diff --git a/docs/examples/routing-policies/wasm/bundle/bundle.wasm b/docs/examples/routing-policies/wasm/bundle/bundle.wasm index d7f09d68fe..6f494149fd 100644 Binary files a/docs/examples/routing-policies/wasm/bundle/bundle.wasm and b/docs/examples/routing-policies/wasm/bundle/bundle.wasm differ diff --git a/pkg/router/policy/preset/names.go b/pkg/router/policy/preset/names.go index 4d0c5bc42a..dde757c0d1 100644 --- a/pkg/router/policy/preset/names.go +++ b/pkg/router/policy/preset/names.go @@ -15,6 +15,7 @@ func Names() []string { "latency-adaptive", "elastic-mux", "probe-and-prune", + "coupled", "adaptive", "geo-avoid", "transport-diverse", diff --git a/pkg/router/policy/preset/preset.go b/pkg/router/policy/preset/preset.go index 95dd85f569..c7d7c8c211 100644 --- a/pkg/router/policy/preset/preset.go +++ b/pkg/router/policy/preset/preset.go @@ -108,6 +108,8 @@ func Decide(name string, ctx Context, cands []Candidate) Spec { return decideElasticMux(ctx) case "probe-and-prune": return decideProbeAndPrune(ctx) + case "coupled": + return decideCoupled(ctx) case "adaptive": return decideAdaptive(ctx, cands) case "app-mux": @@ -234,6 +236,36 @@ func decideProbeAndPrune(ctx Context) Spec { return Spec{} } +// decideCoupled is the EXPERIMENTAL "coupled" preset's decide logic — a +// multipath policy inspired by MPTCP's coupled congestion control (LIA/OLIA). +// It provisions a modest symmetric mux (coupledMux legs) over the multi-hop +// overlay and asks the host to weight bytes toward the best legs (Distribution +// "auto" = inverse-latency), NOT spread them equally across a lossy path. The +// on_tick controller (tickCoupled) then "couples" the aggregate: it grows the +// active set CAUTIOUSLY — at most one promote per tick, and only when NO active +// leg shows rising loss (LIA's coupled increase) — and sheds the WORST leg +// (highest loss, latency-tiebroken) the moment congestion appears, so total +// aggressiveness stays bounded and traffic concentrates on the good legs. +// Latency-sensitive chat stays a single lean route; non-target apps inherit the +// visor default. +func decideCoupled(ctx Context) Spec { + switch ctx.App { + case "skychat", "skychat-client": + return Spec{Mux: 1} + case "vpn-client", "skysocks-client", "skynet-client": + return Spec{ + Mux: coupledMux, + MinHops: 2, + RotationIntervalSeconds: 20, + // Weight bytes toward the lowest-latency/lowest-loss legs (the + // "coupled" bias) rather than round-robin's equal spread across a + // congested leg. + Distribution: "auto", + } + } + return Spec{} +} + // decideAdaptive is the COMPOSITE "adaptive" preset's decide logic — the // intended converged default (the config generator wires "preset:adaptive"). // diff --git a/pkg/router/policy/preset/preset_test.go b/pkg/router/policy/preset/preset_test.go index 8c976c8ee5..1e64cbaf1e 100644 --- a/pkg/router/policy/preset/preset_test.go +++ b/pkg/router/policy/preset/preset_test.go @@ -532,3 +532,91 @@ func TestEngine_OnTick_LedbatGrowsWhenNoQueuing(t *testing.T) { t.Errorf("grow must not shed; got %+v", got) } } + +// TestDecide_Coupled pins the coupled preset's decide shape: a modest symmetric +// mux over the multi-hop overlay with best-leg (auto) byte weighting for target +// apps, a single lean route for chat, defaults for everything else. +func TestDecide_Coupled(t *testing.T) { + if got := Decide("coupled", Context{App: "skysocks-client"}, nil); !reflect.DeepEqual( + got, Spec{Mux: 4, MinHops: 2, RotationIntervalSeconds: 20, Distribution: "auto"}) { + t.Errorf("coupled/proxy: Decide=%+v", got) + } + if got := Decide("coupled", Context{App: "skychat"}, nil); !reflect.DeepEqual(got, Spec{Mux: 1}) { + t.Errorf("coupled/chat: Decide=%+v want {Mux:1}", got) + } + if got := Decide("coupled", Context{App: "other"}, nil); !reflect.DeepEqual(got, Spec{}) { + t.Errorf("coupled/other: Decide=%+v want {}", got) + } +} + +// TestEngine_OnTick_CoupledShedsWorstOnLoss drives the coupled controller with a +// clean 4-wide active set, then makes one leg's retransmits rise, and asserts the +// COUPLED DECREASE: the worst (lossy) leg is shed to standby — a lone demote, +// concentrating traffic on the good legs. +func TestEngine_OnTick_CoupledShedsWorstOnLoss(t *testing.T) { + e := New() + base := func(retrans2 uint64) []LegInfo { + return []LegInfo{ + {Index: 0, TransportID: "t0", Kind: "stcpr", LatencyMs: 50, Alive: true}, + {Index: 1, TransportID: "t1", Kind: "stcpr", LatencyMs: 50, Alive: true}, + {Index: 2, TransportID: "t2", Kind: "stcpr", LatencyMs: 50, Alive: true, Retransmits: retrans2}, + {Index: 3, TransportID: "t3", Kind: "stcpr", LatencyMs: 50, Alive: true}, + } + } + // Tick 1 establishes the retransmit baseline; a clean 4-wide set at the ceiling + // is a no-op (no grow, no loss to shed). + if got := e.OnTick("coupled", base(0)); !reflect.DeepEqual(got, RotationAction{}) { + t.Fatalf("baseline tick must be a no-op; got %+v", got) + } + // Tick 2: leg 2's retransmits jumped → rising loss → shed the worst leg (2). + got := e.OnTick("coupled", base(100)) + if len(got.DemoteToStandby) != 1 || got.DemoteToStandby[0] != 2 { + t.Errorf("coupled must shed the lossy leg (2) to standby; got %+v", got) + } + if got.AddLeg || len(got.PromoteFromStandby) != 0 || len(got.DropLegs) != 0 { + t.Errorf("coupled decrease must be a lone demote (no grow/drop); got %+v", got) + } +} + +// TestEngine_OnTick_CoupledCautiousPromoteWhenClean asserts the COUPLED INCREASE: +// a loss-free active set below the ceiling with a warm spare promotes AT MOST one +// leg (LIA's cautious coupled increase). +func TestEngine_OnTick_CoupledCautiousPromoteWhenClean(t *testing.T) { + e := New() + legs := []LegInfo{ + {Index: 0, TransportID: "t0", Kind: "stcpr", LatencyMs: 50, Alive: true}, + {Index: 1, TransportID: "t1", Kind: "stcpr", LatencyMs: 50, Alive: true}, + {Index: 2, TransportID: "t2", Kind: "stcpr", LatencyMs: 50, Alive: true}, + {Index: 3, TransportID: "s0", Kind: "stcpr", LatencyMs: 45, Alive: true, Standby: true}, + } + got := e.OnTick("coupled", legs) + if len(got.PromoteFromStandby) != 1 || got.PromoteFromStandby[0] != 3 { + t.Errorf("clean active set below the ceiling must cautiously promote the spare; got %+v", got) + } + if got.AddLeg || len(got.DemoteToStandby) != 0 || len(got.DropLegs) != 0 { + t.Errorf("cautious increase must be a lone promote; got %+v", got) + } +} + +// TestEngine_OnTick_CoupledNoGrowUnderLoss is the coupling property: even with a +// warm spare available and the active set below the ceiling, rising loss on an +// active leg FORBIDS growth — the controller sheds the lossy leg instead of +// promoting. (The loss baseline is seeded white-box so the rise reads on the very +// first tick, without an intervening clean tick consuming the spare.) +func TestEngine_OnTick_CoupledNoGrowUnderLoss(t *testing.T) { + e := New() + e.coupledPrevRetransmits["t1"] = 0 + legs := []LegInfo{ + {Index: 0, TransportID: "t0", Kind: "stcpr", LatencyMs: 50, Alive: true}, + {Index: 1, TransportID: "t1", Kind: "stcpr", LatencyMs: 50, Alive: true, Retransmits: 100}, + {Index: 2, TransportID: "t2", Kind: "stcpr", LatencyMs: 50, Alive: true}, + {Index: 3, TransportID: "s0", Kind: "stcpr", LatencyMs: 45, Alive: true, Standby: true}, + } + got := e.OnTick("coupled", legs) + if len(got.PromoteFromStandby) != 0 { + t.Errorf("coupled must NOT grow while an active leg shows rising loss (a spare is available); got %+v", got) + } + if len(got.DemoteToStandby) != 1 || got.DemoteToStandby[0] != 1 { + t.Errorf("coupled should shed the lossy leg (1) instead; got %+v", got) + } +} diff --git a/pkg/router/policy/preset/tick.go b/pkg/router/policy/preset/tick.go index 4e3a2cac32..c435d2861e 100644 --- a/pkg/router/policy/preset/tick.go +++ b/pkg/router/policy/preset/tick.go @@ -60,6 +60,13 @@ type Engine struct { ledbatEWMA map[string]float64 // per-leg EWMA-smoothed one-way-ish delay (ms) ledbatBase map[string]float64 // per-leg base (running-min smoothed) delay (ms) ledbatSeen map[string]bool // transport_ids present this tick (for GC) + + // coupled (MPTCP-style coupled congestion control) + coupledLatEWMA map[string]float64 // per-leg smoothed latency (worst-leg tiebreak) + coupledPrevRetransmits map[string]uint64 // per-leg last-tick retransmit counter (loss delta) + coupledPrevSent map[string]uint64 // per-leg last-tick sent bytes (packet-normalize loss) + coupledSeen map[string]bool // transport_ids present this tick (stale-state GC) + coupledCooldown int // ticks held steady after a grow/shed (anti-churn) } // New returns an Engine with initialized state maps and the @@ -86,6 +93,11 @@ func New() *Engine { ledbatEWMA: map[string]float64{}, ledbatBase: map[string]float64{}, ledbatSeen: map[string]bool{}, + + coupledLatEWMA: map[string]float64{}, + coupledPrevRetransmits: map[string]uint64{}, + coupledPrevSent: map[string]uint64{}, + coupledSeen: map[string]bool{}, } } @@ -103,6 +115,8 @@ func (e *Engine) OnTick(name string, legs []LegInfo) RotationAction { return e.tickElasticMux(legs) case "probe-and-prune": return e.tickProbeAndPrune(legs) + case "coupled": + return e.tickCoupled(legs) case "adaptive": return e.tickAdaptive(legs) case "ledbat": @@ -954,3 +968,155 @@ func (e *Engine) tickLedbat(legs []LegInfo) RotationAction { } return RotationAction{} } + +// --- coupled (MPTCP-style coupled congestion control) --- + +const ( + // coupledMux is the modest symmetric mux the coupled preset provisions and + // the CEILING the cautious coupled-increase grows the active set back up to. + coupledMux = 4 + // coupledFloor is the minimum active width the coupled-decrease will never + // shed below — the aggregate keeps at least this many legs carrying traffic. + coupledFloor = 2 + // coupledAlpha smooths per-leg latency (the worst-leg tiebreak signal). + coupledAlpha = 0.3 + // coupledCooldownTicks holds the active set steady for a few ticks after any + // grow/shed so a transient loss blip can't churn the mux every tick. + coupledCooldownTicks = 3 +) + +// tickCoupled is the coupled congestion controller. The two congestion signals +// are per-leg: the RETRANSMIT delta since last tick (loss, packet-normalized by +// the sent-bytes delta) and the EWMA-smoothed latency. It arbitrates at most one +// structural change per tick, in priority order: +// +// (0) no active legs -> bring one up (promote a spare, else dial) +// (1) active < floor -> restore the floor (recovery; bypasses cooldown) +// cooldown active -> hold steady (anti-churn) +// (2) any active leg's loss RISING -> COUPLED DECREASE: shed the WORST leg +// (highest loss, latency-tiebroken; never leg 0, never below the floor), +// concentrating traffic on the good legs instead of equal-spreading it +// across a lossy one. +// (3) NO active leg showing rising loss AND active < ceiling -> COUPLED +// INCREASE (LIA-cautious): promote AT MOST one warm spare. +// +// Because a shed fires the instant loss appears and a grow fires only when the +// WHOLE active set is loss-free, aggregate aggressiveness stays bounded — the +// "coupling" property. Deterministic (no time.Now/rand) for wasm parity. +func (e *Engine) tickCoupled(legs []LegInfo) RotationAction { + for k := range e.coupledSeen { + delete(e.coupledSeen, k) + } + activeCount := 0 + standbyCount := 0 + promotable := -1 + worstIdx := -1 + worstScore := -1.0 + anyRisingLoss := false + for _, l := range legs { + tid := l.TransportID + if tid != "" { + e.coupledSeen[tid] = true + } + if !l.Alive { + continue + } + if l.Standby { + standbyCount++ + if promotable == -1 || l.Index < promotable { + promotable = l.Index + } + continue + } + activeCount++ + if tid == "" { + continue + } + if l.LatencyMs > 0 { + sample := float64(l.LatencyMs) + if prev, ok := e.coupledLatEWMA[tid]; ok { + e.coupledLatEWMA[tid] = coupledAlpha*sample + (1-coupledAlpha)*prev + } else { + e.coupledLatEWMA[tid] = sample + } + } + var retransDelta, sentDelta uint64 + if prev, ok := e.coupledPrevRetransmits[tid]; ok && l.Retransmits >= prev { + retransDelta = l.Retransmits - prev + } + if prev, ok := e.coupledPrevSent[tid]; ok && l.SentBytes >= prev { + sentDelta = l.SentBytes - prev + } + e.coupledPrevRetransmits[tid] = l.Retransmits + e.coupledPrevSent[tid] = l.SentBytes + if retransDelta > 0 { + anyRisingLoss = true + } + // Loss ratio = retransmits / sent-segments this tick (a ~1KB segment unit; + // +1 avoids divide-by-zero when a leg sent nothing). Loss dominates the + // worst-leg score; smoothed latency breaks ties. Leg 0 (the primary) is + // never a shed candidate. + sentSegs := sentDelta / 1024 + lossRatio := float64(retransDelta) / float64(sentSegs+1) + score := lossRatio*1e6 + e.coupledLatEWMA[tid] + if l.Index != 0 && score > worstScore { + worstScore = score + worstIdx = l.Index + } + } + for tid := range e.coupledLatEWMA { + if !e.coupledSeen[tid] { + delete(e.coupledLatEWMA, tid) + } + } + for tid := range e.coupledPrevRetransmits { + if !e.coupledSeen[tid] { + delete(e.coupledPrevRetransmits, tid) + } + } + for tid := range e.coupledPrevSent { + if !e.coupledSeen[tid] { + delete(e.coupledPrevSent, tid) + } + } + + if e.coupledCooldown > 0 { + e.coupledCooldown-- + } + + // (0) No active legs — bring capacity up (recovery, bypasses cooldown). + if activeCount == 0 { + if promotable >= 0 { + return RotationAction{PromoteFromStandby: []int{promotable}} + } + return RotationAction{AddLeg: true} + } + // (1) Restore the floor if an active leg dropped (recovery, bypasses cooldown). + if activeCount < coupledFloor { + if promotable >= 0 { + return RotationAction{PromoteFromStandby: []int{promotable}} + } + return RotationAction{AddLeg: true} + } + // Anti-churn: hold the set steady during cooldown. + if e.coupledCooldown > 0 { + return RotationAction{} + } + + // (2) Coupled DECREASE: congestion (rising loss on any active leg) -> shed the + // worst leg so aggregate aggressiveness stays bounded. + if anyRisingLoss && activeCount > coupledFloor && worstIdx > 0 { + e.coupledCooldown = coupledCooldownTicks + if standbyCount < nodipStandbyMax { + return RotationAction{DemoteToStandby: []int{worstIdx}} + } + return RotationAction{DropLegs: []int{worstIdx}} + } + // (3) Coupled INCREASE (LIA-cautious): the whole active set is loss-free and we + // are below the ceiling -> promote at most one warm spare. + if !anyRisingLoss && activeCount < coupledMux && promotable >= 0 { + e.coupledCooldown = coupledCooldownTicks + return RotationAction{PromoteFromStandby: []int{promotable}} + } + return RotationAction{} +} diff --git a/pkg/router/policy/wasm/presets/bundle.wasm b/pkg/router/policy/wasm/presets/bundle.wasm index d7f09d68fe..6f494149fd 100755 Binary files a/pkg/router/policy/wasm/presets/bundle.wasm and b/pkg/router/policy/wasm/presets/bundle.wasm differ diff --git a/pkg/router/policy/wasm/presets/manifest.json b/pkg/router/policy/wasm/presets/manifest.json index e95fed3beb..a674893d69 100644 --- a/pkg/router/policy/wasm/presets/manifest.json +++ b/pkg/router/policy/wasm/presets/manifest.json @@ -4,6 +4,7 @@ "latency-adaptive": "latency-adaptive — mux=4 multi-hop that evicts the slowest leg each 30s (when its EWMA-smoothed latency is a >=1.5x-median outlier) until the leg set converges to low-latency disjoint paths, then holds (hysteresis-damped; no churn once converged).", "elastic-mux": "elastic-mux — AIMD scaling of the mux leg count to load: grows a leg (up to 6) when the group is saturated, releases one (down to 2) when idle.", "probe-and-prune": "probe-and-prune — periodically adds one speculative leg over a fresh path, observes its EWMA latency a few ticks, and keeps it only if it beats the current worst leg (continuous explore/exploit).", + "coupled": "coupled (experimental) — MPTCP-style coupled congestion control over a modest mux (4): weights bytes toward the best legs (auto), grows the active set cautiously (at most one promote per tick, only when NO active leg shows rising loss — LIA's coupled increase), and sheds the worst (highest-loss, latency-tiebroken) leg the instant congestion appears, so aggregate aggressiveness stays bounded and traffic concentrates on the good legs.", "adaptive": "adaptive — the app-agnostic performance+stability default (chat excepted): a single lean forward leg and a reverse mux that stays at ONE healthy leg for interactive/idle flows and only widens under sustained bulk load; holds an always-on warm-standby reserve for instant dip-free promotion, keeps gross-latency-outlier and dead legs OUT of the active mux, and rate-limits reshapes (hysteresis + cooldown) so the active set stays stable — one arbitrated action per tick.", "geo-avoid": "geo-avoid — chooses the lowest-latency route whose hops transit NONE of the countries in cli_overrides.avoid_geo (comma-separated ISO codes); defers rather than pick a violating path when none is clean.", "transport-diverse": "transport-diverse — chooses the route whose hops span the most distinct transport types (ties broken by latency), so no single transport technology failing takes the path down; seeds a mux of 2.", diff --git a/pkg/router/policy/wasm/presets/parity_test.go b/pkg/router/policy/wasm/presets/parity_test.go index 0a61771d10..4271848567 100644 --- a/pkg/router/policy/wasm/presets/parity_test.go +++ b/pkg/router/policy/wasm/presets/parity_test.go @@ -154,6 +154,8 @@ func TestDecideParity_NativeMatchesWazero(t *testing.T) { {"latency-adaptive/other", "latency-adaptive", policy.RoutingContext{App: "skychat"}, nil}, {"elastic-mux", "elastic-mux", policy.RoutingContext{App: "skysocks-client"}, nil}, {"probe-and-prune", "probe-and-prune", policy.RoutingContext{App: "skynet-client"}, nil}, + {"coupled", "coupled", policy.RoutingContext{App: "skysocks-client"}, nil}, + {"coupled/chat", "coupled", policy.RoutingContext{App: "skychat"}, nil}, {"adaptive", "adaptive", policy.RoutingContext{App: "vpn-client"}, nil}, // adaptive with real transport-kind metadata: both paths must seed the // most transport-diverse forward candidate (cand b, 2 distinct kinds). @@ -271,11 +273,30 @@ func TestTickParity_NativeMatchesWazero(t *testing.T) { {leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 40, true, true, 0), leg(2, "c", "stcpr", 40, true, true, 0)}, } + // coupled: a clean 4-wide set, then rising loss on one leg (shed), a couple of + // cooldown ticks, then a clean below-ceiling set with a warm spare (cautious + // promote) — exercises both coupled-decrease and coupled-increase. + rl := func(idx int, tid string, lat int, retrans uint64) policy.LegInfo { + return policy.LegInfo{Index: idx, TransportID: tid, Kind: "stcpr", LatencyMs: lat, Alive: true, Retransmits: retrans} + } + sb := func(idx int, tid string, lat int) policy.LegInfo { + return policy.LegInfo{Index: idx, TransportID: tid, Kind: "stcpr", LatencyMs: lat, Alive: true, Standby: true} + } + cpl := [][]policy.LegInfo{ + {rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(2, "c", 50, 0), rl(3, "d", 50, 0)}, + {rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(2, "c", 50, 500), rl(3, "d", 50, 0)}, + {rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(3, "d", 50, 0), sb(2, "c", 50)}, + {rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(3, "d", 50, 0), sb(2, "c", 50)}, + {rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(3, "d", 50, 0), sb(2, "c", 50)}, + {rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(3, "d", 50, 0), sb(2, "c", 50)}, + } + cases := []tickCase{ {"rotating-bw", "rotating-bw", rbw}, {"latency-adaptive", "latency-adaptive", la}, {"elastic-mux", "elastic-mux", em}, {"probe-and-prune", "probe-and-prune", pp}, + {"coupled", "coupled", cpl}, {"adaptive", "adaptive", ad}, {"ledbat", "ledbat", lb}, }