diff --git a/docs/ACCELERATION.md b/docs/ACCELERATION.md index 1e52318..663ed27 100644 --- a/docs/ACCELERATION.md +++ b/docs/ACCELERATION.md @@ -94,9 +94,45 @@ denominator: QUIC can send already-queued bytes during an application wait. Actual transport backpressure remains outside token sleep. When a slower path fills the finite send buffers, predominantly transport-bound samples can age out the old bandwidth maximum. Sleep share is still an application-level -heuristic, not proof that the path is uncongested. The controller does not add -a separate capacity-probing phase: sustained RTT penalties can limit discovery -of spare capacity until conditions or higher delivery observations change. +heuristic, not proof that the path is uncongested. + +For `balanced` and `aggressive`, a bounded capacity-recovery probe can break a +low-rate lock-in after real backpressure subsides while RTT stays elevated. +It requires at least three consecutive valid, loss-free, pacing-limited samples +spanning at least one second on the controller clock, without a material RTT +increase. It runs only when the steady target is below 95% of learned capacity. +The temporary admission rate is at most 1.25 times capacity (`balanced`) or +1.50 times capacity (`aggressive`), also capped at twice the steady target and +the configured maximum. Only a real higher wire sample can raise the capacity +estimate; starting or finishing a probe does not add tokens or change history. +The reported `TargetBytesPerSecond` remains the steady target, not this transient +admission rate. + +The connection-wide byte allowance covers two bursts or two sampling windows, +whichever is larger, with a hard 1 MiB ceiling. The deadline covers service of +that allowance plus one sample window, or two smoothed RTTs, with a 100 ms +minimum and a two-second maximum. Configurations whose complete allowance +cannot fit those byte/time bounds are not probed. Deadline accounting splits +token refill at expiry even without another observation; cancellation refunds +normal tokens but never replenishes the probe allowance. Admission can finish +before the last wire feedback, so a bounded 250–500 ms grace period accepts +that outcome without allowing further probe-rate admission. + +Any newly observed loss, a relative RTT rise above 25% (with a 1 ms noise +tolerance), or transport-bound delivery without a real capacity increase ends +the probe. Idle intervals and counter resets clear qualification. Successful +probes require more than 2% observed capacity growth and are separated by at +least one second or eight smoothed RTTs (the latter capped at 16 seconds). +Unsuccessful attempts back off from two to at most 16 seconds. RTT/loss penalties +continue to determine the steady target; the underlying QUIC controller remains +active. `conservative`, fixed-rate, and bypass modes do not perform these probes. + +This is bounded recovery, not a guarantee of recovering all spare capacity: +large bandwidth-delay products or custom bursts can exceed the probe limits; +long-delayed feedback and persistent congestion can prevent discovery. Tests +cover fixed-RTT synthetic fast/slow/fast phases and a small-window real loopback +QUIC receiver slowdown. They do not establish Internet throughput, fairness, +browser similarity, or performance superiority. This is intentionally not a full BBR state machine. In particular AutoCAR has no transport-visible BDP congestion window, ACK aggregation model, ProbeRTT diff --git a/internal/accel/capacity_probe.go b/internal/accel/capacity_probe.go new file mode 100644 index 0000000..ffc97e7 --- /dev/null +++ b/internal/accel/capacity_probe.go @@ -0,0 +1,170 @@ +package accel + +import ( + "math" + "time" +) + +const ( + maximumProbeBytes = 1 << 20 + maximumProbeDuration = 2 * time.Second + minimumProbeGap = time.Second + maximumProbeBackoff = 16 * time.Second +) + +// capacityProbe is a temporary, connection-wide admission excursion. It never +// writes into the estimator: only ordinary transport observations can establish +// higher capacity. All fields are protected by Controller.mu. +type capacityProbe struct { + active bool + rate int64 + remaining float64 + deadline time.Time + entryCapacity float64 + entryRTT time.Duration + window time.Duration + gap time.Duration + + // The last admitted bytes may not yet appear in wire counters. Allow a + // bounded feedback grace period after admission ends, without more probes. + pending bool + outcomeDeadline time.Time + backoff time.Duration + nextAt time.Time + eligibleSamples int + eligibleSince time.Time + eligibleRTT time.Duration +} + +func (c *Controller) pacingRateLocked() int64 { + if c.probe.active { + return c.probe.rate + } + return c.targetRate +} + +func (c *Controller) probeRTTRoseLocked(rtt time.Duration) bool { + baseline := c.probe.entryRTT + if !c.probe.active && !c.probe.pending && c.probe.eligibleSamples > 0 { + baseline = c.probe.eligibleRTT + } else if !c.probe.active && !c.probe.pending { + return false + } + // An absolute 1ms tolerance avoids treating microsecond scheduler noise on + // loopback as a newly growing queue. Sustained RTT penalties still apply to + // the steady target, independently of this relative abort signal. + return rtt > baseline && rtt-baseline > max(time.Millisecond, baseline/4) +} + +func (c *Controller) resetProbeLocked() { + c.refillLocked(c.clock.Now()) + wasActive := c.probe.active + c.probe = capacityProbe{} + if wasActive { + c.signalStateChangeLocked() + } +} + +func (c *Controller) abortProbeLocked(now time.Time) { + if c.probe.active || c.probe.pending { + c.resolveProbeLocked(now, false) + } + c.probe.eligibleSamples = 0 +} + +func (c *Controller) resolveProbeLocked(now time.Time, success bool) { + wasActive := c.probe.active + c.probe.active = false + c.probe.pending = false + c.probe.eligibleSamples = 0 + if success { + c.probe.backoff = 0 + } else { + c.probe.backoff = min(maximumProbeBackoff, max(2*minimumProbeGap, 2*c.probe.backoff)) + } + c.probe.nextAt = now.Add(max(c.probe.gap, c.probe.backoff)) + if wasActive { + c.signalStateChangeLocked() + } +} + +// Caller has already refilled through now at the old rate. The grace period is +// only for learning the result; no more high-rate admission is allowed. +func (c *Controller) finishProbeLocked(now time.Time) { + c.probe.active = false + c.probe.pending = true + c.probe.outcomeDeadline = now.Add(max(250*time.Millisecond, 2*c.probe.window)) + c.probe.eligibleSamples = 0 + c.signalStateChangeLocked() +} + +func (c *Controller) observeProbeLocked(snapshot Snapshot, now time.Time, limited, loss bool) { + if c.profile == ProfileConservative { + return + } + capacity := c.estimator.bandwidthEstimate() + if c.probe.active || c.probe.pending { + // The normal estimator has consumed this sample first, including real + // backpressure and any higher delivery. Never discard evidence on abort. + if loss || c.probeRTTRoseLocked(snapshot.SmoothedRTT) { + c.abortProbeLocked(now) + } else if capacity > c.probe.entryCapacity*1.02 { + c.resolveProbeLocked(now, true) + } else if !limited { + c.abortProbeLocked(now) + } + return + } + if loss || !limited || c.estimator.count == 0 || float64(c.targetRate) >= capacity*.95 { + c.probe.eligibleSamples = 0 + return + } + if c.probeRTTRoseLocked(snapshot.SmoothedRTT) { + c.probe.eligibleSamples = 0 + } + if c.probe.eligibleSamples == 0 { + c.probe.eligibleSince = now + c.probe.eligibleRTT = snapshot.SmoothedRTT + } + if c.probe.eligibleSamples < 3 { + c.probe.eligibleSamples++ + } + if c.probe.eligibleSamples < 3 || now.Sub(c.probe.eligibleSince) < time.Second || now.Before(c.probe.nextAt) { + return + } + + gain := 1.25 + if c.profile == ProfileAggressive { + gain = 1.50 + } + rate := int64(math.Min(float64(c.maximumRate), math.Min(capacity*gain, 2*float64(c.targetRate)))) + if float64(rate) <= capacity || rate <= c.targetRate { + return + } + window := stableSampleWindow(snapshot.MinRTT) + budget := math.Ceil(math.Max(2*c.burst, 2*float64(rate)*window.Seconds())) + // Do not repeatedly run probes too small/short to produce even a complete + // sample. Extremely large custom bursts or bandwidth-delay products are + // outside this deliberately bounded application-level recovery mechanism. + if budget > maximumProbeBytes { + return + } + service := durationForBytes(budget, rate) + if service > maximumProbeDuration-window { + return + } + duration := max(100*time.Millisecond, service+window) + // Clamp before multiplying potentially untrusted positive RTT durations. + duration = max(duration, 2*min(snapshot.SmoothedRTT, maximumProbeDuration/2)) + gap := max(minimumProbeGap, 8*min(snapshot.SmoothedRTT, maximumProbeBackoff/8)) + c.probe.active = true + c.probe.rate = rate + c.probe.remaining = budget + c.probe.deadline = now.Add(duration) + c.probe.entryCapacity = capacity + c.probe.entryRTT = snapshot.SmoothedRTT + c.probe.window = window + c.probe.gap = gap + c.probe.eligibleSamples = 0 + c.signalStateChangeLocked() +} diff --git a/internal/accel/capacity_probe_safety_test.go b/internal/accel/capacity_probe_safety_test.go new file mode 100644 index 0000000..13be323 --- /dev/null +++ b/internal/accel/capacity_probe_safety_test.go @@ -0,0 +1,391 @@ +package accel + +import ( + "context" + "errors" + "math" + "sync" + "testing" + "time" +) + +func TestCapacityProbeSafetyAdmissionBounds(t *testing.T) { + for _, test := range []struct { + name string + capacity int64 + burst int64 + maximum int64 + rtt time.Duration + want bool + }{ + {name: "ordinary", capacity: 1_000_000, burst: 64 << 10, want: true}, + {name: "window_scaled_budget", capacity: 20_000_000, burst: 64 << 10, want: true}, + {name: "exact_budget_limit", capacity: 41_943_040, burst: 64 << 10, want: true}, + {name: "over_budget_limit", capacity: 41_943_041, burst: 64 << 10}, + {name: "exact_burst_limit", capacity: 1_000_000, burst: 512 << 10, want: true}, + {name: "over_burst_limit", capacity: 1_000_000, burst: 512<<10 + 1}, + {name: "chunk_too_slow", capacity: 100_000, burst: 512 << 10}, + {name: "maximum_prevents_uplift", capacity: 1_000_000, burst: 64 << 10, maximum: 1_000_000}, + {name: "duration_does_not_overflow", capacity: 1_000_000, burst: 64 << 10, rtt: time.Duration(math.MaxInt64), want: true}, + } { + t.Run(test.name, func(t *testing.T) { + clock := newFakeClock() + controller := newProbeSafetyController(t, clock, test.capacity, test.burst, test.maximum) + rtt := test.rtt + if rtt == 0 { + rtt = 4 * time.Millisecond + } + qualifyProbeSafety(t, controller, clock, []time.Duration{rtt, rtt, rtt}) + probe := controller.probe + if probe.active != test.want { + t.Fatalf("probe active=%t, want %t: %+v", probe.active, test.want, probe) + } + if !test.want { + return + } + if probe.remaining > maximumProbeBytes || probe.remaining < 2*controller.burst { + t.Fatalf("probe budget=%g, burst=%g", probe.remaining, controller.burst) + } + if probe.rate <= test.capacity || probe.rate > controller.maximumRate || probe.rate > 2*controller.targetRate { + t.Fatalf("probe rate=%d capacity=%d steady=%d maximum=%d", probe.rate, test.capacity, controller.targetRate, controller.maximumRate) + } + if duration := probe.deadline.Sub(clock.Now()); duration < 100*time.Millisecond || duration > maximumProbeDuration { + t.Fatalf("probe duration=%s", duration) + } + if probe.gap < minimumProbeGap || probe.gap > maximumProbeBackoff { + t.Fatalf("probe gap=%s", probe.gap) + } + if controller.TargetBytesPerSecond() != int64(float64(test.capacity)*.648) { + t.Fatal("probe changed the reported steady target") + } + }) + } +} + +func TestCapacityProbeSafetyQualificationRejectsRisingRTT(t *testing.T) { + clock := newFakeClock() + controller := newProbeSafetyController(t, clock, 1_000_000, 64<<10, 0) + qualifyProbeSafety(t, controller, clock, []time.Duration{100 * time.Millisecond, 200 * time.Millisecond, 400 * time.Millisecond}) + if controller.probe.active { + t.Fatal("probe started while qualifying observations showed a growing queue") + } +} + +func TestCapacityProbeSafetySubwindowRTTResetsQualification(t *testing.T) { + clock := newFakeClock() + controller := newProbeSafetyController(t, clock, 1000, 100, 0) + qualifyProbeSafety(t, controller, clock, []time.Duration{4 * time.Millisecond, 4 * time.Millisecond}) + baseline := Snapshot{At: clock.Now(), MinRTT: time.Millisecond, SmoothedRTT: 4 * time.Millisecond} + mustObserveIdleSnapshot(t, controller, baseline) + if err := clock.Sleep(context.Background(), time.Millisecond); err != nil { + t.Fatal(err) + } + snapshot := baseline + snapshot.At, snapshot.SmoothedRTT = clock.Now(), 6*time.Millisecond + mustObserveIdleSnapshot(t, controller, snapshot) + if controller.probe.active || controller.probe.eligibleSamples != 0 || controller.previous != baseline { + t.Fatalf("early RTT rise did not clear qualification without consuming baseline: %+v", controller.probe) + } +} + +func TestCapacityProbeSafetySharedAdmissionBudget(t *testing.T) { + for _, test := range []struct { + name string + budget float64 + probeChunks int + }{ + {name: "exact_chunks", budget: 200, probeChunks: 2}, + {name: "no_partial_chunk_overshoot", budget: 150, probeChunks: 1}, + } { + t.Run(test.name, func(t *testing.T) { + clock := newFakeClock() + controller := newProbeSafetyController(t, clock, 1000, 100, 0) + startProbeSafety(controller, test.budget) + started := clock.Now() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + var writers sync.WaitGroup + errors := make(chan error, 3) + for range 3 { + writers.Add(1) + go func() { + defer writers.Done() + errors <- controller.Wait(ctx, 100) + }() + } + writers.Wait() + close(errors) + for err := range errors { + if err != nil { + t.Fatal(err) + } + } + want := time.Duration(test.probeChunks)*durationForBytes(100, 1250) + time.Duration(3-test.probeChunks)*durationForBytes(100, 648) + if elapsed := clock.Now().Sub(started); elapsed != want { + t.Fatalf("shared admission elapsed=%s, want %s (only %d chunks may use probe rate)", elapsed, want, test.probeChunks) + } + if controller.probe.active || controller.probe.remaining != test.budget-float64(100*test.probeChunks) { + t.Fatalf("shared budget not exhausted exactly once: %+v", controller.probe) + } + }) + } +} + +func TestCapacityProbeSafetyEntryDoesNotMintTokensOrCapacity(t *testing.T) { + clock := newFakeClock() + controller := newProbeSafetyController(t, clock, 1000, 100, 0) + qualifyProbeSafety(t, controller, clock, []time.Duration{4 * time.Millisecond, 4 * time.Millisecond}) + if err := clock.Sleep(context.Background(), 500*time.Millisecond); err != nil { + t.Fatal(err) + } + controller.mu.Lock() + controller.refillLocked(clock.Now()) + controller.tokens = 17 + controller.observeProbeLocked(Snapshot{MinRTT: time.Millisecond, SmoothedRTT: 4 * time.Millisecond}, clock.Now(), true, false) + controller.mu.Unlock() + if !controller.probe.active || controller.tokens != 17 || controller.estimator.bandwidthEstimate() != 1000 || controller.TargetBytesPerSecond() != 648 { + t.Fatalf("entry changed steady state: probe=%+v tokens=%g capacity=%g steady=%d", controller.probe, controller.tokens, controller.estimator.bandwidthEstimate(), controller.TargetBytesPerSecond()) + } +} + +func TestCapacityProbeSafetyOversleepSplitsRefillWithoutObserve(t *testing.T) { + clock := newPacingSleepClock() + controller := newProbeSafetyController(t, clock, 1000, 1000, 0) + controller.targetRate = 1000 + startProbeSafety(controller, 2000) + controller.probe.rate = 2000 + controller.probe.deadline = clock.Now().Add(100 * time.Millisecond) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- controller.Wait(ctx, 1000) }() + first := clock.nextSleep(t) + if first.duration != 100*time.Millisecond { + t.Fatalf("first sleep=%s, want probe deadline in 100ms", first.duration) + } + clock.advance(250 * time.Millisecond) + close(first.release) + second := clock.nextSleep(t) + // 200 probe tokens plus 150 steady tokens are available after oversleep. + // Refilling all 250ms at the probe rate would incorrectly leave only 500ms. + if second.duration != 650*time.Millisecond { + t.Fatalf("steady remainder=%s, want 650ms", second.duration) + } + controller.mu.Lock() + active := controller.probe.active + controller.mu.Unlock() + if active { + t.Fatal("probe survived its deadline without Observe") + } + clock.advance(second.duration) + close(second.release) + if err := awaitPacingSleepWait(t, done); err != nil { + t.Fatal(err) + } + assertPacingSleep(t, controller, clock.Now(), 900*time.Millisecond, false) +} + +func TestCapacityProbeSafetyCancelRefundDoesNotReplenishBudget(t *testing.T) { + clock := newPacingSleepClock() + controller := newProbeSafetyController(t, clock, 1000, 100, 0) + startProbeSafety(controller, 200) + controller.tokens = 100 + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- controller.Wait(ctx, 200) }() + clock.nextSleep(t) + clock.advance(20 * time.Millisecond) + cancel() + if err := awaitPacingSleepWait(t, done); !errors.Is(err, context.Canceled) { + t.Fatalf("canceled Wait=%v", err) + } + if controller.probe.remaining != 100 || controller.tokens != 100 { + t.Fatalf("cancel budget/tokens=%g/%g, want 100/100", controller.probe.remaining, controller.tokens) + } + if err := controller.Wait(context.Background(), 100); err != nil { + t.Fatal(err) + } + if controller.probe.active || controller.probe.remaining != 0 { + t.Fatalf("refunded tokens replenished probe allowance: %+v", controller.probe) + } + assertPacingSleep(t, controller, clock.Now(), 20*time.Millisecond, false) +} + +func TestCapacityProbeSafetySubwindowCongestionAborts(t *testing.T) { + for _, test := range []struct { + name string + loss uint64 + rtt time.Duration + want bool + }{ + {name: "new_loss", loss: 1, rtt: 4 * time.Millisecond, want: true}, + {name: "exact_rtt_tolerance", rtt: 5 * time.Millisecond}, + {name: "above_rtt_tolerance", rtt: 5*time.Millisecond + time.Nanosecond, want: true}, + } { + t.Run(test.name, func(t *testing.T) { + clock := newFakeClock() + controller := newProbeSafetyController(t, clock, 1000, 100, 0) + baseline := Snapshot{At: clock.Now(), SentBytes: 100, MinRTT: time.Millisecond, SmoothedRTT: 4 * time.Millisecond} + mustObserveIdleSnapshot(t, controller, baseline) + startProbeSafety(controller, 200) + if err := clock.Sleep(context.Background(), time.Millisecond); err != nil { + t.Fatal(err) + } + snapshot := baseline + snapshot.At, snapshot.LostBytes, snapshot.SmoothedRTT = clock.Now(), test.loss, test.rtt + mustObserveIdleSnapshot(t, controller, snapshot) + if got := !controller.probe.active; got != test.want { + t.Fatalf("probe aborted=%t, want %t", got, test.want) + } + if controller.previous != baseline || controller.estimator.count != 1 { + t.Fatal("subwindow abort consumed the delivery baseline") + } + }) + } +} + +func TestCapacityProbeSafetyFinalAdmissionFeedbackIsLearned(t *testing.T) { + for _, delayed := range []bool{false, true} { + name := "higher_sample" + if delayed { + name = "transport_bound_sample" + } + t.Run(name, func(t *testing.T) { + clock := newFakeClock() + controller := newProbeSafetyController(t, clock, 1000, 100, 0) + snapshot := Snapshot{At: clock.Now(), MinRTT: time.Millisecond, SmoothedRTT: 4 * time.Millisecond} + mustObserveIdleSnapshot(t, controller, snapshot) + startProbeSafety(controller, 100) + if err := controller.Wait(context.Background(), 100); err != nil { + t.Fatal(err) + } + if controller.probe.active || !controller.probe.pending { + t.Fatal("last admission did not enter feedback-only phase") + } + if delayed { + if err := clock.Sleep(context.Background(), 100*time.Millisecond); err != nil { + t.Fatal(err) + } + } + snapshot.At, snapshot.SentBytes = clock.Now(), 100 + mustObserveIdleSnapshot(t, controller, snapshot) + if controller.estimator.count != 2 || controller.probe.active || controller.probe.pending { + t.Fatalf("final delivery was discarded or outcome unresolved: count=%d probe=%+v", controller.estimator.count, controller.probe) + } + if delayed { + if controller.estimator.rates[1] >= 1000 || controller.probe.backoff != 2*time.Second { + t.Fatal("ending probe discarded genuine backpressure or failed to back off") + } + } else if controller.estimator.bandwidthEstimate() != 1250 || controller.TargetBytesPerSecond() != 810 || controller.probe.backoff != 0 { + t.Fatalf("post-budget higher sample was not learned: capacity=%g steady=%d backoff=%s", controller.estimator.bandwidthEstimate(), controller.TargetBytesPerSecond(), controller.probe.backoff) + } + }) + } +} + +func TestCapacityProbeSafetyMissingFeedbackBackoffIsBounded(t *testing.T) { + clock := newFakeClock() + controller := newProbeSafetyController(t, clock, 1000, 100, 0) + for _, want := range []time.Duration{2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second, 16 * time.Second} { + controller.mu.Lock() + controller.probe.active = true + controller.probe.gap = time.Second + controller.finishProbeLocked(clock.Now()) + deadline := controller.probe.outcomeDeadline + controller.mu.Unlock() + if err := clock.Sleep(context.Background(), deadline.Sub(clock.Now())); err != nil { + t.Fatal(err) + } + // No Observe is needed to leave the feedback grace period. + if err := controller.Wait(context.Background(), 1); err != nil { + t.Fatal(err) + } + if controller.probe.active || controller.probe.pending || controller.probe.backoff != want || !controller.probe.nextAt.Equal(clock.Now().Add(want)) { + t.Fatalf("missing-feedback outcome=%+v, want %s backoff", controller.probe, want) + } + } +} + +func TestCapacityProbeSafetyIdleUnknownFeedbackAndEpochEndProbe(t *testing.T) { + for _, condition := range []string{"no_bytes", "no_rtt", "idle", "sent_reset", "idle_reset"} { + t.Run(condition, func(t *testing.T) { + clock := newFakeClock() + controller := newProbeSafetyController(t, clock, 1000, 100, 0) + snapshot := Snapshot{At: clock.Now(), SentBytes: 100, MinRTT: time.Millisecond, SmoothedRTT: 4 * time.Millisecond, ApplicationIdleTime: 100 * time.Millisecond} + mustObserveIdleSnapshot(t, controller, snapshot) + startProbeSafety(controller, 200) + if err := clock.Sleep(context.Background(), 20*time.Millisecond); err != nil { + t.Fatal(err) + } + snapshot.At, snapshot.SentBytes = clock.Now(), 120 + switch condition { + case "no_bytes": + snapshot.SentBytes = 100 + case "no_rtt": + snapshot.SmoothedRTT = 0 + case "idle": + snapshot.ApplicationIdleTime += 20 * time.Millisecond + case "sent_reset": + snapshot.SentBytes = 0 + case "idle_reset": + snapshot.ApplicationIdleTime = 0 + } + mustObserveIdleSnapshot(t, controller, snapshot) + if controller.probe.active || controller.probe.pending || controller.probe.eligibleSamples != 0 { + t.Fatalf("%s retained a probe: %+v", condition, controller.probe) + } + if condition == "sent_reset" || condition == "idle_reset" { + if controller.estimator.count != 0 || controller.TargetBytesPerSecond() != 1000 || controller.probe.backoff != 0 { + t.Fatal("new epoch retained old capacity or probe backoff") + } + } else if controller.estimator.count != 1 || controller.TargetBytesPerSecond() != 648 { + t.Fatal("non-capacity observation changed steady state") + } + }) + } +} + +// Seed existing capacity evidence and a steady RTT penalty. Individual tests +// exercise real admission/observation paths; qualification tests isolate only +// the eligibility policy. End-to-end discovery is covered by the closed loop. +func newProbeSafetyController(t *testing.T, clock Clock, capacity, burst, maximum int64) *Controller { + t.Helper() + if maximum == 0 { + maximum = max(defaultMaximumRate, capacity) + } + controller, err := New(Config{Clock: clock, InitialRateBytesPerSecond: capacity, MinRateBytesPerSecond: 1, MaxRateBytesPerSecond: maximum, BurstBytes: burst}) + if err != nil { + t.Fatal(err) + } + controller.estimator.rates[0] = float64(capacity) + controller.estimator.count = 1 + controller.estimator.next = 1 + controller.targetRate = int64(float64(capacity) * .648) + controller.tokens = 0 + return controller +} + +func qualifyProbeSafety(t *testing.T, controller *Controller, clock *fakeClock, rtts []time.Duration) { + t.Helper() + for index, rtt := range rtts { + if index > 0 { + if err := clock.Sleep(context.Background(), 500*time.Millisecond); err != nil { + t.Fatal(err) + } + } + controller.mu.Lock() + controller.refillLocked(clock.Now()) + controller.observeProbeLocked(Snapshot{MinRTT: time.Millisecond, SmoothedRTT: rtt}, clock.Now(), true, false) + controller.mu.Unlock() + } +} + +func startProbeSafety(controller *Controller, budget float64) { + controller.probe = capacityProbe{ + active: true, rate: 1250, remaining: budget, + deadline: controller.clock.Now().Add(time.Second), + entryCapacity: controller.estimator.bandwidthEstimate(), + entryRTT: 4 * time.Millisecond, window: 10 * time.Millisecond, gap: time.Second, + } +} diff --git a/internal/accel/capacity_probe_test.go b/internal/accel/capacity_probe_test.go new file mode 100644 index 0000000..95dbc63 --- /dev/null +++ b/internal/accel/capacity_probe_test.go @@ -0,0 +1,96 @@ +package accel + +import ( + "context" + "strconv" + "testing" + "time" +) + +func TestCapacityProbeRecoversWithPersistentRTTPenalty(t *testing.T) { + for _, profile := range []Profile{ProfileBalanced, ProfileAggressive, ProfileConservative} { + for _, chunk := range []int{32 << 10, 64 << 10} { + t.Run(profile.String()+"/"+strconv.Itoa(chunk), func(t *testing.T) { + clock := newFakeClock() + controller, err := New(Config{Profile: profile, Clock: clock}) + if err != nil { + t.Fatal(err) + } + // RTT never recovers: only real higher delivery can raise the + // estimator. Transport work is independent of actual token Sleep. + snapshot := Snapshot{At: clock.Now(), MinRTT: time.Millisecond, SmoothedRTT: 4 * time.Millisecond} + mustObserveIdleSnapshot(t, controller, snapshot) + type result struct { + capacity float64 + target int64 + higher int + lower int + } + phase := func(name string, chunks int, wire time.Duration) result { + t.Helper() + started := clock.Now() + r := result{} + for range chunks { + if err := controller.Wait(context.Background(), chunk); err != nil { + t.Fatal(err) + } + if err := clock.Sleep(context.Background(), wire); err != nil { + t.Fatal(err) + } + snapshot.At = clock.Now() + snapshot.SentBytes += uint64(chunk) + before := controller.estimator.bandwidthEstimate() + next := controller.estimator.next + mustObserveIdleSnapshot(t, controller, snapshot) + if next != controller.estimator.next { + if controller.estimator.rates[next] > before { + r.higher++ + } else if controller.estimator.rates[next] < before { + r.lower++ + } + } + } + r.capacity = controller.estimator.bandwidthEstimate() + r.target = controller.TargetBytesPerSecond() + t.Logf("%s bytes=%d elapsed=%s capacity=%.0f steady=%d higher=%d lower=%d probe_active=%t", name, chunks*chunk, clock.Now().Sub(started), r.capacity, r.target, r.higher, r.lower, controller.probe.active) + return r + } + fast := phase("fast", (1<<20)/chunk, time.Millisecond) + slow := phase("slow", 24, 120*time.Millisecond) + if slow.capacity >= fast.capacity/2 || slow.lower < deliveryRateWindow { + t.Fatalf("fixture did not learn real lower capacity: fast=%+v slow=%+v", fast, slow) + } + // Empty-bucket recovery rules out relying on old initial tokens. + controller.tokens = 0 + recoveryStart := clock.Now() + recovery := phase("recovery", (4<<20)/chunk, time.Millisecond) + if profile == ProfileConservative { + if recovery.target < slow.target || recovery.higher != 0 || controller.probe.active { + t.Fatalf("conservative behavior changed: slow=%+v recovered=%+v", slow, recovery) + } + return + } + if elapsed := clock.Now().Sub(recoveryStart); elapsed > 20*time.Second { + t.Fatalf("recovery exceeded virtual-time bound: %s", elapsed) + } + if recovery.capacity < slow.capacity*1.5 || float64(recovery.target) < float64(slow.target)*1.5 || recovery.higher < 2 { + t.Fatalf("failed real-sample recovery under unchanged RTT penalty: slow=%+v recovered=%+v", slow, recovery) + } + // A transient probe admission rate must never become the reported + // steady target or estimator without a real observation. + steady := controller.targetRate + history := controller.estimator + if err := controller.Wait(context.Background(), chunk); err != nil { + t.Fatal(err) + } + if controller.targetRate != steady || controller.estimator != history { + t.Fatal("admission manufactured capacity without a wire observation") + } + slowAgain := phase("slow again", 32, 120*time.Millisecond) + if slowAgain.capacity >= recovery.capacity*.75 || slowAgain.lower < deliveryRateWindow { + t.Fatalf("probes hid renewed transport backpressure: recovered=%+v slower=%+v", recovery, slowAgain) + } + }) + } + } +} diff --git a/internal/accel/pacer.go b/internal/accel/pacer.go index c3cd13b..c67f97d 100644 --- a/internal/accel/pacer.go +++ b/internal/accel/pacer.go @@ -129,6 +129,7 @@ type Controller struct { haveSnapshot bool previous Snapshot estimator adaptiveEstimator + probe capacityProbe // Only the admission owner sleeps for missing tokens. Count this interval // once, not the overlapping Wait durations of queued writers. Observation @@ -213,8 +214,8 @@ func (c *Controller) Mode() Mode { return c.mode } // and fixed-rate modes so shared configuration and telemetry stay consistent. func (c *Controller) Profile() Profile { return c.profile } -// TargetBytesPerSecond reports the current application pacing target. It -// returns zero in bypass mode. +// TargetBytesPerSecond reports the steady application pacing target, excluding +// short bounded capacity probes. It returns zero in bypass mode. func (c *Controller) TargetBytesPerSecond() int64 { c.mu.Lock() defer c.mu.Unlock() @@ -280,12 +281,20 @@ func (c *Controller) Observe(snapshot Snapshot) error { // instead of interpreting wrapped counters as a huge delivery sample. c.recordObservationLocked(snapshot, c.clock.Now()) c.estimator.reset() + c.resetProbeLocked() c.setTargetRateLocked(c.initialRate) return nil } elapsed := snapshot.At.Sub(c.previous.At) window := stableSampleWindow(snapshot.MinRTT) + now := c.clock.Now() + c.refillLocked(now) + // Loss and a growing queue must end an excursion even if this writer's + // observation is too early to form a stable delivery sample. + if snapshot.LostBytes > c.previous.LostBytes || c.probeRTTRoseLocked(snapshot.SmoothedRTT) { + c.abortProbeLocked(now) + } // Connection counters are sampled by every writer. Ignore sub-window calls // without advancing the baseline so concurrent streams cannot turn one packet // observed a few microseconds later into a terabyte-per-second rate sample. @@ -293,7 +302,6 @@ func (c *Controller) Observe(snapshot Snapshot) error { return nil } - now := c.clock.Now() observationElapsed := now.Sub(c.previousObservationTime) pacingElapsed := c.pacingTimeLocked(now) - c.previousPacingSleep pacingLimited := observationElapsed > 0 && pacingElapsed > 0 && pacingElapsed >= observationElapsed/2 @@ -312,9 +320,11 @@ func (c *Controller) Observe(snapshot Snapshot) error { // observations above retain both counters until this decision is made. // Require idle time to dominate the interval: a short source gap after // a long, flow-controlled write must not hide genuine path congestion. + c.abortProbeLocked(now) return nil } if sentDelta == 0 || snapshot.MinRTT == 0 || snapshot.SmoothedRTT == 0 { + c.abortProbeLocked(now) return nil } if lostDelta > sentDelta { @@ -332,6 +342,7 @@ func (c *Controller) Observe(snapshot Snapshot) error { target := c.estimator.observe(deliveredRate, lossRatio, snapshot.MinRTT, snapshot.SmoothedRTT, pacingLimited) target = math.Max(float64(c.minimumRate), math.Min(float64(c.maximumRate), target)) c.setTargetRateLocked(int64(math.Round(target))) + c.observeProbeLocked(snapshot, now, pacingLimited, lostDelta != 0) return nil } @@ -409,14 +420,29 @@ func (c *Controller) admit(ctx context.Context, amount float64) error { } c.mu.Lock() - c.refillLocked(c.clock.Now()) + now := c.clock.Now() + c.refillLocked(now) + // Never partially charge a chunk to an exhausted probe allowance. + // Existing tokens remain bounded; ending a probe does not refill them. + if c.probe.active && amount > c.probe.remaining { + c.finishProbeLocked(now) + } if c.tokens >= amount { c.tokens -= amount + if c.probe.active { + c.probe.remaining -= amount + if c.probe.remaining == 0 { + c.finishProbeLocked(now) + } + } c.mu.Unlock() return nil } missing := amount - c.tokens - delay := durationForBytes(missing, c.targetRate) + delay := durationForBytes(missing, c.pacingRateLocked()) + if c.probe.active { + delay = min(delay, c.probe.deadline.Sub(now)) + } stateChange := c.stateChange if c.mode == ModeAdaptive { c.pacingSleepStarted = c.clock.Now() @@ -593,11 +619,24 @@ func durationForBytes(amount float64, rate int64) time.Duration { } func (c *Controller) refillLocked(now time.Time) { + if c.probe.active && !now.Before(c.probe.deadline) { + // Split the interval: a late wakeup must not mint high-rate tokens for + // time beyond the deadline. No Observe call is required for expiry. + c.refillAtRateLocked(c.probe.deadline, c.probe.rate) + c.finishProbeLocked(c.probe.deadline) + } + c.refillAtRateLocked(now, c.pacingRateLocked()) + if c.probe.pending && !now.Before(c.probe.outcomeDeadline) { + c.resolveProbeLocked(now, false) + } +} + +func (c *Controller) refillAtRateLocked(now time.Time, rate int64) { elapsed := now.Sub(c.lastRefill) if elapsed <= 0 { return } - c.tokens += elapsed.Seconds() * float64(c.targetRate) + c.tokens += elapsed.Seconds() * float64(rate) if c.tokens > c.burst { c.tokens = c.burst } diff --git a/internal/tunnel/pacing_recovery_integration_test.go b/internal/tunnel/pacing_recovery_integration_test.go new file mode 100644 index 0000000..de4d896 --- /dev/null +++ b/internal/tunnel/pacing_recovery_integration_test.go @@ -0,0 +1,89 @@ +package tunnel + +import ( + "testing" + "time" +) + +// The adaptive defaults and QUIC path stay unchanged: only the real reader +// changes from fast to slow and back. The small, fixed receive window makes +// its backpressure observable instead of buffering the whole slow phase. +func TestQUICPacingRecoversAfterReceiverBackpressure(t *testing.T) { + f := newFeedbackQUICFixture(t, 1, 16<<10) + fast := runPacingRecoveryPhase(t, f, "fast before", 1<<20, 0) + slow := runPacingRecoveryPhase(t, f, "receiver limited", 16*feedbackChunkSize, 120*time.Millisecond) + if slow.write <= slow.wait { + t.Fatalf("fixture did not establish transport backpressure: Write=%s, Wait=%s", slow.write, slow.wait) + } + // Even both worst-case balanced-profile penalties cannot reduce an + // unchanged capacity target below 36%. Require an actual capacity decline, + // not merely a transient RTT/loss penalty, before checking rediscovery. + if slow.target >= fast.target*36/100 { + t.Fatalf("fixture did not establish lower capacity: target %d -> %d", fast.target, slow.target) + } + + var recovered [4]pacingRecoveryPhase + for index := range recovered { + recovered[index] = runPacingRecoveryPhase(t, f, "fast after", 512<<10, 0) + } + f.checkReverse(t) + // The first two windows allow discovery. Require sustained useful progress + // in both later windows instead of treating one target spike as recovery. + // All bounds are relative to this connection's actual slow-reader phase, + // not a machine-specific Mbps threshold or a race against another mode. + for index := 2; index < len(recovered); index++ { + phase := recovered[index] + if phase.bytesPerSecond <= 1.25*float64(slow.target) { + t.Errorf("recovery window %d did not provide useful progress: %.0f bytes/s, slow steady target %d bytes/s", + index+1, phase.bytesPerSecond, slow.target) + } + // TargetBytesPerSecond excludes temporary probe rates, so this must + // be a sustained target increase. Deterministic controller tests + // separately isolate capacity discovery under persistent RTT penalties. + if phase.target < slow.target*3/2 { + t.Errorf("recovery window %d did not regain a meaningful target: %d, slow target %d", + index+1, phase.target, slow.target) + } + } +} + +type pacingRecoveryPhase struct { + bytesPerSecond float64 + target int64 + wait, write time.Duration +} + +func runPacingRecoveryPhase(t *testing.T, f *feedbackQUICFixture, name string, size int, readDelay time.Duration) pacingRecoveryPhase { + t.Helper() + started := time.Now() + waitBefore := f.streamPacers[0].waitTime.Load() + writeBefore := f.rawStreams[0].writeTime.Load() + writer := f.run(func() error { return writeFeedbackPayload(f.serverStreams[0], size) }) + for read := 0; read < size; read += feedbackChunkSize { + if readDelay > 0 { + timer := time.NewTimer(readDelay) + select { + case <-timer.C: + case <-f.ctx.Done(): + timer.Stop() + t.Fatal(f.ctx.Err()) + } + } + if err := readFeedbackPayload(f.clientStreams[0], min(feedbackChunkSize, size-read)); err != nil { + t.Fatalf("%s: %d/%d verified bytes: %v", name, read, size, err) + } + } + f.await(t, writer) + elapsed := time.Since(started) + phase := pacingRecoveryPhase{ + bytesPerSecond: float64(size) / elapsed.Seconds(), + target: f.pacer.controller.TargetBytesPerSecond(), + wait: time.Duration(f.streamPacers[0].waitTime.Load() - waitBefore), + write: time.Duration(f.rawStreams[0].writeTime.Load() - writeBefore), + } + stats := f.serverStreams[0].conn.ConnectionStats() + t.Logf("%s: bytes=%d duration=%s rate=%.0f bytes/s target=%d Wait=%s Write=%s lost=%d MinRTT=%s SRTT=%s", + name, size, elapsed, phase.bytesPerSecond, phase.target, phase.wait, phase.write, + stats.BytesLost, stats.MinRTT, stats.SmoothedRTT) + return phase +}