From d3852d15f327b6d053089520417231ae2e210c5b Mon Sep 17 00:00:00 2001 From: sechmachine <97589681+sechmachine727@users.noreply.github.com> Date: Thu, 30 Jul 2026 01:46:00 +0700 Subject: [PATCH] fix(gateway): close Phase 3C audit gaps --- cmd/verse-gateway/main.go | 75 +- cmd/verse-gateway/main_test.go | 116 +++ gateway/apollo_audio_fec.go | 58 +- gateway/apollo_native.go | 36 +- gateway/apollo_native_test.go | 60 +- gateway/apollo_rtsp_handshake.go | 43 +- gateway/capability.go | 4 +- gateway/fair_pacer_test.go | 11 + gateway/gateway_test.go | 228 ++++- gateway/provider.go | 9 +- gateway/qualification_contract_test.go | 78 +- gateway/qualification_harness_test.go | 863 +++++++++++++++--- gateway/telemetry.go | 10 +- gateway/transport.go | 53 +- .../.openspec.yaml | 2 + .../design.md | 47 + .../proposal.md | 28 + .../specs/apollo-stream-policy/spec.md | 15 + .../specs/audio-fec-resilience/spec.md | 8 + .../specs/gateway-heartbeat-telemetry/spec.md | 15 + .../specs/gateway-qualification/spec.md | 37 + .../specs/provider-session-lifecycle/spec.md | 19 + .../tasks.md | 23 + 23 files changed, 1619 insertions(+), 219 deletions(-) create mode 100644 cmd/verse-gateway/main_test.go create mode 100644 openspec/changes/phase3c-gateway-audit-remediation/.openspec.yaml create mode 100644 openspec/changes/phase3c-gateway-audit-remediation/design.md create mode 100644 openspec/changes/phase3c-gateway-audit-remediation/proposal.md create mode 100644 openspec/changes/phase3c-gateway-audit-remediation/specs/apollo-stream-policy/spec.md create mode 100644 openspec/changes/phase3c-gateway-audit-remediation/specs/audio-fec-resilience/spec.md create mode 100644 openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-heartbeat-telemetry/spec.md create mode 100644 openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-qualification/spec.md create mode 100644 openspec/changes/phase3c-gateway-audit-remediation/specs/provider-session-lifecycle/spec.md create mode 100644 openspec/changes/phase3c-gateway-audit-remediation/tasks.md diff --git a/cmd/verse-gateway/main.go b/cmd/verse-gateway/main.go index 8eab735..8c8505b 100644 --- a/cmd/verse-gateway/main.go +++ b/cmd/verse-gateway/main.go @@ -90,6 +90,8 @@ func heartbeatLoop(ctx context.Context, client *gateway.ControlPlaneClient, serv ticker := time.NewTicker(2 * time.Second) defer ticker.Stop() var sequence int64 + var sampler heartbeatSampler + _, _ = sampler.sample(time.Now(), server.Metrics()) for { select { case <-ctx.Done(): @@ -104,11 +106,82 @@ func heartbeatLoop(ctx context.Context, client *gateway.ControlPlaneClient, serv state = "draining" } metrics := server.Metrics() - _ = client.Heartbeat(ctx, protocol.GatewayHeartbeat{Version: "1", GatewayID: registration.GatewayID, Sequence: sequence, ObservedAt: time.Now().UTC().Format(time.RFC3339Nano), ActiveConnections: metrics.ActiveSessions, EgressKbps: registration.BandwidthCapacityKbps, State: state}) + observedAt := time.Now() + egressKbps, telemetry := sampler.sample(observedAt, metrics) + _ = client.Heartbeat(ctx, protocol.GatewayHeartbeat{ + Version: "1", GatewayID: registration.GatewayID, Sequence: sequence, + ObservedAt: observedAt.UTC().Format(time.RFC3339Nano), ActiveConnections: metrics.ActiveSessions, + EgressKbps: egressKbps, State: state, Telemetry: telemetry, + }) } } } +type heartbeatSampler struct { + observedAt time.Time + mediaBytes uint64 +} + +func (s *heartbeatSampler) sample(observedAt time.Time, metrics gateway.MetricsSnapshot) (int64, protocol.GatewayTelemetry) { + egressKbps := int64(0) + elapsedMillis := observedAt.Sub(s.observedAt).Milliseconds() + if !s.observedAt.IsZero() && elapsedMillis > 0 && metrics.MediaBytes >= s.mediaBytes { + delta := metrics.MediaBytes - s.mediaBytes + milliseconds := uint64(elapsedMillis) + whole, remainder := delta/milliseconds, delta%milliseconds + if whole > 125_000_000 { + egressKbps = 1_000_000_000 + } else { + rate := whole*8 + remainder*8/milliseconds + if rate > 1_000_000_000 { + rate = 1_000_000_000 + } + egressKbps = int64(rate) + } + } + s.observedAt, s.mediaBytes = observedAt, metrics.MediaBytes + return egressKbps, protocol.GatewayTelemetry{ + AdmittedSessions: boundedMetric(metrics.AdmittedSessions), AdmissionRejects: boundedMetric(metrics.AdmissionRejects), + Reconnects: boundedMetric(metrics.Reconnects), DrainTransitions: boundedMetric(metrics.DrainTransitions), + MediaDrops: boundedMetric(metrics.MediaDrops), MediaPackets: boundedMetric(metrics.MediaPackets), + MediaBytes: boundedMetric(metrics.MediaBytes), QueueDelayMicros: boundedMetric(metrics.QueueDelayNanos / 1000), + ProcessingDelayMicros: boundedMetric(metrics.ProcessingDelayNanos / 1000), ProcessingSamples: boundedMetric(metrics.ProcessingSamples), + PacingDelayMicros: boundedMetric(metrics.PacingDelayNanos / 1000), ProviderErrors: boundedMetric(metrics.ProviderErrors), + InputRejected: boundedMetric(metrics.InputRejected), ControlRttMicros: boundedMetric(metrics.ControlRTTNanos / 1000), + ControlJitterMicros: boundedMetric(metrics.ControlJitterNanos / 1000), ControlLossPpm: boundedMetric(metrics.ControlLossPPM), + PendingReliable: boundedMetric(metrics.PendingReliable), ProviderState: providerStateName(metrics.ProviderState), + } +} + +func boundedMetric(value uint64) int64 { + const maximum = uint64(^uint64(0) >> 1) + if value > maximum { + return int64(maximum) + } + return int64(value) +} + +func providerStateName(value uint64) string { + switch value { + case 1: + return gateway.ProviderStateStarting + case 2: + return gateway.ProviderStateReady + case 3: + return gateway.ProviderStateDisconnected + case 4: + return gateway.ProviderStateTerminating + case 5: + return gateway.ProviderStateTerminated + case 6: + return gateway.ProviderStateCleanup + case 7: + return gateway.ProviderStateFailed + default: + return "unknown" + } +} + func loadTLS(certFile, keyFile, clientCAFile string) (*tls.Config, *tls.Config, error) { certificate, err := tls.LoadX509KeyPair(certFile, keyFile) if err != nil { diff --git a/cmd/verse-gateway/main_test.go b/cmd/verse-gateway/main_test.go new file mode 100644 index 0000000..4715dc7 --- /dev/null +++ b/cmd/verse-gateway/main_test.go @@ -0,0 +1,116 @@ +package main + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "io" + "math/big" + "net/http" + "net/http/httptest" + "testing" + "time" + + "git.sechmachine.io.vn/sechmachine/VerseVDI-Data-Plane/gateway" + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" +) + +func TestHeartbeatReportsMeasuredEgressInsteadOfConfiguredCapacity(t *testing.T) { + heartbeats := make(chan protocol.GatewayHeartbeat, 1) + control := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if request.URL.Path == "/api/v1/gateway/heartbeat" { + heartbeat, err := protocol.DecodeGatewayHeartbeat(mustReadBody(t, request)) + if err != nil { + t.Errorf("decode heartbeat: %v", err) + } else { + heartbeats <- heartbeat + } + } + response.WriteHeader(http.StatusNoContent) + })) + defer control.Close() + + server, err := gateway.NewServer(gateway.ServerConfig{ + ListenAddress: "127.0.0.1:0", TLSConfig: heartbeatTestTLS(t), GatewayID: "gateway-1", + Admission: gateway.AdmissionFunc(func(context.Context, protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) { + return protocol.SessionAuthority{}, gateway.ErrAdmissionRejected + }), + Provider: gateway.NewFakeApollo(gateway.FakeApolloConfig{}), + }) + if err != nil { + t.Fatal(err) + } + defer server.Close() + registration := protocol.GatewayRegistration{GatewayID: "gateway-1", BandwidthCapacityKbps: 100000} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go heartbeatLoop(ctx, gateway.NewControlPlaneClient(control.URL, control.Client()), server, registration) + + select { + case heartbeat := <-heartbeats: + cancel() + if heartbeat.EgressKbps != 0 { + t.Fatalf("idle measured egress = %d Kbps, want 0; configured capacity is not traffic", heartbeat.EgressKbps) + } + case <-time.After(3 * time.Second): + t.Fatal("heartbeat was not published") + } +} + +func TestHeartbeatSamplerUsesByteDeltaAndMonotonicElapsed(t *testing.T) { + var sampler heartbeatSampler + start := time.Now() + if egress, _ := sampler.sample(start, gateway.MetricsSnapshot{MediaBytes: 1000}); egress != 0 { + t.Fatalf("first sample egress = %d, want baseline 0", egress) + } + egress, telemetry := sampler.sample(start.Add(2*time.Second), gateway.MetricsSnapshot{ + AdmittedSessions: 2, AdmissionRejects: 3, Reconnects: 4, DrainTransitions: 5, + MediaDrops: 6, MediaPackets: 7, MediaBytes: 17000, QueueDelayNanos: 9000, + ProcessingDelayNanos: 10000, ProcessingSamples: 11, PacingDelayNanos: 12000, + ProviderErrors: 13, InputRejected: 14, ControlRTTNanos: 15000, + ControlJitterNanos: 16000, ControlLossPPM: 17, PendingReliable: 18, ProviderState: 2, + }) + if egress != 64 || telemetry.MediaBytes != 17000 || telemetry.MediaPackets != 7 || + telemetry.QueueDelayMicros != 9 || telemetry.ProviderState != gateway.ProviderStateReady { + t.Fatalf("sample = egress:%d telemetry:%#v", egress, telemetry) + } +} + +func mustReadBody(t *testing.T, request *http.Request) []byte { + t.Helper() + defer request.Body.Close() + data, err := io.ReadAll(request.Body) + if err != nil { + t.Fatal(err) + } + return data +} + +func heartbeatTestTLS(t *testing.T) *tls.Config { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + template := &x509.Certificate{ + SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "gateway.test"}, + NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(time.Hour), + IsCA: true, BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + certificate := tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key} + pool := x509.NewCertPool() + pool.AddCert(template) + return &tls.Config{ + MinVersion: tls.VersionTLS13, Certificates: []tls.Certificate{certificate}, + ClientAuth: tls.RequireAndVerifyClientCert, ClientCAs: pool, + } +} diff --git a/gateway/apollo_audio_fec.go b/gateway/apollo_audio_fec.go index 7f33493..3e0d0e4 100644 --- a/gateway/apollo_audio_fec.go +++ b/gateway/apollo_audio_fec.go @@ -22,21 +22,46 @@ type apolloAudioAssembler struct { blocks map[uint16]*apolloAudioFECBlock } -func (a *apolloAudioAssembler) Add(codec *apolloMediaCodec, shard apolloAudioShard) ([][]byte, error) { +func (a *apolloAudioAssembler) Add(codec *apolloMediaCodec, shard apolloAudioShard) ([][]byte, bool, error) { if codec == nil || len(shard.payload) == 0 || len(shard.payload) > 1408 || len(shard.payload)%16 != 0 { - return nil, errApolloMedia + return nil, false, errApolloMedia } if a.blocks == nil { a.blocks = make(map[uint16]*apolloAudioFECBlock) } base := shard.base if base&3 != 0 { - return nil, errApolloMedia + return nil, false, errApolloMedia } + index := 0 + if shard.parity { + if shard.parityIndex >= apolloAudioParityShards { + return nil, false, errApolloMedia + } + index = apolloAudioDataShards + int(shard.parityIndex) + } else { + index = int(uint16(shard.sequence - base)) + if index >= apolloAudioDataShards { + return nil, false, errApolloMedia + } + } + evicted := false block := a.blocks[base] if block == nil { if len(a.blocks) >= apolloAudioMaximumBlocks { - return nil, errApolloMedia + var oldest uint16 + var maximumAge uint16 + for candidate := range a.blocks { + age := base - candidate + if age > maximumAge && age < 1<<15 { + oldest, maximumAge = candidate, age + } + } + if maximumAge == 0 { + return nil, false, errApolloMedia + } + delete(a.blocks, oldest) + evicted = true } block = &apolloAudioFECBlock{base: base} a.blocks[base] = block @@ -44,49 +69,40 @@ func (a *apolloAudioAssembler) Add(codec *apolloMediaCodec, shard apolloAudioSha if block.size == 0 { block.size = len(shard.payload) } else if block.size != len(shard.payload) { - return nil, errApolloMedia + return nil, evicted, errApolloMedia } - index := 0 if shard.parity { - if shard.parityIndex >= apolloAudioParityShards { - return nil, errApolloMedia - } - index = apolloAudioDataShards + int(shard.parityIndex) if block.haveFEC && (block.timestamp != shard.timestamp || block.ssrc != shard.ssrc) { - return nil, errApolloMedia + return nil, evicted, errApolloMedia } block.timestamp, block.ssrc, block.haveFEC = shard.timestamp, shard.ssrc, true } else { - index = int(uint16(shard.sequence - base)) - if index >= apolloAudioDataShards { - return nil, errApolloMedia - } if block.haveFEC && (shard.timestamp != block.timestamp+uint32(index*5) || shard.ssrc != block.ssrc) { - return nil, errApolloMedia + return nil, evicted, errApolloMedia } } if block.received[index] { - return nil, errApolloMedia + return nil, evicted, errApolloMedia } block.shards[index] = append([]byte(nil), shard.payload...) block.received[index] = true block.count++ if block.count < apolloAudioDataShards { - return nil, nil + return nil, evicted, nil } if err := reconstructApolloAudioBlock(block); err != nil { - return nil, err + return nil, evicted, err } output := make([][]byte, apolloAudioDataShards) for index := range output { payload, err := codec.openApolloAudioCipher(base+uint16(index), block.shards[index]) if err != nil { - return nil, err + return nil, evicted, err } output[index] = payload } delete(a.blocks, base) - return output, nil + return output, evicted, nil } func reconstructApolloAudioBlock(block *apolloAudioFECBlock) error { diff --git a/gateway/apollo_native.go b/gateway/apollo_native.go index dab57ae..c567dcc 100644 --- a/gateway/apollo_native.go +++ b/gateway/apollo_native.go @@ -43,7 +43,8 @@ func NewNativeApolloBackend() *NativeApolloBackend { func (b *NativeApolloBackend) Management(ctx context.Context, request LaunchRequest) ([]byte, error) { work := request.ProviderWork - if err := work.Validate(); err != nil || request.SessionID == "" || request.SessionID != work.SessionID || work.ProviderProfile != ProviderProfileApollo { + if err := work.Validate(); err != nil || validateApolloStreamPolicy(work.StreamPolicy) != nil || + request.SessionID == "" || request.SessionID != work.SessionID || work.ProviderProfile != ProviderProfileApollo { return nil, ErrProviderMalformed } client, err := newPinnedApolloHTTPClient(work) @@ -199,7 +200,8 @@ func pinnedApolloTLSConfig(work protocol.ProviderSessionWork) (*tls.Config, erro func (b *NativeApolloBackend) Setup(ctx context.Context, request LaunchRequest) ([]byte, error) { work := request.ProviderWork - if err := work.Validate(); err != nil || request.SessionID == "" || request.SessionID != work.SessionID || work.ProviderProfile != ProviderProfileApollo { + if err := work.Validate(); err != nil || validateApolloStreamPolicy(work.StreamPolicy) != nil || + request.SessionID == "" || request.SessionID != work.SessionID || work.ProviderProfile != ProviderProfileApollo { return nil, ErrProviderMalformed } client, err := newPinnedApolloHTTPClient(work) @@ -287,6 +289,7 @@ type nativeApolloSession struct { state protocol.ProviderState pressed map[string]InputEvent closeOnce sync.Once + disconnectOnce sync.Once channelsOnce sync.Once done chan struct{} readDone chan struct{} @@ -493,6 +496,9 @@ func (s *nativeApolloSession) ReleaseAll(ctx context.Context) error { func (s *nativeApolloSession) Terminate(ctx context.Context) error { var cleanupErr error + s.mu.Lock() + disconnected := s.state.State == ProviderStateDisconnected + s.mu.Unlock() s.closeOnce.Do(func() { if err := s.ReleaseAll(ctx); err != nil { cleanupErr = err @@ -518,7 +524,7 @@ func (s *nativeApolloSession) Terminate(ctx context.Context) error { } if cleanupErr == nil { s.closeMediaChannels() - if s.allowApplicationTermination { + if s.allowApplicationTermination && !disconnected { if err := apolloCancelRequest(ctx, s.managementClient, s.managementHost, s.managementPort); err != nil { cleanupErr = err } @@ -536,6 +542,8 @@ func (s *nativeApolloSession) Terminate(ctx context.Context) error { if cleanupErr != nil { s.state.State = ProviderStateCleanup s.state.CleanupPending = true + } else if disconnected { + s.state.State = ProviderStateDisconnected } else { s.state.State = ProviderStateTerminated } @@ -678,11 +686,19 @@ func (s *nativeApolloSession) handleApolloDisconnect(err error) { if err == nil { return } - s.mu.Lock() - if s.state.State != ProviderStateTerminated { + s.disconnectOnce.Do(func() { + s.mu.Lock() + if s.state.State == ProviderStateTerminated { + s.mu.Unlock() + return + } s.state.State = ProviderStateDisconnected - } - s.mu.Unlock() + s.mu.Unlock() + select { + case s.events <- ProviderEvent{Kind: ProviderEventDisconnected}: + default: + } + }) } func (s *nativeApolloSession) closeMediaChannels() { @@ -738,7 +754,11 @@ func (s *nativeApolloSession) readUDPMedia() { if openErr != nil { continue } - payloads, err = s.audioFEC.Add(s.media, shard) + var evicted bool + payloads, evicted, err = s.audioFEC.Add(s.media, shard) + if evicted { + s.mediaDrops.Add(1) + } } if err != nil { continue diff --git a/gateway/apollo_native_test.go b/gateway/apollo_native_test.go index 3c00824..53c0e65 100644 --- a/gateway/apollo_native_test.go +++ b/gateway/apollo_native_test.go @@ -55,6 +55,7 @@ func TestNativeApolloManagementUsesSessionScopedMTLS(t *testing.T) { Version: "1", SessionID: "session-1", GatewayID: "gateway-1", ReconnectSequence: 0, ExpiresAt: "2099-01-01T00:00:00Z", ProviderProfile: ProviderProfileApollo, ProviderIdentity: "apollo-server#sha256:" + hex.EncodeToString(pinned[:]), PolicyVersionID: "policy-1", ApplicationID: "1", ClientID: "paired-client", + StreamPolicy: protocol.ProviderStreamPolicy{ResolutionWidth: 1920, ResolutionHeight: 1080, Fps: 60, Codec: "H264", BitrateKbps: 8000, AudioEnabled: true}, ManagementHost: host, ManagementPort: port, StreamHost: host, StreamPort: 47984, ClientCertificatePem: certificatePEM(t, clientTLS.Certificates[0]), ClientPrivateKeyPem: privateKeyPEM(t, clientTLS.Certificates[0]), @@ -70,6 +71,32 @@ func TestNativeApolloManagementUsesSessionScopedMTLS(t *testing.T) { } } +func TestNativeApolloSetupRejectsUnsupportedStreamPolicyBeforeProviderReadiness(t *testing.T) { + work := protocol.ProviderSessionWork{ + Version: "1", SessionID: "session-1", GatewayID: "gateway-1", + ExpiresAt: "2099-01-01T00:00:00Z", ProviderProfile: ProviderProfileApollo, + ProviderIdentity: "provider#sha256:00", PolicyVersionID: "policy-1", + ApplicationID: "1", ClientID: "client-1", ManagementHost: "127.0.0.1", ManagementPort: 1, + StreamHost: "127.0.0.1", StreamPort: 1, ClientCertificatePem: "invalid", + ClientPrivateKeyPem: "invalid", ServerCertificatePem: "invalid", + 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}, + } { + t.Run(name, func(t *testing.T) { + work.StreamPolicy = policy + _, err := NewNativeApolloBackend().Setup(context.Background(), LaunchRequest{ + SessionID: "session-1", ProviderProfile: ProviderProfileApollo, ProviderWork: work, + }) + if !errors.Is(err, ErrProviderMalformed) { + t.Fatalf("Setup() error = %v, want ErrProviderMalformed before provider readiness", err) + } + }) + } +} + func TestNativeApolloSetupRequiresModernEncryptedRTSPOrder(t *testing.T) { serverTLS, clientTLS := testTLS(t) streamListener, err := net.Listen("tcp", "127.0.0.1:0") @@ -158,8 +185,9 @@ func TestNativeApolloSetupRequiresModernEncryptedRTSPOrder(t *testing.T) { } if method == "ANNOUNCE" { for _, required := range []string{ - "a=x-nv-video[0].clientViewportWd:1920", "a=x-nv-video[0].clientViewportHt:1080", "a=x-nv-video[0].maxFPS:60", - "a=x-nv-video[0].packetSize:1024", "a=x-nv-vqos[0].bw.maximumBitrateKbps:8000", "a=x-nv-audio.surround.numChannels:2", + "a=x-nv-video[0].clientViewportWd:2560", "a=x-nv-video[0].clientViewportHt:1440", "a=x-nv-video[0].maxFPS:120", + "a=x-nv-video[0].packetSize:1024", "a=x-nv-clientSupportHevc:1", "a=x-nv-vqos[0].bitStreamFormat:1", + "a=x-nv-vqos[0].bw.maximumBitrateKbps:32000", "a=x-ml-video.configuredBitrateKbps:40000", "a=x-nv-audio.surround.numChannels:2", "a=x-nv-general.useReliableUdp:13", "a=x-ss-general.encryptionEnabled:7", } { if !strings.Contains(string(plaintext), required+"\r\n") { @@ -255,6 +283,7 @@ func TestNativeApolloSetupRequiresModernEncryptedRTSPOrder(t *testing.T) { Version: "1", SessionID: "session-1", GatewayID: "gateway-1", ReconnectSequence: 0, 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: 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]}}), @@ -625,6 +654,33 @@ func TestNativeApolloSessionRelaysOnlyAuthenticatedEncodedUDPMedia(t *testing.T) t.Fatal("encrypted FEC audio was not recovered") } } + for block := 0; block < apolloAudioMaximumBlocks+1; block++ { + sequence := uint16(100 + block*apolloAudioDataShards) + packet := sourceShapedEncryptedAudioPacketWithHeaders(t, key, keyID, sequence, uint32(sequence)*5, 1, []byte{byte(block)}) + if _, err := audioServer.WriteToUDP(packet, audioClient.LocalAddr().(*net.UDPAddr)); err != nil { + t.Fatal(err) + } + } + for index, want := range [][]byte{{0xa0}, {0xa1}, {0xa2}, {0xa3}} { + sequence := uint16(124 + index) + packet := sourceShapedEncryptedAudioPacketWithHeaders(t, key, keyID, sequence, uint32(sequence)*5, 1, want) + if _, err := audioServer.WriteToUDP(packet, audioClient.LocalAddr().(*net.UDPAddr)); err != nil { + t.Fatal(err) + } + } + for _, want := range [][]byte{{0xa0}, {0xa1}, {0xa2}, {0xa3}} { + select { + case payload := <-session.Audio(): + if string(payload) != string(want) { + t.Fatalf("post-loss audio relay = %x, want %x", payload, want) + } + case <-time.After(time.Second): + t.Fatal("sustained loss permanently stalled newer audio") + } + } + if drops := session.Telemetry().MediaDrops; drops < 2 { + t.Fatalf("stale FEC eviction drops = %d, want at least 2", drops) + } terminateCtx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() if err := session.Terminate(terminateCtx); err != nil { diff --git a/gateway/apollo_rtsp_handshake.go b/gateway/apollo_rtsp_handshake.go index b820914..e71f079 100644 --- a/gateway/apollo_rtsp_handshake.go +++ b/gateway/apollo_rtsp_handshake.go @@ -112,7 +112,10 @@ func (b *NativeApolloBackend) performRTSPHandshake(ctx context.Context, work pro if err != nil { return nil, nil, err } - announceBody := apolloAnnounceProfile() + announceBody, err := apolloAnnounceProfile(work.StreamPolicy) + if err != nil { + return nil, nil, err + } announce, err := request("ANNOUNCE", "streamid=control/13/0", sessionID, []apolloRTSPHeader{{"Content-Type", "application/sdp"}}, announceBody, 6) if err != nil { return nil, nil, err @@ -468,18 +471,33 @@ func apolloRTSPConnectData(message apolloRTSPMessage) (uint32, error) { return uint32(parsed), nil } -func apolloAnnounceProfile() []byte { +func apolloAnnounceProfile(policy protocol.ProviderStreamPolicy) ([]byte, error) { + if err := validateApolloStreamPolicy(policy); err != nil { + return nil, err + } + format, supportsHEVC := int64(0), int64(0) + if policy.Codec == "HEVC" { + format, supportsHEVC = 1, 1 + } + maximumBitrate := policy.BitrateKbps * 80 / 100 + if maximumBitrate > 100000 { + maximumBitrate = 100000 + } return []byte("v=0\r\n" + "o=android 0 0 IN IP4 0.0.0.0\r\n" + "s=NVIDIA Streaming Client\r\n" + - "a=x-nv-video[0].clientViewportWd:1920\r\n" + - "a=x-nv-video[0].clientViewportHt:1080\r\n" + - "a=x-nv-video[0].maxFPS:60\r\n" + + fmt.Sprintf("a=x-nv-video[0].clientViewportWd:%d\r\n", policy.ResolutionWidth) + + fmt.Sprintf("a=x-nv-video[0].clientViewportHt:%d\r\n", policy.ResolutionHeight) + + fmt.Sprintf("a=x-nv-video[0].maxFPS:%d\r\n", policy.Fps) + "a=x-nv-video[0].packetSize:1024\r\n" + "a=x-nv-video[0].videoEncoderSlicesPerFrame:1\r\n" + "a=x-nv-video[0].maxNumReferenceFrames:0\r\n" + - "a=x-nv-vqos[0].bitStreamFormat:0\r\n" + - "a=x-nv-vqos[0].bw.maximumBitrateKbps:8000\r\n" + + fmt.Sprintf("a=x-nv-clientSupportHevc:%d\r\n", supportsHEVC) + + fmt.Sprintf("a=x-nv-vqos[0].bitStreamFormat:%d\r\n", format) + + fmt.Sprintf("a=x-nv-video[0].initialBitrateKbps:%d\r\n", maximumBitrate) + + fmt.Sprintf("a=x-nv-video[0].initialPeakBitrateKbps:%d\r\n", maximumBitrate) + + fmt.Sprintf("a=x-nv-vqos[0].bw.minimumBitrateKbps:%d\r\n", maximumBitrate) + + fmt.Sprintf("a=x-nv-vqos[0].bw.maximumBitrateKbps:%d\r\n", maximumBitrate) + "a=x-nv-vqos[0].fec.minRequiredFecPackets:2\r\n" + "a=x-nv-vqos[0].qosTrafficType:5\r\n" + "a=x-nv-audio.surround.numChannels:2\r\n" + @@ -490,10 +508,17 @@ func apolloAnnounceProfile() []byte { "a=x-nv-general.useReliableUdp:13\r\n" + "a=x-nv-general.featureFlags:167\r\n" + "a=x-ml-general.featureFlags:0\r\n" + - "a=x-ml-video.configuredBitrateKbps:8000\r\n" + + fmt.Sprintf("a=x-ml-video.configuredBitrateKbps:%d\r\n", policy.BitrateKbps) + "a=x-ss-general.encryptionEnabled:7\r\n" + "a=x-ss-video[0].chromaSamplingType:0\r\n" + - "a=x-ss-video[0].intraRefresh:0\r\n") + "a=x-ss-video[0].intraRefresh:0\r\n"), nil +} + +func validateApolloStreamPolicy(policy protocol.ProviderStreamPolicy) error { + if err := policy.Validate(); err != nil || !policy.AudioEnabled || (policy.Codec != "H264" && policy.Codec != "HEVC") { + return ErrProviderMalformed + } + return nil } func validApolloRTSPToken(value string) bool { diff --git a/gateway/capability.go b/gateway/capability.go index 9122b9f..8f72040 100644 --- a/gateway/capability.go +++ b/gateway/capability.go @@ -8,6 +8,8 @@ import ( var ErrNoCapabilityOverlap = errors.New("no capability overlap") +const defaultClientDecode = "h264-hevc-opus" + func DefaultCapabilities() protocol.CapabilityProfile { return protocol.CapabilityProfile{ Transport: "quic-tls13", @@ -15,7 +17,7 @@ func DefaultCapabilities() protocol.CapabilityProfile { Media: "encoded", Audio: "encoded", SourceRateControl: "server", - ClientDecode: "h264-opus", + ClientDecode: defaultClientDecode, } } diff --git a/gateway/fair_pacer_test.go b/gateway/fair_pacer_test.go index c4a1409..bf5560a 100644 --- a/gateway/fair_pacer_test.go +++ b/gateway/fair_pacer_test.go @@ -32,6 +32,17 @@ func TestFairPacerEightFlowSharesAndCapacitySteps(t *testing.T) { assertSyntheticCap(t, half, 500_000) } +func TestFairPacerBoundsCatchupAfterHostStall(t *testing.T) { + start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) + pacer := newFairPacer(8000) + _ = pacer.reserveAt(start, "one", 1000) + resumed := start.Add(100 * time.Millisecond) + next := pacer.reserveAt(resumed, "one", 1000) + if next.Before(resumed.Add(-fairPacerMaximumCatchup)) || next.After(resumed.Add(10*time.Millisecond)) { + t.Fatalf("post-stall reservation = %s, want bounded catchup near %s", next, resumed) + } +} + func runSyntheticPacer(pacer *fairPacer, start, end time.Time, flows []string, next map[string]time.Time) []syntheticPacerDelivery { const packetBytes = 1000 for _, flow := range flows { diff --git a/gateway/gateway_test.go b/gateway/gateway_test.go index 82e428f..8b0db25 100644 --- a/gateway/gateway_test.go +++ b/gateway/gateway_test.go @@ -321,6 +321,56 @@ func TestAdmissionQUICMTLSRelayAndCleanup(t *testing.T) { } } +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" + 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), + Capabilities: capabilities, ProviderProfile: ProviderProfileApollo, ProviderIdentity: fake.config.Identity.Key(), + } + admission := &oneTimeAdmission{ + authority: authority, released: make(chan struct{}), + streamPolicy: protocol.ProviderStreamPolicy{ + ResolutionWidth: 2560, ResolutionHeight: 1440, Fps: 120, + Codec: "HEVC", BitrateKbps: 40000, AudioEnabled: true, + }, + } + server, err := NewServer(ServerConfig{ + ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: "gateway-1", + Capabilities: capabilities, ProviderCapabilities: capabilities, Admission: admission, Provider: fake, + }) + if err != nil { + 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{ + 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: capabilities, + } + if _, err := Dial(context.Background(), server.Addr().String(), clientTLS, request); err == nil { + t.Fatal("gateway accepted HEVC provider work for an H.264-only negotiated profile") + } + select { + case <-admission.released: + case <-time.After(time.Second): + t.Fatal("gateway did not release rejected provider work") + } + if session, _ := fake.LastSession().(*fakeSession); session != nil { + t.Fatal("provider started before policy/capability rejection") + } + _ = server.Close() + if err := <-serveDone; err != nil { + t.Fatal(err) + } +} + func TestRegisteredChannelFramesTraversePublicTransport(t *testing.T) { h := newGatewayTransportHarness(t) @@ -456,6 +506,116 @@ func TestProviderClipboardAuditWaitsForPublicTransportDelivery(t *testing.T) { } } +func TestProviderTerminationEndsPublicGatewaySession(t *testing.T) { + h := newGatewayTransportHarnessWithoutClipboard(t) + h.drainInitialMedia(t) + h.session.EmitEvent(ProviderEvent{Kind: ProviderEventTerminated, Payload: []byte{1, 2, 3, 4}}) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + event, err := h.client.ReceiveProviderEvent(ctx) + cancel() + if err != nil || event.Kind != ProviderEventTerminated { + t.Fatalf("provider termination event = %#v, %v", event, err) + } + h.waitReleased(t) + if states := h.reporter.States(); len(states) == 0 || states[len(states)-1].State != ProviderStateTerminated { + t.Fatalf("provider states = %#v", states) + } + h.assertMediaClosed(t) +} + +func TestEncryptedNativeHostTerminationEndsPublicGatewaySession(t *testing.T) { + serverTLS, clientTLS := testTLS(t) + key := []byte("0123456789abcdef") + native := newNativeApolloSession("session-native-terminal") + control, err := newApolloControlCodec(key) + if err != nil { + t.Fatal(err) + } + native.control = control + close(native.readDone) + session := nativeLifecycleSession{native} + provider := providerStartFunc(func(context.Context, LaunchRequest) (ProviderSession, error) { + if err := native.Ready(context.Background()); err != nil { + return nil, err + } + return session, nil + }) + authority := protocol.SessionAuthority{ + Version: "1", SessionID: native.sessionID, 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} + reporter := &recordingProviderStateReporter{} + server, err := NewServer(ServerConfig{ + ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: authority.GatewayID, + Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), + Admission: admission, ProviderStateReporter: reporter, Provider: provider, + }) + if err != nil { + 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{ + 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) + } + 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) + } + select { + case <-admission.released: + case <-time.After(2 * time.Second): + t.Fatal("native termination 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 TestProviderDisconnectEndsPublicGatewaySessionReconnectable(t *testing.T) { + h := newGatewayTransportHarnessWithoutClipboard(t) + h.drainInitialMedia(t) + h.session.Disconnect() + + 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) + } + h.assertMediaClosed(t) +} + +func TestProviderTerminalCleanupFailureReportsCleanupPending(t *testing.T) { + h := newGatewayTransportHarnessWithoutClipboard(t) + h.drainInitialMedia(t) + h.session.failure = FakeFailureTerminationTimeout + h.session.EmitEvent(ProviderEvent{Kind: ProviderEventTerminated, Payload: []byte{1, 2, 3, 4}}) + + h.waitReleased(t) + if states := h.reporter.States(); len(states) == 0 || states[len(states)-1].State != ProviderStateCleanup || !states[len(states)-1].CleanupPending { + t.Fatalf("provider states = %#v", states) + } + h.assertMediaClosed(t) +} + func testChannelFrame(flowID string, sequence int64, payload []byte) protocol.ChannelFrame { return protocol.ChannelFrame{Version: "1", FlowID: flowID, Sequence: sequence, Flags: 0, FragmentIndex: 0, FragmentCount: 1, TimestampMs: time.Now().UnixMilli(), Payload: base64.StdEncoding.EncodeToString(payload)} } @@ -468,12 +628,32 @@ type gatewayTransportHarness struct { server *Server } +type providerStartFunc func(context.Context, LaunchRequest) (ProviderSession, error) + +func (fn providerStartFunc) Start(ctx context.Context, request LaunchRequest) (ProviderSession, error) { + return fn(ctx, request) +} + +type nativeLifecycleSession struct{ *nativeApolloSession } + +func (s nativeLifecycleSession) Telemetry() ProviderTelemetry { + return ProviderTelemetry{State: s.State().State} +} + func newGatewayTransportHarness(t *testing.T) gatewayTransportHarness { + return newGatewayTransportHarnessWithClipboard(t, true) +} + +func newGatewayTransportHarnessWithoutClipboard(t *testing.T) gatewayTransportHarness { + return newGatewayTransportHarnessWithClipboard(t, false) +} + +func newGatewayTransportHarnessWithClipboard(t *testing.T, clipboardEnabled bool) gatewayTransportHarness { t.Helper() serverTLS, clientTLS := testTLS(t) fake := NewFakeApollo(FakeApolloConfig{Now: time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)}) authority := protocol.SessionAuthority{Version: "1", SessionID: "session-transport", GatewayID: "gateway-1", Audience: "versevdi-gateway", ReconnectSequence: 0, ExpiresAt: time.Now().Add(5 * time.Second).UTC().Format(time.RFC3339Nano), Capabilities: DefaultCapabilities(), ProviderProfile: ProviderProfileApollo, ProviderIdentity: fake.config.Identity.Key()} - admission := &oneTimeAdmission{authority: authority, released: make(chan struct{})} + admission := &oneTimeAdmission{authority: authority, released: make(chan struct{}), disableClipboard: !clipboardEnabled} reporter := &recordingProviderStateReporter{} server, err := NewServer(ServerConfig{ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: "gateway-1", Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), Admission: admission, ProviderStateReporter: reporter, ClipboardAuditReporter: reporter, Provider: fake}) if err != nil { @@ -513,6 +693,28 @@ func (h gatewayTransportHarness) waitReleased(t *testing.T) { } } +func (h gatewayTransportHarness) drainInitialMedia(t *testing.T) { + t.Helper() + for range 2 { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + _, err := h.client.ReceiveFrame(ctx) + cancel() + if err != nil { + t.Fatalf("drain initial media: %v", err) + } + } +} + +func (h gatewayTransportHarness) assertMediaClosed(t *testing.T) { + t.Helper() + h.session.EmitVideo([]byte("must-not-forward")) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if frame, err := h.client.ReceiveFrame(ctx); err == nil { + t.Fatalf("media remained open after provider terminal state: %#v", frame) + } +} + func testTLS(t *testing.T) (*tls.Config, *tls.Config) { t.Helper() caKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) @@ -545,10 +747,12 @@ func testTLS(t *testing.T) (*tls.Config, *tls.Config) { } type oneTimeAdmission struct { - used atomic.Bool - authority protocol.SessionAuthority - releases atomic.Int64 - released chan struct{} + used atomic.Bool + authority protocol.SessionAuthority + releases atomic.Int64 + released chan struct{} + streamPolicy protocol.ProviderStreamPolicy + disableClipboard bool } type recordingProviderStateReporter struct { @@ -594,14 +798,24 @@ func (a *oneTimeAdmission) ProviderWork(_ context.Context, authority protocol.Se if authority != a.authority { return protocol.ProviderSessionWork{}, ErrAdmissionRejected } + streamPolicy := a.streamPolicy + if streamPolicy == (protocol.ProviderStreamPolicy{}) { + streamPolicy = protocol.ProviderStreamPolicy{ResolutionWidth: 1920, ResolutionHeight: 1080, Fps: 60, Codec: "H264", BitrateKbps: 8000, AudioEnabled: true} + } + clipboardPolicy := protocol.ClipboardPolicy{MaxTextBytes: 65536, MaxUpdatesPerMinute: 30} + if !a.disableClipboard { + clipboardPolicy.ClientToProviderEnabled = true + clipboardPolicy.ProviderToClientEnabled = true + } return protocol.ProviderSessionWork{ Version: "1", SessionID: authority.SessionID, GatewayID: authority.GatewayID, ReconnectSequence: authority.ReconnectSequence, ExpiresAt: authority.ExpiresAt, ProviderProfile: ProviderProfileApollo, ProviderIdentity: authority.ProviderIdentity, PolicyVersionID: "policy-1", ApplicationID: "1", ClientID: "paired-client", ManagementHost: "apollo.test", ManagementPort: 47990, - StreamHost: "apollo.test", StreamPort: 47984, ClientCertificatePem: "certificate", + StreamPolicy: streamPolicy, + StreamHost: "apollo.test", StreamPort: 47984, ClientCertificatePem: "certificate", ClientPrivateKeyPem: "private-key", ServerCertificatePem: "server-certificate", - ClipboardPolicy: protocol.ClipboardPolicy{ClientToProviderEnabled: true, ProviderToClientEnabled: true, MaxTextBytes: 65536, MaxUpdatesPerMinute: 30}, + ClipboardPolicy: clipboardPolicy, }, nil } diff --git a/gateway/provider.go b/gateway/provider.go index cd2f3dd..fa88352 100644 --- a/gateway/provider.go +++ b/gateway/provider.go @@ -186,6 +186,7 @@ const ( ProviderEventTerminated ProviderEventKind = iota + 1 ProviderEventRumble ProviderEventHDR + ProviderEventDisconnected ) type ProviderEvent struct { @@ -563,12 +564,17 @@ func (s *fakeSession) Terminate(ctx context.Context) error { s.mu.Unlock() return nil } + disconnected := s.state.State == ProviderStateDisconnected s.state.State = ProviderStateTerminating s.closeOnce.Do(func() { close(s.video) close(s.audio) }) - s.state.State = ProviderStateTerminated + if disconnected { + s.state.State = ProviderStateDisconnected + } else { + s.state.State = ProviderStateTerminated + } s.mu.Unlock() return nil } @@ -587,6 +593,7 @@ func (s *fakeSession) Disconnect() { s.mu.Lock() s.state.State = ProviderStateDisconnected s.mu.Unlock() + s.EmitEvent(ProviderEvent{Kind: ProviderEventDisconnected}) } func (s *fakeSession) ReleaseCount() int { diff --git a/gateway/qualification_contract_test.go b/gateway/qualification_contract_test.go index 2760bcb..6241beb 100644 --- a/gateway/qualification_contract_test.go +++ b/gateway/qualification_contract_test.go @@ -54,6 +54,14 @@ func TestQualificationProtocolVersionIsExplicitAndImmutable(t *testing.T) { } } +func TestQualificationRecordsLinkedToolVersions(t *testing.T) { + versions, err := qualificationToolVersions() + if err != nil || versions["qualification"] != qualificationToolVersion || + versions["go"] == "" || versions["quic-go"] == "" { + t.Fatalf("qualification tool versions = %#v, %v", versions, err) + } +} + func TestQualificationOutputAndStatisticsFailClosed(t *testing.T) { if err := validateQualificationOutputDir("relative/evidence"); err == nil { t.Fatal("relative evidence directory was accepted") @@ -86,15 +94,16 @@ func TestQualificationOutputAndStatisticsFailClosed(t *testing.T) { func TestQualificationShortProcessingWritesRawArtifact(t *testing.T) { profile := qualificationMediaProfile{ - Name: "smoke", Codec: "h264", BitrateKbps: 1000, - Duration: 200 * time.Millisecond, Warmup: time.Millisecond, PacketBytes: 100, + Name: "smoke", Codec: "h264", BitrateKbps: 20000, + Duration: time.Second, Warmup: time.Millisecond, PacketBytes: 1000, } rawPath := filepath.Join(t.TempDir(), "processing.csv.gz") - summary, err := runQualificationProcessing(profile, rawPath) + summary, err := runQualificationProcessing(t, profile, rawPath) if err != nil { t.Fatal(err) } - if summary.Count < 1 || summary.RawSamplesSHA256 == "" || summary.RawSamplesBytes < 1 { + if summary.Count < 1 || summary.RawSamplesSHA256 == "" || summary.RawSamplesBytes < 1 || + summary.ResourceSamples < 2 || summary.RawResourcesSHA256 == "" || summary.RawResourcesBytes < 1 { t.Fatalf("processing summary = %#v", summary) } file, err := os.Open(rawPath) @@ -119,13 +128,14 @@ func TestQualificationShortProcessingWritesRawArtifact(t *testing.T) { } func TestQualificationProcessingPreservesPayload(t *testing.T) { - payload := qualificationPayload(qualificationMediaProfiles()[0]) - processed, elapsed, err := processQualificationPayload(7, payload) + profile := qualificationMediaProfiles()[0] + payload := qualificationPayload(profile) + trace, elapsed, err := newQualificationPath(t, profile.BitrateKbps).traverse(t, payload) if err != nil { t.Fatal(err) } - if !reflect.DeepEqual(processed, payload) { - t.Fatal("encoded payload mutated") + if !trace.PayloadPreserved || !trace.ApolloRecovered || !trace.VerseQUIC { + t.Fatalf("production path trace = %#v", trace) } if elapsed <= 0 { t.Fatalf("processing duration = %s", elapsed) @@ -134,34 +144,66 @@ func TestQualificationProcessingPreservesPayload(t *testing.T) { func TestQualificationImpairmentIsDeterministicAndBounded(t *testing.T) { profile := qualificationImpairmentProfiles()[3] - first, err := runQualificationImpairment(profile, qualificationMediaProfiles()[0], 10_000) + first, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 1000, filepath.Join(t.TempDir(), "first.csv.gz")) if err != nil { t.Fatal(err) } - second, err := runQualificationImpairment(profile, qualificationMediaProfiles()[0], 10_000) + second, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 1000, filepath.Join(t.TempDir(), "second.csv.gz")) if err != nil { t.Fatal(err) } - if !reflect.DeepEqual(first, second) { - t.Fatalf("impairment run is not deterministic:\n%#v\n%#v", first, second) + if first.Dropped != second.Dropped || first.InjectedReordered != second.InjectedReordered { + t.Fatalf("deterministic impairment selection differs:\n%#v\n%#v", first, second) } - if first.Sent != 10_000 || first.Delivered+first.Dropped != first.Sent || - first.ObservedLossPercent < 4.8 || first.ObservedLossPercent > 5.2 || - first.MaxQueuePackets > qualificationImpairmentQueuePackets { + if first.Sent != 1000 || first.Delivered+first.Dropped != first.Sent || + first.ObservedLossPercent < 3.5 || first.ObservedLossPercent > 6.5 || + first.MaxQueuePackets > qualificationImpairmentQueuePackets || first.RawSamplesSHA256 == "" { t.Fatalf("impairment observation = %#v", first) } - if _, err := runQualificationImpairment(profile, qualificationMediaProfiles()[0], qualificationImpairmentMaxPackets+1); err == nil { + if _, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], qualificationImpairmentMaxPackets+1, filepath.Join(t.TempDir(), "invalid.csv.gz")); err == nil { t.Fatal("unbounded impairment packet count was accepted") } + unknown := profile + unknown.Name = "private-simulator" + if _, err := runQualificationImpairment(t, unknown, qualificationMediaProfiles()[0], 1, filepath.Join(t.TempDir(), "unknown.csv.gz")); err == nil { + t.Fatal("unregistered impairment profile was accepted") + } +} + +func TestQualificationSixImpairmentProfilesTraverseProductionPath(t *testing.T) { + profiles := qualificationImpairmentProfiles() + if len(profiles) != 6 { + t.Fatalf("impairment profile count = %d, want exactly 6", len(profiles)) + } + for _, profile := range profiles { + observation, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 40, + filepath.Join(t.TempDir(), profile.Name+".csv.gz")) + if err != nil { + t.Fatalf("%s: %v", profile.Name, err) + } + if observation.Profile != profile.Name || observation.Delivered+observation.Dropped != 40 || + observation.RawSamplesSHA256 == "" || observation.MaxQueuePackets > qualificationImpairmentQueuePackets { + t.Fatalf("%s observation = %#v", profile.Name, observation) + } + } } func TestQualificationUsesPublicQUICAndProductionPacer(t *testing.T) { qualificationTraverseProfiles(t, qualificationMediaProfiles()) - evidence, err := qualificationPacerEvidence() + evidence, err := qualificationPacerEvidence(filepath.Join(t.TempDir(), "fairness.csv.gz")) if err != nil { t.Fatal(err) } - if len(evidence.PerFlowBytes) != 8 || len(evidence.CapacitySteps) != 2 || evidence.JainIndex < 0.99 { + if len(evidence.PerFlowBytes) != 8 || len(evidence.CapacitySteps) != 2 || + len(evidence.Series) != 80 || evidence.RawSamplesSHA256 == "" || evidence.JainIndex < 0.99 { t.Fatalf("pacer evidence = %#v", evidence) } } + +func TestQualificationSmokeTraversesNativeApolloRecoveryQueuePacerAndQUIC(t *testing.T) { + trace := qualificationProductionPathSmoke(t, qualificationMediaProfiles()[0]) + if !trace.ApolloRecovered || !trace.ProductionQueue || !trace.ProductionPacer || + !trace.VerseQUIC || !trace.PayloadPreserved { + t.Fatalf("qualification production-path trace = %#v", trace) + } +} diff --git a/gateway/qualification_harness_test.go b/gateway/qualification_harness_test.go index 3abfa8c..706c9d3 100644 --- a/gateway/qualification_harness_test.go +++ b/gateway/qualification_harness_test.go @@ -6,6 +6,7 @@ import ( "compress/gzip" "context" "crypto/sha256" + "encoding/binary" "encoding/hex" "encoding/json" "errors" @@ -16,18 +17,22 @@ import ( "path/filepath" "regexp" "runtime" + "runtime/debug" + runtimemetrics "runtime/metrics" "sort" "strings" + "sync" "testing" "time" + + protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" ) const ( - qualificationToolVersion = "versevdi-gateway-qualification/v1" + qualificationToolVersion = "versevdi-gateway-qualification/v2" qualificationImpairmentQueuePackets = 64 qualificationImpairmentMaxPackets = 100_000 qualificationImpairmentPacketCount = 10_000 - qualificationMaximumCatchupPackets = 16_384 qualificationProcessingLimit = 5 * time.Millisecond qualificationImpairmentSeed uint64 = 0x3c6a11ce ) @@ -41,6 +46,40 @@ type qualificationMediaProfile struct { PacketBytes int } +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 +} + +type qualificationPath struct { + client *Client + server *Server + session *nativeApolloSession + key []byte + frame uint32 + closeOnce sync.Once + shutdown func() +} + type qualificationImpairmentProfile struct { Name string RTT time.Duration @@ -73,6 +112,15 @@ type qualificationProcessingSummary struct { RawSamples string `json:"raw_samples"` RawSamplesSHA256 string `json:"raw_samples_sha256"` RawSamplesBytes int64 `json:"raw_samples_bytes"` + ResourceSamples int `json:"resource_samples"` + CPUSeconds float64 `json:"cpu_seconds"` + PeakHeapBytes uint64 `json:"peak_heap_bytes"` + PeakGoroutines int `json:"peak_goroutines"` + Mallocs uint64 `json:"mallocs"` + AllocatedBytes uint64 `json:"allocated_bytes"` + RawResources string `json:"raw_resources"` + RawResourcesSHA256 string `json:"raw_resources_sha256"` + RawResourcesBytes int64 `json:"raw_resources_bytes"` } type qualificationImpairmentObservation struct { @@ -96,6 +144,9 @@ type qualificationImpairmentObservation struct { ConfiguredReorder bool `json:"configured_reorder"` ConfiguredCapacitySteps []int `json:"configured_capacity_steps_percent"` CapacityStepObservations []qualificationCapacityStep `json:"capacity_step_observations,omitempty"` + RawSamples string `json:"raw_samples"` + RawSamplesSHA256 string `json:"raw_samples_sha256"` + RawSamplesBytes int64 `json:"raw_samples_bytes"` } type qualificationCapacityStep struct { @@ -105,12 +156,36 @@ type qualificationCapacityStep struct { FiveSecondCap int64 `json:"five_second_cap_bytes"` } +type qualificationResourceSample struct { + Elapsed time.Duration + CPUSeconds float64 + HeapBytes uint64 + Goroutines int + Mallocs uint64 + Allocated uint64 +} + +type qualificationDeliverySample struct { + At time.Time + Bytes int64 +} + type qualificationFairnessEvidence struct { - Evaluation time.Duration `json:"evaluation_ns"` - PerFlowBytes map[string]int64 `json:"per_flow_bytes"` - ShareError map[string]float64 `json:"share_error"` - JainIndex float64 `json:"jain_index"` - CapacitySteps []qualificationCapacityStep `json:"capacity_steps"` + Evaluation time.Duration `json:"evaluation_ns"` + PerFlowBytes map[string]int64 `json:"per_flow_bytes"` + ShareError map[string]float64 `json:"share_error"` + JainIndex float64 `json:"jain_index"` + CapacitySteps []qualificationCapacityStep `json:"capacity_steps"` + Series []qualificationFairnessSeries `json:"per_flow_aggregate_series"` + RawSamples string `json:"raw_samples"` + RawSamplesSHA256 string `json:"raw_samples_sha256"` + RawSamplesBytes int64 `json:"raw_samples_bytes"` +} + +type qualificationFairnessSeries struct { + Elapsed time.Duration `json:"elapsed_ns"` + PerFlowBytes map[string]int64 `json:"per_flow_bytes"` + AggregateBytes int64 `json:"aggregate_bytes"` } type qualificationManifest struct { @@ -122,6 +197,7 @@ type qualificationManifest struct { StartedAt string `json:"started_at"` CompletedAt string `json:"completed_at"` GoVersion string `json:"go_version"` + ToolVersions map[string]string `json:"tool_versions"` OS string `json:"os"` Architecture string `json:"architecture"` Topology string `json:"topology"` @@ -173,29 +249,212 @@ func qualificationPayload(profile qualificationMediaProfile) []byte { return payload } -func processQualificationPayload(sequence uint32, payload []byte) ([]byte, time.Duration, error) { - started := time.Now() - frames, err := FragmentPayload(ChannelVideo, sequence, uint64(started.UnixMilli()), payload) +func newQualificationPath(t *testing.T, pacerKbps int64) *qualificationPath { + t.Helper() + serverTLS, clientTLS := testTLS(t) + key := []byte("0123456789abcdef") + media, err := newApolloMediaCodec(key, 7) if err != nil { - return nil, 0, err + t.Fatal(err) } - recovered := make([]byte, 0, len(payload)) - for _, frame := range frames { - encoded, encodeErr := EncodeFrame(frame) - if encodeErr != nil { - return nil, 0, encodeErr + 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 } - decoded, decodeErr := DecodeFrame(encoded) - if decodeErr != nil { - return nil, 0, decodeErr + return &qualificationNativeSession{nativeApolloSession: session}, nil + }) + 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", + } + 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, ProviderStateReporter: &recordingProviderStateReporter{}, + Provider: provider, PacerKbps: pacerKbps, + }) + 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-qualification", DeviceSignature: strings.Repeat("s", 86), + Capabilities: DefaultCapabilities(), + } + client, err := Dial(context.Background(), server.Addr().String(), clientTLS, request) + if err != nil { + cancel() + _ = server.Close() + t.Fatal(err) + } + path := &qualificationPath{client: client, server: server, session: session, key: key} + path.shutdown = func() { + _ = client.Close() + cancel() + _ = server.Close() + if err := <-serveDone; err != nil { + t.Errorf("serve qualification path: %v", err) } - recovered = append(recovered, decoded.Payload...) } - elapsed := time.Since(started) - if !bytes.Equal(recovered, payload) { - return nil, elapsed, errors.New("qualification payload integrity failure") + t.Cleanup(path.Close) + return path +} + +func (p *qualificationPath) Close() { + if p != nil { + p.closeOnce.Do(p.shutdown) } - return recovered, elapsed, nil +} + +func (p *qualificationPath) traverse(t *testing.T, payload []byte) (qualificationPathTrace, time.Duration, error) { + t.Helper() + started := time.Now() + trace, err := p.emit(t, payload) + if err != nil { + return trace, 0, err + } + recovered, err := p.receivePayload(context.Background()) + if err != nil { + return trace, 0, err + } + trace.VerseQUIC = 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 +} + +func (p *qualificationPath) emit(t *testing.T, payload []byte) (qualificationPathTrace, error) { + t.Helper() + if p == nil || p.session == nil || len(payload) == 0 || len(payload) > 2*apolloVideoShardPayloadSize-8 { + return qualificationPathTrace{}, ErrProviderMalformed + } + 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) + } + } + if !trace.ApolloRecovered { + return trace, errors.New("Apollo FEC produced no recovered payload") + } + return trace, nil +} + +func (p *qualificationPath) receivePayload(parent context.Context) ([]byte, error) { + ctx, cancel := context.WithTimeout(parent, 2*time.Second) + defer cancel() + var recovered []byte + var sequence uint32 + var fragmentCount byte + for { + frame, err := qualificationReceiveFrame(ctx, p.client) + if err != nil { + return nil, err + } + if frame.Channel != ChannelVideo { + continue + } + if fragmentCount == 0 { + sequence, fragmentCount = frame.Sequence, frame.FragmentCount + } + if frame.Sequence != sequence || frame.FragmentIndex != byte(len(recovered)/1179) { + return nil, errors.New("qualification QUIC fragments reordered") + } + recovered = append(recovered, frame.Payload...) + if frame.FragmentIndex+1 == fragmentCount { + break + } + } + 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 { + payload := make([]byte, apolloVideoShardPayloadSize) + payload[0], payload[3] = 0x01, 0x01 + binary.LittleEndian.PutUint16(payload[4:6], uint16(8+len(encoded))) + copy(payload[8:], encoded) + raw := sourceShapedVideoRaw(frame, uint16(frame), frame, 0x07, 1, 0, 0, payload) + return [][]byte{sourceEncryptVideoRaw(t, key, raw, qualificationVideoIV(frame, 0))} + } + combined := make([]byte, 2*apolloVideoShardPayloadSize) + combined[0], combined[3] = 0x01, 0x01 + binary.LittleEndian.PutUint16(combined[4:6], uint16(8+len(encoded)-apolloVideoShardPayloadSize)) + copy(combined[8:], encoded) + first := sourceShapedVideoRaw(frame, uint16(frame*3), frame*3, 0x05, 2, 50, 0, combined[:apolloVideoShardPayloadSize]) + second := sourceShapedVideoRaw(frame, uint16(frame*3+1), frame*3+1, 0x03, 2, 50, 1, combined[apolloVideoShardPayloadSize:]) + parity := make([]byte, len(first)) + for index := range parity { + parity[index] = first[index] ^ sourceGFMultiply(second[index], 142) + } + sourceConfigureVideoShard(parity, frame, uint16(frame*3+2), frame*3+2, 2, 50, 2) + return [][]byte{ + sourceEncryptVideoRaw(t, key, second, qualificationVideoIV(frame, 1)), + sourceEncryptVideoRaw(t, key, parity, qualificationVideoIV(frame, 2)), + } +} + +func qualificationVideoIV(frame uint32, shard byte) string { + return fmt.Sprintf("%09x%01xQV", frame, shard) +} + +func qualificationProductionPathSmoke(t *testing.T, profile qualificationMediaProfile) qualificationPathTrace { + t.Helper() + path := newQualificationPath(t, profile.BitrateKbps) + defer path.Close() + trace, _, err := path.traverse(t, qualificationPayload(profile)) + if err != nil { + t.Fatal(err) + } + return trace } func summarizeQualificationSamples(samples []time.Duration) (qualificationProcessingSummary, error) { @@ -273,10 +532,29 @@ func observeQualificationHistogram(histogram map[string]int, sample time.Duratio } } -func runQualificationImpairment(profile qualificationImpairmentProfile, media qualificationMediaProfile, packetCount int) (qualificationImpairmentObservation, error) { - if packetCount < 1 || packetCount > qualificationImpairmentMaxPackets || media.PacketBytes < 1 || media.PacketBytes > 1179 { +func runQualificationImpairment(t *testing.T, profile qualificationImpairmentProfile, media qualificationMediaProfile, packetCount int, rawPath string) (qualificationImpairmentObservation, error) { + t.Helper() + if packetCount < 1 || packetCount > qualificationImpairmentMaxPackets || media.PacketBytes < 1 || + media.PacketBytes > 1179 || !qualificationKnownImpairment(profile) { return qualificationImpairmentObservation{}, errors.New("qualification impairment bounds invalid") } + file, err := os.OpenFile(rawPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) + if err != nil { + return qualificationImpairmentObservation{}, err + } + compressed := gzip.NewWriter(file) + buffered := bufio.NewWriter(compressed) + closed := false + defer func() { + if !closed { + _ = buffered.Flush() + _ = compressed.Close() + _ = file.Close() + } + }() + if _, err := buffered.WriteString("source_sequence,sent_ns,delivered_ns,processing_ns,outcome,delivery_order,bytes\n"); err != nil { + return qualificationImpairmentObservation{}, err + } state := qualificationImpairmentSeed random := func() uint64 { state ^= state << 13 @@ -294,81 +572,222 @@ func runQualificationImpairment(profile qualificationImpairmentProfile, media qu ConfiguredLossPercent: profile.LossPercent, ConfiguredReorder: profile.Reorder, ConfiguredCapacitySteps: append([]int(nil), profile.CapacitySteps...), } + path := newQualificationPath(t, media.BitrateKbps) + defer path.Close() + started := time.Now() payload := qualificationPayload(media) - var nextService, virtualEnd, previousArrival time.Duration - var totalRTT, totalJitter time.Duration + var deliveries []qualificationDeliverySample + var totalRTT, totalJitter, previousRTT time.Duration + previousDelivered := -1 + deliveryOrder := 0 + type pendingPacket struct { + index int + jitter time.Duration + } + pending := pendingPacket{index: -1} + stepAt := make(map[int]time.Time, len(profile.CapacitySteps)) + deliver := func(packet pendingPacket) error { + if len(profile.CapacitySteps) == 2 { + switch { + case packet.index >= packetCount*2/3 && stepAt[profile.CapacitySteps[1]].IsZero(): + path.server.pacer.setKbps(media.BitrateKbps * int64(100-profile.CapacitySteps[1]) / 100) + stepAt[profile.CapacitySteps[1]] = time.Now() + case packet.index >= packetCount/3 && stepAt[profile.CapacitySteps[0]].IsZero(): + path.server.pacer.setKbps(media.BitrateKbps * int64(100-profile.CapacitySteps[0]) / 100) + stepAt[profile.CapacitySteps[0]] = time.Now() + } + } + sentAt := started.Add(time.Duration(packet.index) * spacing) + target := sentAt.Add(profile.RTT/2 + packet.jitter) + if delay := time.Until(target); delay > 0 { + time.Sleep(delay) + } + current := append([]byte(nil), payload...) + binary.BigEndian.PutUint32(current[len(current)-4:], uint32(packet.index)) + trace, processing, err := path.traverse(t, current) + if err != nil || !trace.PayloadPreserved || !trace.ProductionPacer { + if err == nil { + err = errors.New("impaired packet bypassed production gateway path") + } + return err + } + deliveredAt := time.Now() + rtt := 2 * deliveredAt.Sub(sentAt) + totalRTT += rtt + if previousRTT != 0 { + delta := rtt - previousRTT + if delta < 0 { + delta = -delta + } + totalJitter += delta + } + previousRTT = rtt + if previousDelivered >= 0 && packet.index < previousDelivered { + observation.ObservedOutOfOrder++ + } + previousDelivered = packet.index + observation.Delivered++ + 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, + sentAt.Sub(started).Nanoseconds(), deliveredAt.Sub(started).Nanoseconds(), + processing.Nanoseconds(), deliveryOrder, len(current)); err != nil { + return err + } + if observation.MaxQueuePackets < 1 { + observation.MaxQueuePackets = 1 + } + return nil + } for index := 0; index < packetCount; index++ { - sentAt := time.Duration(index) * spacing jitter := time.Duration(0) if profile.Jitter > 0 { width := uint64(profile.Jitter*2 + 1) jitter = time.Duration(random()%width) - profile.Jitter } - arrival := sentAt + profile.RTT + jitter - if profile.Reorder && index%20 == 19 { - arrival = previousArrival - spacing - observation.InjectedReordered++ - } - if index > 0 && arrival < previousArrival { - observation.ObservedOutOfOrder++ - } - previousArrival = arrival - totalRTT += arrival - sentAt - if jitter < 0 { - totalJitter -= jitter - } else { - totalJitter += jitter - } if float64(random()%10_000) < profile.LossPercent*100 { observation.Dropped++ - continue - } - capacityPercent := 100 - if len(profile.CapacitySteps) == 2 { - if index < packetCount/2 { - capacityPercent -= profile.CapacitySteps[0] - } else { - capacityPercent -= profile.CapacitySteps[1] + if _, err := fmt.Fprintf(buffered, "%d,%d,0,0,dropped,0,0\n", index, time.Duration(index)*spacing); err != nil { + return qualificationImpairmentObservation{}, err } - } - serviceInterval := spacing * 100 / time.Duration(capacityPercent) - queuePackets := 0 - if nextService > arrival { - queuePackets = int((nextService - arrival + serviceInterval - 1) / serviceInterval) - } - if queuePackets >= qualificationImpairmentQueuePackets { - observation.Dropped++ continue } - if queuePackets > observation.MaxQueuePackets { - observation.MaxQueuePackets = queuePackets + packet := pendingPacket{index: index, jitter: jitter} + if profile.Reorder && index%20 == 18 { + pending = packet + continue } - if arrival > nextService { - nextService = arrival - } - nextService += serviceInterval - virtualEnd = nextService - if _, _, err := processQualificationPayload(uint32(index), payload); err != nil { + if err := deliver(packet); err != nil { return qualificationImpairmentObservation{}, err } - observation.Delivered++ + if pending.index >= 0 { + if err := deliver(pending); err != nil { + return qualificationImpairmentObservation{}, err + } + pending.index = -1 + observation.InjectedReordered++ + } + } + if pending.index >= 0 { + if err := deliver(pending); err != nil { + return qualificationImpairmentObservation{}, err + } + } + if err := buffered.Flush(); err != nil { + return qualificationImpairmentObservation{}, err + } + if err := compressed.Close(); err != nil { + return qualificationImpairmentObservation{}, err + } + if err := file.Close(); err != nil { + return qualificationImpairmentObservation{}, err + } + closed = true + if observation.Delivered > 0 { + observation.ObservedRTT = totalRTT / time.Duration(observation.Delivered) + if observation.Delivered > 1 { + observation.ObservedJitter = totalJitter / time.Duration(observation.Delivered-1) + } + observation.ObservedThroughputKbps = float64(observation.Delivered*media.PacketBytes*8) / time.Since(started).Seconds() / 1000 } - observation.ObservedRTT = totalRTT / time.Duration(packetCount) - observation.ObservedJitter = totalJitter / time.Duration(packetCount) observation.ObservedLossPercent = float64(observation.Dropped) * 100 / float64(packetCount) observation.ObservedReorderPercent = float64(observation.ObservedOutOfOrder) * 100 / float64(packetCount) - if virtualEnd > 0 { - observation.ObservedThroughputKbps = float64(observation.Delivered*media.PacketBytes*8) / virtualEnd.Seconds() / 1000 + observation.RawSamples = filepath.Base(rawPath) + observation.RawSamplesSHA256, observation.RawSamplesBytes, err = qualificationFileSHA256(rawPath) + if err != nil { + return qualificationImpairmentObservation{}, err } - if observation.Delivered+observation.Dropped != observation.Sent || observation.MaxQueuePackets > qualificationImpairmentQueuePackets { + for _, reduction := range profile.CapacitySteps { + bytesPerSecond := media.BitrateKbps * int64(100-reduction) * 1000 / 100 / 8 + stepDeliveries := qualificationDeliveriesAfter(deliveries, stepAt[reduction]) + convergence := qualificationMeasuredConvergence(stepDeliveries, stepAt[reduction], bytesPerSecond) + maximum := qualificationMaximumDeliveryBytes(stepDeliveries, 5*time.Second) + observation.CapacityStepObservations = append(observation.CapacityStepObservations, qualificationCapacityStep{ + ReductionPercent: reduction, Convergence: convergence, + MaximumFiveSecond: maximum, FiveSecondCap: bytesPerSecond * 5, + }) + if packetCount >= qualificationImpairmentPacketCount && + (convergence > 10*time.Second || maximum > bytesPerSecond*5*105/100) { + return qualificationImpairmentObservation{}, fmt.Errorf("capacity step %d failed measured convergence=%s five-second=%d", reduction, convergence, maximum) + } + } + if observation.Delivered+observation.Dropped != observation.Sent || + observation.MaxQueuePackets > qualificationImpairmentQueuePackets { return qualificationImpairmentObservation{}, errors.New("qualification impairment accounting invalid") } return observation, nil } -func runQualificationProcessing(profile qualificationMediaProfile, rawPath string) (qualificationProcessingSummary, error) { +func qualificationKnownImpairment(profile qualificationImpairmentProfile) bool { + for _, known := range qualificationImpairmentProfiles() { + if profile.Name != known.Name || profile.RTT != known.RTT || profile.Jitter != known.Jitter || + profile.LossPercent != known.LossPercent || profile.Reorder != known.Reorder || + len(profile.CapacitySteps) != len(known.CapacitySteps) { + continue + } + match := true + for index := range known.CapacitySteps { + match = match && profile.CapacitySteps[index] == known.CapacitySteps[index] + } + if match { + return true + } + } + return false +} + +func qualificationDeliveriesAfter(deliveries []qualificationDeliverySample, start time.Time) []qualificationDeliverySample { + index := sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].At.Before(start) }) + return deliveries[index:] +} + +func qualificationMeasuredConvergence(deliveries []qualificationDeliverySample, start time.Time, targetBytesPerSecond int64) time.Duration { + const window = 250 * time.Millisecond + for offset := time.Duration(0); offset <= 10*time.Second; offset += window { + windowStart := start.Add(offset) + var total int64 + for _, delivery := range deliveries { + if !delivery.At.Before(windowStart) && delivery.At.Before(windowStart.Add(window)) { + total += delivery.Bytes + } + } + rate := total * int64(time.Second) / int64(window) + if rate >= targetBytesPerSecond*90/100 && rate <= targetBytesPerSecond*105/100 { + return offset + window + } + if len(deliveries) > 0 && windowStart.After(deliveries[len(deliveries)-1].At) { + break + } + } + return 11 * time.Second +} + +func qualificationMaximumDeliveryBytes(deliveries []qualificationDeliverySample, window time.Duration) int64 { + var maximum, total int64 + for first, last := 0, 0; first < len(deliveries); first++ { + for last < len(deliveries) && deliveries[last].At.Sub(deliveries[first].At) <= window { + total += deliveries[last].Bytes + last++ + } + if total > maximum { + maximum = total + } + total -= deliveries[first].Bytes + } + return maximum +} + +func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile, rawPath string) (qualificationProcessingSummary, error) { + t.Helper() payload := qualificationPayload(profile) - if err := runQualificationWarmup(profile, payload); err != nil { + if len(payload) < 4 { + 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) + defer path.Close() + if err := runQualificationWarmup(t, path, profile, payload); err != nil { return qualificationProcessingSummary{}, err } file, err := os.OpenFile(rawPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) @@ -396,34 +815,56 @@ func runQualificationProcessing(profile qualificationMediaProfile, rawPath strin targetPackets := bytesPerSecond * profile.Duration.Nanoseconds() / int64(time.Second) / int64(profile.PacketBytes) samples := make([]time.Duration, 0, int(targetPackets)) started := time.Now() - ticker := time.NewTicker(time.Millisecond) - defer ticker.Stop() + resources := []qualificationResourceSample{qualificationRuntimeSample(started)} + lastResourceSample := started var processed int64 for processed < targetPackets { - elapsed := time.Since(started) - expected := targetPackets - if elapsed < profile.Duration { - expected = targetPackets * elapsed.Nanoseconds() / profile.Duration.Nanoseconds() + batch := int64(8) + if remaining := targetPackets - processed; remaining < batch { + batch = remaining } - if backlog := expected - processed; backlog > qualificationMaximumCatchupPackets { - return qualificationProcessingSummary{}, fmt.Errorf("qualification host fell behind by %d packets", backlog) - } - for processed < expected { - _, sample, processErr := processQualificationPayload(uint32(processed), payload) - if processErr != nil { - return qualificationProcessingSummary{}, processErr + 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") + } + } + 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++ } - if processed < targetPackets { - <-ticker.C + 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") + } + if time.Since(lastResourceSample) >= time.Second { + resources = append(resources, qualificationRuntimeSample(started)) + lastResourceSample = time.Now() } } actualDuration := time.Since(started) + resources = append(resources, qualificationRuntimeSample(started)) if err := buffered.Flush(); err != nil { return qualificationProcessingSummary{}, err } @@ -454,6 +895,30 @@ func runQualificationProcessing(profile qualificationMediaProfile, rawPath strin summary.RawSamples = filepath.Base(rawPath) summary.RawSamplesSHA256 = sum summary.RawSamplesBytes = size + resourcePath := strings.TrimSuffix(rawPath, ".csv.gz") + "-resources.csv.gz" + if err := writeQualificationResourceSamples(resourcePath, resources); err != nil { + return qualificationProcessingSummary{}, err + } + resourceSum, resourceSize, err := qualificationFileSHA256(resourcePath) + if err != nil { + return qualificationProcessingSummary{}, err + } + summary.RawResources = filepath.Base(resourcePath) + summary.RawResourcesSHA256 = resourceSum + summary.RawResourcesBytes = resourceSize + summary.ResourceSamples = len(resources) + firstResource, lastResource := resources[0], resources[len(resources)-1] + summary.CPUSeconds = math.Max(0, lastResource.CPUSeconds-firstResource.CPUSeconds) + summary.Mallocs = lastResource.Mallocs - firstResource.Mallocs + summary.AllocatedBytes = lastResource.Allocated - firstResource.Allocated + for _, resource := range resources { + if resource.HeapBytes > summary.PeakHeapBytes { + summary.PeakHeapBytes = resource.HeapBytes + } + if resource.Goroutines > summary.PeakGoroutines { + 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) } @@ -463,18 +928,60 @@ func runQualificationProcessing(profile qualificationMediaProfile, rawPath strin return summary, nil } -func runQualificationWarmup(profile qualificationMediaProfile, payload []byte) error { +func runQualificationWarmup(t *testing.T, path *qualificationPath, profile qualificationMediaProfile, payload []byte) error { + t.Helper() started := time.Now() - var sequence uint32 for time.Since(started) < profile.Warmup { - if _, _, err := processQualificationPayload(sequence, payload); err != nil { + trace, _, err := path.traverse(t, payload) + if err != nil { return err } - sequence++ + if !trace.PayloadPreserved { + return errors.New("qualification warmup payload integrity failure") + } } return nil } +func qualificationRuntimeSample(started time.Time) qualificationResourceSample { + cpu := []runtimemetrics.Sample{{Name: "/cpu/classes/total:cpu-seconds"}} + runtimemetrics.Read(cpu) + var memory runtime.MemStats + runtime.ReadMemStats(&memory) + return qualificationResourceSample{ + Elapsed: time.Since(started), CPUSeconds: cpu[0].Value.Float64(), + HeapBytes: memory.HeapAlloc, Goroutines: runtime.NumGoroutine(), + Mallocs: memory.Mallocs, Allocated: memory.TotalAlloc, + } +} + +func writeQualificationResourceSamples(path string, samples []qualificationResourceSample) error { + file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) + if err != nil { + return err + } + compressed := gzip.NewWriter(file) + buffered := bufio.NewWriter(compressed) + if _, err = buffered.WriteString("elapsed_ns,cpu_seconds,heap_bytes,goroutines,mallocs,allocated_bytes\n"); err == nil { + for _, sample := range samples { + if _, err = fmt.Fprintf(buffered, "%d,%.9f,%d,%d,%d,%d\n", sample.Elapsed.Nanoseconds(), + sample.CPUSeconds, sample.HeapBytes, sample.Goroutines, sample.Mallocs, sample.Allocated); err != nil { + break + } + } + } + if flushErr := buffered.Flush(); err == nil { + err = flushErr + } + if closeErr := compressed.Close(); err == nil { + err = closeErr + } + if closeErr := file.Close(); err == nil { + err = closeErr + } + return err +} + func qualificationClockOverhead() time.Duration { samples := make([]time.Duration, 10_000) for index := range samples { @@ -499,12 +1006,13 @@ func qualificationFileSHA256(path string) (string, int64, error) { return hex.EncodeToString(hash.Sum(nil)), size, nil } -func qualificationPacerEvidence() (qualificationFairnessEvidence, error) { +func qualificationPacerEvidence(rawPath string) (qualificationFairnessEvidence, error) { start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) 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...) evidence := qualificationFairnessEvidence{ Evaluation: 60 * time.Second, PerFlowBytes: make(map[string]int64, len(flows)), ShareError: make(map[string]float64, len(flows)), @@ -537,7 +1045,8 @@ func qualificationPacerEvidence() (qualificationFairnessEvidence, error) { } { pacer.setKbps(step.kbps) deliveries := runSyntheticPacer(pacer, step.start, step.end, flows, next) - convergence := qualificationPacerConvergence(deliveries, step.start, flows) + allDeliveries = append(allDeliveries, deliveries...) + convergence := qualificationPacerConvergence(deliveries, step.start, flows, step.cap) maximum := qualificationMaximumFiveSecondBytes(deliveries) if 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) @@ -547,26 +1056,43 @@ func qualificationPacerEvidence() (qualificationFairnessEvidence, error) { MaximumFiveSecond: maximum, FiveSecondCap: step.cap * 5, }) } + evidence.Series = qualificationFairnessSeriesFor(allDeliveries, start, start.Add(80*time.Second), 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 { + return qualificationFairnessEvidence{}, err + } return evidence, nil } -func qualificationPacerConvergence(deliveries []syntheticPacerDelivery, start time.Time, flows []string) time.Duration { - first := make(map[string]time.Time, len(flows)) - for _, delivery := range deliveries { - if first[delivery.flow].IsZero() { - first[delivery.flow] = delivery.at +func qualificationPacerConvergence(deliveries []syntheticPacerDelivery, start time.Time, flows []string, targetBytesPerSecond int64) time.Duration { + for second := time.Duration(0); second < 10*time.Second; second += time.Second { + windowStart := start.Add(second) + perFlow := make(map[string]int64, len(flows)) + var aggregate int64 + for _, delivery := range deliveries { + if !delivery.at.Before(windowStart) && delivery.at.Before(windowStart.Add(time.Second)) { + perFlow[delivery.flow] += delivery.bytes + aggregate += delivery.bytes + } + } + if aggregate < targetBytesPerSecond*90/100 || aggregate > targetBytesPerSecond*105/100 { + continue + } + targetFlow := targetBytesPerSecond / int64(len(flows)) + converged := true + for _, flow := range flows { + converged = converged && perFlow[flow] >= targetFlow*90/100 && perFlow[flow] <= targetFlow*110/100 + } + if converged { + return second + time.Second } } - var convergence time.Duration - for _, flow := range flows { - if first[flow].IsZero() { - return 11 * time.Second - } - if delay := first[flow].Sub(start); delay > convergence { - convergence = delay - } - } - return convergence + return 11 * time.Second } func qualificationMaximumFiveSecondBytes(deliveries []syntheticPacerDelivery) int64 { @@ -585,6 +1111,49 @@ func qualificationMaximumFiveSecondBytes(deliveries []syntheticPacerDelivery) in return maximum } +func qualificationFairnessSeriesFor(deliveries []syntheticPacerDelivery, 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{ + Elapsed: windowStart.Sub(start), PerFlowBytes: make(map[string]int64, len(flows)), + } + for _, delivery := range deliveries { + if !delivery.at.Before(windowStart) && delivery.at.Before(windowStart.Add(time.Second)) { + sample.PerFlowBytes[delivery.flow] += delivery.bytes + sample.AggregateBytes += delivery.bytes + } + } + series = append(series, sample) + } + return series +} + +func writeQualificationPacerSamples(path string, deliveries []syntheticPacerDelivery, start time.Time) error { + file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) + if err != nil { + return err + } + compressed := gzip.NewWriter(file) + buffered := bufio.NewWriter(compressed) + if _, err = buffered.WriteString("elapsed_ns,flow,bytes\n"); err == nil { + for _, delivery := range deliveries { + if _, err = fmt.Fprintf(buffered, "%d,%s,%d\n", delivery.at.Sub(start).Nanoseconds(), delivery.flow, delivery.bytes); err != nil { + break + } + } + } + if flushErr := buffered.Flush(); err == nil { + err = flushErr + } + if closeErr := compressed.Close(); err == nil { + err = closeErr + } + if closeErr := file.Close(); err == nil { + err = closeErr + } + return err +} + func qualificationProtocolVersion() (string, error) { version := os.Getenv("VERSEVDI_QUALIFICATION_PROTOCOL_VERSION") valid, err := regexp.MatchString(`^v[0-9]+\.[0-9]+\.[0-9]+-[0-9A-Za-z]+(?:[.-][0-9A-Za-z]+)*$`, version) @@ -603,6 +1172,31 @@ func qualificationCandidateCommit() (string, error) { return commit, nil } +func qualificationToolVersions() (map[string]string, error) { + versions := map[string]string{"qualification": qualificationToolVersion, "go": runtime.Version()} + info, ok := debug.ReadBuildInfo() + if ok { + for _, dependency := range info.Deps { + if dependency.Path == "github.com/quic-go/quic-go" { + versions["quic-go"] = dependency.Version + break + } + } + } + if versions["quic-go"] == "" { + module, err := os.ReadFile(filepath.Join("..", "go.mod")) + if err != nil { + return nil, errors.New("qualification QUIC implementation version unavailable") + } + match := regexp.MustCompile(`(?m)^\s*github\.com/quic-go/quic-go\s+(v[^\s]+)`).FindSubmatch(module) + if len(match) != 2 { + return nil, errors.New("qualification QUIC implementation version unavailable") + } + versions["quic-go"] = string(match[1]) + } + return versions, nil +} + func writeQualificationJSON(path string, value any) error { file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) if err != nil { @@ -633,6 +1227,10 @@ func TestSection7Qualification(t *testing.T) { if err != nil { t.Fatal(err) } + toolVersions, err := qualificationToolVersions() + if err != nil { + t.Fatal(err) + } if err := os.Mkdir(output, 0o750); err != nil { t.Fatalf("create new qualification evidence directory: %v", err) } @@ -647,23 +1245,24 @@ func TestSection7Qualification(t *testing.T) { ), CandidateCommit: commit, ProtocolVersion: protocolVersion, StartedAt: started.Format(time.RFC3339Nano), GoVersion: runtime.Version(), - OS: runtime.GOOS, Architecture: runtime.GOARCH, - Topology: "bounded fixture provider -> gateway framing and mTLS/QUIC transport -> fixture client", + 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", Direction: "provider_to_client", - QueueDiscipline: "deterministic virtual FIFO, 64 packets, fixed seed", - Evidence: []string{"local real-time processing", "mTLS/QUIC fixture transport", "virtual impairment", "production fair pacer", "source-shaped Apollo covered by separate frozen test"}, + 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"}, Deferred: []string{"live Apollo", "macOS client", "physical firewall and packet route", "real encoder fidelity", "multi-host scale"}, } for _, profile := range media { raw := filepath.Join(output, "processing-"+profile.Name+".csv.gz") - summary, runErr := runQualificationProcessing(profile, raw) + summary, runErr := runQualificationProcessing(t, profile, raw) if runErr != nil { t.Fatal(runErr) } 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() + fairness, err := qualificationPacerEvidence(filepath.Join(output, "fairness.csv.gz")) if err != nil { t.Fatal(err) } @@ -674,13 +1273,11 @@ func TestSection7Qualification(t *testing.T) { profiles = media } for _, profile := range profiles { - observation, runErr := runQualificationImpairment(impairment, profile, qualificationImpairmentPacketCount) + raw := filepath.Join(output, "impairment-"+impairment.Name+"-"+profile.Name+".csv.gz") + observation, runErr := runQualificationImpairment(t, impairment, profile, qualificationImpairmentPacketCount, raw) if runErr != nil { t.Fatal(runErr) } - if impairment.Name == "constrained" { - observation.CapacityStepObservations = append([]qualificationCapacityStep(nil), fairness.CapacitySteps...) - } manifest.Impairments = append(manifest.Impairments, observation) } } @@ -704,23 +1301,11 @@ func TestSection7Qualification(t *testing.T) { func qualificationTraverseProfiles(t *testing.T, profiles []qualificationMediaProfile) { t.Helper() - harness := newGatewayTransportHarness(t) for _, profile := range profiles { - payload := qualificationPayload(profile) - harness.session.EmitVideo(payload) - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - for { - frame, err := harness.client.ReceiveFrame(ctx) - if err != nil { - cancel() - t.Fatalf("%s fixture transport: %v", profile.Name, err) - } - if frame.Channel == ChannelVideo && bytes.Equal(frame.Payload, payload) { - break - } + trace := qualificationProductionPathSmoke(t, profile) + if !trace.ApolloRecovered || !trace.ProductionQueue || !trace.ProductionPacer || + !trace.VerseQUIC || !trace.PayloadPreserved { + t.Fatalf("%s production path: %#v", profile.Name, trace) } - cancel() } - _ = harness.client.Close() - harness.waitReleased(t) } diff --git a/gateway/telemetry.go b/gateway/telemetry.go index 105449e..7269cdc 100644 --- a/gateway/telemetry.go +++ b/gateway/telemetry.go @@ -132,6 +132,8 @@ type fairPacerFlow struct { lastSeen time.Time } +const fairPacerMaximumCatchup = 5 * time.Millisecond + func newFairPacer(kbps int64) *fairPacer { pacer := &fairPacer{flows: make(map[string]fairPacerFlow)} pacer.setKbps(kbps) @@ -177,9 +179,11 @@ func (p *fairPacer) reserveAt(now time.Time, flow string, bytes int) time.Time { state := p.flows[flow] state.lastSeen = now p.flows[flow] = state - base := now - if state.next.After(base) { - base = state.next + base := state.next + if base.IsZero() { + base = now + } else if lag := now.Sub(base); lag > fairPacerMaximumCatchup { + base = now.Add(-fairPacerMaximumCatchup) } numerator := int64(bytes) * int64(len(p.flows)) * int64(time.Second) delay := time.Duration((numerator + p.bytesPerSecond - 1) / p.bytesPerSecond) diff --git a/gateway/transport.go b/gateway/transport.go index 975cbf2..112d90e 100644 --- a/gateway/transport.go +++ b/gateway/transport.go @@ -19,13 +19,14 @@ import ( ) const ( - defaultHelloLimit = 16 * 1024 - defaultControlLimit = 128 * 1024 - clientControlBacklog = 64 - applicationError = quic.ApplicationErrorCode(0x100) - controlFlowID = "control.ack.v1" - inputFlowID = "input.sequenced.v1" - clipboardFlowID = "clipboard.text.v1" + 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" ) var ( @@ -324,12 +325,26 @@ func (s *Server) validateProviderWork(work protocol.ProviderSessionWork, authori } if work.SessionID != authority.SessionID || work.GatewayID != authority.GatewayID || work.ReconnectSequence != authority.ReconnectSequence || work.ExpiresAt != authority.ExpiresAt || - work.ProviderProfile != authority.ProviderProfile { + work.ProviderProfile != authority.ProviderProfile || !apolloPolicyMatchesCapabilities(work.StreamPolicy, authority.Capabilities) { return ErrAdmissionRejected } return nil } +func apolloPolicyMatchesCapabilities(policy protocol.ProviderStreamPolicy, capabilities protocol.CapabilityProfile) bool { + if validateApolloStreamPolicy(policy) != nil || capabilities.Audio != "encoded" { + return false + } + switch policy.Codec { + case "H264": + return capabilities.ClientDecode == "h264-opus" || capabilities.ClientDecode == defaultClientDecode + case "HEVC": + return capabilities.ClientDecode == "hevc-opus" || capabilities.ClientDecode == defaultClientDecode + default: + return false + } +} + func (s *Server) addSession(session *gatewaySession) { s.mu.Lock() s.sessions[session] = struct{}{} @@ -358,6 +373,7 @@ type gatewaySession struct { pressed map[string]struct{} sequence atomic.Uint32 mediaDrops uint64 + endReason error result chan error } @@ -389,7 +405,7 @@ func (s *gatewaySession) run() { case <-timer.C: s.server.metrics.InputRejected.Add(1) case <-s.ctx.Done(): - case <-s.result: + case s.endReason = <-s.result: } s.cancel() } @@ -404,6 +420,10 @@ func (s *gatewaySession) providerEventLoop() { if !ok { return } + if event.Kind == ProviderEventDisconnected { + s.result <- ErrProviderDisconnected + return + } payload, err := EncodeProviderEvent(event) if err == nil { err = s.sendControl(s.sequence.Add(1), payload) @@ -412,6 +432,17 @@ func (s *gatewaySession) providerEventLoop() { s.result <- err return } + if event.Kind == ProviderEventTerminated { + timer := time.NewTimer(terminalFeedbackDrain) + select { + case <-s.ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + s.result <- ErrProviderTerminated + return + } } } } @@ -771,6 +802,10 @@ func (s *gatewaySession) cleanup() { _ = s.connection.CloseWithError(applicationError, "session closed") return } + if errors.Is(s.endReason, ErrProviderDisconnected) { + state.State = ProviderStateDisconnected + state.CleanupPending = false + } if err := s.server.config.Admission.Release(cleanupCtx, s.authority); err != nil { s.server.metrics.ProviderErrors.Add(1) } diff --git a/openspec/changes/phase3c-gateway-audit-remediation/.openspec.yaml b/openspec/changes/phase3c-gateway-audit-remediation/.openspec.yaml new file mode 100644 index 0000000..f205fc7 --- /dev/null +++ b/openspec/changes/phase3c-gateway-audit-remediation/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-07-29 diff --git a/openspec/changes/phase3c-gateway-audit-remediation/design.md b/openspec/changes/phase3c-gateway-audit-remediation/design.md new file mode 100644 index 0000000..3f482e1 --- /dev/null +++ b/openspec/changes/phase3c-gateway-audit-remediation/design.md @@ -0,0 +1,47 @@ +## Context + +The implementation already contains a source-shaped Apollo fake, native recovery, bounded gateway queues, fair pacing, Verse framing/QUIC, independent client support, lifecycle reporters, and low-cardinality telemetry. Audit defects arise where those existing pieces are bypassed or not connected. + +## Goals / Non-Goals + +**Goals:** + +- Reuse the existing production path for policy, lifecycle, recovery, telemetry, and qualification. +- Delete duplicate qualification simulation. +- Preserve all trust, cleanup, and resource bounds. + +**Non-Goals:** + +- Add codecs, provider transports, dependencies, or a generic lifecycle/telemetry framework. +- Claim live Apollo/macOS/firewall interoperability. +- Run the normative qualification before immutable consumer resolution. + +## 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. +- On a full audio FEC map, evict the oldest block according to existing block ordering and increment existing drop telemetry. +- 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. +- 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. + +## 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. +- [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. +- [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. + +## Migration Plan + +Land focused red/green repairs locally, verify through the temporary Protocol workspace, preserve old artifacts as superseded, and stop at the publication boundary. After a separately authorized immutable Protocol release is pinned, freeze inputs and run the corrected normative qualification once. + +## Open Questions + +None. diff --git a/openspec/changes/phase3c-gateway-audit-remediation/proposal.md b/openspec/changes/phase3c-gateway-audit-remediation/proposal.md new file mode 100644 index 0000000..738f088 --- /dev/null +++ b/openspec/changes/phase3c-gateway-audit-remediation/proposal.md @@ -0,0 +1,28 @@ +## Why + +Fresh audit evidence shows the gateway ignores the immutable launch policy, leaves tunnels alive after provider termination/disconnect, can permanently stall audio after sustained loss, reports configured capacity as measured egress, and qualifies a standalone simulator instead of the production path. + +## 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). +- 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). +- 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 + +### New Capabilities + +- `apollo-stream-policy`: Native Apollo launch consumes the authenticated effective stream policy without downgrade. +- `provider-session-lifecycle`: Provider terminal events close forwarding and report the correct durable lifecycle outcome. +- `audio-fec-resilience`: Bounded Apollo audio recovery continues after permanently incomplete blocks. +- `gateway-heartbeat-telemetry`: Authenticated heartbeat telemetry reports observed traffic and provider-path measurements. + +### Modified Capabilities + +- `gateway-qualification`: Normative evidence must traverse the production gateway path and retain raw resource, impairment, fairness, cap, and convergence observations. + +## Impact + +The native Apollo adapter, transport lifecycle, audio FEC state, heartbeat sampling, qualification driver, focused fixtures, and canonical qualification spec change. No dependency, cgo, sidecar, codec operation, direct provider route, or live interoperability claim is added. 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 new file mode 100644 index 0000000..3c646c0 --- /dev/null +++ b/openspec/changes/phase3c-gateway-audit-remediation/specs/apollo-stream-policy/spec.md @@ -0,0 +1,15 @@ +## ADDED Requirements + +### Requirement: Apollo launch consumes the effective policy +The native Apollo backend SHALL derive ANNOUNCE resolution, frame rate, supported codec, selected bitrate, and audio profile from authenticated `ProviderSessionWork`, and MUST NOT substitute local defaults. + +#### Scenario: Supported HEVC policy reaches Apollo +- **WHEN** provider work selects HEVC at 2560×1440, 120 FPS, 40000 Kbps, with audio enabled +- **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. + +#### 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 diff --git a/openspec/changes/phase3c-gateway-audit-remediation/specs/audio-fec-resilience/spec.md b/openspec/changes/phase3c-gateway-audit-remediation/specs/audio-fec-resilience/spec.md new file mode 100644 index 0000000..f6a7428 --- /dev/null +++ b/openspec/changes/phase3c-gateway-audit-remediation/specs/audio-fec-resilience/spec.md @@ -0,0 +1,8 @@ +## ADDED Requirements + +### Requirement: Bounded audio FEC state advances after loss +The Apollo audio recovery window SHALL remain bounded and SHALL evict the oldest incomplete block when accepting a newer block would otherwise be rejected. + +#### Scenario: Newer complete block follows sustained loss +- **WHEN** more than the bounded number of permanently incomplete audio blocks arrive before a complete newer block +- **THEN** the oldest stale state is dropped, drop telemetry advances, and the newer encoded payload is relayed unchanged 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 new file mode 100644 index 0000000..2cf0f8e --- /dev/null +++ b/openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-heartbeat-telemetry/spec.md @@ -0,0 +1,15 @@ +## ADDED Requirements + +### Requirement: Heartbeat egress is observed +Authenticated gateway heartbeat telemetry SHALL calculate egress from monotonic transmitted-byte deltas over monotonic elapsed time and MUST NOT report configured capacity as measured traffic. + +#### Scenario: Controlled byte delta is sampled +- **WHEN** transmitted bytes increase by a known amount during a known interval +- **THEN** heartbeat egress equals the measured rate while configured capacity remains a separate registration value + +### Requirement: Required telemetry remains bounded and low cardinality +The established authenticated path SHALL expose observed bytes, packets, drops, RTT, loss, jitter, queue delay, processing delay, pacing, reconnect, and provider state without session, route, credential, or payload labels. + +#### 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 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 new file mode 100644 index 0000000..1415eab --- /dev/null +++ b/openspec/changes/phase3c-gateway-audit-remediation/specs/gateway-qualification/spec.md @@ -0,0 +1,37 @@ +## 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. + +#### 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 +- **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. + +#### Scenario: Complete six-profile run +- **WHEN** the frozen candidate runs impairment qualification +- **THEN** one result exists for each named profile, with no Cartesian expansion and with raw observed rather than configured statistics from the real traversal + +#### Scenario: Unsupported or unbounded configuration +- **WHEN** a profile name, packet count, queue bound, loss, reorder, or bandwidth step falls outside the fixed catalog +- **THEN** the harness rejects it before allocating or running traffic + +### Requirement: Fairness and cap qualification +The harness SHALL exercise the production fair pacer with eight equal-tier synthetic sessions for the required 60-second virtual interval, retain every per-flow and aggregate observation, report every share error and Jain's fairness index, and fail above 10% share error. It SHALL apply 25% and 50% capacity steps, measure convergence of observed allocation rather than first delivery, fail convergence beyond ten virtual seconds, and fail aggregate egress above 105% of the cap over any rolling five-second window. + +#### Scenario: Equal-tier and capacity-step evidence +- **WHEN** the frozen candidate runs scheduler qualification +- **THEN** the artifact contains raw per-flow bytes, aggregate-cap series, share errors, Jain's index, measured allocation convergence, and rolling cap observations derived from the production pacer + +### Requirement: Honest qualification boundary +Qualification artifacts SHALL contain no provider endpoint, credential, clipboard text, input payload, secret, or raw media content and SHALL make no claim of live Apollo/macOS/firewall interoperability. The harness SHALL add no codec operation, production dependency, cgo, sidecar, direct provider route, or duplicate processing/impairment simulator. Deterministic smoke evidence SHALL remain distinct from the single normative run on the frozen immutable consumer candidate. + +#### Scenario: Deterministic evidence publication +- **WHEN** qualification completes +- **THEN** the manifest labels fake-provider, path impairment, and local processing evidence separately and leaves live interoperability deferred-owner-e2e 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 new file mode 100644 index 0000000..5114132 --- /dev/null +++ b/openspec/changes/phase3c-gateway-audit-remediation/specs/provider-session-lifecycle/spec.md @@ -0,0 +1,19 @@ +## 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. + +#### 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 + +#### Scenario: Unexpected provider disconnect is reconnectable +- **WHEN** required provider transport disconnects without acknowledged termination +- **THEN** forwarding stops and the Server receives the existing reconnectable lifecycle state rather than a termination claim + +### Requirement: Cleanup failure remains durable +Gateway cleanup MUST preserve `cleanup_pending` when provider input release, transport cleanup, authorized cancellation, or durable reporting fails. + +#### Scenario: Terminal cleanup fails +- **WHEN** a provider terminal event is handled but required cleanup cannot complete +- **THEN** the session is not reported released or reusable and durable state remains cleanup pending diff --git a/openspec/changes/phase3c-gateway-audit-remediation/tasks.md b/openspec/changes/phase3c-gateway-audit-remediation/tasks.md new file mode 100644 index 0000000..8738dd4 --- /dev/null +++ b/openspec/changes/phase3c-gateway-audit-remediation/tasks.md @@ -0,0 +1,23 @@ +## 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 + +## 2. Lifecycle and Telemetry + +- [x] 2.1 Close forwarding and report durable state on provider termination and disconnect +- [x] 2.2 Preserve cleanup-pending on terminal cleanup failure +- [x] 2.3 Report measured heartbeat egress and required bounded 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.4 Pass short fixed-profile, impairment, fairness, race, parser fuzz, and resource smoke checks + +## 4. Immutable Freeze + +- [ ] 4.1 Pin and verify a separately published never-reused Protocol version from an empty cache +- [ ] 4.2 Freeze all normative inputs and run the corrected Section 7 qualification exactly once