From 8a062fec3570d48c2ba800c4c3a6a0bc65b3d8aa Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Wed, 16 Sep 2026 09:54:57 +0200 Subject: [PATCH] Stop reusing replica ids, and stop handing out ports strays still hold MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit's adoption fix made a deeper problem visible instead of fixing it. Replica ids were the model name and a counter, and the counter restarted with the process — so a control plane that came back placed a replica called llama-1 while an engine started before the restart, still running on a different node, also called itself llama-1. Two nodes then claimed the same shard id. Adoption handed it back and forth between them, every placement landed on a port an orphan was still bound to, and the engine failed to bind, which failed the replica, which placed another one on the same port. On the real cluster that loop ran for minutes: couldn't bind HTTP server socket, hostname: 0.0.0.0, port: 20002 Ids now carry four random bytes, which costs nothing and makes the collision impossible. The "already known" branch in adoption goes back to meaning what it says: an agent reconnecting, not a name clash across restarts. Ports a stray still holds are also reserved now. Stopping is not instant and an engine keeps its socket until it exits, so the allocator has to know about sockets no replica claims — otherwise the same loop reappears in the window between telling an agent to stop something and it actually stopping. Two of the tests written for the previous commit had quietly become vacuous under the new ids, asserting things about a literal llama-1 that no longer exists. They capture the id now. Co-Authored-By: Claude Opus 5 (1M context) --- internal/control/reconciler/reconciler.go | 73 +++++++++++++++++++---- internal/control/reconciler/stuck_test.go | 56 ++++++++++++++++- 2 files changed, 117 insertions(+), 12 deletions(-) diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index 149a8c4f..9a43c917 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -2,6 +2,7 @@ package reconciler import ( "context" + crypterand "crypto/rand" "fmt" "log/slog" "math/rand/v2" @@ -129,6 +130,9 @@ type Reconciler struct { // `ctl explain-placement` can answer without re-running placement. lastExplain map[string]placement.Result seq int + // reservedPorts holds ports the fleet reports as in use but no replica + // claims, so the allocator cannot hand out what something is still bound to. + reservedPorts map[int]struct{} wake chan struct{} stop chan struct{} @@ -144,15 +148,16 @@ func New(cfg Config, cat *catalog.Catalog, inv *inventory.Store, disp Dispatcher } return &Reconciler{ cfg: cfg, cat: cat, inv: inv, disp: disp, log: log, - sched: placement.New(inv), - now: time.Now, - replicas: map[string]*Replica{}, - nextAttempt: map[string]time.Time{}, - attempts: map[string]int{}, - lastExplain: map[string]placement.Result{}, - wake: make(chan struct{}, 1), - stop: make(chan struct{}), - done: make(chan struct{}), + sched: placement.New(inv), + now: time.Now, + replicas: map[string]*Replica{}, + nextAttempt: map[string]time.Time{}, + attempts: map[string]int{}, + reservedPorts: map[int]struct{}{}, + lastExplain: map[string]placement.Result{}, + wake: make(chan struct{}, 1), + stop: make(chan struct{}), + done: make(chan struct{}), } } @@ -498,7 +503,7 @@ func (r *Reconciler) startReplica(ctx context.Context, m catalog.Model, pl *plac r.mu.Lock() r.seq++ - id := fmt.Sprintf("%s-%d", m.Name, r.seq) + id := replicaID(m.Name, r.seq) attempt := r.attempts[m.Name] r.mu.Unlock() @@ -587,6 +592,28 @@ func (r *Reconciler) startReplica(ctx context.Context, m catalog.Model, pl *plac return nil } +// replicaID names a replica uniquely for the life of the cluster, not the life +// of this process. +// +// The counter alone is not enough, and the way it fails is ugly. A control +// plane that restarts begins counting again, so the replica it places next is +// called llama-1 — which is also what an engine started before the restart and +// still running on some other node calls itself. Two nodes then claim the same +// shard id, adoption hands the id back and forth between them, and every +// placement lands on a port an orphan is already holding. Observed on a real +// cluster as a bind-conflict loop that never settled. +// +// The random suffix costs nothing and makes the collision impossible. +func replicaID(model string, seq int) string { + var buf [4]byte + if _, err := crypterand.Read(buf[:]); err != nil { + // Randomness failing is not a reason to stop placing replicas; the + // counter alone is still unique within this process. + return fmt.Sprintf("%s-%d", model, seq) + } + return fmt.Sprintf("%s-%x", model, buf) +} + // allocatePorts picks a port shared by every shard of a replica. // // The same port on every node keeps the leader address predictable and the @@ -600,6 +627,13 @@ func (r *Reconciler) allocatePorts(pl *placement.Placement) int { used[s.HTTPPort] = struct{}{} } } + // Ports an agent has reported but nothing owns — a stray mid-shutdown, an + // engine whose replica was just reaped. The process is still listening + // whatever the bookkeeping says, and handing its port to a new shard means + // that shard fails to bind and takes the whole replica with it. + for p := range r.reservedPorts { + used[p] = struct{}{} + } r.mu.Unlock() // Two ports per replica: HTTP and the distributed init rendezvous. @@ -694,6 +728,7 @@ func (r *Reconciler) Adopt(shards []AdoptedShard) (unclaimed []AdoptedShard) { // stray. if existing.hasLiveShard() { unclaimed = append(unclaimed, group...) + r.reservePorts(group) continue } r.log.Warn("a running replica reported by the fleet replaces one we placed", @@ -708,11 +743,13 @@ func (r *Reconciler) Adopt(shards []AdoptedShard) (unclaimed []AdoptedShard) { m, ok := r.cat.Resolve(first.Model) if !ok { unclaimed = append(unclaimed, group...) + r.reservePorts(group) continue } prof, ok := m.ProfileByName(first.Profile) if !ok { unclaimed = append(unclaimed, group...) + r.reservePorts(group) continue } @@ -749,6 +786,22 @@ func (r *Reconciler) Adopt(shards []AdoptedShard) (unclaimed []AdoptedShard) { return unclaimed } +// reservePorts records ports held by shards nothing owns. +// +// They are being stopped, but stopping is not instant and the engine keeps its +// socket until it exits. Handing the port to a new shard in the meantime makes +// that shard fail to bind, which fails its replica, which places another one on +// the same port — a loop that does not end on its own. +// +// Called with the lock held. +func (r *Reconciler) reservePorts(shards []AdoptedShard) { + for _, s := range shards { + if s.HTTPPort > 0 { + r.reservedPorts[s.HTTPPort] = struct{}{} + } + } +} + // AdoptedShard is one running shard an agent reported at registration. type AdoptedShard struct { NodeID string diff --git a/internal/control/reconciler/stuck_test.go b/internal/control/reconciler/stuck_test.go index 74f493c3..07eb64ef 100644 --- a/internal/control/reconciler/stuck_test.go +++ b/internal/control/reconciler/stuck_test.go @@ -29,6 +29,9 @@ func TestAReplicaThatNeverComesUpIsEventuallyFailed(t *testing.T) { if len(reps) != 1 || reps[0].State != Starting { t.Fatalf("expected one starting replica, got %+v", reps) } + // Captured rather than assumed: replica ids carry a random suffix so a + // restarted control plane cannot collide with engines from its last life. + stuckID := reps[0].ID // The agent says nothing at all, which is exactly the case. h.advance(90 * time.Second) @@ -45,7 +48,7 @@ func TestAReplicaThatNeverComesUpIsEventuallyFailed(t *testing.T) { // point of failing it. reps = h.r.Replicas() for _, rep := range reps { - if rep.ID == "llama-1" && rep.State == Starting { + if rep.ID == stuckID && rep.State == Starting { t.Fatal("the replica is still stuck in starting past its deadline") } } @@ -57,7 +60,7 @@ func TestAReplicaThatNeverComesUpIsEventuallyFailed(t *testing.T) { reps = h.r.Replicas() var replaced bool for _, rep := range reps { - if rep.ID != "llama-1" { + if rep.ID != stuckID { replaced = true } } @@ -178,3 +181,52 @@ func TestAnAdoptedShardDoesNotDisplaceOneThatIsActuallyRunning(t *testing.T) { t.Error("a stray replaced the replica that was actually serving") } } + +func TestReplicaIDsDoNotRepeatAcrossRestarts(t *testing.T) { + // The collision that caused the bind-conflict loop on a real cluster. A + // control plane that restarts begins counting again, so the replica it + // places next was called llama-1 — which is also what an engine started + // before the restart, still running on another node, called itself. Two + // nodes claimed the same shard id and every placement landed on a port an + // orphan was already holding. + seen := map[string]bool{} + for i := range 1000 { + // The same counter value every time, which is exactly what a restart + // reproduces. + id := replicaID("llama", 1) + if seen[id] { + t.Fatalf("id %q repeated after %d placements", id, i) + } + seen[id] = true + } +} + +func TestAPortHeldByAStrayIsNotHandedOut(t *testing.T) { + // Stopping is not instant, and an engine keeps its socket until it exits. + // Giving its port to a new shard makes that shard fail to bind, which fails + // its replica, which places another one on the same port. + h := newHarness(t, testModel(1), gpuNode("1", 8)) + ctx := context.Background() + h.r.Reconcile(ctx) + + placed := h.r.Replicas()[0] + h.r.OnShardStatus(placed.ID, placed.Shards[0].ID, supervisor.Ready, "") + + // An agent reports a stray on the same replica id, holding a different + // port. It is disowned — and its port must not become available. + strayPort := placed.Shards[0].HTTPPort + 2 + unclaimed := h.r.Adopt([]AdoptedShard{{ + NodeID: "1", ReplicaID: placed.ID, ShardID: "stray-r0", + Rank: 0, HTTPPort: strayPort, State: supervisor.Ready, + Model: "llama", Profile: "tp4", Endpoint: "http://node1:29999", + }}) + if len(unclaimed) != 1 { + t.Fatalf("the stray was not disowned: %+v", unclaimed) + } + + for range 5 { + if got := h.r.allocatePorts(nil); got == strayPort { + t.Fatalf("port %d was handed out while a stray still holds it", got) + } + } +}