diff --git a/internal/cluster/lo_addr.go b/internal/cluster/lo_addr.go
new file mode 100644
index 0000000..1ff7c23
--- /dev/null
+++ b/internal/cluster/lo_addr.go
@@ -0,0 +1,50 @@
+// LocalAddr helpers: resolve this node's routable LAN address used when a
+// forward targets a loopback (127.0.0.1/0.0.0.0/localhost) and is NOT marked
+// local-only — the task is distributed to the cluster, so the backend address
+// must be reachable from whichever node claims it.
+package cluster
+
+import (
+ "net"
+)
+
+// LanAddr returns this host's first non-loopback IPv4 address ("" if none can
+// be determined, e.g. offline/no interface).
+func LanAddr() string {
+ addrs, err := net.InterfaceAddrs()
+ if err != nil {
+ return ""
+ }
+ for _, a := range addrs {
+ if ipn, ok := a.(*net.IPNet); ok {
+ ip := ipn.IP
+ if ip.IsLoopback() || ip.To4() == nil {
+ continue
+ }
+ return ip.String()
+ }
+ }
+ return ""
+}
+
+// IsLoopbackIP reports whether addr is a loopback/unspecified/localhost value
+// that must be rewritten to a routable address before cluster distribution.
+func IsLoopbackIP(addr string) bool {
+ switch addr {
+ case "", "127.0.0.1", "0.0.0.0", "::1", "::", "localhost":
+ return true
+ }
+ return false
+}
+
+// RewriteForCluster returns addr if it is routable, or the LAN address when it
+// is loopback. Returns original when no LAN address is known (keeps ip as-is).
+func RewriteForCluster(addr string) string {
+ if !IsLoopbackIP(addr) {
+ return addr
+ }
+ if lan := LanAddr(); lan != "" {
+ return lan
+ }
+ return addr
+}
diff --git a/internal/cluster/ring.go b/internal/cluster/ring.go
index ecfe1a1..8f3451b 100644
--- a/internal/cluster/ring.go
+++ b/internal/cluster/ring.go
@@ -49,6 +49,10 @@ type Task struct {
Remote store.Remote `json:"remote"`
Link store.Link `json:"link"`
Created int64 `json:"created"`
+ // Revoke marks a REVOCATION task: instead of creating the forward, the
+ // owning node cancels it (stop worker, drop from topology). Reuses the
+ // same publish channel as creation (round-1 inject, round-2 apply).
+ Revoke bool `json:"revoke,omitempty"`
}
// TopoEntry is an ESTABLISHED forward in the cluster topology: who (OwnerID)
@@ -201,6 +205,25 @@ func (s *State) NextTaskID() string {
return fmt.Sprintf("t%d", s.Seq)
}
+// AddRevoke publishes a REVOCATION task (same channel as creation): when a
+// node wants to cancel an established forward, it injects a Revoke task;
+// the owning node cancels the worker and drops it from topology.
+func (s *State) AddRevoke(local store.Local, remote store.Remote, link store.Link) *Task {
+ t := &Task{
+ ID: s.NextTaskID(),
+ Local: local,
+ Remote: remote,
+ Link: link,
+ Created: time.Now().Unix(),
+ Revoke: true,
+ }
+ if s.PendingTasks == nil {
+ s.PendingTasks = map[string]*Task{}
+ }
+ s.PendingTasks[t.ID] = t
+ return t
+}
+
// AddPending attaches a NEW forward request to the token (round-1 inject).
func (s *State) AddPending(local store.Local, remote store.Remote, link store.Link) *Task {
t := &Task{
@@ -257,6 +280,18 @@ func (s *State) AddTopology(t *Task, ownerID string) *TopoEntry {
return e
}
+// RemoveTopology drops the active forward for the given local/remote/port
+// (returns true if removed — the owning node revokes it).
+func (s *State) RemoveTopology(local, remote string, port int) bool {
+ for id, e := range s.Topology {
+ if e.Local.Name == local && e.Remote.Name == remote && e.Link.RemotePort == port {
+ delete(s.Topology, id)
+ return true
+ }
+ }
+ return false
+}
+
// TopologyList returns active forwards, stable by task id.
func (s *State) TopologyList() []*TopoEntry {
out := make([]*TopoEntry, 0, len(s.Topology))
diff --git a/internal/cluster/ring_engine.go b/internal/cluster/ring_engine.go
index f43064e..5fadf9e 100644
--- a/internal/cluster/ring_engine.go
+++ b/internal/cluster/ring_engine.go
@@ -20,6 +20,7 @@ const (
// (create the frpc worker / forward). Injected to avoid import cycle.
type Handler interface {
Claim(ctx context.Context, tk *Task) error
+ Revoke(ctx context.Context, tk *Task) error
RuntimeLoad() Load
}
@@ -145,6 +146,23 @@ func (e *Engine) phase2(ctx context.Context, tk *Token) error {
if claimed == nil {
break
}
+ // Revocation task: the owning node cancels the forward (stop worker,
+ // drop from topology, log forward.remove). Idempotent if missing.
+ if claimed.Revoke {
+ if e.state.RemoveTopology(claimed.Local.Name, claimed.Remote.Name, claimed.Link.RemotePort) {
+ if e.Handler != nil {
+ if err := e.Handler.Revoke(ctx, claimed); err != nil {
+ log.Printf("ring[%s] revoke %s: %v", e.ID, claimed.ID, err)
+ }
+ }
+ if e.Log != nil {
+ _, _ = e.Log.Append(e.ID, LogForwardRemove, map[string]any{
+ "taskId": claimed.ID, "local": claimed.Local.Name, "remote": claimed.Remote.Name,
+ })
+ }
+ }
+ continue
+ }
if e.Handler != nil {
if err := e.Handler.Claim(ctx, claimed); err != nil {
e.state.PendingTasks[claimed.ID] = claimed
@@ -366,10 +384,35 @@ func (e *Engine) AdoptState(s State) {
// SubmitTask adds a new forward request to pending; it rides the next token
// round and is claimed by the lowest-load member.
+// RevokeTask publishes a revocation for an established forward through the
+// same token channel; the owning node stops the worker and drops topology.
+func (e *Engine) RevokeTask(local store.Local, remote store.Remote, link store.Link) *Task {
+ return e.state.AddRevoke(local, remote, link)
+}
+
func (e *Engine) SubmitTask(local store.Local, remote store.Remote, link store.Link) *Task {
+ if e.HasTask(local.Name, remote.Name, link.RemotePort) {
+ return nil
+ }
return e.state.AddPending(local, remote, link)
}
+// HasTask reports whether a forward with the same local/remote/remotePort is
+// already pending or active in the topology (idempotency guard for resaves).
+func (e *Engine) HasTask(local, remote string, port int) bool {
+ for _, t := range e.state.PendingList() {
+ if t.Local.Name == local && t.Remote.Name == remote && t.Link.RemotePort == port {
+ return true
+ }
+ }
+ for _, t := range e.state.TopologyList() {
+ if t.Local.Name == local && t.Remote.Name == remote && t.Link.RemotePort == port {
+ return true
+ }
+ }
+ return false
+}
+
// IsLeader reports whether this node is the current ring leader.
func (e *Engine) IsLeader() bool { return e.state.LeaderID == e.ID }
diff --git a/internal/cluster/ring_engine_test.go b/internal/cluster/ring_engine_test.go
index 6ab1a40..c5c921a 100644
--- a/internal/cluster/ring_engine_test.go
+++ b/internal/cluster/ring_engine_test.go
@@ -6,8 +6,16 @@ import (
)
type fakeHandler struct {
- load Load
- claim func(ctx context.Context, tk *Task) error
+ load Load
+ claim func(ctx context.Context, tk *Task) error
+ revoke func(ctx context.Context, tk *Task) error
+}
+
+func (h *fakeHandler) Revoke(ctx context.Context, tk *Task) error {
+ if h.revoke != nil {
+ return h.revoke(ctx, tk)
+ }
+ return nil
}
func (h *fakeHandler) RuntimeLoad() Load { return h.load }
diff --git a/internal/cluster/ring_handler.go b/internal/cluster/ring_handler.go
index f7861b6..86490b6 100644
--- a/internal/cluster/ring_handler.go
+++ b/internal/cluster/ring_handler.go
@@ -15,6 +15,8 @@ type AppHandler struct {
LoadFn func() (memPct, netPct float64)
// ClaimFn creates the forward on this node (persist + spawn worker).
ClaimFn func(ctx context.Context, tk *Task) error
+ // RevokeFn cancels the forward on this node (stop worker + drop from store).
+ RevokeFn func(ctx context.Context, tk *Task) error
}
// RuntimeLoad implements Handler.
@@ -34,6 +36,14 @@ func (a *AppHandler) Claim(ctx context.Context, tk *Task) error {
return a.ClaimFn(ctx, tk)
}
+// Revoke implements Handler.
+func (a *AppHandler) Revoke(ctx context.Context, tk *Task) error {
+ if a.RevokeFn == nil {
+ return nil
+ }
+ return a.RevokeFn(ctx, tk)
+}
+
// SampleMemLoad returns a cheap memory-usage percentage (0..100).
func SampleMemLoad() float64 {
var m runtime.MemStats
diff --git a/internal/cluster/ring_revoke_test.go b/internal/cluster/ring_revoke_test.go
new file mode 100644
index 0000000..0e5e8fc
--- /dev/null
+++ b/internal/cluster/ring_revoke_test.go
@@ -0,0 +1,64 @@
+package cluster
+
+import (
+ "context"
+ "testing"
+
+ "webui4frpc/internal/store"
+)
+
+// TestRevokeTaskRemovesTopology: a revoke task published to the ring removes
+// the forward from topology, calls Handler.Revoke, and logs forward.remove.
+func TestRevokeTaskRemovesTopology(t *testing.T) {
+ revoked := false
+ eng := NewEngine("n1", "n1:7500", "u", "p", "0.71.0", nil,
+ &fakeHandler{load: Load{MemPct: 5, NetPct: 5},
+ revoke: func(ctx context.Context, tk *Task) error { revoked = true; return nil }},
+ func(ctx context.Context, next string, tk *Token) error { return nil },
+ "n1:7500", true)
+
+ // establish a forward
+ eng.state.AddPending(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+ // lowest-load is n1 (only node): it claims + builds topology
+ if err := eng.phase2(context.Background(), &Token{Cycle: 1, Phase: PhaseSync, State: eng.state}); err != nil {
+ t.Fatal(err)
+ }
+ if len(eng.state.TopologyList()) != 1 {
+ t.Fatalf("topology after claim = %+v", eng.state.TopologyList())
+ }
+
+ // publish a revoke task pointing at the same forward
+ eng.state.AddRevoke(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+ if err := eng.phase2(context.Background(), &Token{Cycle: 2, Phase: PhaseSync, State: eng.state}); err != nil {
+ t.Fatal(err)
+ }
+ if len(eng.state.TopologyList()) != 0 {
+ t.Fatalf("topology after revoke = %+v", eng.state.TopologyList())
+ }
+ if !revoked {
+ t.Fatal("Handler.Revoke was not called")
+ }
+ // log should contain forward.remove
+ var sawRemove bool
+ for _, e := range eng.Log.Snapshot() {
+ if e.Kind == LogForwardRemove {
+ sawRemove = true
+ }
+ }
+ if !sawRemove {
+ t.Fatalf("log missing forward.remove: %+v", eng.Log.Snapshot())
+ }
+}
+
+// TestRevokeIdempotent: revoking an already-missing forward does not error.
+func TestRevokeIdempotent(t *testing.T) {
+ eng := newTestEngine("n1", true)
+ eng.state.AddRevoke(store.Local{Name: "ghost"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 1})
+ if err := eng.phase2(context.Background(), &Token{Cycle: 1, Phase: PhaseSync, State: eng.state}); err != nil {
+ t.Fatalf("revoke missing: %v", err)
+ }
+ // no topology entry, no panic
+ if len(eng.state.TopologyList()) != 0 {
+ t.Fatal("should be empty")
+ }
+}
diff --git a/internal/httpapi/handlers.go b/internal/httpapi/handlers.go
index 0e8c3a5..5a2ca8e 100644
--- a/internal/httpapi/handlers.go
+++ b/internal/httpapi/handlers.go
@@ -6,6 +6,7 @@ import (
"os"
"strings"
+ "webui4frpc/internal/cluster"
"webui4frpc/internal/store"
)
@@ -42,6 +43,16 @@ func (h *Handler) saveCanvas(w http.ResponseWriter, r *http.Request) {
}
s := h.Store
+ // Rewrite loopback backend addresses for cluster-distributed forwards.
+ // Non-localOnly locals targeting 127.0.0.1/0.0.0.0/localhost must be
+ // reachable from whichever node claims them, so swap in this node LAN addr.
+ // Local-only forwards keep the loopback as-is (they never leave this node).
+ for i := range canvas.Locals {
+ if !canvas.Locals[i].LocalOnly && cluster.IsLoopbackIP(canvas.Locals[i].IP) {
+ canvas.Locals[i].IP = cluster.RewriteForCluster(canvas.Locals[i].IP)
+ }
+ }
+
// Upsert locals.
for _, l := range canvas.Locals {
if err := s.UpsertLocal(l); err != nil {
@@ -49,7 +60,9 @@ func (h *Handler) saveCanvas(w http.ResponseWriter, r *http.Request) {
return
}
}
- // Delete locals not present.
+ // Delete locals not present. A removed localOnly forward is cancelled
+ // locally (stop worker). A removed cluster forward publishes a REVOKE
+ // task through the token so its owning node cancels it everywhere.
if existing, err := s.ListLocals(); err == nil {
keep := map[string]bool{}
for _, l := range canvas.Locals {
@@ -57,6 +70,19 @@ func (h *Handler) saveCanvas(w http.ResponseWriter, r *http.Request) {
}
for _, old := range existing {
if !keep[old.Name] {
+ if old.LocalOnly {
+ if h.Process != nil {
+ if fwd, _ := s.LinksForLocal(old.Name); len(fwd) > 0 {
+ _ = h.Process.Stop(fwd[0].Remote)
+ }
+ }
+ } else if h.Ring != nil {
+ if fwd, _ := s.LinksForLocal(old.Name); len(fwd) > 0 {
+ if rem, ok := s.GetRemote(fwd[0].Remote); ok {
+ h.Ring.RevokeTask(old, rem, store.Link{Local: old.Name, Remote: rem.Name, RemotePort: fwd[0].RemotePort})
+ }
+ }
+ }
_ = s.DeleteLocal(old.Name)
}
}
@@ -87,7 +113,34 @@ func (h *Handler) saveCanvas(w http.ResponseWriter, r *http.Request) {
return
}
- // Restart affected running workers so changes take effect immediately.
+ // Cluster distribution: non-localOnly forwards are submitted as token-ring
+ // tasks (claimed by the lowest-load member). Local-only forwards are NOT
+ // submitted — they stay on this node and are only visible here.
+ if h.Ring != nil {
+ for _, l := range canvas.Locals {
+ if l.LocalOnly {
+ continue
+ }
+ for _, ln := range canvas.Links {
+ if ln.Local != l.Name {
+ continue
+ }
+ var rem store.Remote
+ for _, rr := range canvas.Remotes {
+ if rr.Name == ln.Remote {
+ rem = rr
+ break
+ }
+ }
+ if rem.Name != "" {
+ h.Ring.SubmitTask(l, rem, ln)
+ }
+ }
+ }
+ }
+
+ // Restart affected running workers so changes take effect immediately
+ // (local-only forwards get their worker started right here).
if h.SyncWorkers != nil {
h.SyncWorkers()
}
diff --git a/internal/httpapi/server_test.go b/internal/httpapi/server_test.go
index f329bd7..8e66f30 100644
--- a/internal/httpapi/server_test.go
+++ b/internal/httpapi/server_test.go
@@ -2,6 +2,7 @@ package httpapi
import (
"bytes"
+ "context"
"encoding/json"
"net/http"
"net/http/httptest"
@@ -10,6 +11,7 @@ import (
"strings"
"testing"
+ "webui4frpc/internal/cluster"
"webui4frpc/internal/process"
"webui4frpc/internal/store"
)
@@ -192,3 +194,58 @@ func TestFetchProxyStatesFromAdminAPI(t *testing.T) {
t.Fatalf("connStateAdmin(wait start) = %q, want connecting", got)
}
}
+
+// TestSaveCanvasPublishesRevokeTask: deleting a non-localOnly forward from the
+// canvas publishes a REVOKE task to the ring (owner cancels it cluster-wide).
+func TestSaveCanvasPublishesRevokeTask(t *testing.T) {
+ dir := t.TempDir()
+ st, err := store.New(filepath.Join(dir, "test.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer st.Close()
+ pm := process.NewManager(process.Options{
+ ConfigsDir: filepath.Join(dir, "configs"), LogsDir: filepath.Join(dir, "logs"),
+ BinaryPath: func() string { return "" }, Render: func(string) ([]byte, error) { return []byte(`{}`), nil },
+ AutoRestart: func(string) bool { return false }, RestartInterval: func() int { return 5 },
+ })
+ ring := cluster.NewEngine("n1", "n1:7500", "u", "p", "0.1.0", nil,
+ &cluster.AppHandler{}, func(ctx context.Context, next string, tk *cluster.Token) error { return nil },
+ "n1:7500", true)
+ h := &Handler{Store: st, Process: pm, WorkDir: dir, User: "admin", Password: "pw", Ring: ring}
+ mux, _ := NewServeMux(h)
+ ts := httptest.NewServer(mux)
+ defer ts.Close()
+
+ // establish a cluster forward first via save (non-localOnly)
+ body := `{"locals":[{"name":"web","ip":"127.0.0.1","port":8080,"protocol":"tcp"}],
+ "remotes":[{"name":"frps1","ip":"10.0.0.1","port":7000,"enabled":true}],
+ "links":[{"local":"web","remote":"frps1","remotePort":18081}]}`
+ req, _ := http.NewRequest(http.MethodPut, ts.URL+"/api/manager/canvas", strings.NewReader(body))
+ req.SetBasicAuth("admin", "pw")
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ resp.Body.Close()
+
+ // now save canvas WITHOUT the forward -> should publish revoke
+ body2 := `{"locals":[],"remotes":[],"links":[]}`
+ req2, _ := http.NewRequest(http.MethodPut, ts.URL+"/api/manager/canvas", strings.NewReader(body2))
+ req2.SetBasicAuth("admin", "pw")
+ resp2, err := http.DefaultClient.Do(req2)
+ if err != nil {
+ t.Fatal(err)
+ }
+ resp2.Body.Close()
+ // Ring should now have a pending REVOKE task for web->frps1
+ var sawRevoke bool
+ for _, tk := range ring.State().PendingList() {
+ if tk.Revoke && tk.Local.Name == "web" {
+ sawRevoke = true
+ }
+ }
+ if !sawRevoke {
+ t.Fatalf("expected revoke task for web, pending=%+v", ring.State().PendingList())
+ }
+}
diff --git a/internal/store/store.go b/internal/store/store.go
index 8e53e22..406494d 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -41,6 +41,10 @@ type Local struct {
// The frps load balances the group's proxies (many-to-one backend set).
LBGroup string `json:"lbGroup,omitempty"`
LBGroupKey string `json:"lbGroupKey,omitempty"`
+ // LocalOnly: when true the forward is created directly on this node
+ // (loopback stays 127.0.0.1), it does not enter the cluster token/topology,
+ // and is only visible in this node webui.
+ LocalOnly bool `json:"localOnly,omitempty"`
// Health check (frpc healthCheck): tcp probes LocalIP:LocalPort or an
// http GET to LocalIP:LocalPort + HealthCheckPath. When a group member
@@ -114,6 +118,7 @@ CREATE TABLE IF NOT EXISTS locals (
ip TEXT NOT NULL,
port INTEGER NOT NULL,
protocol TEXT NOT NULL DEFAULT 'tcp',
+ local_only INTEGER NOT NULL DEFAULT 0,
use_encryption INTEGER NOT NULL DEFAULT 0,
use_compression INTEGER NOT NULL DEFAULT 0,
bandwidth_limit TEXT NOT NULL DEFAULT '',
@@ -222,6 +227,7 @@ func (s *Store) migrate() error {
"health_check_timeout INTEGER NOT NULL DEFAULT 0",
"health_check_max_failed INTEGER NOT NULL DEFAULT 0",
"health_check_interval INTEGER NOT NULL DEFAULT 0",
+ "local_only INTEGER NOT NULL DEFAULT 0",
},
"remotes": {
"transport_protocol TEXT NOT NULL DEFAULT ''",
@@ -270,7 +276,7 @@ func (s *Store) Close() error { return s.db.Close() }
// ---- Locals ----
func (s *Store) ListLocals() ([]Local, error) {
- rows, err := s.db.Query("SELECT name, ip, port, protocol, use_encryption, use_compression, bandwidth_limit, pool_count, metadatas, annotations, custom_domains, subdomain, locations, host_header_rewrite, http_headers, basic_auth_user, basic_auth_password, lb_group, lb_group_key, health_check_type, health_check_path, health_check_timeout, health_check_max_failed, health_check_interval FROM locals ORDER BY name")
+ rows, err := s.db.Query("SELECT name, ip, port, protocol, local_only, use_encryption, use_compression, bandwidth_limit, pool_count, metadatas, annotations, custom_domains, subdomain, locations, host_header_rewrite, http_headers, basic_auth_user, basic_auth_password, lb_group, lb_group_key, health_check_type, health_check_path, health_check_timeout, health_check_max_failed, health_check_interval FROM locals ORDER BY name")
if err != nil {
return nil, err
}
@@ -278,13 +284,14 @@ func (s *Store) ListLocals() ([]Local, error) {
var out []Local
for rows.Next() {
var l Local
- var enc, comp int
+ var enc, comp, localOnly int
var metaRaw, annoRaw, hdrRaw, locRaw string
- if err := rows.Scan(&l.Name, &l.IP, &l.Port, &l.Protocol, &enc, &comp, &l.BandwidthLimit, &l.PoolCount, &metaRaw, &annoRaw, &l.CustomDomains, &l.SubDomain, &locRaw, &l.HostHeaderRewrite, &hdrRaw, &l.BasicAuthUser, &l.BasicAuthPassword, &l.LBGroup, &l.LBGroupKey, &l.HealthCheckType, &l.HealthCheckPath, &l.HealthCheckTimeout, &l.HealthCheckMaxFailed, &l.HealthCheckInterval); err != nil {
+ if err := rows.Scan(&l.Name, &l.IP, &l.Port, &l.Protocol, &localOnly, &enc, &comp, &l.BandwidthLimit, &l.PoolCount, &metaRaw, &annoRaw, &l.CustomDomains, &l.SubDomain, &locRaw, &l.HostHeaderRewrite, &hdrRaw, &l.BasicAuthUser, &l.BasicAuthPassword, &l.LBGroup, &l.LBGroupKey, &l.HealthCheckType, &l.HealthCheckPath, &l.HealthCheckTimeout, &l.HealthCheckMaxFailed, &l.HealthCheckInterval); err != nil {
return nil, err
}
l.UseEncryption = enc != 0
l.UseCompression = comp != 0
+ l.LocalOnly = localOnly != 0
l.Metadatas = decodeMap(metaRaw)
l.Annotations = decodeMap(annoRaw)
l.Locations = decodeSlice(locRaw)
@@ -296,14 +303,15 @@ func (s *Store) ListLocals() ([]Local, error) {
func (s *Store) GetLocal(name string) (Local, bool) {
var l Local
- var enc, comp int
+ var enc, comp, localOnly int
var metaRaw, annoRaw, hdrRaw, locRaw string
- row := s.db.QueryRow("SELECT name, ip, port, protocol, use_encryption, use_compression, bandwidth_limit, pool_count, metadatas, annotations, custom_domains, subdomain, locations, host_header_rewrite, http_headers, basic_auth_user, basic_auth_password, lb_group, lb_group_key, health_check_type, health_check_path, health_check_timeout, health_check_max_failed, health_check_interval FROM locals WHERE name = ?", name)
- if err := row.Scan(&l.Name, &l.IP, &l.Port, &l.Protocol, &enc, &comp, &l.BandwidthLimit, &l.PoolCount, &metaRaw, &annoRaw, &l.CustomDomains, &l.SubDomain, &locRaw, &l.HostHeaderRewrite, &hdrRaw, &l.BasicAuthUser, &l.BasicAuthPassword, &l.LBGroup, &l.LBGroupKey, &l.HealthCheckType, &l.HealthCheckPath, &l.HealthCheckTimeout, &l.HealthCheckMaxFailed, &l.HealthCheckInterval); err != nil {
+ row := s.db.QueryRow("SELECT name, ip, port, protocol, local_only, use_encryption, use_compression, bandwidth_limit, pool_count, metadatas, annotations, custom_domains, subdomain, locations, host_header_rewrite, http_headers, basic_auth_user, basic_auth_password, lb_group, lb_group_key, health_check_type, health_check_path, health_check_timeout, health_check_max_failed, health_check_interval FROM locals WHERE name = ?", name)
+ if err := row.Scan(&l.Name, &l.IP, &l.Port, &l.Protocol, &localOnly, &enc, &comp, &l.BandwidthLimit, &l.PoolCount, &metaRaw, &annoRaw, &l.CustomDomains, &l.SubDomain, &locRaw, &l.HostHeaderRewrite, &hdrRaw, &l.BasicAuthUser, &l.BasicAuthPassword, &l.LBGroup, &l.LBGroupKey, &l.HealthCheckType, &l.HealthCheckPath, &l.HealthCheckTimeout, &l.HealthCheckMaxFailed, &l.HealthCheckInterval); err != nil {
return Local{}, false
}
l.UseEncryption = enc != 0
l.UseCompression = comp != 0
+ l.LocalOnly = localOnly != 0
l.Metadatas = decodeMap(metaRaw)
l.Annotations = decodeMap(annoRaw)
l.Locations = decodeSlice(locRaw)
@@ -313,9 +321,9 @@ func (s *Store) GetLocal(name string) (Local, bool) {
func (s *Store) UpsertLocal(l Local) error {
_, err := s.db.Exec(
- "INSERT INTO locals(name, ip, port, protocol, use_encryption, use_compression, bandwidth_limit, pool_count, metadatas, annotations, custom_domains, subdomain, locations, host_header_rewrite, http_headers, basic_auth_user, basic_auth_password, lb_group, lb_group_key, health_check_type, health_check_path, health_check_timeout, health_check_max_failed, health_check_interval) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) "+
- "ON CONFLICT(name) DO UPDATE SET ip=excluded.ip, port=excluded.port, protocol=excluded.protocol, use_encryption=excluded.use_encryption, use_compression=excluded.use_compression, bandwidth_limit=excluded.bandwidth_limit, pool_count=excluded.pool_count, metadatas=excluded.metadatas, annotations=excluded.annotations, custom_domains=excluded.custom_domains, subdomain=excluded.subdomain, locations=excluded.locations, host_header_rewrite=excluded.host_header_rewrite, http_headers=excluded.http_headers, basic_auth_user=excluded.basic_auth_user, basic_auth_password=excluded.basic_auth_password, lb_group=excluded.lb_group, lb_group_key=excluded.lb_group_key, health_check_type=excluded.health_check_type, health_check_path=excluded.health_check_path, health_check_timeout=excluded.health_check_timeout, health_check_max_failed=excluded.health_check_max_failed, health_check_interval=excluded.health_check_interval",
- l.Name, l.IP, l.Port, l.Protocol, boolToInt(l.UseEncryption), boolToInt(l.UseCompression), l.BandwidthLimit, l.PoolCount, encodeMap(l.Metadatas), encodeMap(l.Annotations), l.CustomDomains, l.SubDomain, encodeSlice(l.Locations), l.HostHeaderRewrite, encodeMap(l.HTTPHeaders), l.BasicAuthUser, l.BasicAuthPassword, l.LBGroup, l.LBGroupKey, l.HealthCheckType, l.HealthCheckPath, l.HealthCheckTimeout, l.HealthCheckMaxFailed, l.HealthCheckInterval,
+ "INSERT INTO locals(name, ip, port, protocol, local_only, use_encryption, use_compression, bandwidth_limit, pool_count, metadatas, annotations, custom_domains, subdomain, locations, host_header_rewrite, http_headers, basic_auth_user, basic_auth_password, lb_group, lb_group_key, health_check_type, health_check_path, health_check_timeout, health_check_max_failed, health_check_interval) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) "+
+ "ON CONFLICT(name) DO UPDATE SET ip=excluded.ip, port=excluded.port, protocol=excluded.protocol, local_only=excluded.local_only, use_encryption=excluded.use_encryption, use_compression=excluded.use_compression, bandwidth_limit=excluded.bandwidth_limit, pool_count=excluded.pool_count, metadatas=excluded.metadatas, annotations=excluded.annotations, custom_domains=excluded.custom_domains, subdomain=excluded.subdomain, locations=excluded.locations, host_header_rewrite=excluded.host_header_rewrite, http_headers=excluded.http_headers, basic_auth_user=excluded.basic_auth_user, basic_auth_password=excluded.basic_auth_password, lb_group=excluded.lb_group, lb_group_key=excluded.lb_group_key, health_check_type=excluded.health_check_type, health_check_path=excluded.health_check_path, health_check_timeout=excluded.health_check_timeout, health_check_max_failed=excluded.health_check_max_failed, health_check_interval=excluded.health_check_interval",
+ l.Name, l.IP, l.Port, l.Protocol, boolToInt(l.LocalOnly), boolToInt(l.UseEncryption), boolToInt(l.UseCompression), l.BandwidthLimit, l.PoolCount, encodeMap(l.Metadatas), encodeMap(l.Annotations), l.CustomDomains, l.SubDomain, encodeSlice(l.Locations), l.HostHeaderRewrite, encodeMap(l.HTTPHeaders), l.BasicAuthUser, l.BasicAuthPassword, l.LBGroup, l.LBGroupKey, l.HealthCheckType, l.HealthCheckPath, l.HealthCheckTimeout, l.HealthCheckMaxFailed, l.HealthCheckInterval,
)
return err
}
diff --git a/plan.md b/plan.md
index 9182974..0266ba1 100644
--- a/plan.md
+++ b/plan.md
@@ -118,6 +118,22 @@
- 集群状态（拓扑/任务/leader）可**由日志重放得到**：新节点加入时，通过令牌/邻居拉取完整日志并重放，即可获得与其它节点一致的完整视图（满足"每个节点都掌握完整集群信息"）。
- 该机制与"令牌承载中间配置文件/转发拓扑"正交共存：令牌同时携带〔拓扑快照〕与〔日志增量〕，快照用于即时校验，日志用于一致性追补。
+#### 任务撤销（复用任务发布通道，撤销语义）
+
+- 撤销一个转发 = **发布一个"撤销任务"到令牌**：与任务发布走同一基础通道（round-1 注入、round-2 执行），只是 Type 为撤销、载荷为〔任务 ID + 该转发中间配置〕。
+- 令牌环行时，**持有该转发的节点**（其 owner 记录在拓扑中）收到撤销任务后：取消本地 frpc worker → 从活跃拓扑移除该转发 → 记 `forward.remove` 增量日志，全网随之下一次轮次收敛一致。
+- 撤销任务幂等：若转发已不在拓扑（已被撤销/离线重挂），收到撤销仅记日志、不报错。
+- localOnly 的本地转发撤销不发布到令牌，直接本机取消（因本机即 owner，仅本节点可见）。
+
+#### 画布差异判断（diff）与命令追加
+
+- **在哪个节点修改 webui 画布，就由该节点进行差异判断**：将该节点画布的「期望状态」与当前集群拓扑（该节点持有的完整视图）比对。
+- 差异结果转为**一批转发/撤销命令，追加到令牌任务**：
+ - 画布有、拓扑无 → 追加**新增转发命令**（任务+中间配置）→ 后续由负载最低者摘取创建。
+ - 画布无、拓扑有 → 追加**撤销转发命令**（撤任务+转发的中间配置）→ 持有者收到后取消 worker 并移出拓扑。
+- 画布即唯一编辑入口：编辑节点只发命令，不直接改其它节点；拓扑随令牌轮次收敛全网一致。
+- **其它节点根据集群拓扑生成本地画布存储**：非编辑节点收到令牌同步的完整拓扑后，把拓扑中的转发（locals/remotes/links）重建为本地画布视图，叠加本节点的 localOnly 项。故所有节点的画布展示一致（都源自拓扑），编辑只在入口节点生效。
+
#### 实现里程碑（待按上述设计重建）
- [x] 数据面：Ring State（节点表/转发链/leader/轮次延迟/负载/待办任务）+ 日志（追加式 + 水位）
diff --git a/web/src/components/LocalNode.vue b/web/src/components/LocalNode.vue
index 732a7be..07ef5f8 100644
--- a/web/src/components/LocalNode.vue
+++ b/web/src/components/LocalNode.vue
@@ -27,6 +27,10 @@
+
@@ -106,6 +110,7 @@ const nameField = field('name')
const ipField = field('ip')
const portField = field('port')
const protocolField = field('protocol')
+const localOnlyField = field('localOnly')
const useEncryptionField = field('useEncryption')
const useCompressionField = field('useCompression')
const bandwidthLimitField = field('bandwidthLimit')
diff --git a/web/src/types.ts b/web/src/types.ts
index f5e262e..b86468e 100644
--- a/web/src/types.ts
+++ b/web/src/types.ts
@@ -13,6 +13,8 @@ export interface Local {
ip: string;
port: number;
protocol: string; // tcp | udp | http | https
+ // localOnly: forward stays on this node (loopback kept), no cluster.
+ localOnly?: boolean;
// M1 advanced transport knobs (optional; empty/zero = frpc defaults)
useEncryption?: boolean;