fix(rpc): more fixes in filecoin chain notify API

CHANGELOG.md

@@ -47,6 +47,10 @@
 
 ### 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.
+
+- [#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

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(
@@ -274,6 +296,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<AtomicU64>,
@@ -384,12 +414,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;
                                     }
                                 }
                             },
@@ -399,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);
                 });
             }
@@ -453,3 +493,271 @@ 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;
+
+    /// 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<String>) -> 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".
+    ///
+    /// 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<Box<RawValue>>) {
+        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)
+    }
+
+    /// 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<Box<RawValue>>) -> 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()
+    }
+
+    /// 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<Box<RawValue>>) {
+        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::<String>(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, 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::<String>(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));
+        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
+    /// has its channel closed — like Lotus, so the client knows to
+    /// resubscribe and re-sync — instead of silently losing the overflowed
+    /// events while the channel stays open.
+    #[tokio::test]
+    async fn lagged_consumer_channel_closes() {
+        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: 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());
+        assert_stream_closed(&mut frames).await;
+    }
+}