diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bdd20bd..2ff163ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,15 @@ before 1.0). ### Changed +- **Confirm Class A kind per batch:** a load/write batch is all need-body + (`plan=Some`) or all already-bodied (`plan=None`). Lookup splits loadq + at `header_txs.has_body`; write drain stops on plan polarity. Mixed + stamp is `Corrupt("invariant: confirm batch mixed archived")`. Write + vs tip is all-old (`Ok([])`), all-new (fill zip), or + `Corrupt("invariant: write batch spans tip")` — no prefix strip. + [`docs/invariants.md`](docs/invariants.md), + [`docs/concurrency.md`](docs/concurrency.md). + - **Pin `merge_outs` no-op Arc:** empty / already-covered `checked`+`live` keep the outs Arc (assemble sticky `ptr_eq`). RCU compose borrows `live` instead of cloning script bytes on every retry. diff --git a/crates/rbitcoin-consensus/src/confirm_run/lookup.rs b/crates/rbitcoin-consensus/src/confirm_run/lookup.rs index 8f9c8460..5de7165c 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/lookup.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/lookup.rs @@ -409,6 +409,7 @@ pub(super) fn wire_lookup_phase( let need_fks = query .archive_filter_need_header_fks(&header_fks) .map_err(ConsensusError::from)?; + confirm_archive_kind(header_fks.len(), need_fks.len())?; let filter_ns = t_filter.elapsed().as_nanos() as u64; let t_batch = Instant::now(); let plan = if need_fks.is_empty() { @@ -468,16 +469,6 @@ pub(super) fn wire_lookup_phase( m.tx_fks = fks.clone(); } } - if m.tx_fks.is_empty() { - if let Some(list) = query - .store() - .header_txs - .get_list(m.header_fk) - .map_err(ConsensusError::from)? - { - m.tx_fks = list; - } - } let prev = wire_blocks[i].header.prev_blockhash.to_byte_array(); query.confirm_parent_cache().put_header_plan( m.height.0, @@ -496,6 +487,27 @@ pub(super) fn wire_lookup_phase( Ok((plan, metas, wire_blocks, plan_ns)) } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum ConfirmArchiveKind { + AllNeedBody, + AllHaveBody, +} + +pub(super) fn confirm_archive_kind( + n_headers: usize, + n_need: usize, +) -> Result { + if n_need == 0 { + Ok(ConfirmArchiveKind::AllHaveBody) + } else if n_need == n_headers { + Ok(ConfirmArchiveKind::AllNeedBody) + } else { + Err(ConsensusError::Store(StoreError::Corrupt( + "invariant: confirm batch mixed archived", + ))) + } +} + pub(super) fn create_fks_from_header_ranges( per_header_ranges: &[(rbitcoin_primitives::Fk, rbitcoin_primitives::Fk, u32)], ) -> U64Map> { diff --git a/crates/rbitcoin-consensus/src/confirm_run/mod.rs b/crates/rbitcoin-consensus/src/confirm_run/mod.rs index b5ddcd11..c074f566 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/mod.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/mod.rs @@ -60,11 +60,16 @@ pub use bq_resolve::{ use head_drain::{submit_head_drain, HEAD_DRAIN_THREAD_NAME}; pub use lookup::lookup_stage_stats; pub use lookup::plan_stamp_sub_stats; +#[cfg(test)] +use lookup::ConfirmArchiveKind; +use lookup::{ + confirm_archive_kind, create_fks_from_header_ranges, known_create_txid_lookup, + stamp_parent_pin_archived, +}; pub use lookup::{ confirm_wire_load_from_plan, confirm_wire_lookup_stamp, DenserelsWarmStats, ParentPinStamp, PlanStampOutcome, }; -use lookup::{create_fks_from_header_ranges, known_create_txid_lookup, stamp_parent_pin_archived}; use phases::assemble_run; #[cfg(test)] use phases::{check_bip34, expected_bits_extending, post_commit}; @@ -78,7 +83,10 @@ pub use scripts::{ }; pub use write::confirm_write_phase; #[cfg(test)] -use write::{recent_create_height_slices, recent_create_rows_for_slices, write_height_needed}; +use write::{ + recent_create_height_slices, recent_create_rows_for_slices, write_batch_vs_tip, + write_height_needed, WriteBatchVsTip, +}; /// Pure-write annotate backend from global `RBITCOIN_IO`. #[inline] @@ -359,6 +367,7 @@ pub fn confirm_wire_load_phase_pipelined( let need_fks = query .archive_filter_need_header_fks(&header_fks) .map_err(ConsensusError::from)?; + confirm_archive_kind(header_fks.len(), need_fks.len())?; let mut plan = if need_fks.is_empty() { for (i, m) in metas.iter_mut().enumerate() { if let Some(list) = query @@ -416,16 +425,6 @@ pub fn confirm_wire_load_phase_pipelined( m.tx_fks = fks.clone(); } } - if m.tx_fks.is_empty() { - if let Some(list) = query - .store() - .header_txs - .get_list(m.header_fk) - .map_err(ConsensusError::from)? - { - m.tx_fks = list; - } - } let prev = wire_blocks[i].header.prev_blockhash.to_byte_array(); query.confirm_parent_cache().put_header_plan( m.height.0, @@ -594,8 +593,9 @@ impl ScriptOkBatch { /// /// Scripts enqueue height-ordered tip extensions; write drains the channel /// and merges so Class A + Class C + annotate run once (fewer tip fsyncs). - /// Returns `Err(other)` if not a contiguous height extension (caller keeps - /// `other` for the next batch). + /// Returns `Err(other)` if not a contiguous height extension **or** if + /// `archive_plan` polarity differs (`Some` vs `None`). Caller writes the + /// prefix then keeps `other` for the next meta-batch. pub fn append_contiguous(&mut self, mut other: Self) -> Result<(), Self> { if other.is_empty() { return Ok(()); @@ -619,13 +619,14 @@ impl ScriptOkBatch { { return Err(other); } + if self.archive_plan.is_some() != other.archive_plan.is_some() { + return Err(other); + } self.prepared.append(&mut other.prepared); self.wire_blocks.append(&mut other.wire_blocks); self.batch_parents.extend_from(other.batch_parents); - match (self.archive_plan.as_mut(), other.archive_plan.take()) { - (Some(dst), Some(src)) => dst.append(src), - (None, Some(src)) => self.archive_plan = Some(src), - _ => {} + if let (Some(dst), Some(src)) = (self.archive_plan.as_mut(), other.archive_plan.take()) { + dst.append(src); } Ok(()) } diff --git a/crates/rbitcoin-consensus/src/confirm_run/write.rs b/crates/rbitcoin-consensus/src/confirm_run/write.rs index 745d0ee8..f00d4e1c 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/write.rs @@ -61,6 +61,36 @@ pub(super) fn write_height_needed(tip: Option, height: u32) -> bool { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum WriteBatchVsTip { + AllOld, + AllNew, + SpansTip, +} + +pub(super) fn write_batch_vs_tip( + tip: Option, + heights: impl IntoIterator, +) -> WriteBatchVsTip { + let mut any_old = false; + let mut any_new = false; + for h in heights { + if write_height_needed(tip, h) { + any_new = true; + } else { + any_old = true; + } + if any_old && any_new { + return WriteBatchVsTip::SpansTip; + } + } + if any_new { + WriteBatchVsTip::AllNew + } else { + WriteBatchVsTip::AllOld + } +} + /// Per-height ranges over a write batch's `planned_fks` / pins. /// /// One RecentCreates fifo row per prepared height. A leftover tail (count @@ -140,26 +170,16 @@ pub fn confirm_write_phase( milestone: Milestone, mut batch: ScriptOkBatch, ) -> Result, ConsensusError> { - // Idempotent: skip heights already on the confirmed tip (dup pipeline race). let tip = query.tip_height().map(|h| h.0); - let mut kept = Vec::with_capacity(batch.prepared.len()); - let mut wires = Vec::with_capacity(batch.wire_blocks.len()); - for (p, w) in batch - .prepared - .into_iter() - .zip(batch.wire_blocks.into_iter()) - { - if !write_height_needed(tip, p.height.0) { - continue; + match write_batch_vs_tip(tip, batch.prepared.iter().map(|p| p.height.0)) { + WriteBatchVsTip::AllOld => return Ok(Vec::new()), + WriteBatchVsTip::SpansTip => { + return Err(ConsensusError::Store(rbitcoin_store::StoreError::Corrupt( + "invariant: write batch spans tip", + ))); } - kept.push(p); - wires.push(w); - } - if kept.is_empty() { - return Ok(Vec::new()); + WriteBatchVsTip::AllNew => {} } - batch.prepared = kept; - batch.wire_blocks = wires; let t_wall = Instant::now(); diff --git a/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs b/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs index 5fdea53f..75d915b1 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs @@ -1,6 +1,9 @@ //! Confirm_run unit tests (peeled from confirm_run.rs). -use super::{recent_create_height_slices, recent_create_rows_for_slices, write_height_needed}; +use super::{ + confirm_archive_kind, recent_create_height_slices, recent_create_rows_for_slices, + write_batch_vs_tip, write_height_needed, ConfirmArchiveKind, WriteBatchVsTip, +}; #[test] fn tx_head_drain_thread_is_named_and_reused() { @@ -163,7 +166,7 @@ fn script_ok_append_contiguous_and_gap() { let err = good.append_contiguous(bad).err().expect("len mismatch"); assert_eq!(err.len(), 1); - // archive_plan merge: None + Some, Some + Some, Some + None. + // archive_plan merge: Some+Some concatenates; mixed polarity is leftover. let mut with_plan = batch_one(70); with_plan.archive_plan = Some(rbitcoin_query::ArchiveWritePlan::empty()); let mut next = batch_one(71); @@ -171,35 +174,54 @@ fn script_ok_append_contiguous_and_gap() { assert!(with_plan.append_contiguous(next).is_ok()); assert!(with_plan.archive_plan.is_some()); let mut only_other = batch_one(72); - // Self plan remains Some; other None keeps it. only_other.archive_plan = None; - assert!(with_plan.append_contiguous(only_other).is_ok()); + let err = with_plan + .append_contiguous(only_other) + .err() + .expect("Some+None polarity"); + assert_eq!(err.len(), 1); + assert_eq!(with_plan.len(), 2); assert!(with_plan.archive_plan.is_some()); - // Self None absorbs other's plan. let mut no_plan = batch_one(80); let mut has = batch_one(81); has.archive_plan = Some(rbitcoin_query::ArchiveWritePlan::empty()); - assert!(no_plan.append_contiguous(has).is_ok()); - assert!(no_plan.archive_plan.is_some()); + let err = no_plan + .append_contiguous(has) + .err() + .expect("None+Some polarity"); + assert_eq!(err.len(), 1); + assert_eq!(no_plan.len(), 1); + assert!(no_plan.archive_plan.is_none()); + let n2 = batch_one(81); + assert!(no_plan.append_contiguous(n2).is_ok()); + assert_eq!(no_plan.len(), 2); + assert!(no_plan.archive_plan.is_none()); } -/// Heights at or below tip must be stripped before structural write -/// (dup pipeline race after scripts claim the same tip+1 twice). -/// Write filter + stage entry points + empty scripts purity (one surface). +/// Write vs tip is all-old (no-op), all-new (proceed), or spans tip (Corrupt). /// External three-stage path: rbitcoin-test three_stage_confirm_and_parent_pin_surface. #[test] fn three_stage_write_filter_and_scripts_surface() { let tip = Some(100u32); - let heights = [98u32, 99, 100, 101, 102]; - let kept: Vec = heights - .into_iter() - .filter(|&h| write_height_needed(tip, h)) - .collect(); - assert_eq!(kept, vec![101, 102]); + assert_eq!( + write_batch_vs_tip(tip, [98u32, 99, 100, 101, 102]), + WriteBatchVsTip::SpansTip + ); + assert_eq!( + write_batch_vs_tip(tip, [98u32, 99, 100]), + WriteBatchVsTip::AllOld + ); + assert_eq!( + write_batch_vs_tip(tip, [101u32, 102]), + WriteBatchVsTip::AllNew + ); + assert_eq!( + write_batch_vs_tip(tip, std::iter::empty()), + WriteBatchVsTip::AllOld + ); assert!(!write_height_needed(tip, 100)); assert!(!write_height_needed(Some(0), 0)); assert!(write_height_needed(Some(0), 1)); - // Empty chain: genesis (and all heights) still need write. assert!(write_height_needed(None, 0)); assert!(write_height_needed(None, 1)); @@ -225,6 +247,35 @@ fn three_stage_write_filter_and_scripts_surface() { assert!(ok.batch.wire_blocks.is_empty()); } +#[test] +fn confirm_archive_kind_refuses_mixed() { + assert_eq!( + confirm_archive_kind(3, 0).unwrap(), + ConfirmArchiveKind::AllHaveBody + ); + assert_eq!( + confirm_archive_kind(3, 3).unwrap(), + ConfirmArchiveKind::AllNeedBody + ); + assert_eq!( + confirm_archive_kind(1, 0).unwrap(), + ConfirmArchiveKind::AllHaveBody + ); + assert_eq!( + confirm_archive_kind(1, 1).unwrap(), + ConfirmArchiveKind::AllNeedBody + ); + let err = confirm_archive_kind(3, 2).unwrap_err(); + match err { + crate::error::ConsensusError::Store(rbitcoin_store::StoreError::Corrupt(m)) => { + assert_eq!(m, "invariant: confirm batch mixed archived"); + } + other => panic!("expected mixed archived, got {other:?}"), + } + assert!(confirm_archive_kind(2, 1).is_err()); + assert!(confirm_archive_kind(2, 3).is_err()); +} + fn empty_loaded_batch() -> super::LoadedBatch { super::LoadedBatch { prepared: Vec::new(), @@ -2519,6 +2570,33 @@ fn store_start_states_lookup_load_confirm() { "idempotent Class A skip must not bump class_a_hi" ); + // One-shot mixed need-body + already-bodied must fail closed (split into two calls). + let h_have = h_s1 + 1; + let b_have = mine_cb(b_s1.block_hash(), b_s1.header.time + 600, h_have); + let (header_have, txs_have) = prepare_block_for_archive(&q, ¶ms, &b_have).unwrap(); + q.commit_class_a_only(&header_have, &txs_have).unwrap(); + let h_need = h_have + 1; + let b_need = mine_cb(b_have.block_hash(), b_have.header.time + 600, h_need); + let mixed_arcs = [ + (Height(h_have), Arc::new(b_have.clone()), None), + (Height(h_need), Arc::new(b_need.clone()), None), + ]; + match confirm_wire_lookup_stamp(&q, ¶ms, ms, &mixed_arcs, None) { + Err(crate::error::ConsensusError::Store(rbitcoin_store::StoreError::Corrupt(m))) => { + assert_eq!(m, "invariant: confirm batch mixed archived"); + } + Ok(_) => panic!("mixed lookup stamp must fail closed"), + Err(other) => panic!("expected mixed archived lookup, got {other:?}"), + } + let mixed_run = [(Height(h_have), b_have), (Height(h_need), b_need)]; + match super::confirm_wire_load_phase(&q, ¶ms, ms, &mixed_run, &ScriptPreverified::new()) { + Err(crate::error::ConsensusError::Store(rbitcoin_store::StoreError::Corrupt(m))) => { + assert_eq!(m, "invariant: confirm batch mixed archived"); + } + Ok(_) => panic!("mixed one-shot load must fail closed"), + Err(other) => panic!("expected mixed archived load, got {other:?}"), + } + let _ = std::fs::remove_dir_all(&path); } diff --git a/crates/rbitcoin-net/src/ibd/confirm/mod.rs b/crates/rbitcoin-net/src/ibd/confirm/mod.rs index 7db8f136..cabde5f6 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/mod.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/mod.rs @@ -391,21 +391,32 @@ pub(crate) fn load_stamp_items( .collect() } -/// Split a lookup-wave input-count run into load-sized batch lengths. +/// Split a lookup-wave into load-sized batch lengths. /// -/// Uses the same [`pack_stop_after`] rule as load pack (soft 8000 / hard 144). -pub(crate) fn split_wave_into_load_batches( +/// Stops on [`pack_stop_after`] (soft 8000 / hard 144) and when `has_body` flips. +/// `has_body` is per height, same order as `input_counts`. Empty skips the kind split. +pub(crate) fn split_wave_into_load_batches_kind( input_counts: &[u32], + has_body: &[bool], soft_max_inputs: u32, hard_max_blocks: usize, ) -> Vec { + debug_assert!(has_body.is_empty() || has_body.len() == input_counts.len()); let mut out = Vec::new(); let mut i = 0usize; while i < input_counts.len() { let rest = &input_counts[i..]; + let kind0 = has_body.get(i).copied(); let mut sum = 0u32; let mut n = 0usize; - for &c in rest { + for (j, &c) in rest.iter().enumerate() { + if j > 0 { + if let (Some(k0), Some(&k)) = (kind0, has_body.get(i + j)) { + if k != k0 { + break; + } + } + } sum = sum.saturating_add(c); n += 1; if pack_stop_after(sum, n, soft_max_inputs, hard_max_blocks) { @@ -860,11 +871,19 @@ fn drain_script_ok_write_queue( parts = parts.saturating_add(1); } Err(leftover) => { - // Height gap / leftover union — write the contiguous prefix first. - warn!( - "ibd: write batch drain gap after parts={parts} leftover_blks={}", - leftover.len() - ); + let polarity = + batch.archive_plan.is_some() != leftover.archive_plan.is_some(); + if polarity { + warn!( + "ibd: write batch drain plan polarity after parts={parts} leftover_blks={}", + leftover.len() + ); + } else { + warn!( + "ibd: write batch drain gap after parts={parts} leftover_blks={}", + leftover.len() + ); + } return (batch, parts, Some(leftover)); } } @@ -1664,8 +1683,24 @@ pub(crate) fn spawn_confirm_engine( .iter() .map(|(_, _, w)| block_input_count(w.block.as_ref())) .collect(); - let parts = split_wave_into_load_batches( + let t_kind = Instant::now(); + let kinds: Vec = match wave + .items + .iter() + .map(|(_, hash, _)| hub.query.is_block_archived(hash)) + .collect::, _>>() + { + Ok(k) => k, + Err(e) => { + warn!("ibd: load-batch has_body probe: {e}"); + confirm_thr_stats::add_lookup_other(t_kind.elapsed()); + continue; + } + }; + confirm_thr_stats::add_lookup_other(t_kind.elapsed()); + let parts = split_wave_into_load_batches_kind( &counts, + &kinds, confirm_batch_max_inputs(), CONFIRM_RUN_MAX_BLOCKS, ); diff --git a/crates/rbitcoin-net/src/ibd/confirm/tests.rs b/crates/rbitcoin-net/src/ibd/confirm/tests.rs index 18f96000..8dc47157 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/tests.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/tests.rs @@ -296,7 +296,7 @@ fn pack_confirm_run_len_policy() { #[test] fn split_wave_into_load_batches_is_eight_by_8000() { use super::{ - split_wave_into_load_batches, CONFIRM_BATCH_INPUTS_DEFAULT, CONFIRM_RUN_MAX_BLOCKS, + split_wave_into_load_batches_kind, CONFIRM_BATCH_INPUTS_DEFAULT, CONFIRM_RUN_MAX_BLOCKS, LOAD_QUEUE_CAP_DEFAULT, }; assert_eq!(LOAD_QUEUE_CAP_DEFAULT, 14); @@ -305,25 +305,59 @@ fn split_wave_into_load_batches_is_eight_by_8000() { assert!(super::LoadBatch { items: vec![] }.items.is_empty()); // 8 × 8001 inputs (each block overshoots 8000) → 8 batches of one. let wave: Vec = vec![8001; 8]; - let parts = - split_wave_into_load_batches(&wave, CONFIRM_BATCH_INPUTS_DEFAULT, CONFIRM_RUN_MAX_BLOCKS); + let parts = split_wave_into_load_batches_kind( + &wave, + &[], + CONFIRM_BATCH_INPUTS_DEFAULT, + CONFIRM_RUN_MAX_BLOCKS, + ); assert_eq!(parts, vec![1, 1, 1, 1, 1, 1, 1, 1]); // Exactly 8000 does not stop; two 8000-input blocks are one batch. assert_eq!( - split_wave_into_load_batches(&[8000, 8000], 8000, 144), + split_wave_into_load_batches_kind(&[8000, 8000], &[], 8000, 144), vec![2] ); // Empty / single megablock. - assert!(split_wave_into_load_batches(&[], 8000, 144).is_empty()); - assert_eq!(split_wave_into_load_batches(&[50_000], 8000, 144), vec![1]); + assert!(split_wave_into_load_batches_kind(&[], &[], 8000, 144).is_empty()); + assert_eq!( + split_wave_into_load_batches_kind(&[50_000], &[], 8000, 144), + vec![1] + ); // 144 thin blocks then 144 more → two hard-cap batches. let thin = vec![1u32; 288]; assert_eq!( - split_wave_into_load_batches(&thin, 8000, 144), + split_wave_into_load_batches_kind(&thin, &[], 8000, 144), vec![144, 144] ); } +#[test] +fn split_wave_into_load_batches_stops_at_has_body_change() { + use super::split_wave_into_load_batches_kind; + // Crash prefix already-bodied, suffix need-body: two batches. + let counts = [1u32, 1, 1, 1, 1]; + let has_body = [true, true, false, false, false]; + assert_eq!( + split_wave_into_load_batches_kind(&counts, &has_body, 8000, 144), + vec![2, 3] + ); + // Kind flip inside an 8000-input pack still splits (do not glue kinds). + assert_eq!( + split_wave_into_load_batches_kind(&[4000, 4000], &[true, false], 8000, 144), + vec![1, 1] + ); + // Homogeneous still packs on input cap only. + assert_eq!( + split_wave_into_load_batches_kind(&[8000, 8000], &[false, false], 8000, 144), + vec![2] + ); + assert!(split_wave_into_load_batches_kind(&[], &[], 8000, 144).is_empty()); + assert_eq!( + split_wave_into_load_batches_kind(&[50_000], &[true], 8000, 144), + vec![1] + ); +} + #[test] fn load_recv_is_lookup_order() { use super::LoadBatch; diff --git a/docs/concurrency.md b/docs/concurrency.md index 333ad29e..a3fe1165 100644 --- a/docs/concurrency.md +++ b/docs/concurrency.md @@ -17,11 +17,11 @@ Short map of who may write which tables. **Format is unstable until 1.0.** **Height-ordered unified pipeline (current):** peer → **body queue** (raw only) → **lookup** (in-order from `max(path_lo, lookup_taken_hi+1)`; decode + TipOnly `head_fk`; **dequeue** raw into `loadq=14`). Hole/densify/receive: in-hand = confirmed ∨ BQ hash ∨ `H ≤ lookup_taken_hi` → **load** (recv load-sized batch; stamp + pin + assemble) → scripts → write. Stage IO: [`invariants.md`](./invariants.md). **No** peer→confirm-feed wire retain. **No** hash-only / Class-A-only confirm (bq wire required). Load bind order is in-flight → published `live_union` chain → TipOnly (fence-connected). Lookup owns `live_union`; one `ArcSwap` of the layer-chain head per wave; get walks layers (no union rebuild); a layer stays while any height in its span is on the BQ or overlaps `(tip, taken_hi]` (taken onto loadq); disconnect stores None. Write drain inserts `tx.head` in parallel with Class C on the process-wide `ibd-confirm-head` thread. Drain complete is max inserted **fk** (not tip/fence — those advance during drain). Header-cache GC polls store tip every load pack. In-flight prune is **after pin + scripts handoff**: drain-fk **and** `covers_fk_span` (TipOnly home — `docs/invariants.md`). No leftover pending map. Bodies without a known height are marked missing and re-getdata after the height map is ready — there is **no** dual-track archive-job / ContigPark fallback. Load pack **waits** on `feed.cv` when tip+1 is in `ready` but the BQ is not resolve-complete (no retain/BQ spin). Pack takes the feed mutex to collect candidates and again to mark inflight; one BQ `pack_snapshot` in between. -**Load claim pack size:** soft **Σ `tx.input`** budget (hardcoded **8000**; include overshoot block) or hard **144** blocks. Dense mainnet blocks hit the input soft stop after **typically a few blocks** (often 1–3); early tiny blocks may pack many until the hard cap. Do **not** treat ~32 as pack size (that was 8000/250 mid-chain, not fat-era). +**Load claim pack size:** soft **Σ `tx.input`** budget (hardcoded **8000**; include overshoot block) or hard **144** blocks. Also stop a `LoadBatch` before the next height when `header_txs.has_body` differs from the part's first height (crash some→none). Lookup may still decode a mixed resolve wave; loadq chunks are one kind. Dense mainnet blocks hit the input soft stop after **typically a few blocks** (often 1–3); early tiny blocks may pack many until the hard cap. Do **not** treat ~32 as pack size (that was 8000/250 mid-chain, not fat-era). **IBD lookup resolve wave:** TipOnly `head_fk` bounded by remaining loadq slots × load pack (safety cap **1080** heights / soft **64000** inputs; include-overshoot; 256 k unique-key safety cap). Decode parks as resolved BQ rows; `taken_hi` bumps **per load-batch send**. Unsent tail stays on the BQ (no re-decode). Hard **min 8000** inputs per published layer when more unresolved heights can still join — including `ready=0` / load-frontier / unknown window. Last available thin wave still emits. One published identity layer per wave holds **this wave's** parent identities (re-home union hits + TipOnly misses); `skipped=` is TipOnly avoided, not omitted from the layer. Get walks; splice keeps a layer while its span is on the BQ or overlaps `(tip, taken_hi]`. When `ready >` half the 1-min BQ window, lookup waits for a full wave instead of minting a 1-block layer — unless the first unresolved height is within `path_lo + win/2` **and** the collect is already ≥8000 inputs (load is about to claim it; O(1) from the already-sorted unresolved list). Pack/hold uses enqueue-stamped `n_inputs` (peer CompactSize walk). `wave_intake` does **not** clone raw payloads; decode clones `raw_payload(height)` only on emit. Hold is `decode=0` / `precompute=0`. `lookup_thr wave=(decode= precompute= collect= head=)`: `precompute=` is `from_tx_connect` below milestone or `from_tx` when scripts run; `head=` is TipOnly `get_fk_by_txid_batch`, not load stamp. -**`ibd: perf`:** `load=` is pin+assemble only. Load OS-thread `stamp=` nests `pack=` (plan HashMap) vs `head=` (leftover TipOnly `prep_head_fk_ns`). After a published wave `head=` is ~0. `stamp_sub struct_txid=` must be **0** on IBD (loadq `pres`). Non-zero means load dropped lookup hashes and `from_tx`'d again. Post-scriptq in-flight drop is `prune=` (layer retain only; IBD has no pstore Weak walk). Lookup-wave `decode=` / `precompute=` / `collect=` / `head=` nest under `lookup_thr wave=`. Pin names `thin=` for the vout-map prefix. `script=` is per-batch wave wall on `ibd-confirm` (`jobs=`/`skip=`). Steal claim is an `ArcSwap` snapshot + `fetch_add(32)` (no `WAVES` mutex per job; `in_wave` counts in-flight chunks). The scripts thread publishes another `scriptq` batch when steal is empty (up to 4 in-flight, matching `scriptq`) and parks until a worker unparks it on wave complete (load also unparks on `scriptq` send). `script=` stamps when the wave first reports complete, not at write-queue pop. Write snapshot-drains the whole **writeq** (cap 14) so one `flush_class_c_tip` covers the queued run. `ready>0` + `scriptq=1` + high `stamp=` `head=` means leftover on load, not a hungry script pool. High `stamp=` `pack=` with `head=0` is HashMap CPU on load (leave it there; lookup is already the wave). Do not retune steal or borrow `rbtc-scripts-*` for decode. Sptweak secp +**`ibd: perf`:** `load=` is pin+assemble only. Load OS-thread `stamp=` nests `pack=` (plan HashMap) vs `head=` (leftover TipOnly `prep_head_fk_ns`). After a published wave `head=` is ~0. `stamp_sub struct_txid=` must be **0** on IBD (loadq `pres`). Non-zero means load dropped lookup hashes and `from_tx`'d again. Post-scriptq in-flight drop is `prune=` (layer retain only; IBD has no pstore Weak walk). Lookup-wave `decode=` / `precompute=` / `collect=` / `head=` nest under `lookup_thr wave=`. Pin names `thin=` for the vout-map prefix. `script=` is per-batch wave wall on `ibd-confirm` (`jobs=`/`skip=`). Steal claim is an `ArcSwap` snapshot + `fetch_add(32)` (no `WAVES` mutex per job; `in_wave` counts in-flight chunks). The scripts thread publishes another `scriptq` batch when steal is empty (up to 4 in-flight, matching `scriptq`) and parks until a worker unparks it on wave complete (load also unparks on `scriptq` send). `script=` stamps when the wave first reports complete, not at write-queue pop. Write snapshot-drains the whole **writeq** (cap 14) so one `flush_class_c_tip` covers the queued run. `append_contiguous` stops when `archive_plan` polarity differs (`Some` vs `None`); leftover is the next meta-batch. `ready>0` + `scriptq=1` + high `stamp=` `head=` means leftover on load, not a hungry script pool. High `stamp=` `pack=` with `head=0` is HashMap CPU on load (leave it there; lookup is already the wave). Do not retune steal or borrow `rbtc-scripts-*` for decode. Sptweak secp may publish a **background** wave (`try_for_each_parallel_idle`) claimed only when no foreground wave and no detached job are waiting. Process restart leftover is empty RAM identity (horizon does not survive). diff --git a/docs/invariants.md b/docs/invariants.md index a6ee14fa..16c626eb 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -114,12 +114,13 @@ published idx window. | **S0 fresh tip+1** | absent | present | parents on head | plan=Some: plan_batch stamps fk+txout/spent range+txid + parent vouts; load `txout` by range, copies spent_range | | **S1 already-archived** | body present (plan=None) | present | parents on head | lookup still stamps parent fk+ranges+txid (idx/head); load `txout` only | | **S2 tip-ahead pack** | prior pack uncommitted | — | parents in in_flight | plan uses in_flight create_fk; **must** also stamp ranges (idx) when body exists, or use offline CreatePin | -| **S3 short catch-up** | mixed S0/S1 over gap | present | mostly cold | ordered claim tip+1 only for write; lookup may plan ahead with reserved HWM | +| **S3 short catch-up** | mixed need-body / already-bodied over a height-ordered prefix | present | mostly cold | two (or more) homogeneous batches: lookup splits loadq at `header_txs.has_body`; write drain stops on `archive_plan` polarity. Load/scripts are not splitters. Mixed stamp is `Corrupt("invariant: confirm batch mixed archived")`. Write vs tip is all-old no-op, all-new fill, or `Corrupt("invariant: write batch spans tip")` — no prefix strip | | **S4 cascade fail** | tip+1 blacklisted or write failed | — | — | tip-ahead write may hit `fk mismatch` / `connect height not tip+1` → **soft requeue**, not permanent blacklist | | Error | State | Root | Fix | |-------|-------|------|-----| | `lookup stage miss (load cold denserels forbidden)` | S0/S3 | Load Forbid + parents without plan range | Lookup always fills `external_parents` body; load outs by `txout` range only | +| `invariant: confirm batch mixed archived` | S3 | One stamp/load list spans need-body and already-bodied | Split at the `has_body` change (IBD lookup) or call one-shot twice. Do not hitchhike `get_list` on `plan=Some` | | `put_full_batch fk mismatch` | S4 cascade | Tip-ahead plan after tip+1 reject | Soft requeue for fk mismatch / connect height not tip+1 | | `parent create_fk unresolved` | S2 | Leftover union miss | **Permanent.** Fix publish order. Do not soft-requeue. | | false PrevoutSpent | identity | schema-13 zero pin id | plan reverse map / lookup `txid.body` only |