From e9e6d03ca52f853285ab150fdd0c6a3f3fa88ffa Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 11:33:19 -0700 Subject: [PATCH 01/13] store,net: name io_uring session faults as EngineFault Inner strings like "invariant: io_uring undrained" lack "corrupt record", so from_err_str treated them as Cascade. Recover/halt need EngineFault. Pin each session-fault string on StoreError and the confirm reject map. Co-authored-by: Cursor --- crates/rbitcoin-net/src/ibd/confirm/mod.rs | 4 ++ .../src/ibd/events/confirm_reject_tests.rs | 14 +++++++ crates/rbitcoin-store/src/error.rs | 39 +++++++++++++++++++ 3 files changed, 57 insertions(+) diff --git a/crates/rbitcoin-net/src/ibd/confirm/mod.rs b/crates/rbitcoin-net/src/ibd/confirm/mod.rs index 3318401216..646579165e 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/mod.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/mod.rs @@ -383,6 +383,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 +424,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, 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..89d0fe1239 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,20 @@ 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 wait timeout"), + ConfirmRejectClass::EngineFault + ); } /// `body.rejected ⊆ consensus-invalid set` — Cascade / SoftWire / EngineFault diff --git a/crates/rbitcoin-store/src/error.rs b/crates/rbitcoin-store/src/error.rs index 1f88826b59..088b32385a 100644 --- a/crates/rbitcoin-store/src/error.rs +++ b/crates/rbitcoin-store/src/error.rs @@ -42,6 +42,23 @@ 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 { + m.contains("io_uring undrained") + || m.contains("io_uring wait timeout") + || m.contains("io_uring session poisoned") + || m.contains("io_uring unexpected cqe") + || m.contains("io_uring cq overflow") + || m.contains("io_uring submit_and_wait failed") + } } impl fmt::Display for StoreError { @@ -136,4 +153,26 @@ 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", + ] { + 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("broken").is_uring_session_fault()); + assert!(!StoreError::Io { + path: PathBuf::from("/tmp/x"), + source: io::Error::other("disk"), + } + .is_uring_session_fault()); + } } From 976c884b101abda0497f4584206a41c4b5f3a070 Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 11:40:38 -0700 Subject: [PATCH 02/13] store: drain_all waits while CQEs arrive; abort on 120s stall A 5s spin used to return with leftover SQEs so callers dropped in-flight buffers. Every UringSession now uses DrainBudget: progress resets the zero-CQE clock, slow drains log, and zero completions for RBITCOIN_URING_DRAIN_HARD_SECS (default 120) abort. Tests hit HardCap as undrained instead of abort(). Co-authored-by: Cursor --- crates/rbitcoin-store/src/io_session_iocp.rs | 2 +- crates/rbitcoin-store/src/io_session_pool.rs | 11 +- crates/rbitcoin-store/src/lib.rs | 1 + crates/rbitcoin-store/src/uring_session.rs | 390 +++++++++++++++---- 4 files changed, 328 insertions(+), 76 deletions(-) diff --git a/crates/rbitcoin-store/src/io_session_iocp.rs b/crates/rbitcoin-store/src/io_session_iocp.rs index f63bcda246..af9c070fa3 100644 --- a/crates/rbitcoin-store/src/io_session_iocp.rs +++ b/crates/rbitcoin-store/src/io_session_iocp.rs @@ -198,7 +198,7 @@ impl IocpEngine { 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..ef7791f57c 100644 --- a/crates/rbitcoin-store/src/lib.rs +++ b/crates/rbitcoin-store/src/lib.rs @@ -101,4 +101,5 @@ pub use tx_table::{ encode_packed_tx, encode_packed_tx_with_secret, spend_meta_backend, spent_abs, InputRecord, OutputRecord, TxRecord, }; +pub use uring_session::abort_uring_unusable; pub(crate) use uring_session::IoCtx; diff --git a/crates/rbitcoin-store/src/uring_session.rs b/crates/rbitcoin-store/src/uring_session.rs index 2c59eb1087..08a973eb07 100644 --- a/crates/rbitcoin-store/src/uring_session.rs +++ b/crates/rbitcoin-store/src/uring_session.rs @@ -21,6 +21,7 @@ use crate::io_handle::IoHandle; use crate::io_session_pool::PoolEngine; use std::cell::Cell; use std::path::Path; +use std::time::Duration; /// 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 @@ -32,13 +33,124 @@ pub const DEFAULT_ENTRIES: u32 = 128; 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, +} + +pub(crate) struct DrainBudget { + window: Duration, + hard: Duration, + zero: Duration, + elapsed: Duration, + since_slow_log: Duration, +} + +impl DrainBudget { + pub(crate) fn new(hard: Duration) -> Self { + Self { + window: DRAIN_WINDOW, + 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_window(&mut self, completed: usize) -> DrainVerdict { + self.elapsed += self.window; + self.since_slow_log += self.window; + if completed > 0 { + self.zero = Duration::ZERO; + } else { + self.zero += self.window; + 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; + } + 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); + #[cfg(test)] + { + let _ = secs; + return Duration::from_millis(500); + } + #[cfg(not(test))] + 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: io_uring unusable thread={thread} {reason} — aborting; restart with RBITCOIN_IO=pread if the filesystem or IO scheduling class cannot complete io_uring" + ); + 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)] @@ -510,32 +622,94 @@ impl UringSession { } fn drain_all_harvest(&mut self) -> Result<(), StoreError> { - enum Harvest { - Uds(Vec), + match &self.backend { #[cfg(target_os = "linux")] - Uring, + SessionBackend::Uring(_) => self.drain_all_uring(), + SessionBackend::Pool(_) => self.drain_all_pool(), #[cfg(windows)] - IocpErr(StoreError), + SessionBackend::Iocp(_) => self.drain_all_iocp(), } - 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 apply_drain_window( + &mut self, + budget: &mut DrainBudget, + completed: usize, + ) -> Result<(), StoreError> { + match budget.note_window(completed) { + DrainVerdict::Continue => Ok(()), + DrainVerdict::LogSlow => { + 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(), + 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; + finish_drain_hard_cap(self.pending.len()) + } + } + } + + fn drain_all_pool(&mut self) -> Result<(), StoreError> { + let mut budget = DrainBudget::with_defaults(); + loop { + let inflight = match &self.backend { + SessionBackend::Pool(pool) => pool.inflight(), + _ => 0, + }; + if inflight == 0 { + let uds = match &mut self.backend { + SessionBackend::Pool(pool) => { + pool.harvest_ready().into_iter().map(|(ud, _)| ud).collect() + } + _ => Vec::new(), + }; + return self.apply_drain_cqes(uds); + } + let before = self.pending.len(); + 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() + } + _ => Vec::new(), + }; + self.apply_drain_cqes(uds)?; + let completed = before.saturating_sub(self.pending.len()); + self.apply_drain_window(&mut budget, completed)?; + } + } + + #[cfg(windows)] + fn drain_all_iocp(&mut self) -> Result<(), StoreError> { + let mut budget = DrainBudget::with_defaults(); + loop { + let before = self.pending.len(); + 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)?; } } @@ -572,50 +746,23 @@ impl UringSession { 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(); + 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) { + 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,6 +777,61 @@ 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 { @@ -851,6 +1053,7 @@ 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, } static URING_METERS: UringMeters = UringMeters { @@ -859,6 +1062,7 @@ 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), }; #[derive(Clone, Copy)] @@ -1079,6 +1283,56 @@ mod tests { assert!(UringPending::new().assert_drained().is_ok()); } + #[test] + fn drain_budget_progress_resets_zero_clock() { + let mut b = DrainBudget::new(Duration::from_millis(300)); + assert_eq!(b.note_window(0), DrainVerdict::Continue); + assert_eq!(b.note_window(0), DrainVerdict::Continue); + assert_eq!(b.note_window(1), DrainVerdict::Continue); + assert_eq!(b.note_window(0), DrainVerdict::Continue); + assert_eq!(b.note_window(0), DrainVerdict::Continue); + assert_eq!(b.note_window(0), DrainVerdict::HardCap); + } + + #[test] + fn drain_budget_five_seconds_with_progress_logs_slow_not_hard_cap() { + let mut b = DrainBudget::new(Duration::from_secs(120)); + for _ in 0..49 { + assert_eq!(b.note_window(1), DrainVerdict::Continue); + } + assert_eq!(b.note_window(1), DrainVerdict::LogSlow); + assert_eq!(b.note_window(1), DrainVerdict::Continue); + } + + #[test] + fn drain_budget_five_seconds_zero_progress_logs_slow_before_hard_cap() { + let mut b = DrainBudget::new(Duration::from_secs(120)); + for _ in 0..49 { + assert_eq!(b.note_window(0), DrainVerdict::Continue); + } + assert_eq!(b.note_window(0), DrainVerdict::LogSlow); + assert_eq!(b.note_window(0), DrainVerdict::Continue); + } + + #[test] + fn drain_budget_hard_cap_only_after_hard_duration_of_zero() { + let mut b = DrainBudget::new(Duration::from_millis(200)); + assert_eq!(b.note_window(0), DrainVerdict::Continue); + assert_eq!(b.note_window(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()); + } + /// A CQE whose `user_data` was removed from `pending` is not a silent /// drain success — it is leftover harvest (`unexpected cqe`). #[test] From f021b00efd148b9b4d5e4b8f653b7acb38d3dc6a Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 11:43:16 -0700 Subject: [PATCH 03/13] query: recover io_uring once per 1000-height window Class C leftover above tip is the same repair as open. Credit is a tip gap so a recovered batch of 144 does not mint a new recover. Exhausted does not repair. Drops only this thread's TLS ring. Co-authored-by: Cursor --- crates/rbitcoin-query/src/lib.rs | 35 +++++++++++++++ crates/rbitcoin-query/src/query_tests.rs | 28 ++++++++++++ crates/rbitcoin-store/src/lib.rs | 2 +- crates/rbitcoin-store/src/uring_session.rs | 51 ++++++++++++++++------ 4 files changed, 102 insertions(+), 14 deletions(-) diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 7e5c55b216..7b8e96190a 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,24 @@ 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) -> Result { + let tip = self.tip_height().map(|h| h.0).unwrap_or(0); + 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 Ok(UringRecover::Exhausted); + } + let n = self.store.repair_class_c_above_tip()?; + rbitcoin_store::drop_thread_local(); + self.uring_recover_tip.store(tip, AtomicOrdering::Release); + rbitcoin_store::note_uring_recover(); + rbitcoin_log::warn!("ibd: uring recover tip={tip} n={n} reason={reason}"); + Ok(UringRecover::Recovered) + } + /// 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..86ec3c2349 100644 --- a/crates/rbitcoin-query/src/query_tests.rs +++ b/crates/rbitcoin-query/src/query_tests.rs @@ -36,6 +36,34 @@ 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_clears_leftover_strong_once_per_window() { + 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").unwrap(), UringRecover::Recovered); + assert!(!q.store().strong_tx.is_strong(leftover).unwrap()); + q.store().strong_tx.set_strong(leftover, Fk(1)).unwrap(); + q.store().flush_class_c_tip().unwrap(); + assert_eq!(q.uring_recover("again").unwrap(), UringRecover::Exhausted); + assert!( + q.store().strong_tx.is_strong(leftover).unwrap(), + "Exhausted must not run a second repair" + ); +} + fn temp_query(label: &str) -> (crate::testutil::TempDir, Query) { crate::testutil::tiny_query_labeled(label) } diff --git a/crates/rbitcoin-store/src/lib.rs b/crates/rbitcoin-store/src/lib.rs index ef7791f57c..a60026e924 100644 --- a/crates/rbitcoin-store/src/lib.rs +++ b/crates/rbitcoin-store/src/lib.rs @@ -101,5 +101,5 @@ pub use tx_table::{ encode_packed_tx, encode_packed_tx_with_secret, spend_meta_backend, spent_abs, InputRecord, OutputRecord, TxRecord, }; -pub use uring_session::abort_uring_unusable; pub(crate) use uring_session::IoCtx; +pub use uring_session::{abort_uring_unusable, drop_thread_local, note_uring_recover}; diff --git a/crates/rbitcoin-store/src/uring_session.rs b/crates/rbitcoin-store/src/uring_session.rs index 08a973eb07..0d59becf2f 100644 --- a/crates/rbitcoin-store/src/uring_session.rs +++ b/crates/rbitcoin-store/src/uring_session.rs @@ -19,7 +19,7 @@ use crate::error::StoreError; use crate::io_handle::IoHandle; use crate::io_session_pool::PoolEngine; -use std::cell::Cell; +use std::cell::{Cell, RefCell}; use std::path::Path; use std::time::Duration; @@ -887,6 +887,28 @@ pub(crate) fn require_full_cqe(res: i32, want: usize, path: &Path) -> Result<(), } Ok(()) } + +thread_local! { + static TLS_SESSION: RefCell> = const { RefCell::new(None) }; + static TLS_DEPTH: Cell = const { Cell::new(0) }; +} + +/// Drain and drop this OS thread's TLS completion session. +pub fn drop_thread_local() { + TLS_SESSION.with(|cell| { + if let Some(mut old) = cell.borrow_mut().take() { + let _ = old.drain_all(); + drop(old); + } + }); +} + +pub fn note_uring_recover() { + URING_METERS + .recover_n + .fetch_add(1, 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 @@ -905,23 +927,14 @@ pub fn with_thread_local( let min_entries = min_entries.max(32).min(4096); { - use std::cell::{Cell, RefCell}; - - thread_local! { - static SESSION: RefCell> = const { RefCell::new(None) }; - static DEPTH: Cell = const { Cell::new(0) }; - } - // Gate once; TLS open uses try_open to avoid recursive enabled() probe. if !crate::bulk_io::io_uring_enabled() { return Err(StoreError::Unavailable); } - DEPTH.with(|depth| { + TLS_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 \ @@ -932,11 +945,11 @@ pub fn with_thread_local( struct DepthGuard; impl Drop for DepthGuard { fn drop(&mut self) { - DEPTH.with(|depth| depth.set(0)); + TLS_DEPTH.with(|depth| depth.set(0)); } } let _guard = DepthGuard; - let out = SESSION.with(|cell| -> Result { + let out = TLS_SESSION.with(|cell| -> Result { let mut slot = cell.borrow_mut(); let need_open = match slot.as_ref() { None => true, @@ -1054,6 +1067,7 @@ pub(crate) struct UringMeters { 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 { @@ -1063,6 +1077,7 @@ static URING_METERS: UringMeters = UringMeters { 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)] @@ -1451,6 +1466,16 @@ mod tests { assert!(entries >= DEFAULT_ENTRIES); } + #[test] + fn drop_thread_local_clears_tls_slot() { + if !crate::bulk_io::io_uring_enabled() { + return; + } + let _ = with_thread_local(DEFAULT_ENTRIES, |s| s.entries()); + drop_thread_local(); + let _ = with_thread_local(DEFAULT_ENTRIES, |s| s.entries()); + } + /// `drain_all` must wait until in-flight SQEs complete (not only harvest-ready). #[test] #[cfg(target_os = "linux")] From 819233661efe54f948654b20e93489c5d23db0f0 Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 11:50:05 -0700 Subject: [PATCH 04/13] net,consensus: recover or abort on io_uring session faults Write requeues the same BQ heights after a credited recover; lookup does the same instead of warning every 20s. Credit exhausted or a drain stall aborts and names RBITCOIN_IO=pread. Failed head drain re-notes the taken pending list. ibd: perf gains uring_recover, slow_drain, and lookup_faults. Co-authored-by: Cursor --- .../src/confirm_run/write.rs | 5 +- crates/rbitcoin-net/src/ibd/confirm/mod.rs | 229 ++++++++++++++++-- crates/rbitcoin-net/src/ibd/confirm/tests.rs | 80 ++++++ crates/rbitcoin-net/src/ibd/perf_log.rs | 18 ++ crates/rbitcoin-store/src/lib.rs | 5 +- crates/rbitcoin-store/src/uring_session.rs | 12 + 6 files changed, 323 insertions(+), 26 deletions(-) diff --git a/crates/rbitcoin-consensus/src/confirm_run/write.rs b/crates/rbitcoin-consensus/src/confirm_run/write.rs index c69f849b69..3236531b71 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/write.rs @@ -201,7 +201,7 @@ pub fn confirm_write_phase( let t_head = Instant::now(); let queued = query.store().txs.take_pending_queued(); let drain_max_fk = queued.iter().filter_map(|(_, fk)| fk.get()).max(); - let drain = super::head_drain::submit_head_insert(query.store(), queued); + let drain = super::head_drain::submit_head_insert(query.store(), queued.clone()); let head_sub_ns = t_head.elapsed().as_nanos() as u64; if head_sub_ns > 0 { rbitcoin_query::note_confirm(&query.confirm_stats().write_head_sub_ns, head_sub_ns); @@ -257,6 +257,9 @@ pub fn confirm_write_phase( 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(&queued); + } 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-net/src/ibd/confirm/mod.rs b/crates/rbitcoin-net/src/ibd/confirm/mod.rs index 646579165e..3c886265c1 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/mod.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/mod.rs @@ -474,6 +474,107 @@ impl ConfirmRejectClass { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum WriteStoreFault { + Requeue, + Abort, + Reject(ConfirmRejectClass), +} + +pub(crate) fn classify_write_store_fault( + e: &rbitcoin_consensus::ConsensusError, + recover: rbitcoin_query::UringRecover, +) -> WriteStoreFault { + use rbitcoin_consensus::ConsensusError; + match e { + ConsensusError::Store(se) if se.is_uring_session_fault() => match recover { + rbitcoin_query::UringRecover::Recovered => WriteStoreFault::Requeue, + rbitcoin_query::UringRecover::Exhausted => WriteStoreFault::Abort, + }, + other => WriteStoreFault::Reject(ConfirmRejectClass::from_consensus(other)), + } +} + +pub(crate) fn apply_write_store_fault( + feed: &ConfirmFeed, + heights: &[(u32, [u8; 32])], + action: WriteStoreFault, + load_ahead_reset: &AtomicBool, +) -> WriteStoreFault { + match action { + WriteStoreFault::Requeue => { + load_ahead_reset.store(true, Ordering::Release); + let batch: Vec<_> = heights + .iter() + .map(|(h, raw)| (*h, BlockHash::from_byte_array(*raw), None)) + .collect(); + feed.requeue_wire(&batch); + WriteStoreFault::Requeue + } + WriteStoreFault::Abort => WriteStoreFault::Abort, + WriteStoreFault::Reject(class) => { + load_ahead_reset.store(true, Ordering::Release); + feed.finish(heights.iter().map(|(h, _)| *h)); + WriteStoreFault::Reject(class) + } + } +} + +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, + RecoverContinue, + Abort, + 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, + recover: rbitcoin_query::UringRecover, + ) -> LookupFaultAction { + use rbitcoin_consensus::ConsensusError; + let backpressure = matches!(e, ConsensusError::Store(se) if se.is_io_backpressure()); + if backpressure { + return LookupFaultAction::Ignore; + } + if matches!(e, ConsensusError::Store(se) if se.is_uring_session_fault()) { + return match recover { + rbitcoin_query::UringRecover::Recovered => { + self.consecutive = 0; + LookupFaultAction::RecoverContinue + } + rbitcoin_query::UringRecover::Exhausted => LookupFaultAction::Abort, + }; + } + 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 }, @@ -1449,24 +1550,57 @@ 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. - load_ahead_reset_wb.store(true, Ordering::Release); - feed_wb.finish(heights_hashes.iter().map(|(h, _)| *h)); - loop_stats_wb - .confirm_reject_stops - .fetch_add(1, Ordering::Relaxed); - warn!( - "ibd: confirm write reject @ {height} batch_parts={parts}: {e}" - ); - let _ = emit_confirm_reject( - &event_tx_wb, + let recover = if matches!( + &e, + rbitcoin_consensus::ConsensusError::Store(se) + if se.is_uring_session_fault() + ) { + match hub_wb.query.uring_recover("ibd-confirm-write") { + Ok(r) => r, + Err(re) => { + warn!("ibd: confirm write uring recover: {re}"); + rbitcoin_query::UringRecover::Exhausted + } + } + } else { + rbitcoin_query::UringRecover::Exhausted + }; + let action = classify_write_store_fault(&e, recover); + match apply_write_store_fault( &feed_wb, - height, - hash, - ConfirmRejectClass::from_consensus(&e), - msg, - heights_hashes.len(), - ); + &heights_hashes, + action, + &load_ahead_reset_wb, + ) { + WriteStoreFault::Requeue => { + warn!( + "ibd: confirm write uring recover @ {height} batch_parts={parts}: {e}" + ); + continue; + } + WriteStoreFault::Abort => { + rbitcoin_store::abort_uring_unusable( + "recover credit exhausted on ibd-confirm-write", + ); + } + WriteStoreFault::Reject(class) => { + loop_stats_wb + .confirm_reject_stops + .fetch_add(1, Ordering::Relaxed); + warn!( + "ibd: confirm write reject @ {height} batch_parts={parts}: {e}" + ); + let _ = emit_confirm_reject( + &event_tx_wb, + &feed_wb, + height, + hash, + class, + msg, + heights_hashes.len(), + ); + } + } } } } @@ -1886,6 +2020,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 || { @@ -1893,6 +2028,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; @@ -1950,6 +2086,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 @@ -2010,17 +2147,61 @@ pub(crate) fn spawn_confirm_engine( } feed.notify(); } - Ok(_) => {} + Ok(_) => { + lookup_faults.on_success(); + } Err(e) => { - let backpressure = matches!( + let recover = if matches!( &e, rbitcoin_consensus::ConsensusError::Store(se) - if se.is_io_backpressure() - ); - if backpressure { - debug!("ibd: bq resolve wave: {e}"); + if se.is_uring_session_fault() + ) { + match hub.query.uring_recover("ibd-confirm-lookup") { + Ok(r) => r, + Err(re) => { + warn!("ibd: bq resolve wave uring recover: {re}"); + rbitcoin_query::UringRecover::Exhausted + } + } } else { - warn!("ibd: bq resolve wave: {e}"); + rbitcoin_query::UringRecover::Recovered + }; + match lookup_faults.on_err(&e, recover) { + LookupFaultAction::Ignore => { + debug!("ibd: bq resolve wave: {e}"); + } + LookupFaultAction::RecoverContinue => { + warn!("ibd: bq resolve wave uring recover: {e}"); + } + LookupFaultAction::Abort => { + rbitcoin_store::abort_uring_unusable( + "recover credit exhausted on ibd-confirm-lookup", + ); + } + 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); + let hash = feed + .inner + .lock() + .ok() + .and_then(|g| g.ready.get(&height).map(|(h, _)| *h)) + .unwrap_or_else(|| BlockHash::from_byte_array([0u8; 32])); + let _ = emit_confirm_reject( + &event_tx_lookup, + &feed, + height, + hash, + ConfirmRejectClass::EngineFault, + e.to_string(), + 1, + ); + } } } } diff --git a/crates/rbitcoin-net/src/ibd/confirm/tests.rs b/crates/rbitcoin-net/src/ibd/confirm/tests.rs index 0d29b5a481..99ec766f3b 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/tests.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/tests.rs @@ -1245,3 +1245,83 @@ fn plan_epoch_stale_after_clear() { ); assert!(!feed.plan_epoch_stale(99), "unknown height is not stale"); } + +#[test] +fn write_store_fault_uring_requeues_then_aborts() { + use super::{apply_write_store_fault, classify_write_store_fault, WriteStoreFault}; + use rbitcoin_consensus::ConsensusError; + use rbitcoin_query::UringRecover; + use rbitcoin_store::StoreError; + use std::sync::atomic::AtomicBool; + + let undrained = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring undrained")); + assert_eq!( + classify_write_store_fault(&undrained, UringRecover::Recovered), + WriteStoreFault::Requeue + ); + assert_eq!( + classify_write_store_fault(&undrained, UringRecover::Exhausted), + WriteStoreFault::Abort + ); + let io = ConsensusError::Store(StoreError::io("/tmp/x", std::io::Error::other("disk"))); + assert_eq!( + classify_write_store_fault(&io, UringRecover::Recovered), + WriteStoreFault::Reject(super::ConfirmRejectClass::EngineFault) + ); + + let feed = ConfirmFeed::new(); + let reset = AtomicBool::new(false); + let raw = [1u8; 32]; + { + let mut g = feed.inner.lock().unwrap(); + g.inflight.insert(10); + } + assert_eq!( + apply_write_store_fault(&feed, &[(10, raw)], WriteStoreFault::Requeue, &reset), + WriteStoreFault::Requeue + ); + assert!(reset.load(std::sync::atomic::Ordering::Acquire)); + let g = feed.inner.lock().unwrap(); + assert!(g.ready.contains_key(&10)); + assert!(!g.inflight.contains(&10)); +} + +#[test] +fn lookup_fault_policy_recover_abort_and_io_halt() { + use super::{LookupFaultAction, LookupFaultPolicy}; + use rbitcoin_consensus::ConsensusError; + use rbitcoin_query::UringRecover; + use rbitcoin_store::StoreError; + + let mut p = LookupFaultPolicy::default(); + let uring = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring undrained")); + assert_eq!( + p.on_err(&uring, UringRecover::Recovered), + LookupFaultAction::RecoverContinue + ); + assert_eq!( + p.on_err(&uring, UringRecover::Exhausted), + LookupFaultAction::Abort + ); + let bp = ConsensusError::Store(StoreError::BudgetFull("io_uring SQ")); + assert_eq!( + p.on_err(&bp, UringRecover::Recovered), + 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, UringRecover::Recovered), + LookupFaultAction::Warn + ); + } + assert_eq!( + p.on_err(&io, UringRecover::Recovered), + LookupFaultAction::RejectEngineFault + ); + p.on_success(); + assert_eq!( + p.on_err(&io, UringRecover::Recovered), + LookupFaultAction::Warn + ); +} 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-store/src/lib.rs b/crates/rbitcoin-store/src/lib.rs index a60026e924..08132e2828 100644 --- a/crates/rbitcoin-store/src/lib.rs +++ b/crates/rbitcoin-store/src/lib.rs @@ -102,4 +102,7 @@ pub use tx_table::{ OutputRecord, TxRecord, }; pub(crate) use uring_session::IoCtx; -pub use uring_session::{abort_uring_unusable, drop_thread_local, note_uring_recover}; +pub use uring_session::{ + abort_uring_unusable, drop_thread_local, 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 0d59becf2f..12805bd933 100644 --- a/crates/rbitcoin-store/src/uring_session.rs +++ b/crates/rbitcoin-store/src/uring_session.rs @@ -909,6 +909,18 @@ pub fn note_uring_recover() { .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 From d5a091f4c091404825684ed63b054f853dc595cc Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 11:51:20 -0700 Subject: [PATCH 05/13] docs: document progress drain, recover credit, and pread restart Operators saw 19h of undrained warnings under idle IO. Document the 120s hard-cap abort, 1000-height recover, Class C repair, and that there is no mid-run libc fallback. Co-authored-by: Cursor --- CHANGELOG.md | 6 ++++++ OPERATOR.md | 8 +++++++- docs/README.md | 2 +- docs/crash-recovery.md | 2 ++ docs/env-knobs.md | 1 + docs/io-modality.md | 13 +++++++++++-- 6 files changed, 28 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d9624fc316..5ec8f36578 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,12 @@ 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. Write/lookup recover once + per 1000 heights (same Class C repair as open) instead of a 19h warn loop. + Restart with `RBITCOIN_IO=pread` if uring 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..a63b9dad2e 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: io_uring 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/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..fd42b62436 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 io_uring recover (IBD write/lookup session fault after drain idle) runs that same `repair_class_c_above_tip` and 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/io-modality.md b/docs/io-modality.md index baaa91c3a4..a694410d65 100644 --- a/docs/io-modality.md +++ b/docs/io-modality.md @@ -49,11 +49,20 @@ 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 +opens a new one. `submit_and_wait_one` is capped (5 s). `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 the process (buffers still +pinned). 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). +unavailable also falls back; operator restart after a drain abort). + +IBD write/lookup: a session fault after a successful drain recovers once per +1000-height window (`repair_class_c_above_tip`, drop this thread's ring, +requeue the same BQ bodies). A second stall in that window aborts and names +`RBITCOIN_IO=pread`. ### Do not flatten custom machines From bf54551926769e721fe39f714f034eb27e0e4717 Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:01:30 -0700 Subject: [PATCH 06/13] store: exhaust pool drain matches per backend Darwin SessionBackend is Pool-only, so `_` in drain_all_pool was unreachable under -D warnings and failed macos CI. Co-authored-by: Cursor --- crates/rbitcoin-store/src/uring_session.rs | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/crates/rbitcoin-store/src/uring_session.rs b/crates/rbitcoin-store/src/uring_session.rs index 12805bd933..111d1f4218 100644 --- a/crates/rbitcoin-store/src/uring_session.rs +++ b/crates/rbitcoin-store/src/uring_session.rs @@ -663,14 +663,20 @@ impl UringSession { loop { let inflight = match &self.backend { SessionBackend::Pool(pool) => pool.inflight(), - _ => 0, + #[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() } - _ => Vec::new(), + #[cfg(target_os = "linux")] + SessionBackend::Uring(_) => Vec::new(), + #[cfg(windows)] + SessionBackend::Iocp(_) => Vec::new(), }; return self.apply_drain_cqes(uds); } @@ -680,7 +686,10 @@ impl UringSession { let _ = pool.wait_one_cqe_timeout(DRAIN_WINDOW); pool.harvest_ready().into_iter().map(|(ud, _)| ud).collect() } - _ => Vec::new(), + #[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()); From 4df33c02dab2ba73346c69d5b959b7928760f021 Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:43:59 -0700 Subject: [PATCH 07/13] store: drain budget uses elapsed time; Drop does not abort Assumed 100ms windows let EINTR spin hit the hard cap. note() now takes a real dt. submit_and_wait_one shares that budget. Session Drop waits then poisons without abort(). Session-fault match is any Corrupt io_uring message except Linux-only/unavailable. drop_thread_local is gone: poisoned TLS already reopens. Co-authored-by: Cursor --- crates/rbitcoin-store/src/error.rs | 11 +- crates/rbitcoin-store/src/io_session_iocp.rs | 7 + crates/rbitcoin-store/src/lib.rs | 3 +- crates/rbitcoin-store/src/uring_session.rs | 283 +++++++++++-------- 4 files changed, 179 insertions(+), 125 deletions(-) diff --git a/crates/rbitcoin-store/src/error.rs b/crates/rbitcoin-store/src/error.rs index 088b32385a..1e8d754c78 100644 --- a/crates/rbitcoin-store/src/error.rs +++ b/crates/rbitcoin-store/src/error.rs @@ -52,12 +52,8 @@ impl StoreError { } pub fn is_uring_session_fault_msg(m: &str) -> bool { - m.contains("io_uring undrained") - || m.contains("io_uring wait timeout") - || m.contains("io_uring session poisoned") - || m.contains("io_uring unexpected cqe") - || m.contains("io_uring cq overflow") - || m.contains("io_uring submit_and_wait failed") + let s = m.to_ascii_lowercase(); + s.contains("io_uring") && !s.contains("linux-only") && !s.contains("unavailable") } } @@ -163,11 +159,14 @@ mod tests { "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"), diff --git a/crates/rbitcoin-store/src/io_session_iocp.rs b/crates/rbitcoin-store/src/io_session_iocp.rs index af9c070fa3..443bffed70 100644 --- a/crates/rbitcoin-store/src/io_session_iocp.rs +++ b/crates/rbitcoin-store/src/io_session_iocp.rs @@ -196,6 +196,13 @@ impl IocpEngine { Ok(()) } + pub(crate) fn wait_one_cqe_timeout(&mut self, timeout_ms: u32) -> bool { + if self.ready.is_empty() { + self.poll(timeout_ms); + } + !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(100); diff --git a/crates/rbitcoin-store/src/lib.rs b/crates/rbitcoin-store/src/lib.rs index 08132e2828..5ad3705880 100644 --- a/crates/rbitcoin-store/src/lib.rs +++ b/crates/rbitcoin-store/src/lib.rs @@ -103,6 +103,5 @@ pub use tx_table::{ }; pub(crate) use uring_session::IoCtx; pub use uring_session::{ - abort_uring_unusable, drop_thread_local, note_uring_recover, uring_recover_count, - uring_slow_drain_count, + 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 111d1f4218..f15fd34e6a 100644 --- a/crates/rbitcoin-store/src/uring_session.rs +++ b/crates/rbitcoin-store/src/uring_session.rs @@ -19,20 +19,16 @@ use crate::error::StoreError; use crate::io_handle::IoHandle; use crate::io_session_pool::PoolEngine; -use std::cell::{Cell, RefCell}; +use std::cell::Cell; use std::path::Path; -use std::time::Duration; +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); - /// 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")] @@ -49,8 +45,14 @@ pub(crate) enum DrainVerdict { HardCap, } +enum WaitOneWindow { + Ready, + TimedOut, + Idle, + Enter(StoreError), +} + pub(crate) struct DrainBudget { - window: Duration, hard: Duration, zero: Duration, elapsed: Duration, @@ -60,7 +62,6 @@ pub(crate) struct DrainBudget { impl DrainBudget { pub(crate) fn new(hard: Duration) -> Self { Self { - window: DRAIN_WINDOW, hard, zero: Duration::ZERO, elapsed: Duration::ZERO, @@ -76,13 +77,13 @@ impl DrainBudget { self.elapsed } - pub(crate) fn note_window(&mut self, completed: usize) -> DrainVerdict { - self.elapsed += self.window; - self.since_slow_log += self.window; + 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.window; + self.zero = self.zero.saturating_add(dt); if self.zero >= self.hard { return DrainVerdict::HardCap; } @@ -100,18 +101,15 @@ fn drain_hard_cap() -> Duration { if let Some(d) = TEST_DRAIN_HARD.with(|c| c.get()) { return d; } - 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); - #[cfg(test)] - { - let _ = secs; - return Duration::from_millis(500); - } - #[cfg(not(test))] - Duration::from_secs(secs) + 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)] @@ -134,7 +132,7 @@ pub fn abort_uring_unusable(reason: &str) -> ! { let thread = std::thread::current(); let thread = thread.name().unwrap_or("unnamed"); rbitcoin_log::error!( - "store: io_uring unusable thread={thread} {reason} — aborting; restart with RBITCOIN_IO=pread if the filesystem or IO scheduling class cannot complete io_uring" + "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(); } @@ -498,45 +496,74 @@ 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")); + } + 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 + } + } } } @@ -609,7 +636,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; @@ -621,44 +652,60 @@ impl UringSession { Ok(()) } - fn drain_all_harvest(&mut self) -> Result<(), StoreError> { + fn drain_all_harvest(&mut self, fail_closed: bool) -> Result<(), StoreError> { match &self.backend { #[cfg(target_os = "linux")] - SessionBackend::Uring(_) => self.drain_all_uring(), - SessionBackend::Pool(_) => self.drain_all_pool(), + SessionBackend::Uring(_) => self.drain_all_uring(fail_closed), + SessionBackend::Pool(_) => self.drain_all_pool(fail_closed), #[cfg(windows)] - SessionBackend::Iocp(_) => self.drain_all_iocp(), + SessionBackend::Iocp(_) => self.drain_all_iocp(fail_closed), } } + 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_window(completed) { + match budget.note(dt, completed) { DrainVerdict::Continue => Ok(()), DrainVerdict::LogSlow => { - 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(), - budget.elapsed(), - ); + self.note_slow_drain(budget.elapsed()); Ok(()) } DrainVerdict::HardCap => { self.poisoned = true; - finish_drain_hard_cap(self.pending.len()) + 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) -> Result<(), StoreError> { + fn drain_all_pool(&mut self, fail_closed: bool) -> Result<(), StoreError> { let mut budget = DrainBudget::with_defaults(); loop { let inflight = match &self.backend { @@ -681,6 +728,7 @@ impl UringSession { 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); @@ -693,15 +741,16 @@ impl UringSession { }; self.apply_drain_cqes(uds)?; let completed = before.saturating_sub(self.pending.len()); - self.apply_drain_window(&mut budget, completed)?; + self.apply_drain_window(&mut budget, completed, t0.elapsed(), fail_closed)?; } } #[cfg(windows)] - fn drain_all_iocp(&mut self) -> Result<(), StoreError> { + 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(()), @@ -718,7 +767,7 @@ impl UringSession { return Ok(()); } let completed = before.saturating_sub(self.pending.len()); - self.apply_drain_window(&mut budget, completed)?; + self.apply_drain_window(&mut budget, completed, t0.elapsed(), fail_closed)?; } } @@ -751,7 +800,7 @@ 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; @@ -759,12 +808,15 @@ impl UringSession { 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; } let completed = before.saturating_sub(self.pending.len()); - if let Err(e) = self.apply_drain_window(&mut budget, completed) { + if let Err(e) = + self.apply_drain_window(&mut budget, completed, t0.elapsed(), fail_closed) + { enter_err = Some(e); break; } @@ -845,7 +897,7 @@ impl UringSession { impl Drop for UringSession { fn drop(&mut self) { - let _ = self.drain_all(); + let _ = self.drain_all_inner(false); } } @@ -897,21 +949,6 @@ pub(crate) fn require_full_cqe(res: i32, want: usize, path: &Path) -> Result<(), Ok(()) } -thread_local! { - static TLS_SESSION: RefCell> = const { RefCell::new(None) }; - static TLS_DEPTH: Cell = const { Cell::new(0) }; -} - -/// Drain and drop this OS thread's TLS completion session. -pub fn drop_thread_local() { - TLS_SESSION.with(|cell| { - if let Some(mut old) = cell.borrow_mut().take() { - let _ = old.drain_all(); - drop(old); - } - }); -} - pub fn note_uring_recover() { URING_METERS .recover_n @@ -948,12 +985,19 @@ pub fn with_thread_local( let min_entries = min_entries.max(32).min(4096); { + use std::cell::{Cell, RefCell}; + + thread_local! { + static SESSION: RefCell> = const { RefCell::new(None) }; + static DEPTH: Cell = const { Cell::new(0) }; + } + // Gate once; TLS open uses try_open to avoid recursive enabled() probe. if !crate::bulk_io::io_uring_enabled() { return Err(StoreError::Unavailable); } - TLS_DEPTH.with(|depth| { + DEPTH.with(|depth| { let d = depth.get(); if d != 0 { panic!( @@ -966,11 +1010,11 @@ pub fn with_thread_local( struct DepthGuard; impl Drop for DepthGuard { fn drop(&mut self) { - TLS_DEPTH.with(|depth| depth.set(0)); + DEPTH.with(|depth| depth.set(0)); } } let _guard = DepthGuard; - let out = TLS_SESSION.with(|cell| -> Result { + let out = SESSION.with(|cell| -> Result { let mut slot = cell.borrow_mut(); let need_open = match slot.as_ref() { None => true, @@ -1321,40 +1365,52 @@ mod tests { #[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_window(0), DrainVerdict::Continue); - assert_eq!(b.note_window(0), DrainVerdict::Continue); - assert_eq!(b.note_window(1), DrainVerdict::Continue); - assert_eq!(b.note_window(0), DrainVerdict::Continue); - assert_eq!(b.note_window(0), DrainVerdict::Continue); - assert_eq!(b.note_window(0), DrainVerdict::HardCap); + 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_window(1), DrainVerdict::Continue); + assert_eq!(b.note(dt, 1), DrainVerdict::Continue); } - assert_eq!(b.note_window(1), DrainVerdict::LogSlow); - assert_eq!(b.note_window(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_window(0), DrainVerdict::Continue); + assert_eq!(b.note(dt, 0), DrainVerdict::Continue); } - assert_eq!(b.note_window(0), DrainVerdict::LogSlow); - assert_eq!(b.note_window(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)); - assert_eq!(b.note_window(0), DrainVerdict::Continue); - assert_eq!(b.note_window(0), DrainVerdict::HardCap); + for _ in 0..10_000 { + assert_ne!(b.note(Duration::from_nanos(1), 0), DrainVerdict::HardCap); + } } #[test] @@ -1362,11 +1418,14 @@ mod tests { 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:?}"), + 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); }); - assert!(session.is_poisoned()); } /// A CQE whose `user_data` was removed from `pending` is not a silent @@ -1487,16 +1546,6 @@ mod tests { assert!(entries >= DEFAULT_ENTRIES); } - #[test] - fn drop_thread_local_clears_tls_slot() { - if !crate::bulk_io::io_uring_enabled() { - return; - } - let _ = with_thread_local(DEFAULT_ENTRIES, |s| s.entries()); - drop_thread_local(); - let _ = with_thread_local(DEFAULT_ENTRIES, |s| s.entries()); - } - /// `drain_all` must wait until in-flight SQEs complete (not only harvest-ready). #[test] #[cfg(target_os = "linux")] From f148cfc84bb1c55c27ec4512cdf70e04103629ad Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:44:10 -0700 Subject: [PATCH 08/13] query: recover is credit-only; CAS the 1000-height window Lookup must not clear strong bits while write sets them before the fence. In-process recover no longer runs Class C repair. Two threads cannot both pass the tip-gap check. Co-authored-by: Cursor --- crates/rbitcoin-query/src/lib.rs | 27 ++++++++++++++---------- crates/rbitcoin-query/src/query_tests.rs | 20 ++++++++++++------ 2 files changed, 29 insertions(+), 18 deletions(-) diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 7b8e96190a..82deb782bb 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/crates/rbitcoin-query/src/lib.rs @@ -1036,19 +1036,24 @@ impl Query { /// 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) -> Result { + pub fn uring_recover(&self, reason: &'static str) -> UringRecover { let tip = self.tip_height().map(|h| h.0).unwrap_or(0); - 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 Ok(UringRecover::Exhausted); + 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; + } } - let n = self.store.repair_class_c_above_tip()?; - rbitcoin_store::drop_thread_local(); - self.uring_recover_tip.store(tip, AtomicOrdering::Release); - rbitcoin_store::note_uring_recover(); - rbitcoin_log::warn!("ibd: uring recover tip={tip} n={n} reason={reason}"); - Ok(UringRecover::Recovered) } /// Highest height on the RAM fence. Not the in-flight prune HWM diff --git a/crates/rbitcoin-query/src/query_tests.rs b/crates/rbitcoin-query/src/query_tests.rs index 86ec3c2349..cf0a360727 100644 --- a/crates/rbitcoin-query/src/query_tests.rs +++ b/crates/rbitcoin-query/src/query_tests.rs @@ -47,21 +47,27 @@ fn uring_recover_credit_gap() { } #[test] -fn uring_recover_clears_leftover_strong_once_per_window() { +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").unwrap(), UringRecover::Recovered); - assert!(!q.store().strong_tx.is_strong(leftover).unwrap()); - q.store().strong_tx.set_strong(leftover, Fk(1)).unwrap(); - q.store().flush_class_c_tip().unwrap(); - assert_eq!(q.uring_recover("again").unwrap(), UringRecover::Exhausted); + assert_eq!(q.uring_recover("test"), UringRecover::Recovered); assert!( q.store().strong_tx.is_strong(leftover).unwrap(), - "Exhausted must not run a second repair" + "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); } fn temp_query(label: &str) -> (crate::testutil::TempDir, Query) { From 8080f7e75980f79c338cfae8f6a807f4a2d23941 Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:44:10 -0700 Subject: [PATCH 09/13] consensus: name session faults; restore pending-head on drain Err ConsensusError::is_uring_session_fault covers leftover cqe and submit failed. Head insert returns the taken batch on Err so write does not clone the drain list on the success path. Co-authored-by: Cursor --- .../src/confirm_run/head_drain.rs | 35 +++++++++++++++---- .../rbitcoin-consensus/src/confirm_run/mod.rs | 2 +- .../src/confirm_run/write.rs | 6 ++-- .../src/confirm_run/write_idempotent_tests.rs | 9 +++++ crates/rbitcoin-consensus/src/error.rs | 20 +++++++++++ 5 files changed, 62 insertions(+), 10 deletions(-) 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 3236531b71..277f2cae80 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/write.rs @@ -201,7 +201,7 @@ pub fn confirm_write_phase( let t_head = Instant::now(); let queued = query.store().txs.take_pending_queued(); let drain_max_fk = queued.iter().filter_map(|(_, fk)| fk.get()).max(); - let drain = super::head_drain::submit_head_insert(query.store(), queued.clone()); + let drain = super::head_drain::submit_head_insert(query.store(), queued); let head_sub_ns = t_head.elapsed().as_nanos() as u64; if head_sub_ns > 0 { rbitcoin_query::note_confirm(&query.confirm_stats().write_head_sub_ns, head_sub_ns); @@ -252,13 +252,13 @@ 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(&queued); + 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)?; 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( From 95d697e1f8d07fdc854573d7df2ecfb7c19613f6 Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:44:10 -0700 Subject: [PATCH 10/13] net: recover on load/scripts/tip-connect; skip zero-hash reject One helper returns Option recover. Write classify takes that Option. Lookup EngineFault skips emit when ready has no hash. Load stamp/pin and scripts requeue on credited recover. Tip connect retries instead of note_invalid_block. Co-authored-by: Cursor --- crates/rbitcoin-net/src/chain.rs | 19 +- crates/rbitcoin-net/src/ibd/confirm/mod.rs | 219 ++++++++++++------ crates/rbitcoin-net/src/ibd/confirm/tests.rs | 55 +++-- .../src/ibd/events/confirm_reject_tests.rs | 6 +- 4 files changed, 197 insertions(+), 102 deletions(-) diff --git a/crates/rbitcoin-net/src/chain.rs b/crates/rbitcoin-net/src/chain.rs index 025e76354a..c37e4edcef 100644 --- a/crates/rbitcoin-net/src/chain.rs +++ b/crates/rbitcoin-net/src/chain.rs @@ -2039,15 +2039,28 @@ 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() => { + match self.query.uring_recover("tip-connect") { + rbitcoin_query::UringRecover::Recovered => continue, + rbitcoin_query::UringRecover::Exhausted => { + rbitcoin_store::abort_uring_unusable( + "recover credit exhausted on tip-connect", + ); + } + } + } + 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 3c886265c1..90dd4a2acf 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/mod.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/mod.rs @@ -483,15 +483,12 @@ pub(crate) enum WriteStoreFault { pub(crate) fn classify_write_store_fault( e: &rbitcoin_consensus::ConsensusError, - recover: rbitcoin_query::UringRecover, + recover: Option, ) -> WriteStoreFault { - use rbitcoin_consensus::ConsensusError; - match e { - ConsensusError::Store(se) if se.is_uring_session_fault() => match recover { - rbitcoin_query::UringRecover::Recovered => WriteStoreFault::Requeue, - rbitcoin_query::UringRecover::Exhausted => WriteStoreFault::Abort, - }, - other => WriteStoreFault::Reject(ConfirmRejectClass::from_consensus(other)), + match recover { + Some(rbitcoin_query::UringRecover::Recovered) => WriteStoreFault::Requeue, + Some(rbitcoin_query::UringRecover::Exhausted) => WriteStoreFault::Abort, + None => WriteStoreFault::Reject(ConfirmRejectClass::from_consensus(e)), } } @@ -500,7 +497,7 @@ pub(crate) fn apply_write_store_fault( heights: &[(u32, [u8; 32])], action: WriteStoreFault, load_ahead_reset: &AtomicBool, -) -> WriteStoreFault { +) { match action { WriteStoreFault::Requeue => { load_ahead_reset.store(true, Ordering::Release); @@ -509,17 +506,44 @@ pub(crate) fn apply_write_store_fault( .map(|(h, raw)| (*h, BlockHash::from_byte_array(*raw), None)) .collect(); feed.requeue_wire(&batch); - WriteStoreFault::Requeue } - WriteStoreFault::Abort => WriteStoreFault::Abort, - WriteStoreFault::Reject(class) => { + WriteStoreFault::Abort => {} + WriteStoreFault::Reject(_) => { load_ahead_reset.store(true, Ordering::Release); feed.finish(heights.iter().map(|(h, _)| *h)); - WriteStoreFault::Reject(class) } } } +fn try_uring_recover( + query: &rbitcoin_query::Query, + e: &rbitcoin_consensus::ConsensusError, + reason: &'static str, +) -> Option { + if !e.is_uring_session_fault() { + return None; + } + Some(query.uring_recover(reason)) +} + +fn try_uring_recover_msg( + query: &rbitcoin_query::Query, + msg: &str, + reason: &'static str, +) -> Option { + if !rbitcoin_store::StoreError::is_uring_session_fault_msg(msg) { + return None; + } + Some(query.uring_recover(reason)) +} + +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)] @@ -544,27 +568,27 @@ impl LookupFaultPolicy { pub(crate) fn on_err( &mut self, e: &rbitcoin_consensus::ConsensusError, - recover: rbitcoin_query::UringRecover, + recover: Option, ) -> LookupFaultAction { use rbitcoin_consensus::ConsensusError; let backpressure = matches!(e, ConsensusError::Store(se) if se.is_io_backpressure()); if backpressure { return LookupFaultAction::Ignore; } - if matches!(e, ConsensusError::Store(se) if se.is_uring_session_fault()) { - return match recover { - rbitcoin_query::UringRecover::Recovered => { - self.consecutive = 0; - LookupFaultAction::RecoverContinue + match recover { + Some(rbitcoin_query::UringRecover::Recovered) => { + self.consecutive = 0; + LookupFaultAction::RecoverContinue + } + Some(rbitcoin_query::UringRecover::Exhausted) => LookupFaultAction::Abort, + None => { + self.consecutive = self.consecutive.saturating_add(1); + if self.consecutive >= LOOKUP_FAULT_HALT_AFTER { + LookupFaultAction::RejectEngineFault + } else { + LookupFaultAction::Warn } - rbitcoin_query::UringRecover::Exhausted => LookupFaultAction::Abort, - }; - } - self.consecutive = self.consecutive.saturating_add(1); - if self.consecutive >= LOOKUP_FAULT_HALT_AFTER { - LookupFaultAction::RejectEngineFault - } else { - LookupFaultAction::Warn + } } } } @@ -1550,28 +1574,15 @@ pub(crate) fn spawn_confirm_engine( feed_wb.finish(heights_hashes.iter().map(|(h, _)| *h)); continue; } - let recover = if matches!( - &e, - rbitcoin_consensus::ConsensusError::Store(se) - if se.is_uring_session_fault() - ) { - match hub_wb.query.uring_recover("ibd-confirm-write") { - Ok(r) => r, - Err(re) => { - warn!("ibd: confirm write uring recover: {re}"); - rbitcoin_query::UringRecover::Exhausted - } - } - } else { - rbitcoin_query::UringRecover::Exhausted - }; + let recover = try_uring_recover(&hub_wb.query, &e, "ibd-confirm-write"); let action = classify_write_store_fault(&e, recover); - match apply_write_store_fault( + apply_write_store_fault( &feed_wb, &heights_hashes, action, &load_ahead_reset_wb, - ) { + ); + match action { WriteStoreFault::Requeue => { warn!( "ibd: confirm write uring recover @ {height} batch_parts={parts}: {e}" @@ -1667,6 +1678,27 @@ pub(crate) fn spawn_confirm_engine( info!("ibd: confirm scripts aborted: {msg}"); return false; } + if let Some(r) = try_uring_recover(&hub_sc.query, &e, "ibd-confirm-scripts") { + match r { + rbitcoin_query::UringRecover::Recovered => { + let req: Vec<(u32, BlockHash, Option)> = meta + .heights_hashes + .iter() + .map(|(h, raw)| { + (*h, BlockHash::from_byte_array(*raw), None) + }) + .collect(); + feed_sc.requeue_wire(&req); + warn!("ibd: confirm scripts uring recover @ {}: {e}", meta.first_h); + return true; + } + rbitcoin_query::UringRecover::Exhausted => { + rbitcoin_store::abort_uring_unusable( + "recover credit exhausted on ibd-confirm-scripts", + ); + } + } + } let (height, hash) = meta .heights_hashes .first() @@ -1836,6 +1868,30 @@ pub(crate) fn spawn_confirm_engine( let _ = scripts.join(); return; } + if let Some(r) = + try_uring_recover(&hub_load.query, &e, "ibd-confirm-load") + { + match r { + rbitcoin_query::UringRecover::Recovered => { + let req: Vec<(u32, BlockHash, Option)> = + wire_batch + .iter() + .map(|(h, ha, _)| (*h, *ha, None)) + .collect(); + feed_load.requeue_wire(&req); + warn!( + "ibd: confirm load stamp uring recover {first} @ {expect_h}: {e}", + first = wire_batch[0].1 + ); + continue; + } + rbitcoin_query::UringRecover::Exhausted => { + rbitcoin_store::abort_uring_unusable( + "recover credit exhausted on ibd-confirm-load", + ); + } + } + } let first_hash = wire_batch[0].1; if wire_batch.len() > 1 { let tail: Vec<(u32, BlockHash, Option)> = @@ -1978,6 +2034,31 @@ pub(crate) fn spawn_confirm_engine( let _ = scripts.join(); return; } + if let Some(r) = try_uring_recover_msg( + &hub_load.query, + &msg, + "ibd-confirm-load", + ) { + match r { + rbitcoin_query::UringRecover::Recovered => { + let req: Vec<(u32, BlockHash, Option)> = + heights_hashes + .iter() + .map(|(h, ha)| (*h, *ha, None)) + .collect(); + feed_load.requeue_wire(&req); + warn!( + "ibd: confirm load uring recover {first_hash} @ {expect_h}: {e}" + ); + continue; + } + rbitcoin_query::UringRecover::Exhausted => { + rbitcoin_store::abort_uring_unusable( + "recover credit exhausted on ibd-confirm-load", + ); + } + } + } if heights_hashes.len() > 1 { let tail: Vec<(u32, BlockHash, Option)> = heights_hashes @@ -2151,21 +2232,8 @@ pub(crate) fn spawn_confirm_engine( lookup_faults.on_success(); } Err(e) => { - let recover = if matches!( - &e, - rbitcoin_consensus::ConsensusError::Store(se) - if se.is_uring_session_fault() - ) { - match hub.query.uring_recover("ibd-confirm-lookup") { - Ok(r) => r, - Err(re) => { - warn!("ibd: bq resolve wave uring recover: {re}"); - rbitcoin_query::UringRecover::Exhausted - } - } - } else { - rbitcoin_query::UringRecover::Recovered - }; + let recover = + try_uring_recover(&hub.query, &e, "ibd-confirm-lookup"); match lookup_faults.on_err(&e, recover) { LookupFaultAction::Ignore => { debug!("ibd: bq resolve wave: {e}"); @@ -2186,21 +2254,24 @@ pub(crate) fn spawn_confirm_engine( 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); - let hash = feed - .inner - .lock() - .ok() - .and_then(|g| g.ready.get(&height).map(|(h, _)| *h)) - .unwrap_or_else(|| BlockHash::from_byte_array([0u8; 32])); - let _ = emit_confirm_reject( - &event_tx_lookup, - &feed, - height, - hash, - ConfirmRejectClass::EngineFault, - e.to_string(), - 1, - ); + 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 99ec766f3b..277f82b6cd 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/tests.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/tests.rs @@ -1256,16 +1256,21 @@ fn write_store_fault_uring_requeues_then_aborts() { let undrained = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring undrained")); assert_eq!( - classify_write_store_fault(&undrained, UringRecover::Recovered), + classify_write_store_fault(&undrained, Some(UringRecover::Recovered)), WriteStoreFault::Requeue ); assert_eq!( - classify_write_store_fault(&undrained, UringRecover::Exhausted), + classify_write_store_fault(&undrained, Some(UringRecover::Exhausted)), WriteStoreFault::Abort ); + let leftover = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring leftover cqe")); + assert_eq!( + classify_write_store_fault(&leftover, Some(UringRecover::Recovered)), + WriteStoreFault::Requeue + ); let io = ConsensusError::Store(StoreError::io("/tmp/x", std::io::Error::other("disk"))); assert_eq!( - classify_write_store_fault(&io, UringRecover::Recovered), + classify_write_store_fault(&io, None), WriteStoreFault::Reject(super::ConfirmRejectClass::EngineFault) ); @@ -1276,10 +1281,7 @@ fn write_store_fault_uring_requeues_then_aborts() { let mut g = feed.inner.lock().unwrap(); g.inflight.insert(10); } - assert_eq!( - apply_write_store_fault(&feed, &[(10, raw)], WriteStoreFault::Requeue, &reset), - WriteStoreFault::Requeue - ); + apply_write_store_fault(&feed, &[(10, raw)], WriteStoreFault::Requeue, &reset); assert!(reset.load(std::sync::atomic::Ordering::Acquire)); let g = feed.inner.lock().unwrap(); assert!(g.ready.contains_key(&10)); @@ -1296,32 +1298,37 @@ fn lookup_fault_policy_recover_abort_and_io_halt() { let mut p = LookupFaultPolicy::default(); let uring = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring undrained")); assert_eq!( - p.on_err(&uring, UringRecover::Recovered), + p.on_err(&uring, Some(UringRecover::Recovered)), LookupFaultAction::RecoverContinue ); assert_eq!( - p.on_err(&uring, UringRecover::Exhausted), + p.on_err(&uring, Some(UringRecover::Exhausted)), LookupFaultAction::Abort ); - let bp = ConsensusError::Store(StoreError::BudgetFull("io_uring SQ")); + let leftover = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring leftover cqe")); assert_eq!( - p.on_err(&bp, UringRecover::Recovered), - LookupFaultAction::Ignore + p.on_err(&leftover, Some(UringRecover::Recovered)), + LookupFaultAction::RecoverContinue ); + let bp = ConsensusError::Store(StoreError::BudgetFull("io_uring SQ")); + assert_eq!(p.on_err(&bp, None), 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, UringRecover::Recovered), - LookupFaultAction::Warn - ); + assert_eq!(p.on_err(&io, None), LookupFaultAction::Warn); } - assert_eq!( - p.on_err(&io, UringRecover::Recovered), - LookupFaultAction::RejectEngineFault - ); + assert_eq!(p.on_err(&io, None), LookupFaultAction::RejectEngineFault); p.on_success(); - assert_eq!( - p.on_err(&io, UringRecover::Recovered), - LookupFaultAction::Warn - ); + assert_eq!(p.on_err(&io, None), 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 89d0fe1239..7b653ce1e9 100644 --- a/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs +++ b/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs @@ -133,7 +133,11 @@ fn confirm_reject_class_matches_substring_table() { ConfirmRejectClass::EngineFault ); assert_eq!( - ConfirmRejectClass::from_err_str("invariant: io_uring wait timeout"), + ConfirmRejectClass::from_err_str("invariant: io_uring leftover cqe"), + ConfirmRejectClass::EngineFault + ); + assert_eq!( + ConfirmRejectClass::from_err_str("io_uring submit failed"), ConfirmRejectClass::EngineFault ); } From 56e8f19a7f5e3b3b4bbaac3c6d568fd27327cdaf Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:44:10 -0700 Subject: [PATCH 11/13] docs: recover is credit-only; Drop and BQ dequeue named Open still repairs leftover strong. In-process recover does not. Lookup 8 consecutive Io dequeues BQ; session-fault recover does not. Co-authored-by: Cursor --- CHANGELOG.md | 8 +++++--- OPERATOR.md | 2 +- docs/crash-recovery.md | 2 +- docs/invariants.md | 1 + docs/io-modality.md | 32 ++++++++++++++++++-------------- 5 files changed, 26 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ec8f36578..63d262be12 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,9 +22,11 @@ before 1.0). - **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. Write/lookup recover once - per 1000 heights (same Class C repair as open) instead of a 19h warn loop. - Restart with `RBITCOIN_IO=pread` if uring cannot complete. [`OPERATOR.md`](OPERATOR.md) + 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 diff --git a/OPERATOR.md b/OPERATOR.md index a63b9dad2e..90f39920f5 100644 --- a/OPERATOR.md +++ b/OPERATOR.md @@ -179,7 +179,7 @@ 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: io_uring unusable`), +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 diff --git a/docs/crash-recovery.md b/docs/crash-recovery.md index fd42b62436..627db5727c 100644 --- a/docs/crash-recovery.md +++ b/docs/crash-recovery.md @@ -24,7 +24,7 @@ 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 io_uring recover (IBD write/lookup session fault after drain idle) runs that same `repair_class_c_above_tip` and does **not** truncate Class A. Tip stays the commit point. +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. 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 a694410d65..bf104b1b94 100644 --- a/docs/io-modality.md +++ b/docs/io-modality.md @@ -49,20 +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_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 the process (buffers still -pinned). 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: a session fault after a successful drain recovers once per -1000-height window (`repair_class_c_above_tip`, drop this thread's ring, -requeue the same BQ bodies). A second stall in that window aborts and names -`RBITCOIN_IO=pread`. +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 From 526931804b08cadc46eac735ea0f03be47a0d4c1 Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:48:57 -0700 Subject: [PATCH 12/13] store: drop Linux-only wait-one Enter on Darwin/Windows WaitOneWindow::Enter is only built from io_uring enter errors. IOCP wait_one_cqe had no callers after the timeout drain loop. Both failed -D warnings on macos/windows CI. Co-authored-by: Cursor --- crates/rbitcoin-store/src/io_session_iocp.rs | 7 ------- crates/rbitcoin-store/src/uring_session.rs | 2 ++ 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/crates/rbitcoin-store/src/io_session_iocp.rs b/crates/rbitcoin-store/src/io_session_iocp.rs index 443bffed70..a387059257 100644 --- a/crates/rbitcoin-store/src/io_session_iocp.rs +++ b/crates/rbitcoin-store/src/io_session_iocp.rs @@ -189,13 +189,6 @@ impl IocpEngine { self.ready.drain(..).collect() } - pub(crate) fn wait_one_cqe(&mut self) -> Result<(), StoreError> { - if self.ready.is_empty() { - self.poll(u32::MAX); - } - Ok(()) - } - pub(crate) fn wait_one_cqe_timeout(&mut self, timeout_ms: u32) -> bool { if self.ready.is_empty() { self.poll(timeout_ms); diff --git a/crates/rbitcoin-store/src/uring_session.rs b/crates/rbitcoin-store/src/uring_session.rs index f15fd34e6a..5d49c063eb 100644 --- a/crates/rbitcoin-store/src/uring_session.rs +++ b/crates/rbitcoin-store/src/uring_session.rs @@ -49,6 +49,7 @@ enum WaitOneWindow { Ready, TimedOut, Idle, + #[cfg(target_os = "linux")] Enter(StoreError), } @@ -512,6 +513,7 @@ impl UringSession { 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); From 6b52dac20024e91abdde8bd361657f04ed3b866e Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:58:05 -0700 Subject: [PATCH 13/13] net,query: one recover-or-abort path for session faults Write, load, scripts, lookup, and tip-connect each matched Recovered vs Exhausted. Credit/abort lives on Query; IBD requeue sites share one helper. Lookup's 8-strike halt no longer carries recover arms. Co-authored-by: Cursor --- crates/rbitcoin-net/src/chain.rs | 10 +- crates/rbitcoin-net/src/ibd/confirm/mod.rs | 332 +++++++------------ crates/rbitcoin-net/src/ibd/confirm/tests.rs | 86 ++--- crates/rbitcoin-query/src/lib.rs | 11 + crates/rbitcoin-query/src/query_tests.rs | 7 + 5 files changed, 183 insertions(+), 263 deletions(-) diff --git a/crates/rbitcoin-net/src/chain.rs b/crates/rbitcoin-net/src/chain.rs index c37e4edcef..529c0d81a1 100644 --- a/crates/rbitcoin-net/src/chain.rs +++ b/crates/rbitcoin-net/src/chain.rs @@ -2050,14 +2050,8 @@ impl ChainHub { ) { Ok(fk) => return Ok(fk), Err(e) if e.is_uring_session_fault() => { - match self.query.uring_recover("tip-connect") { - rbitcoin_query::UringRecover::Recovered => continue, - rbitcoin_query::UringRecover::Exhausted => { - rbitcoin_store::abort_uring_unusable( - "recover credit exhausted on tip-connect", - ); - } - } + self.query.uring_recover_or_abort("tip-connect"); + continue; } Err(e) => return Err(e), } diff --git a/crates/rbitcoin-net/src/ibd/confirm/mod.rs b/crates/rbitcoin-net/src/ibd/confirm/mod.rs index 90dd4a2acf..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) { @@ -474,67 +479,19 @@ impl ConfirmRejectClass { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum WriteStoreFault { - Requeue, - Abort, - Reject(ConfirmRejectClass), -} - -pub(crate) fn classify_write_store_fault( - e: &rbitcoin_consensus::ConsensusError, - recover: Option, -) -> WriteStoreFault { - match recover { - Some(rbitcoin_query::UringRecover::Recovered) => WriteStoreFault::Requeue, - Some(rbitcoin_query::UringRecover::Exhausted) => WriteStoreFault::Abort, - None => WriteStoreFault::Reject(ConfirmRejectClass::from_consensus(e)), - } -} - -pub(crate) fn apply_write_store_fault( - feed: &ConfirmFeed, - heights: &[(u32, [u8; 32])], - action: WriteStoreFault, - load_ahead_reset: &AtomicBool, -) { - match action { - WriteStoreFault::Requeue => { - load_ahead_reset.store(true, Ordering::Release); - let batch: Vec<_> = heights - .iter() - .map(|(h, raw)| (*h, BlockHash::from_byte_array(*raw), None)) - .collect(); - feed.requeue_wire(&batch); - } - WriteStoreFault::Abort => {} - WriteStoreFault::Reject(_) => { - load_ahead_reset.store(true, Ordering::Release); - feed.finish(heights.iter().map(|(h, _)| *h)); - } - } -} - -fn try_uring_recover( +fn requeue_on_uring_recover( query: &rbitcoin_query::Query, - e: &rbitcoin_consensus::ConsensusError, - reason: &'static str, -) -> Option { - if !e.is_uring_session_fault() { - return None; - } - Some(query.uring_recover(reason)) -} - -fn try_uring_recover_msg( - query: &rbitcoin_query::Query, - msg: &str, + feed: &ConfirmFeed, + session_fault: bool, reason: &'static str, -) -> Option { - if !rbitcoin_store::StoreError::is_uring_session_fault_msg(msg) { - return None; + heights: impl IntoIterator, +) -> bool { + if !session_fault { + return false; } - Some(query.uring_recover(reason)) + query.uring_recover_or_abort(reason); + feed.requeue_hashes(heights); + true } pub(crate) fn lookup_ready_hash(feed: &ConfirmFeed, height: u32) -> Option { @@ -554,8 +511,6 @@ pub(crate) struct LookupFaultPolicy { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum LookupFaultAction { Ignore, - RecoverContinue, - Abort, Warn, RejectEngineFault, } @@ -565,30 +520,17 @@ impl LookupFaultPolicy { self.consecutive = 0; } - pub(crate) fn on_err( - &mut self, - e: &rbitcoin_consensus::ConsensusError, - recover: Option, - ) -> LookupFaultAction { + 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; } - match recover { - Some(rbitcoin_query::UringRecover::Recovered) => { - self.consecutive = 0; - LookupFaultAction::RecoverContinue - } - Some(rbitcoin_query::UringRecover::Exhausted) => LookupFaultAction::Abort, - None => { - self.consecutive = self.consecutive.saturating_add(1); - if self.consecutive >= LOOKUP_FAULT_HALT_AFTER { - LookupFaultAction::RejectEngineFault - } else { - LookupFaultAction::Warn - } - } + self.consecutive = self.consecutive.saturating_add(1); + if self.consecutive >= LOOKUP_FAULT_HALT_AFTER { + LookupFaultAction::RejectEngineFault + } else { + LookupFaultAction::Warn } } } @@ -1574,44 +1516,37 @@ pub(crate) fn spawn_confirm_engine( feed_wb.finish(heights_hashes.iter().map(|(h, _)| *h)); continue; } - let recover = try_uring_recover(&hub_wb.query, &e, "ibd-confirm-write"); - let action = classify_write_store_fault(&e, recover); - apply_write_store_fault( + if requeue_on_uring_recover( + &hub_wb.query, &feed_wb, - &heights_hashes, - action, - &load_ahead_reset_wb, - ); - match action { - WriteStoreFault::Requeue => { - warn!( - "ibd: confirm write uring recover @ {height} batch_parts={parts}: {e}" - ); - continue; - } - WriteStoreFault::Abort => { - rbitcoin_store::abort_uring_unusable( - "recover credit exhausted on ibd-confirm-write", - ); - } - WriteStoreFault::Reject(class) => { - loop_stats_wb - .confirm_reject_stops - .fetch_add(1, Ordering::Relaxed); - warn!( - "ibd: confirm write reject @ {height} batch_parts={parts}: {e}" - ); - let _ = emit_confirm_reject( - &event_tx_wb, - &feed_wb, - height, - hash, - class, - msg, - heights_hashes.len(), - ); - } + 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 + .confirm_reject_stops + .fetch_add(1, Ordering::Relaxed); + 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, + class, + msg, + heights_hashes.len(), + ); } } } @@ -1678,26 +1613,17 @@ pub(crate) fn spawn_confirm_engine( info!("ibd: confirm scripts aborted: {msg}"); return false; } - if let Some(r) = try_uring_recover(&hub_sc.query, &e, "ibd-confirm-scripts") { - match r { - rbitcoin_query::UringRecover::Recovered => { - let req: Vec<(u32, BlockHash, Option)> = meta - .heights_hashes - .iter() - .map(|(h, raw)| { - (*h, BlockHash::from_byte_array(*raw), None) - }) - .collect(); - feed_sc.requeue_wire(&req); - warn!("ibd: confirm scripts uring recover @ {}: {e}", meta.first_h); - return true; - } - rbitcoin_query::UringRecover::Exhausted => { - rbitcoin_store::abort_uring_unusable( - "recover credit exhausted on ibd-confirm-scripts", - ); - } - } + 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 @@ -1868,29 +1794,18 @@ pub(crate) fn spawn_confirm_engine( let _ = scripts.join(); return; } - if let Some(r) = - try_uring_recover(&hub_load.query, &e, "ibd-confirm-load") - { - match r { - rbitcoin_query::UringRecover::Recovered => { - let req: Vec<(u32, BlockHash, Option)> = - wire_batch - .iter() - .map(|(h, ha, _)| (*h, *ha, None)) - .collect(); - feed_load.requeue_wire(&req); - warn!( - "ibd: confirm load stamp uring recover {first} @ {expect_h}: {e}", - first = wire_batch[0].1 - ); - continue; - } - rbitcoin_query::UringRecover::Exhausted => { - rbitcoin_store::abort_uring_unusable( - "recover credit exhausted on ibd-confirm-load", - ); - } - } + 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 { @@ -2034,30 +1949,17 @@ pub(crate) fn spawn_confirm_engine( let _ = scripts.join(); return; } - if let Some(r) = try_uring_recover_msg( + if requeue_on_uring_recover( &hub_load.query, - &msg, + &feed_load, + rbitcoin_store::StoreError::is_uring_session_fault_msg(&msg), "ibd-confirm-load", + heights_hashes.iter().copied(), ) { - match r { - rbitcoin_query::UringRecover::Recovered => { - let req: Vec<(u32, BlockHash, Option)> = - heights_hashes - .iter() - .map(|(h, ha)| (*h, *ha, None)) - .collect(); - feed_load.requeue_wire(&req); - warn!( - "ibd: confirm load uring recover {first_hash} @ {expect_h}: {e}" - ); - continue; - } - rbitcoin_query::UringRecover::Exhausted => { - rbitcoin_store::abort_uring_unusable( - "recover credit exhausted on ibd-confirm-load", - ); - } - } + warn!( + "ibd: confirm load uring recover {first_hash} @ {expect_h}: {e}" + ); + continue; } if heights_hashes.len() > 1 { let tail: Vec<(u32, BlockHash, Option)> = @@ -2232,44 +2134,40 @@ pub(crate) fn spawn_confirm_engine( lookup_faults.on_success(); } Err(e) => { - let recover = - try_uring_recover(&hub.query, &e, "ibd-confirm-lookup"); - match lookup_faults.on_err(&e, recover) { - LookupFaultAction::Ignore => { - debug!("ibd: bq resolve wave: {e}"); - } - LookupFaultAction::RecoverContinue => { - warn!("ibd: bq resolve wave uring recover: {e}"); - } - LookupFaultAction::Abort => { - rbitcoin_store::abort_uring_unusable( - "recover credit exhausted on ibd-confirm-lookup", - ); - } - 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)" - ); + 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 { + 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 277f82b6cd..7dcd35ea51 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/tests.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/tests.rs @@ -1247,78 +1247,88 @@ fn plan_epoch_stale_after_clear() { } #[test] -fn write_store_fault_uring_requeues_then_aborts() { - use super::{apply_write_store_fault, classify_write_store_fault, WriteStoreFault}; +fn write_session_fault_is_engine_fault_and_requeue_puts_ready() { use rbitcoin_consensus::ConsensusError; - use rbitcoin_query::UringRecover; use rbitcoin_store::StoreError; - use std::sync::atomic::AtomicBool; let undrained = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring undrained")); assert_eq!( - classify_write_store_fault(&undrained, Some(UringRecover::Recovered)), - WriteStoreFault::Requeue - ); - assert_eq!( - classify_write_store_fault(&undrained, Some(UringRecover::Exhausted)), - WriteStoreFault::Abort + super::ConfirmRejectClass::from_consensus(&undrained), + super::ConfirmRejectClass::EngineFault ); let leftover = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring leftover cqe")); assert_eq!( - classify_write_store_fault(&leftover, Some(UringRecover::Recovered)), - WriteStoreFault::Requeue + super::ConfirmRejectClass::from_consensus(&leftover), + super::ConfirmRejectClass::EngineFault ); let io = ConsensusError::Store(StoreError::io("/tmp/x", std::io::Error::other("disk"))); assert_eq!( - classify_write_store_fault(&io, None), - WriteStoreFault::Reject(super::ConfirmRejectClass::EngineFault) + super::ConfirmRejectClass::from_consensus(&io), + super::ConfirmRejectClass::EngineFault ); let feed = ConfirmFeed::new(); - let reset = AtomicBool::new(false); - let raw = [1u8; 32]; + let hash = bitcoin::BlockHash::from_byte_array([1u8; 32]); { let mut g = feed.inner.lock().unwrap(); g.inflight.insert(10); } - apply_write_store_fault(&feed, &[(10, raw)], WriteStoreFault::Requeue, &reset); - assert!(reset.load(std::sync::atomic::Ordering::Acquire)); + 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 lookup_fault_policy_recover_abort_and_io_halt() { +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_query::UringRecover; use rbitcoin_store::StoreError; let mut p = LookupFaultPolicy::default(); - let uring = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring undrained")); - assert_eq!( - p.on_err(&uring, Some(UringRecover::Recovered)), - LookupFaultAction::RecoverContinue - ); - assert_eq!( - p.on_err(&uring, Some(UringRecover::Exhausted)), - LookupFaultAction::Abort - ); - let leftover = ConsensusError::Store(StoreError::Corrupt("invariant: io_uring leftover cqe")); - assert_eq!( - p.on_err(&leftover, Some(UringRecover::Recovered)), - LookupFaultAction::RecoverContinue - ); let bp = ConsensusError::Store(StoreError::BudgetFull("io_uring SQ")); - assert_eq!(p.on_err(&bp, None), LookupFaultAction::Ignore); + 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, None), LookupFaultAction::Warn); + assert_eq!(p.on_err(&io), LookupFaultAction::Warn); } - assert_eq!(p.on_err(&io, None), LookupFaultAction::RejectEngineFault); + assert_eq!(p.on_err(&io), LookupFaultAction::RejectEngineFault); p.on_success(); - assert_eq!(p.on_err(&io, None), LookupFaultAction::Warn); + assert_eq!(p.on_err(&io), LookupFaultAction::Warn); } #[test] diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 82deb782bb..072fbef772 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/crates/rbitcoin-query/src/lib.rs @@ -1056,6 +1056,17 @@ impl Query { } } + /// 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 cf0a360727..6ce292e2c5 100644 --- a/crates/rbitcoin-query/src/query_tests.rs +++ b/crates/rbitcoin-query/src/query_tests.rs @@ -70,6 +70,13 @@ fn uring_recover_cas_second_claim_at_same_tip_is_exhausted() { 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) }