Merge pull request #946 from telemt/http-conveyor

Bounded WEB HTTP conveyor with replay-safe streaming downlink added

Cargo.lock

@@ -2780,7 +2780,7 @@ dependencies = [
 
 [[package]]
 name = "telemt"
-version = "3.5.12"
+version = "3.5.13"
 dependencies = [
  "aes",
  "anyhow",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "telemt"
-version = "3.5.12"
+version = "3.5.13"
 edition = "2024"
 rust-version = "1.88"
 

src/api/config_edit.rs

@@ -453,3 +453,7 @@ mod base_path_tests;
 #[cfg(test)]
 #[path = "config_edit/tests.rs"]
 mod tests;
+
+#[cfg(test)]
+#[path = "config_edit/conveyor_tests.rs"]
+mod conveyor_tests;

src/api/config_edit/conveyor_tests.rs

@@ -0,0 +1,27 @@
+use super::*;
+
+#[tokio::test]
+async fn conveyor_api_is_boolean_hot_and_rejects_invalid_patch_atomically() {
+    let directory = tempfile::tempdir().unwrap();
+    let path = directory.path().join("config.toml");
+    std::fs::write(&path, "[web]\nconveyor = true\n").unwrap();
+    for enabled in [false, true] {
+        let active = ProxyConfig::load(&path).unwrap();
+        let patch = serde_json::json!({"web": {"conveyor": enabled}});
+        let mut response = apply_patch_to_path(&path, &patch, None).await.unwrap();
+        let desired = ProxyConfig::load(&path).unwrap();
+        let resolved = reconcile_runtime_effect(&mut response, &active, &desired).unwrap();
+        assert_eq!(resolved.effective.web.conveyor, enabled);
+        assert!(response.runtime_reload_required);
+        assert!(!response.process_restart_required);
+        let (managed, _) = read_managed_config(&path).await.unwrap();
+        assert_eq!(managed["web"]["conveyor"].as_bool(), Some(enabled));
+        let previous = std::fs::read(&path).unwrap();
+        let invalid = serde_json::json!({"web": {"conveyor": "false"}});
+        let error = apply_patch_to_path(&path, &invalid, None)
+            .await
+            .unwrap_err();
+        assert_eq!(error.status, hyper::StatusCode::BAD_REQUEST);
+        assert_eq!(std::fs::read(&path).unwrap(), previous);
+    }
+}

src/api/config_edit/tests.rs

@@ -176,7 +176,8 @@ async fn patch_web_debug_is_hot_and_limits_are_process_deferred() {
     assert!(desired.web.debug.sideband);
 
     let limits_patch: Json = serde_json::json!({
-        "web": {"limits": {"max_http_connections": 2049}}
+        // Extra HTTP heads must fit alongside the default-on conveyor metadata reservation.
+        "web": {"limits": {"max_http_connections": 2049, "memory_envelope_bytes": 1610612736u64}}
     });
     let limits = apply_patch_to_path(&path, &limits_patch, None)
         .await

src/config/hot_reload.rs

@@ -72,3 +72,7 @@ mod tests;
 #[cfg(test)]
 #[path = "hot_reload/carrier_method_tests.rs"]
 mod carrier_method_tests;
+
+#[cfg(test)]
+#[path = "hot_reload/conveyor_tests.rs"]
+mod conveyor_tests;

src/config/hot_reload/conveyor_tests.rs

@@ -0,0 +1,20 @@
+use super::*;
+
+#[test]
+fn conveyor_reload_is_hot_in_both_directions_and_preserves_process_limits() {
+    let mut active = ProxyConfig::default();
+    for enabled in [false, true] {
+        let mut desired = active.clone();
+        desired.web.conveyor = enabled;
+        assert_eq!(classify_config_changes(&active, &desired).changed, ["web"]);
+        assert!(!classify_config_changes(&active, &desired).restart_required);
+        desired.web.limits.max_body_readers += 1;
+        let applied = overlay_hot_fields(&active, &desired);
+        assert_eq!(applied.web.conveyor, enabled);
+        assert_eq!(
+            applied.web.limits.max_body_readers,
+            active.web.limits.max_body_readers
+        );
+        active = applied;
+    }
+}

src/config/load/strict_keys.rs

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

src/config/load/validate_web/memory.rs

@@ -4,7 +4,8 @@ const WEB_DEBUG_RENDERERS: usize = 2;
 const WEB_DEBUG_STATUS_PAGE_BYTES: usize = 8 * 1024 * 1024;
 const WEB_DEBUG_GROUP_SCRATCH_BYTES: usize = 4 * 1024 * 1024;
 const WEB_CARRIER_LEARNING_ENTRY_BYTES: usize = 512;
-const WEB_LANE_STATE_BYTES: usize = 512;
+// Includes the lane plus bounded conveyor slots, Notify ownership, and both hash tables.
+const WEB_LANE_STATE_BYTES: usize = 1024;
 const WEB_OVERLOAD_CONNECTION_BYTES: usize = 4 * 1024;
 // Each profile capability is stored in its vhost and in the global containment table.
 const WEB_CAPABILITY_INDEX_ENTRY_BYTES: usize = 64;

src/config/tests/load_basic_tests/web_tests.rs

@@ -7,6 +7,9 @@ mod base_path_tests;
 #[path = "web_tests/carrier_method_tests.rs"]
 mod carrier_method_tests;
 
+#[path = "web_tests/conveyor_tests.rs"]
+mod conveyor_tests;
+
 const WEB_CONFIG: &str = r#"
 [access.users]
 alice = "000102030405060708090a0b0c0d0e0f"

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

@@ -0,0 +1,44 @@
+use super::*;
+
+#[test]
+fn conveyor_defaults_on_and_roundtrips_without_changing_capabilities() {
+    let default = load_config_from_temp_toml(WEB_CONFIG);
+    assert!(default.web.conveyor);
+    assert!(ProxyConfig::default().web.conveyor);
+    let capabilities = default.web.runtime.as_ref().unwrap().capabilities.clone();
+    for enabled in [false, true] {
+        let configured = WEB_CONFIG.replace(
+            "carrier = \"https-lanes\"",
+            &format!("carrier = \"https-lanes\"\nconveyor = {enabled}"),
+        );
+        let config =
+            load_config_from_temp_toml(&format!("[general]\nconfig_strict = true\n{configured}",));
+        assert_eq!(config.web.conveyor, enabled);
+        assert_eq!(
+            config.web.runtime.as_ref().unwrap().capabilities,
+            capabilities
+        );
+        let json = serde_json::to_value(&config.web).unwrap();
+        assert_eq!(json["conveyor"], enabled);
+        assert_eq!(
+            serde_json::from_value::<WebConfig>(json).unwrap().conveyor,
+            enabled
+        );
+        let toml = toml::to_string(&config.web).unwrap();
+        assert_eq!(
+            toml::from_str::<WebConfig>(&toml).unwrap().conveyor,
+            enabled
+        );
+    }
+}
+
+#[test]
+fn conveyor_rejects_non_boolean_values() {
+    for value in ["\"true\"", "1", "0", "[]", "{}"] {
+        let configured = WEB_CONFIG.replace(
+            "carrier = \"https-lanes\"",
+            &format!("carrier = \"https-lanes\"\nconveyor = {value}"),
+        );
+        assert!(load_config_error_from_temp_toml(&configured).contains("conveyor"));
+    }
+}

src/config/types/web.rs

@@ -416,6 +416,9 @@ pub struct WebConfig {
     /// Page-owned HTTP method for HTTPS uplink and downlink, including retries.
     #[serde(default)]
     pub carrier_method: WebCarrierMethod,
+    /// Enables negotiated bounded HTTP uplink pipelining for new WEB sessions.
+    #[serde(default = "default_web_conveyor")]
+    pub conveyor: bool,
     /// Ordered carriers considered by server-side negotiation before the fallback carrier.
     #[serde(default)]
     pub carriers: WebCarriers,
@@ -475,6 +478,7 @@ impl Default for WebConfig {
             enabled: false,
             carrier: WebCarrier::default(),
             carrier_method: WebCarrierMethod::default(),
+            conveyor: default_web_conveyor(),
             carriers: WebCarriers::default(),
             carrier_learning: default_web_carrier_learning(),
             carrier_negotiation_aggressiveness: WebCarrierNegotiationAggressiveness::default(),

src/config/types/web/defaults.rs

@@ -2,6 +2,11 @@ pub(super) fn default_web_static_index() -> String {
     "index.html".to_string()
 }
 
+/// Enables explicitly negotiated HTTP pipelining unless the operator disables it.
+pub(super) fn default_web_conveyor() -> bool {
+    true
+}
+
 macro_rules! usize_default {
     ($name:ident, $value:expr) => {
         pub(super) fn $name() -> usize {

src/web/bridge.rs

@@ -65,6 +65,8 @@ pub(crate) fn render(
         .replace("__REQUEST_RUNTIME__", REQUEST_RUNTIME)
         .replace("__BUFFER_RUNTIME__", BUFFER_RUNTIME)
         .replace("__RECOVERY_RUNTIME__", RECOVERY_RUNTIME)
+        .replace("__DOWNLINK_RUNTIME__", DOWNLINK_RUNTIME)
+        .replace("__CONVEYOR_RUNTIME__", CONVEYOR_RUNTIME)
         .replace("__RUNTIME__", RUNTIME)
         .replace(
             "__DIAGNOSTIC_BINDING__;\n",
@@ -174,6 +176,8 @@ const RESPONSE_RUNTIME: &str = include_str!("bridge/response.js");
 const REQUEST_RUNTIME: &str = include_str!("bridge/request.js");
 const BUFFER_RUNTIME: &str = include_str!("bridge/buffers.js");
 const RECOVERY_RUNTIME: &str = include_str!("bridge/recovery.js");
+const DOWNLINK_RUNTIME: &str = include_str!("bridge/downlink.js");
+const CONVEYOR_RUNTIME: &str = include_str!("bridge/conveyor.js");
 const RUNTIME: &str = include_str!("bridge/runtime.js");
 
 // Rendered wire-contract tests remain separate from the embedded document.

src/web/bridge/behavior_tests.js

@@ -0,0 +1,265 @@
+'use strict';
+const assert=require('node:assert/strict');
+const fs=require('node:fs');
+const vm=require('node:vm');
+const path=require('node:path');
+
+function renderedPage(){
+ if(process.argv.includes('--stdin'))return fs.readFileSync(0,'utf8');
+ let page=fs.readFileSync(path.join(__dirname,'document.html'),'utf8');
+ const modules={RESPONSE:'response',REQUEST:'request',BUFFER:'buffers',RECOVERY:'recovery',CONVEYOR:'conveyor',DOWNLINK:'downlink'};
+ page=page.replace('__DIAGNOSTIC_RUNTIME__','');
+ for(const [key,file] of Object.entries(modules))if(page.includes('__'+key+'_RUNTIME__'))page=page.replace('__'+key+'_RUNTIME__',()=>fs.readFileSync(path.join(__dirname,file+'.js'),'utf8'));
+ page=page.replace('__RUNTIME__',()=>fs.readFileSync(path.join(__dirname,'runtime.js'),'utf8'));
+ const values={BOOTSTRAP:'A'.repeat(43),HOST:'proxy.example.com',BASE_PREFIX:'',CARRIER_METHOD:'POST',NEGOTIATION_ENABLED:'true',CANDIDATE_COUNT:4,CARRIER_DEADLINES:'3,5,8,12',LONG_POLL_SECS:25,BRIDGE_REQUEST_SECS:10,BRIDGE_RETRY_SECS:90,BRIDGE_RECOVERY_SECS:15,WEBSOCKET_OPEN_SECS:15,RECONNECT_GRACE_SECS:120,CARRIER_PROBE_COALESCE_MS:0,BATCH_LIMIT:2097152,QUEUE_LIMIT:33554432,QUEUE_ITEMS:16384,MAX_STREAMS:1024,STATUS_FUNCTION:"state=>{if(port&&!closed)port.postMessage({t:'status',state})}",HELLO_TIMEOUT_CALLBACK:"()=>fail('timeout')",PAGEHIDE_CALLBACK:'()=>close(true)'};
+ return page.replace(/__([A-Z_]+)__/g,(all,key)=>key.startsWith('DIAGNOSTIC_')?'':String(values[key]??''));
+}
+function frame(type,id=0,payload=[]){
+ const data=new Uint8Array(8+payload.length),view=new DataView(data.buffer);
+ data[0]=type;data[1]=id>>>16;data[2]=id>>>8;data[3]=id;view.setUint32(4,payload.length);data.set(payload,8);return data.buffer;
+}
+function join(...frames){const data=new Uint8Array(frames.reduce((n,f)=>n+f.byteLength,0));let offset=0;for(const f of frames){data.set(new Uint8Array(f),offset);offset+=f.byteLength}return data.buffer}
+async function flush(){for(let i=0;i<40;i++)await Promise.resolve()}
+function environment(page,android=false){
+ let now=1000,nextTimer=1;const timers=new Map(),events=new Map(),requests=[],received=[];
+ const port={onmessage:null,start(){},close(){},postMessage(value,transfer){received.push(structuredClone(value,{transfer:transfer||[]}))}};
+ const context={ArrayBuffer,Uint8Array,DataView,TextDecoder,TextEncoder,URL,Headers,AbortController,ReadableStream,structuredClone,console,
+  location:{hash:android?'#android='+'A'.repeat(43):'',pathname:'/',search:'?bridge=test'},history:{replaceState(){}},parent:{},
+  performance:{now:()=>now},Date:class extends Date{static now(){return now}},document:{visibilityState:'visible',addEventListener(){}},
+  addEventListener:(name,fn)=>events.set(name,fn),setTimeout:(fn,delay)=>{const id=nextTimer++;timers.set(id,{fn,at:now+delay});return id},clearTimeout:id=>timers.delete(id),
+  fetch:(url,options)=>new Promise((resolve,reject)=>{const request={url,options,resolve,reject};requests.push(request);options.signal?.addEventListener('abort',()=>reject(new Error('aborted')),{once:true})}),
+  WebSocket:class {static OPEN=1;static CLOSING=2;constructor(){throw new Error('unexpected websocket')}}
+ };
+ if(android)context.TelegramWebProxy={onmessage:null,postMessage(value){received.push(typeof value==='string'?JSON.parse(value):structuredClone(value))}};
+ vm.createContext(context);
+ for(const match of page.matchAll(/<script\b[^>]*>([\s\S]*?)<\/script>/g))vm.runInContext(match[1],context);
+ if(!android)events.get('message')({source:context.parent,origin:'http://127.0.0.1:12345',data:{t:'tproxy-init',v:1},ports:[port]});
+ const send=data=>android?context.TelegramWebProxy.onmessage({data}):port.onmessage({data});
+ const pending=suffix=>requests.filter(r=>r.url.endsWith(suffix)&&!r.answered);
+ const answer=(request,status,body=null,headers={})=>{request.answered=true;request.resolve({status,headers:new Headers(headers),body:body===null?null:new ReadableStream({start(c){c.enqueue(new Uint8Array(body));c.close()}})})};
+ return {context,send,pending,answer,received,requests,async tick(ms){const until=now+ms;for(;;){const entry=[...timers].filter(([,t])=>t.at<=until).sort((a,b)=>a[1].at-b[1].at)[0];if(!entry)break;now=entry[1].at;timers.delete(entry[0]);entry[1].fn();await flush()}now=until;await flush()},close(){events.get('pagehide')();}};
+}
+async function session(page,carrier='https',window=1,android=false){
+ const env=environment(page,android);env.send(frame(16,0,[1]));await flush();
+ const request=env.pending('/api/v1/session')[0];assert.ok(request,'session requested');
+ const headers={'X-Session-Token':'B'.repeat(43),'X-Down-Cursor':'0','X-Carrier-Mode':carrier,'X-Carrier-Attempt':'1','X-Carrier-Candidate-Count':'4','X-Carrier-Deadline':'12','X-Carrier-State':'provisional'};
+ if(window!==null)headers['X-Telemt-Up-Window']=String(window);
+ env.answer(request,200,frame(17),headers);await flush();return env;
+}
+const tests=[];
+function test(name,run){tests.push({name,run})}
+function ack(env,request){env.answer(request,204,null,{'X-Up-Ack':request.options.headers['X-Up-Seq']})}
+function binary(env){return env.received.filter(value=>value instanceof ArrayBuffer&&new Uint8Array(value)[0]!==17)}
+function streaming(env,request,next='1',length=null){
+ let controller;const body=new ReadableStream({start(value){controller=value}}),headers={'X-Down-Cursor':next};if(length!==null)headers['Content-Length']=String(length);
+ request.answered=true;request.resolve({status:200,headers:new Headers(headers),body});return controller;
+}
+test('already queued OPEN and DATA share the zero-wait startup request',async page=>{
+ const env=await session(page);const body=join(frame(1,1),frame(2,1,[7,8]));env.send(body);await flush();
+ assert.equal(env.pending('/api/v1/up')[0].options.body.byteLength,body.byteLength);env.close();await flush();
+});
+test('negotiated conveyor sends DATA before the OPEN acknowledgement',async page=>{
+ for(const android of [false,true])for(const carrier of ['https','https-lanes']){
+  const env=await session(page,carrier,4,android);env.send(frame(1,1));await flush();env.send(frame(2,1,[9]));await flush();
+  assert.equal(env.pending('/api/v1/up').length,2,carrier+' startup must not wait for ACK1');env.close();await flush();
+ }
+});
+test('unsupported server and disabled conveyor remain serialized',async page=>{
+ for(const window of [null,1]){
+  const env=await session(page,'https',window);env.send(frame(1,1));await flush();env.send(frame(2,1,[9]));await flush();
+  assert.equal(env.pending('/api/v1/up').length,1);env.close();await flush();
+ }
+});
+test('lost ACK1 cannot slide the window after ACK2 through ACK4',async page=>{
+ const env=await session(page,'https',4);env.send(frame(1,1));await flush();
+ for(let i=0;i<6;i++){env.send(frame(2,1,[i]));await flush()}
+ const up=env.pending('/api/v1/up');assert.equal(up.length,4);
+ for(const request of up.slice(1)){ack(env,request);await flush()}
+ assert.equal(env.requests.filter(r=>r.url.endsWith('/api/v1/up')).length,4);
+ ack(env,up[0]);await flush();
+ const next=env.pending('/api/v1/up')[0];assert.equal(next.options.headers['X-Up-Seq'],'5');assert.equal(next.options.headers['X-Telemt-Up-Confirmed'],'4');
+ env.close();await flush();
+});
+test('early downlink cannot deliver native data before a verified commit',async page=>{
+ const env=await session(page,'https',4);env.send(frame(1,1));await flush();
+ const down=env.pending('/api/v1/down')[0];assert.ok(down,'down starts before ACK');const data=frame(2,1,[4]);
+ const stream=streaming(env,down,'1',data.byteLength);stream.enqueue(new Uint8Array(data));await flush();assert.equal(binary(env).length,0);
+ ack(env,env.pending('/api/v1/up')[0]);await flush();assert.equal(binary(env).length,1);
+ assert.equal(env.requests.filter(r=>r.url.endsWith('/api/v1/down')).length,1,'cursor cannot advance before EOF');
+ stream.close();await flush();assert.equal(env.pending('/api/v1/down')[0].options.headers['X-Down-Cursor'],'1');env.close();await flush();
+});
+test('DATA and WINDOW prefixes are delivered once across a mid-body network failure',async page=>{
+ for(const carrier of ['https','https-lanes']){
+  const env=await session(page,carrier,4);env.send(frame(1,1));await flush();ack(env,env.pending('/api/v1/up')[0]);await flush();
+  const first=frame(2,1,[1,2]),credit=frame(4,1,[0,0,0,2]),last=frame(3,1),body=join(first,credit,last);
+  const stream=streaming(env,env.pending('/api/v1/down')[0],'1',body.byteLength);
+  stream.enqueue(new Uint8Array(join(first,credit)));await flush();assert.equal(binary(env).length,2);
+  stream.error(new Error('connection lost'));await flush();
+  const replay=env.pending('/api/v1/down')[0];assert.ok(replay);assert.equal(replay.options.headers['X-Down-Cursor'],'0');
+  env.answer(replay,200,body,{'X-Down-Cursor':'1','Content-Length':String(body.byteLength)});await flush();
+  assert.deepEqual(binary(env).map(data=>new Uint8Array(data)[0]),[2,4,3]);
+  assert.equal(env.pending('/api/v1/down')[0].options.headers['X-Down-Cursor'],'1');
+  assert.equal(env.requests.filter(r=>r.url.endsWith('/api/v1/down')).length,3,'successful replay is consumed without another old-cursor request');env.close();await flush();
+ }
+});
+test('changed replay prefix fails closed without duplicate native frames',async page=>{
+ const env=await session(page,'https',4);env.send(frame(1,1));await flush();ack(env,env.pending('/api/v1/up')[0]);await flush();
+ const data=frame(2,1,[7]),tail=frame(2,1,[8]),body=join(data,tail),stream=streaming(env,env.pending('/api/v1/down')[0],'1',body.byteLength);
+ stream.enqueue(new Uint8Array(data));await flush();stream.error(new Error('lost'));await flush();
+ const changed=join(frame(2,1,[9]),tail);env.answer(env.pending('/api/v1/down')[0],200,changed,{'X-Down-Cursor':'1','Content-Length':String(changed.byteLength)});await flush();
+ assert.equal(binary(env).length,1);assert.ok(env.received.some(value=>value&&value.t==='close'));env.close();await flush();
+});
+test('partial frame boundaries never escape to native and cancellation owns every lease',async page=>{
+ const env=await session(page,'https-lanes',4,true);env.send(frame(1,1));await flush();ack(env,env.pending('/api/v1/up')[0]);await flush();
+ const data=frame(2,1,[1,2,3]),stream=streaming(env,env.pending('/api/v1/down')[0],'1',data.byteLength);
+ stream.enqueue(new Uint8Array(data,0,5));await flush();assert.equal(binary(env).length,0);
+ stream.enqueue(new Uint8Array(data,5));await flush();assert.equal(binary(env).length,1);
+ env.close();await flush();assert.equal(binary(env).length,1);
+});
+test('post-commit lane batching keeps OPEN and DATA from one native message together',async page=>{
+ const env=await session(page,'https-lanes',1);env.send(frame(1,1));await flush();ack(env,env.pending('/api/v1/up')[0]);await flush();
+ const body=join(frame(1,2),frame(2,2,[7]));env.send(body);await flush();
+ assert.equal(env.pending('/api/v1/up')[0].options.body.byteLength,body.byteLength);env.close();await flush();
+});
+
+test('page-owned HTTP method and identity survive frozen retries',async page=>{
+ const expected=/const carrierMethod='(POST|PUT)'/.exec(page)[1],env=await session(page,'https',4);
+ assert.equal(env.requests.find(r=>r.url.endsWith('/api/v1/session')).options.method,'POST');
+ env.send(frame(1,1));await flush();const first=env.pending('/api/v1/up')[0];ack(env,first);await flush();
+ env.send(frame(2,1,[7]));await flush();const original=env.pending('/api/v1/up')[0];
+ original.answered=true;original.reject(new Error('lost'));await flush();const replay=env.pending('/api/v1/up')[0];
+ assert.equal(replay.options.body,original.options.body);assert.deepEqual(replay.options.headers,original.options.headers);
+ assert.equal(replay.options.method,expected);ack(env,replay);await flush();
+ for(const request of env.requests.filter(r=>/\/api\/v1\/(up|down)$/.test(r.url)))assert.equal(request.options.method,expected);
+ env.close();await flush();assert.equal(env.requests.find(r=>r.options.method==='DELETE').options.method,'DELETE');
+});
+
+test('RTT scales time but adds no OPEN ACK barrier or coalescing timer',async page=>{
+ for(const rtt of [40,100,400]){
+  const env=await session(page,'https',4);env.send(frame(1,1));await flush();await env.tick(1);
+  env.send(frame(2,1,[5]));await flush();assert.equal(env.pending('/api/v1/up').length,2);
+  await env.tick(rtt-1);for(const request of env.pending('/api/v1/up'))ack(env,request);await flush();
+  assert.equal(env.received.filter(value=>value instanceof ArrayBuffer&&new Uint8Array(value)[0]===17).length,1);
+  assert.ok(env.received.filter(value=>value&&value.t==='status').every(value=>Object.keys(value).sort().join(',')==='state,t'));
+  env.close();await flush();
+ }
+});
+
+test('simultaneous failed lane polls each consume their own exact replay',async page=>{
+ const env=await session(page,'https-lanes',4);env.send(frame(1,1));await flush();ack(env,env.pending('/api/v1/up')[0]);await flush();
+ env.send(frame(1,2));await flush();ack(env,env.pending('/api/v1/up')[0]);await flush();
+ const streams=[];
+ for(const request of env.pending('/api/v1/down')){
+  const id=Number(request.options.headers['X-Lane-ID']),data=frame(2,id,[id]),body=join(data,frame(4,id,[0,0,0,1]));
+  const response=streaming(env,request,'1',body.byteLength);response.enqueue(new Uint8Array(data));streams.push({id,response,body});
+ }
+ await flush();assert.equal(binary(env).length,2);for(const item of streams)item.response.error(new Error('lost'));await flush();
+ for(let round=0;round<3;round++){
+  for(const request of env.pending('/api/v1/down').filter(r=>r.options.headers['X-Down-Cursor']==='0')){
+   const body=streams.find(item=>String(item.id)===request.options.headers['X-Lane-ID']).body;
+   env.answer(request,200,body,{'X-Down-Cursor':'1','Content-Length':String(body.byteLength)});
+  }await flush();
+ }
+ assert.equal(binary(env).length,4);assert.equal(env.pending('/api/v1/down').length,2);
+ assert.ok(env.pending('/api/v1/down').every(r=>r.options.headers['X-Down-Cursor']==='1'));env.close();await flush();
+});
+
+test('incremental decoder rejects changed cursor, invalid WINDOW and cross-lane frames',async page=>{
+ for(const invalid of ['cursor','credit','lane']){
+  const env=await session(page,'https-lanes',4);env.send(frame(1,1));await flush();ack(env,env.pending('/api/v1/up')[0]);await flush();
+  const body=invalid==='credit'?frame(4,1,[0,0,0,0]):frame(2,invalid==='lane'?2:1,[1]);
+  env.answer(env.pending('/api/v1/down')[0],200,body,{'X-Down-Cursor':invalid==='cursor'?'2':'1','Content-Length':String(body.byteLength)});await flush();
+  assert.equal(binary(env).length,0);assert.ok(env.received.some(value=>value&&value.t==='close'));env.close();await flush();
+ }
+});
+
+test('a replay without Content-Length cannot truncate an already known batch',async page=>{
+ const env=await session(page,'https',4);env.send(frame(1,1));await flush();ack(env,env.pending('/api/v1/up')[0]);await flush();
+ const prefix=frame(2,1,[1]),tail=frame(2,1,[2]),body=join(prefix,tail),response=streaming(env,env.pending('/api/v1/down')[0],'1',body.byteLength);
+ response.enqueue(new Uint8Array(prefix));await flush();response.error(new Error('lost'));await flush();
+ env.answer(env.pending('/api/v1/down')[0],200,prefix,{'X-Down-Cursor':'1'});await flush();
+ assert.equal(binary(env).length,1);assert.ok(env.received.some(value=>value&&value.t==='close'));
+ assert.ok(!env.requests.some(r=>r.url.endsWith('/api/v1/down')&&r.options.headers['X-Down-Cursor']==='1'));env.close();await flush();
+});
+
+test('a pipelined startup flight adopts the recovery budget once another ACK commits',async page=>{
+ const env=await session(page,'https',4);env.send(frame(1,1));await flush();env.send(frame(2,1,[7]));await flush();
+ const [head,tail]=env.pending('/api/v1/up');ack(env,head);await flush();tail.answered=true;tail.reject(new Error('lost'));await flush();
+ const replay=env.pending('/api/v1/up')[0];assert.ok(replay,'post-commit recovery replays immediately instead of staying in the startup backoff');
+ assert.equal(replay.options.body,tail.options.body);ack(env,replay);await flush();env.close();await flush();
+});
+
+test('candidate rollback restores every unacknowledged OPEN and DATA without a second WELCOME',async page=>{
+ const env=await session(page,'https',4);env.send(frame(1,1));await flush();env.send(frame(2,1,[7]));await flush();
+ const retired=env.requests.filter(r=>/\/api\/v1\/(up|down)$/.test(r.url));await env.tick(3001);
+ const next=env.pending('/api/v1/session')[0];assert.ok(next,'candidate fallback creates a new session');
+ env.answer(next,200,frame(17),{'X-Session-Token':'C'.repeat(43),'X-Down-Cursor':'0','X-Carrier-Mode':'https-lanes','X-Carrier-Attempt':'2','X-Carrier-Candidate-Count':'4','X-Carrier-Deadline':'12','X-Carrier-State':'provisional','X-Telemt-Up-Window':'4'});await flush();
+ const first=env.requests.find(r=>r.url.endsWith('/api/v1/up')&&r.options.headers.Authorization==='Bearer '+'C'.repeat(43));
+ assert.ok(first);assert.deepEqual(new Uint8Array(first.options.body),new Uint8Array(join(frame(1,1),frame(2,1,[7]))));
+ assert.ok(retired.every(r=>r.options.signal.aborted));assert.equal(env.received.filter(value=>value instanceof ArrayBuffer&&new Uint8Array(value)[0]===17).length,1);
+ ack(env,first);await flush();env.close();await flush();
+});
+
+test('downlink prefix reservations are bounded, cancellable and released on retirement',async page=>{
+ const env=environment(page),budget=env.context.TelemtBridgeDownlink.budget(()=>16),first=await budget.acquire(16);
+ const controller=new AbortController();let acquired=false;
+ const parked=budget.acquire(8,controller.signal).then(release=>{acquired=true;release()},()=>{});await flush();assert.equal(acquired,false);
+ controller.abort();await parked;first();assert.equal(budget.used(),0);
+ const release=await budget.acquire(16);release();release();assert.equal(budget.used(),0);env.close();await flush();
+});
+
+test('two interrupted and differently chunked replays never duplicate DATA WINDOW or CLOSE',async page=>{
+ const env=await session(page,'https',4);env.send(frame(1,1));await flush();ack(env,env.pending('/api/v1/up')[0]);await flush();
+ const data=frame(2,1,[1]),credit=frame(4,1,[0,0,0,1]),close=frame(3,1),body=join(data,credit,close);
+ const first=streaming(env,env.pending('/api/v1/down')[0],'1',body.byteLength);
+ first.enqueue(new Uint8Array(data));await flush();first.error(new Error('lost once'));await flush();
+ const second=streaming(env,env.pending('/api/v1/down')[0],'1',body.byteLength);
+ const prefix=new Uint8Array(join(data,credit));second.enqueue(prefix.subarray(0,3));second.enqueue(prefix.subarray(3));await flush();
+ assert.deepEqual(binary(env).map(value=>new Uint8Array(value)[0]),[2,4]);second.error(new Error('lost twice'));await flush();await env.tick(1000);
+ const last=streaming(env,env.pending('/api/v1/down')[0],'1',body.byteLength);
+ for(const byte of new Uint8Array(body))last.enqueue(new Uint8Array([byte]));last.close();
+ for(let i=0;i<20&&(!env.pending('/api/v1/down').length);i++)await flush();
+ assert.deepEqual(binary(env).map(value=>new Uint8Array(value)[0]),[2,4,3]);assert.equal(env.pending('/api/v1/down')[0].options.headers['X-Down-Cursor'],'1');
+ env.close();await flush();
+});
+
+test('multiple precommit lanes survive fallback without losing stream ordering',async page=>{
+ const env=await session(page,'https-lanes',4);env.send(join(frame(1,1),frame(1,2)));await flush();
+ env.send(join(frame(2,2,[2]),frame(2,1,[1])));await flush();await env.tick(3001);
+ const next=env.pending('/api/v1/session')[0];env.answer(next,200,frame(17),{'X-Session-Token':'C'.repeat(43),'X-Down-Cursor':'0','X-Carrier-Mode':'https','X-Carrier-Attempt':'2','X-Carrier-Candidate-Count':'4','X-Carrier-Deadline':'12','X-Carrier-State':'provisional','X-Telemt-Up-Window':'4'});await flush();
+ const requests=env.requests.filter(r=>r.url.endsWith('/api/v1/up')&&r.options.headers.Authorization==='Bearer '+'C'.repeat(43));
+ const frames=[];for(const request of requests){const bytes=new Uint8Array(request.options.body);for(let offset=0;offset<bytes.length;){const view=new DataView(bytes.buffer,offset);frames.push([bytes[offset],bytes[offset+3]]);offset+=8+view.getUint32(4);}}
+ for(const id of [1,2])assert.deepEqual(frames.filter(value=>value[1]===id),[[1,id],[2,id]]);
+ assert.equal(frames.length,4);env.close();await flush();
+});
+
+test('receiver retirement releases shared prefix capacity without exposing a partial frame',async page=>{
+ const env=environment(page),module=env.context.TelemtBridgeDownlink,budget=module.budget(()=>18),received=[];
+ function receiver(id){
+  let cursor='0',stream;const controller=new AbortController(),response={status:200,headers:new Headers({'X-Down-Cursor':'1','Content-Length':'18'}),body:new ReadableStream({start(value){stream=value}})};
+  const owner=module.create({failure:(reason,message)=>Object.assign(new Error(message),{telemtReason:reason}),alive:()=>true,cursor:()=>cursor,lane:id,limit:()=>18,budget,cancel(){},ready:()=>Promise.resolve(),deliver:data=>received.push(new Uint8Array(data)),advance:value=>{cursor=value},finished(){}});
+  const result=owner.read(response,controller.signal).catch(()=>{});return {owner,result,stream,controller,cursor:()=>cursor};
+ }
+ const a=receiver(1);a.stream.enqueue(new Uint8Array(frame(2,1,[1])));await flush();
+ const b=receiver(2);b.stream.enqueue(new Uint8Array(join(frame(2,2,[2]),frame(2,2,[3]))));b.stream.close();await flush();
+ assert.equal(budget.used(),18);assert.equal(received.length,1);assert.equal(b.cursor(),'0');
+ a.owner.close();a.controller.abort();a.stream.error(new Error('retired'));await a.result;await b.result;
+ assert.equal(received.length,3);assert.equal(b.cursor(),'1');assert.equal(budget.used(),0);
+ b.owner.close();env.close();await flush();
+});
+
+test('native close cancels every HTTP request and pagehide cannot repeat cleanup',async page=>{
+ for(const android of [false,true])for(const carrier of ['https','https-lanes']){
+  const env=await session(page,carrier,4,android);env.send(frame(1,1));await flush();env.send(frame(2,1,[7]));await flush();
+  const up=env.pending('/api/v1/up'),down=env.pending('/api/v1/down'),requests=up.concat(down);
+  assert.equal(up.length,2);assert.equal(down.length,1);assert.ok(requests.every(request=>!request.options.signal.aborted));
+  env.send({t:'close'});await flush();
+  assert.ok(requests.every(request=>request.options.signal.aborted));
+  assert.equal(env.requests.filter(request=>request.options.method==='DELETE').length,1);
+  const count=env.requests.length;env.close();await env.tick(120000);
+  assert.equal(env.requests.filter(request=>request.options.method==='DELETE').length,1);
+  assert.equal(env.requests.length,count);
+ }
+});
+
+(async()=>{const page=renderedPage();let failed=0;for(const {name,run} of tests){let timer;try{await Promise.race([run(page),new Promise((_,reject)=>{timer=setTimeout(()=>reject(new Error('test made no bounded progress')),5000)})]);console.log('ok - '+name)}catch(error){failed++;console.error('not ok - '+name+'\n'+error.stack)}finally{clearTimeout(timer)}}if(failed)process.exitCode=1})();

src/web/bridge/behavior_tests.rs

@@ -0,0 +1,26 @@
+use std::io::Write;
+use std::process::{Command, Stdio};
+
+/// Exercises the actual generated page with deterministic network and native-boundary doubles.
+pub(super) fn run(page: &super::BridgePage) {
+    let mut child = Command::new("node")
+        .args(["-e", include_str!("behavior_tests.js"), "--", "--stdin"])
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .expect("Node.js 22+ is required for executable WEB bridge regressions");
+    child
+        .stdin
+        .take()
+        .unwrap()
+        .write_all(page.body.as_bytes())
+        .unwrap();
+    let output = child.wait_with_output().unwrap();
+    assert!(
+        output.status.success(),
+        "bridge regressions failed:\n{}\n{}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+}

src/web/bridge/conveyor.js

@@ -0,0 +1,106 @@
+(()=>{'use strict';
+// HTTP request ownership is separate from the native WINDOW credit protocol.
+function create(settings){
+ const channels=new Map(),window=settings.window,token=settings.token,downBudget=globalThis.TelemtBridgeDownlink.budget(settings.queueLimit);
+ let closed=false,order=0,ready=settings.committed(),resolveReady;
+ const readyPromise=new Promise(resolve=>{resolveReady=resolve});if(ready)resolveReady();
+ const alive=channel=>!closed&&!channel.closed&&settings.alive(channel.lane);
+ const markReady=()=>{if(!ready){ready=true;resolveReady();settings.commit()}};
+ function waitReady(signal){
+  if(ready)return Promise.resolve();
+  return new Promise((resolve,reject)=>{const abort=()=>reject(new Error('carrier retired'));if(signal&&signal.aborted){abort();return}if(signal)signal.addEventListener('abort',abort,{once:true});readyPromise.then(()=>{if(signal)signal.removeEventListener('abort',abort);resolve()})});
+ }
+ function get(id,type){
+  let channel=channels.get(id);if(channel)return channel;
+  const lane=id===null?null:settings.lane(id,type);if(id!==null&&!lane)return null;
+  channel={id,lane,pending:[],flights:new Map(),acks:new Set(),next:1,confirmed:0,cursor:'0',controller:null,polling:false,closed:false};
+  channel.receiver=globalThis.TelemtBridgeDownlink.create({
+   failure:settings.failure,alive:()=>alive(channel),cursor:()=>channel.cursor,lane:id,limit:settings.batchLimit,budget:downBudget,
+   cancel:settings.cancel,ready:waitReady,deliver:settings.deliver,advance:next=>{channel.cursor=next},finished:()=>{if(lane)settings.finish(lane,false)}
+  });channels.set(id,channel);return channel;
+ }
+ function enqueue(data,defer){
+  if(closed)return;
+  const entries=settings.lanes?settings.buffers.splitFrames(data):[{id:null,type:0,data}],touched=new Set();
+  for(const entry of entries){
+   const channel=get(entry.id,entry.type);if(!channel)continue;
+   if(!settings.buffers.reserve(entry.data,channel.lane))throw settings.failure('capacity','carrier queue full');
+   channel.pending.push(entry.data);touched.add(channel);
+  }
+  if(!defer)for(const channel of touched)pump(channel);
+ }
+ function checkAck(response,sequence){
+  if(response.status!==204)throw settings.failure('http','uplink rejected');
+  if(response.headers.get('X-Up-Ack')!==String(sequence))throw settings.failure('protocol','uplink acknowledgement rejected');
+ }
+ function acknowledge(channel,sequence,lease){
+  if(!alive(channel)||channel.flights.get(sequence)!==lease)return;
+  settings.buffers.settleBatch(lease);channel.flights.delete(sequence);channel.acks.add(sequence);
+  while(channel.acks.delete(channel.confirmed+1))channel.confirmed++;
+  markReady();settings.traffic(lease.total,0);pump(channel);
+ }
+ async function send(channel,sequence,lease){
+  const headers={'X-Up-Seq':String(sequence)};if(channel.id!==null)headers['X-Lane-ID']=String(channel.id);
+  if(window>1)headers['X-Telemt-Up-Confirmed']=String(channel.confirmed);
+  const frozen=settings.options(settings.method,token,lease.body,headers,lease.controller.signal);
+  try{
+   while(alive(channel)&&!lease.cancelled){
+    try{const response=await settings.request('/api/v1/up',frozen,null,()=>ready?1:9);if(!alive(channel))return;checkAck(response,sequence);acknowledge(channel,sequence,lease);return}
+    catch(error){
+     if(!alive(channel)||lease.cancelled)return;
+     if(!ready){settings.failed(error);return}
+     let replayed=false;
+     const recovered=await settings.recover(error,async(signal,remaining)=>{
+      const response=await settings.request('/api/v1/up',Object.assign({},frozen,{signal}),remaining,2);
+      if(!alive(channel)||lease.cancelled)return;checkAck(response,sequence);replayed=true;acknowledge(channel,sequence,lease);
+     });
+     if(!recovered||!alive(channel)||replayed)return;
+    }
+   }
+  }catch(error){if(alive(channel)&&!lease.cancelled)settings.failed(error)}
+ }
+ function pump(channel){
+  while(alive(channel)&&channel.pending.length&&channel.next<=channel.confirmed+window){
+   const lease=settings.buffers.takeBatch(channel.pending,channel.lane),sequence=channel.next++;
+   lease.controller=new AbortController();lease.order=order++;channel.flights.set(sequence,lease);send(channel,sequence,lease);
+   if(!channel.polling)poll(channel);
+  }
+ }
+ async function poll(channel){
+  if(channel.polling||!alive(channel))return;channel.polling=true;
+  try{
+   while(alive(channel)){
+    const cursor=channel.cursor,headers={'X-Down-Cursor':cursor};if(channel.id!==null)headers['X-Lane-ID']=String(channel.id);
+    channel.controller=new AbortController();const frozen=settings.options(settings.method,token,null,headers,channel.controller.signal);
+    try{
+     const response=await settings.request('/api/v1/down',frozen,null,1,channel.receiver.read);
+     if(!alive(channel))return;if(response.status!==200&&response.status!==204)throw settings.failure('http','downlink rejected');
+     if(ready)settings.connected();
+    }catch(error){
+     if(!alive(channel))return;
+     if(!ready){settings.failed(error);return}
+     const recovered=await settings.recover(error,async(signal,remaining)=>{
+      const response=await settings.request('/api/v1/down',Object.assign({},frozen,{signal}),remaining,2,channel.receiver.read);
+      if(response.status!==200&&response.status!==204)throw settings.failure('http','downlink replay rejected');
+     });
+     if(!recovered||!alive(channel))return;
+    }
+   }
+  }catch(error){if(alive(channel))settings.failed(error)}
+  finally{channel.polling=false;channel.controller=null}
+ }
+ function retire(channel,restore){
+  channel.closed=true;if(channel.controller)channel.controller.abort();channel.receiver.close();
+  const saved=[];
+  for(const lease of channel.flights.values()){if(restore)saved.push({order:lease.order,data:lease.body});settings.buffers.cancelBatch(lease);}
+  channel.flights.clear();channel.acks.clear();
+  for(const data of channel.pending)if(restore)saved.push({order:order++,data});
+  settings.buffers.releasePending(channel.pending,channel.lane);return saved;
+ }
+ return {enqueue,flush(){for(const channel of channels.values())pump(channel)},finish(lane){const channel=channels.get(lane.id);if(channel){channels.delete(lane.id);retire(channel,false)}},close(restore){
+  if(closed)return [];closed=true;const saved=[];for(const channel of channels.values())saved.push(...retire(channel,restore));channels.clear();
+  return saved.sort((a,b)=>a.order-b.order).map(entry=>entry.data);
+ }};
+}
+globalThis.TelemtBridgeConveyor=Object.freeze({create});
+})();

src/web/bridge/document.html

@@ -20,6 +20,12 @@ __BUFFER_RUNTIME__
 __RECOVERY_RUNTIME__
 </script>
 <script nonce="__NONCE__">
+__DOWNLINK_RUNTIME__
+</script>
+<script nonce="__NONCE__">
+__CONVEYOR_RUNTIME__
+</script>
+<script nonce="__NONCE__">
 __RUNTIME__
 </script>
 </body>

src/web/bridge/downlink.js

@@ -0,0 +1,86 @@
+(()=>{'use strict';
+// Prefix ownership survives HTTP retries, but never a session or lane replacement.
+function budget(limit){
+ let used=0;const waiters=[];
+ function wake(){for(const next of waiters.slice())next()}
+ function acquire(size,signal){
+  if(size>limit())return Promise.reject(new Error('downlink budget exceeded'));
+  return new Promise((resolve,reject)=>{
+   const remove=()=>{const at=waiters.indexOf(tryAcquire);if(at>=0)waiters.splice(at,1);if(signal)signal.removeEventListener('abort',abort)};
+   const abort=()=>{remove();reject(new Error('downlink aborted'))};
+   const tryAcquire=()=>{if(signal&&signal.aborted){abort();return}if(size<=limit()-used){remove();used+=size;let released=false;resolve(()=>{if(!released){released=true;used-=size;wake()}})}};
+   waiters.push(tryAcquire);if(signal)signal.addEventListener('abort',abort,{once:true});tryAcquire();
+  });
+ }
+ return {acquire,used:()=>used};
+}
+function create(settings){
+ let transaction=null,closed=false;
+ const protocol=message=>settings.failure('protocol',message);
+ const current=signal=>!closed&&settings.alive()&&!(signal&&signal.aborted);
+ function dispose(){if(transaction){transaction.release();transaction=null}}
+ function shape(type,id,size){
+  if(type===2)return id!==0&&size>0&&size<=1048576;
+  if(type===3)return id!==0&&size===0;
+  if(type===4)return id!==0&&size===4;
+  if(type===5)return id===0&&size<=64;
+  return type===31&&id===0&&size===0;
+ }
+ async function read(response,signal){
+  if(!current(signal))throw new Error('downlink retired');
+  if(response.status===204){
+   settings.cancel(response);if(transaction)throw protocol('missing replay batch');
+   const cursor=response.headers.get('X-Down-Cursor');
+   if(cursor!==null&&cursor!==settings.cursor())throw protocol('changed empty cursor');
+   if(response.headers.get('X-Lane-Closed')==='1')settings.finished();
+   return new ArrayBuffer(0);
+  }
+  const next=response.headers.get('X-Down-Cursor')||'',base=settings.cursor();
+  if(!/^[1-9]\d*$/.test(next)||BigInt(next)!==BigInt(base)+1n||BigInt(next)>18446744073709551615n)throw protocol('invalid downlink cursor');
+  const length=response.headers.get('Content-Length');let declared=null;
+  if(length!==null){if(!/^(0|[1-9]\d*)$/.test(length))throw protocol('invalid response length');declared=Number(length);if(!Number.isSafeInteger(declared)||declared===0||declared>settings.limit())throw protocol('response body overflow');}
+  if(transaction&&(transaction.base!==base||transaction.next!==next||(transaction.declared!==null&&declared!==null&&transaction.declared!==declared)))throw protocol('changed replay metadata');
+  if(!transaction){
+   const capacity=declared===null?settings.limit():declared,release=await settings.budget.acquire(capacity,signal);
+   if(!current(signal)){release();throw new Error('downlink retired');}
+   transaction={base,next,declared,bytes:new Uint8Array(capacity),retained:0,delivered:0,release};
+  }
+  const owner=transaction;
+  if(owner.declared===null&&declared!==null)owner.declared=declared;
+  await settings.ready(signal);
+  if(!current(signal)||transaction!==owner)throw new Error('downlink retired');
+  if(!response.body)throw protocol('missing response body');
+  const reader=response.body.getReader();let position=0,offset=0,frames=0,chunks=0,complete=false;
+  try{
+   for(;;){
+    let part;try{part=await reader.read();}catch(error){throw settings.failure('network','response stream interrupted');}
+    if(!current(signal)||transaction!==owner)throw new Error('downlink retired');
+    if(part.done)break;
+    const data=part.value;if(!(data instanceof Uint8Array)||++chunks>65536||data.byteLength>owner.bytes.byteLength-position)throw protocol('response body overflow');
+    const compared=Math.min(data.byteLength,Math.max(0,owner.retained-position));
+    for(let i=0;i<compared;i++)if(data[i]!==owner.bytes[position+i])throw protocol('changed replay prefix');
+    owner.bytes.set(data.subarray(compared),position+compared);position+=data.byteLength;owner.retained=Math.max(owner.retained,position);
+    while(position-offset>=8){
+     const view=new DataView(owner.bytes.buffer,offset),type=view.getUint8(0),id=(view.getUint8(1)<<16)|(view.getUint8(2)<<8)|view.getUint8(3),size=view.getUint32(4),end=offset+8+size;
+     if(!shape(type,id,size)||(settings.lane!==null&&id!==settings.lane)||end>owner.bytes.byteLength)throw protocol('invalid downlink frame');
+     if(end>position)break;if(++frames>4096)throw protocol('too many downlink frames');
+     if(type===4&&view.getUint32(8)===0)throw protocol('invalid WINDOW credit');
+     if(end>owner.delivered){
+      if(offset!==owner.delivered)throw protocol('invalid replay boundary');
+      const frame=owner.bytes.slice(offset,end).buffer;
+      settings.deliver(frame);owner.delivered=end;
+     }
+     offset=end;
+    }
+   }
+   if(!position||offset!==position||position<owner.retained||(owner.declared!==null&&position!==owner.declared))throw protocol('incomplete downlink batch');
+   settings.advance(next);complete=true;dispose();return new ArrayBuffer(0);
+  }finally{
+   if(!complete)try{const cancelled=reader.cancel();if(cancelled)cancelled.catch(()=>{});}catch(error){}
+   reader.releaseLock();
+  }
+ }
+ return {read,close(){closed=true;dispose()}};
+}
+globalThis.TelemtBridgeDownlink=Object.freeze({create,budget});
+})();

src/web/bridge/request.js

@@ -26,12 +26,13 @@ function create(settings){
   if(status===204&&(path==='/api/v1/up'||path==='/api/v1/down'))return {limit:0,exact:true,reason:'protocol'};
   return {limit:0,exact:true,reason:'http'};
  }
- async function send(path,frozenOptions,remainingBudget,maxAttempts){
+ async function send(path,frozenOptions,remainingBudget,maxAttempts,receiver){
   let delay=250,attempt=0,lastReason='network';maxAttempts=maxAttempts||9;
+  const attempts=()=>typeof maxAttempts==='function'?maxAttempts():maxAttempts;
   const initialBudget=remainingBudget?Math.min(settings.retryMs(),remainingBudget()):settings.retryMs();
   const deadline=Date.now()+Math.max(0,initialBudget),external=frozenOptions.signal;
   const attemptLimit=path==='/api/v1/down'?settings.longPollMs()+settings.requestMs():settings.requestMs();
-  while(attempt<maxAttempts){
+  while(attempt<attempts()){
    if(settings.closed()||(external&&external.aborted))throw new Error('request aborted');
    const remaining=Math.min(deadline-Date.now(),remainingBudget?remainingBudget():Infinity);if(remaining<=0)break;attempt++;
    const controller=new AbortController(),abort=()=>controller.abort();let timedOut=false;
@@ -45,12 +46,12 @@ function create(settings){
      lastReason='http';wait=retryAfterMs(fetched);settings.cancel(fetched);
     }else{
      const policy=responsePolicy(path,fetched.status);let body;
-     try{body=await settings.read(fetched,policy.limit,policy.exact,controller.signal)}
+     try{body=receiver&&(fetched.status===200||fetched.status===204)?await receiver(fetched,controller.signal):await settings.read(fetched,policy.limit,policy.exact,controller.signal);}
      catch(error){
       controller.abort();
       if(external&&external.aborted)throw error;
       if(timedOut)throw settings.failure('timeout','response deadline exceeded');
-      throw settings.failure(policy.reason,error&&error.message);
+      throw settings.failure(settings.reason(error,policy.reason),error&&error.message);
      }
      response={status:fetched.status,headers:fetched.headers,body};return response;
     }
@@ -60,7 +61,7 @@ function create(settings){
     if(timedOut)lastReason='timeout';
     if(settings.reason(error,'')==='protocol')throw error;
    }finally{clearTimeout(timer);if(external)external.removeEventListener('abort',abort)}
-   const after=Math.min(deadline-Date.now(),remainingBudget?remainingBudget():Infinity);if(attempt>=maxAttempts||after<=0)break;
+   const after=Math.min(deadline-Date.now(),remainingBudget?remainingBudget():Infinity);if(attempt>=attempts()||after<=0)break;
    settings.retrying();
    const backoff=wait||delay+Math.floor(Math.random()*Math.max(1,delay/4));
    await pause(Math.min(backoff,after),external);delay=Math.min(delay*2,2000);

src/web/bridge/runtime.js

@@ -19,7 +19,7 @@ let laneQueueLimit=Math.min(queueLimit,8388608),laneItemLimit=Math.min(queueItem
 const fragment=location.hash,androidNonce=/^#android=([A-Za-z0-9_-]{43})$/.exec(fragment)?.[1]||'',recoveryPath=location.pathname+location.search;
 history.replaceState(null,'',location.pathname);
 let initialized=false,closed=false,port=null,sessionToken='',cleanupToken='',createStarted=false,socket=null,socketReady=false,carrier='';
-let upSequence=1,downCursor='0',upRunning=false,upLease=null,pollController=null;
+let upSequence=1,downCursor='0',upRunning=false,upLease=null,httpCarrier=null,upWindow=1;
 let helloFrame=null,helloTimer=null,welcomeSent=false,carrierAttempt=1,carrierFailure='',carrierCommitted=false,terminalFailure='';
 let negotiationStartedAt=0,carrierTimer=null,probeTimer=null,attemptController=null,attemptEpoch=1,candidateRunning=false,switching=false,currentAttempt=null;
 let recoveryController=null,recoveryCommit=null,recoveryReplaced=false,lastSchedulerWall=Date.now(),lastSchedulerMonotonic=performance.now(),schedulerGapPending=0,schedulerTimer=null;
@@ -44,7 +44,7 @@ const {reserve,release,releasePending,frameBound,splitFrames,acceptNativeFrames,
 function detachLease(lease){if(lease.lane){if(lease.lane.upLease===lease)lease.lane.upLease=null}else if(upLease===lease)upLease=null}
 function settleBatch(lease){if(!buffers.settleBatch(lease))return false;detachLease(lease);return true}
 function cancelBatch(lease){if(!lease||lease.settled)return;buffers.cancelBatch(lease);detachLease(lease)}
-const attemptHeaders=(attempt,failure)=>negotiationEnabled?Object.assign({'X-Carrier-Capabilities':carrierCapabilities,'X-Carrier-Attempt':String(attempt)},failure?{'X-Carrier-Failure':failure}:{}):{};
+const attemptHeaders=(attempt,failure)=>Object.assign({'X-Telemt-Up-Window':'4'},negotiationEnabled?Object.assign({'X-Carrier-Capabilities':carrierCapabilities,'X-Carrier-Attempt':String(attempt)},failure?{'X-Carrier-Failure':failure}:{}):{});
 function finishOldRecovery(){
  recoveryReplaced=false;resetScheduler();status('connected');
  while(recoveryPending.length&&!closed){const data=recoveryPending.shift();release(data.byteLength,1,null);queueCarrier(data)}
@@ -58,8 +58,9 @@ function resolveRecoveryCommit(){
  commit.signal.removeEventListener('abort',commit.abort);commit.resolve();
 }
 function retireCarrier(policy){
+ stopHttp(false);upWindow=1;
  recoveryReplaced=true;attemptEpoch++;if(carrierTimer)clearTimeout(carrierTimer);carrierTimer=null;clearProbeTimer();
- if(attemptController)attemptController.abort();attemptController=null;if(pollController)pollController.abort();pollController=null;
+ if(attemptController)attemptController.abort();attemptController=null;
  if(socket){const previous=socket;socket=null;previous.close()}socketReady=false;cancelBatch(upLease);releasePending(upPending,null);
  for(const lane of lanes.values()){
   if(lane.controller)lane.controller.abort();cancelBatch(lane.upLease);releasePending(lane.pending,lane);if(lane.socket)lane.socket.close();
@@ -118,9 +119,10 @@ function fail(reason){
 }
 function knownCarrier(value){return value==='https'||value==='https-lanes'||value==='websocket'||value==='websocket-lanes'}
 function sessionEcho(response,expectedAttempt,states,exactAttempt){
- const selected=response.headers.get('X-Carrier-Mode')||'',echo=response.headers.get('X-Carrier-Attempt')||'';
+ const selected=response.headers.get('X-Carrier-Mode')||'',echo=response.headers.get('X-Carrier-Attempt')||'',windowHeader=response.headers.get('X-Telemt-Up-Window');
+ if(windowHeader!==null&&!/^[1-4]$/.test(windowHeader))throw new Error('invalid conveyor window');const upWindow=windowHeader===null?1:Number(windowHeader);
  if(!knownCarrier(selected))throw new Error('invalid carrier mode');
- if(!negotiationEnabled){if(echo!=='')throw new Error('unexpected carrier attempt');return {selected,state:''}}
+ if(!negotiationEnabled){if(echo!=='')throw new Error('unexpected carrier attempt');return {selected,state:'',upWindow}}
  const count=response.headers.get('X-Carrier-Candidate-Count')||'',deadline=response.headers.get('X-Carrier-Deadline')||'',state=response.headers.get('X-Carrier-State')||'';
  if(!/^[1-4]$/.test(count)||!/^[1-9]\d*$/.test(deadline)||!states.includes(state))throw new Error('invalid carrier state');
  const echoedAttempt=Number(echo),parsedCount=Number(count),parsedDeadline=Number(deadline);
@@ -129,7 +131,7 @@ function sessionEcho(response,expectedAttempt,states,exactAttempt){
  if(!negotiatedFrozen){negotiatedCandidateCount=parsedCount;negotiatedFinalDeadline=parsedDeadline;negotiatedFrozen=true}
  else if(parsedCount!==negotiatedCandidateCount||parsedDeadline!==negotiatedFinalDeadline)throw new Error('changed carrier bounds');
  if(echoedAttempt>negotiatedCandidateCount)throw new Error('carrier attempt exceeds candidates');
- return {selected,state};
+ return {selected,state,upWindow};
 }
 function armCarrierDeadline(epoch){
  if(!negotiationStartedAt||epoch!==attemptEpoch)return;
@@ -140,8 +142,8 @@ function armCarrierDeadline(epoch){
 }
 function clearProbeTimer(){if(probeTimer){clearTimeout(probeTimer.timer);probeTimer=null}}
 function resetCandidate(){
+ stopHttp(true);upWindow=1;
  clearProbeTimer();
- if(pollController)pollController.abort();pollController=null;
  if(socket){const previous=socket;socket=null;previous.close()}socketReady=false;
  cancelBatch(upLease);releasePending(upPending,null);
  for(const lane of lanes.values()){
@@ -198,8 +200,8 @@ function startCandidate(probe,epoch){
  else advanceCarrier('protocol',epoch);
 }
 function maybeStartCandidate(){
- if(closed||carrierCommitted||!sessionToken||candidateRunning)return;const epoch=attemptEpoch;
+ if(closed||carrierCommitted||!sessionToken)return;if(candidateRunning){try{flushHttpPending()}catch(error){fail(failureReason(error,'protocol'))}return}const epoch=attemptEpoch;
- let probe;try{probe=findProbe(probeCoalesceMs>0)}catch(error){fail('protocol');return}if(!probe)return;
+ let probe;try{probe=findProbe(true)}catch(error){fail('protocol');return}if(!probe)return;
  if(!probeCoalesceMs||probe.hasData){startCandidate(probe,epoch);return}
  if(probeTimer)return;const owner={epoch,timer:null};
  owner.timer=setTimeout(()=>{if(probeTimer!==owner||closed||owner.epoch!==attemptEpoch)return;probeTimer=null;let current;try{current=findProbe(false)}catch(error){fail('protocol');return}startCandidate(current,owner.epoch)},probeCoalesceMs);
@@ -219,74 +221,58 @@ async function createSession(epoch){
   const token=response.headers.get('X-Session-Token')||'',cursor=response.headers.get('X-Down-Cursor')||'';
   if(!token||cursor!=='0'){advanceCarrier('protocol',epoch);return}
   const welcome=response.body;if(closed||epoch!==attemptEpoch)return;
-  carrier=selected;sessionToken=token;cleanupToken=token;downCursor=cursor;
+  carrier=selected;sessionToken=token;cleanupToken=token;downCursor=cursor;upWindow=echo.upWindow;
   if(!welcomeSent){welcomeSent=true;port.postMessage(welcome,[welcome]);status('connecting')}
   if(carrier==='websocket')openCandidateSocket(null,null,epoch);
   maybeStartCandidate();
  }catch(error){if(closed||epoch!==attemptEpoch)return;advanceCarrier(failureReason(error,'network'),epoch)}
 }
-async function probeHttp(probe,laneID,epoch){
+function stopHttp(restore){
- try{
+ if(!httpCarrier)return;const saved=httpCarrier.close(restore);httpCarrier=null;
-  const headers={'X-Up-Seq':'1'},token=sessionToken,controller=attemptController,body=probe.data;if(laneID!==null)headers['X-Lane-ID']=String(laneID);
+ for(const data of saved){if(!reserve(data,null)){fail('capacity');return}pending.push(data)}
-  const response=await request('/api/v1/up',options(carrierMethod,token,body,headers,controller.signal));
+}
-  if(closed||epoch!==attemptEpoch)return
+function flushHttpPending(){
-  if(response.status!==204){advanceCarrier('http',epoch);return}
+ while(httpCarrier&&pending.length&&!closed){const data=pending.shift();release(data.byteLength,1,null);httpCarrier.enqueue(data,true)}
-  if(response.headers.get('X-Up-Ack')!=='1'){advanceCarrier('protocol',epoch);return}
+ if(httpCarrier&&!closed)httpCarrier.flush();
-  if(laneID===null)upSequence=2;else ensureLane(laneID).sequence=2;
+}
-  commitCarrier(probe,epoch);
+function probeHttp(probe,laneID,epoch){
- }catch(error){if(!closed&&epoch===attemptEpoch)advanceCarrier(failureReason(error,'network'),epoch)}
+ const token=sessionToken;
+ httpCarrier=globalThis.TelemtBridgeConveyor.create({
+  token,window:upWindow,method:carrierMethod,lanes:carrier==='https-lanes',buffers,options,request,
+  queueLimit:()=>queueLimit,batchLimit:()=>batchLimit,committed:()=>carrierCommitted,failure,cancel:responseBody.cancel,
+  alive:lane=>!closed&&epoch===attemptEpoch&&token===sessionToken&&(!lane||lanes.get(lane.id)===lane),
+  commit:()=>commitCarrier(null,epoch),recover:recoverTransport,connected:()=>status('connected'),
+  traffic:(up,down)=>port.postMessage({t:'traffic',up,down}),
+  deliver:data=>{observeServerFrames(data);port.postMessage({t:'traffic',up:0,down:data.byteLength});port.postMessage(data,[data])},
+  lane:(id,type)=>{
+   let lane=lanes.get(id);if(!lane&&(type===2||type===3||type===4))return null;
+   if(!lane&&closedLanes.has(id))throw failure('protocol','closed lane reused');
+   if(!lane&&id!==0&&type!==1)throw failure('protocol','lane did not begin with OPEN');
+   return lane||ensureLane(id);
+  },
+  finish:finishLane,failed:error=>{if(closed||epoch!==attemptEpoch)return;if(carrierCommitted)fail(failureReason(error,'network'));else advanceCarrier(failureReason(error,'network'),epoch)}
+ });
+ try{flushHttpPending()}catch(error){fail(failureReason(error,'protocol'))}
 }
 function commitCarrier(probe,epoch){
  if(closed||carrierCommitted||epoch!==attemptEpoch)return;
  if(switching){fail('protocol');return}
- clearProbeTimer();try{consumeProbe(probe)}catch(error){fail('protocol');return}
+ clearProbeTimer();try{if(probe)consumeProbe(probe)}catch(error){fail('protocol');return}
  carrierCommitted=true;candidateRunning=false;if(carrierTimer)clearTimeout(carrierTimer);carrierTimer=null;
  resetScheduler();
  attemptController=null;currentAttempt=null;
  status('connected');
- if(carrier==='https')poll();
- else if(carrier==='https-lanes'){const lane=lanes.get(probe.id);if(lane&&!lane.polling)pollLane(lane)}
  while(pending.length&&!closed){const data=pending.shift();release(data.byteLength,1,null);queueCarrier(data)}
  resolveRecoveryCommit();
 }
 function queueCarrier(data){
  if(closed)return;
  try{
-  if(carrier==='https')queueUp(data);
+  if(carrier==='https'||carrier==='https-lanes')httpCarrier.enqueue(data);
   else if(carrier==='websocket')queueSocket(data);
   else for(const value of splitFrames(data)){if(closed)break;queueLane(value)}
  }catch(error){fail('protocol')}
 }
-function queueUp(data){if(!reserve(data,null)){fail('capacity');return}upPending.push(data);runUp()}
-async function runUp(){
- if(upRunning)return;upRunning=true;const ownerEpoch=attemptEpoch;let lease=null;
- try{
-  while(!closed&&sessionToken&&upPending.length){
-   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(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(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;
-     });
-     if(!recovered||closed||lease.cancelled||sessionToken!==token)return;
-     if(!replayed)continue;
-     break;
-    }
-   }
-   if(!settleBatch(lease))return;port.postMessage({t:'traffic',up:lease.total,down:0});upSequence++;lease=null;
-  }
- }catch(error){if(!closed&&!(lease&&lease.cancelled))fail(failureReason(error,'network'))}
- finally{if(ownerEpoch===attemptEpoch){upRunning=false;if(!closed&&sessionToken&&upPending.length)runUp()}}
-}
 function sendCandidateSocket(next){
  const state=next.telemt;if(!state||state.sent||next.readyState!==WebSocket.OPEN||!state.probe)return;
  let probe=state.probe;
@@ -340,30 +326,6 @@ async function runSocketUp(){
  }catch(error){if(!closed&&!(lease&&lease.cancelled))recoverTransport(error,null)}
  finally{if(ownerEpoch===attemptEpoch){upRunning=false;if(!closed&&socketReady&&upPending.length)runSocketUp()}}
 }
-async function poll(){
- while(!closed&&sessionToken){
-  const token=sessionToken,cursor=downCursor;
-  try{
-   pollController=new AbortController();
-   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');
-   const next=response.headers.get('X-Down-Cursor')||'',data=response.body;
-   if(!next||!data.byteLength)throw failure('protocol','invalid downlink response');
-   if(closed)return;
-   observeServerFrames(data);port.postMessage({t:'traffic',up:0,down:data.byteLength});port.postMessage(data,[data]);downCursor=next;status('connected');
-  }catch(error){
-   if(closed)return;
-   const recovered=await recoverTransport(error,async(signal,remaining)=>{
-    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');
-   });
-   if(!recovered||closed||sessionToken!==token)return;
-  }
- }
-}
 function ensureLane(id){
  let lane=lanes.get(id);
  if(!lane){lane={id,sequence:1,cursor:'0',pending:[],bytes:0,items:0,running:false,upLease:null,polling:false,controller:null,socket:null,ready:false,remoteClosed:false};lanes.set(id,lane)}
@@ -375,7 +337,7 @@ function rememberLaneClosed(id){
  closedLanes.add(id);closedLaneOrder.push(id);
 }
 function finishLane(lane,notifyClient){
- if(lanes.get(lane.id)!==lane)return;
+ if(lanes.get(lane.id)!==lane)return;if(httpCarrier)httpCarrier.finish(lane);
  if(lane.controller)lane.controller.abort();lane.controller=null;cancelBatch(lane.upLease);
  if(lane.socket&&lane.socket.readyState<WebSocket.CLOSING)lane.socket.close();
  releasePending(lane.pending,lane);lanes.delete(lane.id);rememberLaneClosed(lane.id);
@@ -389,7 +351,7 @@ function queueLane(value){
  lane=lane||ensureLane(value.id);
  if(!reserve(value.data,lane)){fail('capacity');return}
  lane.pending.push(value.data);
- if(carrier==='websocket-lanes'){openLaneSocket(lane);runLaneSocketUp(lane)}else runLaneUp(lane);
+ openLaneSocket(lane);runLaneSocketUp(lane);
 }
 function openLaneSocket(lane){
  if(lane.socket||closed)return;lane.socket=new WebSocket(socketURL(),'tproxy-lane-v1.'+sessionToken+'.'+String(lane.id));lane.socket.binaryType='arraybuffer';
@@ -416,74 +378,12 @@ async function runLaneSocketUp(lane){
  }catch(error){if(!closed&&lanes.get(lane.id)===lane&&!(lease&&lease.cancelled))finishLane(lane,true)}
  finally{lane.running=false;if(!closed&&lanes.get(lane.id)===lane&&lane.ready&&lane.pending.length)runLaneSocketUp(lane)}
 }
-async function runLaneUp(lane){
- if(lane.running)return;lane.running=true;let lease=null;
- try{
-  while(!closed&&sessionToken&&lane.pending.length){
-   lease=takeBatch(lane.pending,lane);lane.upLease=lease;lease.controller=new AbortController();
-   const sequence=String(lane.sequence),laneID=String(lane.id),token=sessionToken;
-   for(;;){
-    try{
-     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(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;
-     });
-     if(!recovered||closed||lease.cancelled||sessionToken!==token||lanes.get(lane.id)!==lane)return;
-     if(!replayed)continue;
-     break;
-    }
-   }
-   if(!settleBatch(lease))return;port.postMessage({t:'traffic',up:lease.total,down:0});lane.sequence++;lease=null;
-   if(!lane.polling)pollLane(lane);
-  }
- }catch(error){if(!closed&&lanes.get(lane.id)===lane&&!(lease&&lease.cancelled))fail(failureReason(error,'network'))}
- finally{lane.running=false;if(!closed&&lanes.get(lane.id)===lane&&sessionToken&&lane.pending.length)runLaneUp(lane)}
-}
-async function pollLane(lane){
- if(!lane||lane.polling)return;lane.polling=true;let restart=false,failedToken='',failedCursor='',failedLaneID='';
- try{
-  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(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}
-    status('connected');continue;
-   }
-   if(response.status!==200)throw failure('http','lane downlink rejected');
-   const next=response.headers.get('X-Down-Cursor')||'',data=response.body;
-   if(!next||!data.byteLength)throw failure('protocol','invalid lane downlink response');
-   for(const value of splitFrames(data))if(value.id!==lane.id)throw new Error('cross-lane frame');
-   if(closed)return;
-   observeServerFrames(data);port.postMessage({t:'traffic',up:0,down:data.byteLength});port.postMessage(data,[data]);lane.cursor=next;status('connected');
-  }
-  }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(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');
-    });
-    restart=recovered&&!closed&&sessionToken===failedToken&&lanes.get(lane.id)===lane;
-   }
-  }
- finally{lane.polling=false;lane.controller=null;if(restart)pollLane(lane)}
-}
 function deleteSession(){
  const token=cleanupToken||sessionToken,headers=canonicalFailures.includes(terminalFailure)?{'X-Carrier-Failure':terminalFailure}:null;
  if(token)fetch(relayBase+'/api/v1/session',options('DELETE',token,null,headers,undefined,true)).catch(()=>{});
 }
 function close(notifyServer){
- if(closed)return;closed=true;if(recoveryController)recoveryController.cancel();rejectRecoveryCommit(failure('network','bridge closed'));if(helloTimer)clearTimeout(helloTimer);helloTimer=null;if(carrierTimer)clearTimeout(carrierTimer);clearProbeTimer();if(schedulerTimer)clearTimeout(schedulerTimer);schedulerTimer=null;if(attemptController)attemptController.abort();if(pollController)pollController.abort();
+ if(closed)return;closed=true;stopHttp(false);if(recoveryController)recoveryController.cancel();rejectRecoveryCommit(failure('network','bridge closed'));if(helloTimer)clearTimeout(helloTimer);helloTimer=null;if(carrierTimer)clearTimeout(carrierTimer);clearProbeTimer();if(schedulerTimer)clearTimeout(schedulerTimer);schedulerTimer=null;if(attemptController)attemptController.abort();
  if(socket)socket.close();cancelBatch(upLease);releasePending(upPending,null);
  for(const lane of lanes.values()){
   if(lane.controller)lane.controller.abort();cancelBatch(lane.upLease);releasePending(lane.pending,lane);if(lane.socket)lane.socket.close();

src/web/bridge/tests.rs

@@ -31,20 +31,8 @@ fn carrier_method_is_page_owned_and_used_by_every_https_request() {
                 .contains(&format!("const carrierMethod='{}';", method.as_str()))
         );
         assert_eq!(page.body.matches("carrierMethod=").count(), 1);
-        assert_eq!(page.body.matches("options(carrierMethod,").count(), 9);
+        // Execute the rendered POST/PUT page, including retries, instead of counting call sites.
-        let requests: Vec<_> = page
+        behavior_tests::run(&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
@@ -68,6 +56,9 @@ fn carrier_method_is_page_owned_and_used_by_every_https_request() {
     }
 }
 
+#[path = "behavior_tests.rs"]
+mod behavior_tests;
+
 fn render_page(bootstrap: &str, candidate_count: usize) -> BridgePage {
     render(
         "proxy.example.com",
@@ -289,8 +280,9 @@ fn disabled_negotiation_does_not_arm_a_carrier_deadline() {
 fn retry_and_attempt_state_are_frozen_before_fetch() {
     let page = render_page("EEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE", 4);
     assert!(
-        page.body
+        page.body.contains(
-            .contains("async function send(path,frozenOptions,remainingBudget,maxAttempts)")
+            "async function send(path,frozenOptions,remainingBudget,maxAttempts,receiver)"
+        )
     );
     assert!(!page.body.contains("makeOptions"));
     assert!(page.body.contains(

src/web/http.rs

@@ -36,6 +36,8 @@ mod decoy;
 mod diagnostic;
 // Downlink long-poll handling remains isolated from request routing.
 mod down;
+// Uplink admission and conveyor waits own their bounded request bodies.
+mod up;
 // Canonical request parsing rejects ambiguous credentials before routing.
 mod request;
 // Positive-only recovery representation stays separate from ordinary bridge rendering.
@@ -68,6 +70,7 @@ use response::{
     insert_header, service_unavailable,
 };
 use session::handle_session;
+use up::handle_up;
 
 type BoxError = Box<dyn Error + Send + Sync>;
 type HttpBody = UnsyncBoxBody<Bytes, BoxError>;
@@ -457,76 +460,6 @@ 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 !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)
-    else {
-        return serve_decoy(request, vhost, true, &runtime).await;
-    };
-    let Ok(session) = runtime.get_session(token_hash, &vhost.host) else {
-        return serve_decoy(request, vhost, true, &runtime).await;
-    };
-    if session.carrier().uses_websocket() {
-        return serve_decoy(request, vhost, true, &runtime).await;
-    }
-    if let Some(trace) = request_trace(&request) {
-        trace.set_route(TraceRoute::Uplink);
-        trace.bind_identity(session.trace_identity());
-    }
-    let Some(lane_id) = carrier_lane(&request, session.carrier()) else {
-        return serve_decoy(request, vhost, true, &runtime).await;
-    };
-    let limit = session.limits().max_body_bytes;
-    let CollectedBody {
-        request,
-        body,
-        _body_budget,
-    } = match collect_body(
-        request,
-        &runtime,
-        Duration::from_secs(session.timeouts().body_secs),
-        limit,
-        false,
-    )
-    .await
-    {
-        Ok(result) => result,
-        Err(CollectBodyError::Limit) => return service_unavailable(),
-        Err(CollectBodyError::Invalid(request)) => {
-            return serve_decoy(request, vhost, true, &runtime).await;
-        }
-    };
-    if let Some(trace) = request_trace(&request) {
-        trace.record_frames(TraceDirection::Request, &body, session.limits());
-    }
-    let result = match lane_id {
-        Some(lane_id) => session.process_up_lane(lane_id, sequence, &body),
-        None => session.process_up(sequence, &body),
-    };
-    match result {
-        Ok(ack) => {
-            let mut response = carrier_empty(StatusCode::NO_CONTENT);
-            insert_header(
-                &mut response,
-                HeaderName::from_static("x-up-ack"),
-                &ack.to_string(),
-            );
-            response
-        }
-        Err(ManagerError::Backpressure | ManagerError::Concurrent | ManagerError::Limit) => {
-            service_unavailable()
-        }
-        Err(_) => serve_decoy(request, vhost, true, &runtime).await,
-    }
-}
-
 fn strip_query<B>(request: &mut Request<B>) {
     if request.uri().query().is_some()
         && let Ok(uri) = request.uri().path().parse()

src/web/http/carrier_method_tests.rs

@@ -2,6 +2,9 @@ use super::super::session_policy_tests::{open_keepalive, read_http_response};
 use super::super::*;
 use crate::config::WebCarrierMethod;
 
+#[path = "conveyor_tests.rs"]
+mod conveyor_tests;
+
 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",

src/web/http/conveyor_tests.rs

@@ -0,0 +1,358 @@
+use super::*;
+
+async fn create_with_headers(
+    listener: &TcpListener,
+    runtime: &Arc<WebProcessRuntime>,
+    bootstrap: &str,
+    headers: &str,
+) -> Vec<u8> {
+    request(
+        listener,
+        runtime,
+        carrier_request(
+            "POST",
+            "/api/v1/session",
+            bootstrap,
+            &format!("Content-Type: application/octet-stream\r\n{headers}"),
+            &frame::encode(FrameType::Hello, 0, &[1]),
+        ),
+    )
+    .await
+}
+
+fn up(token: &str, seq: u64, confirmed: &str, lane: &str) -> Vec<u8> {
+    carrier_request(
+        "POST",
+        "/api/v1/up",
+        token,
+        &format!("Content-Type: application/octet-stream\r\nX-Up-Seq: {seq}\r\n{confirmed}{lane}"),
+        &frame::encode(FrameType::Pong, 0, &[]),
+    )
+}
+
+#[tokio::test]
+async fn conveyor_negotiation_is_opt_in_bounded_and_invisible_to_native_welcome() {
+    for enabled in [false, true] {
+        for carrier in [
+            WebCarrier::Https,
+            WebCarrier::HttpsLanes,
+            WebCarrier::Websocket,
+        ] {
+            let capability = [71; 32];
+            let mut config = runtime_config(capability, carrier);
+            config.web.conveyor = enabled;
+            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();
+            for offered in [None, Some(1), Some(2), Some(4)] {
+                let page = bridge_page(&listener, &runtime, capability, "/").await;
+                let bootstrap = bootstrap_from(&page);
+                let offer = offered.map_or(String::new(), |value| {
+                    format!("X-Telemt-Up-Window: {value}\r\n")
+                });
+                let response = create_with_headers(&listener, &runtime, bootstrap, &offer).await;
+                let (headers, body) = split_response(&response);
+                assert!(headers.starts_with(b"HTTP/1.1 200"));
+                assert_eq!(body, frame::encode(FrameType::Welcome, 0, &[]));
+                if let Some(offered) = offered {
+                    let window = if enabled && !carrier.uses_websocket() {
+                        offered
+                    } else {
+                        1
+                    };
+                    assert_eq!(
+                        response_header(headers, "x-telemt-up-window"),
+                        window.to_string()
+                    );
+                } else {
+                    assert!(
+                        !String::from_utf8_lossy(headers)
+                            .to_lowercase()
+                            .contains("x-telemt-up-window")
+                    );
+                }
+                let replay = create_with_headers(&listener, &runtime, bootstrap, &offer).await;
+                let (replayed, _) = split_response(&replay);
+                assert_eq!(
+                    response_header(headers, "x-session-token"),
+                    response_header(replayed, "x-session-token")
+                );
+            }
+            runtime.shutdown().await;
+            generation.stop_sessions().await;
+            generation.stop_background_tasks().await;
+        }
+    }
+}
+
+#[tokio::test]
+async fn malformed_window_does_not_consume_bootstrap_or_enable_a_conveyor() {
+    let capability = [72; 32];
+    let generation = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https));
+    let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let page = bridge_page(&listener, &runtime, capability, "/").await;
+    let bootstrap = bootstrap_from(&page);
+    for offer in ["0", "5", "04", "-1", "4, 4", "4\r\nX-Telemt-Up-Window: 4"] {
+        let response = create_with_headers(
+            &listener,
+            &runtime,
+            bootstrap,
+            &format!("X-Telemt-Up-Window: {offer}\r\n"),
+        )
+        .await;
+        assert!(
+            response.starts_with(b"HTTP/1.1 404"),
+            "{offer}: {:?}",
+            response
+        );
+    }
+    let response =
+        create_with_headers(&listener, &runtime, bootstrap, "X-Telemt-Up-Window: 4\r\n").await;
+    assert!(response.starts_with(b"HTTP/1.1 200"));
+    runtime.shutdown().await;
+    generation.stop_sessions().await;
+    generation.stop_background_tasks().await;
+}
+
+#[tokio::test]
+async fn reload_changes_only_new_sessions_and_exact_creation_replay_keeps_its_window() {
+    let capability = [73; 32];
+    let config = runtime_config(capability, WebCarrier::Https);
+    let generation = test_runtime_generation(1, config.clone());
+    let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let page = bridge_page(&listener, &runtime, capability, "/").await;
+    let bootstrap = bootstrap_from(&page);
+    let offer = "X-Telemt-Up-Window: 4\r\n";
+    let original = create_with_headers(&listener, &runtime, bootstrap, offer).await;
+    let (original_headers, _) = split_response(&original);
+    let token = response_header(original_headers, "x-session-token");
+    let mut disabled = config;
+    disabled.web.conveyor = false;
+    let replacement = test_runtime_generation(2, disabled);
+    runtime.activate_generation(Arc::clone(&replacement));
+    let replay = create_with_headers(&listener, &runtime, bootstrap, offer).await;
+    let (headers, _) = split_response(&replay);
+    assert_eq!(response_header(headers, "x-telemt-up-window"), "4");
+    assert_eq!(response_header(headers, "x-session-token"), token);
+    let old_up = request(
+        &listener,
+        &runtime,
+        up(token, 1, "X-Telemt-Up-Confirmed: 0\r\n", ""),
+    )
+    .await;
+    assert!(old_up.starts_with(b"HTTP/1.1 204"));
+    let page = bridge_page(&listener, &runtime, capability, "/").await;
+    let created = create_with_headers(&listener, &runtime, bootstrap_from(&page), offer).await;
+    let (headers, _) = split_response(&created);
+    assert_eq!(response_header(headers, "x-telemt-up-window"), "1");
+    let token = response_header(headers, "x-session-token");
+    let mode_error = request(
+        &listener,
+        &runtime,
+        up(token, 1, "X-Telemt-Up-Confirmed: 0\r\n", ""),
+    )
+    .await;
+    assert!(mode_error.starts_with(b"HTTP/1.1 409"));
+    let legacy = request(&listener, &runtime, up(token, 1, "", "")).await;
+    assert!(legacy.starts_with(b"HTTP/1.1 204"));
+    runtime.shutdown().await;
+    for generation in [generation, replacement] {
+        generation.stop_sessions().await;
+        generation.stop_background_tasks().await;
+    }
+}
+
+#[tokio::test]
+async fn future_http_request_is_not_acked_until_head_applies_for_both_carriers() {
+    for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] {
+        let capability = [74; 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 page = bridge_page(&listener, &runtime, capability, "/").await;
+        let created = create_with_headers(
+            &listener,
+            &runtime,
+            bootstrap_from(&page),
+            "X-Telemt-Up-Window: 4\r\n",
+        )
+        .await;
+        let (headers, _) = split_response(&created);
+        let token = response_header(headers, "x-session-token");
+        let lane = if carrier.uses_lanes() {
+            "X-Lane-ID: 0\r\n"
+        } else {
+            ""
+        };
+        let (mut tail, cancellation, task) = open_keepalive(&listener, &runtime).await;
+        tail.write_all(&up(token, 2, "X-Telemt-Up-Confirmed: 0\r\n", lane))
+            .await
+            .unwrap();
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(20), tail.read_u8())
+                .await
+                .is_err()
+        );
+        let head = request(
+            &listener,
+            &runtime,
+            up(token, 1, "X-Telemt-Up-Confirmed: 0\r\n", lane),
+        )
+        .await;
+        assert!(head.starts_with(b"HTTP/1.1 204"));
+        let response = read_http_response(&mut tail).await;
+        let (headers, _) = split_response(&response);
+        assert!(headers.starts_with(b"HTTP/1.1 204"));
+        assert_eq!(response_header(headers, "x-up-ack"), "2");
+        cancellation.cancel();
+        task.await.unwrap();
+        runtime.shutdown().await;
+        generation.stop_sessions().await;
+        generation.stop_background_tasks().await;
+    }
+}
+
+#[tokio::test]
+async fn parked_uplink_keeps_its_session_deadline_after_reload() {
+    let capability = [75; 32];
+    let mut config = runtime_config(capability, WebCarrier::Https);
+    config.web.timeouts.body_secs = 4;
+    let generation = test_runtime_generation(1, config.clone());
+    let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation))));
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let page = bridge_page(&listener, &runtime, capability, "/").await;
+    let response = create_with_headers(
+        &listener,
+        &runtime,
+        bootstrap_from(&page),
+        "X-Telemt-Up-Window: 4\r\n",
+    )
+    .await;
+    let (headers, _) = split_response(&response);
+    let token = response_header(headers, "x-session-token");
+    config.web.timeouts.body_secs = 1;
+    config.web.timeouts.http_idle_secs = 2;
+    let replacement = test_runtime_generation(2, config);
+    runtime.activate_generation(Arc::clone(&replacement));
+    let (mut tail, cancellation, task) = open_keepalive(&listener, &runtime).await;
+    tail.write_all(&up(token, 2, "X-Telemt-Up-Confirmed: 0\r\n", ""))
+        .await
+        .unwrap();
+    tokio::time::sleep(std::time::Duration::from_millis(2200)).await;
+    assert!(
+        !task.is_finished(),
+        "connection idle timeout must not replace the frozen parked-UP deadline"
+    );
+    let head = request(
+        &listener,
+        &runtime,
+        up(token, 1, "X-Telemt-Up-Confirmed: 0\r\n", ""),
+    )
+    .await;
+    assert!(head.starts_with(b"HTTP/1.1 204"));
+    assert!(
+        read_http_response(&mut tail)
+            .await
+            .starts_with(b"HTTP/1.1 204")
+    );
+    cancellation.cancel();
+    task.await.unwrap();
+    runtime.shutdown().await;
+    for generation in [generation, replacement] {
+        generation.stop_sessions().await;
+        generation.stop_background_tasks().await;
+    }
+}
+
+#[tokio::test]
+async fn empty_long_poll_cannot_steal_the_reserved_head_body_capacity() {
+    let capability = [76; 32];
+    let mut config = runtime_config(capability, WebCarrier::Https);
+    config.web.limits.max_body_readers = 3;
+    config.web.limits.max_body_bytes = 32;
+    config.web.limits.max_body_bytes_global = 96;
+    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 page = bridge_page(&listener, &runtime, capability, "/").await;
+    let response = create_with_headers(
+        &listener,
+        &runtime,
+        bootstrap_from(&page),
+        "X-Telemt-Up-Window: 4\r\n",
+    )
+    .await;
+    let (headers, _) = split_response(&response);
+    let token = response_header(headers, "x-session-token");
+    let (mut down, down_cancel, down_task) = open_keepalive(&listener, &runtime).await;
+    down.write_all(&carrier_request(
+        "POST",
+        "/api/v1/down",
+        token,
+        "X-Down-Cursor: 0\r\n",
+        &[],
+    ))
+    .await
+    .unwrap();
+    let used = |resource| {
+        runtime
+            .capacity_snapshot()
+            .resources
+            .into_iter()
+            .find(|value| value.resource == resource)
+            .unwrap()
+            .used
+    };
+    tokio::time::timeout(std::time::Duration::from_secs(1), async {
+        while used("lane_polls") != 1 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(
+        used("body_bytes"),
+        0,
+        "an empty long poll owns no buffered request body"
+    );
+    let mut tails = Vec::new();
+    for sequence in [2, 3] {
+        let (mut client, cancel, task) = open_keepalive(&listener, &runtime).await;
+        client
+            .write_all(&up(token, sequence, "X-Telemt-Up-Confirmed: 0\r\n", ""))
+            .await
+            .unwrap();
+        tails.push((client, cancel, task));
+    }
+    tokio::time::timeout(std::time::Duration::from_secs(1), async {
+        while used("body_bytes") != 64 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    let head = request(
+        &listener,
+        &runtime,
+        up(token, 1, "X-Telemt-Up-Confirmed: 0\r\n", ""),
+    )
+    .await;
+    assert!(head.starts_with(b"HTTP/1.1 204"));
+    for (mut client, cancel, task) in tails {
+        assert!(
+            read_http_response(&mut client)
+                .await
+                .starts_with(b"HTTP/1.1 204")
+        );
+        cancel.cancel();
+        task.await.unwrap();
+    }
+    down_cancel.cancel();
+    down_task.await.unwrap();
+    runtime.shutdown().await;
+    generation.stop_sessions().await;
+    generation.stop_background_tasks().await;
+}

src/web/http/down.rs

@@ -65,6 +65,8 @@ pub(super) async fn handle_down(
     if !body.is_empty() {
         return serve_decoy(request, vhost, true, &runtime).await;
     }
+    // Empty polls must not retain even one byte from a conveyor head's body reserve.
+    drop(_body_budget);
     let Some(_down_poll) = runtime.try_lane_poll(false) else {
         return service_unavailable();
     };

src/web/http/session.rs

@@ -76,6 +76,17 @@ pub(super) async fn handle_session(
     let Some(carrier_request) = carrier_request(&request, &vhost.host) else {
         return serve_decoy(request, vhost, true, &runtime).await;
     };
+    let up_window = if request.headers().contains_key("x-telemt-up-window") {
+        let Some(window) = super::request::canonical_u64_header(&request, "x-telemt-up-window")
+            .filter(|window| (1..=4).contains(window))
+        else {
+            return serve_decoy(request, vhost, true, &runtime).await;
+        };
+        Some(window as u8)
+    } else {
+        None
+    };
+    let carrier_request = carrier_request.with_up_window(up_window);
     let ip_learning_eligible = carrier_ip_learning_eligible(&request, client_ip);
     let Some((trace_session_id, profile, body_timeout)) =
         runtime.bootstrap_trace_identity(token_hash, &vhost.host)
@@ -124,6 +135,13 @@ pub(super) async fn handle_session(
             }
             let mut response = full_response(StatusCode::OK, welcome);
             carrier_headers(&mut response);
+            if let Some(window) = result.up_window {
+                insert_header(
+                    &mut response,
+                    HeaderName::from_static("x-telemt-up-window"),
+                    &window.to_string(),
+                );
+            }
             insert_header(
                 &mut response,
                 HeaderName::from_static("x-session-token"),

src/web/http/up.rs

@@ -0,0 +1,109 @@
+use super::*;
+use crate::web::session::ConveyorError;
+
+/// Handles one bounded legacy or explicitly negotiated conveyor uplink.
+pub(super) async fn handle_up(
+    request: Request<RequestBody>,
+    runtime: Arc<WebProcessRuntime>,
+    vhost: Arc<WebRuntimeVhost>,
+    token_hash: crate::web::manager::TokenHash,
+) -> HttpResponse {
+    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)
+    else {
+        return serve_decoy(request, vhost, true, &runtime).await;
+    };
+    let Ok(session) = runtime.get_session(token_hash, &vhost.host) else {
+        return serve_decoy(request, vhost, true, &runtime).await;
+    };
+    if session.carrier().uses_websocket() {
+        return serve_decoy(request, vhost, true, &runtime).await;
+    }
+    if let Some(trace) = request_trace(&request) {
+        trace.set_route(TraceRoute::Uplink);
+        trace.bind_identity(session.trace_identity());
+    }
+    let Some(lane_id) = carrier_lane(&request, session.carrier()) else {
+        return serve_decoy(request, vhost, true, &runtime).await;
+    };
+
+    let confirmed = if request.headers().contains_key("x-telemt-up-confirmed") {
+        let Some(value) = canonical_u64_header(&request, "x-telemt-up-confirmed") else {
+            return serve_decoy(request, vhost, true, &runtime).await;
+        };
+        Some(value)
+    } else {
+        None
+    };
+    let claim = match session.claim_conveyor(lane_id, sequence, confirmed) {
+        Ok(claim) => claim,
+        Err(ConveyorError::Stale | ConveyorError::Mode) => {
+            return carrier_empty(StatusCode::CONFLICT);
+        }
+        Err(ConveyorError::Manager(
+            ManagerError::Backpressure | ManagerError::Concurrent | ManagerError::Limit,
+        )) => {
+            return service_unavailable();
+        }
+        Err(_) => return serve_decoy(request, vhost, true, &runtime).await,
+    };
+    let limit = session.limits().max_body_bytes;
+    let CollectedBody {
+        request,
+        body,
+        _body_budget,
+    } = match collect_body(
+        request,
+        &runtime,
+        Duration::from_secs(session.timeouts().body_secs),
+        limit,
+        false,
+    )
+    .await
+    {
+        Ok(result) => result,
+        Err(CollectBodyError::Limit) => return service_unavailable(),
+        Err(CollectBodyError::Invalid(request)) => {
+            return serve_decoy(request, vhost, true, &runtime).await;
+        }
+    };
+    if let Some(trace) = request_trace(&request) {
+        trace.record_frames(TraceDirection::Request, &body, session.limits());
+    }
+    let result = if let Some(claim) = claim {
+        let _deadline_lease = match super::request_deadline(&request) {
+            Some(deadline) => {
+                match deadline.lease_for(Duration::from_secs(session.timeouts().body_secs)) {
+                    Some(lease) => Some(lease),
+                    None => return service_unavailable(),
+                }
+            }
+            None => None,
+        };
+        claim.process(&body).await
+    } else {
+        match lane_id {
+            Some(lane_id) => session.process_up_lane(lane_id, sequence, &body),
+            None => session.process_up(sequence, &body),
+        }
+        .map_err(ConveyorError::Manager)
+    };
+    match result {
+        Ok(ack) => {
+            let mut response = carrier_empty(StatusCode::NO_CONTENT);
+            insert_header(
+                &mut response,
+                HeaderName::from_static("x-up-ack"),
+                &ack.to_string(),
+            );
+            response
+        }
+        Err(ConveyorError::Stale | ConveyorError::Mode) => carrier_empty(StatusCode::CONFLICT),
+        Err(ConveyorError::Manager(
+            ManagerError::Backpressure | ManagerError::Concurrent | ManagerError::Limit,
+        )) => service_unavailable(),
+        Err(_) => serve_decoy(request, vhost, true, &runtime).await,
+    }
+}

src/web/manager.rs

@@ -177,6 +177,8 @@ impl ManagerError {
 
 /// Successful idempotent session creation result.
 pub(crate) struct CreateResult {
+    /// HTTP window echoed only to bridge pages advertising the extension.
+    pub(crate) up_window: Option<u8>,
     /// Opaque bearer token for the created or replayed session.
     pub(crate) token: String,
     /// Carrier frozen into the created or replayed session.
@@ -230,6 +232,7 @@ pub(crate) struct WebProcessRuntime {
     lane_aux_polls: Arc<Semaphore>,
     body_readers: Arc<Semaphore>,
     body_bytes: Arc<Semaphore>,
+    conveyor_waiters: Arc<Semaphore>,
     stream_handshakes: Arc<Semaphore>,
     websocket_connections: Arc<Semaphore>,
     websockets: Mutex<websocket::WebSocketRegistry>,
@@ -299,6 +302,9 @@ impl WebProcessRuntime {
             lane_aux_polls: Arc::new(Semaphore::new(lane_aux_poll_limit)),
             body_readers: Arc::new(Semaphore::new(limits.max_body_readers)),
             body_bytes: Arc::new(Semaphore::new(limits.max_body_bytes_global)),
+            conveyor_waiters: Arc::new(Semaphore::new(crate::web::session::conveyor_waiter_limit(
+                &limits,
+            ))),
             stream_handshakes: Arc::new(Semaphore::new(limits.max_stream_handshakes)),
             websocket_connections: Arc::new(Semaphore::new(websocket_connections)),
             websockets: Mutex::new(websocket::WebSocketRegistry::default()),

src/web/manager/budget.rs

@@ -337,6 +337,11 @@ impl Drop for WebSocketBudgetLease {
 }
 
 impl WebProcessRuntime {
+    /// Reserves a parked future uplink without consuming the head request's capacity.
+    pub(crate) fn try_conveyor_waiter(&self) -> Option<OwnedSemaphorePermit> {
+        Arc::clone(&self.conveyor_waiters).try_acquire_owned().ok()
+    }
+
     /// Reserves one body reader and its declared bounded body allocation.
     pub(crate) fn try_body_budget(
         &self,

src/web/manager/lifecycle.rs

@@ -136,6 +136,7 @@ impl WebProcessRuntime {
         self.lane_aux_polls.close();
         self.body_readers.close();
         self.body_bytes.close();
+        self.conveyor_waiters.close();
         self.stream_handshakes.close();
         self.websocket_connections.close();
         let sessions = {

src/web/manager/negotiation.rs

@@ -130,6 +130,7 @@ pub(crate) struct CarrierRequest {
     attempt: Option<u8>,
     failure: Option<CarrierFailure>,
     user_agent_hash: [u8; 32],
+    up_window: Option<u8>,
 }
 
 impl CarrierRequest {
@@ -141,6 +142,7 @@ impl CarrierRequest {
             attempt: None,
             failure: None,
             user_agent_hash,
+            up_window: None,
         }
     }
 
@@ -152,6 +154,7 @@ impl CarrierRequest {
             attempt: None,
             failure: None,
             user_agent_hash,
+            up_window: None,
         }
     }
 
@@ -169,9 +172,21 @@ impl CarrierRequest {
             attempt: Some(attempt),
             failure,
             user_agent_hash,
+            up_window: None,
         }
     }
 
+    /// Attaches the separately validated HTTP conveyor capability.
+    pub(crate) const fn with_up_window(mut self, window: Option<u8>) -> Self {
+        self.up_window = window;
+        self
+    }
+
+    /// Returns the separately advertised HTTP conveyor capability.
+    pub(crate) const fn up_window(self) -> Option<u8> {
+        self.up_window
+    }
+
     /// Returns whether this request participates in server-side negotiation.
     pub(crate) const fn is_automatic(self) -> bool {
         self.capabilities.is_some()
@@ -215,6 +230,7 @@ impl CarrierRequest {
         self.class == other.class
             && self.capabilities_bits() == other.capabilities_bits()
             && self.user_agent_hash == other.user_agent_hash
+            && self.up_window == other.up_window
     }
 
     /// Checks the complete idempotent identity of one exact attempt request.

src/web/manager/session_creation.rs

@@ -125,6 +125,7 @@ impl WebProcessRuntime {
                     entry.carrier_phase.as_str()
                 };
                 let result = CreateResult {
+                    up_window: carrier_request.up_window().map(|_| session.up_window()),
                     token: entry.session_token.as_str().to_owned(),
                     carrier: session.carrier(),
                     attempt: carrier_request.attempt(),
@@ -357,6 +358,11 @@ impl WebProcessRuntime {
             issued_timeouts.clone(),
             Some(user_registration),
         );
+        session.configure_conveyor(if config.web.conveyor {
+            carrier_request.up_window().unwrap_or(1)
+        } else {
+            1
+        });
         state.sessions.insert(session_hash, Arc::clone(&session));
         *state.sessions_per_ip.entry(client_ip).or_insert(0) += 1;
         *state.sessions_per_profile.entry(profile_key).or_insert(0) += 1;
@@ -399,6 +405,7 @@ impl WebProcessRuntime {
         }
         let identity = session.trace_identity();
         let result = CreateResult {
+            up_window: carrier_request.up_window().map(|_| session.up_window()),
             token: session_token,
             carrier,
             attempt: carrier_request.attempt(),

src/web/manager/session_creation/replacement.rs

@@ -104,6 +104,7 @@ impl WebProcessRuntime {
             replacement.old_session.timeouts().clone(),
             Some(user_registration),
         );
+        session.configure_conveyor(replacement.old_session.conveyor_offer());
         let Some(supersede) = replacement.old_session.prepare_carrier_supersede() else {
             drop(user_publication);
             drop(state);
@@ -149,6 +150,7 @@ impl WebProcessRuntime {
             crate::web::session::SessionCloseReason::CarrierSuperseded,
         );
         let result = CreateResult {
+            up_window: replacement.request.up_window().map(|_| session.up_window()),
             token: session_token,
             carrier: replacement.carrier,
             attempt: Some(replacement.attempt),

src/web/session.rs

@@ -54,6 +54,9 @@ use lifecycle::SessionNegotiationPhase;
 pub(crate) use lifecycle::{SessionCloseOutcome, SessionCloseReason};
 // Uplink batches own exactly-once sequencing and client-frame validation.
 mod uplink;
+// HTTP conveyor ownership bounds reordered requests without changing native frames.
+mod conveyor;
+pub(crate) use conveyor::{ConveyorError, conveyor_waiter_limit};
 // Logical stream polling owns cancellation-safe waker registration.
 mod stream_io;
 
@@ -160,6 +163,7 @@ impl CarrierLane {
 }
 
 struct SessionState {
+    conveyor: conveyor::ConveyorState,
     streams: HashMap<u32, StreamState>,
     closing_streams: HashMap<u32, u64>,
     next_stream_instance: u64,
@@ -295,6 +299,7 @@ impl WebSession {
             timeouts,
             _user_registration: user_registration,
             state: Mutex::new(SessionState {
+                conveyor: conveyor::ConveyorState::default(),
                 streams: HashMap::new(),
                 closing_streams: HashMap::new(),
                 next_stream_instance: 1,

src/web/session/conveyor.rs

@@ -0,0 +1,421 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use sha2::{Digest, Sha256};
+use subtle::ConstantTimeEq;
+use tokio::sync::{Notify, OwnedSemaphorePermit};
+
+use super::{SessionCloseReason, SessionNegotiationPhase, WebSession};
+use crate::config::WebLimitsConfig;
+use crate::web::manager::{ManagerError, TokenHash};
+use crate::web::telemetry::WebSessionLifecycleObservation;
+
+const MAX_WINDOW: usize = 4;
+
+/// Preserves one head body and handler capacity beyond the bounded down-poll plane.
+pub(crate) fn conveyor_waiter_limit(limits: &WebLimitsConfig) -> usize {
+    (limits.max_http_handlers / 4)
+        .min(limits.max_body_readers.saturating_sub(1))
+        .min((limits.max_body_bytes_global / limits.max_body_bytes.max(1)).saturating_sub(1))
+}
+
+/// HTTP-only failures distinguish expired exchanges from terminal protocol faults.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum ConveyorError {
+    /// The bridge already confirmed this sequence, so its digest was retired.
+    Stale,
+    /// A negotiated mode cannot change within the same bearer session.
+    Mode,
+    /// Existing carrier failure semantics remain authoritative.
+    Manager(ManagerError),
+}
+
+impl From<ManagerError> for ConveyorError {
+    fn from(error: ManagerError) -> Self {
+        Self::Manager(error)
+    }
+}
+
+#[derive(Clone, Copy)]
+struct Slot {
+    sequence: u64,
+    owner: u64,
+    digest: Option<TokenHash>,
+    committed: bool,
+}
+
+struct Channel {
+    confirmed: u64,
+    committed: u64,
+    slots: [Option<Slot>; MAX_WINDOW],
+    notify: Arc<Notify>,
+}
+
+impl Channel {
+    fn new() -> Self {
+        Self {
+            confirmed: 0,
+            committed: 0,
+            slots: [None; MAX_WINDOW],
+            notify: Arc::new(Notify::new()),
+        }
+    }
+}
+
+/// Session-owned replay metadata; bodies stay in their bounded HTTP handlers.
+pub(super) struct ConveyorState {
+    offered: u8,
+    mode: Option<bool>,
+    next_owner: u64,
+    channels: HashMap<Option<u32>, Channel>,
+}
+
+impl Default for ConveyorState {
+    fn default() -> Self {
+        Self {
+            offered: 1,
+            mode: None,
+            next_owner: 1,
+            channels: HashMap::new(),
+        }
+    }
+}
+
+impl ConveyorState {
+    /// Publishes commit history under the same lock as frame application.
+    pub(super) fn commit(&mut self, lane: Option<u32>, sequence: u64, digest: TokenHash) {
+        if self.mode != Some(true) {
+            return;
+        }
+        if let Some(channel) = self.channels.get_mut(&lane) {
+            let index = sequence as usize % MAX_WINDOW;
+            channel.slots[index] = Some(Slot {
+                sequence,
+                digest: Some(digest),
+                committed: true,
+                owner: 0,
+            });
+            channel.committed = sequence;
+        }
+    }
+
+    /// Detaches retired-lane ordering state for notification after unlocking.
+    pub(super) fn remove(&mut self, lane: Option<u32>) -> Option<Arc<Notify>> {
+        self.channels.remove(&lane).map(|channel| channel.notify)
+    }
+
+    /// Detaches all waiter notifications for the session's deferred close effects.
+    pub(super) fn clear(&mut self) -> Vec<Arc<Notify>> {
+        self.channels
+            .drain()
+            .map(|(_, channel)| channel.notify)
+            .collect()
+    }
+}
+
+/// An exact request owner; its body and ordinary body permit stay in the HTTP handler.
+pub(crate) struct ConveyorClaim {
+    session: Arc<WebSession>,
+    lane: Option<u32>,
+    sequence: u64,
+    owner: u64,
+    notify: Arc<Notify>,
+    _waiter: Option<OwnedSemaphorePermit>,
+}
+
+impl WebSession {
+    /// Freezes the offer before the newly created session is published by its manager.
+    pub(crate) fn configure_conveyor(&self, requested: u8) {
+        let mut state = self.state.lock();
+        if state.conveyor.mode.is_none() {
+            state.conveyor.offered = if conveyor_waiter_limit(&self.limits) == 0 {
+                1
+            } else {
+                requested.clamp(1, MAX_WINDOW as u8)
+            };
+        }
+    }
+
+    /// Returns the immutable chain ceiling even across a WebSocket candidate.
+    pub(crate) fn conveyor_offer(&self) -> u8 {
+        self.state.lock().conveyor.offered
+    }
+
+    /// Returns the HTTP window without changing carrier capability negotiation.
+    pub(crate) fn up_window(&self) -> u8 {
+        if self.carrier().uses_websocket() {
+            1
+        } else {
+            self.conveyor_offer()
+        }
+    }
+
+    /// Claims an HTTP sequence before allocating a body, or locks in legacy serialization.
+    pub(crate) fn claim_conveyor(
+        self: &Arc<Self>,
+        lane: Option<u32>,
+        sequence: u64,
+        confirmed: Option<u64>,
+    ) -> Result<Option<ConveyorClaim>, ConveyorError> {
+        let mut state = self.state.lock();
+        self.ensure_carrier_active_locked(&state)?;
+        if state.closed {
+            return Err(ManagerError::Closed.into());
+        }
+        let enabled = confirmed.is_some();
+        if (enabled && state.conveyor.offered <= 1)
+            || state.conveyor.mode.is_some_and(|mode| mode != enabled)
+        {
+            return Err(ConveyorError::Mode);
+        }
+        if !enabled {
+            state.conveyor.mode = Some(false);
+            return Ok(None);
+        }
+        let confirmed = confirmed.unwrap_or(0);
+        let window = u64::from(state.conveyor.offered);
+        let existing = state.conveyor.channels.get(&lane);
+        // Lane eviction retires its bounded digest history, not the parent session.
+        if existing.is_none() && lane.is_some() && confirmed > 0 {
+            return Err(ConveyorError::Stale);
+        }
+        let committed = existing.map_or(0, |channel| channel.committed);
+        let floor = existing
+            .map_or(0, |channel| channel.confirmed)
+            .max(confirmed);
+        if confirmed > committed || sequence == 0 || sequence > floor.saturating_add(window) {
+            drop(state);
+            self.close(SessionCloseReason::Protocol);
+            return Err(ManagerError::Protocol.into());
+        }
+        if sequence <= floor {
+            return Err(ConveyorError::Stale);
+        }
+        let index = sequence as usize % MAX_WINDOW;
+        if existing
+            .and_then(|channel| channel.slots[index])
+            .is_some_and(|slot| slot.sequence == sequence && !slot.committed)
+        {
+            return Err(ManagerError::Concurrent.into());
+        }
+        if existing.is_none() {
+            let limit = self
+                .profile
+                .max_streams_per_session
+                .saturating_add(self.limits.max_tombstones_per_session)
+                .saturating_add(1);
+            let provisional = state
+                .conveyor
+                .channels
+                .keys()
+                .filter(|id| id.is_some_and(|id| !state.carrier_lanes.contains_key(&id)))
+                .count();
+            if state.conveyor.channels.len() >= limit
+                || (lane.is_some_and(|id| !state.carrier_lanes.contains_key(&id))
+                    && state.carrier_lanes.len().saturating_add(provisional) >= limit)
+            {
+                return Err(ManagerError::Limit.into());
+            }
+            if lane.is_some_and(|id| {
+                state.closed_streams.contains(&id) && !state.carrier_lanes.contains_key(&id)
+            }) {
+                return Err(ConveyorError::Stale);
+            }
+        }
+        let owner = state.conveyor.next_owner;
+        let next_owner = owner.checked_add(1).ok_or(ManagerError::Limit)?;
+        // All fallible validation precedes the permit so no early return drops it under state.
+        let waiter = if sequence > committed.saturating_add(1) {
+            Some(
+                self.manager
+                    .upgrade()
+                    .ok_or(ManagerError::Closed)?
+                    .try_conveyor_waiter()
+                    .ok_or(ManagerError::Backpressure)?,
+            )
+        } else {
+            None
+        };
+        state.conveyor.mode = Some(true);
+        state.conveyor.next_owner = next_owner;
+        let channel = state
+            .conveyor
+            .channels
+            .entry(lane)
+            .or_insert_with(Channel::new);
+        channel.confirmed = floor;
+        for slot in &mut channel.slots {
+            if slot.is_some_and(|slot| slot.sequence <= floor) {
+                *slot = None;
+            }
+        }
+        let replay = sequence <= channel.committed;
+        if !replay {
+            channel.slots[index] = Some(Slot {
+                sequence,
+                owner,
+                digest: None,
+                committed: false,
+            });
+        }
+        Ok(Some(ConveyorClaim {
+            session: Arc::clone(self),
+            lane,
+            sequence,
+            owner: if replay { 0 } else { owner },
+            notify: Arc::clone(&channel.notify),
+            _waiter: waiter,
+        }))
+    }
+}
+
+impl ConveyorClaim {
+    /// Revalidates exact ownership under the frame-application transaction's lock.
+    pub(super) fn validate_locked(
+        &self,
+        state: &super::SessionState,
+        digest: &TokenHash,
+    ) -> Result<(), ManagerError> {
+        let channel = state
+            .conveyor
+            .channels
+            .get(&self.lane)
+            .filter(|channel| Arc::ptr_eq(&channel.notify, &self.notify))
+            .ok_or(ManagerError::Closed)?;
+        let slot = channel.slots[self.sequence as usize % MAX_WINDOW]
+            .filter(|slot| {
+                slot.sequence == self.sequence
+                    && slot.owner == self.owner
+                    && !slot.committed
+                    && slot
+                        .digest
+                        .is_some_and(|value| bool::from(value.ct_eq(digest)))
+            })
+            .ok_or(ManagerError::Closed)?;
+        if channel.committed.checked_add(1) != Some(slot.sequence) {
+            return Err(ManagerError::Concurrent);
+        }
+        Ok(())
+    }
+
+    /// Applies only the next sequence; duplicates verify the retained digest without replaying frames.
+    pub(crate) async fn process(&self, body: &[u8]) -> Result<u64, ConveyorError> {
+        let digest: TokenHash = Sha256::digest(body).into();
+        let operation = async {
+            loop {
+                let notified = self.notify.notified();
+                tokio::pin!(notified);
+                notified.as_mut().enable();
+                let ready = {
+                    let mut state = self.session.state.lock();
+                    self.session.ensure_carrier_active_locked(&state)?;
+                    if state.closed {
+                        return Err(ManagerError::Closed.into());
+                    }
+                    let Some(channel) = state
+                        .conveyor
+                        .channels
+                        .get_mut(&self.lane)
+                        .filter(|channel| Arc::ptr_eq(&channel.notify, &self.notify))
+                    else {
+                        return Err(ConveyorError::Stale);
+                    };
+                    if self.sequence <= channel.confirmed {
+                        return Err(ConveyorError::Stale);
+                    }
+                    let Some(slot) = &mut channel.slots[self.sequence as usize % MAX_WINDOW] else {
+                        return Err(ConveyorError::Stale);
+                    };
+                    if slot.sequence != self.sequence
+                        || (!slot.committed && slot.owner != self.owner)
+                    {
+                        return Err(ConveyorError::Stale);
+                    }
+                    if slot
+                        .digest
+                        .is_some_and(|previous| !bool::from(previous.ct_eq(&digest)))
+                    {
+                        drop(state);
+                        self.session.close(SessionCloseReason::Protocol);
+                        return Err(ManagerError::Protocol.into());
+                    }
+                    slot.digest = Some(digest);
+                    if slot.committed {
+                        self.session.touch_peer_locked(
+                            &mut state,
+                            Instant::now(),
+                            WebSessionLifecycleObservation::HttpActivityAfterGap,
+                        );
+                        if self.session.automatic_carrier
+                            && state.negotiation_phase != SessionNegotiationPhase::Committed
+                        {
+                            return Err(ManagerError::Backpressure.into());
+                        }
+                        return Ok(self.sequence);
+                    }
+                    channel.committed.checked_add(1) == Some(self.sequence)
+                        && (self.lane.is_some()
+                            || !self
+                                .session
+                                .up_active
+                                .load(std::sync::atomic::Ordering::Acquire))
+                };
+                if ready {
+                    let result = match self.lane {
+                        Some(lane) => self.session.process_up_lane_inner(
+                            lane,
+                            self.sequence,
+                            body,
+                            Some((self, digest)),
+                        ),
+                        None => self
+                            .session
+                            .process_claimed_up(self.sequence, body, self, digest),
+                    };
+                    self.notify.notify_waiters();
+                    return result.map_err(ConveyorError::from);
+                }
+                tokio::select! {
+                    _ = self.session.cancel.cancelled() => return Err(ManagerError::Closed.into()),
+                    _ = notified => {}
+                }
+            }
+        };
+        tokio::time::timeout(
+            Duration::from_secs(self.session.timeouts.body_secs),
+            operation,
+        )
+        .await
+        .map_err(|_| ConveyorError::Manager(ManagerError::Backpressure))?
+    }
+}
+
+impl Drop for ConveyorClaim {
+    fn drop(&mut self) {
+        if self.owner == 0 {
+            return;
+        }
+        let mut state = self.session.state.lock();
+        if let Some(channel) = state
+            .conveyor
+            .channels
+            .get_mut(&self.lane)
+            .filter(|channel| Arc::ptr_eq(&channel.notify, &self.notify))
+        {
+            let slot = &mut channel.slots[self.sequence as usize % MAX_WINDOW];
+            if slot.is_some_and(|slot| !slot.committed && slot.owner == self.owner) {
+                *slot = None;
+            }
+            if channel.committed == 0 && channel.slots.iter().all(Option::is_none) {
+                state.conveyor.channels.remove(&self.lane);
+            }
+        }
+        drop(state);
+        self.notify.notify_waiters();
+    }
+}
+
+#[cfg(test)]
+#[path = "conveyor/tests.rs"]
+mod tests;

src/web/session/conveyor/tests.rs

@@ -0,0 +1,524 @@
+use super::*;
+use std::collections::VecDeque;
+use std::future::Future;
+use std::task::{Context, Poll, Wake, Waker};
+
+use crate::config::{ProxyConfig, WebCarrier, WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig};
+use crate::maestro::generation::test_runtime_generation;
+use crate::web::frame::{self, FrameType};
+use crate::web::manager::{CarrierClientClass, WebProcessRuntime};
+use crate::web::session::{CarrierLane, StreamState};
+use arc_swap::ArcSwap;
+
+fn fixture(
+    carrier: WebCarrier,
+    limits: WebLimitsConfig,
+    automatic: bool,
+) -> (Arc<WebSession>, Arc<WebProcessRuntime>) {
+    let mut config = ProxyConfig::default();
+    config.web.limits = limits.clone();
+    let manager =
+        WebProcessRuntime::start(Arc::new(ArcSwap::from(test_runtime_generation(1, config))));
+    let profile = Arc::new(WebRuntimeProfile {
+        host: "proxy.example.com".into(),
+        public_addr: "203.0.113.10:443".parse().unwrap(),
+        user: "default".into(),
+        secret_mode: WebSecretMode::Plain,
+        carrier,
+        carrier_negotiation_enabled: automatic,
+        carrier_learning: false,
+        carriers: Arc::from([carrier]),
+        carrier_negotiation_deadlines_secs: [3, 5, 8, 12],
+        capability: [0; 32],
+        credential_id: [0; 16],
+        key_fingerprint: "0000000000000000".into(),
+        max_sessions: 2,
+        max_streams: 2,
+        max_streams_per_session: 2,
+    });
+    let session = WebSession::new(
+        Arc::downgrade(&manager),
+        [1; 32],
+        "192.0.2.1".parse().unwrap(),
+        1,
+        profile,
+        [2; 32],
+        carrier,
+        1,
+        [3; 32],
+        None,
+        if automatic {
+            CarrierClientClass::Bridge
+        } else {
+            CarrierClientClass::Legacy
+        },
+        None,
+        automatic,
+        false,
+        limits,
+        WebTimeoutsConfig::default(),
+        None,
+    );
+    session.configure_conveyor(4);
+    (session, manager)
+}
+
+fn claim(session: &Arc<WebSession>, lane: Option<u32>, seq: u64, ack: u64) -> ConveyorClaim {
+    session
+        .claim_conveyor(lane, seq, Some(ack))
+        .unwrap()
+        .unwrap()
+}
+
+fn stream(session: &WebSession, carrier: WebCarrier) -> Option<u32> {
+    let mut state = session.state.lock();
+    state.streams.insert(
+        7,
+        StreamState {
+            instance: 1,
+            inbound: VecDeque::new(),
+            receive_window: frame::INITIAL_STREAM_WINDOW,
+            send_credit: 0,
+            read_waker: None,
+            write_waker: None,
+        },
+    );
+    if carrier.uses_lanes() {
+        state.carrier_lanes.insert(7, CarrierLane::new(7));
+        Some(7)
+    } else {
+        None
+    }
+}
+
+async fn cleanup(session: Arc<WebSession>, manager: Arc<WebProcessRuntime>) {
+    session.close(SessionCloseReason::ApiClose);
+    session.wait().await;
+    manager.shutdown().await;
+}
+
+#[tokio::test]
+async fn reordered_data_and_window_apply_once_in_sequence_for_both_carriers() {
+    for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] {
+        let (session, manager) = fixture(carrier, WebLimitsConfig::default(), false);
+        let lane = stream(&session, carrier);
+        let first = frame::encode(FrameType::Data, 7, b"first");
+        let second = frame::encode(FrameType::Data, 7, b"second");
+        let credit = frame::encode(FrameType::Window, 7, &5u32.to_be_bytes());
+        let second_claim = claim(&session, lane, 2, 0);
+        let credit_claim = claim(&session, lane, 3, 0);
+        let mut second_wait = Box::pin(second_claim.process(&second));
+        let mut credit_wait = Box::pin(credit_claim.process(&credit));
+        assert!(futures::poll!(&mut second_wait).is_pending());
+        assert!(futures::poll!(&mut credit_wait).is_pending());
+        assert!(session.state.lock().streams[&7].inbound.is_empty());
+        assert_eq!(claim(&session, lane, 1, 0).process(&first).await, Ok(1));
+        assert!(futures::poll!(&mut credit_wait).is_pending());
+        assert_eq!(second_wait.await, Ok(2));
+        assert_eq!(credit_wait.await, Ok(3));
+        for (seq, body) in [(1, &first), (2, &second), (3, &credit)] {
+            assert_eq!(claim(&session, lane, seq, 0).process(body).await, Ok(seq));
+        }
+        {
+            let state = session.state.lock();
+            let stream = &state.streams[&7];
+            assert_eq!(stream.inbound.len(), 2);
+            assert_eq!(stream.inbound[0].bytes.as_ref(), b"first");
+            assert_eq!(stream.inbound[1].bytes.as_ref(), b"second");
+            assert_eq!(stream.send_credit, 5);
+            assert_eq!(state.conveyor.channels[&lane].committed, 3);
+        }
+        cleanup(session, manager).await;
+    }
+}
+
+#[tokio::test]
+async fn confirmed_floor_retires_only_acknowledged_history_and_never_replays_it() {
+    let (session, manager) = fixture(WebCarrier::Https, WebLimitsConfig::default(), false);
+    let body = frame::encode(FrameType::Pong, 0, b"same");
+    for seq in 1..=4 {
+        assert_eq!(claim(&session, None, seq, 0).process(&body).await, Ok(seq));
+    }
+    // ACK2 through ACK4 do not constitute confirmation of a lost ACK1.
+    assert_eq!(claim(&session, None, 1, 0).process(&body).await, Ok(1));
+    assert_eq!(claim(&session, None, 5, 4).process(&body).await, Ok(5));
+    assert!(matches!(
+        session.claim_conveyor(None, 1, Some(0)),
+        Err(ConveyorError::Stale)
+    ));
+    assert!(!session.state.lock().closed);
+    assert_eq!(session.state.lock().last_up_sequence, 5);
+    assert_eq!(
+        session.state.lock().conveyor.channels[&None]
+            .slots
+            .iter()
+            .flatten()
+            .count(),
+        1
+    );
+    cleanup(session, manager).await;
+}
+
+#[tokio::test]
+async fn digest_mismatch_and_window_violations_are_terminal() {
+    for invalid in [0, 1, 2] {
+        let (session, manager) = fixture(WebCarrier::Https, WebLimitsConfig::default(), false);
+        let body = frame::encode(FrameType::Pong, 0, b"first");
+        assert_eq!(claim(&session, None, 1, 0).process(&body).await, Ok(1));
+        if invalid == 0 {
+            let changed = frame::encode(FrameType::Pong, 0, b"changed");
+            assert_eq!(
+                claim(&session, None, 1, 0).process(&changed).await,
+                Err(ConveyorError::Manager(ManagerError::Protocol))
+            );
+        } else {
+            let (sequence, confirmed) = if invalid == 1 { (5, 0) } else { (3, 2) };
+            assert!(matches!(
+                session.claim_conveyor(None, sequence, Some(confirmed)),
+                Err(ConveyorError::Manager(ManagerError::Protocol))
+            ));
+        }
+        assert!(session.state.lock().closed);
+        cleanup(session, manager).await;
+    }
+}
+
+#[tokio::test]
+async fn duplicate_pending_claims_and_cancelled_owners_do_not_consume_or_erase_new_slots() {
+    let (session, manager) = fixture(WebCarrier::Https, WebLimitsConfig::default(), false);
+    let first = claim(&session, None, 2, 0);
+    assert!(matches!(
+        session.claim_conveyor(None, 2, Some(0)),
+        Err(ConveyorError::Manager(ManagerError::Concurrent))
+    ));
+    drop(first);
+    assert!(session.state.lock().conveyor.channels.is_empty());
+    let cancelled = claim(&session, None, 2, 0);
+    session.with_state_effects(|state, effects| {
+        effects.notify(state.conveyor.remove(None).unwrap());
+    });
+    let replacement = claim(&session, None, 2, 0);
+    drop(cancelled);
+    assert_eq!(
+        session.state.lock().conveyor.channels[&None].slots[2]
+            .unwrap()
+            .owner,
+        replacement.owner
+    );
+    drop(replacement);
+    assert!(session.state.lock().conveyor.channels.is_empty());
+    cleanup(session, manager).await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn missing_head_times_out_without_committing_and_returns_waiter_capacity() {
+    let (session, manager) = fixture(WebCarrier::Https, WebLimitsConfig::default(), false);
+    let body = frame::encode(FrameType::Pong, 0, &[]);
+    let tail = claim(&session, None, 2, 0);
+    let mut waiting = Box::pin(tail.process(&body));
+    assert!(futures::poll!(&mut waiting).is_pending());
+    tokio::time::advance(Duration::from_secs(session.timeouts.body_secs)).await;
+    assert_eq!(
+        waiting.await,
+        Err(ConveyorError::Manager(ManagerError::Backpressure))
+    );
+    assert_eq!(session.state.lock().last_up_sequence, 0);
+    drop(tail);
+    let mut permits = Vec::new();
+    while let Some(permit) = manager.try_conveyor_waiter() {
+        permits.push(permit);
+    }
+    assert_eq!(
+        permits.len(),
+        conveyor_waiter_limit(&WebLimitsConfig::default())
+    );
+    drop(permits);
+    cleanup(session, manager).await;
+}
+
+#[test]
+fn waiter_ceiling_always_preserves_head_handler_reader_and_bytes() {
+    let defaults = WebLimitsConfig::default();
+    assert_eq!(conveyor_waiter_limit(&defaults), 31);
+    for (handlers, readers, bodies, expected) in [
+        (8, 4, 8, 2),
+        (100, 2, 8, 1),
+        (100, 8, 2, 1),
+        (3, 8, 8, 0),
+        (8, 1, 8, 0),
+        (8, 8, 1, 0),
+    ] {
+        let limits = WebLimitsConfig {
+            max_http_handlers: handlers,
+            max_body_readers: readers,
+            max_body_bytes: 32,
+            max_body_bytes_global: bodies * 32,
+            ..defaults.clone()
+        };
+        assert_eq!(conveyor_waiter_limit(&limits), expected);
+    }
+}
+
+#[tokio::test]
+async fn future_request_saturation_keeps_head_admissible_before_any_body_is_read() {
+    let limits = WebLimitsConfig {
+        max_http_handlers: 8,
+        max_body_readers: 3,
+        max_body_bytes: 32,
+        max_body_bytes_global: 96,
+        ..WebLimitsConfig::default()
+    };
+    let (session, manager) = fixture(WebCarrier::HttpsLanes, limits, false);
+    let a = claim(&session, Some(7), 2, 0);
+    let b = claim(&session, Some(8), 2, 0);
+    let a_body = manager.try_body_budget(32).unwrap();
+    let b_body = manager.try_body_budget(32).unwrap();
+    assert!(matches!(
+        session.claim_conveyor(Some(9), 2, Some(0)),
+        Err(ConveyorError::Manager(ManagerError::Backpressure))
+    ));
+    assert!(
+        !session
+            .state
+            .lock()
+            .conveyor
+            .channels
+            .contains_key(&Some(9))
+    );
+    let head = claim(&session, Some(7), 1, 0);
+    let head_body = manager
+        .try_body_budget(32)
+        .expect("head reader and byte reserve");
+    assert!(!session.state.lock().carrier_lanes.contains_key(&7));
+    drop((a, b, head, a_body, b_body, head_body));
+    cleanup(session, manager).await;
+}
+
+#[tokio::test]
+async fn mode_is_frozen_and_control_only_ack_cannot_commit_an_automatic_carrier() {
+    for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] {
+        let (session, manager) = fixture(carrier, WebLimitsConfig::default(), true);
+        let lane = carrier.uses_lanes().then_some(0);
+        let body = frame::encode(FrameType::Pong, 0, &[]);
+        for _ in 0..2 {
+            assert_eq!(
+                claim(&session, lane, 1, 0).process(&body).await,
+                Err(ConveyorError::Manager(ManagerError::Backpressure))
+            );
+        }
+        assert!(!session.is_carrier_committed());
+        assert!(matches!(
+            session.claim_conveyor(lane, 2, None),
+            Err(ConveyorError::Mode)
+        ));
+        assert!(!session.state.lock().closed);
+        cleanup(session, manager).await;
+    }
+    let (session, manager) = fixture(WebCarrier::Https, WebLimitsConfig::default(), false);
+    assert!(session.claim_conveyor(None, 1, None).unwrap().is_none());
+    assert!(matches!(
+        session.claim_conveyor(None, 1, Some(0)),
+        Err(ConveyorError::Mode)
+    ));
+    cleanup(session, manager).await;
+}
+
+#[tokio::test]
+async fn first_data_on_unknown_lane_is_not_silently_acknowledged() {
+    let (session, manager) = fixture(WebCarrier::HttpsLanes, WebLimitsConfig::default(), false);
+    let body = frame::encode(FrameType::Data, 7, b"not opened");
+    assert_eq!(
+        claim(&session, Some(7), 1, 0).process(&body).await,
+        Err(ConveyorError::Manager(ManagerError::Protocol))
+    );
+    assert!(session.state.lock().closed);
+    cleanup(session, manager).await;
+}
+
+struct LockProbe {
+    session: Arc<WebSession>,
+    notified: std::sync::atomic::AtomicBool,
+}
+
+impl Wake for LockProbe {
+    fn wake(self: Arc<Self>) {
+        self.wake_by_ref();
+    }
+    fn wake_by_ref(self: &Arc<Self>) {
+        assert!(
+            self.session.state.try_lock().is_some(),
+            "wake under session lock"
+        );
+        self.notified
+            .store(true, std::sync::atomic::Ordering::Release);
+    }
+}
+
+#[tokio::test]
+async fn close_and_lane_retirement_wake_parked_requests_only_after_unlock() {
+    for close in [false, true] {
+        let (session, manager) = fixture(WebCarrier::HttpsLanes, WebLimitsConfig::default(), false);
+        let body = frame::encode(FrameType::Data, 7, b"future");
+        let tail = claim(&session, Some(7), 2, 0);
+        let mut waiting = Box::pin(tail.process(&body));
+        let probe = Arc::new(LockProbe {
+            session: Arc::clone(&session),
+            notified: std::sync::atomic::AtomicBool::new(false),
+        });
+        let waker = Waker::from(Arc::clone(&probe));
+        assert!(matches!(
+            waiting.as_mut().poll(&mut Context::from_waker(&waker)),
+            Poll::Pending
+        ));
+        if close {
+            session.close(SessionCloseReason::ApiClose);
+        } else {
+            session.with_state_effects(|state, effects| {
+                session.release_lane_locked(state, effects, 7);
+            });
+        }
+        assert!(probe.notified.load(std::sync::atomic::Ordering::Acquire));
+        assert!(waiting.await.is_err());
+        assert!(session.state.lock().conveyor.channels.is_empty());
+        drop(tail);
+        cleanup(session, manager).await;
+    }
+}
+
+#[tokio::test]
+async fn delayed_retry_after_lane_history_retirement_is_request_local() {
+    let (session, manager) = fixture(WebCarrier::HttpsLanes, WebLimitsConfig::default(), false);
+    let lane = stream(&session, WebCarrier::HttpsLanes);
+    let body = frame::encode(FrameType::Window, 7, &1u32.to_be_bytes());
+    for seq in 1..=2 {
+        assert_eq!(
+            claim(&session, lane, seq, seq - 1).process(&body).await,
+            Ok(seq)
+        );
+    }
+    session.with_state_effects(|state, effects| {
+        state.streams.remove(&7);
+        session.release_lane_locked(state, effects, 7);
+    });
+    assert!(matches!(
+        session.claim_conveyor(lane, 2, Some(1)),
+        Err(ConveyorError::Stale)
+    ));
+    assert!(!session.state.lock().closed);
+    cleanup(session, manager).await;
+}
+
+#[tokio::test]
+async fn verified_replay_renews_peer_activity_without_reapplying_credit() {
+    let (session, manager) = fixture(WebCarrier::Https, WebLimitsConfig::default(), false);
+    stream(&session, WebCarrier::Https);
+    let body = frame::encode(FrameType::Window, 7, &1u32.to_be_bytes());
+    assert_eq!(claim(&session, None, 1, 0).process(&body).await, Ok(1));
+    let now = std::time::Instant::now();
+    session.state.lock().activity =
+        crate::web::session::activity::SessionActivity::new(now - Duration::from_secs(60));
+    assert_eq!(claim(&session, None, 1, 0).process(&body).await, Ok(1));
+    assert!(session.state.lock().activity.peer_idle(now) < Duration::from_secs(1));
+    assert_eq!(session.state.lock().streams[&7].send_credit, 1);
+    cleanup(session, manager).await;
+}
+
+#[tokio::test]
+async fn full_lane_registry_still_admits_first_control_channel_request() {
+    let limits = WebLimitsConfig {
+        max_tombstones_per_session: 0,
+        ..WebLimitsConfig::default()
+    };
+    let (session, manager) = fixture(WebCarrier::HttpsLanes, limits, false);
+    stream(&session, WebCarrier::HttpsLanes);
+    session
+        .state
+        .lock()
+        .carrier_lanes
+        .insert(8, CarrierLane::new(8));
+    let body = frame::encode(FrameType::Pong, 0, &[]);
+    assert_eq!(claim(&session, Some(0), 1, 0).process(&body).await, Ok(1));
+    cleanup(session, manager).await;
+}
+
+#[test]
+fn per_channel_memory_estimate_covers_window_notify_and_hash_table_slack() {
+    // The memory-envelope validator reserves 512 additional bytes per bounded lane slot.
+    let key_and_value = std::mem::size_of::<(Option<u32>, Channel)>();
+    let table_slack = key_and_value.div_ceil(7);
+    let notify_allocation = std::mem::size_of::<Notify>() + 2 * std::mem::size_of::<usize>();
+    assert!(key_and_value + table_slack + notify_allocation + 32 <= 512);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn concurrent_windows_preserve_contiguous_application_under_scheduler_pressure() {
+    let (session, manager) = fixture(WebCarrier::Https, WebLimitsConfig::default(), false);
+    let body = frame::encode(FrameType::Pong, 0, &[]);
+    for window in 0..256u64 {
+        let confirmed = window * 4;
+        let mut tasks = Vec::new();
+        for offset in (1..=4).rev() {
+            let claim = claim(&session, None, confirmed + offset, confirmed);
+            let body = body.clone();
+            tasks.push(tokio::spawn(async move {
+                tokio::task::yield_now().await;
+                claim.process(&body).await
+            }));
+        }
+        for (task, offset) in tasks.into_iter().zip((1..=4).rev()) {
+            assert_eq!(task.await.unwrap(), Ok(confirmed + offset));
+        }
+        let state = session.state.lock();
+        assert_eq!(state.last_up_sequence, confirmed + 4);
+        assert_eq!(state.conveyor.channels[&None].committed, confirmed + 4);
+        assert_eq!(
+            state.conveyor.channels[&None]
+                .slots
+                .iter()
+                .flatten()
+                .count(),
+            4
+        );
+    }
+    cleanup(session, manager).await;
+}
+
+#[tokio::test]
+async fn next_head_waits_for_previous_synchronous_uplink_guard_to_drop() {
+    let (session, manager) = fixture(WebCarrier::Https, WebLimitsConfig::default(), false);
+    let body = frame::encode(FrameType::Pong, 0, &[]);
+    let head = claim(&session, None, 1, 0);
+    session
+        .up_active
+        .store(true, std::sync::atomic::Ordering::Release);
+    let mut waiting = Box::pin(head.process(&body));
+    assert!(futures::poll!(&mut waiting).is_pending());
+    session
+        .up_active
+        .store(false, std::sync::atomic::Ordering::Release);
+    head.notify.notify_waiters();
+    assert_eq!(waiting.await, Ok(1));
+    cleanup(session, manager).await;
+}
+
+#[tokio::test]
+async fn new_lane_future_data_waits_for_open_without_creating_a_backend_early() {
+    let (session, manager) = fixture(WebCarrier::HttpsLanes, WebLimitsConfig::default(), true);
+    let body = frame::encode(FrameType::Data, 7, b"first inner bytes");
+    let future = claim(&session, Some(7), 2, 0);
+    let mut waiting = Box::pin(future.process(&body));
+    assert!(futures::poll!(&mut waiting).is_pending());
+    assert!(session.state.lock().streams.is_empty());
+    assert!(!session.state.lock().carrier_lanes.contains_key(&7));
+    assert_eq!(session.tasks_live(), 0);
+    assert!(!session.is_carrier_committed());
+    let open = frame::encode(FrameType::Open, 7, &[]);
+    assert_eq!(claim(&session, Some(7), 1, 0).process(&open).await, Ok(1));
+    assert!(session.is_carrier_committed());
+    assert_eq!(waiting.await, Ok(2));
+    assert_eq!(
+        session.state.lock().streams[&7].inbound[0].bytes,
+        b"first inner bytes"[..]
+    );
+    cleanup(session, manager).await;
+}

src/web/session/lane_uplink.rs

@@ -20,6 +20,17 @@ impl WebSession {
         lane_id: u32,
         sequence: u64,
         body: &[u8],
+    ) -> Result<u64, ManagerError> {
+        self.process_up_lane_inner(lane_id, sequence, body, None)
+    }
+
+    /// Applies one lane batch only while its optional conveyor claim still owns the turn.
+    pub(super) fn process_up_lane_inner(
+        self: &Arc<Self>,
+        lane_id: u32,
+        sequence: u64,
+        body: &[u8],
+        claim: Option<(&super::conveyor::ConveyorClaim, TokenHash)>,
     ) -> Result<u64, ManagerError> {
         if self.carrier() != WebCarrier::HttpsLanes || lane_id > frame::MAX_STREAM_ID {
             return Err(ManagerError::Protocol);
@@ -42,7 +53,8 @@ impl WebSession {
             self.close(SessionCloseReason::Protocol);
             return Err(ManagerError::Protocol);
         }
-        let digest: TokenHash = Sha256::digest(body).into();
+        let digest: TokenHash =
+            claim.map_or_else(|| Sha256::digest(body).into(), |(_, digest)| digest);
         let mut opened = Vec::new();
         let mut committed = false;
         let mut healthy = None;
@@ -53,9 +65,13 @@ impl WebSession {
                 return Err(ManagerError::Closed);
             }
             self.ensure_carrier_active_locked(&state)?;
+            if let Some((claim, _)) = claim {
+                claim.validate_locked(&state, &digest)?;
+            }
             let new_lane = !state.carrier_lanes.contains_key(&lane_id);
             if new_lane {
-                if lane_id != 0
+                if claim.is_none()
+                    && lane_id != 0
                     && frames
                         .first()
                         .is_some_and(|value| value.frame_type != FrameType::Open)
@@ -190,6 +206,7 @@ impl WebSession {
                 }
             }
             if applied {
+                state.conveyor.commit(Some(lane_id), sequence, digest);
                 (committed, healthy) = self.record_uplink_progress_locked(&mut state, progress);
             }
             applied.then_some(sequence).ok_or(ManagerError::Closed)

src/web/session/lanes.rs

@@ -469,12 +469,16 @@ impl WebSession {
         }
     }
 
+    /// Retires exact lane queues and ordering history, deferring every waiter notification.
     pub(super) fn release_lane_locked(
         &self,
         state: &mut SessionState,
         effects: &mut DeferredSessionEffects,
         lane_id: u32,
     ) {
+        if let Some(notify) = state.conveyor.remove(Some(lane_id)) {
+            effects.notify(notify);
+        }
         let Some(mut lane) = state.carrier_lanes.remove(&lane_id) else {
             return;
         };

src/web/session/lifecycle.rs

@@ -279,6 +279,9 @@ impl WebSession {
         let recovery_closed_before_commit = self.recovery && !state.recovery_committed;
         state.close_requested = Some(reason);
         state.closed = true;
+        for notify in state.conveyor.clear() {
+            effects.notify(notify);
+        }
         if reason == SessionCloseReason::CarrierSuperseded {
             state.negotiation_phase = SessionNegotiationPhase::Superseded;
         }

src/web/session/uplink.rs

@@ -39,6 +39,7 @@ impl WebSession {
             sequence,
             body,
             WebSessionLifecycleObservation::HttpActivityAfterGap,
+            None,
         )?;
         if self.automatic_carrier && !progressed && !self.is_carrier_committed() {
             return Err(ManagerError::Backpressure);
@@ -56,15 +57,37 @@ impl WebSession {
             sequence,
             body,
             WebSessionLifecycleObservation::WebSocketActivityAfterGap,
+            None,
         )
         .map(|(_, progress)| progress)
     }
 
+    /// Applies an admitted conveyor head without weakening the carrier commit gate.
+    pub(super) fn process_claimed_up(
+        self: &Arc<Self>,
+        sequence: u64,
+        body: &[u8],
+        claim: &super::conveyor::ConveyorClaim,
+        digest: TokenHash,
+    ) -> Result<u64, ManagerError> {
+        let (acknowledged, progressed) = self.process_up_inner(
+            sequence,
+            body,
+            WebSessionLifecycleObservation::HttpActivityAfterGap,
+            Some((claim, digest)),
+        )?;
+        if self.automatic_carrier && !progressed && !self.is_carrier_committed() {
+            return Err(ManagerError::Backpressure);
+        }
+        Ok(acknowledged)
+    }
+
     fn process_up_inner(
         self: &Arc<Self>,
         sequence: u64,
         body: &[u8],
         observation: WebSessionLifecycleObservation,
+        claim: Option<(&super::conveyor::ConveyorClaim, TokenHash)>,
     ) -> Result<(u64, bool), ManagerError> {
         if !self.carrier().is_multiplexed() {
             return Err(ManagerError::Protocol);
@@ -95,7 +118,8 @@ impl WebSession {
             self.close(SessionCloseReason::Protocol);
             return Err(ManagerError::Protocol);
         }
-        let digest: TokenHash = Sha256::digest(body).into();
+        let digest: TokenHash =
+            claim.map_or_else(|| Sha256::digest(body).into(), |(_, digest)| digest);
         let mut opened = Vec::new();
         let mut committed = false;
         let mut healthy = None;
@@ -106,6 +130,9 @@ impl WebSession {
                 return Err(ManagerError::Closed);
             }
             self.ensure_carrier_active_locked(&state)?;
+            if let Some((claim, _)) = claim {
+                claim.validate_locked(&state, &digest)?;
+            }
             if sequence == state.last_up_sequence && sequence != 0 {
                 return if bool::from(state.last_up_digest.ct_eq(&digest)) {
                     self.touch_peer_locked(&mut state, Instant::now(), observation);
@@ -155,6 +182,7 @@ impl WebSession {
             } else {
                 state.last_up_sequence = sequence;
                 state.last_up_digest = digest;
+                state.conveyor.commit(None, sequence, digest);
                 (committed, healthy) = self.record_uplink_progress_locked(&mut state, progress);
                 Ok((sequence, progress.any()))
             }