Generation Fence against Cancellation race after HTTP-Runner Fresh Reincarnation

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

src/web/bridge/runtime.js

@@ -252,7 +252,7 @@ function queueCarrier(data){
 }
 function queueUp(data){if(!reserve(data,null)){fail('capacity');return}upPending.push(data);runUp()}
 async function runUp(){
- if(upRunning)return;upRunning=true;let lease=null;
+ 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;
@@ -278,7 +278,7 @@ async function runUp(){
    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{upRunning=false;if(!closed&&sessionToken&&upPending.length)runUp()}
+ 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;
@@ -323,7 +323,7 @@ async function waitSocket(next,size,limit,signal){
  if(closed||(signal&&signal.aborted)||next.readyState!==WebSocket.OPEN)throw new Error('websocket closed');
 }
 async function runSocketUp(){
- if(upRunning||!socketReady)return;upRunning=true;let lease=null;
+ if(upRunning||!socketReady)return;upRunning=true;const ownerEpoch=attemptEpoch;let lease=null;
  try{
   while(!closed&&socketReady&&upPending.length){
    lease=takeBatch(upPending,null);upLease=lease;lease.controller=new AbortController();
@@ -331,7 +331,7 @@ async function runSocketUp(){
    if(!settleBatch(lease))return;port.postMessage({t:'traffic',up:lease.total,down:0});lease=null;
   }
  }catch(error){if(!closed&&!(lease&&lease.cancelled))recoverTransport(error,null)}
- finally{upRunning=false;if(!closed&&socketReady&&upPending.length)runSocketUp()}
+ finally{if(ownerEpoch===attemptEpoch){upRunning=false;if(!closed&&socketReady&&upPending.length)runSocketUp()}}
 }
 async function poll(){
  while(!closed&&sessionToken){
@@ -339,6 +339,7 @@ async function poll(){
   try{
    pollController=new AbortController();
    const response=await request('/api/v1/down',options('POST',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;
@@ -445,6 +446,7 @@ async function pollLane(lane){
    const controller=new AbortController(),laneID=String(lane.id),token=sessionToken,cursor=lane.cursor;lane.controller=controller;
    failedToken=token;failedCursor=cursor;failedLaneID=laneID;
    const response=await request('/api/v1/down',options('POST',token,null,{'X-Down-Cursor':cursor,'X-Lane-ID':laneID},controller.signal),null,1);
+   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;

src/web/telemetry/lifecycle.rs

@@ -115,14 +115,14 @@ pub(super) const fn session_close_slot(
     carrier: WebCarrier,
     reason: SessionCloseReason,
 ) -> usize {
-    carrier as usize * SessionCloseReason::ALL.len() + reason as usize
+    carrier.index() * SessionCloseReason::ALL.len() + reason as usize
 }
 
 pub(super) const fn session_observation_slot(
     carrier: WebCarrier,
     observation: WebSessionLifecycleObservation,
 ) -> usize {
-    carrier as usize * WebSessionLifecycleObservation::ALL.len() + observation as usize
+    carrier.index() * WebSessionLifecycleObservation::ALL.len() + observation as usize
 }
 
 pub(super) fn load(counter: &AtomicU64) -> u64 {

src/web/telemetry/tests.rs

@@ -1,6 +1,7 @@
 use super::*;
 use crate::config::WebCarrier;
 use crate::web::manager::CarrierFailure;
+use crate::web::session::SessionCloseReason;
 
 #[test]
 fn fixed_counter_sets_and_acceptor_guard_are_exact() {
@@ -23,6 +24,12 @@ fn fixed_counter_sets_and_acceptor_guard_are_exact() {
         WebCarrier::Https,
         WebCarrierLearningOutcome::Recorded,
     );
+    telemetry.record_session_closed(WebCarrier::Https, SessionCloseReason::ApiClose);
+    telemetry.record_session_observation(
+        WebCarrier::Https,
+        WebSessionLifecycleObservation::RequestAfterClose,
+    );
+    telemetry.record_bridge_recovery(WebBridgeRecoveryEvent::BootstrapIssued);
     assert_eq!(telemetry.rejection_counters().len(), WebRejectionReason::ALL.len());
     assert_eq!(
         telemetry.overload_counters().len(),
@@ -46,6 +53,18 @@ fn fixed_counter_sets_and_acceptor_guard_are_exact() {
         telemetry.carrier_learning_counters().len(),
         WebCarrier::ALL.len() * WebCarrierLearningOutcome::ALL.len()
     );
+    assert_eq!(
+        telemetry.session_close_counters().len(),
+        WebCarrier::ALL.len() * SessionCloseReason::ALL.len()
+    );
+    assert_eq!(
+        telemetry.session_observation_counters().len(),
+        WebCarrier::ALL.len() * WebSessionLifecycleObservation::ALL.len()
+    );
+    assert_eq!(
+        telemetry.bridge_recovery_counters().len(),
+        WebBridgeRecoveryEvent::ALL.len()
+    );
     assert_eq!(
         telemetry.rejection_total(WebRejectionReason::HttpConnectionCapacity),
         1
@@ -54,6 +73,15 @@ fn fixed_counter_sets_and_acceptor_guard_are_exact() {
         telemetry.last_decoy().map(|value| value.0),
         Some("connect_refused")
     );
+    assert_eq!(telemetry.aggregates().sessions_closed, 1);
+    assert_eq!(
+        telemetry
+            .session_close_counters()
+            .into_iter()
+            .map(|counter| counter.total)
+            .sum::<u64>(),
+        1
+    );
     drop(guard);
     assert_eq!(telemetry.live_acceptors(), 0);
 }