diff --git a/docs/examples/routing-policies/wasm/bundle/main.go b/docs/examples/routing-policies/wasm/bundle/main.go index a13863fb25..c236f49ce2 100644 --- a/docs/examples/routing-policies/wasm/bundle/main.go +++ b/docs/examples/routing-policies/wasm/bundle/main.go @@ -108,6 +108,7 @@ type rotationActionWire struct { ExcludeHops []string `json:"exclude_hops,omitempty"` DemoteToStandby []int `json:"demote_to_standby,omitempty"` PromoteFromStandby []int `json:"promote_from_standby,omitempty"` + AddForwardLeg bool `json:"add_forward_leg,omitempty"` } // engine holds the adaptive tick controllers' per-transport_id state for this @@ -269,6 +270,7 @@ func actionToWire(a preset.RotationAction) rotationActionWire { ExcludeHops: a.ExcludeHops, DemoteToStandby: a.DemoteToStandby, PromoteFromStandby: a.PromoteFromStandby, + AddForwardLeg: a.AddForwardLeg, } } diff --git a/pkg/router/dial_hook.go b/pkg/router/dial_hook.go index 4bcc679a02..21481f68d1 100644 --- a/pkg/router/dial_hook.go +++ b/pkg/router/dial_hook.go @@ -352,6 +352,14 @@ type RotationAction struct { // valid. DemoteToStandby []int PromoteFromStandby []int + + // AddForwardLeg requests one more FORWARD-ONLY aux leg — appended + // addFwd=true / addRev=false so it adds upstream send capacity without + // enlarging the reverse/download set. The rotation loop dials it via the + // forward-only add callback (router.addOneAuxSendLeg). The forward- + // direction mirror of AddLeg; emitted by the adaptive preset under + // sustained upload (SentBytes) saturation. ExcludeHops applies to it too. + AddForwardLeg bool } // RotationHook fires periodically per active route group, giving diff --git a/pkg/router/policy/bridge.go b/pkg/router/policy/bridge.go index 33db906dcd..90ba518b22 100644 --- a/pkg/router/policy/bridge.go +++ b/pkg/router/policy/bridge.go @@ -403,8 +403,9 @@ func parseRotationAction(v starlark.Value) (RotationAction, error) { return RotationAction{}, fmt.Errorf("expected struct, got %s", v.Type()) } out := RotationAction{ - AddLeg: readBoolField(s, "add_leg"), - ExcludeHops: readStrListField(s, "exclude_hops"), + AddLeg: readBoolField(s, "add_leg"), + AddForwardLeg: readBoolField(s, "add_forward_leg"), + ExcludeHops: readStrListField(s, "exclude_hops"), } if drops, err := s.Attr("drop_legs"); err == nil && drops != nil { if _, isNone := drops.(starlark.NoneType); !isNone { diff --git a/pkg/router/policy/hook.go b/pkg/router/policy/hook.go index c5fe1afff4..e891408825 100644 --- a/pkg/router/policy/hook.go +++ b/pkg/router/policy/hook.go @@ -395,6 +395,7 @@ func (h *Hook) OnTick(info router.DialInfo, legs []router.LegInfo) router.Rotati ExcludeHops: append([]string(nil), action.ExcludeHops...), DemoteToStandby: append([]int(nil), action.DemoteToStandby...), PromoteFromStandby: append([]int(nil), action.PromoteFromStandby...), + AddForwardLeg: action.AddForwardLeg, } } diff --git a/pkg/router/policy/preset/preset.go b/pkg/router/policy/preset/preset.go index c7d7c8c211..5c98d61cf9 100644 --- a/pkg/router/policy/preset/preset.go +++ b/pkg/router/policy/preset/preset.go @@ -92,6 +92,11 @@ type RotationAction struct { ExcludeHops []string DemoteToStandby []int PromoteFromStandby []int + // AddForwardLeg requests one more FORWARD-ONLY aux leg (the router appends + // it addFwd=true / addRev=false) — extra upstream send capacity that does + // NOT enlarge the reverse/download set. The adaptive preset emits this on + // sustained upload saturation, the forward-direction mirror of AddLeg. + AddForwardLeg bool } // Decide dispatches to the named preset's decide logic. The name diff --git a/pkg/router/policy/preset/preset_test.go b/pkg/router/policy/preset/preset_test.go index 1e64cbaf1e..e6da8f97ca 100644 --- a/pkg/router/policy/preset/preset_test.go +++ b/pkg/router/policy/preset/preset_test.go @@ -620,3 +620,174 @@ func TestEngine_OnTick_CoupledNoGrowUnderLoss(t *testing.T) { t.Errorf("coupled should shed the lossy leg (1) instead; got %+v", got) } } + +// --- adaptive BIDIRECTIONAL sizing (forward/upload widening) --- +// +// These exercise the SentBytes-driven forward controller added alongside the +// existing RecvBytes-driven reverse controller. adaptiveSim is a faithful +// mini-router: it feeds the engine a leg snapshot, applies the returned +// RotationAction to the leg set (append on AddLeg/AddForwardLeg, flip Standby on +// promote/demote, remove on drop), then advances the cumulative byte counters +// for the next tick — so the engine sees leg growth react to its own actions. +type adaptiveSim struct { + e *Engine + legs []LegInfo + nextTID int + sawAddFwd bool + sawAddLeg bool + sawPromote bool + sentPerTick uint64 // bytes added to each ACTIVE leg's SentBytes each tick + recvPerTick uint64 // bytes added to each ACTIVE leg's RecvBytes each tick +} + +func newAdaptiveSim(sentPerTick, recvPerTick uint64) *adaptiveSim { + s := &adaptiveSim{e: New(), sentPerTick: sentPerTick, recvPerTick: recvPerTick, nextTID: 1} + // Steady start: one active forward/primary leg (leg 0). No standby — the + // simplest shape that isolates the sizing controllers. + s.legs = []LegInfo{{Index: 0, TransportID: "t0", Kind: "stcpr", LatencyMs: 40, Alive: true}} + return s +} + +func (s *adaptiveSim) activeCount() int { + n := 0 + for _, l := range s.legs { + if l.Alive && !l.Standby { + n++ + } + } + return n +} + +func (s *adaptiveSim) reindex() { + for i := range s.legs { + s.legs[i].Index = i + } +} + +func (s *adaptiveSim) step() RotationAction { //nolint:unparam // test helper: return kept for call-site clarity + act := s.e.OnTick("adaptive", s.legs) + if act.AddForwardLeg { + s.sawAddFwd = true + } + if act.AddLeg { + s.sawAddLeg = true + } + if len(act.PromoteFromStandby) > 0 { + s.sawPromote = true + } + // Apply promote/demote first (index-stable), then drops (compact), then adds. + for _, idx := range act.PromoteFromStandby { + if idx >= 0 && idx < len(s.legs) { + s.legs[idx].Standby = false + } + } + for _, idx := range act.DemoteToStandby { + if idx >= 0 && idx < len(s.legs) { + s.legs[idx].Standby = true + } + } + if len(act.DropLegs) > 0 { + drop := map[int]bool{} + for _, idx := range act.DropLegs { + drop[idx] = true + } + var kept []LegInfo + for i, l := range s.legs { + if !drop[i] { + kept = append(kept, l) + } + } + s.legs = kept + s.reindex() + } + if act.AddLeg || act.AddForwardLeg { + tid := "t" + string(rune('a'+s.nextTID)) //nolint:gosec // G115: bounded test rune + s.nextTID++ + s.legs = append(s.legs, LegInfo{Index: len(s.legs), TransportID: tid, Kind: "stcpr", LatencyMs: 40, Alive: true}) + } + // Advance cumulative counters on the ACTIVE legs for the next snapshot. + for i := range s.legs { + if s.legs[i].Alive && !s.legs[i].Standby { + s.legs[i].SentBytes += s.sentPerTick + s.legs[i].RecvBytes += s.recvPerTick + } + } + return act +} + +// TestEngine_OnTick_AdaptiveForwardWidensOnUpload asserts an upload-heavy flow +// (growing SentBytes, flat RecvBytes) widens the FORWARD mux via AddForwardLeg +// while the reverse controller stays completely lean (no AddLeg, reverse target +// unchanged at adaptRevActive). +func TestEngine_OnTick_AdaptiveForwardWidensOnUpload(t *testing.T) { + s := newAdaptiveSim(1_000_000, 0) // heavy upload, zero download + for i := 0; i < 20; i++ { + s.step() + } + if !s.sawAddFwd { + t.Fatalf("upload-heavy flow must widen forward via AddForwardLeg; never saw one") + } + if s.sawAddLeg { + t.Errorf("upload-heavy flow must NOT grow the reverse/full-duplex set (AddLeg)") + } + if s.e.adaptTarget != adaptRevActive { + t.Errorf("reverse target must stay lean at %d; got %d", adaptRevActive, s.e.adaptTarget) + } + if s.e.adaptFwdTarget <= adaptFwdActive { + t.Errorf("forward target must have widened above %d; got %d", adaptFwdActive, s.e.adaptFwdTarget) + } + if s.activeCount() <= 1 { + t.Errorf("forward widening must have added at least one active send leg; active=%d", s.activeCount()) + } +} + +// TestEngine_OnTick_AdaptiveReverseWidensOnDownload is the REGRESSION guard: a +// download-heavy flow (growing RecvBytes, flat SentBytes) must still widen the +// reverse set via AddLeg exactly as before, and must NOT trip the new forward +// controller (no AddForwardLeg; forward target stays at adaptFwdActive). +func TestEngine_OnTick_AdaptiveReverseWidensOnDownload(t *testing.T) { + s := newAdaptiveSim(0, 1_000_000) // zero upload, heavy download + for i := 0; i < 20; i++ { + s.step() + } + if !s.sawAddLeg && !s.sawPromote { + t.Fatalf("download-heavy flow must widen the reverse set (AddLeg/promote); saw neither") + } + if s.sawAddFwd { + t.Errorf("download-heavy flow must NOT trip the forward controller (AddForwardLeg)") + } + if s.e.adaptFwdTarget != adaptFwdActive { + t.Errorf("forward target must stay lean at %d on a download flow; got %d", adaptFwdActive, s.e.adaptFwdTarget) + } + if s.e.adaptTarget <= adaptRevActive { + t.Errorf("reverse target must have widened above %d; got %d", adaptRevActive, s.e.adaptTarget) + } +} + +// TestEngine_OnTick_AdaptiveForwardCollapsesOnIdle asserts a forward-widened +// flow collapses back to the lean single forward leg once the upload goes idle +// (forward target returns to adaptFwdActive and the active set shrinks). +func TestEngine_OnTick_AdaptiveForwardCollapsesOnIdle(t *testing.T) { + s := newAdaptiveSim(1_000_000, 0) + for i := 0; i < 20; i++ { // widen under upload + s.step() + } + if s.e.adaptFwdTarget <= adaptFwdActive { + t.Fatalf("precondition: forward must be widened; got fwdTarget=%d", s.e.adaptFwdTarget) + } + grown := s.activeCount() + // Upload stops: flat SentBytes and RecvBytes from here on. + s.sentPerTick, s.recvPerTick = 0, 0 + for i := 0; i < 40; i++ { + s.step() + } + if s.e.adaptFwdTarget != adaptFwdActive { + t.Errorf("idle must collapse forward target back to %d; got %d", adaptFwdActive, s.e.adaptFwdTarget) + } + if s.activeCount() >= grown { + t.Errorf("idle must shed the extra forward legs; active stayed %d (peak %d)", s.activeCount(), grown) + } + if s.activeCount() != 1 { + t.Errorf("idle steady state must be a single active forward leg; active=%d", s.activeCount()) + } +} diff --git a/pkg/router/policy/preset/tick.go b/pkg/router/policy/preset/tick.go index c435d2861e..7a2ccf058d 100644 --- a/pkg/router/policy/preset/tick.go +++ b/pkg/router/policy/preset/tick.go @@ -50,6 +50,21 @@ type Engine struct { adaptSeeded bool adaptIdleCount int adaptTarget int + // forward (upload / SentBytes) sizing — the exact mirror of the reverse + // (RecvBytes) machine above, driven by SentBytes deltas. adaptFwdTarget is + // the forward active width (seeded to adaptFwdActive); an upload-heavy flow + // widens it under sustained SENT saturation and collapses it back when the + // upload goes idle, using the SAME EWMA/peak/hysteresis/cooldown constants + // as the reverse side. When SentBytes never advances the whole forward + // machine stays dormant (adaptFwdSeeded never trips), so a download-only or + // idle flow behaves byte-identically to the reverse-only controller. + adaptPrevSent map[string]uint64 + adaptFwdThroughputEWMA float64 + adaptFwdPeak float64 + adaptFwdSeeded bool + adaptFwdIdleCount int + adaptFwdTarget int + adaptFwdSatTicks int // health + anti-churn state adaptCooldown int // ticks to hold the active set steady after a reshape adaptSatTicks int // consecutive saturated ticks (grow only on a sustained signal) @@ -85,11 +100,13 @@ func New() *Engine { prevKnownTIDs: map[string]bool{}, adaptLatEWMA: map[string]float64{}, adaptPrevRecv: map[string]uint64{}, + adaptPrevSent: map[string]uint64{}, adaptSeen: map[string]bool{}, adaptAliveIdx: map[string]int{}, adaptUnhealthy: map[string]int{}, adaptStall: map[string]int{}, adaptTarget: adaptRevActive, + adaptFwdTarget: adaptFwdActive, ledbatEWMA: map[string]float64{}, ledbatBase: map[string]float64{}, ledbatSeen: map[string]bool{}, @@ -508,9 +525,13 @@ func (e *Engine) tickProbeAndPrune(legs []LegInfo) RotationAction { // --- adaptive (composite) --- const ( - // adaptFwdActive is the lean forward (upstream / request) leg count: a - // single low-latency route, so the request path never pays mux head-of-line - // cost. adaptRevActive is the STEADY active reverse width — deliberately 1: + // adaptFwdActive is the STEADY lean forward (upstream / request) leg count: a + // single low-latency route, so an interactive / idle request path pays no mux + // head-of-line cost. Like the reverse side it is only a STEADY floor: the + // forward mux WIDENS under sustained UPLOAD (SentBytes) saturation — tickAdaptive + // grows adaptFwdTarget and emits AddForwardLeg (forward-only aux legs) — then + // collapses back to adaptFwdActive when the upload goes idle. adaptRevActive is + // the STEADY active reverse width — deliberately 1: // an interactive / idle flow rides ONE good leg (leg 0) and is never // scattered+reordered across high-variance legs. The reverse mux only WIDENS // under sustained bulk load (tickAdaptive promotes warm spares), then shrinks @@ -577,6 +598,7 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { delete(e.adaptAliveIdx, k) } var rawTotal float64 + var rawSent float64 aliveCount := 0 standbyCount := 0 newestAliveIdx := -1 @@ -618,6 +640,12 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { movedRecv[tid] = true } e.adaptPrevRecv[tid] = l.RecvBytes + // Forward (upload) throughput — the SentBytes mirror of the RecvBytes + // accumulation above. Feeds the independent forward sizing machine. + if prev, ok := e.adaptPrevSent[tid]; ok && l.SentBytes > prev { + rawSent += float64(l.SentBytes - prev) + } + e.adaptPrevSent[tid] = l.SentBytes } for tid := range e.adaptLatEWMA { if !e.adaptSeen[tid] { @@ -629,6 +657,11 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { delete(e.adaptPrevRecv, tid) } } + for tid := range e.adaptPrevSent { + if !e.adaptSeen[tid] { + delete(e.adaptPrevSent, tid) + } + } saturated, idle := false, false if !e.adaptSeeded { @@ -654,6 +687,35 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { e.adaptIdleCount = 0 } + // Forward (upload) saturation/idle — the SentBytes mirror of the block + // above, using the SAME EWMA smoothing, peak decay, and 0.80/0.25 + // saturated/idle thresholds. Stays fully dormant (fwdSaturated / fwdIdle + // both false, adaptFwdSeeded never set) while SentBytes is flat, so a + // download-only or idle flow evolves exactly as before. + fwdSaturated, fwdIdle := false, false + if !e.adaptFwdSeeded { + if rawSent > 0 { + e.adaptFwdThroughputEWMA = rawSent + e.adaptFwdPeak = rawSent + e.adaptFwdSeeded = true + } + } else { + e.adaptFwdThroughputEWMA = adaptAlpha*rawSent + (1-adaptAlpha)*e.adaptFwdThroughputEWMA + e.adaptFwdPeak *= adaptPeakDecay + if e.adaptFwdThroughputEWMA > e.adaptFwdPeak { + e.adaptFwdPeak = e.adaptFwdThroughputEWMA + } + if e.adaptFwdPeak > 0 { + fwdSaturated = e.adaptFwdThroughputEWMA >= 0.80*e.adaptFwdPeak + fwdIdle = e.adaptFwdThroughputEWMA < 0.25*e.adaptFwdPeak + } + } + if fwdIdle { + e.adaptFwdIdleCount++ + } else { + e.adaptFwdIdleCount = 0 + } + // Anti-churn cooldown: after any reshape, hold the active set steady for a // few ticks so a transient signal can't reshape the mux every tick (each // reshape disrupts in-flight flows). @@ -666,6 +728,12 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { } else { e.adaptSatTicks = 0 } + // Forward sustained-saturation streak (upload analog). + if fwdSaturated { + e.adaptFwdSatTicks++ + } else { + e.adaptFwdSatTicks = 0 + } // Active-set latency median (EWMA) for gross-outlier classification, plus the // per-active-leg health streaks. A leg (never leg 0) that reads a @@ -741,6 +809,20 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { } } + // desiredActive is the combined active-leg width the group converges to: + // the reverse steady/grown target PLUS any extra forward legs the upload + // machine has grown (forwardExtra). Folding the forward growth into the + // convergence target is what stops the reverse-side park/drop-recovery + // rules from immediately tearing down a leg the forward machine just added. + // When the forward machine is dormant (SentBytes flat) forwardExtra is 0 + // and desiredActive == adaptTarget, so every rule below behaves exactly as + // the reverse-only controller did. + forwardExtra := e.adaptFwdTarget - adaptFwdActive + desiredActive := e.adaptTarget + forwardExtra + if desiredActive > adaptCap { + desiredActive = adaptCap + } + // (0) No active legs — bring one up, preferring a healthy warm spare. if aliveCount == 0 { if healthyPromotable >= 0 { @@ -756,7 +838,7 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { // (active fell below the steady target): promote a HEALTHY warm spare // INSTANTLY and re-establish a replacement to restore the floor — zero // re-dial dip. Restoring lost capacity outranks every optimization. - if aliveCount < e.adaptTarget { + if aliveCount < desiredActive { if healthyPromotable >= 0 { return RotationAction{PromoteFromStandby: []int{healthyPromotable}, AddLeg: true} } @@ -790,7 +872,7 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { // load). This is what makes an idle / interactive flow settle onto ONE leg. // No cooldown: park is self-limiting (only fires while over-provisioned) and // stops once active == target, so it converges quickly at dial without churn. - if !saturated && aliveCount > e.adaptTarget && standbyCount < adaptStandbyMax && newestAliveIdx > 0 { + if !saturated && !fwdSaturated && aliveCount > desiredActive && standbyCount < adaptStandbyMax && newestAliveIdx > 0 { return RotationAction{DemoteToStandby: []int{newestAliveIdx}} } @@ -799,7 +881,10 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { // wide mux is reserved for real bulk load and never drains the reserve below // adaptStandbyMin. if e.adaptSatTicks >= adaptHysteresis && aliveCount < adaptCap { - e.adaptTarget = aliveCount + 1 + // Grow the REVERSE target by one. Subtract forwardExtra so the reverse + // width tracks only the download legs even when forward growth has + // enlarged aliveCount (a no-op when the forward machine is dormant). + e.adaptTarget = aliveCount - forwardExtra + 1 if e.adaptTarget > adaptCap { e.adaptTarget = adaptCap } @@ -811,12 +896,31 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { return RotationAction{AddLeg: true} } + // (4b) Grow the FORWARD width on SUSTAINED upload saturation — the exact + // mirror of (4), driven by SentBytes. Emits AddForwardLeg so the router + // appends the aux leg forward-only (appendRouteAsymmetric addFwd=true, + // addRev=false): the extra upstream send capacity is added WITHOUT enlarging + // the reverse/download set. Bounded by the same adaptCap as the reverse side. + // A fresh leg (not a warm-standby promote) is used because the standby pool + // is full-duplex — promoting one would also grow the reverse set. + if e.adaptFwdSatTicks >= adaptHysteresis && aliveCount < adaptCap && desiredActive < adaptCap { + e.adaptFwdTarget++ + if e.adaptFwdTarget > adaptCap { + e.adaptFwdTarget = adaptCap + } + e.adaptFwdSatTicks = 0 + e.adaptCooldown = adaptReshapeCooldown + return RotationAction{AddForwardLeg: true} + } + // (5) Shrink on SUSTAINED idle back toward the single-leg steady target, so a // finished bulk transfer releases its extra legs (parked to the reserve, or - // dropped once it is full). Never leg 0. - if e.adaptIdleCount >= adaptHysteresis && aliveCount > adaptRevActive && newestAliveIdx > 0 { + // dropped once it is full). Never leg 0. The reverse-portion width + // (aliveCount-forwardExtra) gates this so forward-grown legs aren't shrunk + // here (rule 5b owns them). + if e.adaptIdleCount >= adaptHysteresis && aliveCount-forwardExtra > adaptRevActive && newestAliveIdx > 0 { e.adaptIdleCount = 0 - e.adaptTarget = aliveCount - 1 + e.adaptTarget = aliveCount - forwardExtra - 1 if e.adaptTarget < adaptRevActive { e.adaptTarget = adaptRevActive } @@ -827,6 +931,23 @@ func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { return RotationAction{DropLegs: []int{newestAliveIdx}} } + // (5b) Shrink the FORWARD width on SUSTAINED upload idle back toward the lean + // single forward leg — the mirror of (5), driven by SentBytes idle. Parks the + // newest leg to the reserve (or drops it once the reserve is full). Never + // leg 0. + if e.adaptFwdIdleCount >= adaptHysteresis && e.adaptFwdTarget > adaptFwdActive && aliveCount > adaptFwdActive && newestAliveIdx > 0 { + e.adaptFwdIdleCount = 0 + e.adaptFwdTarget-- + if e.adaptFwdTarget < adaptFwdActive { + e.adaptFwdTarget = adaptFwdActive + } + e.adaptCooldown = adaptReshapeCooldown + if standbyCount < adaptStandbyMax { + return RotationAction{DemoteToStandby: []int{newestAliveIdx}} + } + return RotationAction{DropLegs: []int{newestAliveIdx}} + } + return RotationAction{} } diff --git a/pkg/router/policy/presethook/presethook.go b/pkg/router/policy/presethook/presethook.go index 47fdfd9776..a92c26de1c 100644 --- a/pkg/router/policy/presethook/presethook.go +++ b/pkg/router/policy/presethook/presethook.go @@ -139,6 +139,7 @@ func (h *Hook) OnTick(_ router.DialInfo, legs []router.LegInfo) router.RotationA ExcludeHops: action.ExcludeHops, DemoteToStandby: action.DemoteToStandby, PromoteFromStandby: action.PromoteFromStandby, + AddForwardLeg: action.AddForwardLeg, } } diff --git a/pkg/router/policy/types.go b/pkg/router/policy/types.go index f96f2725c3..091b696d1e 100644 --- a/pkg/router/policy/types.go +++ b/pkg/router/policy/types.go @@ -215,4 +215,9 @@ type RotationAction struct { // router.RotationAction and docs/warm_standby_legs_rfc.md. DemoteToStandby []int PromoteFromStandby []int + // AddForwardLeg requests one more FORWARD-ONLY aux leg (router appends it + // addFwd=true / addRev=false): extra upstream send capacity that does not + // enlarge the reverse/download set. The forward-direction mirror of AddLeg, + // emitted by the adaptive preset on sustained upload saturation. + AddForwardLeg bool } diff --git a/pkg/router/policy/wasm/abi.go b/pkg/router/policy/wasm/abi.go index 0733a446de..1002f13e8e 100644 --- a/pkg/router/policy/wasm/abi.go +++ b/pkg/router/policy/wasm/abi.go @@ -125,4 +125,5 @@ type RotationActionWire struct { ExcludeHops []string `json:"exclude_hops,omitempty"` DemoteToStandby []int `json:"demote_to_standby,omitempty"` PromoteFromStandby []int `json:"promote_from_standby,omitempty"` + AddForwardLeg bool `json:"add_forward_leg,omitempty"` } diff --git a/pkg/router/policy/wasm/evaluator.go b/pkg/router/policy/wasm/evaluator.go index 34015485cf..1199faabde 100644 --- a/pkg/router/policy/wasm/evaluator.go +++ b/pkg/router/policy/wasm/evaluator.go @@ -315,6 +315,7 @@ func rotationFromWire(w RotationActionWire) policy.RotationAction { ExcludeHops: append([]string(nil), w.ExcludeHops...), DemoteToStandby: append([]int(nil), w.DemoteToStandby...), PromoteFromStandby: append([]int(nil), w.PromoteFromStandby...), + AddForwardLeg: w.AddForwardLeg, } } diff --git a/pkg/router/policy/wasm/presets/bundle.wasm b/pkg/router/policy/wasm/presets/bundle.wasm index 6f494149fd..49934227c7 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/parity_test.go b/pkg/router/policy/wasm/presets/parity_test.go index 4271848567..6042ceabb6 100644 --- a/pkg/router/policy/wasm/presets/parity_test.go +++ b/pkg/router/policy/wasm/presets/parity_test.go @@ -84,14 +84,15 @@ type normAct struct { ExcludeHops []string DemoteToStandby []int PromoteFromStandby []int + AddForwardLeg bool } func actFromPolicy(a policy.RotationAction) normAct { - return normAct{DropLegs: nzi(a.DropLegs), AddLeg: a.AddLeg, ExcludeHops: nz(a.ExcludeHops), DemoteToStandby: nzi(a.DemoteToStandby), PromoteFromStandby: nzi(a.PromoteFromStandby)} + return normAct{DropLegs: nzi(a.DropLegs), AddLeg: a.AddLeg, ExcludeHops: nz(a.ExcludeHops), DemoteToStandby: nzi(a.DemoteToStandby), PromoteFromStandby: nzi(a.PromoteFromStandby), AddForwardLeg: a.AddForwardLeg} } func actFromPreset(a preset.RotationAction) normAct { - return normAct{DropLegs: nzi(a.DropLegs), AddLeg: a.AddLeg, ExcludeHops: nz(a.ExcludeHops), DemoteToStandby: nzi(a.DemoteToStandby), PromoteFromStandby: nzi(a.PromoteFromStandby)} + return normAct{DropLegs: nzi(a.DropLegs), AddLeg: a.AddLeg, ExcludeHops: nz(a.ExcludeHops), DemoteToStandby: nzi(a.DemoteToStandby), PromoteFromStandby: nzi(a.PromoteFromStandby), AddForwardLeg: a.AddForwardLeg} } func nz(s []string) []string { @@ -291,6 +292,24 @@ func TestTickParity_NativeMatchesWazero(t *testing.T) { {rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(3, "d", 50, 0), sb(2, "c", 50)}, } + // adaptive UPLOAD-heavy: sustained SentBytes growth on the primary leg with + // flat RecvBytes drives the forward (upload) controller — exercises the new + // SentBytes path and its AddForwardLeg emission across the wire so native and + // wazero must agree step-for-step on the forward-direction sizing too. + legS := func(idx int, tid string, sent uint64) policy.LegInfo { + return policy.LegInfo{Index: idx, TransportID: tid, Kind: "stcpr", LatencyMs: 40, Alive: true, SentBytes: sent} + } + up := [][]policy.LegInfo{ + {legS(0, "a", 1_000_000)}, + {legS(0, "a", 2_000_000)}, + {legS(0, "a", 3_000_000)}, + {legS(0, "a", 4_000_000)}, + {legS(0, "a", 5_000_000)}, + {legS(0, "a", 6_000_000), legS(1, "b", 0)}, + {legS(0, "a", 7_000_000), legS(1, "b", 1_000_000)}, + {legS(0, "a", 8_000_000), legS(1, "b", 2_000_000)}, + } + cases := []tickCase{ {"rotating-bw", "rotating-bw", rbw}, {"latency-adaptive", "latency-adaptive", la}, @@ -299,6 +318,7 @@ func TestTickParity_NativeMatchesWazero(t *testing.T) { {"coupled", "coupled", cpl}, {"adaptive", "adaptive", ad}, {"ledbat", "ledbat", lb}, + {"adaptive-upload", "adaptive", up}, } for _, tc := range cases { diff --git a/pkg/router/route_group.go b/pkg/router/route_group.go index 2cc99d8c56..50fed229c0 100644 --- a/pkg/router/route_group.go +++ b/pkg/router/route_group.go @@ -212,7 +212,12 @@ type RouteGroup struct { // See pkg/router/dial_hook.go RotationHook. rotationHook RotationHook rotationApplyAdd func(excludeHops []string) - rotationInterval time.Duration + // rotationApplyAddForward dials one FORWARD-ONLY aux leg + // (appendRouteAsymmetric addFwd=true/addRev=false) for a + // RotationAction.AddForwardLeg — extra upstream send capacity that leaves + // the reverse/download set untouched. Nil disables forward widening. + rotationApplyAddForward func(excludeHops []string) + rotationInterval time.Duration // selfHealAdd restores the multiplexed degree in the background when a // leg dies. pruneDeadTransports drops the dead leg (surviving legs @@ -748,11 +753,14 @@ func (rg *RouteGroup) SetLegChangeHook(hook LegChangeHook, info DialInfo) { // forward leg with the policy's ExcludeHops as the disjoint- // intermediate filter. It runs in the rotation goroutine's // own context so a slow setup-node dial doesn't block other -// route groups' rotation. -func (rg *RouteGroup) SetRotation(hook RotationHook, applyAdd func(excludeHops []string), interval time.Duration) { +// route groups' rotation. applyAddForward is the FORWARD-ONLY +// analog (appendRouteAsymmetric addFwd=true/addRev=false), dialed +// for a RotationAction.AddForwardLeg; nil disables forward widening. +func (rg *RouteGroup) SetRotation(hook RotationHook, applyAdd, applyAddForward func(excludeHops []string), interval time.Duration) { rg.mu.Lock() rg.rotationHook = hook rg.rotationApplyAdd = applyAdd + rg.rotationApplyAddForward = applyAddForward rg.rotationInterval = interval rg.mu.Unlock() // Start the rotation goroutine now (startOffServiceLoops fired @@ -1239,6 +1247,7 @@ func (rg *RouteGroup) rotationServiceFn(_ time.Duration) { rg.mu.Lock() hook := rg.rotationHook applyAdd := rg.rotationApplyAdd + applyAddForward := rg.rotationApplyAddForward info := rg.legChangeInfo legs := rg.snapshotLegs() rg.mu.Unlock() @@ -1246,7 +1255,7 @@ func (rg *RouteGroup) rotationServiceFn(_ time.Duration) { return } action := hook.OnTick(info, legs) - if len(action.DropLegs) == 0 && !action.AddLeg && + if len(action.DropLegs) == 0 && !action.AddLeg && !action.AddForwardLeg && len(action.DemoteToStandby) == 0 && len(action.PromoteFromStandby) == 0 { return } @@ -1297,6 +1306,17 @@ func (rg *RouteGroup) rotationServiceFn(_ time.Duration) { if action.AddLeg && applyAdd != nil { applyAdd(action.ExcludeHops) } + + // Forward-only widen: extra upstream send leg that leaves the reverse set + // untouched. Falls back to the full-duplex add when no forward-only callback + // is wired (older dial path) so the widen still happens. + if action.AddForwardLeg { + if applyAddForward != nil { + applyAddForward(action.ExcludeHops) + } else if applyAdd != nil { + applyAdd(action.ExcludeHops) + } + } } // dropLegsByIndex closes the transports at the policy-supplied diff --git a/pkg/router/router_dial.go b/pkg/router/router_dial.go index 445be7ecb3..17c2f73421 100644 --- a/pkg/router/router_dial.go +++ b/pkg/router/router_dial.go @@ -703,7 +703,26 @@ func (r *router) finishDial( // Periodic rotation (policy on_tick) reuses the same callback. if rh, ok := r.conf.DialHook.(RotationHook); ok && rh != nil && opts.RotationIntervalSeconds > 0 { interval := time.Duration(opts.RotationIntervalSeconds) * time.Second - nrg.rg.SetRotation(rh, applyAdd, interval) + // Forward-only add-leg callback: the adaptive preset's AddForwardLeg + // (upload-saturation widen) dials an aux leg addFwd=true/addRev=false + // so the extra upstream send capacity does not enlarge the + // reverse/download set. Distinct from applyAdd (full-duplex) so the + // two directions size independently. + applyAddForward := func(excludeHops []string) { + addCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + excludePKs := make([]cipher.PubKey, 0, len(excludeHops)) + for _, h := range excludeHops { + var pk cipher.PubKey + if err := pk.Set(h); err == nil { + excludePKs = append(excludePKs, pk) + } + } + if err := r.addOneAuxSendLeg(addCtx, nrgCapture, &optsCopy, fwdDescCopy, excludePKs); err != nil { + r.logger.WithError(err).Debug("Mux forward add-leg failed; route group keeps current leg set") + } + } + nrg.rg.SetRotation(rh, applyAdd, applyAddForward, interval) } } @@ -2477,6 +2496,29 @@ func (r *router) establishMuxRoutes( // leg set); the rotation goroutine logs and waits for the next // tick. func (r *router) addOneAuxForwardLeg(ctx context.Context, nrg *NoiseRouteGroup, opts *DialOptions, forwardDesc routing.RouteDescriptor, extraExcludePKs []cipher.PubKey) error { + return r.addOneAuxLeg(ctx, nrg, opts, forwardDesc, extraExcludePKs, true) +} + +// addOneAuxSendLeg dials one additional FORWARD-ONLY aux leg +// (appendRouteAsymmetric addFwd=true / addRev=false) and appends it to nrg. +// Used by the rotation hook for a RotationAction.AddForwardLeg — the adaptive +// preset's upload-saturation widen. It adds upstream send capacity WITHOUT a +// paired reverse rule, so it does not enlarge the reverse/download set. +// +// Caveat (see the full-duplex note in addOneAuxLeg): a leg with no local +// reverse (consume) rule black-holes any DOWNLOAD the far end spreads onto it. +// That is acceptable here precisely because this leg is grown for an +// upload-dominant flow (little reverse traffic) and the leg-dataprogress / +// leg-liveness prunes evict it if the far end mis-spreads bulk download onto +// it. It is the "forward actuation can't fully mirror reverse" corner: the warm +// standby pool is full-duplex, so a forward-only widen must be a fresh leg. +func (r *router) addOneAuxSendLeg(ctx context.Context, nrg *NoiseRouteGroup, opts *DialOptions, forwardDesc routing.RouteDescriptor, extraExcludePKs []cipher.PubKey) error { + return r.addOneAuxLeg(ctx, nrg, opts, forwardDesc, extraExcludePKs, false) +} + +// addOneAuxLeg is the shared implementation behind addOneAuxForwardLeg +// (addRev=true, full-duplex) and addOneAuxSendLeg (addRev=false, forward-only). +func (r *router) addOneAuxLeg(ctx context.Context, nrg *NoiseRouteGroup, opts *DialOptions, forwardDesc routing.RouteDescriptor, extraExcludePKs []cipher.PubKey, addRev bool) error { if nrg == nil || nrg.rg == nil { return fmt.Errorf("route group nil") } @@ -2582,20 +2624,22 @@ func (r *router) addOneAuxForwardLeg(ctx context.Context, nrg *NoiseRouteGroup, if err != nil { return fmt.Errorf("rotation add-leg: setup-node dial: %w", err) } - // Append the replacement leg FULL-DUPLEX (forward + reverse). Forward-only - // deletes the initiator's consume (reverse) rule for this leg — but the - // setup-node dial already installed that rule on the far end, which marks the - // leg ready on our forward handshake and then spreads its bulk (download) - // stream onto it. With the initiator's consume rule gone those packets are - // dropped (errRouteDescNotExist), so the leg black-holes (recv=0) and, because - // the reorder buffer is lossless, the missing sequences head-of-line-stall the - // primary leg too — a net 0-byte transfer. aliveLegCount counts forward legs, - // so a send-only leg still satisfies the degree target while being a download - // blackhole. Keeping the reverse rule lets the aggregated download land here. - if err := r.appendRouteAsymmetric(nrg, muxRules, true, true); err != nil { + // Append the leg. Full-duplex (addRev=true, the default rotation/self-heal + // path): forward-only would delete the initiator's consume (reverse) rule for + // this leg — but the setup-node dial already installed that rule on the far + // end, which marks the leg ready on our forward handshake and then spreads its + // bulk (download) stream onto it. With the initiator's consume rule gone those + // packets are dropped (errRouteDescNotExist), so the leg black-holes (recv=0) + // and, because the reorder buffer is lossless, the missing sequences + // head-of-line-stall the primary leg too — a net 0-byte transfer. So the + // full-duplex path keeps the reverse rule. addRev=false is used ONLY by the + // adaptive preset's forward-only (AddForwardLeg) upload widen, where the flow + // is upload-dominant so little download lands here, and the data-progress / + // liveness prunes evict the leg if the far end mis-spreads download onto it. + if err := r.appendRouteAsymmetric(nrg, muxRules, true, addRev); err != nil { return fmt.Errorf("rotation add-leg: append: %w", err) } - log.Infof("Rotation aux leg established via tp %s", muxRules.Forward.NextTransportID()) + log.Infof("Rotation aux leg established (addRev=%v) via tp %s", addRev, muxRules.Forward.NextTransportID()) return nil }