Rust: preserve resume bootstrap events until subscription handoff

rust/src/session.rs

@@ -328,6 +328,9 @@ pub struct Session {
     open_canvases: Arc<parking_lot::RwLock<Vec<OpenCanvasInstance>>>,
     /// Broadcast channel for runtime event subscribers — see [`Session::subscribe`].
     event_tx: tokio::sync::broadcast::Sender<SessionEvent>,
+    /// Resume-only queue that retains routed events until the first
+    /// post-resume subscriber catches up and activates live broadcast delivery.
+    resume_bootstrap: Option<Arc<crate::subscription::ResumeBootstrap>>,
     github_token_registration:
         ParkingLotMutex<Option<crate::github_token::GitHubTokenRegistration>>,
     /// Identity of this session's router registration.
@@ -419,8 +422,27 @@ impl Session {
     /// loop or any combinator from `tokio_stream::StreamExt` /
     /// `futures::StreamExt`.
     ///
-    /// Each subscriber maintains its own queue. If a consumer cannot keep
-    /// up, the oldest events are dropped and `recv` returns
+    /// On a session returned by [`Client::resume_session`], the first
+    /// subscription also receives every routed event retained while resume
+    /// startup had no active [`PreparedSession`] subscriber. That bootstrap
+    /// prefix is lossless and ordered before live events. It is a one-shot
+    /// handoff: later subscriptions observe newly dispatched events even while
+    /// the first subscriber is catching up, and dropping
+    /// the first subscription before draining it discards its remaining
+    /// bootstrap events.
+    ///
+    /// Bootstrap retention is unbounded until the first subscriber catches
+    /// up: callers that never subscribe or cannot catch up can retain an
+    /// arbitrarily large backlog. Resume consumers should subscribe and drain
+    /// promptly. Ownership is assigned by the first `subscribe()` call, not
+    /// by the first poll. Stopping
+    /// the event loop releases an unclaimed backlog; dropping a claimed
+    /// subscription releases its unread backlog.
+    /// This retention covers events routed to this session, not events lost
+    /// to overflow in the client-global notification router.
+    /// After the bootstrap handoff, each subscriber maintains its own finite
+    /// queue. If a consumer cannot keep
+    /// up, the oldest live events are dropped and `recv` returns
     /// [`RecvErrorKind::Lagged`](crate::subscription::RecvErrorKind::Lagged)
     /// reporting the count of skipped events. Slow consumers do not block
     /// the session's event loop.
@@ -438,7 +460,10 @@ impl Session {
     /// # }
     /// ```
     pub fn subscribe(&self) -> crate::subscription::EventSubscription {
-        crate::subscription::EventSubscription::new(self.event_tx.subscribe())
+        match &self.resume_bootstrap {
+            Some(bootstrap) => bootstrap.subscribe(&self.event_tx),
+            None => crate::subscription::EventSubscription::new(self.event_tx.subscribe()),
+        }
     }
 
     /// The underlying Client (for advanced use cases).
@@ -1155,11 +1180,16 @@ impl Client {
     ///
     /// # Event delivery
     ///
-    /// Equivalent to `prepare_resume_session(config)?.start().await`, and
-    /// carries the same startup-event caveat documented on
-    /// [`create_session`](Self::create_session). Use
-    /// [`prepare_resume_session`](Self::prepare_resume_session) when
-    /// startup events matter.
+    /// When no active [`PreparedSession`] subscription exists at startup,
+    /// routed events emitted during resume are retained in an ordered,
+    /// unbounded bootstrap queue. The first
+    /// [`Session::subscribe`] call receives that complete prefix before
+    /// switching atomically to normal live delivery. Later subscribers are
+    /// live-only.
+    ///
+    /// Use [`prepare_resume_session`](Self::prepare_resume_session) when the
+    /// event consumer must be installed before protocol activity begins or
+    /// when multiple startup observers are required.
     pub async fn resume_session(&self, config: ResumeSessionConfig) -> Result<Session, Error> {
         self.prepare_resume_session(config)?.start().await
     }
@@ -1431,6 +1461,7 @@ impl Client {
             capabilities.clone(),
             open_canvases.clone(),
             event_tx.clone(),
+            None,
             shutdown.clone(),
             external_tools_shutdown.clone(),
         );
@@ -1469,6 +1500,7 @@ impl Client {
             capabilities,
             open_canvases,
             event_tx,
+            resume_bootstrap: None,
             github_token_registration: ParkingLotMutex::new(github_token_registration),
             registration_token,
         };
@@ -1627,6 +1659,11 @@ impl Client {
 
         let capabilities = Arc::new(parking_lot::RwLock::new(SessionCapabilities::default()));
         let setup_start = Instant::now();
+        // An active prepared subscription already owns startup delivery. The
+        // implicit queue is only needed by the compatibility resume wrapper,
+        // where Session::subscribe cannot be called until startup returns.
+        let resume_bootstrap = (event_tx.receiver_count() == 0)
+            .then(|| crate::subscription::ResumeBootstrap::new(&event_tx));
         let registration = self.register_session(&session_id);
         let registration_token = registration.token;
         let channels = registration.channels;
@@ -1648,6 +1685,7 @@ impl Client {
             capabilities.clone(),
             open_canvases.clone(),
             event_tx.clone(),
+            resume_bootstrap.clone(),
             shutdown.clone(),
             external_tools_shutdown.clone(),
         );
@@ -1760,6 +1798,7 @@ impl Client {
             capabilities,
             open_canvases,
             event_tx,
+            resume_bootstrap,
             github_token_registration: ParkingLotMutex::new(github_token_registration),
             registration_token,
         };
@@ -1818,16 +1857,20 @@ impl Client {
 ///
 /// # Buffering
 ///
-/// The broadcast buffer is finite —
-/// [`DEFAULT_EVENT_BUFFER_CAPACITY`] unless
+/// Subscriptions taken directly from this prepared handle use the finite
+/// broadcast buffer — [`DEFAULT_EVENT_BUFFER_CAPACITY`] unless
 /// [`SessionConfig::event_buffer_capacity`] /
-/// [`ResumeSessionConfig::event_buffer_capacity`] overrides it. Subscribers
-/// that fall behind observe
+/// [`ResumeSessionConfig::event_buffer_capacity`] overrides it. Those
+/// subscribers that fall behind observe
 /// [`Lagged`](crate::subscription::Lagged) instead of applying backpressure
 /// to the event loop. Consumers that need a lossless view of a large
 /// startup burst must either configure a capacity that covers it or drain
 /// the subscription concurrently with [`start`](Self::start).
 ///
+/// If a resume starts with no active prepared subscription, the eventual
+/// [`Session`] instead retains routed startup events in the resume bootstrap
+/// queue documented on [`Session::subscribe`].
+///
 /// # Server-assigned session IDs
 ///
 /// For cloud sessions without a caller-supplied session ID, the CLI assigns
@@ -2056,6 +2099,7 @@ fn spawn_event_loop(
     capabilities: Arc<parking_lot::RwLock<SessionCapabilities>>,
     open_canvases: Arc<parking_lot::RwLock<Vec<OpenCanvasInstance>>>,
     event_tx: tokio::sync::broadcast::Sender<SessionEvent>,
+    resume_bootstrap: Option<Arc<crate::subscription::ResumeBootstrap>>,
     shutdown: CancellationToken,
     external_tools_shutdown: CancellationToken,
 ) -> JoinHandle<()> {
@@ -2097,7 +2141,7 @@ fn spawn_event_loop(
                     _ = shutdown.cancelled() => break,
                     Some(notification) = notifications.recv() => {
                         handle_notification(
-                            &session_id, &client, &handlers, &command_handlers, notification, &idle_waiter, &capabilities, &open_canvases, &event_tx, &shutdown, &external_tools_shutdown, &pending_external_tools,
+                            &session_id, &client, &handlers, &command_handlers, notification, &idle_waiter, &capabilities, &open_canvases, &event_tx, resume_bootstrap.as_ref(), &shutdown, &external_tools_shutdown, &pending_external_tools,
                         ).await;
                     }
                     Some(request) = requests.recv() => {
@@ -2146,6 +2190,9 @@ fn spawn_event_loop(
                     else => break,
                 }
             }
+            if let Some(bootstrap) = &resume_bootstrap {
+                bootstrap.release_unclaimed();
+            }
             // Channels closed or shutdown signaled — fail any pending
             // send_and_wait so the caller observes a clean error.
             if let Some(waiter) = idle_waiter.lock().take() {
@@ -2263,6 +2310,7 @@ async fn handle_notification(
     capabilities: &Arc<parking_lot::RwLock<SessionCapabilities>>,
     open_canvases: &Arc<parking_lot::RwLock<Vec<OpenCanvasInstance>>>,
     event_tx: &tokio::sync::broadcast::Sender<SessionEvent>,
+    resume_bootstrap: Option<&Arc<crate::subscription::ResumeBootstrap>>,
     shutdown: &CancellationToken,
     external_tools_shutdown: &CancellationToken,
     pending_external_tools: &PendingExternalTools,
@@ -2364,10 +2412,14 @@ async fn handle_notification(
         }
     }
 
-    // Fan out the event to runtime subscribers (`Session::subscribe`). `send`
-    // only errors when there are no receivers, which is the normal case
-    // before any consumer subscribes.
-    let _ = event_tx.send(event.clone());
+    // Resume startup queues routed events until the first post-resume
+    // subscriber catches up. All other paths retain the existing bounded
+    // broadcast behavior.
+    if let Some(bootstrap) = resume_bootstrap {
+        bootstrap.publish(event_tx, event.clone());
+    } else {
+        let _ = event_tx.send(event.clone());
+    }
 
     tracing::debug!(
         elapsed_ms = dispatch_start.elapsed().as_millis(),