IBD: recover or abort on io_uring drain stall (#421)

CHANGELOG.md

@@ -20,6 +20,14 @@ before 1.0).
 
 ### Fixed
 
+- **IBD io_uring drain stall:** `drain_all` no longer returns after 5 s with
+  leftover SQEs (that freed in-flight buffers). Every TLS session waits while
+  CQEs arrive; a 120 s zero-completion stall aborts explicit drain (session
+  `Drop` does not abort). Write/lookup/load/scripts and tip-connect recover once
+  per 1000 heights (credit only; Class C leftover waits for open repair)
+  instead of a 19h warn loop. Restart with `RBITCOIN_IO=pread` if completions
+  cannot complete. [`OPERATOR.md`](OPERATOR.md)
+  / [`docs/io-modality.md`](docs/io-modality.md).
 - **`cmpct_differential` fill vs Core extra-txn:** Core v31 latches IBD in
   `UpdateIBDStatus` (`LoadChainTip` / connect), not `setmocktime`. A fresh
   regtest genesis stays in IBD under the default 24h `-maxtipage`, so P2P

OPERATOR.md

@@ -177,6 +177,11 @@ rbitcoin use otherwise-idle resources. This is an opt-in host policy, not a
 performance setting: sync can slow substantially under contention, and an idle
 I/O class can starve while higher-priority storage work continues.
 
+Idle `IOSchedulingClass` (and some filesystems, including bcachefs) can
+starve `io_uring` completions. Expect `store: io_uring drain slow` then at most
+one recover per ~1000 heights. If the node aborts (`store: completion session unusable`),
+restart as-is or with `RBITCOIN_IO=pread`. There is no mid-IBD libc fallback.
+
 For a regular systemd installation, create a service drop-in with
 `systemctl edit rbitcoin.service`:
 
@@ -512,7 +517,8 @@ Full modality matrix: [`docs/io-modality.md`](docs/io-modality.md).
 when available). Table transport is always **fd pread/pwrite**. Compact Class C
 is L2 write-behind; see [`docs/io-modality.md`](docs/io-modality.md). Per-path
 env overrides are **removed**. If `uring` is selected but setup fails, demote to
-**pread** / **pwrite**.
+**pread** / **pwrite**. If a live ring stops completing (`drain slow`, then abort),
+restart with **`RBITCOIN_IO=pread`** — the process does not switch backends itself.
 
 | Env | Values | Note |
 |-----|--------|------|

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<VecDeque<Job>>, cv: &Condvar) -> Job {
 
 pub(crate) struct HeadDrainHandle {
     rx: Option<Receiver<Result<u64, StoreError>>>,
+    restore: Option<Arc<Mutex<Option<Vec<([u8; 32], rbitcoin_primitives::Fk)>>>>>,
     #[cfg(test)]
     named: Arc<Mutex<Option<(ThreadId, String)>>>,
 }
 
 impl HeadDrainHandle {
-    pub(crate) fn join(mut self) -> Result<u64, StoreError> {
-        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<u64, StoreError>,
+        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<u64, StoreError> {
@@ -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
 }

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;

crates/rbitcoin-consensus/src/confirm_run/write.rs

@@ -252,11 +252,14 @@ pub fn confirm_write_phase(
     })();
 
     let t_join = Instant::now();
-    let drain_res = drain.join();
+    let (drain_res, restore) = drain.join_restore();
     let drain_join_ns = t_join.elapsed().as_nanos() as u64;
     if drain_join_ns > 0 {
         rbitcoin_query::note_confirm(&query.confirm_stats().write_drain_join_ns, drain_join_ns);
     }
+    if drain_res.is_err() {
+        query.store().txs.head_note_pending(&restore);
+    }
     let (out, n_blocks, structural_ns, struct_ph, class_c_ns, spend_ann_ns) = overlap?;
     drain_res.map_err(ConsensusError::from)?;
     if let Some(fk) = drain_max_fk {

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() {

crates/rbitcoin-consensus/src/error.rs

@@ -60,6 +60,12 @@ impl From<StoreError> 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(

crates/rbitcoin-net/src/chain.rs

@@ -2039,15 +2039,22 @@ impl ChainHub {
         tip_accept_stats_reset(&self.query);
         let t_wall = std::time::Instant::now();
         let now = self.clock.now_secs();
-        rbitcoin_consensus::with_now(now, || {
-            accept_and_connect_block_preverified(
+        let _ = rbitcoin_consensus::with_now(now, || loop {
+            match accept_and_connect_block_preverified(
                 &self.query,
                 &self.params,
                 Height(height),
                 &block,
                 self.milestone,
                 &preverified,
-            )
+            ) {
+                Ok(fk) => return Ok(fk),
+                Err(e) if e.is_uring_session_fault() => {
+                    self.query.uring_recover_or_abort("tip-connect");
+                    continue;
+                }
+                Err(e) => return Err(e),
+            }
         })
         .map_err(|e| {
             let reason = rbitcoin_consensus::block_reject_reason(&e);