confirm: homogeneous Class A kind per load/write batch

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.

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<ConfirmArchiveKind, ConsensusError> {
+    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<Vec<rbitcoin_primitives::Fk>> {

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(())
     }

crates/rbitcoin-consensus/src/confirm_run/write.rs

@@ -61,6 +61,36 @@ pub(super) fn write_height_needed(tip: Option<u32>, 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<u32>,
+    heights: impl IntoIterator<Item = u32>,
+) -> 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<Vec<rbitcoin_primitives::Fk>, 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();
 

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,43 +166,62 @@ 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);
     next.archive_plan = Some(rbitcoin_query::ArchiveWritePlan::empty());
     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<u32> = 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, &params, &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, &params, 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, &params, 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);
 }