diff --git a/cmd/wasm-visor/main.go b/cmd/wasm-visor/main.go index fc24a100c4..a53475ee5d 100644 --- a/cmd/wasm-visor/main.go +++ b/cmd/wasm-visor/main.go @@ -78,6 +78,8 @@ import ( "github.com/skycoin/skywire/pkg/logging" "github.com/skycoin/skywire/pkg/rfclient" "github.com/skycoin/skywire/pkg/router" + "github.com/skycoin/skywire/pkg/router/policy/preset" + "github.com/skycoin/skywire/pkg/router/policy/presethook" "github.com/skycoin/skywire/pkg/routing" "github.com/skycoin/skywire/pkg/skyenv" "github.com/skycoin/skywire/pkg/transport" @@ -149,6 +151,36 @@ func pageHTTPS() bool { return loc.Get("protocol").String() == "https:" } +// wasmRoutingPolicyPreset reads the routing-policy preset name from the page URL +// (?routing_policy=), validating it against the built-in preset set. It +// returns "" (no policy) when the param is absent, empty, or names an unknown +// preset — so the default is unchanged no-policy behavior and a typo can't +// silently apply the app-mux fallback. This is the wasm-visor's analog of the +// native visor's routing.policy config field. +func wasmRoutingPolicyPreset() string { + loc := js.Global().Get("location") + if !loc.Truthy() { + return "" + } + search := loc.Get("search") + if !search.Truthy() { + return "" + } + usp := js.Global().Get("URLSearchParams") + if !usp.Truthy() { + return "" + } + v := usp.New(search.String()).Call("get", "routing_policy") + if !v.Truthy() { + return "" + } + name := v.String() + if !preset.Has(name) { + return "" + } + return name +} + func main() { ctx = context.Background() // One binary, several roles (see shell_js.go / tpviz_js.go): the tab loads @@ -554,6 +586,19 @@ func bootEdge(skHex, seedPKHex, seedWSURL, discDmsgAddr, cfgOverrideJSON string) // svc.MinHops is 1 (origination enabled; a direct transport still downgrades to // a 0-intermediate-hop path). vlog("router: New + serve…") + // Routing policy: a browser leaf can't host wazero (wasm-in-wasm), so it + // can't run the embedded bundle.wasm the native visor evaluates for + // "preset:". Instead it runs the SAME preset decide/tick logic compiled + // in as native Go (pkg/router/policy/preset, the single source of truth) via + // presethook, wired here as the router's DialHook — the one integration point + // (pkg/visor/init_router.go builds the wazero-backed equivalent on native). + // Opt-in and configurable: the preset name comes from the ?routing_policy= + // query param; unset ⇒ nil DialHook ⇒ unchanged (no-policy) behavior. + var dialHook router.DialHook + if name := wasmRoutingPolicyPreset(); name != "" { + dialHook = presethook.New(name, nil) + vlog(fmt.Sprintf("router: routing policy preset %q active (native-Go preset engine)", name)) + } r, err := visorcore.BuildRouter(ctx, visorcore.RouterDeps{ DmsgC: dmsgC, PubKey: pk, @@ -563,6 +608,7 @@ func bootEdge(skHex, seedPKHex, seedWSURL, discDmsgAddr, cfgOverrideJSON string) RouteGroupDialer: rgDialer, SetupNodes: svc.RouteSetupNodes, MinHops: svc.MinHops, + DialHook: dialHook, // Route-setup hook: on an app dial (min_hops=1, no --existing-tp) the // router races this against the route-finder to create a DIRECT transport // to the peer — the browser edge's missing half (a native visor registers diff --git a/docs/examples/routing-policies/wasm/bundle/go.mod b/docs/examples/routing-policies/wasm/bundle/go.mod index f3bb35fbcb..fab0c989d7 100644 --- a/docs/examples/routing-policies/wasm/bundle/go.mod +++ b/docs/examples/routing-policies/wasm/bundle/go.mod @@ -1,3 +1,12 @@ module skywire-routing-policy-bundle -go 1.21 +go 1.26.4 + +// The preset decide/tick logic is the single source of truth in the main +// skywire module (pkg/router/policy/preset); this bundle is a thin wasm ABI +// shim over it. The replace points at the repo root so `tinygo build` compiles +// the in-tree package. preset imports only the Go stdlib, so no go.sum / external +// downloads are pulled into this module. +require github.com/skycoin/skywire v0.0.0 + +replace github.com/skycoin/skywire => ../../../../.. diff --git a/docs/examples/routing-policies/wasm/bundle/main.go b/docs/examples/routing-policies/wasm/bundle/main.go index e7941668d2..a13863fb25 100644 --- a/docs/examples/routing-policies/wasm/bundle/main.go +++ b/docs/examples/routing-policies/wasm/bundle/main.go @@ -8,28 +8,39 @@ // runtime is paid once; each extra preset adds only its own few KB of // logic. // +// This file is now a THIN SHIM. The preset decide/tick logic itself lives +// ONCE, as pure Go, in github.com/skycoin/skywire/pkg/router/policy/preset +// (the single source of truth). This shim keeps only the wasm ABI: the +// JSON wire types, the alloc/free/read/write linear-memory glue, and the +// //export decide_route / on_tick entry points. Each entry point unmarshals +// the wire input, converts it to the shared preset types, calls into the +// preset package, and marshals the result back out. The NATIVE visor runs +// the compiled bundle.wasm via wazero; the WASM (browser/TinyGo) visor +// compiles the SAME preset package in directly (it cannot host wazero — +// wasm-in-wasm), so both paths make byte-identical decisions. +// // This is the module embedded at pkg/router/policy/wasm/presets/ // bundle.wasm and selected by config as "preset:" (see // pkg/visor/policy_loader.go). The per-preset standalone examples next // to this dir (app-mux/, rotating-bw/) stay as pedagogical single-preset // modules; this one is their union. // -// Build with TinyGo: -// -// cd docs/examples/routing-policies/wasm/bundle -// tinygo build -target=wasi -no-debug -opt=2 -o bundle.wasm . +// Build with TinyGo (from this directory): // -// then copy it over pkg/router/policy/wasm/presets/bundle.wasm. +// tinygo build -target=wasi -no-debug -opt=2 \ +// -o ../../../../pkg/router/policy/wasm/presets/bundle.wasm . package main import ( "encoding/json" - "sort" - "strings" "unsafe" + + "github.com/skycoin/skywire/pkg/router/policy/preset" ) -// Wire types — kept in sync with pkg/router/policy/wasm/abi.go. +// Wire types — kept in sync with pkg/router/policy/wasm/abi.go. These are the +// JSON envelopes the host (wazero) marshals across linear memory; the guest +// converts them to/from the pure preset.* types. type candidateWire struct { Hops []string `json:"hops"` @@ -99,6 +110,13 @@ type rotationActionWire struct { PromoteFromStandby []int `json:"promote_from_standby,omitempty"` } +// engine holds the adaptive tick controllers' per-transport_id state for this +// module instance. One package-global instance mirrors the bundle's original +// package-global tick state (the host instantiates one wazero module per policy +// load, so this is created fresh per load); the native visor constructs its own +// preset.Engine per evaluator the same way. +var engine = preset.New() + // Required: host-driven memory management. //export alloc @@ -133,1595 +151,166 @@ func decideRoute(inPtr, inLen uint32) uint64 { if err := json.Unmarshal(readInput(inPtr, inLen), &input); err != nil { return 0 } - var spec routeSpecWire - switch input.Preset { - case "rotating-bw": - spec = decideRotatingBW(input.Ctx) - case "latency-adaptive": - spec = decideLatencyAdaptive(input.Ctx) - case "elastic-mux": - spec = decideElasticMux(input.Ctx) - case "probe-and-prune": - spec = decideProbeAndPrune(input.Ctx) - case "adaptive": - spec = decideAdaptive(input.Ctx) - case "app-mux": - spec = decideAppMux(input.Ctx) - case "geo-avoid": - spec = decideGeoAvoid(input.Ctx, input.Candidates) - case "transport-diverse": - spec = decideTransportDiverse(input.Ctx, input.Candidates) - case "trust-tiered": - spec = decideTrustTiered(input.Ctx, input.Candidates) - case "time-of-day": - spec = decideTimeOfDay(input.Ctx) - default: - // Empty / unknown preset: fall back to the per-app app-mux - // behavior so a bare bundle still does something sensible. - spec = decideAppMux(input.Ctx) - } - out, err := json.Marshal(spec) + spec := preset.Decide(input.Preset, ctxToPreset(input.Ctx), candsToPreset(input.Candidates)) + out, err := json.Marshal(specToWire(spec)) if err != nil { return 0 } return writeOutput(out) } -// decideAppMux is the verbatim app-mux preset logic: per-app static mux -// + min_hops; latency-sensitive apps stay single-route, bandwidth apps -// get parallel legs. -func decideAppMux(ctx routingContextWire) routeSpecWire { - switch ctx.App { - case "vpn-client": - return routeSpecWire{Mux: 4, MinHops: 2} - case "skychat": - // Chat is latency-sensitive — single route, lowest mux. - return routeSpecWire{Mux: 1} - default: - // Everything else: visor defaults (empty spec). - return routeSpecWire{} - } -} - -// decideRotatingBW is the rotating-bw (privacy) preset: targetMux active legs -// over multi-hop with byte load spread EQUALLY across them (traffic-analysis -// resistance — no single relay sees a large fraction of the flow), plus one -// extra warm-standby leg so on_tick can rotate the active set every 90s -// WITHOUT a tear-and-rebuild dip (see tickRotatingBW). -func decideRotatingBW(ctx routingContextWire) routeSpecWire { - // rotating-bw is OPT-IN per app (proxy/vpn/skynet start --routing-policy, or - // visor app arg routing-policy). Apply it to WHATEVER app/session it is - // active for — do NOT gate on the built-in binary names. A named proxy - // session dials under its SESSION name (e.g. "g8"), not "skysocks-client", - // so the old `switch ctx.App { case "skysocks-client", ... }` silently made - // the policy a NO-OP for every custom-named session (empty spec → no mux, no - // min_hops, no rotation — the reason the rotation never fired live). Only - // skip genuinely latency-sensitive apps, where a multi-hop mux is the wrong - // shape. - switch ctx.App { - case "skychat", "skychat-client": - return routeSpecWire{Mux: 1} - } - // min_hops=2 already says "no direct transport" (direct is 0 intermediates); - // the visor treats min_hops>=2 as an implicit avoid_direct so the dial flows - // to the overlay path where rotation can act. Without it the mux/distribution - // is silently dropped by the direct-dial fast path. - return routeSpecWire{ - // targetMux active + 1 warm standby (see tickRotatingBW). - Mux: targetMux + 1, - MinHops: 2, - RotationIntervalSeconds: 90, - // Equal (round-robin) byte spread across the active legs — the privacy - // property. "auto"/latency-weighting pins bytes to the fastest leg, - // defeating the spread; the policy sets this to override the muxMode - // default. - Distribution: "round-robin", - } -} - -// decideLatencyAdaptive is the latency-adaptive preset's decide logic: an -// ASYMMETRIC spec for bandwidth/proxy apps — a single lean, direct-ok -// upstream leg (uploads are small) paired with a 4-way multi-hop -// downstream fan-out (bulk downloads). RotationIntervalSeconds=30 keeps -// on_tick firing so the downstream set can be re-evaluated and the -// slowest leg evicted; Distribution "auto" lets the host weight bytes -// toward the faster legs. Non-target apps get the empty spec (defaults). -func decideLatencyAdaptive(ctx routingContextWire) routeSpecWire { - switch ctx.App { - case "vpn-client", "skysocks-client", "skynet-client": - // Symmetric mux=4: the on_tick evict-slowest logic acts on the - // route group's forward legs (rg.tps) — the only legs the tick - // hook sees (reverse-only legs live in rg.rvs and are invisible - // to on_tick). An asymmetric 1-up/4-down shape would put the 4 - // adaptive legs on the reverse side where on_tick can't manage - // them, making the eviction a no-op. Symmetric mux gives on_tick - // 4 real legs to converge over. (The lean-upstream refinement - // awaits a router change exposing reverse legs to the tick hook.) - // - // Mux is targetMux+1 (5): one leg beyond the 4-wide active target is a - // WARM RESERVE the on_tick controller parks on standby, so evicting the - // slowest leg is a hot-swap (promote the spare, demote the outlier) with - // no width dip — the no-dip discipline shared with rotating-bw. - return routeSpecWire{ - Mux: 5, - MinHops: 2, - RotationIntervalSeconds: 30, - Distribution: "auto", - } - } - return routeSpecWire{} -} - -// decideElasticMux is the elastic-mux preset's decide logic. It starts -// deliberately MODEST — a 2-way mux over multi-hop — and lets the on_tick -// AIMD controller grow or shrink the leg count to match observed load. -// Unlike the static-mux presets (which pin a fixed width), the initial -// Mux here is only a seed: RotationIntervalSeconds=20 makes on_tick fire -// often enough to react to load swings, and Distribution "auto" lets the -// host weight bytes toward the faster legs while the count floats. -// Non-target apps get the empty spec (defaults). -func decideElasticMux(ctx routingContextWire) routeSpecWire { - switch ctx.App { - case "vpn-client", "skysocks-client", "skynet-client": - return routeSpecWire{ - Mux: 2, - MinHops: 2, - RotationIntervalSeconds: 20, - Distribution: "auto", - } - } - return routeSpecWire{} -} - -// decideProbeAndPrune is the probe-and-prune preset's decide logic. It -// holds a steady 3-way mux over multi-hop as the "established" set that -// on_tick continuously refines: every 30s the tick controller adds one -// speculative leg over a fresh path, watches it for a few ticks, and -// keeps it only if it beats the current worst leg (else discards it), so -// the established set drifts toward lower latency without ever growing -// past its target. Distribution "auto" weights bytes toward the faster -// legs meanwhile. Non-target apps get the empty spec (defaults). -func decideProbeAndPrune(ctx routingContextWire) routeSpecWire { - switch ctx.App { - case "vpn-client", "skysocks-client", "skynet-client": - return routeSpecWire{ - Mux: 3, - MinHops: 2, - RotationIntervalSeconds: 30, - Distribution: "auto", - } - } - return routeSpecWire{} -} - -// decideAdaptive is the COMPOSITE "adaptive" preset's decide logic — the -// intended converged default. It returns a sensible middle spec that the -// tickAdaptive controller then steers along all three performance -// dimensions at once: a 3-way disjoint multi-hop mux (min_hops=2), latency- -// weighted byte distribution ("auto"), re-evaluated every 20s so on_tick -// fires often enough to react to load swings, latency drift, and probe -// results. The Mux of 3 is only the STARTING width — tickAdaptive's AIMD -// dimension grows it toward the cap under load and shrinks it back toward -// the floor when idle. Non-target apps get the empty spec (defaults). -func decideAdaptive(ctx routingContextWire) routeSpecWire { - switch ctx.App { - case "vpn-client", "skysocks-client", "skynet-client": - // NOTE: no MinHops here on purpose. min-hops is a PRIVACY CONSTRAINT the - // operator owns (session --min-hops / visor config), not a knob for the - // performance policy to impose. Leaving it 0 means "inherit" — the - // router's EffectiveMinHops takes the strictest of the operator's floor - // and this spec, so adaptive optimizes WITHIN the operator's chosen - // floor: with the default floor it can use a fast low-hop path; with a - // privacy floor set it fans the mux out over multi-hop. Hardcoding - // MinHops=2 forced every dial onto slow multi-hop legs even when a fast - // direct path existed — measurably worse, not better. - return routeSpecWire{ - Mux: adaptDecideMux, - RotationIntervalSeconds: 20, - Distribution: "auto", - } - } - return routeSpecWire{} -} - -// targetMux is the mux size the rotating-bw policy aims to maintain. -// Kept in sync with the Mux value returned from decideRotatingBW so -// on_tick can reason about "are we at target?" -const targetMux = 4 - -// --- conditional presets: constrain WHICH path is chosen, by route metadata --- -// -// Unlike the performance presets (which return a mux/min-hops SHAPE and let the -// host pick candidates), these pick a specific forward candidate via -// spec.Chosen so the selected route provably honors a CONSTRAINT expressed over -// the candidate's own metadata — the per-hop country (hops_geo), the per-hop -// transport types (transport_kinds), the hop PKs (hops) — or the wall clock -// (now_unix_nano). They are decide-time only (no on_tick): the constraint is a -// property of the chosen route, not something that drifts over the session. -// Parameters come from cli_overrides so the operator configures the constraint -// (avoid_geo / trusted_pks / business_hours) without a new preset. - -// splitSet parses a comma/space-separated override value into a lowercased -// membership set. Empty input yields an empty (never-nil) set. -func splitSet(s string) map[string]bool { - out := map[string]bool{} - for _, f := range strings.FieldsFunc(s, func(r rune) bool { return r == ',' || r == ' ' }) { - if f != "" { - out[strings.ToLower(f)] = true - } - } - return out -} - -// decideGeoAvoid picks the lowest-latency forward candidate whose hops transit -// NONE of the blocked countries in cli_overrides["avoid_geo"] (comma-separated -// ISO country codes). Constraint: the chosen route never passes through a -// blocked geography. If no candidate is clean (or none are offered) it returns -// the empty spec, deferring to the router rather than forcing a violating path. -func decideGeoAvoid(ctx routingContextWire, cands []candidateWire) routeSpecWire { - blocked := splitSet(ctx.CLIOverrides["avoid_geo"]) - if len(blocked) == 0 || len(cands) == 0 { - return routeSpecWire{} // nothing to enforce - } - var best *candidateWire - for i := range cands { - if candidateTransitsBlockedGeo(cands[i], blocked) { - continue - } - if best == nil || cands[i].EstLatencyMs < best.EstLatencyMs { - c := cands[i] - best = &c - } - } - if best == nil { - return routeSpecWire{} // every candidate violates; let the router decide - } - return routeSpecWire{Chosen: best, MinHops: 2} -} - -// candidateTransitsBlockedGeo reports whether any of the candidate's per-hop -// countries is in the blocked set. -func candidateTransitsBlockedGeo(c candidateWire, blocked map[string]bool) bool { - for _, g := range c.HopsGeo { - if blocked[strings.ToLower(g)] { - return true - } - } - return false -} - -// decideTransportDiverse picks the forward candidate whose hops span the MOST -// distinct transport types (ties broken by lower latency). Constraint: the -// chosen route maximizes transport-type diversity, so no single transport -// technology failing (e.g. all-stcpr under a firewall change) takes the whole -// path down. Empty spec when no candidates are offered. -func decideTransportDiverse(ctx routingContextWire, cands []candidateWire) routeSpecWire { - var best *candidateWire - bestDiversity := -1 - for i := range cands { - d := distinctCount(cands[i].TransportKinds) - switch { - case d > bestDiversity: - c := cands[i] - best, bestDiversity = &c, d - case d == bestDiversity && best != nil && cands[i].EstLatencyMs < best.EstLatencyMs: - c := cands[i] - best = &c - } - } - if best == nil { - return routeSpecWire{} - } - // A mux of 2 over the diverse path keeps a spare leg on a different carrier. - return routeSpecWire{Chosen: best, Mux: 2, MinHops: 2} -} - -// distinctCount counts unique (case-folded) entries in a slice. -func distinctCount(xs []string) int { - seen := map[string]bool{} - for _, x := range xs { - seen[strings.ToLower(x)] = true - } - return len(seen) -} - -// decideTrustTiered prefers routes that transit ONLY trusted intermediaries -// (cli_overrides["trusted_pks"], comma-separated hop PKs). It is TIERED: a -// fully-trusted candidate wins outright (lowest latency among them); if none is -// fully trusted it falls back to the candidate with the MOST trusted hops, so -// the route is as trusted as the topology allows rather than failing closed. -// Empty spec when no trust set is configured or no candidates are offered. -func decideTrustTiered(ctx routingContextWire, cands []candidateWire) routeSpecWire { - trusted := splitSet(ctx.CLIOverrides["trusted_pks"]) - if len(trusted) == 0 || len(cands) == 0 { - return routeSpecWire{} - } - var bestFull, bestPartial *candidateWire - bestPartialScore := -1 - for i := range cands { - n := trustedHopCount(cands[i], trusted) - if len(cands[i].Hops) > 0 && n == len(cands[i].Hops) { - if bestFull == nil || cands[i].EstLatencyMs < bestFull.EstLatencyMs { - c := cands[i] - bestFull = &c - } - } - if n > bestPartialScore { - c := cands[i] - bestPartial, bestPartialScore = &c, n - } - } - if bestFull != nil { - return routeSpecWire{Chosen: bestFull, MinHops: 2} - } - if bestPartial != nil { - return routeSpecWire{Chosen: bestPartial, MinHops: 2} - } - return routeSpecWire{} -} - -// trustedHopCount counts how many of a candidate's hops are in the trusted set. -func trustedHopCount(c candidateWire, trusted map[string]bool) int { - n := 0 - for _, h := range c.Hops { - if trusted[strings.ToLower(h)] { - n++ - } - } - return n -} - -// decideTimeOfDay switches the route SHAPE by wall-clock hour (UTC), derived -// from ctx.now_unix_nano — no clock import, just arithmetic on the host-stamped -// timestamp. During the configured business-hours window -// (cli_overrides["business_hours"] = "START-END", default "9-17") it returns a -// lean single-route shape (latency-sensitive daytime traffic); outside it, -// a wide privacy mux with byte-spread and rotation (bulk/overnight). Constraint: -// the emitted shape is a deterministic function of the hour. -func decideTimeOfDay(ctx routingContextWire) routeSpecWire { - startH, endH := parseHourRange(ctx.CLIOverrides["business_hours"], 9, 17) - if inHourRange(hourOfDayUTC(ctx.NowUnixNano), startH, endH) { - return routeSpecWire{Mux: 1} // business hours: lean + low-latency - } - // off-hours: privacy-wide mux (mirrors rotating-bw's shape) - return routeSpecWire{Mux: 4, MinHops: 2, Distribution: "round-robin", RotationIntervalSeconds: 90} -} - -// hourOfDayUTC returns the UTC hour (0-23) of a unix-nanosecond timestamp using -// only integer arithmetic (avoids a "time" import in the wasm guest). -func hourOfDayUTC(unixNano int64) int { - if unixNano <= 0 { - return 0 - } - secOfDay := (unixNano / 1e9) % 86400 - return int(secOfDay / 3600) -} - -// parseHourRange parses "START-END" (24h) from an override, falling back to the -// given defaults on empty/malformed input. -func parseHourRange(s string, defStart, defEnd int) (int, int) { - a, b, ok := strings.Cut(strings.TrimSpace(s), "-") - if !ok { - return defStart, defEnd - } - start, okA := atoiHour(a) - end, okB := atoiHour(b) - if !okA || !okB { - return defStart, defEnd - } - return start, end -} - -// atoiHour parses a 0-23 hour string; returns ok=false otherwise. -func atoiHour(s string) (int, bool) { - s = strings.TrimSpace(s) - if s == "" { - return 0, false - } - n := 0 - for _, r := range s { - if r < '0' || r > '9' { - return 0, false - } - n = n*10 + int(r-'0') - } - if n < 0 || n > 23 { - return 0, false - } - return n, true -} - -// inHourRange reports whether hour is within [start, end), handling a window -// that wraps past midnight (start > end, e.g. 22-6). -func inHourRange(hour, start, end int) bool { - if start <= end { - return hour >= start && hour < end - } - return hour >= start || hour < end // wraps midnight -} - //export on_tick func onTick(inPtr, inLen uint32) uint64 { var input tickInputWire if err := json.Unmarshal(readInput(inPtr, inLen), &input); err != nil { return 0 } - // rotating-bw, latency-adaptive, elastic-mux, probe-and-prune and the - // composite adaptive preset have tick logic; app-mux and unknown - // presets are static, so they take no rotation action. - switch input.Preset { - case "rotating-bw": - return tickRotatingBW(input) - case "latency-adaptive": - return tickLatencyAdaptive(input) - case "elastic-mux": - return tickElasticMux(input) - case "probe-and-prune": - return tickProbeAndPrune(input) - case "adaptive": - return tickAdaptive(input) - default: - return 0 - } -} - -// reliableMuxKind reports whether a transport type is reliable enough to ANCHOR -// a mux active set. stcpr/sudph/squicr/stcp are direct, well-behaved for -// sustained multiplexed throughput. webrtc (SCTP-over-DTLS), ws, wt and the dmsg -// relay are NOT — webrtc especially has poor sustained throughput and drops -// under load; a mux made entirely of webrtc legs collapses. They are still -// USABLE as bonus capacity (promoted for throughput), just never the anchor. -func reliableMuxKind(kind string) bool { - switch kind { - case "stcpr", "sudph", "squicr", "stcp": - return true - } - return false -} - -// tickRotatingBW manages the mux by STANDBY, never by dropping. Legs, once -// established, are never torn down — the tick only flips each leg's standby flag -// (setLegStandby: no teardown, no setup round-trip, in-flight bytes drain). The -// policy keeps ~targetMux legs ACTIVE, anchored on reliable transport types, and -// PARKS the rest on warm standby; it promotes a webrtc/other leg only when it -// needs the aggregate throughput (not enough reliable legs to reach targetMux). -// This is why an all-reliable anchor keeps the group alive even as fragile -// bonus legs come and go, and it spreads bytes across the reliable set over time -// for the privacy property — all with no throughput dip. One flag-flip per tick -// (the ABI applies one RotationAction). Decisions are on MEASUREMENT: leg Kind + -// standby state come from the host per-leg telemetry (hook.go). -func tickRotatingBW(input tickInputWire) uint64 { - var relAct, relSb, fragAct, fragSb []int - for _, l := range input.Legs { - if !l.Alive { - continue - } - rel := reliableMuxKind(l.Kind) - switch { - case l.Standby && rel: - relSb = append(relSb, l.Index) - case l.Standby: - fragSb = append(fragSb, l.Index) - case rel: - relAct = append(relAct, l.Index) - default: - fragAct = append(fragAct, l.Index) - } - } - active := len(relAct) + len(fragAct) - alive := active + len(relSb) + len(fragSb) - if alive == 0 { - return 0 - } - lo := func(s []int) int { - m := s[0] - for _, v := range s { - if v < m { - m = v - } - } - return m - } - hi := func(s []int) int { - m := s[0] - for _, v := range s { - if v > m { - m = v - } - } - return m - } - - // NEVER emit AddLeg / DropLegs here — leg GROWTH and death-replacement are the - // route group's own self-heal job (it tops up to the mux degree on any leg - // death; aliveLegCount counts standby legs, so parking one never triggers a - // re-grow). on_tick chasing the count with adds only fought self-heal and made - // a flapping webrtc leg thrash (drop→add→drop). on_tick's ONLY job is to pick - // which established legs are ACTIVE vs parked — pure standby flips, no churn. - var a rotationActionWire - switch { - case len(relAct) >= 1 && len(fragAct) > 0: - // A reliable leg is already carrying — PARK a fragile (webrtc) active leg - // so the reliable anchor(s) carry the stream and a webrtc drop can't - // disrupt it. Fragile legs stay warm on standby (never dropped), promoted - // only if the reliable anchor is lost (below). One per tick; repeats until - // the active set is reliable-only. - a = rotationActionWire{DemoteToStandby: []int{hi(fragAct)}} - case len(fragAct) > 0 && len(relSb) > 0: - // No reliable leg active yet, but a reliable spare exists — swap it in for - // a fragile one (promote reliable, park fragile). No drop, no dip. - a = rotationActionWire{PromoteFromStandby: []int{lo(relSb)}, DemoteToStandby: []int{hi(fragAct)}} - case len(relAct) < targetMux && len(relSb) > 0: - // Below the reliable-active target — promote a reliable spare. - a = rotationActionWire{PromoteFromStandby: []int{lo(relSb)}} - case len(relAct) == 0 && len(fragAct) == 0 && len(fragSb) > 0: - // No reliable legs at all and nothing active — promote a fragile spare so - // SOME leg carries (a slow/flaky leg beats a dead group). Bonus, not anchor. - a = rotationActionWire{PromoteFromStandby: []int{lo(fragSb)}} - case len(relAct) > targetMux: - // More reliable active than target — park the newest to settle at target. - a = rotationActionWire{DemoteToStandby: []int{hi(relAct)}} - case len(fragAct) == 0 && len(relSb) > 0 && len(relAct) > 0: - // Steady state, reliable-only active with a reliable spare: gentle crawl — - // rotate the reliable active set through the spare (spread bytes for - // privacy). No dip, no drop. - a = rotationActionWire{PromoteFromStandby: []int{lo(relSb)}, DemoteToStandby: []int{lo(relAct)}} - default: - return 0 - } - out, err := json.Marshal(a) + action := engine.OnTick(input.Preset, legsToPreset(input.Legs)) + out, err := json.Marshal(actionToWire(action)) if err != nil { return 0 } return writeOutput(out) } -// latAdaptEWMA holds the exponentially-weighted moving average of each -// leg's latency_ms, keyed by the leg's STABLE transport_id (not its -// index, which shifts as the tps[] slice compacts on drop/add). This is -// the per-leg state that lets tickLatencyAdaptive act on a smoothed -// signal rather than the raw instantaneous latency — on the live mesh a -// single leg's latency_ms jumps around wildly tick-to-tick (e.g. -// 75→245→13000→180), and evicting on the raw value chases that noise. -// Keys are pruned each tick to the current alive-or-present leg set so a -// dropped transport's entry doesn't linger. -var latAdaptEWMA = map[string]float64{} - -// latAdaptSeen is a scratch set reused each tick to record which -// transport_ids appear in the current leg snapshot, so stale keys can be -// pruned from latAdaptEWMA. Package-global (rather than tick-local) only -// to avoid a per-tick allocation; it is cleared at the top of every tick. -var latAdaptSeen = map[string]bool{} - -// latAdaptAlpha is the EWMA smoothing factor. 0.3 weights the newest -// sample at 30% and the running average at 70% — enough to track a -// genuine sustained latency shift within a few ticks while damping a -// lone transient spike to a fraction of its raw magnitude. -const latAdaptAlpha = 0.3 - -// tickLatencyAdaptive is the latency-adaptive rotation logic. It differs -// from rotating-bw in what it evicts and WHEN: -// -// rotating-bw drops the OLDEST alive leg every single tick — an -// unconditional churn that keeps the set drifting for -// traffic-analysis resistance. -// -// latency-adaptive instead drops the SLOWEST alive leg, and ONLY when -// that leg is a clear outlier — its SMOOTHED latency is at least 1.5x -// the median of the alive legs' smoothed latencies. This hysteresis -// band means: once the set has converged to a cluster of comparably- -// fast legs (worst < 1.5x median), NO leg qualifies and the policy -// holds — no churn. It converges toward a low-latency disjoint set and -// then stops thrashing, the opposite of rotating-bw's every-tick -// rotation. -// -// The eviction decision runs on an EWMA of each leg's latency, keyed by -// the leg's stable transport_id, rather than on the raw per-tick -// latency_ms. Raw latency on the live mesh is noisy — one leg can read -// 75ms one tick and 13000ms the next with no real change in its quality — -// and evicting on the instantaneous value chases that noise, tearing down -// a perfectly good leg on a transient spike. Smoothing filters those -// spikes so eviction fires only on a PERSISTENT latency difference. The -// stable transport_id key is what makes a per-leg average meaningful at -// all: the leg's index shifts when the slice compacts, so index-keyed -// state would smear one leg's history onto another after any drop. -// -// alive_count == 0 → no-op (nothing measured yet) -// alive_count < target_mux → add only (recover toward target; never -// shrink below it) -// alive_count >= target_mux && -// smoothed_worst > 0 && -// smoothed_worst >= -// 1.5 * smoothed_median → drop slowest + add + exclude its hops -// (evict the outlier; exclude its -// intermediates so the replacement differs) -// otherwise (converged) → no-op (hold; do not churn) -func tickLatencyAdaptive(input tickInputWire) uint64 { - const targetMux = 4 - - // Update the per-leg EWMA and record which transport_ids are present - // this tick so stale keys can be pruned afterward. - for k := range latAdaptSeen { - delete(latAdaptSeen, k) - } - for _, l := range input.Legs { - tid := l.TransportID - if tid == "" { - continue - } - latAdaptSeen[tid] = true - // Only fold in a real measurement; latency_ms==0 means "unknown" - // and must not drag the average toward zero. - if l.Alive && l.LatencyMs > 0 { - sample := float64(l.LatencyMs) - if prev, ok := latAdaptEWMA[tid]; ok { - latAdaptEWMA[tid] = latAdaptAlpha*sample + (1-latAdaptAlpha)*prev - } else { - // Seed with the first sample so the average starts on the - // leg's own latency rather than climbing from zero. - latAdaptEWMA[tid] = sample - } - } - } - // Prune EWMA keys for transport_ids no longer in the leg set so a - // dropped leg's history can't resurface if its index is later reused. - for tid := range latAdaptEWMA { - if !latAdaptSeen[tid] { - delete(latAdaptEWMA, tid) - } - } - - // Evict-slowest-outlier logic, run on the SMOOTHED latencies over the - // ACTIVE legs only — warm-standby spares aren't carrying traffic and - // aren't judged. Membership changes are standby flips (no teardown): the - // group runs targetMux active + a warm reserve, so evicting the outlier is - // a hot-swap (promote a spare, demote the outlier) with no width dip. - sets := classifyLegs(input.Legs) - worstIdx := -1 - worstSmoothed := -1.0 - var smoothed []float64 - for _, l := range input.Legs { - if !l.Alive || l.Standby { - continue - } - sm, ok := latAdaptEWMA[l.TransportID] - if !ok { - // No smoothed value yet (leg never reported a latency) — it - // can't be judged an outlier this tick. - continue - } - smoothed = append(smoothed, sm) - if sm > worstSmoothed { - worstSmoothed = sm - worstIdx = l.Index - } - } - // Below target width: recover by promoting a warm spare (no dip), or add a - // fresh leg when none is parked. Covers activeCount==0. - if sets.activeCount < targetMux { - return growActive(sets) - } - // Above target width (the warm reserve is still active): park the oldest as - // a standby spare so a later eviction can hot-swap without a dip. - if sets.activeCount > targetMux { - return shedActive(sets, sets.oldestActive) - } - // Need at least a couple of smoothed samples to compute a meaningful - // median; otherwise hold. - if len(smoothed) < 2 || worstIdx < 0 || worstSmoothed <= 0 { - return 0 - } - - // Median of the active legs' smoothed latencies. - sort.Float64s(smoothed) - median := medianSorted(smoothed) +// --- wire <-> preset conversions --- - // Hysteresis: only evict a clear outlier (smoothed worst >= 1.5x smoothed - // median). Hot-swap it out (promote a warm spare + demote the outlier) so - // the width never dips; the demoted leg stays warm for a later re-promote. - // Once the set has converged this is false for every leg, so we hold. - if median > 0 && worstSmoothed >= 1.5*median { - return swapActive(sets, worstIdx) +func ctxToPreset(c routingContextWire) preset.Context { + return preset.Context{ + App: c.App, + PeerPK: c.PeerPK, + Port: c.Port, + NowUnixNano: c.NowUnixNano, + CLIOverrides: c.CLIOverrides, + IsDirectDial: c.IsDirectDial, + TransportKind: c.TransportKind, + ReverseCandidates: candsToPreset(c.ReverseCandidates), } - // Converged: all active legs comparably fast — hold. - return 0 } -// --- elastic-mux: AIMD scaling of the mux leg count to load --- -// -// elastic-mux treats the mux width as a control variable, not a constant. -// Its on_tick runs a classic AIMD (additive-increase / multiplicative-… -// here just single-step-decrease) controller over the group's aggregate -// received throughput: -// -// SATURATED (throughput near the observed peak) → ADD one leg. Under -// real load more disjoint paths raise the aggregate the group can pull, -// so we grow — additively, one leg per tick — up to a hard cap. -// IDLE (throughput far below peak, for two consecutive ticks) → DROP one -// leg. The width bought under load is now wasted transport state and -// fleet bandwidth, so we release it — one leg at a time, down to a floor. -// Otherwise → hold. -// -// The controller drives its decisions off an EWMA of the per-tick -// received-byte DELTA (recv_bytes is a monotonic counter; the increment -// since last tick is this interval's throughput), NOT the raw delta. -// Raw per-tick byte deltas on the live mesh are extremely spiky — a bulk -// download arrives in bursts, so one tick reads near-zero and the next -// reads a full window — and an AIMD loop driven by the raw value would -// oscillate (grow, shrink, grow) chasing that jitter. Smoothing the -// delta with an EWMA (α=0.3) gives a stable load signal the controller -// can compare against a slowly-decaying running peak without thrashing. - -// elasticPrevRecv holds each leg's last-seen recv_bytes counter, keyed by -// the STABLE transport_id (index shifts as legs drop/add), so the next -// tick can compute that leg's byte delta. Stale keys are pruned each tick. -var elasticPrevRecv = map[string]uint64{} - -// elasticSeen is a scratch set (reused, cleared each tick) recording which -// transport_ids appear this tick, so stale elasticPrevRecv keys can be -// pruned. Package-global only to avoid a per-tick allocation. -var elasticSeen = map[string]bool{} - -// elasticThroughputEWMA is the smoothed aggregate received throughput -// (sum of alive legs' per-tick recv_bytes deltas), the load signal the -// AIMD controller acts on. elasticPeak is the running high-water mark of -// that smoothed signal, decayed a little each tick so a transient burst -// long past stops holding the bar artificially high. elasticIdleCount is -// the consecutive-idle tick counter (so a single quiet tick can't shrink -// the group). elasticSeeded guards the one-time seed on the first tick -// that observes non-zero throughput. -var ( - elasticThroughputEWMA float64 - elasticPeak float64 - elasticIdleCount int - elasticSeeded bool -) - -const ( - // elasticFloor / elasticCap bound the leg count AIMD may reach. - elasticFloor = 2 - elasticCap = 6 - // elasticAlpha is the EWMA smoothing factor for the throughput signal - // (newest delta weighted 30%, running average 70%). - elasticAlpha = 0.3 - // elasticPeakDecay bleeds the running peak down ~2%/tick so the - // saturation bar tracks a sustained drop in achievable throughput - // instead of being pinned forever by one historical burst. - elasticPeakDecay = 0.98 -) - -// tickElasticMux is the elastic-mux AIMD controller. See the block comment -// above for the load model and why the signal is a smoothed byte-delta. -// -// alive_count == 0 → no-op (nothing measured) -// smoothed >= 0.80*peak && alive < cap → add one leg (grow to load) -// smoothed < 0.25*peak for >=2 ticks -// && alive > floor → drop oldest leg (release) -// otherwise → hold -func tickElasticMux(input tickInputWire) uint64 { - // Compute this interval's aggregate received byte-delta over alive - // legs, refreshing each leg's prev-counter, and record presence for - // stale-key pruning. Also find the oldest (lowest-index) alive leg to - // release when shrinking, and count alive legs. - for k := range elasticSeen { - delete(elasticSeen, k) - } - var rawTotal float64 - // active/standby partition: only ACTIVE legs carry traffic and count - // toward the AIMD width; standby spares are the warm reserve a grow - // promotes back with no dip. - sets := classifyLegs(input.Legs) - for _, l := range input.Legs { - if !l.Alive { - continue - } - tid := l.TransportID - if tid == "" { - continue - } - elasticSeen[tid] = true - // Only fold an ACTIVE leg's non-negative delta into the load signal. - // Standby spares don't send; their counter is still tracked so it - // doesn't read as a reset when the leg is later re-promoted. - if prev, ok := elasticPrevRecv[tid]; ok && l.RecvBytes >= prev && !l.Standby { - rawTotal += float64(l.RecvBytes - prev) - } - elasticPrevRecv[tid] = l.RecvBytes - } - // Prune prev-counters for transport_ids no longer present. - for tid := range elasticPrevRecv { - if !elasticSeen[tid] { - delete(elasticPrevRecv, tid) - } - } - if sets.activeCount == 0 { - // Nothing active: promote a warm spare if one is parked, else hold. - if sets.promotable >= 0 { - return growActive(sets) - } - return 0 - } - - // Fold the raw delta into the smoothed signal and maintain the peak. - // Seed both on the first tick that sees real throughput so the signal - // starts on a genuine value rather than ramping up from zero. - if !elasticSeeded { - if rawTotal > 0 { - elasticThroughputEWMA = rawTotal - elasticPeak = rawTotal - elasticSeeded = true - } - // No load signal yet — hold and keep measuring. - return 0 - } - elasticThroughputEWMA = elasticAlpha*rawTotal + (1-elasticAlpha)*elasticThroughputEWMA - elasticPeak *= elasticPeakDecay - if elasticThroughputEWMA > elasticPeak { - elasticPeak = elasticThroughputEWMA +func candsToPreset(cs []candidateWire) []preset.Candidate { + if cs == nil { + return nil } - if elasticPeak <= 0 { - return 0 - } - - smoothed := elasticThroughputEWMA - // Track consecutive idle ticks independently of whether we can act on - // them, so the "two quiet ticks" requirement is about load, not width. - if smoothed < 0.25*elasticPeak { - elasticIdleCount++ - } else { - elasticIdleCount = 0 + out := make([]preset.Candidate, len(cs)) + for i, c := range cs { + out[i] = candToPreset(c) } + return out +} - // Additive increase: saturated and below the cap → grow one leg. Promote a - // warm spare (parked by an earlier shrink) with no dip when one exists; - // otherwise add a fresh leg under genuine load. - if smoothed >= 0.80*elasticPeak && sets.activeCount < elasticCap { - return growActive(sets) +func candToPreset(c candidateWire) preset.Candidate { + return preset.Candidate{ + Hops: c.Hops, + HopsGeo: c.HopsGeo, + EstLatencyMs: c.EstLatencyMs, + TransportKinds: c.TransportKinds, } - // Decrease: sustained idle and above the floor → release one leg by PARKING - // it as a warm standby (no teardown; re-promoted instantly when load - // returns), falling back to a real drop only once the standby pool is full. - // Reset the idle counter after acting so we step down gracefully (one leg - // every two idle ticks) rather than collapsing to the floor. - if elasticIdleCount >= 2 && sets.activeCount > elasticFloor { - elasticIdleCount = 0 - return shedActive(sets, sets.oldestActive) - } - // Hold. - return 0 } -// --- probe-and-prune: continuous explore/exploit over a fresh path --- -// -// probe-and-prune keeps a fixed-width "established" set (probeTarget legs) -// but never stops looking for a better path than the ones it holds. Its -// on_tick is a small state machine: -// -// EXPLORE: when idle and exactly at target width, add ONE speculative -// leg over a fresh path (the host picks the candidate; on_tick can only -// request an add). The next tick identifies that new leg by diffing the -// current transport_ids against the pre-add "known" set, and puts it on -// probation (probeTID / probeAge). -// OBSERVE: let the probe run probeObserveTicks ticks so its EWMA latency -// (same stable-transport_id-keyed smoothing latency-adaptive uses) is a -// real measurement, not first-packet noise. -// EXPLOIT: compare the probe's smoothed latency to the WORST (highest- -// EWMA) established leg. If the probe is better, it GRADUATES — drop the -// worst established leg, and the probe takes its place. If not, the -// experiment FAILED — drop the probe itself. Either way width returns to -// target and the cycle repeats. -// -// Net effect: the established set ratchets toward lower latency one path -// at a time, spending at most one extra leg's worth of transport state -// while a probe is in flight, and never growing past target permanently. - -// probeEWMA is the per-leg smoothed latency, keyed by stable transport_id -// (same rationale as latAdaptEWMA). probeSeen is the reused prune scratch -// set; probeAliveIdx maps transport_id → current leg index each tick so a -// DropLegs decision made about a transport can be translated to the index -// the host expects. prevKnownTIDs snapshots the established set just before -// an add so the following tick can diff out the newly-added probe leg. -var ( - probeEWMA = map[string]float64{} - probeSeen = map[string]bool{} - probeAliveIdx = map[string]int{} - prevKnownTIDs = map[string]bool{} -) - -// probeTID is the transport_id of the leg currently on probation ("" = -// none). probeAge counts ticks since the probe was adopted. probePending -// is set the tick we request the add, so the next tick knows to adopt the -// newly-appeared leg as the probe. -var ( - probeTID string - probeAge int - probePending bool -) - -const ( - // probeTarget is the established-set width probe-and-prune maintains - // (matches decideProbeAndPrune's Mux). A probe transiently makes it - // target+1. - probeTarget = 3 - // probeObserveTicks is how many ticks a probe runs before it is judged, - // so its EWMA latency reflects steady behavior, not startup noise. - probeObserveTicks = 3 - // probeAlpha is the latency EWMA smoothing factor (as in latency-…). - probeAlpha = 0.3 -) - -// tickProbeAndPrune is the probe-and-prune explore/exploit controller. See -// the block comment above for the state machine. -func tickProbeAndPrune(input tickInputWire) uint64 { - // Update per-leg latency EWMA, map transport_id → index, count alive, - // and record presence for stale-key pruning. - for k := range probeSeen { - delete(probeSeen, k) - } - for k := range probeAliveIdx { - delete(probeAliveIdx, k) - } - sets := classifyLegs(input.Legs) - activeCount := 0 - for _, l := range input.Legs { - tid := l.TransportID - if tid == "" { - continue - } - probeSeen[tid] = true - if !l.Alive { - continue - } - // Fold in only real measurements; latency_ms==0 means "unknown". EWMA - // tracks any alive leg (harmless for a standby spare). - if l.LatencyMs > 0 { - sample := float64(l.LatencyMs) - if prev, ok := probeEWMA[tid]; ok { - probeEWMA[tid] = probeAlpha*sample + (1-probeAlpha)*prev - } else { - probeEWMA[tid] = sample - } - } - // Only ACTIVE legs form the "established" set the probe machine - // manages; a warm standby (a previously graduated-out incumbent, - // parked not torn down) is neither an established leg nor the probe. - if l.Standby { - continue - } - activeCount++ - probeAliveIdx[tid] = l.Index +func legsToPreset(ls []legInfoWire) []preset.LegInfo { + if ls == nil { + return nil } - // Prune EWMA keys for transport_ids no longer present. - for tid := range probeEWMA { - if !probeSeen[tid] { - delete(probeEWMA, tid) + out := make([]preset.LegInfo, len(ls)) + for i, l := range ls { + out[i] = preset.LegInfo{ + Index: l.Index, + Kind: l.Kind, + TransportID: l.TransportID, + LatencyMs: l.LatencyMs, + Alive: l.Alive, + Standby: l.Standby, + SentBytes: l.SentBytes, + RecvBytes: l.RecvBytes, + Retransmits: l.Retransmits, + Hops: l.Hops, } } + return out +} - // Adopt: if we requested an add last tick, the new leg should now be - // present — it is the alive transport_id absent from the pre-add known - // set. Put it on probation. - if probePending { - probePending = false - for tid := range probeAliveIdx { - if !prevKnownTIDs[tid] { - probeTID = tid - probeAge = 0 - break - } - } +func specToWire(s preset.Spec) routeSpecWire { + w := routeSpecWire{ + Mux: s.Mux, + ForwardMux: s.ForwardMux, + ReverseMux: s.ReverseMux, + MinHops: s.MinHops, + ForwardMinHops: s.ForwardMinHops, + ReverseMinHops: s.ReverseMinHops, + Fallback: s.Fallback, + Distribution: s.Distribution, + RotationIntervalSeconds: s.RotationIntervalSeconds, } - - // Active probe: observe, then graduate-or-discard. - if probeTID != "" { - idx, alive := probeAliveIdx[probeTID] - if !alive { - // Probe died on its own — abandon the experiment. - probeTID = "" - probeAge = 0 - return 0 - } - probeAge++ - if probeAge < probeObserveTicks { - return 0 - } - // Judge: probe's smoothed latency vs the worst established leg. - probeSm, okProbe := probeEWMA[probeTID] - worstIdx := -1 - worstSm := -1.0 - for tid, i := range probeAliveIdx { - if tid == probeTID { - continue - } - sm, ok := probeEWMA[tid] - if !ok { - continue - } - if sm > worstSm { - worstSm = sm - worstIdx = i - } - } - if !okProbe || worstIdx < 0 { - // Not enough latency signal to judge yet — keep observing. - return 0 - } - probeTID = "" - if probeSm < worstSm { - // Probe graduates: PARK the worst established leg as a warm standby - // (no teardown — it re-promotes instantly if a leg later dies) - // rather than tearing it down; the probe stays and joins the - // established set next tick. Evict-by-demote, no capacity dip. - return shedActive(sets, worstIdx) - } - // Failed experiment: drop the SPECULATIVE probe leg (it was added fresh - // this cycle, so removing it just returns the group to target — no - // working leg is lost), keeping the incumbents. - return writeAction(rotationActionWire{DropLegs: []int{idx}}) + if s.Chosen != nil { + c := candToWire(*s.Chosen) + w.Chosen = &c } - - // Explore: no probe in flight and exactly at target ACTIVE width → snapshot - // the known set and request one speculative leg over a fresh path. - if !probePending && activeCount == probeTarget { - for k := range prevKnownTIDs { - delete(prevKnownTIDs, k) - } - for tid := range probeAliveIdx { - prevKnownTIDs[tid] = true - } - probePending = true - return writeAction(rotationActionWire{AddLeg: true}) + if s.ReverseChosen != nil { + c := candToWire(*s.ReverseChosen) + w.ReverseChosen = &c } - return 0 + return w } -// medianSorted returns the median of an already-sorted float slice (mean of -// the two middle elements for an even count). Shared by the latency-adaptive -// and adaptive controllers so the outlier test uses one definition. -func medianSorted(s []float64) float64 { - n := len(s) - if n == 0 { - return 0 - } - if n%2 == 1 { - return s[n/2] +func candToWire(c preset.Candidate) candidateWire { + return candidateWire{ + Hops: c.Hops, + HopsGeo: c.HopsGeo, + EstLatencyMs: c.EstLatencyMs, + TransportKinds: c.TransportKinds, } - return (s[n/2-1] + s[n/2]) / 2 } -// --- shared no-dip controller helpers ------------------------------------- -// -// Every adaptive controller manages an ACTIVE sending set drawn from a -// slightly larger warm pool. Membership changes are expressed as warm-standby -// flips, never teardowns: a demoted leg keeps its rules alive and re-promotes -// instantly, so the active width never dips during an eviction or a load -// swing. GROWTH of the underlying leg pool stays the route group's self-heal -// job (aliveLegCount counts standby, so parking never re-grows); on_tick only -// picks which legs are active. This is the rotating-bw discipline, factored -// out so latency-adaptive / elastic-mux / probe-and-prune / adaptive share it. - -// nodipStandbyMax caps the warm-standby pool a shed/park may hold before it -// falls back to a real teardown, so parked-but-unused legs don't accumulate. -const nodipStandbyMax = 2 - -// legSets is the per-tick active/standby partition of a route group's legs. -// active = alive && !standby (carrying traffic); standby = alive && standby -// (warm spares, promoted with no setup round-trip). -type legSets struct { - activeCount int - standbyCount int - oldestActive int // lowest-index active leg (the "oldest" to shed); -1 if none - promotable int // lowest-index standby leg (spare to promote); -1 if none -} - -// classifyLegs partitions the leg snapshot into active/standby counts and the -// two indices the controllers act on (oldest active to shed, lowest standby to -// promote). -func classifyLegs(legs []legInfoWire) legSets { - s := legSets{oldestActive: -1, promotable: -1} - for _, l := range legs { - if !l.Alive { - continue - } - if l.Standby { - s.standbyCount++ - if s.promotable == -1 || l.Index < s.promotable { - s.promotable = l.Index - } - continue - } - s.activeCount++ - if s.oldestActive == -1 || l.Index < s.oldestActive { - s.oldestActive = l.Index - } +func actionToWire(a preset.RotationAction) rotationActionWire { + return rotationActionWire{ + DropLegs: a.DropLegs, + AddLeg: a.AddLeg, + ExcludeHops: a.ExcludeHops, + DemoteToStandby: a.DemoteToStandby, + PromoteFromStandby: a.PromoteFromStandby, } - return s } -// growActive raises the active width by one with no setup dip when a warm -// spare is parked (instant promote); otherwise it requests a genuinely fresh -// leg (real capacity addition under load — not a tear-and-rebuild). -func growActive(s legSets) uint64 { - if s.promotable >= 0 { - return writeAction(rotationActionWire{PromoteFromStandby: []int{s.promotable}}) - } - return writeAction(rotationActionWire{AddLeg: true}) -} +// --- thin per-preset wrappers used by main_test.go (the conditional-preset +// constraint tests). They delegate to the shared preset package so the tests +// exercise the single source of truth through the wire types. --- -// swapActive evicts leg idx and fills its slot from the warm reserve in the -// SAME tick so the active width never dips (hot-swap). When no spare is parked -// it parks idx alone and lets a subsequent grow refill from the reserve — -// still no teardown, just a transient one-leg dip until self-heal/grow catches -// up. -func swapActive(s legSets, idx int) uint64 { - if s.promotable >= 0 { - return writeAction(rotationActionWire{ - PromoteFromStandby: []int{s.promotable}, - DemoteToStandby: []int{idx}, - }) - } - return writeAction(rotationActionWire{DemoteToStandby: []int{idx}}) +func decideGeoAvoid(ctx routingContextWire, cands []candidateWire) routeSpecWire { + return specToWire(preset.Decide("geo-avoid", ctxToPreset(ctx), candsToPreset(cands))) } -// shedActive removes leg idx from the active set with no capacity dip: parks it -// as a warm standby (instant re-promote) while the pool has room, else tears it -// down. Used by the idle-shrink / prune dimensions — the parked legs become the -// warm reserve a later grow promotes back. -func shedActive(s legSets, idx int) uint64 { - if s.standbyCount < nodipStandbyMax { - return writeAction(rotationActionWire{DemoteToStandby: []int{idx}}) - } - return writeAction(rotationActionWire{DropLegs: []int{idx}}) +func decideTransportDiverse(ctx routingContextWire, cands []candidateWire) routeSpecWire { + return specToWire(preset.Decide("transport-diverse", ctxToPreset(ctx), candsToPreset(cands))) } -// --- adaptive: the COMPOSITE performance default (size+membership+explore) --- -// -// adaptive is the intended converged default. It runs the three standalone -// performance controllers — elastic-mux (SIZE: AIMD the leg count to load), -// latency-adaptive (MEMBERSHIP: evict the slowest leg toward a low-latency -// set), and probe-and-prune (EXPLORE: speculatively try a fresh path and keep -// it only if it is better) — under ONE arbitrated on_tick that performs AT -// MOST ONE structural action (an add OR a drop) per tick. -// -// Why one action per tick: each sub-controller can independently want to add -// or drop a leg on the same tick (load says "grow", latency says "evict the -// outlier", the explorer says "probe"). Letting them all act would churn the -// route group — multiple simultaneous adds/drops tear legs up and down faster -// than the host can build them and faster than the EWMA signals can settle, -// which is exactly the thrash the smoothing was added to avoid. Arbitrating -// to a single action per tick lets each dimension make steady, observable -// progress: the group takes one deliberate step, the next tick re-measures on -// the settled state, and the highest-priority need is served first. -// -// Arbitration priority each tick (first match fires, then return): -// -// 1. aliveCount == 0 → hold (nothing measured yet). -// 2. RECOVER (aliveCount < floor) → add one leg. Safety -// first: a group starved below the floor is refilled before anything -// else is considered. -// 3. MEMBERSHIP (latency): worst alive leg's EWMA latency is a >=1.5x -// outlier over the alive-leg median → drop it + add + exclude its hops. -// Correctness of the set (drop a genuinely bad path) outranks sizing. -// 4. GROW (load): smoothed throughput >= 0.80*peak and aliveCount < cap → -// add one leg. Under real saturation more disjoint paths raise the -// aggregate the group can pull. -// 5. SHRINK (idle): smoothed throughput < 0.25*peak for >=2 consecutive -// ticks and aliveCount > floor → drop the oldest leg (release wasted -// width). Reset the idle counter so we step down one leg at a time. -// 6. EXPLORE: otherwise, when the set is stable (no probe in flight, -// aliveCount == target), every exploreEvery ticks start a probe — add a -// speculative leg, observe its EWMA latency adaptObserve ticks, then keep -// it (drop the worst established leg) only if it beats that worst, else -// prune the probe. This is probe-and-prune's explore/exploit loop. -// 7. else → hold. -// -// Rules 3/4/5 are gated on "no probe in flight": while an experiment is mid- -// flight (the group is transiently target+1 wide) a membership/grow/shrink -// action would corrupt the probe's accounting and defeat the one-step-at-a- -// time discipline, so structural changes pause until the probe graduates or -// is pruned. The RECOVER rule is never gated — a starved group is refilled -// regardless. The size target starts at the decide Mux (3) and grow/shrink -// move it within [floor, cap] so EXPLORE always probes at the current settled -// width. -// -// All three dimensions share ONE set of per-transport_id state, refreshed at -// the top of EVERY tick before arbitration: a latency EWMA (adaptLatEWMA), a -// received-byte-delta throughput EWMA + decaying peak (adaptThroughputEWMA / -// adaptPeak), the consecutive-idle counter (adaptIdleCount), and the probe -// state machine (adaptProbeTID / adaptProbeAge / adaptProbePending / -// adaptPrevKnown). Keys are pruned to the present leg set each tick so a -// dropped transport's history can't linger or smear onto a reused index. - -var ( - // adaptLatEWMA is the per-leg smoothed latency, keyed by stable - // transport_id (same rationale as latAdaptEWMA/probeEWMA). - adaptLatEWMA = map[string]float64{} - // adaptPrevRecv is each leg's last-seen recv_bytes counter, keyed by - // transport_id, so the next tick can compute its throughput delta. - adaptPrevRecv = map[string]uint64{} - // adaptSeen is the reused scratch set of transport_ids present this tick - // (for stale-key pruning); adaptAliveIdx maps transport_id → current leg - // index for alive legs so a DropLegs decision can be expressed as the - // index the host expects. adaptPrevKnown snapshots the established set - // just before a probe add so the next tick can diff out the new leg. - adaptSeen = map[string]bool{} - adaptAliveIdx = map[string]int{} - adaptPrevKnown = map[string]bool{} -) - -var ( - // adaptThroughputEWMA / adaptPeak are the smoothed aggregate received - // throughput and its slowly-decaying high-water mark (the AIMD load - // signal). adaptSeeded guards the one-time seed on the first tick with - // real throughput. adaptIdleCount is the consecutive-idle tick counter. - adaptThroughputEWMA float64 - adaptPeak float64 - adaptSeeded bool - adaptIdleCount int - // adaptTick counts ticks (drives the explore cadence). adaptTarget is the - // current steady width EXPLORE probes at; it starts at the decide Mux and - // grow/shrink move it within [floor, cap]. - adaptTick int - adaptTarget = adaptDecideMux - // Probe state machine (see probe-and-prune): adaptProbeTID is the leg on - // probation ("" = none), adaptProbeAge counts ticks since adoption, - // adaptProbePending is set the tick we request the speculative add. - adaptProbeTID string - adaptProbeAge int - adaptProbePending bool -) - -const ( - // adaptDecideMux is the starting mux width decideAdaptive returns and the - // initial size target. Start LEAN at 1: the router picks the single fastest - // available path, and the on_tick controller GROWS (up to adaptCap) only - // when that leg saturates. Starting wide (mux>1) forced every dial to drag - // traffic across disjoint sibling legs, which on a fleet with unequal legs - // (a fast direct/low-hop path next to slow multi-hop ones) is a throughput - // LOSS — the mux reorder buffer stalls on the slowest leg. Lean-start + - // grow-on-load keeps the common case fast and only pays for extra legs when - // there is load to justify them. - adaptDecideMux = 1 - // adaptFloor / adaptCap bound the leg count the size dimension may reach. - // Floor 1 so an unsaturated session stays at its single fast leg; resilience - // comes from grow-on-failure + warm standbys, not from always-on wide mux. - adaptFloor = 1 - adaptCap = 6 - // adaptAlpha is the EWMA smoothing factor for both the latency and the - // throughput signals (newest sample weighted 30%). - adaptAlpha = 0.3 - // adaptPeakDecay bleeds the throughput peak down ~2%/tick so the - // saturation bar tracks a sustained drop rather than one historical burst. - adaptPeakDecay = 0.98 - // adaptExploreEvery is the explore cadence: start a probe every N ticks - // when the set is stable. - adaptExploreEvery = 6 - // adaptObserve is how many ticks a probe runs before it is judged. - adaptObserve = 3 - // adaptStandbyMax caps the warm-standby pool. SHRINK parks an idle leg as a - // warm standby (rules kept alive, not sending) instead of tearing it down, - // so a later GROW / eviction can promote it with no setup-node round-trip - // and no capacity dip. Once the pool is full SHRINK really drops the leg. - adaptStandbyMax = 2 -) - -// tickAdaptive is the composite arbitrated controller. See the block comment -// above for the load/latency/explore model and the priority ordering. -func tickAdaptive(input tickInputWire) uint64 { - adaptTick++ - - // --- Phase 1: refresh the shared per-transport_id state (every tick, - // before arbitration). One pass computes the latency EWMA, the aggregate - // received byte-delta, the alive count / oldest-alive index / alive - // index map, and records presence for stale-key pruning. - for k := range adaptSeen { - delete(adaptSeen, k) - } - for k := range adaptAliveIdx { - delete(adaptAliveIdx, k) - } - var rawTotal float64 - aliveCount := 0 - standbyCount := 0 - oldestAliveIdx := -1 - promotableIdx := -1 - for _, l := range input.Legs { - tid := l.TransportID - if tid != "" { - adaptSeen[tid] = true - } - if !l.Alive { - continue - } - // A warm standby is alive (rules kept, kept-alive) but not sending: it - // is a spare, not part of the active sending width. Record it as - // promotable and skip the active-leg accounting below (adaptAliveIdx, - // EWMA, throughput) so the size/latency/explore dimensions reason only - // over the legs actually carrying traffic. - if l.Standby { - standbyCount++ - if promotableIdx == -1 || l.Index < promotableIdx { - promotableIdx = l.Index - } - continue - } - aliveCount++ - if oldestAliveIdx == -1 || l.Index < oldestAliveIdx { - oldestAliveIdx = l.Index - } - if tid == "" { - continue - } - adaptAliveIdx[tid] = l.Index - // Latency EWMA — fold in only real measurements (0 == unknown). - if l.LatencyMs > 0 { - sample := float64(l.LatencyMs) - if prev, ok := adaptLatEWMA[tid]; ok { - adaptLatEWMA[tid] = adaptAlpha*sample + (1-adaptAlpha)*prev - } else { - adaptLatEWMA[tid] = sample - } - } - // Throughput: accumulate this interval's received byte-delta. Only a - // non-negative delta is real; a counter that went backwards means a - // reset/new leg, not throughput. - if prev, ok := adaptPrevRecv[tid]; ok && l.RecvBytes >= prev { - rawTotal += float64(l.RecvBytes - prev) - } - adaptPrevRecv[tid] = l.RecvBytes - } - // Prune state for transport_ids no longer present. - for tid := range adaptLatEWMA { - if !adaptSeen[tid] { - delete(adaptLatEWMA, tid) - } - } - for tid := range adaptPrevRecv { - if !adaptSeen[tid] { - delete(adaptPrevRecv, tid) - } - } - - // Fold the raw delta into the smoothed throughput signal and maintain the - // decaying peak. Seed on the first tick with real throughput so the signal - // starts on a genuine value rather than ramping from zero. saturated/idle - // stay false until seeded, so the size dimension holds until load appears. - saturated, idle := false, false - if !adaptSeeded { - if rawTotal > 0 { - adaptThroughputEWMA = rawTotal - adaptPeak = rawTotal - adaptSeeded = true - } - } else { - adaptThroughputEWMA = adaptAlpha*rawTotal + (1-adaptAlpha)*adaptThroughputEWMA - adaptPeak *= adaptPeakDecay - if adaptThroughputEWMA > adaptPeak { - adaptPeak = adaptThroughputEWMA - } - if adaptPeak > 0 { - saturated = adaptThroughputEWMA >= 0.80*adaptPeak - idle = adaptThroughputEWMA < 0.25*adaptPeak - } - } - // Track consecutive idle ticks independently of whether we can act on them - // (the "two quiet ticks" requirement is about load, not width). - if idle { - adaptIdleCount++ - } else { - adaptIdleCount = 0 - } - - // --- Phase 2: arbitration — at most ONE structural action, by priority. - - // 1. Nothing active — promote a warm spare if one is parked, else nothing - // to act on. (All legs standby is degenerate but recoverable without setup.) - if aliveCount == 0 { - if promotableIdx >= 0 { - return writeAction(rotationActionWire{PromoteFromStandby: []int{promotableIdx}}) - } - return 0 - } - // 2. RECOVER: starved below the floor. Promote a warm standby instantly if - // one exists (no setup, no dip); else add a fresh leg. Never gated on a probe. - if aliveCount < adaptFloor { - if promotableIdx >= 0 { - return writeAction(rotationActionWire{PromoteFromStandby: []int{promotableIdx}}) - } - return writeAction(rotationActionWire{AddLeg: true}) - } - - // A probe transiently makes the group target+1 wide; while it is in flight - // the size/membership dimensions pause so they can't corrupt its accounting. - probeInFlight := adaptProbePending || adaptProbeTID != "" - if !probeInFlight { - // 3. MEMBERSHIP: evict the slowest-EWMA leg iff it is a >=1.5x-median - // outlier. If a warm spare is parked, HOT-SWAP: promote it and demote - // the outlier in the SAME tick — the width never dips and the promoted - // leg needs no setup-node round-trip. The demoted leg stays warm, so a - // transient degradation can be undone by promoting it back later. Only - // when the pool is empty do we fall back to a cold drop+add. - if idx, hops, ok := adaptWorstOutlier(input.Legs); ok { - if promotableIdx >= 0 { - return writeAction(rotationActionWire{ - PromoteFromStandby: []int{promotableIdx}, - DemoteToStandby: []int{idx}, - }) - } - return writeAction(rotationActionWire{ - DropLegs: []int{idx}, - AddLeg: true, - ExcludeHops: hops, - }) - } - // 4. GROW: saturated and below the cap. Promote a warm spare instantly - // if one is parked (no setup, no dip); else add one leg to load. - if saturated && aliveCount < adaptCap { - adaptTarget = aliveCount + 1 - if adaptTarget > adaptCap { - adaptTarget = adaptCap - } - if promotableIdx >= 0 { - return writeAction(rotationActionWire{PromoteFromStandby: []int{promotableIdx}}) - } - return writeAction(rotationActionWire{AddLeg: true}) - } - // 5. SHRINK: sustained idle and above the floor. Park the oldest leg as - // a WARM STANDBY (rules kept alive, ready to promote) rather than tear - // it down — until the pool is full, then really drop it. Reset the idle - // counter so we step down one leg per two idle ticks rather than - // collapsing to the floor at once. - if adaptIdleCount >= 2 && aliveCount > adaptFloor { - adaptIdleCount = 0 - adaptTarget = aliveCount - 1 - if adaptTarget < adaptFloor { - adaptTarget = adaptFloor - } - if standbyCount < adaptStandbyMax { - return writeAction(rotationActionWire{DemoteToStandby: []int{oldestAliveIdx}}) - } - return writeAction(rotationActionWire{DropLegs: []int{oldestAliveIdx}}) - } - } - - // 6/7. EXPLORE (probe/exploit machine) or hold. Hand the active/standby - // partition down so a graduating probe parks (not tears down) the incumbent - // it displaces. - sets := legSets{ - activeCount: aliveCount, - standbyCount: standbyCount, - oldestActive: oldestAliveIdx, - promotable: promotableIdx, - } - return adaptExplore(sets) +func decideTrustTiered(ctx routingContextWire, cands []candidateWire) routeSpecWire { + return specToWire(preset.Decide("trust-tiered", ctxToPreset(ctx), candsToPreset(cands))) } -// adaptWorstOutlier reports the index and hops of the slowest alive leg when -// its smoothed latency is a >=1.5x-median outlier over the alive legs' -// smoothed latencies (needs >=2 smoothed samples), else ok=false. This is -// latency-adaptive's eviction test, run over the shared adaptLatEWMA. -func adaptWorstOutlier(legs []legInfoWire) (int, []string, bool) { - worstIdx := -1 - worstSm := -1.0 - var worstHops []string - var smoothed []float64 - for _, l := range legs { - if !l.Alive || l.Standby { - continue - } - sm, ok := adaptLatEWMA[l.TransportID] - if !ok { - continue - } - smoothed = append(smoothed, sm) - if sm > worstSm { - worstSm = sm - worstIdx = l.Index - worstHops = l.Hops - } - } - if len(smoothed) < 2 || worstIdx < 0 || worstSm <= 0 { - return -1, nil, false - } - sort.Float64s(smoothed) - median := medianSorted(smoothed) - if median > 0 && worstSm >= 1.5*median { - return worstIdx, worstHops, true - } - return -1, nil, false +func decideTimeOfDay(ctx routingContextWire) routeSpecWire { + return specToWire(preset.Decide("time-of-day", ctxToPreset(ctx), nil)) } -// adaptExplore is the probe-and-prune explore/exploit machine over the shared -// probe state. It adopts a pending probe, observes it adaptObserve ticks, then -// keeps it (dropping the worst established leg) only if it beats that worst — -// else prunes the probe. When no probe is in flight and the set is stable at -// target, it starts a fresh probe on the explore cadence. -func adaptExplore(sets legSets) uint64 { - aliveCount := sets.activeCount - // Adopt: a probe add requested last tick should now be present — the alive - // transport_id absent from the pre-add known set. Put it on probation. - if adaptProbePending { - adaptProbePending = false - for tid := range adaptAliveIdx { - if !adaptPrevKnown[tid] { - adaptProbeTID = tid - adaptProbeAge = 0 - break - } - } - } - - // Active probe: observe, then graduate-or-discard. - if adaptProbeTID != "" { - idx, alive := adaptAliveIdx[adaptProbeTID] - if !alive { - // Probe died on its own — abandon the experiment. - adaptProbeTID = "" - adaptProbeAge = 0 - return 0 - } - adaptProbeAge++ - if adaptProbeAge < adaptObserve { - return 0 - } - // Judge: probe's smoothed latency vs the worst established leg. - probeSm, okProbe := adaptLatEWMA[adaptProbeTID] - worstIdx := -1 - worstSm := -1.0 - for tid, i := range adaptAliveIdx { - if tid == adaptProbeTID { - continue - } - sm, ok := adaptLatEWMA[tid] - if !ok { - continue - } - if sm > worstSm { - worstSm = sm - worstIdx = i - } - } - if !okProbe || worstIdx < 0 { - // Not enough latency signal to judge yet — keep observing. - return 0 - } - adaptProbeTID = "" - if probeSm < worstSm { - // Probe graduates: PARK the worst established leg as a warm standby - // (no teardown — it re-promotes instantly if a leg later dies) - // rather than tearing it down; the probe stays and joins the - // established set next tick. Evict-by-demote, no capacity dip. - return shedActive(sets, worstIdx) - } - // Failed experiment: drop the SPECULATIVE probe leg (added fresh this - // cycle, so removing it just returns the group to target — no working - // leg lost), keeping the incumbents. - return writeAction(rotationActionWire{DropLegs: []int{idx}}) - } - - // Explore: on the cadence, when stable at the target width, snapshot the - // known set and request one speculative leg over a fresh path. - if adaptTick%adaptExploreEvery == 0 && aliveCount == adaptTarget { - for k := range adaptPrevKnown { - delete(adaptPrevKnown, k) - } - for tid := range adaptAliveIdx { - adaptPrevKnown[tid] = true - } - adaptProbePending = true - return writeAction(rotationActionWire{AddLeg: true}) +// distinctCount counts unique (case-folded) entries — used by main_test.go to +// assert transport-diverse picked the most-diverse route. +func distinctCount(xs []string) int { + seen := map[string]bool{} + for _, x := range xs { + seen[toLower(x)] = true } - // Hold. - return 0 + return len(seen) } -// writeAction marshals a rotation action and packs it for the host. -func writeAction(action rotationActionWire) uint64 { - out, err := json.Marshal(action) - if err != nil { - return 0 +// toLower is a tiny ASCII lowercaser so distinctCount stays dependency-free. +func toLower(s string) string { + b := []byte(s) + for i, c := range b { + if c >= 'A' && c <= 'Z' { + b[i] = c + ('a' - 'A') + } } - return writeOutput(out) + return string(b) } // main is required by the WASI target but isn't called by the host at diff --git a/pkg/router/policy/preset/names.go b/pkg/router/policy/preset/names.go new file mode 100644 index 0000000000..45fa6a3f6d --- /dev/null +++ b/pkg/router/policy/preset/names.go @@ -0,0 +1,36 @@ +// Package preset pkg/router/policy/preset/names.go c2-net-routing +// enumerates the built-in preset names Decide / Engine.OnTick +// dispatch on, so callers (the wasm-visor's config selection, the +// CLI, tests) can validate a requested preset without importing the +// wazero-side manifest. +package preset + +// Names returns the built-in preset names, in a stable order. Kept in sync +// with the switch in Decide (preset.go) and OnTick (tick.go) — the same set +// the wazero bundle's manifest.json lists. +func Names() []string { + return []string{ + "app-mux", + "rotating-bw", + "latency-adaptive", + "elastic-mux", + "probe-and-prune", + "adaptive", + "geo-avoid", + "transport-diverse", + "trust-tiered", + "time-of-day", + } +} + +// Has reports whether name is a built-in preset. Unknown names still Decide to +// the app-mux fallback (matching the bundle), but callers that want to reject +// typos up front use this. +func Has(name string) bool { + for _, n := range Names() { + if n == name { + return true + } + } + return false +} diff --git a/pkg/router/policy/preset/preset.go b/pkg/router/policy/preset/preset.go new file mode 100644 index 0000000000..b74722664b --- /dev/null +++ b/pkg/router/policy/preset/preset.go @@ -0,0 +1,468 @@ +// Package preset pkg/router/policy/preset/preset.go c2-net-routing +// is the SINGLE SOURCE OF TRUTH for the built-in routing-policy +// presets' decide/tick logic, as pure Go with no wasm ABI, no +// wazero, and no starlark. It carries the preset input/output +// types as plain Go structs and the per-preset decide functions +// plus the name dispatch (Decide) and the stateful tick engine +// (Engine.OnTick). +// +// Two callers compile this same logic in and both must agree +// byte-for-byte on every decision: +// +// - The TinyGo wasm bundle +// (docs/examples/routing-policies/wasm/bundle/main.go) is a +// thin JSON/pointer shim over this package; the compiled +// bundle.wasm is what the NATIVE visor runs via wazero. +// - The native-Go preset evaluator +// (pkg/router/policy/preset/evaluator.go, tinygo-tagged) calls +// this package directly so the WASM visor — which cannot host +// wazero (wasm-in-wasm) — runs the identical preset logic. +// +// The types here intentionally mirror the flat wire structs in +// pkg/router/policy/wasm/abi.go field-for-field, but as clean Go +// structs with no json tags and no dependency on the policy +// package (which pulls in starlark and cannot compile under +// TinyGo). Callers convert their own types to/from these. +package preset + +import "strings" + +// Candidate mirrors policy.Candidate / CandidateWire: a concrete +// route option the policy may pick or shape. +type Candidate struct { + Hops []string + HopsGeo []string + EstLatencyMs int + TransportKinds []string +} + +// Context mirrors policy.RoutingContext / RoutingContextWire: the +// per-dial context a decide function reads. NowUnixNano is a +// unix-nanosecond timestamp (not time.Time) so the wasm guest and +// the native path share one integer-arithmetic clock. +type Context struct { + App string + PeerPK string + Port uint16 + NowUnixNano int64 + CLIOverrides map[string]string + IsDirectDial bool + TransportKind string + ReverseCandidates []Candidate +} + +// LegInfo mirrors policy.LegInfo / LegInfoWire: a per-leg +// telemetry snapshot the tick engine reasons over. +type LegInfo struct { + Index int + Kind string + TransportID string + LatencyMs int + Alive bool + Standby bool + SentBytes uint64 + RecvBytes uint64 + Retransmits uint64 + Hops []string +} + +// Spec mirrors policy.RouteSpec / RouteSpecWire: the shape a +// decide function returns. A zero Spec means "use the visor +// default." +type Spec struct { + Chosen *Candidate + ReverseChosen *Candidate + Mux int + ForwardMux int + ReverseMux int + MinHops int + ForwardMinHops int + ReverseMinHops int + Fallback string + Distribution string + RotationIntervalSeconds int +} + +// RotationAction mirrors policy.RotationAction / RotationActionWire: +// the at-most-one structural change a tick returns. The zero value +// is "no-op this tick." +type RotationAction struct { + DropLegs []int + AddLeg bool + ExcludeHops []string + DemoteToStandby []int + PromoteFromStandby []int +} + +// Decide dispatches to the named preset's decide logic. The name +// is the preset selected by config ("preset:"); unknown or +// empty falls back to app-mux so a bare bundle still does something +// sensible — identical to the wasm bundle's default case. +func Decide(name string, ctx Context, cands []Candidate) Spec { + switch name { + case "rotating-bw": + return decideRotatingBW(ctx) + case "latency-adaptive": + return decideLatencyAdaptive(ctx) + case "elastic-mux": + return decideElasticMux(ctx) + case "probe-and-prune": + return decideProbeAndPrune(ctx) + case "adaptive": + return decideAdaptive(ctx) + case "app-mux": + return decideAppMux(ctx) + case "geo-avoid": + return decideGeoAvoid(ctx, cands) + case "transport-diverse": + return decideTransportDiverse(ctx, cands) + case "trust-tiered": + return decideTrustTiered(ctx, cands) + case "time-of-day": + return decideTimeOfDay(ctx) + default: + return decideAppMux(ctx) + } +} + +// decideAppMux is the verbatim app-mux preset logic: per-app static mux +// + min_hops; latency-sensitive apps stay single-route, bandwidth apps +// get parallel legs. +func decideAppMux(ctx Context) Spec { + switch ctx.App { + case "vpn-client": + return Spec{Mux: 4, MinHops: 2} + case "skychat": + // Chat is latency-sensitive — single route, lowest mux. + return Spec{Mux: 1} + default: + // Everything else: visor defaults (empty spec). + return Spec{} + } +} + +// targetMux is the mux size the rotating-bw policy aims to maintain. +// Kept in sync with the Mux value returned from decideRotatingBW so +// on_tick can reason about "are we at target?" +const targetMux = 4 + +// decideRotatingBW is the rotating-bw (privacy) preset: targetMux active legs +// over multi-hop with byte load spread EQUALLY across them (traffic-analysis +// resistance — no single relay sees a large fraction of the flow), plus one +// extra warm-standby leg so on_tick can rotate the active set every 90s +// WITHOUT a tear-and-rebuild dip (see tickRotatingBW). +func decideRotatingBW(ctx Context) Spec { + // rotating-bw is OPT-IN per app (proxy/vpn/skynet start --routing-policy, or + // visor app arg routing-policy). Apply it to WHATEVER app/session it is + // active for — do NOT gate on the built-in binary names. A named proxy + // session dials under its SESSION name (e.g. "g8"), not "skysocks-client", + // so the old `switch ctx.App { case "skysocks-client", ... }` silently made + // the policy a NO-OP for every custom-named session (empty spec → no mux, no + // min_hops, no rotation — the reason the rotation never fired live). Only + // skip genuinely latency-sensitive apps, where a multi-hop mux is the wrong + // shape. + switch ctx.App { + case "skychat", "skychat-client": + return Spec{Mux: 1} + } + // min_hops=2 already says "no direct transport" (direct is 0 intermediates); + // the visor treats min_hops>=2 as an implicit avoid_direct so the dial flows + // to the overlay path where rotation can act. Without it the mux/distribution + // is silently dropped by the direct-dial fast path. + return Spec{ + // targetMux active + 1 warm standby (see tickRotatingBW). + Mux: targetMux + 1, + MinHops: 2, + RotationIntervalSeconds: 90, + // Equal (round-robin) byte spread across the active legs — the privacy + // property. "auto"/latency-weighting pins bytes to the fastest leg, + // defeating the spread; the policy sets this to override the muxMode + // default. + Distribution: "round-robin", + } +} + +// decideLatencyAdaptive is the latency-adaptive preset's decide logic: a +// symmetric 4-wide multi-hop mux (plus one warm reserve) for bandwidth/proxy +// apps, re-evaluated every 30s so on_tick can evict the slowest leg; +// Distribution "auto" lets the host weight bytes toward the faster legs. +// Non-target apps get the empty spec (defaults). +func decideLatencyAdaptive(ctx Context) Spec { + switch ctx.App { + case "vpn-client", "skysocks-client", "skynet-client": + return Spec{ + Mux: 5, + MinHops: 2, + RotationIntervalSeconds: 30, + Distribution: "auto", + } + } + return Spec{} +} + +// decideElasticMux is the elastic-mux preset's decide logic. It starts +// deliberately MODEST — a 2-way mux over multi-hop — and lets the on_tick +// AIMD controller grow or shrink the leg count to match observed load. +func decideElasticMux(ctx Context) Spec { + switch ctx.App { + case "vpn-client", "skysocks-client", "skynet-client": + return Spec{ + Mux: 2, + MinHops: 2, + RotationIntervalSeconds: 20, + Distribution: "auto", + } + } + return Spec{} +} + +// decideProbeAndPrune is the probe-and-prune preset's decide logic. It holds a +// steady 3-way mux over multi-hop as the "established" set that on_tick +// continuously refines. +func decideProbeAndPrune(ctx Context) Spec { + switch ctx.App { + case "vpn-client", "skysocks-client", "skynet-client": + return Spec{ + Mux: 3, + MinHops: 2, + RotationIntervalSeconds: 30, + Distribution: "auto", + } + } + return Spec{} +} + +// decideAdaptive is the COMPOSITE "adaptive" preset's decide logic — the +// intended converged default. It returns a lean seed spec that tickAdaptive +// then steers along size/latency/explore at once. No MinHops here on purpose: +// min-hops is a privacy constraint the operator owns; leaving it 0 means +// "inherit" so adaptive optimizes within the operator's chosen floor. +func decideAdaptive(ctx Context) Spec { + switch ctx.App { + case "vpn-client", "skysocks-client", "skynet-client": + return Spec{ + Mux: adaptDecideMux, + RotationIntervalSeconds: 20, + Distribution: "auto", + } + } + return Spec{} +} + +// --- conditional presets: constrain WHICH path is chosen, by route metadata --- + +// splitSet parses a comma/space-separated override value into a lowercased +// membership set. Empty input yields an empty (never-nil) set. +func splitSet(s string) map[string]bool { + out := map[string]bool{} + for _, f := range strings.FieldsFunc(s, func(r rune) bool { return r == ',' || r == ' ' }) { + if f != "" { + out[strings.ToLower(f)] = true + } + } + return out +} + +// decideGeoAvoid picks the lowest-latency forward candidate whose hops transit +// NONE of the blocked countries in cli_overrides["avoid_geo"] (comma-separated +// ISO country codes). If no candidate is clean (or none are offered) it returns +// the empty spec, deferring to the router rather than forcing a violating path. +func decideGeoAvoid(ctx Context, cands []Candidate) Spec { + blocked := splitSet(ctx.CLIOverrides["avoid_geo"]) + if len(blocked) == 0 || len(cands) == 0 { + return Spec{} // nothing to enforce + } + var best *Candidate + for i := range cands { + if candidateTransitsBlockedGeo(cands[i], blocked) { + continue + } + if best == nil || cands[i].EstLatencyMs < best.EstLatencyMs { + c := cands[i] + best = &c + } + } + if best == nil { + return Spec{} // every candidate violates; let the router decide + } + return Spec{Chosen: best, MinHops: 2} +} + +// candidateTransitsBlockedGeo reports whether any of the candidate's per-hop +// countries is in the blocked set. +func candidateTransitsBlockedGeo(c Candidate, blocked map[string]bool) bool { + for _, g := range c.HopsGeo { + if blocked[strings.ToLower(g)] { + return true + } + } + return false +} + +// decideTransportDiverse picks the forward candidate whose hops span the MOST +// distinct transport types (ties broken by lower latency). Empty spec when no +// candidates are offered. +func decideTransportDiverse(_ Context, cands []Candidate) Spec { + var best *Candidate + bestDiversity := -1 + for i := range cands { + d := distinctCount(cands[i].TransportKinds) + switch { + case d > bestDiversity: + c := cands[i] + best, bestDiversity = &c, d + case d == bestDiversity && best != nil && cands[i].EstLatencyMs < best.EstLatencyMs: + c := cands[i] + best = &c + } + } + if best == nil { + return Spec{} + } + // A mux of 2 over the diverse path keeps a spare leg on a different carrier. + return Spec{Chosen: best, Mux: 2, MinHops: 2} +} + +// distinctCount counts unique (case-folded) entries in a slice. +func distinctCount(xs []string) int { + seen := map[string]bool{} + for _, x := range xs { + seen[strings.ToLower(x)] = true + } + return len(seen) +} + +// decideTrustTiered prefers routes that transit ONLY trusted intermediaries +// (cli_overrides["trusted_pks"], comma-separated hop PKs). Tiered: a +// fully-trusted candidate wins outright (lowest latency among them); else it +// falls back to the candidate with the MOST trusted hops. +func decideTrustTiered(ctx Context, cands []Candidate) Spec { + trusted := splitSet(ctx.CLIOverrides["trusted_pks"]) + if len(trusted) == 0 || len(cands) == 0 { + return Spec{} + } + var bestFull, bestPartial *Candidate + bestPartialScore := -1 + for i := range cands { + n := trustedHopCount(cands[i], trusted) + if len(cands[i].Hops) > 0 && n == len(cands[i].Hops) { + if bestFull == nil || cands[i].EstLatencyMs < bestFull.EstLatencyMs { + c := cands[i] + bestFull = &c + } + } + if n > bestPartialScore { + c := cands[i] + bestPartial, bestPartialScore = &c, n + } + } + if bestFull != nil { + return Spec{Chosen: bestFull, MinHops: 2} + } + if bestPartial != nil { + return Spec{Chosen: bestPartial, MinHops: 2} + } + return Spec{} +} + +// trustedHopCount counts how many of a candidate's hops are in the trusted set. +func trustedHopCount(c Candidate, trusted map[string]bool) int { + n := 0 + for _, h := range c.Hops { + if trusted[strings.ToLower(h)] { + n++ + } + } + return n +} + +// decideTimeOfDay switches the route SHAPE by wall-clock hour (UTC), derived +// from ctx.NowUnixNano. During the configured business-hours window +// (cli_overrides["business_hours"] = "START-END", default "9-17") it returns a +// lean single-route shape; outside it, a wide privacy mux with byte-spread and +// rotation. +func decideTimeOfDay(ctx Context) Spec { + startH, endH := parseHourRange(ctx.CLIOverrides["business_hours"], 9, 17) + if inHourRange(hourOfDayUTC(ctx.NowUnixNano), startH, endH) { + return Spec{Mux: 1} // business hours: lean + low-latency + } + // off-hours: privacy-wide mux (mirrors rotating-bw's shape) + return Spec{Mux: 4, MinHops: 2, Distribution: "round-robin", RotationIntervalSeconds: 90} +} + +// hourOfDayUTC returns the UTC hour (0-23) of a unix-nanosecond timestamp using +// only integer arithmetic (avoids a "time" import in the wasm guest). +func hourOfDayUTC(unixNano int64) int { + if unixNano <= 0 { + return 0 + } + secOfDay := (unixNano / 1e9) % 86400 + return int(secOfDay / 3600) +} + +// parseHourRange parses "START-END" (24h) from an override, falling back to the +// given defaults on empty/malformed input. +func parseHourRange(s string, defStart, defEnd int) (int, int) { + a, b, ok := strings.Cut(strings.TrimSpace(s), "-") + if !ok { + return defStart, defEnd + } + start, okA := atoiHour(a) + end, okB := atoiHour(b) + if !okA || !okB { + return defStart, defEnd + } + return start, end +} + +// atoiHour parses a 0-23 hour string; returns ok=false otherwise. +func atoiHour(s string) (int, bool) { + s = strings.TrimSpace(s) + if s == "" { + return 0, false + } + n := 0 + for _, r := range s { + if r < '0' || r > '9' { + return 0, false + } + n = n*10 + int(r-'0') + } + if n < 0 || n > 23 { + return 0, false + } + return n, true +} + +// inHourRange reports whether hour is within [start, end), handling a window +// that wraps past midnight (start > end, e.g. 22-6). +func inHourRange(hour, start, end int) bool { + if start <= end { + return hour >= start && hour < end + } + return hour >= start || hour < end // wraps midnight +} + +// medianSorted returns the median of an already-sorted float slice (mean of +// the two middle elements for an even count). +func medianSorted(s []float64) float64 { + n := len(s) + if n == 0 { + return 0 + } + if n%2 == 1 { + return s[n/2] + } + return (s[n/2-1] + s[n/2]) / 2 +} + +// reliableMuxKind reports whether a transport type is reliable enough to ANCHOR +// a mux active set. stcpr/sudph/squicr/stcp are direct, well-behaved for +// sustained multiplexed throughput; webrtc/ws/wt/dmsg are not. +func reliableMuxKind(kind string) bool { + switch kind { + case "stcpr", "sudph", "squicr", "stcp": + return true + } + return false +} diff --git a/pkg/router/policy/preset/preset_test.go b/pkg/router/policy/preset/preset_test.go new file mode 100644 index 0000000000..f46e53f73b --- /dev/null +++ b/pkg/router/policy/preset/preset_test.go @@ -0,0 +1,89 @@ +package preset + +import ( + "reflect" + "testing" +) + +func TestDecide_ShapePresets(t *testing.T) { + cases := []struct { + name string + ctx Context + want Spec + }{ + {"app-mux/vpn", Context{App: "vpn-client"}, Spec{Mux: 4, MinHops: 2}}, + {"app-mux/skychat", Context{App: "skychat"}, Spec{Mux: 1}}, + {"app-mux/other", Context{App: "x"}, Spec{}}, + {"rotating-bw", Context{App: "skysocks-client"}, Spec{Mux: 5, MinHops: 2, RotationIntervalSeconds: 90, Distribution: "round-robin"}}, + {"rotating-bw/chat", Context{App: "skychat"}, Spec{Mux: 1}}, + {"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: 1, RotationIntervalSeconds: 20, Distribution: "auto"}}, + } + for _, tc := range cases { + presetName, _, _ := splitName(tc.name) + got := Decide(presetName, tc.ctx, nil) + if !reflect.DeepEqual(got, tc.want) { + t.Errorf("%s: Decide=%+v want %+v", tc.name, got, tc.want) + } + } +} + +// splitName maps a "preset/variant" test label to the preset name. +func splitName(label string) (string, string, bool) { + for i := 0; i < len(label); i++ { + if label[i] == '/' { + return label[:i], label[i+1:], true + } + } + return label, "", false +} + +func TestDecide_GeoAvoid(t *testing.T) { + cands := []Candidate{ + {Hops: []string{"a"}, HopsGeo: []string{"US"}, EstLatencyMs: 10}, + {Hops: []string{"b"}, HopsGeo: []string{"DE"}, EstLatencyMs: 50}, + } + got := Decide("geo-avoid", Context{CLIOverrides: map[string]string{"avoid_geo": "US"}}, cands) + if got.Chosen == nil || got.Chosen.HopsGeo[0] != "DE" { + t.Fatalf("geo-avoid should pick the clean DE route; got %+v", got.Chosen) + } + // No clean candidate → defer. + got = Decide("geo-avoid", Context{CLIOverrides: map[string]string{"avoid_geo": "US,DE"}}, cands) + if got.Chosen != nil { + t.Fatalf("geo-avoid should defer when all violate; got %+v", got.Chosen) + } +} + +func TestDecide_TimeOfDay(t *testing.T) { + const h = int64(3600) * 1_000_000_000 + if got := Decide("time-of-day", Context{NowUnixNano: 11 * h}, nil); got.Mux != 1 { + t.Errorf("business hours should be lean (mux 1); got %+v", got) + } + off := Decide("time-of-day", Context{NowUnixNano: 3 * h}, nil) + if off.Mux != 4 || off.Distribution != "round-robin" || off.RotationIntervalSeconds == 0 { + t.Errorf("off-hours should be a wide rotating mux; got %+v", off) + } +} + +func TestEngine_OnTick_UnknownIsNoop(t *testing.T) { + e := New() + if got := e.OnTick("app-mux", []LegInfo{{Index: 0, Alive: true}}); !reflect.DeepEqual(got, RotationAction{}) { + t.Errorf("app-mux has no tick logic; want no-op, got %+v", got) + } +} + +func TestEngine_OnTick_RotatingBWParksFragile(t *testing.T) { + e := New() + // A reliable active leg + a fragile (webrtc) active leg → park the fragile. + legs := []LegInfo{ + {Index: 0, TransportID: "t0", Kind: "stcpr", Alive: true}, + {Index: 1, TransportID: "t1", Kind: "webrtc", Alive: true}, + } + got := e.OnTick("rotating-bw", legs) + want := RotationAction{DemoteToStandby: []int{1}} + if !reflect.DeepEqual(got, want) { + t.Errorf("rotating-bw should park the fragile active leg; got %+v want %+v", got, want) + } +} diff --git a/pkg/router/policy/preset/tick.go b/pkg/router/policy/preset/tick.go new file mode 100644 index 0000000000..77dbdac3e1 --- /dev/null +++ b/pkg/router/policy/preset/tick.go @@ -0,0 +1,732 @@ +// Package preset pkg/router/policy/preset/tick.go c2-net-routing +// carries the stateful on_tick controllers for the adaptive +// presets. In the wasm bundle these controllers keep their state +// in package globals (one module instance per policy load); here +// the same state lives on an Engine value so the native path gets +// one independent controller per evaluator instance. A single +// package-global Engine in the bundle shim reproduces the bundle's +// original global-state behavior exactly. +package preset + +import "sort" + +// Engine holds the per-transport_id smoothing state and probe/AIMD +// bookkeeping the adaptive tick controllers accumulate across +// ticks. Construct with New; each Engine is independent, so one +// route group's controller state never smears onto another's. +// +// Engine is NOT safe for concurrent use — the router serializes +// on_tick per route group, mirroring the single-threaded wazero +// module the bundle compiles to. +type Engine struct { + // latency-adaptive + latAdaptEWMA map[string]float64 + latAdaptSeen map[string]bool + + // elastic-mux + elasticPrevRecv map[string]uint64 + elasticSeen map[string]bool + elasticThroughputEWMA float64 + elasticPeak float64 + elasticIdleCount int + elasticSeeded bool + + // probe-and-prune + probeEWMA map[string]float64 + probeSeen map[string]bool + probeAliveIdx map[string]int + prevKnownTIDs map[string]bool + probeTID string + probeAge int + probePending bool + + // adaptive (composite) + adaptLatEWMA map[string]float64 + adaptPrevRecv map[string]uint64 + adaptSeen map[string]bool + adaptAliveIdx map[string]int + adaptPrevKnown map[string]bool + adaptThroughputEWMA float64 + adaptPeak float64 + adaptSeeded bool + adaptIdleCount int + adaptTick int + adaptTarget int + adaptProbeTID string + adaptProbeAge int + adaptProbePending bool +} + +// New returns an Engine with initialized state maps and the +// adaptive size target seeded to the decide mux, matching the +// bundle's package-global initializers. +func New() *Engine { + return &Engine{ + latAdaptEWMA: map[string]float64{}, + latAdaptSeen: map[string]bool{}, + elasticPrevRecv: map[string]uint64{}, + elasticSeen: map[string]bool{}, + probeEWMA: map[string]float64{}, + probeSeen: map[string]bool{}, + probeAliveIdx: map[string]int{}, + prevKnownTIDs: map[string]bool{}, + adaptLatEWMA: map[string]float64{}, + adaptPrevRecv: map[string]uint64{}, + adaptSeen: map[string]bool{}, + adaptAliveIdx: map[string]int{}, + adaptPrevKnown: map[string]bool{}, + adaptTarget: adaptDecideMux, + } +} + +// OnTick dispatches to the named preset's tick controller. Presets +// without tick logic (app-mux, the conditional presets, unknown) +// return the zero RotationAction (no-op) — identical to the +// bundle's on_tick default case. +func (e *Engine) OnTick(name string, legs []LegInfo) RotationAction { + switch name { + case "rotating-bw": + return e.tickRotatingBW(legs) + case "latency-adaptive": + return e.tickLatencyAdaptive(legs) + case "elastic-mux": + return e.tickElasticMux(legs) + case "probe-and-prune": + return e.tickProbeAndPrune(legs) + case "adaptive": + return e.tickAdaptive(legs) + default: + return RotationAction{} + } +} + +// --- shared no-dip controller helpers --- + +// nodipStandbyMax caps the warm-standby pool a shed/park may hold before it +// falls back to a real teardown. +const nodipStandbyMax = 2 + +// legSets is the per-tick active/standby partition of a route group's legs. +type legSets struct { + activeCount int + standbyCount int + oldestActive int + promotable int +} + +// classifyLegs partitions the leg snapshot into active/standby counts and the +// two indices the controllers act on. +func classifyLegs(legs []LegInfo) legSets { + s := legSets{oldestActive: -1, promotable: -1} + for _, l := range legs { + if !l.Alive { + continue + } + if l.Standby { + s.standbyCount++ + if s.promotable == -1 || l.Index < s.promotable { + s.promotable = l.Index + } + continue + } + s.activeCount++ + if s.oldestActive == -1 || l.Index < s.oldestActive { + s.oldestActive = l.Index + } + } + return s +} + +// growActive raises the active width by one with no setup dip when a warm +// spare is parked; otherwise requests a genuinely fresh leg. +func growActive(s legSets) RotationAction { + if s.promotable >= 0 { + return RotationAction{PromoteFromStandby: []int{s.promotable}} + } + return RotationAction{AddLeg: true} +} + +// swapActive evicts leg idx and fills its slot from the warm reserve in the +// same tick (hot-swap); parks idx alone when no spare is available. +func swapActive(s legSets, idx int) RotationAction { + if s.promotable >= 0 { + return RotationAction{ + PromoteFromStandby: []int{s.promotable}, + DemoteToStandby: []int{idx}, + } + } + return RotationAction{DemoteToStandby: []int{idx}} +} + +// shedActive removes leg idx from the active set with no capacity dip while the +// pool has room, else tears it down. +func shedActive(s legSets, idx int) RotationAction { + if s.standbyCount < nodipStandbyMax { + return RotationAction{DemoteToStandby: []int{idx}} + } + return RotationAction{DropLegs: []int{idx}} +} + +// --- rotating-bw --- + +func (e *Engine) tickRotatingBW(legs []LegInfo) RotationAction { + var relAct, relSb, fragAct, fragSb []int + for _, l := range legs { + if !l.Alive { + continue + } + rel := reliableMuxKind(l.Kind) + switch { + case l.Standby && rel: + relSb = append(relSb, l.Index) + case l.Standby: + fragSb = append(fragSb, l.Index) + case rel: + relAct = append(relAct, l.Index) + default: + fragAct = append(fragAct, l.Index) + } + } + active := len(relAct) + len(fragAct) + alive := active + len(relSb) + len(fragSb) + if alive == 0 { + return RotationAction{} + } + lo := func(s []int) int { + m := s[0] + for _, v := range s { + if v < m { + m = v + } + } + return m + } + hi := func(s []int) int { + m := s[0] + for _, v := range s { + if v > m { + m = v + } + } + return m + } + + switch { + case len(relAct) >= 1 && len(fragAct) > 0: + return RotationAction{DemoteToStandby: []int{hi(fragAct)}} + case len(fragAct) > 0 && len(relSb) > 0: + return RotationAction{PromoteFromStandby: []int{lo(relSb)}, DemoteToStandby: []int{hi(fragAct)}} + case len(relAct) < targetMux && len(relSb) > 0: + return RotationAction{PromoteFromStandby: []int{lo(relSb)}} + case len(relAct) == 0 && len(fragAct) == 0 && len(fragSb) > 0: + return RotationAction{PromoteFromStandby: []int{lo(fragSb)}} + case len(relAct) > targetMux: + return RotationAction{DemoteToStandby: []int{hi(relAct)}} + case len(fragAct) == 0 && len(relSb) > 0 && len(relAct) > 0: + return RotationAction{PromoteFromStandby: []int{lo(relSb)}, DemoteToStandby: []int{lo(relAct)}} + default: + return RotationAction{} + } +} + +// --- latency-adaptive --- + +const latAdaptAlpha = 0.3 + +func (e *Engine) tickLatencyAdaptive(legs []LegInfo) RotationAction { + const targetMux = 4 + + for k := range e.latAdaptSeen { + delete(e.latAdaptSeen, k) + } + for _, l := range legs { + tid := l.TransportID + if tid == "" { + continue + } + e.latAdaptSeen[tid] = true + if l.Alive && l.LatencyMs > 0 { + sample := float64(l.LatencyMs) + if prev, ok := e.latAdaptEWMA[tid]; ok { + e.latAdaptEWMA[tid] = latAdaptAlpha*sample + (1-latAdaptAlpha)*prev + } else { + e.latAdaptEWMA[tid] = sample + } + } + } + for tid := range e.latAdaptEWMA { + if !e.latAdaptSeen[tid] { + delete(e.latAdaptEWMA, tid) + } + } + + sets := classifyLegs(legs) + worstIdx := -1 + worstSmoothed := -1.0 + var smoothed []float64 + for _, l := range legs { + if !l.Alive || l.Standby { + continue + } + sm, ok := e.latAdaptEWMA[l.TransportID] + if !ok { + continue + } + smoothed = append(smoothed, sm) + if sm > worstSmoothed { + worstSmoothed = sm + worstIdx = l.Index + } + } + if sets.activeCount < targetMux { + return growActive(sets) + } + if sets.activeCount > targetMux { + return shedActive(sets, sets.oldestActive) + } + if len(smoothed) < 2 || worstIdx < 0 || worstSmoothed <= 0 { + return RotationAction{} + } + + sort.Float64s(smoothed) + median := medianSorted(smoothed) + + if median > 0 && worstSmoothed >= 1.5*median { + return swapActive(sets, worstIdx) + } + return RotationAction{} +} + +// --- elastic-mux --- + +const ( + elasticFloor = 2 + elasticCap = 6 + elasticAlpha = 0.3 + elasticPeakDecay = 0.98 +) + +func (e *Engine) tickElasticMux(legs []LegInfo) RotationAction { + for k := range e.elasticSeen { + delete(e.elasticSeen, k) + } + var rawTotal float64 + sets := classifyLegs(legs) + for _, l := range legs { + if !l.Alive { + continue + } + tid := l.TransportID + if tid == "" { + continue + } + e.elasticSeen[tid] = true + if prev, ok := e.elasticPrevRecv[tid]; ok && l.RecvBytes >= prev && !l.Standby { + rawTotal += float64(l.RecvBytes - prev) + } + e.elasticPrevRecv[tid] = l.RecvBytes + } + for tid := range e.elasticPrevRecv { + if !e.elasticSeen[tid] { + delete(e.elasticPrevRecv, tid) + } + } + if sets.activeCount == 0 { + if sets.promotable >= 0 { + return growActive(sets) + } + return RotationAction{} + } + + if !e.elasticSeeded { + if rawTotal > 0 { + e.elasticThroughputEWMA = rawTotal + e.elasticPeak = rawTotal + e.elasticSeeded = true + } + return RotationAction{} + } + e.elasticThroughputEWMA = elasticAlpha*rawTotal + (1-elasticAlpha)*e.elasticThroughputEWMA + e.elasticPeak *= elasticPeakDecay + if e.elasticThroughputEWMA > e.elasticPeak { + e.elasticPeak = e.elasticThroughputEWMA + } + if e.elasticPeak <= 0 { + return RotationAction{} + } + + smoothed := e.elasticThroughputEWMA + if smoothed < 0.25*e.elasticPeak { + e.elasticIdleCount++ + } else { + e.elasticIdleCount = 0 + } + + if smoothed >= 0.80*e.elasticPeak && sets.activeCount < elasticCap { + return growActive(sets) + } + if e.elasticIdleCount >= 2 && sets.activeCount > elasticFloor { + e.elasticIdleCount = 0 + return shedActive(sets, sets.oldestActive) + } + return RotationAction{} +} + +// --- probe-and-prune --- + +const ( + probeTarget = 3 + probeObserveTicks = 3 + probeAlpha = 0.3 +) + +func (e *Engine) tickProbeAndPrune(legs []LegInfo) RotationAction { + for k := range e.probeSeen { + delete(e.probeSeen, k) + } + for k := range e.probeAliveIdx { + delete(e.probeAliveIdx, k) + } + sets := classifyLegs(legs) + activeCount := 0 + for _, l := range legs { + tid := l.TransportID + if tid == "" { + continue + } + e.probeSeen[tid] = true + if !l.Alive { + continue + } + if l.LatencyMs > 0 { + sample := float64(l.LatencyMs) + if prev, ok := e.probeEWMA[tid]; ok { + e.probeEWMA[tid] = probeAlpha*sample + (1-probeAlpha)*prev + } else { + e.probeEWMA[tid] = sample + } + } + if l.Standby { + continue + } + activeCount++ + e.probeAliveIdx[tid] = l.Index + } + for tid := range e.probeEWMA { + if !e.probeSeen[tid] { + delete(e.probeEWMA, tid) + } + } + + if e.probePending { + e.probePending = false + for tid := range e.probeAliveIdx { + if !e.prevKnownTIDs[tid] { + e.probeTID = tid + e.probeAge = 0 + break + } + } + } + + if e.probeTID != "" { + idx, alive := e.probeAliveIdx[e.probeTID] + if !alive { + e.probeTID = "" + e.probeAge = 0 + return RotationAction{} + } + e.probeAge++ + if e.probeAge < probeObserveTicks { + return RotationAction{} + } + probeSm, okProbe := e.probeEWMA[e.probeTID] + worstIdx := -1 + worstSm := -1.0 + for tid, i := range e.probeAliveIdx { + if tid == e.probeTID { + continue + } + sm, ok := e.probeEWMA[tid] + if !ok { + continue + } + if sm > worstSm { + worstSm = sm + worstIdx = i + } + } + if !okProbe || worstIdx < 0 { + return RotationAction{} + } + e.probeTID = "" + if probeSm < worstSm { + return shedActive(sets, worstIdx) + } + return RotationAction{DropLegs: []int{idx}} + } + + if !e.probePending && activeCount == probeTarget { + for k := range e.prevKnownTIDs { + delete(e.prevKnownTIDs, k) + } + for tid := range e.probeAliveIdx { + e.prevKnownTIDs[tid] = true + } + e.probePending = true + return RotationAction{AddLeg: true} + } + return RotationAction{} +} + +// --- adaptive (composite) --- + +const ( + adaptDecideMux = 1 + adaptFloor = 1 + adaptCap = 6 + adaptAlpha = 0.3 + adaptPeakDecay = 0.98 + adaptExploreEvery = 6 + adaptObserve = 3 + adaptStandbyMax = 2 +) + +func (e *Engine) tickAdaptive(legs []LegInfo) RotationAction { + e.adaptTick++ + + for k := range e.adaptSeen { + delete(e.adaptSeen, k) + } + for k := range e.adaptAliveIdx { + delete(e.adaptAliveIdx, k) + } + var rawTotal float64 + aliveCount := 0 + standbyCount := 0 + oldestAliveIdx := -1 + promotableIdx := -1 + for _, l := range legs { + tid := l.TransportID + if tid != "" { + e.adaptSeen[tid] = true + } + if !l.Alive { + continue + } + if l.Standby { + standbyCount++ + if promotableIdx == -1 || l.Index < promotableIdx { + promotableIdx = l.Index + } + continue + } + aliveCount++ + if oldestAliveIdx == -1 || l.Index < oldestAliveIdx { + oldestAliveIdx = l.Index + } + if tid == "" { + continue + } + e.adaptAliveIdx[tid] = l.Index + if l.LatencyMs > 0 { + sample := float64(l.LatencyMs) + if prev, ok := e.adaptLatEWMA[tid]; ok { + e.adaptLatEWMA[tid] = adaptAlpha*sample + (1-adaptAlpha)*prev + } else { + e.adaptLatEWMA[tid] = sample + } + } + if prev, ok := e.adaptPrevRecv[tid]; ok && l.RecvBytes >= prev { + rawTotal += float64(l.RecvBytes - prev) + } + e.adaptPrevRecv[tid] = l.RecvBytes + } + for tid := range e.adaptLatEWMA { + if !e.adaptSeen[tid] { + delete(e.adaptLatEWMA, tid) + } + } + for tid := range e.adaptPrevRecv { + if !e.adaptSeen[tid] { + delete(e.adaptPrevRecv, tid) + } + } + + saturated, idle := false, false + if !e.adaptSeeded { + if rawTotal > 0 { + e.adaptThroughputEWMA = rawTotal + e.adaptPeak = rawTotal + e.adaptSeeded = true + } + } else { + e.adaptThroughputEWMA = adaptAlpha*rawTotal + (1-adaptAlpha)*e.adaptThroughputEWMA + e.adaptPeak *= adaptPeakDecay + if e.adaptThroughputEWMA > e.adaptPeak { + e.adaptPeak = e.adaptThroughputEWMA + } + if e.adaptPeak > 0 { + saturated = e.adaptThroughputEWMA >= 0.80*e.adaptPeak + idle = e.adaptThroughputEWMA < 0.25*e.adaptPeak + } + } + if idle { + e.adaptIdleCount++ + } else { + e.adaptIdleCount = 0 + } + + if aliveCount == 0 { + if promotableIdx >= 0 { + return RotationAction{PromoteFromStandby: []int{promotableIdx}} + } + return RotationAction{} + } + if aliveCount < adaptFloor { + if promotableIdx >= 0 { + return RotationAction{PromoteFromStandby: []int{promotableIdx}} + } + return RotationAction{AddLeg: true} + } + + probeInFlight := e.adaptProbePending || e.adaptProbeTID != "" + if !probeInFlight { + if idx, hops, ok := e.adaptWorstOutlier(legs); ok { + if promotableIdx >= 0 { + return RotationAction{ + PromoteFromStandby: []int{promotableIdx}, + DemoteToStandby: []int{idx}, + } + } + return RotationAction{ + DropLegs: []int{idx}, + AddLeg: true, + ExcludeHops: hops, + } + } + if saturated && aliveCount < adaptCap { + e.adaptTarget = aliveCount + 1 + if e.adaptTarget > adaptCap { + e.adaptTarget = adaptCap + } + if promotableIdx >= 0 { + return RotationAction{PromoteFromStandby: []int{promotableIdx}} + } + return RotationAction{AddLeg: true} + } + if e.adaptIdleCount >= 2 && aliveCount > adaptFloor { + e.adaptIdleCount = 0 + e.adaptTarget = aliveCount - 1 + if e.adaptTarget < adaptFloor { + e.adaptTarget = adaptFloor + } + if standbyCount < adaptStandbyMax { + return RotationAction{DemoteToStandby: []int{oldestAliveIdx}} + } + return RotationAction{DropLegs: []int{oldestAliveIdx}} + } + } + + sets := legSets{ + activeCount: aliveCount, + standbyCount: standbyCount, + oldestActive: oldestAliveIdx, + promotable: promotableIdx, + } + return e.adaptExplore(sets) +} + +func (e *Engine) adaptWorstOutlier(legs []LegInfo) (int, []string, bool) { + worstIdx := -1 + worstSm := -1.0 + var worstHops []string + var smoothed []float64 + for _, l := range legs { + if !l.Alive || l.Standby { + continue + } + sm, ok := e.adaptLatEWMA[l.TransportID] + if !ok { + continue + } + smoothed = append(smoothed, sm) + if sm > worstSm { + worstSm = sm + worstIdx = l.Index + worstHops = l.Hops + } + } + if len(smoothed) < 2 || worstIdx < 0 || worstSm <= 0 { + return -1, nil, false + } + sort.Float64s(smoothed) + median := medianSorted(smoothed) + if median > 0 && worstSm >= 1.5*median { + return worstIdx, worstHops, true + } + return -1, nil, false +} + +func (e *Engine) adaptExplore(sets legSets) RotationAction { + aliveCount := sets.activeCount + if e.adaptProbePending { + e.adaptProbePending = false + for tid := range e.adaptAliveIdx { + if !e.adaptPrevKnown[tid] { + e.adaptProbeTID = tid + e.adaptProbeAge = 0 + break + } + } + } + + if e.adaptProbeTID != "" { + idx, alive := e.adaptAliveIdx[e.adaptProbeTID] + if !alive { + e.adaptProbeTID = "" + e.adaptProbeAge = 0 + return RotationAction{} + } + e.adaptProbeAge++ + if e.adaptProbeAge < adaptObserve { + return RotationAction{} + } + probeSm, okProbe := e.adaptLatEWMA[e.adaptProbeTID] + worstIdx := -1 + worstSm := -1.0 + for tid, i := range e.adaptAliveIdx { + if tid == e.adaptProbeTID { + continue + } + sm, ok := e.adaptLatEWMA[tid] + if !ok { + continue + } + if sm > worstSm { + worstSm = sm + worstIdx = i + } + } + if !okProbe || worstIdx < 0 { + return RotationAction{} + } + e.adaptProbeTID = "" + if probeSm < worstSm { + return shedActive(sets, worstIdx) + } + return RotationAction{DropLegs: []int{idx}} + } + + if e.adaptTick%adaptExploreEvery == 0 && aliveCount == e.adaptTarget { + for k := range e.adaptPrevKnown { + delete(e.adaptPrevKnown, k) + } + for tid := range e.adaptAliveIdx { + e.adaptPrevKnown[tid] = true + } + e.adaptProbePending = true + return RotationAction{AddLeg: true} + } + return RotationAction{} +} diff --git a/pkg/router/policy/presethook/presethook.go b/pkg/router/policy/presethook/presethook.go new file mode 100644 index 0000000000..f83d24f9dc --- /dev/null +++ b/pkg/router/policy/presethook/presethook.go @@ -0,0 +1,240 @@ +// Package presethook pkg/router/policy/presethook/presethook.go c2-net-routing +// adapts a built-in routing-policy preset (pkg/router/policy/preset) to the +// router.DialHook / RouteSelectingHook / RotationHook interfaces WITHOUT the +// starlark evaluator or the wazero runtime. +// +// The native visor runs presets by loading the embedded bundle.wasm and +// evaluating it in wazero (pkg/router/policy/wasm/presets, wired through +// pkg/visor/policy_loader.go + pkg/router/policy.Hook). That path pulls in +// starlark and a host wasm runtime, neither of which compiles for the browser / +// TinyGo wasm-visor — and a wasm module cannot host wazero (wasm-in-wasm) +// anyway. This package is the wasm-visor's equivalent: it calls the SAME preset +// decide/tick logic (compiled in as native Go, the single source of truth) and +// projects the result onto the same router hook interfaces the policy.Hook +// implements, so the router integration point is identical. +// +// It imports only pkg/router (for the hook interface types, already compiled +// into the wasm-visor) and pkg/router/policy/preset (stdlib-only), so it is +// TinyGo-safe. The native visor keeps using the wazero path; this is selected +// on the wasm side (see cmd/wasm-visor). +package presethook + +import ( + "context" + + "github.com/skycoin/skywire/pkg/router" + "github.com/skycoin/skywire/pkg/router/policy/preset" +) + +// Provider supplies per-hop metadata the conditional presets read during route +// selection: the intermediary's country (geo-avoid) and transport kind +// (transport-diverse). The wasm-visor may pass nil (→ NopProvider), in which +// case geo/kind are unknown and those two presets defer rather than constrain; +// trust-tiered needs only the hop PKs and works regardless. Mirrors the +// enrichment pkg/router/policy.Hook does via its policy.Provider on native. +type Provider interface { + Geo(pk string) string // ISO country code, or "??" if unknown + Kind(pk string) string // transport kind ("stcpr"/…), or "" if unknown +} + +// NopProvider returns geo "??" and empty kind for every PK — the safe default +// when the wasm-visor has no metadata source wired. +func NopProvider() Provider { return nopProvider{} } + +type nopProvider struct{} + +func (nopProvider) Geo(string) string { return "??" } +func (nopProvider) Kind(string) string { return "" } + +// Hook adapts one preset to the router hook interfaces. Construct with New. It +// carries a preset.Engine so the stateful tick controllers (rotating-bw, +// latency-adaptive, elastic-mux, probe-and-prune, adaptive) accumulate their +// per-transport_id state across ticks exactly as the wazero module does. +// +// Not safe for concurrent Decide/OnTick across route groups sharing one Hook — +// construct one Hook per policy scope, as the router does per route group. +type Hook struct { + name string + engine *preset.Engine + provider Provider +} + +// New returns a Hook bound to the named preset. An unknown name still decides +// via the preset package's app-mux fallback (matching the bundle); callers that +// want to reject typos check preset.Has first. provider may be nil. +func New(name string, provider Provider) *Hook { + if provider == nil { + provider = NopProvider() + } + return &Hook{name: name, engine: preset.New(), provider: provider} +} + +// Name returns the preset this hook runs. +func (h *Hook) Name() string { return h.name } + +// BeforeDial implements router.DialHook: it maps the preset's decide-time route +// SHAPE (mux / min_hops / distribution / rotation) onto a DialAdjustment. The +// candidate-picking presets (geo-avoid etc.) return no shape here and act in +// SelectRoute instead. +func (h *Hook) BeforeDial(_ context.Context, info router.DialInfo) (router.DialAdjustment, error) { + spec := preset.Decide(h.name, dialInfoToCtx(info), nil) + // MinHops >= 2 (any direction) implies "no direct" — the direct path is a + // single 0-intermediate hop — so surface AvoidDirect, matching policy.Hook. + avoidDirect := spec.MinHops >= 2 || spec.ForwardMinHops >= 2 || spec.ReverseMinHops >= 2 + return router.DialAdjustment{ + MuxRoutes: spec.Mux, + ForwardMuxRoutes: spec.ForwardMux, + ReverseMuxRoutes: spec.ReverseMux, + MinHops: spec.MinHops, + ForwardMinHops: spec.ForwardMinHops, + ReverseMinHops: spec.ReverseMinHops, + Fallback: spec.Fallback, + RotationIntervalSeconds: spec.RotationIntervalSeconds, + AvoidDirect: avoidDirect, + Distribution: distributionFor(spec.Distribution), + }, nil +} + +// SelectRoute implements router.RouteSelectingHook: it enriches the router's +// bare candidates with per-hop geo/kind (via the Provider), runs the preset's +// decide over them, and translates the chosen candidate back to an index by hop +// equality — the same protocol policy.Hook uses. Presets that return no Chosen +// defer (index -1) to the router's built-in pick. +func (h *Hook) SelectRoute(_ context.Context, info router.DialInfo, forward, reverse []router.CandidateInfo) (router.RouteSelection, error) { + if len(forward) == 0 { + return router.RouteSelection{Chosen: -1, ReverseChosen: -1}, nil + } + fwd := h.enrich(forward) + rev := h.enrich(reverse) + ctx := dialInfoToCtx(info) + ctx.ReverseCandidates = rev + spec := preset.Decide(h.name, ctx, fwd) + if spec.Fallback == "drop" { + return router.RouteSelection{Drop: true, Chosen: -1, ReverseChosen: -1}, nil + } + sel := router.RouteSelection{Chosen: -1, ReverseChosen: -1, Distribution: distributionFor(spec.Distribution)} + if spec.Chosen != nil { + sel.Chosen = matchCandidate(fwd, *spec.Chosen) + } + if spec.ReverseChosen != nil { + sel.ReverseChosen = matchCandidate(rev, *spec.ReverseChosen) + } + return sel, nil +} + +// OnTick implements router.RotationHook: it runs the preset's stateful tick +// controller over the current leg snapshot and returns the structural action +// (promote/demote/add/drop) for the route group to apply. +func (h *Hook) OnTick(_ router.DialInfo, legs []router.LegInfo) router.RotationAction { + action := h.engine.OnTick(h.name, legsToPreset(legs)) + return router.RotationAction{ + DropLegs: action.DropLegs, + AddLeg: action.AddLeg, + ExcludeHops: action.ExcludeHops, + DemoteToStandby: action.DemoteToStandby, + PromoteFromStandby: action.PromoteFromStandby, + } +} + +// enrich converts router CandidateInfo to preset.Candidate, filling per-hop geo +// codes and the distinct transport kinds along the path from the Provider. +func (h *Hook) enrich(cs []router.CandidateInfo) []preset.Candidate { + if len(cs) == 0 { + return nil + } + out := make([]preset.Candidate, 0, len(cs)) + for _, c := range cs { + geo := make([]string, len(c.Hops)) + var kinds []string + seen := map[string]struct{}{} + for i, pk := range c.Hops { + geo[i] = h.provider.Geo(pk) + if k := h.provider.Kind(pk); k != "" { + if _, ok := seen[k]; !ok { + seen[k] = struct{}{} + kinds = append(kinds, k) + } + } + } + out = append(out, preset.Candidate{ + Hops: append([]string(nil), c.Hops...), + HopsGeo: geo, + EstLatencyMs: c.EstLatencyMs, + TransportKinds: kinds, + }) + } + return out +} + +func dialInfoToCtx(info router.DialInfo) preset.Context { + return preset.Context{ + App: info.AppName, + PeerPK: info.PeerPK.Hex(), + Port: uint16(info.RPort), + CLIOverrides: info.CLIOverrides, + IsDirectDial: info.IsDirectDial, + TransportKind: info.TransportKind, + } +} + +func legsToPreset(legs []router.LegInfo) []preset.LegInfo { + if len(legs) == 0 { + return nil + } + out := make([]preset.LegInfo, len(legs)) + for i, l := range legs { + out[i] = preset.LegInfo{ + Index: l.Index, + Kind: l.Kind, + TransportID: l.TransportID, + LatencyMs: l.LatencyMs, + Alive: l.Alive, + Standby: l.Standby, + SentBytes: l.SentBytes, + RecvBytes: l.RecvBytes, + Retransmits: l.Retransmits, + Hops: l.Hops, + } + } + return out +} + +// matchCandidate finds want in pool by hop equality (the stable, un-fabricable +// key), returning -1 when no candidate matches so the router keeps its own pick. +func matchCandidate(pool []preset.Candidate, want preset.Candidate) int { + for i := range pool { + if hopsEqual(pool[i].Hops, want.Hops) { + return i + } + } + return -1 +} + +func hopsEqual(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +// distributionFor maps the descriptor strings the presets emit ("round-robin", +// "auto", "") onto a router.DistributionConfig. The presets only ever emit +// these; any other value maps to Unset (defer to the visor-wide muxMode). This +// is the minimal parse the wasm path needs — the native path's full descriptor +// grammar (pkg/router/policy.ParseDistribution) lives in the starlark-tainted +// policy package and isn't needed here. +func distributionFor(desc string) router.DistributionConfig { + switch desc { + case "round-robin": + return router.DistributionConfig{Mode: router.DistributionRoundRobin} + case "auto": + return router.DistributionConfig{Mode: router.DistributionAuto} + default: + return router.DistributionConfig{Mode: router.DistributionUnset} + } +} diff --git a/pkg/router/policy/presethook/presethook_test.go b/pkg/router/policy/presethook/presethook_test.go new file mode 100644 index 0000000000..563e6abc28 --- /dev/null +++ b/pkg/router/policy/presethook/presethook_test.go @@ -0,0 +1,83 @@ +package presethook + +import ( + "context" + "reflect" + "testing" + + "github.com/skycoin/skywire/pkg/router" +) + +func TestBeforeDial_RotatingBWShape(t *testing.T) { + h := New("rotating-bw", nil) + adj, err := h.BeforeDial(context.Background(), router.DialInfo{AppName: "skysocks-client"}) + if err != nil { + t.Fatalf("BeforeDial: %v", err) + } + want := router.DialAdjustment{ + MuxRoutes: 5, + MinHops: 2, + RotationIntervalSeconds: 90, + AvoidDirect: true, + Distribution: router.DistributionConfig{Mode: router.DistributionRoundRobin}, + } + if !reflect.DeepEqual(adj, want) { + t.Errorf("rotating-bw BeforeDial:\n got %+v\nwant %+v", adj, want) + } +} + +func TestBeforeDial_AppMuxOtherIsNoop(t *testing.T) { + h := New("app-mux", nil) + adj, err := h.BeforeDial(context.Background(), router.DialInfo{AppName: "unknown"}) + if err != nil { + t.Fatalf("BeforeDial: %v", err) + } + if !reflect.DeepEqual(adj, router.DialAdjustment{}) { + t.Errorf("app-mux for a non-target app must be a no-op adjustment; got %+v", adj) + } +} + +// staticProvider supplies fixed geo/kind for named hops. +type staticProvider struct { + geo map[string]string + kind map[string]string +} + +func (p staticProvider) Geo(pk string) string { + if g, ok := p.geo[pk]; ok { + return g + } + return "??" +} +func (p staticProvider) Kind(pk string) string { return p.kind[pk] } + +func TestSelectRoute_GeoAvoidPicksCleanRoute(t *testing.T) { + prov := staticProvider{geo: map[string]string{"a": "US", "b": "DE"}} + h := New("geo-avoid", prov) + fwd := []router.CandidateInfo{ + {Hops: []string{"a"}, EstLatencyMs: 10}, // US — blocked + {Hops: []string{"b"}, EstLatencyMs: 50}, // DE — clean + } + sel, err := h.SelectRoute(context.Background(), + router.DialInfo{AppName: "skysocks-client", CLIOverrides: map[string]string{"avoid_geo": "US"}}, + fwd, nil) + if err != nil { + t.Fatalf("SelectRoute: %v", err) + } + if sel.Chosen != 1 { + t.Fatalf("geo-avoid should choose the clean DE candidate (index 1); got %d", sel.Chosen) + } +} + +func TestOnTick_RotatingBWParksFragile(t *testing.T) { + h := New("rotating-bw", nil) + legs := []router.LegInfo{ + {Index: 0, TransportID: "t0", Kind: "stcpr", Alive: true}, + {Index: 1, TransportID: "t1", Kind: "webrtc", Alive: true}, + } + act := h.OnTick(router.DialInfo{AppName: "skysocks-client"}, legs) + want := router.RotationAction{DemoteToStandby: []int{1}} + if !reflect.DeepEqual(act, want) { + t.Errorf("OnTick rotating-bw: got %+v want %+v", act, want) + } +} diff --git a/pkg/router/policy/wasm/presets/bundle.wasm b/pkg/router/policy/wasm/presets/bundle.wasm index c0a844f3ff..9eaed2c38f 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 new file mode 100644 index 0000000000..ce9dd53bd6 --- /dev/null +++ b/pkg/router/policy/wasm/presets/parity_test.go @@ -0,0 +1,288 @@ +package presets + +// parity_test.go is the gate-1 "byte-identical decisions" property extended to +// the NATIVE-Go preset path. The compiled bundle.wasm (run here via wazero) and +// the pure-Go pkg/router/policy/preset package are two compilations of ONE +// source of truth; the wasm-visor cannot host wazero (wasm-in-wasm) and so runs +// the native path instead. These tests assert the two agree — field-for-field +// on every decide, and step-for-step over a multi-tick on_tick sequence (the +// stateful controllers) — for representative inputs of every preset. If they +// ever diverge, the bundle was rebuilt from a different revision of the preset +// package than the one compiled into the visor, and this fails loudly. + +import ( + "context" + "reflect" + "testing" + "time" + + "github.com/skycoin/skywire/pkg/router/policy" + "github.com/skycoin/skywire/pkg/router/policy/preset" + policywasm "github.com/skycoin/skywire/pkg/router/policy/wasm" +) + +// unixNano builds a UTC time.Time from a unix-nanosecond value so time-of-day +// decide cases feed both paths the same clock. +func unixNano(ns int64) time.Time { return time.Unix(0, ns).UTC() } + +// normSpec is the comparable projection of a route-spec, built from either the +// wazero policy.RouteSpec or the native preset.Spec so the two can be compared +// directly (slices normalized nil==empty via the candidate projection). +type normSpec struct { + Chosen, ReverseChosen *normCand + Mux, ForwardMux, ReverseMux int + MinHops int + ForwardMinHops, ReverseMinHops int + Fallback, Distribution string + RotationIntervalSeconds int +} + +type normCand struct { + Hops []string + HopsGeo []string + EstLatencyMs int + TransportKinds []string +} + +func normCandFromPolicy(c *policy.Candidate) *normCand { + if c == nil { + return nil + } + return &normCand{Hops: nz(c.Hops), HopsGeo: nz(c.HopsGeo), EstLatencyMs: c.EstLatencyMs, TransportKinds: nz(c.TransportKinds)} +} + +func normCandFromPreset(c *preset.Candidate) *normCand { + if c == nil { + return nil + } + return &normCand{Hops: nz(c.Hops), HopsGeo: nz(c.HopsGeo), EstLatencyMs: c.EstLatencyMs, TransportKinds: nz(c.TransportKinds)} +} + +func specFromPolicy(s policy.RouteSpec) normSpec { + return normSpec{ + Chosen: normCandFromPolicy(s.Chosen), ReverseChosen: normCandFromPolicy(s.ReverseChosen), + Mux: s.Mux, ForwardMux: s.ForwardMux, ReverseMux: s.ReverseMux, + MinHops: s.MinHops, ForwardMinHops: s.ForwardMinHops, ReverseMinHops: s.ReverseMinHops, + Fallback: s.Fallback, Distribution: s.Distribution, RotationIntervalSeconds: s.RotationIntervalSeconds, + } +} + +func specFromPreset(s preset.Spec) normSpec { + return normSpec{ + Chosen: normCandFromPreset(s.Chosen), ReverseChosen: normCandFromPreset(s.ReverseChosen), + Mux: s.Mux, ForwardMux: s.ForwardMux, ReverseMux: s.ReverseMux, + MinHops: s.MinHops, ForwardMinHops: s.ForwardMinHops, ReverseMinHops: s.ReverseMinHops, + Fallback: s.Fallback, Distribution: s.Distribution, RotationIntervalSeconds: s.RotationIntervalSeconds, + } +} + +// normAct is the comparable projection of a rotation action (slices normalized +// nil==empty so the wazero JSON round-trip and the native path compare equal). +type normAct struct { + DropLegs []int + AddLeg bool + ExcludeHops []string + DemoteToStandby []int + PromoteFromStandby []int +} + +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)} +} + +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)} +} + +func nz(s []string) []string { + if len(s) == 0 { + return nil + } + return s +} + +func nzi(s []int) []int { + if len(s) == 0 { + return nil + } + return s +} + +// decideCase is one representative decide input shared by both paths. +type decideCase struct { + name string + preset string + rctx policy.RoutingContext + cands []policy.Candidate +} + +func toPresetCtx(r policy.RoutingContext) preset.Context { + return preset.Context{ + App: r.App, PeerPK: r.PeerPK, Port: r.Port, NowUnixNano: r.Now.UnixNano(), + CLIOverrides: r.CLIOverrides, IsDirectDial: r.IsDirectDial, TransportKind: r.TransportKind, + ReverseCandidates: toPresetCands(r.ReverseCandidates), + } +} + +func toPresetCands(cs []policy.Candidate) []preset.Candidate { + if cs == nil { + return nil + } + out := make([]preset.Candidate, len(cs)) + for i, c := range cs { + out[i] = preset.Candidate{Hops: c.Hops, HopsGeo: c.HopsGeo, EstLatencyMs: c.EstLatencyMs, TransportKinds: c.TransportKinds} + } + return out +} + +// TestDecideParity_NativeMatchesWazero asserts preset.Decide == the wazero +// bundle's Decide for representative inputs of every preset. +func TestDecideParity_NativeMatchesWazero(t *testing.T) { + cands := []policy.Candidate{ + {Hops: []string{"a"}, HopsGeo: []string{"US"}, EstLatencyMs: 10, TransportKinds: []string{"stcpr"}}, + {Hops: []string{"b"}, HopsGeo: []string{"DE"}, EstLatencyMs: 50, TransportKinds: []string{"stcpr", "sudph"}}, + {Hops: []string{"trustA", "trustB"}, HopsGeo: []string{"FR", "NL"}, EstLatencyMs: 30, TransportKinds: []string{"sudph"}}, + } + const oneHour = int64(3600) * 1_000_000_000 + cases := []decideCase{ + {"app-mux/vpn", "app-mux", policy.RoutingContext{App: "vpn-client"}, nil}, + {"app-mux/skychat", "app-mux", policy.RoutingContext{App: "skychat"}, nil}, + {"app-mux/other", "app-mux", policy.RoutingContext{App: "unknown"}, nil}, + {"rotating-bw/proxy", "rotating-bw", policy.RoutingContext{App: "skysocks-client"}, nil}, + {"rotating-bw/chat", "rotating-bw", policy.RoutingContext{App: "skychat"}, nil}, + {"latency-adaptive", "latency-adaptive", policy.RoutingContext{App: "vpn-client"}, nil}, + {"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}, + {"adaptive", "adaptive", policy.RoutingContext{App: "vpn-client"}, nil}, + {"geo-avoid", "geo-avoid", policy.RoutingContext{App: "skysocks-client", CLIOverrides: map[string]string{"avoid_geo": "US"}}, cands}, + {"geo-avoid/noclean", "geo-avoid", policy.RoutingContext{App: "skysocks-client", CLIOverrides: map[string]string{"avoid_geo": "US,DE,FR"}}, cands}, + {"transport-diverse", "transport-diverse", policy.RoutingContext{App: "skysocks-client"}, cands}, + {"trust-tiered", "trust-tiered", policy.RoutingContext{App: "skysocks-client", CLIOverrides: map[string]string{"trusted_pks": "trustA,trustB"}}, cands}, + {"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}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + l, err := policywasm.NewLoaderBytes(tc.preset, Bundle(), policywasm.WithPreset(tc.preset)) + if err != nil { + t.Fatalf("NewLoaderBytes: %v", err) + } + defer l.Close() //nolint:errcheck + + wz, err := l.Decide(context.Background(), tc.rctx, tc.cands) + if err != nil { + t.Fatalf("wazero Decide: %v", err) + } + nat := preset.Decide(tc.preset, toPresetCtx(tc.rctx), toPresetCands(tc.cands)) + + if got, want := specFromPreset(nat), specFromPolicy(wz); !reflect.DeepEqual(got, want) { + t.Errorf("decide parity mismatch for %s:\n native: %+v\n wazero: %+v", tc.preset, got, want) + } + }) + } +} + +// tickCase is a preset plus a fixed sequence of leg snapshots fed to both the +// wazero on_tick and the native Engine.OnTick tick-by-tick. +type tickCase struct { + name string + preset string + steps [][]policy.LegInfo +} + +func toPresetLegs(ls []policy.LegInfo) []preset.LegInfo { + if ls == nil { + return nil + } + out := make([]preset.LegInfo, len(ls)) + for i, l := range ls { + out[i] = preset.LegInfo{ + Index: l.Index, Kind: l.Kind, TransportID: l.TransportID, LatencyMs: l.LatencyMs, + Alive: l.Alive, Standby: l.Standby, SentBytes: l.SentBytes, RecvBytes: l.RecvBytes, + Retransmits: l.Retransmits, Hops: l.Hops, + } + } + return out +} + +// TestTickParity_NativeMatchesWazero drives the SAME leg-snapshot sequence +// through the wazero on_tick and the native Engine.OnTick and asserts each +// step's RotationAction is identical — proving the stateful controllers +// (EWMA smoothing, AIMD peak/idle counters, probe state machine) evolve +// byte-identically across the two compilations. +func TestTickParity_NativeMatchesWazero(t *testing.T) { + leg := func(idx int, tid, kind string, lat int, alive, standby bool, recv uint64, hops ...string) policy.LegInfo { + return policy.LegInfo{Index: idx, TransportID: tid, Kind: kind, LatencyMs: lat, Alive: alive, Standby: standby, RecvBytes: recv, Hops: hops} + } + + // rotating-bw: a reliable + fragile active mix, converging toward reliable-only. + rbw := [][]policy.LegInfo{ + {leg(0, "t0", "stcpr", 50, true, false, 0), leg(1, "t1", "webrtc", 80, true, false, 0)}, + {leg(0, "t0", "stcpr", 50, true, false, 0), leg(1, "t1", "webrtc", 80, true, true, 0)}, + {leg(0, "t0", "stcpr", 50, true, false, 0)}, + } + // latency-adaptive: an outlier leg that should eventually be swapped. + la := [][]policy.LegInfo{ + {leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 45, true, false, 0), leg(2, "c", "stcpr", 50, true, false, 0), leg(3, "d", "stcpr", 900, true, false, 0)}, + {leg(0, "a", "stcpr", 42, true, false, 0), leg(1, "b", "stcpr", 44, true, false, 0), leg(2, "c", "stcpr", 51, true, false, 0), leg(3, "d", "stcpr", 950, true, false, 0)}, + {leg(0, "a", "stcpr", 41, true, false, 0), leg(1, "b", "stcpr", 46, true, false, 0), leg(2, "c", "stcpr", 49, true, false, 0), leg(3, "d", "stcpr", 980, true, false, 0)}, + } + // elastic-mux: growing then idle throughput. + em := [][]policy.LegInfo{ + {leg(0, "a", "stcpr", 30, true, false, 1000), leg(1, "b", "stcpr", 30, true, false, 1000)}, + {leg(0, "a", "stcpr", 30, true, false, 9000), leg(1, "b", "stcpr", 30, true, false, 9000)}, + {leg(0, "a", "stcpr", 30, true, false, 20000), leg(1, "b", "stcpr", 30, true, false, 20000)}, + {leg(0, "a", "stcpr", 30, true, false, 20001), leg(1, "b", "stcpr", 30, true, false, 20001)}, + {leg(0, "a", "stcpr", 30, true, false, 20001), leg(1, "b", "stcpr", 30, true, false, 20001)}, + {leg(0, "a", "stcpr", 30, true, false, 20001), leg(1, "b", "stcpr", 30, true, false, 20001)}, + } + // probe-and-prune: steady 3-wide established set to trigger an explore add. + pp := [][]policy.LegInfo{ + {leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 45, true, false, 0), leg(2, "c", "stcpr", 50, true, false, 0)}, + {leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 45, true, false, 0), leg(2, "c", "stcpr", 50, true, false, 0), leg(3, "d", "stcpr", 30, true, false, 0)}, + {leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 45, true, false, 0), leg(2, "c", "stcpr", 50, true, false, 0), leg(3, "d", "stcpr", 30, true, false, 0)}, + {leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 45, true, false, 0), leg(2, "c", "stcpr", 50, true, false, 0), leg(3, "d", "stcpr", 30, true, false, 0)}, + } + // adaptive: mixed load + latency to exercise the arbitration priority. + ad := [][]policy.LegInfo{ + {leg(0, "a", "stcpr", 40, true, false, 1000)}, + {leg(0, "a", "stcpr", 40, true, false, 9000)}, + {leg(0, "a", "stcpr", 40, true, false, 20000), leg(1, "b", "stcpr", 45, true, false, 20000)}, + {leg(0, "a", "stcpr", 41, true, false, 30000), leg(1, "b", "stcpr", 900, true, false, 30000)}, + {leg(0, "a", "stcpr", 42, true, false, 30001), leg(1, "b", "stcpr", 950, true, false, 30001)}, + {leg(0, "a", "stcpr", 43, true, false, 30001), leg(1, "b", "stcpr", 980, true, false, 30001)}, + } + + 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}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + l, err := policywasm.NewLoaderBytes(tc.preset, Bundle(), policywasm.WithPreset(tc.preset)) + if err != nil { + t.Fatalf("NewLoaderBytes: %v", err) + } + defer l.Close() //nolint:errcheck + eng := preset.New() + + for i, legs := range tc.steps { + wz, err := l.OnTick(context.Background(), policy.RoutingContext{App: "skysocks-client"}, legs) + if err != nil { + t.Fatalf("wazero OnTick step %d: %v", i, err) + } + nat := eng.OnTick(tc.preset, toPresetLegs(legs)) + if got, want := actFromPreset(nat), actFromPolicy(wz); !reflect.DeepEqual(got, want) { + t.Errorf("tick parity mismatch %s step %d:\n native: %+v\n wazero: %+v", tc.preset, i, got, want) + } + } + }) + } +}