SH compact, header anti-DoS, and IBD stamp/pin rewind

CHANGELOG.md

@@ -27,6 +27,21 @@ before 1.0).
 - **IBD `lookup_taken_hi` rewind:** merkle/witness SoftWire, Cascade,
   EngineFault, and ConsensusInvalid rewind the lookup consume high-water to
   the confirmed tip so densify can re-getdata. Previously only BadPrev did.
+- **SH overflow compact / tip append:** `compact_sealed_ovf` installs L1
+  before dropping L0 so `locate_head` never sees both empty. Occupancy walk
+  includes compacted L1. Tip append probes sealed heads when `live_count == 0`
+  but head slots are still occupied (crash mid-finish).
+- **Equal-length reorg work:** `disconnect_to` truncates `chain_work_prefix` to
+  `keep_height+1` so the next most-work compare is rebuilt from the new branch.
+- **Header anti-DoS:** `ensure_header` runs POW / nBits / MTP when the parent is
+  on the best chain (claimed POW otherwise). Pending ranking and getdata skip
+  headers whose claimed nBits the hash does not meet, and fail closed on a
+  pending-header hole.
+- **IBD stamp/pin fail:** re-offer the rest of the wave into the **body
+  queue** (lookup never reads `feed.ready` wire), `clear_all` in-flight
+  identity, rewind `lookup_taken_hi` to the confirmed tip, and bump the feed
+  epoch so later same-wave loadq chunks are stale (those chunks re-offer on
+  drop).
 - **IBD session-fault resume:** after Class C, a uring session fault on spend
   annotate or `tx.head` drain finishes annotate+drain on the write thread
   (tip `connect_at` retries `finish_post_commit`; IBD does the same in place

crates/rbitcoin-net/src/chain.rs

@@ -7,11 +7,11 @@ use crate::cache::BlockCache;
 use crate::error::NetError;
 use bitcoin::block::Header;
 use bitcoin::hashes::Hash;
-use bitcoin::{Block, BlockHash, ScriptBuf, Transaction, Work};
+use bitcoin::{Block, BlockHash, ScriptBuf, Target, Transaction, Work};
 use rbitcoin_consensus::{
     accept_and_connect_block_preverified, confirm_wire_load_from_plan as consensus_load_from_plan,
     confirm_wire_load_phase_pipelined, confirm_write_phase, genesis_block, header_to_record,
-    mine_regtest_paying, ChainParams, Milestone, PlanStampOutcome, ScriptOkBatch,
+    mine_regtest_paying, validate_header, ChainParams, Milestone, PlanStampOutcome, ScriptOkBatch,
     ScriptPreverified, WireLoadPipeline,
 };
 use rbitcoin_log::info;
@@ -380,10 +380,19 @@ impl ChainHub {
         *self.minimum_chain_work.write().unwrap() = w;
     }
 
+    /// Claimed nBits meet `pow_limit` and the hash meets that target.
+    pub(crate) fn header_claimed_pow_ok(&self, header: &Header) -> bool {
+        let target = Target::from_compact(header.bits);
+        target <= self.params.pow_limit && header.validate_pow(target).is_ok()
+    }
+
     /// Work of `header` hanging off a known parent (tip-extend, header-only
     /// chain, or side), else just the header's own work.
     pub fn work_with_header(&self, header: &Header) -> Work {
-        let mut extra = vec![header.work()];
+        let mut extra = Vec::new();
+        if self.header_claimed_pow_ok(header) {
+            extra.push(header.work());
+        }
         let mut prev = header.prev_blockhash;
         for _ in 0..10_000 {
             if prev.to_byte_array() == [0u8; 32] {
@@ -404,7 +413,9 @@ impl ChainHub {
             let Some(hdr) = self.header_of(&prev) else {
                 return crate::most_work::sum_work(extra.into_iter());
             };
-            extra.push(hdr.work());
+            if self.header_claimed_pow_ok(&hdr) {
+                extra.push(hdr.work());
+            }
             prev = hdr.prev_blockhash;
         }
         crate::most_work::sum_work(extra.into_iter())
@@ -1029,6 +1040,20 @@ impl ChainHub {
                 }
             }
         };
+        if header.prev_blockhash.to_byte_array() != [0u8; 32] {
+            let parent = header.prev_blockhash.to_byte_array();
+            if let Some(ph) = self.query.height_of_hash(&parent).ok().flatten() {
+                validate_header(
+                    self.query.as_ref(),
+                    &self.params,
+                    Height(ph.0.saturating_add(1)),
+                    header,
+                )
+                .map_err(|e| NetError::Consensus(e.to_string()))?;
+            } else if !self.header_claimed_pow_ok(header) {
+                return Err(NetError::Consensus("invalid proof of work".into()));
+            }
+        }
         let rec = header_to_record(prev_fk, header);
         let fk = self
             .query
@@ -2142,9 +2167,26 @@ impl ChainHub {
         }
         self.query
             .drop_sh_pending_from(Height(keep_height.saturating_add(1)));
+        self.chain_work_prefix
+            .write()
+            .unwrap()
+            .truncate(keep_height as usize + 1);
         Ok(())
     }
 
+    #[cfg(test)]
+    pub(crate) fn test_chain_work_prefix_len(&self) -> usize {
+        self.chain_work_prefix.read().unwrap().len()
+    }
+
+    #[cfg(test)]
+    pub(crate) fn test_poison_chain_work_prefix_last(&self) {
+        let mut p = self.chain_work_prefix.write().unwrap();
+        if let Some(last) = p.last_mut() {
+            *last = Work::from_be_bytes([0xff; 32]);
+        }
+    }
+
     fn block_at_height(&self, height: u32) -> Result<Option<Block>, NetError> {
         if let Some(h) = self.cache.hash_at_height(height) {
             if let Some(b) = self.cache.get_block(&h) {
@@ -2923,6 +2965,54 @@ mod tests {
         let _ = std::fs::remove_dir_all(dir);
     }
 
+    #[test]
+    fn disconnect_truncates_chain_work_prefix_to_keep_height() {
+        let (dir, hub) = tmp_hub();
+        hub.ensure_genesis().unwrap();
+        let gen = hub.tip_hash().unwrap();
+        let a1 = mine(gen, 1_300_030_000, 1);
+        hub.accept_block(a1.clone()).unwrap();
+        let a2 = mine(a1.block_hash(), 1_300_030_100, 2);
+        hub.accept_block(a2.clone()).unwrap();
+        let _ = hub.chain_work().unwrap();
+        assert_eq!(hub.test_chain_work_prefix_len(), 3);
+        hub.test_poison_chain_work_prefix_last();
+        hub.rewind_to_height(1).unwrap();
+        assert_eq!(
+            hub.test_chain_work_prefix_len(),
+            2,
+            "equal-length reorg must not keep the losing branch's prefix"
+        );
+        let b2 = mine_distinct(a1.block_hash(), 1_300_030_200, 2, &[a2.block_hash()]);
+        hub.accept_block(b2).unwrap();
+        let mut acc = Work::from_be_bytes([0u8; 32]);
+        for h in 0..=2 {
+            acc = acc + hub.query.wire_header_at_height(Height(h)).unwrap().work();
+        }
+        assert_eq!(
+            hub.chain_work().unwrap(),
+            acc,
+            "prefix must be rebuilt from the winner, not the poisoned loser"
+        );
+        assert_ne!(hub.chain_work().unwrap(), Work::from_be_bytes([0xff; 32]));
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
+    #[test]
+    fn ensure_header_rejects_claimed_hard_bits_without_pow() {
+        let (dir, hub) = tmp_hub();
+        hub.ensure_genesis().unwrap();
+        let gen = hub.tip_hash().unwrap();
+        let mut bad = mine(gen, 1_300_031_000, 1).header;
+        bad.bits = CompactTarget::from_consensus(0x1d00ffff);
+        bad.nonce = 0;
+        assert!(
+            hub.ensure_header(&bad).is_err(),
+            "persist must not accept nBits/POW that confirm would reject"
+        );
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
     #[test]
     fn tip_follow_accept_logs_update_tip_per_block() {
         // Shipped path: accept_block → connect_at → log_update_tip (info).

crates/rbitcoin-net/src/ibd/confirm/mod.rs

@@ -518,6 +518,45 @@ fn requeue_on_uring_recover(
     true
 }
 
+fn reoffer_blocks_to_body_queue<'a>(
+    hub: &ChainHub,
+    items: impl IntoIterator<Item = (u32, BlockHash, &'a bitcoin::Block)>,
+) {
+    use bitcoin::consensus::encode::serialize;
+    for (h, hash, block) in items {
+        if hub.has_block(&hash) {
+            continue;
+        }
+        let payload = serialize(block);
+        let header_fk = hub
+            .query
+            .get_header_by_hash(&hash.to_byte_array())
+            .ok()
+            .flatten()
+            .map(|(fk, _)| fk.0)
+            .unwrap_or(0);
+        let _ = hub
+            .query
+            .block_queue_offer(h, hash.to_byte_array(), header_fk, &payload);
+    }
+}
+
+/// Stamp/pin fail: drop speculative fks, bump the feed epoch, re-offer tail to BQ.
+fn load_fail_rewind_wave<'a>(
+    feed: &ConfirmFeed,
+    hub: &ChainHub,
+    lookup_ahead: &mut LoadAheadState,
+    first_h: u32,
+    tail: impl IntoIterator<Item = (u32, BlockHash, &'a bitcoin::Block)>,
+) {
+    let tail: Vec<_> = tail.into_iter().collect();
+    reoffer_blocks_to_body_queue(hub, tail.iter().copied());
+    lookup_ahead.clear_all(hub);
+    feed.finish(std::iter::once(first_h));
+    feed.clear();
+    hub.query.set_lookup_taken_hi(hub.tip_height());
+}
+
 pub(crate) fn lookup_ready_hash(feed: &ConfirmFeed, height: u32) -> Option<BlockHash> {
     feed.inner
         .lock()
@@ -1785,6 +1824,13 @@ pub(crate) fn spawn_confirm_engine(
                         "ibd: confirm load drop stale plan epoch={claim_epoch} live={}",
                         feed_load.epoch()
                     );
+                    reoffer_blocks_to_body_queue(
+                        &hub_load,
+                        lb.items.iter().filter_map(|(h, raw, w)| {
+                            let hash = BlockHash::from_byte_array(*raw);
+                            (!hub_load.has_block(&hash)).then_some((*h, hash, w.block.as_ref()))
+                        }),
+                    );
                     feed_load.finish(lb.items.iter().map(|(h, _, _)| *h));
                     continue;
                 }
@@ -1885,18 +1931,15 @@ pub(crate) fn spawn_confirm_engine(
                             continue;
                         }
                         let first_hash = wire_batch[0].1;
-                        if wire_batch.len() > 1 {
-                            let tail: Vec<(u32, BlockHash, Option<bitcoin::Block>)> =
-                                wire_batch
-                                    .iter()
-                                    .skip(1)
-                                    .filter(|(_, ha, _)| !hub_load.has_block(ha))
-                                    .map(|(h, ha, _)| (*h, *ha, None))
-                                    .collect();
-                            feed_load.requeue_wire(&tail);
-                        }
-                        feed_load.finish(std::iter::once(expect_h));
-                        lookup_ahead.clear_all(&hub_load);
+                        load_fail_rewind_wave(
+                            &feed_load,
+                            &hub_load,
+                            &mut lookup_ahead,
+                            expect_h,
+                            wire_batch.iter().skip(1).filter_map(|(h, ha, w)| {
+                                (!hub_load.has_block(ha)).then_some((*h, *ha, w.block.as_ref()))
+                            }),
+                        );
                         loop_stats_load
                             .confirm_reject_stops
                             .fetch_add(1, Ordering::Relaxed);
@@ -1948,7 +1991,6 @@ pub(crate) fn spawn_confirm_engine(
                     .map(|(h, ha, _)| (*h, *ha))
                     .collect();
                 let first_hash = heights_hashes[0].1;
-                let _ = wire_batch;
 
                 struct LiveGuard<'a> {
                     stats: &'a LoopStats,
@@ -2039,17 +2081,15 @@ pub(crate) fn spawn_confirm_engine(
                             );
                             continue;
                         }
-                        if heights_hashes.len() > 1 {
-                            let tail: Vec<(u32, BlockHash, Option<bitcoin::Block>)> =
-                                heights_hashes
-                                    .iter()
-                                    .skip(1)
-                                    .filter(|(_, ha)| !hub_load.has_block(ha))
-                                    .map(|(h, ha)| (*h, *ha, None))
-                                    .collect();
-                            feed_load.requeue_wire(&tail);
-                        }
-                        feed_load.finish(std::iter::once(expect_h));
+                        load_fail_rewind_wave(
+                            &feed_load,
+                            &hub_load,
+                            &mut lookup_ahead,
+                            expect_h,
+                            wire_batch.iter().skip(1).filter_map(|(h, ha, w)| {
+                                (!hub_load.has_block(ha)).then_some((*h, *ha, w.block.as_ref()))
+                            }),
+                        );
                         loop_stats_load
                             .confirm_reject_stops
                             .fetch_add(1, Ordering::Relaxed);
@@ -2068,6 +2108,7 @@ pub(crate) fn spawn_confirm_engine(
                             break;
                         }
                         std::thread::sleep(Duration::from_millis(10));
+                        continue;
                     }
                 }
                 // Body HWM only — in-flight drop is the marked last-batch path above.

crates/rbitcoin-net/src/ibd/confirm/tests.rs

@@ -1459,6 +1459,97 @@ fn write_session_fault_after_class_c_finishes_annotate_in_place() {
     assert!(!hub.query.block_queue_has_height(1));
 }
 
+#[test]
+fn load_fail_rewind_reoffers_tail_to_bq() {
+    use super::{load_fail_rewind_wave, ConfirmFeed, LoadAheadState};
+    use bitcoin::consensus::encode::serialize;
+    use rbitcoin_query::ArchiveWritePlan;
+    use std::collections::HashSet;
+
+    let (_dir, hub) = crate::chain::tiny_regtest_hub_labeled("load-fail-rewind");
+    hub.ensure_genesis().unwrap();
+    let mut st = LoadAheadState::new(&hub);
+    let body0 = hub.query.tx_body_count();
+    let mut plan = ArchiveWritePlan::empty();
+    plan.planned_fks = vec![Fk(body0.saturating_add(10).max(10))];
+    st.note_lookup_ok(&plan, 10, [1u8; 32]);
+    let pin = test_pin(body0.saturating_add(10).max(10));
+    st.in_flight
+        .note_pins(std::iter::once((plan.planned_fks[0], &pin)), Some(10));
+    assert!(st.in_flight.entry_count() > 0);
+
+    let feed = ConfirmFeed::new();
+    {
+        let mut g = feed.inner.lock().unwrap();
+        g.inflight.insert(10);
+        g.inflight.insert(11);
+    }
+    let prev = hub.tip_hash().unwrap();
+    let body = rbitcoin_consensus::mine_empty_regtest(prev, 1_300_000_000, 1);
+    let hash = body.block_hash();
+    hub.query.set_lookup_taken_hi(Some(11));
+    load_fail_rewind_wave(&feed, &hub, &mut st, 10, std::iter::once((11, hash, &body)));
+    assert_eq!(
+        st.in_flight.entry_count(),
+        0,
+        "pin/stamp fail must clear_all"
+    );
+    assert_eq!(
+        feed.epoch(),
+        1,
+        "epoch bump drops in-channel same-wave loadq"
+    );
+    assert_eq!(
+        hub.query.lookup_taken_hi(),
+        hub.tip_height(),
+        "lookup must be able to select BQ heights again"
+    );
+    assert!(
+        hub.query.block_queue_has_height(11),
+        "tail wire belongs on the body queue, not feed.ready"
+    );
+    assert_eq!(
+        hub.query.block_queue_payload(11).unwrap().as_deref(),
+        Some(serialize(&body).as_slice())
+    );
+    let skip = HashSet::new();
+    assert_eq!(
+        hub.query.block_queue_unresolved_heights(11, &skip, 4),
+        vec![11],
+        "taken_hi rewind + BQ offer must make the tail claimable"
+    );
+    let g = feed.inner.lock().unwrap();
+    assert!(!g.ready.contains_key(&10));
+    assert!(
+        !g.ready.contains_key(&11),
+        "production lookup does not read feed.ready wire"
+    );
+}
+
+#[test]
+fn stale_loadq_reoffers_decoded_bodies_to_bq() {
+    use super::reoffer_blocks_to_body_queue;
+    use bitcoin::consensus::encode::serialize;
+
+    let (_dir, hub) = crate::chain::tiny_regtest_hub_labeled("stale-loadq-bq");
+    hub.ensure_genesis().unwrap();
+    let prev = hub.tip_hash().unwrap();
+    let body = rbitcoin_consensus::mine_empty_regtest(prev, 1_300_000_100, 1);
+    let hash = body.block_hash();
+    hub.query.set_lookup_taken_hi(Some(12));
+    reoffer_blocks_to_body_queue(&hub, std::iter::once((12, hash, &body)));
+    assert!(hub.query.block_queue_has_height(12));
+    assert_eq!(
+        hub.query.block_queue_payload(12).unwrap().as_deref(),
+        Some(serialize(&body).as_slice())
+    );
+    assert_eq!(
+        hub.query.lookup_taken_hi(),
+        Some(12),
+        "stale drop restores BQ only; stamp/pin fail already rewound taken_hi"
+    );
+}
+
 #[test]
 fn load_session_fault_after_note_lookup_ok_clears_speculative_fks() {
     use super::LoadAheadState;