Stop reusing replica ids, and stop handing out ports strays still hold

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) <noreply@anthropic.com>

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

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