From 5944b51d1c50df1364af15c3be4b5f36154c605c Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Fri, 4 Sep 2026 08:25:49 -0700 Subject: [PATCH] net: FeeFilter min_relay is atomic, not inner.read Tip-follow handshake called min_relay_sat_kvb() on the session, which took a blocking mempool inner read. After #320's assert_not_reactor that panicked tokio-rt-worker and left follow_live=0. Overlay lives on an AtomicU64. rebroadcast_unbroadcast uses try_contains so blocksonly re-INV does not park either. --- CHANGELOG.md | 5 +++ crates/rbitcoin-net/src/tx_relay.rs | 60 ++++++++++++++++++++++++++++- docs/concurrency.md | 4 +- 3 files changed, 65 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 938f664b..19afc33a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,11 @@ before 1.0). ### Fixed +- **Tip-follow FeeFilter no longer panics on the reactor:** session handshake + called `min_relay_sat_kvb()`, which took a blocking `inner` read after + #320's `assert_not_reactor`. The overlay is an atomic; `rebroadcast_unbroadcast` + uses `try_contains` instead of blocking `contains`. + - **Tip-follow reactor never parks on mempool `inner`:** INV/getdata/compact use `try_read` (busy write skips that item). Accept commit, orphan promote, package, and reorg strip no longer hold `inner` write across diff --git a/crates/rbitcoin-net/src/tx_relay.rs b/crates/rbitcoin-net/src/tx_relay.rs index fefc354a..2806a7bd 100644 --- a/crates/rbitcoin-net/src/tx_relay.rs +++ b/crates/rbitcoin-net/src/tx_relay.rs @@ -335,6 +335,9 @@ pub struct MempoolHub { accept_gen: Mutex>, /// Core `-mempoolexpiry` in seconds (default 336h). expiry_secs: AtomicU64, + /// Core `-minrelaytxfee` overlay (sat/kvB). Session FeeFilter reads this + /// without taking `inner`. + min_relay_sat_kvb: AtomicU64, } impl MempoolHub { @@ -420,6 +423,9 @@ impl MempoolHub { next_accept_gen: AtomicU64::new(1), accept_gen: Mutex::new(HashMap::new()), expiry_secs: AtomicU64::new(DEFAULT_MEMPOOL_EXPIRY_SECS), + min_relay_sat_kvb: AtomicU64::new( + rbitcoin_consensus::policy::MIN_RELAY_FEE_RATE_SAT_PER_KVB, + ), fee_deltas: Mutex::new(HashMap::new()), template_updates: AtomicU64::new(0), }; @@ -1787,11 +1793,12 @@ impl MempoolHub { /// Core `-minrelaytxfee` overlay (sat/kvB). `0` admits any non-negative fee. pub fn set_min_relay_sat_kvb(&self, sat_kvb: u64) { + self.min_relay_sat_kvb.store(sat_kvb, Ordering::Release); self.lock_write().set_min_relay_sat_kvb(sat_kvb); } pub fn min_relay_sat_kvb(&self) -> u64 { - self.lock_read().min_relay_sat_kvb() + self.min_relay_sat_kvb.load(Ordering::Acquire) } /// In-mempool ancestors of `txid`, **excluding** itself (Core RPC). @@ -1888,7 +1895,7 @@ impl MempoolHub { pub fn rebroadcast_unbroadcast(&self) { let ids: Vec = self.unbroadcast.lock().unwrap().iter().copied().collect(); for txid in ids { - if !self.contains(&txid) { + if !self.try_contains(&txid) { continue; } let _ = self.announce.send(MempoolAnnounce { @@ -2860,6 +2867,55 @@ mod tests { let _ = std::fs::remove_dir_all(&store_dir); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn min_relay_sat_kvb_does_not_panic_on_tokio_worker() { + if std::env::var_os("RBITCOIN_HEAD_SCALE").is_none() { + std::env::set_var("RBITCOIN_HEAD_SCALE", "tiny"); + } + let store_dir = tmp(); + let mp_dir = tmp(); + let q = Query::open_or_create(&store_dir).unwrap(); + let hub = MempoolHub::open(&mp_dir, Arc::new(q)).unwrap(); + hub.set_min_relay_sat_kvb(250); + let join = tokio::spawn(async move { + let name = std::thread::current().name().unwrap_or("").to_string(); + let rate = hub.min_relay_sat_kvb(); + (name, rate) + }); + let (name, rate) = join.await.expect("join worker"); + assert!( + name.starts_with("tokio-rt-worker"), + "spawned task must run on a tokio worker, got {name:?}" + ); + assert_eq!(rate, 250); + let _ = std::fs::remove_dir_all(&mp_dir); + let _ = std::fs::remove_dir_all(&store_dir); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn rebroadcast_unbroadcast_does_not_panic_on_tokio_worker() { + if std::env::var_os("RBITCOIN_HEAD_SCALE").is_none() { + std::env::set_var("RBITCOIN_HEAD_SCALE", "tiny"); + } + let store_dir = tmp(); + let mp_dir = tmp(); + let q = Query::open_or_create(&store_dir).unwrap(); + let hub = MempoolHub::open(&mp_dir, Arc::new(q)).unwrap(); + hub.note_unbroadcast(Txid::from_byte_array([1u8; 32])); + let join = tokio::spawn(async move { + let name = std::thread::current().name().unwrap_or("").to_string(); + hub.rebroadcast_unbroadcast(); + name + }); + let name = join.await.expect("join worker"); + assert!( + name.starts_with("tokio-rt-worker"), + "spawned task must run on a tokio worker, got {name:?}" + ); + let _ = std::fs::remove_dir_all(&mp_dir); + let _ = std::fs::remove_dir_all(&store_dir); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn blocking_contains_refuses_tokio_worker() { if std::env::var_os("RBITCOIN_HEAD_SCALE").is_none() { diff --git a/docs/concurrency.md b/docs/concurrency.md index caaa6f30..9d648364 100644 --- a/docs/concurrency.md +++ b/docs/concurrency.md @@ -51,7 +51,7 @@ Three thread kinds only. **Tokio workers must not wait on a `std` mutex/rwlock, | Role | Notes | |------|--------| -| `peer_session` (split read/write) | Serve + reconstruct compact/body. Offers reconstructed blocks to **`tip-accept`** (does **not** take `connect_lock` or run confirm on the tokio worker). P2P `tx` is `accept_tx_async` (blocking pool). INV / getdata / compact fill use mempool **`try_read` only** (busy write → skip that item; never park). 50 ms tick calls `PeerHub::on_session_heartbeat` (headers-sync stall timeout). Inbound accept is `inbound_connect_and_handshake` (60 s VERSION/VERACK); timeout drops the `max_inbound` permit. | +| `peer_session` (split read/write) | Serve + reconstruct compact/body. Offers reconstructed blocks to **`tip-accept`** (does **not** take `connect_lock` or run confirm on the tokio worker). P2P `tx` is `accept_tx_async` (blocking pool). INV / getdata / compact fill use mempool **`try_read` only** (busy write → skip that item; never park). Handshake `FeeFilter` reads **atomic** `-minrelaytxfee` (no `inner`). 50 ms tick calls `PeerHub::on_session_heartbeat` (headers-sync stall timeout). Inbound accept is `inbound_connect_and_handshake` (60 s VERSION/VERACK); timeout drops the `max_inbound` permit. | | `tip-accept` | **One** process-wide OS thread. Queue depth 8. Sole production thread for `accept_block` / `accept_branch` / `accept_received_block` / `generate_to_script` / `connect_lock` at tip. Confirm is **`confirm_wire_run_preverified`** (lookup stamp, load pin/assemble, `confirm_scripts_phase` → `rbtc-scripts-*`, write + `ibd-confirm-head` drain). TLS uring is this thread’s `with_thread_local` session. SIGINT stays on tokio; the current job finishes, then the session sees shutdown. Dropping the session’s join future **detaches** (does not condvar-wait on the worker). | | `rbtc-sh-wb` | **One** Class B scripthash appender. Used for tip follow **and** short catch-up when a durable SH head already exists. Confirm enqueues RAM records; `connect_at` / `note_confirmed_tip` **release** after `tip_tx`. This thread `put_create_batch_append` only for released heights, then advances `sh_indexed_through`. Apply errors re-queue and halt. Post-IBD Class A collect uses pack sessions (not a second appender); it runs while still Direct so this thread no-ops. | | Electrum / Esplora | Confirmed SH reads join durable index **plus a RAM SH head** (pending jobs keyed by scripthash) and pin that visible height (live tip while jobs sit, never above published tip). A tx is in mempool overlay **or** SH (pending/durable), not both and not neither. Reorg reaccepts then drops pending. Headers subscribe is live tip. | @@ -89,7 +89,7 @@ spend annotations. | `tx.head` segment seal | Roll opens the next OA immediately; BDZ+fuse8 runs on a sidecar. Lookup probes every unsealed OA until publish. Write joins the sidecar only on the *next* roll, `flush`, or `Drop` (not on the rolling insert). | | `header.head` overflow | Insert past 7/8 rolls `header.head.gN` (new empty file). Occupied rewrite is open-only: undersized single gen writes `header.head.grow` then rename. | | `ChainHub::confirmed` | `RwLock` for O(1) `has_block` (IBD assign path) | -| `MempoolHub::inner` | One `RwLock`. **Write** is graph mutation + durable slot append only (microseconds at ~12k live). No Query / script verify / compact reconstruct under write. Prepare may hold a **read** across Query. Blocking `read`/`write` is `assert_not_reactor`. Session INV/getdata/compact use `try_read`; miss = skip that item (re-getdata / notfound / reconstruct without those txs). Compact fill siphashes live txid/wtxid and clones **matching** bodies only — never `list_live()`. Do not CoW a 12k-txid presence set on every accept. Lock order: `inner` → `sh_index` / `unbroadcast` / `relay_*`. | +| `MempoolHub::inner` | One `RwLock`. **Write** is graph mutation + durable slot append only (microseconds at ~12k live). No Query / script verify / compact reconstruct under write. Prepare may hold a **read** across Query. Blocking `read`/`write` is `assert_not_reactor`. Session INV/getdata/compact use `try_read`; miss = skip that item (re-getdata / notfound / reconstruct without those txs). Compact fill siphashes live txid/wtxid and clones **matching** bodies only — never `list_live()`. Handshake `FeeFilter` / `-minrelaytxfee` is an **atomic**, not `inner`. Do not CoW a 12k-txid presence set on every accept. Lock order: `inner` → `sh_index` / `unbroadcast` / `relay_*`. | There is **no** global “pause queries during confirm write.” Tip-as-commit + `is_confirmed_strong` define query visibility ([`crash-recovery.md`](./crash-recovery.md)).