fix(transport): stop ambiguous admission failover
Verify Data Plane / gateway (push) Successful in 4m54s
Verify Data Plane / gateway (push) Successful in 4m54s
This commit is contained in:
@@ -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");
|
||||
|
||||
Reference in New Issue
Block a user