diff --git a/gateway/qualification_contract_test.go b/gateway/qualification_contract_test.go index 63df641..b23b5ab 100644 --- a/gateway/qualification_contract_test.go +++ b/gateway/qualification_contract_test.go @@ -7,6 +7,7 @@ import ( "encoding/binary" "errors" "io" + "net" "os" "path/filepath" "reflect" @@ -67,6 +68,72 @@ func TestQualificationRecordsLinkedToolVersions(t *testing.T) { } } +func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) { + key := bytes.Repeat([]byte{0x3c}, 16) + encoded := make([]byte, 1000*apolloVideoShardPayloadSize-8) + packets := qualificationSourceVideoPackets(t, key, 1, encoded) + if len(packets) != 1000 || len(packets[0]) != 1072 { + t.Fatalf("source vector = %d packets of %d bytes, want 1000 packets of 1072 bytes", len(packets), len(packets[0])) + } + packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(len(packets[0])) + if packetsPerMillisecond != 93 || batchSize != 61 { + t.Fatalf("Apollo pacing vector = %d packets/ms, batch %d; want 93 and 61", packetsPerMillisecond, batchSize) + } + + receiver, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + defer receiver.Close() + sender, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + defer sender.Close() + + fixture := &qualificationApolloFixture{video: sender, failures: make(chan error, 1)} + remote := *receiver.LocalAddr().(*net.UDPAddr) + fixture.videoRemote.Store(&remote) + drained := make(chan struct{}) + go func() { + defer close(drained) + buffer := make([]byte, 2048) + for range len(packets) + 1 { + if _, _, readErr := receiver.ReadFromUDP(buffer); readErr != nil { + return + } + } + }() + + started := time.Now() + if err := fixture.sendVideo(context.Background(), packets); err != nil { + t.Fatal(err) + } + if err := fixture.sendVideo(context.Background(), packets[:1]); err != nil { + t.Fatal(err) + } + elapsed := time.Since(started) + wantCarry := 10 * time.Millisecond // floor(1000 / 93) ms at Apollo's pinned 80%-of-1-Gbps rate. + if elapsed < wantCarry { + t.Fatalf("source fixture sent the next frame after %s, before Apollo pacing carry %s", elapsed, wantCarry) + } + select { + case <-drained: + case <-time.After(time.Second): + t.Fatal("source-shaped UDP receiver did not drain the fixed vector") + } + fixture.videoNext = time.Now().Add(time.Second) + beforeCancel := fixture.sentPackets.Load() + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + if err := fixture.sendVideo(cancelled, packets[:1]); !errors.Is(err, context.Canceled) { + t.Fatalf("cancelled pacing wait returned %v, want context.Canceled", err) + } + if fixture.sentPackets.Load() != beforeCancel { + t.Fatal("cancelled pacing wait emitted a UDP shard") + } +} + func TestQualificationOutputAndStatisticsFailClosed(t *testing.T) { if err := validateQualificationOutputDir("relative/evidence"); err == nil { t.Fatal("relative evidence directory was accepted") diff --git a/gateway/qualification_harness_test.go b/gateway/qualification_harness_test.go index e39a4b7..f89b9a0 100644 --- a/gateway/qualification_harness_test.go +++ b/gateway/qualification_harness_test.go @@ -39,15 +39,18 @@ import ( ) const ( - qualificationToolVersion = "versevdi-gateway-qualification/v6" - qualificationImpairmentQueuePackets = nativeApolloVideoQueuePackets - qualificationImpairmentMaxPackets = 100_000 - qualificationImpairmentPacketCount = 10_000 - qualificationProcessingLimit = 5 * time.Millisecond - qualificationImpairmentSeed uint64 = 0x3c6a11ce - qualificationClockOverheadMethod = "median of 1000 batches of 100 monotonic time reads" - qualificationGatewayCPUScope = "isolated gateway subprocess; bounded recorder/control included, fixture and client driver excluded" - qualificationResourceMethod = "RUSAGE_SELF user+system CPU; runtime/metrics heap objects, allocated objects/bytes, and live goroutines sampled once per second" + qualificationToolVersion = "versevdi-gateway-qualification/v7" + qualificationImpairmentQueuePackets = nativeApolloVideoQueuePackets + qualificationImpairmentMaxPackets = 100_000 + qualificationImpairmentPacketCount = 10_000 + qualificationProcessingLimit = 5 * time.Millisecond + qualificationImpairmentSeed uint64 = 0x3c6a11ce + qualificationClockOverheadMethod = "median of 1000 batches of 100 monotonic time reads" + qualificationGatewayCPUScope = "isolated gateway subprocess; bounded recorder/control included, fixture and client driver excluded" + qualificationResourceMethod = "RUSAGE_SELF user+system CPU; runtime/metrics heap objects, allocated objects/bytes, and live goroutines sampled once per second" + qualificationApolloVideoRateBitsPerSecond = 1_000_000_000 * 80 / 100 + qualificationApolloVideoBatchBytes = 64 * 1024 + qualificationApolloVideoBatchPackets = 64 ) type qualificationMediaProfile struct { @@ -425,6 +428,8 @@ type qualificationApolloFixture struct { closeOnce sync.Once sentPackets atomic.Uint64 work protocol.ProviderSessionWork + videoPaceMu sync.Mutex + videoNext time.Time controlImpairmentMu sync.Mutex controlRTT time.Duration @@ -754,15 +759,76 @@ func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][] } } remote := f.videoRemote.Load() - for _, packet := range packets { - if _, err := f.video.WriteToUDP(packet, remote); err != nil { - return err - } - f.sentPackets.Add(1) + if len(packets) == 0 { + return nil } + packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(len(packets[0])) + if packetsPerMillisecond == 0 || batchSize == 0 { + return ErrProviderMalformed + } + + f.videoPaceMu.Lock() + defer f.videoPaceMu.Unlock() + frameStart := time.Now() + if f.videoNext.After(frameStart) { + frameStart = f.videoNext + } + framePackets, groupPackets := 0, 0 + for batchStart := 0; batchStart < len(packets); batchStart += batchSize { + if framePackets == 0 || groupPackets >= packetsPerMillisecond { + due := frameStart.Add(time.Millisecond * time.Duration(framePackets) / time.Duration(packetsPerMillisecond)) + if err := qualificationWaitContext(ctx, due); err != nil { + return err + } + groupPackets = 0 + } + batchEnd := min(batchStart+batchSize, len(packets)) + for _, packet := range packets[batchStart:batchEnd] { + if len(packet) != len(packets[0]) { + return ErrProviderMalformed + } + if _, err := f.video.WriteToUDP(packet, remote); err != nil { + return err + } + f.sentPackets.Add(1) + } + currentBatch := batchEnd - batchStart + framePackets += currentBatch + groupPackets += currentBatch + } + f.videoNext = frameStart.Add(time.Millisecond * time.Duration(framePackets) / time.Duration(packetsPerMillisecond)) return nil } +func qualificationApolloVideoPacing(packetBytes int) (packetsPerMillisecond, batchSize int) { + if packetBytes <= 0 { + return 0, 0 + } + packetsPerMillisecond = qualificationApolloVideoRateBitsPerSecond / 1000 / packetBytes / 8 + batchSize = min(qualificationApolloVideoBatchBytes/packetBytes, qualificationApolloVideoBatchPackets) + return packetsPerMillisecond, batchSize +} + +func qualificationWaitContext(ctx context.Context, due time.Time) error { + delay := time.Until(due) + if delay <= 0 { + select { + case <-ctx.Done(): + return ctx.Err() + default: + return nil + } + } + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + func (f *qualificationApolloFixture) streamKey() ([]byte, error) { key := f.key.Load() if key == nil || len(*key) != 16 { diff --git a/openspec/changes/truthful-fixed-frame-qualification/design.md b/openspec/changes/truthful-fixed-frame-qualification/design.md index 2d65c88..5ed2c92 100644 --- a/openspec/changes/truthful-fixed-frame-qualification/design.md +++ b/openspec/changes/truthful-fixed-frame-qualification/design.md @@ -2,6 +2,8 @@ The current harness sends one fixed 1,179-byte payload per logical sample. It reaches the production path but does not represent encoded frames at 60/120 FPS or exercise realistic fragmentation, reassembly, queue bytes, and keyframe pressure. +The complete-frame fixture also must preserve the pinned Apollo source schedule. For each frame it derives packets per millisecond from the raw UDP block size at 80% of 1 Gbps, limits source batches to both 64 KiB and 64 packets, and carries the next-send time into the following frame. Waiting is context-cancellable. This is qualification-fixture behavior only; production transport and queue behavior remain unchanged. + ## Goals / Non-Goals **Goals:** diff --git a/openspec/changes/truthful-fixed-frame-qualification/proposal.md b/openspec/changes/truthful-fixed-frame-qualification/proposal.md index ba2d78f..687718a 100644 --- a/openspec/changes/truthful-fixed-frame-qualification/proposal.md +++ b/openspec/changes/truthful-fixed-frame-qualification/proposal.md @@ -2,6 +2,8 @@ The existing fixed-profile harness treats each 1,179-byte datagram as an encoded frame, so its reported frame rate, frame boundaries, bitrate, queue pressure, and processing evidence do not model the named 60/120 FPS profiles. +The v6 complete-frame fixture subsequently exposed a source-fidelity defect on ordinary Linux runners: it emitted every UDP shard in one tight loop, unlike pinned Apollo's bounded intra-frame rate and batch schedule. The affected v6 qualification evidence remains retained but is superseded for candidate-readiness purposes. + ## What Changes - Generate deterministic variable-size encoded frame units at the named frame rates and target bitrates, including bounded keyframes. diff --git a/openspec/changes/truthful-fixed-frame-qualification/specs/gateway-qualification/spec.md b/openspec/changes/truthful-fixed-frame-qualification/specs/gateway-qualification/spec.md index 4e63ab4..83617ae 100644 --- a/openspec/changes/truthful-fixed-frame-qualification/specs/gateway-qualification/spec.md +++ b/openspec/changes/truthful-fixed-frame-qualification/specs/gateway-qualification/spec.md @@ -3,6 +3,8 @@ ### Requirement: Fixed media processing qualification The qualification harness SHALL drive pinned-mTLS Apollo management, encrypted RTSP, ENet, and provider UDP through native source validation, `readUDPMedia`, recovery/FEC, byte/count/latency-bounded production queues, the production fair pacer, Protocol complete-frame fragmentation, Verse framing/QUIC, and an independent bounded client reassembler for 1080p60 H.264 at 20 Mbps, 1440p120 HEVC at 50 Mbps, and 4K60 HEVC at 80 Mbps. The source fixture SHALL emit deterministic variable-size complete encoded frame units at the named 60/120 FPS rate, preserve exact target bytes over each fixed interval, and include bounded larger keyframes without codec operation. After a recorded warm-up, the frozen candidate SHALL run each profile for ten wall-clock minutes, preserve every frame's bytes and boundary, retain every monotonic processing sample plus bounded provider-queue observations, and report frame count, frame rate, bitrate, count, min, median, p90, p95, p99, max, mean, standard deviation, measured batched monotonic-clock overhead and method, and observed bitrate. Processing begins at complete provider-frame receipt and ends at QUIC handoff, excluding client transit and pacing. Queue delay SHALL measure provider-queue residence, processing SHALL measure gateway work before pacing, and pacing delay SHALL measure scheduler waiting. CPU, heap, allocations, and goroutines SHALL be measured from the isolated gateway process only; CPU SHALL be actual OS user plus system consumption and MUST NOT include idle wall capacity or unrelated parent fixture/client work. Successive profiles SHALL use independent resource-counter baselines. Any bypass, payload or boundary mutation, frame-rate/count mismatch, wall-duration violation, bitrate outside both lower and upper bounds, unexplained clean-path loss, zero or unbounded clock overhead, or p95 above 5 ms SHALL fail. +Within each complete frame the source fixture SHALL reproduce pinned Apollo's source schedule by deriving packets per millisecond from the raw UDP block size at 80% of 1 Gbps, bounding each source batch to the smaller of 64 KiB or 64 packets, carrying the next-send time across frames, and making pacing waits context-cancellable. + #### Scenario: Healthy fixed profile - **WHEN** a frozen candidate runs one fixed profile for the normative duration in the isolated qualification command - **THEN** the harness emits compressed raw frame/path and gateway-process resource samples plus a summary tied to the exact command, CPU scope, timing-overhead method, topology, source commit, immutable Protocol version, environment, and payload hash @@ -10,3 +12,7 @@ The qualification harness SHALL drive pinned-mTLS Apollo management, encrypted R #### Scenario: Processing gate failure - **WHEN** any production path stage lacks a per-frame observation, stage accounting does not balance, payload or frame boundaries change, duration, frame-rate, frame-count, or bitrate bounds fail, measured p95 exceeds 5 ms, parent work changes gateway CPU, idle capacity is reported as consumed CPU, or timing overhead is absent - **THEN** the qualification command exits unsuccessfully without recording a passing candidate + +#### Scenario: Source-shaped Apollo pacing is preserved +- **WHEN** the fixture emits 1,072-byte encrypted video shards for consecutive complete frames +- **THEN** it uses 93 packets per millisecond, batches at most 61 shards, carries the integer next-send offset into the following frame, and emits no shard after a cancelled pacing wait diff --git a/openspec/changes/truthful-fixed-frame-qualification/tasks.md b/openspec/changes/truthful-fixed-frame-qualification/tasks.md index 540ba4e..c75e935 100644 --- a/openspec/changes/truthful-fixed-frame-qualification/tasks.md +++ b/openspec/changes/truthful-fixed-frame-qualification/tasks.md @@ -17,3 +17,9 @@ ## 4. Frozen qualification - [x] 4.1 Run the single normative Section 7 qualification after immutable Protocol consumer resolution + +## 5. Pinned Apollo source-fidelity remediation + +- [x] 5.1 Retain the v6 qualification attempt and mark its passing result superseded by the tight-loop source defect +- [ ] 5.2 Rate- and batch-shape complete-frame UDP emission from the pinned Apollo behavior and verify focused, race, resource, full, artifact, and CI gates +- [ ] 5.3 Freeze the v7 harness descendant and run one separately approved replacement normative Section 7 qualification