package gateway import ( "bufio" "bytes" "compress/gzip" "context" "crypto/sha256" "encoding/binary" "encoding/hex" "encoding/json" "errors" "fmt" "io" "math" "os" "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/v2" qualificationImpairmentQueuePackets = 64 qualificationImpairmentMaxPackets = 100_000 qualificationImpairmentPacketCount = 10_000 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 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 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"` 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 { 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"` RawSamples string `json:"raw_samples"` RawSamplesSHA256 string `json:"raw_samples_sha256"` RawSamplesBytes int64 `json:"raw_samples_bytes"` } 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 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"` 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 { 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"` ToolVersions map[string]string `json:"tool_versions"` 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 newQualificationPath(t *testing.T, pacerKbps int64) *qualificationPath { t.Helper() serverTLS, clientTLS := testTLS(t) key := []byte("0123456789abcdef") media, err := newApolloMediaCodec(key, 7) if err != nil { t.Fatal(err) } 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 } 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) } } t.Cleanup(path.Close) return path } func (p *qualificationPath) Close() { if p != nil { p.closeOnce.Do(p.shutdown) } } 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) { 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(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 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...), } path := newQualificationPath(t, media.BitrateKbps) defer path.Close() started := time.Now() payload := qualificationPayload(media) 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++ { jitter := time.Duration(0) if profile.Jitter > 0 { width := uint64(profile.Jitter*2 + 1) jitter = time.Duration(random()%width) - profile.Jitter } if float64(random()%10_000) < profile.LossPercent*100 { observation.Dropped++ if _, err := fmt.Fprintf(buffered, "%d,%d,0,0,dropped,0,0\n", index, time.Duration(index)*spacing); err != nil { return qualificationImpairmentObservation{}, err } continue } packet := pendingPacket{index: index, jitter: jitter} if profile.Reorder && index%20 == 18 { pending = packet continue } if err := deliver(packet); err != nil { return qualificationImpairmentObservation{}, err } 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.ObservedLossPercent = float64(observation.Dropped) * 100 / float64(packetCount) observation.ObservedReorderPercent = float64(observation.ObservedOutOfOrder) * 100 / float64(packetCount) observation.RawSamples = filepath.Base(rawPath) observation.RawSamplesSHA256, observation.RawSamplesBytes, err = qualificationFileSHA256(rawPath) if err != nil { return qualificationImpairmentObservation{}, err } 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 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 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) 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() resources := []qualificationResourceSample{qualificationRuntimeSample(started)} lastResourceSample := started var processed int64 for processed < targetPackets { batch := int64(8) if remaining := targetPackets - processed; remaining < batch { batch = remaining } 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 += 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 } 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 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) } if err := enforceQualificationProcessingGate(summary); err != nil { return qualificationProcessingSummary{}, err } return summary, nil } func runQualificationWarmup(t *testing.T, path *qualificationPath, profile qualificationMediaProfile, payload []byte) error { t.Helper() started := time.Now() for time.Since(started) < profile.Warmup { trace, _, err := path.traverse(t, payload) if err != nil { return err } 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 { 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(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)), } 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) 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) } evidence.CapacitySteps = append(evidence.CapacitySteps, qualificationCapacityStep{ ReductionPercent: step.reduction, Convergence: convergence, 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, 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 } } return 11 * time.Second } 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 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) if err != nil || !valid { return "", errors.New("VERSEVDI_QUALIFICATION_PROTOCOL_VERSION must be an immutable prerelease tag") } return version, nil } 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 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 { 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) } protocolVersion, err := qualificationProtocolVersion() 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) } 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 VERSEVDI_QUALIFICATION_PROTOCOL_VERSION=%s GOWORK=off go test ./gateway -run '^TestSection7Qualification$' -count=1 -timeout 45m -v", output, commit, protocolVersion, ), CandidateCommit: commit, ProtocolVersion: protocolVersion, StartedAt: started.Format(time.RFC3339Nano), GoVersion: runtime.Version(), 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: "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(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(filepath.Join(output, "fairness.csv.gz")) 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 { 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) } 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() for _, profile := range profiles { trace := qualificationProductionPathSmoke(t, profile) if !trace.ApolloRecovered || !trace.ProductionQueue || !trace.ProductionPacer || !trace.VerseQUIC || !trace.PayloadPreserved { t.Fatalf("%s production path: %#v", profile.Name, trace) } } }