diff --git a/pkg/router/policy/preset/tick.go b/pkg/router/policy/preset/tick.go index b2c9cb3ecd..ef27082c0a 100644 --- a/pkg/router/policy/preset/tick.go +++ b/pkg/router/policy/preset/tick.go @@ -584,7 +584,19 @@ const ( // out instead of aggregating; the reorder/aggregation (reorder.go, // datagram_route_group.go) doesn't scale to many legs (#86 family). // Both are being worked; the pool is uncapped deliberately to surface them. - adaptStandbyMax = 60 + // + // TRUE UNCAP 2026-08-26: raised 60 -> 512 so the standby pool is the full + // disjoint set the topology offers (a warm visor exposes ~480 disjoint + // intermediates to a busy exit), not an arbitrary ceiling. 512 sits above any + // realistic single-exit disjoint count, so the binding limit is the topology, + // discovered by the self-heal's no-progress backoff (route_group.go) — it + // fills to what actually establishes and stops, re-probing as new transports + // come online. Risk (1) above is contained two ways: establishMuxRoutes now + // caps its FOREGROUND initial dial (initialForegroundMux) so the dial returns + // fast on a lean mux, and the background self-heal fills the rest one leg at a + // time with the no-progress backoff — so uncapping the pool never becomes a + // dial storm at connect time. + adaptStandbyMax = 512 adaptStandbyMin = 1 // Health + anti-churn. A leg is a gross-outlier (kept OUT of the active mux, // where the no-skip reorder buffer would head-of-line-stall on it) when its diff --git a/pkg/router/policy/wasm/presets/bundle.wasm b/pkg/router/policy/wasm/presets/bundle.wasm index 217707f29b..7f335c2fd6 100644 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/presets_test.go b/pkg/router/policy/wasm/presets/presets_test.go index bae936fda4..3d0aaaaa01 100644 --- a/pkg/router/policy/wasm/presets/presets_test.go +++ b/pkg/router/policy/wasm/presets/presets_test.go @@ -347,10 +347,11 @@ func TestAdaptiveDecides(t *testing.T) { // dip-free promotion. The forward-lean / reverse-wide split is applied at SEND // time, not baked into asymmetric rule setup. Per-direction ForwardMux/ // ReverseMux stay 0 (symmetric Mux drives it). Mux = adaptRevActive(1) + - // adaptStandbyMax(60) = 61 (uncapped pool, 2026-08-26); this literal mirrors - // the native const in pkg/router/policy/preset/tick.go — keep them in sync. - if spec.Mux != 61 { - t.Errorf("Mux = %d, want 61 (1 active + 60 warm standby, full-duplex)", spec.Mux) + // adaptStandbyMax(512) = 513 (true uncap, 2026-08-26 — the full disjoint pool + // the topology offers, filled in the background); this literal mirrors the + // native const in pkg/router/policy/preset/tick.go — keep them in sync. + if spec.Mux != 513 { + t.Errorf("Mux = %d, want 513 (1 active + 512 warm standby, full-duplex)", spec.Mux) } if spec.ForwardMux != 0 { t.Errorf("ForwardMux = %d, want 0 (symmetric Mux drives adaptive)", spec.ForwardMux) diff --git a/pkg/router/route_group.go b/pkg/router/route_group.go index 50fed229c0..70f5b2f894 100644 --- a/pkg/router/route_group.go +++ b/pkg/router/route_group.go @@ -802,6 +802,13 @@ func (rg *RouteGroup) aliveLegCount() int { return n } +// selfHealNoProgressLimit is how many consecutive self-heal dials may fail to +// grow the live leg count before the heal concludes the destination's disjoint- +// intermediate set is exhausted for now and stops (instead of hammering the +// setup node for the full uncapped target). A later leg death or newly-online +// transport re-triggers the heal, so this is a backoff, not a cap. +const selfHealNoProgressLimit = 4 + // maybeSelfHeal restores the multiplexed degree after a leg drop. If the live // leg count fell below target and no replacement is already in flight, it // dials replacement aux legs in the background until the degree is restored @@ -827,21 +834,48 @@ func (rg *RouteGroup) maybeSelfHeal() { } go func() { defer rg.healInFlight.Store(false) - // Each add(nil) blocks ~one setup-node dial and, on success, - // appends one leg. Re-check the live count between attempts and - // stop as soon as the degree is restored, the group closes, or we - // hit the attempt cap (target+1 gives a little headroom for dials - // that fail on a bad intermediate before one lands). + // Each add(nil) blocks ~one setup-node dial and, on success, appends one + // leg. Re-check the live count between attempts and stop as soon as the + // degree is restored or the group closes. + // + // NO-PROGRESS BACKOFF: with the standby pool uncapped (target ~513), the + // achievable degree is bounded by the destination's disjoint-intermediate + // set, which is usually far below target. Once the pool is filled to what + // the topology offers, every further add fails ("failure code 1: transport + // already in the group" / "setup-node dial: context deadline exceeded") + // and re-dialing target-more times would hammer the setup node for minutes + // (the observed storm). So compare the live count before/after each add: + // after selfHealNoProgressLimit consecutive adds that grow the degree by + // nothing, the disjoint set is exhausted for now — settle at the degree we + // have and stop. A later leg death (which frees an intermediate) or newly- + // online transports (which open fresh disjoint paths) re-trigger this and + // the pool grows again, so the target is never a hard cap — the fill just + // tracks the topology instead of storming past it. + noProgress := 0 for attempt := 0; attempt < target+1; attempt++ { - if rg.isClosed() || rg.aliveLegCount() >= target { + before := rg.aliveLegCount() + if rg.isClosed() || before >= target { return } if rg.logger != nil { - rg.logger.WithField("alive", rg.aliveLegCount()). + rg.logger.WithField("alive", before). WithField("target", target). Debug("Mux self-heal: dialing replacement leg to restore degree") } add(nil) + if rg.aliveLegCount() > before { + noProgress = 0 + continue + } + noProgress++ + if noProgress >= selfHealNoProgressLimit { + if rg.logger != nil { + rg.logger.WithField("alive", rg.aliveLegCount()). + WithField("target", target). + Debug("Mux self-heal: no disjoint path available right now; settling at current degree") + } + return + } } }() } diff --git a/pkg/router/router_dial.go b/pkg/router/router_dial.go index 1ad0705c44..b9cd3902f4 100644 --- a/pkg/router/router_dial.go +++ b/pkg/router/router_dial.go @@ -2202,6 +2202,17 @@ func hopPath(path []routing.Hop) string { // stale-entry leaks. The most useful case is fwd=1 + rev=N for // download-heavy workloads (1 forward upstream + N reverse legs that // aggregate the bulk payload). +// initialForegroundMux bounds how many mux legs establishMuxRoutes sets up +// SYNCHRONOUSLY at dial time. The standby pool is uncapped (adaptStandbyMax=512), +// but dialing hundreds of legs at connect would storm the setup node and stall +// the connection before it serves; so the foreground dial builds a lean mux (a +// few active + a small warm reserve) and returns, and the background self-heal +// fills the rest of the disjoint pool one leg at a time (see SetSelfHeal / +// maybeSelfHeal). Chosen well above the adaptive active cap (adaptCap=8) so a +// download can grow onto warm legs immediately, but small enough that the +// initial dial is a handful of parallel setups, not a storm. +const initialForegroundMux = 16 + func (r *router) establishMuxRoutes( ctx context.Context, nrg *NoiseRouteGroup, @@ -2224,6 +2235,17 @@ func (r *router) establishMuxRoutes( if revCount > maxCount { maxCount = revCount } + // Cap the FOREGROUND initial dial. With the standby pool uncapped + // (adaptStandbyMax=512 → Mux ~513), planning+dialing every leg here — at + // connect time, Phase-2 in parallel — would be a setup-node dial storm that + // stalls the connection before it serves a byte. Instead set up a lean mux + // now (a few active + a small warm reserve) so the dial returns fast, and let + // the BACKGROUND self-heal (SetSelfHeal target = full Mux) fill the rest of + // the disjoint pool one leg at a time. The reverse/forward split is preserved + // below; this only bounds how many legs are attempted synchronously. + if maxCount > initialForegroundMux { + maxCount = initialForegroundMux + } if maxCount <= 1 || nrg.rg.mux == nil { return }