Skip to content

feat(router): live mux control (mode/cap/width) + goodput weighted-ramp #4234

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
0pcom merged 3 commits into from
Aug 26, 2026
Merged

feat(router): live mux control (mode/cap/width) + goodput weighted-ramp #4234

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7176b96
feat(router): live mux control (mode/cap/width) + goodput weighted-ra…
0pcom Aug 26, 2026
62ad08e
fix(router): update preset tests for runtime-tunable mux caps + capac…
0pcom Aug 26, 2026
ebaec1a
fix(lint): SetMinHops LogCall deferred-return not called (SA9010)
0pcom Aug 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 83 additions & 13 deletions cmd/skywire-cli/commands/proxy/mux_ops.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,8 @@ import (
"io"
"os"

"strconv"

"github.com/google/uuid"
"github.com/spf13/cobra"

Expand All @@ -62,6 +64,66 @@ func init() {
addMuxSub(muxAddCmd, "mux-add")
addMuxSub(muxRmCmd, "mux-rm")
addMuxSub(muxModeCmd, "mux-mode")
addMuxSub(muxCapCmd, "mux-cap")
addMuxSub(muxWidthCmd, "mux-width")
}

var muxCapCmd = &cobra.Command{
Use: "cap <n>",
Short: "Set the adaptive mux active-width ceiling at runtime",
Long: `Set the MAXIMUM number of ACTIVE mux legs the adaptive engine may grow to
under sustained load — the aggregation ceiling. Applies LIVE to this visor's
adaptive route groups on their next tick (no restart). Send-side is a per-visor
decision, so set it independently on each end (e.g. over the pty to the exit).

Example:
skywire cli proxy mux cap 60 # allow aggregation up to 60 active legs`,
Args: cobra.ExactArgs(1),
DisableFlagsInUseLine: true,
Run: func(cmd *cobra.Command, args []string) {
n, err := strconv.Atoi(args[0])
if err != nil || n < 1 {
internal.PrintFatalError(cmd.Flags(), fmt.Errorf("cap must be a positive integer, got %q", args[0]))
}
rpcClient, err := clirpc.Client(cmd.Flags())
if err != nil {
internal.PrintFatalError(cmd.Flags(), fmt.Errorf("unable to create RPC client: %w", err))
}
defer rpcClient.Close() //nolint:errcheck,gosec
if err := rpcClient.SetMuxCap(n); err != nil {
internal.PrintFatalError(cmd.Flags(), fmt.Errorf("SetMuxCap: %w", err))
}
internal.Catch(cmd.Flags(), cliout.Print(cmd, cliproxy.MuxOp{Op: "cap", App: muxOpsApp, Mode: args[0]}))
},
}

var muxWidthCmd = &cobra.Command{
Use: "width <n>",
Short: "Set the adaptive mux steady active download width at runtime",
Long: `Set the STEADY active download width — the floor number of active mux legs
the adaptive engine converges to when idle (more than one spreads a bulk flow
proactively before saturation instead of ramping from a single leg). Applies
LIVE on the next tick; clamped to [1, cap]. Set per-visor, per-end.

Example:
skywire cli proxy mux width 8 # keep 8 legs active by default`,
Args: cobra.ExactArgs(1),
DisableFlagsInUseLine: true,
Run: func(cmd *cobra.Command, args []string) {
n, err := strconv.Atoi(args[0])
if err != nil || n < 1 {
internal.PrintFatalError(cmd.Flags(), fmt.Errorf("width must be a positive integer, got %q", args[0]))
}
rpcClient, err := clirpc.Client(cmd.Flags())
if err != nil {
internal.PrintFatalError(cmd.Flags(), fmt.Errorf("unable to create RPC client: %w", err))
}
defer rpcClient.Close() //nolint:errcheck,gosec
if err := rpcClient.SetMuxWidth(n); err != nil {
internal.PrintFatalError(cmd.Flags(), fmt.Errorf("SetMuxWidth: %w", err))
}
internal.Catch(cmd.Flags(), cliout.Print(cmd, cliproxy.MuxOp{Op: "width", App: muxOpsApp, Mode: args[0]}))
},
}

// routePair mirrors the shape 'cli route calc --json' emits.
Expand Down Expand Up @@ -195,30 +257,38 @@ Example:
}

var muxModeCmd = &cobra.Command{
Use: "mode <auto|equal>",
Use: "mode <auto|equal|capacity>",
Short: "Change mux scheduler weighting at runtime",
Long: `Set the mux transport-selection mode for the visor.

auto - latency-weighted: lower-latency legs get more packets.
Best when the legs have different RTTs (the typical case)
because it minimizes head-of-line stalls in SACK reorder.
equal - round-robin: each leg gets equal share. Useful when legs
have similar latency and you want to verify aggregation
behavior without the auto-mode masking it.
auto - latency-weighted: lower-latency legs get more packets.
Best when the legs have different RTTs (the typical case)
because it minimizes head-of-line stalls in SACK reorder.
equal - round-robin: each leg gets equal share. Useful when legs
have similar latency and you want to verify aggregation
behavior without the auto-mode masking it.
capacity - goodput-weighted: each leg's share tracks its recently-
measured throughput (bytes/sec), so a fast leg carries more
and a slow one carries little — the thin-spread aggregation
mode. A just-promoted leg starts at a small cold-leg floor
share and ramps as its goodput proves out.

Affects every active and future mux'd route group on this visor.
The setting persists to skywire-config.json so it survives restart.
Affects every active and future mux'd route group on this visor
IMMEDIATELY (the router re-applies the mode to live route groups). The
setting persists to skywire-config.json so it survives restart.

Example:
skywire cli proxy mux mode equal # before measuring aggregation
skywire cli proxy mux mode capacity # goodput-weighted thin spread
skywire cli proxy mux info --watch 1s
skywire cli proxy mux mode auto # back to weighted`,
skywire cli proxy mux mode auto # back to latency-weighted`,
Args: cobra.ExactArgs(1),
DisableFlagsInUseLine: true,
Run: func(cmd *cobra.Command, args []string) {
mode := args[0]
if mode != "auto" && mode != "equal" {
internal.PrintFatalError(cmd.Flags(), fmt.Errorf("mode must be 'auto' or 'equal', got %q", mode))
switch mode {
case "auto", "equal", "capacity":
default:
internal.PrintFatalError(cmd.Flags(), fmt.Errorf("mode must be 'auto', 'equal', or 'capacity', got %q", mode))
}
rpcClient, err := clirpc.Client(cmd.Flags())
if err != nil {
Expand Down
9 changes: 7 additions & 2 deletions pkg/router/policy/preset/preset.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -335,9 +335,14 @@ func decideAdaptive(ctx Context, cands []Candidate) Spec {
// Symmetric Mux => establishMuxRoutes builds every leg FULL-DUPLEX
// (fwdCount == revCount). See the Shape note above for why bidirectional
// setup replaced the old ForwardMux=1 / ReverseMux=N asymmetry.
Mux: adaptRevActive + adaptStandbyMax,
Mux: AdaptRevActive() + adaptStandbyMax,
RotationIntervalSeconds: 20,
Distribution: "auto",
// Goodput-weighted thin spread: each active leg's share tracks its
// recently-measured throughput, with a cold-leg floor so a fresh leg
// ramps in (see rebuildWeights). Live-tunable to auto/equal via the
// mux-control RPC. This is the aggregation default — a slow leg carries
// little and can't stall the reorder frontier.
Distribution: "capacity",
}
if anyKnownTransportKind(cands) {
spec.Chosen = mostTransportDiverse(cands)
Expand Down
105 changes: 72 additions & 33 deletions pkg/router/policy/preset/preset_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,9 @@ func TestDecide_ShapePresets(t *testing.T) {
{"latency-adaptive", Context{App: "vpn-client"}, Spec{Mux: 5, MinHops: 2, RotationIntervalSeconds: 30, Distribution: "auto"}},
{"elastic-mux", Context{App: "skynet-client"}, Spec{Mux: 2, MinHops: 2, RotationIntervalSeconds: 20, Distribution: "auto"}},
{"probe-and-prune", Context{App: "skynet-client"}, Spec{Mux: 3, MinHops: 2, RotationIntervalSeconds: 30, Distribution: "auto"}},
{"adaptive", Context{App: "vpn-client"}, Spec{Mux: adaptRevActive + adaptStandbyMax, RotationIntervalSeconds: 20, Distribution: "auto"}},
{"adaptive", Context{App: "vpn-client"}, Spec{Mux: AdaptRevActive() + adaptStandbyMax, RotationIntervalSeconds: 20, Distribution: "capacity"}},
{"adaptive/chat", Context{App: "skychat"}, Spec{Mux: 1}},
{"adaptive/custom-session", Context{App: "g8"}, Spec{Mux: adaptRevActive + adaptStandbyMax, RotationIntervalSeconds: 20, Distribution: "auto"}},
{"adaptive/custom-session", Context{App: "g8"}, Spec{Mux: AdaptRevActive() + adaptStandbyMax, RotationIntervalSeconds: 20, Distribution: "capacity"}},
{"ledbat", Context{App: "skysocks-client"}, Spec{Mux: 3, MinHops: 2, RotationIntervalSeconds: 20, Distribution: "auto"}},
{"ledbat/chat", Context{App: "skychat"}, Spec{Mux: 1}},
}
Expand Down Expand Up @@ -80,8 +80,8 @@ func TestDecide_Adaptive(t *testing.T) {
if got.Chosen == nil || got.Chosen.Hops[0] != "b" {
t.Fatalf("adaptive should seed the most transport-diverse route (b); got %+v", got.Chosen)
}
if got.Mux != adaptRevActive+adaptStandbyMax || got.ForwardMux != 0 || got.ReverseMux != 0 ||
got.RotationIntervalSeconds != 20 || got.Distribution != "auto" ||
if got.Mux != AdaptRevActive()+adaptStandbyMax || got.ForwardMux != 0 || got.ReverseMux != 0 ||
got.RotationIntervalSeconds != 20 || got.Distribution != "capacity" ||
got.MinHops != 0 {
t.Errorf("adaptive seed shape changed: %+v", got)
}
Expand Down Expand Up @@ -119,7 +119,7 @@ func TestDecide_Adaptive(t *testing.T) {
// App-agnostic: an unknown / custom-named session still gets the adaptive
// bidirectional mux (not the empty spec) — the fix must apply to EVERY app that
// dials a route group, not a hardcoded allowlist.
if got := Decide("adaptive", Context{App: "some-custom-app"}, nil); got.Mux != adaptRevActive+adaptStandbyMax || got.ForwardMux != 0 || got.ReverseMux != 0 {
if got := Decide("adaptive", Context{App: "some-custom-app"}, nil); got.Mux != AdaptRevActive()+adaptStandbyMax || got.ForwardMux != 0 || got.ReverseMux != 0 {
t.Errorf("adaptive must apply to any non-chat app; got %+v", got)
}
}
Expand All @@ -128,24 +128,24 @@ func TestDecide_Adaptive(t *testing.T) {
// PROACTIVELY parks the surplus reverse legs as warm standby (down to the
// steady active target) instead of waiting for an idle signal, and never parks
// leg 0 (the primary / forward leg the router refuses to demote). The decide
// seeds adaptRevActive+adaptStandbyMax reverse legs; the router brings them up
// seeds AdaptRevActive()+adaptStandbyMax reverse legs; the router brings them up
// active; the first ticks must demote the newest surplus legs to standby.
func TestEngine_OnTick_AdaptiveHoldsWarmStandby(t *testing.T) {
e := New()
// adaptRevActive+adaptStandbyMax legs all active (as the router first
// AdaptRevActive()+adaptStandbyMax legs all active (as the router first
// establishes them — every leg is born active). Steady active target =
// adaptRevActive.
total := adaptRevActive + adaptStandbyMax
// AdaptRevActive().
total := AdaptRevActive() + adaptStandbyMax
legs := make([]LegInfo, total)
for i := range legs {
legs[i] = LegInfo{Index: i, TransportID: string(rune('a' + i)), Kind: "stcpr", LatencyMs: 40, Alive: true}
}

// ONE tick parks the WHOLE surplus (total-adaptRevActive legs) at once — the
// ONE tick parks the WHOLE surplus (total-AdaptRevActive() legs) at once — the
// bulk fast-converge that stops a wide uncapped mux from lingering as a
// head-of-line-stalling active set. Leg 0 (primary) is never parked.
act := e.OnTick("adaptive", legs)
wantParked := total - adaptRevActive
wantParked := total - AdaptRevActive()
if len(act.DemoteToStandby) != wantParked {
t.Fatalf("expected a bulk park of %d surplus legs in one tick, got %d: %+v",
wantParked, len(act.DemoteToStandby), act)
Expand All @@ -165,7 +165,7 @@ func TestEngine_OnTick_AdaptiveHoldsWarmStandby(t *testing.T) {
t.Fatalf("expected %d distinct parked legs, got %d", wantParked, len(parked))
}

// Steady state: adaptRevActive active + adaptStandbyMax standby → no further
// Steady state: AdaptRevActive() active + adaptStandbyMax standby → no further
// structural change (no dip, no churn).
if act := e.OnTick("adaptive", legs); !reflect.DeepEqual(act, RotationAction{}) {
t.Errorf("at steady active target the adaptive tick must be a no-op; got %+v", act)
Expand Down Expand Up @@ -322,12 +322,18 @@ func TestEngine_OnTick_AdaptiveSwapsBadPrimary(t *testing.T) {
}

// TestEngine_OnTick_AdaptiveCapsActiveUnderLoad asserts the active set is capped
// at adaptCap even when the group reads SATURATED. During the uncap fill, legs
// at AdaptCap() even when the group reads SATURATED. During the uncap fill, legs
// are born active and route-setup traffic keeps the group saturated — the park
// must still cap active at adaptCap (parking the slowest excess) so the reorder
// buffer never spans more than adaptCap legs; only born-active excess is shed,
// never below adaptCap under load.
// must still cap active at AdaptCap() (parking the slowest excess) so the reorder
// buffer never spans more than AdaptCap() legs; only born-active excess is shed,
// never below AdaptCap() under load.
func TestEngine_OnTick_AdaptiveCapsActiveUnderLoad(t *testing.T) {
// Pin the runtime-tunable cap to 8 so this test exercises the capping logic
// deterministically regardless of the shipped default (which is now 60).
restoreCap := AdaptCap()
SetAdaptCap(8)
defer SetAdaptCap(restoreCap)

e := seedSaturatedAdaptive(1)
const n = 20
legs := make([]LegInfo, n)
Expand All @@ -342,10 +348,10 @@ func TestEngine_OnTick_AdaptiveCapsActiveUnderLoad(t *testing.T) {
legs[i] = LegInfo{Index: i, TransportID: tid, Kind: "stcpr", LatencyMs: 20 + i*10, Alive: true, RecvBytes: recv}
}
act := e.OnTick("adaptive", legs)
wantParked := n - adaptCap
wantParked := n - AdaptCap()
if len(act.DemoteToStandby) != wantParked {
t.Fatalf("under load, active must be capped at adaptCap=%d: want %d parked, got %d: %+v",
adaptCap, wantParked, len(act.DemoteToStandby), act)
t.Fatalf("under load, active must be capped at AdaptCap()=%d: want %d parked, got %d: %+v",
AdaptCap(), wantParked, len(act.DemoteToStandby), act)
}
parked := map[int]bool{}
for _, idx := range act.DemoteToStandby {
Expand All @@ -354,14 +360,14 @@ func TestEngine_OnTick_AdaptiveCapsActiveUnderLoad(t *testing.T) {
}
parked[idx] = true
}
// The adaptCap kept-active legs are the fastest (leg 0 + legs 1..adaptCap-1);
// the n-adaptCap slowest (highest index) are parked.
for i := adaptCap; i < n; i++ {
// The AdaptCap() kept-active legs are the fastest (leg 0 + legs 1..AdaptCap()-1);
// the n-AdaptCap() slowest (highest index) are parked.
for i := AdaptCap(); i < n; i++ {
if !parked[i] {
t.Errorf("slowest excess legs must be parked to cap active at adaptCap: leg %d (%dms) not parked", i, 20+i*10)
t.Errorf("slowest excess legs must be parked to cap active at AdaptCap(): leg %d (%dms) not parked", i, 20+i*10)
}
}
for i := 1; i < adaptCap; i++ {
for i := 1; i < AdaptCap(); i++ {
if parked[i] {
t.Errorf("fastest legs must stay active under load: leg %d (%dms) was parked", i, 20+i*10)
}
Expand Down Expand Up @@ -461,7 +467,7 @@ func TestEngine_OnTick_AdaptiveStandbyFloor(t *testing.T) {
// fall below adaptStandbyMin across many heavy-load ticks, and the active
// width must grow.
eng := New()
active, standby := adaptRevActive, adaptStandbyMax
active, standby := AdaptRevActive(), adaptStandbyMax
var recv uint64
build := func() []LegInfo {
legs := make([]LegInfo, 0, active+standby)
Expand Down Expand Up @@ -500,8 +506,8 @@ func TestEngine_OnTick_AdaptiveStandbyFloor(t *testing.T) {
t.Fatalf("tick %d: standby reserve %d fell below floor %d (action %+v)", tick, standby, adaptStandbyMin, act)
}
}
if active <= adaptRevActive {
t.Errorf("sustained saturation should grow the active width beyond the %d-leg seed; active=%d", adaptRevActive, active)
if active <= AdaptRevActive() {
t.Errorf("sustained saturation should grow the active width beyond the %d-leg seed; active=%d", AdaptRevActive(), active)
}
if standby < adaptStandbyMin {
t.Errorf("standby reserve ended below floor: %d < %d", standby, adaptStandbyMin)
Expand Down Expand Up @@ -551,8 +557,19 @@ func TestEngine_OnTick_AdaptiveEvictsGrossOutlier(t *testing.T) {
// width and then stops reshaping (a long tail of no-ops). This is the fix for
// the observed "active set churns constantly, disrupting in-flight flows."
func TestEngine_OnTick_AdaptiveStableUnderSteady(t *testing.T) {
// Pin the runtime-tunable widths to the values this test's convergence/churn
// assertions were written against (cap 8, floor 1), independent of the shipped
// defaults (now 60 / 4). Set the cap first so the low floor is not clamped, and
// restore in reverse.
restoreCap := AdaptCap()
SetAdaptCap(8)
defer SetAdaptCap(restoreCap)
restoreW := AdaptRevActive()
SetAdaptRevActive(1)
defer SetAdaptRevActive(restoreW)

eng := New()
active, standby := adaptRevActive, adaptStandbyMax
active, standby := AdaptRevActive(), adaptStandbyMax
var recv uint64
build := func() []LegInfo {
legs := make([]LegInfo, 0, active+standby)
Expand Down Expand Up @@ -916,8 +933,19 @@ func (s *adaptiveSim) step() RotationAction { //nolint:unparam // test helper: r
// 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).
// unchanged at AdaptRevActive()).
func TestEngine_OnTick_AdaptiveForwardWidensOnUpload(t *testing.T) {
// Pin the widths to the values this test asserts against (cap 8, floor 1),
// independent of the shipped defaults (now 60 / 4). Pin BEFORE newAdaptiveSim
// so New() seeds adaptTarget from the pinned floor. Set cap first, restore in
// reverse.
restoreCap := AdaptCap()
SetAdaptCap(8)
defer SetAdaptCap(restoreCap)
restoreW := AdaptRevActive()
SetAdaptRevActive(1)
defer SetAdaptRevActive(restoreW)

s := newAdaptiveSim(1_000_000, 0) // heavy upload, zero download
for i := 0; i < 20; i++ {
s.step()
Expand All @@ -928,8 +956,8 @@ func TestEngine_OnTick_AdaptiveForwardWidensOnUpload(t *testing.T) {
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.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)
Expand Down Expand Up @@ -957,15 +985,26 @@ func TestEngine_OnTick_AdaptiveReverseWidensOnDownload(t *testing.T) {
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)
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) {
// Pin the widths to the values this test asserts against (cap 8, floor 1) so
// the idle steady state collapses to a SINGLE active leg, independent of the
// shipped defaults (now 60 / 4). Pin BEFORE newAdaptiveSim. Set cap first,
// restore in reverse.
restoreCap := AdaptCap()
SetAdaptCap(8)
defer SetAdaptCap(restoreCap)
restoreW := AdaptRevActive()
SetAdaptRevActive(1)
defer SetAdaptRevActive(restoreW)

s := newAdaptiveSim(1_000_000, 0)
for i := 0; i < 20; i++ { // widen under upload
s.step()
Expand Down
Loading
Loading