Native carrier negotiation + attempt deadline mapping fixed

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

src/web/http/negotiation_tests.rs

@@ -5,6 +5,8 @@ use std::time::Duration;
 use sha2::{Digest, Sha256};
 
 const CAPABILITIES: &str = "https,https-lanes,websocket,websocket-lanes";
+const NATIVE_USER_AGENT_HEADER: &str =
+    "User-Agent: Telegram/3951 CFNetwork/3896.100.1.2.1 Darwin/27.0.0\r\n";
 
 fn issue_bootstrap(runtime: &Arc<WebProcessRuntime>, client_ip: &str) -> String {
     let profile = runtime
@@ -27,6 +29,16 @@ fn create_request(
     hello: &[u8],
     attempt: Option<u8>,
     failure: Option<&str>,
+) -> Vec<u8> {
+    create_request_with_headers(bootstrap, hello, attempt, failure, "")
+}
+
+fn create_request_with_headers(
+    bootstrap: &str,
+    hello: &[u8],
+    attempt: Option<u8>,
+    failure: Option<&str>,
+    extra_headers: &str,
 ) -> Vec<u8> {
     let negotiation = attempt.map_or_else(String::new, |attempt| {
         let failure = failure
@@ -37,7 +49,7 @@ fn create_request(
         )
     });
     let mut request = format!(
-        "POST /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {bootstrap}\r\nContent-Type: application/octet-stream\r\n{negotiation}Content-Length: {}\r\nConnection: close\r\n\r\n",
+        "POST /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {bootstrap}\r\nContent-Type: application/octet-stream\r\n{negotiation}{extra_headers}Content-Length: {}\r\nConnection: close\r\n\r\n",
         hello.len()
     )
     .into_bytes();
@@ -60,6 +72,17 @@ fn token_hash(token: &str) -> crate::web::manager::TokenHash {
     Sha256::digest(raw).into()
 }
 
+fn assert_no_negotiation_headers(headers: &[u8]) {
+    for header in [
+        "x-carrier-attempt",
+        "x-carrier-candidate-count",
+        "x-carrier-deadline",
+        "x-carrier-state",
+    ] {
+        assert!(optional_response_header(headers, header).is_none());
+    }
+}
+
 #[tokio::test]
 async fn absent_carriers_reject_negotiation_and_preserve_legacy_creation() {
     let capability = [41; 32];
@@ -94,6 +117,121 @@ async fn absent_carriers_reject_negotiation_and_preserve_legacy_creation() {
     generation.stop_background_tasks().await;
 }
 
+#[tokio::test]
+async fn metadata_free_native_client_can_use_each_fixed_carrier() {
+    for (index, carrier) in WebCarrier::ALL.into_iter().enumerate() {
+        let capability = [50 + index as u8; 32];
+        let generation = test_runtime_generation(1, runtime_config(capability, carrier));
+        let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let bootstrap = issue_bootstrap(&runtime, "192.0.2.10");
+        let hello = frame::encode(FrameType::Hello, 0, &[1]);
+
+        let response = request(
+            &listener,
+            &runtime,
+            create_request_with_headers(
+                &bootstrap,
+                &hello,
+                None,
+                None,
+                NATIVE_USER_AGENT_HEADER,
+            ),
+        )
+        .await;
+        let (headers, _) = split_response(&response);
+        assert!(headers.starts_with(b"HTTP/1.1 200"));
+        assert_eq!(response_header(headers, "x-carrier-mode"), carrier.as_str());
+        assert_no_negotiation_headers(headers);
+
+        runtime.shutdown().await;
+        generation.stop_sessions().await;
+        generation.stop_background_tasks().await;
+    }
+}
+
+#[tokio::test]
+async fn metadata_free_native_client_uses_fallback_when_candidates_are_enabled() {
+    let capability = [55; 32];
+    let generation = test_runtime_generation(
+        1,
+        negotiation_runtime_config(
+            capability,
+            WebCarrier::HttpsLanes,
+            false,
+            Arc::from([WebCarrier::Websocket, WebCarrier::HttpsLanes]),
+        ),
+    );
+    let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let bootstrap = issue_bootstrap(&runtime, "192.0.2.10");
+    let hello = frame::encode(FrameType::Hello, 0, &[1]);
+
+    let response = request(
+        &listener,
+        &runtime,
+        create_request_with_headers(
+            &bootstrap,
+            &hello,
+            None,
+            None,
+            NATIVE_USER_AGENT_HEADER,
+        ),
+    )
+    .await;
+    let (headers, _) = split_response(&response);
+    assert!(headers.starts_with(b"HTTP/1.1 200"));
+    assert_eq!(response_header(headers, "x-carrier-mode"), "https-lanes");
+    assert_no_negotiation_headers(headers);
+
+    runtime.shutdown().await;
+    generation.stop_sessions().await;
+    generation.stop_background_tasks().await;
+}
+
+#[tokio::test]
+async fn explicit_native_capabilities_participate_in_automatic_selection() {
+    let capability = [56; 32];
+    let generation = test_runtime_generation(
+        1,
+        negotiation_runtime_config(
+            capability,
+            WebCarrier::Https,
+            false,
+            Arc::from([WebCarrier::WebsocketLanes, WebCarrier::Https]),
+        ),
+    );
+    let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let bootstrap = issue_bootstrap(&runtime, "192.0.2.10");
+    let hello = frame::encode(FrameType::Hello, 0, &[1]);
+
+    let response = request(
+        &listener,
+        &runtime,
+        create_request_with_headers(
+            &bootstrap,
+            &hello,
+            Some(1),
+            None,
+            NATIVE_USER_AGENT_HEADER,
+        ),
+    )
+    .await;
+    let (headers, _) = split_response(&response);
+    assert!(headers.starts_with(b"HTTP/1.1 200"));
+    assert_eq!(
+        response_header(headers, "x-carrier-mode"),
+        "websocket-lanes"
+    );
+    assert_eq!(response_header(headers, "x-carrier-attempt"), "1");
+    assert_eq!(response_header(headers, "x-carrier-candidate-count"), "2");
+
+    runtime.shutdown().await;
+    generation.stop_sessions().await;
+    generation.stop_background_tasks().await;
+}
+
 #[tokio::test]
 async fn negotiation_replays_replaces_and_freezes_after_carrier_commit() {
     let capability = [42; 32];
@@ -225,6 +363,65 @@ async fn negotiation_replays_replaces_and_freezes_after_carrier_commit() {
     generation.stop_background_tasks().await;
 }
 
+#[tokio::test]
+async fn timed_out_attempt_replays_before_successor_own_deadline() {
+    let capability = [57; 32];
+    let generation = test_runtime_generation(
+        1,
+        negotiation_runtime_config_with_deadlines(
+            capability,
+            WebCarrier::HttpsLanes,
+            false,
+            Arc::from([WebCarrier::Https, WebCarrier::HttpsLanes]),
+            [1, 5, 8, 12],
+        ),
+    );
+    let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let bootstrap = issue_bootstrap(&runtime, "192.0.2.10");
+    let hello = frame::encode(FrameType::Hello, 0, &[1]);
+    let first_request = create_request(&bootstrap, &hello, Some(1), None);
+
+    let first = request(&listener, &runtime, first_request.clone()).await;
+    let (first_headers, _) = split_response(&first);
+    assert!(first_headers.starts_with(b"HTTP/1.1 200"));
+    assert_eq!(response_header(first_headers, "x-carrier-mode"), "https");
+    let first_token = response_header(first_headers, "x-session-token").to_string();
+
+    tokio::time::sleep(Duration::from_millis(1_100)).await;
+
+    let replay = request(&listener, &runtime, first_request).await;
+    let (replay_headers, _) = split_response(&replay);
+    assert!(replay_headers.starts_with(b"HTTP/1.1 200"));
+    assert_eq!(
+        response_header(replay_headers, "x-session-token"),
+        first_token
+    );
+    assert_eq!(response_header(replay_headers, "x-carrier-state"), "provisional");
+
+    let second = request(
+        &listener,
+        &runtime,
+        create_request(&bootstrap, &hello, Some(2), Some("timeout")),
+    )
+    .await;
+    let (second_headers, _) = split_response(&second);
+    assert!(second_headers.starts_with(b"HTTP/1.1 200"));
+    assert_eq!(
+        response_header(second_headers, "x-carrier-mode"),
+        "https-lanes"
+    );
+    assert_eq!(response_header(second_headers, "x-carrier-attempt"), "2");
+    assert_ne!(
+        response_header(second_headers, "x-session-token"),
+        first_token
+    );
+
+    runtime.shutdown().await;
+    generation.stop_sessions().await;
+    generation.stop_background_tasks().await;
+}
+
 #[tokio::test]
 async fn https_lane_downlink_can_arrive_before_its_uplink_open() {
     let capability = [43; 32];

src/web/http/request.rs

@@ -151,11 +151,7 @@ pub(super) fn carrier_request<B>(request: &Request<B>, host: &str) -> Option<Car
                 } else {
                     CarrierClientClass::Bridge
                 },
-                if native_ios {
-                    CarrierCapabilities::ios()
-                } else {
-                    capabilities
-                },
+                capabilities,
                 attempt,
                 failure,
                 user_agent_hash,
@@ -171,11 +167,7 @@ pub(super) fn carrier_request<B>(request: &Request<B>, host: &str) -> Option<Car
                 } else {
                     CarrierClientClass::BrowserHint
                 },
-                if native_ios {
-                    CarrierCapabilities::ios()
-                } else {
-                    CarrierCapabilities::all()
-                },
+                CarrierCapabilities::all(),
                 attempt,
                 failure,
                 user_agent_hash,
@@ -473,11 +465,23 @@ mod tests {
     }
 
     #[test]
-    fn native_ios_capability_claims_cannot_enable_parallel_carriers() {
-        let request = Request::builder()
+    fn native_ios_user_agent_classifies_without_overriding_capabilities() {
+        let metadata_free = Request::builder()
+            .header(
+                header::USER_AGENT,
+                "Telemt/1 CFNetwork/1498.700.2 Darwin/23.6.0",
+            )
+            .body(())
+            .unwrap();
+        let parsed = carrier_request(&metadata_free, "proxy.example.com").unwrap();
+        assert_eq!(parsed.class(), CarrierClientClass::Ios);
+        assert!(!parsed.is_automatic());
+        assert!(!parsed.uses_capabilities());
+
+        let automatic = Request::builder()
             .header(
                 "x-carrier-capabilities",
-                "https,https-lanes,websocket,websocket-lanes",
+                "https,https-lanes",
             )
             .header("x-carrier-attempt", "1")
             .header(
@@ -486,10 +490,11 @@ mod tests {
             )
             .body(())
             .unwrap();
-        let parsed = carrier_request(&request, "proxy.example.com").unwrap();
+        let parsed = carrier_request(&automatic, "proxy.example.com").unwrap();
         assert_eq!(parsed.class(), CarrierClientClass::Ios);
+        assert!(parsed.is_automatic());
         assert!(parsed.supports(WebCarrier::Https));
-        assert!(!parsed.supports(WebCarrier::HttpsLanes));
+        assert!(parsed.supports(WebCarrier::HttpsLanes));
         assert!(!parsed.supports(WebCarrier::Websocket));
         assert!(!parsed.supports(WebCarrier::WebsocketLanes));
     }

src/web/http/tests.rs

@@ -22,6 +22,8 @@ mod legacy_tests;
 #[path = "negotiation_tests.rs"]
 mod negotiation_tests;
 
+const TEST_CARRIER_DEADLINES_SECS: [u64; 4] = [3, 5, 8, 12];
+
 pub(super) fn runtime_config(capability: [u8; 32], carrier: WebCarrier) -> ProxyConfig {
     runtime_config_with_carriers(capability, carrier, false, true, Arc::from([carrier]))
 }
@@ -41,6 +43,41 @@ fn runtime_config_with_carriers(
     carrier_negotiation_enabled: bool,
     carrier_learning: bool,
     carriers: Arc<[WebCarrier]>,
+) -> ProxyConfig {
+    runtime_config_with_carriers_and_deadlines(
+        capability,
+        carrier,
+        carrier_negotiation_enabled,
+        carrier_learning,
+        carriers,
+        TEST_CARRIER_DEADLINES_SECS,
+    )
+}
+
+pub(super) fn negotiation_runtime_config_with_deadlines(
+    capability: [u8; 32],
+    carrier: WebCarrier,
+    carrier_learning: bool,
+    carriers: Arc<[WebCarrier]>,
+    carrier_negotiation_deadlines_secs: [u64; 4],
+) -> ProxyConfig {
+    runtime_config_with_carriers_and_deadlines(
+        capability,
+        carrier,
+        true,
+        carrier_learning,
+        carriers,
+        carrier_negotiation_deadlines_secs,
+    )
+}
+
+fn runtime_config_with_carriers_and_deadlines(
+    capability: [u8; 32],
+    carrier: WebCarrier,
+    carrier_negotiation_enabled: bool,
+    carrier_learning: bool,
+    carriers: Arc<[WebCarrier]>,
+    carrier_negotiation_deadlines_secs: [u64; 4],
 ) -> ProxyConfig {
     let profile = Arc::new(WebRuntimeProfile {
         host: "proxy.example.com".to_string(),
@@ -51,7 +88,7 @@ fn runtime_config_with_carriers(
         carrier_negotiation_enabled,
         carrier_learning,
         carriers: Arc::clone(&carriers),
-        carrier_negotiation_deadlines_secs: [3, 5, 8, 12],
+        carrier_negotiation_deadlines_secs,
         capability,
         key_fingerprint: "0000000000000000".to_string(),
         max_sessions: 4,
@@ -97,6 +134,8 @@ fn runtime_config_with_carriers(
         WebCarriers::Disabled
     };
     config.web.carrier_learning = carrier_learning;
+    config.web.timeouts.carrier_negotiation_deadlines_secs =
+        carrier_negotiation_deadlines_secs;
     config.web.limits.max_bootstraps_per_ip = 1;
     config.web.timeouts.shutdown_secs = 1;
     config.web.runtime = Some(Arc::new(WebRuntimeConfig {

src/web/manager/negotiation.rs

@@ -11,7 +11,7 @@ pub(crate) enum CarrierClientClass {
     Bridge,
     /// Strict same-origin browser metadata survived while the marker did not.
     BrowserHint,
-    /// A native iOS client that supports only the serialized HTTPS carrier.
+    /// A native iOS client classified for diagnostics and learning only.
     Ios,
 }
 
@@ -67,7 +67,7 @@ impl CarrierFailure {
     }
 }
 
-/// Fixed carrier capability set sent by the generated bridge.
+/// Validated carrier capability set sent by a negotiation-capable client.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(crate) struct CarrierCapabilities(u8);
 
@@ -77,11 +77,6 @@ impl CarrierCapabilities {
         Self(0b1111)
     }
 
-    /// Returns the only carrier implemented by the native iOS client.
-    pub(crate) const fn ios() -> Self {
-        Self(0b0001)
-    }
-
     /// Builds a set from a validated bit representation.
     pub(crate) const fn from_bits(bits: u8) -> Option<Self> {
         if bits != 0 && bits & !0b1111 == 0 {
@@ -105,7 +100,6 @@ pub(crate) struct CarrierRequest {
     attempt: Option<u8>,
     failure: Option<CarrierFailure>,
     user_agent_hash: [u8; 32],
-    initial_only: bool,
 }
 
 impl CarrierRequest {
@@ -117,19 +111,17 @@ impl CarrierRequest {
             attempt: None,
             failure: None,
             user_agent_hash,
-            initial_only: false,
         }
     }
 
-    /// Constructs a known fixed-capability client without retry negotiation.
+    /// Constructs a metadata-free native client without inferring capabilities.
     pub(crate) const fn ios(user_agent_hash: [u8; 32]) -> Self {
         Self {
             class: CarrierClientClass::Ios,
-            capabilities: Some(CarrierCapabilities::ios()),
+            capabilities: None,
             attempt: None,
             failure: None,
             user_agent_hash,
-            initial_only: true,
         }
     }
 
@@ -147,13 +139,12 @@ impl CarrierRequest {
             attempt: Some(attempt),
             failure,
             user_agent_hash,
-            initial_only: false,
         }
     }
 
     /// Returns whether this request participates in server-side negotiation.
     pub(crate) const fn is_automatic(self) -> bool {
-        self.capabilities.is_some() && !self.initial_only
+        self.capabilities.is_some()
     }
 
     /// Returns whether server capability filtering applies to this request.
@@ -194,7 +185,6 @@ impl CarrierRequest {
         self.class == other.class
             && self.capabilities_bits() == other.capabilities_bits()
             && self.user_agent_hash == other.user_agent_hash
-            && self.initial_only == other.initial_only
     }
 
     /// Checks the complete idempotent identity of one exact attempt request.
@@ -209,6 +199,21 @@ impl CarrierRequest {
     }
 }
 
+/// Returns the cumulative deadline slot assigned to one carrier attempt.
+pub(super) const fn carrier_attempt_deadline_index(
+    candidate_count: u8,
+    attempt: u8,
+) -> Option<usize> {
+    if candidate_count == 0 || candidate_count > 4 || attempt == 0 || attempt > candidate_count {
+        return None;
+    }
+    if attempt == candidate_count {
+        Some(3)
+    } else {
+        Some((attempt - 1) as usize)
+    }
+}
+
 /// Secret-independent evidence owner frozen into an automatic session.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(crate) struct CarrierLearningContext {
@@ -225,3 +230,29 @@ pub(crate) struct CarrierLearningContext {
     /// Whether the authoritative client address is safe to use as learning evidence.
     pub(crate) ip_learning_eligible: bool,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::carrier_attempt_deadline_index;
+
+    #[test]
+    fn final_candidate_uses_the_final_cumulative_deadline_slot() {
+        assert_eq!(carrier_attempt_deadline_index(1, 1), Some(3));
+        assert_eq!(carrier_attempt_deadline_index(2, 1), Some(0));
+        assert_eq!(carrier_attempt_deadline_index(2, 2), Some(3));
+        assert_eq!(carrier_attempt_deadline_index(3, 1), Some(0));
+        assert_eq!(carrier_attempt_deadline_index(3, 2), Some(1));
+        assert_eq!(carrier_attempt_deadline_index(3, 3), Some(3));
+        assert_eq!(carrier_attempt_deadline_index(4, 1), Some(0));
+        assert_eq!(carrier_attempt_deadline_index(4, 2), Some(1));
+        assert_eq!(carrier_attempt_deadline_index(4, 3), Some(2));
+        assert_eq!(carrier_attempt_deadline_index(4, 4), Some(3));
+    }
+
+    #[test]
+    fn invalid_candidate_or_attempt_counts_have_no_deadline_slot() {
+        for (candidate_count, attempt) in [(0, 1), (5, 1), (1, 0), (1, 2), (3, 4)] {
+            assert_eq!(carrier_attempt_deadline_index(candidate_count, attempt), None);
+        }
+    }
+}

src/web/manager/session_creation.rs

@@ -11,6 +11,7 @@ use super::state::{
     CarrierChainPhase, decrement_map, matching_profile, new_unique_token, profile_key,
     remember_closed_token_locked, remove_expired_locked,
 };
+use super::negotiation::carrier_attempt_deadline_index;
 use super::session_admission::admit_initial;
 use super::{
     CarrierLearningContext, CarrierRequest, CreateResult, ManagerError, TokenHash, WebProcessRuntime,
@@ -154,7 +155,10 @@ impl WebProcessRuntime {
             else {
                 return Err(ManagerError::Protocol);
             };
-            let deadline_index = usize::from(next_attempt.saturating_sub(2));
+            let candidate_count = u8::try_from(entry.carrier_candidates.len())
+                .map_err(|_| ManagerError::Protocol)?;
+            let deadline_index = carrier_attempt_deadline_index(candidate_count, next_attempt)
+                .ok_or(ManagerError::Protocol)?;
             if entry.carrier_started_at.is_some_and(|started| {
                 now.saturating_duration_since(started)
                     >= Duration::from_secs(