From 6c78ad386fc85e7bd7ba013108a2f8c85c7c09c4 Mon Sep 17 00:00:00 2001 From: Aryan Tikarya Date: Mon, 17 Aug 2026 14:19:41 +0530 Subject: [PATCH 1/3] fix(rpc): close ChainNotify channel on lagged consumers --- CHANGELOG.md | 2 + src/rpc/channel.rs | 118 +++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 116 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cd781b1b05f..57c400c28ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,8 @@ ### Fixed +- [#5795](https://github.com/ChainSafe/forest/issues/5795): `Filecoin.ChainNotify` now closes the subscription channel when a client falls too far behind instead of silently dropping head changes, matching Lotus, so clients can detect the gap and resubscribe. + - [#7473](https://github.com/ChainSafe/forest/pull/7473): Fixed `eth_estimateGas` under-estimating gas for nested contract calls (EIP-150's 63/64 rule), which could make transactions fail on chain with `SYS_OUT_OF_GAS`; the estimate is now raised until the message succeeds. Genuine reverts return an `execution reverted` error (JSON-RPC code `3`) with the decoded reason and data, matching Lotus. - [#7412](https://github.com/ChainSafe/forest/issues/7412): Fixes quicknet "unchained" logic to fetch the `max_beacon_round` for all covered epochs diff --git a/src/rpc/channel.rs b/src/rpc/channel.rs index 9d0aa86a565..9a031200508 100644 --- a/src/rpc/channel.rs +++ b/src/rpc/channel.rs @@ -274,6 +274,14 @@ fn close_channel_response(channel_id: ChannelId) -> MethodResponse { ) } +/// Sends the bare `xrpc.ch.close` notification for this channel, ignoring +/// send failures (the connection may already be gone). +async fn send_close(sink: &SubscriptionSink) { + if let Ok(payload) = to_raw_value(&close_payload(sink.channel_id())) { + let _ = sink.send(payload).await; + } +} + #[derive(Debug, Clone)] pub struct RpcModule { id_provider: Arc, @@ -384,12 +392,19 @@ impl RpcModule { } } Err(RecvError::Closed) => { - if let Ok(payload) = to_raw_value(&close_payload(sink.channel_id())) { - let _ = sink.send(payload).await; - } + send_close(&sink).await; break; } - Err(RecvError::Lagged(_)) => { + Err(RecvError::Lagged(n)) => { + // Events were lost: close the channel (like Lotus) + // so the client knows to resubscribe and resync, + // instead of silently continuing with a gap. + tracing::warn!( + "closing channel {}: subscriber lagged by {n} messages", + sink.channel_id() + ); + send_close(&sink).await; + break; } } }, @@ -453,3 +468,98 @@ impl RpcModule { ) } } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::{Value, json}; + use std::time::Duration; + use tokio::sync::broadcast; + + const TEST_METHOD: &str = "test.channel"; + const RECV_TIMEOUT: Duration = Duration::from_secs(5); + /// Capacity of the per-test event source; the lag test overflows it to + /// force a `Lagged` observation. + const SOURCE_CAPACITY: usize = 4; + /// Buffer size of the per-call frame stream returned by `raw_json_request`. + const STREAM_BUF_SIZE: usize = 256; + + /// Subscribe with the given request id; returns the allocated channel id + /// and the stream of frames sent to this "connection". + /// + /// Every `raw_json_request` call gets its own frame stream, but they all + /// share `ConnectionId(0)`. The duplicate subscribe response that + /// `accept()` writes to the transport sink is swallowed by + /// `raw_json_request` itself, so the stream carries notification frames + /// only. + async fn subscribe( + methods: &Methods, + request_id: u64, + ) -> (ChannelId, mpsc::Receiver>) { + let request = format!( + r#"{{"jsonrpc":"2.0","id":{request_id},"method":"{TEST_METHOD}","params":[]}}"# + ); + let (response, frames) = methods + .raw_json_request(&request, STREAM_BUF_SIZE) + .await + .unwrap(); + let response: Value = serde_json::from_str(response.get()).unwrap(); + assert_eq!(response.get("id"), Some(&json!(request_id))); + let channel_id = response + .get("result") + .and_then(Value::as_u64) + .unwrap_or_else(|| panic!("channel id must be a bare u64: {response}")); + (channel_id, frames) + } + + async fn next_frame(frames: &mut mpsc::Receiver>) -> Value { + let frame = tokio::time::timeout(RECV_TIMEOUT, frames.recv()) + .await + .expect("timed out waiting for a frame") + .expect("stream closed while waiting for a frame"); + serde_json::from_str(frame.get()).unwrap() + } + + fn close_frame(channel_id: ChannelId) -> Value { + json!({"jsonrpc": "2.0", "method": "xrpc.ch.close", "params": [channel_id]}) + } + + /// Regression test: a subscriber that falls behind the broadcast source + /// has its channel closed — like Lotus, so the client knows to + /// resubscribe and resync — instead of silently losing the overflowed + /// events while the channel stays open. + #[tokio::test] + async fn lagged_consumer_channel_closes() { + // The receiver handed to the pump is already lagged before the pump + // ever polls it: overflow it first, then subscribe. This makes the + // `Lagged` observation deterministic regardless of task scheduling. + let (events, lagged_rx) = broadcast::channel(SOURCE_CAPACITY); + for n in 0..SOURCE_CAPACITY + 2 { + events.send(format!("event-{n}")).unwrap(); + } + let lagged_rx = Mutex::new(Some(lagged_rx)); + let mut module = RpcModule::default(); + module + .register_channel(TEST_METHOD, move |_params| { + lagged_rx.lock().take().expect("single subscriber") + }) + .unwrap(); + let methods: Methods = module.into(); + + let (channel_id, mut frames) = subscribe(&methods, 1).await; + + // no value frames arrive — the client is told the channel is gone + assert_eq!(next_frame(&mut frames).await, close_frame(channel_id)); + + // the pump exited and dropped its receiver — the source has no + // subscribers left, and no stray frames follow the close (the frame + // stream stays open because the pump's registry entry is not yet + // cleaned up on exit) + assert!(events.send("after-close".into()).is_err()); + tokio::task::yield_now().await; + assert!(matches!( + frames.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + )); + } +} From 1d3d05bde0b14d1b4a35d6f6d5191d55778e3f8d Mon Sep 17 00:00:00 2001 From: Aryan Tikarya Date: Mon, 17 Aug 2026 14:22:28 +0530 Subject: [PATCH 2/3] test(rpc): cover ChainNotify wire protocol and event semantics --- src/rpc/channel.rs | 180 ++++++++++++++++++++- src/rpc/methods/chain.rs | 338 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 512 insertions(+), 6 deletions(-) diff --git a/src/rpc/channel.rs b/src/rpc/channel.rs index 9a031200508..31a7d9d273d 100644 --- a/src/rpc/channel.rs +++ b/src/rpc/channel.rs @@ -484,6 +484,21 @@ mod tests { /// Buffer size of the per-call frame stream returned by `raw_json_request`. const STREAM_BUF_SIZE: usize = 256; + /// A [`Methods`] with one channel method: every subscriber gets a fresh + /// receiver from the same `events` broadcast source. + /// + /// The callback keeps only a receiver prototype — not a sender clone — so + /// the test's `events` sender stays the single sender and dropping it + /// closes the source (exercised by the close tests). + fn test_methods(events: &broadcast::Sender) -> Methods { + let mut module = RpcModule::default(); + let prototype = events.subscribe(); + module + .register_channel(TEST_METHOD, move |_params| prototype.resubscribe()) + .unwrap(); + module.into() + } + /// Subscribe with the given request id; returns the allocated channel id /// and the stream of frames sent to this "connection". /// @@ -512,6 +527,23 @@ mod tests { (channel_id, frames) } + /// Request id used for the `xrpc.cancel` calls themselves; non-null so + /// the error path's id echo is observable. + const CANCEL_REQUEST_ID: u64 = 999; + + /// Send an `xrpc.cancel` for the given original request id and return the + /// raw response. + async fn cancel(methods: &Methods, target_request_id: u64) -> Value { + let request = format!( + r#"{{"jsonrpc":"2.0","id":{CANCEL_REQUEST_ID},"method":"{CANCEL_METHOD_NAME}","params":[{target_request_id}]}}"# + ); + let (response, _) = methods + .raw_json_request(&request, STREAM_BUF_SIZE) + .await + .unwrap(); + serde_json::from_str(response.get()).unwrap() + } + async fn next_frame(frames: &mut mpsc::Receiver>) -> Value { let frame = tokio::time::timeout(RECV_TIMEOUT, frames.recv()) .await @@ -520,19 +552,157 @@ mod tests { serde_json::from_str(frame.get()).unwrap() } + /// Assert the stream ends without yielding another frame. This only + /// resolves once the channel's pump task has exited and dropped its sink, + /// so it doubles as a synchronization point on pump shutdown. + async fn assert_stream_closed(frames: &mut mpsc::Receiver>) { + let frame = tokio::time::timeout(RECV_TIMEOUT, frames.recv()) + .await + .expect("timed out waiting for the stream to close"); + assert!( + frame.is_none(), + "expected the stream to close, got frame: {}", + frame.unwrap().get() + ); + } + + fn val_frame(channel_id: ChannelId, payload: &str) -> Value { + json!({"jsonrpc": "2.0", "method": NOTIF_METHOD_NAME, "params": [channel_id, payload]}) + } + fn close_frame(channel_id: ChannelId) -> Value { json!({"jsonrpc": "2.0", "method": "xrpc.ch.close", "params": [channel_id]}) } + /// The response shape `xrpc.cancel` currently produces: an `id:null` + /// response wrapping the close notification (see #4453). + fn close_response(channel_id: ChannelId) -> Value { + json!({"jsonrpc": "2.0", "id": null, "result": close_frame(channel_id)}) + } + + #[tokio::test] + async fn subscribe_returns_u64_channel_id() { + let (events, _) = broadcast::channel::(SOURCE_CAPACITY); + let methods = test_methods(&events); + + let (first_channel, _first_frames) = subscribe(&methods, 1).await; + let (second_channel, _second_frames) = subscribe(&methods, 2).await; + + assert_eq!(second_channel, first_channel + 1); + } + + #[tokio::test] + async fn value_framing_positional() { + let (events, _) = broadcast::channel(SOURCE_CAPACITY); + let methods = test_methods(&events); + let (channel_id, mut frames) = subscribe(&methods, 1).await; + + events.send("head-change".into()).unwrap(); + drop(events); + + // Exactly one `xrpc.ch.val` frame with positional params + // `[channelId, payload]`, then the close from the dropped source — + // proving the send produced no extra frames. + assert_eq!( + next_frame(&mut frames).await, + val_frame(channel_id, "head-change") + ); + assert_eq!(next_frame(&mut frames).await, close_frame(channel_id)); + } + + #[tokio::test] + async fn two_channels_one_conn_independent() { + let (events, _) = broadcast::channel(SOURCE_CAPACITY); + let methods = test_methods(&events); + let (first_channel, mut first_frames) = subscribe(&methods, 1).await; + let (second_channel, mut second_frames) = subscribe(&methods, 2).await; + assert_ne!(first_channel, second_channel); + + // both channels deliver the same event + events.send("both".into()).unwrap(); + assert_eq!( + next_frame(&mut first_frames).await, + val_frame(first_channel, "both") + ); + assert_eq!( + next_frame(&mut second_frames).await, + val_frame(second_channel, "both") + ); + + // cancelling #1 closes only #1 (wait for its pump to exit before the + // next send, so the event cannot race the pump shutdown) + assert_eq!(cancel(&methods, 1).await, close_response(first_channel)); + assert_stream_closed(&mut first_frames).await; + + // ... while #2 still delivers + events.send("second-only".into()).unwrap(); + assert_eq!( + next_frame(&mut second_frames).await, + val_frame(second_channel, "second-only") + ); + } + + #[tokio::test] + async fn hundred_channel_fanout() { + let (events, _) = broadcast::channel(SOURCE_CAPACITY); + let methods = test_methods(&events); + + let mut channels = Vec::new(); + for request_id in 1..=100 { + channels.push(subscribe(&methods, request_id).await); + } + + events.send("fan-out".into()).unwrap(); + + let mut seen = ahash::HashSet::default(); + for (channel_id, frames) in &mut channels { + assert_eq!(next_frame(frames).await, val_frame(*channel_id, "fan-out")); + assert!(seen.insert(*channel_id), "channel ids must be unique"); + } + } + + #[tokio::test] + async fn cancel_unknown_id_errors() { + let (events, _) = broadcast::channel(SOURCE_CAPACITY); + let methods = test_methods(&events); + let (channel_id, mut frames) = subscribe(&methods, 1).await; + + let response = cancel(&methods, 99).await; + assert!( + response.get("error").is_some(), + "cancelling an unknown id must return an error response: {response}" + ); + assert!(response.get("result").is_none()); + // the error path echoes the cancel request's own id + assert_eq!(response.get("id"), Some(&json!(CANCEL_REQUEST_ID))); + + // the live channel is unaffected + events.send("still-open".into()).unwrap(); + assert_eq!( + next_frame(&mut frames).await, + val_frame(channel_id, "still-open") + ); + } + + /// When the event source closes, the client gets a bare `xrpc.ch.close` + /// notification. + #[tokio::test] + async fn source_closed_sends_bare_close() { + let (events, _) = broadcast::channel::(SOURCE_CAPACITY); + let methods = test_methods(&events); + let (channel_id, mut frames) = subscribe(&methods, 1).await; + + drop(events); + + assert_eq!(next_frame(&mut frames).await, close_frame(channel_id)); + } + /// Regression test: a subscriber that falls behind the broadcast source /// has its channel closed — like Lotus, so the client knows to - /// resubscribe and resync — instead of silently losing the overflowed + /// resubscribe and re-sync — instead of silently losing the overflowed /// events while the channel stays open. #[tokio::test] async fn lagged_consumer_channel_closes() { - // The receiver handed to the pump is already lagged before the pump - // ever polls it: overflow it first, then subscribe. This makes the - // `Lagged` observation deterministic regardless of task scheduling. let (events, lagged_rx) = broadcast::channel(SOURCE_CAPACITY); for n in 0..SOURCE_CAPACITY + 2 { events.send(format!("event-{n}")).unwrap(); @@ -548,7 +718,7 @@ mod tests { let (channel_id, mut frames) = subscribe(&methods, 1).await; - // no value frames arrive — the client is told the channel is gone + // no value frames arrive, the client is told the channel is gone assert_eq!(next_frame(&mut frames).await, close_frame(channel_id)); // the pump exited and dropped its receiver — the source has no diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index dd03bf009d1..caf464a1e63 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -2078,7 +2078,7 @@ mod tests { use super::*; use crate::daemon::db_util::RangeSpec; use crate::{ - blocks::{Chain4U, RawBlockHeader, chain4u}, + blocks::{Chain4U, HeaderBuilder, RawBlockHeader, chain4u}, db::{ MemoryDB, car::{AnyCar, ManyCar}, @@ -2088,6 +2088,7 @@ mod tests { use PathChange::{Apply, Revert}; use rstest::rstest; use std::sync::Arc; + use std::time::Duration; #[rstest] #[case(Some(0), None, Some(RangeSpec::To(0)))] @@ -2421,4 +2422,339 @@ mod tests { "expected change (left) does not match actual change (right)" ) } + + const BATCH_TIMEOUT: Duration = Duration::from_secs(5); + + async fn next_batch(rx: &mut Subscriber>) -> Vec { + tokio::time::timeout(BATCH_TIMEOUT, rx.recv()) + .await + .expect("timed out waiting for a head-change batch") + .expect("chain notify channel closed") + } + + /// Open a ChainNotify subscription and consume the immediate `current` + /// event, asserting it matches the store head. + async fn open_notify(cs: &ChainStore) -> Subscriber> { + let mut rx = chain_notify_inner(cs); + assert_eq!( + next_batch(&mut rx).await, + vec![ApiHeadChange { + change: "current".into(), + tipset: cs.heaviest_tipset(), + }] + ); + rx + } + + fn applied(ts: impl MakeTipset) -> ApiHeadChange { + ApiHeadChange { + change: "apply".into(), + tipset: ts.make_tipset(), + } + } + + fn reverted(ts: impl MakeTipset) -> ApiHeadChange { + ApiHeadChange { + change: "revert".into(), + tipset: ts.make_tipset(), + } + } + + #[tokio::test] + async fn current_first_matches_store_head() { + let cs = ChainStore::calibnet(); + + let mut rx = chain_notify_inner(&cs); + + assert_eq!( + next_batch(&mut rx).await, + vec![ApiHeadChange { + change: "current".into(), + tipset: cs.heaviest_tipset(), + }] + ); + } + + #[tokio::test] + async fn linear_applies_chain() { + let cs = ChainStore::calibnet(); + let db = Chain4U::with_blockstore(cs.db_owned()); + chain4u! { + in db; + [_genesis = cs.genesis_block_header()] + -> [a] -> [b] -> [c] -> [d] + }; + + let mut rx = open_notify(&cs).await; + + cs.set_heaviest_tipset(a.make_tipset()).unwrap(); + cs.set_heaviest_tipset(b.make_tipset()).unwrap(); + cs.set_heaviest_tipset(c.make_tipset()).unwrap(); + cs.set_heaviest_tipset(d.make_tipset()).unwrap(); + + // one single-apply batch per head move, in chain order; equality with + // the constructed tipsets pins increasing heights and parent linkage + assert_eq!(next_batch(&mut rx).await, vec![applied(a)]); + assert_eq!(next_batch(&mut rx).await, vec![applied(b)]); + assert_eq!(next_batch(&mut rx).await, vec![applied(c)]); + assert_eq!(next_batch(&mut rx).await, vec![applied(d)]); + } + + #[tokio::test] + async fn fork_switch_one_batch_reverts_then_applies() { + let cs = ChainStore::calibnet(); + let db = Chain4U::with_blockstore(cs.db_owned()); + chain4u! { + in db; + [_genesis = cs.genesis_block_header()] + -> [a] -> [b] -> [c] + }; + chain4u! { + from [a] in db; + [b2] -> [c2] -> [d2] + }; + + let mut rx = open_notify(&cs).await; + + cs.set_heaviest_tipset(a.make_tipset()).unwrap(); + assert_eq!(next_batch(&mut rx).await, vec![applied(a)]); + + // advancing over several epochs delivers one multi-apply batch + cs.set_heaviest_tipset(c.make_tipset()).unwrap(); + assert_eq!(next_batch(&mut rx).await, vec![applied(b), applied(c)]); + + // switching to the longer fork delivers a single batch: reverts + // newest-first down to the common ancestor, then applies oldest-first + cs.set_heaviest_tipset(d2.make_tipset()).unwrap(); + assert_eq!( + next_batch(&mut rx).await, + vec![ + reverted(c), + reverted(b), + applied(b2), + applied(c2), + applied(d2) + ] + ); + } + + #[tokio::test] + async fn rewind_pure_revert_batch() { + let cs = ChainStore::calibnet(); + let db = Chain4U::with_blockstore(cs.db_owned()); + chain4u! { + in db; + [_genesis = cs.genesis_block_header()] + -> [a] -> [b] -> [c] -> [d] + }; + + let mut rx = open_notify(&cs).await; + + cs.set_heaviest_tipset(a.make_tipset()).unwrap(); + assert_eq!(next_batch(&mut rx).await, vec![applied(a)]); + + cs.set_heaviest_tipset(d.make_tipset()).unwrap(); + assert_eq!( + next_batch(&mut rx).await, + vec![applied(b), applied(c), applied(d)] + ); + + // rewinding head to an ancestor (what the admin `ChainSetHead` does) + // delivers one pure-revert batch, newest-first, with no applies + cs.set_heaviest_tipset(a.make_tipset()).unwrap(); + assert_eq!( + next_batch(&mut rx).await, + vec![reverted(d), reverted(c), reverted(b)] + ); + } + + #[tokio::test] + async fn null_round_height_gaps() { + let cs = ChainStore::calibnet(); + let db = Chain4U::with_blockstore(cs.db_owned()); + chain4u! { + in db; + [_genesis = cs.genesis_block_header()] + -> [a] -> [b] + -> [c = HeaderBuilder::new().with_epoch(5)] // epochs 3 and 4 are null rounds + -> [d] + }; + assert_eq!(c.epoch, 5); + assert_eq!(c.parents, b.make_tipset().key().clone()); + + let mut rx = open_notify(&cs).await; + + cs.set_heaviest_tipset(a.make_tipset()).unwrap(); + cs.set_heaviest_tipset(b.make_tipset()).unwrap(); + cs.set_heaviest_tipset(c.make_tipset()).unwrap(); + cs.set_heaviest_tipset(d.make_tipset()).unwrap(); + + // consecutive applies jump from epoch 2 to epoch 5 over the null + // rounds; no filler events are emitted for the missing epochs + assert_eq!(next_batch(&mut rx).await, vec![applied(a)]); + assert_eq!(next_batch(&mut rx).await, vec![applied(b)]); + assert_eq!(next_batch(&mut rx).await, vec![applied(c)]); + assert_eq!(next_batch(&mut rx).await, vec![applied(d)]); + } + + #[tokio::test] + async fn over_finality_fallback_single_apply() { + let cs = ChainStore::calibnet(); + let finality = cs.chain_config().policy.chain_finality; + let db = Chain4U::with_blockstore(cs.db_owned()); + chain4u! { + in db; + [_genesis = cs.genesis_block_header()] + -> [a] -> [b] -> [c] + }; + chain4u! { + from [a] in db; + [far = HeaderBuilder::new().with_epoch(finality + 4)] + }; + + let mut rx = open_notify(&cs).await; + + cs.set_heaviest_tipset(a.make_tipset()).unwrap(); + assert_eq!(next_batch(&mut rx).await, vec![applied(a)]); + + cs.set_heaviest_tipset(c.make_tipset()).unwrap(); + assert_eq!(next_batch(&mut rx).await, vec![applied(b), applied(c)]); + + // `far` forks off `a`, so a real path from `c` would start with + // reverts of `c` and `b`. But the head jump is wider than chain + // finality, so the path computation bails out and the batch degrades + // to a single apply with no reverts. + cs.set_heaviest_tipset(far.make_tipset()).unwrap(); + assert_eq!(next_batch(&mut rx).await, vec![applied(far)]); + } + + /// End-to-end weld: a real jsonrpsee server serving just the pubsub + /// module (no auth stack), a raw WebSocket client, and a real + /// `ChainStore`. A fork switch must arrive as one correctly framed + /// `xrpc.ch.val` notification carrying the revert+apply batch. + #[tokio::test] + async fn ws_weld_end_to_end() { + use crate::rpc::channel::{NOTIF_METHOD_NAME, RpcModule as FilRpcModule}; + use futures::{SinkExt, StreamExt}; + use serde_json::Value; + use tokio_tungstenite::tungstenite::Message; + + /// Assert `xrpc.ch.val` framing (positional `[channelId, payload]` + /// params) and decode the payload. + fn decode_val_frame(frame: Value) -> (u64, Vec) { + assert_eq!( + frame.get("method").and_then(Value::as_str), + Some(NOTIF_METHOD_NAME), + "not an xrpc.ch.val frame: {frame}" + ); + let params = frame + .get("params") + .and_then(Value::as_array) + .unwrap_or_else(|| panic!("params must be a positional array: {frame}")); + let [channel_id, payload] = params.as_slice() else { + panic!("params must be [channelId, payload]: {frame}"); + }; + let channel_id = channel_id.as_u64().expect("channel id must be a u64"); + let batch = serde_json::from_value(payload.clone()) + .expect("payload must decode as Vec"); + (channel_id, batch) + } + + let cs = ChainStore::calibnet(); + let db = Chain4U::with_blockstore(cs.db_owned()); + chain4u! { + in db; + [_genesis = cs.genesis_block_header()] + -> [a] -> [b] -> [c] + }; + chain4u! { + from [a] in db; + [b2] -> [c2] -> [d2] + }; + + // serve only the pubsub module (no auth stack) + let mut pubsub = FilRpcModule::default(); + pubsub + .register_channel("Filecoin.ChainNotify", { + let chain_store = cs.shallow_clone(); + move |_params| chain_notify_inner(&chain_store) + }) + .unwrap(); + let server = jsonrpsee::server::Server::builder() + .build("127.0.0.1:0") + .await + .unwrap(); + let addr = server.local_addr().unwrap(); + let _server_handle = server.start(pubsub); + + let (mut ws, _) = tokio_tungstenite::connect_async(format!("ws://{addr}")) + .await + .unwrap(); + + async fn next_ws_json( + ws: &mut ( + impl StreamExt> + + Unpin + ), + ) -> Value { + loop { + let message = tokio::time::timeout(BATCH_TIMEOUT, ws.next()) + .await + .expect("timed out waiting for a websocket frame") + .expect("websocket closed") + .unwrap(); + if message.is_text() { + return serde_json::from_str(message.into_text().unwrap().as_str()).unwrap(); + } + } + } + + // subscribe: the response carries a bare u64 channel id + ws.send(Message::text( + r#"{"jsonrpc":"2.0","id":1,"method":"Filecoin.ChainNotify","params":[]}"#, + )) + .await + .unwrap(); + let response = next_ws_json(&mut ws).await; + let channel_id = response + .get("result") + .and_then(Value::as_u64) + .unwrap_or_else(|| panic!("channel id must be a bare u64: {response}")); + + // the first frame is the `current` event for the store head + let (frame_channel, batch) = decode_val_frame(next_ws_json(&mut ws).await); + assert_eq!(frame_channel, channel_id); + assert_eq!( + batch, + vec![ApiHeadChange { + change: "current".into(), + tipset: cs.heaviest_tipset(), + }] + ); + + cs.set_heaviest_tipset(a.make_tipset()).unwrap(); + let (frame_channel, batch) = decode_val_frame(next_ws_json(&mut ws).await); + assert_eq!(frame_channel, channel_id); + assert_eq!(batch, vec![applied(a)]); + + cs.set_heaviest_tipset(c.make_tipset()).unwrap(); + let (frame_channel, batch) = decode_val_frame(next_ws_json(&mut ws).await); + assert_eq!(frame_channel, channel_id); + assert_eq!(batch, vec![applied(b), applied(c)]); + + // fork switch: one frame carrying reverts-then-applies + cs.set_heaviest_tipset(d2.make_tipset()).unwrap(); + let (frame_channel, batch) = decode_val_frame(next_ws_json(&mut ws).await); + assert_eq!(frame_channel, channel_id); + assert_eq!( + batch, + vec![ + reverted(c), + reverted(b), + applied(b2), + applied(c2), + applied(d2) + ] + ); + } } From 789e89f50b0bab69a0edb937352566cd8d0250aa Mon Sep 17 00:00:00 2001 From: Aryan Tikarya Date: Tue, 18 Aug 2026 19:19:01 +0530 Subject: [PATCH 3/3] fix(rpc): drop channel registry entry when the pump exits [skip ci] --- CHANGELOG.md | 2 ++ src/rpc/channel.rs | 50 ++++++++++++++++++++++++++++++++++++---------- 2 files changed, 41 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 57c400c28ab..1ae88ae4a0b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,6 +49,8 @@ - [#5795](https://github.com/ChainSafe/forest/issues/5795): `Filecoin.ChainNotify` now closes the subscription channel when a client falls too far behind instead of silently dropping head changes, matching Lotus, so clients can detect the gap and resubscribe. +- [#5795](https://github.com/ChainSafe/forest/issues/5795): RPC channel subscriptions no longer leak their server-side registry entry when the channel closes without an explicit cancel (source closed, lagged consumer, or client disconnect). + - [#7473](https://github.com/ChainSafe/forest/pull/7473): Fixed `eth_estimateGas` under-estimating gas for nested contract calls (EIP-150's 63/64 rule), which could make transactions fail on chain with `SYS_OUT_OF_GAS`; the estimate is now raised until the message succeeds. Genuine reverts return an `execution reverted` error (JSON-RPC code `3`) with the decoded reason and data, matching Lotus. - [#7412](https://github.com/ChainSafe/forest/issues/7412): Fixes quicknet "unchained" logic to fetch the `max_beacon_round` for all covered epochs diff --git a/src/rpc/channel.rs b/src/rpc/channel.rs index 31a7d9d273d..8de51c6ed39 100644 --- a/src/rpc/channel.rs +++ b/src/rpc/channel.rs @@ -144,7 +144,7 @@ impl PendingSubscriptionSink { if success { let (tx, rx) = mpsc::channel(1); self.subscribers.lock().insert( - (self.connection_id, id), + (self.connection_id, id.clone()), (self.inner.clone(), rx, self.channel_id), ); tracing::debug!( @@ -157,6 +157,9 @@ impl PendingSubscriptionSink { method: self.method, unsubscribe: IsUnsubscribed(tx), channel_id: self.channel_id, + subscribers: self.subscribers, + connection_id: self.connection_id, + id, }) } else { panic!( @@ -193,6 +196,12 @@ pub struct SubscriptionSink { unsubscribe: IsUnsubscribed, /// Channel identifier. channel_id: ChannelId, + /// Shared subscriber registry, for cleanup when the pump exits. + subscribers: Subscribers, + /// Connection identifier (registry key). + connection_id: ConnectionId, + /// ID of the subscription call (registry key). + id: Id<'static>, } impl SubscriptionSink { @@ -238,6 +247,19 @@ impl SubscriptionSink { _ = self.unsubscribe.unsubscribed() => (), } } + + /// Removes this channel's subscriber-registry entry, unless the entry + /// already belongs to a newer channel that reused the same request id. + fn unregister(&self) { + let key = (self.connection_id, self.id.clone()); + let mut subscribers = self.subscribers.lock(); + if subscribers + .get(&key) + .is_some_and(|(_, _, channel_id)| *channel_id == self.channel_id) + { + subscribers.remove(&key); + } + } } fn create_notif_message( @@ -414,6 +436,9 @@ impl RpcModule { } } + // Every exit path drops the registry entry (a no-op when + // cancel already removed it). + sink.unregister(); tracing::debug!("Send notification task ended (chann_id={})", sink.channel_id); }); } @@ -685,7 +710,9 @@ mod tests { } /// When the event source closes, the client gets a bare `xrpc.ch.close` - /// notification. + /// notification, the stream ends, and the registry entry is gone — a + /// later cancel finds nothing instead of "closing" the dead channel a + /// second time. #[tokio::test] async fn source_closed_sends_bare_close() { let (events, _) = broadcast::channel::(SOURCE_CAPACITY); @@ -695,6 +722,13 @@ mod tests { drop(events); assert_eq!(next_frame(&mut frames).await, close_frame(channel_id)); + assert_stream_closed(&mut frames).await; + + let response = cancel(&methods, 1).await; + assert!( + response.get("error").is_some(), + "cancel after the channel closed must fail: {response}" + ); } /// Regression test: a subscriber that falls behind the broadcast source @@ -721,15 +755,9 @@ mod tests { // no value frames arrive, the client is told the channel is gone assert_eq!(next_frame(&mut frames).await, close_frame(channel_id)); - // the pump exited and dropped its receiver — the source has no - // subscribers left, and no stray frames follow the close (the frame - // stream stays open because the pump's registry entry is not yet - // cleaned up on exit) + // the pump exited: it dropped its receiver (the source has no + // subscribers left) and its registry entry, so the stream ends assert!(events.send("after-close".into()).is_err()); - tokio::task::yield_now().await; - assert!(matches!( - frames.try_recv(), - Err(mpsc::error::TryRecvError::Empty) - )); + assert_stream_closed(&mut frames).await; } }