From 7d5bca5b72ecc7ed6854b456913f941bea955c0f Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Tue, 6 Oct 2026 11:50:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(cluster):=20=E8=B4=9F=E8=BD=BD=E5=9D=87?= =?UTF-8?q?=E8=A1=A1=E8=BF=81=E7=A7=BB=20=E2=80=94=E2=80=94=20=E5=AE=9A?= =?UTF-8?q?=E5=90=91=E6=8A=95=E9=80=92=20+=20=E4=B8=A4=E9=98=B6=E6=AE=B5?= =?UTF-8?q?=E6=8F=90=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 承接 875e580（只落决策逻辑）。本提交补齐运行时：周期检查、迟滞判定、 两阶段迁移，以及让迁移任务**定向**到指定节点。 ## 为什么必须是定向的（核心安全点） loadSnapshot() 的 Forwards 来自 ForwardsOwnedBy，所以旧 owner 一旦被摘掉条目， 它立刻变成最闲节点。而普通创建任务是"任何最闲者认领"—— 于是它会把刚迁走的 转发**当场抢回**，迁移变成"只重启一次 worker"的空操作。 故 Task 新增 Migrate + Target 字段，runCommands 的选择循环照 Restart 的范式 处理：Target == 自己才认领，否则 `continue` 让任务随环前进。 ## 为什么必须两阶段（不能一次发 revoke + submit） 1. 旧 owner 变最闲后会把未定向的任务抢回（上面那条）； 2. PendingList 按 `t` **字符串**排序（`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 | 210 +++++++++++++++++++++++++++++++ internal/cluster/ring.go | 7 ++ internal/cluster/ring_balance.go | 210 +++++++++++++++++++++++++++++-- internal/cluster/ring_engine.go | 66 +++++++++- 4 files changed, 484 insertions(+), 9 deletions(-) diff --git a/internal/cluster/balance_test.go b/internal/cluster/balance_test.go index 36f9c53..a0936e3 100644 --- a/internal/cluster/balance_test.go +++ b/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") + } +} diff --git a/internal/cluster/ring.go b/internal/cluster/ring.go index b4bbf20..5227bec 100644 --- a/internal/cluster/ring.go +++ b/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"` diff --git a/internal/cluster/ring_balance.go b/internal/cluster/ring_balance.go index 540abb5..4cb4137 100644 --- a/internal/cluster/ring_balance.go +++ b/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 - score float64 + id string + 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 - score float64 + id string + score float64 + forwards int } var overs []over for _, d := range dests { - if d.score > threshold { - overs = append(overs, over{d.id, d.score}) + if d.score > threshold && d.forwards >= minForwards { + 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 +} diff --git a/internal/cluster/ring_engine.go b/internal/cluster/ring_engine.go index 3717940..923c7b5 100644 --- a/internal/cluster/ring_engine.go +++ b/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.