diff --git a/docs/examples/routing-policies/wasm/bundle/bundle.wasm b/docs/examples/routing-policies/wasm/bundle/bundle.wasm index 51c22f44d5..d7f09d68fe 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 45fa6a3f6d..4d0c5bc42a 100644 --- a/pkg/router/policy/preset/names.go +++ b/pkg/router/policy/preset/names.go @@ -20,6 +20,7 @@ func Names() []string { "transport-diverse", "trust-tiered", "time-of-day", + "ledbat", } } diff --git a/pkg/router/policy/preset/preset.go b/pkg/router/policy/preset/preset.go index 3c7718fcc8..95dd85f569 100644 --- a/pkg/router/policy/preset/preset.go +++ b/pkg/router/policy/preset/preset.go @@ -120,6 +120,8 @@ func Decide(name string, ctx Context, cands []Candidate) Spec { return decideTrustTiered(ctx, cands) case "time-of-day": return decideTimeOfDay(ctx) + case "ledbat": + return decideLedbat(ctx) default: return decideAppMux(ctx) } @@ -330,6 +332,36 @@ func mostTransportDiverse(cands []Candidate) *Candidate { return best } +// ledbatMux is the small symmetric mux the ledbat preset provisions and the +// hard CAP its on_tick controller may grow the active set back up to. It is +// deliberately lean: ledbat is a background / scavenger policy, so it holds few +// legs and yields (shrinks toward one) the moment it detects it is queuing. +const ledbatMux = 3 + +// decideLedbat is the EXPERIMENTAL ledbat preset's decide logic — a delay-based, +// background/scavenger multipath policy inspired by LEDBAT congestion control +// (RFC 6817). It provisions a small symmetric mux (ledbatMux legs) over +// multi-hop with "auto" (latency-weighted) byte distribution, re-evaluated every +// 20s so on_tick can shrink or grow the active set from the measured queuing +// delay. Latency-sensitive chat stays a single lean route (same idiom as the +// other presets). The scavenger behavior itself lives in tickLedbat: it starts +// at the small provisioned width and BACKS OFF toward a single leg whenever a +// leg's smoothed delay rises meaningfully above its own base (min) delay — +// yielding capacity to other traffic — and grows back up to ledbatMux only while +// every active leg reads near its base (no self-induced queuing). +func decideLedbat(ctx Context) Spec { + switch ctx.App { + case "skychat", "skychat-client": + return Spec{Mux: 1} + } + return Spec{ + Mux: ledbatMux, + MinHops: 2, + RotationIntervalSeconds: 20, + Distribution: "auto", + } +} + // --- conditional presets: constrain WHICH path is chosen, by route metadata --- // splitSet parses a comma/space-separated override value into a lowercased diff --git a/pkg/router/policy/preset/preset_test.go b/pkg/router/policy/preset/preset_test.go index 947092473c..8c976c8ee5 100644 --- a/pkg/router/policy/preset/preset_test.go +++ b/pkg/router/policy/preset/preset_test.go @@ -22,6 +22,8 @@ func TestDecide_ShapePresets(t *testing.T) { {"adaptive", Context{App: "vpn-client"}, Spec{ForwardMux: 1, ReverseMux: 3, RotationIntervalSeconds: 20, Distribution: "auto"}}, {"adaptive/chat", Context{App: "skychat"}, Spec{Mux: 1}}, {"adaptive/custom-session", Context{App: "g8"}, Spec{ForwardMux: 1, ReverseMux: 3, RotationIntervalSeconds: 20, Distribution: "auto"}}, + {"ledbat", Context{App: "skysocks-client"}, Spec{Mux: 3, MinHops: 2, RotationIntervalSeconds: 20, Distribution: "auto"}}, + {"ledbat/chat", Context{App: "skychat"}, Spec{Mux: 1}}, } for _, tc := range cases { presetName, _, _ := splitName(tc.name) @@ -427,3 +429,106 @@ func TestEngine_OnTick_RotatingBWParksFragile(t *testing.T) { t.Errorf("rotating-bw should park the fragile active leg; got %+v want %+v", got, want) } } + +// TestEngine_OnTick_LedbatBacksOffOnQueuingDelay drives the ledbat scavenger +// with three active legs where leg 2's delay climbs far above its base (min) +// delay while legs 0 and 1 stay near theirs. Once leg 2's smoothed queuing delay +// exceeds the target, the controller must BACK OFF by parking the highest-delay +// non-primary active leg (leg 2) to warm standby — yielding capacity — and never +// tear a leg down or touch leg 0. +func TestEngine_OnTick_LedbatBacksOffOnQueuingDelay(t *testing.T) { + e := New() + // Legs 0,1 hold a steady ~40ms; leg 2 starts at its base 40ms then spikes to + // 300ms so its EWMA rises well past base+target (60ms). + steady := func(lat2 int) []LegInfo { + return []LegInfo{ + {Index: 0, TransportID: "t0", Kind: "stcpr", LatencyMs: 40, Alive: true}, + {Index: 1, TransportID: "t1", Kind: "stcpr", LatencyMs: 40, Alive: true}, + {Index: 2, TransportID: "t2", Kind: "stcpr", LatencyMs: lat2, Alive: true}, + } + } + // First tick seeds base=40 for every leg (EWMA==sample), no queuing yet. + if got := e.OnTick("ledbat", steady(40)); !reflect.DeepEqual(got, RotationAction{}) { + t.Fatalf("seed tick: at base delay, no queuing → no-op; got %+v", got) + } + // Now leg 2 congests. Feed the spike until its EWMA queuing delay crosses the + // target and the controller backs off (parks leg 2). Bounded loop. + var got RotationAction + for i := 0; i < 10; i++ { + got = e.OnTick("ledbat", steady(300)) + if len(got.DemoteToStandby) > 0 { + break + } + if !reflect.DeepEqual(got, RotationAction{}) { + t.Fatalf("pre-backoff tick %d: expected no-op or the backoff demote; got %+v", i, got) + } + } + if len(got.DemoteToStandby) != 1 || got.DemoteToStandby[0] != 2 { + t.Fatalf("ledbat must park the highest-queuing-delay non-primary leg (2); got %+v", got) + } + if got.DropLegs != nil { + t.Errorf("back-off parks, never tears down; got DropLegs=%v", got.DropLegs) + } + if len(got.PromoteFromStandby) != 0 || got.AddLeg { + t.Errorf("back-off must not grow; got %+v", got) + } +} + +// TestEngine_OnTick_LedbatNeverParksBelowFloor asserts the yield floor: even when +// the sole non-primary active leg is congesting, ledbat shrinks only to a single +// active leg (leg 0) and then holds — it never parks leg 0, so the flow always +// keeps one leg making progress. +func TestEngine_OnTick_LedbatNeverParksBelowFloor(t *testing.T) { + e := New() + legs := func(lat1 int, oneStandby bool) []LegInfo { + return []LegInfo{ + {Index: 0, TransportID: "t0", Kind: "stcpr", LatencyMs: 40, Alive: true}, + {Index: 1, TransportID: "t1", Kind: "stcpr", LatencyMs: lat1, Alive: true, Standby: oneStandby}, + } + } + // Seed base=40 for both legs. + e.OnTick("ledbat", legs(40, false)) + // Congest leg 1 until it is parked, then confirm at the floor (leg 0 only + // active, leg 1 standby) the controller does NOT keep demoting. + parked := false + for i := 0; i < 10 && !parked; i++ { + got := e.OnTick("ledbat", legs(300, false)) + if len(got.DemoteToStandby) == 1 && got.DemoteToStandby[0] == 1 { + parked = true + } else if len(got.DemoteToStandby) > 0 { + t.Fatalf("must only ever park leg 1 (never leg 0); got %+v", got) + } + } + if !parked { + t.Fatal("leg 1 never parked despite sustained queuing delay") + } + // At the floor (1 active), a still-congested snapshot must be a no-op — leg 0 + // is never parked. + if got := e.OnTick("ledbat", legs(300, true)); len(got.DemoteToStandby) != 0 || got.DropLegs != nil { + t.Errorf("at the yield floor ledbat must hold, never park leg 0; got %+v", got) + } +} + +// TestEngine_OnTick_LedbatGrowsWhenNoQueuing asserts the grow rule: with a parked +// reserve leg and every active leg reading near its base (no self-induced +// queuing), the controller re-promotes a parked leg — up to the ledbatMux cap — +// drawing only on the reserve (no fresh dial). +func TestEngine_OnTick_LedbatGrowsWhenNoQueuing(t *testing.T) { + e := New() + // Leg 0 active near base, legs 1 and 2 parked. No queuing anywhere. + legs := []LegInfo{ + {Index: 0, TransportID: "t0", Kind: "stcpr", LatencyMs: 40, Alive: true}, + {Index: 1, TransportID: "t1", Kind: "stcpr", LatencyMs: 40, Alive: true, Standby: true}, + {Index: 2, TransportID: "t2", Kind: "stcpr", LatencyMs: 40, Alive: true, Standby: true}, + } + got := e.OnTick("ledbat", legs) + if len(got.PromoteFromStandby) != 1 || got.PromoteFromStandby[0] != 1 { + t.Fatalf("no queuing + below cap → promote the lowest-index parked leg (1); got %+v", got) + } + if got.AddLeg { + t.Errorf("grow must draw on the reserve, never dial fresh; got AddLeg=true") + } + if len(got.DemoteToStandby) != 0 || got.DropLegs != nil { + t.Errorf("grow must not shed; got %+v", got) + } +} diff --git a/pkg/router/policy/preset/tick.go b/pkg/router/policy/preset/tick.go index c6337d8a8a..4e3a2cac32 100644 --- a/pkg/router/policy/preset/tick.go +++ b/pkg/router/policy/preset/tick.go @@ -55,6 +55,11 @@ type Engine struct { adaptSatTicks int // consecutive saturated ticks (grow only on a sustained signal) adaptUnhealthy map[string]int // per-active-leg consecutive gross-outlier-latency ticks adaptStall map[string]int // per-active-leg consecutive no-throughput ticks under load + + // ledbat (delay-based scavenger) + 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) } // New returns an Engine with initialized state maps and the @@ -78,6 +83,9 @@ func New() *Engine { adaptUnhealthy: map[string]int{}, adaptStall: map[string]int{}, adaptTarget: adaptRevActive, + ledbatEWMA: map[string]float64{}, + ledbatBase: map[string]float64{}, + ledbatSeen: map[string]bool{}, } } @@ -97,6 +105,8 @@ func (e *Engine) OnTick(name string, legs []LegInfo) RotationAction { return e.tickProbeAndPrune(legs) case "adaptive": return e.tickAdaptive(legs) + case "ledbat": + return e.tickLedbat(legs) default: return RotationAction{} } @@ -823,3 +833,124 @@ func legUnhealthyLat(lat, median float64) bool { } return false } + +// --- ledbat (delay-based scavenger) --- + +const ( + // ledbatAlpha is the EWMA smoothing factor for the per-leg delay signal — + // matched to the other controllers so a leg's smoothed delay reacts at the + // same rate everywhere. + ledbatAlpha = 0.3 + // ledbatTargetMs is the LEDBAT queuing-delay target (RFC 6817 uses 100ms; a + // tighter 60ms here makes the scavenger yield sooner). When a leg's smoothed + // delay rises more than this above its own base (min) delay, the flow is + // judged to be queuing — i.e. causing congestion — and the controller backs + // off. While every active leg stays within the target of its base, there is + // no self-induced queuing and the controller may grow. + ledbatTargetMs = 60.0 + // ledbatMinActive is the yield floor: the controller shrinks the active set + // toward this width under congestion but never below it, so the flow always + // keeps one leg making progress. + ledbatMinActive = 1 +) + +// tickLedbat is the ledbat preset's on_tick controller: a delay-based, +// background/scavenger congestion response over the mux active set. It tracks +// each leg's EWMA-smoothed delay and a per-leg base (running-min) delay, then +// derives the queuing delay (smoothed - base): +// +// - BACK OFF (yield): if ANY active leg's queuing delay exceeds ledbatTargetMs +// — the flow is causing congestion — demote the HIGHEST-queuing-delay active +// leg (never leg 0) to warm standby, shrinking the active set toward +// ledbatMinActive. Parked, not torn down, so it can be re-promoted for free +// when the congestion clears. +// - GROW: if every active leg reads within the target of its base (no +// self-induced queuing) and the active set is below ledbatMux, promote one +// warm standby leg back. Growth only ever draws on the parked reserve — the +// scavenger never dials fresh legs beyond its lean provisioned width. +// +// One arbitrated action per tick, like the other controllers. Pure integer/ +// float arithmetic and map state only (no time.Now / rand) so it is +// deterministic and identical under wazero and native compilation. +func (e *Engine) tickLedbat(legs []LegInfo) RotationAction { + for k := range e.ledbatSeen { + delete(e.ledbatSeen, k) + } + for _, l := range legs { + tid := l.TransportID + if tid == "" { + continue + } + e.ledbatSeen[tid] = true + if !l.Alive || l.LatencyMs <= 0 { + continue + } + sample := float64(l.LatencyMs) + if prev, ok := e.ledbatEWMA[tid]; ok { + e.ledbatEWMA[tid] = ledbatAlpha*sample + (1-ledbatAlpha)*prev + } else { + e.ledbatEWMA[tid] = sample + } + sm := e.ledbatEWMA[tid] + if b, ok := e.ledbatBase[tid]; !ok || sm < b { + e.ledbatBase[tid] = sm + } + } + for tid := range e.ledbatEWMA { + if !e.ledbatSeen[tid] { + delete(e.ledbatEWMA, tid) + } + } + for tid := range e.ledbatBase { + if !e.ledbatSeen[tid] { + delete(e.ledbatBase, tid) + } + } + + sets := classifyLegs(legs) + // No active legs — bring one up from the parked reserve (no fresh dial). + if sets.activeCount == 0 { + if sets.promotable >= 0 { + return RotationAction{PromoteFromStandby: []int{sets.promotable}} + } + return RotationAction{} + } + + // Queuing delay across the active set. maxQ (including leg 0) decides whether + // the flow is congesting; worstIdx (excluding leg 0) is the demote target so + // the primary leg is never parked. + maxQ := 0.0 + worstIdx := -1 + worstQ := 0.0 + for _, l := range legs { + if !l.Alive || l.Standby { + continue + } + sm, ok := e.ledbatEWMA[l.TransportID] + if !ok { + continue + } + q := sm - e.ledbatBase[l.TransportID] + if q > maxQ { + maxQ = q + } + if l.Index == 0 { + continue + } + if worstIdx == -1 || q > worstQ { + worstIdx, worstQ = l.Index, q + } + } + + // BACK OFF: congestion detected — park the worst active (non-primary) leg, + // yielding capacity, until we reach the floor. + if maxQ > ledbatTargetMs && sets.activeCount > ledbatMinActive && worstIdx >= 0 { + return RotationAction{DemoteToStandby: []int{worstIdx}} + } + + // GROW: no self-induced queuing and below the cap — re-promote one parked leg. + if maxQ <= ledbatTargetMs && sets.activeCount < ledbatMux && sets.promotable >= 0 { + return RotationAction{PromoteFromStandby: []int{sets.promotable}} + } + return RotationAction{} +} diff --git a/pkg/router/policy/wasm/presets/bundle.wasm b/pkg/router/policy/wasm/presets/bundle.wasm index 51c22f44d5..d7f09d68fe 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 3fb704612b..e95fed3beb 100644 --- a/pkg/router/policy/wasm/presets/manifest.json +++ b/pkg/router/policy/wasm/presets/manifest.json @@ -8,5 +8,6 @@ "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.", "trust-tiered": "trust-tiered — prefers routes transiting only trusted intermediaries (cli_overrides.trusted_pks); tiered fallback to the most-trusted route available rather than failing closed.", - "time-of-day": "time-of-day — switches route shape by wall-clock hour (cli_overrides.business_hours, default 9-17): lean single route during business hours, wide rotating privacy mux off-hours." + "time-of-day": "time-of-day — switches route shape by wall-clock hour (cli_overrides.business_hours, default 9-17): lean single route during business hours, wide rotating privacy mux off-hours.", + "ledbat": "ledbat — EXPERIMENTAL delay-based background/scavenger multipath (LEDBAT-inspired): a lean mux=3 that tracks each leg's queuing delay (smoothed delay above its base/min) and BACKS OFF — parks the highest-delay active leg toward a single leg — once queuing exceeds ~60ms, yielding to other traffic, then re-promotes parked legs (up to 3) only while every active leg stays near its base. Does not change any default." } diff --git a/pkg/router/policy/wasm/presets/parity_test.go b/pkg/router/policy/wasm/presets/parity_test.go index 4ccb1c54d0..0a61771d10 100644 --- a/pkg/router/policy/wasm/presets/parity_test.go +++ b/pkg/router/policy/wasm/presets/parity_test.go @@ -165,6 +165,8 @@ func TestDecideParity_NativeMatchesWazero(t *testing.T) { {"time-of-day/biz", "time-of-day", policy.RoutingContext{App: "skysocks-client", Now: unixNano(11 * oneHour)}, nil}, {"time-of-day/off", "time-of-day", policy.RoutingContext{App: "skysocks-client", Now: unixNano(3 * oneHour)}, nil}, {"time-of-day/window", "time-of-day", policy.RoutingContext{App: "skysocks-client", Now: unixNano(23 * oneHour), CLIOverrides: map[string]string{"business_hours": "22-6"}}, nil}, + {"ledbat", "ledbat", policy.RoutingContext{App: "skysocks-client"}, nil}, + {"ledbat/chat", "ledbat", policy.RoutingContext{App: "skychat"}, nil}, } for _, tc := range cases { @@ -259,12 +261,23 @@ func TestTickParity_NativeMatchesWazero(t *testing.T) { {leg(0, "a", "stcpr", 43, true, false, 30001), leg(1, "b", "stcpr", 980, true, false, 30001)}, } + // ledbat: leg 2's delay climbs above its base while legs 0,1 stay near theirs + // (the scavenger backs off and parks leg 2); a final recovered snapshot with + // parked spares exercises the re-grow path — every step must agree. + lb := [][]policy.LegInfo{ + {leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 40, true, false, 0), leg(2, "c", "stcpr", 40, true, false, 0)}, + {leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 40, true, false, 0), leg(2, "c", "stcpr", 300, true, false, 0)}, + {leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 40, true, false, 0), leg(2, "c", "stcpr", 300, true, false, 0)}, + {leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 40, true, true, 0), leg(2, "c", "stcpr", 40, true, true, 0)}, + } + cases := []tickCase{ {"rotating-bw", "rotating-bw", rbw}, {"latency-adaptive", "latency-adaptive", la}, {"elastic-mux", "elastic-mux", em}, {"probe-and-prune", "probe-and-prune", pp}, {"adaptive", "adaptive", ad}, + {"ledbat", "ledbat", lb}, } for _, tc := range cases {