From 79887efb982ef947bfab4f2285f0cd364d3a7b22 Mon Sep 17 00:00:00 2001 From: Aryan Tikarya Date: Mon, 17 Aug 2026 14:19:41 +0530 Subject: [PATCH 1/4] 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 c056ead5d7c..ad56a0400c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,6 +49,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 272f455d22680439c5504402c99dfc21861842ae Mon Sep 17 00:00:00 2001 From: Aryan Tikarya Date: Mon, 17 Aug 2026 14:22:28 +0530 Subject: [PATCH 2/4] 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 9d33aa8cee3..a777d38aa79 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -2087,7 +2087,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}, @@ -2097,6 +2097,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)))] @@ -2430,4 +2431,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 75201ece6be1db81396ca96212d86aa76337446f Mon Sep 17 00:00:00 2001 From: Aryan Tikarya Date: Tue, 18 Aug 2026 21:58:45 +0530 Subject: [PATCH 3/4] fix merge linter issue --- src/rpc/methods/chain.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index a777d38aa79..ea78d16a917 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -2448,7 +2448,7 @@ mod tests { assert_eq!( next_batch(&mut rx).await, vec![ApiHeadChange { - change: "current".into(), + change: HeadChangeType::Current, tipset: cs.heaviest_tipset(), }] ); @@ -2457,14 +2457,14 @@ mod tests { fn applied(ts: impl MakeTipset) -> ApiHeadChange { ApiHeadChange { - change: "apply".into(), + change: HeadChangeType::Apply, tipset: ts.make_tipset(), } } fn reverted(ts: impl MakeTipset) -> ApiHeadChange { ApiHeadChange { - change: "revert".into(), + change: HeadChangeType::Revert, tipset: ts.make_tipset(), } } @@ -2478,7 +2478,7 @@ mod tests { assert_eq!( next_batch(&mut rx).await, vec![ApiHeadChange { - change: "current".into(), + change: HeadChangeType::Current, tipset: cs.heaviest_tipset(), }] ); @@ -2736,7 +2736,7 @@ mod tests { assert_eq!( batch, vec![ApiHeadChange { - change: "current".into(), + change: HeadChangeType::Current, tipset: cs.heaviest_tipset(), }] ); From efe5fa55bf3b35db022796e83a6938cedc5b7913 Mon Sep 17 00:00:00 2001 From: Aryan Tikarya Date: Thu, 20 Aug 2026 18:01:26 +0530 Subject: [PATCH 4/4] test(rpc): unify subscription test timeouts --- src/rpc/channel.rs | 4 +++- src/rpc/methods/chain.rs | 22 +++++++++------------- 2 files changed, 12 insertions(+), 14 deletions(-) diff --git a/src/rpc/channel.rs b/src/rpc/channel.rs index 31a7d9d273d..b370888e932 100644 --- a/src/rpc/channel.rs +++ b/src/rpc/channel.rs @@ -477,7 +477,9 @@ mod tests { use tokio::sync::broadcast; const TEST_METHOD: &str = "test.channel"; - const RECV_TIMEOUT: Duration = Duration::from_secs(5); + /// Upper bound on waiting for one frame: a passing test never waits it + /// out (frames are already queued), it only caps failure time. + const RECV_TIMEOUT: Duration = Duration::from_secs(1); /// Capacity of the per-test event source; the lag test overflows it to /// force a `Lagged` observation. const SOURCE_CAPACITY: usize = 4; diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index ea78d16a917..13be62bccc9 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -2300,10 +2300,7 @@ mod tests { let mut rx = chain_notify_inner(&cs); // First message is the current head. - let first = tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv()) - .await - .unwrap() - .unwrap(); + let first = next_batch(&mut rx).await; assert_eq!(first.len(), 1); assert_eq!(first[0].change, HeadChangeType::Current); @@ -2314,14 +2311,13 @@ mod tests { // Every applied tipset must be delivered, in order, with none dropped. let mut applied = vec![]; for _ in 0..3 { - match tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()).await { - Ok(Ok(msg)) => applied.extend( - msg.into_iter() - .filter(|c| c.change == HeadChangeType::Apply) - .map(|c| c.tipset), - ), - _ => break, - } + applied.extend( + next_batch(&mut rx) + .await + .into_iter() + .filter(|c| c.change == HeadChangeType::Apply) + .map(|c| c.tipset), + ); } assert_eq!( applied, @@ -2432,7 +2428,7 @@ mod tests { ) } - const BATCH_TIMEOUT: Duration = Duration::from_secs(5); + const BATCH_TIMEOUT: Duration = Duration::from_secs(1); async fn next_batch(rx: &mut Subscriber>) -> Vec { tokio::time::timeout(BATCH_TIMEOUT, rx.recv())