From f37f53eab28d960150382862b159df22800d0e01 Mon Sep 17 00:00:00 2001 From: sechmachine <97589681+sechmachine727@users.noreply.github.com> Date: Wed, 29 Jul 2026 22:13:48 +0700 Subject: [PATCH] test(gateway): add Section 7 qualification harness --- gateway/qualification_contract_test.go | 152 ++++ gateway/qualification_harness_test.go | 723 ++++++++++++++++++ .../.openspec.yaml | 2 + .../phase3c-gateway-qualification/design.md | 54 ++ .../phase3c-gateway-qualification/proposal.md | 35 + .../specs/gateway-qualification/spec.md | 63 ++ .../phase3c-gateway-qualification/tasks.md | 24 + 7 files changed, 1053 insertions(+) create mode 100644 gateway/qualification_contract_test.go create mode 100644 gateway/qualification_harness_test.go create mode 100644 openspec/changes/phase3c-gateway-qualification/.openspec.yaml create mode 100644 openspec/changes/phase3c-gateway-qualification/design.md create mode 100644 openspec/changes/phase3c-gateway-qualification/proposal.md create mode 100644 openspec/changes/phase3c-gateway-qualification/specs/gateway-qualification/spec.md create mode 100644 openspec/changes/phase3c-gateway-qualification/tasks.md diff --git a/gateway/qualification_contract_test.go b/gateway/qualification_contract_test.go new file mode 100644 index 0000000..785d519 --- /dev/null +++ b/gateway/qualification_contract_test.go @@ -0,0 +1,152 @@ +package gateway + +import ( + "compress/gzip" + "io" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + "time" +) + +func TestQualificationCatalogMatchesSection7(t *testing.T) { + media := qualificationMediaProfiles() + if len(media) != 3 { + t.Fatalf("media profile count = %d, want 3", len(media)) + } + wantMedia := []qualificationMediaProfile{ + {Name: "1080p60-h264", Codec: "h264", BitrateKbps: 20000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179}, + {Name: "1440p120-hevc", Codec: "hevc", BitrateKbps: 50000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179}, + {Name: "4k60-hevc", Codec: "hevc", BitrateKbps: 80000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179}, + } + if !reflect.DeepEqual(media, wantMedia) { + t.Fatalf("media profiles = %#v, want %#v", media, wantMedia) + } + + impairments := qualificationImpairmentProfiles() + wantImpairments := []qualificationImpairmentProfile{ + {Name: "baseline", RTT: 20 * time.Millisecond}, + {Name: "latency", RTT: 150 * time.Millisecond}, + {Name: "jitter", RTT: 50 * time.Millisecond, Jitter: 30 * time.Millisecond}, + {Name: "loss", RTT: 50 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 5}, + {Name: "reorder", RTT: 100 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 1, Reorder: true}, + {Name: "constrained", RTT: 50 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 2, Reorder: true, CapacitySteps: []int{25, 50}}, + } + if !reflect.DeepEqual(impairments, wantImpairments) { + t.Fatalf("impairment profiles = %#v, want %#v", impairments, wantImpairments) + } +} + +func TestQualificationOutputAndStatisticsFailClosed(t *testing.T) { + if err := validateQualificationOutputDir("relative/evidence"); err == nil { + t.Fatal("relative evidence directory was accepted") + } + if err := validateQualificationOutputDir(filepath.Join(t.TempDir(), "evidence")); err != nil { + t.Fatalf("absolute evidence directory rejected: %v", err) + } + + summary, err := summarizeQualificationSamples([]time.Duration{ + time.Millisecond, 2 * time.Millisecond, 3 * time.Millisecond, + 4 * time.Millisecond, 5 * time.Millisecond, + }) + if err != nil { + t.Fatal(err) + } + if summary.Count != 5 || summary.Min != time.Millisecond || summary.Median != 3*time.Millisecond || + summary.P90 != 5*time.Millisecond || summary.P95 != 5*time.Millisecond || + summary.P99 != 5*time.Millisecond || summary.Max != 5*time.Millisecond || + summary.Mean != 3*time.Millisecond || summary.Histogram["le_5ms"] != 5 { + t.Fatalf("summary = %#v", summary) + } + if err := enforceQualificationProcessingGate(summary); err != nil { + t.Fatalf("5 ms p95 rejected: %v", err) + } + summary.P95++ + if err := enforceQualificationProcessingGate(summary); err == nil { + t.Fatal("p95 above 5 ms was accepted") + } +} + +func TestQualificationShortProcessingWritesRawArtifact(t *testing.T) { + profile := qualificationMediaProfile{ + Name: "smoke", Codec: "h264", BitrateKbps: 1000, + Duration: 200 * time.Millisecond, Warmup: time.Millisecond, PacketBytes: 100, + } + rawPath := filepath.Join(t.TempDir(), "processing.csv.gz") + summary, err := runQualificationProcessing(profile, rawPath) + if err != nil { + t.Fatal(err) + } + if summary.Count < 1 || summary.RawSamplesSHA256 == "" || summary.RawSamplesBytes < 1 { + t.Fatalf("processing summary = %#v", summary) + } + file, err := os.Open(rawPath) + if err != nil { + t.Fatal(err) + } + defer file.Close() + reader, err := gzip.NewReader(file) + if err != nil { + t.Fatal(err) + } + raw, err := io.ReadAll(reader) + if err != nil { + t.Fatal(err) + } + if err := reader.Close(); err != nil { + t.Fatal(err) + } + if !strings.HasPrefix(string(raw), "elapsed_ns,processing_ns\n") || strings.Count(string(raw), "\n") != int(summary.Count)+1 { + t.Fatalf("raw sample rows do not match summary count: %q", raw) + } +} + +func TestQualificationProcessingPreservesPayload(t *testing.T) { + payload := qualificationPayload(qualificationMediaProfiles()[0]) + processed, elapsed, err := processQualificationPayload(7, payload) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(processed, payload) { + t.Fatal("encoded payload mutated") + } + if elapsed <= 0 { + t.Fatalf("processing duration = %s", elapsed) + } +} + +func TestQualificationImpairmentIsDeterministicAndBounded(t *testing.T) { + profile := qualificationImpairmentProfiles()[3] + first, err := runQualificationImpairment(profile, qualificationMediaProfiles()[0], 10_000) + if err != nil { + t.Fatal(err) + } + second, err := runQualificationImpairment(profile, qualificationMediaProfiles()[0], 10_000) + 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.Sent != 10_000 || first.Delivered+first.Dropped != first.Sent || + first.ObservedLossPercent < 4.8 || first.ObservedLossPercent > 5.2 || + first.MaxQueuePackets > qualificationImpairmentQueuePackets { + t.Fatalf("impairment observation = %#v", first) + } + if _, err := runQualificationImpairment(profile, qualificationMediaProfiles()[0], qualificationImpairmentMaxPackets+1); err == nil { + t.Fatal("unbounded impairment packet count was accepted") + } +} + +func TestQualificationUsesPublicQUICAndProductionPacer(t *testing.T) { + qualificationTraverseProfiles(t, qualificationMediaProfiles()) + evidence, err := qualificationPacerEvidence() + if err != nil { + t.Fatal(err) + } + if len(evidence.PerFlowBytes) != 8 || len(evidence.CapacitySteps) != 2 || evidence.JainIndex < 0.99 { + t.Fatalf("pacer evidence = %#v", evidence) + } +} diff --git a/gateway/qualification_harness_test.go b/gateway/qualification_harness_test.go new file mode 100644 index 0000000..de48852 --- /dev/null +++ b/gateway/qualification_harness_test.go @@ -0,0 +1,723 @@ +package gateway + +import ( + "bufio" + "bytes" + "compress/gzip" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "math" + "os" + "path/filepath" + "runtime" + "runtime/debug" + "sort" + "strings" + "testing" + "time" +) + +const ( + qualificationToolVersion = "versevdi-gateway-qualification/v1" + qualificationImpairmentQueuePackets = 64 + qualificationImpairmentMaxPackets = 100_000 + qualificationImpairmentPacketCount = 10_000 + qualificationMaximumCatchupPackets = 16_384 + qualificationProcessingLimit = 5 * time.Millisecond + qualificationImpairmentSeed uint64 = 0x3c6a11ce +) + +type qualificationMediaProfile struct { + Name string + Codec string + BitrateKbps int64 + Duration time.Duration + Warmup time.Duration + PacketBytes int +} + +type qualificationImpairmentProfile struct { + Name string + RTT time.Duration + Jitter time.Duration + LossPercent float64 + Reorder bool + CapacitySteps []int +} + +type qualificationProcessingSummary struct { + Profile string `json:"profile"` + Codec string `json:"codec"` + ConfiguredBitrateKbps int64 `json:"configured_bitrate_kbps"` + ObservedBitrateKbps float64 `json:"observed_bitrate_kbps"` + Warmup time.Duration `json:"warmup_ns"` + ConfiguredDuration time.Duration `json:"configured_duration_ns"` + ActualDuration time.Duration `json:"actual_duration_ns"` + Count int64 `json:"count"` + Min time.Duration `json:"min_ns"` + Median time.Duration `json:"median_ns"` + P90 time.Duration `json:"p90_ns"` + P95 time.Duration `json:"p95_ns"` + P99 time.Duration `json:"p99_ns"` + Max time.Duration `json:"max_ns"` + Mean time.Duration `json:"mean_ns"` + StandardDeviation time.Duration `json:"standard_deviation_ns"` + ClockOverhead time.Duration `json:"clock_overhead_ns"` + Histogram map[string]int `json:"histogram"` + PayloadSHA256 string `json:"payload_sha256"` + RawSamples string `json:"raw_samples"` + RawSamplesSHA256 string `json:"raw_samples_sha256"` + RawSamplesBytes int64 `json:"raw_samples_bytes"` +} + +type qualificationImpairmentObservation struct { + Profile string `json:"profile"` + MediaProfile string `json:"media_profile"` + Seed uint64 `json:"seed"` + Sent int `json:"sent"` + Delivered int `json:"delivered"` + Dropped int `json:"dropped"` + InjectedReordered int `json:"injected_reordered"` + ObservedOutOfOrder int `json:"observed_out_of_order"` + ObservedRTT time.Duration `json:"observed_rtt_ns"` + ObservedJitter time.Duration `json:"observed_jitter_ns"` + ObservedLossPercent float64 `json:"observed_loss_percent"` + ObservedReorderPercent float64 `json:"observed_reorder_percent"` + ObservedThroughputKbps float64 `json:"observed_throughput_kbps"` + MaxQueuePackets int `json:"max_queue_packets"` + ConfiguredRTT time.Duration `json:"configured_rtt_ns"` + ConfiguredJitter time.Duration `json:"configured_jitter_ns"` + ConfiguredLossPercent float64 `json:"configured_loss_percent"` + ConfiguredReorder bool `json:"configured_reorder"` + ConfiguredCapacitySteps []int `json:"configured_capacity_steps_percent"` + CapacityStepObservations []qualificationCapacityStep `json:"capacity_step_observations,omitempty"` +} + +type qualificationCapacityStep struct { + ReductionPercent int `json:"reduction_percent"` + Convergence time.Duration `json:"convergence_ns"` + MaximumFiveSecond int64 `json:"maximum_five_second_bytes"` + FiveSecondCap int64 `json:"five_second_cap_bytes"` +} + +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"` +} + +type qualificationManifest struct { + Status string `json:"status"` + ToolVersion string `json:"tool_version"` + Command string `json:"command"` + CandidateCommit string `json:"candidate_commit"` + ProtocolVersion string `json:"protocol_version"` + StartedAt string `json:"started_at"` + CompletedAt string `json:"completed_at"` + GoVersion string `json:"go_version"` + OS string `json:"os"` + Architecture string `json:"architecture"` + Topology string `json:"topology"` + Direction string `json:"direction"` + QueueDiscipline string `json:"queue_discipline"` + Evidence []string `json:"evidence_classification"` + Deferred []string `json:"deferred"` + Processing []qualificationProcessingSummary `json:"processing"` + Impairments []qualificationImpairmentObservation `json:"impairments"` + Fairness qualificationFairnessEvidence `json:"fairness"` +} + +func qualificationMediaProfiles() []qualificationMediaProfile { + return []qualificationMediaProfile{ + {Name: "1080p60-h264", Codec: "h264", BitrateKbps: 20000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179}, + {Name: "1440p120-hevc", Codec: "hevc", BitrateKbps: 50000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179}, + {Name: "4k60-hevc", Codec: "hevc", BitrateKbps: 80000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179}, + } +} + +func qualificationImpairmentProfiles() []qualificationImpairmentProfile { + return []qualificationImpairmentProfile{ + {Name: "baseline", RTT: 20 * time.Millisecond}, + {Name: "latency", RTT: 150 * time.Millisecond}, + {Name: "jitter", RTT: 50 * time.Millisecond, Jitter: 30 * time.Millisecond}, + {Name: "loss", RTT: 50 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 5}, + {Name: "reorder", RTT: 100 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 1, Reorder: true}, + {Name: "constrained", RTT: 50 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 2, Reorder: true, CapacitySteps: []int{25, 50}}, + } +} + +func validateQualificationOutputDir(path string) error { + if path == "" || !filepath.IsAbs(path) || filepath.Clean(path) == string(filepath.Separator) { + return errors.New("qualification evidence directory must be a non-root absolute path") + } + return nil +} + +func qualificationPayload(profile qualificationMediaProfile) []byte { + payload := make([]byte, profile.PacketBytes) + if profile.Codec == "h264" { + copy(payload, []byte{0, 0, 1, 0x65}) + } else { + copy(payload, []byte{0, 0, 1, 0x26}) + } + for index := 4; index < len(payload); index++ { + payload[index] = byte(index*31 + len(profile.Name)) + } + return payload +} + +func processQualificationPayload(sequence uint32, payload []byte) ([]byte, time.Duration, error) { + started := time.Now() + frames, err := FragmentPayload(ChannelVideo, sequence, uint64(started.UnixMilli()), payload) + if err != nil { + return nil, 0, err + } + recovered := make([]byte, 0, len(payload)) + for _, frame := range frames { + encoded, encodeErr := EncodeFrame(frame) + if encodeErr != nil { + return nil, 0, encodeErr + } + decoded, decodeErr := DecodeFrame(encoded) + if decodeErr != nil { + return nil, 0, decodeErr + } + recovered = append(recovered, decoded.Payload...) + } + elapsed := time.Since(started) + if !bytes.Equal(recovered, payload) { + return nil, elapsed, errors.New("qualification payload integrity failure") + } + return recovered, elapsed, nil +} + +func summarizeQualificationSamples(samples []time.Duration) (qualificationProcessingSummary, error) { + if len(samples) == 0 { + return qualificationProcessingSummary{}, errors.New("qualification has no processing samples") + } + var mean, m2 float64 + histogram := newQualificationHistogram() + for index, sample := range samples { + value := float64(sample) + delta := value - mean + mean += delta / float64(index+1) + m2 += delta * (value - mean) + observeQualificationHistogram(histogram, sample) + } + sort.Slice(samples, func(first, second int) bool { return samples[first] < samples[second] }) + return qualificationProcessingSummary{ + Count: int64(len(samples)), + Min: samples[0], + Median: qualificationPercentile(samples, 0.50), + P90: qualificationPercentile(samples, 0.90), + P95: qualificationPercentile(samples, 0.95), + P99: qualificationPercentile(samples, 0.99), + Max: samples[len(samples)-1], + Mean: time.Duration(mean), + StandardDeviation: time.Duration(math.Sqrt(m2 / float64(len(samples)))), + Histogram: histogram, + }, nil +} + +func qualificationPercentile(samples []time.Duration, percentile float64) time.Duration { + index := int(math.Ceil(percentile*float64(len(samples)))) - 1 + if index < 0 { + index = 0 + } + return samples[index] +} + +func enforceQualificationProcessingGate(summary qualificationProcessingSummary) error { + if summary.Count < 1 || summary.P95 > qualificationProcessingLimit { + return fmt.Errorf("processing p95 %s exceeds %s", summary.P95, qualificationProcessingLimit) + } + return nil +} + +func newQualificationHistogram() map[string]int { + return map[string]int{ + "le_1us": 0, "le_5us": 0, "le_10us": 0, "le_25us": 0, + "le_50us": 0, "le_100us": 0, "le_250us": 0, "le_500us": 0, + "le_1ms": 0, "le_2ms": 0, "le_5ms": 0, "gt_5ms": 0, + } +} + +func observeQualificationHistogram(histogram map[string]int, sample time.Duration) { + buckets := []struct { + name string + limit time.Duration + }{ + {"le_1us", time.Microsecond}, {"le_5us", 5 * time.Microsecond}, + {"le_10us", 10 * time.Microsecond}, {"le_25us", 25 * time.Microsecond}, + {"le_50us", 50 * time.Microsecond}, {"le_100us", 100 * time.Microsecond}, + {"le_250us", 250 * time.Microsecond}, {"le_500us", 500 * time.Microsecond}, + {"le_1ms", time.Millisecond}, {"le_2ms", 2 * time.Millisecond}, + {"le_5ms", 5 * time.Millisecond}, + } + observed := false + for _, bucket := range buckets { + if sample <= bucket.limit { + histogram[bucket.name]++ + observed = true + } + } + if !observed { + histogram["gt_5ms"]++ + } +} + +func runQualificationImpairment(profile qualificationImpairmentProfile, media qualificationMediaProfile, packetCount int) (qualificationImpairmentObservation, error) { + if packetCount < 1 || packetCount > qualificationImpairmentMaxPackets || media.PacketBytes < 1 || media.PacketBytes > 1179 { + return qualificationImpairmentObservation{}, errors.New("qualification impairment bounds invalid") + } + state := qualificationImpairmentSeed + random := func() uint64 { + state ^= state << 13 + state ^= state >> 7 + state ^= state << 17 + return state + } + spacing := time.Duration(int64(time.Second) * int64(media.PacketBytes) * 8 / (media.BitrateKbps * 1000)) + if spacing < time.Nanosecond { + spacing = time.Nanosecond + } + observation := qualificationImpairmentObservation{ + Profile: profile.Name, MediaProfile: media.Name, Seed: qualificationImpairmentSeed, + Sent: packetCount, ConfiguredRTT: profile.RTT, ConfiguredJitter: profile.Jitter, + ConfiguredLossPercent: profile.LossPercent, ConfiguredReorder: profile.Reorder, + ConfiguredCapacitySteps: append([]int(nil), profile.CapacitySteps...), + } + payload := qualificationPayload(media) + var nextService, virtualEnd, previousArrival time.Duration + var totalRTT, totalJitter time.Duration + 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] + } + } + 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 + } + if arrival > nextService { + nextService = arrival + } + nextService += serviceInterval + virtualEnd = nextService + if _, _, err := processQualificationPayload(uint32(index), payload); err != nil { + return qualificationImpairmentObservation{}, err + } + observation.Delivered++ + } + 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 + } + 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) { + payload := qualificationPayload(profile) + if err := runQualificationWarmup(profile, payload); err != nil { + return qualificationProcessingSummary{}, err + } + file, err := os.OpenFile(rawPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) + if err != nil { + return qualificationProcessingSummary{}, err + } + compressed, err := gzip.NewWriterLevel(file, gzip.BestSpeed) + if err != nil { + _ = file.Close() + return qualificationProcessingSummary{}, err + } + buffered := bufio.NewWriterSize(compressed, 1<<20) + closed := false + defer func() { + if !closed { + _ = buffered.Flush() + _ = compressed.Close() + _ = file.Close() + } + }() + if _, err := buffered.WriteString("elapsed_ns,processing_ns\n"); err != nil { + return qualificationProcessingSummary{}, err + } + bytesPerSecond := profile.BitrateKbps * 1000 / 8 + 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() + var processed int64 + for processed < targetPackets { + elapsed := time.Since(started) + expected := targetPackets + if elapsed < profile.Duration { + expected = targetPackets * elapsed.Nanoseconds() / profile.Duration.Nanoseconds() + } + 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 + } + 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 + } + } + actualDuration := time.Since(started) + if err := buffered.Flush(); err != nil { + return qualificationProcessingSummary{}, err + } + if err := compressed.Close(); err != nil { + return qualificationProcessingSummary{}, err + } + if err := file.Close(); err != nil { + return qualificationProcessingSummary{}, err + } + closed = true + summary, err := summarizeQualificationSamples(samples) + if err != nil { + return qualificationProcessingSummary{}, err + } + sum, size, err := qualificationFileSHA256(rawPath) + if err != nil { + return qualificationProcessingSummary{}, err + } + summary.Profile = profile.Name + summary.Codec = profile.Codec + summary.ConfiguredBitrateKbps = profile.BitrateKbps + summary.ObservedBitrateKbps = float64(processed*int64(profile.PacketBytes)*8) / actualDuration.Seconds() / 1000 + summary.Warmup = profile.Warmup + summary.ConfiguredDuration = profile.Duration + summary.ActualDuration = actualDuration + summary.ClockOverhead = qualificationClockOverhead() + summary.PayloadSHA256 = fmt.Sprintf("%x", sha256.Sum256(payload)) + summary.RawSamples = filepath.Base(rawPath) + summary.RawSamplesSHA256 = sum + summary.RawSamplesBytes = size + if summary.ObservedBitrateKbps < float64(profile.BitrateKbps)*0.95 { + return qualificationProcessingSummary{}, fmt.Errorf("observed bitrate %.2f below profile %d", summary.ObservedBitrateKbps, profile.BitrateKbps) + } + if err := enforceQualificationProcessingGate(summary); err != nil { + return qualificationProcessingSummary{}, err + } + return summary, nil +} + +func runQualificationWarmup(profile qualificationMediaProfile, payload []byte) error { + started := time.Now() + var sequence uint32 + for time.Since(started) < profile.Warmup { + if _, _, err := processQualificationPayload(sequence, payload); err != nil { + return err + } + sequence++ + } + return nil +} + +func qualificationClockOverhead() time.Duration { + samples := make([]time.Duration, 10_000) + for index := range samples { + started := time.Now() + samples[index] = time.Since(started) + } + summary, _ := summarizeQualificationSamples(samples) + return summary.Median +} + +func qualificationFileSHA256(path string) (string, int64, error) { + file, err := os.Open(path) + if err != nil { + return "", 0, err + } + defer file.Close() + hash := sha256.New() + size, err := io.Copy(hash, file) + if err != nil { + return "", 0, err + } + return hex.EncodeToString(hash.Sum(nil)), size, nil +} + +func qualificationPacerEvidence() (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) + evidence := qualificationFairnessEvidence{ + Evaluation: 60 * time.Second, PerFlowBytes: make(map[string]int64, len(flows)), + ShareError: make(map[string]float64, len(flows)), + } + var total, squares float64 + for _, delivery := range baseline { + evidence.PerFlowBytes[delivery.flow] += delivery.bytes + } + for _, flow := range flows { + total += float64(evidence.PerFlowBytes[flow]) + squares += float64(evidence.PerFlowBytes[flow]) * float64(evidence.PerFlowBytes[flow]) + } + target := total / float64(len(flows)) + for _, flow := range flows { + evidence.ShareError[flow] = math.Abs(float64(evidence.PerFlowBytes[flow])-target) / target + if evidence.ShareError[flow] > 0.10 { + return qualificationFairnessEvidence{}, fmt.Errorf("flow %s share error %.4f", flow, evidence.ShareError[flow]) + } + } + evidence.JainIndex = total * total / (float64(len(flows)) * squares) + for _, step := range []struct { + reduction int + kbps int64 + start time.Time + end time.Time + cap int64 + }{ + {25, 6000, start.Add(60 * time.Second), start.Add(70 * time.Second), 750_000}, + {50, 4000, start.Add(70 * time.Second), start.Add(80 * time.Second), 500_000}, + } { + pacer.setKbps(step.kbps) + deliveries := runSyntheticPacer(pacer, step.start, step.end, flows, next) + convergence := qualificationPacerConvergence(deliveries, step.start, flows) + 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) + } + evidence.CapacitySteps = append(evidence.CapacitySteps, qualificationCapacityStep{ + ReductionPercent: step.reduction, Convergence: convergence, + MaximumFiveSecond: maximum, FiveSecondCap: step.cap * 5, + }) + } + 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 + } + } + 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 +} + +func qualificationMaximumFiveSecondBytes(deliveries []syntheticPacerDelivery) int64 { + sort.Slice(deliveries, func(first, second int) bool { return deliveries[first].at.Before(deliveries[second].at) }) + var maximum, total int64 + for first, last := 0, 0; first < len(deliveries); first++ { + for last < len(deliveries) && deliveries[last].at.Sub(deliveries[first].at) <= 5*time.Second { + total += deliveries[last].bytes + last++ + } + if total > maximum { + maximum = total + } + total -= deliveries[first].bytes + } + return maximum +} + +func qualificationProtocolVersion() string { + info, ok := debug.ReadBuildInfo() + if !ok { + return "unknown" + } + for _, dependency := range info.Deps { + if dependency.Path == "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol" { + return dependency.Version + } + } + return "unknown" +} + +func qualificationCandidateCommit() (string, error) { + commit := os.Getenv("VERSEVDI_QUALIFICATION_COMMIT") + decoded, err := hex.DecodeString(commit) + if err != nil || len(decoded) != 20 || strings.ToLower(commit) != commit { + return "", errors.New("VERSEVDI_QUALIFICATION_COMMIT must be a lowercase full SHA-1") + } + return commit, 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 { + return err + } + encoder := json.NewEncoder(file) + encoder.SetIndent("", " ") + if err := encoder.Encode(value); err != nil { + _ = file.Close() + return err + } + return file.Close() +} + +func TestSection7Qualification(t *testing.T) { + output := os.Getenv("VERSEVDI_QUALIFICATION_DIR") + if output == "" { + t.Skip("set VERSEVDI_QUALIFICATION_DIR to run the 30-minute frozen-candidate qualification") + } + if err := validateQualificationOutputDir(output); err != nil { + t.Fatal(err) + } + commit, err := qualificationCandidateCommit() + if err != nil { + t.Fatal(err) + } + if err := os.Mkdir(output, 0o750); err != nil { + t.Fatalf("create new qualification evidence directory: %v", err) + } + started := time.Now().UTC() + media := qualificationMediaProfiles() + qualificationTraverseProfiles(t, media) + manifest := qualificationManifest{ + Status: "running", ToolVersion: qualificationToolVersion, + Command: fmt.Sprintf("VERSEVDI_QUALIFICATION_DIR=%s VERSEVDI_QUALIFICATION_COMMIT=%s GOWORK=off go test ./gateway -run '^TestSection7Qualification$' -count=1 -timeout 45m -v", output, commit), + CandidateCommit: commit, ProtocolVersion: qualificationProtocolVersion(), + 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", + 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"}, + 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) + 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() + if err != nil { + t.Fatal(err) + } + manifest.Fairness = fairness + for _, impairment := range qualificationImpairmentProfiles() { + profiles := media[:1] + if impairment.Name == "baseline" { + profiles = media + } + for _, profile := range profiles { + observation, runErr := runQualificationImpairment(impairment, profile, qualificationImpairmentPacketCount) + if runErr != nil { + t.Fatal(runErr) + } + if impairment.Name == "constrained" { + observation.CapacityStepObservations = append([]qualificationCapacityStep(nil), fairness.CapacitySteps...) + } + manifest.Impairments = append(manifest.Impairments, observation) + } + } + manifest.Status = "passed" + manifest.CompletedAt = time.Now().UTC().Format(time.RFC3339Nano) + manifestPath := filepath.Join(output, "manifest.json") + if err := writeQualificationJSON(manifestPath, manifest); err != nil { + t.Fatal(err) + } + manifestBytes, err := os.ReadFile(manifestPath) + if err != nil { + t.Fatal(err) + } + for _, forbidden := range []string{"private_key", "client_private", "clipboard text", "apollo.test", "provider endpoint", "live interoperability passed"} { + if bytes.Contains(bytes.ToLower(manifestBytes), []byte(forbidden)) { + t.Fatalf("qualification manifest contains forbidden boundary text %q", forbidden) + } + } + t.Logf("qualification manifest: %s", manifestPath) +} + +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 + } + } + cancel() + } + _ = harness.client.Close() + harness.waitReleased(t) +} diff --git a/openspec/changes/phase3c-gateway-qualification/.openspec.yaml b/openspec/changes/phase3c-gateway-qualification/.openspec.yaml new file mode 100644 index 0000000..f205fc7 --- /dev/null +++ b/openspec/changes/phase3c-gateway-qualification/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-07-29 diff --git a/openspec/changes/phase3c-gateway-qualification/design.md b/openspec/changes/phase3c-gateway-qualification/design.md new file mode 100644 index 0000000..25b0761 --- /dev/null +++ b/openspec/changes/phase3c-gateway-qualification/design.md @@ -0,0 +1,54 @@ +## Context + +P3C-029 through P3C-033 require raw, reproducible processing, payload-rate, +fairness, cap, and impairment evidence. Existing focused tests cover the +framer, bounded queues, native Apollo fake, and fair pacer, but do not emit the +normative ten-minute or six-profile artifacts. + +## Goals / Non-Goals + +**Goals:** + +- Run the three 20/50/80 Mbps encoded-payload profiles for ten wall-clock + minutes each after a recorded warm-up. +- Measure the existing gateway framing path with a monotonic clock and retain + compressed raw latency samples plus full summary statistics. +- Run the exact six Section 7.2 configurations once using a deterministic, + bounded virtual packet discipline and retain configured and observed values. +- Exercise one real mTLS/QUIC fake-provider traversal for every media profile + and reuse the real fair-pacer implementation for fairness and cap evidence. + +**Non-Goals:** + +- Live Apollo/macOS/firewall qualification, real encoder fidelity, codec + processing, host network mutation, or multi-host scale. +- A production impairment framework, new gateway API, dependency, cgo, or + sidecar. + +## Decisions + +- Implement the harness as an opt-in `go test` in package `gateway`. This + keeps qualification access to the actual unexported fair pacer without + adding a production API. Normal suites skip the long run unless an explicit + absolute evidence directory is supplied. +- Use wall-clock duration and target-rate pacing for performance profiles. + Measure only receive-to-framed-payload processing; pacing wait and raw-file + writes stay outside the measured interval and are reported separately. +- Stream every raw sample into gzip-compressed CSV while retaining one bounded + duration slice per profile for exact percentiles. +- Use a fixed-seed virtual FIFO for impairment. It records no host claim and + identifies its queue discipline and deterministic topology explicitly. +- Treat any payload mutation, p95 above 5 ms, catalog mismatch, fairness error + above 10%, cap excess above 5%, or step convergence beyond ten seconds as a + hard command failure. + +## Risks / Trade-offs + +- [Local virtual impairment cannot prove deployed route behavior] → label every + artifact deterministic and retain live Apollo/macOS/firewall as + deferred-owner-e2e. +- [Raw samples can be large] → stream gzip output and bound in-memory samples + to the exact profile packet budget. +- [Host load can invalidate latency] → record OS, architecture, Go version, + timing overhead, actual duration, packet count, and observed bitrate; fail + rather than substitute configured capacity for measured egress. diff --git a/openspec/changes/phase3c-gateway-qualification/proposal.md b/openspec/changes/phase3c-gateway-qualification/proposal.md new file mode 100644 index 0000000..0ba9d7a --- /dev/null +++ b/openspec/changes/phase3c-gateway-qualification/proposal.md @@ -0,0 +1,35 @@ +## Why + +The Phase 3C gateway candidate has deterministic transport and scheduler tests +but no executable artifact generator for the normative ten-minute media +measurements and six bounded impairment profiles. Without that evidence, +P3C-029 through P3C-033 cannot be frozen truthfully. + +## What Changes + +- Add one stdlib-only qualification command for the three fixed encoded-media + profiles and the exact six Section 7.2 impairment profiles. +- Emit bounded machine-readable configuration, raw observations, summaries, + environment, topology, direction, queue discipline, and tool version. +- Fail the command when payload integrity, the 5 ms processing p95, impairment + bounds, fairness, capacity-step convergence, or aggregate cap gates fail. +- Keep live Apollo, macOS, physical firewall, real encoder fidelity, and real + multi-host scale explicitly deferred-owner-e2e. + +## Capabilities + +### New Capabilities + +- `gateway-qualification`: Deterministic P3C-029 through P3C-033 media, + processing, fairness, cap, and impairment evidence generation. + +### Modified Capabilities + +None. + +## Impact + +The Data Plane gains a qualification-only Go command, focused tests, and +documented evidence output. It adds no dependency, production transport +abstraction, provider route, codec operation, cgo, sidecar, or Connection +Server code. diff --git a/openspec/changes/phase3c-gateway-qualification/specs/gateway-qualification/spec.md b/openspec/changes/phase3c-gateway-qualification/specs/gateway-qualification/spec.md new file mode 100644 index 0000000..635b6dc --- /dev/null +++ b/openspec/changes/phase3c-gateway-qualification/specs/gateway-qualification/spec.md @@ -0,0 +1,63 @@ +## ADDED Requirements + +### Requirement: Fixed media processing qualification +The qualification harness SHALL run 1080p60 H.264 at 20 Mbps, 1440p120 HEVC +at 50 Mbps, and 4K60 HEVC at 80 Mbps for ten wall-clock minutes each after a +recorded warm-up. It SHALL preserve encoded payload bytes, record every +monotonic processing sample, report count, min, median, p90, p95, p99, max, +mean, standard deviation, timing overhead, and observed bitrate, and fail when +any p95 exceeds 5 ms. + +#### Scenario: Healthy fixed profile +- **WHEN** a frozen candidate runs one fixed profile for the normative duration +- **THEN** the harness emits compressed raw samples and a summary tied to the + exact source commit, Protocol version, environment, and payload hash. + +#### Scenario: Processing gate failure +- **WHEN** 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. Baseline SHALL cover all three +media profiles and the other profiles SHALL cover 1080p60. Each artifact SHALL +record tool version, exact command/configuration, 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 observed rather than configured statistics. + +#### 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 the simulation. + +### 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, report every +share error and Jain's fairness index, and fail above 10% share error. It SHALL +apply 25% and 50% capacity steps, 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 per-flow bytes, share errors, Jain's index, + step 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, raw media content, or claim of live +Apollo/macOS/firewall interoperability. The harness SHALL add no codec +operation, production dependency, cgo, sidecar, or direct provider route. + +#### Scenario: Deterministic evidence publication +- **WHEN** qualification completes +- **THEN** the manifest labels fake-provider, virtual impairment, and local + processing evidence separately and leaves live interoperability + deferred-owner-e2e. diff --git a/openspec/changes/phase3c-gateway-qualification/tasks.md b/openspec/changes/phase3c-gateway-qualification/tasks.md new file mode 100644 index 0000000..a80822d --- /dev/null +++ b/openspec/changes/phase3c-gateway-qualification/tasks.md @@ -0,0 +1,24 @@ +## 1. Contract and focused regressions + +- [x] 1.1 Add fixed catalog tests for the three media profiles, ten-minute + duration, exact six impairment profiles, and bounded output configuration. +- [x] 1.2 Add summary, payload-integrity, fairness, cap, and failure-threshold + tests before implementing the harness. + +## 2. Qualification harness + +- [x] 2.1 Implement opt-in real-duration processing measurement with compressed + raw samples, full statistics, timing overhead, and environment metadata. +- [x] 2.2 Implement deterministic bounded impairment observations and reuse the + production fair pacer for fairness and capacity-step evidence. +- [x] 2.3 Add one mTLS/QUIC fake-provider traversal per fixed encoded profile + and prove the artifact boundary contains no provider route or secret. + +## 3. Freeze and evidence + +- [x] 3.1 Run focused red/green checks, strict OpenSpec validation, `make + verify`, race/fuzz/resource checks, and freeze the harness commit. +- [ ] 3.2 Run the opt-in ten-minute and six-profile command exactly once + against the frozen candidate and archive raw artifacts and hashes. +- [ ] 3.3 Sync the canonical specification, archive the completed change, and + revalidate strictly without claiming live Apollo/macOS/firewall evidence.