ibd: stop a confirm lookup wave on the confirm cancel

changelog.d/lookup-wave-cancel.md

@@ -0,0 +1,11 @@
+Fixed
+
+- **IBD teardown no longer waits out a whole confirm lookup wave.** A wave
+  resolves up to 64k parent txids in one TipOnly batch. On a busy HDD
+  datadir that batch took most of an hour, and stop, SIGTERM and a catch-up
+  give-up all waited for it (`still waiting for confirm engine` for 58
+  minutes in the field). The batch still runs once (one open probe, one
+  sealed walk, one `create.loc` batch) and checks the confirm cancel before
+  each sealed segment and each identity shot. A live sync never sets that
+  flag. A cancelled wave ends the lookup thread instead of counting as a
+  lookup fault.

crates/rbitcoin-ibd/src/ibd/confirm/mod.rs

@@ -796,6 +796,8 @@ pub enum LookupFaultAction {
     Ignore,
     Warn,
     RejectEngineFault,
+    /// The wave saw the confirm cancel (IBD teardown): leave the lookup loop.
+    Stop,
 }
 
 impl LookupFaultPolicy {
@@ -804,6 +806,9 @@ impl LookupFaultPolicy {
     }
 
     pub fn on_err(&mut self, e: &rbitcoin_query::ConfirmError) -> LookupFaultAction {
+        if e.is_cancelled() {
+            return LookupFaultAction::Stop;
+        }
         let backpressure = matches!(
             e,
             rbitcoin_query::ConfirmError::Store(se) if se.is_io_backpressure()
@@ -3011,6 +3016,10 @@ pub fn spawn_confirm_engine(
                                 warn!("ibd: bq resolve wave uring recover: {e}");
                             } else {
                                 match lookup_faults.on_err(&e) {
+                                    LookupFaultAction::Stop => {
+                                        info!("ibd: confirm lookup cancelled mid-wave");
+                                        break;
+                                    }
                                     LookupFaultAction::Ignore => {
                                         debug!("ibd: bq resolve wave: {e}");
                                     }

crates/rbitcoin-ibd/src/ibd/perf_log.rs

@@ -371,7 +371,7 @@ pub struct IbdPerfSample {
     pub lookup_decode_ms: u64,
     /// Lookup-wave `from_tx_wire` on payload slices (`precompute=`).
     pub lookup_precompute_ms: u64,
-    /// Lookup-wave TipOnly `get_fk_by_txid_batch` (`wave=… head=`). Not load stamp.
+    /// Lookup-wave TipOnly `get_fk_by_txid_batch_cancellable` (`wave=… head=`). Not load stamp.
     pub lookup_wave_head_ms: u64,
     /// TipOnly CPU (slot/fuse probe + fence snapshot) inside `head=`.
     pub lookup_wave_head_probe_ms: u64,
@@ -380,7 +380,7 @@ pub struct IbdPerfSample {
     /// TipOnly `txid.body` / identity preads this window.
     pub lookup_wave_head_preads: u64,
     /// Lookup-wave `create.loc` fill inside TipOnly `head=` (`wave=… loc=`).
-    /// One batch after identity; not on the probe ring.
+    /// One batch after the sealed walk; not on the probe ring.
     pub lookup_wave_spent_ms: u64,
     pub plan_parents: u64,
     pub plan_already: u64,

crates/rbitcoin-ibd/tests/ibd/confirm.rs

@@ -1652,6 +1652,24 @@ fn lookup_fault_policy_io_halt() {
     assert_eq!(p.on_err(&io), LookupFaultAction::Warn);
 }
 
+/// A wave cut short by the confirm cancel (IBD teardown) is not a fault: the
+/// lookup thread stops, and repeats never add up to a halt that would emit a
+/// reject on the way out.
+#[test]
+fn lookup_fault_policy_stops_on_cancel() {
+    use rbitcoin_ibd::ibd::confirm::{
+        LookupFaultAction, LookupFaultPolicy, LOOKUP_FAULT_HALT_AFTER,
+    };
+    use rbitcoin_store::StoreError;
+
+    let mut p = LookupFaultPolicy::default();
+    let cancelled = rbitcoin_query::ConfirmError::Store(StoreError::Cancelled("txid batch"));
+    for _ in 0..=LOOKUP_FAULT_HALT_AFTER {
+        assert_eq!(p.on_err(&cancelled), LookupFaultAction::Stop);
+    }
+    assert_eq!(p.consecutive, 0);
+}
+
 #[test]
 fn lookup_ready_hash_none_when_missing() {
     let feed = ConfirmFeed::new();

crates/rbitcoin-query/src/confirm_run/bq_resolve.rs

@@ -1,8 +1,10 @@
 //! BQ-ahead TipOnly parent resolve (lookup wave).
 //!
 //! Pack/hold uses enqueue-stamped Σ inputs (no `Block` decode). Emit runs one
-//! `Store::get_fk_by_txid_batch` (TipOnly) across the selected heights
-//! (soft **64000** inputs / hard **1080** blocks). Hits ride the wave as
+//! `Store::get_fk_by_txid_batch_cancellable` (TipOnly) across the selected
+//! heights (soft **64000** inputs / hard **1080** blocks). The confirm cancel
+//! is asked before each sealed segment and each identity shot. Hits ride the
+//! wave as
 //! [`crate::BatchParentIds`] (fk + body_range + spent_range). Does not
 //! claim, structure, or stamp. Same-wave creates are omitted from TipOnly need.
 
@@ -136,10 +138,10 @@ pub struct BqResolveWaveStats {
     pub precompute_ns: u64,
     /// `push_resolve_keys` (this wave).
     pub collect_ns: u64,
-    /// TipOnly `get_fk_by_txid_batch` (this wave).
+    /// TipOnly `get_fk_by_txid_batch_cancellable` (this wave).
     pub head_ns: u64,
     /// TipOnly `create.loc` fill is inside [`Self::head_ns`] (`idx_ns`): one
-    /// batch after identity waves, not on the held probe ring.
+    /// batch after the sealed walk, not on the held probe ring.
     pub spent_ns: u64,
 }
 
@@ -362,7 +364,7 @@ pub fn confirm_bq_resolve_wave_capped(
     if !need.is_empty() {
         let rows = query
             .store()
-            .get_fk_by_txid_batch(&need)
+            .get_fk_by_txid_batch_cancellable(&need, &|| query.confirm_cancelled())
             .map_err(ConfirmError::from)?;
         for (txid, row) in rows {
             if let Some((fk, pair)) = row {

crates/rbitcoin-query/tests/confirm_run/bq_resolve.rs

@@ -934,3 +934,87 @@ fn take_after_a_rearm_leaves_its_rows_queued() {
     assert!(!q.block_queue_has_height(1) && !q.block_queue_has_height(2));
     let _ = std::fs::remove_dir_all(&path);
 }
+
+/// IBD teardown sets the confirm cancel, then joins the lookup thread. A wave
+/// under that flag returns cancelled instead of resolving its TipOnly need: a
+/// full wave is up to 64k cold reads, most of an hour on a busy HDD.
+#[test]
+fn wave_under_confirm_cancel_returns_cancelled() {
+    let (path, q) = tmp_query();
+    let params = ChainParams::regtest();
+    let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest);
+    accept_and_connect_block(&q, &params, Height::GENESIS, &genesis, Milestone::NONE).unwrap();
+    let g_cb = genesis.txdata[0].compute_txid();
+    let b1 = mine_with_txs(
+        genesis.block_hash(),
+        genesis.header.time + 600,
+        1,
+        vec![spend_op_true(g_cb, 0, Amount::from_sat(49_0000_0000))],
+    );
+    q.block_queue_enqueue(1, b1.block_hash().to_byte_array(), 1, &serialize(&b1))
+        .unwrap();
+    q.request_confirm_cancel();
+    let err = match confirm_bq_resolve_wave_capped(
+        &q,
+        &params,
+        Milestone::NONE,
+        &[1],
+        BQ_RESOLVE_WAVE_MAX_BLOCKS,
+        BQ_RESOLVE_WAVE_MAX_INPUTS,
+    ) {
+        Ok(_) => panic!("a wave under the confirm cancel must not resolve"),
+        Err(e) => e,
+    };
+    assert!(err.is_cancelled(), "want cancelled, got {err:?}");
+    // Promote and dequeue sit after the `?`, so a cancelled wave leaves the
+    // height queued for the next session.
+    assert!(q.block_queue_has_height(1));
+    let _ = std::fs::remove_dir_all(&path);
+}
+
+/// The cancellable batch is one resolve of the whole need. Uncancelled it
+/// returns the rows the one-shot batch does. The cancel is asked before the
+/// resolve and before each identity shot, so a flag set on the first ask
+/// returns cancelled and never resolves.
+#[test]
+fn txid_batch_asks_cancel_before_the_identity_shot() {
+    let (path, q) = tmp_query();
+    let params = ChainParams::regtest();
+    let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest);
+    accept_and_connect_block(&q, &params, Height::GENESIS, &genesis, Milestone::NONE).unwrap();
+    let mut txids = vec![genesis.txdata[0].compute_txid().to_byte_array()];
+    let mut prev = genesis.block_hash();
+    let mut time = genesis.header.time;
+    for h in 1..=2u32 {
+        time += 600;
+        let b = mine_with_txs(prev, time, h, vec![]);
+        accept_and_connect_block(&q, &params, Height(h), &b, Milestone::NONE).unwrap();
+        txids.push(b.txdata[0].compute_txid().to_byte_array());
+        prev = b.block_hash();
+    }
+    let whole = q.store().get_fk_by_txid_batch(&txids).unwrap();
+    assert!(
+        whole.iter().all(|(_, row)| row.is_some()),
+        "connected coinbases resolve"
+    );
+    let once = q
+        .store()
+        .get_fk_by_txid_batch_cancellable(&txids, &|| false)
+        .unwrap();
+    assert_eq!(once, whole, "one resolve returns the one-shot rows");
+
+    let asked = std::cell::Cell::new(0u32);
+    let err = q
+        .store()
+        .get_fk_by_txid_batch_cancellable(&txids, &|| {
+            asked.set(asked.get() + 1);
+            true
+        })
+        .unwrap_err();
+    assert!(
+        matches!(err, rbitcoin_store::StoreError::Cancelled(_)),
+        "{err:?}"
+    );
+    assert_eq!(asked.get(), 1, "the resolve asks before any identity shot");
+    let _ = std::fs::remove_dir_all(&path);
+}

crates/rbitcoin-store/src/head_resolve_denserels.rs

@@ -210,17 +210,33 @@ pub fn resolve_fk_and_range_batch_opts(
     txids: &[[u8; 32]],
     heights: Option<&HeightFence>,
     tip_only: bool,
+) -> Result<Vec<crate::tx_table::TxidFkRange>, StoreError> {
+    resolve_fk_and_range_batch_cancellable(table, txids, heights, tip_only, &|| false)
+}
+
+/// Same machine as [`resolve_fk_and_range_batch_opts`]. `cancelled` is asked
+/// before each sealed segment and before each identity shot, so a stop waits
+/// for one segment rather than the whole need. One open probe, one sealed
+/// walk, one `create.loc` batch.
+pub fn resolve_fk_and_range_batch_cancellable(
+    table: &TxTable,
+    txids: &[[u8; 32]],
+    heights: Option<&HeightFence>,
+    tip_only: bool,
+    cancelled: &dyn Fn() -> bool,
 ) -> Result<Vec<crate::tx_table::TxidFkRange>, StoreError> {
     if txids.is_empty() {
         crate::head_resolve_stats::clear_leftover_miss();
         return Ok(Vec::new());
     }
     match io_backend::read_io_backend() {
         ReadIoBackend::Uring => map_uring_resolve(
-            resolve_fk_and_range_uring(table, txids, heights, tip_only),
-            || resolve_fk_and_range_pread(table, txids, heights, tip_only),
+            resolve_fk_and_range_uring(table, txids, heights, tip_only, cancelled),
+            || resolve_fk_and_range_pread(table, txids, heights, tip_only, cancelled),
         ),
-        ReadIoBackend::Pread => resolve_fk_and_range_pread(table, txids, heights, tip_only),
+        ReadIoBackend::Pread => {
+            resolve_fk_and_range_pread(table, txids, heights, tip_only, cancelled)
+        }
     }
 }
 
@@ -269,6 +285,7 @@ pub fn resolve_identity_core(
     heights: Option<&HeightFence>,
     tip_only: bool,
     session: Option<&mut UringSession>,
+    cancelled: &dyn Fn() -> bool,
 ) -> Result<IdentityHits, StoreError> {
     crate::head_resolve_stats::add_keys(txids.len() as u64);
 
@@ -308,12 +325,16 @@ pub fn resolve_identity_core(
         &mut local_age,
         &mut ctx,
         &mut had_id,
+        cancelled,
     )?;
 
     if any_unfinished(&picked, &connected, heights) {
         let n_segs = table.head.segment_count();
         let mut active = unfinished_mask(&picked, &connected, heights);
         for si in (0..n_segs).rev() {
+            if cancelled() {
+                return Err(StoreError::Cancelled("txid batch"));
+            }
             if !active.iter().any(|on| *on) {
                 break;
             }
@@ -340,6 +361,7 @@ pub fn resolve_identity_core(
                 &mut local_age,
                 &mut ctx,
                 &mut had_id,
+                cancelled,
             )?;
             for (i, on) in active.iter_mut().enumerate() {
                 *on = !key_finished(i, &picked, &connected, heights);
@@ -372,8 +394,9 @@ pub fn resolve_fk_and_range_pread(
     txids: &[[u8; 32]],
     heights: Option<&HeightFence>,
     tip_only: bool,
+    cancelled: &dyn Fn() -> bool,
 ) -> Result<Vec<crate::tx_table::TxidFkRange>, StoreError> {
-    let ident = resolve_identity_core(table, txids, heights, tip_only, None)?;
+    let ident = resolve_identity_core(table, txids, heights, tip_only, None, cancelled)?;
     attach_loc_to_identity(table, txids, ident)
 }
 
@@ -476,6 +499,7 @@ pub fn id_idx_wave(
     local_age: &mut [u64; crate::head_resolve_stats::AGE_CAP],
     ctx: &mut crate::IoCtx<'_>,
     had_id: &mut [bool],
+    cancelled: &dyn Fn() -> bool,
 ) -> Result<(), StoreError> {
     use crate::head_resolve_pick::{next_id_shot, walk_id_prefix, ID_FILL_CHUNK};
     use crate::int_map::{U64Map, U64Set};
@@ -493,6 +517,9 @@ pub fn id_idx_wave(
     let mut id_map: U64Map<[u8; 32]> = U64Map::default();
 
     for take in [ID_FILL_CHUNK, usize::MAX] {
+        if cancelled() {
+            return Err(StoreError::Cancelled("txid batch"));
+        }
         let shot = next_id_shot(cands_by_key, &filled, &skip, take);
         let mut need: Vec<Fk> = Vec::new();
         {
@@ -577,9 +604,10 @@ pub fn resolve_fk_and_range_uring(
     txids: &[[u8; 32]],
     heights: Option<&HeightFence>,
     tip_only: bool,
+    cancelled: &dyn Fn() -> bool,
 ) -> Result<Vec<crate::tx_table::TxidFkRange>, StoreError> {
     let ident = uring_session::with_thread_local(uring_session::DEFAULT_ENTRIES, |session| {
-        resolve_identity_core(table, txids, heights, tip_only, Some(session))
+        resolve_identity_core(table, txids, heights, tip_only, Some(session), cancelled)
     })??;
     attach_loc_to_identity(table, txids, ident)
 }

crates/rbitcoin-store/src/store.rs

@@ -1146,6 +1146,27 @@ impl Store {
         &self,
         txids: &[[u8; 32]],
         mode: TxidResolveMode,
+    ) -> Result<Vec<crate::tx_table::TxidFkRange>, StoreError> {
+        self.resolve_txid_batch(txids, mode, &|| false)
+    }
+
+    /// [`Self::get_fk_by_txid_batch`] that asks `cancelled` before each sealed
+    /// segment and each identity shot. One open probe, one sealed walk, one
+    /// `create.loc` batch. A confirm lookup wave is up to 64k cold reads; a
+    /// stop now waits for one segment rather than the wave.
+    pub fn get_fk_by_txid_batch_cancellable(
+        &self,
+        txids: &[[u8; 32]],
+        cancelled: &dyn Fn() -> bool,
+    ) -> Result<Vec<crate::tx_table::TxidFkRange>, StoreError> {
+        self.resolve_txid_batch(txids, TxidResolveMode::TipOnly, cancelled)
+    }
+
+    fn resolve_txid_batch(
+        &self,
+        txids: &[[u8; 32]],
+        mode: TxidResolveMode,
+        cancelled: &dyn Fn() -> bool,
     ) -> Result<Vec<crate::tx_table::TxidFkRange>, StoreError> {
         // Snapshot: leftover IO is 0.4–2s. Holding the fence read lock blocks
         // `height_fence_extend`. Confirm extends before `set_many`, so tip
@@ -1154,11 +1175,16 @@ impl Store {
         let t_fence = std::time::Instant::now();
         let fence = self.fence().clone();
         crate::head_resolve_stats::add_probe(t_fence.elapsed().as_nanos() as u64);
-        crate::head_resolve_denserels::resolve_fk_and_range_batch_with_tip(
+        let tip_only = matches!(mode, TxidResolveMode::TipOnly);
+        if cancelled() {
+            return Err(StoreError::Cancelled("txid batch"));
+        }
+        crate::head_resolve_denserels::resolve_fk_and_range_batch_cancellable(
             &self.txs,
-            &fence,
             txids,
-            matches!(mode, TxidResolveMode::TipOnly),
+            Some(&fence),
+            tip_only,
+            cancelled,
         )
     }
 

crates/rbitcoin-store/tests/head_resolve_denserels.rs

@@ -110,7 +110,7 @@ fn resolve_uring_unavailable_falls_back_to_pread() {
 #[test]
 fn uring_fk_and_range_matches_pread() {
     let (dir, t, txids) = seed_table(40);
-    let pread = resolve_fk_and_range_pread(&t, &txids, None, false).unwrap();
+    let pread = resolve_fk_and_range_pread(&t, &txids, None, false, &|| false).unwrap();
     // Public entry (uring when available, else pread) must match pure pread.
     let via = resolve_fk_and_range_batch(&t, &txids).unwrap();
     assert_eq!(pread.len(), via.len());
@@ -334,7 +334,10 @@ fn loc_batch_matches_serial_body_range() {
         assert_eq!(got.map(|p| p.txout), Some(exp), "fk={}", fk.0);
     }
     let got =
-        resolve_fk_and_range_pread(&t, &[txids[0], txids[1], txids[1099]], None, false).unwrap();
+        resolve_fk_and_range_pread(&t, &[txids[0], txids[1], txids[1099]], None, false, &|| {
+            false
+        })
+        .unwrap();
     assert_eq!(got[0].1, Some((first, batch[0].unwrap())));
     assert_eq!(got[1].1, Some((near, batch[1].unwrap())));
     assert_eq!(got[2].1, Some((far, batch[2].unwrap())));