diff --git a/src/config/types/web.rs b/src/config/types/web.rs index 9f917fc..224e5ae 100644 --- a/src/config/types/web.rs +++ b/src/config/types/web.rs @@ -337,7 +337,7 @@ pub struct WebTimeoutsConfig { /// Lifetime of an unused bootstrap credential and closed-token replay marker. #[serde(default = "default_web_bootstrap_lifetime_secs")] pub bootstrap_lifetime_secs: u64, - /// Maximum carrier inactivity before a session is closed. + /// Maximum validated peer inactivity before a session is closed. #[serde(default = "default_web_reconnect_grace_secs")] pub reconnect_grace_secs: u64, /// Maximum idle lifetime of a WEB HTTP keep-alive connection. diff --git a/src/web/bridge/runtime.js b/src/web/bridge/runtime.js index 63dfb3d..8f6f6c2 100644 --- a/src/web/bridge/runtime.js +++ b/src/web/bridge/runtime.js @@ -49,7 +49,7 @@ function cancelBatch(lease){if(!lease||lease.settled)return;buffers.cancelBatch( const attemptHeaders=(attempt,failure)=>negotiationEnabled?Object.assign({'X-Carrier-Capabilities':carrierCapabilities,'X-Carrier-Attempt':String(attempt)},failure?{'X-Carrier-Failure':failure}:{}):{}; function finishOldRecovery(){ recoveryReplaced=false;status('connected','committed','',0); - for(const data of recoveryPending.splice(0)){release(data.byteLength,1,null);queueCarrier(data)} + while(recoveryPending.length&&!closed){const data=recoveryPending.shift();release(data.byteLength,1,null);queueCarrier(data)} } function rejectRecoveryCommit(error){ const commit=recoveryCommit;if(!commit)return;recoveryCommit=null; @@ -86,7 +86,7 @@ function replaceCarrier(policy,signal,remaining){ return new Promise((resolve,reject)=>{ const abort=()=>rejectRecoveryCommit(failure('timeout','recovery deadline')); recoveryCommit={resolve,reject,signal,abort};signal.addEventListener('abort',abort,{once:true}); - if(remaining()<=0){abort();return}createSession(attemptEpoch); + if(signal.aborted||remaining()<=0){abort();return}createSession(attemptEpoch); }); } function recoverTransport(error,replay){ @@ -239,14 +239,15 @@ function commitCarrier(probe,epoch){ status('connected','committed','',0); if(carrier==='https')poll(); else if(carrier==='https-lanes'){const lane=lanes.get(probe.id);if(lane&&!lane.polling)pollLane(lane)} - for(const data of pending.splice(0)){release(data.byteLength,1,null);queueCarrier(data)} + 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); else if(carrier==='websocket')queueSocket(data); - else for(const value of splitFrames(data))queueLane(value); + 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()} @@ -262,12 +263,15 @@ async function runUp(){ if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','uplink acknowledgement rejected'); break; }catch(error){ + let replayed=false; const recovered=await recoverTransport(error,async(signal,remaining)=>{ const response=await request('/api/v1/up',options('POST',token,lease.body,{'X-Up-Seq':sequence},signal),remaining,2); 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; } } @@ -416,12 +420,15 @@ async function runLaneUp(lane){ if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','lane uplink acknowledgement rejected'); break; }catch(error){ + let replayed=false; const recovered=await recoverTransport(error,async(signal,remaining)=>{ const response=await request('/api/v1/up',options('POST',token,lease.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID},signal),remaining,2); 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; } } diff --git a/src/web/manager/session_creation/replacement.rs b/src/web/manager/session_creation/replacement.rs index 06a6519..cdc3581 100644 --- a/src/web/manager/session_creation/replacement.rs +++ b/src/web/manager/session_creation/replacement.rs @@ -140,6 +140,7 @@ impl WebProcessRuntime { deadline_secs: Some(entry.profile.carrier_negotiation_deadlines_secs[3]), carrier_state: Some(CarrierChainPhase::Provisional.as_str()), }; + let predecessor_session_id = entry.predecessor_session_id; if let Some(index) = state.session_index.get_mut(&replacement.trace_session_id) && index.session_hash == old_hash { @@ -198,7 +199,7 @@ impl WebProcessRuntime { replacement.scores, None, ); - self.trace.record_lifecycle( + self.trace.record_lifecycle_with_context( None, Some(client_ip), identity, @@ -208,6 +209,10 @@ impl WebProcessRuntime { .request .failure() .map(|failure| failure.as_str()), + crate::web::trace::TraceLifecycleContext { + peer_gap_ms: None, + predecessor_session_id, + }, ); Ok(result) } diff --git a/src/web/session/lanes.rs b/src/web/session/lanes.rs index ae65b56..0329b42 100644 --- a/src/web/session/lanes.rs +++ b/src/web/session/lanes.rs @@ -319,6 +319,11 @@ impl WebSession { return Err(ManagerError::Limit); }; state.lane_open_waits += 1; + self.touch_peer_locked( + &mut state, + Instant::now(), + WebSessionLifecycleObservation::HttpActivityAfterGap, + ); LaneOpenWaitGuard { session: self, _auxiliary: auxiliary, diff --git a/src/web/trace/types.rs b/src/web/trace/types.rs index 746ef78..412e4a8 100644 --- a/src/web/trace/types.rs +++ b/src/web/trace/types.rs @@ -221,7 +221,7 @@ pub(crate) struct TraceWebSocketRecord { pub(crate) lane_id: Option, } -/// Closed WEB lifecycle event category. +/// WEB lifecycle event category. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum TraceLifecycleEvent { /// A bridge bootstrap was issued. @@ -329,7 +329,7 @@ pub(crate) struct TraceLifecycleRecord { pub(crate) event: TraceLifecycleEvent, /// Logical stream identifier when applicable. pub(crate) stream_id: Option, - /// Closed outcome or rejection reason. + /// Terminal outcome or rejection reason. pub(crate) reason: Option<&'static str>, /// Carrier negotiation detail when this is a carrier lifecycle event. pub(crate) carrier: Option,