fix(gateway): bound impairment catchup

gateway/qualification_contract_test.go

@@ -346,6 +346,24 @@ func TestQualificationLossOnlyDoesNotImplicitlyReorder(t *testing.T) {
 	}
 }
 
+func TestQualificationSourceShaperCarriesAtMostOnePacketOfCatchup(t *testing.T) {
+	spacing := time.Millisecond
+	started := time.Unix(0, 0)
+	now := started.Add(100 * spacing)
+	first := qualificationBoundedRelease(started, time.Time{}, now, spacing)
+	if first != now.Add(-spacing) {
+		t.Fatalf("first catch-up release = %s, want %s", first, now.Add(-spacing))
+	}
+	second := qualificationBoundedRelease(started.Add(spacing), first.Add(spacing), now, spacing)
+	if second != now {
+		t.Fatalf("second catch-up release = %s, want %s", second, now)
+	}
+	third := qualificationBoundedRelease(started.Add(2*spacing), second.Add(spacing), now, spacing)
+	if third != now.Add(spacing) {
+		t.Fatalf("catch-up debt was reset: third release = %s, want %s", third, now.Add(spacing))
+	}
+}
+
 func TestQualificationExplicitReorderIsBoundedAndAttributed(t *testing.T) {
 	observation, err := runQualificationImpairment(
 		t,

gateway/qualification_harness_test.go

@@ -280,6 +280,17 @@ func qualificationMediaPacerKbps(profile qualificationMediaProfile, reduction in
 	return (payloadKbps*int64(profile.PacketBytes+frameHeaderSize) + int64(profile.PacketBytes) - 1) / int64(profile.PacketBytes)
 }
 
+func qualificationBoundedRelease(target, next, now time.Time, spacing time.Duration) time.Time {
+	release := target
+	if next.After(release) {
+		release = next
+	}
+	if now.Sub(release) > spacing {
+		return now.Add(-spacing)
+	}
+	return release
+}
+
 type qualificationTracingBackend struct {
 	native   *NativeApolloBackend
 	setups   atomic.Uint64
@@ -1326,20 +1337,13 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
 
 	stepAt := make(map[int]time.Time, len(profile.CapacitySteps))
 	emitted := 0
-	var previousRelease time.Time
+	var nextRelease time.Time
-	maxCatchup := spacing
 	for _, packet := range jobs {
-		release := started.Add(packet.target)
+		release := qualificationBoundedRelease(started.Add(packet.target), nextRelease, time.Now(), spacing)
-		if minimum := previousRelease.Add(spacing); !previousRelease.IsZero() && release.Before(minimum) {
-			release = minimum
-		}
-		if lag := time.Since(release); lag > maxCatchup {
-			release = release.Add(lag - maxCatchup)
-		}
 		if delay := time.Until(release); delay > 0 {
 			time.Sleep(delay)
 		}
-		previousRelease = release
+		nextRelease = release.Add(spacing)
 		if len(profile.CapacitySteps) == 2 {
 			switch {
 			case packet.index >= packetCount*2/3 && stepAt[profile.CapacitySteps[1]].IsZero():