Merge pull request #943 from telemt/carrier-methods

Configurable HTTP Methods for Bridge Carriers + Fix autonomous ME recovery from Direct fallback

src/api/config_edit/tests.rs

@@ -1,5 +1,54 @@
 use super::*;
 
+#[tokio::test]
+async fn carrier_method_api_defaults_and_patches_are_hot() {
+    let (path, _directory) = temp_config("[web]\nenabled = false\n");
+    let (value, _) = read_managed_config(&path).await.unwrap();
+    assert_eq!(value["web"]["carrier_method"].as_str(), Some("post"));
+    for token in ["put", "post"] {
+        let active = ProxyConfig::load(&path).unwrap();
+        let patch = serde_json::json!({"web": {"carrier_method": token}});
+        let mut response = apply_patch_to_path(&path, &patch, None).await.unwrap();
+        let desired = ProxyConfig::load(&path).unwrap();
+        reconcile_runtime_effect(&mut response, &active, &desired).unwrap();
+        assert!(!response.restart_required);
+        assert!(response.runtime_reload_required);
+        assert!(!response.process_restart_required);
+        assert!(response.deferred_process_fields.is_empty());
+        let (value, revision) = read_managed_config(&path).await.unwrap();
+        assert_eq!(value["web"]["carrier_method"].as_str(), Some(token));
+        assert_eq!(revision, response.revision);
+        let written = tokio::fs::read_to_string(&path).await.unwrap();
+        assert!(written.contains(&format!("carrier_method = \"{token}\"")));
+    }
+}
+
+#[tokio::test]
+async fn carrier_method_api_rejects_invalid_values_without_writing() {
+    let (path, _directory) = temp_config("[web]\ncarrier_method = \"put\"\n");
+    let original = tokio::fs::read(&path).await.unwrap();
+    let revision = crate::api::config_store::current_revision(&path)
+        .await
+        .unwrap();
+    for value in [
+        serde_json::json!("PUT"),
+        serde_json::json!("patch"),
+        serde_json::json!(true),
+        serde_json::json!(42),
+    ] {
+        let patch = serde_json::json!({"web": {"carrier_method": value}});
+        let error = apply_patch_to_path(&path, &patch, None).await.unwrap_err();
+        assert_eq!(error.status, hyper::StatusCode::BAD_REQUEST);
+        assert_eq!(tokio::fs::read(&path).await.unwrap(), original);
+        assert_eq!(
+            crate::api::config_store::current_revision(&path)
+                .await
+                .unwrap(),
+            revision
+        );
+    }
+}
+
 #[test]
 fn json_object_converts_to_toml_table() {
     let j: Json = serde_json::json!({"censorship": {"tls_domain": "a.com"}, "default_dc": 2});

src/config/defaults.rs

@@ -59,7 +59,7 @@ const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_BUDGET_PER_CORE: u16 = 16;
 const DEFAULT_ME_POOL_DRAIN_SOFT_EVICT_COOLDOWN_MS: u64 = 1000;
 const DEFAULT_USER_MAX_UNIQUE_IPS_WINDOW_SECS: u64 = 30;
 const DEFAULT_ACCEPT_PERMIT_TIMEOUT_MS: u64 = 250;
-const DEFAULT_CONNTRACK_CONTROL_ENABLED: bool = true;
+const DEFAULT_CONNTRACK_CONTROL_ENABLED: bool = false;
 const DEFAULT_CONNTRACK_PRESSURE_HIGH_WATERMARK_PCT: u8 = 85;
 const DEFAULT_CONNTRACK_PRESSURE_LOW_WATERMARK_PCT: u8 = 70;
 const DEFAULT_CONNTRACK_DELETE_BUDGET_PER_SEC: u64 = 4096;
@@ -244,6 +244,7 @@ pub(crate) fn default_accept_permit_timeout_ms() -> u64 {
     DEFAULT_ACCEPT_PERMIT_TIMEOUT_MS
 }
 
+/// Keeps privileged conntrack control disabled unless explicitly requested.
 pub(crate) fn default_conntrack_control_enabled() -> bool {
     DEFAULT_CONNTRACK_CONTROL_ENABLED
 }

src/config/hot_reload.rs

@@ -17,6 +17,7 @@
 //! | `network` | `dns_overrides`                | Applied immediately                            |
 //! | `access`  | All user/quota fields          | Effective immediately                          |
 //! | `web`     | Carrier, timing, and debug policy | Applied to newly issued sessions             |
+//! | `web`     | `carrier_method`               | Applied to newly rendered bridge pages        |
 //! Fields that require re-binding sockets (`server.listeners`, legacy
 //! `server.port`, `censorship.*`, `network.*`, `use_middle_proxy`) are **not**
 //! applied; a warning is emitted. SYN limiter rules are process-owned and are
@@ -67,3 +68,7 @@ use watcher::{ReloadState, reload_config};
 mod base_path_tests;
 #[cfg(test)]
 mod tests;
+// Carrier method reloads preserve page-owned requests and process-owned limits.
+#[cfg(test)]
+#[path = "hot_reload/carrier_method_tests.rs"]
+mod carrier_method_tests;

src/config/hot_reload/carrier_method_tests.rs

@@ -0,0 +1,50 @@
+use super::*;
+
+use crate::config::WebCarrierMethod;
+
+#[test]
+fn carrier_method_reload_is_hot_and_preserves_process_limits() {
+    let mut active = ProxyConfig::default();
+    for method in [WebCarrierMethod::Put, WebCarrierMethod::Post] {
+        let mut desired = active.clone();
+        desired.web.carrier_method = method;
+        assert_eq!(classify_config_changes(&active, &desired).changed, ["web"]);
+        assert!(!classify_config_changes(&active, &desired).restart_required);
+        desired.web.limits.max_http_connections += 1;
+        let applied = overlay_hot_fields(&active, &desired);
+        assert_eq!(applied.web.carrier_method, method);
+        assert_eq!(
+            applied.web.limits.max_http_connections,
+            active.web.limits.max_http_connections
+        );
+        active = applied;
+    }
+}
+
+#[test]
+fn carrier_method_reload_publishes_both_directions_and_keeps_last_good_value() {
+    let directory = tempfile::tempdir().unwrap();
+    let path = directory.path().join("config.toml");
+    let source = |method| format!("[web]\ncarrier_method = \"{method}\"\n");
+    std::fs::write(&path, source("post")).unwrap();
+    let initial = Arc::new(ProxyConfig::load(&path).unwrap());
+    let initial_hash = ProxyConfig::load_with_metadata(&path)
+        .unwrap()
+        .rendered_hash;
+    let (config_tx, _config_rx) = watch::channel(Arc::clone(&initial));
+    let (log_tx, _log_rx) = watch::channel(initial.general.log_level.clone());
+    let mut reload_state = ReloadState::new(Some(initial_hash));
+    for (token, method) in [
+        ("put", WebCarrierMethod::Put),
+        ("post", WebCarrierMethod::Post),
+    ] {
+        std::fs::write(&path, source(token)).unwrap();
+        reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap();
+        let applied = config_tx.borrow().clone();
+        assert_eq!(applied.web.carrier_method, method);
+        std::fs::write(&path, source("PATCH")).unwrap();
+        reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state);
+        let unchanged = config_tx.borrow().clone();
+        assert!(Arc::ptr_eq(&unchanged, &applied));
+    }
+}

src/config/load/strict_keys.rs

@@ -263,6 +263,7 @@ const LISTENER_CONFIG_KEYS: &[&str] = &[
 const WEB_CONFIG_KEYS: &[&str] = &[
     "enabled",
     "carrier",
+    "carrier_method",
     "carriers",
     "carrier_learning",
     "carrier_negotiation_aggressiveness",

src/config/tests/load_basic_tests/conntrack_tests.rs

@@ -1,5 +1,71 @@
 use super::*;
 
+#[test]
+fn conntrack_control_is_opt_in_by_default() {
+    assert!(!default_conntrack_control_enabled());
+    assert!(
+        !ProxyConfig::default()
+            .server
+            .conntrack_control
+            .inline_conntrack_control
+    );
+    let cfg = load_config_from_temp_toml(
+        r#"
+        [censorship]
+        tls_domain = "example.com"
+
+        [access.users]
+        user = "00000000000000000000000000000000"
+        "#,
+    );
+    assert!(!cfg.server.conntrack_control.inline_conntrack_control);
+    assert!(
+        !cfg.server
+            .conntrack_control
+            .inline_conntrack_control_explicit
+    );
+    assert_eq!(
+        serde_json::to_value(&cfg.server.conntrack_control).unwrap()["inline_conntrack_control"],
+        false,
+    );
+}
+
+#[test]
+fn conntrack_control_preserves_explicit_enablement() {
+    for enabled in [false, true] {
+        let cfg = load_config_from_temp_toml(&format!(
+            r#"
+            [server.conntrack_control]
+            inline_conntrack_control = {enabled}
+            backend = "nftables"
+
+            [censorship]
+            tls_domain = "example.com"
+
+            [access.users]
+            user = "00000000000000000000000000000000"
+            "#,
+        ));
+        assert_eq!(
+            cfg.server.conntrack_control.inline_conntrack_control,
+            enabled
+        );
+        assert!(
+            cfg.server
+                .conntrack_control
+                .inline_conntrack_control_explicit
+        );
+        assert_eq!(
+            cfg.server.conntrack_control.backend,
+            ConntrackBackend::Nftables
+        );
+        assert_eq!(
+            serde_json::to_value(&cfg.server.conntrack_control).unwrap()["inline_conntrack_control"],
+            enabled,
+        );
+    }
+}
+
 #[test]
 fn conntrack_pressure_high_watermark_out_of_range_is_rejected() {
     let toml = r#"

src/config/tests/load_basic_tests/web_tests.rs

@@ -3,6 +3,10 @@ use super::*;
 #[path = "web_tests/base_path_tests.rs"]
 mod base_path_tests;
 
+// Method selection must not alter capability identity or accepted config literals.
+#[path = "web_tests/carrier_method_tests.rs"]
+mod carrier_method_tests;
+
 const WEB_CONFIG: &str = r#"
 [access.users]
 alice = "000102030405060708090a0b0c0d0e0f"

src/config/tests/load_basic_tests/web_tests/carrier_method_tests.rs

@@ -0,0 +1,48 @@
+use super::*;
+
+#[test]
+fn carrier_method_defaults_and_roundtrips_without_changing_capabilities() {
+    let default = load_config_from_temp_toml(WEB_CONFIG);
+    assert_eq!(default.web.carrier_method, WebCarrierMethod::Post);
+    let capabilities = default.web.runtime.as_ref().unwrap().capabilities.clone();
+    for (token, method) in [
+        ("post", WebCarrierMethod::Post),
+        ("put", WebCarrierMethod::Put),
+    ] {
+        let configured = WEB_CONFIG.replace(
+            "carrier = \"https-lanes\"",
+            &format!("carrier = \"https-lanes\"\ncarrier_method = \"{token}\""),
+        );
+        let source = format!("[general]\nconfig_strict = true\n{configured}");
+        let config = load_config_from_temp_toml(&source);
+        assert_eq!(config.web.carrier_method, method);
+        assert_eq!(
+            config.web.runtime.as_ref().unwrap().capabilities,
+            capabilities
+        );
+        let json = serde_json::to_value(&config.web).unwrap();
+        assert_eq!(json["carrier_method"], token);
+        let decoded: WebConfig = serde_json::from_value(json).unwrap();
+        assert_eq!(decoded.carrier_method, method);
+        let serialized = toml::to_string(&config.web).unwrap();
+        let decoded: WebConfig = toml::from_str(&serialized).unwrap();
+        assert_eq!(decoded.carrier_method, method);
+    }
+}
+
+#[test]
+fn carrier_method_rejects_unknown_tokens_types_and_aliases() {
+    for value in ["\"POST\"", "\"PUT\"", "\"patch\"", "true", "42", "[]"] {
+        let source = WEB_CONFIG.replace(
+            "carrier = \"https-lanes\"",
+            &format!("carrier = \"https-lanes\"\ncarrier_method = {value}"),
+        );
+        assert!(load_config_error_from_temp_toml(&source).contains("carrier_method"));
+    }
+    let configured = WEB_CONFIG.replace(
+        "carrier = \"https-lanes\"",
+        "carrier = \"https-lanes\"\nhttp_method = \"put\"",
+    );
+    let source = format!("[general]\nconfig_strict = true\n{configured}");
+    assert!(load_config_error_from_temp_toml(&source).contains("http_method"));
+}

src/config/types.rs

@@ -63,7 +63,7 @@ pub(crate) use web::{
 };
 pub(crate) use web_carrier::WEB_CARRIER_LEARNING_MIN_ENTRIES;
 #[allow(unused_imports)]
-pub use web_carrier::{WebCarrier, WebCarriers};
+pub use web_carrier::{WebCarrier, WebCarrierMethod, WebCarriers};
 pub(crate) use web_debug::web_debug_fits_limits;
 pub use web_debug::{WebDebugBodyCapture, WebDebugConfig};
 

src/config/types/web.rs

@@ -6,7 +6,7 @@ use std::sync::Arc;
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 
-use super::web_carrier::{WebCarrier, WebCarriers};
+use super::web_carrier::{WebCarrier, WebCarrierMethod, WebCarriers};
 use super::web_debug::WebDebugConfig;
 
 // Serialized WEB defaults remain separate from the runtime data model.
@@ -413,6 +413,9 @@ pub struct WebConfig {
     /// Sole carrier when negotiation is disabled and final fallback when enabled.
     #[serde(default)]
     pub carrier: WebCarrier,
+    /// Page-owned HTTP method for HTTPS uplink and downlink, including retries.
+    #[serde(default)]
+    pub carrier_method: WebCarrierMethod,
     /// Ordered carriers considered by server-side negotiation before the fallback carrier.
     #[serde(default)]
     pub carriers: WebCarriers,
@@ -471,6 +474,7 @@ impl Default for WebConfig {
         Self {
             enabled: false,
             carrier: WebCarrier::default(),
+            carrier_method: WebCarrierMethod::default(),
             carriers: WebCarriers::default(),
             carrier_learning: default_web_carrier_learning(),
             carrier_negotiation_aggressiveness: WebCarrierNegotiationAggressiveness::default(),

src/config/types/web_carrier.rs

@@ -3,6 +3,27 @@ use serde::{Deserialize, Serialize};
 /// Minimum restart-owned entries required for one complete learning sample.
 pub(crate) const WEB_CARRIER_LEARNING_MIN_ENTRIES: usize = 3;
 
+/// HTTP method emitted by newly rendered HTTPS carrier bridges.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum WebCarrierMethod {
+    /// Preserve the original bridge request method.
+    #[default]
+    Post,
+    /// Allow idempotent-method connection reuse in Linux WebKitGTK.
+    Put,
+}
+
+impl WebCarrierMethod {
+    /// Returns the HTTP token embedded in the page-owned carrier policy.
+    pub(crate) const fn as_str(self) -> &'static str {
+        match self {
+            Self::Post => "POST",
+            Self::Put => "PUT",
+        }
+    }
+}
+
 /// Carrier selected for one newly issued WEB relay session.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case")]

src/conntrack_control/firewall/actor.rs

@@ -26,15 +26,21 @@ const RETRY_DELAYS: [Duration; 6] = [
     Duration::from_secs(30),
 ];
 
+/// Reports whether a generation's desired firewall policy was confirmed.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(super) enum ReconcileOutcome {
+    /// The applied policy matches the accepted generation's desired policy.
     Applied,
+    /// The attempt failed without confirming the desired policy.
     Failed,
 }
 
+/// Associates a reconciliation result with its accepted runtime generation.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(super) struct ReconcileStatus {
+    /// Runtime generation whose desired policy was attempted.
     pub(super) generation: u64,
+    /// Result of the latest completed attempt for this generation.
     pub(super) outcome: ReconcileOutcome,
 }
 
@@ -51,25 +57,14 @@ pub(crate) struct FirewallAuthority {
 }
 
 impl FirewallAuthority {
-    /// Starts the single process-owned firewall reconciler.
-    pub(crate) fn spawn(control_plane: &ProcessControlPlane) -> Result<Self, String> {
-        let (desired_tx, desired_rx) = watch::channel(None);
-        let (status_tx, status_rx) = watch::channel(None);
-        let terminal = CancellationToken::new();
-        let closed = Arc::new(AtomicBool::new(false));
-        let completed_flag = Arc::new(AtomicBool::new(false));
-        let cleanup_succeeded = Arc::new(AtomicBool::new(false));
-        let completed = Arc::new(Notify::new());
-        let actor = FirewallReconciler::new(
-            SystemCommandRunner,
-            desired_rx,
-            status_tx,
-            terminal.clone(),
-            Arc::clone(&closed),
-            Arc::clone(&completed_flag),
-            Arc::clone(&cleanup_succeeded),
-            Arc::clone(&completed),
-        );
+    /// Starts reconciliation only when enabled and CAP_NET_ADMIN is available.
+    pub(crate) fn spawn(
+        control_plane: &ProcessControlPlane,
+        config: &ProxyConfig,
+    ) -> Result<Option<Self>, String> {
+        let Some((authority, actor)) = Self::prepare(config, SystemCommandRunner) else {
+            return Ok(None);
+        };
         control_plane
             .spawn_cooperative(move |process_cancellation| async move {
                 actor.run(process_cancellation).await;
@@ -78,7 +73,37 @@ impl FirewallAuthority {
                 "process control-plane admission closed before conntrack firewall startup"
                     .to_string()
             })?;
-        Ok(Self {
+        Ok(Some(authority))
+    }
+
+    // Admission precedes allocation so an opted-out process never owns firewall cleanup.
+    // Keeping construction separate permits fake runners without changing async Send bounds.
+    fn prepare<R>(config: &ProxyConfig, runner: R) -> Option<(Self, FirewallReconciler<R>)>
+    where
+        R: FirewallCommandRunner + 'static,
+    {
+        if !config.server.conntrack_control.inline_conntrack_control || !runner.has_cap_net_admin()
+        {
+            return None;
+        }
+        let (desired_tx, desired_rx) = watch::channel(None);
+        let (status_tx, status_rx) = watch::channel(None);
+        let terminal = CancellationToken::new();
+        let closed = Arc::new(AtomicBool::new(false));
+        let completed_flag = Arc::new(AtomicBool::new(false));
+        let cleanup_succeeded = Arc::new(AtomicBool::new(false));
+        let completed = Arc::new(Notify::new());
+        let actor = FirewallReconciler::new(
+            runner,
+            desired_rx,
+            status_tx,
+            terminal.clone(),
+            Arc::clone(&closed),
+            Arc::clone(&completed_flag),
+            Arc::clone(&cleanup_succeeded),
+            Arc::clone(&completed),
+        );
+        let authority = Self {
             desired_tx,
             status_rx,
             terminal,
@@ -86,7 +111,8 @@ impl FirewallAuthority {
             completed_flag,
             cleanup_succeeded,
             completed,
-        })
+        };
+        Some((authority, actor))
     }
 
     /// Publishes policy only after its runtime generation becomes active.
@@ -169,6 +195,7 @@ impl Drop for CompletionGuard {
     }
 }
 
+/// Serializes confirmed-state transitions and retains cleanup ownership until shutdown.
 pub(super) struct FirewallReconciler<R> {
     runner: R,
     desired_rx: watch::Receiver<Option<DesiredState>>,
@@ -186,6 +213,7 @@ impl<R> FirewallReconciler<R>
 where
     R: FirewallCommandRunner + 'static,
 {
+    /// Constructs an admitted actor with unknown state pending startup recovery.
     pub(super) fn new(
         runner: R,
         desired_rx: watch::Receiver<Option<DesiredState>>,
@@ -214,6 +242,7 @@ where
         }
     }
 
+    /// Reconciles accepted generations and completes bounded cleanup on cancellation.
     pub(super) async fn run(mut self, process_cancellation: CancellationToken) {
         let mut current = None;
         let mut retry_index = 0usize;
@@ -337,6 +366,7 @@ where
         let _completion = &self.completion;
     }
 
+    /// Accepts fenced publications without replacing telemetry with stale desired state.
     pub(super) fn take_latest_desired(&mut self) -> Option<DesiredState> {
         let next = self.desired_rx.borrow_and_update().clone()?;
         if next.generation < self.last_generation {
@@ -369,3 +399,8 @@ where
         Some(next)
     }
 }
+
+// Exercises startup admission and cleanup ownership with fake helper processes.
+#[cfg(test)]
+#[path = "tests/startup_admission.rs"]
+mod startup_admission_tests;

src/conntrack_control/firewall/tests/startup_admission.rs

@@ -0,0 +1,267 @@
+use std::sync::Mutex;
+use std::sync::atomic::AtomicUsize;
+
+use crate::config::{ConntrackBackend, ConntrackMode};
+
+use super::super::command::{CommandErrorKind, CommandSpec, classify_command_error};
+use super::super::transaction::reconcile_once;
+use super::*;
+
+const NETLINK_ERROR: &str =
+    "src/mnl.c:68: Unable to initialize Netlink socket: Address family not supported by protocol";
+const PERMISSION_ERROR: &str = "iptables v1.8.11 (nf_tables): Could not fetch rule set generation id: Permission denied (you must be root)";
+
+#[derive(Clone)]
+struct StartupRunner {
+    has_cap_net_admin: bool,
+    cap_probes: Arc<AtomicUsize>,
+    calls: Arc<Mutex<Vec<CommandSpec>>>,
+    failure: Option<&'static str>,
+}
+
+impl StartupRunner {
+    fn new(has_cap_net_admin: bool, failure: Option<&'static str>) -> Self {
+        Self {
+            has_cap_net_admin,
+            cap_probes: Arc::new(AtomicUsize::new(0)),
+            calls: Arc::new(Mutex::new(Vec::new())),
+            failure,
+        }
+    }
+
+    fn calls(&self) -> Vec<CommandSpec> {
+        self.calls.lock().unwrap().clone()
+    }
+}
+
+impl FirewallCommandRunner for StartupRunner {
+    fn available(&self, _binary: &str) -> bool {
+        true
+    }
+
+    fn has_cap_net_admin(&self) -> bool {
+        self.cap_probes.fetch_add(1, Ordering::Relaxed);
+        self.has_cap_net_admin
+    }
+
+    async fn run(&self, spec: CommandSpec) -> Result<(), CommandError> {
+        self.calls.lock().unwrap().push(spec.clone());
+        if let Some(message) = self.failure {
+            return Err(CommandError {
+                kind: classify_command_error(spec.binary, &spec.args, message),
+                message: message.to_string(),
+            });
+        }
+        if (spec.binary == "nft" && spec.args.first().map(String::as_str) == Some("delete"))
+            || (matches!(spec.binary, "iptables" | "ip6tables")
+                && matches!(
+                    spec.args.get(2).map(String::as_str),
+                    Some("-C" | "-D" | "-F" | "-X")
+                ))
+        {
+            return Err(CommandError {
+                kind: CommandErrorKind::NotFound,
+                message: "injected absent owned object".to_string(),
+            });
+        }
+        Ok(())
+    }
+}
+
+fn start_with_runner(
+    config: &ProxyConfig,
+    runner: StartupRunner,
+    scope: &ProcessControlPlane,
+) -> Option<FirewallAuthority> {
+    let (authority, actor) = FirewallAuthority::prepare(config, runner)?;
+    assert!(
+        scope
+            .spawn_cooperative(move |cancellation| actor.run(cancellation))
+            .is_ok()
+    );
+    Some(authority)
+}
+
+#[tokio::test(start_paused = true)]
+async fn disabled_startup_never_acquires_cleanup_ownership() {
+    for explicit in [false, true] {
+        for has_cap in [true, false] {
+            for backend in [
+                ConntrackBackend::Auto,
+                ConntrackBackend::Nftables,
+                ConntrackBackend::Iptables,
+            ] {
+                for mode in [
+                    ConntrackMode::Tracked,
+                    ConntrackMode::Notrack,
+                    ConntrackMode::Hybrid,
+                ] {
+                    for message in [NETLINK_ERROR, PERMISSION_ERROR] {
+                        let mut config = ProxyConfig::default();
+                        if explicit {
+                            config.server.conntrack_control.inline_conntrack_control = false;
+                            config
+                                .server
+                                .conntrack_control
+                                .inline_conntrack_control_explicit = true;
+                        }
+                        config.server.conntrack_control.backend = backend;
+                        config.server.conntrack_control.mode = mode;
+                        let runner = StartupRunner::new(has_cap, Some(message));
+                        let scope = ProcessControlPlane::new();
+                        let authority = start_with_runner(&config, runner.clone(), &scope);
+                        assert!(
+                            authority.is_none(),
+                            "{explicit} {has_cap} {backend:?} {mode:?}"
+                        );
+                        assert_eq!(runner.cap_probes.load(Ordering::Relaxed), 0);
+                        tokio::time::advance(Duration::from_secs(90)).await;
+                        assert!(scope.shutdown(Duration::from_secs(1)).await);
+                        assert!(runner.calls().is_empty());
+                    }
+                }
+            }
+        }
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn enabled_startup_without_capability_creates_no_firewall_work() {
+    let mut config = ProxyConfig::default();
+    config.server.conntrack_control.inline_conntrack_control = true;
+    for mode in [ConntrackMode::Tracked, ConntrackMode::Notrack] {
+        config.server.conntrack_control.mode = mode;
+        let runner = StartupRunner::new(false, Some(PERMISSION_ERROR));
+        let scope = ProcessControlPlane::new();
+        assert!(start_with_runner(&config, runner.clone(), &scope).is_none());
+        assert_eq!(runner.cap_probes.load(Ordering::Relaxed), 1);
+        tokio::time::advance(Duration::from_secs(90)).await;
+        assert!(scope.shutdown(Duration::from_secs(1)).await);
+        assert!(runner.calls().is_empty());
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn admitted_startup_preserves_recovery_idle_and_shutdown_cleanup() {
+    for mode in [ConntrackMode::Tracked, ConntrackMode::Notrack] {
+        let mut config = ProxyConfig::default();
+        config.server.conntrack_control.inline_conntrack_control = true;
+        config.server.conntrack_control.mode = mode;
+        config.server.conntrack_control.backend = ConntrackBackend::Nftables;
+        let runner = StartupRunner::new(true, None);
+        let scope = ProcessControlPlane::new();
+        let authority = start_with_runner(&config, runner.clone(), &scope).unwrap();
+        let stats = Arc::new(Stats::new());
+        assert!(
+            authority
+                .publish_initial(1, Arc::new(config), stats.clone())
+                .await
+        );
+        assert!(stats.get_conntrack_rule_apply_ok());
+        for binary in ["nft", "iptables", "ip6tables"] {
+            assert!(runner.calls().iter().any(|call| call.binary == binary));
+        }
+        let calls = runner.calls().len();
+        tokio::time::advance(Duration::from_secs(90)).await;
+        tokio::task::yield_now().await;
+        assert_eq!(runner.calls().len(), calls);
+        assert!(authority.shutdown_and_clear().await);
+        assert!(!stats.get_conntrack_rule_apply_ok());
+        assert!(runner.calls().len() > calls);
+        assert!(scope.shutdown(Duration::from_secs(1)).await);
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn admitted_process_cancellation_preserves_cleanup() {
+    let mut config = ProxyConfig::default();
+    config.server.conntrack_control.inline_conntrack_control = true;
+    let runner = StartupRunner::new(true, None);
+    let scope = ProcessControlPlane::new();
+    let authority = start_with_runner(&config, runner.clone(), &scope).unwrap();
+    assert!(scope.shutdown(Duration::from_secs(1)).await);
+    assert!(authority.shutdown_and_clear().await);
+    assert!(!runner.calls().is_empty());
+}
+
+#[test]
+fn netlink_and_permission_errors_are_not_classified_as_absence() {
+    for spec in [
+        CommandSpec::new("nft", ["delete", "table", "inet", "telemt_conntrack"]),
+        CommandSpec::new("nft", ["-f", "-"]),
+        CommandSpec::new(
+            "iptables",
+            ["-t", "raw", "-D", "PREROUTING", "-j", "TELEMT_NOTRACK"],
+        ),
+        CommandSpec::new("iptables", ["-t", "raw", "-F", "TELEMT_NT_A"]),
+        CommandSpec::new("ip6tables", ["-t", "raw", "-X", "TELEMT_NT_B"]),
+        CommandSpec::new("iptables-restore", ["--noflush"]),
+        CommandSpec::new("ip6tables-restore", ["--noflush"]),
+    ] {
+        for message in [
+            NETLINK_ERROR,
+            PERMISSION_ERROR,
+            "iptables: Failed to initialize nft: Address family not supported by protocol",
+        ] {
+            assert_eq!(
+                classify_command_error(spec.binary, &spec.args, message),
+                CommandErrorKind::Failed,
+                "{spec:?} {message}",
+            );
+        }
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn admitted_genuine_failures_remain_unknown_and_retry() {
+    for message in [NETLINK_ERROR, PERMISSION_ERROR] {
+        let mut config = ProxyConfig::default();
+        config.server.conntrack_control.inline_conntrack_control = true;
+        config.server.conntrack_control.mode = ConntrackMode::Notrack;
+        config.server.conntrack_control.backend = ConntrackBackend::Nftables;
+        let runner = StartupRunner::new(true, Some(message));
+        let mut applied = AppliedState::Unknown;
+        let requested = DesiredState {
+            generation: 1,
+            policy: DesiredPolicy::from_config(&config),
+            stats: Arc::new(Stats::new()),
+        };
+        let error = reconcile_once(&runner, &runner, &mut applied, &requested)
+            .await
+            .unwrap_err();
+        assert!(error.message.contains(message));
+        assert_eq!(applied, AppliedState::Unknown);
+        assert!(runner.calls().iter().all(|call| call.stdin.is_none()));
+
+        let scope = ProcessControlPlane::new();
+        let authority = start_with_runner(&config, runner.clone(), &scope).unwrap();
+        let stats = Arc::new(Stats::new());
+        assert!(
+            !authority
+                .publish_initial(1, Arc::new(config), stats.clone())
+                .await
+        );
+        assert!(!stats.get_conntrack_rule_apply_ok());
+        assert_eq!(stats.get_conntrack_rule_reconcile_error_total(), 1);
+        let mut status = authority.status_rx.clone();
+        assert_eq!(
+            status.borrow_and_update().as_ref().unwrap().outcome,
+            ReconcileOutcome::Failed,
+        );
+        let calls = runner.calls().len();
+        tokio::time::advance(Duration::from_millis(999)).await;
+        tokio::task::yield_now().await;
+        assert_eq!(runner.calls().len(), calls);
+        tokio::time::advance(Duration::from_millis(1)).await;
+        status.changed().await.unwrap();
+        assert_eq!(
+            status.borrow().as_ref().unwrap().outcome,
+            ReconcileOutcome::Failed,
+        );
+        assert_eq!(stats.get_conntrack_rule_reconcile_error_total(), 2);
+        assert!(!authority.shutdown_and_clear().await);
+        assert!(authority.completed_flag.load(Ordering::Acquire));
+        assert!(scope.shutdown(Duration::from_secs(1)).await);
+        assert!(runner.calls().iter().all(|call| call.stdin.is_none()));
+    }
+}

src/maestro/admission.rs

@@ -13,6 +13,11 @@ use super::generation::RuntimeTaskScope;
 const STARTUP_FALLBACK_AFTER: Duration = Duration::from_secs(80);
 const RUNTIME_FALLBACK_AFTER: Duration = Duration::from_secs(6);
 
+// Admission regressions cover notification loss without network startup.
+#[cfg(test)]
+mod tests;
+
+/// Keeps generation admission and routing synchronized with periodic ME readiness.
 pub(crate) async fn configure_admission_gate(
     config: &Arc<ProxyConfig>,
     me_pool: Option<Arc<MePool>>,
@@ -76,11 +81,18 @@ pub(crate) async fn configure_admission_gate(
                 } else {
                     Some(Instant::now())
                 };
+                let mut config_watch_open = true;
+                let mut me_ready_watch_open = true;
                 loop {
                     tokio::select! {
-                        changed = config_rx_gate.changed() => {
+                        changed = config_rx_gate.changed(), if config_watch_open => {
                             if changed.is_err() {
-                                break;
+                                config_watch_open = false;
+                                warn!(
+                                    watch_channel = "config",
+                                    "Admission config watch closed; continuing readiness polling with last configuration"
+                                );
+                                continue;
                             }
                             let cfg = config_rx_gate.borrow_and_update().clone();
                             admission_poll_ms = cfg.general.me_admission_poll_ms.max(1);
@@ -88,9 +100,13 @@ pub(crate) async fn configure_admission_gate(
                             fast_fallback_enabled = cfg.general.me2dc_fallback && cfg.general.me2dc_fast;
                             continue;
                         }
-                        changed = me_ready_rx_gate.changed() => {
+                        changed = me_ready_rx_gate.changed(), if me_ready_watch_open => {
                             if changed.is_err() {
-                                break;
+                                me_ready_watch_open = false;
+                                warn!(
+                                    watch_channel = "me_ready",
+                                    "Admission ME readiness watch closed; continuing periodic polling"
+                                );
                             }
                         }
                         _ = tokio::time::sleep(Duration::from_millis(admission_poll_ms)) => {}

src/maestro/admission/tests.rs

@@ -0,0 +1,210 @@
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use tracing::{Event, Subscriber};
+use tracing_subscriber::Layer;
+use tracing_subscriber::layer::{Context, SubscriberExt};
+
+use super::*;
+use crate::transport::middle_proxy::admission_test_support::IdlePoolFixture;
+
+const POLL_MS: u64 = 20;
+
+struct ClosureWarnings(Arc<AtomicUsize>);
+
+impl<S: Subscriber> Layer<S> for ClosureWarnings {
+    fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
+        struct ChannelField(bool);
+        impl tracing::field::Visit for ChannelField {
+            fn record_debug(
+                &mut self,
+                field: &tracing::field::Field,
+                _value: &dyn std::fmt::Debug,
+            ) {
+                self.0 |= field.name() == "watch_channel";
+            }
+        }
+        if event.metadata().level() == &tracing::Level::WARN {
+            let mut visitor = ChannelField(false);
+            event.record(&mut visitor);
+            if visitor.0 {
+                self.0.fetch_add(1, Ordering::Relaxed);
+            }
+        }
+    }
+}
+
+struct GateFixture {
+    pool: IdlePoolFixture,
+    route: Arc<RouteRuntimeController>,
+    admission: watch::Receiver<bool>,
+    config_tx: Option<watch::Sender<Arc<ProxyConfig>>>,
+    ready_tx: Option<watch::Sender<u64>>,
+    config: Arc<ProxyConfig>,
+    scope: RuntimeTaskScope,
+}
+
+impl GateFixture {
+    async fn new() -> Self {
+        let pool = IdlePoolFixture::new(false).await;
+        let mut cfg = ProxyConfig::default();
+        cfg.general.use_middle_proxy = true;
+        cfg.general.me2dc_fallback = true;
+        cfg.general.me2dc_fast = true;
+        cfg.general.me_admission_poll_ms = POLL_MS;
+        let config = Arc::new(cfg);
+        let (config_tx, config_rx) = watch::channel(config.clone());
+        let (ready_tx, ready_rx) = watch::channel(0);
+        let (admission_tx, mut admission) = watch::channel(false);
+        let route = Arc::new(RouteRuntimeController::new(RelayRouteMode::Direct));
+        let scope = RuntimeTaskScope::new();
+        configure_admission_gate(
+            &config,
+            Some(pool.pool()),
+            Arc::new(RwLock::new(Some(pool.pool()))),
+            route.clone(),
+            &admission_tx,
+            config_rx,
+            ready_rx,
+            scope.clone(),
+        )
+        .await;
+        drop(admission_tx);
+        admission.borrow_and_update();
+        tokio::task::yield_now().await;
+        Self {
+            pool,
+            route,
+            admission,
+            config_tx: Some(config_tx),
+            ready_tx: Some(ready_tx),
+            config,
+            scope,
+        }
+    }
+
+    async fn poll(&self) {
+        tokio::task::yield_now().await;
+        tokio::time::advance(Duration::from_millis(POLL_MS)).await;
+        tokio::task::yield_now().await;
+    }
+
+    async fn expect_route(&self, mode: RelayRouteMode) {
+        let mut rx = self.route.subscribe();
+        tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                if rx.borrow_and_update().mode == mode {
+                    return;
+                }
+                rx.changed().await.unwrap();
+            }
+        })
+        .await
+        .expect("the generation must keep sampling readiness without client traffic");
+    }
+}
+
+async fn recovery_after_watch_closure(close_config: bool, close_ready: bool) {
+    let warnings = Arc::new(AtomicUsize::new(0));
+    let subscriber = tracing_subscriber::registry().with(ClosureWarnings(warnings.clone()));
+    let _default = tracing::subscriber::set_default(subscriber);
+    let mut fixture = GateFixture::new().await;
+    assert!(*fixture.admission.borrow());
+    if close_config {
+        fixture.config_tx.take();
+    }
+    if close_ready {
+        fixture.ready_tx.take();
+    }
+    for _ in 0..4 {
+        tokio::task::yield_now().await;
+    }
+    let expected_warnings = usize::from(close_config) + usize::from(close_ready);
+    fixture.pool.set_ready(true);
+    for _ in 0..4 {
+        tokio::task::yield_now().await;
+    }
+    assert_eq!(fixture.route.snapshot().mode, RelayRouteMode::Direct);
+    fixture.poll().await;
+    fixture.expect_route(RelayRouteMode::Middle).await;
+    fixture.pool.set_ready(false);
+    fixture.poll().await;
+    fixture.expect_route(RelayRouteMode::Direct).await;
+    fixture.pool.set_ready(true);
+    fixture.poll().await;
+    fixture.expect_route(RelayRouteMode::Middle).await;
+    for _ in 0..5 {
+        fixture.poll().await;
+    }
+    assert_eq!(warnings.load(Ordering::Relaxed), expected_warnings);
+    fixture.scope.stop().await;
+    assert!(fixture.admission.has_changed().is_err());
+}
+
+#[tokio::test(start_paused = true)]
+async fn idle_gate_recovers_after_readiness_watch_closes() {
+    recovery_after_watch_closure(false, true).await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn idle_gate_recovers_after_config_watch_closes() {
+    recovery_after_watch_closure(true, false).await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn idle_gate_recovers_after_both_watches_close_without_spinning() {
+    recovery_after_watch_closure(true, true).await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn closed_config_watch_preserves_last_effective_fallback_policy() {
+    let mut fixture = GateFixture::new().await;
+    let mut cfg = fixture.config.as_ref().clone();
+    cfg.general.me2dc_fallback = false;
+    fixture
+        .config_tx
+        .as_ref()
+        .unwrap()
+        .send_replace(Arc::new(cfg));
+    fixture.poll().await;
+    fixture.expect_route(RelayRouteMode::Middle).await;
+    assert!(!*fixture.admission.borrow_and_update());
+    fixture.config_tx.take();
+    fixture.ready_tx.take();
+    fixture.poll().await;
+    fixture.pool.set_ready(true);
+    fixture.poll().await;
+    tokio::time::timeout(Duration::from_secs(1), fixture.admission.changed())
+        .await
+        .expect("restored readiness must open admission after notification loss")
+        .unwrap();
+    assert!(*fixture.admission.borrow_and_update());
+    fixture.pool.set_ready(false);
+    fixture.poll().await;
+    tokio::time::timeout(Duration::from_secs(1), fixture.admission.changed())
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(!*fixture.admission.borrow_and_update());
+    assert_eq!(fixture.route.snapshot().mode, RelayRouteMode::Middle);
+    fixture.scope.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn generation_cancellation_stops_gate_after_notification_loss() {
+    let mut fixture = GateFixture::new().await;
+    fixture.config_tx.take();
+    fixture.ready_tx.take();
+    fixture.poll().await;
+    fixture.pool.set_ready(true);
+    fixture.poll().await;
+    fixture.expect_route(RelayRouteMode::Middle).await;
+    fixture.scope.stop().await;
+    assert!(fixture.admission.has_changed().is_err());
+    let state = fixture.route.snapshot();
+    fixture.pool.set_ready(false);
+    for _ in 0..3 {
+        fixture.poll().await;
+    }
+    assert_eq!(fixture.route.snapshot(), state);
+    assert!(*fixture.admission.borrow());
+}

src/maestro/orchestrator.rs

@@ -375,17 +375,21 @@ pub(super) async fn run_telemt_core(
 
     #[cfg(target_os = "linux")]
     let conntrack_firewall = {
-        let authority = crate::conntrack_control::FirewallAuthority::spawn(&process_control_plane)
-            .map_err(std::io::Error::other)?;
-        if !authority
-            .publish_initial(1, runtime.config.clone(), stats.clone())
-            .await
+        let authority = crate::conntrack_control::FirewallAuthority::spawn(
+            &process_control_plane,
+            &runtime.config,
+        )
+        .map_err(std::io::Error::other)?;
+        if let Some(authority) = &authority
+            && !authority
+                .publish_initial(1, runtime.config.clone(), stats.clone())
+                .await
         {
             warn!(
                 "Initial conntrack firewall reconciliation failed; background retries remain active"
             );
         }
-        Some(authority)
+        authority
     };
     #[cfg(not(target_os = "linux"))]
     let conntrack_firewall = None::<crate::conntrack_control::FirewallAuthority>;

src/service/mod.rs

@@ -104,11 +104,12 @@ pub fn generate_service_file(init_system: InitSystem, opts: &ServiceOptions) ->
         InitSystem::Systemd => generate_systemd_unit(opts),
         InitSystem::OpenRC => generate_openrc_script(opts),
         InitSystem::FreeBSDRc => generate_freebsd_rc_script(opts),
-        InitSystem::Unknown => generate_systemd_unit(opts), // Default to systemd format
+        // Preserve systemd generation when init detection is unavailable.
+        InitSystem::Unknown => generate_systemd_unit(opts),
     }
 }
 
-/// Generates an enhanced systemd unit file.
+/// Generates a hardened systemd unit with Netlink access for optional firewall helpers.
 fn generate_systemd_unit(opts: &ServiceOptions) -> String {
     let user_line = opts.user.map(|u| format!("User={}", u)).unwrap_or_default();
     let group_line = opts
@@ -151,14 +152,14 @@ PrivateDevices=true
 ProtectKernelTunables=true
 ProtectKernelModules=true
 ProtectControlGroups=true
-RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
+RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX AF_NETLINK
 RestrictNamespaces=true
 RestrictRealtime=true
 RestrictSUIDSGID=true
 MemoryDenyWriteExecute=true
 LockPersonality=true
 
-# Allow binding to privileged ports and writing to specific paths
+# Allow privileged port binding, optional firewall helpers, and runtime state writes
 AmbientCapabilities=CAP_NET_BIND_SERVICE CAP_NET_ADMIN
 CapabilityBoundingSet=CAP_NET_BIND_SERVICE CAP_NET_ADMIN
 ReadWritePaths=/etc/telemt /var/run /var/lib/telemt
@@ -233,9 +234,11 @@ fn generate_freebsd_rc_script(opts: &ServiceOptions) -> String {
 # Add the following lines to /etc/rc.conf to enable telemt:
 #
 # telemt_enable="YES"
-# telemt_config="/etc/telemt/config.toml"  # optional
-# telemt_user="telemt"                      # optional
-# telemt_group="telemt"                     # optional
+# The configuration path can be overridden in rc.conf.
+# telemt_config="/etc/telemt/config.toml"
+# The service user and group can be overridden in rc.conf.
+# telemt_user="telemt"
+# telemt_group="telemt"
 #
 
 . /etc/rc.subr
@@ -355,6 +358,28 @@ mod tests {
         assert!(unit.contains("PIDFile="));
     }
 
+    #[test]
+    fn systemd_unit_allows_netlink_without_removing_hardening() {
+        let unit = generate_systemd_unit(&ServiceOptions::default());
+        let families = unit
+            .lines()
+            .find_map(|line| line.strip_prefix("RestrictAddressFamilies="))
+            .unwrap();
+        assert_eq!(
+            families.split_whitespace().collect::<Vec<_>>(),
+            ["AF_INET", "AF_INET6", "AF_UNIX", "AF_NETLINK"],
+        );
+        for directive in [
+            "NoNewPrivileges=true",
+            "ProtectSystem=strict",
+            "RestrictNamespaces=true",
+            "AmbientCapabilities=CAP_NET_BIND_SERVICE CAP_NET_ADMIN",
+            "CapabilityBoundingSet=CAP_NET_BIND_SERVICE CAP_NET_ADMIN",
+        ] {
+            assert!(unit.lines().any(|line| line == directive), "{directive}");
+        }
+    }
+
     #[test]
     fn test_openrc_script_generation() {
         let opts = ServiceOptions::default();

src/transport/middle_proxy/health/recovery.rs

@@ -1,5 +1,10 @@
 use super::*;
 
+// Outage regressions use loopback sockets and explicit family-state snapshots.
+#[cfg(test)]
+mod tests;
+
+/// Reports whether replacing an endpoint could disrupt a bound client.
 pub(super) fn has_bound_clients_on_endpoint(
     writer_ids: &[u64],
     bound_clients_by_writer: &HashMap<u64, usize>,
@@ -9,6 +14,7 @@ pub(super) fn has_bound_clients_on_endpoint(
         .any(|writer_id| bound_clients_by_writer.get(writer_id).copied().unwrap_or(0) > 0)
 }
 
+/// Restores missing single-endpoint coverage under the bounded outage retry policy.
 pub(super) async fn recover_single_endpoint_outage(
     pool: &Arc<MePool>,
     rng: &Arc<SecureRandom>,
@@ -53,13 +59,21 @@ pub(super) async fn recover_single_endpoint_outage(
     pool.stats
         .increment_me_single_endpoint_outage_reconnect_attempt_total();
 
+    let generation = pool.current_generation();
     let bypass_quarantine = pool.single_endpoint_outage_disable_quarantine();
     let attempt_ok = if bypass_quarantine {
         pool.stats
             .increment_me_single_endpoint_quarantine_bypass_total();
         match tokio::time::timeout(
             pool.reconnect_runtime.me_one_timeout,
-            pool.connect_one_for_dc(endpoint, key.0, rng.as_ref()),
+            pool.connect_one_with_generation_contour_for_dc_with_intent(
+                endpoint,
+                rng.as_ref(),
+                generation,
+                WriterContour::Active,
+                key.0,
+                WriterOpenIntent::Coverage,
+            ),
         )
         .await
         {
@@ -88,7 +102,14 @@ pub(super) async fn recover_single_endpoint_outage(
         let one_endpoint = [endpoint];
         match tokio::time::timeout(
             pool.reconnect_runtime.me_one_timeout,
-            pool.connect_endpoints_round_robin(key.0, &one_endpoint, rng.as_ref()),
+            pool.connect_endpoints_round_robin_with_generation_contour(
+                key.0,
+                &one_endpoint,
+                rng.as_ref(),
+                generation,
+                WriterContour::Active,
+                WriterOpenIntent::Coverage,
+            ),
         )
         .await
         {
@@ -142,6 +163,7 @@ pub(super) async fn recover_single_endpoint_outage(
     );
 }
 
+/// Rotates an idle single-endpoint shadow without removing the existing coverage first.
 pub(super) async fn maybe_rotate_single_endpoint_shadow(
     pool: &Arc<MePool>,
     rng: &Arc<SecureRandom>,

src/transport/middle_proxy/health/recovery/tests.rs

@@ -0,0 +1,148 @@
+use std::sync::atomic::Ordering;
+
+use super::*;
+use crate::transport::middle_proxy::admission_test_support::unregistered_writer;
+use crate::transport::middle_proxy::pool_writer_security_tests::make_pool;
+
+async fn single_endpoint_reaches_tcp_with_a_full_cap(bypass_quarantine: bool) {
+    let pool = make_pool().await;
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let endpoint = listener.local_addr().unwrap();
+    let donor = SocketAddr::new(endpoint.ip(), 1);
+    pool.update_proxy_maps(
+        HashMap::from([
+            (2, vec![(endpoint.ip(), endpoint.port())]),
+            (3, vec![(donor.ip(), donor.port())]),
+        ]),
+        None,
+    )
+    .await;
+    pool.floor_runtime
+        .me_adaptive_floor_cpu_cores_override
+        .store(1, Ordering::Relaxed);
+    pool.floor_runtime
+        .me_adaptive_floor_max_active_writers_per_core
+        .store(1, Ordering::Relaxed);
+    pool.floor_runtime
+        .me_adaptive_floor_max_active_writers_global
+        .store(1, Ordering::Relaxed);
+    pool.single_endpoint_runtime
+        .me_single_endpoint_outage_disable_quarantine
+        .store(bypass_quarantine, Ordering::Relaxed);
+    let (writer, _receiver) = unregistered_writer(
+        &pool,
+        1,
+        3,
+        donor,
+        pool.current_generation(),
+        WriterContour::Active,
+    );
+    pool.writers.write().await.push(writer);
+    if bypass_quarantine {
+        pool.endpoint_quarantine
+            .lock()
+            .await
+            .insert(endpoint, Instant::now() + Duration::from_secs(60));
+    }
+    let rng = Arc::new(SecureRandom::new());
+    let key = (2, IpFamily::V4);
+    let mut backoff = HashMap::new();
+    let mut next_attempt = HashMap::new();
+    let semaphore = Arc::new(Semaphore::new(1));
+    let required = pool.required_writers_for_dc(1);
+    // TCP acceptance proves capacity admission; closing the peer deliberately rejects the handshake.
+    tokio::time::timeout(Duration::from_secs(1), async {
+        tokio::join!(
+            recover_single_endpoint_outage(
+                &pool,
+                &rng,
+                key,
+                endpoint,
+                required,
+                &mut backoff,
+                &mut next_attempt,
+                &semaphore,
+            ),
+            async {
+                let (stream, _) = listener.accept().await.unwrap();
+                drop(stream);
+            },
+        );
+    })
+    .await
+    .expect("missing-DC recovery must reach TCP even when another DC fills the cap");
+    assert_eq!(
+        pool.stats
+            .get_me_single_endpoint_outage_reconnect_attempt_total(),
+        1
+    );
+    assert_eq!(
+        pool.stats
+            .get_me_single_endpoint_outage_reconnect_success_total(),
+        0
+    );
+    assert_eq!(
+        pool.stats.get_me_single_endpoint_quarantine_bypass_total(),
+        u64::from(bypass_quarantine)
+    );
+    assert!(next_attempt.contains_key(&key));
+    assert!(backoff.contains_key(&key));
+    assert_eq!(semaphore.available_permits(), 1);
+    assert_eq!(
+        pool.writer_connect_active_reserved.load(Ordering::Acquire),
+        0
+    );
+}
+
+#[tokio::test]
+async fn single_endpoint_quarantine_bypass_recovers_coverage_at_cap() {
+    single_endpoint_reaches_tcp_with_a_full_cap(true).await;
+}
+
+#[tokio::test]
+async fn single_endpoint_round_robin_recovers_coverage_at_cap() {
+    single_endpoint_reaches_tcp_with_a_full_cap(false).await;
+}
+
+#[tokio::test]
+async fn restored_coverage_preserves_family_suppression_hysteresis() {
+    let pool = make_pool().await;
+    let now = MePool::now_epoch_secs();
+    let until = now + 3600;
+    pool.set_family_runtime_state(
+        IpFamily::V4,
+        MeFamilyRuntimeState::Suppressed,
+        now,
+        until,
+        5,
+        0,
+    );
+    super::super::update_family_runtime_state(&pool, IpFamily::V4, false);
+    assert_eq!(
+        pool.family_runtime_state(IpFamily::V4),
+        MeFamilyRuntimeState::Suppressed
+    );
+    assert_eq!(pool.family_suppressed_until_epoch_secs(IpFamily::V4), until);
+    assert_eq!(pool.family_recover_success_streak(IpFamily::V4), 0);
+
+    pool.set_family_runtime_state(
+        IpFamily::V4,
+        MeFamilyRuntimeState::Suppressed,
+        now,
+        now - 1,
+        5,
+        0,
+    );
+    super::super::update_family_runtime_state(&pool, IpFamily::V4, false);
+    assert_eq!(
+        pool.family_runtime_state(IpFamily::V4),
+        MeFamilyRuntimeState::Recovering
+    );
+    assert_eq!(pool.family_recover_success_streak(IpFamily::V4), 1);
+    super::super::update_family_runtime_state(&pool, IpFamily::V4, false);
+    assert_eq!(
+        pool.family_runtime_state(IpFamily::V4),
+        MeFamilyRuntimeState::Healthy
+    );
+    assert_eq!(pool.family_fail_streak(IpFamily::V4), 0);
+}

src/transport/middle_proxy/mod.rs

@@ -49,6 +49,11 @@ mod send;
 mod send_adversarial_tests;
 mod wire;
 
+/// Network-free writer fixtures shared by admission and recovery regressions.
+#[cfg(test)]
+#[path = "tests/admission_test_support.rs"]
+pub(crate) mod admission_test_support;
+
 use bytes::Bytes;
 use tokio::sync::OwnedSemaphorePermit;
 

src/transport/middle_proxy/pool/writer_admission.rs

@@ -61,13 +61,13 @@ impl Drop for WriterOpenReservation<'_> {
 }
 
 impl MePool {
-    /// Computes the authoritative active-writer floor across enabled families and DCs.
+    /// Computes reservation coverage across configured families, including suppressed ones.
     pub(in crate::transport::middle_proxy) async fn active_coverage_required_total(&self) -> usize {
-        let now_epoch_secs = Self::now_epoch_secs();
         let mut required_total = 0usize;
         let endpoint_snapshot = self.endpoint_snapshot.load_full();
 
-        if self.family_enabled_for_drain_coverage(IpFamily::V4, now_epoch_secs) {
+        // Suppression must not remove the capacity needed to recover active coverage.
+        if self.decision.ipv4_me {
             for addrs in endpoint_snapshot.map_v4.values() {
                 let mut endpoints = HashSet::<SocketAddr>::new();
                 for (ip, port) in addrs.iter().copied() {
@@ -79,7 +79,7 @@ impl MePool {
             }
         }
 
-        if self.family_enabled_for_drain_coverage(IpFamily::V6, now_epoch_secs) {
+        if self.decision.ipv6_me {
             for addrs in endpoint_snapshot.map_v6.values() {
                 let mut endpoints = HashSet::<SocketAddr>::new();
                 for (ip, port) in addrs.iter().copied() {

src/transport/middle_proxy/pool_writer.rs

@@ -33,6 +33,10 @@ mod publication;
 mod replacement;
 mod runtime;
 
+// Recovery regressions exercise the serialized production publication boundary.
+#[cfg(test)]
+mod recovery_tests;
+
 struct PreparedWriter<'a> {
     writer: MeWriter,
     tx: mpsc::Sender<WriterCommand>,

src/transport/middle_proxy/pool_writer/publication.rs

@@ -84,7 +84,17 @@ impl MePool {
             crate::network::IpFamily::V6
         };
         let now_epoch_secs = Self::now_epoch_secs();
-        if !self.family_enabled_for_drain_coverage(family, now_epoch_secs) {
+        let configured = match family {
+            crate::network::IpFamily::V4 => self.decision.ipv4_me,
+            crate::network::IpFamily::V6 => self.decision.ipv6_me,
+        };
+        // Recovery coverage must remain publishable during suppression; the serialized
+        // per-group floor check below still bounds it and generation authority is unchanged.
+        let recovery_coverage =
+            contour == WriterContour::Active && intent == WriterOpenIntent::Coverage;
+        if !configured
+            || (self.is_family_temporarily_suppressed(family, now_epoch_secs) && !recovery_coverage)
+        {
             return Err(ProxyError::Proxy(
                 "ME writer family lost publication authority".into(),
             ));

src/transport/middle_proxy/pool_writer/recovery_tests.rs

@@ -0,0 +1,466 @@
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+
+use super::*;
+use crate::network::{IpFamily, probe::NetworkDecision};
+use crate::transport::middle_proxy::admission_test_support::unregistered_writer;
+use crate::transport::middle_proxy::pool::MeFamilyRuntimeState;
+use crate::transport::middle_proxy::pool_writer_security_tests::{
+    make_pool, make_pool_with_decision,
+};
+
+fn cap_at_one(pool: &MePool) {
+    pool.floor_runtime
+        .me_adaptive_floor_cpu_cores_override
+        .store(1, Ordering::Relaxed);
+    pool.floor_runtime
+        .me_adaptive_floor_max_active_writers_per_core
+        .store(1, Ordering::Relaxed);
+    pool.floor_runtime
+        .me_adaptive_floor_max_active_writers_global
+        .store(1, Ordering::Relaxed);
+}
+
+fn suppress(pool: &MePool, family: IpFamily) {
+    let now = MePool::now_epoch_secs();
+    pool.set_family_runtime_state(
+        family,
+        MeFamilyRuntimeState::Suppressed,
+        now,
+        now + 3600,
+        5,
+        0,
+    );
+}
+
+fn prepared<'a>(
+    pool: &'a Arc<MePool>,
+    writer: MeWriter,
+    intent: WriterOpenIntent,
+    reservation: WriterOpenReservation<'a>,
+    task_started: Arc<AtomicBool>,
+) -> PreparedWriter<'a> {
+    let cancel = writer.cancel.clone();
+    PreparedWriter {
+        tx: writer.tx.clone(),
+        byte_budget: writer.byte_budget.clone(),
+        writer,
+        task_registration: pool.lifecycle.try_register().unwrap(),
+        writer_task: Box::pin(async move {
+            task_started.store(true, Ordering::Release);
+            cancel.cancelled().await;
+        }),
+        intent,
+        _open_reservation: reservation,
+    }
+}
+
+#[tokio::test]
+async fn suppressed_active_coverage_publishes_only_the_missing_floor() {
+    let pool = make_pool().await;
+    let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443);
+    pool.update_proxy_maps(HashMap::from([(2, vec![(addr.ip(), addr.port())])]), None)
+        .await;
+    cap_at_one(&pool);
+    suppress(&pool, IpFamily::V4);
+    let required = pool.required_writers_for_dc(1);
+    let mut receivers = Vec::new();
+    for id in 1..=required as u64 {
+        let reservation = pool
+            .reserve_writer_open(WriterContour::Active, WriterOpenIntent::Coverage, 2, addr)
+            .await
+            .expect("missing coverage must have a reservation despite suppression");
+        let (writer, receiver) = unregistered_writer(
+            &pool,
+            id,
+            2,
+            addr,
+            pool.current_generation(),
+            WriterContour::Active,
+        );
+        receivers.push(receiver);
+        pool.publish_connected_writer(prepared(
+            &pool,
+            writer,
+            WriterOpenIntent::Coverage,
+            reservation,
+            Arc::new(AtomicBool::new(false)),
+        ))
+        .await
+        .expect("authenticated coverage must publish without waiting for suppression expiry");
+        assert_eq!(pool.conn_count.load(Ordering::Relaxed), id as usize);
+        assert_eq!(
+            pool.writer_connect_active_reserved.load(Ordering::Acquire),
+            0
+        );
+    }
+    assert!(
+        pool.reserve_writer_open(WriterContour::Active, WriterOpenIntent::Coverage, 2, addr)
+            .await
+            .is_none()
+    );
+    let (extra, _receiver) = unregistered_writer(
+        &pool,
+        99,
+        2,
+        addr,
+        pool.current_generation(),
+        WriterContour::Active,
+    );
+    assert!(
+        pool.authorize_writer_publication_capacity(
+            &extra,
+            WriterContour::Active,
+            WriterOpenIntent::Coverage,
+            &pool.writers.read().await,
+        )
+        .is_err()
+    );
+    assert_eq!(
+        pool.family_runtime_state(IpFamily::V4),
+        MeFamilyRuntimeState::Suppressed
+    );
+    assert!(pool.shutdown_until(Duration::from_secs(1)).await);
+}
+
+#[tokio::test]
+async fn concurrent_suppressed_coverage_publications_cannot_overfill_a_group() {
+    let pool = make_pool().await;
+    let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443);
+    pool.update_proxy_maps(HashMap::from([(2, vec![(addr.ip(), addr.port())])]), None)
+        .await;
+    cap_at_one(&pool);
+    suppress(&pool, IpFamily::V4);
+    let required = pool.required_writers_for_dc(1);
+    let mut receivers = Vec::new();
+    let mut publications = Vec::new();
+    for id in 1..=(required + 1) as u64 {
+        let reservation = pool
+            .reserve_writer_open(WriterContour::Active, WriterOpenIntent::Coverage, 2, addr)
+            .await
+            .unwrap();
+        let (writer, receiver) = unregistered_writer(
+            &pool,
+            id,
+            2,
+            addr,
+            pool.current_generation(),
+            WriterContour::Active,
+        );
+        receivers.push(receiver);
+        publications.push(pool.publish_connected_writer(prepared(
+            &pool,
+            writer,
+            WriterOpenIntent::Coverage,
+            reservation,
+            Arc::new(AtomicBool::new(false)),
+        )));
+    }
+    let guard = pool.writers.write().await;
+    let results = futures::future::join_all(publications);
+    tokio::pin!(results);
+    tokio::select! {
+        biased;
+        results = &mut results => panic!("publication bypassed the writer barrier: {results:?}"),
+        _ = tokio::task::yield_now() => {}
+    }
+    drop(guard);
+    let results = results.await;
+    assert_eq!(
+        results.iter().filter(|result| result.is_ok()).count(),
+        required
+    );
+    assert_eq!(results.iter().filter(|result| result.is_err()).count(), 1);
+    assert_eq!(pool.writers.read().await.len(), required);
+    assert_eq!(pool.conn_count.load(Ordering::Relaxed), required);
+    assert_eq!(
+        pool.registry.writer_idle_since_snapshot().await.len(),
+        required
+    );
+    assert_eq!(
+        pool.writer_connect_active_reserved.load(Ordering::Acquire),
+        0
+    );
+    assert!(pool.shutdown_until(Duration::from_secs(1)).await);
+}
+
+#[tokio::test]
+async fn committed_warm_writer_uses_resolved_active_coverage_authority() {
+    let pool = make_pool().await;
+    let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443);
+    pool.update_proxy_maps(HashMap::from([(2, vec![(addr.ip(), addr.port())])]), None)
+        .await;
+    suppress(&pool, IpFamily::V4);
+    let reservation = pool
+        .reserve_writer_open(WriterContour::Warm, WriterOpenIntent::Coverage, 2, addr)
+        .await
+        .unwrap();
+    let (writer, _receiver) = unregistered_writer(
+        &pool,
+        1,
+        2,
+        addr,
+        pool.current_generation(),
+        WriterContour::Warm,
+    );
+    pool.publish_connected_writer(prepared(
+        &pool,
+        writer,
+        WriterOpenIntent::Coverage,
+        reservation,
+        Arc::new(AtomicBool::new(false)),
+    ))
+    .await
+    .expect("a committed generation must resolve warm coverage to its active role");
+    assert_eq!(
+        pool.writers.read().await[0].contour.load(Ordering::Acquire),
+        WriterContour::Active.as_u8()
+    );
+    assert_eq!(pool.writer_connect_warm_reserved.load(Ordering::Acquire), 0);
+    assert!(pool.shutdown_until(Duration::from_secs(1)).await);
+}
+
+#[tokio::test]
+async fn suppressed_families_keep_their_full_coverage_reservation_budget() {
+    let pool = make_pool_with_decision(NetworkDecision {
+        ipv4_me: true,
+        ipv6_me: true,
+        ..NetworkDecision::default()
+    })
+    .await;
+    let v4 = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443);
+    let v6 = SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 443);
+    pool.update_proxy_maps(
+        HashMap::from([(2, vec![(v4.ip(), v4.port())])]),
+        Some(HashMap::from([(2, vec![(v6.ip(), v6.port())])])),
+    )
+    .await;
+    cap_at_one(&pool);
+    suppress(&pool, IpFamily::V4);
+    suppress(&pool, IpFamily::V6);
+    let required = pool.required_writers_for_dc(1);
+    assert_eq!(pool.active_coverage_required_total().await, required * 4);
+
+    let mut receivers = Vec::new();
+    let mut writers = pool.writers.write().await;
+    for (index, (dc, addr)) in [(2, v4), (-2, v4), (-2, v6)].into_iter().enumerate() {
+        for offset in 0..required {
+            let (writer, receiver) = unregistered_writer(
+                &pool,
+                (index * required + offset + 1) as u64,
+                dc,
+                addr,
+                pool.current_generation(),
+                WriterContour::Active,
+            );
+            writers.push(writer);
+            receivers.push(receiver);
+        }
+    }
+    drop(writers);
+    let reservation = pool
+        .reserve_writer_open(WriterContour::Active, WriterOpenIntent::Coverage, 2, v6)
+        .await
+        .expect("other covered groups must not consume the suppressed family's floor budget");
+    assert_eq!(
+        pool.writer_connect_active_reserved.load(Ordering::Acquire),
+        1
+    );
+    drop(reservation);
+    assert_eq!(
+        pool.writer_connect_active_reserved.load(Ordering::Acquire),
+        0
+    );
+}
+
+#[tokio::test]
+async fn coverage_reservations_remain_bounded_and_release_on_drop() {
+    let pool = make_pool().await;
+    let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443);
+    pool.update_proxy_maps(HashMap::from([(2, vec![(addr.ip(), addr.port())])]), None)
+        .await;
+    cap_at_one(&pool);
+    suppress(&pool, IpFamily::V4);
+    let limit = pool.required_writers_for_dc(1) * 2
+        + pool
+            .reconnect_runtime
+            .me_reconnect_max_concurrent_per_dc
+            .max(1) as usize;
+    let mut reservations = Vec::new();
+    for _ in 0..limit {
+        reservations.push(
+            pool.reserve_writer_open(WriterContour::Active, WriterOpenIntent::Coverage, 2, addr)
+                .await
+                .expect("the configured coverage budget must remain available"),
+        );
+    }
+    assert!(
+        pool.reserve_writer_open(WriterContour::Active, WriterOpenIntent::Coverage, 2, addr,)
+            .await
+            .is_none()
+    );
+    assert_eq!(
+        pool.writer_connect_active_reserved.load(Ordering::Acquire),
+        limit
+    );
+    drop(reservations);
+    assert_eq!(
+        pool.writer_connect_active_reserved.load(Ordering::Acquire),
+        0
+    );
+}
+
+#[tokio::test]
+async fn suppression_exception_does_not_admit_growth_replacement_or_pending_warm() {
+    let pool = make_pool().await;
+    let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443);
+    pool.update_proxy_maps(HashMap::from([(2, vec![(addr.ip(), addr.port())])]), None)
+        .await;
+    suppress(&pool, IpFamily::V4);
+    for (contour, intent) in [
+        (WriterContour::Active, WriterOpenIntent::Normal),
+        (WriterContour::Active, WriterOpenIntent::Replacement),
+        (WriterContour::Warm, WriterOpenIntent::Coverage),
+        (WriterContour::Warm, WriterOpenIntent::Normal),
+    ] {
+        let (writer, _receiver) =
+            unregistered_writer(&pool, 1, 2, addr, pool.current_generation(), contour);
+        assert!(
+            pool.authorize_writer_publication_capacity(&writer, contour, intent, &[])
+                .is_err()
+        );
+    }
+}
+
+#[tokio::test]
+async fn coverage_exception_never_enables_a_disabled_family() {
+    let pool = make_pool().await;
+    let v4 = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443);
+    let v6 = SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 443);
+    pool.update_proxy_maps(
+        HashMap::from([(2, vec![(v4.ip(), v4.port())])]),
+        Some(HashMap::from([(2, vec![(v6.ip(), v6.port())])])),
+    )
+    .await;
+    suppress(&pool, IpFamily::V6);
+    assert_eq!(
+        pool.active_coverage_required_total().await,
+        pool.required_writers_for_dc(1) * 2
+    );
+    let (writer, _receiver) = unregistered_writer(
+        &pool,
+        1,
+        2,
+        v6,
+        pool.current_generation(),
+        WriterContour::Active,
+    );
+    assert!(
+        pool.authorize_writer_publication_capacity(
+            &writer,
+            WriterContour::Active,
+            WriterOpenIntent::Coverage,
+            &[],
+        )
+        .is_err()
+    );
+}
+
+#[tokio::test]
+async fn suppressed_coverage_cannot_publish_a_stale_generation() {
+    let pool = make_pool().await;
+    let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443);
+    pool.update_proxy_maps(HashMap::from([(2, vec![(addr.ip(), addr.port())])]), None)
+        .await;
+    suppress(&pool, IpFamily::V4);
+    let reservation = pool
+        .reserve_writer_open(WriterContour::Active, WriterOpenIntent::Coverage, 2, addr)
+        .await
+        .unwrap();
+    let (writer, _receiver) = unregistered_writer(
+        &pool,
+        1,
+        2,
+        addr,
+        pool.current_generation().saturating_sub(1),
+        WriterContour::Active,
+    );
+    let task_started = Arc::new(AtomicBool::new(false));
+    let error = pool
+        .publish_connected_writer(prepared(
+            &pool,
+            writer,
+            WriterOpenIntent::Coverage,
+            reservation,
+            task_started.clone(),
+        ))
+        .await
+        .unwrap_err();
+    assert!(
+        error
+            .to_string()
+            .contains("generation lost publication authority")
+    );
+    assert!(pool.writers.read().await.is_empty());
+    assert!(pool.registry.writer_idle_since_snapshot().await.is_empty());
+    assert_eq!(pool.conn_count.load(Ordering::Relaxed), 0);
+    assert_eq!(
+        pool.writer_connect_active_reserved.load(Ordering::Acquire),
+        0
+    );
+    assert!(!task_started.load(Ordering::Acquire));
+    assert!(pool.shutdown_until(Duration::from_secs(1)).await);
+}
+
+#[tokio::test]
+async fn suppressed_coverage_rechecks_endpoints_after_publication_waits() {
+    let pool = make_pool().await;
+    let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443);
+    pool.update_proxy_maps(HashMap::from([(2, vec![(addr.ip(), addr.port())])]), None)
+        .await;
+    suppress(&pool, IpFamily::V4);
+    let reservation = pool
+        .reserve_writer_open(WriterContour::Active, WriterOpenIntent::Coverage, 2, addr)
+        .await
+        .unwrap();
+    let (writer, _receiver) = unregistered_writer(
+        &pool,
+        1,
+        2,
+        addr,
+        pool.current_generation(),
+        WriterContour::Active,
+    );
+    let task_started = Arc::new(AtomicBool::new(false));
+    let publication = pool.publish_connected_writer(prepared(
+        &pool,
+        writer,
+        WriterOpenIntent::Coverage,
+        reservation,
+        task_started.clone(),
+    ));
+    tokio::pin!(publication);
+    let guard = pool.writers.write().await;
+    tokio::select! {
+        biased;
+        result = &mut publication => panic!("publication bypassed the writer barrier: {result:?}"),
+        _ = tokio::task::yield_now() => {}
+    }
+    pool.update_proxy_maps(HashMap::from([(2, vec![(addr.ip(), 444)])]), None)
+        .await;
+    drop(guard);
+    let error = publication.await.unwrap_err();
+    assert!(
+        error
+            .to_string()
+            .contains("target changed before publication")
+    );
+    assert!(pool.writers.read().await.is_empty());
+    assert!(pool.registry.writer_idle_since_snapshot().await.is_empty());
+    assert_eq!(pool.conn_count.load(Ordering::Relaxed), 0);
+    assert_eq!(
+        pool.writer_connect_active_reserved.load(Ordering::Acquire),
+        0
+    );
+    assert!(!task_started.load(Ordering::Acquire));
+    assert!(pool.shutdown_until(Duration::from_secs(1)).await);
+}

src/transport/middle_proxy/tests/admission_test_support.rs

@@ -0,0 +1,98 @@
+use std::collections::HashMap;
+use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, Ordering};
+use std::time::Instant;
+
+use tokio::sync::mpsc;
+use tokio_util::sync::CancellationToken;
+
+use super::codec::WriterCommand;
+use super::pool::{MePool, MeWriter, WriterContour};
+use super::pool_writer_security_tests::make_pool;
+
+/// Owns idle writer visibility without spawning ME transports or client sessions.
+pub(crate) struct IdlePoolFixture {
+    pool: Arc<MePool>,
+    draining: Vec<Arc<AtomicBool>>,
+    _receivers: Vec<mpsc::Receiver<WriterCommand>>,
+}
+
+impl IdlePoolFixture {
+    /// Builds both signed DC groups with a controllable readiness quorum.
+    pub(crate) async fn new(ready: bool) -> Self {
+        let pool = make_pool().await;
+        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443);
+        pool.update_proxy_maps(HashMap::from([(2, vec![(addr.ip(), addr.port())])]), None)
+            .await;
+        let mut draining = Vec::new();
+        let mut receivers = Vec::new();
+        {
+            let mut writers = pool.writers.write().await;
+            for (id, dc) in [(1, 2), (2, -2)] {
+                let (writer, receiver) = unregistered_writer(
+                    &pool,
+                    id,
+                    dc,
+                    addr,
+                    pool.current_generation(),
+                    WriterContour::Active,
+                );
+                writer.draining.store(!ready, Ordering::Release);
+                draining.push(writer.draining.clone());
+                receivers.push(receiver);
+                writers.push(writer);
+            }
+        }
+        Self {
+            pool,
+            draining,
+            _receivers: receivers,
+        }
+    }
+
+    /// Returns the pool sampled by the production admission gate.
+    pub(crate) fn pool(&self) -> Arc<MePool> {
+        self.pool.clone()
+    }
+
+    /// Changes the fixture quorum without notifying the admission gate.
+    pub(crate) fn set_ready(&self, ready: bool) {
+        for draining in &self.draining {
+            draining.store(!ready, Ordering::Release);
+        }
+    }
+}
+
+/// Builds a writer whose command receiver remains owned by the test.
+pub(super) fn unregistered_writer(
+    pool: &Arc<MePool>,
+    id: u64,
+    dc: i32,
+    addr: SocketAddr,
+    generation: u64,
+    contour: WriterContour,
+) -> (MeWriter, mpsc::Receiver<WriterCommand>) {
+    let (tx, rx) = mpsc::channel(8);
+    (
+        MeWriter {
+            id,
+            addr,
+            source_ip: addr.ip(),
+            writer_dc: dc,
+            generation,
+            contour: Arc::new(AtomicU8::new(contour.as_u8())),
+            created_at: Instant::now(),
+            tx,
+            byte_budget: pool.new_writer_byte_budget(),
+            cancel: CancellationToken::new(),
+            degraded: Arc::new(AtomicBool::new(false)),
+            rtt_ema_ms_x10: Arc::new(AtomicU32::new(0)),
+            draining: Arc::new(AtomicBool::new(false)),
+            draining_started_at_epoch_secs: Arc::new(AtomicU64::new(0)),
+            drain_deadline_epoch_secs: Arc::new(AtomicU64::new(0)),
+            allow_drain_fallback: Arc::new(AtomicBool::new(false)),
+        },
+        rx,
+    )
+}

src/web/bridge.rs

@@ -1,5 +1,6 @@
 use base64::Engine as _;
 
+use crate::config::WebCarrierMethod;
 use crate::crypto::SecureRandom;
 
 /// Browser security policy for the transient Telegram Desktop bridge page.
@@ -34,6 +35,7 @@ pub(crate) fn render(
     reconnect_grace_secs: u64,
     carrier_probe_coalesce_ms: u64,
     bridge_diagnostics_enabled: bool,
+    carrier_method: WebCarrierMethod,
     rng: &SecureRandom,
 ) -> BridgePage {
     let mut nonce = [0u8; 18];
@@ -124,6 +126,7 @@ pub(crate) fn render(
         .replace("__HOST__", host)
         .replace("__BASE_PREFIX__", base_prefix)
         .replace("__BOOTSTRAP__", bootstrap)
+        .replace("__CARRIER_METHOD__", carrier_method.as_str())
         .replace("__BATCH_LIMIT__", &batch_limit.to_string())
         .replace("__QUEUE_LIMIT__", &queue_limit.to_string())
         .replace("__QUEUE_ITEMS__", &queue_items.to_string())

src/web/bridge/runtime.js

@@ -7,6 +7,8 @@ const responseBody=globalThis.TelemtBridgeResponse;if(!responseBody)throw new Er
 const requestSupport=globalThis.TelemtBridgeRequest;if(!requestSupport)throw new Error('missing request runtime');
 const bufferSupport=globalThis.TelemtBridgeBuffers;if(!bufferSupport)throw new Error('missing buffer runtime');
 const recoverySupport=globalThis.TelemtBridgeRecovery;if(!recoverySupport)throw new Error('missing recovery runtime');
+// Keep the method page-owned so recovery and config rollback cannot change frozen retries.
+const carrierMethod='__CARRIER_METHOD__';
 let negotiationEnabled=__NEGOTIATION_ENABLED__,candidateCount=__CANDIDATE_COUNT__,candidateDeadlines=[__CARRIER_DEADLINES__];
 let longPollMs=__LONG_POLL_SECS__*1000,bridgeRequestMs=__BRIDGE_REQUEST_SECS__*1000,bridgeRetryMs=__BRIDGE_RETRY_SECS__*1000;
 let bridgeRecoveryMs=__BRIDGE_RECOVERY_SECS__*1000,websocketOpenMs=__WEBSOCKET_OPEN_SECS__*1000,reconnectGraceMs=__RECONNECT_GRACE_SECS__*1000;
@@ -226,7 +228,7 @@ async function createSession(epoch){
 async function probeHttp(probe,laneID,epoch){
  try{
   const headers={'X-Up-Seq':'1'},token=sessionToken,controller=attemptController,body=probe.data;if(laneID!==null)headers['X-Lane-ID']=String(laneID);
-  const response=await request('/api/v1/up',options('POST',token,body,headers,controller.signal));
+  const response=await request('/api/v1/up',options(carrierMethod,token,body,headers,controller.signal));
   if(closed||epoch!==attemptEpoch)return
   if(response.status!==204){advanceCarrier('http',epoch);return}
   if(response.headers.get('X-Up-Ack')!=='1'){advanceCarrier('protocol',epoch);return}
@@ -263,14 +265,14 @@ async function runUp(){
    lease=takeBatch(upPending,null);upLease=lease;lease.controller=new AbortController();const sequence=String(upSequence),token=sessionToken;
    for(;;){
     try{
-     const response=await request('/api/v1/up',options('POST',token,lease.body,{'X-Up-Seq':sequence},lease.controller.signal),null,1);
+     const response=await request('/api/v1/up',options(carrierMethod,token,lease.body,{'X-Up-Seq':sequence},lease.controller.signal),null,1);
      if(response.status!==204)throw failure('http','uplink rejected');
      if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','uplink acknowledgement rejected');
      break;
     }catch(error){
      let replayed=false;
      const recovered=await recoverTransport(error,async(signal,remaining)=>{
-      const response=await request('/api/v1/up',options('POST',token,lease.body,{'X-Up-Seq':sequence},signal),remaining,2);
+      const response=await request('/api/v1/up',options(carrierMethod,token,lease.body,{'X-Up-Seq':sequence},signal),remaining,2);
       if(response.status!==204)throw failure('http','uplink replay rejected');
       if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','uplink replay acknowledgement rejected');
       replayed=true;
@@ -343,7 +345,7 @@ async function poll(){
   const token=sessionToken,cursor=downCursor;
   try{
    pollController=new AbortController();
-   const response=await request('/api/v1/down',options('POST',token,null,{'X-Down-Cursor':cursor},pollController.signal),null,1);
+   const response=await request('/api/v1/down',options(carrierMethod,token,null,{'X-Down-Cursor':cursor},pollController.signal),null,1);
    if(closed||sessionToken!==token)return;
    if(response.status===204){status('connected');continue}
    if(response.status!==200)throw failure('http','downlink rejected');
@@ -354,7 +356,7 @@ async function poll(){
   }catch(error){
    if(closed)return;
    const recovered=await recoverTransport(error,async(signal,remaining)=>{
-    const response=await request('/api/v1/down',options('POST',token,null,{'X-Down-Cursor':cursor},signal),remaining,2);
+    const response=await request('/api/v1/down',options(carrierMethod,token,null,{'X-Down-Cursor':cursor},signal),remaining,2);
     if(response.status===204)return;
     if(response.status!==200||!response.body.byteLength||!response.headers.get('X-Down-Cursor'))throw failure('http','downlink replay rejected');
    });
@@ -422,14 +424,14 @@ async function runLaneUp(lane){
    const sequence=String(lane.sequence),laneID=String(lane.id),token=sessionToken;
    for(;;){
     try{
-     const response=await request('/api/v1/up',options('POST',token,lease.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID},lease.controller.signal),null,1);
+     const response=await request('/api/v1/up',options(carrierMethod,token,lease.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID},lease.controller.signal),null,1);
      if(response.status!==204)throw failure('http','lane uplink rejected');
      if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','lane uplink acknowledgement rejected');
      break;
     }catch(error){
      let replayed=false;
      const recovered=await recoverTransport(error,async(signal,remaining)=>{
-      const response=await request('/api/v1/up',options('POST',token,lease.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID},signal),remaining,2);
+      const response=await request('/api/v1/up',options(carrierMethod,token,lease.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID},signal),remaining,2);
       if(response.status!==204)throw failure('http','lane uplink replay rejected');
       if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','lane uplink replay acknowledgement rejected');
       replayed=true;
@@ -451,7 +453,7 @@ async function pollLane(lane){
   while(!closed&&sessionToken&&lanes.get(lane.id)===lane){
    const controller=new AbortController(),laneID=String(lane.id),token=sessionToken,cursor=lane.cursor;lane.controller=controller;
    failedToken=token;failedCursor=cursor;failedLaneID=laneID;
-   const response=await request('/api/v1/down',options('POST',token,null,{'X-Down-Cursor':cursor,'X-Lane-ID':laneID},controller.signal),null,1);
+   const response=await request('/api/v1/down',options(carrierMethod,token,null,{'X-Down-Cursor':cursor,'X-Lane-ID':laneID},controller.signal),null,1);
    if(closed||sessionToken!==token||lanes.get(lane.id)!==lane)return;
    if(response.status===204){
     if(response.headers.get('X-Lane-Closed')==='1'){finishLane(lane,false);return}
@@ -467,7 +469,7 @@ async function pollLane(lane){
   }catch(error){
    if(!closed&&lanes.get(lane.id)===lane){
     const recovered=await recoverTransport(error,async(signal,remaining)=>{
-     const response=await request('/api/v1/down',options('POST',failedToken,null,{'X-Down-Cursor':failedCursor,'X-Lane-ID':failedLaneID},signal),remaining,2);
+     const response=await request('/api/v1/down',options(carrierMethod,failedToken,null,{'X-Down-Cursor':failedCursor,'X-Lane-ID':failedLaneID},signal),remaining,2);
      if(response.status===204)return;
      if(response.status!==200||!response.body.byteLength||!response.headers.get('X-Down-Cursor'))throw failure('http','lane downlink replay rejected');
     });

src/web/bridge/tests.rs

@@ -1,5 +1,73 @@
 use super::*;
 
+#[test]
+fn carrier_method_is_page_owned_and_used_by_every_https_request() {
+    for method in [WebCarrierMethod::Post, WebCarrierMethod::Put] {
+        let page = render(
+            "proxy.example.com",
+            "/telegram/web/",
+            "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
+            2 * 1024 * 1024,
+            32 * 1024 * 1024,
+            16 * 1024,
+            1024,
+            true,
+            4,
+            [3, 5, 8, 12],
+            25,
+            10,
+            90,
+            15,
+            15,
+            120,
+            0,
+            true,
+            method,
+            &SecureRandom::new(),
+        );
+        assert!(!page.body.contains("__"));
+        assert!(
+            page.body
+                .contains(&format!("const carrierMethod='{}';", method.as_str()))
+        );
+        assert_eq!(page.body.matches("carrierMethod=").count(), 1);
+        assert_eq!(page.body.matches("options(carrierMethod,").count(), 9);
+        let requests: Vec<_> = page
+            .body
+            .lines()
+            .filter(|line| {
+                line.contains("request('/api/v1/up'") || line.contains("request('/api/v1/down'")
+            })
+            .collect();
+        assert_eq!(requests.len(), 9);
+        assert!(
+            requests
+                .iter()
+                .all(|line| line.contains("options(carrierMethod,"))
+        );
+        assert_eq!(page.body.matches("options('POST',bootstrap,").count(), 2);
+        assert!(
+            page.body
+                .contains("fetch(relayBase+'/api/v1/diagnostic',{method:'POST'")
+        );
+        assert!(
+            page.body
+                .contains("options('DELETE',token,null,headers,undefined,true)")
+        );
+        assert!(
+            page.body
+                .contains("method:'GET',signal:requestController.signal")
+        );
+        assert!(
+            page.body
+                .contains("exactKeys(value,['v','bootstrap','limits','timeouts','negotiation'])")
+        );
+        assert!(!page.body.contains("policy.carrier_method"));
+        assert!(page.body.contains("port.postMessage({t:'status',state})"));
+        assert!(!page.body.contains("port.postMessage({t:'status',state,"));
+    }
+}
+
 fn render_page(bootstrap: &str, candidate_count: usize) -> BridgePage {
     render(
         "proxy.example.com",
@@ -20,6 +88,7 @@ fn render_page(bootstrap: &str, candidate_count: usize) -> BridgePage {
         120,
         0,
         false,
+        WebCarrierMethod::Post,
         &SecureRandom::new(),
     )
 }
@@ -44,6 +113,7 @@ fn render_diagnostic_page(bootstrap: &str) -> BridgePage {
         120,
         0,
         true,
+        WebCarrierMethod::Post,
         &SecureRandom::new(),
     )
 }
@@ -95,6 +165,7 @@ fn rendered_page_resolves_carriers_against_the_exact_base_path() {
         120,
         0,
         true,
+        WebCarrierMethod::Post,
         &SecureRandom::new(),
     );
 
@@ -148,6 +219,7 @@ fn rendered_page_embeds_the_configured_bridge_timing_policy() {
         119,
         4,
         false,
+        WebCarrierMethod::Post,
         &SecureRandom::new(),
     );
 
@@ -201,6 +273,7 @@ fn disabled_negotiation_does_not_arm_a_carrier_deadline() {
         120,
         0,
         false,
+        WebCarrierMethod::Post,
         &SecureRandom::new(),
     );
     assert!(page.body.contains(

src/web/http.rs

@@ -44,9 +44,9 @@ mod recovery;
 mod response;
 // Session creation and replacement negotiation remain separate from request routing.
 mod session;
-// RFC 6455 upgrade validation and carrier drivers remain isolated from HTTP routing.
 #[cfg(test)]
 mod tests;
+// RFC 6455 upgrade validation and carrier drivers remain isolated from HTTP routing.
 mod websocket;
 // Enabled-debug integration coverage remains separate from carrier behavior tests.
 #[cfg(test)]
@@ -389,6 +389,7 @@ async fn handle_root(
         config.web.timeouts.reconnect_grace_secs,
         config.web.timeouts.carrier_probe_coalesce_ms,
         config.web.debug.bridge_diagnostics_enabled(),
+        config.web.carrier_method,
         &generation.rng,
     );
     let mut response = full_response(StatusCode::OK, Bytes::from(page.body));
@@ -456,14 +457,13 @@ async fn handle_api(
         _ => serve_decoy(request, vhost, true, &runtime).await,
     }
 }
-
 async fn handle_up(
     request: Request<RequestBody>,
     runtime: Arc<WebProcessRuntime>,
     vhost: Arc<WebRuntimeVhost>,
     token_hash: crate::web::manager::TokenHash,
 ) -> HttpResponse {
-    if request.method() != Method::POST || !binary_content_type(&request) {
+    if !matches!(*request.method(), Method::POST | Method::PUT) || !binary_content_type(&request) {
         return serve_decoy(request, vhost, true, &runtime).await;
     }
     let Some(sequence) = canonical_u64_header(&request, "x-up-seq").filter(|value| *value != 0)

src/web/http/carrier_method_tests.rs

@@ -0,0 +1,480 @@
+use super::super::session_policy_tests::{open_keepalive, read_http_response};
+use super::super::*;
+use crate::config::WebCarrierMethod;
+
+fn carrier_request(method: &str, path: &str, token: &str, headers: &str, body: &[u8]) -> Vec<u8> {
+    let mut bytes = format!(
+        "{method} {path} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.60\r\nAuthorization: Bearer {token}\r\n{headers}Content-Length: {}\r\nConnection: close\r\n\r\n",
+        body.len()
+    )
+    .into_bytes();
+    bytes.extend_from_slice(body);
+    bytes
+}
+
+async fn bridge_page(
+    listener: &TcpListener,
+    runtime: &Arc<WebProcessRuntime>,
+    capability: [u8; 32],
+    base: &str,
+) -> Vec<u8> {
+    let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability);
+    request(
+        listener,
+        runtime,
+        format!(
+            "GET {base}?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.60\r\nConnection: close\r\n\r\n"
+        )
+        .into_bytes(),
+    )
+    .await
+}
+
+fn bootstrap_from(page: &[u8]) -> &str {
+    let (headers, body) = split_response(page);
+    assert!(headers.starts_with(b"HTTP/1.1 200"));
+    std::str::from_utf8(body)
+        .unwrap()
+        .split_once("bootstrap=\"")
+        .unwrap()
+        .1
+        .split_once('"')
+        .unwrap()
+        .0
+}
+
+async fn create_session(
+    listener: &TcpListener,
+    runtime: &Arc<WebProcessRuntime>,
+    bootstrap: &str,
+    base: &str,
+) -> String {
+    let hello = frame::encode(FrameType::Hello, 0, &[1]);
+    let bytes = carrier_request(
+        "POST",
+        &format!("{base}api/v1/session"),
+        bootstrap,
+        "Content-Type: application/octet-stream\r\n",
+        &hello,
+    );
+    let response = request(listener, runtime, bytes).await;
+    let (headers, _) = split_response(&response);
+    assert!(headers.starts_with(b"HTTP/1.1 200"));
+    response_header(headers, "x-session-token").to_string()
+}
+
+async fn session_token(
+    listener: &TcpListener,
+    runtime: &Arc<WebProcessRuntime>,
+    capability: [u8; 32],
+    base: &str,
+) -> String {
+    let page = bridge_page(listener, runtime, capability, base).await;
+    create_session(listener, runtime, bootstrap_from(&page), base).await
+}
+
+async fn assert_put_accepted(path: &str) {
+    for (carrier, base, method) in [WebCarrier::Https, WebCarrier::HttpsLanes]
+        .into_iter()
+        .flat_map(|carrier| {
+            ["/", "/telegram/web/"].into_iter().flat_map(move |base| {
+                [WebCarrierMethod::Post, WebCarrierMethod::Put]
+                    .map(|method| (carrier, base, method))
+            })
+        })
+    {
+        let capability = [60; 32];
+        let mut config = runtime_config_with_base(capability, carrier, base);
+        config.web.carrier_method = method;
+        config.web.timeouts.long_poll_secs = 0;
+        let generation = test_runtime_generation(1, config);
+        let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let token = session_token(&listener, &runtime, capability, base).await;
+        let lane = if carrier.uses_lanes() {
+            "X-Lane-ID: 0\r\n"
+        } else {
+            ""
+        };
+        let pong = frame::encode(FrameType::Pong, 0, &[]);
+        let (headers, body) = if path.ends_with("/up") {
+            (
+                format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{lane}"),
+                pong.as_ref(),
+            )
+        } else {
+            (format!("X-Down-Cursor: 0\r\n{lane}"), &[][..])
+        };
+        let path = format!("{base}{}", path.trim_start_matches('/'));
+        let bytes = carrier_request("PUT", &path, &token, &headers, body);
+        let response = request(&listener, &runtime, bytes).await;
+        assert!(
+            response.starts_with(b"HTTP/1.1 204"),
+            "{carrier:?}: {}",
+            String::from_utf8_lossy(&response)
+        );
+        runtime.shutdown().await;
+        generation.stop_sessions().await;
+        generation.stop_background_tasks().await;
+    }
+}
+
+#[tokio::test]
+async fn put_uplink_is_accepted_for_both_https_carriers() {
+    assert_put_accepted("/api/v1/up").await;
+}
+
+#[tokio::test]
+async fn put_downlink_is_accepted_for_both_https_carriers() {
+    assert_put_accepted("/api/v1/down").await;
+}
+
+#[tokio::test]
+async fn carrier_method_mixed_retries_share_sequence_and_cursor_state() {
+    for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] {
+        for method in [WebCarrierMethod::Post, WebCarrierMethod::Put] {
+            let capability = [61; 32];
+            let mut config = runtime_config(capability, carrier);
+            config.web.carrier_method = method;
+            config.web.timeouts.long_poll_secs = 0;
+            // Exhausted admission queues a deterministic CLOSE without starting a relay task.
+            config.web.limits.max_streams_global = 0;
+            let generation = test_runtime_generation(1, config);
+            let runtime =
+                WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let token = session_token(&listener, &runtime, capability, "/").await;
+            for (index, stream_id) in [7, 8].into_iter().enumerate() {
+                let lane = if carrier.uses_lanes() {
+                    format!("X-Lane-ID: {stream_id}\r\n")
+                } else {
+                    String::new()
+                };
+                let sequence = if carrier.uses_lanes() { 1 } else { index + 1 };
+                let cursor = if carrier.uses_lanes() { 0 } else { index };
+                let open = frame::encode(FrameType::Open, stream_id, &[]);
+                let up_headers = format!(
+                    "Content-Type: application/octet-stream\r\nX-Up-Seq: {sequence}\r\n{lane}"
+                );
+                let methods = if index == 0 {
+                    ["POST", "PUT"]
+                } else {
+                    ["PUT", "POST"]
+                };
+                for verb in methods {
+                    let bytes = carrier_request(verb, "/api/v1/up", &token, &up_headers, &open);
+                    let response = request(&listener, &runtime, bytes).await;
+                    let (headers, body) = split_response(&response);
+                    assert!(headers.starts_with(b"HTTP/1.1 204"));
+                    assert_eq!(response_header(headers, "x-up-ack"), sequence.to_string());
+                    assert!(body.is_empty());
+                }
+                for verb in methods {
+                    let headers = format!("X-Down-Cursor: {cursor}\r\n{lane}");
+                    let bytes = carrier_request(verb, "/api/v1/down", &token, &headers, &[]);
+                    let response = request(&listener, &runtime, bytes).await;
+                    let (headers, body) = split_response(&response);
+                    assert!(headers.starts_with(b"HTTP/1.1 200"));
+                    assert_eq!(
+                        response_header(headers, "x-down-cursor"),
+                        (cursor + 1).to_string()
+                    );
+                    assert_eq!(body, frame::encode(FrameType::Close, stream_id, &[]));
+                }
+                let headers = format!("X-Down-Cursor: {}\r\n{lane}", cursor + 1);
+                let bytes = carrier_request("PUT", "/api/v1/down", &token, &headers, &[]);
+                let response = request(&listener, &runtime, bytes).await;
+                assert!(response.starts_with(b"HTTP/1.1 204"));
+            }
+            runtime.shutdown().await;
+            generation.stop_sessions().await;
+            generation.stop_background_tasks().await;
+        }
+    }
+}
+
+#[tokio::test]
+async fn carrier_method_changed_body_retry_keeps_protocol_failure() {
+    for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] {
+        let capability = [62; 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 token = session_token(&listener, &runtime, capability, "/").await;
+        let lane = if carrier.uses_lanes() {
+            "X-Lane-ID: 0\r\n"
+        } else {
+            ""
+        };
+        let headers = format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{lane}");
+        let pong = frame::encode(FrameType::Pong, 0, &[]);
+        let bytes = carrier_request("POST", "/api/v1/up", &token, &headers, &pong);
+        let response = request(&listener, &runtime, bytes).await;
+        assert!(response.starts_with(b"HTTP/1.1 204"));
+        let mut changed = pong.to_vec();
+        changed.extend_from_slice(&pong);
+        let bytes = carrier_request("PUT", "/api/v1/up", &token, &headers, &changed);
+        let response = request(&listener, &runtime, bytes).await;
+        assert_private_decoy(&response);
+        runtime.shutdown().await;
+        generation.stop_sessions().await;
+        generation.stop_background_tasks().await;
+    }
+}
+
+fn assert_private_decoy(response: &[u8]) {
+    let (headers, body) = split_response(response);
+    assert!(headers.starts_with(b"HTTP/1.1 404"));
+    assert_eq!(response_header(headers, "cache-control"), "no-store");
+    assert_eq!(body, b"not found\n");
+    let headers = std::str::from_utf8(headers).unwrap().to_ascii_lowercase();
+    for name in ["x-up-ack:", "x-down-cursor:", "x-session-token:", "allow:"] {
+        assert!(!headers.contains(name));
+    }
+}
+
+#[tokio::test]
+async fn carrier_method_put_preserves_authenticated_request_shape_checks() {
+    for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] {
+        let capability = [63; 32];
+        let base = "/telegram/web/";
+        let generation =
+            test_runtime_generation(1, runtime_config_with_base(capability, carrier, base));
+        let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let token = session_token(&listener, &runtime, capability, base).await;
+        let pong = frame::encode(FrameType::Pong, 0, &[]);
+        let lane = if carrier.uses_lanes() {
+            "X-Lane-ID: 0\r\n"
+        } else {
+            ""
+        };
+        let up = format!("{base}api/v1/up");
+        let down = format!("{base}api/v1/down");
+        let up_headers = format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{lane}");
+        let down_headers = format!("X-Down-Cursor: 0\r\n{lane}");
+        let mut invalid = Vec::new();
+        for media in [
+            "",
+            "Content-Type: text/plain\r\n",
+            "Content-Type: application/octet-stream\r\nContent-Type: application/octet-stream\r\n",
+        ] {
+            let headers = format!("{media}X-Up-Seq: 1\r\n{lane}");
+            invalid.push(carrier_request("PUT", &up, &token, &headers, &pong));
+        }
+        for sequence in ["0", "01", ""] {
+            let sequence = if sequence.is_empty() {
+                String::new()
+            } else {
+                format!("X-Up-Seq: {sequence}\r\n")
+            };
+            let headers = format!("Content-Type: application/octet-stream\r\n{sequence}{lane}");
+            invalid.push(carrier_request("PUT", &up, &token, &headers, &pong));
+        }
+        let invalid_lane = if carrier.uses_lanes() {
+            ""
+        } else {
+            "X-Lane-ID: 0\r\n"
+        };
+        let headers =
+            format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{invalid_lane}");
+        invalid.push(carrier_request("PUT", &up, &token, &headers, &pong));
+        let headers = format!("{down_headers}Content-Type: application/octet-stream\r\n");
+        invalid.push(carrier_request("PUT", &down, &token, &headers, &[]));
+        invalid.push(carrier_request("PUT", &down, &token, &down_headers, &[0]));
+        let headers = format!("X-Down-Cursor: 01\r\n{lane}");
+        invalid.push(carrier_request("PUT", &down, &token, &headers, &[]));
+        for verb in ["GET", "PATCH"] {
+            invalid.push(carrier_request(verb, &up, &token, &up_headers, &pong));
+            invalid.push(carrier_request(verb, &down, &token, &down_headers, &[]));
+        }
+        for path in [
+            "/api/v1/up",
+            "/telegram/web/api/v1/up/",
+            "/telegram/web/api/v1/up?q=1",
+            "/telegram/web/api/v1//up",
+            "/telegram/web/api/v1/%75p",
+        ] {
+            invalid.push(carrier_request("PUT", path, &token, &up_headers, &pong));
+        }
+        let wrong_host = String::from_utf8(carrier_request("PUT", &up, &token, &up_headers, &pong))
+            .unwrap()
+            .replace("Host: proxy.example.com", "Host: other.example.com")
+            .into_bytes();
+        invalid.push(wrong_host);
+        let path = format!("{base}api/v1/session");
+        invalid.push(carrier_request("PUT", &path, &token, &up_headers, &pong));
+        for bytes in invalid {
+            let response = request(&listener, &runtime, bytes).await;
+            assert_private_decoy(&response);
+        }
+        // Rejected shapes must not consume the first sequence or close the valid session.
+        let bytes = carrier_request("PUT", &up, &token, &up_headers, &pong);
+        let response = request(&listener, &runtime, bytes).await;
+        assert!(response.starts_with(b"HTTP/1.1 204"));
+        runtime.shutdown().await;
+        generation.stop_sessions().await;
+        generation.stop_background_tasks().await;
+    }
+}
+
+#[tokio::test]
+async fn carrier_method_post_and_put_reuse_one_private_http_connection() {
+    for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] {
+        let capability = [64; 32];
+        let mut config = runtime_config(capability, carrier);
+        config.web.timeouts.long_poll_secs = 0;
+        config.web.limits.max_streams_global = 0;
+        let generation = test_runtime_generation(1, config);
+        let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let token = session_token(&listener, &runtime, capability, "/").await;
+        let (mut client, cancellation, task) = open_keepalive(&listener, &runtime).await;
+        let lane = if carrier.uses_lanes() {
+            "X-Lane-ID: 7\r\n"
+        } else {
+            ""
+        };
+        let open = frame::encode(FrameType::Open, 7, &[]);
+        let close = frame::encode(FrameType::Close, 7, &[]);
+        for verb in ["POST", "PUT", "PUT", "POST"] {
+            for (path, headers, body, ack_header) in [
+                (
+                    "/api/v1/up",
+                    format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{lane}"),
+                    open.as_ref(),
+                    "x-up-ack",
+                ),
+                (
+                    "/api/v1/down",
+                    format!("X-Down-Cursor: 0\r\n{lane}"),
+                    &[][..],
+                    "x-down-cursor",
+                ),
+            ] {
+                let bytes = carrier_request(verb, path, &token, &headers, body);
+                let bytes = String::from_utf8(bytes)
+                    .unwrap()
+                    .replace("Connection: close\r\n", "Connection: keep-alive\r\n")
+                    .into_bytes();
+                client.write_all(&bytes).await.unwrap();
+                let response = tokio::time::timeout(
+                    std::time::Duration::from_secs(5),
+                    read_http_response(&mut client),
+                )
+                .await
+                .unwrap();
+                let (headers, body) = split_response(&response);
+                assert_eq!(response_header(headers, ack_header), "1");
+                assert!(
+                    !std::str::from_utf8(headers)
+                        .unwrap()
+                        .to_ascii_lowercase()
+                        .contains("connection: close")
+                );
+                if path.ends_with("/up") {
+                    assert!(headers.starts_with(b"HTTP/1.1 204"));
+                    assert!(body.is_empty());
+                } else {
+                    assert!(headers.starts_with(b"HTTP/1.1 200"));
+                    assert_eq!(
+                        response_header(headers, "content-length"),
+                        close.len().to_string()
+                    );
+                    assert_eq!(body, close);
+                }
+            }
+        }
+        cancellation.cancel();
+        task.await.unwrap();
+        runtime.shutdown().await;
+        generation.stop_sessions().await;
+        generation.stop_background_tasks().await;
+    }
+}
+
+#[tokio::test]
+async fn carrier_method_reload_preserves_old_pages_bootstraps_sessions_and_recovery() {
+    for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] {
+        let capability = [65; 32];
+        let mut config = runtime_config(capability, carrier);
+        config.web.limits.max_bootstraps_per_ip = 6;
+        let generation = test_runtime_generation(1, config.clone());
+        let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&generation)));
+        let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime));
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let post_page = bridge_page(&listener, &runtime, capability, "/").await;
+        let old_session =
+            create_session(&listener, &runtime, bootstrap_from(&post_page), "/").await;
+        let unused_post_page = bridge_page(&listener, &runtime, capability, "/").await;
+        config.web.carrier_method = WebCarrierMethod::Put;
+        let put_generation = test_runtime_generation(2, config.clone());
+        active_runtime.store(Arc::clone(&put_generation));
+        let post_session =
+            create_session(&listener, &runtime, bootstrap_from(&unused_post_page), "/").await;
+        let put_page = bridge_page(&listener, &runtime, capability, "/").await;
+        assert!(String::from_utf8_lossy(&put_page).contains("const carrierMethod='PUT';"));
+        config.web.carrier_method = WebCarrierMethod::Post;
+        let rollback_generation = test_runtime_generation(3, config);
+        active_runtime.store(Arc::clone(&rollback_generation));
+        let put_session = create_session(&listener, &runtime, bootstrap_from(&put_page), "/").await;
+        let rollback_page = bridge_page(&listener, &runtime, capability, "/").await;
+        assert!(String::from_utf8_lossy(&rollback_page).contains("const carrierMethod='POST';"));
+        assert!(String::from_utf8_lossy(&post_page).contains("const carrierMethod='POST';"));
+        let lane = if carrier.uses_lanes() {
+            "X-Lane-ID: 0\r\n"
+        } else {
+            ""
+        };
+        let headers = format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{lane}");
+        let pong = frame::encode(FrameType::Pong, 0, &[]);
+        for token in [&old_session, &post_session, &put_session] {
+            for verb in ["PUT", "POST"] {
+                let bytes = carrier_request(verb, "/api/v1/up", token, &headers, &pong);
+                let response = request(&listener, &runtime, bytes).await;
+                assert!(response.starts_with(b"HTTP/1.1 204"));
+            }
+        }
+        let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability);
+        let bytes = carrier_request(
+            "GET",
+            &format!("/?bridge={encoded}"),
+            &put_session,
+            "Accept: application/vnd.telemt.web-recovery+json\r\n",
+            &[],
+        );
+        let recovery = request(&listener, &runtime, bytes).await;
+        let (recovery_headers, body) = split_response(&recovery);
+        assert!(recovery_headers.starts_with(b"HTTP/1.1 200"));
+        assert_eq!(
+            response_header(recovery_headers, "content-type"),
+            "application/vnd.telemt.web-recovery+json"
+        );
+        let document: serde_json::Value = serde_json::from_slice(body).unwrap();
+        let keys: std::collections::BTreeSet<_> = document
+            .as_object()
+            .unwrap()
+            .keys()
+            .map(String::as_str)
+            .collect();
+        assert_eq!(
+            keys,
+            ["v", "bootstrap", "limits", "timeouts", "negotiation"]
+                .into_iter()
+                .collect()
+        );
+        let bootstrap = document["bootstrap"].as_str().unwrap();
+        let recovered = create_session(&listener, &runtime, bootstrap, "/").await;
+        let bytes = carrier_request("PUT", "/api/v1/up", &recovered, &headers, &pong);
+        let response = request(&listener, &runtime, bytes).await;
+        assert!(response.starts_with(b"HTTP/1.1 204"));
+        let bytes = carrier_request("PUT", "/api/v1/up", &put_session, &headers, &pong);
+        let retired = request(&listener, &runtime, bytes).await;
+        assert_private_decoy(&retired);
+        runtime.shutdown().await;
+        for generation in [generation, put_generation, rollback_generation] {
+            generation.stop_sessions().await;
+            generation.stop_background_tasks().await;
+        }
+    }
+}

src/web/http/down.rs

@@ -22,7 +22,7 @@ pub(super) async fn handle_down(
     vhost: Arc<WebRuntimeVhost>,
     token_hash: TokenHash,
 ) -> HttpResponse {
-    if request.method() != hyper::Method::POST
+    if !matches!(*request.method(), hyper::Method::POST | hyper::Method::PUT)
         || request.headers().contains_key(header::CONTENT_TYPE)
     {
         return serve_decoy(request, vhost, true, &runtime).await;

src/web/http/legacy_tests.rs

@@ -1,5 +1,9 @@
 use super::*;
 
+// HTTP method compatibility keeps platform workarounds outside session state.
+#[path = "carrier_method_tests.rs"]
+mod carrier_method_tests;
+
 #[tokio::test]
 async fn windows_restricted_webview_empty_cookie_preserves_the_carrier_flow() {
     for (index, carrier) in [WebCarrier::Https, WebCarrier::HttpsLanes]

src/web/http/session_policy_tests.rs

@@ -1,6 +1,7 @@
 use super::*;
 
-async fn open_keepalive(
+/// Opens one reusable private HTTP connection for integration scenarios.
+pub(super) async fn open_keepalive(
     listener: &TcpListener,
     runtime: &Arc<WebProcessRuntime>,
 ) -> (TcpStream, CancellationToken, tokio::task::JoinHandle<()>) {
@@ -21,7 +22,8 @@ async fn open_keepalive(
     (client.unwrap(), cancellation, task)
 }
 
-async fn read_http_response(client: &mut TcpStream) -> Vec<u8> {
+/// Reads one length-delimited response without waiting for connection closure.
+pub(super) async fn read_http_response(client: &mut TcpStream) -> Vec<u8> {
     let mut response = Vec::new();
     while !response.ends_with(b"\r\n\r\n") {
         assert!(response.len() < 16 * 1024);