net: FeeFilter min_relay is atomic, not inner.read

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

crates/rbitcoin-net/src/tx_relay.rs

@@ -335,6 +335,9 @@ pub struct MempoolHub {
     accept_gen: Mutex<HashMap<Wtxid, u64>>,
     /// 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<Txid> = 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() {

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<HashSet>` for O(1) `has_block` (IBD assign path) |
-| `MempoolHub::inner` | One `RwLock<ActiveMempool>`. **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<ActiveMempool>`. **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)).