diff --git a/CHANGELOG.md b/CHANGELOG.md index 38c21e19..183ad44a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/crates/rbitcoin-net/src/chain.rs b/crates/rbitcoin-net/src/chain.rs index c685c66e..47e6d69b 100644 --- a/crates/rbitcoin-net/src/chain.rs +++ b/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, 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). diff --git a/crates/rbitcoin-net/src/ibd/confirm/mod.rs b/crates/rbitcoin-net/src/ibd/confirm/mod.rs index fca9bc04..f770fcba 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/mod.rs +++ b/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, +) { + 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, +) { + 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 { 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)> = - 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)> = - 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. diff --git a/crates/rbitcoin-net/src/ibd/confirm/tests.rs b/crates/rbitcoin-net/src/ibd/confirm/tests.rs index 266fb892..39e510d7 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/tests.rs +++ b/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; diff --git a/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs b/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs index d71d4ded..0ed4a32c 100644 --- a/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs +++ b/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs @@ -1748,14 +1748,16 @@ fn apply_peer_event_body_and_control_surface() { } } fn dummy_header(prev: BlockHash, n: u8) -> Header { - Header { + let mut h = Header { version: Version::from_consensus(4), prev_blockhash: prev, merkle_root: bitcoin::TxMerkleNode::from_byte_array([n; 32]), time: 1_300_000_000 + u32::from(n), bits: CompactTarget::from_consensus(0x207fffff), nonce: u32::from(n), - } + }; + rbitcoin_consensus::grind_regtest_pow(&mut h); + h } let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("ev-apply"); @@ -1982,14 +1984,16 @@ fn apply_peer_event_repeat_headers_skips_ensure_header_fk() { } } fn dummy_header(prev: BlockHash, n: u8) -> Header { - Header { + let mut h = Header { version: Version::from_consensus(4), prev_blockhash: prev, merkle_root: bitcoin::TxMerkleNode::from_byte_array([n; 32]), time: 1_300_000_000 + u32::from(n), bits: CompactTarget::from_consensus(0x207fffff), nonce: u32::from(n), - } + }; + rbitcoin_consensus::grind_regtest_pow(&mut h); + h } let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("ev-hdr-repeat"); @@ -2150,6 +2154,7 @@ fn apply_peer_event_block_framed_bq_horizon_and_headers_done() { txdata: vec![coinbase(height)], }; b.header.merkle_root = b.compute_merkle_root().unwrap(); + rbitcoin_consensus::grind_regtest_pow(&mut b.header); b } fn ser(b: &Block) -> Vec { @@ -2405,6 +2410,7 @@ fn block_framed_raw_offers_body_queue_with_confirm_feed() { txdata: vec![coinbase(height)], }; b.header.merkle_root = b.compute_merkle_root().unwrap(); + rbitcoin_consensus::grind_regtest_pow(&mut b.header); b } @@ -2504,14 +2510,16 @@ fn known_headers_re_admit_to_ordered_after_tip_drain() { } } fn dummy_header(prev: BlockHash, n: u32) -> Header { - Header { + let mut h = Header { version: Version::from_consensus(4), prev_blockhash: prev, merkle_root: bitcoin::TxMerkleNode::from_byte_array([n as u8; 32]), time: 1_300_000_000 + n, bits: CompactTarget::from_consensus(0x207fffff), nonce: n, - } + }; + rbitcoin_consensus::grind_regtest_pow(&mut h); + h } let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("ev-readmit"); @@ -2653,14 +2661,16 @@ fn path_slot_first_wins_chained_via_headers() { } } fn dummy_header(prev: BlockHash, n: u8) -> Header { - Header { + let mut h = Header { version: Version::from_consensus(4), prev_blockhash: prev, merkle_root: bitcoin::TxMerkleNode::from_byte_array([n; 32]), time: 1_300_000_000 + u32::from(n), bits: CompactTarget::from_consensus(0x207fffff), nonce: u32::from(n), - } + }; + rbitcoin_consensus::grind_regtest_pow(&mut h); + h } let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("path-slot"); diff --git a/crates/rbitcoin-net/src/peer.rs b/crates/rbitcoin-net/src/peer.rs index e143d725..eb41002f 100644 --- a/crates/rbitcoin-net/src/peer.rs +++ b/crates/rbitcoin-net/src/peer.rs @@ -3447,7 +3447,10 @@ fn work_of_header_path( std::iter::once(base).chain(extra), )); } - let hdr = pending.get(&h)?; + let hdr = pending.get(&h).copied().or_else(|| hub.header_of(&h))?; + if !hub.header_claimed_pow_ok(&hdr) { + return None; + } extra.push(hdr.work()); h = hdr.prev_blockhash; if h.to_byte_array() == [0u8; 32] { @@ -3470,6 +3473,9 @@ fn fetchable_header_path_bodies( if !header_path_meets_minwork(hub, pending, tip) { return Vec::new(); } + if work_of_header_path(hub, pending, tip).is_none() { + return Vec::new(); + } if matches!( announced_work_cmp(hub, pending, tip), Some(std::cmp::Ordering::Less) diff --git a/crates/rbitcoin-net/src/peer_tests.rs b/crates/rbitcoin-net/src/peer_tests.rs index 2967e41f..63940b4b 100644 --- a/crates/rbitcoin-net/src/peer_tests.rs +++ b/crates/rbitcoin-net/src/peer_tests.rs @@ -156,14 +156,15 @@ fn should_poll_peer_headers_skips_behind_and_weaker_fork() { should_poll_peer_headers(&hub, Some(BlockHash::from_byte_array([0xee; 32]))), "unknown best-known still poll until we can classify the branch" ); - let fork = bitcoin::block::Header { + let mut fork = bitcoin::block::Header { version: bitcoin::block::Version::from_consensus(4), prev_blockhash: gen, merkle_root: bitcoin::TxMerkleNode::from_byte_array([0x22; 32]), - time: 1, + time: 1_300_000_000, bits: bitcoin::CompactTarget::from_consensus(0x207f_ffff), nonce: 99, }; + rbitcoin_consensus::grind_regtest_pow(&mut fork); hub.ensure_header(&fork).unwrap(); assert!( !should_poll_peer_headers(&hub, Some(fork.block_hash())), @@ -4653,12 +4654,12 @@ fn announced_tip_is_hopeless_less_and_288_behind() { } #[test] -fn shorter_higher_work_fork_is_not_hopeless() { +fn claimed_hard_bits_without_pow_does_not_getdata() { use bitcoin::block::{Header, Version}; use bitcoin::{CompactTarget, TxMerkleNode}; use rbitcoin_primitives::Height; - let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("short-high-work-fork"); + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("claimed-hard-bits"); hub.ensure_genesis().unwrap(); hub.generate_to_script(5, bitcoin::ScriptBuf::from_bytes(vec![0x51]), vec![]) .unwrap(); @@ -4675,21 +4676,80 @@ fn shorter_higher_work_fork_is_not_hopeless() { let mut pending = HashMap::new(); pending.insert(tip, hard); let work_cmp = announced_work_cmp(&hub, &pending, tip); - assert_eq!( + assert_ne!( work_cmp, Some(std::cmp::Ordering::Greater), - "one mainnet-diff header must outwork 300 regtest blocks" + "claimed mainnet nBits without POW must not outwork the tip" ); - let announced_h = announced_headers_height(&hub, &pending, tip); + let want = + fetchable_header_path_bodies(&hub, &pending, tip, &PendingBlocks::new(), &HashSet::new()); assert!( - !announced_tip_is_hopeless(hub.tip_height().unwrap(), announced_h, work_cmp), - "shorter higher-work path must not be hopeless" + want.is_empty(), + "must not getdata a shorter path that only looks higher-work via claimed nBits" ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn pending_header_hole_does_not_getdata() { + use bitcoin::block::{Header, Version}; + use bitcoin::{CompactTarget, TxMerkleNode}; + + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("pending-hole"); + hub.ensure_genesis().unwrap(); + hub.generate_to_script(2, bitcoin::ScriptBuf::from_bytes(vec![0x51]), vec![]) + .unwrap(); + let orphan = Header { + version: Version::from_consensus(4), + prev_blockhash: BlockHash::from_byte_array([0x11; 32]), + merkle_root: TxMerkleNode::from_byte_array([0x5b; 32]), + time: 1_300_000_000, + bits: CompactTarget::from_consensus(0x207f_ffff), + nonce: 0, + }; + let tip = orphan.block_hash(); + let mut pending = HashMap::new(); + pending.insert(tip, orphan); let want = fetchable_header_path_bodies(&hub, &pending, tip, &PendingBlocks::new(), &HashSet::new()); - assert!( - !want.is_empty(), - "must not skip bodies on a shorter higher-work path" + assert!(want.is_empty(), "a path we cannot walk must not getdata"); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn shorter_higher_work_fork_still_getdata() { + use bitcoin::block::{Header, Version}; + use bitcoin::{CompactTarget, TxMerkleNode}; + use rbitcoin_primitives::Height; + + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("short-high-work-fork"); + hub.ensure_genesis().unwrap(); + hub.generate_to_script(5, bitcoin::ScriptBuf::from_bytes(vec![0x51]), vec![]) + .unwrap(); + let gen = hub.query.wire_header_at_height(Height(0)).unwrap(); + let mut hard = Header { + version: Version::from_consensus(4), + prev_blockhash: gen.block_hash(), + merkle_root: TxMerkleNode::from_byte_array([0x5c; 32]), + time: gen.time.saturating_add(600), + bits: CompactTarget::from_consensus(0x1f7f_ffff), + nonce: 0, + }; + rbitcoin_consensus::grind_regtest_pow(&mut hard); + let tip = hard.block_hash(); + let mut pending = HashMap::new(); + pending.insert(tip, hard); + assert_eq!( + announced_work_cmp(&hub, &pending, tip), + Some(std::cmp::Ordering::Greater), + "one harder-than-regtest header must outwork five easy blocks" + ); + let want = + fetchable_header_path_bodies(&hub, &pending, tip, &PendingBlocks::new(), &HashSet::new()); + assert_eq!( + want, + vec![tip], + "legitimate shorter higher-work path still fetches" ); let _ = std::fs::remove_dir_all(dir); } diff --git a/crates/rbitcoin-store/src/scripthash.rs b/crates/rbitcoin-store/src/scripthash.rs index 2552e9dd..05506f47 100644 --- a/crates/rbitcoin-store/src/scripthash.rs +++ b/crates/rbitcoin-store/src/scripthash.rs @@ -1344,6 +1344,19 @@ impl ScriptHashTable { })?; } } + { + let g = self.ovf_l1.lock().unwrap(); + if let Some(l1) = g.as_ref() { + let body = self.ovf_file(); + l1.head.for_each_occupied(|_k, val| { + let entries = self.collect_entries_from(body, &val)?; + for fk in entries { + f(fk); + } + Ok(()) + })?; + } + } self.ingest.lock().unwrap().for_each_occupied(|_key, val| { let entries = self.collect_entries_from(self.ovf_file(), &val)?; for fk in entries { @@ -1443,7 +1456,7 @@ impl ScriptHashTable { let t_seed = std::time::Instant::now(); // Cold body (no prior creates): skip N head gets — empty table probes. - if self.entry_count() > 0 { + if self.entry_count() > 0 || !self.head_is_empty() { let mut missing: Vec<[u8; 32]> = Vec::new(); { let mut seen_miss = std::collections::HashSet::new(); @@ -1724,8 +1737,8 @@ impl ScriptHashTable { } } - /// K-way merge of sealed global ovf heads. Body offs unchanged. Readers - /// keep the old `Vec` until this lock is released after rename. + /// K-way merge of sealed global ovf heads. Body offs unchanged. Install L1 + /// before dropping L0 so locate_head never sees both empty. pub fn compact_sealed_ovf(&self) -> Result<(), StoreError> { if self.ovf_l1.lock().unwrap().is_some() { self.warn_l1_frozen(); @@ -1777,12 +1790,12 @@ impl ScriptHashTable { fp.push(".fuse8"); fuse.write_to(&PathBuf::from(fp))?; } + *self.ovf_l1.lock().unwrap() = Some(OvfL1 { head, fuse }); let old = { let mut g = self.sealed_ovf.lock().unwrap(); std::mem::take(&mut *g) }; drop(old); - *self.ovf_l1.lock().unwrap() = Some(OvfL1 { head, fuse }); for p in old_paths { let _ = std::fs::remove_file(&p); let mut idx = p.as_os_str().to_os_string(); diff --git a/crates/rbitcoin-store/src/scripthash_tests.rs b/crates/rbitcoin-store/src/scripthash_tests.rs index 83ce44b6..79b88613 100644 --- a/crates/rbitcoin-store/src/scripthash_tests.rs +++ b/crates/rbitcoin-store/src/scripthash_tests.rs @@ -527,6 +527,36 @@ fn sh_heads_insert_capped_caps_and_keeps_latest() { assert!(heads.contains_key(&last), "latest insert must stay"); } +#[test] +fn append_after_zero_live_count_keeps_sealed_home() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let sh = script_hash(&[0x7e]); + let mut session = t.bulk_session(1).unwrap(); + session.put_chain(sh, &[Fk(1)]).unwrap(); + let _ = session.finish().unwrap(); + assert!(matches!(t.key_home(&sh).unwrap(), KeyHome::Main)); + t.test_zero_live_count_keep_head().unwrap(); + assert_eq!(t.entry_count(), 0); + assert!(!t.head_is_empty()); + put_create(&t, rec(sh, 2, 0)); + assert!( + matches!(t.key_home(&sh).unwrap(), KeyHome::Main), + "crash mid-finish (live_count=0, heads occupied) must still probe sealed main" + ); + let fks: Vec<_> = t + .entries(&sh) + .unwrap() + .into_iter() + .map(|(_, r)| r.create_tx_fk) + .collect(); + assert!( + fks.contains(&Fk(1)) && fks.contains(&Fk(2)), + "append must not dual-home ingest over sealed rows: {fks:?}" + ); + let _ = std::fs::remove_dir_all(&dir); +} + fn dummy_sh_head_key(i: u64) -> [u8; 32] { let mut k = [0xEE; 32]; k[..8].copy_from_slice(&i.to_le_bytes()); @@ -1042,6 +1072,12 @@ fn compact_merges_two_sealed_global_ovf_files() { t.key_home(&first_new).unwrap(), KeyHome::SealedOvf )); + let mut walked = 0u64; + t.for_each_live_create(|_| walked += 1).unwrap(); + assert_eq!( + walked, 421, + "occupancy walk must include compacted overflow L1" + ); t.compact_sealed_ovf().unwrap(); assert!(t.ovf_l1.lock().unwrap().is_some()); diff --git a/docs/concurrency.md b/docs/concurrency.md index 008dcae5..da2b1b54 100644 --- a/docs/concurrency.md +++ b/docs/concurrency.md @@ -15,7 +15,7 @@ Short map of who may write which tables. **Format is unstable until 1.0.** **IoSession TLS:** one completion session per OS thread (`with_thread_local`). Harvest / poison / drain / do-not-flatten: [`io-modality.md`](./io-modality.md). `RBITCOIN_IO=pread` disables the session. SH k-way merge submits 256 KiB ahead preads on that TLS session and waits only when promote needs a page that has not completed. -**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` (post-lookup reject except Cancelled rewinds taken_hi to the confirmed tip; BadPrev most-work rewind then sets the LCA) → **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 same-batch → in-flight → load-batch skeleton → Corrupt (plan=None leftover TipOnly). Lookup does not own a published identity chain. 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 the last load batch of a lookup wave finishes its in-flight read**, using the drain+fence height snapshotted before that wave's TipOnly (`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. +**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` (post-lookup reject except Cancelled rewinds taken_hi to the confirmed tip; stamp/pin fail re-offers the rest of the wave into the body queue and rewinds taken_hi immediately; epoch-stale loadq chunks re-offer on drop; BadPrev most-work rewind then sets the LCA) → **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 same-batch → in-flight → load-batch skeleton → Corrupt (plan=None leftover TipOnly). Lookup does not own a published identity chain. 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 the last load batch of a lookup wave finishes its in-flight read**, using the drain+fence height snapshotted before that wave's TipOnly (`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. 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).