Give each IBD thread one role.

changelog.d/ibd-thread-roles.md

@@ -0,0 +1,7 @@
+Changed
+
+- The IBD header thread is named `ibd-header` for the catch-up, then joined.
+- `rbtc-sh-wb` starts when `--sh-index` has the index in Tip. A durable head starts it before catch-up. Direct catch-up starts it only after materialize. It stays down when materialize is cancelled or fails.
+- IBD Class C barriers share one `rbtc-tip-flush` thread for the confirm engine. Tip mode still flushes that barrier on the tip-accept thread.
+- Heavy pure CPU leaves the blocking pool for `rbtc-scripts-*`: v2 encode when `encode_is_cpu_heavy`, compact short ids after the body is in hand, and Electrum or Esplora JSON of at least 4 KiB, including `GET /block-template`. Shorter encodes stay on the caller. Block templates still send `cache-control: no-store`. Block and header decode stay on the blocking pool.
+- A panic inside one of those script-pool jobs does not kill the worker. A dropped JSON encode is an error, not `{}`. A dropped compact-block encode is a peer protocol error, not a header fallback.

crates/rbitcoin-consensus/src/script_pool.rs

@@ -14,12 +14,14 @@
 //! No rayon / crossbeam.
 
 use arc_swap::ArcSwap;
-use std::cell::Cell;
 use std::collections::VecDeque;
+use std::future::Future;
 use std::ops::Range;
+use std::pin::Pin;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::mpsc;
 use std::sync::{Arc, Condvar, Mutex, OnceLock};
+use std::task::{Context, Poll, Waker};
 use std::thread;
 
 /// Jobs claimed per steal. Amortizes `next` / `in_wave` / `Arc<Wave>` traffic
@@ -28,13 +30,9 @@ pub const STEAL_CHUNK: usize = 32;
 
 use crate::error::ConsensusError;
 
-thread_local! {
-    static ON_STEAL_WORKER: Cell<bool> = const { Cell::new(false) };
-}
-
 #[doc(hidden)]
 pub fn on_steal_worker() -> bool {
-    ON_STEAL_WORKER.with(|c| c.get())
+    rbitcoin_primitives::on_script_worker()
 }
 
 /// Type-erased `f(&items[i])`. `ctx` is valid until the publisher drops the
@@ -589,7 +587,7 @@ pub fn workers() -> &'static ScriptWorkers {
             let Ok(h) = thread::Builder::new()
                 .name(format!("rbtc-scripts-{i}"))
                 .spawn(move || {
-                    ON_STEAL_WORKER.with(|c| c.set(true));
+                    rbitcoin_primitives::set_script_worker(true);
                     loop {
                         if steal_or_job(pool) {
                             continue;
@@ -725,10 +723,88 @@ where
     wake_steal_workers();
 }
 
+struct CpuState<T> {
+    value: Option<T>,
+    waker: Option<Waker>,
+    done: bool,
+}
+
+struct CpuFinish<T> {
+    state: Arc<Mutex<CpuState<T>>>,
+    value: Option<T>,
+}
+
+impl<T> Drop for CpuFinish<T> {
+    fn drop(&mut self) {
+        let waker = {
+            let mut slot = self.state.lock().unwrap_or_else(|p| p.into_inner());
+            if let Some(value) = self.value.take() {
+                slot.value = Some(value);
+            }
+            slot.done = true;
+            slot.waker.take()
+        };
+        if let Some(waker) = waker {
+            waker.wake();
+        }
+    }
+}
+
+/// A [`run_cpu`] join. `None` means the worker dropped the reply.
+#[doc(hidden)]
+pub struct CpuJob<T> {
+    state: Arc<Mutex<CpuState<T>>>,
+}
+
+impl<T> Future for CpuJob<T> {
+    type Output = Option<T>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
+        let mut slot = self.state.lock().unwrap_or_else(|p| p.into_inner());
+        if let Some(value) = slot.value.take() {
+            return Poll::Ready(Some(value));
+        }
+        if slot.done {
+            return Poll::Ready(None);
+        }
+        slot.waker = Some(cx.waker().clone());
+        Poll::Pending
+    }
+}
+
+/// Run `work` on `rbtc-scripts-*` and await it.
+///
+/// The caller parks. It does not hold a blocking-pool thread. A published
+/// steal chunk is claimed before this job. `None` means the guest panicked
+/// and the worker dropped the reply. The worker stays in the pool.
+pub fn run_cpu<T, F>(work: F) -> CpuJob<T>
+where
+    T: Send + 'static,
+    F: FnOnce() -> T + Send + 'static,
+{
+    let state = Arc::new(Mutex::new(CpuState {
+        value: None,
+        waker: None,
+        done: false,
+    }));
+    let finish = CpuFinish {
+        state: Arc::clone(&state),
+        value: None,
+    };
+    spawn_detached(move || {
+        let mut finish = finish;
+        if let Ok(value) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(work)) {
+            finish.value = Some(value);
+        }
+    });
+    CpuJob { state }
+}
+
 /// Run `work` on the shared `rbtc-scripts` pool and join the result.
 ///
 /// Used by mempool accept so the peer/tokio stack never runs the interpreter
-/// (even for a single input). Returns `None` if the pool is gone.
+/// (even for a single input). Returns `None` if the pool is gone. Callers
+/// that can await should use [`run_cpu`] so they do not hold a blocking thread.
 pub fn run_detached_join<T, F>(work: F) -> Option<T>
 where
     T: Send + 'static,

crates/rbitcoin-consensus/tests/script_pool.rs

@@ -855,3 +855,106 @@ fn second_stage_keeps_the_first_publisher_wake() {
     a.join().expect("stage a");
     b.join().expect("stage b");
 }
+
+static CPU_ORDER: Mutex<Vec<u8>> = Mutex::new(Vec::new());
+
+fn note_cpu_wave(_: &u32) -> Result<(), ConsensusError> {
+    CPU_ORDER.lock().unwrap_or_else(|p| p.into_inner()).push(1);
+    Ok(())
+}
+
+fn block_on_cpu<T>(job: CpuJob<T>) -> Option<T> {
+    use std::future::Future;
+    use std::pin::Pin;
+    use std::task::{Context, Poll, Wake, Waker};
+    struct Unpark(thread::Thread);
+    impl Wake for Unpark {
+        fn wake(self: Arc<Self>) {
+            self.0.unpark();
+        }
+    }
+    let mut job = Pin::new(Box::new(job));
+    let waker = Waker::from(Arc::new(Unpark(thread::current())));
+    let mut cx = Context::from_waker(&waker);
+    loop {
+        match job.as_mut().poll(&mut cx) {
+            Poll::Ready(v) => return v,
+            Poll::Pending => thread::park(),
+        }
+    }
+}
+
+/// `None` when `limit` passes with the job still pending. The waiter thread
+/// is left behind on timeout so a dead script worker cannot hang the suite.
+fn block_on_cpu_timeout<T: Send + 'static>(job: CpuJob<T>, limit: Duration) -> Option<Option<T>> {
+    let (tx, rx) = mpsc::channel();
+    thread::spawn(move || {
+        let _ = tx.send(block_on_cpu(job));
+    });
+    rx.recv_timeout(limit).ok()
+}
+
+/// `run_cpu` finishes on `rbtc-scripts-*`. The caller parks; it does not
+/// take a blocking-pool thread. A published steal chunk runs before the guest.
+#[test]
+fn run_cpu_uses_one_script_worker_after_the_wave() {
+    let _gate = STEAL_TEST.lock().unwrap_or_else(|p| p.into_inner());
+    CPU_ORDER.lock().unwrap_or_else(|p| p.into_inner()).clear();
+    let n = worker_spawn_count();
+    let occupy = OccupyGate::occupy_n(n.saturating_sub(1));
+    let wave = start_for_each_pooled(vec![1u32], note_cpu_wave, 1)
+        .unwrap()
+        .expect("published wave");
+    let name = block_on_cpu(run_cpu(|| {
+        CPU_ORDER.lock().unwrap_or_else(|p| p.into_inner()).push(2);
+        thread::current().name().unwrap_or("").to_string()
+    }))
+    .expect("script worker");
+    wave.finish().unwrap();
+    occupy.release();
+    assert!(name.starts_with("rbtc-scripts-"), "guest ran on {name}");
+    assert_eq!(
+        CPU_ORDER
+            .lock()
+            .unwrap_or_else(|p| p.into_inner())
+            .as_slice(),
+        &[1, 2],
+        "the steal chunk finishes before the guest"
+    );
+}
+
+/// A panicking guest drops the reply and leaves that worker in the pool.
+/// The other workers are held, so the next job can run only on the one
+/// that just panicked.
+#[test]
+fn run_cpu_panic_keeps_the_worker() {
+    let _gate = STEAL_TEST.lock().unwrap_or_else(|p| p.into_inner());
+    let n = worker_spawn_count();
+    let occupy = OccupyGate::occupy_n(n.saturating_sub(1));
+    let lost = block_on_cpu_timeout(
+        run_cpu(|| -> u8 { panic!("run_cpu guest") }),
+        Duration::from_secs(2),
+    );
+    assert_eq!(lost, Some(None), "a panicking guest must drop the reply");
+    let next = block_on_cpu_timeout(run_cpu(|| 7u8), Duration::from_secs(2));
+    occupy.release();
+    assert_eq!(
+        next,
+        Some(Some(7)),
+        "the script worker must take the next job"
+    );
+}
+
+/// A guest that enters store IO trips the script-worker guard.
+#[test]
+fn run_cpu_guest_cannot_touch_store_io() {
+    let _gate = STEAL_TEST.lock().unwrap_or_else(|p| p.into_inner());
+    let hit = block_on_cpu(run_cpu(|| {
+        std::panic::catch_unwind(|| {
+            rbitcoin_primitives::assert_not_script_worker("store read");
+        })
+        .is_err()
+    }))
+    .expect("script worker");
+    assert!(hit, "store read on a script worker must panic");
+}

crates/rbitcoin-electrum/src/json_cpu.rs

@@ -0,0 +1,84 @@
+//! JSON replies. Under 4 KiB encodes on the caller. At 4 KiB and above,
+//! encoding runs on `rbtc-scripts-*`. The store join stays where it was.
+
+use serde_json::Value;
+use std::sync::Arc;
+
+/// Encoded bodies at or above this leave the caller for `rbtc-scripts-*`.
+pub const JSON_CPU_MIN: usize = 4096;
+
+/// True when a lower bound of the encoded size reaches [`JSON_CPU_MIN`].
+/// The walk stops at that bound.
+#[doc(hidden)]
+pub fn json_is_cpu_heavy(v: &Value) -> bool {
+    let mut left = JSON_CPU_MIN;
+    walk(v, &mut left)
+}
+
+fn take(left: &mut usize, n: usize) -> bool {
+    if n >= *left {
+        *left = 0;
+        true
+    } else {
+        *left -= n;
+        false
+    }
+}
+
+fn walk(v: &Value, left: &mut usize) -> bool {
+    if *left == 0 {
+        return true;
+    }
+    match v {
+        Value::Null => take(left, 4),
+        Value::Bool(true) => take(left, 4),
+        Value::Bool(false) => take(left, 5),
+        Value::Number(_) => take(left, 1),
+        Value::String(s) => take(left, s.len().saturating_add(2)),
+        Value::Array(items) => {
+            if take(left, 1) {
+                return true;
+            }
+            for (i, item) in items.iter().enumerate() {
+                if i > 0 && take(left, 1) {
+                    return true;
+                }
+                if walk(item, left) {
+                    return true;
+                }
+            }
+            take(left, 1)
+        }
+        Value::Object(map) => {
+            if take(left, 1) {
+                return true;
+            }
+            for (i, (key, item)) in map.iter().enumerate() {
+                if i > 0 && take(left, 1) {
+                    return true;
+                }
+                if take(left, key.len().saturating_add(3)) || walk(item, left) {
+                    return true;
+                }
+            }
+            take(left, 1)
+        }
+    }
+}
+
+/// Encode `v`. A heavy body hops to `rbtc-scripts-*`. A light body encodes here.
+///
+/// The `Arc` moves into the script job. Callers that already own the value
+/// do not clone it on the waiting task. `Err` means the worker dropped the
+/// reply.
+#[doc(hidden)]
+pub async fn encode_json_body(v: Arc<Value>) -> Result<Vec<u8>, ()> {
+    if !json_is_cpu_heavy(&v) {
+        return Ok(serde_json::to_vec(v.as_ref()).unwrap_or_else(|_| b"{}".to_vec()));
+    }
+    rbitcoin_consensus::script_pool::run_cpu(move || {
+        serde_json::to_vec(v.as_ref()).unwrap_or_else(|_| b"{}".to_vec())
+    })
+    .await
+    .ok_or(())
+}

crates/rbitcoin-electrum/src/lib.rs

@@ -12,6 +12,8 @@
 
 #![forbid(unsafe_code)]
 
+#[doc(hidden)]
+pub mod json_cpu;
 #[doc(hidden)]
 pub mod server;
 #[doc(hidden)]
@@ -21,6 +23,7 @@ pub mod tweaks;
 #[doc(hidden)]
 pub mod unspent;
 
+pub use json_cpu::{encode_json_body, json_is_cpu_heavy, JSON_CPU_MIN};
 pub use server::{
     batch_window, electrum_scripthash_hex, parse_electrum_request_line, perf_totals, run_electrum,
     sample_reset_perf, ElectrumConfig, ElectrumHandle, ServeLimits, TipNotify,