fix(gateway): recover bounded pacing debt

gateway/fair_pacer_test.go

@@ -41,6 +41,83 @@ func TestFairPacerBoundsCatchupAfterHostStall(t *testing.T) {
 	if next.Before(resumed.Add(-fairPacerMaximumCatchup)) || next.After(resumed.Add(10*time.Millisecond)) {
 		t.Fatalf("post-stall reservation = %s, want bounded catchup near %s", next, resumed)
 	}
+	pacer.mu.Lock()
+	debt := pacer.flows["one"].debt
+	pacer.mu.Unlock()
+	if debt <= 0 || debt > nativeApolloVideoQueueLatency-fairPacerMaximumCatchup {
+		t.Fatalf("post-stall debt = %s, want bounded valid schedule debt", debt)
+	}
+}
+
+func TestFairPacerRepaysBoundedDebtAfterHostStall(t *testing.T) {
+	start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
+	pacer := newFairPacer(8000)
+	next := make(map[string]time.Time)
+	deliveries := runSyntheticPacerWithStall(
+		pacer, start, start.Add(6*time.Second), []string{"one"}, next,
+		start.Add(time.Second), 100*time.Millisecond,
+	)
+	if total := syntheticDeliveryBytes(deliveries); total < 5_990_000 || total > 6_010_000 {
+		t.Fatalf("post-stall delivery bytes = %d, want nominal throughput after bounded debt repayment", total)
+	}
+	pacer.mu.Lock()
+	remaining := pacer.flows["one"].debt
+	pacer.mu.Unlock()
+	if remaining != 0 {
+		t.Fatalf("post-stall debt = %s after repayment, want zero", remaining)
+	}
+	assertSyntheticCap(t, deliveries, 1_000_000)
+	t.Logf("single-flow debt repaid: bytes=%d remaining=%s", syntheticDeliveryBytes(deliveries), remaining)
+}
+
+func TestFairPacerRepaysSimultaneousEightFlowDebtAcrossCapacitySteps(t *testing.T) {
+	start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
+	flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
+
+	tests := []struct {
+		name           string
+		kbps           int64
+		bytesPerSecond int64
+		minimumBytes   int64
+	}{
+		{name: "baseline", kbps: 8000, bytesPerSecond: 1_000_000, minimumBytes: 9_980_000},
+		{name: "quarter", kbps: 6000, bytesPerSecond: 750_000, minimumBytes: 7_480_000},
+		{name: "half", kbps: 4000, bytesPerSecond: 500_000, minimumBytes: 4_980_000},
+	}
+	for _, test := range tests {
+		pacer := newFairPacer(8000)
+		next := make(map[string]time.Time, len(flows))
+		_ = runSyntheticPacer(pacer, start, start.Add(time.Second), flows, next)
+		resumed := start.Add(1100 * time.Millisecond)
+		for _, flow := range flows {
+			next[flow] = pacer.reserveAt(resumed, flow, 1000)
+		}
+		assertSyntheticDebt(t, pacer, flows, true)
+		pacer.setKbps(test.kbps)
+		deliveries := runSyntheticPacer(pacer, resumed, resumed.Add(10*time.Second), flows, next)
+		if total := syntheticDeliveryBytes(deliveries); total < test.minimumBytes {
+			t.Fatalf("%s post-stall delivery bytes = %d, want at least %d", test.name, total, test.minimumBytes)
+		}
+		assertSyntheticFairness(t, deliveries, flows)
+		assertSyntheticCap(t, deliveries, test.bytesPerSecond)
+		assertSyntheticDebt(t, pacer, flows, false)
+		t.Logf("%s eight-flow debt repaid: bytes=%d cap=%d", test.name, syntheticDeliveryBytes(deliveries), test.bytesPerSecond*5*105/100)
+	}
+}
+
+func assertSyntheticDebt(t *testing.T, pacer *fairPacer, flows []string, wantDebt bool) {
+	t.Helper()
+	pacer.mu.Lock()
+	defer pacer.mu.Unlock()
+	for _, flow := range flows {
+		debt := pacer.flows[flow].debt
+		if wantDebt && (debt <= 0 || debt > nativeApolloVideoQueueLatency-fairPacerMaximumCatchup) {
+			t.Fatalf("flow %s active debt = %s, want bounded nonzero debt", flow, debt)
+		}
+		if !wantDebt && debt != 0 {
+			t.Fatalf("flow %s debt = %s after repayment, want zero", flow, debt)
+		}
+	}
 }
 
 func runSyntheticPacer(pacer *fairPacer, start, end time.Time, flows []string, next map[string]time.Time) []syntheticPacerDelivery {
@@ -67,6 +144,50 @@ func runSyntheticPacer(pacer *fairPacer, start, end time.Time, flows []string, n
 	}
 }
 
+func runSyntheticPacerWithStall(pacer *fairPacer, start, end time.Time, flows []string, next map[string]time.Time, stallAt time.Time, stall time.Duration) []syntheticPacerDelivery {
+	const packetBytes = 1000
+	for _, flow := range flows {
+		if next[flow].IsZero() {
+			next[flow] = pacer.reserveAt(start, flow, packetBytes)
+		}
+	}
+	now := start
+	stalled := false
+	var deliveries []syntheticPacerDelivery
+	for {
+		flow := ""
+		target := end.Add(time.Nanosecond)
+		for _, candidate := range flows {
+			if next[candidate].Before(target) {
+				flow, target = candidate, next[candidate]
+			}
+		}
+		if target.After(end) {
+			return deliveries
+		}
+		if !stalled && !target.Before(stallAt) {
+			now = stallAt.Add(stall)
+			stalled = true
+		}
+		if now.Before(target) {
+			now = target
+		}
+		if now.After(end) {
+			return deliveries
+		}
+		deliveries = append(deliveries, syntheticPacerDelivery{at: now, flow: flow, bytes: packetBytes})
+		next[flow] = pacer.reserveAt(now, flow, packetBytes)
+	}
+}
+
+func syntheticDeliveryBytes(deliveries []syntheticPacerDelivery) int64 {
+	var total int64
+	for _, delivery := range deliveries {
+		total += delivery.bytes
+	}
+	return total
+}
+
 func assertSyntheticFairness(t *testing.T, deliveries []syntheticPacerDelivery, flows []string) {
 	t.Helper()
 	counts := make(map[string]int64, len(flows))

gateway/qualification_contract_test.go

@@ -719,6 +719,184 @@ func TestQualificationCarriesCompleteLargeFramesThroughPublicPath(t *testing.T)
 	}
 }
 
+func TestQualificationPacerRepaysThreeMediaLoopStallsThroughPublicPath(t *testing.T) {
+	profile := qualificationMediaProfiles()[0]
+	const totalFrames = 480
+	barriers := []int64{30, 180, 330}
+	type stall struct {
+		public  chan struct{}
+		blocked chan uint64
+		release chan struct{}
+	}
+	stalls := make([]stall, len(barriers))
+	for index := range stalls {
+		stalls[index] = stall{public: make(chan struct{}), blocked: make(chan uint64, 1), release: make(chan struct{})}
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+	var path *qualificationPath
+	var completed int64
+	var observationMu sync.Mutex
+	var maximumQueueDelay time.Duration
+	observer := func(observation mediaTimingObservation) {
+		observationMu.Lock()
+		maximumQueueDelay = max(maximumQueueDelay, observation.QueueDelay)
+		completed++
+		current := completed
+		observationMu.Unlock()
+		for index, barrier := range barriers {
+			if current != barrier {
+				continue
+			}
+			select {
+			case <-stalls[index].public:
+			case <-ctx.Done():
+				return
+			}
+			baseline := path.session.mediaRecovered.Load()
+			select {
+			case stalls[index].blocked <- baseline:
+			case <-ctx.Done():
+				return
+			}
+			select {
+			case <-stalls[index].release:
+			case <-ctx.Done():
+			}
+			return
+		}
+	}
+	path = newQualificationPathWithNativeBackendAndObserver(
+		t, profile, qualificationFramePacerKbps(profile), nil, NewNativeApolloBackend(), observer,
+	)
+	defer path.Close()
+
+	spacing := time.Second / time.Duration(profile.FPS)
+	started := time.Now().Add(10 * time.Millisecond)
+	producerDone := make(chan error, 1)
+	go func() {
+		var nextRelease time.Time
+		for index := int64(0); index < totalFrames; index++ {
+			release := qualificationBoundedRelease(
+				started.Add(time.Duration(index)*spacing), nextRelease, time.Now(), spacing,
+			)
+			if err := qualificationWaitContext(ctx, release); err != nil {
+				producerDone <- err
+				return
+			}
+			nextRelease = release.Add(spacing)
+			if _, err := path.emit(t, qualificationFramePayload(profile, index)); err != nil {
+				producerDone <- err
+				return
+			}
+		}
+		producerDone <- nil
+	}()
+
+	var expectedVersePackets uint64
+	receiverDone := make(chan error, 1)
+	go func() {
+		for index := int64(0); index < totalFrames; index++ {
+			payload, err := path.receivePayload(ctx)
+			if err != nil {
+				receiverDone <- err
+				return
+			}
+			expected := qualificationFramePayload(profile, index)
+			if !bytes.Equal(payload, expected) {
+				receiverDone <- fmt.Errorf("public payload changed at frame %d", index)
+				return
+			}
+			fragments := (len(payload) + frameV2PayloadSize - 1) / frameV2PayloadSize
+			expectedVersePackets += uint64(fragments)
+			for barrierIndex, barrier := range barriers {
+				if index+1 == barrier {
+					close(stalls[barrierIndex].public)
+				}
+			}
+		}
+		receiverDone <- nil
+	}()
+
+	for index := range stalls {
+		var baseline uint64
+		select {
+		case baseline = <-stalls[index].blocked:
+		case <-ctx.Done():
+			t.Fatalf("media loop did not block at public frame %d: %v", barriers[index], ctx.Err())
+		}
+		target := baseline + 6
+		for path.session.mediaRecovered.Load() < target {
+			select {
+			case <-ctx.Done():
+				t.Fatalf("stall %d recovered %d, want at least %d: %v", index, path.session.mediaRecovered.Load(), target, ctx.Err())
+			default:
+				runtime.Gosched()
+			}
+		}
+		close(stalls[index].release)
+	}
+	if err := <-producerDone; err != nil {
+		t.Fatalf("source fixture: %v", err)
+	}
+	if err := <-receiverDone; err != nil {
+		t.Fatalf("independent client: %v; expected=%d sent=%d source=%d ingress=%d recovered=%d enqueued=%d provider_drops=%d gateway_drops=%d queue_count=%d queue_bytes=%d",
+			err, path.expectedSourceUDP.Load(), path.fixture.sentPackets.Load(), path.sourceUDP.Load(),
+			path.session.mediaIngress.Load(), path.session.mediaRecovered.Load(), path.session.mediaEnqueued.Load(),
+			path.session.Telemetry().MediaDrops, path.server.Metrics().MediaDrops,
+			path.session.mediaQueueMaximum.Load(), path.session.mediaQueueMaximumBytes.Load())
+	}
+	for {
+		observationMu.Lock()
+		observed := completed
+		observationMu.Unlock()
+		if observed == totalFrames {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			t.Fatalf("media observer completed %d frames, want %d: %v", observed, totalFrames, ctx.Err())
+		default:
+			runtime.Gosched()
+		}
+	}
+
+	if source := path.expectedSourceUDP.Load(); source != path.fixture.sentPackets.Load() || source != path.sourceUDP.Load() || source != path.session.mediaIngress.Load() {
+		t.Fatalf("source accounting expected=%d sent=%d source=%d ingress=%d", source, path.fixture.sentPackets.Load(), path.sourceUDP.Load(), path.session.mediaIngress.Load())
+	}
+	metrics := path.server.Metrics()
+	observationMu.Lock()
+	completedFrames := completed
+	queueDelay := maximumQueueDelay
+	observationMu.Unlock()
+	if path.backend.setups.Load() != 1 || path.backend.opens.Load() != 1 || completedFrames != totalFrames ||
+		path.session.mediaRecovered.Load() != totalFrames || path.session.mediaEnqueued.Load() != totalFrames ||
+		metrics.ProcessingSamples != totalFrames || metrics.MediaPackets != expectedVersePackets ||
+		path.server.pacer.reservations.Load() != expectedVersePackets ||
+		path.session.Telemetry().MediaDrops != 0 || path.server.Metrics().MediaDrops != 0 {
+		t.Fatalf("media accounting setup=%d open=%d completed=%d recovered=%d enqueued=%d processing=%d media=%d pacer=%d provider_drops=%d gateway_drops=%d",
+			path.backend.setups.Load(), path.backend.opens.Load(), completedFrames,
+			path.session.mediaRecovered.Load(), path.session.mediaEnqueued.Load(), metrics.ProcessingSamples,
+			metrics.MediaPackets, path.server.pacer.reservations.Load(), path.session.Telemetry().MediaDrops, metrics.MediaDrops)
+	}
+	if path.session.mediaQueueMaximum.Load() > nativeApolloVideoQueuePackets ||
+		path.session.mediaQueueMaximumBytes.Load() > nativeApolloVideoQueueBytes || queueDelay > nativeApolloVideoQueueLatency {
+		t.Fatalf("queue bounds count=%d bytes=%d residence=%s", path.session.mediaQueueMaximum.Load(), path.session.mediaQueueMaximumBytes.Load(), queueDelay)
+	}
+	t.Logf("three-stall public path: frames=%d source=%d ingress=%d recovered=%d enqueued=%d drops=%d/%d queue=%d/%d residence=%s verse_packets=%d",
+		totalFrames, path.sourceUDP.Load(), path.session.mediaIngress.Load(), path.session.mediaRecovered.Load(),
+		path.session.mediaEnqueued.Load(), path.session.Telemetry().MediaDrops, metrics.MediaDrops,
+		path.session.mediaQueueMaximum.Load(), path.session.mediaQueueMaximumBytes.Load(), queueDelay, metrics.MediaPackets)
+
+	path.Close()
+	select {
+	case <-path.session.readDone:
+	case <-time.After(time.Second):
+		t.Fatal("native media workers did not stop")
+	}
+}
+
 func TestQualificationLateSourceBatchDoesNotCollapseThroughPublicPath(t *testing.T) {
 	profile := qualificationMediaProfiles()[2]
 	path := newQualificationPath(t, profile, profile.BitrateKbps)

gateway/qualification_harness_test.go

@@ -1059,6 +1059,10 @@ func newQualificationPathWithImpairment(t *testing.T, profile qualificationMedia
 }
 
 func newQualificationPathWithNativeBackend(t *testing.T, profile qualificationMediaProfile, pacerKbps int64, impairment *qualificationImpairmentProfile, native *NativeApolloBackend) *qualificationPath {
+	return newQualificationPathWithNativeBackendAndObserver(t, profile, pacerKbps, impairment, native, nil)
+}
+
+func newQualificationPathWithNativeBackendAndObserver(t *testing.T, profile qualificationMediaProfile, pacerKbps int64, impairment *qualificationImpairmentProfile, native *NativeApolloBackend, observer func(mediaTimingObservation)) *qualificationPath {
 	t.Helper()
 	serverTLS, clientTLS := testTLS(t)
 	fixture := newQualificationApolloFixture(t, serverTLS, clientTLS, "qualification-session", profile)
@@ -1081,7 +1085,7 @@ func newQualificationPathWithNativeBackend(t *testing.T, profile qualificationMe
 		ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: authority.GatewayID,
 		Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(),
 		Admission: admission, ProviderStateReporter: &recordingProviderStateReporter{},
-		Provider: provider, PacerKbps: pacerKbps,
+		Provider: provider, PacerKbps: pacerKbps, mediaObserver: observer,
 	})
 	if err != nil {
 		t.Fatal(err)

gateway/telemetry.go

@@ -126,6 +126,7 @@ type fairPacer struct {
 type fairPacerFlow struct {
 	next     time.Time
 	lastSeen time.Time
+	debt     time.Duration
 }
 
 const fairPacerMaximumCatchup = 5 * time.Millisecond
@@ -180,9 +181,14 @@ func (p *fairPacer) reserveAt(now time.Time, flow string, bytes int) time.Time {
 		base = now
 	} else if lag := now.Sub(base); lag > fairPacerMaximumCatchup {
 		base = now.Add(-fairPacerMaximumCatchup)
+		state.debt = min(state.debt+lag-fairPacerMaximumCatchup, nativeApolloVideoQueueLatency-fairPacerMaximumCatchup)
 	}
 	numerator := int64(bytes) * int64(len(p.flows)) * int64(time.Second)
 	delay := time.Duration((numerator + p.bytesPerSecond - 1) / p.bytesPerSecond)
+	if repayment := min(delay/21, state.debt); repayment > 0 {
+		delay -= repayment
+		state.debt -= repayment
+	}
 	state.next = base.Add(delay)
 	p.flows[flow] = state
 	return state.next