feat(router): ECF predictive mux scheduler (proxy mux mode ecf)

cmd/skywire-cli/commands/proxy/mux_ops.go

@@ -257,7 +257,7 @@ Example:
 }
 
 var muxModeCmd = &cobra.Command{
-	Use:   "mode <auto|equal|capacity>",
+	Use:   "mode <auto|equal|capacity|ecf>",
 	Short: "Change mux scheduler weighting at runtime",
 	Long: `Set the mux transport-selection mode for the visor.
 
@@ -272,12 +272,22 @@ var muxModeCmd = &cobra.Command{
              and a slow one carries little — the thin-spread aggregation
              mode. A just-promoted leg starts at a small cold-leg floor
              share and ramps as its goodput proves out.
+  ecf      - Earliest Completion First: predictive hold-back. Sends on
+             the fastest leg while it has send capacity and only spills
+             onto a slower leg when that leg would deliver its frame
+             sooner than the fast leg can drain its own backlog —
+             otherwise it holds the frame on the fast leg. Unlike
+             capacity (which still sprays a share onto slow legs and
+             head-of-line-stalls the reorder buffer on them), ECF
+             aggregates across heterogeneous legs without paying the
+             slow-leg HoL cost.
 
 Affects every active and future mux'd route group on this visor
 IMMEDIATELY (the router re-applies the mode to live route groups). The
 setting persists to skywire-config.json so it survives restart.
 
 Example:
+  skywire cli proxy mux mode ecf        # predictive earliest-completion-first
   skywire cli proxy mux mode capacity   # goodput-weighted thin spread
   skywire cli proxy mux info --watch 1s
   skywire cli proxy mux mode auto       # back to latency-weighted`,
@@ -286,9 +296,9 @@ Example:
 	Run: func(cmd *cobra.Command, args []string) {
 		mode := args[0]
 		switch mode {
-		case "auto", "equal", "capacity":
+		case "auto", "equal", "capacity", "ecf":
 		default:
-			internal.PrintFatalError(cmd.Flags(), fmt.Errorf("mode must be 'auto', 'equal', or 'capacity', got %q", mode))
+			internal.PrintFatalError(cmd.Flags(), fmt.Errorf("mode must be 'auto', 'equal', 'capacity', or 'ecf', got %q", mode))
 		}
 		rpcClient, err := clirpc.Client(cmd.Flags())
 		if err != nil {

pkg/router/dial_hook.go

@@ -254,6 +254,17 @@ const (
 	// for aggregating bandwidth across a disjoint multi-leg route.
 	// This is the adaptive default's bulk-spread mode.
 	DistributionCapacity
+	// DistributionECF is the predictive Earliest-Completion-First
+	// hold-back scheduler (router.WeightModeECF). Unlike
+	// DistributionCapacity, which sprays a fraction of frames onto
+	// slower legs in proportion to their goodput — and so keeps
+	// head-of-line-stalling the in-order reorder buffer on those slow
+	// legs — ECF sends on the fastest leg while it has send capacity
+	// and only spills onto a slower leg when the slow leg would deliver
+	// its frame sooner than the fast leg can drain its own backlog. It
+	// is the mode that actually aggregates across heterogeneous legs
+	// without paying the slow-leg HoL cost. Per-packet, O(legs).
+	DistributionECF
 )
 
 // LegChangeHook is an optional hook fired by the route group

pkg/router/policy/distribution.go

@@ -48,6 +48,8 @@ func ParseDistribution(s string) (router.DistributionConfig, error) {
 		return router.DistributionConfig{Mode: router.DistributionLatencyAdaptive}, nil
 	case "capacity", "capacity-weighted":
 		return router.DistributionConfig{Mode: router.DistributionCapacity}, nil
+	case "ecf":
+		return router.DistributionConfig{Mode: router.DistributionECF}, nil
 	}
 
 	// Prefix-based.

pkg/router/policy/presethook/presethook.go

@@ -243,6 +243,8 @@ func distributionFor(desc string) router.DistributionConfig {
 		return router.DistributionConfig{Mode: router.DistributionAuto}
 	case "capacity":
 		return router.DistributionConfig{Mode: router.DistributionCapacity}
+	case "ecf":
+		return router.DistributionConfig{Mode: router.DistributionECF}
 	default:
 		return router.DistributionConfig{Mode: router.DistributionUnset}
 	}

pkg/router/route_group.go

@@ -1129,6 +1129,8 @@ func (rg *RouteGroup) applyDistribution(cfg DistributionConfig) {
 		wm = WeightModeLatencyAdaptive
 	case DistributionCapacity:
 		wm = WeightModeCapacity
+	case DistributionECF:
+		wm = WeightModeECF
 	case DistributionDSCPPriority:
 		wm = WeightModeDSCPPriority
 		rg.mux.tpSelector.SetDSCPThreshold(cfg.DSCPThreshold)

pkg/router/route_mux.go

@@ -94,6 +94,16 @@ type legCounters struct {
 	lastRateNano      int64
 	goodputUpBps      float64
 	goodputDownBps    float64
+	// ECF (WeightModeECF) per-leg state, maintained by rebuildWeights' ECF
+	// branch (under legMu, off the data path). ecfLastSentBytes snapshots the
+	// sent counter at the previous ECF refresh so the delta over the refresh
+	// window is this leg's send rate (kept separate from lastTotalBytes /
+	// lastRateSentBytes so the ECF sampler never disturbs the capacity or
+	// telemetry samplers). ecfRttMs / ecfJitterMs are the EWMA'd mean RTT and
+	// jitter (sigma) the ECF predicate consumes.
+	ecfLastSentBytes uint64
+	ecfRttMs         float64
+	ecfJitterMs      float64
 }
 
 // routeMux encapsulates route multiplexing state and logic.
@@ -160,6 +170,11 @@ type routeMux struct {
 	// Guarded by legMu. See docs/warm_standby_legs_rfc.md.
 	standby []bool
 
+	// ecfLastRebuildNano is the wall-clock (UnixNano) of the previous ECF-state
+	// refresh, used to turn each leg's sent-byte delta into a bytes/sec rate.
+	// Touched only under legMu in rebuildWeights' ECF branch.
+	ecfLastRebuildNano int64
+
 	// standbyNewLegs makes every NEWLY-grown aux leg (index > 0) enter the
 	// warm-standby pool instead of going straight into the active send set.
 	// The primary leg (index 0) is never affected. Set only when a promoting
@@ -246,7 +261,8 @@ func (m *routeMux) selectTransport(tps []*transport.ManagedTransport, fwd []rout
 		case WeightModeSizeThreshold,
 			WeightModeSticky5Tuple,
 			WeightModeLatencyAdaptive,
-			WeightModeDSCPPriority:
+			WeightModeDSCPPriority,
+			WeightModeECF:
 			idx := m.tpSelector.SelectForPayload(payload)
 			if idx < len(tps) {
 				tp := tps[idx]
@@ -855,5 +871,67 @@ func (m *routeMux) rebuildWeights(tps []*transport.ManagedTransport) {
 		m.legMu.Unlock()
 		m.tpSelector.SetCapacityWeights(weights)
 	}
+	// ECF mode: build the per-leg {rate, RTT, jitter, ready, BDP} snapshot the
+	// predictive scheduler reasons over. Rate is the sent-byte delta over the
+	// refresh window (computed here, not from snapshotLegs, so it works even
+	// when nothing is observing the telemetry page). RTT is the leg's first-hop
+	// transport latency (tp.GetLatency(), ms) — the end-to-end route latency
+	// would be more accurate but is not reachable from the mux; noted as a
+	// follow-up. Jitter is an EWMA of |RTT-mean|, the ECF sigma margin.
+	if m.tpSelector.Mode() == WeightModeECF {
+		m.legMu.Lock()
+		now := time.Now().UnixNano()
+		var elapsed float64
+		if m.ecfLastRebuildNano != 0 {
+			elapsed = float64(now-m.ecfLastRebuildNano) / float64(time.Second)
+		}
+		states := make([]ecfLegState, len(m.legs))
+		for i, lc := range m.legs {
+			if lc == nil {
+				continue
+			}
+			// Send rate over the refresh window (bytes/sec).
+			sent := atomic.LoadUint64(&lc.sentBytes)
+			var rate float64
+			if elapsed > 0 {
+				rate = float64(byteDelta(sent, lc.ecfLastSentBytes)) / elapsed
+			}
+			lc.ecfLastSentBytes = sent
+			// RTT EWMA + jitter (sigma) EWMA.
+			var rttMs float64
+			if i < len(tps) && tps[i] != nil {
+				rttMs = tps[i].GetLatency()
+			}
+			if rttMs > 0 {
+				if lc.ecfRttMs == 0 {
+					lc.ecfRttMs = rttMs
+				} else {
+					dev := rttMs - lc.ecfRttMs
+					if dev < 0 {
+						dev = -dev
+					}
+					lc.ecfJitterMs = ecfJitterAlpha*dev + (1-ecfJitterAlpha)*lc.ecfJitterMs
+					lc.ecfRttMs = ecfRttAlpha*rttMs + (1-ecfRttAlpha)*lc.ecfRttMs
+				}
+			}
+			ready := true
+			if i < len(m.standby) && m.standby[i] {
+				ready = false
+			}
+			if i < len(m.ready) && !m.ready[i] {
+				ready = false
+			}
+			states[i] = ecfLegState{
+				rttMs:     lc.ecfRttMs,
+				jitterMs:  lc.ecfJitterMs,
+				rateBps:   rate,
+				cwndBytes: rate * lc.ecfRttMs / 1000.0,
+				ready:     ready,
+			}
+		}
+		m.ecfLastRebuildNano = now
+		m.legMu.Unlock()
+		m.tpSelector.SetECFState(states)
+	}
 	m.tpSelector.Rebuild(tps)
 }