fix(gateway): sustain qualified media traversal

gateway/apollo_native.go

@@ -28,7 +28,7 @@ import (
 )
 
 const (
-	nativeApolloVideoQueuePackets = 64
+	nativeApolloVideoQueuePackets = 256
 	nativeApolloAudioQueuePackets = 16
 	nativeApolloEventQueuePackets = 16
 )

gateway/qualification_contract_test.go

@@ -109,7 +109,8 @@ func TestQualificationShortProcessingWritesRawArtifact(t *testing.T) {
 	}
 	if summary.Count < 1 || summary.RawSamplesSHA256 == "" || summary.RawSamplesBytes < 1 ||
 		summary.ResourceSamples < 2 || summary.RawResourcesSHA256 == "" || summary.RawResourcesBytes < 1 ||
-		summary.CPUScope != qualificationGatewayCPUScope || summary.ClockOverhead <= 0 || summary.ClockMethod == "" {
+		summary.CPUScope != qualificationGatewayCPUScope || summary.ResourceMethod != qualificationResourceMethod ||
+		summary.ClockOverhead <= 0 || summary.ClockMethod == "" {
 		t.Fatalf("processing summary = %#v", summary)
 	}
 	file, err := os.Open(rawPath)
@@ -151,6 +152,20 @@ func TestQualificationShortProcessingSubprocessCoversFixedProfiles(t *testing.T)
 	}
 }
 
+func TestQualificationSustainedProcessingKeepsCleanPathBounded(t *testing.T) {
+	profile := qualificationMediaProfiles()[2]
+	profile.Duration = 2 * time.Minute
+	profile.Warmup = 100 * time.Millisecond
+	summary, err := runQualificationProcessing(t, profile, filepath.Join(t.TempDir(), "4k60-hevc.csv.gz"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if summary.Count < 1 || summary.ObservedBitrateKbps < float64(profile.BitrateKbps)*0.95 ||
+		summary.ObservedBitrateKbps > float64(profile.BitrateKbps)*1.05 {
+		t.Fatalf("sustained processing summary = %#v", summary)
+	}
+}
+
 func TestQualificationProcessingResourcesExcludeParentDriverCPU(t *testing.T) {
 	profile := qualificationMediaProfile{
 		Name: "resource-isolation", Codec: "h264", BitrateKbps: 1000,
@@ -429,11 +444,11 @@ func TestQualificationGatewaySubprocessResourcesResetAndTrackWork(t *testing.T)
 	if idle.CPUSeconds > 50*time.Millisecond.Seconds() || secondIdle.CPUSeconds > 50*time.Millisecond.Seconds() {
 		t.Fatalf("idle gateway CPU was reported as consumed work: first=%.6fs second=%.6fs", idle.CPUSeconds, secondIdle.CPUSeconds)
 	}
-	if work.CPUSeconds <= idle.CPUSeconds || work.Count != 500 || work.Mallocs == 0 ||
+	if work.CPUSeconds <= idle.CPUSeconds || work.Count != 500 || work.AllocatedObjects == 0 ||
 		work.AllocatedBytes == 0 || work.PeakHeapBytes == 0 || work.PeakGoroutines == 0 {
 		t.Fatalf("gateway work resource sample = %#v idle=%#v", work, idle)
 	}
-	if secondIdle.Mallocs >= work.Mallocs || secondIdle.AllocatedBytes >= work.AllocatedBytes {
+	if secondIdle.AllocatedObjects >= work.AllocatedObjects || secondIdle.AllocatedBytes >= work.AllocatedBytes {
 		t.Fatalf("successive recording inherited counters: work=%#v second=%#v", work, secondIdle)
 	}
 	if work.ClockOverhead <= 0 || work.ClockMethod != qualificationClockOverheadMethod {

gateway/qualification_harness_test.go

@@ -25,6 +25,7 @@ import (
 	"regexp"
 	"runtime"
 	"runtime/debug"
+	runtimemetrics "runtime/metrics"
 	"sort"
 	"strconv"
 	"strings"
@@ -38,14 +39,15 @@ import (
 )
 
 const (
-	qualificationToolVersion                   = "versevdi-gateway-qualification/v5"
-	qualificationImpairmentQueuePackets        = 64
+	qualificationToolVersion                   = "versevdi-gateway-qualification/v6"
+	qualificationImpairmentQueuePackets        = 256
 	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"
 )
 
 type qualificationMediaProfile struct {
@@ -122,9 +124,10 @@ type qualificationProcessingSummary struct {
 	ResourceSamples       int            `json:"resource_samples"`
 	CPUSeconds            float64        `json:"cpu_seconds"`
 	CPUScope              string         `json:"cpu_scope"`
+	ResourceMethod        string         `json:"resource_method"`
 	PeakHeapBytes         uint64         `json:"peak_heap_bytes"`
 	PeakGoroutines        int            `json:"peak_goroutines"`
-	Mallocs               uint64         `json:"mallocs"`
+	AllocatedObjects      uint64         `json:"allocated_objects"`
 	AllocatedBytes        uint64         `json:"allocated_bytes"`
 	RawResources          string         `json:"raw_resources"`
 	RawResourcesSHA256    string         `json:"raw_resources_sha256"`
@@ -183,12 +186,12 @@ type qualificationCapacityStep struct {
 }
 
 type qualificationResourceSample struct {
-	Elapsed    time.Duration
-	CPUSeconds float64
-	HeapBytes  uint64
-	Goroutines int
-	Mallocs    uint64
-	Allocated  uint64
+	Elapsed          time.Duration
+	CPUSeconds       float64
+	HeapBytes        uint64
+	Goroutines       int
+	AllocatedObjects uint64
+	AllocatedBytes   uint64
 }
 
 type qualificationDeliverySample struct {
@@ -1720,10 +1723,11 @@ func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile,
 	summary.RawResourcesBytes = resourceSize
 	summary.ResourceSamples = record.ResourceSamples
 	summary.CPUScope = qualificationGatewayCPUScope
+	summary.ResourceMethod = qualificationResourceMethod
 	summary.CPUSeconds = record.CPUSeconds
 	summary.PeakHeapBytes = record.PeakHeapBytes
 	summary.PeakGoroutines = record.PeakGoroutines
-	summary.Mallocs = record.Mallocs
+	summary.AllocatedObjects = record.AllocatedObjects
 	summary.AllocatedBytes = record.AllocatedBytes
 	if summary.CPUSeconds < 0 || summary.ClockOverhead <= 0 || summary.ClockMethod == "" {
 		return qualificationProcessingSummary{}, errors.New("process CPU usage unavailable")
@@ -1813,12 +1817,17 @@ func qualificationRuntimeSample(started time.Time) qualificationResourceSample {
 		cpuSeconds = float64(usage.Utime.Sec+usage.Stime.Sec) +
 			float64(usage.Utime.Usec+usage.Stime.Usec)/1_000_000
 	}
-	var memory runtime.MemStats
-	runtime.ReadMemStats(&memory)
+	samples := []runtimemetrics.Sample{
+		{Name: "/memory/classes/heap/objects:bytes"},
+		{Name: "/sched/goroutines:goroutines"},
+		{Name: "/gc/heap/allocs:objects"},
+		{Name: "/gc/heap/allocs:bytes"},
+	}
+	runtimemetrics.Read(samples)
 	return qualificationResourceSample{
 		Elapsed: time.Since(started), CPUSeconds: cpuSeconds,
-		HeapBytes: memory.HeapAlloc, Goroutines: runtime.NumGoroutine(),
-		Mallocs: memory.Mallocs, Allocated: memory.TotalAlloc,
+		HeapBytes: samples[0].Value.Uint64(), Goroutines: int(samples[1].Value.Uint64()),
+		AllocatedObjects: samples[2].Value.Uint64(), AllocatedBytes: samples[3].Value.Uint64(),
 	}
 }
 
@@ -1829,10 +1838,10 @@ func writeQualificationResourceSamples(path string, samples []qualificationResou
 	}
 	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 {
+	if _, err = buffered.WriteString("elapsed_ns,cpu_seconds,heap_object_bytes,goroutines,allocated_objects,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 {
+				sample.CPUSeconds, sample.HeapBytes, sample.Goroutines, sample.AllocatedObjects, sample.AllocatedBytes); err != nil {
 				break
 			}
 		}
@@ -2201,7 +2210,7 @@ func TestSection7Qualification(t *testing.T) {
 		OS:           runtime.GOOS, Architecture: runtime.GOARCH,
 		Topology:        "parent source-shaped encrypted Apollo fixture -> isolated gateway subprocess for processing/resource evidence -> public Verse client decoder; impairment uses the same native recovery/FEC, bounded queue, production pacer, framing, and QUIC path",
 		Direction:       "provider_to_client",
-		QueueDiscipline: "ordered fixed-seed source delay queue with one-serialization-interval catch-up, bounded 64-packet native video queue, production equal-tier fair pacer",
+		QueueDiscipline: "ordered fixed-seed source delay queue with one-serialization-interval catch-up, bounded 256-packet native video queue, production equal-tier fair pacer",
 		Evidence:        []string{"deterministic source-shaped Apollo recovery", "isolated gateway-process resources", "local real-time production path", "mTLS/QUIC fixture transport", "attributed path impairment", "production fair pacer"},
 		Deferred:        []string{"live Apollo", "macOS client", "physical firewall and packet route", "real encoder fidelity", "multi-host scale"},
 	}

gateway/qualification_process_test.go

@@ -19,6 +19,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -71,7 +72,7 @@ type qualificationProcessRecordResult struct {
 	CPUSeconds       float64
 	PeakHeapBytes    uint64
 	PeakGoroutines   int
-	Mallocs          uint64
+	AllocatedObjects uint64
 	AllocatedBytes   uint64
 	RecordingElapsed time.Duration
 }
@@ -162,11 +163,19 @@ func (r *qualificationProcessRecorder) observe(observation mediaTimingObservatio
 
 func (r *qualificationProcessRecorder) writeTimings() {
 	defer close(r.timingDone)
+	line := make([]byte, 0, 96)
 	for sample := range r.timingSamples {
 		if r.recordErr == nil {
-			_, r.recordErr = fmt.Fprintf(r.rawBuffered, "%d,%d,%d,%d\n",
-				sample.elapsed.Nanoseconds(), sample.observation.QueueDelay.Nanoseconds(),
-				sample.observation.ProcessingDelay.Nanoseconds(), sample.observation.PacingDelay.Nanoseconds())
+			line = line[:0]
+			line = strconv.AppendInt(line, sample.elapsed.Nanoseconds(), 10)
+			line = append(line, ',')
+			line = strconv.AppendInt(line, sample.observation.QueueDelay.Nanoseconds(), 10)
+			line = append(line, ',')
+			line = strconv.AppendInt(line, sample.observation.ProcessingDelay.Nanoseconds(), 10)
+			line = append(line, ',')
+			line = strconv.AppendInt(line, sample.observation.PacingDelay.Nanoseconds(), 10)
+			line = append(line, '\n')
+			_, r.recordErr = r.rawBuffered.Write(line)
 		}
 		r.samples++
 	}
@@ -233,8 +242,8 @@ func (r *qualificationProcessRecorder) stop() (qualificationProcessRecordResult,
 	}
 	first, last := resources[0], resources[len(resources)-1]
 	result.CPUSeconds = max(last.CPUSeconds-first.CPUSeconds, 0)
-	result.Mallocs = last.Mallocs - first.Mallocs
-	result.AllocatedBytes = last.Allocated - first.Allocated
+	result.AllocatedObjects = last.AllocatedObjects - first.AllocatedObjects
+	result.AllocatedBytes = last.AllocatedBytes - first.AllocatedBytes
 	for _, sample := range resources {
 		result.PeakHeapBytes = max(result.PeakHeapBytes, sample.HeapBytes)
 		result.PeakGoroutines = max(result.PeakGoroutines, sample.Goroutines)