diff --git a/CHANGELOG.md b/CHANGELOG.md index d9624fc316..63d262be12 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,14 @@ before 1.0). ### Fixed +- **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 + `Drop` does not abort). Write/lookup/load/scripts and tip-connect recover once + per 1000 heights (credit only; Class C leftover waits for open repair) + instead of a 19h warn loop. Restart with `RBITCOIN_IO=pread` if completions + cannot complete. [`OPERATOR.md`](OPERATOR.md) + / [`docs/io-modality.md`](docs/io-modality.md). - **`cmpct_differential` fill vs Core extra-txn:** Core v31 latches IBD in `UpdateIBDStatus` (`LoadChainTip` / connect), not `setmocktime`. A fresh regtest genesis stays in IBD under the default 24h `-maxtipage`, so P2P diff --git a/OPERATOR.md b/OPERATOR.md index d88940bc99..90f39920f5 100644 --- a/OPERATOR.md +++ b/OPERATOR.md @@ -177,6 +177,11 @@ rbitcoin use otherwise-idle resources. This is an opt-in host policy, not a performance setting: sync can slow substantially under contention, and an idle I/O class can starve while higher-priority storage work continues. +Idle `IOSchedulingClass` (and some filesystems, including bcachefs) can +starve `io_uring` completions. Expect `store: io_uring drain slow` then at most +one recover per ~1000 heights. If the node aborts (`store: completion session unusable`), +restart as-is or with `RBITCOIN_IO=pread`. There is no mid-IBD libc fallback. + For a regular systemd installation, create a service drop-in with `systemctl edit rbitcoin.service`: @@ -512,7 +517,8 @@ Full modality matrix: [`docs/io-modality.md`](docs/io-modality.md). when available). Table transport is always **fd pread/pwrite**. Compact Class C is L2 write-behind; see [`docs/io-modality.md`](docs/io-modality.md). Per-path env overrides are **removed**. If `uring` is selected but setup fails, demote to -**pread** / **pwrite**. +**pread** / **pwrite**. If a live ring stops completing (`drain slow`, then abort), +restart with **`RBITCOIN_IO=pread`** — the process does not switch backends itself. | Env | Values | Note | |-----|--------|------| diff --git a/crates/rbitcoin-consensus/src/confirm_run/head_drain.rs b/crates/rbitcoin-consensus/src/confirm_run/head_drain.rs index 41a9cfa9a9..2ee74fa6a4 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/head_drain.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/head_drain.rs @@ -7,9 +7,7 @@ use rbitcoin_store::StoreError; use std::collections::VecDeque; use std::panic::{self, AssertUnwindSafe}; use std::sync::mpsc::{self, Receiver}; -#[cfg(test)] -use std::sync::Arc; -use std::sync::{Condvar, Mutex, OnceLock}; +use std::sync::{Arc, Condvar, Mutex, OnceLock}; use std::thread; #[cfg(test)] use std::thread::ThreadId; @@ -56,13 +54,27 @@ fn recv_job(jobs: &Mutex>, cv: &Condvar) -> Job { pub(crate) struct HeadDrainHandle { rx: Option>>, + restore: Option>>>>, #[cfg(test)] named: Arc>>, } impl HeadDrainHandle { - pub(crate) fn join(mut self) -> Result { - self.recv_result() + /// Join and, on insert failure, return the batch so the write thread can + /// put it back on pending-head (no clone on the success path). + pub(crate) fn join_restore( + mut self, + ) -> ( + Result, + Vec<([u8; 32], rbitcoin_primitives::Fk)>, + ) { + let r = self.recv_result(); + let queued = self + .restore + .take() + .and_then(|a| a.lock().unwrap_or_else(|p| p.into_inner()).take()) + .unwrap_or_default(); + (r, queued) } fn recv_result(&mut self) -> Result { @@ -129,6 +141,7 @@ where pool.cv.notify_one(); HeadDrainHandle { rx: Some(rx), + restore: None, #[cfg(test)] named, } @@ -155,5 +168,15 @@ pub(crate) fn submit_head_insert( batch: Vec<([u8; 32], rbitcoin_primitives::Fk)>, ) -> HeadDrainHandle { let ptr = SendStorePtr::from_store(store); - submit_head_drain(move || ptr.insert(&batch)) + let restore = Arc::new(Mutex::new(None)); + let restore_job = Arc::clone(&restore); + let mut handle = submit_head_drain(move || match ptr.insert(&batch) { + Ok(n) => Ok(n), + Err(e) => { + *restore_job.lock().unwrap_or_else(|p| p.into_inner()) = Some(batch); + Err(e) + } + }); + handle.restore = Some(restore); + handle } diff --git a/crates/rbitcoin-consensus/src/confirm_run/mod.rs b/crates/rbitcoin-consensus/src/confirm_run/mod.rs index 3af21ff039..6b4ad76cea 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/mod.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/mod.rs @@ -55,7 +55,7 @@ pub use bq_resolve::{ BQ_RESOLVE_WAVE_MAX_INPUTS, }; #[cfg(test)] -use head_drain::{submit_head_drain, HEAD_DRAIN_THREAD_NAME}; +use head_drain::{submit_head_drain, submit_head_insert, HEAD_DRAIN_THREAD_NAME}; #[cfg(test)] use lookup::confirm_archive_kind; use lookup::known_create_txid_lookup; diff --git a/crates/rbitcoin-consensus/src/confirm_run/write.rs b/crates/rbitcoin-consensus/src/confirm_run/write.rs index c69f849b69..277f2cae80 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/write.rs @@ -252,11 +252,14 @@ pub fn confirm_write_phase( })(); let t_join = Instant::now(); - let drain_res = drain.join(); + let (drain_res, restore) = drain.join_restore(); let drain_join_ns = t_join.elapsed().as_nanos() as u64; if drain_join_ns > 0 { rbitcoin_query::note_confirm(&query.confirm_stats().write_drain_join_ns, drain_join_ns); } + if drain_res.is_err() { + query.store().txs.head_note_pending(&restore); + } let (out, n_blocks, structural_ns, struct_ph, class_c_ns, spend_ann_ns) = overlap?; drain_res.map_err(ConsensusError::from)?; if let Some(fk) = drain_max_fk { 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 634f799136..73a2ce08fc 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs @@ -21,6 +21,15 @@ fn tx_head_drain_thread_is_named_and_reused() { assert_eq!(id1, id2, "drain must keep one OS thread across batches"); } +#[test] +fn head_insert_join_restore_empty_ok() { + use super::submit_head_insert; + let (_d, q) = tmp_query(); + let (r, queued) = submit_head_insert(q.store(), Vec::new()).join_restore(); + assert_eq!(r.unwrap(), 0); + assert!(queued.is_empty()); +} + /// Batch append: contiguous heights merge; gap returns Err(other). #[test] fn script_ok_append_contiguous_and_gap() { diff --git a/crates/rbitcoin-consensus/src/error.rs b/crates/rbitcoin-consensus/src/error.rs index 90bf374c83..931015bd59 100644 --- a/crates/rbitcoin-consensus/src/error.rs +++ b/crates/rbitcoin-consensus/src/error.rs @@ -60,6 +60,12 @@ impl From for ConsensusError { } } +impl ConsensusError { + pub fn is_uring_session_fault(&self) -> bool { + matches!(self, ConsensusError::Store(se) if se.is_uring_session_fault()) + } +} + /// Core `SCRIPT_VERIFY_*` parenthetical for `*-script-verify-flag-failed (…)`. pub fn script_flag_paren(token: &str) -> &str { let inner = token.split(" txid=").next().unwrap_or(token); @@ -157,6 +163,20 @@ mod tests { )); } + #[test] + fn uring_session_fault_covers_leftover_and_submit() { + for m in [ + "invariant: io_uring leftover cqe", + "io_uring submit failed", + "invariant: io_uring undrained", + ] { + let e = ConsensusError::Store(StoreError::Corrupt(m)); + assert!(e.is_uring_session_fault(), "{m}"); + } + assert!(!ConsensusError::Store(StoreError::NotFound).is_uring_session_fault()); + assert!(!ConsensusError::PrevoutSpent.is_uring_session_fault()); + } + #[test] fn archive_unresolved_parent_is_missing_prevout_not_corrupt() { let e = ConsensusError::from(StoreError::Corrupt( diff --git a/crates/rbitcoin-net/src/chain.rs b/crates/rbitcoin-net/src/chain.rs index 025e76354a..529c0d81a1 100644 --- a/crates/rbitcoin-net/src/chain.rs +++ b/crates/rbitcoin-net/src/chain.rs @@ -2039,15 +2039,22 @@ impl ChainHub { tip_accept_stats_reset(&self.query); let t_wall = std::time::Instant::now(); let now = self.clock.now_secs(); - rbitcoin_consensus::with_now(now, || { - accept_and_connect_block_preverified( + let _ = rbitcoin_consensus::with_now(now, || loop { + match accept_and_connect_block_preverified( &self.query, &self.params, Height(height), &block, self.milestone, &preverified, - ) + ) { + Ok(fk) => return Ok(fk), + Err(e) if e.is_uring_session_fault() => { + self.query.uring_recover_or_abort("tip-connect"); + continue; + } + Err(e) => return Err(e), + } }) .map_err(|e| { let reason = rbitcoin_consensus::block_reject_reason(&e); diff --git a/crates/rbitcoin-net/src/ibd/confirm/mod.rs b/crates/rbitcoin-net/src/ibd/confirm/mod.rs index 3318401216..79ba0a9193 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/mod.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/mod.rs @@ -296,6 +296,11 @@ impl ConfirmFeed { self.cv.notify_one(); } + pub(crate) fn requeue_hashes(&self, items: impl IntoIterator) { + let batch: Vec<_> = items.into_iter().map(|(h, ha)| (h, ha, None)).collect(); + self.requeue_wire(&batch); + } + /// Write (or permanent reject) finished — height may be re-offered only after /// tip moves past it (or a future requeue path). pub(crate) fn finish(&self, heights: impl IntoIterator) { @@ -383,6 +388,9 @@ impl ConfirmRejectClass { { return Self::Cascade; } + if rbitcoin_store::StoreError::is_uring_session_fault_msg(&s) { + return Self::EngineFault; + } if s.contains("parent create_fk unresolved") || s.contains("spend annotate missing pin denserels") || s.contains("corrupt record") @@ -421,6 +429,7 @@ impl ConfirmRejectClass { | ConsensusError::BadHeader(_) | ConsensusError::MissingPrevout => Self::ConsensusInvalid, ConsensusError::Store(StoreError::Cancelled(_)) => Self::Cancelled, + ConsensusError::Store(e) if e.is_uring_session_fault() => Self::EngineFault, ConsensusError::Store(StoreError::Io { .. }) => Self::EngineFault, ConsensusError::Store(StoreError::Corrupt(m)) => Self::from_err_str(m), ConsensusError::Store(_) => Self::EngineFault, @@ -470,6 +479,68 @@ impl ConfirmRejectClass { } } +fn requeue_on_uring_recover( + query: &rbitcoin_query::Query, + feed: &ConfirmFeed, + session_fault: bool, + reason: &'static str, + heights: impl IntoIterator, +) -> bool { + if !session_fault { + return false; + } + query.uring_recover_or_abort(reason); + feed.requeue_hashes(heights); + true +} + +pub(crate) fn lookup_ready_hash(feed: &ConfirmFeed, height: u32) -> Option { + feed.inner + .lock() + .ok() + .and_then(|g| g.ready.get(&height).map(|(h, _)| *h)) +} + +const LOOKUP_FAULT_HALT_AFTER: u32 = 8; + +#[derive(Debug, Default)] +pub(crate) struct LookupFaultPolicy { + consecutive: u32, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum LookupFaultAction { + Ignore, + Warn, + RejectEngineFault, +} + +impl LookupFaultPolicy { + pub(crate) fn on_success(&mut self) { + self.consecutive = 0; + } + + pub(crate) fn on_err(&mut self, e: &rbitcoin_consensus::ConsensusError) -> LookupFaultAction { + use rbitcoin_consensus::ConsensusError; + let backpressure = matches!(e, ConsensusError::Store(se) if se.is_io_backpressure()); + if backpressure { + return LookupFaultAction::Ignore; + } + self.consecutive = self.consecutive.saturating_add(1); + if self.consecutive >= LOOKUP_FAULT_HALT_AFTER { + LookupFaultAction::RejectEngineFault + } else { + LookupFaultAction::Warn + } + } +} + +static LOOKUP_WAVE_FAULTS: AtomicU64 = AtomicU64::new(0); + +pub(crate) fn take_lookup_wave_faults() -> u64 { + LOOKUP_WAVE_FAULTS.swap(0, Ordering::Relaxed) +} + pub(crate) enum ConfirmEvent { /// Tip advanced; hash is the confirmed block. Accepted { hash: BlockHash }, @@ -1445,7 +1516,19 @@ pub(crate) fn spawn_confirm_engine( feed_wb.finish(heights_hashes.iter().map(|(h, _)| *h)); continue; } - // Reset load-ahead so reserved create fks / last_loaded do not drift. + if requeue_on_uring_recover( + &hub_wb.query, + &feed_wb, + e.is_uring_session_fault(), + "ibd-confirm-write", + meta.iter().copied(), + ) { + load_ahead_reset_wb.store(true, Ordering::Release); + warn!( + "ibd: confirm write uring recover @ {height} batch_parts={parts}: {e}" + ); + continue; + } load_ahead_reset_wb.store(true, Ordering::Release); feed_wb.finish(heights_hashes.iter().map(|(h, _)| *h)); loop_stats_wb @@ -1454,12 +1537,13 @@ pub(crate) fn spawn_confirm_engine( warn!( "ibd: confirm write reject @ {height} batch_parts={parts}: {e}" ); + let class = ConfirmRejectClass::from_consensus(&e); let _ = emit_confirm_reject( &event_tx_wb, &feed_wb, height, hash, - ConfirmRejectClass::from_consensus(&e), + class, msg, heights_hashes.len(), ); @@ -1529,6 +1613,18 @@ pub(crate) fn spawn_confirm_engine( info!("ibd: confirm scripts aborted: {msg}"); return false; } + if requeue_on_uring_recover( + &hub_sc.query, + &feed_sc, + e.is_uring_session_fault(), + "ibd-confirm-scripts", + meta.heights_hashes.iter().map(|(h, raw)| { + (*h, BlockHash::from_byte_array(*raw)) + }), + ) { + warn!("ibd: confirm scripts uring recover @ {}: {e}", meta.first_h); + return true; + } let (height, hash) = meta .heights_hashes .first() @@ -1698,6 +1794,19 @@ pub(crate) fn spawn_confirm_engine( let _ = scripts.join(); return; } + if requeue_on_uring_recover( + &hub_load.query, + &feed_load, + e.is_uring_session_fault(), + "ibd-confirm-load", + wire_batch.iter().map(|(h, ha, _)| (*h, *ha)), + ) { + warn!( + "ibd: confirm load stamp uring recover {first} @ {expect_h}: {e}", + first = wire_batch[0].1 + ); + continue; + } let first_hash = wire_batch[0].1; if wire_batch.len() > 1 { let tail: Vec<(u32, BlockHash, Option)> = @@ -1840,6 +1949,18 @@ pub(crate) fn spawn_confirm_engine( let _ = scripts.join(); return; } + if requeue_on_uring_recover( + &hub_load.query, + &feed_load, + rbitcoin_store::StoreError::is_uring_session_fault_msg(&msg), + "ibd-confirm-load", + heights_hashes.iter().copied(), + ) { + warn!( + "ibd: confirm load uring recover {first_hash} @ {expect_h}: {e}" + ); + continue; + } if heights_hashes.len() > 1 { let tail: Vec<(u32, BlockHash, Option)> = heights_hashes @@ -1882,6 +2003,7 @@ pub(crate) fn spawn_confirm_engine( .expect("spawn ibd-confirm-load"); let queues_lookup = Arc::clone(&queues); + let event_tx_lookup = event_tx.clone(); let lookup_join = std::thread::Builder::new() .name("ibd-confirm-lookup".into()) .spawn(move || { @@ -1889,6 +2011,7 @@ pub(crate) fn spawn_confirm_engine( let queues_lookup = queues_lookup; let stats = hub.query.confirm_stats_arc(); let mut disco_seen = 0u64; + let mut lookup_faults = LookupFaultPolicy::default(); loop { if feed.stopped() { break; @@ -1946,6 +2069,7 @@ pub(crate) fn spawn_confirm_engine( max_inputs, ) { Ok(wave) if !wave.items.is_empty() => { + lookup_faults.on_success(); did = true; let counts: Vec = wave .items @@ -2006,17 +2130,47 @@ pub(crate) fn spawn_confirm_engine( } feed.notify(); } - Ok(_) => {} + Ok(_) => { + lookup_faults.on_success(); + } Err(e) => { - let backpressure = matches!( - &e, - rbitcoin_consensus::ConsensusError::Store(se) - if se.is_io_backpressure() - ); - if backpressure { - debug!("ibd: bq resolve wave: {e}"); + if e.is_uring_session_fault() { + hub.query.uring_recover_or_abort("ibd-confirm-lookup"); + lookup_faults.on_success(); + warn!("ibd: bq resolve wave uring recover: {e}"); } else { - warn!("ibd: bq resolve wave: {e}"); + match lookup_faults.on_err(&e) { + LookupFaultAction::Ignore => { + debug!("ibd: bq resolve wave: {e}"); + } + LookupFaultAction::Warn => { + LOOKUP_WAVE_FAULTS.fetch_add(1, Ordering::Relaxed); + warn!("ibd: bq resolve wave: {e}"); + } + LookupFaultAction::RejectEngineFault => { + LOOKUP_WAVE_FAULTS.fetch_add(1, Ordering::Relaxed); + warn!("ibd: bq resolve wave halt: {e}"); + let height = wave_h.first().copied().unwrap_or(path_lo); + match lookup_ready_hash(&feed, height) { + Some(hash) => { + let _ = emit_confirm_reject( + &event_tx_lookup, + &feed, + height, + hash, + ConfirmRejectClass::EngineFault, + e.to_string(), + 1, + ); + } + None => { + warn!( + "ibd: bq resolve wave halt skip emit @{height} (no ready hash)" + ); + } + } + } + } } } } diff --git a/crates/rbitcoin-net/src/ibd/confirm/tests.rs b/crates/rbitcoin-net/src/ibd/confirm/tests.rs index 0d29b5a481..7dcd35ea51 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/tests.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/tests.rs @@ -1245,3 +1245,100 @@ fn plan_epoch_stale_after_clear() { ); assert!(!feed.plan_epoch_stale(99), "unknown height is not stale"); } + +#[test] +fn write_session_fault_is_engine_fault_and_requeue_puts_ready() { + use rbitcoin_consensus::ConsensusError; + use rbitcoin_store::StoreError; + + let undrained = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring undrained")); + assert_eq!( + super::ConfirmRejectClass::from_consensus(&undrained), + super::ConfirmRejectClass::EngineFault + ); + let leftover = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring leftover cqe")); + assert_eq!( + super::ConfirmRejectClass::from_consensus(&leftover), + super::ConfirmRejectClass::EngineFault + ); + let io = ConsensusError::Store(StoreError::io("/tmp/x", std::io::Error::other("disk"))); + assert_eq!( + super::ConfirmRejectClass::from_consensus(&io), + super::ConfirmRejectClass::EngineFault + ); + + let feed = ConfirmFeed::new(); + let hash = bitcoin::BlockHash::from_byte_array([1u8; 32]); + { + let mut g = feed.inner.lock().unwrap(); + g.inflight.insert(10); + } + feed.requeue_hashes(std::iter::once((10, hash))); + let g = feed.inner.lock().unwrap(); + assert!(g.ready.contains_key(&10)); + assert!(!g.inflight.contains(&10)); +} + +#[test] +fn requeue_on_uring_recover_credits_then_skips_non_fault() { + let (_d, q) = rbitcoin_query::testutil::tiny_query_labeled("requeue-uring"); + let feed = ConfirmFeed::new(); + let hash = bitcoin::BlockHash::from_byte_array([3u8; 32]); + { + let mut g = feed.inner.lock().unwrap(); + g.inflight.insert(7); + } + assert!(super::requeue_on_uring_recover( + &q, + &feed, + true, + "test", + std::iter::once((7, hash)), + )); + { + let g = feed.inner.lock().unwrap(); + assert!(g.ready.contains_key(&7)); + assert!(!g.inflight.contains(&7)); + } + assert!(!super::requeue_on_uring_recover( + &q, + &feed, + false, + "test", + std::iter::empty(), + )); + assert_eq!( + q.uring_recover("again"), + rbitcoin_query::UringRecover::Exhausted + ); +} + +#[test] +fn lookup_fault_policy_io_halt() { + use super::{LookupFaultAction, LookupFaultPolicy}; + use rbitcoin_consensus::ConsensusError; + use rbitcoin_store::StoreError; + + let mut p = LookupFaultPolicy::default(); + let bp = ConsensusError::Store(StoreError::BudgetFull("io_uring SQ")); + assert_eq!(p.on_err(&bp), LookupFaultAction::Ignore); + let io = ConsensusError::Store(StoreError::io("/tmp/x", std::io::Error::other("disk"))); + for _ in 0..7 { + assert_eq!(p.on_err(&io), LookupFaultAction::Warn); + } + assert_eq!(p.on_err(&io), LookupFaultAction::RejectEngineFault); + p.on_success(); + assert_eq!(p.on_err(&io), LookupFaultAction::Warn); +} + +#[test] +fn lookup_ready_hash_none_when_missing() { + let feed = ConfirmFeed::new(); + assert!(super::lookup_ready_hash(&feed, 10).is_none()); + let h = bitcoin::BlockHash::from_byte_array([2u8; 32]); + { + let mut g = feed.inner.lock().unwrap(); + g.ready.insert(10, (h, None)); + } + assert_eq!(super::lookup_ready_hash(&feed, 10), Some(h)); +} 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 0e6d617147..7b653ce1e9 100644 --- a/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs +++ b/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs @@ -122,6 +122,24 @@ fn confirm_reject_class_matches_substring_table() { ))), ConfirmRejectClass::EngineFault ); + assert_eq!( + ConfirmRejectClass::from_consensus(&ConsensusError::Store(StoreError::Corrupt( + "invariant: io_uring undrained" + ))), + ConfirmRejectClass::EngineFault + ); + assert_eq!( + ConfirmRejectClass::from_err_str("invariant: io_uring undrained"), + ConfirmRejectClass::EngineFault + ); + assert_eq!( + ConfirmRejectClass::from_err_str("invariant: io_uring leftover cqe"), + ConfirmRejectClass::EngineFault + ); + assert_eq!( + ConfirmRejectClass::from_err_str("io_uring submit failed"), + ConfirmRejectClass::EngineFault + ); } /// `body.rejected ⊆ consensus-invalid set` — Cascade / SoftWire / EngineFault diff --git a/crates/rbitcoin-net/src/ibd/perf_log.rs b/crates/rbitcoin-net/src/ibd/perf_log.rs index eccafff52f..32dc86e207 100644 --- a/crates/rbitcoin-net/src/ibd/perf_log.rs +++ b/crates/rbitcoin-net/src/ibd/perf_log.rs @@ -411,6 +411,9 @@ pub(crate) struct IbdPerfSample { pub owned: ProcessOwnedSizes, /// Confirm load/scripts/write queue contents + feed. pub conf_pipe: ConfirmPipelineSizes, + pub uring_recover_n: u64, + pub uring_slow_drain: u64, + pub lookup_faults: u64, } impl Default for IbdPerfSample { @@ -622,6 +625,9 @@ impl Default for IbdPerfSample { work: WorkStructureSizes::default(), owned: ProcessOwnedSizes::default(), conf_pipe: ConfirmPipelineSizes::default(), + uring_recover_n: 0, + uring_slow_drain: 0, + lookup_faults: 0, } } } @@ -1063,6 +1069,9 @@ pub(crate) fn sample( work, owned, conf_pipe, + uring_recover_n: rbitcoin_store::uring_recover_count(), + uring_slow_drain: rbitcoin_store::uring_slow_drain_count(), + lookup_faults: super::confirm::take_lookup_wave_faults(), } } @@ -1240,6 +1249,9 @@ pub(crate) fn format_info(s: &IbdPerfSample) -> String { s.conf_write_q_hwm, s.conf_write_q_cap, )); + append_nz(&mut out, "uring_recover", s.uring_recover_n); + append_nz(&mut out, "slow_drain", s.uring_slow_drain); + append_nz(&mut out, "lookup_faults", s.lookup_faults); let _ = thr_lookup_wait; if s.stamp_struct_ms > 0 || s.stamp_prepare_ms > 0 @@ -2092,8 +2104,14 @@ mod tests { s.dominant = "confirm"; s.live = Some((100, 32, 8000, 1500)); s.confirm_reject_stops = 2; + s.uring_recover_n = 1; + s.uring_slow_drain = 2; + s.lookup_faults = 3; let line = format_info(&s); assert!(line.starts_with("ibd: perf "), "{line}"); + assert!(line.contains("uring_recover=1"), "{line}"); + assert!(line.contains("slow_drain=2"), "{line}"); + assert!(line.contains("lookup_faults=3"), "{line}"); assert!(line.contains("inflight=3/256"), "{line}"); assert!(!line.contains("body_soft="), "{line}"); assert!(!line.contains("body_pend="), "{line}"); diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 7e5c55b216..072fbef772 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/crates/rbitcoin-query/src/lib.rs @@ -59,6 +59,20 @@ pub use confirm_stats::{ pub type QueryError = StoreError; +/// Result of [`Query::uring_recover`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum UringRecover { + Recovered, + Exhausted, +} + +pub(crate) fn uring_recover_credit(last_tip: Option, tip: u32) -> bool { + match last_tip { + None => true, + Some(last) => tip.saturating_sub(last) >= Query::URING_RECOVER_MIN_TIP_GAP, + } +} + /// Cheap process-owned cache occupancy for IBD `ibd: sizes` (O(1) lens + brief locks). /// /// `conf_plans` is header plan occupancy in ConfirmParentCache. Pipeline pins / @@ -270,6 +284,8 @@ pub struct Query { disconnect_gen: AtomicU64, /// Confirm / archive / load window meters (`ibd: perf`). One instance per Query. confirm_stats: Arc, + /// Tip height of last in-process io_uring recover (`u32::MAX` = none). + uring_recover_tip: AtomicU32, } /// In-process hash→height map for the confirmed tip chain (~33 MiB raw at 1e6 tips). @@ -357,6 +373,7 @@ impl Query { disconnect_height: AtomicU32::new(0), disconnect_gen: AtomicU64::new(0), confirm_stats: Arc::new(ConfirmStats::default()), + uring_recover_tip: AtomicU32::new(u32::MAX), }; if let Some(tip) = q.tip_height() { let _ = q.ensure_height_by_hash_index(tip); @@ -1016,6 +1033,40 @@ impl Query { self.store.tip_height() } + /// Minimum tip advance between in-process io_uring recovers. + pub const URING_RECOVER_MIN_TIP_GAP: u32 = 1000; + + pub fn uring_recover(&self, reason: &'static str) -> UringRecover { + let tip = self.tip_height().map(|h| h.0).unwrap_or(0); + loop { + let last = self.uring_recover_tip.load(AtomicOrdering::Acquire); + let last_opt = if last == u32::MAX { None } else { Some(last) }; + if !uring_recover_credit(last_opt, tip) { + return UringRecover::Exhausted; + } + if self + .uring_recover_tip + .compare_exchange(last, tip, AtomicOrdering::AcqRel, AtomicOrdering::Acquire) + .is_ok() + { + rbitcoin_store::note_uring_recover(); + rbitcoin_log::warn!("ibd: uring recover tip={tip} reason={reason}"); + return UringRecover::Recovered; + } + } + } + + /// Take recover credit, or abort. Returns only after a credited recover. + pub fn uring_recover_or_abort(&self, reason: &'static str) { + match self.uring_recover(reason) { + UringRecover::Recovered => {} + UringRecover::Exhausted => { + let msg = format!("recover credit exhausted on {reason}"); + rbitcoin_store::abort_uring_unusable(&msg); + } + } + } + /// Highest height on the RAM fence. Not the in-flight prune HWM /// ([`Self::drain_and_fence_hi`] — drain can lag this). pub fn fence_tip_height(&self) -> Option { diff --git a/crates/rbitcoin-query/src/query_tests.rs b/crates/rbitcoin-query/src/query_tests.rs index f77c42afa7..6ce292e2c5 100644 --- a/crates/rbitcoin-query/src/query_tests.rs +++ b/crates/rbitcoin-query/src/query_tests.rs @@ -36,6 +36,47 @@ fn query_open_clears_strong_above_tip() { let _ = std::fs::remove_dir_all(&dir); } +#[test] +fn uring_recover_credit_gap() { + assert!(uring_recover_credit(None, 0)); + assert!(!uring_recover_credit(Some(0), 0)); + assert!(!uring_recover_credit(Some(0), 144)); + assert!(!uring_recover_credit(Some(0), 999)); + assert!(uring_recover_credit(Some(0), 1000)); + assert!(uring_recover_credit(Some(100), 1100)); +} + +#[test] +fn uring_recover_credit_once_per_window_does_not_repair() { + let (_d, q) = temp_query("uring-recover"); + let leftover = Fk(99); + q.store().strong_tx.set_strong(leftover, Fk(1)).unwrap(); + q.store().flush_class_c_tip().unwrap(); + assert!(q.store().strong_tx.is_strong(leftover).unwrap()); + assert_eq!(q.uring_recover("test"), UringRecover::Recovered); + assert!( + q.store().strong_tx.is_strong(leftover).unwrap(), + "in-process recover must not mutate Class C; leftover strong waits for open repair" + ); + q.store().strong_tx.set_strong(leftover, Fk(1)).unwrap(); + q.store().flush_class_c_tip().unwrap(); + assert_eq!(q.uring_recover("again"), UringRecover::Exhausted); +} + +#[test] +fn uring_recover_cas_second_claim_at_same_tip_is_exhausted() { + let (_d, q) = temp_query("uring-recover-cas"); + assert_eq!(q.uring_recover("a"), UringRecover::Recovered); + assert_eq!(q.uring_recover("b"), UringRecover::Exhausted); +} + +#[test] +fn uring_recover_or_abort_takes_credit() { + let (_d, q) = temp_query("uring-recover-or-abort"); + q.uring_recover_or_abort("test"); + assert_eq!(q.uring_recover("again"), UringRecover::Exhausted); +} + fn temp_query(label: &str) -> (crate::testutil::TempDir, Query) { crate::testutil::tiny_query_labeled(label) } diff --git a/crates/rbitcoin-store/src/error.rs b/crates/rbitcoin-store/src/error.rs index 1f88826b59..1e8d754c78 100644 --- a/crates/rbitcoin-store/src/error.rs +++ b/crates/rbitcoin-store/src/error.rs @@ -42,6 +42,19 @@ impl StoreError { pub fn is_io_backpressure(&self) -> bool { matches!(self, StoreError::BudgetFull(m) if m.contains("SQ")) } + + /// Completion-session failure (not on-disk corruption, not SQ backpressure). + pub fn is_uring_session_fault(&self) -> bool { + match self { + StoreError::Corrupt(m) => Self::is_uring_session_fault_msg(m), + _ => false, + } + } + + pub fn is_uring_session_fault_msg(m: &str) -> bool { + let s = m.to_ascii_lowercase(); + s.contains("io_uring") && !s.contains("linux-only") && !s.contains("unavailable") + } } impl fmt::Display for StoreError { @@ -136,4 +149,29 @@ mod tests { ); assert!(!StoreError::Corrupt("broken").is_io_backpressure()); } + + #[test] + fn uring_session_fault_strings() { + for m in [ + "invariant: io_uring undrained", + "invariant: io_uring wait timeout", + "invariant: io_uring session poisoned", + "invariant: io_uring unexpected cqe", + "invariant: io_uring cq overflow", + "io_uring submit_and_wait failed", + "invariant: io_uring leftover cqe", + "io_uring submit failed", + ] { + assert!(StoreError::Corrupt(m).is_uring_session_fault(), "{m}"); + } + assert!(!StoreError::BudgetFull("io_uring SQ").is_uring_session_fault()); + assert!(!StoreError::Unavailable.is_uring_session_fault()); + assert!(!StoreError::Corrupt("io_uring is Linux-only").is_uring_session_fault()); + assert!(!StoreError::Corrupt("broken").is_uring_session_fault()); + assert!(!StoreError::Io { + path: PathBuf::from("/tmp/x"), + source: io::Error::other("disk"), + } + .is_uring_session_fault()); + } } diff --git a/crates/rbitcoin-store/src/io_session_iocp.rs b/crates/rbitcoin-store/src/io_session_iocp.rs index f63bcda246..a387059257 100644 --- a/crates/rbitcoin-store/src/io_session_iocp.rs +++ b/crates/rbitcoin-store/src/io_session_iocp.rs @@ -189,16 +189,16 @@ impl IocpEngine { self.ready.drain(..).collect() } - pub(crate) fn wait_one_cqe(&mut self) -> Result<(), StoreError> { + pub(crate) fn wait_one_cqe_timeout(&mut self, timeout_ms: u32) -> bool { if self.ready.is_empty() { - self.poll(u32::MAX); + self.poll(timeout_ms); } - Ok(()) + !self.ready.is_empty() } pub(crate) fn wait_idle(&mut self) -> Result<(), StoreError> { // Caller harvests until pending is empty; one long wait then drain. - self.poll(50); + self.poll(100); Ok(()) } diff --git a/crates/rbitcoin-store/src/io_session_pool.rs b/crates/rbitcoin-store/src/io_session_pool.rs index b088c861ca..d70d8818fa 100644 --- a/crates/rbitcoin-store/src/io_session_pool.rs +++ b/crates/rbitcoin-store/src/io_session_pool.rs @@ -87,6 +87,10 @@ impl PoolEngine { } } + pub(crate) fn inflight(&self) -> usize { + self.cq.inflight.load(Ordering::Acquire) + } + pub(crate) fn push_pread( &self, handle: IoHandle, @@ -183,13 +187,6 @@ impl PoolEngine { } !q.is_empty() } - - pub(crate) fn wait_idle(&self) { - let mut q = self.cq.cqes.lock().unwrap_or_else(|e| e.into_inner()); - while self.cq.inflight.load(Ordering::Acquire) != 0 { - q = self.cq.cqe_cv.wait(q).unwrap_or_else(|e| e.into_inner()); - } - } } fn worker_loop(shared: Arc) { diff --git a/crates/rbitcoin-store/src/lib.rs b/crates/rbitcoin-store/src/lib.rs index d13a6a0c94..5ad3705880 100644 --- a/crates/rbitcoin-store/src/lib.rs +++ b/crates/rbitcoin-store/src/lib.rs @@ -102,3 +102,6 @@ pub use tx_table::{ OutputRecord, TxRecord, }; pub(crate) use uring_session::IoCtx; +pub use uring_session::{ + abort_uring_unusable, note_uring_recover, uring_recover_count, uring_slow_drain_count, +}; diff --git a/crates/rbitcoin-store/src/uring_session.rs b/crates/rbitcoin-store/src/uring_session.rs index 2c59eb1087..5d49c063eb 100644 --- a/crates/rbitcoin-store/src/uring_session.rs +++ b/crates/rbitcoin-store/src/uring_session.rs @@ -21,24 +21,135 @@ use crate::io_handle::IoHandle; use crate::io_session_pool::PoolEngine; use std::cell::Cell; use std::path::Path; +use std::sync::OnceLock; +use std::time::{Duration, Instant}; /// Default SQ/CQ depth for all store io_uring sessions (bulk, plan head-resolve, /// spend annotate). TLS rings open at this size; [`with_thread_local`] may grow /// if a caller requests more (none currently do). pub const DEFAULT_ENTRIES: u32 = 128; -/// Cap for [`UringSession::submit_and_wait_one`]. Lost CQEs must fail the wave -/// instead of parking the lookup thread for the rest of IBD. -const WAIT_ONE_TIMEOUT_SECS: u64 = 5; -const WAIT_ONE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(WAIT_ONE_TIMEOUT_SECS); - -/// [`UringSession::drain_all`] wait: 100 ms × 50 = 5 s. Short enough that a -/// poisoned ring fails the wave; long enough that 128×4 KiB g-page preads -/// under Class A write traffic are not a false `undrained`. +/// One `drain_all` wait window. Progress in a window resets the zero-CQE clock. +const DRAIN_WINDOW: Duration = Duration::from_millis(100); #[cfg(target_os = "linux")] const DRAIN_WAIT_NS: u32 = 100_000_000; -#[cfg(target_os = "linux")] -const DRAIN_MAX_SPINS: u32 = 50; +const DRAIN_SLOW_AFTER: Duration = Duration::from_secs(5); +const DRAIN_HARD_DEFAULT_SECS: u64 = 120; + +/// Drain wait policy: keep waiting while CQEs arrive; abort only after +/// [`DrainBudget::hard`] of zero completions. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum DrainVerdict { + Continue, + LogSlow, + HardCap, +} + +enum WaitOneWindow { + Ready, + TimedOut, + Idle, + #[cfg(target_os = "linux")] + Enter(StoreError), +} + +pub(crate) struct DrainBudget { + hard: Duration, + zero: Duration, + elapsed: Duration, + since_slow_log: Duration, +} + +impl DrainBudget { + pub(crate) fn new(hard: Duration) -> Self { + Self { + hard, + zero: Duration::ZERO, + elapsed: Duration::ZERO, + since_slow_log: Duration::ZERO, + } + } + + pub(crate) fn with_defaults() -> Self { + Self::new(drain_hard_cap()) + } + + pub(crate) fn elapsed(&self) -> Duration { + self.elapsed + } + + pub(crate) fn note(&mut self, dt: Duration, completed: usize) -> DrainVerdict { + self.elapsed = self.elapsed.saturating_add(dt); + self.since_slow_log = self.since_slow_log.saturating_add(dt); + if completed > 0 { + self.zero = Duration::ZERO; + } else { + self.zero = self.zero.saturating_add(dt); + if self.zero >= self.hard { + return DrainVerdict::HardCap; + } + } + if self.elapsed >= DRAIN_SLOW_AFTER && self.since_slow_log >= DRAIN_SLOW_AFTER { + self.since_slow_log = Duration::ZERO; + return DrainVerdict::LogSlow; + } + DrainVerdict::Continue + } +} + +fn drain_hard_cap() -> Duration { + #[cfg(test)] + if let Some(d) = TEST_DRAIN_HARD.with(|c| c.get()) { + return d; + } + static CAP: OnceLock = OnceLock::new(); + *CAP.get_or_init(|| { + let secs = std::env::var("RBITCOIN_URING_DRAIN_HARD_SECS") + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|&n| n > 0) + .unwrap_or(DRAIN_HARD_DEFAULT_SECS); + Duration::from_secs(secs) + }) +} + +#[cfg(test)] +thread_local! { + static TEST_DRAIN_HARD: Cell> = const { Cell::new(None) }; +} + +#[cfg(test)] +pub(crate) fn with_drain_hard_cap(hard: Duration, f: impl FnOnce() -> R) -> R { + TEST_DRAIN_HARD.with(|c| { + let prev = c.replace(Some(hard)); + let out = f(); + c.set(prev); + out + }) +} + +/// FATAL then abort. Operator may restart with `RBITCOIN_IO=pread`. +pub fn abort_uring_unusable(reason: &str) -> ! { + let thread = std::thread::current(); + let thread = thread.name().unwrap_or("unnamed"); + rbitcoin_log::error!( + "store: completion session unusable thread={thread} {reason} — aborting; restart with RBITCOIN_IO=pread if the filesystem or IO scheduling class cannot complete bulk IO" + ); + std::process::abort(); +} + +fn finish_drain_hard_cap(pending: usize) -> Result<(), StoreError> { + let thread = std::thread::current(); + let thread = thread.name().unwrap_or("unnamed"); + let reason = format!("drain hard cap pending={pending} thread={thread}"); + #[cfg(test)] + { + let _ = reason; + return Err(StoreError::Corrupt("invariant: io_uring undrained")); + } + #[cfg(not(test))] + abort_uring_unusable(&reason); +} /// Which completion backend [`UringSession`] opens. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -386,45 +497,75 @@ impl UringSession { /// Submit pending SQEs and wait for at least one CQE. Does not harvest. /// - /// Bounded: a lost CQE / phantom pending must not park lookup forever. + /// Slow-log after 5 s of zero completions; poison at the drain hard cap. + /// A lost CQE / phantom pending must not park lookup forever. pub fn submit_and_wait_one(&mut self) -> Result<(), StoreError> { self.check_live()?; if self.pending.is_empty() { return Ok(()); } + let mut budget = DrainBudget::with_defaults(); + loop { + let t0 = Instant::now(); + match self.wait_one_window() { + WaitOneWindow::Ready => return Ok(()), + WaitOneWindow::Idle => { + self.poisoned = true; + return Err(StoreError::Corrupt("invariant: io_uring wait timeout")); + } + #[cfg(target_os = "linux")] + WaitOneWindow::Enter(err) => { + self.poisoned = true; + return Err(err); + } + WaitOneWindow::TimedOut => match budget.note(t0.elapsed(), 0) { + DrainVerdict::Continue => {} + DrainVerdict::LogSlow => self.note_slow_drain(budget.elapsed()), + DrainVerdict::HardCap => { + self.poisoned = true; + return Err(StoreError::Corrupt("invariant: io_uring wait timeout")); + } + }, + } + } + } + + fn wait_one_window(&mut self) -> WaitOneWindow { match &mut self.backend { #[cfg(target_os = "linux")] SessionBackend::Uring(ring) => { ring.submission().sync(); - let ts = io_uring::types::Timespec::new().sec(WAIT_ONE_TIMEOUT_SECS); + let ts = io_uring::types::Timespec::new().nsec(DRAIN_WAIT_NS); let args = io_uring::types::SubmitArgs::new().timespec(&ts); - loop { - match ring.submitter().submit_with_args(1, &args) { - Ok(_) => return Ok(()), - Err(e) => { - if e.raw_os_error() == Some(libc::ETIME) { - self.poisoned = true; - return Err(StoreError::Corrupt( - "invariant: io_uring wait timeout", - )); - } - if let Some(err) = map_enter_err(&e) { - self.poisoned = true; - return Err(err); - } + match ring.submitter().submit_with_args(1, &args) { + Ok(_) => WaitOneWindow::Ready, + Err(e) if e.raw_os_error() == Some(libc::ETIME) => WaitOneWindow::TimedOut, + Err(e) => { + if let Some(err) = map_enter_err(&e) { + WaitOneWindow::Enter(err) + } else { + WaitOneWindow::TimedOut } } } } SessionBackend::Pool(pool) => { - if !pool.wait_one_cqe_timeout(WAIT_ONE_TIMEOUT) { - self.poisoned = true; - return Err(StoreError::Corrupt("invariant: io_uring wait timeout")); + if pool.wait_one_cqe_timeout(DRAIN_WINDOW) { + WaitOneWindow::Ready + } else if pool.inflight() == 0 { + WaitOneWindow::Idle + } else { + WaitOneWindow::TimedOut } - Ok(()) } #[cfg(windows)] - SessionBackend::Iocp(eng) => eng.wait_one_cqe(), + SessionBackend::Iocp(eng) => { + if eng.wait_one_cqe_timeout(100) { + WaitOneWindow::Ready + } else { + WaitOneWindow::TimedOut + } + } } } @@ -497,7 +638,11 @@ impl UringSession { /// harvest of only-ready CQEs is not enough — the kernel may still write /// into buffers for unfinished SQEs (use-after-free → SIGSEGV). pub fn drain_all(&mut self) -> Result<(), StoreError> { - self.drain_all_harvest()?; + self.drain_all_inner(true) + } + + fn drain_all_inner(&mut self, fail_closed: bool) -> Result<(), StoreError> { + self.drain_all_harvest(fail_closed)?; if let Err(e) = self.pending.assert_drained() { let already = self.poisoned; self.poisoned = true; @@ -509,33 +654,122 @@ impl UringSession { Ok(()) } - fn drain_all_harvest(&mut self) -> Result<(), StoreError> { - enum Harvest { - Uds(Vec), + fn drain_all_harvest(&mut self, fail_closed: bool) -> Result<(), StoreError> { + match &self.backend { #[cfg(target_os = "linux")] - Uring, + SessionBackend::Uring(_) => self.drain_all_uring(fail_closed), + SessionBackend::Pool(_) => self.drain_all_pool(fail_closed), #[cfg(windows)] - IocpErr(StoreError), + SessionBackend::Iocp(_) => self.drain_all_iocp(fail_closed), } - let harvested = match &mut self.backend { - #[cfg(target_os = "linux")] - SessionBackend::Uring(_) => Harvest::Uring, - SessionBackend::Pool(pool) => { - pool.wait_idle(); - Harvest::Uds(pool.harvest_ready().into_iter().map(|(ud, _)| ud).collect()) + } + + fn note_slow_drain(&self, waited: Duration) { + URING_METERS + .slow_drain + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let thread = std::thread::current(); + let thread = thread.name().unwrap_or("unnamed"); + rbitcoin_log::warn!( + "store: io_uring drain slow pending={} waited={:?} thread={thread}", + self.pending.len(), + waited, + ); + } + + fn apply_drain_window( + &mut self, + budget: &mut DrainBudget, + completed: usize, + dt: Duration, + fail_closed: bool, + ) -> Result<(), StoreError> { + match budget.note(dt, completed) { + DrainVerdict::Continue => Ok(()), + DrainVerdict::LogSlow => { + self.note_slow_drain(budget.elapsed()); + Ok(()) } - #[cfg(windows)] - SessionBackend::Iocp(eng) => match eng.wait_idle() { - Ok(()) => Harvest::Uds(eng.harvest_ready().into_iter().map(|(ud, _)| ud).collect()), - Err(e) => Harvest::IocpErr(e), - }, - }; - match harvested { - Harvest::Uds(uds) => self.apply_drain_cqes(uds), - #[cfg(target_os = "linux")] - Harvest::Uring => self.drain_all_uring(), - #[cfg(windows)] - Harvest::IocpErr(e) => Err(e), + DrainVerdict::HardCap => { + self.poisoned = true; + if fail_closed { + finish_drain_hard_cap(self.pending.len()) + } else { + let thread = std::thread::current(); + let thread = thread.name().unwrap_or("unnamed"); + rbitcoin_log::warn!( + "store: io_uring drain hard cap on drop pending={} thread={thread}", + self.pending.len() + ); + Err(StoreError::Corrupt("invariant: io_uring undrained")) + } + } + } + } + + fn drain_all_pool(&mut self, fail_closed: bool) -> Result<(), StoreError> { + let mut budget = DrainBudget::with_defaults(); + loop { + let inflight = match &self.backend { + SessionBackend::Pool(pool) => pool.inflight(), + #[cfg(target_os = "linux")] + SessionBackend::Uring(_) => 0, + #[cfg(windows)] + SessionBackend::Iocp(_) => 0, + }; + if inflight == 0 { + let uds = match &mut self.backend { + SessionBackend::Pool(pool) => { + pool.harvest_ready().into_iter().map(|(ud, _)| ud).collect() + } + #[cfg(target_os = "linux")] + SessionBackend::Uring(_) => Vec::new(), + #[cfg(windows)] + SessionBackend::Iocp(_) => Vec::new(), + }; + return self.apply_drain_cqes(uds); + } + let before = self.pending.len(); + let t0 = Instant::now(); + let uds = match &mut self.backend { + SessionBackend::Pool(pool) => { + let _ = pool.wait_one_cqe_timeout(DRAIN_WINDOW); + pool.harvest_ready().into_iter().map(|(ud, _)| ud).collect() + } + #[cfg(target_os = "linux")] + SessionBackend::Uring(_) => Vec::new(), + #[cfg(windows)] + SessionBackend::Iocp(_) => Vec::new(), + }; + self.apply_drain_cqes(uds)?; + let completed = before.saturating_sub(self.pending.len()); + self.apply_drain_window(&mut budget, completed, t0.elapsed(), fail_closed)?; + } + } + + #[cfg(windows)] + fn drain_all_iocp(&mut self, fail_closed: bool) -> Result<(), StoreError> { + let mut budget = DrainBudget::with_defaults(); + loop { + let before = self.pending.len(); + let t0 = Instant::now(); + let wait_res = match &mut self.backend { + SessionBackend::Iocp(eng) => eng.wait_idle(), + _ => Ok(()), + }; + wait_res?; + let uds = match &mut self.backend { + SessionBackend::Iocp(eng) => { + eng.harvest_ready().into_iter().map(|(ud, _)| ud).collect() + } + _ => Vec::new(), + }; + self.apply_drain_cqes(uds)?; + if self.pending.is_empty() { + return Ok(()); + } + let completed = before.saturating_sub(self.pending.len()); + self.apply_drain_window(&mut budget, completed, t0.elapsed(), fail_closed)?; } } @@ -568,54 +802,30 @@ impl UringSession { } #[cfg(target_os = "linux")] - fn drain_all_uring(&mut self) -> Result<(), StoreError> { + fn drain_all_uring(&mut self, fail_closed: bool) -> Result<(), StoreError> { let mut unexpected = false; let mut overflow = 0u32; let mut enter_err = None; - { - let SessionBackend::Uring(ring) = &mut self.backend else { - return Ok(()); - }; - ring.submission().sync(); - let mut spins = 0u32; - while !self.pending.is_empty() && spins < DRAIN_MAX_SPINS { - spins += 1; - let ts = io_uring::types::Timespec::new().nsec(DRAIN_WAIT_NS); - let args = io_uring::types::SubmitArgs::new().timespec(&ts); - match ring.submitter().submit_with_args(1, &args) { - Ok(_) => {} - Err(e) if e.raw_os_error() == Some(libc::ETIME) => {} - Err(e) => { - if let Some(err) = map_enter_err(&e) { - enter_err = Some(err); - break; - } - } - } - overflow = { - let cq = ring.completion(); - cq.overflow() - }; - ring.completion().sync(); - for cqe in ring.completion() { - if self.pending.expect_cqe(cqe.user_data()).is_err() { - unexpected = true; - } - } - if overflow != 0 || unexpected { - break; - } + let mut budget = DrainBudget::with_defaults(); + self.uring_sync_sq(); + while !self.pending.is_empty() { + let before = self.pending.len(); + let t0 = Instant::now(); + self.uring_drain_window(&mut unexpected, &mut overflow, &mut enter_err); + if enter_err.is_some() || unexpected || overflow != 0 { + break; } - if enter_err.is_none() && !unexpected && overflow == 0 { - let _ = ring.submit(); - ring.completion().sync(); - for cqe in ring.completion() { - if self.pending.expect_cqe(cqe.user_data()).is_err() { - unexpected = true; - } - } + let completed = before.saturating_sub(self.pending.len()); + if let Err(e) = + self.apply_drain_window(&mut budget, completed, t0.elapsed(), fail_closed) + { + enter_err = Some(e); + break; } } + if enter_err.is_none() && !unexpected && overflow == 0 { + self.uring_final_harvest(&mut unexpected); + } if let Some(err) = enter_err { self.poisoned = true; return Err(err); @@ -630,11 +840,66 @@ impl UringSession { } Ok(()) } + + #[cfg(target_os = "linux")] + fn uring_sync_sq(&mut self) { + if let SessionBackend::Uring(ring) = &mut self.backend { + ring.submission().sync(); + } + } + + #[cfg(target_os = "linux")] + fn uring_drain_window( + &mut self, + unexpected: &mut bool, + overflow: &mut u32, + enter_err: &mut Option, + ) { + let SessionBackend::Uring(ring) = &mut self.backend else { + return; + }; + let ts = io_uring::types::Timespec::new().nsec(DRAIN_WAIT_NS); + let args = io_uring::types::SubmitArgs::new().timespec(&ts); + match ring.submitter().submit_with_args(1, &args) { + Ok(_) => {} + Err(e) if e.raw_os_error() == Some(libc::ETIME) => {} + Err(e) => { + if let Some(err) = map_enter_err(&e) { + *enter_err = Some(err); + return; + } + } + } + *overflow = { + let cq = ring.completion(); + cq.overflow() + }; + ring.completion().sync(); + for cqe in ring.completion() { + if self.pending.expect_cqe(cqe.user_data()).is_err() { + *unexpected = true; + } + } + } + + #[cfg(target_os = "linux")] + fn uring_final_harvest(&mut self, unexpected: &mut bool) { + let SessionBackend::Uring(ring) = &mut self.backend else { + return; + }; + let _ = ring.submit(); + ring.completion().sync(); + for cqe in ring.completion() { + if self.pending.expect_cqe(cqe.user_data()).is_err() { + *unexpected = true; + } + } + } } impl Drop for UringSession { fn drop(&mut self) { - let _ = self.drain_all(); + let _ = self.drain_all_inner(false); } } @@ -685,6 +950,25 @@ pub(crate) fn require_full_cqe(res: i32, want: usize, path: &Path) -> Result<(), } Ok(()) } + +pub fn note_uring_recover() { + URING_METERS + .recover_n + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); +} + +pub fn uring_slow_drain_count() -> u64 { + URING_METERS + .slow_drain + .load(std::sync::atomic::Ordering::Relaxed) +} + +pub fn uring_recover_count() -> u64 { + URING_METERS + .recover_n + .load(std::sync::atomic::Ordering::Relaxed) +} + /// Run `f` with this **OS thread's** long-lived io_uring session. /// /// - Opens once on first use; reopens only if `min_entries` exceeds the current @@ -718,8 +1002,6 @@ pub fn with_thread_local( DEPTH.with(|depth| { let d = depth.get(); if d != 0 { - // Hard error: nested temp rings caused plan head_fk regression - // (probe + record_range under the plan machine). Fail loud. panic!( "nested thread-local io_uring (depth={d}); \ fold IO into the outer with_thread_local machine \ @@ -851,6 +1133,8 @@ pub(crate) struct UringMeters { #[cfg(any(test, target_os = "linux"))] pub cq_overflow: std::sync::atomic::AtomicU64, pub idx_range_missing: std::sync::atomic::AtomicU64, + pub slow_drain: std::sync::atomic::AtomicU64, + pub recover_n: std::sync::atomic::AtomicU64, } static URING_METERS: UringMeters = UringMeters { @@ -859,6 +1143,8 @@ static URING_METERS: UringMeters = UringMeters { #[cfg(any(test, target_os = "linux"))] cq_overflow: std::sync::atomic::AtomicU64::new(0), idx_range_missing: std::sync::atomic::AtomicU64::new(0), + slow_drain: std::sync::atomic::AtomicU64::new(0), + recover_n: std::sync::atomic::AtomicU64::new(0), }; #[derive(Clone, Copy)] @@ -1079,6 +1365,71 @@ mod tests { assert!(UringPending::new().assert_drained().is_ok()); } + #[test] + fn drain_budget_progress_resets_zero_clock() { + let dt = Duration::from_millis(100); + let mut b = DrainBudget::new(Duration::from_millis(300)); + assert_eq!(b.note(dt, 0), DrainVerdict::Continue); + assert_eq!(b.note(dt, 0), DrainVerdict::Continue); + assert_eq!(b.note(dt, 1), DrainVerdict::Continue); + assert_eq!(b.note(dt, 0), DrainVerdict::Continue); + assert_eq!(b.note(dt, 0), DrainVerdict::Continue); + assert_eq!(b.note(dt, 0), DrainVerdict::HardCap); + } + + #[test] + fn drain_budget_five_seconds_with_progress_logs_slow_not_hard_cap() { + let dt = Duration::from_millis(100); + let mut b = DrainBudget::new(Duration::from_secs(120)); + for _ in 0..49 { + assert_eq!(b.note(dt, 1), DrainVerdict::Continue); + } + assert_eq!(b.note(dt, 1), DrainVerdict::LogSlow); + assert_eq!(b.note(dt, 1), DrainVerdict::Continue); + } + + #[test] + fn drain_budget_five_seconds_zero_progress_logs_slow_before_hard_cap() { + let dt = Duration::from_millis(100); + let mut b = DrainBudget::new(Duration::from_secs(120)); + for _ in 0..49 { + assert_eq!(b.note(dt, 0), DrainVerdict::Continue); + } + assert_eq!(b.note(dt, 0), DrainVerdict::LogSlow); + assert_eq!(b.note(dt, 0), DrainVerdict::Continue); + } + + #[test] + fn drain_budget_hard_cap_only_after_hard_duration_of_zero() { + let dt = Duration::from_millis(100); + let mut b = DrainBudget::new(Duration::from_millis(200)); + assert_eq!(b.note(dt, 0), DrainVerdict::Continue); + assert_eq!(b.note(dt, 0), DrainVerdict::HardCap); + } + + #[test] + fn drain_budget_eintr_spin_does_not_hit_hard_cap() { + let mut b = DrainBudget::new(Duration::from_millis(200)); + for _ in 0..10_000 { + assert_ne!(b.note(Duration::from_nanos(1), 0), DrainVerdict::HardCap); + } + } + + #[test] + fn drain_all_leftover_pending_hits_hard_cap_on_linux_or_pool() { + let mut session = UringSession::try_open(32) + .unwrap_or_else(|_| UringSession::try_open_kind(SessionKind::Pool, 32).expect("pool")); + session.pending.insert(1).unwrap(); + with_drain_hard_cap(Duration::from_millis(200), || { + match session.drain_all() { + Err(StoreError::Corrupt("invariant: io_uring undrained")) => {} + other => panic!("expected undrained after hard cap, got {other:?}"), + } + assert!(session.is_poisoned()); + drop(session); + }); + } + /// A CQE whose `user_data` was removed from `pending` is not a silent /// drain success — it is leftover harvest (`unexpected cqe`). #[test] diff --git a/docs/README.md b/docs/README.md index 6af08d92ba..52f5d13d25 100644 --- a/docs/README.md +++ b/docs/README.md @@ -31,7 +31,7 @@ quality backlog). | [`architecture.md`](./architecture.md) | Why this node is different (Core / Fulcrum contrasts). No stage-IO table copy. | | [`concurrency.md`](./concurrency.md) | Writer roles, publish order, body-queue, pins. Links invariants for leftover union. | | [`invariants.md`](./invariants.md) | Confirm stage IO (the **only** copy), leftover union, store start states S0–S4, no silent fallbacks. | -| [`crash-recovery.md`](./crash-recovery.md) | Tip-as-commit write order, kill-9, open repair. | +| [`crash-recovery.md`](./crash-recovery.md) | Tip-as-commit write order, kill-9, open repair, in-process uring recover. | | [`ibd-memory.md`](./ibd-memory.md) | Process RAM vs page cache; body-queue soft assign; production evict APIs. | | [`io-modality.md`](./io-modality.md) | `RBITCOIN_IO`, fd vs uring, TLS harvest, do-not-flatten machines, host A/B. | | [`heads.md`](./heads.md) | Which head file / module (tx / header / SH). | diff --git a/docs/crash-recovery.md b/docs/crash-recovery.md index 18fdc50c63..627db5727c 100644 --- a/docs/crash-recovery.md +++ b/docs/crash-recovery.md @@ -24,6 +24,8 @@ On open (in order): 2. **Tip-window revalidate** (Core `checkblocks=6`): first drop any trailing null `confirmed[]` slots (HWM ahead of last real tip), then the last six confirmed heights — `prev_fk`/hash chain, `header_txs` range bounds, merkle root from `txid.body`, and those six runs all-strong. On failure: clear bad Class A association and/or shrink tip to last good height, rebuild the fence, flush confirmed. 3. One `repair_class_c_above_tip`: unstrong bits **not on the fence** via complement ranges (holes + suffix until a zero page). Does **not** walk every set bit. Logs `class_c repair cleared= ranges= ms=` even when zero. +In-process completion-session recover (IBD write/lookup/load/scripts and tip connect) consumes a 1000-height credit and requeues; it does **not** run `repair_class_c_above_tip` (lookup would race the write thread's strong-before-fence window). Abandoned leftover strong is the same as kill-9 and is repaired on the next open. Does **not** truncate Class A. Tip stays the commit point. + Open revalidation runs in `Query::open_or_create` **before** P2P can extend tip. ### L2 write-behind + body queue (phase 6) diff --git a/docs/env-knobs.md b/docs/env-knobs.md index caff693e82..1a24020913 100644 --- a/docs/env-knobs.md +++ b/docs/env-knobs.md @@ -34,6 +34,7 @@ for signet/mainnet sync. **Not** CLI. | `RBITCOIN_SH_FORCE_REBUILD` | off | Sticky SH rebuild (also in OPERATOR) | | `RBITCOIN_SH_MERGE_WORKERS` | min(n-cpu, free-RAM/2 GiB) | Unsorted SH pack (`1` = serial). Unset = auto (see [`ibd-memory.md`](./ibd-memory.md)) | | `RBITCOIN_P2P_MAX_INBOUND` | 125 | Only if `--maxinbound` / conf omitted | +| `RBITCOIN_URING_DRAIN_HARD_SECS` | 120 | `drain_all` zero-CQE abort (Linux/pool/IOCP). Raise if completions are slow but still arriving | ## Hardcoded (no env) diff --git a/docs/invariants.md b/docs/invariants.md index 668eae9872..28fe56f102 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -124,6 +124,7 @@ published idx window. | `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 | Skeleton / in-flight miss | **Engine fault.** Requeue once, then halt IBD. Do not blacklist. | +| Lookup 8 consecutive non-session `Io` | live | Read-side store IO | **EngineFault** reject: `apply_engine_fault_reject` **dequeues** that height's body-queue payload and requeues the hash once; a second occurrence on the same hash **halts** IBD. Session-fault recover does **not** dequeue. Do not emit a reject when the feed has no ready hash. | | false PrevoutSpent | identity | schema-13 zero pin id | plan reverse map / lookup `txid.body` only | ## Why there is no leftover pending map diff --git a/docs/io-modality.md b/docs/io-modality.md index baaa91c3a4..bf104b1b94 100644 --- a/docs/io-modality.md +++ b/docs/io-modality.md @@ -49,11 +49,24 @@ if the session is poisoned or leftover cannot drain (probe / idx / BDZ g-pages / rel preads share the ring). Held idx fill fails closed on that error (no libc fallback on a dirty ring). Undrained / unexpected / wait-timeout **poisons** the session and drops the TLS ring so the next wave -opens a new one. `submit_and_wait_one` is capped (5 s). Drain before SQE -buffers drop (spend annotate `DrainOnDrop`). Per-op short/errno on a live -session still libc-completes that op; libc fail is `StoreError::io`. -`RBITCOIN_IO=pread` is the only whole-batch pread fallback (session -unavailable also falls back). +opens a new one. `submit_and_wait_one` shares the drain budget (slow-log at +5 s, poison at the hard cap). `drain_all` on every TLS session (Linux, pool, +IOCP) waits while CQEs keep arriving; after 5 s it logs `store: io_uring +drain slow`. Zero completions for `RBITCOIN_URING_DRAIN_HARD_SECS` (default +120) abort **explicit** `drain_all` / `DrainOnDrop` (buffers still pinned). +Session `Drop` waits the same budget then poisons without aborting so a +stalled device cannot `abort` from a destructor. There is **no** runtime +switch to `pread`. Drain before SQE buffers drop (spend annotate +`DrainOnDrop`). Per-op short/errno on a live session still libc-completes +that op; libc fail is `StoreError::io`. `RBITCOIN_IO=pread` is the only +whole-batch pread fallback (session unavailable also falls back; operator +restart after a drain abort). + +IBD write / lookup / load / scripts and post-IBD tip connect: a session fault +after a successful drain recovers once per 1000-height window (credit CAS; +**no** in-process Class C repair — leftover strong is the open-repair case). +The next wave opens a new TLS ring. A second stall in that window aborts and +names `RBITCOIN_IO=pread`. ### Do not flatten custom machines