fix(gateway): preserve paced qualification traffic

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) {

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 {
+		qualificationWaitUntil(release)
-			time.Sleep(delay)
-		}
 		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 {
+	if actualDuration < profile.Duration || actualDuration > maximumDuration {
-		return qualificationProcessingSummary{}, fmt.Errorf("wall-clock duration %s outside [%s,%s]", actualDuration, profile.Duration, profile.Duration*105/100+250*time.Millisecond)
+		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"},
 	}

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",
+	r.timingSamples <- qualificationProcessTimingSample{
-		time.Since(r.started).Nanoseconds(), observation.QueueDelay.Nanoseconds(),
+		elapsed: time.Since(r.started), observation: observation,
-		observation.ProcessingDelay.Nanoseconds(), observation.PacingDelay.Nanoseconds())
+	}
-	r.samples++
+}
+
+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))

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.

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