WEB Bounded Bridge Recovery added

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

src/web/session/backend_tests.rs

@@ -131,6 +131,7 @@ fn test_runtime_with_dc(
         crate::web::manager::CarrierClientClass::Legacy,
         None,
         false,
+        false,
         limits,
         timeouts,
     );

src/web/session/downlink_tests.rs

@@ -47,6 +47,7 @@ fn session() -> (Arc<WebSession>, Arc<WebProcessRuntime>) {
         crate::web::manager::CarrierClientClass::Legacy,
         None,
         false,
+        false,
         WebLimitsConfig::default(),
         timeouts,
     );

src/web/session/lanes.rs

@@ -6,8 +6,8 @@ use tokio::sync::OwnedSemaphorePermit;
 
 use super::lane_downlink::take_lane_down_batch;
 use super::{
-    CarrierLaneIdentity, PendingClass, PollResult, QUEUE_ITEM_COST, QueuedFrame, SessionCloseReason,
-    SessionState, WebSession, remember_closed,
+    CarrierLaneIdentity, PendingClass, PollResult, QUEUE_ITEM_COST, QueuedFrame,
+    SessionCloseReason, SessionState, WebSession, remember_closed,
 };
 use crate::web::frame::{self, FrameType};
 use crate::web::manager::ManagerError;

src/web/session/lanes/tests.rs

@@ -62,6 +62,7 @@ fn new_session_with_automatic(
         },
         None,
         automatic,
+        false,
         limits,
         WebTimeoutsConfig::default(),
     )

src/web/session/lifecycle.rs

@@ -97,6 +97,7 @@ struct ReleasedQueues {
     control_bytes: usize,
     control_items: usize,
     closed_before_health: bool,
+    recovery_closed_before_commit: bool,
     reason: SessionCloseReason,
     peer_gap: Duration,
 }
@@ -253,6 +254,7 @@ impl WebSession {
         let closed_before_health = self.automatic_carrier
             && state.negotiation_phase == SessionNegotiationPhase::Committed
             && self.reject_carrier_health_on_close();
+        let recovery_closed_before_commit = self.recovery && !state.recovery_committed;
         state.close_requested = Some(reason);
         state.closed = true;
         if reason == SessionCloseReason::CarrierSuperseded {
@@ -305,6 +307,7 @@ impl WebSession {
             control_bytes,
             control_items,
             closed_before_health,
+            recovery_closed_before_commit,
             reason,
             peer_gap,
         }
@@ -326,6 +329,11 @@ impl WebSession {
                     crate::web::telemetry::WebCarrierLearningOutcome::ClosedBeforeHealth,
                 );
             }
+            if released.recovery_closed_before_commit {
+                manager.telemetry().record_bridge_recovery(
+                    crate::web::telemetry::WebBridgeRecoveryEvent::ClosedBeforeCommit,
+                );
+            }
             manager.release_pending(
                 self.profile_key,
                 released.data_bytes,
@@ -350,10 +358,7 @@ impl WebSession {
                     Some(released.reason.as_str()),
                     crate::web::trace::TraceLifecycleContext {
                         peer_gap_ms: Some(
-                            released
-                                .peer_gap
-                                .as_millis()
-                                .min(u128::from(u64::MAX)) as u64,
+                            released.peer_gap.as_millis().min(u128::from(u64::MAX)) as u64
                         ),
                         predecessor_session_id: None,
                     },

src/web/session/negotiation.rs

@@ -107,6 +107,14 @@ impl WebSession {
         state: &mut SessionState,
         progress: AppliedProgress,
     ) -> (bool, Option<CarrierHealthClaim>) {
+        if self.recovery && progress.any() && !state.recovery_committed {
+            state.recovery_committed = true;
+            if let Some(manager) = self.manager.upgrade() {
+                manager.telemetry().record_bridge_recovery(
+                    crate::web::telemetry::WebBridgeRecoveryEvent::Committed,
+                );
+            }
+        }
         if !self.automatic_carrier || !progress.any() {
             return (false, None);
         }
@@ -331,6 +339,7 @@ mod tests {
             CarrierClientClass::Bridge,
             None,
             true,
+            false,
             WebLimitsConfig::default(),
             WebTimeoutsConfig::default(),
         )
@@ -371,9 +380,17 @@ mod tests {
         state.carrier_health_uplink = true;
         state.carrier_health_downlink = true;
         state.carrier_health_activity_at = Some(now - Duration::from_secs(2));
-        assert!(session.carrier_health_ready_locked(&mut state, now).is_none());
+        assert!(
+            session
+                .carrier_health_ready_locked(&mut state, now)
+                .is_none()
+        );
         state.carrier_health_activity_at = Some(now);
-        assert!(session.carrier_health_ready_locked(&mut state, now).is_some());
+        assert!(
+            session
+                .carrier_health_ready_locked(&mut state, now)
+                .is_some()
+        );
     }
 
     #[test]
@@ -391,9 +408,17 @@ mod tests {
         state.websocket_commit_ack_owner = Some(7);
         state.websocket_commit_ack_written = true;
         state.carrier_health_uplink = true;
-        assert!(session.carrier_health_ready_locked(&mut state, now).is_none());
+        assert!(
+            session
+                .carrier_health_ready_locked(&mut state, now)
+                .is_none()
+        );
         state.websocket_probe_claimed = true;
-        assert!(session.carrier_health_ready_locked(&mut state, now).is_some());
+        assert!(
+            session
+                .carrier_health_ready_locked(&mut state, now)
+                .is_some()
+        );
     }
 
     #[test]
@@ -407,7 +432,11 @@ mod tests {
         state.carrier_health_downlink = true;
         state.carrier_health_activity_at = Some(now);
 
-        assert!(session.carrier_health_ready_locked(&mut state, now).is_none());
+        assert!(
+            session
+                .carrier_health_ready_locked(&mut state, now)
+                .is_none()
+        );
         assert_eq!(
             session.carrier_health_publication_state(),
             CarrierHealthPublicationState::Awaiting
@@ -421,8 +450,16 @@ mod tests {
         arm_http_health(&session, now);
         let mut state = session.state.lock();
 
-        assert!(session.carrier_health_ready_locked(&mut state, now).is_some());
-        assert!(session.carrier_health_ready_locked(&mut state, now).is_none());
+        assert!(
+            session
+                .carrier_health_ready_locked(&mut state, now)
+                .is_some()
+        );
+        assert!(
+            session
+                .carrier_health_ready_locked(&mut state, now)
+                .is_none()
+        );
         drop(state);
         assert_eq!(
             session.carrier_health_publication_state(),
@@ -469,8 +506,7 @@ mod tests {
 
             assert!(matches!(
                 session.carrier_health_publication_state(),
-                CarrierHealthPublicationState::Published
-                    | CarrierHealthPublicationState::Rejected
+                CarrierHealthPublicationState::Published | CarrierHealthPublicationState::Rejected
             ));
             assert!(!session.publish_carrier_health());
             assert_eq!(

src/web/session/state.rs

@@ -0,0 +1,44 @@
+use std::collections::VecDeque;
+
+use super::{CarrierLane, CarrierLaneIdentity, InboundChunk, QUEUE_ITEM_COST, SessionState};
+
+pub(super) fn inbound_queue_cost(queue: &VecDeque<InboundChunk>) -> (usize, usize) {
+    let bytes = queue.iter().fold(0usize, |total, chunk| {
+        total.saturating_add(chunk.bytes.len().saturating_sub(chunk.offset) + QUEUE_ITEM_COST)
+    });
+    (bytes, queue.len())
+}
+
+pub(super) fn remember_closed(
+    state: &mut SessionState,
+    stream_id: u32,
+    limit: usize,
+) -> Option<u32> {
+    if !state.closed_streams.insert(stream_id) {
+        return None;
+    }
+    state.closed_order.push_back(stream_id);
+    let mut evicted = None;
+    while state.closed_order.len() > limit {
+        if let Some(oldest) = state.closed_order.pop_front() {
+            state.closed_streams.remove(&oldest);
+            evicted = Some(oldest);
+        }
+    }
+    evicted
+}
+
+pub(super) fn insert_carrier_lane(
+    state: &mut SessionState,
+    lane_id: u32,
+) -> Option<CarrierLaneIdentity> {
+    if state.carrier_lanes.contains_key(&lane_id) {
+        return None;
+    }
+    let instance = state.next_lane_instance;
+    state.next_lane_instance = instance.checked_add(1)?;
+    state
+        .carrier_lanes
+        .insert(lane_id, CarrierLane::new(instance));
+    Some(CarrierLaneIdentity { lane_id, instance })
+}

src/web/session/uplink_tests.rs

@@ -45,6 +45,7 @@ fn session_with_automatic(automatic: bool) -> Arc<WebSession> {
         },
         None,
         automatic,
+        false,
         WebLimitsConfig::default(),
         WebTimeoutsConfig::default(),
     )

src/web/session/websocket/tests.rs

@@ -18,7 +18,8 @@ struct TestRuntime {
 
 impl TestRuntime {
     async fn shutdown(self) {
-        self.session.close(super::super::SessionCloseReason::ApiClose);
+        self.session
+            .close(super::super::SessionCloseReason::ApiClose);
         self.session.wait().await;
         self.manager.shutdown().await;
         self.generation.stop_sessions().await;
@@ -71,6 +72,7 @@ fn runtime(admission: bool) -> TestRuntime {
         crate::web::manager::CarrierClientClass::Legacy,
         None,
         false,
+        false,
         limits,
         timeouts,
     );