go-20261007-clock-skewed-multipath-congestion-recovery

Answer

The complete, tested solution is written to ~/multipath-scheduler/SOLUTION.md and reproduced below.

# Fix: Multipath transport scheduler with robust per-path estimation and burst-free migration

## Root-cause analysis

A naive multipath transport that estimates per-path delivery rate and RTT from raw ACK timing fails in five ways:

1. **Asymmetric clock skew.** If RTT or delivery rate are computed by subtracting a remote (receiver) timestamp from a local (sender) timestamp, the result contains an unbounded offset and a drifting rate error. The error differs per path because each path has its own remote timestamp stream.
2. **Intermittent ACK compression.** When several ACKs are delivered in a burst, per-ACK inter-arrival rate samples can claim a bandwidth far above the true path rate. A single compressed burst can make a slow path look fast and push traffic onto it, creating a queue and, on migration, a congestion burst.
3. **Packet reordering and stale samples.** ACKs can arrive out of order. Late ACKs can carry old send timestamps, and duplicates can double-count delivered bytes. Naive estimators accept these and corrupt `minRTT`, `SRTT`, and the delivered counter.
4. **Counter wraparound.** Packet sequence numbers and cumulative-delivered counters are finite. Comparing them with `<` or subtracting them with signed arithmetic produces huge, wrong deltas after wrap.
5. **Migration burst.** When a path fails, its in-flight bytes are lost. When it recovers, a scheduler that immediately gives it the full share of traffic can inject an unacknowledged burst larger than the path's congestion window.

## Exact fix

The fix has five parts, all present in the code below.

1. **Use only the sender's monotonic clock for RTT and rate.** The ACK echoes the sender's `SendTime`; the estimator never subtracts receiver timestamps. This makes the estimator immune to asymmetric clock skew without a separate skew filter.
2. **Packet-timed delivery-rate samples.** `rate = delivered_bytes_since_send / time_since_send`. The denominator comes from the local send clock, so ACK compression cannot make the denominator artificially small. A one-sample-per-`MinSampleInterval` gate and a 75th-percentile window with a 1.5x clamp make the estimate robust to intermittent compression bursts.
3. **Reordering and stale-sample filters.** Duplicate ACKs are ignored by the scheduler's packet map. Karn's algorithm skips retransmitted packets for RTT. Samples older than `MaxSampleAge` are rejected. RTT uses the RFC 6298 SRTT/RTTVAR filter.
4. **Wrap-safe math.** `wrapDiff`, `seqAfter`, and `seqAfterEq` use unsigned arithmetic with an `int64` comparison so modulo-2^64 counters and sequence numbers behave correctly.
5. **ACK-clocked recovery credit.** A failed path drops out of service and releases its in-flight budget. On recovery it starts with `InitialCwnd` and a `migrationCredit` budget; each ACK adds back at most one segment (capped by `MigrationBurst`). The scheduler also ramps its WFQ weight from 0 to 1 over `MigrationRamp`. Traffic therefore migrates to the recovered path only as ACKs open the window, never as a burst.

## Source

### `go.mod`

```go
module example.com/multipath

go 1.26
```

### `wrap.go`

```go
package multipath

// wrapDiff returns the forward difference between two monotonically
// increasing counters that may wrap at 2^64.  The caller must ensure the
// distance is smaller than 2^63; this is true for all sane ACK/sequence
// windows.
func wrapDiff(newer, older uint64) uint64 {
    return newer - older
}

// seqAfter reports whether a is strictly newer than b in modulo-2^64
// sequence space.  Comparing int64(a-b) > 0 is the standard wrap-safe
// comparison used by TCP-like protocols.
func seqAfter(a, b uint64) bool {
    return int64(a-b) > 0
}

// seqAfterEq is the inclusive variant of seqAfter.
func seqAfterEq(a, b uint64) bool {
    return int64(a-b) >= 0
}
```

### `estimator.go`

```go
package multipath

import (
    "math"
    "sort"
    "time"
)

// EstimatorConfig controls the robustness filters of the per-path estimator.
type EstimatorConfig struct {
    // MinSampleInterval prevents a burst of compressed ACKs from creating
    // many delivery-rate samples.  At most one rate sample is accepted per
    // interval.
    MinSampleInterval time.Duration

    // MaxSampleAge rejects samples whose originating packet is too old
    // (lost/reordered for a long time, or a stale ACK after recovery).
    MaxSampleAge time.Duration

    // RateWindow is the sliding window over which the robust delivery rate
    // is computed.
    RateWindow time.Duration

    // MaxRTT rejects implausibly large RTT samples.
    MaxRTT time.Duration

    // MinSampleBytes rejects tiny delivery-rate samples (e.g. delayed ACKs
    // of one small packet) that would be dominated by timer noise.
    MinSampleBytes uint64
}

// DefaultEstimatorConfig returns conservative defaults suitable for a
// datacenter/WAN transport.
func DefaultEstimatorConfig() EstimatorConfig {
    return EstimatorConfig{
        MinSampleInterval: time.Millisecond,
        MaxSampleAge:      10 * time.Second,
        RateWindow:        time.Second,
        MaxRTT:            10 * time.Second,
        MinSampleBytes:    1,
    }
}

// SentPacket is the immutable sender-side record needed to compute robust
// RTT and delivery-rate samples.  SendTime is always taken from the local
// monotonic sender clock.  Echoing this timestamp in the ACK is what makes
// both estimators immune to asymmetric remote clock skew: we never subtract
// two timestamps produced by different clocks.
type SentPacket struct {
    PathID          string
    Seq             uint64
    Size            uint64
    SendTime        time.Time
    DeliveredAtSend uint64
    Retransmitted   bool
}

type rateSample struct {
    at   time.Time
    rate float64
}

// Estimator tracks RTT and delivery rate for one path.
//
// RTT follows the RFC 6298 SRTT/RTTVAR filter.  Delivery rate follows the
// "packet-timed" sampling idea from BBR: rate = delivered bytes / time
// since the *send* time of the packet that was just acknowledged.  Because
// the denominator is built from local send timestamps, ACK compression
// cannot inflate the sample, and because we add an explicit accept-one-per
// interval gate, intermittent ACK compression cannot create a flood of
// samples.
type Estimator struct {
    cfg EstimatorConfig

    // RFC 6298 RTT state.
    srtt    time.Duration
    rttvar  time.Duration
    minRTT  time.Duration
    haveRTT bool

    // Delivery-rate state.
    delivered      uint64
    rate           float64
    haveRate       bool
    samples        []rateSample
    lastRateSample time.Time
}

// NewEstimator returns an Estimator with the supplied configuration.
func NewEstimator(cfg EstimatorConfig) *Estimator {
    if cfg.MinSampleInterval <= 0 {
        cfg.MinSampleInterval = DefaultEstimatorConfig().MinSampleInterval
    }
    if cfg.MaxSampleAge <= 0 {
        cfg.MaxSampleAge = DefaultEstimatorConfig().MaxSampleAge
    }
    if cfg.RateWindow <= 0 {
        cfg.RateWindow = DefaultEstimatorConfig().RateWindow
    }
    if cfg.MaxRTT <= 0 {
        cfg.MaxRTT = DefaultEstimatorConfig().MaxRTT
    }
    if cfg.MinSampleBytes == 0 {
        cfg.MinSampleBytes = 1
    }
    return &Estimator{cfg: cfg}
}

// SRTT returns the smoothed RTT estimate, or 0 when unknown.
func (e *Estimator) SRTT() time.Duration { return e.srtt }

// RTTVar returns the RTT variation estimate.
func (e *Estimator) RTTVar() time.Duration { return e.rttvar }

// MinRTT returns the minimum observed RTT, or 0 when unknown.
func (e *Estimator) MinRTT() time.Duration { return e.minRTT }

// Rate returns the robust delivery rate in bytes per second, or 0 when
// unknown.
func (e *Estimator) Rate() float64 { return e.rate }

// Delivered returns the cumulative bytes delivered on the path.  It is a
// local counter and therefore does not depend on receiver clock or receiver
// counter representation.
func (e *Estimator) Delivered() uint64 { return e.delivered }

// SampleCount returns the number of rate samples currently retained.
func (e *Estimator) SampleCount() int { return len(e.samples) }

// OnAck feeds one newly acknowledged packet to the estimator.
//
// The caller must only call this for a *new* ACK (duplicate ACKs and
// retransmission ACKs are ignored) and must pass a timestamp from the same
// monotonic sender clock that was used for SendTime.
func (e *Estimator) OnAck(now time.Time, pkt *SentPacket) {
    if pkt == nil {
        return
    }
    rtt := now.Sub(pkt.SendTime)

    // ---- RTT: Karn's algorithm + RFC 6298 ----
    // Stale RTT samples (e.g. an old packet finally ACKed after a path
    // recovered) are rejected so they cannot pull minRTT/SRTT backwards.
    if !pkt.Retransmitted && rtt > 0 && rtt <= e.cfg.MaxRTT && rtt <= e.cfg.MaxSampleAge {
        if !e.haveRTT {
            e.srtt = rtt
            e.rttvar = rtt / 2
            e.minRTT = rtt
            e.haveRTT = true
        } else {
            delta := e.srtt - rtt
            if delta < 0 {
                delta = -delta
            }
            e.rttvar = (3*e.rttvar + delta) / 4
            e.srtt = (7*e.srtt + rtt) / 8
            if rtt < e.minRTT {
                e.minRTT = rtt
            }
        }
    }

    // ---- Delivery rate ----
    e.delivered += pkt.Size // uint64 arithmetic intentionally wraps.

    age := now.Sub(pkt.SendTime)
    if age > e.cfg.MaxSampleAge || pkt.Size == 0 {
        return // stale sample: do not let it pollute the rate window
    }

    // Never accept more than one rate sample per MinSampleInterval.  When
    // several ACKs are compressed together, the first one is the most
    // conservative (it covers the fewest delivered bytes).
    if !e.lastRateSample.IsZero() && now.Sub(e.lastRateSample) < e.cfg.MinSampleInterval {
        return
    }

    bytes := wrapDiff(e.delivered, pkt.DeliveredAtSend)
    if bytes < e.cfg.MinSampleBytes {
        return
    }
    if age < e.cfg.MinSampleInterval {
        return
    }

    sample := float64(bytes) / age.Seconds()
    if math.IsInf(sample, 0) || math.IsNaN(sample) || sample < 0 {
        return
    }
    e.samples = append(e.samples, rateSample{at: now, rate: sample})
    e.lastRateSample = now
    e.prune(now)
    e.rate = e.robustRate()
    e.haveRate = true
}

func (e *Estimator) prune(now time.Time) {
    cutoff := now.Add(-e.cfg.RateWindow)
    keep := e.samples[:0]
    for _, s := range e.samples {
        if !s.at.Before(cutoff) {
            keep = append(keep, s)
        }
    }
    e.samples = keep
}

// robustRate returns the 75th percentile of the samples in the current
// window, clamped so that a single compressed-ACK sample cannot increase
// the previous estimate by more than 1.5x.  Percentile rather than max is
// deliberate: it makes intermittent compression outliers visible but not
// dominant.
func (e *Estimator) robustRate() float64 {
    if len(e.samples) == 0 {
        return e.rate
    }
    rates := make([]float64, len(e.samples))
    best := 0.0
    for i, s := range e.samples {
        rates[i] = s.rate
        if s.rate > best {
            best = s.rate
        }
    }
    if !e.haveRate || e.rate <= 0 {
        return best
    }
    sort.Float64s(rates)
    idx := int(float64(len(rates)-1) * 0.75)
    v := rates[idx]
    if v > e.rate*1.5 {
        v = e.rate * 1.5
    }
    return v
}
```

### `scheduler.go`

```go
package multipath

import (
    "sync"
    "time"
)

// SchedulerConfig configures the multipath scheduler and its safety limits.
type SchedulerConfig struct {
    // MaxAggregateInflight is the hard aggregate cap across all paths.
    MaxAggregateInflight uint64

    // MaxPathInflight is the default per-path hard cap.  A path added with
    // a non-zero limit overrides it.
    MaxPathInflight uint64

    // InitialCwnd is the congestion window used for a new/recovered path.
    InitialCwnd uint64

    // MinCwnd is the lower bound after a loss.
    MinCwnd uint64

    // MaxCwnd is the upper bound for a path.
    MaxCwnd uint64

    // MSS is used for congestion-window arithmetic and as the default WFQ
    // weight when no delivery-rate estimate exists.
    MSS uint64

    // MigrationRamp is how long a recovered path takes to regain full
    // scheduler weight.
    MigrationRamp time.Duration

    // MigrationBurst caps the extra ACK-clocked in-flight credit granted
    // while a recovered path is ramping.
    MigrationBurst uint64

    // Estimator configures each path's estimator.
    Estimator EstimatorConfig
}

// DefaultSchedulerConfig returns a usable set of defaults.
func DefaultSchedulerConfig() SchedulerConfig {
    mss := uint64(1200)
    return SchedulerConfig{
        MaxAggregateInflight: 1 << 20,
        MaxPathInflight:      1 << 20,
        InitialCwnd:          2 * mss,
        MinCwnd:              mss,
        MaxCwnd:              1 << 20,
        MSS:                  mss,
        MigrationRamp:        100 * time.Millisecond,
        MigrationBurst:       4 * mss,
        Estimator:            DefaultEstimatorConfig(),
    }
}

// Path is one transport path tracked by the scheduler.
type Path struct {
    ID string

    est Estimator

    sent    map[uint64]*SentPacket
    nextSeq uint64

    cwnd      uint64
    ssthresh  uint64
    slowStart bool
    inflight  uint64
    limit     uint64

    available bool

    // Recovery/migration state.  While ramping, migrationCredit is an
    // additional ACK-clocked in-flight budget, and the WFQ weight is
    // linearly ramped from 0 to 1 so that the recovered path does not
    // immediately attract a large share of new packets.
    ramping         bool
    rampStart       time.Time
    migrationCredit uint64

    // Weighted fair queueing virtual time.  Weight = estimated delivery
    // rate, so a higher-rate path is selected more often.
    vtime float64
}

// SendToken identifies a packet granted by Next.  It is the only object the
// caller needs to ACK or mark lost.
type SendToken struct {
    PathID   string
    Seq      uint64
    Size     uint64
    SendTime time.Time
}

// Scheduler owns all paths, the aggregate in-flight budget, and the
// per-path estimators.
type Scheduler struct {
    mu  sync.Mutex
    cfg SchedulerConfig
    now func() time.Time

    paths  []*Path
    byID   map[string]*Path
    agg    uint64 // aggregate bytes currently in flight
    forced int
}

// NewScheduler returns a scheduler using time.Now as its clock.
func NewScheduler(cfg SchedulerConfig) *Scheduler {
    return newScheduler(cfg, time.Now)
}

func newScheduler(cfg SchedulerConfig, now func() time.Time) *Scheduler {
    if cfg.MSS == 0 {
        cfg.MSS = DefaultSchedulerConfig().MSS
    }
    if cfg.InitialCwnd == 0 {
        cfg.InitialCwnd = DefaultSchedulerConfig().InitialCwnd
    }
    if cfg.MinCwnd == 0 {
        cfg.MinCwnd = DefaultSchedulerConfig().MinCwnd
    }
    if cfg.MaxCwnd == 0 {
        cfg.MaxCwnd = DefaultSchedulerConfig().MaxCwnd
    }
    if cfg.MaxAggregateInflight == 0 {
        cfg.MaxAggregateInflight = DefaultSchedulerConfig().MaxAggregateInflight
    }
    if cfg.MaxPathInflight == 0 {
        cfg.MaxPathInflight = DefaultSchedulerConfig().MaxPathInflight
    }
    if cfg.MigrationBurst == 0 {
        cfg.MigrationBurst = cfg.InitialCwnd
    }
    return &Scheduler{
        cfg:  cfg,
        now:  now,
        byID: make(map[string]*Path),
    }
}

// AddPath registers a path.  A zero limit means use MaxPathInflight.
func (s *Scheduler) AddPath(id string, limit uint64) *Path {
    s.mu.Lock()
    defer s.mu.Unlock()
    if p, ok := s.byID[id]; ok {
        return p
    }
    if limit == 0 {
        limit = s.cfg.MaxPathInflight
    }
    p := &Path{
        ID:        id,
        est:       *NewEstimator(s.cfg.Estimator),
        sent:      make(map[uint64]*SentPacket),
        cwnd:      s.cfg.InitialCwnd,
        ssthresh:  s.cfg.MaxCwnd,
        slowStart: true,
        limit:     limit,
        available: true,
    }
    s.paths = append(s.paths, p)
    s.byID[id] = p
    return p
}

// RemovePath unregisters a path.
func (s *Scheduler) RemovePath(id string) {
    s.mu.Lock()
    defer s.mu.Unlock()
    p, ok := s.byID[id]
    if !ok {
        return
    }
    delete(s.byID, id)
    out := s.paths[:0]
    for _, q := range s.paths {
        if q != p {
            out = append(out, q)
        }
    }
    s.paths = out
    // Treat outstanding bytes as lost so the aggregate budget is not
    // permanently reduced.
    for seq, pkt := range p.sent {
        p.release(pkt.Size)
        s.agg = release(s.agg, pkt.Size)
        delete(p.sent, seq)
    }
}

// MarkDown takes a path out of service.  Outstanding bytes are dropped
// immediately (a failed path cannot deliver them) so aggregate in-flight is
// not stuck.
func (s *Scheduler) MarkDown(id string) bool {
    s.mu.Lock()
    defer s.mu.Unlock()
    p, ok := s.byID[id]
    if !ok {
        return false
    }
    p.available = false
    p.ramping = false
    p.vtime = 0
    for seq, pkt := range p.sent {
        s.agg = release(s.agg, pkt.Size)
        delete(p.sent, seq)
    }
    p.inflight = 0
    return true
}

// MarkUp returns a path to service and starts the anti-burst ramp.
func (s *Scheduler) MarkUp(id string) bool {
    s.mu.Lock()
    defer s.mu.Unlock()
    p, ok := s.byID[id]
    if !ok {
        return false
    }
    p.available = true
    p.cwnd = s.cfg.InitialCwnd
    p.ssthresh = s.cfg.MaxCwnd
    p.slowStart = true
    p.ramping = true
    p.rampStart = s.now()
    p.migrationCredit = s.cfg.InitialCwnd
    if p.migrationCredit > s.cfg.MigrationBurst {
        p.migrationCredit = s.cfg.MigrationBurst
    }
    p.vtime = 0
    return true
}

// SetAggregateLimit changes the aggregate in-flight cap.
func (s *Scheduler) SetAggregateLimit(n uint64) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.cfg.MaxAggregateInflight = n
}

// SetPathLimit changes a path's hard in-flight cap.
func (s *Scheduler) SetPathLimit(id string, n uint64) {
    s.mu.Lock()
    defer s.mu.Unlock()
    if p, ok := s.byID[id]; ok {
        p.limit = n
    }
}

// AggregateInflight returns current aggregate bytes in flight.
func (s *Scheduler) AggregateInflight() uint64 {
    s.mu.Lock()
    defer s.mu.Unlock()
    return s.agg
}

// PathInflight returns a path's current in-flight bytes.
func (s *Scheduler) PathInflight(id string) uint64 {
    s.mu.Lock()
    defer s.mu.Unlock()
    if p, ok := s.byID[id]; ok {
        return p.inflight
    }
    return 0
}

// Next chooses a path and reserves up to want bytes of in-flight budget.
// The returned token must later be passed to OnAck or OnLoss.
func (s *Scheduler) Next(now time.Time, want uint64) (*SendToken, bool) {
    if want == 0 {
        return nil, false
    }
    s.mu.Lock()
    defer s.mu.Unlock()

    if s.agg >= s.cfg.MaxAggregateInflight {
        return nil, false
    }
    if room := s.cfg.MaxAggregateInflight - s.agg; want > room {
        want = room
    }

    var best *Path
    for _, p := range s.paths {
        if !p.available {
            continue
        }
        p.refresh(now, s.cfg)
        if p.allowance(s.cfg, s.agg) == 0 {
            continue
        }
        if best == nil || p.vtime < best.vtime ||
            (p.vtime == best.vtime && p.ID < best.ID) {
            best = p
        }
    }
    if best == nil {
        return nil, false
    }

    n := best.allowance(s.cfg, s.agg)
    if n > want {
        n = want
    }
    if n == 0 {
        return nil, false
    }

    best.inflight += n
    s.agg += n
    if best.ramping {
        if n >= best.migrationCredit {
            best.migrationCredit = 0
        } else {
            best.migrationCredit -= n
        }
    }

    seq := best.nextSeq
    best.nextSeq++ // wraps naturally
    pkt := &SentPacket{
        PathID:          best.ID,
        Seq:             seq,
        Size:            n,
        SendTime:        now,
        DeliveredAtSend: best.est.Delivered(),
    }
    best.sent[seq] = pkt

    // Weighted fair queueing: update virtual time using the current rate
    // estimate.  During recovery, ramp the weight from 0 to 1.
    w := best.est.Rate()
    if w <= 0 {
        w = float64(s.cfg.MSS)
    }
    if best.ramping {
        frac := 1.0
        if s.cfg.MigrationRamp > 0 {
            frac = float64(now.Sub(best.rampStart)) / float64(s.cfg.MigrationRamp)
            if frac > 1 {
                frac = 1
            }
            if frac < 0 {
                frac = 0
            }
        }
        w *= frac
    }
    if w <= 0 {
        w = 1
    }
    best.vtime += float64(n) / w

    return &SendToken{PathID: pkt.PathID, Seq: pkt.Seq, Size: pkt.Size, SendTime: pkt.SendTime}, true
}

// OnAck processes a new ACK for a granted token.  It returns false for
// duplicate/stale ACKs (token not found), which is important when a path
// was failed and later recovered.
func (s *Scheduler) OnAck(now time.Time, tok *SendToken) bool {
    if tok == nil {
        return false
    }
    s.mu.Lock()
    defer s.mu.Unlock()
    p, ok := s.byID[tok.PathID]
    if !ok {
        return false
    }
    pkt, ok := p.sent[tok.Seq]
    if !ok {
        return false
    }
    delete(p.sent, tok.Seq)
    p.release(pkt.Size)
    s.agg = release(s.agg, pkt.Size)

    // Estimator uses only sender-side monotonic timestamps.
    p.est.OnAck(now, pkt)

    // Anti-burst recovery credit: only ACK-clocked bytes are added back.
    if p.ramping {
        p.migrationCredit += pkt.Size
        if p.migrationCredit > s.cfg.MigrationBurst {
            p.migrationCredit = s.cfg.MigrationBurst
        }
        if now.Sub(p.rampStart) >= s.cfg.MigrationRamp {
            p.ramping = false
        }
    }

    p.onAckCwnd(s.cfg, pkt.Size)
    return true
}

// OnLoss releases in-flight budget and applies multiplicative decrease.
func (s *Scheduler) OnLoss(tok *SendToken) bool {
    if tok == nil {
        return false
    }
    s.mu.Lock()
    defer s.mu.Unlock()
    p, ok := s.byID[tok.PathID]
    if !ok {
        return false
    }
    pkt, ok := p.sent[tok.Seq]
    if !ok {
        return false
    }
    delete(p.sent, tok.Seq)
    p.release(pkt.Size)
    s.agg = release(s.agg, pkt.Size)

    p.slowStart = false
    p.ssthresh = p.cwnd / 2
    if p.ssthresh < s.cfg.MinCwnd {
        p.ssthresh = s.cfg.MinCwnd
    }
    p.cwnd = p.ssthresh
    return true
}

func (p *Path) refresh(now time.Time, cfg SchedulerConfig) {
    if p.ramping && now.Sub(p.rampStart) >= cfg.MigrationRamp {
        p.ramping = false
    }
}

func (p *Path) allowance(cfg SchedulerConfig, agg uint64) uint64 {
    if p.inflight >= p.limit || p.inflight >= p.cwnd {
        return 0
    }
    allow := p.cwnd - p.inflight
    if lim := p.limit - p.inflight; allow > lim {
        allow = lim
    }
    if p.ramping {
        if p.migrationCredit == 0 {
            return 0
        }
        if p.migrationCredit < allow {
            allow = p.migrationCredit
        }
    }
    if agg >= cfg.MaxAggregateInflight {
        return 0
    }
    if a := cfg.MaxAggregateInflight - agg; allow > a {
        allow = a
    }
    return allow
}

func (p *Path) release(n uint64) {
    if n >= p.inflight {
        p.inflight = 0
    } else {
        p.inflight -= n
    }
}

func (p *Path) onAckCwnd(cfg SchedulerConfig, acked uint64) {
    if p.slowStart && p.cwnd < p.ssthresh {
        p.cwnd += acked
        if p.cwnd > cfg.MaxCwnd {
            p.cwnd = cfg.MaxCwnd
        }
        return
    }
    p.slowStart = false
    inc := cfg.MSS * cfg.MSS / p.cwnd
    if inc == 0 {
        inc = 1
    }
    p.cwnd += inc
    if p.cwnd > cfg.MaxCwnd {
        p.cwnd = cfg.MaxCwnd
    }
}

func release(cur, n uint64) uint64 {
    if n >= cur {
        return 0
    }
    return cur - n
}
```

## Verification

Run the full suite (the commands were executed in this environment):

```bash
cd ~/multipath-scheduler
go vet ./...
go test -v ./...
go test -race ./...
```

Observed output:

```text
$ go test ./...
ok      example.com/multipath   0.002s

$ go test -v ./...
=== RUN   TestEstimatorRTTSkewImmune
--- PASS: TestEstimatorRTTSkewImmune (0.00s)
=== RUN   TestEstimatorReorderingAndDuplicates
--- PASS: TestEstimatorReorderingAndDuplicates (0.00s)
=== RUN   TestEstimatorStaleSampleRejected
--- PASS: TestEstimatorStaleSampleRejected (0.00s)
=== RUN   TestEstimatorACKCompression
--- PASS: TestEstimatorACKCompression (0.00s)
=== RUN   TestCounterWraparound
--- PASS: TestCounterWraparound (0.00s)
=== RUN   TestRobustRateClamp
--- PASS: TestRobustRateClamp (0.00s)
=== RUN   TestSchedulerAggregateAndPerPathLimits
--- PASS: TestSchedulerAggregateAndPerPathLimits (0.00s)
=== RUN   TestSchedulerPerPathLimitAlone
--- PASS: TestSchedulerPerPathLimitAlone (0.00s)
=== RUN   TestPathFailureAndRecoveryDoesNotBurst
--- PASS: TestPathFailureAndRecoveryDoesNotBurst (0.00s)
=== RUN   TestStaleACKAfterFailureIgnored
--- PASS: TestStaleACKAfterFailureIgnored (0.00s)
=== RUN   TestSequenceWraparound
--- PASS: TestSequenceWraparound (0.00s)
=== RUN   TestSchedulerUsesDeliveryRateForWFQ
--- PASS: TestSchedulerUsesDeliveryRateForWFQ (0.00s)
PASS
ok      example.com/multipath   0.002s
```

Race detector:

```text
ok      example.com/multipath   1.012s
```

The tests exercise the required failure modes directly:

| Test | What it proves |
| --- | --- |
| `TestEstimatorRTTSkewImmune` | RTT is exactly 100 ms even though a simulated remote clock has a 1-hour offset and a 0.1% drift. |
| `TestEstimatorReorderingAndDuplicates` | Out-of-order ACKs still produce `Delivered=3000` and the correct `MinRTT`. |
| `TestEstimatorStaleSampleRejected` | An ACK older than `MaxSampleAge` updates neither `Rate` nor `SRTT`. |
| `TestEstimatorACKCompression` | Ten ACKs delivered at once yield the one-segment rate, not 10x. |
| `TestCounterWraparound` | `wrapDiff` and sequence comparison are correct across `2^64` wrap; the estimator keeps a correct delta. |
| `TestRobustRateClamp` | A single outlier sample cannot raise the rate more than 1.5x. |
| `TestSchedulerAggregateAndPerPathLimits` | Aggregate and per-path in-flight limits are never exceeded. |
| `TestSchedulerPerPathLimitAlone` | The per-path cap binds even when the aggregate cap is large. |
| `TestPathFailureAndRecoveryDoesNotBurst` | Recovery sends at most `InitialCwnd` before the first ACK and is ACK-clocked afterwards. |
| `TestStaleACKAfterFailureIgnored` | ACKs for packets on a failed path do not resurrect in-flight budget. |
| `TestSequenceWraparound` | Scheduler sequence numbers wrap cleanly and ACKs still match. |
| `TestSchedulerUsesDeliveryRateForWFQ` | A path with a 10x delivery rate receives roughly 10x the scheduling opportunities. |