From 823c6cbc7fa30e5a50463869774aced10524d034 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sat, 3 Oct 2026 16:02:18 +0300 Subject: [PATCH 1/3] Bounded WEB HTTP conveyor with replay-safe streaming downlink added Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com> --- src/api/config_edit.rs | 4 + src/api/config_edit/conveyor_tests.rs | 27 + src/api/config_edit/tests.rs | 3 +- src/config/hot_reload.rs | 4 + src/config/hot_reload/conveyor_tests.rs | 20 + src/config/load/strict_keys.rs | 1 + src/config/load/validate_web/memory.rs | 3 +- .../tests/load_basic_tests/web_tests.rs | 3 + .../web_tests/conveyor_tests.rs | 44 ++ src/config/types/web.rs | 4 + src/config/types/web/defaults.rs | 5 + src/web/bridge.rs | 4 + src/web/bridge/behavior_tests.js | 251 +++++++++ src/web/bridge/behavior_tests.rs | 26 + src/web/bridge/conveyor.js | 106 ++++ src/web/bridge/document.html | 6 + src/web/bridge/downlink.js | 86 +++ src/web/bridge/request.js | 11 +- src/web/bridge/runtime.js | 183 ++---- src/web/bridge/tests.rs | 24 +- src/web/http.rs | 73 +-- src/web/http/carrier_method_tests.rs | 3 + src/web/http/conveyor_tests.rs | 358 ++++++++++++ src/web/http/down.rs | 2 + src/web/http/session.rs | 18 + src/web/http/up.rs | 109 ++++ src/web/manager.rs | 6 + src/web/manager/budget.rs | 5 + src/web/manager/lifecycle.rs | 1 + src/web/manager/negotiation.rs | 16 + src/web/manager/session_creation.rs | 7 + .../manager/session_creation/replacement.rs | 2 + src/web/session.rs | 5 + src/web/session/conveyor.rs | 421 ++++++++++++++ src/web/session/conveyor/tests.rs | 524 ++++++++++++++++++ src/web/session/lane_uplink.rs | 21 +- src/web/session/lanes.rs | 4 + src/web/session/lifecycle.rs | 3 + src/web/session/uplink.rs | 30 +- 39 files changed, 2186 insertions(+), 237 deletions(-) create mode 100644 src/api/config_edit/conveyor_tests.rs create mode 100644 src/config/hot_reload/conveyor_tests.rs create mode 100644 src/config/tests/load_basic_tests/web_tests/conveyor_tests.rs create mode 100644 src/web/bridge/behavior_tests.js create mode 100644 src/web/bridge/behavior_tests.rs create mode 100644 src/web/bridge/conveyor.js create mode 100644 src/web/bridge/downlink.js create mode 100644 src/web/http/conveyor_tests.rs create mode 100644 src/web/http/up.rs create mode 100644 src/web/session/conveyor.rs create mode 100644 src/web/session/conveyor/tests.rs diff --git a/src/api/config_edit.rs b/src/api/config_edit.rs index b1e8719..809dac1 100644 --- a/src/api/config_edit.rs +++ b/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; diff --git a/src/api/config_edit/conveyor_tests.rs b/src/api/config_edit/conveyor_tests.rs new file mode 100644 index 0000000..a137f66 --- /dev/null +++ b/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); + } +} diff --git a/src/api/config_edit/tests.rs b/src/api/config_edit/tests.rs index e5dbcf7..624a820 100644 --- a/src/api/config_edit/tests.rs +++ b/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 diff --git a/src/config/hot_reload.rs b/src/config/hot_reload.rs index 101eac0..8cb4d78 100644 --- a/src/config/hot_reload.rs +++ b/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; diff --git a/src/config/hot_reload/conveyor_tests.rs b/src/config/hot_reload/conveyor_tests.rs new file mode 100644 index 0000000..b0af200 --- /dev/null +++ b/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; + } +} diff --git a/src/config/load/strict_keys.rs b/src/config/load/strict_keys.rs index 6b5fa66..e02c5db 100644 --- a/src/config/load/strict_keys.rs +++ b/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", diff --git a/src/config/load/validate_web/memory.rs b/src/config/load/validate_web/memory.rs index 3a8d161..491e053 100644 --- a/src/config/load/validate_web/memory.rs +++ b/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; diff --git a/src/config/tests/load_basic_tests/web_tests.rs b/src/config/tests/load_basic_tests/web_tests.rs index 1191676..16b085f 100644 --- a/src/config/tests/load_basic_tests/web_tests.rs +++ b/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" diff --git a/src/config/tests/load_basic_tests/web_tests/conveyor_tests.rs b/src/config/tests/load_basic_tests/web_tests/conveyor_tests.rs new file mode 100644 index 0000000..941ea03 --- /dev/null +++ b/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::(json).unwrap().conveyor, + enabled + ); + let toml = toml::to_string(&config.web).unwrap(); + assert_eq!( + toml::from_str::(&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")); + } +} diff --git a/src/config/types/web.rs b/src/config/types/web.rs index 4f8cb5a..4aaf874 100644 --- a/src/config/types/web.rs +++ b/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(), diff --git a/src/config/types/web/defaults.rs b/src/config/types/web/defaults.rs index 10e91e6..8680e80 100644 --- a/src/config/types/web/defaults.rs +++ b/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 { diff --git a/src/web/bridge.rs b/src/web/bridge.rs index 5a57389..c0a485a 100644 --- a/src/web/bridge.rs +++ b/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. diff --git a/src/web/bridge/behavior_tests.js b/src/web/bridge/behavior_tests.js new file mode 100644 index 0000000..cdbc863 --- /dev/null +++ b/src/web/bridge/behavior_tests.js @@ -0,0 +1,251 @@ +'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(/]*>([\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;offsetvalue[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(); +}); + +(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})(); diff --git a/src/web/bridge/behavior_tests.rs b/src/web/bridge/behavior_tests.rs new file mode 100644 index 0000000..093d791 --- /dev/null +++ b/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) + ); +} diff --git a/src/web/bridge/conveyor.js b/src/web/bridge/conveyor.js new file mode 100644 index 0000000..1c138a4 --- /dev/null +++ b/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}); +})(); diff --git a/src/web/bridge/document.html b/src/web/bridge/document.html index 7c98b5d..38ebabc 100644 --- a/src/web/bridge/document.html +++ b/src/web/bridge/document.html @@ -20,6 +20,12 @@ __BUFFER_RUNTIME__ __RECOVERY_RUNTIME__ + + diff --git a/src/web/bridge/downlink.js b/src/web/bridge/downlink.js new file mode 100644 index 0000000..98ef42c --- /dev/null +++ b/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=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{})}catch(error){} + reader.releaseLock(); + } + } + return {read,close(){closed=true;dispose()}}; +} +globalThis.TelemtBridgeDownlink=Object.freeze({create,budget}); +})(); diff --git a/src/web/bridge/request.js b/src/web/bridge/request.js index 27bd638..4041fe4 100644 --- a/src/web/bridge/request.js +++ b/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(attemptcontroller.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); diff --git a/src/web/bridge/runtime.js b/src/web/bridge/runtime.js index 88991eb..45d16ab 100644 --- a/src/web/bridge/runtime.js +++ b/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,pollController=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,6 +58,7 @@ 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(socket){const previous=socket;socket=null;previous.close()}socketReady=false;cancelBatch(upLease);releasePending(upPending,null); @@ -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,6 +142,7 @@ 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; @@ -198,8 +201,8 @@ function startCandidate(probe,epoch){ else advanceCarrier('protocol',epoch); } function maybeStartCandidate(){ - if(closed||carrierCommitted||!sessionToken||candidateRunning)return;const epoch=attemptEpoch; - let probe;try{probe=findProbe(probeCoalesceMs>0)}catch(error){fail('protocol');return}if(!probe)return; + if(closed||carrierCommitted||!sessionToken)return;if(candidateRunning){try{flushHttpPending()}catch(error){fail(failureReason(error,'protocol'))}return}const epoch=attemptEpoch; + 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 +222,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){ - try{ - const headers={'X-Up-Seq':'1'},token=sessionToken,controller=attemptController,body=probe.data;if(laneID!==null)headers['X-Lane-ID']=String(laneID); - const response=await request('/api/v1/up',options(carrierMethod,token,body,headers,controller.signal)); - if(closed||epoch!==attemptEpoch)return - if(response.status!==204){advanceCarrier('http',epoch);return} - if(response.headers.get('X-Up-Ack')!=='1'){advanceCarrier('protocol',epoch);return} - if(laneID===null)upSequence=2;else ensureLane(laneID).sequence=2; - commitCarrier(probe,epoch); - }catch(error){if(!closed&&epoch===attemptEpoch)advanceCarrier(failureReason(error,'network'),epoch)} +function stopHttp(restore){ + if(!httpCarrier)return;const saved=httpCarrier.close(restore);httpCarrier=null; + for(const data of saved){if(!reserve(data,null)){fail('capacity');return}pending.push(data)} +} +function flushHttpPending(){ + while(httpCarrier&&pending.length&&!closed){const data=pending.shift();release(data.byteLength,1,null);httpCarrier.enqueue(data,true)} + if(httpCarrier&&!closed)httpCarrier.flush(); +} +function probeHttp(probe,laneID,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 +327,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 +338,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{ - 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(pollController)pollController.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(); diff --git a/src/web/bridge/tests.rs b/src/web/bridge/tests.rs index 9342b0a..7eca039 100644 --- a/src/web/bridge/tests.rs +++ b/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); - let requests: Vec<_> = page - .body - .lines() - .filter(|line| { - line.contains("request('/api/v1/up'") || line.contains("request('/api/v1/down'") - }) - .collect(); - assert_eq!(requests.len(), 9); - assert!( - requests - .iter() - .all(|line| line.contains("options(carrierMethod,")) - ); + // Execute the rendered POST/PUT page, including retries, instead of counting call sites. + behavior_tests::run(&page); 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 - .contains("async function send(path,frozenOptions,remainingBudget,maxAttempts)") + page.body.contains( + "async function send(path,frozenOptions,remainingBudget,maxAttempts,receiver)" + ) ); assert!(!page.body.contains("makeOptions")); assert!(page.body.contains( diff --git a/src/web/http.rs b/src/web/http.rs index 9a24d5a..bb387c7 100644 --- a/src/web/http.rs +++ b/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; type HttpBody = UnsyncBoxBody; @@ -457,76 +460,6 @@ async fn handle_api( _ => serve_decoy(request, vhost, true, &runtime).await, } } -async fn handle_up( - request: Request, - runtime: Arc, - vhost: Arc, - 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(request: &mut Request) { if request.uri().query().is_some() && let Ok(uri) = request.uri().path().parse() diff --git a/src/web/http/carrier_method_tests.rs b/src/web/http/carrier_method_tests.rs index 4ec4d62..45cbb9c 100644 --- a/src/web/http/carrier_method_tests.rs +++ b/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 { 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", diff --git a/src/web/http/conveyor_tests.rs b/src/web/http/conveyor_tests.rs new file mode 100644 index 0000000..09f9984 --- /dev/null +++ b/src/web/http/conveyor_tests.rs @@ -0,0 +1,358 @@ +use super::*; + +async fn create_with_headers( + listener: &TcpListener, + runtime: &Arc, + bootstrap: &str, + headers: &str, +) -> Vec { + 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 { + 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; +} diff --git a/src/web/http/down.rs b/src/web/http/down.rs index 9f188fe..d612a5a 100644 --- a/src/web/http/down.rs +++ b/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(); }; diff --git a/src/web/http/session.rs b/src/web/http/session.rs index 89d1f21..665a891 100644 --- a/src/web/http/session.rs +++ b/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"), diff --git a/src/web/http/up.rs b/src/web/http/up.rs new file mode 100644 index 0000000..3bc0f39 --- /dev/null +++ b/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, + runtime: Arc, + vhost: Arc, + 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, + } +} diff --git a/src/web/manager.rs b/src/web/manager.rs index f0f0816..bd9bba5 100644 --- a/src/web/manager.rs +++ b/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, /// 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, body_readers: Arc, body_bytes: Arc, + conveyor_waiters: Arc, stream_handshakes: Arc, websocket_connections: Arc, websockets: Mutex, @@ -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()), diff --git a/src/web/manager/budget.rs b/src/web/manager/budget.rs index efae15e..170c13b 100644 --- a/src/web/manager/budget.rs +++ b/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 { + 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, diff --git a/src/web/manager/lifecycle.rs b/src/web/manager/lifecycle.rs index b1b9d06..dabd3eb 100644 --- a/src/web/manager/lifecycle.rs +++ b/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 = { diff --git a/src/web/manager/negotiation.rs b/src/web/manager/negotiation.rs index 1231afb..95f5ad2 100644 --- a/src/web/manager/negotiation.rs +++ b/src/web/manager/negotiation.rs @@ -130,6 +130,7 @@ pub(crate) struct CarrierRequest { attempt: Option, failure: Option, user_agent_hash: [u8; 32], + up_window: Option, } 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) -> Self { + self.up_window = window; + self + } + + /// Returns the separately advertised HTTP conveyor capability. + pub(crate) const fn up_window(self) -> Option { + 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. diff --git a/src/web/manager/session_creation.rs b/src/web/manager/session_creation.rs index ce822c4..9b29373 100644 --- a/src/web/manager/session_creation.rs +++ b/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(), diff --git a/src/web/manager/session_creation/replacement.rs b/src/web/manager/session_creation/replacement.rs index 170f3e6..fac719c 100644 --- a/src/web/manager/session_creation/replacement.rs +++ b/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), diff --git a/src/web/session.rs b/src/web/session.rs index 3c74c20..99ff42b 100644 --- a/src/web/session.rs +++ b/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, closing_streams: HashMap, 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, diff --git a/src/web/session/conveyor.rs b/src/web/session/conveyor.rs new file mode 100644 index 0000000..451deee --- /dev/null +++ b/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 for ConveyorError { + fn from(error: ManagerError) -> Self { + Self::Manager(error) + } +} + +#[derive(Clone, Copy)] +struct Slot { + sequence: u64, + owner: u64, + digest: Option, + committed: bool, +} + +struct Channel { + confirmed: u64, + committed: u64, + slots: [Option; MAX_WINDOW], + notify: Arc, +} + +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, + next_owner: u64, + channels: HashMap, 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, 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) -> Option> { + 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> { + 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, + lane: Option, + sequence: u64, + owner: u64, + notify: Arc, + _waiter: Option, +} + +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, + lane: Option, + sequence: u64, + confirmed: Option, + ) -> Result, 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 { + 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; diff --git a/src/web/session/conveyor/tests.rs b/src/web/session/conveyor/tests.rs new file mode 100644 index 0000000..f279b57 --- /dev/null +++ b/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, Arc) { + 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, lane: Option, seq: u64, ack: u64) -> ConveyorClaim { + session + .claim_conveyor(lane, seq, Some(ack)) + .unwrap() + .unwrap() +} + +fn stream(session: &WebSession, carrier: WebCarrier) -> Option { + 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, manager: Arc) { + 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, + notified: std::sync::atomic::AtomicBool, +} + +impl Wake for LockProbe { + fn wake(self: Arc) { + self.wake_by_ref(); + } + fn wake_by_ref(self: &Arc) { + 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, Channel)>(); + let table_slack = key_and_value.div_ceil(7); + let notify_allocation = std::mem::size_of::() + 2 * std::mem::size_of::(); + 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; +} diff --git a/src/web/session/lane_uplink.rs b/src/web/session/lane_uplink.rs index 40d2b59..c3b1d52 100644 --- a/src/web/session/lane_uplink.rs +++ b/src/web/session/lane_uplink.rs @@ -20,6 +20,17 @@ impl WebSession { lane_id: u32, sequence: u64, body: &[u8], + ) -> Result { + 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, + lane_id: u32, + sequence: u64, + body: &[u8], + claim: Option<(&super::conveyor::ConveyorClaim, TokenHash)>, ) -> Result { 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) diff --git a/src/web/session/lanes.rs b/src/web/session/lanes.rs index abdf41e..b1185ef 100644 --- a/src/web/session/lanes.rs +++ b/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; }; diff --git a/src/web/session/lifecycle.rs b/src/web/session/lifecycle.rs index fc345c3..cc0c565 100644 --- a/src/web/session/lifecycle.rs +++ b/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; } diff --git a/src/web/session/uplink.rs b/src/web/session/uplink.rs index ecaf63e..8757a4f 100644 --- a/src/web/session/uplink.rs +++ b/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, + sequence: u64, + body: &[u8], + claim: &super::conveyor::ConveyorClaim, + digest: TokenHash, + ) -> Result { + 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, 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())) } From dcce897d8b08e2e5e4160c4aa6eeb79c81aa1c87 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sat, 3 Oct 2026 16:05:22 +0300 Subject: [PATCH 2/3] Bump -> 3.5.13 --- Cargo.lock | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index da89d11..022910c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2780,7 +2780,7 @@ dependencies = [ [[package]] name = "telemt" -version = "3.5.12" +version = "3.5.13" dependencies = [ "aes", "anyhow", diff --git a/Cargo.toml b/Cargo.toml index 9408d67..135b926 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "telemt" -version = "3.5.12" +version = "3.5.13" edition = "2024" rust-version = "1.88" From e2f1c3f3a95ad088e995b4891c4cbba913cee09f Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sat, 3 Oct 2026 20:29:18 +0300 Subject: [PATCH 3/3] Bridge JS semicolon warnings fixed --- src/web/bridge/behavior_tests.js | 16 +++++++++++++++- src/web/bridge/conveyor.js | 2 +- src/web/bridge/downlink.js | 8 ++++---- src/web/bridge/request.js | 2 +- src/web/bridge/runtime.js | 7 +++---- 5 files changed, 24 insertions(+), 11 deletions(-) diff --git a/src/web/bridge/behavior_tests.js b/src/web/bridge/behavior_tests.js index cdbc863..821883b 100644 --- a/src/web/bridge/behavior_tests.js +++ b/src/web/bridge/behavior_tests.js @@ -228,7 +228,7 @@ test('multiple precommit lanes survive fallback without losing stream ordering', 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;offsetvalue[1]===id),[[1,id],[2,id]]); assert.equal(frames.length,4);env.close();await flush(); }); @@ -248,4 +248,18 @@ test('receiver retirement releases shared prefix capacity without exposing a par 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})(); diff --git a/src/web/bridge/conveyor.js b/src/web/bridge/conveyor.js index 1c138a4..5abdb05 100644 --- a/src/web/bridge/conveyor.js +++ b/src/web/bridge/conveyor.js @@ -92,7 +92,7 @@ function create(settings){ 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)} + 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; diff --git a/src/web/bridge/downlink.js b/src/web/bridge/downlink.js index 98ef42c..5859c1e 100644 --- a/src/web/bridge/downlink.js +++ b/src/web/bridge/downlink.js @@ -38,11 +38,11 @@ function create(settings){ 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(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')} + 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; @@ -53,7 +53,7 @@ function create(settings){ 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')} + 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'); @@ -76,7 +76,7 @@ function create(settings){ if(!position||offset!==position||position{})}catch(error){} + if(!complete)try{const cancelled=reader.cancel();if(cancelled)cancelled.catch(()=>{});}catch(error){} reader.releaseLock(); } } diff --git a/src/web/bridge/request.js b/src/web/bridge/request.js index 4041fe4..01ddf80 100644 --- a/src/web/bridge/request.js +++ b/src/web/bridge/request.js @@ -46,7 +46,7 @@ function create(settings){ lastReason='http';wait=retryAfterMs(fetched);settings.cancel(fetched); }else{ const policy=responsePolicy(path,fetched.status);let body; - try{body=receiver&&(fetched.status===200||fetched.status===204)?await receiver(fetched,controller.signal):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; diff --git a/src/web/bridge/runtime.js b/src/web/bridge/runtime.js index 45d16ab..1271bee 100644 --- a/src/web/bridge/runtime.js +++ b/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,httpCarrier=null,upWindow=1; +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; @@ -60,7 +60,7 @@ function resolveRecoveryCommit(){ 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(); @@ -144,7 +144,6 @@ function clearProbeTimer(){if(probeTimer){clearTimeout(probeTimer.timer);probeTi 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()){ @@ -384,7 +383,7 @@ function deleteSession(){ if(token)fetch(relayBase+'/api/v1/session',options('DELETE',token,null,headers,undefined,true)).catch(()=>{}); } function close(notifyServer){ - 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(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();