store: two-scan SH extract (unique keys, then fuse-hit postings)

CHANGELOG.md

@@ -33,6 +33,34 @@ before 1.0).
 
 ### Changed
 
+- **Pruned SH materialize is two-pass extract:** each collect worker owns a
+  contiguous create-fk span and unsized maps capped at 1.5 GiB
+  (`SH_EXTRACT_WORKER_RAM_BYTES`; 64 B/key pass 1, `80n+8f` pass 2). After
+  each 64 k-fk loc/body batch, spill the largest shard map while over
+  budget (`SHKSP01` under `keys/NN/`, first-fk delta singles). Collect spills
+  are tmp+rename without `sync_all` (no `DONE.keys` still wipes unsorted).
+  Merge folds those spills into one map, one walk to pack8
+  `scripthash.head/NN` (singles `inline_one`, multis Empty; file exists is
+  not pack-done) and fuse8 of dupes to `multi/NN.fuse8` (on disk so other
+  shards do not keep it resident), then unlinks `keys/NN/`. The folded
+  map is consumed into the pack8 records and dropped before fuse8 and BDZ.
+  Pass 2 keeps fuse8 only (no BDZ): same static spans; fuse-hit creates
+  fold into per-worker `key16 → Vec<fk>` maps and spill-largest as
+  `SHPST01` under `post/NN/`. A `post/NN` file, or a spill whose magic is
+  not `SHPST01`, is Corrupt. Pack folds those spills into one map
+  (~0.3–0.5 GiB/shard; a few GiB for 8 workers), then `slot_for_key16` +
+  2+ bodies (grouped by MPHF slot); `len == 1` after fold is `fp_singles`. `DONE.post` is
+  `SHPOST02` last_fk. Keys already unlinked when Class A grows before
+  pack: full recollect (MPHF tags are not key16). All extract phases
+  share `sh_extract_workers()` = min(CPUs, max(1, free RAM / 1.5 GiB));
+  `RBITCOIN_SH_MERGE_WORKERS` still overrides. Collect maps are the
+  1.5 GiB worker cap (spilled and dropped before merge BDZ). Progress is
+  `scanned=` finished fks. Output and hit counters flush once per 64 k-fk
+  batch. Spills share one writer (1-slot queue). One
+  `keys merge start`; live `keys merge shard=` with `fold=` `bdz=`; pack
+  shard lines when each worker finishes. Previous `DONE` / 24 B `NN`
+  unsorted is deleted and pass 1 restarts.
+
 - **PR cargo-mutants is 4 in-diff shards:** `ci.yml` `mutants`, after
   fmt/clippy/test, in parallel with coverage. Advisory
   (`continue-on-error`), not a merge gate. One 30-minute job was canceled

OPERATOR.md

@@ -895,13 +895,23 @@ Tip-follow readiness is **independent** of SH materialize (`tip_follow_ready` 
 
 ### Abort / resume (tip materialize)
 
-Keep **`store/scripthash.unsorted/`** until all shards seal. SIGINT / SIGTERM
-mid-cold keeps every **sealed** `scripthash.head/NN`; restart with the same
+Keep **`store/scripthash.unsorted/`** until all shards seal. Extra disk during
+build is **`SHKSP01` spills** (one rec per unique key per worker map) plus
+**`SHPST01` post spills** (one rec per unique multi key per map, delta fks).
+SIGINT / SIGTERM
+mid-cold keeps every **RAM-published** `scripthash.head/NN`; restart with the same
 `--datadir --sh-index` packs **unsealed** shards only (holes stay). Incomplete
-collect (no `DONE`) restarts the Class A pass. `DONE` names the Class A
-`create_fk` scanned; restart appends new creates into unsorted files when no
+pass 1 (no `DONE.keys`) restarts the first Class A scan. A previous layout
+(`DONE` / 24 B `NN` files, or `keys/NN` / `post/NN` as a file) with no valid
+`DONE.keys` is deleted and pass 1 starts over. A spill whose magic is not
+`SHKSP01` or `SHPST01` is Corrupt — wipe `store/scripthash.unsorted` and
+rematerialize. `DONE.keys` / `DONE.post` name the Class A
+`create_fk` scanned; restart appends new creates when no
 shards are sealed, or tail-appends onto the durable head after pack when any
-`head/NN` already exists. Do not delete unsorted files
+`head/NN` is already published. Extract phases (collect, merge, BDZ, fuse,
+pass 2, pack) share one worker cap (`store: scripthash … workers=`;
+`RBITCOIN_SH_MERGE_WORKERS` override). Do not
+delete unsorted files
 to “start over” unless you intend a full Class A collect
 (`RBITCOIN_SH_FORCE_REBUILD`). Leftover `scripthash.runs` are discarded at tip
 (never k-way rematerialized).
@@ -910,8 +920,8 @@ to “start over” unless you intend a full Class A collect
 |------|-------------------|
 | SIGTERM / SIGINT mid pack | Resume. Sealed `head/NN` stays; unsealed shards re-pack from unsorted files. |
 | Kill-9 mid pack | Same idea; unfinished shard work is redone. Open follows [`docs/crash-recovery.md`](docs/crash-recovery.md) (scripthash Direct). |
-| `DONE` then more Class A, no sealed shards | Append the new fk span into unsorted files, then pack. |
-| `DONE` then more Class A, some/all shards sealed | Pack remaining unsealed files; Class A tail onto the durable head (Direct) or write-behind (Tip). |
+| `DONE.keys` / `DONE.post` then more Class A, no sealed shards | Append the new fk span into keys then postings, then pack. |
+| `DONE.keys` / `DONE.post` then more Class A, some/all shards sealed | Pack remaining unsealed `post/NN`; Class A tail onto the durable head (Direct) or write-behind (Tip). |
 | Empty SH head + leftover catalog | Wipe leftover runs + SEAL, then Class A collect into unsorted shards. |
 | Durable SH head + leftover runs | Discard leftover runs (keep SEAL); write-behind fills HWM lag. |
 | Corrupt SH (leftover live OA, mixed body, refuse line) | Wipe `store/scripthash*` only, keep Class A, rematerialize with `--sh-index`. |

SCHEMA.md

@@ -227,7 +227,10 @@ itself changed.
     scripthash.ovf/NNNNNN[.fuse8][.idx]                  # L0 SHSR pack8
     scripthash.ovf/NNNNNN.mphf|.val|.fuse8               # L1 promoted ovf (at most one)
     scripthash.runs              # leftover catalog (key_len=40); discarded at tip
-    scripthash.unsorted/NN       # tip collect: raw 24 B recs (prefix16+fk), unsorted; DONE=SHUNSRT3+last_fk+counts; unlinked after seal
+    scripthash.unsorted/keys/NN/  # pass 1 dir of SHKSP01 spills (000000…); merge identity-map fold + one walk to head+multi fuse, unlinks; DONE.keys=SHKEYS02 last_fk marker
+    scripthash.unsorted/multi/NN.fuse8  # throwaway fuse8 of 0xFFFF keys for pass 2
+    scripthash.unsorted/post/NN/  # pass 2 dir of SHPST01 spills (000000…); pack folds to one map then slot_for_key16 + 2+ body; DONE.post=SHPOST02 last_fk; unlinked after pack
+    # previous DONE / 24 B NN (SHUNSRT3) with no SHKEYS02 is wiped and pass 1 restarts; no SCHEMA_VERSION bump
     sp_tweaks.idx/  sp_tweaks.body/   # optional BIP-352 (schema 17 dirs; leftover files unlinked)
 
 <datadir-cold>/                  # only when --datadir-cold is set
@@ -699,8 +702,17 @@ compact still merges **heads only** — all ovf keys share
 ### Query join
 
 Heights, value, spentness, vouts: expand from Class A outputs (match full scripthash) + spend annotations + Class C.  
-IBD may stage creates in **unsorted per-shard files** (24 B `{sh_prefix16\|create_fk}`)
-and unique-sort + pack durable SH at tip entry. Leftover schema-16 `key_len=32`
+IBD may stage creates in **two Class A `txout` scans** under `scripthash.unsorted/`:
+workers write `SHKSP01` spill files under `keys/NN/`
+(n_multi key16s already `0`, then first-fk-sorted uleb(delta)‖key16 singles),
+then one map-fold walk per shard into pack8 `scripthash.head/NN` and
+`multi/NN.fuse8` (`DONE.keys` = `SHKEYS02` last_fk marker). A previous
+`DONE` / 24 B `NN` layout with no valid `DONE.keys` is deleted and pass 1
+restarts. A spill whose magic is not `SHKSP01` is Corrupt — wipe
+`scripthash.unsorted` and rematerialize. Fuse-hit `SHPST01` post spills
+under `post/NN/` (`DONE.post` = `SHPOST02` last_fk). A `post/NN` file, or a
+spill whose magic is not `SHPST01`, is Corrupt. Pack folds those spills, then
+rewrites 2+ into body and skips 1-fk fuse FPs. Leftover schema-16 `key_len=32`
 catalogs are refused.
 
 **Decision:** inline for 1-use scripts (`SH_INLINE_CAP = 1`, ~95 % of keys); geometric slabs for

crates/rbitcoin-query/src/catchup.rs

@@ -860,7 +860,10 @@ mod tests {
         let done_last = rbitcoin_store::unsorted_done_last_fk(&udir, n_shards).unwrap();
         extend_direct_chain(&q, 2);
         assert!(q.store.txs.count() > done_last);
-        assert!(!q.store.scripthash.has_durable_index());
+        assert!(
+            !q.store.scripthash.unsealed_main_shards().is_empty(),
+            "RAM merge writes head files; pack seal is later"
+        );
         let _ = q.finalize_sh_runs().unwrap();
         let sh_old = rbitcoin_store::script_hash(&[0x51, 0]);
         let sh_new = rbitcoin_store::script_hash(&[0x51, 3]);

crates/rbitcoin-query/src/sh_builder.rs

@@ -1,5 +1,5 @@
-//! Post-IBD scripthash collect: one Class A pass into unsorted per-shard
-//! files, then in-place unique-sort + seal.
+//! Post-IBD scripthash collect: two Class A `txout` scans (unique key16,
+//! then fuse-hit postings), then pack + seal.
 //!
 //! Direct confirm does **not** enqueue SH. A durable head never enters this
 //! path (write-behind / `recover_sh_writebehind` instead). Leftover

crates/rbitcoin-store/src/file.rs

@@ -54,9 +54,16 @@ pub(crate) fn tmp_sidecar_path(path: &Path) -> PathBuf {
     }
 }
 
+/// Write `bytes` to a sibling tmp, then rename over `path`. No `sync_all`.
+pub(crate) fn write_tmp_rename(path: &Path, bytes: &[u8]) -> Result<(), StoreError> {
+    write_tmp_file(path, false, |f| {
+        f.write_all(bytes).map_err(|e| StoreError::io(path, e))
+    })
+}
+
 /// Write `bytes` to a sibling tmp, `sync_all`, then rename over `path`.
 pub(crate) fn write_synced_tmp_rename(path: &Path, bytes: &[u8]) -> Result<(), StoreError> {
-    write_synced_tmp_file(path, |f| {
+    write_tmp_file(path, true, |f| {
         f.write_all(bytes).map_err(|e| StoreError::io(path, e))
     })
 }
@@ -65,6 +72,13 @@ pub(crate) fn write_synced_tmp_rename(path: &Path, bytes: &[u8]) -> Result<(), S
 ///
 /// The durable name is not created or truncated until rename.
 pub(crate) fn write_synced_tmp_file<F>(path: &Path, write: F) -> Result<(), StoreError>
+where
+    F: FnOnce(&mut File) -> Result<(), StoreError>,
+{
+    write_tmp_file(path, true, write)
+}
+
+fn write_tmp_file<F>(path: &Path, sync: bool, write: F) -> Result<(), StoreError>
 where
     F: FnOnce(&mut File) -> Result<(), StoreError>,
 {
@@ -77,7 +91,9 @@ where
     {
         let mut f = File::create(&tmp).map_err(|e| StoreError::io(&tmp, e))?;
         write(&mut f)?;
-        f.sync_all().map_err(|e| StoreError::io(&tmp, e))?;
+        if sync {
+            f.sync_all().map_err(|e| StoreError::io(&tmp, e))?;
+        }
     }
     std::fs::rename(&tmp, path).map_err(|e| StoreError::io(path, e))?;
     Ok(())
@@ -1410,4 +1426,21 @@ mod advise_tests {
         assert_eq!(std::fs::read(&dest).unwrap(), b"sealed");
         let _ = std::fs::remove_dir_all(&dir);
     }
+
+    #[test]
+    fn write_tmp_rename_installs_dest_only() {
+        static N: AtomicU64 = AtomicU64::new(0);
+        let id = N.fetch_add(1, Ordering::Relaxed);
+        let dir = std::env::temp_dir().join(format!("rbitcoin-tmp-rename-{id}"));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        let dest = dir.join("spill.bin");
+        let tmp = tmp_sidecar_path(&dest);
+        assert!(!dest.exists());
+        write_tmp_rename(&dest, b"spill").unwrap();
+        assert!(dest.exists());
+        assert!(!tmp.exists());
+        assert_eq!(std::fs::read(&dest).unwrap(), b"spill");
+        let _ = std::fs::remove_dir_all(&dir);
+    }
 }