Carrier negotiation + WebSocket lifecycle

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

src/web/session/backend.rs

@@ -42,26 +42,13 @@ impl WebSession {
             drop(completion);
             return false;
         }
-        let Ok(connection_permit) = generation.max_connections.clone().try_acquire_owned() else {
-            manager.record_stream_rejected();
-            self.trace_lifecycle(
-                crate::web::trace::TraceLifecycleEvent::StreamRejected,
-                Some(stream.id),
-                Some("connection_limit"),
-            );
-            completion
-                .retain_rejected
-                .store(retain_reservation_on_reject, Ordering::Release);
-            drop(completion);
-            return false;
-        };
+        let connection_permits = Arc::clone(&generation.max_connections);
         let deps = generation.client_runtime_deps();
         let replay_checker = Arc::clone(&generation.replay_checker);
         let session = Arc::clone(self);
         let cancel = self.cancel.clone();
         let retain_rejected = Arc::clone(&completion.retain_rejected);
         let future = async move {
-            let _connection_permit = connection_permit;
             let _completion = completion;
             session.trace_lifecycle(
                 crate::web::trace::TraceLifecycleEvent::StreamAdmitted,
@@ -77,6 +64,7 @@ impl WebSession {
                     logical_stream,
                     deps,
                     replay_checker,
+                    connection_permits,
                     peer_port,
                 ) => {}
             }
@@ -219,6 +207,7 @@ async fn run_stream(
     stream: WebLogicalStream,
     deps: crate::proxy::authenticated::ClientRuntimeDeps,
     replay_checker: Arc<crate::stats::ReplayChecker>,
+    connection_permits: Arc<tokio::sync::Semaphore>,
     peer_port: u16,
 ) {
     use tokio::io::AsyncReadExt;
@@ -267,6 +256,15 @@ async fn run_stream(
     let Some(manager) = session.manager.upgrade() else {
         return;
     };
+    let Ok(_connection_permit) = connection_permits.try_acquire_owned() else {
+        manager.record_stream_rejected();
+        session.trace_lifecycle(
+            crate::web::trace::TraceLifecycleEvent::StreamRejected,
+            Some(stream_identity.id),
+            Some("connection_limit_after_first_byte"),
+        );
+        return;
+    };
     let Some(handshake_permit) = manager.try_stream_handshake() else {
         session.trace_lifecycle(
             crate::web::trace::TraceLifecycleEvent::StreamRejected,

src/web/session/backend_tests.rs

@@ -206,6 +206,58 @@ async fn open_without_data_does_not_start_the_inner_handshake_timeout() {
     }
 }
 
+#[tokio::test(start_paused = true)]
+async fn silent_open_does_not_consume_generation_connection_capacity() {
+    let runtime = test_runtime(WebCarrier::Https, 1);
+    let available = runtime.generation.max_connections.available_permits();
+
+    assert_eq!(runtime.process_frame(1, 1, FrameType::Open, &[]), Ok(1));
+    settle_tasks().await;
+
+    assert_eq!(
+        runtime.generation.max_connections.available_permits(),
+        available
+    );
+
+    assert_eq!(runtime.process_frame(1, 2, FrameType::Data, &[0x5a]), Ok(2));
+    settle_tasks().await;
+    assert_eq!(
+        runtime.generation.max_connections.available_permits(),
+        available - 1
+    );
+
+    runtime.shutdown().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn first_inner_byte_fails_closed_when_connection_capacity_is_exhausted() {
+    let runtime = test_runtime(WebCarrier::Https, 1);
+    let capacity = runtime.generation.max_connections.available_permits();
+    let permit = runtime
+        .generation
+        .max_connections
+        .clone()
+        .try_acquire_many_owned(capacity as u32)
+        .unwrap();
+
+    assert_eq!(runtime.process_frame(1, 1, FrameType::Open, &[]), Ok(1));
+    settle_tasks().await;
+    assert!(runtime.session.state.lock().streams.contains_key(&1));
+
+    assert_eq!(runtime.process_frame(1, 2, FrameType::Data, &[0x5a]), Ok(2));
+    settle_tasks().await;
+
+    assert!(!runtime.session.state.lock().streams.contains_key(&1));
+    assert_eq!(runtime.generation.max_connections.available_permits(), 0);
+
+    drop(permit);
+    assert_eq!(
+        runtime.generation.max_connections.available_permits(),
+        capacity
+    );
+    runtime.shutdown().await;
+}
+
 #[tokio::test(start_paused = true)]
 async fn the_first_inner_byte_starts_the_handshake_timeout() {
     let runtime = test_runtime(WebCarrier::Https, 1);

src/web/session/lifecycle.rs

@@ -206,6 +206,7 @@ impl WebSession {
                         self.client_ip,
                         self.profile_key,
                         &self.profile.host,
+                        Duration::from_secs(self.timeouts.bootstrap_lifetime_secs),
                     );
                 }
             }