Concurrent Uplinks w/ wrong single-flight recovery fixes

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

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.

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;
     }
    }

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)
     }

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,

src/web/trace/types.rs

@@ -221,7 +221,7 @@ pub(crate) struct TraceWebSocketRecord {
     pub(crate) lane_id: Option<u32>,
 }
 
-/// 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<u32>,
-    /// 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<TraceCarrierDetail>,