diff --git a/cmd/skywire-cli/commands/proxy/mux_set.go b/cmd/skywire-cli/commands/proxy/mux_set.go index a123ef7c01..7b540b0add 100644 --- a/cmd/skywire-cli/commands/proxy/mux_set.go +++ b/cmd/skywire-cli/commands/proxy/mux_set.go @@ -37,8 +37,69 @@ import ( clirpc "github.com/skycoin/skywire/cmd/skywire-cli/commands/rpc" "github.com/skycoin/skywire/pkg/cliout" "github.com/skycoin/skywire/pkg/cliout/cliproxy" + "github.com/skycoin/skywire/pkg/visor" ) +// legReconcile is the outcome of reconcileLegs: the first-hop transport ids of +// legs added and (with prune) removed, plus how many targets were already present. +type legReconcile struct { + added, removed []string + existing int +} + +// reconcileLegs makes app's mux legs be AT LEAST (prune=false) or EXACTLY +// (prune=true) the target set, keyed by each leg's first-hop transport id. +// It is the shared engine behind `proxy mux set` and `proxy start --route`. +// The route group must already exist (start the proxy first). Per-leg RPC +// errors are logged to stderr and skipped rather than aborting the batch. +func reconcileLegs(rpcClient visor.API, app string, srcPort uint16, targets []routePair, prune bool) (legReconcile, error) { + var res legReconcile + want := make(map[uuid.UUID]routePair, len(targets)) + for _, t := range targets { + if len(t.Forward) == 0 || len(t.Reverse) == 0 { + return res, fmt.Errorf("target leg missing forward or reverse hops") + } + want[t.Forward[0].TpID] = t + } + + infos, err := rpcClient.RouteGroupMuxInfo(app) + if err != nil { + return res, fmt.Errorf("RouteGroupMuxInfo: %w", err) + } + current, err := currentLegTpIDs(infos, srcPort) + if err != nil { + return res, err + } + + // Add target legs that aren't present yet. + for tp, t := range want { + if _, ok := current[tp]; ok { + res.existing++ + continue + } + if err := rpcClient.AddMuxRoute(app, t.Forward, t.Reverse, srcPort); err != nil { + fmt.Fprintf(os.Stderr, " add leg (first tp=%s): %v\n", tp, err) + continue + } + res.added = append(res.added, fmt.Sprint(tp)) + } + + // Prune current legs absent from the target set. + if prune { + for tp := range current { + if _, ok := want[tp]; ok { + continue + } + if err := rpcClient.RemoveMuxRoute(app, tp, srcPort); err != nil { + fmt.Fprintf(os.Stderr, " remove leg (tp=%s): %v\n", tp, err) + continue + } + res.removed = append(res.removed, fmt.Sprint(tp)) + } + } + return res, nil +} + var ( muxSetApp string muxSetSrcPort uint16 @@ -149,13 +210,6 @@ Example: if err != nil { internal.PrintFatalError(cmd.Flags(), err) } - want := make(map[uuid.UUID]routePair, len(targets)) - for _, t := range targets { - if len(t.Forward) == 0 || len(t.Reverse) == 0 { - internal.PrintFatalError(cmd.Flags(), fmt.Errorf("target leg missing forward or reverse hops")) - } - want[t.Forward[0].TpID] = t - } rpcClient, err := clirpc.Client(cmd.Flags()) if err != nil { @@ -163,58 +217,23 @@ Example: } defer rpcClient.Close() //nolint:errcheck,gosec - infos, err := rpcClient.RouteGroupMuxInfo(muxSetApp) - if err != nil { - internal.PrintFatalError(cmd.Flags(), fmt.Errorf("RouteGroupMuxInfo: %w", err)) - } - current, err := currentLegTpIDs(infos, muxSetSrcPort) + res, err := reconcileLegs(rpcClient, muxSetApp, muxSetSrcPort, targets, muxSetPrune) if err != nil { internal.PrintFatalError(cmd.Flags(), err) } - - // Add target legs that aren't present yet. - added, present := 0, 0 - var addedTps []string - for tp, t := range want { - if _, ok := current[tp]; ok { - present++ - continue - } - if err := rpcClient.AddMuxRoute(muxSetApp, t.Forward, t.Reverse, muxSetSrcPort); err != nil { - fmt.Fprintf(os.Stderr, " add leg (first tp=%s): %v\n", tp, err) - continue - } - added++ - addedTps = append(addedTps, fmt.Sprint(tp)) - if !cliout.JSONMode(cmd) { + if !cliout.JSONMode(cmd) { + for _, tp := range res.added { fmt.Printf("+ added leg (first tp=%s)\n", tp) } - } - - // Prune current legs absent from the target. - var removedTps []string - removed := 0 - if muxSetPrune { - for tp := range current { - if _, ok := want[tp]; ok { - continue - } - if err := rpcClient.RemoveMuxRoute(muxSetApp, tp, muxSetSrcPort); err != nil { - fmt.Fprintf(os.Stderr, " remove leg (tp=%s): %v\n", tp, err) - continue - } - removed++ - removedTps = append(removedTps, fmt.Sprint(tp)) - if !cliout.JSONMode(cmd) { - fmt.Printf("- removed leg (tp=%s)\n", tp) - } + for _, tp := range res.removed { + fmt.Printf("- removed leg (tp=%s)\n", tp) } } internal.Catch(cmd.Flags(), cliout.Print(cmd, cliproxy.MuxSet{ - App: muxSetApp, Target: len(want), - Added: addedTps, Removed: removedTps, - Existing: present, Note: pruneNote(muxSetPrune), + App: muxSetApp, Target: len(targets), + Added: res.added, Removed: res.removed, + Existing: res.existing, Note: pruneNote(muxSetPrune), })) }, } diff --git a/cmd/skywire-cli/commands/proxy/proxy.go b/cmd/skywire-cli/commands/proxy/proxy.go index 8919dcb2cd..ee63083242 100644 --- a/cmd/skywire-cli/commands/proxy/proxy.go +++ b/cmd/skywire-cli/commands/proxy/proxy.go @@ -89,6 +89,7 @@ func init() { startCmd.Flags().StringVar(&muxMode, "mux-mode", "auto", "mux weight distribution mode: auto (latency-based) or equal (round-robin)") startCmd.Flags().Uint16Var(&minHops, "min-hops", 1, "minimum routing hops for this session (1=no minimum). Set on the visor before app start; rolled back is not automatic — restart visor or re-run with --min-hops=1 to revert.") startCmd.Flags().IntVar(&startTunnels, "tunnels", 1, "number of independent tunnels (route group + noise + yamux each) to stripe browser connections across; 1 = today's behavior. >1 AGGREGATES bandwidth: each extra tunnel is auto-steered by the visor onto a DIFFERENT first-hop transport (disjoint path) so their throughputs sum. Best paired with --mux 1 (one leg per tunnel).") + startCmd.Flags().StringVar(&startRoute, "route", "", "pin explicit route(s) chosen by you instead of the route finder: a JSON file of {forward,reverse} hop pairs ('cli route calc --count N --json' shape). Once the proxy is up its mux legs are reconciled to these — each pinned route is added as a leg and any AUX auto legs are pruned. NOTE: the auto PRIMARY leg (index 0) is privileged and cannot yet be pruned, so it remains alongside the pinned legs; full primary override is the dial-level follow-up. One pair = one pinned route; N pairs = N disjoint legs. Pair with --mux N. Tip: 'route calc --source tps' avoids stale-transport install failures.") startCmd.Flags().BoolVarP(&startVerbose, "verbose", "v", false, "stream the visor's logs scoped to this app's session (app stdout + tagged router/mux/setup events); ctrl+c stops the proxy and exits") startCmd.Flags().StringVar(&startVerboseLevel, "verbose-level", "debug", "minimum log level when --verbose is set: trace|debug|info|warn|error") startCmd.Flags().BoolVar(&reconnect, "reconnect", true, "in-process reconnect on route-group collapse: proxy keeps re-dialing with backoff instead of dropping the SOCKS5 listener; --reconnect=false restores exit-on-failure") @@ -425,6 +426,36 @@ var startCmd = &cobra.Command{ } } + // --route: pin the session to EXACTLY the supplied route(s). The app + // dialed an auto primary route to come up (above); now reconcile its mux + // legs to be exactly the caller's — add each pinned leg, then prune the + // auto route and any extras. Full manual override of intermediate + // selection (the route finder is bypassed for the working set), sharing + // the engine behind `proxy mux set --prune`. + if appReachedRunning && startRoute != "" { + targets, rErr := readRoutePairs(startRoute) + if rErr != nil { + internal.PrintFatalError(cmd.Flags(), fmt.Errorf("--route: %w", rErr)) + } + // The route group is up once the app reached Running; allow a brief + // lag before it is queryable. + var res legReconcile + for i := 0; i < 10; i++ { + res, rErr = reconcileLegs(rpcClient, clientName, 0, targets, true) + if rErr == nil { + break + } + time.Sleep(500 * time.Millisecond) + } + if rErr != nil { + internal.PrintFatalError(cmd.Flags(), fmt.Errorf("--route reconcile: %w", rErr)) + } + if !startVerbose { + fmt.Printf("route pinned: %d leg(s) added, %d pruned, %d already present\n", + len(res.added), len(res.removed), res.existing) + } + } + // SIGINT during the startup polling loop (above) exits the // loop with appReachedRunning=false. In --verbose mode the // SignalContext goroutine intentionally doesn't call KillApp diff --git a/cmd/skywire-cli/commands/proxy/root.go b/cmd/skywire-cli/commands/proxy/root.go index 5834012272..d1ca6563c8 100644 --- a/cmd/skywire-cli/commands/proxy/root.go +++ b/cmd/skywire-cli/commands/proxy/root.go @@ -122,6 +122,7 @@ var ( muxMode string minHops uint16 startTunnels int + startRoute string startVerbose bool startVerboseLevel string reconnect bool