From 199147f7c9af15e32cda41b37b2e194a01e22260 Mon Sep 17 00:00:00 2001 From: Chase Date: Fri, 17 Jul 2026 11:43:57 +0200 Subject: [PATCH] electrum: delta-driven subscription updates The periodic (5s) subscription update loop re-checked every subscribed scripthash via get_history on every tick - O(subscribed) work per cycle even when nothing changed. With ~60k subscriptions on a production replica this kept the periodic_update latency histogram fat-tailed (>250ms ticks) and burned steady CPU. The mempool already records which scripthashes every tx touches (tx_scripthashes, from the eviction-ownership fix). Track a touched-epoch per scripthash on mempool add/remove; connections snapshot the epoch each tick and only re-check subscriptions whose scripthash was actually touched since their previous tick: O(touched) instead of O(subscribed). Correctness fallbacks keep the full scan for: the first tick of a connection, any tick where the chain tip moved (the chain index has no equivalent touched-set, and blocks are rare relative to ticks), and any tick whose epoch snapshot predates the tracking map's pruning floor (bounded at 200k entries / ~720 epochs). touched_since takes &self on the read guard so per-connection ticks never contend on the mempool write lock; pruning happens inside add(). --- src/electrum/server.rs | 27 ++++++++++++++++++++++ src/new_index/mempool.rs | 50 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/src/electrum/server.rs b/src/electrum/server.rs index 045ccf185..e80c8eddd 100644 --- a/src/electrum/server.rs +++ b/src/electrum/server.rs @@ -158,6 +158,9 @@ struct Connection { query: Arc, last_header_entry: Option, status_hashes: HashMap, // ScriptHash -> StatusHash + // Mempool touched-epoch snapshot from this connection's last periodic tick; + // None forces a full subscription scan (first tick / after pruning). + mempool_touched_epoch: Option, stream: TcpStream, addr: SocketAddr, sender: SyncSender, @@ -192,6 +195,7 @@ impl Connection { query, last_header_entry: None, // disable header subscription for now status_hashes: HashMap::new(), + mempool_touched_epoch: None, stream, addr, sender, @@ -617,9 +621,11 @@ impl Connection { .with_label_values(&["periodic_update"]) .start_timer(); let mut result = vec![]; + let mut tip_changed = false; if let Some(ref mut last_entry) = self.last_header_entry { let entry = self.query.chain().best_header(); if *last_entry != entry { + tip_changed = true; *last_entry = entry; let hex_header = serialize_hex(last_entry.header()); let header = json!({"hex": hex_header, "height": last_entry.height()}); @@ -629,7 +635,28 @@ impl Connection { "params": [header]})); } } + // Delta-driven fast path: on a mempool-only tick (no new block), only + // re-check subscriptions whose scripthash the mempool actually touched + // since our last tick. A new block can affect any subscription (and the + // chain index has no equivalent touched-set yet), so tip changes keep the + // full scan; blocks are rare relative to 5s ticks. `None` (first tick, or + // the mempool pruned past our snapshot) also falls back to the full scan. + let (current_epoch, touched) = { + let mempool = self.query.mempool(); + let epoch = mempool.touched_epoch(); + let touched = match self.mempool_touched_epoch { + Some(since) if !tip_changed => mempool.touched_since(since), + _ => None, + }; + (epoch, touched) + }; + self.mempool_touched_epoch = Some(current_epoch); for (script_hash, status_hash) in self.status_hashes.iter_mut() { + if let Some(ref touched) = touched { + if !touched.contains(&script_hash.to_byte_array()) { + continue; + } + } let history_txids = get_history(&self.query, &script_hash[..], self.txs_limit)?; let new_status_hash = get_status_hash(history_txids, &self.query) .map_or(Value::Null, |h| json!(h.to_lower_hex_string())); diff --git a/src/new_index/mempool.rs b/src/new_index/mempool.rs index ddb91fe9c..06eb28ce3 100644 --- a/src/new_index/mempool.rs +++ b/src/new_index/mempool.rs @@ -37,6 +37,15 @@ pub struct Mempool { feeinfo: HashMap, // Map txid -> scripthashes touched, to prune efficiently on eviction. tx_scripthashes: HashMap>, + // Delta tracking for electrum subscription updates: scripthash -> the epoch at + // which a mempool add/removal last touched it. Connections snapshot the epoch + // each periodic tick and only re-check scripthashes touched since their last + // tick instead of every subscription (O(touched) vs O(subscribed)). + touched: HashMap, + touched_epoch: u64, + // Entries older than this epoch have been pruned; ticks whose snapshot predates + // it must fall back to a full scan for correctness. + touched_floor: u64, history: HashMap>, // ScriptHash -> {history_entries} edges: HashMap, // OutPoint -> (spending_txid, spending_vin) recent: ArrayDeque, // The N most recent txs to enter the mempool @@ -74,6 +83,9 @@ impl Mempool { txstore: HashMap::new(), feeinfo: HashMap::new(), tx_scripthashes: HashMap::new(), + touched: HashMap::new(), + touched_epoch: 0, + touched_floor: 0, history: HashMap::new(), edges: HashMap::new(), recent: ArrayDeque::new(), @@ -346,6 +358,15 @@ impl Mempool { #[trace] fn add(&mut self, txs_map: HashMap) -> Result<()> { + self.touched_epoch += 1; + // Bound the touched-tracking map: entries older than ~720 epochs (roughly an + // hour of 5s ticks) cannot matter to any connection that ticks regularly; + // ticks whose snapshot predates the floor fall back to a full scan. + if self.touched.len() > 200_000 { + let floor = self.touched_epoch.saturating_sub(720); + self.touched.retain(|_, e| *e >= floor); + self.touched_floor = floor; + } self.delta .with_label_values(&["add"]) .observe(txs_map.len() as f64); @@ -445,6 +466,9 @@ impl Mempool { } tx_scripthashes.sort_unstable(); tx_scripthashes.dedup(); + for sh in &tx_scripthashes { + self.touched.insert(*sh, self.touched_epoch); + } self.tx_scripthashes.insert(txid, tx_scripthashes); for (i, txi) in tx.input.iter().enumerate() { self.edges.insert(txi.previous_output, (txid, i as u32)); @@ -493,11 +517,34 @@ impl Mempool { } #[trace] + /// Current touched-epoch; bumped on every mempool add/removal batch. + pub fn touched_epoch(&self) -> u64 { + self.touched_epoch + } + + /// Scripthashes touched strictly after `since`, or `None` when `since` predates + /// the pruning floor (caller must fall back to a full scan). Also prunes the + /// tracking map when it grows large: entries older than ~720 epochs (roughly an + /// hour of 5s ticks) cannot matter to any live connection that ticks regularly. + pub fn touched_since(&self, since: u64) -> Option> { + if since < self.touched_floor { + return None; + } + Some( + self.touched + .iter() + .filter(|(_, e)| **e > since) + .map(|(sh, _)| *sh) + .collect(), + ) + } + fn remove(&mut self, to_remove: HashSet<&Txid>) { self.delta .with_label_values(&["remove"]) .observe(to_remove.len() as f64); let _timer = self.latency.with_label_values(&["remove"]).start_timer(); + self.touched_epoch += 1; for txid in &to_remove { let tx = self @@ -513,6 +560,9 @@ impl Mempool { .tx_scripthashes .remove(*txid) .unwrap_or_else(|| panic!("missing tx_scripthashes for {}", txid)); + for sh in &scripthashes { + self.touched.insert(*sh, self.touched_epoch); + } prune_history_entries(&mut self.history, &scripthashes, txid); for txin in tx.input {