From e3a978888ac0e503e3690797e407f1f608f65554 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Mon, 24 Aug 2026 01:42:58 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20per-forward=20worker=20=E6=A8=A1?= =?UTF-8?q?=E5=9E=8B=20=E2=80=94=20=E6=AF=8F=E6=9D=A1=E8=BD=AC=E5=8F=91?= =?UTF-8?q?=E7=8B=AC=E7=AB=8B=20frpc=20=E9=85=8D=E7=BD=AE+=E7=8B=AC?= =?UTF-8?q?=E7=AB=8B=E8=BF=9B=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心改动（ring 协议不变，Task 本来就是 {Local,Remote,Link} 三元组）: - process.WorkerKey: worker 键改为 'local~remote~port' 三元组 - renderForward: 每条 forward 渲染只含 1 个 proxy 的独立 frpc 配置 (替代 renderRemote 把该 remote 全部 forwards 塞进一个进程) - ClaimFn/RevokeFn: 认领/撤销只操作这一条 forward 自己的进程 - SyncWorkers/auto-start: 只拉起 localOnly 转发的独立进程, 集群转发由 ring 认领节点拉起 (消除三台争抢 proxy already exists) - RemoteStatus/StartRemote/StopRemote/RestartRemote: remote 级聚合辅助, 保持 /profiles API 形状不变, 前端零改动 - handleStatus/logs: 按 worker key 解析真实 remote 名 效果: 三台节点的 frpc 各自只注册自己拥有的 proxy, 单条转发故障不再波及兄弟 --- cmd/webui4frpc/main.go | 179 +++++++++++++++----------- internal/httpapi/handlers.go | 47 +++++-- internal/httpapi/handlers_canvas.go | 6 +- internal/httpapi/handlers_forwards.go | 13 +- internal/httpapi/handlers_logs.go | 9 +- internal/httpapi/server.go | 29 ++++- internal/process/process.go | 82 ++++++++++++ internal/process/worker_key.go | 39 ++++++ internal/process/worker_key_test.go | 37 ++++++ 9 files changed, 340 insertions(+), 101 deletions(-) create mode 100644 internal/process/worker_key.go create mode 100644 internal/process/worker_key_test.go diff --git a/cmd/webui4frpc/main.go b/cmd/webui4frpc/main.go index 24adf97..ea60b43 100644 --- a/cmd/webui4frpc/main.go +++ b/cmd/webui4frpc/main.go @@ -8,6 +8,7 @@ import ( "context" "encoding/json" "flag" + "fmt" "log" "net" "net/http" @@ -90,7 +91,7 @@ func main() { } return fallbackBin }, - Render: renderRemote(st), + Render: renderForward(st), AutoRestart: func(name string) bool { s, _ := st.Settings() if !s.RestartOnExit { @@ -172,44 +173,32 @@ func main() { // disabled (by RevokeFn); clear it so renderRemote renders the // proxy back in. No-op for a fresh claim. _ = st.SetLinkDisabled(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort, false) - // Start (or restart) the worker for the remote we just claimed. - // Restart re-renders the frpc config with all of this remote's - // links (including the just-upserted one) and respawns the process. - // We deliberately do NOT iterate every stored remote here: this - // node's store may hold remotes for cluster forwards it does NOT - // own (the submitter persists the whole canvas), and starting - // those here would duplicate workers across the cluster. + // Start (or restart) the per-forward worker for exactly this link. + // Each forward has its own frpc process (keyed by the forward + // triple); restarting only this key leaves sibling forwards' + // processes untouched. if tk.Remote.Enabled { - if _, has := pm.Status(tk.Remote.Name); has { - _ = pm.Restart(tk.Remote.Name) + key := process.WorkerKey(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort) + if _, has := pm.Status(key); has { + _ = pm.Restart(key) } else { - _ = pm.Start(tk.Remote.Name) + _ = pm.Start(key) } } log.Printf("ring[%s] claimed task %s: %s→%s:%d", selfID, tk.ID, tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort) return nil }, - // Revoke cancels the forward NON-destructively: flag the exact link - // disabled (so renderRemote drops just its proxy and a later forward- - // centric start can re-enable it) and right-size the worker — stop it - // when the remote has no remaining enabled forwards, else restart so - // sibling proxies survive. The old delete-local/remote/drop-all- - // matching-links form served the topology-derived-canvas model but - // broke per-forward stop/start: it wiped sibling forwards sharing the - // local or remote (group stop) and made re-start fail (link gone). + // Revoke cancels the forward NON-destructively: stop THIS forward's + // own worker process (per-forward model — no siblings share it) and + // flag the exact link disabled so a later forward-centric start can + // re-enable it. The old delete-local/remote/drop-all-matching-links + // form served the topology-derived-canvas model but broke per-forward + // stop/start: it wiped sibling forwards sharing the local or remote. RevokeFn: func(ctx context.Context, tk *cluster.Task) error { _ = st.SetLinkDisabled(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort, true) - fwd, _ := st.LinksForRemote(tk.Remote.Name) - enabled := 0 - for _, f := range fwd { - if !f.Disabled { - enabled++ - } - } - if enabled == 0 { - _ = pm.Stop(tk.Remote.Name) - } else if _, running := pm.Status(tk.Remote.Name); running { - _ = pm.Restart(tk.Remote.Name) + key := process.WorkerKey(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort) + if _, running := pm.Status(key); running { + _ = pm.Stop(key) } log.Printf("ring[%s] revoked task %s: %s→%s:%d", selfID, tk.ID, tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort) return nil @@ -297,53 +286,72 @@ func main() { return path, ver, nil }, SyncWorkers: func() { - // Only forwards that stay on this node (localOnly) get their worker - // started here. Cluster-distributed forwards (non-localOnly) are claimed - // and spawned by the owning cluster node — starting them locally too - // would duplicate the worker. - remotes, err := st.ListRemotes() + // PER-FORWARD worker model: every enabled, non-disabled link on THIS + // node gets its own frpc process keyed by the forward triple. + // - localOnly forwards: always owned here (loopback stays local). + // - cluster forwards: only started by the ring claim (ClaimFn) — + // SyncWorkers deliberately skips them so a node that merely holds + // a canvas copy does not spawn workers for forwards it doesn't own + // (the old "every node registers every proxy" bug). + links, err := st.ListLinks() if err != nil { return } - for _, r := range remotes { - if !r.Enabled { + for _, ln := range links { + loc, ok := st.GetLocal(ln.Local) + if !ok || !loc.LocalOnly { + continue // cluster-owned or unknown local + } + rem, ok := st.GetRemote(ln.Remote) + if !ok || !rem.Enabled { continue } - // worker starts here only if every link of this remote is localOnly - fwd, err := st.LinksForRemote(r.Name) - if err != nil { + key := process.WorkerKey(ln.Local, ln.Remote, ln.RemotePort) + if ln.Disabled { + _ = pm.Stop(key) // per-forward stop kills exactly its process continue } - allLocal := len(fwd) > 0 - for _, f := range fwd { - loc, ok := st.GetLocal(f.Service) - if !ok || !loc.LocalOnly { - allLocal = false - break - } - } - if !allLocal { - // has at least one cluster-distributed forward; cluster owns it - continue - } - if _, has := pm.Status(r.Name); has { - _ = pm.Restart(r.Name) + if _, has := pm.Status(key); has { + _ = pm.Restart(key) } else { - _ = pm.Start(r.Name) + _ = pm.Start(key) } } }, } - // Auto-start enabled remotes on boot. + // Auto-start localOnly forwards on boot. Cluster forwards are NOT started + // here — they are re-submitted by the leader's startup auto-submit below + // and claimed (spawning their own worker) wherever the ring places them. s, _ := st.Settings() if s.AutoStartProfiles { - remotes, _ := st.ListRemotes() - for _, r := range remotes { - if r.Enabled { - _ = pm.Start(r.Name) + links, _ := st.ListLinks() + remotesByName := map[string]store.Remote{} + if remotes, err := st.ListRemotes(); err == nil { + for _, r := range remotes { + remotesByName[r.Name] = r } } + localsByName := map[string]store.Local{} + if locals, err := st.ListLocals(); err == nil { + for _, l := range locals { + localsByName[l.Name] = l + } + } + for _, ln := range links { + if ln.Disabled { + continue + } + loc, ok := localsByName[ln.Local] + if !ok || !loc.LocalOnly { + continue + } + rem, ok := remotesByName[ln.Remote] + if !ok || !rem.Enabled { + continue + } + _ = pm.Start(process.WorkerKey(ln.Local, ln.Remote, ln.RemotePort)) + } } handler, err := httpapi.NewServeMux(h) @@ -633,35 +641,44 @@ func splitCSV(s string) []string { return out } -// renderRemote builds the process.Render closure from store data. -func renderRemote(st *store.Store) func(string) ([]byte, error) { - return func(remoteName string) ([]byte, error) { +// renderForward builds the process.Render closure for the PER-FORWARD worker +// model: each (local, remote, remotePort) link gets its OWN frpc config +// containing exactly ONE proxy. The worker key is the forward triple +// (process.WorkerKey), so every node only ever registers its own proxies on +// frps — this eliminates the multi-node "proxy already exists" fight of the +// old per-remote model, where every node rendered ALL forwards of a remote. +// A disabled link renders an empty proxy list: frpc exits cleanly with no +// proxies (the supervisor sees a clean exit and stops restarting it), which +// is exactly the per-forward stop semantics. +func renderForward(st *store.Store) func(string) ([]byte, error) { + return func(key string) ([]byte, error) { + localName, remoteName, port, ok := process.ParseWorkerKey(key) + if !ok { + return nil, fmt.Errorf("invalid worker key %q", key) + } rem, ok := st.GetRemote(remoteName) if !ok { return nil, store.ErrNotFound } - forwards, err := st.LinksForRemote(remoteName) - if err != nil { - return nil, err + loc, ok := st.GetLocal(localName) + if !ok { + return nil, store.ErrNotFound } - proxies := make([]render.Proxy, 0, len(forwards)) - for _, f := range forwards { - // A disabled forward is omitted from the generated frpc config so - // a per-forward stop drops just this proxy on worker restart, - // leaving sibling forwards on the same remote untouched. - if f.Disabled { - continue - } - loc, ok := st.GetLocal(f.Service) - if !ok { - continue + disabled := false + for _, ln := range mustListLinks(st) { + if ln.Local == localName && ln.Remote == remoteName && ln.RemotePort == port { + disabled = ln.Disabled + break } + } + proxies := []render.Proxy(nil) + if !disabled { proxies = append(proxies, render.Proxy{ Name: loc.Name, Type: loc.Protocol, LocalIP: loc.IP, LocalPort: loc.Port, - RemotePort: f.RemotePort, + RemotePort: port, UseEncryption: loc.UseEncryption, UseCompression: loc.UseCompression, BandwidthLimit: loc.BandwidthLimit, @@ -690,6 +707,12 @@ func renderRemote(st *store.Store) func(string) ([]byte, error) { } } +// mustListLinks is ListLinks with errors swallowed (best-effort reads). +func mustListLinks(st *store.Store) []store.Link { + links, _ := st.ListLinks() + return links +} + // multiFlag collects repeated string flags (-peer a -peer b ...). type multiFlag []string diff --git a/internal/httpapi/handlers.go b/internal/httpapi/handlers.go index f81512e..8badb51 100644 --- a/internal/httpapi/handlers.go +++ b/internal/httpapi/handlers.go @@ -2,11 +2,13 @@ package httpapi import ( "encoding/json" + "log" "net/http" "os" "strings" "webui4frpc/internal/cluster" + "webui4frpc/internal/process" "webui4frpc/internal/store" ) @@ -127,8 +129,11 @@ func (h *Handler) applyCanvas(w http.ResponseWriter, r *http.Request, canvas *ca if !keep[old.Name] { if old.LocalOnly { if h.Process != nil { + // Per-forward model: stop every localOnly worker of this local. if fwd, _ := s.LinksForLocal(old.Name); len(fwd) > 0 { - _ = h.Process.Stop(fwd[0].Remote) + for _, f := range fwd { + _ = h.Process.Stop(process.WorkerKey(old.Name, f.Remote, f.RemotePort)) + } } } } else if h.Ring != nil { @@ -324,9 +329,13 @@ func (h *Handler) handleRemoteUpsert(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusInternalServerError) return } - // Start the worker if enabled. + // Start the per-forward workers for this remote if enabled. if rem.Enabled { - _ = h.Process.Start(rem.Name) + // In the per-forward model, merely upserting a remote does not start + // any workers — the actual links control which forwards run. The + // caller is expected to save the canvas (PUT /canvas) or start + // individual forwards via the forwards page. + log.Printf("remote %s upserted (enabled=%v); use status page forwards to start", rem.Name, rem.Enabled) } writeJSON(w, http.StatusOK, rem) } @@ -341,7 +350,7 @@ func (h *Handler) handleRemoteDelete(w http.ResponseWriter, r *http.Request) { http.Error(w, "remote name required", http.StatusBadRequest) return } - _ = h.Process.Stop(name) + _ = h.Process.StopRemote(name) if err := h.Store.DeleteRemote(name); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return @@ -367,8 +376,10 @@ func (h *Handler) handleProfile(w http.ResponseWriter, r *http.Request) { switch action { case "": - // GET profile status - st, has := h.Process.Status(name) + // GET profile status — aggregate per-forward workers into a single + // remote-centric view. "hasProcess" is true when ANY forward worker + // for this remote is running. + st, has := h.Process.RemoteStatus(name) forwards, _ := h.Store.LinksForRemote(name) writeJSON(w, http.StatusOK, map[string]any{ "name": name, "process": st, "hasProcess": has, "forwards": forwards, @@ -384,11 +395,11 @@ func (h *Handler) handleProfile(w http.ResponseWriter, r *http.Request) { } var err error if action == "start" { - err = h.Process.Start(name) + err = h.Process.StartRemote(name) } else if action == "stop" { - err = h.Process.Stop(name) + err = h.Process.StopRemote(name) } else { - err = h.Process.Restart(name) + err = h.Process.RestartRemote(name) } if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) @@ -399,7 +410,15 @@ func (h *Handler) handleProfile(w http.ResponseWriter, r *http.Request) { methodNotAllowed(w) } case "config": - data, err := os.ReadFile(h.Process.ConfigPath(name)) + // Per-forward model: no single config file for a remote. Return the + // first forward's config as a hint; the status page already shows + // per-forward configs. + workers := h.Process.RemoteWorkers(name) + if len(workers) == 0 { + http.NotFound(w, r) + return + } + data, err := os.ReadFile(h.Process.ConfigPath(workers[0])) if err != nil { http.NotFound(w, r) return @@ -407,7 +426,13 @@ func (h *Handler) handleProfile(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") _, _ = w.Write(data) case "logs": - path := h.Process.LogPath(name) + // Per-forward model: return the first per-forward log for this remote. + workers := h.Process.RemoteWorkers(name) + if len(workers) == 0 { + http.Error(w, "no workers for this remote", http.StatusNotFound) + return + } + path := h.Process.LogPath(workers[0]) data, err := tailFile(path, 64*1024) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) diff --git a/internal/httpapi/handlers_canvas.go b/internal/httpapi/handlers_canvas.go index 7c97bbc..c5f4c72 100644 --- a/internal/httpapi/handlers_canvas.go +++ b/internal/httpapi/handlers_canvas.go @@ -9,6 +9,7 @@ import ( "time" "webui4frpc/internal/cluster" + "webui4frpc/internal/process" "webui4frpc/internal/store" ) @@ -76,8 +77,11 @@ func (h *Handler) handleLocalByName(w http.ResponseWriter, r *http.Request) { case http.MethodDelete: if l.LocalOnly { if h.Process != nil { + // Per-forward model: stop every localOnly worker of this local. if fwd, _ := h.Store.LinksForLocal(name); len(fwd) > 0 { - _ = h.Process.Stop(fwd[0].Remote) + for _, f := range fwd { + _ = h.Process.Stop(process.WorkerKey(name, f.Remote, f.RemotePort)) + } } } } else if h.Ring != nil { diff --git a/internal/httpapi/handlers_forwards.go b/internal/httpapi/handlers_forwards.go index 7807381..c43b069 100644 --- a/internal/httpapi/handlers_forwards.go +++ b/internal/httpapi/handlers_forwards.go @@ -4,6 +4,7 @@ import ( "encoding/json" "net/http" + "webui4frpc/internal/process" "webui4frpc/internal/store" ) @@ -74,10 +75,11 @@ func (h *Handler) startForward(local, remote string, port int) error { ln.Disabled = false if loc.LocalOnly { if h.Process != nil { - if _, has := h.Process.Status(remote); has { - _ = h.Process.Restart(remote) + key := process.WorkerKey(local, remote, port) + if _, has := h.Process.Status(key); has { + _ = h.Process.Restart(key) } else { - _ = h.Process.Start(remote) + _ = h.Process.Start(key) } } return nil @@ -109,8 +111,9 @@ func (h *Handler) stopForward(local, remote string, port int) error { } if loc.LocalOnly { if h.Process != nil { - if _, has := h.Process.Status(remote); has { - _ = h.Process.Restart(remote) + key := process.WorkerKey(local, remote, port) + if _, has := h.Process.Status(key); has { + _ = h.Process.Restart(key) } } return nil diff --git a/internal/httpapi/handlers_logs.go b/internal/httpapi/handlers_logs.go index fb83aaf..e6a34e9 100644 --- a/internal/httpapi/handlers_logs.go +++ b/internal/httpapi/handlers_logs.go @@ -6,6 +6,8 @@ import ( "net/http" "sync" "time" + + "webui4frpc/internal/process" ) // workerLog is one frpc worker's log tail on a node. @@ -34,8 +36,13 @@ func (h *Handler) localWorkerLogs() nodeLogsResp { for _, name := range h.Process.ListWorkers() { st, _ := h.Process.Status(name) lines, _ := tailFile(h.Process.LogPath(name), 64*1024) + // Parse the worker key to get the real remote name for display. + remoteName := name + if _, r, _, ok := process.ParseWorkerKey(name); ok { + remoteName = r + } out.Workers = append(out.Workers, workerLog{ - Name: name, Remote: name, State: st.State, Lines: lines, + Name: name, Remote: remoteName, State: st.State, Lines: lines, }) } return out diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index f2c9d37..eb53264 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -257,7 +257,7 @@ func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) { binary = h.BinaryPath() } for _, rv := range remotes { - st, has := h.Process.Status(rv.Name) + st, has := h.Process.RemoteStatus(rv.Name) fwd, _ := h.Store.LinksForRemote(rv.Name) p := profileStatus{ Name: rv.Name, Enabled: rv.Enabled, Status: st, HasProc: has, Forwards: fwd, @@ -271,10 +271,12 @@ func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) { } } // Prefer admin-derived conn state; fall back to log inference. + // In the per-forward model multiple workers may exist for one remote; + // use the first worker's log when available. if p.AdminEnabled && len(p.ProxyStates) > 0 { p.ConnState = connStateAdmin(p.ProxyStates) } else { - p.ConnState = connStateOf(h.Process.LogPath(rv.Name), rv.Enabled, st) + p.ConnState = connStateOf(firstWorkerLog(h.Process, rv.Name), rv.Enabled, st) } profiles = append(profiles, p) } @@ -300,11 +302,17 @@ func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) { } ts := make([]localTargetStatus, 0, len(targets)) for _, tg := range targets { - st, _ := h.Process.Status(tg.Remote) + // Per-forward model: this target's worker is the specific + // (local,remote,port) forward; active = its own process running. + st, has := h.Process.Status(process.WorkerKey(l.Name, tg.Remote, tg.RemotePort)) + state := "stopped" + if has { + state = st.State + } ts = append(ts, localTargetStatus{ Remote: tg.Remote, RemotePort: tg.RemotePort, - WorkerState: st.State, + WorkerState: state, }) } localStatuses = append(localStatuses, localStatus{Local: l, Targets: ts}) @@ -346,7 +354,8 @@ func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) { if loc.LocalOnly { fs.Kind = "local" fs.OwnerID = selfID - if st, has := h.Process.Status(ln.Remote); has { + // Per-forward model: check this exact forward's worker process. + if st, has := h.Process.Status(process.WorkerKey(ln.Local, ln.Remote, ln.RemotePort)); has { fs.Active = !ln.Disabled && st.State == "running" } } else { @@ -727,6 +736,16 @@ func frpcVersionOf(binPath string) string { return "" } +// firstWorkerLog returns the log path of the first per-forward worker of a +// remote (worker keys are "local~remote~port"), or "" when none exist. Used +// by status aggregation to infer frps connectivity from worker logs. +func firstWorkerLog(pm *process.Manager, remote string) string { + for _, key := range pm.RemoteWorkers(remote) { + return pm.LogPath(key) + } + return "" +} + // connStateOf derives the frps connection state from the local worker process // state plus a probe of its log. We never control the remote frps: the worker is // our local frpc; "connected" means frpc has successfully logged in to frps. diff --git a/internal/process/process.go b/internal/process/process.go index 5407b8b..29d755e 100644 --- a/internal/process/process.go +++ b/internal/process/process.go @@ -287,6 +287,88 @@ func (m *Manager) ListWorkers() []string { return out } +// RemoteWorkers returns the worker keys of all per-forward workers whose key +// parses to the given remote name. In the per-forward model every link has +// its own process keyed "local~remote~port"; this aggregates them so remote- +// centric callers (status page, profile endpoints) can report an overall +// view without knowing the key format. +func (m *Manager) RemoteWorkers(remote string) []string { + var out []string + for _, key := range m.ListWorkers() { + if _, r, _, ok := ParseWorkerKey(key); ok && r == remote { + out = append(out, key) + } + } + return out +} + +// RemoteStatus summarizes a remote's per-forward workers into one Status: +// running when ANY forward worker runs; crashed/starting/restarting take +// priority in that order; stopped (has=false) when no worker exists. +// This keeps the legacy per-remote status shape working on top of the +// per-forward worker model. +func (m *Manager) RemoteStatus(remote string) (Status, bool) { + keys := m.RemoteWorkers(remote) + if len(keys) == 0 { + return Status{State: "stopped"}, false + } + summary := Status{State: "stopped"} + worst := -1 // rank: running=0 < others=1 < crashed=2 + for _, k := range keys { + st, _ := m.Status(k) + rank := 1 + switch st.State { + case "running": + rank = 0 + case "crashed": + rank = 2 + } + if rank > worst || (rank == worst && summary.State == "stopped") { + if rank != worst || st.State != "stopped" { + summary = st + worst = rank + } + } + } + return summary, true +} + +// StartRemote starts every per-forward worker of one remote. Missing workers +// are skipped; the first error wins. +func (m *Manager) StartRemote(remote string) error { + var firstErr error + for _, key := range m.RemoteWorkers(remote) { + if err := m.Start(key); err != nil && firstErr == nil { + firstErr = err + } + } + return firstErr +} + +// RestartRemote restarts every per-forward worker belonging to one remote +// (used where the old model had a single per-remote worker). Missing workers +// are skipped; the first error wins. +func (m *Manager) RestartRemote(remote string) error { + var firstErr error + for _, key := range m.RemoteWorkers(remote) { + if err := m.Restart(key); err != nil && firstErr == nil { + firstErr = err + } + } + return firstErr +} + +// StopRemote stops every per-forward worker of one remote. +func (m *Manager) StopRemote(remote string) error { + var firstErr error + for _, key := range m.RemoteWorkers(remote) { + if err := m.Stop(key); err != nil && firstErr == nil { + firstErr = err + } + } + return firstErr +} + // StopAll stops all workers (used on manager shutdown). func (m *Manager) StopAll() { m.mu.Lock() diff --git a/internal/process/worker_key.go b/internal/process/worker_key.go new file mode 100644 index 0000000..e44d097 --- /dev/null +++ b/internal/process/worker_key.go @@ -0,0 +1,39 @@ +package process + +import ( + "fmt" + "strconv" + "strings" +) + +// WorkerKey is the unique identity of ONE forward's worker process: +// +// "~~" +// +// The per-forward worker model gives every link its own frpc process, config +// file and log file. This fixes the multi-node proxy-name fight of the old +// per-remote model (every node rendered the SAME proxy list and raced to +// register them on frps -> "proxy already exists"), and isolates faults: one +// crashing forward can no longer take its siblings down. +// +// '~' is a legal filename character on every supported OS (including Windows, +// unlike ':') and is rare in user-chosen node names. ParseWorkerKey rejects +// malformed keys so callers fail loudly instead of silently mis-rendering. +func WorkerKey(local, remote string, port int) string { + return fmt.Sprintf("%s~%s~%d", local, remote, port) +} + +// ParseWorkerKey splits a worker key back into its (local, remote, port) +// forward triple. ok=false for anything that is not a well-formed key +// (e.g. legacy per-remote worker names). +func ParseWorkerKey(key string) (local, remote string, port int, ok bool) { + parts := strings.Split(key, "~") + if len(parts) != 3 { + return "", "", 0, false + } + p, err := strconv.Atoi(parts[2]) + if err != nil || parts[0] == "" || parts[1] == "" || p <= 0 { + return "", "", 0, false + } + return parts[0], parts[1], p, true +} diff --git a/internal/process/worker_key_test.go b/internal/process/worker_key_test.go new file mode 100644 index 0000000..3b619f8 --- /dev/null +++ b/internal/process/worker_key_test.go @@ -0,0 +1,37 @@ +package process + +import "testing" + +func TestWorkerKeyRoundTrip(t *testing.T) { + cases := []struct { + local, remote string + port int + }{ + {"web", "frps1", 18080}, + {"db", "aliyun-frps", 5432}, + {"svc-1", "node.a.example", 80}, + } + for _, c := range cases { + key := WorkerKey(c.local, c.remote, c.port) + l, r, p, ok := ParseWorkerKey(key) + if !ok || l != c.local || r != c.remote || p != c.port { + t.Fatalf("roundtrip %q -> (%q,%q,%d,%v)", key, l, r, p, ok) + } + } +} + +func TestParseWorkerKeyRejectsBad(t *testing.T) { + bad := []string{ + "", "srv-a", // legacy per-remote names / empty + "a~b", // missing port + "a~b~x", // non-numeric port + "a~b~0", // port out of range + "~b~1", // empty local + "a~~1", // empty remote + } + for _, key := range bad { + if _, _, _, ok := ParseWorkerKey(key); ok { + t.Fatalf("ParseWorkerKey(%q) accepted, want reject", key) + } + } +}