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