feat(cluster): 负载均衡迁移 —— 定向投递 + 两阶段提交

承接 875e580（只落决策逻辑）。本提交补齐运行时：周期检查、迟滞判定、
两阶段迁移，以及让迁移任务**定向**到指定节点。

## 为什么必须是定向的（核心安全点）

loadSnapshot() 的 Forwards 来自 ForwardsOwnedBy，所以旧 owner 一旦被摘掉条目，
它立刻变成最闲节点。而普通创建任务是"任何最闲者认领"—— 于是它会把刚迁走的
转发**当场抢回**，迁移变成"只重启一次 worker"的空操作。

故 Task 新增 Migrate + Target 字段，runCommands 的选择循环照 Restart 的范式
处理：Target == 自己才认领，否则 `continue` 让任务随环前进。

## 为什么必须两阶段（不能一次发 revoke + submit）

1. 旧 owner 变最闲后会把未定向的任务抢回（上面那条）；
2. PendingList 按 `t<ID>` **字符串**排序（`t10 < t5`），同轮内两个任务的先后
   不可控。

Balancer 因此是显式状态机：phase1 发布 revoke 并等待旧条目真正消失，确认后才
发布 phase2 的定向 migrate。两个超时兜底：revoke 超时则放弃；migrate 超时
（target 没认领）则退回**无定向** SubmitTask，让"任何最闲者"接手 —— 转发不会
被丢，只是延迟。这与环本身的行为一致（实测：只要还有更闲的节点，pending 任务
最终会被认领；只有"全员都更忙"时才滞留）。

## 两个被测试逼出来的设计修正

- **绝对下限 minForwards=2**：只有"超均值倍数"在小集群失真——3 节点 1 条转发时
  owner Score≈105 而两台空闲≈5，20 倍于均值会触发迁移，但那是完全正常的分布，
  迁它只会断一条活跃连接。倍数负责看形状，下限负责"有没有东西值得搬"。
- **Migrate 也要绕过 duplicate-claim 守卫**，否则旧条目还在时目标节点会拒绝
  认领（这条守卫正是为了防止同一 remotePort 注册两次 → frps 报
  "proxy already exists"）。

## 测试（本项目第六次假绿的教训）

新增 6 条，全绿。做双向验证时发现
TestMigrateTaskIsAddressedNotClaimedByLightest **删掉定向分支后仍然 PASS**，
查因：`loadSnapshot()每次 OnToken 都用 Handler.RuntimeLoad() 刷新 mem/net，
我手工设的 State.Nodes[].Load 被静默覆盖 ⇒ n2/n3 始终平手 ⇒ "最闲者"恰好
就是 target，定向性从未被验证。

改为**用三个 Handler 的不同 RuntimeLoad 制造真实差异**（n3 最闲、n2 是较重的
target）后，删掉定向分支如期变红：

    n3 (the LIGHTEST node) claimed 1 migrate task(s); the task is addressed
    to n2 and must ignore load entirely

⇒ 这条纪律再次生效：测试必须制造出能让错误实现暴露的**不对称**，否则"恰好通过"
比"失败"更危险。

go build / go vet / go test ./... 全绿，gofmt 干净。

internal/cluster/balance_test.go

@@ -1,6 +1,7 @@
 package cluster
 
 import (
+	"context"
 	"encoding/json"
 	"testing"
 
@@ -144,3 +145,212 @@ func TestPlanBalanceIgnoresDeadNodes(t *testing.T) {
 		t.Fatal("chose an OFFLINE node as the migration destination")
 	}
 }
+
+// migrateRing builds a 3-node ring where n1 owns every forward, plus n2/n3 idle.
+func migrateRing(t *testing.T, ownCount int) (*Engine, *Engine, *Engine) {
+	t.Helper()
+	mk := func(id string, leader bool) *Engine {
+		h := &fakeHandler{load: Load{MemPct: 5, NetPct: 5}}
+		return NewEngine(id, id+":7500", "u", "p", "0.1.3", nil, h,
+			func(ctx context.Context, next string, tk *Token) error { return nil },
+			id+":7500", leader, "")
+	}
+	n1, n2, n3 := mk("n1", true), mk("n2", false), mk("n3", false)
+
+	seed := State{
+		LeaderID: "n1",
+		Nodes: []Node{
+			{ID: "n1", Addr: "n1:7500", Alive: true, IsLeader: true, Load: Load{Forwards: ownCount, MemPct: 5}},
+			{ID: "n2", Addr: "n2:7500", Alive: true, Load: Load{MemPct: 5}},
+			{ID: "n3", Addr: "n3:7500", Alive: true, Load: Load{MemPct: 5}},
+		},
+		Topology:     map[string]*TopoEntry{},
+		PendingTasks: map[string]*Task{},
+	}
+	for _, e := range []*Engine{n1, n2, n3} {
+		var own State
+		b, _ := json.Marshal(seed)
+		if err := json.Unmarshal(b, &own); err != nil {
+			t.Fatal(err)
+		}
+		e.AdoptState(own)
+	}
+	// n1 claims `ownCount` forwards.
+	for i := 0; i < ownCount; i++ {
+		n1.state.AddTopology(&Task{
+			ID:    "f" + string(rune('a'+i)),
+			Local: store.Local{Name: "web"}, Remote: store.Remote{Name: "frps1"},
+			Link: store.Link{RemotePort: 100 + i},
+		}, "n1")
+	}
+	var snap State
+	b, _ := json.Marshal(n1.state)
+	_ = json.Unmarshal(b, &snap)
+	for _, e := range []*Engine{n1, n2, n3} {
+		e.AdoptState(snap)
+	}
+	return n1, n2, n3
+}
+
+// TestMigrateTaskIsAddressedNotClaimedByLightest is the core safety test.
+//
+// After the revoke lands, the node we took the forward from becomes the
+// lightest — an untargeted creation task would be claimed straight back, making
+// the migration a no-op that merely restarts the worker. Migrate tasks must be
+// delivered to their named target only.
+//
+// The forward is removed from the topology first (as the balancer's phase 1
+// does), so this exercises the realistic post-revoke state.
+func TestMigrateTaskIsAddressedNotClaimedByLightest(t *testing.T) {
+	n1, n2, n3 := migrateRing(t, 3)
+	n1.State().RemoveTopology("web", "frps1", 100) // phase 1 has landed
+
+	n1.submitMigrate(store.Local{Name: "web"}, store.Remote{Name: "frps1"},
+		store.Link{RemotePort: 100}, "n2")
+
+	// Give each node a DISTINCT load via its handler. loadSnapshot() refreshes
+	// mem/net from Handler.RuntimeLoad() on every OnToken, so seeding
+	// State.Nodes[].Load by hand is silently overwritten — which is why an
+	// earlier version of this test had n2 and n3 tied and therefore passed even
+	// with the targeted delivery removed (the "lightest" node happened to be the
+	// target). n3 is now strictly the lightest, so an untargeted rule would hand
+	// the task to n3 and fail here.
+	claimedBy := map[string]int{}
+	mk := func(id string, mem float64) Handler {
+		return &fakeHandler{load: Load{MemPct: mem, NetPct: 0},
+			claim: func(ctx context.Context, tk *Task) error {
+				claimedBy[id]++
+				return nil
+			}}
+	}
+	n1.Handler = mk("n1", 50)
+	n2.Handler = mk("n2", 40) // the TARGET — heavier than n3
+	n3.Handler = mk("n3", 1)  // strictly the lightest
+	var snap State
+	b, _ := json.Marshal(n1.State())
+	_ = json.Unmarshal(b, &snap)
+	for _, e := range []*Engine{n1, n2, n3} {
+		e.AdoptState(snap)
+	}
+	for i := 0; i < 3; i++ {
+		for _, e := range []*Engine{n1, n2, n3} {
+			if _, err := e.OnToken(context.Background(), &Token{
+				Cycle: int64(i + 1), State: *e.State(), SentAt: int64(i + 1),
+			}); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+	if claimedBy["n1"] != 0 {
+		t.Errorf("n1 claimed %d migrate task(s); the old owner must never take it back",
+			claimedBy["n1"])
+	}
+	if claimedBy["n3"] != 0 {
+		t.Errorf("n3 (the LIGHTEST node) claimed %d migrate task(s); the task is addressed "+
+			"to n2 and must ignore load entirely", claimedBy["n3"])
+	}
+	if claimedBy["n2"] != 1 {
+		t.Errorf("the target n2 claimed %d time(s), want exactly 1", claimedBy["n2"])
+	}
+}
+
+// TestMigrateRefusedWhileEntryStillOwned pins the duplicate guard: a migrate that
+// arrives before the revoke completed must NOT create a second topology entry
+// for the same forward. This is the failure that would otherwise leave two nodes
+// running workers for one remotePort — which frps rejects with
+// "proxy already exists" (observed 14k times before that was fixed).
+func TestMigrateRefusedWhileEntryStillOwned(t *testing.T) {
+	n1, _, _ := migrateRing(t, 3)
+	claimed := 0
+	n1.Handler = &fakeHandler{load: Load{MemPct: 5, NetPct: 5},
+		claim: func(ctx context.Context, tk *Task) error { claimed++; return nil }}
+
+	// Entry still owned by n1 — the revoke has NOT landed.
+	n1.submitMigrate(store.Local{Name: "web"}, store.Remote{Name: "frps1"},
+		store.Link{RemotePort: 100}, "n1")
+
+	if _, err := n1.OnToken(context.Background(), &Token{Cycle: 1, State: *n1.State(), SentAt: 1}); err != nil {
+		t.Fatal(err)
+	}
+	if claimed != 0 {
+		t.Errorf("Claim ran %d time(s) while the forward was still owned; "+
+			"this would create a duplicate proxy", claimed)
+	}
+	if n := len(n1.State().TopologyList()); n != 3 {
+		t.Errorf("topology has %d entries, want 3 (no duplicate created)", n)
+	}
+}
+
+// TestBalancerWaitsForRevokeBeforePublishing covers the two-phase guarantee: no
+// migrate task may exist while the old entry is still present, because that is
+// what would let the old owner race the new destination.
+func TestBalancerWaitsForRevokeBeforePublishing(t *testing.T) {
+	n1, _, _ := migrateRing(t, 3)
+	b := NewBalancer(n1, 2, 0)
+
+	// First step starts phase 1 (revoke) and publishes nothing else.
+	_ = b.Step(1000)
+	if b.revoking == nil {
+		t.Fatal("first step should have started a revoke")
+	}
+	for _, tk := range n1.State().PendingList() {
+		if tk.Migrate {
+			t.Fatal("a migrate task was published before the revoke completed")
+		}
+	}
+
+	// Phase 1 completes (the owner drops the entry) → phase 2 publishes.
+	n1.State().RemoveTopology("web", "frps1", 100)
+	_ = b.Step(1010)
+	if b.revoking != nil {
+		t.Error("revoke phase should have finished")
+	}
+	found := false
+	for _, tk := range n1.State().PendingList() {
+		if tk.Migrate && tk.Target != "" {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatal("after the revoke completed, a targeted migrate task should exist")
+	}
+}
+
+// TestBalancerHysteresis: an imbalance must persist before anything moves, so a
+// momentary CPU spike never restarts a worker.
+func TestBalancerHysteresis(t *testing.T) {
+	n1, _, _ := migrateRing(t, 3)
+	b := NewBalancer(n1, 2, 600) // must be over for 10 minutes
+
+	_ = b.Step(1000) // first over-threshold observation: only starts the timer
+	for _, tk := range n1.State().PendingList() {
+		if tk.Revoke || tk.Migrate {
+			t.Fatal("acted before the sustained window elapsed")
+		}
+	}
+	// Still inside the window.
+	_ = b.Step(1300)
+	if b.revoking != nil {
+		t.Fatal("acted 5 minutes in; the window is 10")
+	}
+	// Past the window.
+	_ = b.Step(1700)
+	if b.revoking == nil {
+		t.Fatal("did not act after the imbalance persisted past the window")
+	}
+}
+
+// TestBalancerQuietWhenBalanced: the threshold means a healthy ring is never
+// touched, which is the whole reason for not running this every hour blindly.
+func TestBalancerQuietWhenBalanced(t *testing.T) {
+	n1, _, _ := migrateRing(t, 1)
+	b := NewBalancer(n1, 2, 0)
+	for i := 0; i < 5; i++ {
+		if msg := b.Step(int64(1000 + i*600)); msg != "" {
+			t.Fatalf("a balanced ring produced action: %q", msg)
+		}
+	}
+	if b.InFlight() {
+		t.Error("balancer started work on a balanced ring")
+	}
+}

internal/cluster/ring.go

@@ -68,6 +68,13 @@ type Task struct {
 	// duplicate-claim guard would discard it. The owner applies this one
 	// unconditionally: re-mark the entry enabled and (re)spawn the worker.
 	Restart bool `json:"restart,omitempty"`
+	// Migrate marks a LOAD-BALANCING task and Target names the node that must take
+	// the forward over. Unlike a plain creation task — which any lowest-load node
+	// may claim — a migrate is addressed: it exists precisely because the node that
+	// is currently lightest (usually the node we just took the forward FROM) would
+	// otherwise claim it straight back, undoing the migration.
+	Migrate bool   `json:"migrate,omitempty"`
+	Target  string `json:"target,omitempty"`
 	// RemoveNode: node ID to remove from the ring; the node self-removes when
 	// the command reaches it via the token.
 	RemoveNode string `json:"removeNode,omitempty"`

internal/cluster/ring_balance.go

@@ -1,5 +1,12 @@
 package cluster
 
+import (
+	"log"
+	"strconv"
+
+	"webui4frpc/internal/store"
+)
+
 // fwdKey identifies a forward for balancing purposes.
 type fwdKey struct {
 	local, remote string
@@ -20,11 +27,23 @@ type fwdKey struct {
 // BalanceTracker) can pass its bookkeeping through without changing this
 // signature, but this function itself is stateless about time: it answers
 // "given the current snapshot, what (if anything) should move right now".
+// minForwards is an absolute floor: never migrate unless the overloaded node
+// holds at least this many forwards.
+//
+// The ratio test alone is misleading on small rings. With 3 nodes and a single
+// forward, the owner scores ~105 (Forwards*100) and the two idle nodes ~5, so the
+// owner sits at 20x the mean and trips any ratio — yet one forward on one node is
+// the perfectly correct distribution, there is nothing to gain by moving it, and
+// the move would only drop a live connection. The ratio detects *shape*; this
+// floor says there must be something worth moving.
+const minForwards = 2
+
 func planBalance(s *State, minOverRatio float64) (*TopoEntry, string, bool) {
 	// Candidate destinations: alive, non-leader-free (any), scored by Load.
 	type cand struct {
-		id    string
+		id       string
-		score float64
+		score    float64
+		forwards int
 	}
 	var dests []cand
 	total := 0
@@ -33,7 +52,7 @@ func planBalance(s *State, minOverRatio float64) (*TopoEntry, string, bool) {
 		if !n.Alive {
 			continue
 		}
-		dests = append(dests, cand{n.ID, n.Load.Score()})
+		dests = append(dests, cand{n.ID, n.Load.Score(), n.Load.Forwards})
 		total += n.Load.Forwards
 	}
 	if len(dests) < 2 {
@@ -42,15 +61,17 @@ func planBalance(s *State, minOverRatio float64) (*TopoEntry, string, bool) {
 	avg := float64(total) / float64(len(dests))
 	threshold := avg * minOverRatio
 
-	// Overloaded owners, worst first.
+	// Overloaded owners, worst first. The absolute floor is applied here so a node
+	// holding one forward in a three-node ring is never considered hot.
 	type over struct {
-		id    string
+		id       string
-		score float64
+		score    float64
+		forwards int
 	}
 	var overs []over
 	for _, d := range dests {
-		if d.score > threshold {
+		if d.score > threshold && d.forwards >= minForwards {
-			overs = append(overs, over{d.id, d.score})
+			overs = append(overs, over{d.id, d.score, d.forwards})
 		}
 	}
 	if len(overs) == 0 {
@@ -99,3 +120,176 @@ func planBalance(s *State, minOverRatio float64) (*TopoEntry, string, bool) {
 	}
 	return pick, dest.id, true
 }
+
+// Balancer moves forwards off an overloaded node, one at a time.
+//
+// It is a TWO-PHASE state machine, and the split is a safety requirement rather
+// than a stylistic choice:
+//
+//	phase 1 (revoking): ask the current owner to stop. Wait until the topology
+//	    entry is actually gone before publishing anything.
+//	phase 2 (publishing): hand the forward to a NAMED destination.
+//
+// Publishing both at once does not work, for two independent reasons:
+//
+//  1. loadSnapshot() derives Forwards from ForwardsOwnedBy, so the moment the old
+//     entry is dropped the old owner becomes the lightest node. A plain creation
+//     task is claimed by "whoever is lightest" — so the node we just took the
+//     forward FROM would immediately claim it back and the migration would be a
+//     no-op that merely restarts the worker.
+//  2. PendingList sorts by task id as a STRING (t10 < t5), so the revoke and the
+//     submit could execute in either order within one pass.
+//
+// Migrate tasks solve (1) by carrying an explicit Target; this type solves (2) by
+// refusing to publish phase 2 before phase 1 has completed.
+type Balancer struct {
+	engine *Engine
+
+	// ratio is how far above the average a node's Score must be, relative to the
+	// lightest node, before anything moves.
+	ratio float64
+	// sustainedSec is how long an imbalance must persist before we act. This is
+	// what stops a transient CPU/network spike from restarting workers.
+	sustainedSec int64
+
+	// overSince records when the currently-overloaded node first crossed the
+	// threshold. Zero means "not currently over".
+	overSince map[string]int64
+	// pending migration awaiting phase 1 (the forward to revoke).
+	revoking *pendingMigration
+	// migration awaiting phase 2 (the forward to publish, and where to).
+	publishing *pendingMigration
+}
+
+type pendingMigration struct {
+	local      string
+	remote     string
+	port       int
+	fromOwner  string
+	toOwner    string
+	deadlineAt int64 // give up (re-publish for anyone) after this unix second
+}
+
+// NewBalancer builds a balancer. ratio <= 1 or sustainedSec < 0 are clamped to
+// sane minimums so a bad configuration cannot turn into a migration storm.
+func NewBalancer(e *Engine, ratio float64, sustainedSec int64) *Balancer {
+	if ratio <= 1 {
+		ratio = 2
+	}
+	if sustainedSec < 0 {
+		sustainedSec = 0
+	}
+	return &Balancer{
+		engine: e, ratio: ratio, sustainedSec: sustainedSec,
+		overSince: map[string]int64{},
+	}
+}
+
+// InFlight reports whether a migration is currently between phases.
+func (b *Balancer) InFlight() bool {
+	return b.revoking != nil || b.publishing != nil
+}
+
+// Step advances the state machine by one cycle. now is the current unix second.
+// It returns a short human-readable description when something happened.
+func (b *Balancer) Step(now int64) string {
+	// --- phase 2: did phase 1 finish? ---
+	if b.revoking != nil {
+		if b.engine.state.TopologyOwner(&Task{
+			Local:  store.Local{Name: b.revoking.local},
+			Remote: store.Remote{Name: b.revoking.remote},
+			Link:   store.Link{RemotePort: b.revoking.port},
+		}) != "" {
+			// Still owned: the revoke has not completed yet. Wait, unless it has
+			// been far too long, in which case something is wrong with the ring.
+			if now < b.revoking.deadlineAt {
+				return ""
+			}
+			log.Printf("ring[%s] migrate revoke for %s→%s:%d timed out; abandoning",
+				b.engine.ID, b.revoking.local, b.revoking.remote, b.revoking.port)
+			b.revoking = nil
+			return ""
+		}
+		// Entry gone — the old worker is stopped. Publish for the named target.
+		m := b.revoking
+		b.revoking = nil
+		b.publishing = m
+		b.publishMigration()
+		return "revoke complete, migrating " + m.local + ":" + strconv.Itoa(m.port) + " → " + m.toOwner
+	}
+
+	// --- phase 1: wait for an in-flight publish, or start a new migration ---
+	if b.publishing != nil {
+		// The migrate task carries itself in the ring; nothing to do but let it
+		// ride. Clear once the forward is back in the topology.
+		if b.engine.state.TopologyOwner(&Task{
+			Local:  store.Local{Name: b.publishing.local},
+			Remote: store.Remote{Name: b.publishing.remote},
+			Link:   store.Link{RemotePort: b.publishing.port},
+		}) == b.publishing.toOwner {
+			done := b.publishing
+			b.publishing = nil
+			return "migrated " + done.local + ":" + strconv.Itoa(done.port) + " to " + done.toOwner
+		}
+		if now >= b.publishing.deadlineAt {
+			// Nobody claimed it. Submit an UNTARGETED task so whichever node is
+			// lightest picks it up — the ring keeps the forward alive rather than
+			// stranding it, which is the same guarantee the rest of the protocol
+			// relies on.
+			m := b.publishing
+			b.publishing = nil
+			b.engine.SubmitTask(store.Local{Name: m.local}, store.Remote{Name: m.remote},
+				store.Link{RemotePort: m.port})
+			return "migrate target did not claim " + m.local + ":" + strconv.Itoa(m.port) + "; re-opened for any node"
+		}
+		return ""
+	}
+
+	// --- nothing in flight: look for work ---
+	pick, dest, hot := b.engine.findBalanceCandidate(b.ratio)
+	if hot != "" {
+		if _, seen := b.overSince[hot]; !seen {
+			b.overSince[hot] = now
+		}
+	} else {
+		b.overSince = map[string]int64{}
+	}
+	if pick == nil || dest == "" {
+		return ""
+	}
+	if since, ok := b.overSince[pick.OwnerID]; !ok || now-since < int64(b.sustainedSec) {
+		return ""
+	}
+	// Phase 1: publish the revoke and remember what we are waiting for.
+	m := &pendingMigration{
+		local: pick.Local.Name, remote: pick.Remote.Name, port: pick.Link.RemotePort,
+		fromOwner: pick.OwnerID, toOwner: dest,
+		// Generous deadline: a token round is ~2s, and the owner may be busy.
+		deadlineAt: now + 120,
+	}
+	b.revoking = m
+	b.engine.RevokeTask(pick.Local, pick.Remote, pick.Link)
+	// The revoke task must be able to reach the owner even though the balancer
+	// runs on the leader: it travels in the ring like any other command.
+	return "revoking " + m.local + ":" + strconv.Itoa(m.port) + " from " + m.fromOwner
+}
+
+func (b *Balancer) publishMigration() {
+	m := b.publishing
+	if m == nil {
+		return
+	}
+	b.engine.submitMigrate(store.Local{Name: m.local}, store.Remote{Name: m.remote},
+		store.Link{RemotePort: m.port}, m.toOwner)
+}
+
+// findBalanceCandidate exposes planBalance to the engine's live state.
+func (e *Engine) findBalanceCandidate(ratio float64) (*TopoEntry, string, string) {
+	pick, dest, ok := planBalance(&e.state, ratio)
+	if !ok {
+		return nil, "", ""
+	}
+	// Report the overloaded node even when there is nothing movable, so the
+	// hysteresis timer can start for a node whose forwards are all stopped.
+	return pick, dest, pick.OwnerID
+}

internal/cluster/ring_engine.go

@@ -406,6 +406,17 @@ func (e *Engine) runCommands(ctx context.Context, tk *Token) error {
 				}
 				continue // owned elsewhere — ride to the owner
 			}
+			if t.Migrate {
+				// Addressed to one node. The whole point is that "whoever is
+				// lightest" is the WRONG answer here: the node we just revoked the
+				// forward from is the one that became lightest, and it would claim
+				// the task straight back.
+				if t.Target == e.ID {
+					target = t
+					break
+				}
+				continue // ride to the target
+			}
 			if t.Restart {
 				// Owner-directed, exactly like a revoke: only the node holding the
 				// forward may act. Selecting it anywhere else would let a non-owner
@@ -538,7 +549,7 @@ func (e *Engine) runCommands(ctx context.Context, tk *Token) error {
 		// exists precisely to re-activate a forward whose entry is still present
 		// (a stop keeps the entry as the flag's carrier), so applying the guard
 		// would swallow every re-enable and leave the forward stopped forever.
-		if !claimed.Restart {
+		if !claimed.Restart && !claimed.Migrate {
 			if owner := e.state.TopologyOwner(claimed); owner != "" {
 				log.Printf("ring[%s] drop duplicate task %s: %s→%s:%d already owned by %s",
 					e.ID, claimed.ID, claimed.Local.Name, claimed.Remote.Name,
@@ -546,6 +557,35 @@ func (e *Engine) runCommands(ctx context.Context, tk *Token) error {
 				continue
 			}
 		}
+		if claimed.Migrate {
+			// We are the designated destination. The forward must NOT already be
+			// here: the two-phase migration guarantees the old entry was gone before
+			// this task was published. If one somehow exists, drop the task rather
+			// than creating a second topology entry for the same forward.
+			if owner := e.state.TopologyOwner(claimed); owner != "" {
+				log.Printf("ring[%s] drop migrate %s: already owned by %s", e.ID, claimed.ID, owner)
+				continue
+			}
+			if e.Handler != nil {
+				if err := e.Handler.Claim(ctx, claimed); err != nil {
+					// Put it back so another cycle can retry — the forward is
+					// currently owned by nobody, so losing this task would strand it.
+					e.state.PendingTasks[claimed.ID] = claimed
+					log.Printf("ring[%s] migrate %s failed: %v", e.ID, claimed.ID, err)
+					break
+				}
+			}
+			e.state.AddTopology(claimed, e.ID)
+			if e.Log != nil {
+				_, _ = e.Log.Append(e.ID, LogForwardAdd, map[string]any{
+					"taskId": claimed.ID, "local": claimed.Local.Name, "remote": claimed.Remote.Name,
+					"migratedFrom": e.state.TopologyOwner(claimed),
+				})
+			}
+			log.Printf("ring[%s] migrated in %s: %s→%s:%d", e.ID, claimed.ID,
+				claimed.Local.Name, claimed.Remote.Name, claimed.Link.RemotePort)
+			continue
+		}
 		if claimed.Restart {
 			// Only reached when this node owns the forward (or it is already gone
 			// and we are the fallback consumer — see the selection loop).
@@ -960,6 +1000,30 @@ func (e *Engine) SubmitRestart(local store.Local, remote store.Remote, link stor
 	return t
 }
 
+// submitMigrate publishes a load-balancing task addressed to one specific node.
+//
+// It is addressed on purpose. A plain creation task is claimed by "whichever node
+// is currently lightest", and after a revoke the node we just took the forward
+// from IS the lightest — so an untargeted task would be claimed straight back by
+// the old owner and the migration would achieve nothing but a worker restart.
+func (e *Engine) submitMigrate(local store.Local, remote store.Remote, link store.Link, target string) *Task {
+	link.Disabled = false
+	t := &Task{
+		ID:      e.state.NextTaskID(),
+		Local:   local,
+		Remote:  remote,
+		Link:    link,
+		Created: time.Now().Unix(),
+		Migrate: true,
+		Target:  target,
+	}
+	if e.state.PendingTasks == nil {
+		e.state.PendingTasks = map[string]*Task{}
+	}
+	e.state.PendingTasks[t.ID] = t
+	return t
+}
+
 // HasActiveTask reports whether a forward with this key is genuinely in flight
 // or running: a non-revocation pending task, or a topology entry that is not
 // marked disabled.