feat(routing-policy): experimental 'ledbat' delay-based scavenger preset

pkg/router/policy/preset/names.go

@@ -20,6 +20,7 @@ func Names() []string {
 		"transport-diverse",
 		"trust-tiered",
 		"time-of-day",
+		"ledbat",
 	}
 }
 

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

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)
+	}
+}

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{}
+}

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."
 }

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 {