From 70667d5aeed065b7dfdc2ccd3ea4b0e188ed34f0 Mon Sep 17 00:00:00 2001 From: sechmachine <97589681+sechmachine727@users.noreply.github.com> Date: Thu, 30 Jul 2026 04:48:02 +0700 Subject: [PATCH] fix(gateway): enforce audited production traversal --- gateway/apollo_native.go | 74 +- gateway/apollo_native_test.go | 102 ++- gateway/apollo_rtsp_handshake.go | 27 +- gateway/capability.go | 28 +- gateway/feedback.go | 27 +- gateway/gateway_test.go | 289 +++++- gateway/provider.go | 70 +- gateway/qualification_contract_test.go | 6 +- gateway/qualification_harness_test.go | 863 ++++++++++++++---- gateway/transport.go | 140 ++- .../design.md | 11 +- .../proposal.md | 6 +- .../specs/apollo-stream-policy/spec.md | 10 +- .../specs/gateway-heartbeat-telemetry/spec.md | 7 + .../specs/gateway-qualification/spec.md | 6 +- .../specs/provider-session-lifecycle/spec.md | 4 +- .../tasks.md | 15 +- 17 files changed, 1368 insertions(+), 317 deletions(-) diff --git a/gateway/apollo_native.go b/gateway/apollo_native.go index c567dcc..3c6609c 100644 --- a/gateway/apollo_native.go +++ b/gateway/apollo_native.go @@ -198,12 +198,16 @@ func pinnedApolloTLSConfig(work protocol.ProviderSessionWork) (*tls.Config, erro }, nil } -func (b *NativeApolloBackend) Setup(ctx context.Context, request LaunchRequest) ([]byte, error) { +func (b *NativeApolloBackend) Setup(ctx context.Context, request LaunchRequest, management []byte) ([]byte, error) { work := request.ProviderWork if err := work.Validate(); err != nil || validateApolloStreamPolicy(work.StreamPolicy) != nil || request.SessionID == "" || request.SessionID != work.SessionID || work.ProviderProfile != ProviderProfileApollo { return nil, ErrProviderMalformed } + info, err := ParseManagementXML(management) + if err != nil || validateApolloProviderStreamPolicy(info, work.StreamPolicy) != nil { + return nil, ErrProviderMalformed + } client, err := newPinnedApolloHTTPClient(work) if err != nil { return nil, err @@ -281,10 +285,11 @@ type nativeApolloSession struct { audioPing []byte videoPing []byte sessionID string - video chan []byte - audio chan []byte + video chan ProviderMedia + audio chan ProviderMedia events chan ProviderEvent mu sync.Mutex + mediaMu sync.Mutex controlMu sync.Mutex state protocol.ProviderState pressed map[string]InputEvent @@ -299,10 +304,15 @@ type nativeApolloSession struct { allowApplicationTermination bool terminationErr error mediaDrops atomic.Uint64 + mediaQuiesced atomic.Bool + mediaIngress atomic.Uint64 + mediaRecovered atomic.Uint64 + mediaEnqueued atomic.Uint64 + mediaQueueMaximum atomic.Uint64 } func newNativeApolloSession(sessionID string) *nativeApolloSession { - return &nativeApolloSession{sessionID: sessionID, video: make(chan []byte, 16), audio: make(chan []byte, 16), events: make(chan ProviderEvent, 16), state: protocol.ProviderState{Version: "1", SessionID: sessionID, State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}}, pressed: make(map[string]InputEvent), done: make(chan struct{}), readDone: make(chan struct{})} + return &nativeApolloSession{sessionID: sessionID, video: make(chan ProviderMedia, 16), audio: make(chan ProviderMedia, 16), events: make(chan ProviderEvent, 16), state: protocol.ProviderState{Version: "1", SessionID: sessionID, State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}}, pressed: make(map[string]InputEvent), done: make(chan struct{}), readDone: make(chan struct{})} } func newNativeApolloProviderSession(ctx context.Context, setup *apolloRTSPSetup) (*nativeApolloSession, error) { @@ -405,8 +415,8 @@ func (s *nativeApolloSession) Ready(context.Context) error { return nil } -func (s *nativeApolloSession) Video() <-chan []byte { return s.video } -func (s *nativeApolloSession) Audio() <-chan []byte { return s.audio } +func (s *nativeApolloSession) Video() <-chan ProviderMedia { return s.video } +func (s *nativeApolloSession) Audio() <-chan ProviderMedia { return s.audio } func (s *nativeApolloSession) Events() <-chan ProviderEvent { return s.events } func (s *nativeApolloSession) Input(ctx context.Context, event InputEvent) error { @@ -646,6 +656,7 @@ func (s *nativeApolloSession) handleApolloControlPayload(_ uint8, _ bool, payloa s.handleApolloDisconnect(ErrProviderMalformed) return } + s.quiesceMedia() s.mu.Lock() s.state.State = ProviderStateTerminated s.mu.Unlock() @@ -686,6 +697,7 @@ func (s *nativeApolloSession) handleApolloDisconnect(err error) { if err == nil { return } + s.quiesceMedia() s.disconnectOnce.Do(func() { s.mu.Lock() if s.state.State == ProviderStateTerminated { @@ -701,13 +713,45 @@ func (s *nativeApolloSession) handleApolloDisconnect(err error) { }) } +func (s *nativeApolloSession) quiesceMedia() { + if !s.mediaQuiesced.CompareAndSwap(false, true) { + return + } + if s.audioConn != nil { + _ = s.audioConn.Close() + } + if s.videoConn != nil { + _ = s.videoConn.Close() + } +} + func (s *nativeApolloSession) closeMediaChannels() { s.channelsOnce.Do(func() { + s.mediaMu.Lock() + defer s.mediaMu.Unlock() close(s.video) close(s.audio) }) } +func (s *nativeApolloSession) enqueueMedia(output chan ProviderMedia, payload []byte, receivedAt time.Time) bool { + s.mediaMu.Lock() + defer s.mediaMu.Unlock() + if len(payload) == 0 || s.mediaQuiesced.Load() { + return false + } + s.mediaRecovered.Add(1) + media := ProviderMedia{Payload: payload, ReceivedAt: receivedAt, EnqueuedAt: time.Now()} + if pushLatest(output, media) { + s.mediaDrops.Add(1) + } + s.mediaEnqueued.Add(1) + depth := uint64(len(output)) + for maximum := s.mediaQueueMaximum.Load(); depth > maximum && !s.mediaQueueMaximum.CompareAndSwap(maximum, depth); maximum = s.mediaQueueMaximum.Load() { + } + return true +} + func (s *nativeApolloSession) readUDPMedia() { if s.media == nil { close(s.readDone) @@ -716,14 +760,18 @@ func (s *nativeApolloSession) readUDPMedia() { } var readers sync.WaitGroup readers.Add(2) - read := func(conn *net.UDPConn, output chan []byte, video bool) { + read := func(conn *net.UDPConn, output chan ProviderMedia, video bool) { defer readers.Done() buffer := make([]byte, apolloMediaMaximumPacket+1) for { + if s.mediaQuiesced.Load() { + return + } if err := conn.SetReadDeadline(time.Now().Add(250 * time.Millisecond)); err != nil { return } count, err := conn.Read(buffer) + receivedAt := time.Now() if err != nil { if networkErr, ok := err.(net.Error); ok && networkErr.Timeout() { select { @@ -738,6 +786,10 @@ func (s *nativeApolloSession) readUDPMedia() { if count > apolloMediaMaximumPacket { continue } + if s.mediaQuiesced.Load() { + return + } + s.mediaIngress.Add(1) var payloads [][]byte if video { shard, openErr := s.media.OpenVideo(buffer[:count]) @@ -764,11 +816,7 @@ func (s *nativeApolloSession) readUDPMedia() { continue } for _, payload := range payloads { - if len(payload) != 0 { - if pushLatest(output, payload) { - s.mediaDrops.Add(1) - } - } + s.enqueueMedia(output, payload, receivedAt) } } } @@ -781,7 +829,7 @@ func (s *nativeApolloSession) readUDPMedia() { }() } -func pushLatest(channel chan []byte, payload []byte) bool { +func pushLatest[T any](channel chan T, payload T) bool { select { case channel <- payload: return false diff --git a/gateway/apollo_native_test.go b/gateway/apollo_native_test.go index 53c0e65..2ec1e73 100644 --- a/gateway/apollo_native_test.go +++ b/gateway/apollo_native_test.go @@ -82,14 +82,18 @@ func TestNativeApolloSetupRejectsUnsupportedStreamPolicyBeforeProviderReadiness( ClipboardPolicy: protocol.ClipboardPolicy{MaxTextBytes: 65536, MaxUpdatesPerMinute: 30}, } for name, policy := range map[string]protocol.ProviderStreamPolicy{ - "audio-disabled": {ResolutionWidth: 1920, ResolutionHeight: 1080, Fps: 60, Codec: "H264", BitrateKbps: 8000, AudioEnabled: false}, - "av1": {ResolutionWidth: 3840, ResolutionHeight: 2160, Fps: 60, Codec: "AV1", BitrateKbps: 50000, AudioEnabled: true}, + "audio-disabled": {ResolutionWidth: 1920, ResolutionHeight: 1080, Fps: 60, Codec: "H264", BitrateKbps: 8000, AudioEnabled: false}, + "av1": {ResolutionWidth: 3840, ResolutionHeight: 2160, Fps: 60, Codec: "AV1", BitrateKbps: 50000, AudioEnabled: true}, + "h264-resolution": {ResolutionWidth: 4097, ResolutionHeight: 2160, Fps: 60, Codec: "H264", BitrateKbps: 50000, AudioEnabled: true}, + "hevc-resolution": {ResolutionWidth: 8193, ResolutionHeight: 4320, Fps: 60, Codec: "HEVC", BitrateKbps: 80000, AudioEnabled: true}, + "fps": {ResolutionWidth: 1920, ResolutionHeight: 1080, Fps: 241, Codec: "H264", BitrateKbps: 8000, AudioEnabled: true}, + "bitrate-cap": {ResolutionWidth: 1920, ResolutionHeight: 1080, Fps: 60, Codec: "H264", BitrateKbps: 125001, AudioEnabled: true}, } { t.Run(name, func(t *testing.T) { work.StreamPolicy = policy _, err := NewNativeApolloBackend().Setup(context.Background(), LaunchRequest{ SessionID: "session-1", ProviderProfile: ProviderProfileApollo, ProviderWork: work, - }) + }, nil) if !errors.Is(err, ErrProviderMalformed) { t.Fatalf("Setup() error = %v, want ErrProviderMalformed before provider readiness", err) } @@ -97,6 +101,73 @@ func TestNativeApolloSetupRejectsUnsupportedStreamPolicyBeforeProviderReadiness( } } +func TestNativeApolloRejectsProviderCapabilityMismatchBeforeInventoryOrLaunch(t *testing.T) { + serverTLS, clientTLS := testTLS(t) + var paths []string + management := httptest.NewUnstartedServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + paths = append(paths, request.URL.Path) + switch request.URL.Path { + case "/serverinfo": + _, _ = response.Write([]byte("apollo-server10")) + case "/applist": + _, _ = response.Write([]byte("42")) + default: + http.Error(response, "unexpected provider request", http.StatusBadRequest) + } + })) + management.TLS = serverTLS + management.StartTLS() + defer management.Close() + host, portText, err := net.SplitHostPort(management.Listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + port, err := strconv.ParseInt(portText, 10, 64) + if err != nil { + t.Fatal(err) + } + pinned := sha256.Sum256(serverTLS.Certificates[0].Certificate[0]) + work := protocol.ProviderSessionWork{ + Version: "1", SessionID: "session-source-policy", GatewayID: "gateway-1", + ExpiresAt: "2099-01-01T00:00:00Z", ProviderProfile: ProviderProfileApollo, + ProviderIdentity: "apollo-server#sha256:" + hex.EncodeToString(pinned[:]), PolicyVersionID: "policy-1", + ApplicationID: "42", ClientID: "paired-client", + StreamPolicy: protocol.ProviderStreamPolicy{ + ResolutionWidth: 2560, ResolutionHeight: 1440, Fps: 120, + Codec: "HEVC", BitrateKbps: 40000, AudioEnabled: true, + }, + ManagementHost: host, ManagementPort: port, StreamHost: host, StreamPort: 47984, + ClientCertificatePem: certificatePEM(t, clientTLS.Certificates[0]), + ClientPrivateKeyPem: privateKeyPEM(t, clientTLS.Certificates[0]), + ServerCertificatePem: certificatePEM(t, tls.Certificate{Certificate: [][]byte{serverTLS.Certificates[0].Certificate[1]}}), + ClipboardPolicy: protocol.ClipboardPolicy{MaxTextBytes: 65536, MaxUpdatesPerMinute: 30}, + } + adapter := NewApolloAdapter(NewNativeApolloBackend(), ProviderIdentity{UniqueID: "apollo-server", Fingerprint: "sha256:" + hex.EncodeToString(pinned[:])}) + if _, err := adapter.Start(context.Background(), LaunchRequest{ + SessionID: work.SessionID, ProviderProfile: ProviderProfileApollo, + ProviderIdentity: work.ProviderIdentity, ProviderWork: work, + }); err == nil { + t.Fatal("Start() accepted a provider/source policy mismatch") + } + if got := strings.Join(paths, ","); got != "/serverinfo" { + t.Fatalf("provider requests before capability rejection = %s", got) + } +} + +func TestApolloProviderCapabilityRejectsHEVCLumaDowngrade(t *testing.T) { + info := ManagementInfo{ + ServerCodecModeSupport: 0x101, HasServerCodecModeSupport: true, + MaxLumaPixelsHEVC: 1920 * 1080, HasMaxLumaPixelsHEVC: true, + } + policy := protocol.ProviderStreamPolicy{ + ResolutionWidth: 3840, ResolutionHeight: 2160, Fps: 60, + Codec: "HEVC", BitrateKbps: 80000, AudioEnabled: true, + } + if err := validateApolloProviderStreamPolicy(info, policy); !errors.Is(err, ErrProviderMalformed) { + t.Fatalf("provider HEVC luma downgrade error = %v", err) + } +} + func TestNativeApolloSetupRequiresModernEncryptedRTSPOrder(t *testing.T) { serverTLS, clientTLS := testTLS(t) streamListener, err := net.Listen("tcp", "127.0.0.1:0") @@ -213,7 +284,7 @@ func TestNativeApolloSetupRequiresModernEncryptedRTSPOrder(t *testing.T) { } switch request.URL.Path { case "/serverinfo": - _, _ = response.Write([]byte("apollo-server")) + _, _ = response.Write([]byte("apollo-server2571869449984")) case "/applist": if request.URL.Query().Get("uniqueid") != "paired-client" { http.Error(response, "wrong client", http.StatusBadRequest) @@ -291,7 +362,7 @@ func TestNativeApolloSetupRequiresModernEncryptedRTSPOrder(t *testing.T) { ProviderApplicationTerminationAllowed: true, } backend := NewNativeApolloBackend() - response, err := backend.Setup(context.Background(), LaunchRequest{SessionID: "session-1", ProviderProfile: ProviderProfileApollo, ProviderWork: work}) + response, err := backend.Setup(context.Background(), LaunchRequest{SessionID: "session-1", ProviderProfile: ProviderProfileApollo, ProviderWork: work}, []byte("apollo-server2571869449984")) if err != nil { t.Fatalf("Setup() error = %v", err) } @@ -474,7 +545,8 @@ func TestNativeApolloSetupRequiresModernEncryptedRTSPOrder(t *testing.T) { t.Fatal("encrypted host termination was not forwarded") } select { - case payload := <-session.Video(): + case media := <-session.Video(): + payload := media.Payload if len(payload) != 1001 || payload[0] != 'A' || payload[1000] != 'B' { t.Fatalf("source-shaped video relay = %x", payload) } @@ -482,7 +554,8 @@ func TestNativeApolloSetupRequiresModernEncryptedRTSPOrder(t *testing.T) { t.Fatal("source-shaped video was not relayed") } select { - case payload := <-session.Audio(): + case media := <-session.Audio(): + payload := media.Payload if string(payload) != "A" { t.Fatalf("source-shaped audio relay = %x", payload) } @@ -608,7 +681,8 @@ func TestNativeApolloSessionRelaysOnlyAuthenticatedEncodedUDPMedia(t *testing.T) } select { - case payload := <-session.Video(): + case media := <-session.Video(): + payload := media.Payload if string(payload) != string([]byte{0x01, 0x02, 0x03}) { t.Fatalf("video relay = %x, want encoded payload", payload) } @@ -617,7 +691,8 @@ func TestNativeApolloSessionRelaysOnlyAuthenticatedEncodedUDPMedia(t *testing.T) } for _, want := range wantAudio { select { - case payload := <-session.Audio(): + case media := <-session.Audio(): + payload := media.Payload if string(payload) != string(want) { t.Fatalf("audio relay = %x, want %x", payload, want) } @@ -632,7 +707,8 @@ func TestNativeApolloSessionRelaysOnlyAuthenticatedEncodedUDPMedia(t *testing.T) } } select { - case payload := <-session.Video(): + case media := <-session.Video(): + payload := media.Payload if len(payload) != 1001 || payload[0] != 'A' || payload[999] != 'A' || payload[1000] != 'B' { t.Fatalf("FEC video relay = %x", payload) } @@ -646,7 +722,8 @@ func TestNativeApolloSessionRelaysOnlyAuthenticatedEncodedUDPMedia(t *testing.T) } for _, want := range [][]byte{{'A'}, {'B'}, {'C'}, {'D'}} { select { - case payload := <-session.Audio(): + case media := <-session.Audio(): + payload := media.Payload if string(payload) != string(want) { t.Fatalf("FEC audio relay = %x, want %x", payload, want) } @@ -670,7 +747,8 @@ func TestNativeApolloSessionRelaysOnlyAuthenticatedEncodedUDPMedia(t *testing.T) } for _, want := range [][]byte{{0xa0}, {0xa1}, {0xa2}, {0xa3}} { select { - case payload := <-session.Audio(): + case media := <-session.Audio(): + payload := media.Payload if string(payload) != string(want) { t.Fatalf("post-loss audio relay = %x, want %x", payload, want) } diff --git a/gateway/apollo_rtsp_handshake.go b/gateway/apollo_rtsp_handshake.go index e71f079..4c25ad7 100644 --- a/gateway/apollo_rtsp_handshake.go +++ b/gateway/apollo_rtsp_handshake.go @@ -515,7 +515,32 @@ func apolloAnnounceProfile(policy protocol.ProviderStreamPolicy) ([]byte, error) } func validateApolloStreamPolicy(policy protocol.ProviderStreamPolicy) error { - if err := policy.Validate(); err != nil || !policy.AudioEnabled || (policy.Codec != "H264" && policy.Codec != "HEVC") { + if err := policy.Validate(); err != nil || !policy.AudioEnabled || policy.BitrateKbps > 125000 || + (policy.Codec != "H264" && policy.Codec != "HEVC") { + return ErrProviderMalformed + } + if (policy.Codec == "H264" && (policy.ResolutionWidth > 4096 || policy.ResolutionHeight > 4096)) || + (policy.Codec == "HEVC" && (policy.ResolutionWidth > 8192 || policy.ResolutionHeight > 8192)) { + return ErrProviderMalformed + } + return nil +} + +func validateApolloProviderStreamPolicy(info ManagementInfo, policy protocol.ProviderStreamPolicy) error { + if validateApolloStreamPolicy(policy) != nil || !info.HasServerCodecModeSupport || !info.HasMaxLumaPixelsHEVC { + return ErrProviderMalformed + } + switch policy.Codec { + case "H264": + if info.ServerCodecModeSupport&0x1 == 0 { + return ErrProviderMalformed + } + case "HEVC": + luma := uint64(policy.ResolutionWidth) * uint64(policy.ResolutionHeight) + if info.ServerCodecModeSupport&0x100 == 0 || info.MaxLumaPixelsHEVC == 0 || luma > info.MaxLumaPixelsHEVC { + return ErrProviderMalformed + } + default: return ErrProviderMalformed } return nil diff --git a/gateway/capability.go b/gateway/capability.go index 8f72040..04d1593 100644 --- a/gateway/capability.go +++ b/gateway/capability.go @@ -8,8 +8,6 @@ import ( var ErrNoCapabilityOverlap = errors.New("no capability overlap") -const defaultClientDecode = "h264-hevc-opus" - func DefaultCapabilities() protocol.CapabilityProfile { return protocol.CapabilityProfile{ Transport: "quic-tls13", @@ -17,29 +15,19 @@ func DefaultCapabilities() protocol.CapabilityProfile { Media: "encoded", Audio: "encoded", SourceRateControl: "server", - ClientDecode: defaultClientDecode, + ClientDecode: []string{"hevc-opus", "h264-opus"}, } } +func capabilityProfileUnset(profile protocol.CapabilityProfile) bool { + return profile.Transport == "" && profile.Framing == "" && profile.Media == "" && + profile.Audio == "" && profile.SourceRateControl == "" && len(profile.ClientDecode) == 0 +} + func IntersectCapabilities(profiles ...protocol.CapabilityProfile) (protocol.CapabilityProfile, error) { - if len(profiles) == 0 { + selected, err := protocol.IntersectCapabilityProfiles(profiles...) + if err != nil { return protocol.CapabilityProfile{}, ErrNoCapabilityOverlap } - for _, profile := range profiles { - if err := profile.Validate(); err != nil { - return protocol.CapabilityProfile{}, ErrNoCapabilityOverlap - } - } - selected := profiles[0] - for _, profile := range profiles[1:] { - if selected.Transport != profile.Transport || - selected.Framing != profile.Framing || - selected.Media != profile.Media || - selected.Audio != profile.Audio || - selected.SourceRateControl != profile.SourceRateControl || - selected.ClientDecode != profile.ClientDecode { - return protocol.CapabilityProfile{}, ErrNoCapabilityOverlap - } - } return selected, nil } diff --git a/gateway/feedback.go b/gateway/feedback.go index 6c8bd0b..71d882e 100644 --- a/gateway/feedback.go +++ b/gateway/feedback.go @@ -10,14 +10,15 @@ const ( ) const ( - gatewayFeedbackHeaderSize = 8 - gatewayFeedbackClient = 0 - gatewayFeedbackGateway = 1 - gatewayFeedbackIDR = 1 - gatewayFeedbackFEC = 2 - gatewayFeedbackTerminated = 0x10 - gatewayFeedbackRumble = 0x11 - gatewayFeedbackHDR = 0x12 + gatewayFeedbackHeaderSize = 8 + gatewayFeedbackClient = 0 + gatewayFeedbackGateway = 1 + gatewayFeedbackIDR = 1 + gatewayFeedbackFEC = 2 + gatewayFeedbackTerminated = 0x10 + gatewayFeedbackRumble = 0x11 + gatewayFeedbackHDR = 0x12 + gatewayFeedbackDisconnected = 0x13 ) type gatewayFeedbackMessage struct { @@ -43,6 +44,11 @@ func EncodeProviderEvent(event ProviderEvent) ([]byte, error) { return nil, ErrProviderMalformed } return encodeGatewayFeedback(gatewayFeedbackGateway, gatewayFeedbackHDR, event.Payload) + case ProviderEventDisconnected: + if len(event.Payload) != 0 { + return nil, ErrProviderMalformed + } + return encodeGatewayFeedback(gatewayFeedbackGateway, gatewayFeedbackDisconnected, nil) default: return nil, ErrProviderMalformed } @@ -120,6 +126,11 @@ func DecodeProviderEvent(data []byte) (ProviderEvent, error) { return ProviderEvent{}, ErrProviderMalformed } return ProviderEvent{Kind: ProviderEventHDR, Payload: message.payload}, nil + case gatewayFeedbackDisconnected: + if len(message.payload) != 0 { + return ProviderEvent{}, ErrProviderMalformed + } + return ProviderEvent{Kind: ProviderEventDisconnected}, nil default: return ProviderEvent{}, ErrProviderMalformed } diff --git a/gateway/gateway_test.go b/gateway/gateway_test.go index 8b0db25..b07e924 100644 --- a/gateway/gateway_test.go +++ b/gateway/gateway_test.go @@ -14,6 +14,7 @@ import ( "math/big" "net" "os" + "reflect" "strings" "sync" "sync/atomic" @@ -97,6 +98,13 @@ func TestClientFeedbackUsesFixedProtocolVGFVector(t *testing.T) { if _, err := DecodeClientFeedback([]byte{'F', 'B', 'R', 'K', 0}); !errors.Is(err, ErrProviderMalformed) { t.Fatalf("legacy feedback accepted: %v", err) } + disconnected, err := EncodeProviderEvent(ProviderEvent{Kind: ProviderEventDisconnected}) + if err != nil || hex.EncodeToString(disconnected) != "5647463101130000" { + t.Fatalf("disconnected event vector = %x, %v", disconnected, err) + } + if event, err := DecodeProviderEvent(disconnected); err != nil || event.Kind != ProviderEventDisconnected { + t.Fatalf("decoded disconnected event = %#v, %v", event, err) + } } func TestCapabilityIntersectionAndBoundedQueue(t *testing.T) { @@ -124,6 +132,24 @@ func TestCapabilityIntersectionAndBoundedQueue(t *testing.T) { } } +func TestNewServerRejectsPartiallyConfiguredCapabilities(t *testing.T) { + serverTLS, _ := testTLS(t) + fake := NewFakeApollo(FakeApolloConfig{Now: time.Now()}) + server, err := NewServer(ServerConfig{ + TLSConfig: serverTLS, GatewayID: "gateway-1", + Capabilities: protocol.CapabilityProfile{SourceRateControl: "server"}, + ProviderCapabilities: DefaultCapabilities(), + Admission: &oneTimeAdmission{}, + Provider: fake, + }) + if server != nil { + _ = server.Close() + } + if err == nil { + t.Fatal("partial capability configuration was silently replaced with defaults") + } +} + func TestSyntheticImpairmentPacingAndResourceBounds(t *testing.T) { payload := make([]byte, 1179*16+1) if _, err := FragmentPayload(ChannelVideo, 1, 0, payload); !errors.Is(err, ErrFrameFragmentedLimit) { @@ -187,10 +213,10 @@ func TestApolloFixturesAndLifecycle(t *testing.T) { if err != nil { t.Fatal(err) } - if got := <-session.Video(); string(got) != string(video) { + if got := <-session.Video(); string(got.Payload) != string(video) { t.Fatalf("video changed: %x", got) } - if got := <-session.Audio(); string(got) != string(audio) { + if got := <-session.Audio(); string(got.Payload) != string(audio) { t.Fatalf("audio changed: %x", got) } if err := session.Input(context.Background(), InputEvent{Sequence: 1, Device: "keyboard", Code: 7, Pressed: true}); err != nil { @@ -321,11 +347,96 @@ func TestAdmissionQUICMTLSRelayAndCleanup(t *testing.T) { } } +func TestGatewayTelemetrySeparatesQueueProcessingAndPacing(t *testing.T) { + serverTLS, clientTLS := testTLS(t) + session := &fakeSession{ + video: make(chan ProviderMedia, 1), + audio: make(chan ProviderMedia), + events: make(chan ProviderEvent, 1), + clipboardWrites: make(chan string, 1), + state: protocol.ProviderState{ + Version: "1", SessionID: "session-timing", State: ProviderStateStarting, + Channels: []string{"video", "audio", "input", "feedback"}, + }, + pressed: make(map[string]struct{}), + } + provider := providerStartFunc(func(context.Context, LaunchRequest) (ProviderSession, error) { + enqueuedAt := time.Now() + session.video <- ProviderMedia{Payload: bytesRepeat(0x5a, 2000), ReceivedAt: enqueuedAt, EnqueuedAt: enqueuedAt} + time.Sleep(60 * time.Millisecond) + session.mu.Lock() + session.state.State = ProviderStateReady + session.mu.Unlock() + return session, nil + }) + authority := protocol.SessionAuthority{ + Version: "1", SessionID: "session-timing", GatewayID: "gateway-1", Audience: "versevdi-gateway", + ExpiresAt: time.Now().Add(5 * time.Second).UTC().Format(time.RFC3339Nano), + Capabilities: DefaultCapabilities(), ProviderProfile: ProviderProfileApollo, + ProviderIdentity: "apollo-fixture-1#sha256:fixture-apollo-1", + } + admission := &oneTimeAdmission{authority: authority, released: make(chan struct{}), disableClipboard: true} + server, err := NewServer(ServerConfig{ + ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: authority.GatewayID, + Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), + Admission: admission, Provider: provider, PacerKbps: 24, + }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + serveDone := make(chan error, 1) + go func() { serveDone <- server.Serve(ctx) }() + request := protocol.TunnelAdmissionRequest{ + Version: "1", SessionID: authority.SessionID, GatewayID: authority.GatewayID, Audience: authority.Audience, + Grant: strings.Repeat("g", 64), ClientNonce: "nonce-0000000001", DeviceSignature: strings.Repeat("s", 86), + Capabilities: DefaultCapabilities(), + } + client, err := Dial(context.Background(), server.Addr().String(), clientTLS, request) + if err != nil { + t.Fatal(err) + } + receiveCtx, receiveCancel := context.WithTimeout(context.Background(), 2*time.Second) + for index := byte(0); index < 2; index++ { + frame, err := client.ReceiveFrame(receiveCtx) + if err != nil { + t.Fatal(err) + } + if frame.FragmentIndex != index || frame.FragmentCount != 2 { + t.Fatalf("timing frame = %#v", frame) + } + } + receiveCancel() + metrics := server.Metrics() + if metrics.ProcessingSamples != 1 { + t.Fatalf("timing samples = %d, want one provider unit", metrics.ProcessingSamples) + } + if metrics.MediaPackets != 2 { + t.Fatalf("media packets = %d, want two fragments", metrics.MediaPackets) + } + queue, processing, pacing := time.Duration(metrics.QueueDelayNanos), time.Duration(metrics.ProcessingDelayNanos), time.Duration(metrics.PacingDelayNanos) + if queue < 40*time.Millisecond || queue > 150*time.Millisecond { + t.Fatalf("queue residence = %s, want the controlled 60ms provider queue wait", queue) + } + if processing >= 100*time.Millisecond { + t.Fatalf("processing = %s, pacing leaked into gateway processing", processing) + } + if pacing < 500*time.Millisecond { + t.Fatalf("pacing = %s, want the controlled scheduler wait", pacing) + } + _ = client.Close() + cancel() + _ = server.Close() + if err := <-serveDone; err != nil { + t.Fatal(err) + } +} + func TestGatewayRejectsProviderWorkOutsideNegotiatedDecodeProfile(t *testing.T) { serverTLS, clientTLS := testTLS(t) fake := NewFakeApollo(FakeApolloConfig{Now: time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)}) capabilities := DefaultCapabilities() - capabilities.ClientDecode = "h264-opus" + capabilities.ClientDecode = []string{"h264-opus"} authority := protocol.SessionAuthority{ Version: "1", SessionID: "session-policy", GatewayID: "gateway-1", Audience: "versevdi-gateway", ExpiresAt: time.Now().Add(5 * time.Second).UTC().Format(time.RFC3339Nano), @@ -371,6 +482,75 @@ func TestGatewayRejectsProviderWorkOutsideNegotiatedDecodeProfile(t *testing.T) } } +func TestGatewayNegotiatesRegisteredProfilesWithIndependentClient(t *testing.T) { + for _, test := range []struct { + name string + clientProfiles []string + selected string + codec string + }{ + {name: "h264-only", clientProfiles: []string{"h264-opus"}, selected: "h264-opus", codec: "H264"}, + {name: "hevc-only", clientProfiles: []string{"hevc-opus"}, selected: "hevc-opus", codec: "HEVC"}, + {name: "policy-selects-hevc", clientProfiles: []string{"h264-opus", "hevc-opus"}, selected: "hevc-opus", codec: "HEVC"}, + } { + t.Run(test.name, func(t *testing.T) { + serverTLS, clientTLS := testTLS(t) + fake := NewFakeApollo(FakeApolloConfig{Now: time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)}) + clientCapabilities := DefaultCapabilities() + clientCapabilities.ClientDecode = test.clientProfiles + authority := protocol.SessionAuthority{ + Version: "1", SessionID: "session-profile-" + test.name, GatewayID: "gateway-1", Audience: "versevdi-gateway", + ExpiresAt: time.Now().Add(5 * time.Second).UTC().Format(time.RFC3339Nano), + Capabilities: clientCapabilities, ProviderProfile: ProviderProfileApollo, ProviderIdentity: fake.config.Identity.Key(), + } + admission := &oneTimeAdmission{ + authority: authority, released: make(chan struct{}), disableClipboard: true, + streamPolicy: protocol.ProviderStreamPolicy{ + ResolutionWidth: 1920, ResolutionHeight: 1080, Fps: 60, + Codec: test.codec, BitrateKbps: 8000, AudioEnabled: true, + }, + } + started := make(chan LaunchRequest, 1) + provider := providerStartFunc(func(ctx context.Context, request LaunchRequest) (ProviderSession, error) { + started <- request + return fake.Start(ctx, request) + }) + server, err := NewServer(ServerConfig{ + ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: authority.GatewayID, + Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), + Admission: admission, Provider: provider, + }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + serveDone := make(chan error, 1) + go func() { serveDone <- server.Serve(ctx) }() + request := protocol.TunnelAdmissionRequest{ + Version: "1", SessionID: authority.SessionID, GatewayID: authority.GatewayID, Audience: authority.Audience, + Grant: strings.Repeat("g", 64), ClientNonce: "nonce-0000000001", DeviceSignature: strings.Repeat("s", 86), + Capabilities: clientCapabilities, + } + client, err := Dial(context.Background(), server.Addr().String(), clientTLS, request) + if err != nil { + cancel() + _ = server.Close() + t.Fatal(err) + } + launch := <-started + if !reflect.DeepEqual(launch.Capabilities.ClientDecode, []string{test.selected}) { + t.Fatalf("provider selected profiles = %v", launch.Capabilities.ClientDecode) + } + _ = client.Close() + cancel() + _ = server.Close() + if err := <-serveDone; err != nil { + t.Fatal(err) + } + }) + } +} + func TestRegisteredChannelFramesTraversePublicTransport(t *testing.T) { h := newGatewayTransportHarness(t) @@ -525,9 +705,60 @@ func TestProviderTerminationEndsPublicGatewaySession(t *testing.T) { } func TestEncryptedNativeHostTerminationEndsPublicGatewaySession(t *testing.T) { + h := newNativeGatewayLifecycleHarness(t, "session-native-terminal") + h.native.handleApolloControlPayload(apolloChannelGeneric, true, sourceSealHostControl(t, h.key, 0, apolloControlTypeTerm, []byte{1, 2, 3, 4})) + tryQueueNativeMedia(h.native, h.native.video, []byte("queued-video")) + tryQueueNativeMedia(h.native, h.native.audio, []byte("queued-audio")) + + eventCtx, eventCancel := context.WithTimeout(context.Background(), time.Second) + event, err := h.client.ReceiveProviderEvent(eventCtx) + eventCancel() + if err != nil || event.Kind != ProviderEventTerminated { + t.Fatalf("native provider termination = %#v, %v", event, err) + } + tryQueueNativeMedia(h.native, h.native.video, []byte("new-video")) + tryQueueNativeMedia(h.native, h.native.audio, []byte("new-audio")) + h.assertNoMedia(t) + h.waitReleased(t) + if states := h.reporter.States(); len(states) == 0 || states[len(states)-1].State != ProviderStateTerminated { + t.Fatalf("provider states = %#v", states) + } +} + +func TestNativeENetDisconnectQuiescesPublicGatewaySession(t *testing.T) { + h := newNativeGatewayLifecycleHarness(t, "session-native-disconnect") + h.native.handleApolloDisconnect(ErrProviderDisconnected) + tryQueueNativeMedia(h.native, h.native.video, []byte("queued-video")) + tryQueueNativeMedia(h.native, h.native.audio, []byte("queued-audio")) + + eventCtx, eventCancel := context.WithTimeout(context.Background(), time.Second) + event, err := h.client.ReceiveProviderEvent(eventCtx) + eventCancel() + if err != nil || event.Kind != ProviderEventDisconnected { + t.Fatalf("native provider disconnect = %#v, %v", event, err) + } + tryQueueNativeMedia(h.native, h.native.video, []byte("new-video")) + tryQueueNativeMedia(h.native, h.native.audio, []byte("new-audio")) + h.assertNoMedia(t) + h.waitReleased(t) + if states := h.reporter.States(); len(states) == 0 || states[len(states)-1].State != ProviderStateDisconnected || states[len(states)-1].CleanupPending { + t.Fatalf("provider states = %#v", states) + } +} + +type nativeGatewayLifecycleHarness struct { + native *nativeApolloSession + key []byte + client *Client + admission *oneTimeAdmission + reporter *recordingProviderStateReporter +} + +func newNativeGatewayLifecycleHarness(t *testing.T, sessionID string) nativeGatewayLifecycleHarness { + t.Helper() serverTLS, clientTLS := testTLS(t) key := []byte("0123456789abcdef") - native := newNativeApolloSession("session-native-terminal") + native := newNativeApolloSession(sessionID) control, err := newApolloControlCodec(key) if err != nil { t.Fatal(err) @@ -557,7 +788,6 @@ func TestEncryptedNativeHostTerminationEndsPublicGatewaySession(t *testing.T) { t.Fatal(err) } ctx, cancel := context.WithCancel(context.Background()) - defer cancel() serveDone := make(chan error, 1) go func() { serveDone <- server.Serve(ctx) }() request := protocol.TunnelAdmissionRequest{ @@ -569,28 +799,39 @@ func TestEncryptedNativeHostTerminationEndsPublicGatewaySession(t *testing.T) { if err != nil { t.Fatal(err) } - native.handleApolloControlPayload(apolloChannelGeneric, true, sourceSealHostControl(t, key, 0, apolloControlTypeTerm, []byte{1, 2, 3, 4})) - eventCtx, eventCancel := context.WithTimeout(context.Background(), time.Second) - event, err := client.ReceiveProviderEvent(eventCtx) - eventCancel() - if err != nil || event.Kind != ProviderEventTerminated { - t.Fatalf("native provider termination = %#v, %v", event, err) - } + t.Cleanup(func() { + _ = client.Close() + cancel() + _ = server.Close() + if err := <-serveDone; err != nil { + t.Errorf("serve: %v", err) + } + }) + return nativeGatewayLifecycleHarness{native: native, key: key, client: client, admission: admission, reporter: reporter} +} + +func (h nativeGatewayLifecycleHarness) waitReleased(t *testing.T) { + t.Helper() select { - case <-admission.released: + case <-h.admission.released: case <-time.After(2 * time.Second): - t.Fatal("native termination did not release admission") + t.Fatal("native terminal state did not release admission") } - if states := reporter.States(); len(states) == 0 || states[len(states)-1].State != ProviderStateTerminated { - t.Fatalf("provider states = %#v", states) - } - _ = client.Close() - _ = server.Close() - if err := <-serveDone; err != nil { - t.Fatal(err) +} + +func (h nativeGatewayLifecycleHarness) assertNoMedia(t *testing.T) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + if frame, err := h.client.ReceiveFrame(ctx); err == nil { + t.Fatalf("media crossed after native terminal signal: channel=%d payload=%q", frame.Channel, frame.Payload) } } +func tryQueueNativeMedia(session *nativeApolloSession, channel chan ProviderMedia, payload []byte) { + session.enqueueMedia(channel, payload, time.Now()) +} + func TestProviderDisconnectEndsPublicGatewaySessionReconnectable(t *testing.T) { h := newGatewayTransportHarnessWithoutClipboard(t) h.drainInitialMedia(t) @@ -752,6 +993,7 @@ type oneTimeAdmission struct { releases atomic.Int64 released chan struct{} streamPolicy protocol.ProviderStreamPolicy + providerWork *protocol.ProviderSessionWork disableClipboard bool } @@ -795,9 +1037,12 @@ func (a *oneTimeAdmission) Admit(context.Context, protocol.TunnelAdmissionReques } func (a *oneTimeAdmission) ProviderWork(_ context.Context, authority protocol.SessionAuthority) (protocol.ProviderSessionWork, error) { - if authority != a.authority { + if !reflect.DeepEqual(authority, a.authority) { return protocol.ProviderSessionWork{}, ErrAdmissionRejected } + if a.providerWork != nil { + return *a.providerWork, nil + } streamPolicy := a.streamPolicy if streamPolicy == (protocol.ProviderStreamPolicy{}) { streamPolicy = protocol.ProviderStreamPolicy{ResolutionWidth: 1920, ResolutionHeight: 1080, Fps: 60, Codec: "H264", BitrateKbps: 8000, AudioEnabled: true} diff --git a/gateway/provider.go b/gateway/provider.go index fa88352..2f7403b 100644 --- a/gateway/provider.go +++ b/gateway/provider.go @@ -5,6 +5,7 @@ import ( "encoding/xml" "errors" "fmt" + "strconv" "strings" "sync" "time" @@ -66,8 +67,12 @@ func (i ProviderIdentity) Validate(now time.Time, expected ProviderIdentity) err } type ManagementInfo struct { - Identity ProviderIdentity - Name string + Identity ProviderIdentity + Name string + ServerCodecModeSupport uint32 + MaxLumaPixelsHEVC uint64 + HasServerCodecModeSupport bool + HasMaxLumaPixelsHEVC bool } func ParseManagementXML(data []byte) (ManagementInfo, error) { @@ -82,6 +87,8 @@ func ParseManagementXML(data []byte) (ManagementInfo, error) { NotBefore string `xml:"not_before"` NotAfter string `xml:"not_after"` Name string `xml:"name"` + CodecModes string `xml:"ServerCodecModeSupport"` + MaxHEVCLuma string `xml:"MaxLumaPixelsHEVC"` } decoder := xml.NewDecoder(strings.NewReader(string(data))) decoder.Strict = true @@ -108,7 +115,24 @@ func ParseManagementXML(data []byte) (ManagementInfo, error) { if identity.UniqueID == "" || len(identity.UniqueID) > 128 || len(identity.Fingerprint) > 256 { return ManagementInfo{}, ErrProviderMalformed } - return ManagementInfo{Identity: identity, Name: document.Name}, nil + info := ManagementInfo{Identity: identity, Name: document.Name} + if document.CodecModes != "" { + value, parseErr := strconv.ParseUint(document.CodecModes, 10, 32) + if parseErr != nil { + return ManagementInfo{}, ErrProviderMalformed + } + info.ServerCodecModeSupport = uint32(value) + info.HasServerCodecModeSupport = true + } + if document.MaxHEVCLuma != "" { + value, parseErr := strconv.ParseUint(document.MaxHEVCLuma, 10, 64) + if parseErr != nil { + return ManagementInfo{}, ErrProviderMalformed + } + info.MaxLumaPixelsHEVC = value + info.HasMaxLumaPixelsHEVC = true + } + return info, nil } type RTSPResponse struct { @@ -206,14 +230,20 @@ type ProviderTelemetry struct { MediaDrops uint64 } +type ProviderMedia struct { + Payload []byte + ReceivedAt time.Time + EnqueuedAt time.Time +} + type Provider interface { Start(context.Context, LaunchRequest) (ProviderSession, error) } type ProviderSession interface { Ready(context.Context) error - Video() <-chan []byte - Audio() <-chan []byte + Video() <-chan ProviderMedia + Audio() <-chan ProviderMedia Events() <-chan ProviderEvent Input(context.Context, InputEvent) error Feedback(context.Context, Feedback) error @@ -227,7 +257,7 @@ type ProviderSession interface { type ApolloBackend interface { Management(context.Context, LaunchRequest) ([]byte, error) - Setup(context.Context, LaunchRequest) ([]byte, error) + Setup(context.Context, LaunchRequest, []byte) ([]byte, error) Open(context.Context, LaunchRequest, RTSPResponse) (ProviderSession, error) } @@ -268,7 +298,7 @@ func (a *ApolloAdapter) Start(ctx context.Context, request LaunchRequest) (Provi if request.ProviderIdentity != "" && info.Identity.UniqueID != expected.UniqueID { return nil, ErrProviderIdentity } - rawRTSP, err := a.backend.Setup(ctx, request) + rawRTSP, err := a.backend.Setup(ctx, request, management) if err != nil { return nil, err } @@ -352,7 +382,7 @@ func (f *FakeApollo) Management(context.Context, LaunchRequest) ([]byte, error) return []byte(fmt.Sprintf("%s%s%s%sfixture-apollo", identity.UniqueID, identity.Fingerprint, f.config.Now.Add(-time.Hour).Format(time.RFC3339), f.config.Now.Add(time.Hour).Format(time.RFC3339))), nil } -func (f *FakeApollo) Setup(context.Context, LaunchRequest) ([]byte, error) { +func (f *FakeApollo) Setup(context.Context, LaunchRequest, []byte) ([]byte, error) { if f.config.Failure == FakeFailureMalformed { return []byte("RTSP/1.0 200 OK\r\n\r\n"), nil } @@ -362,8 +392,8 @@ func (f *FakeApollo) Setup(context.Context, LaunchRequest) ([]byte, error) { func (f *FakeApollo) Open(_ context.Context, request LaunchRequest, _ RTSPResponse) (ProviderSession, error) { session := &fakeSession{ failure: f.config.Failure, - video: make(chan []byte, 16), - audio: make(chan []byte, 16), + video: make(chan ProviderMedia, 16), + audio: make(chan ProviderMedia, 16), events: make(chan ProviderEvent, 16), clipboardWrites: make(chan string, 1), state: protocol.ProviderState{Version: "1", SessionID: request.SessionID, State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}}, @@ -405,8 +435,8 @@ func (f *FakeApollo) DisconnectProvider() { type fakeSession struct { mu sync.Mutex failure FakeFailure - video chan []byte - audio chan []byte + video chan ProviderMedia + audio chan ProviderMedia events chan ProviderEvent state protocol.ProviderState pressed map[string]struct{} @@ -432,8 +462,8 @@ func (s *fakeSession) Ready(ctx context.Context) error { return nil } -func (s *fakeSession) Video() <-chan []byte { return s.video } -func (s *fakeSession) Audio() <-chan []byte { return s.audio } +func (s *fakeSession) Video() <-chan ProviderMedia { return s.video } +func (s *fakeSession) Audio() <-chan ProviderMedia { return s.audio } func (s *fakeSession) Events() <-chan ProviderEvent { return s.events } func (s *fakeSession) EmitEvent(event ProviderEvent) { @@ -449,15 +479,17 @@ func (s *fakeSession) EmitVideo(payload []byte) { if s.state.State == ProviderStateTerminating || s.state.State == ProviderStateTerminated || s.state.State == ProviderStateDisconnected { return } + now := time.Now() + media := ProviderMedia{Payload: append([]byte(nil), payload...), ReceivedAt: now, EnqueuedAt: now} select { - case s.video <- append([]byte(nil), payload...): + case s.video <- media: default: select { case <-s.video: default: } select { - case s.video <- append([]byte(nil), payload...): + case s.video <- media: default: } } @@ -469,15 +501,17 @@ func (s *fakeSession) EmitAudio(payload []byte) { if s.state.State == ProviderStateTerminating || s.state.State == ProviderStateTerminated || s.state.State == ProviderStateDisconnected { return } + now := time.Now() + media := ProviderMedia{Payload: append([]byte(nil), payload...), ReceivedAt: now, EnqueuedAt: now} select { - case s.audio <- append([]byte(nil), payload...): + case s.audio <- media: default: select { case <-s.audio: default: } select { - case s.audio <- append([]byte(nil), payload...): + case s.audio <- media: default: } } diff --git a/gateway/qualification_contract_test.go b/gateway/qualification_contract_test.go index 6241beb..0807748 100644 --- a/gateway/qualification_contract_test.go +++ b/gateway/qualification_contract_test.go @@ -130,7 +130,7 @@ func TestQualificationShortProcessingWritesRawArtifact(t *testing.T) { func TestQualificationProcessingPreservesPayload(t *testing.T) { profile := qualificationMediaProfiles()[0] payload := qualificationPayload(profile) - trace, elapsed, err := newQualificationPath(t, profile.BitrateKbps).traverse(t, payload) + trace, elapsed, err := newQualificationPath(t, profile, profile.BitrateKbps).traverse(t, payload) if err != nil { t.Fatal(err) } @@ -190,12 +190,12 @@ func TestQualificationSixImpairmentProfilesTraverseProductionPath(t *testing.T) func TestQualificationUsesPublicQUICAndProductionPacer(t *testing.T) { qualificationTraverseProfiles(t, qualificationMediaProfiles()) - evidence, err := qualificationPacerEvidence(filepath.Join(t.TempDir(), "fairness.csv.gz")) + evidence, err := qualificationPacerEvidence(t, filepath.Join(t.TempDir(), "fairness.csv.gz"), 2*time.Second, 2*time.Second) if err != nil { t.Fatal(err) } if len(evidence.PerFlowBytes) != 8 || len(evidence.CapacitySteps) != 2 || - len(evidence.Series) != 80 || evidence.RawSamplesSHA256 == "" || evidence.JainIndex < 0.99 { + len(evidence.Series) != 6 || evidence.RawSamplesSHA256 == "" || evidence.JainIndex < 0.99 { t.Fatalf("pacer evidence = %#v", evidence) } } diff --git a/gateway/qualification_harness_test.go b/gateway/qualification_harness_test.go index 706c9d3..7cfc9da 100644 --- a/gateway/qualification_harness_test.go +++ b/gateway/qualification_harness_test.go @@ -5,7 +5,10 @@ import ( "bytes" "compress/gzip" "context" + "crypto/aes" + "crypto/cipher" "crypto/sha256" + "crypto/tls" "encoding/binary" "encoding/hex" "encoding/json" @@ -13,15 +16,21 @@ import ( "fmt" "io" "math" + "net" + "net/http" + "net/http/httptest" "os" "path/filepath" + "reflect" "regexp" "runtime" "runtime/debug" runtimemetrics "runtime/metrics" "sort" + "strconv" "strings" "sync" + "sync/atomic" "testing" "time" @@ -29,7 +38,7 @@ import ( ) const ( - qualificationToolVersion = "versevdi-gateway-qualification/v2" + qualificationToolVersion = "versevdi-gateway-qualification/v3" qualificationImpairmentQueuePackets = 64 qualificationImpairmentMaxPackets = 100_000 qualificationImpairmentPacketCount = 10_000 @@ -47,35 +56,28 @@ type qualificationMediaProfile struct { } type qualificationPathTrace struct { - ApolloRecovered bool - ProductionQueue bool - ProductionPacer bool - VerseQUIC bool - PayloadPreserved bool -} - -type qualificationNativeSession struct { - *nativeApolloSession -} - -func (s *qualificationNativeSession) Telemetry() ProviderTelemetry { - return ProviderTelemetry{State: s.State().State, MediaDrops: s.mediaDrops.Load()} -} - -func (s *qualificationNativeSession) Terminate(context.Context) error { - s.mu.Lock() - s.state.State = ProviderStateTerminated - s.mu.Unlock() - s.closeMediaChannels() - return nil + NativeSetup bool + NativeOpen bool + NativeUDPIngress bool + ApolloRecovered bool + ProductionQueue bool + ProductionMediaLoop bool + ProductionPacer bool + VerseQUIC bool + PublicClientDecode bool + PayloadPreserved bool } type qualificationPath struct { client *Client server *Server session *nativeApolloSession + fixture *qualificationApolloFixture + backend *qualificationTracingBackend key []byte + flow string frame uint32 + bootTrace atomic.Bool closeOnce sync.Once shutdown func() } @@ -249,30 +251,521 @@ func qualificationPayload(profile qualificationMediaProfile) []byte { return payload } -func newQualificationPath(t *testing.T, pacerKbps int64) *qualificationPath { +type qualificationTracingBackend struct { + native *NativeApolloBackend + setups atomic.Uint64 + opens atomic.Uint64 + sessions sync.Map +} + +func (b *qualificationTracingBackend) Management(ctx context.Context, request LaunchRequest) ([]byte, error) { + return b.native.Management(ctx, request) +} + +func (b *qualificationTracingBackend) Setup(ctx context.Context, request LaunchRequest, management []byte) ([]byte, error) { + response, err := b.native.Setup(ctx, request, management) + if err == nil { + b.setups.Add(1) + } + return response, err +} + +func (b *qualificationTracingBackend) Open(ctx context.Context, request LaunchRequest, response RTSPResponse) (ProviderSession, error) { + session, err := b.native.Open(ctx, request, response) + if err == nil { + native, ok := session.(*nativeApolloSession) + if !ok { + return nil, ErrProviderMalformed + } + b.opens.Add(1) + b.sessions.Store(request.SessionID, native) + } + return session, err +} + +func (b *qualificationTracingBackend) session(sessionID string) *nativeApolloSession { + value, _ := b.sessions.Load(sessionID) + session, _ := value.(*nativeApolloSession) + return session +} + +type qualificationApolloFixture struct { + sessionID string + management *httptest.Server + stream net.Listener + control *net.UDPConn + audio *net.UDPConn + video *net.UDPConn + videoRemote atomic.Pointer[net.UDPAddr] + key atomic.Pointer[[]byte] + keyReady chan []byte + failures chan error + closed atomic.Bool + closeOnce sync.Once + sentPackets atomic.Uint64 + work protocol.ProviderSessionWork +} + +func newQualificationApolloFixture(t *testing.T, serverTLS, clientTLS *tls.Config, sessionID string, profile qualificationMediaProfile) *qualificationApolloFixture { t.Helper() - serverTLS, clientTLS := testTLS(t) - key := []byte("0123456789abcdef") - media, err := newApolloMediaCodec(key, 7) + fixture := &qualificationApolloFixture{sessionID: sessionID, keyReady: make(chan []byte, 1), failures: make(chan error, 8)} + var err error + fixture.stream, err = net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatal(err) } - session := newNativeApolloSession("qualification-session") - session.media = media - provider := providerStartFunc(func(ctx context.Context, _ LaunchRequest) (ProviderSession, error) { - if err := session.Ready(ctx); err != nil { - return nil, err + fixture.control, err = net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + fixture.Close() + t.Fatal(err) + } + fixture.audio, err = net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + fixture.Close() + t.Fatal(err) + } + fixture.video, err = net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + fixture.Close() + t.Fatal(err) + } + streamHost, streamPortText, err := net.SplitHostPort(fixture.stream.Addr().String()) + if err != nil { + fixture.Close() + t.Fatal(err) + } + streamPort, err := strconv.ParseInt(streamPortText, 10, 64) + if err != nil { + fixture.Close() + t.Fatal(err) + } + fixture.management = httptest.NewUnstartedServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if request.TLS == nil || len(request.TLS.PeerCertificates) != 1 { + http.Error(response, "mTLS required", http.StatusUnauthorized) + return + } + switch request.URL.Path { + case "/serverinfo": + _, _ = response.Write([]byte("qualification-apollo2571869449984")) + case "/applist": + _, _ = response.Write([]byte("42")) + case "/launch": + key, decodeErr := hex.DecodeString(request.URL.Query().Get("rikey")) + if decodeErr != nil || len(key) != 16 { + http.Error(response, "invalid RIK", http.StatusBadRequest) + return + } + copyKey := append([]byte(nil), key...) + fixture.key.Store(©Key) + select { + case fixture.keyReady <- copyKey: + default: + } + _, _ = response.Write([]byte("rtspenc://" + fixture.stream.Addr().String() + "")) + default: + http.NotFound(response, request) + } + })) + fixture.management.TLS = serverTLS + fixture.management.StartTLS() + managementHost, managementPortText, err := net.SplitHostPort(fixture.management.Listener.Addr().String()) + if err != nil { + fixture.Close() + t.Fatal(err) + } + managementPort, err := strconv.ParseInt(managementPortText, 10, 64) + if err != nil { + fixture.Close() + t.Fatal(err) + } + codec, width, height, fps := "H264", int64(1920), int64(1080), int64(60) + if profile.Codec == "hevc" { + codec = "HEVC" + if strings.Contains(profile.Name, "1440") { + width, height, fps = 2560, 1440, 120 + } else { + width, height, fps = 3840, 2160, 60 + } + } + pinned := sha256.Sum256(serverTLS.Certificates[0].Certificate[0]) + fixture.work = protocol.ProviderSessionWork{ + Version: "1", SessionID: sessionID, GatewayID: "gateway-1", + ExpiresAt: "2099-01-01T00:00:00Z", ProviderProfile: ProviderProfileApollo, + ProviderIdentity: "qualification-apollo#sha256:" + hex.EncodeToString(pinned[:]), + PolicyVersionID: "qualification-policy", ApplicationID: "42", ClientID: "qualification-client", + StreamPolicy: protocol.ProviderStreamPolicy{ + ResolutionWidth: width, ResolutionHeight: height, Fps: fps, + Codec: codec, BitrateKbps: profile.BitrateKbps, AudioEnabled: true, + }, + ManagementHost: managementHost, ManagementPort: managementPort, + StreamHost: streamHost, StreamPort: streamPort, + ClientCertificatePem: certificatePEM(t, clientTLS.Certificates[0]), + ClientPrivateKeyPem: privateKeyPEM(t, clientTLS.Certificates[0]), + ServerCertificatePem: certificatePEM(t, tls.Certificate{Certificate: [][]byte{serverTLS.Certificates[0].Certificate[1]}}), + ClipboardPolicy: protocol.ClipboardPolicy{MaxTextBytes: 65536, MaxUpdatesPerMinute: 30}, + } + go fixture.serveRTSP() + go fixture.serveControl() + go fixture.serveMedia(fixture.audio, false) + go fixture.serveMedia(fixture.video, true) + t.Cleanup(fixture.Close) + return fixture +} + +func (f *qualificationApolloFixture) fail(err error) { + if err == nil || f.closed.Load() { + return + } + select { + case f.failures <- err: + default: + } +} + +func (f *qualificationApolloFixture) serveRTSP() { + key := <-f.keyReady + codec, err := newEncryptedRTSPCodec(key) + if err != nil { + f.fail(err) + return + } + methods := []string{"OPTIONS", "DESCRIBE", "SETUP", "SETUP", "SETUP", "ANNOUNCE", "PLAY"} + targets := []string{"rtspenc://" + f.stream.Addr().String(), "rtspenc://" + f.stream.Addr().String(), "streamid=audio/0/0", "streamid=video/0/0", "streamid=control/13/0", "streamid=control/13/0", "/"} + describe := "a=x-ss-general.featureFlags:1\r\na=x-ss-general.encryptionSupported:7\r\na=x-ss-general.encryptionRequested:1\r\na=fmtp:97 surround-params=21101\r\n" + responses := []string{ + "RTSP/1.0 200 OK\r\nCSeq: 1\r\n\r\n", + fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 2\r\nContent-Type: application/sdp\r\nContent-Length: %d\r\n\r\n%s", len(describe), describe), + fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 3\r\nSession: %s;timeout=90\r\nTransport: unicast;server_port=%d\r\nX-SS-Ping-Payload: 0123456789abcdef\r\n\r\n", f.sessionID, f.audio.LocalAddr().(*net.UDPAddr).Port), + fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 4\r\nSession: %s\r\nTransport: unicast;server_port=%d\r\nX-SS-Ping-Payload: fedcba9876543210\r\n\r\n", f.sessionID, f.video.LocalAddr().(*net.UDPAddr).Port), + fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 5\r\nSession: %s\r\nTransport: unicast;server_port=%d\r\nX-SS-Connect-Data: 305419896\r\n\r\n", f.sessionID, f.control.LocalAddr().(*net.UDPAddr).Port), + fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 6\r\nSession: %s\r\n\r\n", f.sessionID), + fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 7\r\nSession: %s\r\n\r\n", f.sessionID), + } + for index, method := range methods { + connection, acceptErr := f.stream.Accept() + if acceptErr != nil { + f.fail(acceptErr) + return + } + header := make([]byte, encryptedRTSPHeaderSize) + if _, err = io.ReadFull(connection, header); err != nil { + _ = connection.Close() + f.fail(err) + return + } + length := binary.BigEndian.Uint32(header[:4]) & 0x7fffffff + if length > maxApolloRTSPHeaders+maxApolloRTSPBody { + _ = connection.Close() + f.fail(ErrProviderMalformed) + return + } + frame := make([]byte, encryptedRTSPHeaderSize+int(length)) + copy(frame, header) + if _, err = io.ReadFull(connection, frame[encryptedRTSPHeaderSize:]); err != nil { + _ = connection.Close() + f.fail(err) + return + } + plaintext, openErr := codec.OpenClient(frame) + firstLine, _, _ := strings.Cut(string(plaintext), "\r\n") + if openErr != nil || firstLine != method+" "+targets[index]+" RTSP/1.0" { + _ = connection.Close() + f.fail(ErrProviderMalformed) + return + } + responseFrame, sealErr := hostEncryptedRTSPFrameNoTest(key, uint32(index+1), []byte(responses[index])) + if sealErr != nil { + _ = connection.Close() + f.fail(sealErr) + return + } + _, err = connection.Write(responseFrame) + _ = connection.Close() + if err != nil { + f.fail(err) + return + } + } +} + +func hostEncryptedRTSPFrameNoTest(key []byte, sequence uint32, plaintext []byte) ([]byte, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, err + } + aead, err := cipher.NewGCM(block) + if err != nil { + return nil, err + } + nonce := encryptedRTSPNonce(sequence, 'H', 'R') + sealed := aead.Seal(nil, nonce[:], plaintext, nil) + frame := make([]byte, encryptedRTSPHeaderSize+len(plaintext)) + binary.BigEndian.PutUint32(frame[:4], uint32(len(plaintext))|0x80000000) + binary.BigEndian.PutUint32(frame[4:8], sequence) + copy(frame[8:24], sealed[len(plaintext):]) + copy(frame[24:], sealed[:len(plaintext)]) + return frame, nil +} + +func (f *qualificationApolloFixture) serveControl() { + buffer := make([]byte, apolloENetMaximumPacket) + count, remote, err := f.control.ReadFromUDP(buffer) + if err != nil { + f.fail(err) + return + } + connect := buffer[:count] + if count != 52 || connect[4] != apolloENetConnect|apolloENetAcknowledged || + binary.BigEndian.Uint32(connect[20:24]) != apolloENetChannels { + f.fail(ErrProviderMalformed) + return + } + verify := make([]byte, 48) + apolloENetHeader(verify, 0, 0, time.Now()) + verify[4] = apolloENetVerifyConnect | apolloENetAcknowledged + verify[5] = 0xff + binary.BigEndian.PutUint16(verify[6:8], 1) + binary.BigEndian.PutUint16(verify[8:10], 7) + verify[10], verify[11] = 2, 3 + binary.BigEndian.PutUint32(verify[12:16], 1400) + binary.BigEndian.PutUint32(verify[16:20], 32768) + binary.BigEndian.PutUint32(verify[20:24], apolloENetChannels) + binary.BigEndian.PutUint32(verify[44:48], binary.BigEndian.Uint32(connect[44:48])) + if _, err = f.control.WriteToUDP(verify, remote); err != nil { + f.fail(err) + return + } + for { + count, _, err = f.control.ReadFromUDP(buffer) + if err != nil { + f.fail(err) + return + } + packet := buffer[:count] + if len(packet) < 8 { + f.fail(ErrProviderMalformed) + return + } + command, channel := packet[4]&apolloENetCommandMask, packet[5] + sequence := binary.BigEndian.Uint16(packet[6:8]) + switch command { + case 1: + case apolloENetSendReliable, apolloENetPing, apolloENetDisconnect: + if _, err = f.control.WriteToUDP(sourceShapedENetAcknowledgePacket(7, 2, channel, sequence), remote); err != nil { + f.fail(err) + return + } + if command == apolloENetDisconnect { + return + } + case apolloENetSendUnsequenced: + default: + f.fail(ErrProviderMalformed) + return + } + } +} + +func (f *qualificationApolloFixture) serveMedia(socket *net.UDPConn, video bool) { + buffer := make([]byte, apolloMediaMaximumPacket) + for { + _, remote, err := socket.ReadFromUDP(buffer) + if err != nil { + f.fail(err) + return + } + if video && f.videoRemote.Load() == nil { + copyRemote := *remote + f.videoRemote.Store(©Remote) + } + } +} + +func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][]byte) error { + for f.videoRemote.Load() == nil { + select { + case err := <-f.failures: + return err + case <-ctx.Done(): + return ctx.Err() + case <-time.After(time.Millisecond): + } + } + remote := f.videoRemote.Load() + for _, packet := range packets { + if _, err := f.video.WriteToUDP(packet, remote); err != nil { + return err + } + f.sentPackets.Add(1) + } + return nil +} + +func (f *qualificationApolloFixture) streamKey() ([]byte, error) { + key := f.key.Load() + if key == nil || len(*key) != 16 { + return nil, ErrProviderMalformed + } + return append([]byte(nil), (*key)...), nil +} + +func (f *qualificationApolloFixture) Close() { + if f == nil { + return + } + f.closeOnce.Do(func() { + f.closed.Store(true) + if f.management != nil { + f.management.Close() + } + if f.stream != nil { + _ = f.stream.Close() + } + for _, socket := range []*net.UDPConn{f.control, f.audio, f.video} { + if socket != nil { + _ = socket.Close() + } } - return &qualificationNativeSession{nativeApolloSession: session}, nil }) +} + +type qualificationAdmissionRecord struct { + authority protocol.SessionAuthority + work protocol.ProviderSessionWork + used bool +} + +type qualificationAdmission struct { + mu sync.Mutex + records map[string]*qualificationAdmissionRecord +} + +func (a *qualificationAdmission) Admit(_ context.Context, request protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) { + a.mu.Lock() + defer a.mu.Unlock() + record := a.records[request.SessionID] + if record == nil || record.used || request.GatewayID != record.authority.GatewayID || + request.Audience != record.authority.Audience || !reflect.DeepEqual(request.Capabilities, record.authority.Capabilities) { + return protocol.SessionAuthority{}, ErrAdmissionRejected + } + record.used = true + return record.authority, nil +} + +func (a *qualificationAdmission) ProviderWork(_ context.Context, authority protocol.SessionAuthority) (protocol.ProviderSessionWork, error) { + a.mu.Lock() + defer a.mu.Unlock() + record := a.records[authority.SessionID] + if record == nil || !record.used || !reflect.DeepEqual(authority, record.authority) { + return protocol.ProviderSessionWork{}, ErrAdmissionRejected + } + return record.work, nil +} + +func (*qualificationAdmission) Release(context.Context, protocol.SessionAuthority) error { return nil } + +type qualificationFleet struct { + t *testing.T + server *Server + paths []*qualificationPath + clients []*Client + cancel context.CancelFunc + serveDone chan error + closeOnce sync.Once +} + +func newQualificationFleet(t *testing.T, count int, profile qualificationMediaProfile, pacerKbps int64) *qualificationFleet { + t.Helper() + serverTLS, clientTLS := testTLS(t) + backend := &qualificationTracingBackend{native: NewNativeApolloBackend()} + admission := &qualificationAdmission{records: make(map[string]*qualificationAdmissionRecord, count)} + fixtures := make([]*qualificationApolloFixture, 0, count) + for index := 0; index < count; index++ { + sessionID := fmt.Sprintf("qualification-flow-%d", index+1) + fixture := newQualificationApolloFixture(t, serverTLS, clientTLS, sessionID, profile) + authority := protocol.SessionAuthority{ + Version: "1", SessionID: sessionID, GatewayID: "gateway-1", Audience: "versevdi-gateway", + ExpiresAt: time.Now().Add(2 * time.Minute).UTC().Format(time.RFC3339Nano), + Capabilities: DefaultCapabilities(), ProviderProfile: ProviderProfileApollo, + ProviderIdentity: fixture.work.ProviderIdentity, + } + work := fixture.work + work.ExpiresAt = authority.ExpiresAt + admission.records[sessionID] = &qualificationAdmissionRecord{authority: authority, work: work} + fixtures = append(fixtures, fixture) + } + server, err := NewServer(ServerConfig{ + ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: "gateway-1", + Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), + Admission: admission, ProviderStateReporter: &recordingProviderStateReporter{}, + Provider: NewApolloAdapter(backend, ProviderIdentity{}), PacerKbps: pacerKbps, + }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + fleet := &qualificationFleet{t: t, server: server, cancel: cancel, serveDone: make(chan error, 1)} + go func() { fleet.serveDone <- server.Serve(ctx) }() + for index, fixture := range fixtures { + sessionID := fmt.Sprintf("qualification-flow-%d", index+1) + request := protocol.TunnelAdmissionRequest{ + Version: "1", SessionID: sessionID, GatewayID: "gateway-1", Audience: "versevdi-gateway", + Grant: strings.Repeat(string(rune('a'+index)), 64), ClientNonce: fmt.Sprintf("nonce-fleet-%06d", index), + DeviceSignature: strings.Repeat("s", 86), Capabilities: DefaultCapabilities(), + } + client, err := Dial(context.Background(), server.Addr().String(), clientTLS, request) + if err != nil { + fleet.Close() + t.Fatal(err) + } + session := backend.session(sessionID) + key, keyErr := fixture.streamKey() + if keyErr != nil || session == nil { + fleet.Close() + t.Fatalf("qualification fleet session %s unavailable: %v", sessionID, keyErr) + } + fleet.clients = append(fleet.clients, client) + fleet.paths = append(fleet.paths, &qualificationPath{ + client: client, server: server, session: session, fixture: fixture, + backend: backend, key: key, shutdown: func() {}, + }) + } + t.Cleanup(fleet.Close) + return fleet +} + +func (f *qualificationFleet) Close() { + if f == nil { + return + } + f.closeOnce.Do(func() { + for _, client := range f.clients { + _ = client.Close() + } + f.cancel() + _ = f.server.Close() + if err := <-f.serveDone; err != nil { + f.t.Errorf("serve qualification fleet: %v", err) + } + }) +} + +func newQualificationPath(t *testing.T, profile qualificationMediaProfile, pacerKbps int64) *qualificationPath { + t.Helper() + serverTLS, clientTLS := testTLS(t) + fixture := newQualificationApolloFixture(t, serverTLS, clientTLS, "qualification-session", profile) + backend := &qualificationTracingBackend{native: NewNativeApolloBackend()} + provider := NewApolloAdapter(backend, ProviderIdentity{}) authority := protocol.SessionAuthority{ Version: "1", SessionID: "qualification-session", GatewayID: "gateway-1", Audience: "versevdi-gateway", ReconnectSequence: 0, ExpiresAt: time.Now().Add(45 * time.Minute).UTC().Format(time.RFC3339Nano), Capabilities: DefaultCapabilities(), ProviderProfile: ProviderProfileApollo, - ProviderIdentity: "apollo-fixture#sha256:qualification", + ProviderIdentity: fixture.work.ProviderIdentity, } - admission := &oneTimeAdmission{authority: authority, released: make(chan struct{}), disableClipboard: true} + work := fixture.work + work.ExpiresAt = authority.ExpiresAt + admission := &oneTimeAdmission{authority: authority, released: make(chan struct{}), disableClipboard: true, providerWork: &work} server, err := NewServer(ServerConfig{ ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: authority.GatewayID, Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), @@ -297,7 +790,15 @@ func newQualificationPath(t *testing.T, pacerKbps int64) *qualificationPath { _ = server.Close() t.Fatal(err) } - path := &qualificationPath{client: client, server: server, session: session, key: key} + session := backend.session(authority.SessionID) + key, err := fixture.streamKey() + if err != nil || session == nil { + _ = client.Close() + cancel() + _ = server.Close() + t.Fatalf("native qualification session unavailable: %v", err) + } + path := &qualificationPath{client: client, server: server, session: session, fixture: fixture, backend: backend, key: key} path.shutdown = func() { _ = client.Close() cancel() @@ -318,7 +819,10 @@ func (p *qualificationPath) Close() { func (p *qualificationPath) traverse(t *testing.T, payload []byte) (qualificationPathTrace, time.Duration, error) { t.Helper() - started := time.Now() + beforeMetrics := p.server.Metrics() + beforeIngress := p.session.mediaIngress.Load() + beforeRecovered := p.session.mediaRecovered.Load() + beforeEnqueued := p.session.mediaEnqueued.Load() trace, err := p.emit(t, payload) if err != nil { return trace, 0, err @@ -327,12 +831,20 @@ func (p *qualificationPath) traverse(t *testing.T, payload []byte) (qualificatio if err != nil { return trace, 0, err } - trace.VerseQUIC = true + afterMetrics := p.server.Metrics() + trace.NativeUDPIngress = p.session.mediaIngress.Load() > beforeIngress + trace.ApolloRecovered = p.session.mediaRecovered.Load() > beforeRecovered + trace.ProductionQueue = p.session.mediaEnqueued.Load() > beforeEnqueued + trace.ProductionMediaLoop = afterMetrics.ProcessingSamples == beforeMetrics.ProcessingSamples+1 + trace.ProductionPacer = afterMetrics.PacingDelayNanos > beforeMetrics.PacingDelayNanos + trace.VerseQUIC = afterMetrics.MediaPackets > beforeMetrics.MediaPackets + trace.PublicClientDecode = true trace.PayloadPreserved = bytes.Equal(recovered, payload) - metrics := p.server.Metrics() - trace.ProductionPacer = metrics.ProcessingSamples > 0 && metrics.MediaPackets > 0 && - metrics.PacingDelayNanos > 0 && metrics.QueueDelayNanos > 0 - return trace, time.Since(started), nil + if p.bootTrace.CompareAndSwap(false, true) { + trace.NativeSetup = p.backend.setups.Load() == 1 + trace.NativeOpen = p.backend.opens.Load() == 1 + } + return trace, time.Duration(afterMetrics.ProcessingDelayNanos - beforeMetrics.ProcessingDelayNanos), nil } func (p *qualificationPath) emit(t *testing.T, payload []byte) (qualificationPathTrace, error) { @@ -342,25 +854,12 @@ func (p *qualificationPath) emit(t *testing.T, payload []byte) (qualificationPat } p.frame++ packets := qualificationSourceVideoPackets(t, p.key, p.frame, payload) - trace := qualificationPathTrace{} - for _, packet := range packets { - shard, err := p.session.media.OpenVideo(packet) - if err != nil { - return trace, err - } - recovered, err := p.session.videoFEC.Add(shard) - if err != nil { - return trace, err - } - if len(recovered) != 0 { - trace.ApolloRecovered = true - trace.ProductionQueue = !pushLatest(p.session.video, recovered) - } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := p.fixture.sendVideo(ctx, packets); err != nil { + return qualificationPathTrace{}, err } - if !trace.ApolloRecovered { - return trace, errors.New("Apollo FEC produced no recovered payload") - } - return trace, nil + return qualificationPathTrace{}, nil } func (p *qualificationPath) receivePayload(parent context.Context) ([]byte, error) { @@ -370,7 +869,7 @@ func (p *qualificationPath) receivePayload(parent context.Context) ([]byte, erro var sequence uint32 var fragmentCount byte for { - frame, err := qualificationReceiveFrame(ctx, p.client) + frame, err := p.client.ReceiveFrame(ctx) if err != nil { return nil, err } @@ -391,30 +890,6 @@ func (p *qualificationPath) receivePayload(parent context.Context) ([]byte, erro return recovered, nil } -func qualificationReceiveFrame(ctx context.Context, client *Client) (Frame, error) { - if client == nil || client.connection == nil { - return Frame{}, ErrProviderMalformed - } - raw, err := client.connection.ReceiveDatagram(ctx) - if err != nil { - return Frame{}, err - } - if len(raw) < frameHeaderSize || len(raw) > maxFrameSize || - raw[0] != 'V' || raw[1] != 'D' || raw[2] != 1 || raw[3] != ChannelVideo || raw[4] != 0 { - return Frame{}, ErrProviderMalformed - } - length := int(binary.BigEndian.Uint16(raw[19:21])) - if length > 1179 || len(raw) != frameHeaderSize+length || raw[18] == 0 || - raw[18] > maxFragmentCount || raw[17] >= raw[18] { - return Frame{}, ErrProviderMalformed - } - return Frame{ - Channel: raw[3], Sequence: binary.BigEndian.Uint32(raw[5:9]), - FragmentIndex: raw[17], FragmentCount: raw[18], - Payload: append([]byte(nil), raw[frameHeaderSize:]...), - }, nil -} - func qualificationSourceVideoPackets(t *testing.T, key []byte, frame uint32, encoded []byte) [][]byte { t.Helper() if len(encoded) <= apolloVideoShardPayloadSize-8 { @@ -448,7 +923,7 @@ func qualificationVideoIV(frame uint32, shard byte) string { func qualificationProductionPathSmoke(t *testing.T, profile qualificationMediaProfile) qualificationPathTrace { t.Helper() - path := newQualificationPath(t, profile.BitrateKbps) + path := newQualificationPath(t, profile, profile.BitrateKbps) defer path.Close() trace, _, err := path.traverse(t, qualificationPayload(profile)) if err != nil { @@ -457,6 +932,17 @@ func qualificationProductionPathSmoke(t *testing.T, profile qualificationMediaPr return trace } +func TestQualificationTraversesNativePublicGatewayPath(t *testing.T) { + profile := qualificationMediaProfiles()[0] + profile.BitrateKbps = 100000 + trace := qualificationProductionPathSmoke(t, profile) + if !trace.NativeSetup || !trace.NativeOpen || !trace.NativeUDPIngress || !trace.ApolloRecovered || + !trace.ProductionQueue || !trace.ProductionMediaLoop || !trace.ProductionPacer || + !trace.VerseQUIC || !trace.PublicClientDecode || !trace.PayloadPreserved { + t.Fatalf("qualification path skipped production stages: %#v", trace) + } +} + func summarizeQualificationSamples(samples []time.Duration) (qualificationProcessingSummary, error) { if len(samples) == 0 { return qualificationProcessingSummary{}, errors.New("qualification has no processing samples") @@ -552,7 +1038,7 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro _ = file.Close() } }() - if _, err := buffered.WriteString("source_sequence,sent_ns,delivered_ns,processing_ns,outcome,delivery_order,bytes\n"); err != nil { + if _, err := buffered.WriteString("source_sequence,sent_ns,delivered_ns,processing_ns,outcome,delivery_order,bytes,queue_packets\n"); err != nil { return qualificationImpairmentObservation{}, err } state := qualificationImpairmentSeed @@ -572,7 +1058,7 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro ConfiguredLossPercent: profile.LossPercent, ConfiguredReorder: profile.Reorder, ConfiguredCapacitySteps: append([]int(nil), profile.CapacitySteps...), } - path := newQualificationPath(t, media.BitrateKbps) + path := newQualificationPath(t, media, media.BitrateKbps) defer path.Close() started := time.Now() payload := qualificationPayload(media) @@ -630,14 +1116,13 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro deliveryOrder++ wireBytes := int64(len(current) + frameHeaderSize) deliveries = append(deliveries, qualificationDeliverySample{At: deliveredAt, Bytes: wireBytes}) - if _, err := fmt.Fprintf(buffered, "%d,%d,%d,%d,delivered,%d,%d\n", packet.index, + queuePackets := int(path.session.mediaQueueMaximum.Load()) + if _, err := fmt.Fprintf(buffered, "%d,%d,%d,%d,delivered,%d,%d,%d\n", packet.index, sentAt.Sub(started).Nanoseconds(), deliveredAt.Sub(started).Nanoseconds(), - processing.Nanoseconds(), deliveryOrder, len(current)); err != nil { + processing.Nanoseconds(), deliveryOrder, len(current), queuePackets); err != nil { return err } - if observation.MaxQueuePackets < 1 { - observation.MaxQueuePackets = 1 - } + observation.MaxQueuePackets = max(observation.MaxQueuePackets, queuePackets) return nil } for index := 0; index < packetCount; index++ { @@ -648,7 +1133,7 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro } if float64(random()%10_000) < profile.LossPercent*100 { observation.Dropped++ - if _, err := fmt.Fprintf(buffered, "%d,%d,0,0,dropped,0,0\n", index, time.Duration(index)*spacing); err != nil { + if _, err := fmt.Fprintf(buffered, "%d,%d,0,0,dropped,0,0,%d\n", index, time.Duration(index)*spacing, path.session.mediaQueueMaximum.Load()); err != nil { return qualificationImpairmentObservation{}, err } continue @@ -693,6 +1178,13 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro } observation.ObservedLossPercent = float64(observation.Dropped) * 100 / float64(packetCount) observation.ObservedReorderPercent = float64(observation.ObservedOutOfOrder) * 100 / float64(packetCount) + if packetCount >= qualificationImpairmentPacketCount { + lowerThroughput := float64(media.BitrateKbps) * (1 - profile.LossPercent/100) * 0.90 + upperThroughput := float64(media.BitrateKbps) * 1.05 + if observation.ObservedThroughputKbps < lowerThroughput || observation.ObservedThroughputKbps > upperThroughput { + return qualificationImpairmentObservation{}, fmt.Errorf("observed throughput %.2f outside [%.2f,%.2f]", observation.ObservedThroughputKbps, lowerThroughput, upperThroughput) + } + } observation.RawSamples = filepath.Base(rawPath) observation.RawSamplesSHA256, observation.RawSamplesBytes, err = qualificationFileSHA256(rawPath) if err != nil { @@ -744,6 +1236,8 @@ func qualificationDeliveriesAfter(deliveries []qualificationDeliverySample, star func qualificationMeasuredConvergence(deliveries []qualificationDeliverySample, start time.Time, targetBytesPerSecond int64) time.Duration { const window = 250 * time.Millisecond + const requiredWindows = 4 + consecutive := 0 for offset := time.Duration(0); offset <= 10*time.Second; offset += window { windowStart := start.Add(offset) var total int64 @@ -754,7 +1248,12 @@ func qualificationMeasuredConvergence(deliveries []qualificationDeliverySample, } rate := total * int64(time.Second) / int64(window) if rate >= targetBytesPerSecond*90/100 && rate <= targetBytesPerSecond*105/100 { - return offset + window + consecutive++ + if consecutive == requiredWindows { + return offset + window + } + } else { + consecutive = 0 } if len(deliveries) > 0 && windowStart.After(deliveries[len(deliveries)-1].At) { break @@ -785,7 +1284,7 @@ func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile, return qualificationProcessingSummary{}, errors.New("qualification payload too small") } pacerKbps := (profile.BitrateKbps*int64(profile.PacketBytes+frameHeaderSize) + int64(profile.PacketBytes) - 1) / int64(profile.PacketBytes) - path := newQualificationPath(t, pacerKbps) + path := newQualificationPath(t, profile, pacerKbps) defer path.Close() if err := runQualificationWarmup(t, path, profile, payload); err != nil { return qualificationProcessingSummary{}, err @@ -812,52 +1311,30 @@ func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile, return qualificationProcessingSummary{}, err } bytesPerSecond := profile.BitrateKbps * 1000 / 8 - targetPackets := bytesPerSecond * profile.Duration.Nanoseconds() / int64(time.Second) / int64(profile.PacketBytes) + targetBytes := bytesPerSecond * profile.Duration.Nanoseconds() / int64(time.Second) + targetPackets := (targetBytes + int64(profile.PacketBytes) - 1) / int64(profile.PacketBytes) samples := make([]time.Duration, 0, int(targetPackets)) started := time.Now() resources := []qualificationResourceSample{qualificationRuntimeSample(started)} lastResourceSample := started var processed int64 - for processed < targetPackets { - batch := int64(8) - if remaining := targetPackets - processed; remaining < batch { - batch = remaining + for processed < targetPackets || time.Since(started) < profile.Duration { + current := append([]byte(nil), payload...) + binary.BigEndian.PutUint32(current[len(current)-4:], uint32(processed)) + trace, sample, traverseErr := path.traverse(t, current) + if traverseErr != nil { + return qualificationProcessingSummary{}, traverseErr } - startedAt := make([]time.Time, batch) - for index := int64(0); index < batch; index++ { - current := append([]byte(nil), payload...) - binary.BigEndian.PutUint32(current[len(current)-4:], uint32(processed+index)) - startedAt[index] = time.Now() - trace, emitErr := path.emit(t, current) - if emitErr != nil { - return qualificationProcessingSummary{}, emitErr - } - if !trace.ApolloRecovered || !trace.ProductionQueue { - return qualificationProcessingSummary{}, errors.New("qualification bypassed Apollo recovery or bounded provider queue") - } + if !trace.NativeUDPIngress || !trace.ApolloRecovered || !trace.ProductionQueue || + !trace.ProductionMediaLoop || !trace.ProductionPacer || !trace.VerseQUIC || + !trace.PublicClientDecode || !trace.PayloadPreserved { + return qualificationProcessingSummary{}, errors.New("qualification bypassed the production provider-to-client path") } - for index := int64(0); index < batch; index++ { - recovered, receiveErr := path.receivePayload(context.Background()) - if receiveErr != nil { - return qualificationProcessingSummary{}, receiveErr - } - want := append([]byte(nil), payload...) - binary.BigEndian.PutUint32(want[len(want)-4:], uint32(processed+index)) - if !bytes.Equal(recovered, want) { - return qualificationProcessingSummary{}, errors.New("qualification payload integrity failure") - } - sample := time.Since(startedAt[index]) - samples = append(samples, sample) - if _, writeErr := fmt.Fprintf(buffered, "%d,%d\n", time.Since(started).Nanoseconds(), sample.Nanoseconds()); writeErr != nil { - return qualificationProcessingSummary{}, writeErr - } - } - processed += batch - metrics := path.server.Metrics() - if metrics.ProcessingSamples < uint64(processed) || metrics.MediaPackets < uint64(processed) || - metrics.PacingDelayNanos == 0 || metrics.QueueDelayNanos == 0 { - return qualificationProcessingSummary{}, errors.New("qualification bypassed production pacing, framing, or QUIC") + samples = append(samples, sample) + if _, writeErr := fmt.Fprintf(buffered, "%d,%d\n", time.Since(started).Nanoseconds(), sample.Nanoseconds()); writeErr != nil { + return qualificationProcessingSummary{}, writeErr } + processed++ if time.Since(lastResourceSample) >= time.Second { resources = append(resources, qualificationRuntimeSample(started)) lastResourceSample = time.Now() @@ -919,8 +1396,11 @@ func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile, summary.PeakGoroutines = resource.Goroutines } } - if summary.ObservedBitrateKbps < float64(profile.BitrateKbps)*0.95 { - return qualificationProcessingSummary{}, fmt.Errorf("observed bitrate %.2f below profile %d", summary.ObservedBitrateKbps, profile.BitrateKbps) + if actualDuration < profile.Duration || actualDuration > profile.Duration*105/100+250*time.Millisecond { + return qualificationProcessingSummary{}, fmt.Errorf("wall-clock duration %s outside [%s,%s]", actualDuration, profile.Duration, profile.Duration*105/100+250*time.Millisecond) + } + if summary.ObservedBitrateKbps < float64(profile.BitrateKbps)*0.95 || summary.ObservedBitrateKbps > float64(profile.BitrateKbps)*1.05 { + return qualificationProcessingSummary{}, fmt.Errorf("observed bitrate %.2f outside profile bounds for %d", summary.ObservedBitrateKbps, profile.BitrateKbps) } if err := enforceQualificationProcessingGate(summary); err != nil { return qualificationProcessingSummary{}, err @@ -1006,15 +1486,66 @@ func qualificationFileSHA256(path string) (string, int64, error) { return hex.EncodeToString(hash.Sum(nil)), size, nil } -func qualificationPacerEvidence(rawPath string) (qualificationFairnessEvidence, error) { - start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) +type qualificationFlowDelivery struct { + at time.Time + flow string + bytes int64 +} + +func runQualificationFleetStage(t *testing.T, fleet *qualificationFleet, profile qualificationMediaProfile, duration time.Duration) ([]qualificationFlowDelivery, error) { + t.Helper() + if duration <= 0 { + return nil, errors.New("qualification fleet duration must be positive") + } + end := time.Now().Add(duration) + payload := qualificationPayload(profile) + deliveries := make([]qualificationFlowDelivery, 0, int(duration/time.Millisecond)) + var sequence uint32 + for time.Now().Before(end) { + for _, path := range fleet.paths { + if !time.Now().Before(end) { + break + } + current := append([]byte(nil), payload...) + binary.BigEndian.PutUint32(current[len(current)-4:], sequence) + sequence++ + trace, _, err := path.traverse(t, current) + if err != nil { + return nil, err + } + if !trace.NativeUDPIngress || !trace.ApolloRecovered || !trace.ProductionQueue || + !trace.ProductionMediaLoop || !trace.ProductionPacer || !trace.VerseQUIC || + !trace.PublicClientDecode || !trace.PayloadPreserved { + return nil, errors.New("fairness traffic bypassed production gateway traversal") + } + deliveries = append(deliveries, qualificationFlowDelivery{ + at: time.Now(), flow: path.flow, bytes: int64(len(current) + frameHeaderSize), + }) + } + } + return deliveries, nil +} + +func qualificationPacerEvidence(t *testing.T, rawPath string, baselineDuration, stepDuration time.Duration) (qualificationFairnessEvidence, error) { + t.Helper() flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"} - pacer := newFairPacer(8000) - next := make(map[string]time.Time, len(flows)) - baseline := runSyntheticPacer(pacer, start, start.Add(60*time.Second), flows, next) - allDeliveries := append([]syntheticPacerDelivery(nil), baseline...) + profile := qualificationMediaProfile{Name: "fairness-h264", Codec: "h264", BitrateKbps: 8000, PacketBytes: 1000} + fleet := newQualificationFleet(t, len(flows), profile, 8000) + defer fleet.Close() + for index, path := range fleet.paths { + path.flow = flows[index] + if _, _, err := path.traverse(t, qualificationPayload(profile)); err != nil { + return qualificationFairnessEvidence{}, err + } + } + start := time.Now() + baseline, err := runQualificationFleetStage(t, fleet, profile, baselineDuration) + if err != nil { + return qualificationFairnessEvidence{}, err + } + allDeliveries := append([]qualificationFlowDelivery(nil), baseline...) evidence := qualificationFairnessEvidence{ - Evaluation: 60 * time.Second, PerFlowBytes: make(map[string]int64, len(flows)), + Evaluation: baselineDuration, PerFlowBytes: make(map[string]int64, len(flows)), ShareError: make(map[string]float64, len(flows)), } var total, squares float64 @@ -1036,19 +1567,21 @@ func qualificationPacerEvidence(rawPath string) (qualificationFairnessEvidence, for _, step := range []struct { reduction int kbps int64 - start time.Time - end time.Time cap int64 }{ - {25, 6000, start.Add(60 * time.Second), start.Add(70 * time.Second), 750_000}, - {50, 4000, start.Add(70 * time.Second), start.Add(80 * time.Second), 500_000}, + {25, 6000, 750_000}, + {50, 4000, 500_000}, } { - pacer.setKbps(step.kbps) - deliveries := runSyntheticPacer(pacer, step.start, step.end, flows, next) + fleet.server.pacer.setKbps(step.kbps) + stepStart := time.Now() + deliveries, runErr := runQualificationFleetStage(t, fleet, profile, stepDuration) + if runErr != nil { + return qualificationFairnessEvidence{}, runErr + } allDeliveries = append(allDeliveries, deliveries...) - convergence := qualificationPacerConvergence(deliveries, step.start, flows, step.cap) + convergence := qualificationPacerConvergence(deliveries, stepStart, flows, step.cap) maximum := qualificationMaximumFiveSecondBytes(deliveries) - if convergence > 10*time.Second || maximum > step.cap*5*105/100 { + if stepDuration >= 10*time.Second && (convergence > 10*time.Second || maximum > step.cap*5*105/100) { return qualificationFairnessEvidence{}, fmt.Errorf("capacity step %d failed convergence=%s five-second=%d", step.reduction, convergence, maximum) } evidence.CapacitySteps = append(evidence.CapacitySteps, qualificationCapacityStep{ @@ -1056,11 +1589,11 @@ func qualificationPacerEvidence(rawPath string) (qualificationFairnessEvidence, MaximumFiveSecond: maximum, FiveSecondCap: step.cap * 5, }) } - evidence.Series = qualificationFairnessSeriesFor(allDeliveries, start, start.Add(80*time.Second), flows) + end := start.Add(baselineDuration + 2*stepDuration) + evidence.Series = qualificationFairnessSeriesFor(allDeliveries, start, end, flows) if err := writeQualificationPacerSamples(rawPath, allDeliveries, start); err != nil { return qualificationFairnessEvidence{}, err } - var err error evidence.RawSamples = filepath.Base(rawPath) evidence.RawSamplesSHA256, evidence.RawSamplesBytes, err = qualificationFileSHA256(rawPath) if err != nil { @@ -1069,7 +1602,8 @@ func qualificationPacerEvidence(rawPath string) (qualificationFairnessEvidence, return evidence, nil } -func qualificationPacerConvergence(deliveries []syntheticPacerDelivery, start time.Time, flows []string, targetBytesPerSecond int64) time.Duration { +func qualificationPacerConvergence(deliveries []qualificationFlowDelivery, start time.Time, flows []string, targetBytesPerSecond int64) time.Duration { + consecutive := 0 for second := time.Duration(0); second < 10*time.Second; second += time.Second { windowStart := start.Add(second) perFlow := make(map[string]int64, len(flows)) @@ -1081,6 +1615,7 @@ func qualificationPacerConvergence(deliveries []syntheticPacerDelivery, start ti } } if aggregate < targetBytesPerSecond*90/100 || aggregate > targetBytesPerSecond*105/100 { + consecutive = 0 continue } targetFlow := targetBytesPerSecond / int64(len(flows)) @@ -1089,13 +1624,18 @@ func qualificationPacerConvergence(deliveries []syntheticPacerDelivery, start ti converged = converged && perFlow[flow] >= targetFlow*90/100 && perFlow[flow] <= targetFlow*110/100 } if converged { - return second + time.Second + consecutive++ + if consecutive == 2 { + return second + time.Second + } + } else { + consecutive = 0 } } return 11 * time.Second } -func qualificationMaximumFiveSecondBytes(deliveries []syntheticPacerDelivery) int64 { +func qualificationMaximumFiveSecondBytes(deliveries []qualificationFlowDelivery) int64 { sort.Slice(deliveries, func(first, second int) bool { return deliveries[first].at.Before(deliveries[second].at) }) var maximum, total int64 for first, last := 0, 0; first < len(deliveries); first++ { @@ -1111,7 +1651,7 @@ func qualificationMaximumFiveSecondBytes(deliveries []syntheticPacerDelivery) in return maximum } -func qualificationFairnessSeriesFor(deliveries []syntheticPacerDelivery, start, end time.Time, flows []string) []qualificationFairnessSeries { +func qualificationFairnessSeriesFor(deliveries []qualificationFlowDelivery, start, end time.Time, flows []string) []qualificationFairnessSeries { series := make([]qualificationFairnessSeries, 0, int(end.Sub(start)/time.Second)) for windowStart := start; windowStart.Before(end); windowStart = windowStart.Add(time.Second) { sample := qualificationFairnessSeries{ @@ -1128,7 +1668,7 @@ func qualificationFairnessSeriesFor(deliveries []syntheticPacerDelivery, start, return series } -func writeQualificationPacerSamples(path string, deliveries []syntheticPacerDelivery, start time.Time) error { +func writeQualificationPacerSamples(path string, deliveries []qualificationFlowDelivery, start time.Time) error { file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) if err != nil { return err @@ -1247,7 +1787,7 @@ func TestSection7Qualification(t *testing.T) { StartedAt: started.Format(time.RFC3339Nano), GoVersion: runtime.Version(), ToolVersions: toolVersions, OS: runtime.GOOS, Architecture: runtime.GOARCH, - Topology: "source-shaped encrypted Apollo fixture -> native recovery/FEC -> bounded provider queue -> production fair pacer -> Verse framing over mTLS/QUIC -> independent fixture client", + Topology: "source-shaped encrypted Apollo fixture -> native recovery/FEC -> bounded provider queue -> production fair pacer -> Verse framing over mTLS/QUIC -> public Verse client decoder", Direction: "provider_to_client", QueueDiscipline: "bounded 16-packet native provider queue, production equal-tier fair pacer, deterministic fixed-seed source impairment", Evidence: []string{"deterministic source-shaped Apollo recovery", "local real-time production path", "mTLS/QUIC fixture transport", "path impairment", "production fair pacer"}, @@ -1262,7 +1802,7 @@ func TestSection7Qualification(t *testing.T) { manifest.Processing = append(manifest.Processing, summary) t.Logf("%s count=%d p95=%s observed=%.2f kbps", profile.Name, summary.Count, summary.P95, summary.ObservedBitrateKbps) } - fairness, err := qualificationPacerEvidence(filepath.Join(output, "fairness.csv.gz")) + fairness, err := qualificationPacerEvidence(t, filepath.Join(output, "fairness.csv.gz"), 60*time.Second, 10*time.Second) if err != nil { t.Fatal(err) } @@ -1303,8 +1843,9 @@ func qualificationTraverseProfiles(t *testing.T, profiles []qualificationMediaPr t.Helper() for _, profile := range profiles { trace := qualificationProductionPathSmoke(t, profile) - if !trace.ApolloRecovered || !trace.ProductionQueue || !trace.ProductionPacer || - !trace.VerseQUIC || !trace.PayloadPreserved { + if !trace.NativeSetup || !trace.NativeOpen || !trace.NativeUDPIngress || !trace.ApolloRecovered || + !trace.ProductionQueue || !trace.ProductionMediaLoop || !trace.ProductionPacer || + !trace.VerseQUIC || !trace.PublicClientDecode || !trace.PayloadPreserved { t.Fatalf("%s production path: %#v", profile.Name, trace) } } diff --git a/gateway/transport.go b/gateway/transport.go index 112d90e..ddbb067 100644 --- a/gateway/transport.go +++ b/gateway/transport.go @@ -9,6 +9,7 @@ import ( "fmt" "io" "net" + "slices" "sync" "sync/atomic" "time" @@ -19,14 +20,13 @@ import ( ) const ( - defaultHelloLimit = 16 * 1024 - defaultControlLimit = 128 * 1024 - clientControlBacklog = 64 - applicationError = quic.ApplicationErrorCode(0x100) - terminalFeedbackDrain = 100 * time.Millisecond - controlFlowID = "control.ack.v1" - inputFlowID = "input.sequenced.v1" - clipboardFlowID = "clipboard.text.v1" + defaultHelloLimit = 16 * 1024 + defaultControlLimit = 128 * 1024 + clientControlBacklog = 64 + applicationError = quic.ApplicationErrorCode(0x100) + controlFlowID = "control.ack.v1" + inputFlowID = "input.sequenced.v1" + clipboardFlowID = "clipboard.text.v1" ) var ( @@ -101,12 +101,15 @@ func NewServer(config ServerConfig) (*Server, error) { if err := validateServerTLS(config.TLSConfig); err != nil { return nil, err } - if config.Capabilities == (protocol.CapabilityProfile{}) { + if capabilityProfileUnset(config.Capabilities) { config.Capabilities = DefaultCapabilities() } - if config.ProviderCapabilities == (protocol.CapabilityProfile{}) { + if capabilityProfileUnset(config.ProviderCapabilities) { config.ProviderCapabilities = DefaultCapabilities() } + if config.Capabilities.Validate() != nil || config.ProviderCapabilities.Validate() != nil { + return nil, ErrNoCapabilityOverlap + } if config.ProviderProfile == "" { config.ProviderProfile = ProviderProfileApollo } @@ -241,6 +244,13 @@ func (s *Server) handleConnection(parent context.Context, connection *quic.Conn) _ = writeStableError(stream, "no_capability_overlap", err, false) return } + selected, err = selectApolloPolicyCapabilities(work.StreamPolicy, selected) + if err != nil { + _ = s.config.Admission.Release(context.Background(), authority) + s.metrics.AdmissionRejects.Add(1) + _ = writeStableError(stream, "no_capability_overlap", err, false) + return + } clipboard, err := newClipboardGate(work.ClipboardPolicy, time.Now) if err != nil { _ = s.config.Admission.Release(context.Background(), authority) @@ -335,13 +345,26 @@ func apolloPolicyMatchesCapabilities(policy protocol.ProviderStreamPolicy, capab if validateApolloStreamPolicy(policy) != nil || capabilities.Audio != "encoded" { return false } + required := apolloPolicyProfile(policy) + return required != "" && slices.Contains(capabilities.ClientDecode, required) +} + +func selectApolloPolicyCapabilities(policy protocol.ProviderStreamPolicy, capabilities protocol.CapabilityProfile) (protocol.CapabilityProfile, error) { + if !apolloPolicyMatchesCapabilities(policy, capabilities) { + return protocol.CapabilityProfile{}, ErrNoCapabilityOverlap + } + capabilities.ClientDecode = []string{apolloPolicyProfile(policy)} + return capabilities, nil +} + +func apolloPolicyProfile(policy protocol.ProviderStreamPolicy) string { switch policy.Codec { case "H264": - return capabilities.ClientDecode == "h264-opus" || capabilities.ClientDecode == defaultClientDecode + return "h264-opus" case "HEVC": - return capabilities.ClientDecode == "hevc-opus" || capabilities.ClientDecode == defaultClientDecode + return "hevc-opus" default: - return false + return "" } } @@ -370,9 +393,12 @@ type gatewaySession struct { cleanupOnce sync.Once inputMu sync.Mutex controlWriteMu sync.Mutex + outputMu sync.Mutex pressed map[string]struct{} sequence atomic.Uint32 mediaDrops uint64 + mediaQuiesced bool + terminalSent atomic.Bool endReason error result chan error } @@ -408,6 +434,13 @@ func (s *gatewaySession) run() { case s.endReason = <-s.result: } s.cancel() + if s.terminalSent.Load() { + s.cleanup() + select { + case <-s.connection.Context().Done(): + case <-timer.C: + } + } } func (s *gatewaySession) providerEventLoop() { @@ -420,28 +453,32 @@ func (s *gatewaySession) providerEventLoop() { if !ok { return } - if event.Kind == ProviderEventDisconnected { - s.result <- ErrProviderDisconnected - return + terminal := event.Kind == ProviderEventTerminated || event.Kind == ProviderEventDisconnected + if terminal { + s.outputMu.Lock() + s.mediaQuiesced = true } payload, err := EncodeProviderEvent(event) if err == nil { err = s.sendControl(s.sequence.Add(1), payload) } + if terminal { + s.outputMu.Unlock() + } if err != nil { s.result <- err return } - if event.Kind == ProviderEventTerminated { - timer := time.NewTimer(terminalFeedbackDrain) - select { - case <-s.ctx.Done(): - timer.Stop() - return - case <-timer.C: - } + if terminal { + s.terminalSent.Store(true) + } + switch event.Kind { + case ProviderEventTerminated: s.result <- ErrProviderTerminated return + case ProviderEventDisconnected: + s.result <- ErrProviderDisconnected + return } } } @@ -604,21 +641,21 @@ func (s *gatewaySession) mediaLoop() { select { case <-s.ctx.Done(): return - case payload, ok := <-video: + case media, ok := <-video: if !ok { video = nil continue } - if err := s.sendMedia(ChannelVideo, payload); err != nil { + if err := s.forwardMedia(ChannelVideo, media); err != nil { s.result <- err return } - case payload, ok := <-audio: + case media, ok := <-audio: if !ok { audio = nil continue } - if err := s.sendMedia(ChannelAudio, payload); err != nil { + if err := s.forwardMedia(ChannelAudio, media); err != nil { s.result <- err return } @@ -627,12 +664,31 @@ func (s *gatewaySession) mediaLoop() { s.result <- ErrProviderDisconnected } -func (s *gatewaySession) sendMedia(channel byte, payload []byte) error { +func (s *gatewaySession) forwardMedia(channel byte, media ProviderMedia) error { + s.outputMu.Lock() + defer s.outputMu.Unlock() + state := s.provider.State().State + if s.mediaQuiesced || state == ProviderStateTerminated || state == ProviderStateDisconnected { + s.mediaQuiesced = true + return nil + } + return s.sendMedia(channel, media) +} + +func (s *gatewaySession) sendMedia(channel byte, media ProviderMedia) error { + dequeuedAt := time.Now() + if media.EnqueuedAt.IsZero() || media.EnqueuedAt.After(dequeuedAt) { + media.EnqueuedAt = dequeuedAt + } + if media.ReceivedAt.IsZero() || media.ReceivedAt.After(media.EnqueuedAt) { + media.ReceivedAt = media.EnqueuedAt + } processingStarted := time.Now() - frames, err := FragmentPayload(channel, s.sequence.Add(1), uint64(time.Now().UnixMilli()), payload) + frames, err := FragmentPayload(channel, s.sequence.Add(1), uint64(time.Now().UnixMilli()), media.Payload) if err != nil { return err } + var pacingDelay time.Duration for _, frame := range frames { encoded, err := EncodeFrame(frame) if err != nil { @@ -642,16 +698,18 @@ func (s *gatewaySession) sendMedia(channel byte, payload []byte) error { if err := s.server.pacer.wait(s.ctx, s.authority.SessionID, len(encoded)); err != nil { return err } - s.server.metrics.PacingDelayNanos.Add(uint64(time.Since(pacingStarted))) - s.server.metrics.QueueDelayNanos.Add(uint64(time.Since(pacingStarted))) + pacingDelay += time.Since(pacingStarted) if err := s.connection.SendDatagram(encoded); err != nil { return err } s.server.metrics.MediaPackets.Add(1) s.server.metrics.MediaBytes.Add(uint64(len(encoded))) - s.server.metrics.ProcessingDelayNanos.Add(uint64(time.Since(processingStarted))) - s.server.metrics.ProcessingSamples.Add(1) } + processingDelay := media.EnqueuedAt.Sub(media.ReceivedAt) + time.Since(processingStarted) - pacingDelay + s.server.metrics.QueueDelayNanos.Add(uint64(dequeuedAt.Sub(media.EnqueuedAt))) + s.server.metrics.ProcessingDelayNanos.Add(uint64(max(processingDelay, 0))) + s.server.metrics.PacingDelayNanos.Add(uint64(pacingDelay)) + s.server.metrics.ProcessingSamples.Add(1) return nil } @@ -799,7 +857,9 @@ func (s *gatewaySession) cleanup() { } else if err := s.server.config.Admission.Release(cleanupCtx, s.authority); err != nil { s.server.metrics.ProviderErrors.Add(1) } - _ = s.connection.CloseWithError(applicationError, "session closed") + if !s.terminalSent.Load() { + _ = s.connection.CloseWithError(applicationError, "session closed") + } return } if errors.Is(s.endReason, ErrProviderDisconnected) { @@ -812,7 +872,9 @@ func (s *gatewaySession) cleanup() { if err := s.server.reportProviderState(cleanupCtx, state); err != nil { s.server.metrics.ProviderErrors.Add(1) } - _ = s.connection.CloseWithError(applicationError, "session closed") + if !s.terminalSent.Load() { + _ = s.connection.CloseWithError(applicationError, "session closed") + } }) } @@ -1001,7 +1063,11 @@ func (c *Client) ReceiveProviderEvent(ctx context.Context) (ProviderEvent, error if len(payload) > 1024 { return ProviderEvent{}, ErrProviderMalformed } - return DecodeProviderEvent(payload) + event, err := DecodeProviderEvent(payload) + if err == nil && (event.Kind == ProviderEventTerminated || event.Kind == ProviderEventDisconnected) { + _ = c.Close() + } + return event, err } func (c *Client) ReceiveClipboard(ctx context.Context) (protocol.GatewayClipboardText, error) { diff --git a/openspec/changes/phase3c-gateway-audit-remediation/design.md b/openspec/changes/phase3c-gateway-audit-remediation/design.md index 3f482e1..7b8210f 100644 --- a/openspec/changes/phase3c-gateway-audit-remediation/design.md +++ b/openspec/changes/phase3c-gateway-audit-remediation/design.md @@ -19,11 +19,13 @@ The implementation already contains a source-shaped Apollo fake, native recovery ## Decisions - Format ANNOUNCE from `ProviderStreamPolicy` using the pinned Moonlight common-c bitrate and codec attributes. H.264 and HEVC with audio enabled are supported; AV1 and audio disabled fail before management/network readiness. -- Treat the current exact `client_decode` string as the negotiated decode profile. The default advertises the bounded H.264+HEVC set, and provider work must be a member of it. -- Consume existing provider events in the gateway session loop. Termination and disconnect cancel forwarding, then reuse current cleanup/release/reporting machinery and its cleanup-pending result. +- Advertise ordered registered `hevc-opus` and `h264-opus` profiles and delegate policy-compatible selection to generated Protocol intersection behavior. There is no Data Plane capability grammar. +- Before `/applist` or `/launch`, validate the selected policy against source-backed Apollo `/serverinfo` codec flags and HEVC luma bounds plus reviewed source limits for dimensions, frame rate, bitrate, and audio where Apollo exposes no dynamic field. Reject rather than cap or downgrade. +- Quiesce provider media sockets and the bounded forwarding path before emitting an existing terminal or disconnected event. Reuse current cleanup/release/reporting machinery and its cleanup-pending result; final control delivery has no fixed drain delay. - On a full audio FEC map, evict the oldest block according to existing block ordering and increment existing drop telemetry. +- Carry provider receipt and queue-enqueue timestamps through the existing bounded media value. Queue residence, active processing, and scheduler pacing are sampled separately, once per complete provider media unit. - Sample existing process counters at heartbeat time; calculate rate from byte and monotonic-time deltas while leaving configured capacity in registration. -- Build qualification on the existing native/provider fixture and public QUIC client path. Production stage observations replace the standalone codec and arithmetic impairment simulator; short smoke gates freeze the wiring, while normative durations remain deferred. +- Build qualification on source-shaped pinned-mTLS Apollo management, encrypted RTSP, ENet, and provider UDP plus the public QUIC client path. Per-traversal stage deltas replace the standalone codec/parser and arithmetic impairment simulator; short smoke gates freeze the wiring, while normative durations remain deferred. - Preserve the production fair-pacer schedule across short host-timer overshoots so measured allocation can catch up within the already bounded provider queue instead of accumulating timer granularity as lost capacity. @@ -31,9 +33,10 @@ The implementation already contains a source-shaped Apollo fake, native recovery ## Risks / Trade-offs - [Apollo cannot represent disabled audio truthfully] → Reject it rather than silently streaming stereo. -- [Provider event races with media] → Cancel the session first and let bounded cleanup serialize final release/reporting. +- [Provider event races with queued or new media] → Quiesce ingestion and serialize forwarding with terminal event delivery before cleanup. - [Counter reset or zero elapsed time] → Emit zero measured rate and establish a new baseline. - [Corrected qualification is more expensive] → Run only short smoke tests until the immutable candidate is frozen. +- [Apollo exposes incomplete dynamic capability detail] → Use only source-backed fields and explicit reviewed bounds; never infer support by silent capping. - [Pacer catch-up can emit a short burst after timer overshoot] → Clamp schedule debt to five milliseconds in addition to the existing 16-packet provider queue. diff --git a/openspec/changes/phase3c-gateway-audit-remediation/proposal.md b/openspec/changes/phase3c-gateway-audit-remediation/proposal.md index 738f088..ee9c40e 100644 --- a/openspec/changes/phase3c-gateway-audit-remediation/proposal.md +++ b/openspec/changes/phase3c-gateway-audit-remediation/proposal.md @@ -4,10 +4,10 @@ Fresh audit evidence shows the gateway ignores the immutable launch policy, leav ## What Changes -- Apply the effective policy to Apollo ANNOUNCE and reject unsupported or downgraded profiles before readiness (P3C-009, P3C-016, P3C-038). -- Convert provider termination/disconnect into bounded tunnel and durable lifecycle transitions while preserving cleanup-pending semantics (P3C-018–021, P3C-027). +- Negotiate registered decode profiles through the shared Protocol intersection, apply the effective policy to Apollo ANNOUNCE, and reject client or provider/source mismatch before launch (P3C-009, P3C-016, P3C-038). +- Quiesce media immediately on provider termination/disconnect, deliver the final typed event reliably, and preserve bounded tunnel and durable lifecycle transitions including cleanup-pending (P3C-018–021, P3C-027). - Evict bounded stale audio FEC blocks so newer recoverable media continues (P3C-001, P3C-026). -- Derive heartbeat egress and required low-cardinality telemetry from observed counters (P3C-022, P3C-028). +- Derive heartbeat egress and semantically separated queue, processing, and pacing observations from the production path (P3C-022, P3C-028). - Replace standalone processing/impairment simulation with a driver around the source-shaped provider, production queues/pacer/framing, QUIC, and an independent client (P3C-029–033, VER-008, VER-010). ## Capabilities diff --git a/openspec/changes/phase3c-gateway-audit-remediation/specs/apollo-stream-policy/spec.md b/openspec/changes/phase3c-gateway-audit-remediation/specs/apollo-stream-policy/spec.md index 3c646c0..2428ee1 100644 --- a/openspec/changes/phase3c-gateway-audit-remediation/specs/apollo-stream-policy/spec.md +++ b/openspec/changes/phase3c-gateway-audit-remediation/specs/apollo-stream-policy/spec.md @@ -8,8 +8,12 @@ The native Apollo backend SHALL derive ANNOUNCE resolution, frame rate, supporte - **THEN** the encrypted ANNOUNCE carries those settings and the source-backed HEVC and bitrate attributes ### Requirement: Provider policy cannot downgrade -The gateway MUST reject an invalid, unsupported, audio-disabled, AV1, or capability-mismatched Apollo policy before provider readiness because the current native path cannot truthfully honor those combinations. +The gateway MUST use the generated Protocol intersection to select only a registered profile compatible with the immutable policy. It MUST reject invalid, unsupported, no-overlap, downgrade, audio-disabled, AV1, or provider/source-mismatched Apollo policy before `/applist`, `/launch`, or provider readiness because the current native path cannot truthfully honor those combinations. #### Scenario: Unsupported policy fails closed -- **WHEN** authenticated provider work selects audio disabled, AV1, or a codec outside the negotiated client-decode profile -- **THEN** setup fails before readiness without falling back to H.264, stereo, or another local default +- **WHEN** authenticated provider work selects audio disabled, AV1, a codec outside the registered peer intersection, or a resolution, frame rate, bitrate, audio, or codec combination outside source-backed Apollo support +- **THEN** setup fails before application discovery or launch without falling back to H.264, stereo, a cap, or another local default + +#### Scenario: Independent peers negotiate one registered profile +- **WHEN** a production gateway and independent client advertise overlapping registered H.264 or HEVC profiles +- **THEN** admission selects the first policy-compatible common profile using shared Protocol behavior diff --git a/openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-heartbeat-telemetry/spec.md b/openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-heartbeat-telemetry/spec.md index 2cf0f8e..3db9023 100644 --- a/openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-heartbeat-telemetry/spec.md +++ b/openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-heartbeat-telemetry/spec.md @@ -13,3 +13,10 @@ The established authenticated path SHALL expose observed bytes, packets, drops, #### Scenario: Telemetry snapshot is published - **WHEN** the gateway emits a heartbeat after forwarding traffic - **THEN** it carries the bounded process-level observations and no high-cardinality or secret-bearing value + +### Requirement: Production delays have disjoint sample semantics +Queue delay SHALL measure provider-queue residence, processing delay SHALL measure provider recovery plus framing and QUIC handoff work excluding queue and pacing, and pacing delay SHALL measure scheduler waiting only. The gateway SHALL advance processing samples once per complete provider media unit even when it emits multiple Verse frames. + +#### Scenario: Known production waits +- **WHEN** one provider media unit has controlled enqueue, processing, and pacing intervals and fragments across multiple frames +- **THEN** each cumulative total reports only its intended interval and exactly one processing sample is retained through authenticated Server persistence diff --git a/openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-qualification/spec.md b/openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-qualification/spec.md index 1415eab..3139d1e 100644 --- a/openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-qualification/spec.md +++ b/openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-qualification/spec.md @@ -1,18 +1,18 @@ ## MODIFIED Requirements ### Requirement: Fixed media processing qualification -The qualification harness SHALL drive the source-shaped provider fixture through Apollo recovery/FEC, bounded production queues, the production fair pacer, Verse framing/QUIC, and an independent client for 1080p60 H.264 at 20 Mbps, 1440p120 HEVC at 50 Mbps, and 4K60 HEVC at 80 Mbps. After a recorded warm-up, the frozen candidate SHALL run each profile for ten wall-clock minutes, preserve encoded payload bytes, retain every monotonic processing sample plus bounded CPU, memory, goroutine, and allocation observations, and report count, min, median, p90, p95, p99, max, mean, standard deviation, timing overhead, and observed bitrate. Any payload mutation or p95 above 5 ms SHALL fail. +The qualification harness SHALL drive pinned-mTLS Apollo management, encrypted RTSP, ENet, and provider UDP through native source validation, `readUDPMedia`, recovery/FEC, bounded production queues, the production fair pacer, Verse framing/QUIC, and a public or independent client decoder for 1080p60 H.264 at 20 Mbps, 1440p120 HEVC at 50 Mbps, and 4K60 HEVC at 80 Mbps. After a recorded warm-up, the frozen candidate SHALL run each profile for ten wall-clock minutes, preserve encoded payload bytes, retain every monotonic processing sample plus bounded CPU, memory, goroutine, allocation, and provider-queue observations, and report count, min, median, p90, p95, p99, max, mean, standard deviation, timing overhead, and observed bitrate. Processing begins at complete provider-unit receipt and ends at QUIC handoff, excluding client transit. Any bypass, payload mutation, wall-duration violation, bitrate outside both lower and upper bounds, or p95 above 5 ms SHALL fail. #### Scenario: Healthy fixed profile - **WHEN** a frozen candidate runs one fixed profile for the normative duration - **THEN** the harness emits compressed raw path and resource samples plus a summary tied to the exact command, topology, source commit, immutable Protocol version, environment, and payload hash #### Scenario: Processing gate failure -- **WHEN** any production path stage is bypassed, payload integrity fails, or measured p95 exceeds 5 ms +- **WHEN** any production path stage lacks a per-traversal observation, payload integrity fails, duration or bitrate bounds fail, or measured p95 exceeds 5 ms - **THEN** the qualification command exits unsuccessfully without recording a passing candidate ### Requirement: Bounded impairment qualification -The harness SHALL run exactly the baseline, latency, jitter, loss, reorder, and constrained Section 7.2 profiles once against traffic traversing the production gateway path. Baseline SHALL cover all three media profiles and the other profiles SHALL cover 1080p60. Each artifact SHALL retain raw impairment observations and record tool version, exact command/configuration, environment, candidate commit, immutable Protocol version, direction, queue discipline, topology, fixed seed, and observed RTT, jitter, loss, reorder, throughput, drops, and capacity-step statistics. +The harness SHALL run exactly the baseline, latency, jitter, loss, reorder, and constrained Section 7.2 profiles once by applying impairment at the source-shaped provider UDP boundary while traffic traverses the production gateway path. Baseline SHALL cover all three media profiles and the other profiles SHALL cover 1080p60. Each artifact SHALL retain raw impairment and queue observations and record tool version, exact command/configuration, environment, candidate commit, immutable Protocol version, direction, queue discipline, topology, fixed seed, and observed RTT, jitter, loss, reorder, throughput, drops, and capacity-step statistics. #### Scenario: Complete six-profile run - **WHEN** the frozen candidate runs impairment qualification diff --git a/openspec/changes/phase3c-gateway-audit-remediation/specs/provider-session-lifecycle/spec.md b/openspec/changes/phase3c-gateway-audit-remediation/specs/provider-session-lifecycle/spec.md index 5114132..8314c8e 100644 --- a/openspec/changes/phase3c-gateway-audit-remediation/specs/provider-session-lifecycle/spec.md +++ b/openspec/changes/phase3c-gateway-audit-remediation/specs/provider-session-lifecycle/spec.md @@ -1,11 +1,11 @@ ## ADDED Requirements ### Requirement: Provider terminal events end forwarding -Encrypted provider termination and unexpected provider disconnect SHALL stop media forwarding, close the Verse tunnel within a bounded interval, release the session reservation, and report the appropriate durable provider/session state. +Encrypted provider termination and unexpected provider disconnect SHALL quiesce provider ingestion and queued/new media forwarding before the existing reliable typed terminal event is delivered, close the Verse tunnel within a bounded interval, release the session reservation, and report the appropriate durable provider/session state. A fixed drain delay MUST NOT stand in for reliable control delivery. #### Scenario: Host termination closes the tunnel - **WHEN** the native provider emits an authenticated termination event -- **THEN** no later provider media is delivered and the client tunnel, reservation, and durable lifecycle transition complete +- **THEN** queued and newly injected media cannot cross the Verse transport after observation, and the client tunnel, reservation, and durable lifecycle transition complete #### Scenario: Unexpected provider disconnect is reconnectable - **WHEN** required provider transport disconnects without acknowledged termination diff --git a/openspec/changes/phase3c-gateway-audit-remediation/tasks.md b/openspec/changes/phase3c-gateway-audit-remediation/tasks.md index 8738dd4..3448ffc 100644 --- a/openspec/changes/phase3c-gateway-audit-remediation/tasks.md +++ b/openspec/changes/phase3c-gateway-audit-remediation/tasks.md @@ -1,20 +1,21 @@ ## 1. Native Policy and Media - [x] 1.1 Drive supported immutable policy values into encrypted Apollo ANNOUNCE -- [x] 1.2 Reject unsupported and capability-mismatched policy before readiness -- [x] 1.3 Evict oldest incomplete audio FEC state and pass sustained-loss relay regression +- [x] 1.2 Negotiate registered profiles through shared Protocol intersection behavior +- [x] 1.3 Reject unsupported client or provider/source policy before `/applist`, `/launch`, or readiness +- [x] 1.4 Evict oldest incomplete audio FEC state and pass sustained-loss relay regression ## 2. Lifecycle and Telemetry -- [x] 2.1 Close forwarding and report durable state on provider termination and disconnect +- [x] 2.1 Quiesce queued and new media before terminal event delivery and report durable state - [x] 2.2 Preserve cleanup-pending on terminal cleanup failure -- [x] 2.3 Report measured heartbeat egress and required bounded telemetry +- [x] 2.3 Report measured heartbeat egress and separately sampled queue, processing, and pacing telemetry ## 3. Qualification Path -- [x] 3.1 Add a red-to-green end-to-end production-path smoke gate -- [x] 3.2 Remove duplicate processing and impairment simulation -- [x] 3.3 Retain raw processing, impairment, fairness, cap, convergence, and resource observations +- [x] 3.1 Add a red-to-green native UDP-to-public-client production-path smoke gate with per-traversal stage evidence +- [x] 3.2 Remove direct native internals, private QUIC/parser, and arithmetic impairment shortcuts +- [x] 3.3 Retain raw processing, impairment, fairness, cap, convergence, queue, and resource observations - [x] 3.4 Pass short fixed-profile, impairment, fairness, race, parser fuzz, and resource smoke checks ## 4. Immutable Freeze