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

CHANGELOG.md

@@ -11,6 +11,11 @@ before 1.0).
 
 ### Fixed
 
+- **SH collect status counts finished fks:** static worker spans made `fk=`
+  a max, so a late span looked done. `scanned=` is the sum. End-of-pass
+  spills share one writer (1-slot queue) so the 1.5 GiB flushes do not hit
+  disk together. A `flush span=` line marks that stall.
+
 - **Orphan parent GETDATA flush cap is 100:** Core
   `MAX_PEER_TX_REQUEST_IN_FLIGHT`. A 25-tx ancestor package's missing
   parents fit one GetData (was 16).
@@ -33,6 +38,32 @@ 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; leftover
+  `keys/NN` file / other magics refuse). 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/`.
+  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/`. Leftover `post/NN` file / 18 B windows
+  refuse. 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; `len == 1` after
+  fold is `fp_singles`. `DONE.post` stays `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 (dropped before merge BDZ). Start logs include
+  `workers=` `budget_MiB=`. Progress is `scanned=` finished fks. 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.
+
 - **Weekly cargo-mutants:** `mutants.yml` — PR `--workspace --in-diff`
   advisory (`continue-on-error`); Sunday 8-shard `--workspace` sweep.
   Must use `--workspace` (`default-members` is node). Snapshot lists:

OPERATOR.md

@@ -895,13 +895,22 @@ 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) —
+not 18 B × every fuse-hit create. Leftover `post/NN` file from older binaries
+**refuse** — wipe `store/scripthash.unsorted` and rematerialize. 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. Leftover in-process
+keys from older binaries (`keys/NN` file / `SHKEYU02` / framed 18 B / `SHUNSRT3`)
+**refuse** — 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 +919,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

@@ -224,7 +224,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; leftover keys/NN file / other magics refuse
+    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; leftover post/NN file / 18 B windows refuse; DONE.post=SHPOST02 last_fk; unlinked after pack
+    # leftover scripthash.unsorted/mphf/ and 24 B NN (SHUNSRT3) are wiped; 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
@@ -726,8 +729,16 @@ 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; leftover
+`keys/NN` file / `SHKEYU02` / framed 18 B / 16 B lists / wrong magic
+**refuse** — wipe `scripthash.unsorted` / rematerialize), then fuse-hit `SHPST01` post spills under `post/NN/`
+(`DONE.post` = `SHPOST02` last_fk; leftover `post/NN` file / 18 B windows refuse). Pack folds those spills, then
+rewrites 2+ into body and skips 1-fk fuse FPs. Leftover 24 B `NN` /
+`SHUNSRT3` / `SHKEYS01` refuse pass-1 keys bodies (wipe unsorted). 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 windowed 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);
+    }
 }

crates/rbitcoin-store/src/scripthash.rs

@@ -390,7 +390,7 @@
     matches!(f.read_exact(&mut magic), Ok(())) && magic == *b"SHSR"
 }
 
-fn sorted_main_shard_path(dir: &Path, shard: usize, n_shards: usize) -> PathBuf {
+pub(crate) fn sorted_main_shard_path(dir: &Path, shard: usize, n_shards: usize) -> PathBuf {
     let p = dir.join("scripthash.head");
     if n_shards <= 1 && p.is_file() {
         p
@@ -399,6 +399,18 @@
     }
 }
 
+/// Dummy 32 B key in `shard` whose bytes 4..8 encode `slot` (postings pack grouping).
+fn pack_key_for_slot(shard: usize, n_shards: usize, slot: u32) -> [u8; 32] {
+    let mut k = [0u8; 32];
+    let n = n_shards.max(1);
+    if n > 1 {
+        let bits = n.trailing_zeros();
+        k[0] = (shard as u8) << (8 - bits as u8);
+    }
+    k[4..8].copy_from_slice(&slot.to_le_bytes());
+    k
+}
+
 fn open_sorted_main_shards(
     dir: &Path,
     n_shards: usize,
@@ -2499,6 +2511,7 @@
             resume_from_shard: 0,
             active_shard: None,
             recs: Vec::new(),
+            slot_words: Vec::new(),
             key_budget,
             body_buf: Vec::with_capacity(BULK_BODY_FLUSH),
             body_write_off: bump,
@@ -2547,6 +2560,7 @@
             resume_from_shard: 0,
             active_shard: Some(shard),
             recs: Vec::new(),
+            slot_words: Vec::new(),
             key_budget,
             body_buf: Vec::with_capacity(BULK_BODY_FLUSH),
             body_write_off: bump,
@@ -2609,6 +2623,7 @@
             resume_from_shard: progress.next_shard,
             active_shard: None,
             recs: Vec::new(),
+            slot_words: Vec::new(),
             key_budget,
             body_buf: Vec::with_capacity(BULK_BODY_FLUSH),
             body_write_off: bump,
@@ -2638,6 +2653,18 @@
         self.allocs[0].lock().unwrap().bump
     }
 
+    /// Pass-1 BDZ singles already inline in `head/NN` (not yet packed body).
+    pub(crate) fn set_extract_inline_creates(
+        &self,
+        shard: usize,
+        n: u64,
+    ) -> Result<(), StoreError> {
+        let body = self.shard_body(shard);
+        let mut g = self.shard_alloc(shard).lock().unwrap();
+        g.live_count = n;
+        write_alloc_header(body, &g)
+    }
+
     /// Seal `recs` as sorted main shard `shard` and publish alloc HWM.
     pub fn publish_sorted_shard(
         &self,
@@ -2683,7 +2710,15 @@
         if let Some(parent) = path.parent() {
             let _ = std::fs::create_dir_all(parent);
         }
-        let sealed = MphfHead::write_pack8(&path, &pack.recs)?;
+        let sealed = if MphfHead::exists(&path) {
+            let h = MphfHead::open(&path)?;
+            if !pack.slot_words.is_empty() {
+                h.rewrite_val_slots(&pack.slot_words)?;
+            }
+            h
+        } else {
+            MphfHead::write_pack8(&path, &pack.recs)?
+        };
         self.install_sorted_main(shard, sealed);
         let body = self.shard_body(shard);
         if bump > body.logical_len() {
@@ -2725,6 +2760,8 @@
     active_shard: Option<usize>,
     /// Packed recs for [`Self::bulk_session`] (sorted at shard seal; 16 B key order).
     recs: Vec<(crate::scripthash_layout::ShHeadKey, u64)>,
+    /// MPHF slot + pack8 when packing from windowed postings (pass 2).
+    slot_words: Vec<(u32, u64)>,
     /// Unique-key budget (log / tests). Does not pre-size an OA table.
     key_budget: u64,
     /// Sequential slab bytes; flushed at [`BULK_BODY_FLUSH`] or before a
@@ -2755,17 +2792,21 @@
 /// One shard packed onto its live body, ready for ordered head publish.
 pub struct ShShardPack {
     pub recs: Vec<(crate::scripthash_layout::ShHeadKey, u64)>,
+    pub slot_words: Vec<(u32, u64)>,
     pub creates: u64,
     pub max_fk: u64,
     pub keys: u64,
     pub bump: u64,
     pub body_flush_ns: u64,
     pub pack_ns: u64,
+    /// Fuse-hit postings with one unique create_fk (left inline; no body).
+    pub fp_singles: u64,
 }
 
 /// One unfinished key in [`ScriptHashBulkSession`] (≤ one delta page of FKs).
 struct BulkOpenKey {
     key: [u8; 32],
+    slot: Option<u32>,
     buf: Vec<u64>,
     stream_used: usize,
     n_total: u32,
@@ -2875,16 +2916,18 @@
         }
         self.finish_key()?;
         self.flush_body()?;
-        let persist_live = self.live_count;
+        let persist_live = self.committed_live_count.saturating_add(self.live_count);
         self.persist_session_alloc(persist_live, self.bump)?;
         let pack = ShShardPack {
             recs: std::mem::take(&mut self.recs),
+            slot_words: std::mem::take(&mut self.slot_words),
             creates: self.live_count,
             max_fk: self.max_fk,
             keys: self.keys_written,
             bump: self.bump,
             body_flush_ns: self.body_flush_ns,
             pack_ns: self.pack_ns,
+            fp_singles: 0,
         };
         self.finished = true;
         Ok(pack)
@@ -2909,6 +2952,7 @@
             let buf = self.take_fk_scratch();
             self.open_key = Some(BulkOpenKey {
                 key,
+                slot: None,
                 buf,
                 stream_used: 0,
                 n_total: 0,
@@ -2959,6 +3003,22 @@
         Ok(())
     }
 
+    /// Stream one create_fk for MPHF `slot` (postings pack; keys grouped by slot).
+    pub fn push_sorted_slot_fk(&mut self, slot: u32, fk: Fk) -> Result<(), StoreError> {
+        if fk.is_null() {
+            return Ok(());
+        }
+        let si = self.active_shard.ok_or(StoreError::Corrupt(
+            "scripthash pack slot stream requires pack_shard_session",
+        ))?;
+        let key = pack_key_for_slot(si, self.table.head_shard_count(), slot);
+        self.push_sorted_fk(key, fk)?;
+        if let Some(open) = self.open_key.as_mut() {
+            open.slot = Some(slot);
+        }
+        Ok(())
+    }
+
     /// Seal the open key (inline / slab / last page).
     pub fn finish_key(&mut self) -> Result<(), StoreError> {
         if self.open_key.is_none() {
@@ -3015,6 +3075,9 @@
         self.live_count = self.live_count.saturating_add(u64::from(n));
         self.keys_written = self.keys_written.saturating_add(1);
         let rec = (head_key_from_full(&open.key), pack8(&val)?);
+        if let Some(slot) = open.slot {
+            self.slot_words.push((slot, rec.1));
+        }
         self.recs.push(rec);
         self.peak_table_bytes = self
             .peak_table_bytes