diff --git a/gateway/apollo_native.go b/gateway/apollo_native.go index 39d04a8..9c43902 100644 --- a/gateway/apollo_native.go +++ b/gateway/apollo_native.go @@ -27,6 +27,12 @@ import ( protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol" ) +const ( + nativeApolloVideoQueuePackets = 64 + nativeApolloAudioQueuePackets = 16 + nativeApolloEventQueuePackets = 16 +) + // NativeApolloBackend keeps provider sockets inside the gateway process. The // session-scoped Server work is the sole source of provider endpoint and mTLS // material; it is never serialized into a client manifest or authority. @@ -313,7 +319,16 @@ type nativeApolloSession struct { } func newNativeApolloSession(sessionID string) *nativeApolloSession { - return &nativeApolloSession{sessionID: sessionID, video: make(chan ProviderMedia, 16), audio: make(chan ProviderMedia, 16), events: make(chan ProviderEvent, 16), state: protocol.ProviderState{Version: "1", SessionID: sessionID, State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}}, pressed: make(map[string]InputEvent), done: make(chan struct{}), readDone: make(chan struct{})} + return &nativeApolloSession{ + sessionID: sessionID, + video: make(chan ProviderMedia, nativeApolloVideoQueuePackets), + audio: make(chan ProviderMedia, nativeApolloAudioQueuePackets), + events: make(chan ProviderEvent, nativeApolloEventQueuePackets), + state: protocol.ProviderState{Version: "1", SessionID: sessionID, State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}}, + pressed: make(map[string]InputEvent), + done: make(chan struct{}), + readDone: make(chan struct{}), + } } func newNativeApolloProviderSession(ctx context.Context, setup *apolloRTSPSetup) (*nativeApolloSession, error) { diff --git a/gateway/qualification_harness_test.go b/gateway/qualification_harness_test.go index 765aec2..2a01ff8 100644 --- a/gateway/qualification_harness_test.go +++ b/gateway/qualification_harness_test.go @@ -291,6 +291,21 @@ func qualificationBoundedRelease(target, next, now time.Time, spacing time.Durat return release } +func qualificationWaitUntil(target time.Time) { + const preciseWindow = 500 * time.Microsecond + for { + delay := time.Until(target) + if delay <= 0 { + return + } + if delay > preciseWindow { + time.Sleep(delay - preciseWindow) + } else if delay > 50*time.Microsecond { + runtime.Gosched() + } + } +} + type qualificationTracingBackend struct { native *NativeApolloBackend setups atomic.Uint64 @@ -1340,9 +1355,7 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro var nextRelease time.Time for _, packet := range jobs { release := qualificationBoundedRelease(started.Add(packet.target), nextRelease, time.Now(), spacing) - if delay := time.Until(release); delay > 0 { - time.Sleep(delay) - } + qualificationWaitUntil(release) nextRelease = release.Add(spacing) if len(profile.CapacitySteps) == 2 { switch { @@ -1590,20 +1603,71 @@ func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile, bytesPerSecond := profile.BitrateKbps * 1000 / 8 targetBytes := bytesPerSecond * profile.Duration.Nanoseconds() / int64(time.Second) targetPackets := (targetBytes + int64(profile.PacketBytes) - 1) / int64(profile.PacketBytes) + spacing := time.Duration(int64(time.Second) * int64(profile.PacketBytes) * 8 / (profile.BitrateKbps * 1000)) + maximumDuration := profile.Duration*105/100 + 250*time.Millisecond started := time.Now() + receiveCtx, receiveCancel := context.WithDeadline(context.Background(), started.Add(maximumDuration+2*time.Second)) + defer receiveCancel() + receivedDone := make(chan error, 1) + go func() { + for index := int64(0); index < targetPackets; index++ { + recovered, receiveErr := path.receivePayload(receiveCtx) + if receiveErr != nil { + receivedDone <- receiveErr + return + } + if len(recovered) != len(payload) { + receivedDone <- fmt.Errorf( + "qualification processing payload length = %d, want %d at sequence %d", + len(recovered), len(payload), index, + ) + return + } + sequence := binary.BigEndian.Uint32(recovered[len(recovered)-4:]) + if sequence != uint32(index) { + receivedDone <- fmt.Errorf( + "qualification processing sequence = %d, want %d", sequence, index, + ) + return + } + expected := append([]byte(nil), payload...) + binary.BigEndian.PutUint32(expected[len(expected)-4:], uint32(index)) + if !bytes.Equal(recovered, expected) { + receivedDone <- fmt.Errorf("qualification processing payload bytes changed at sequence %d", index) + return + } + } + receivedDone <- nil + }() var processed int64 - for processed < targetPackets || time.Since(started) < profile.Duration { + var nextRelease time.Time + for processed < targetPackets { + select { + case receiveErr := <-receivedDone: + if receiveErr != nil { + snapshot, _ := path.process.snapshot() + return qualificationProcessingSummary{}, fmt.Errorf("%w; gateway snapshot=%#v", receiveErr, snapshot) + } + return qualificationProcessingSummary{}, errors.New("qualification processing receiver ended early") + default: + } + release := qualificationBoundedRelease( + started.Add(time.Duration(processed)*spacing), nextRelease, time.Now(), spacing, + ) + qualificationWaitUntil(release) + nextRelease = release.Add(spacing) current := append([]byte(nil), payload...) binary.BigEndian.PutUint32(current[len(current)-4:], uint32(processed)) if _, err := path.emit(t, current); err != nil { return qualificationProcessingSummary{}, err } - recovered, err := path.receivePayload(context.Background()) - if err != nil || !bytes.Equal(recovered, current) { - return qualificationProcessingSummary{}, errors.New("qualification processing payload integrity failure") - } processed++ } + if err := <-receivedDone; err != nil { + snapshot, _ := path.process.snapshot() + return qualificationProcessingSummary{}, fmt.Errorf("%w; gateway snapshot=%#v", err, snapshot) + } + qualificationWaitUntil(started.Add(profile.Duration)) actualDuration := time.Since(started) record, err := path.process.stopRecording() if err != nil { @@ -1664,8 +1728,8 @@ func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile, if summary.CPUSeconds < 0 || summary.ClockOverhead <= 0 || summary.ClockMethod == "" { return qualificationProcessingSummary{}, errors.New("process CPU usage unavailable") } - if actualDuration < profile.Duration || actualDuration > profile.Duration*105/100+250*time.Millisecond { - return qualificationProcessingSummary{}, fmt.Errorf("wall-clock duration %s outside [%s,%s]", actualDuration, profile.Duration, profile.Duration*105/100+250*time.Millisecond) + if actualDuration < profile.Duration || actualDuration > maximumDuration { + return qualificationProcessingSummary{}, fmt.Errorf("wall-clock duration %s outside [%s,%s]", actualDuration, profile.Duration, maximumDuration) } if summary.ObservedBitrateKbps < float64(profile.BitrateKbps)*0.95 || summary.ObservedBitrateKbps > float64(profile.BitrateKbps)*1.05 { return qualificationProcessingSummary{}, fmt.Errorf("observed bitrate %.2f outside profile bounds for %d", summary.ObservedBitrateKbps, profile.BitrateKbps) @@ -2137,7 +2201,7 @@ func TestSection7Qualification(t *testing.T) { OS: runtime.GOOS, Architecture: runtime.GOARCH, Topology: "parent source-shaped encrypted Apollo fixture -> isolated gateway subprocess for processing/resource evidence -> public Verse client decoder; impairment uses the same native recovery/FEC, bounded queue, production pacer, framing, and QUIC path", Direction: "provider_to_client", - QueueDiscipline: "ordered fixed-seed source delay queue with one-serialization-interval catch-up, bounded 16-packet native provider queue, production equal-tier fair pacer", + QueueDiscipline: "ordered fixed-seed source delay queue with one-serialization-interval catch-up, bounded 64-packet native video queue, production equal-tier fair pacer", Evidence: []string{"deterministic source-shaped Apollo recovery", "isolated gateway-process resources", "local real-time production path", "mTLS/QUIC fixture transport", "attributed path impairment", "production fair pacer"}, Deferred: []string{"live Apollo", "macOS client", "physical firewall and packet route", "real encoder fidelity", "multi-host scale"}, } diff --git a/gateway/qualification_process_test.go b/gateway/qualification_process_test.go index dbfac9b..b133372 100644 --- a/gateway/qualification_process_test.go +++ b/gateway/qualification_process_test.go @@ -76,6 +76,11 @@ type qualificationProcessRecordResult struct { RecordingElapsed time.Duration } +type qualificationProcessTimingSample struct { + elapsed time.Duration + observation mediaTimingObservation +} + type qualificationProcessRecorder struct { mu sync.Mutex active bool @@ -89,6 +94,8 @@ type qualificationProcessRecorder struct { recordErr error tickerStop chan struct{} tickerDone chan struct{} + timingSamples chan qualificationProcessTimingSample + timingDone chan struct{} clock time.Duration } @@ -135,6 +142,9 @@ func (r *qualificationProcessRecorder) start(request qualificationProcessRecordR r.clock = clock r.tickerStop = make(chan struct{}) r.tickerDone = make(chan struct{}) + r.timingSamples = make(chan qualificationProcessTimingSample, 4096) + r.timingDone = make(chan struct{}) + go r.writeTimings() go r.sampleResources() return nil } @@ -142,13 +152,24 @@ func (r *qualificationProcessRecorder) start(request qualificationProcessRecordR func (r *qualificationProcessRecorder) observe(observation mediaTimingObservation) { r.mu.Lock() defer r.mu.Unlock() - if !r.active || r.recordErr != nil { + if !r.active { return } - _, r.recordErr = fmt.Fprintf(r.rawBuffered, "%d,%d,%d,%d\n", - time.Since(r.started).Nanoseconds(), observation.QueueDelay.Nanoseconds(), - observation.ProcessingDelay.Nanoseconds(), observation.PacingDelay.Nanoseconds()) - r.samples++ + r.timingSamples <- qualificationProcessTimingSample{ + elapsed: time.Since(r.started), observation: observation, + } +} + +func (r *qualificationProcessRecorder) writeTimings() { + defer close(r.timingDone) + for sample := range r.timingSamples { + if r.recordErr == nil { + _, r.recordErr = fmt.Fprintf(r.rawBuffered, "%d,%d,%d,%d\n", + sample.elapsed.Nanoseconds(), sample.observation.QueueDelay.Nanoseconds(), + sample.observation.ProcessingDelay.Nanoseconds(), sample.observation.PacingDelay.Nanoseconds()) + } + r.samples++ + } } func (r *qualificationProcessRecorder) sampleResources() { @@ -177,9 +198,12 @@ func (r *qualificationProcessRecorder) stop() (qualificationProcessRecordResult, } r.active = false stop, done := r.tickerStop, r.tickerDone + timings, timingDone := r.timingSamples, r.timingDone + close(timings) r.mu.Unlock() close(stop) <-done + <-timingDone r.mu.Lock() r.resources = append(r.resources, qualificationRuntimeSample(r.started)) diff --git a/openspec/changes/correct-gateway-qualification-evidence/design.md b/openspec/changes/correct-gateway-qualification-evidence/design.md index 3c2584c..ee23bc0 100644 --- a/openspec/changes/correct-gateway-qualification-evidence/design.md +++ b/openspec/changes/correct-gateway-qualification-evidence/design.md @@ -19,9 +19,12 @@ The qualification driver already reaches the production Apollo-to-QUIC path, but ## Decisions - Reuse the existing source-boundary shaper, preserve source order unless explicit reorder is enabled, and limit catch-up to one media serialization interval. Record the fixed-seed applied-delay standard deviation separately from the jitter observed after ordered traversal. +- Drive processing sends at the configured source rate while a separate public-client receive loop validates ordered payload delivery. Use cooperative scheduling with a bounded high-resolution final wait in the parent driver so sub-millisecond packet spacing does not depend on host sleep granularity. - Assign stable source sequence identifiers and retain per-stage counts so injected loss, provider/FEC drop, queue replacement, QUIC failure, and client miss are disjoint. - Reuse the established gateway child-test pattern for the actual gateway server; the Apollo fixture and QUIC client remain in the parent driver. A token-protected loopback test control endpoint starts and stops bounded child-owned recording and returns aggregate stage state. - Stream queue, processing, and pacing samples from the production `sendMedia` boundary to child-owned raw evidence. Sample child `RUSAGE_SELF`, Go heap, allocations, and goroutines once per second with independent per-run baselines. +- Buffer at most 4,096 child-owned timing samples before the gzip writer; drain every sample before recording stops and backpressure on sustained writer overload instead of dropping evidence or compressing synchronously in the media loop. +- Use the existing reviewed 64-packet qualification bound for native video (about 7.5 ms at 80 Mbps) after public-path counters proved the 16-packet queue replaced 1–5 clean-path units during ordinary scheduler pauses. Keep audio at 16 packets and retain latest-unit replacement at both bounds. - Measure clock overhead as the median elapsed time per read across 1,000 batches of 100 monotonic reads and record that method. ## Risks / Trade-offs @@ -29,4 +32,5 @@ The qualification driver already reaches the production Apollo-to-QUIC path, but - [Ordered release suppresses some delivered jitter] → Retain both the applied fixed-seed delay distribution and the separately observed ordered-traversal jitter. - [Stage attribution double-counts a unit] → Record one terminal outcome per source sequence and validate accounting equality. - [Process sampling perturbs qualification] → Use bounded low-rate samples and include the sampling method in evidence. +- [The source driver consumes CPU for precise pacing] → Keep it in the parent process excluded by the gateway-only resource sampler, and yield cooperatively until the final 50 microseconds. - [Shared private runners cannot sustain the reviewed 20/50/80 Mbps gates] → Run the complete verifier on the registered on-demand xhigh runner; the frozen qualification remains authoritative for the normative duration. diff --git a/openspec/changes/correct-gateway-qualification-evidence/tasks.md b/openspec/changes/correct-gateway-qualification-evidence/tasks.md index 1052fe4..e7b6fe4 100644 --- a/openspec/changes/correct-gateway-qualification-evidence/tasks.md +++ b/openspec/changes/correct-gateway-qualification-evidence/tasks.md @@ -19,5 +19,5 @@ - [x] 4.1 Run focused production-path, race, fuzz, cancellation, slow-reader, amplification, parser-resource, and bounded soak checks - [x] 4.2 Run strict OpenSpec validation, normal-module verification, and reproducible Linux artifact inspection -- [x] 4.3 Freeze all executable inputs and run the corrected normative Section 7 qualification once for the candidate +- [ ] 4.3 Freeze all executable inputs and run the corrected normative Section 7 qualification once for the candidate - [ ] 4.4 Preserve failed attempts and append superseding evidence and ledger rows without rewriting RC8