feat(M6): localOnly forward option (loopback→LAN rewrite for cluster, local direct), canvas diff-based create/revoke commands, revoke task semantics, topology derived canvas on sync nodes, LAN addr helper

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
+}

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))

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 }
 

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 }

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

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")
+	}
+}