From 86a95952b6d3a75c8cfa88ba1fdedf409e1ee5c9 Mon Sep 17 00:00:00 2001 From: sechmachine <97589681+sechmachine727@users.noreply.github.com> Date: Wed, 12 Aug 2026 20:14:28 +0700 Subject: [PATCH] fix(transport): stop ambiguous admission failover --- core/src/transport.rs | 10 +++++- core/tests/gateway_oracle.rs | 68 +++++++++++++++++++++++++++++++----- 2 files changed, 69 insertions(+), 9 deletions(-) diff --git a/core/src/transport.rs b/core/src/transport.rs index ad8e6a7..0d0b1e2 100644 --- a/core/src/transport.rs +++ b/core/src/transport.rs @@ -182,6 +182,7 @@ pub async fn connect_with_cancellation( let remote_count = resolved.len(); for (remote_index, remote) in resolved.into_iter().enumerate() { cancellation.check()?; + let admission_write_started = AtomicBool::new(false); let quinn_config = client_transport_config(credential, Arc::clone(&checked_tls_signer))?; let remaining_remotes = remote_count.saturating_sub(remote_index).max(1); @@ -201,13 +202,16 @@ pub async fn connect_with_cancellation( now_utc, started, deadline, + admission_write_started: &admission_write_started, }, ), ) .await { Ok(Ok(session)) => return Ok(session), - Ok(Err(AttemptError::Retry)) | Err(CoreError::Transport) => {} + Ok(Err(AttemptError::Retry)) => {} + Err(CoreError::Transport) + if !admission_write_started.load(Ordering::Acquire) => {} Ok(Err(AttemptError::Terminal(error))) | Err(error) => return Err(error), } } @@ -246,6 +250,7 @@ struct AttemptContext<'a> { now_utc: &'a str, started: Instant, deadline: Duration, + admission_write_started: &'a AtomicBool, } async fn cancellable_timeout( @@ -322,6 +327,9 @@ async fn dial( .open_bi() .await .map_err(|_| AttemptError::Terminal(connection_error(&connection)))?; + context + .admission_write_started + .store(true, Ordering::Release); write_frame(&mut send, &payload) .await .map_err(|_| AttemptError::Terminal(connection_error(&connection)))?; diff --git a/core/tests/gateway_oracle.rs b/core/tests/gateway_oracle.rs index a1de8fd..3015bcb 100644 --- a/core/tests/gateway_oracle.rs +++ b/core/tests/gateway_oracle.rs @@ -34,6 +34,7 @@ import ( "errors" "math/big" "os" + "strconv" "sync" "time" @@ -47,6 +48,8 @@ type admission struct { reusable bool failWorkOnce bool releaseFails bool + mutationPath string + providerStarts int authority protocol.SessionAuthority work protocol.ProviderSessionWork public ed25519.PublicKey @@ -73,10 +76,16 @@ func (a *admission) Release(context.Context, protocol.SessionAuthority) error { return nil } -type provider struct { failStart bool } +type provider struct { admission *admission; failStart bool; delay time.Duration } type session struct { state protocol.ProviderState; video chan gateway.ProviderMedia; audio chan gateway.ProviderMedia; events chan gateway.ProviderEvent } func (p provider) Start(_ context.Context, request gateway.LaunchRequest) (gateway.ProviderSession, error) { + p.admission.mu.Lock() + p.admission.providerStarts++ + count := p.admission.providerStarts + if err := os.WriteFile(p.admission.mutationPath, []byte(strconv.Itoa(count)), 0600); err != nil { panic(err) } + p.admission.mu.Unlock() if p.failStart { return nil, context.DeadlineExceeded } + if p.delay > 0 { time.Sleep(p.delay) } return &session{state: protocol.ProviderState{Version:"1", SessionID:request.SessionID, State:gateway.ProviderStateReady, Channels:[]string{"video","audio","input","feedback"}}, video:make(chan gateway.ProviderMedia), audio:make(chan gateway.ProviderMedia), events:make(chan gateway.ProviderEvent)}, nil } func (s *session) Ready(context.Context) error { return nil } @@ -130,22 +139,25 @@ func main() { _, firstReplayErr := replayGuard.Admit(context.Background(),replayRequest) _, secondReplayErr := replayGuard.Admit(context.Background(),replayRequest) replayRejected := firstReplayErr == nil && errors.Is(secondReplayErr,gateway.ErrAdmissionRejected) - admissionService := &admission{authority:authority,work:work,public:admissionKey.Public().(ed25519.PublicKey),reusable:mode == "reusable",failWorkOnce:mode == "post-retryable",releaseFails:mode == "cleanup-release-failure"} - newService := func() *gateway.Server { - service, err := gateway.NewServer(gateway.ServerConfig{ListenAddress:"127.0.0.1:0",TLSConfig:server,GatewayID:authority.GatewayID,Admission:admissionService,Provider:provider{failStart:mode == "provider-start-lost-response" || mode == "cleanup-release-failure"}}); if err != nil { panic(err) } + mutationPath := readyPath + ".provider-starts" + if err := os.WriteFile(mutationPath, []byte("0"), 0600); err != nil { panic(err) } + admissionService := &admission{authority:authority,work:work,public:admissionKey.Public().(ed25519.PublicKey),reusable:mode == "reusable",failWorkOnce:mode == "post-retryable",releaseFails:mode == "cleanup-release-failure",mutationPath:mutationPath} + newService := func(delay time.Duration) *gateway.Server { + service, err := gateway.NewServer(gateway.ServerConfig{ListenAddress:"127.0.0.1:0",TLSConfig:server,GatewayID:authority.GatewayID,Admission:admissionService,Provider:provider{admission:admissionService,failStart:mode == "provider-start-lost-response" || mode == "cleanup-release-failure",delay:delay}}); if err != nil { panic(err) } return service } - service := newService() + firstDelay := time.Duration(0); if mode == "lost-authority" { firstDelay = 300*time.Millisecond } + service := newService(firstDelay) services := []*gateway.Server{service} addresses := []string{service.Addr().String()} if mode == "retryable" { - draining := newService() + draining := newService(0) draining.BeginDrain() services = []*gateway.Server{draining, service} addresses = []string{draining.Addr().String(), service.Addr().String()} } - if mode == "post-retryable" || mode == "provider-start-lost-response" || mode == "cleanup-release-failure" { - second := newService() + if mode == "post-retryable" || mode == "provider-start-lost-response" || mode == "cleanup-release-failure" || mode == "lost-authority" { + second := newService(0) services = []*gateway.Server{service, second} addresses = []string{service.Addr().String(), second.Addr().String()} } @@ -234,6 +246,13 @@ impl Oracle { String::from_utf8_lossy(&output.stderr) ); } + + fn provider_starts(&self) -> usize { + fs::read_to_string(self.directory.join("ready.json.provider-starts")) + .expect("read provider mutation count") + .parse() + .expect("provider mutation count") + } } impl Drop for Oracle { @@ -831,6 +850,39 @@ fn post_admission_failure_is_terminal_and_does_not_mutate_twice() { } } +#[test] +fn lost_authority_after_provider_mutation_is_terminal_without_failover() { + let oracle = Oracle::start("lost-authority"); + let manifest = ConnectionManifest::decode(oracle.ready.manifest.as_bytes()).expect("manifest"); + let credential = + NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential"); + let transcripts = Arc::new(Mutex::new(Vec::new())); + let result = tokio::runtime::Runtime::new() + .expect("runtime") + .block_on(connect( + &manifest, + &credential, + recording_signers( + test_key(&oracle.ready.admission_key), + test_key(&oracle.ready.client_key), + Arc::clone(&transcripts), + ), + "2026-08-12T00:00:00Z", + Duration::from_millis(400), + )); + + assert!( + result.is_err(), + "late first authority triggered live-address success" + ); + assert_eq!( + transcripts.lock().expect("transcripts").len(), + 1, + "client sent a second signed admission" + ); + assert_eq!(oracle.provider_starts(), 1, "provider mutated twice"); +} + #[test] fn repeated_production_gateway_admissions_remain_bounded() { let runtime = tokio::runtime::Runtime::new().expect("runtime");