WEB WS Downlink correctness + peer-lease fixed

Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>

src/web/http/websocket/driver.rs

@@ -134,7 +134,7 @@ async fn run_multiplex(
     let maximum_message = session.limits().carrier_batch_bytes;
     let mut active = false;
     loop {
-        let down = session.poll_down(cursor);
+        let down = session.poll_down_websocket(cursor);
         tokio::pin!(down);
         let event = tokio::select! {
             _ = cancellation.cancelled() => return Err(()),

src/web/session/downlink.rs

@@ -15,6 +15,14 @@ use crate::web::telemetry::WebSessionLifecycleObservation;
 impl WebSession {
     /// Polls pending downlink frames with cursor replay and newest-poll-wins semantics.
     pub(crate) async fn poll_down(&self, cursor: u64) -> Result<PollResult, ManagerError> {
+        self.poll_down_inner(cursor, true).await
+    }
+
+    async fn poll_down_inner(
+        &self,
+        cursor: u64,
+        peer_activity: bool,
+    ) -> Result<PollResult, ManagerError> {
         if !self.carrier().is_multiplexed() {
             return Err(ManagerError::Protocol);
         }
@@ -30,11 +38,13 @@ impl WebSession {
                         next_cursor: unacked.next_cursor,
                         lane_closed: false,
                     };
-                    self.touch_peer_locked(
+                    if peer_activity {
-                        &mut state,
+                        self.touch_peer_locked(
-                        Instant::now(),
+                            &mut state,
-                        WebSessionLifecycleObservation::HttpActivityAfterGap,
+                            Instant::now(),
-                    );
+                            WebSessionLifecycleObservation::HttpActivityAfterGap,
+                        );
+                    }
                     return Ok(result);
                 }
                 if cursor != unacked.next_cursor {
@@ -59,11 +69,13 @@ impl WebSession {
                 return Err(ManagerError::Protocol);
             };
             state.down_epoch = epoch;
-            self.touch_peer_locked(
+            if peer_activity {
-                &mut state,
+                self.touch_peer_locked(
-                Instant::now(),
+                    &mut state,
-                WebSessionLifecycleObservation::HttpActivityAfterGap,
+                    Instant::now(),
-            );
+                    WebSessionLifecycleObservation::HttpActivityAfterGap,
+                );
+            }
             let healthy = self.carrier_health_ready_locked(&mut state, Instant::now());
             (state.down_epoch, healthy)
         };
@@ -133,6 +145,14 @@ impl WebSession {
         }
     }
 
+    /// Polls multiplexed WebSocket downlink without renewing the peer lease.
+    pub(crate) async fn poll_down_websocket(
+        &self,
+        cursor: u64,
+    ) -> Result<PollResult, ManagerError> {
+        self.poll_down_inner(cursor, false).await
+    }
+
     /// Reserves session and process queue capacity while the session lock is held.
     pub(super) fn reserve_locked(
         &self,

src/web/session/downlink_tests.rs

@@ -142,3 +142,19 @@ async fn newer_poll_supersedes_older_poll_without_closing_session() {
     session.close(super::SessionCloseReason::ApiClose);
     manager.shutdown().await;
 }
+
+#[tokio::test]
+async fn websocket_downlink_poll_does_not_extend_the_peer_lease() {
+    let (session, manager) = session();
+    session
+        .state
+        .lock()
+        .activity
+        .touch_peer(Instant::now() - Duration::from_secs(121));
+    queue_close(&session);
+
+    session.poll_down_websocket(0).await.unwrap();
+
+    assert!(session.close_if_due(Instant::now()));
+    manager.shutdown().await;
+}