Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,18 @@ before 1.0).

### Fixed

- **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
when every hash is connected). Load recover after `note_lookup_ok` rebuilds
create-fk HWM from durable Class A (`clear_all`) so retried plans are not
stamped past an abandoned pack.
- **1p1c child-fail rollback:** rolling back a below-min-relay parent also
evicts mempool txs that spend it (a concurrent spender in that window
cannot stay).
- **RPC `maxfeerate`:** prevout-sum overflow is over-cap (reject), not
fail-open. Missing prevouts still skip the cap so `accept_tx` can say
missing-inputs.
- **IBD io_uring drain stall:** `drain_all` no longer returns after 5 s with
leftover SQEs (that freed in-flight buffers). Every TLS session waits while
CQEs arrive; a 120 s zero-completion stall aborts explicit drain (session
Expand Down
2 changes: 1 addition & 1 deletion TESTING.md
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

- Put HOLD / wait hooks in a shipped function other tests also call (`confirm_scripts_phase`).
- Assert process-global last-writer meters as the contract. Confirm / query / IBD window meters are instance-owned (`Query::confirm_stats`, take-and-reset). Pin two engines, not crate-root atomics. Store head-resolve window meters and `last_union_miss` / leftover probe diag remain process-global; use pin/layout, error strings, or a pure formatter.
- Thread-local `test_take_*` IO probes on store hot paths (assert session/table instance stats or file state).
- Thread-local `test_take_*` IO probes on store hot paths (assert session/table instance stats or file state). Class A three-stem append is `UringSession` max-batch pwrite SQEs (`tls_take_max_batch_pwrite_n` after `with_thread_local`), not `test_take_pwrite_waves`.
- `std::env::set_var` without the crate lock (or pass the knob as an argument).
- Bind a fixed port (use `:0`) or share a `/tmp` path (use `rbitcoin_store::testutil::TempDir` / `tiny_store`, `rbitcoin_query::testutil::tiny_query`, net `tiny_regtest_hub`, or `rbitcoin_test::TestDatadir`).

Expand Down
3 changes: 2 additions & 1 deletion crates/rbitcoin-consensus/src/confirm_run/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,8 @@ use phases::assemble_run;
use phases::{check_bip34, expected_bits_extending, post_commit};
use pin::{ensure_spend_abs_layouts, pin_for_wire_batch};
pub use scripts::{confirm_scripts_phase, drive_script_waves_with};
pub use write::confirm_write_phase;
pub(crate) use write::finish_post_commit;
pub use write::{confirm_write_phase, finish_post_commit_hashes};
#[cfg(test)]
use write::{write_batch_vs_tip, write_height_needed, WriteBatchVsTip};

Expand Down
147 changes: 146 additions & 1 deletion crates/rbitcoin-consensus/src/confirm_run/write.rs
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,15 @@ pub fn confirm_write_phase(
) -> Result<Vec<rbitcoin_primitives::Fk>, ConsensusError> {
let tip = query.tip_height().map(|h| h.0);
match write_batch_vs_tip(tip, batch.prepared.iter().map(|p| p.height.0)) {
WriteBatchVsTip::AllOld => return Ok(Vec::new()),
WriteBatchVsTip::AllOld => {
let items: Vec<(u32, [u8; 32])> = batch
.prepared
.iter()
.map(|p| (p.height.0, p.hash))
.collect();
finish_post_commit_hashes(query, &items)?;
return Ok(Vec::new());
}
WriteBatchVsTip::SpansTip => {
return Err(ConsensusError::Store(rbitcoin_store::StoreError::Corrupt(
"invariant: write batch spans tip",
Expand Down Expand Up @@ -254,6 +262,143 @@ pub fn confirm_write_phase(
Ok(out)
}

/// Replay spend annotate + `tx.head` drain after Class C already committed.
///
/// Session-fault retry must not skip this: `height_of_hash` matching is not
/// "write finished."
pub(crate) fn finish_post_commit(
query: &Query,
height: u32,
hash: &[u8; 32],
) -> Result<(), ConsensusError> {
finish_post_commit_hashes(query, &[(height, *hash)])
}

/// IBD write-thread in-place retry: `(height, hash)` must already be connected
/// at that height.
pub fn finish_post_commit_hashes(
query: &Query,
items: &[(u32, [u8; 32])],
) -> Result<(), ConsensusError> {
let queued = query.store().txs.take_pending_queued();
if queued.is_empty() && !query.spend_index_enabled() {
return Ok(());
}
let drain_max_fk = queued.iter().filter_map(|(_, fk)| fk.get()).max();
let drain = if queued.is_empty() {
None
} else {
Some(super::head_drain::submit_head_insert(query.store(), queued))
};

let annotate_res = (|| -> Result<(), ConsensusError> {
if !query.spend_index_enabled() {
return Ok(());
}
let mut jobs = Vec::new();
for &(height, hash) in items {
jobs.extend(annotate_jobs_from_connected_hash(query, height, &hash)?);
}
post_commit(query, &jobs)?;
Ok(())
})();

if let Some(drain) = drain {
let (drain_res, restore) = drain.join_restore();
if drain_res.is_err() {
query.store().txs.head_note_pending(&restore);
}
annotate_res?;
drain_res.map_err(ConsensusError::from)?;
if let Some(fk) = drain_max_fk {
query.note_head_drain_fk(fk);
}
} else {
annotate_res?;
}
Ok(())
}

fn annotate_jobs_from_connected_hash(
query: &Query,
height: u32,
hash: &[u8; 32],
) -> Result<Vec<crate::block::SpendAnnotateJob>, ConsensusError> {
match query.height_of_hash(hash).map_err(ConsensusError::from)? {
Some(h) if h.0 == height => {}
_ => return Ok(Vec::new()),
}
let Some((hfk, _)) = query
.get_header_by_hash(hash)
.map_err(ConsensusError::from)?
else {
return Ok(Vec::new());
};
let Some(tx_fks) = query
.store()
.header_txs
.get_list(hfk)
.map_err(ConsensusError::from)?
else {
return Ok(Vec::new());
};
let mut jobs = Vec::new();
for &spend_fk in &tx_fks {
let (_meta, ins, _outs) = query
.store()
.get_tx_full(spend_fk)
.map_err(ConsensusError::from)?;
for inp in ins {
if inp.is_coinbase() {
continue;
}
let create_fk = if inp.create_fk.is_null() {
query
.store()
.get_fk_by_txid_tip(&inp.prev_txid)
.map_err(ConsensusError::from)?
.unwrap_or(rbitcoin_primitives::Fk::NULL)
} else {
inp.create_fk
};
if create_fk.is_null() {
continue;
}
let (off, len) = query
.store()
.tx_spent_range(create_fk)
.map_err(ConsensusError::from)?;
let abs = rbitcoin_store::spent_abs(off, inp.prev_index);
if abs.saturating_add(rbitcoin_store::OutputRecord::SPENT_SLOT_LEN as u64)
> off.saturating_add(len)
{
return Err(ConsensusError::Store(StoreError::Corrupt(
"invariant: finish_post_commit spent slot OOB",
)));
}
let (multi, field) = query
.store()
.txs
.get_output_spender_meta(create_fk, inp.prev_index)
.map_err(ConsensusError::from)?;
let flags = if multi {
rbitcoin_store::output_flags::MULTI_SPENDER
} else {
0
};
jobs.push(crate::block::SpendAnnotateJob {
abs,
field,
flags,
create_fk,
vout: inp.prev_index,
spend_fk,
});
}
}
Ok(jobs)
}

/// After Class A commit, set body_range (+ spent.idx) for **pinned** creates
/// still missing layout. Body ranges come from the write's one `tx_body_range_batch`
/// (same batch as layout fill). Spent holes use one `tx_spent_range_batch`.
Expand Down
128 changes: 128 additions & 0 deletions crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,12 @@ fn three_stage_write_filter_and_scripts_surface() {
assert!(batch.is_empty());
assert_eq!(batch.approx_wire_bytes(), 0);
assert_eq!(batch.parent_count(), 0);
let cur = std::thread::current();
let name = cur.name().unwrap_or("").to_string();
assert!(
!name.starts_with("rbtc-scripts"),
"confirm_scripts_phase must run on the caller, got {name}"
);
let ok = confirm_scripts_phase(batch).expect("empty scripts ok");
assert!(ok.batch.prepared.is_empty());
assert!(ok.batch.wire_blocks.is_empty());
Expand Down Expand Up @@ -255,6 +261,12 @@ fn drive_script_waves_ordered_without_coordinator_threads() {
&rx,
|_, _| {},
|ok, meta| {
let cur = thread::current();
let name = cur.name().unwrap_or("").to_string();
assert!(
!name.starts_with("rbtc-scripts"),
"script publisher must not be a steal worker, got {name}"
);
heights_w.lock().unwrap().push(meta.first_h);
assert!(ok.batch.is_empty());
true
Expand Down Expand Up @@ -2173,3 +2185,119 @@ fn one_shot_load_matches_stamp_then_load_from_plan() {
let _ = std::fs::remove_dir_all(&path_a);
let _ = std::fs::remove_dir_all(&path_b);
}

/// Class C can commit (`height_of_hash` matches) while spend annotate is still
/// missing. Retry must finish annotate, not take the already-at-height `Ok`.
#[test]
fn already_at_height_retries_post_commit_spend_annotate() {
use crate::{accept_and_connect_block_preverified, ChainParams, Milestone, ScriptPreverified};
use bitcoin::hashes::Hash;
use bitcoin::{block::Header, Block, BlockHash, CompactTarget, TxMerkleNode};
use rbitcoin_primitives::{Fk, Height};
use rbitcoin_query::testutil::FixtureChain;
use rbitcoin_query::TxApply;
use rbitcoin_store::{HeaderRecord, InputRecord, OutputRecord, TxRecord};

let (path, q) = tmp_query();
q.set_spend_index(false);

let h0 = HeaderRecord {
prev_fk: Fk::NULL,
version: 1,
timestamp: 1,
bits: 0x207fffff,
nonce: 0,
Comment thread
reardencode marked this conversation as resolved.
Dismissed
merkle_root: [0xab; 32],
hash: [0xab; 32],
};
let mut txid0 = [0u8; 32];
txid0[31] = 0xcb;
let ta0 = TxApply {
tx: TxRecord {
txid: txid0,
version: 1,
locktime: 0,
input_start_fk: Fk::NULL,
input_count: 1,
output_start_fk: Fk::NULL,
output_count: 1,
},
inputs: vec![InputRecord::coinbase(u32::MAX, vec![0x01], vec![])],
outputs: vec![OutputRecord::unspent(50_0000_0000, vec![0x51])],
};
let hfk0 = q.connect_block(Height(0), &h0, &[ta0]).unwrap();
let create_fk = q.block_tx_fks(Height(0)).unwrap()[0];

let hash1 = rbitcoin_store::block_header_hash(1, &h0.hash, &[0x11; 32], 2, 0x207fffff, 1);
Comment thread
reardencode marked this conversation as resolved.
Dismissed
let h1 = HeaderRecord {
prev_fk: hfk0,
version: 1,
timestamp: 2,
bits: 0x207fffff,
nonce: 1,
Comment thread
reardencode marked this conversation as resolved.
Dismissed
merkle_root: [0x11; 32],
hash: hash1,
};
let mut spend_txid = [0u8; 32];
spend_txid[0] = 0x11;
spend_txid[31] = 0xcd;
let ta1 = TxApply {
tx: TxRecord {
txid: spend_txid,
version: 1,
locktime: 0,
input_start_fk: Fk::NULL,
input_count: 1,
output_start_fk: Fk::NULL,
output_count: 1,
},
inputs: vec![InputRecord {
prev_txid: txid0,
create_fk,
prev_index: 0,
sequence: u32::MAX,
script_sig: vec![],
witness: vec![],
}],
outputs: vec![OutputRecord::unspent(49_0000_0000, vec![0x51])],
};
q.connect_block(Height(1), &h1, &[ta1]).unwrap();
let spend_fk = q.block_tx_fks(Height(1)).unwrap()[0];

let (multi, field) = q.store().txs.get_output_spender_meta(create_fk, 0).unwrap();
assert!(!multi);
assert!(
field.is_null(),
"spend_index off: Class C must not spend-annotate"
);
q.set_spend_index(true);

let block1 = Block {
header: Header {
version: bitcoin::block::Version::from_consensus(1),
prev_blockhash: BlockHash::from_byte_array(h0.hash),
merkle_root: TxMerkleNode::from_byte_array(h1.merkle_root),
time: 2,
bits: CompactTarget::from_consensus(0x207fffff),
nonce: 1,
},
txdata: vec![],
};
assert_eq!(block1.block_hash().to_byte_array(), hash1);
assert_eq!(q.height_of_hash(&hash1).unwrap(), Some(Height(1)));

accept_and_connect_block_preverified(
&q,
&ChainParams::regtest(),
Height(1),
&block1,
Milestone::NONE,
&ScriptPreverified::new(),
)
.expect("already-at-height retry");

let (multi2, field2) = q.store().txs.get_output_spender_meta(create_fk, 0).unwrap();
assert!(!multi2);
assert_eq!(field2, spend_fk, "post_commit must annotate after Class C");
let _ = std::fs::remove_dir_all(&path);
}
5 changes: 3 additions & 2 deletions crates/rbitcoin-consensus/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -153,8 +153,8 @@ pub use confirm_run::{
confirm_bq_resolve_wave_capped, confirm_scripts_phase, confirm_wire_load_from_plan,
confirm_wire_load_phase, confirm_wire_load_phase_pipelined, confirm_wire_lookup_stamp,
confirm_wire_run, confirm_wire_run_preverified, confirm_write_phase, drive_script_waves_with,
take_wave_items_for_load, ConfirmLoadOutcome, ConfirmScriptOutcome, LoadedBatch,
PlanStampOutcome, ScriptOkBatch, ScriptPreverified, WireLoadPipeline,
finish_post_commit_hashes, take_wave_items_for_load, ConfirmLoadOutcome, ConfirmScriptOutcome,
LoadedBatch, PlanStampOutcome, ScriptOkBatch, ScriptPreverified, WireLoadPipeline,
BQ_RESOLVE_WAVE_MAX_BLOCKS, BQ_RESOLVE_WAVE_MAX_INPUTS,
};

Expand Down Expand Up @@ -205,6 +205,7 @@ pub fn accept_and_connect_block_preverified(
.get_header_by_hash(&hash)
.map_err(ConsensusError::from)?
{
confirm_run::finish_post_commit(query, height.0, &hash)?;
return Ok(fk);
}
}
Expand Down
19 changes: 19 additions & 0 deletions crates/rbitcoin-consensus/src/script_pool.rs
Original file line number Diff line number Diff line change
Expand Up @@ -691,6 +691,25 @@ mod tests {
assert!(format!("{err}").contains("boom"));
}

fn boom_at_seven(i: &u32) -> Result<(), ConsensusError> {
if *i == 7 {
Err(ConsensusError::BadBlock("boom"))
} else {
Ok(())
}
}

#[test]
fn owned_wave_first_error_surfaces() {
let _gate = STEAL_TEST.lock().unwrap_or_else(|p| p.into_inner());
let err = match start_for_each_owned((0..32u32).collect(), boom_at_seven) {
Ok(Some(w)) => w.finish().expect_err("owned wave must fail"),
Ok(None) => panic!("expected a published owned wave"),
Err(e) => e,
};
assert!(format!("{err}").contains("boom"));
}

#[test]
fn empty_and_single() {
let empty: Vec<u32> = vec![];
Expand Down
Loading