From 50faaba8c511a501aea2a56411331545c9d01b51 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Thu, 3 Sep 2026 19:00:06 +0300 Subject: [PATCH] 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 | 10 ++++++---- src/web/telemetry/lifecycle.rs | 4 ++-- src/web/telemetry/tests.rs | 28 ++++++++++++++++++++++++++++ 3 files changed, 36 insertions(+), 6 deletions(-) diff --git a/src/web/bridge/runtime.js b/src/web/bridge/runtime.js index 8f6f6c2..dbd1152 100644 --- a/src/web/bridge/runtime.js +++ b/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; diff --git a/src/web/telemetry/lifecycle.rs b/src/web/telemetry/lifecycle.rs index 826af91..16fee77 100644 --- a/src/web/telemetry/lifecycle.rs +++ b/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 { diff --git a/src/web/telemetry/tests.rs b/src/web/telemetry/tests.rs index 960c34b..eb031a9 100644 --- a/src/web/telemetry/tests.rs +++ b/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::(), + 1 + ); drop(guard); assert_eq!(telemetry.live_acceptors(), 0); }