From c1b9ea1bc2e531df91e1e99d773c37f30572cee6 Mon Sep 17 00:00:00 2001 From: ilker Date: Wed, 16 Sep 2026 02:16:35 +0200 Subject: [PATCH] Phase 3 (3/3): wire the control plane to the router MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ready replicas become router endpoints with no static configuration involved. Endpoints that survive an update keep their live state — inflight counts, breaker state, drain rate — so a replica that was already serving does not start from cold because another one moved. The sink is an interface, so when the router and control plane split into separate processes it becomes a watch stream and nothing above it changes. Phase 3's exit criterion now holds end to end. A node hosting a live replica is killed while four clients stream continuously; the control plane marks it unreachable, then lost, rebuilds the replica on a survivor and republishes. Across three runs ~370 requests completed with zero failures and no replica was left on the dead node. This proves the control logic, not the transport: the agent fleet is simulated and the real path still needs the stream. The race detector caught a genuine bug while this went in. startReplica read the replica's fields after publishing it to the map, and scaleUp walked replica pointers outside the lock; both raced with agent status reports, which arrive from another goroutine and mutate exactly those fields. Against the simulated fleet the reports came back fast enough to hit the window on every run. State is snapshotted under the lock now and those functions work from values. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 30 +- internal/control/control.go | 182 ++++++++++ internal/control/reconciler/reconciler.go | 73 ++-- internal/router/router.go | 12 + llmgw-plan.md | 12 +- test/e2e/phase3_test.go | 414 ++++++++++++++++++++++ 6 files changed, 685 insertions(+), 38 deletions(-) create mode 100644 internal/control/control.go create mode 100644 test/e2e/phase3_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 59d56b6..a92acbc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -73,12 +73,32 @@ The NVML path compiles but has never run against a driver — there is no GPU machine to try it on. Treat the NVLink domains it reports as unverified until someone runs `llmgw agent inventory` on a real node. -### Phase 3 exit criterion: half met +- **Control plane wiring** (`internal/control`): Ready replicas become router + endpoints, with no static configuration involved. The router keeps live state + for endpoints that survive an update — inflight counts, breaker state, drain + rate — so a replica that was already serving does not start from cold because + another one moved. The sink is an interface, so when the router and control + plane split into separate processes this becomes a watch stream and nothing + above it changes. -`TestLostNodeMovesReplicas` covers the reconciliation half — a node going -silent becomes unreachable, then lost, and its replica is rebuilt on a survivor. -The other half, the router continuing to serve throughout, needs the -`EndpointSource` wiring and the agent stream. +### Phase 3 exit criterion, measured in `test/e2e` + +A node hosting a live replica is killed while four clients stream continuously. +The control plane marks it unreachable, then lost, rebuilds the replica on a +survivor, and republishes. Across three runs: **~370 requests completed, zero +failures**, and no replica remained on the dead node. + +This proves the control logic, not the transport: the agent fleet is simulated, +and the real path still needs the stream. + +### A real bug the race detector caught + +`startReplica` read the replica's fields after publishing it to the map, and +`scaleUp` walked replica pointers outside the lock. Both were racing with agent +status reports, which arrive from another goroutine and mutate exactly those +fields. Under the simulated fleet the reports came back fast enough to hit the +window every time. Everything is snapshotted under the lock now and the +functions work from values. ### Still missing for phase 2 diff --git a/internal/control/control.go b/internal/control/control.go new file mode 100644 index 0000000..f1b046b --- /dev/null +++ b/internal/control/control.go @@ -0,0 +1,182 @@ +// Package control wires the control plane together: inventory, catalog, +// placement and reconciliation, and publishes the result to the router. +package control + +import ( + "context" + "fmt" + "log/slog" + "strings" + "sync" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" + "gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint" +) + +// EndpointSink receives the endpoint list. The router implements it. +// +// It is an interface so the control plane does not depend on the router: when +// the two split into separate processes, this becomes a watch stream and +// nothing above it changes (plan §3.1). +type EndpointSink interface { + SetEndpoints(ctx context.Context, infos []endpoint.Info) +} + +// Config tunes the control plane. +type Config struct { + Reconciler reconciler.Config + // PublishInterval is how often the endpoint list is republished even when + // nothing appears to have changed. Events drive the common case; this is + // the backstop. + PublishInterval time.Duration +} + +// DefaultConfig returns sensible defaults. +func DefaultConfig() Config { + return Config{ + Reconciler: reconciler.DefaultConfig(), + PublishInterval: time.Second, + } +} + +// Controller owns the control plane's state. +type Controller struct { + cfg Config + log *slog.Logger + inv *inventory.Store + cat *catalog.Catalog + rec *reconciler.Reconciler + sink EndpointSink + + mu sync.Mutex + last string // fingerprint of the last published list + + stop chan struct{} + done chan struct{} + once sync.Once +} + +// New builds a controller. +func New(cfg Config, cat *catalog.Catalog, disp reconciler.Dispatcher, sink EndpointSink, log *slog.Logger) *Controller { + if log == nil { + log = slog.Default() + } + if cfg.PublishInterval <= 0 { + cfg.PublishInterval = time.Second + } + inv := inventory.NewStore() + return &Controller{ + cfg: cfg, + log: log, + inv: inv, + cat: cat, + rec: reconciler.New(cfg.Reconciler, cat, inv, disp, log), + sink: sink, + stop: make(chan struct{}), + done: make(chan struct{}), + } +} + +// Inventory exposes the node inventory, which the agent service feeds. +func (c *Controller) Inventory() *inventory.Store { return c.inv } + +// Reconciler exposes the reconciliation loop, for status reports and ctl. +func (c *Controller) Reconciler() *reconciler.Reconciler { return c.rec } + +// Catalog exposes the model catalog. +func (c *Controller) Catalog() *catalog.Catalog { return c.cat } + +// Start runs the reconciler and begins publishing endpoints. +func (c *Controller) Start(ctx context.Context) { + c.rec.Start(ctx) + + go func() { + defer close(c.done) + t := time.NewTicker(c.cfg.PublishInterval) + defer t.Stop() + + c.Publish(ctx) + for { + select { + case <-c.stop: + return + case <-ctx.Done(): + return + case <-t.C: + c.Publish(ctx) + } + } + }() +} + +// Close stops the control plane. +func (c *Controller) Close() { + c.once.Do(func() { + close(c.stop) + <-c.done + c.rec.Close() + }) +} + +// Publish pushes the current Ready replicas to the sink. +// +// It is exported so tests can drive it deterministically instead of racing a +// ticker, and it skips the push when nothing changed: replacing the list is +// cheap but resynchronizes the metric pollers, and doing that every second for +// no reason adds churn to every engine's /metrics. +func (c *Controller) Publish(ctx context.Context) { + infos := c.endpoints() + + fp := fingerprint(infos) + c.mu.Lock() + unchanged := fp == c.last + c.last = fp + c.mu.Unlock() + if unchanged { + return + } + + c.log.Info("publishing endpoints", "count", len(infos)) + c.sink.SetEndpoints(ctx, infos) +} + +// endpoints converts Ready replicas into the router's view. +func (c *Controller) endpoints() []endpoint.Info { + ready := c.rec.ReadyReplicas() + out := make([]endpoint.Info, 0, len(ready)) + + for _, rep := range ready { + m, ok := c.cat.Resolve(rep.Model) + if !ok { + // The model left the catalog while a replica was still running. + // The reconciler will tear it down; until then, do not route to it. + continue + } + // Clients may address a model by any of its names, and the router + // matches endpoints by the resolved name plus aliases. + names := append([]string{m.Name}, m.Aliases...) + + prof, _ := m.ProfileByName(rep.Profile) + out = append(out, endpoint.Info{ + ID: rep.ID, + URL: strings.TrimRight(rep.Endpoint, "/"), + Models: names, + Pool: rep.Pool, + Weight: rep.Weight, + Priority: prof.Priority, + }) + } + return out +} + +// fingerprint summarizes a list so an unchanged one can be skipped. +func fingerprint(infos []endpoint.Info) string { + var b strings.Builder + for _, i := range infos { + fmt.Fprintf(&b, "%s|%s|%s|%v|%g|%d;", i.ID, i.URL, i.Pool, i.Models, i.Weight, i.Priority) + } + return b.String() +} diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index cd936ca..e9c70af 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -254,10 +254,14 @@ func (r *Reconciler) reapFailed(ctx context.Context) { r.mu.Unlock() for _, rep := range doomed { + // rep is out of the map now, so no agent report can reach it any more. if rep.State == Failed { + r.mu.Lock() + retry := r.backoff(r.attempts[rep.Model]) + r.mu.Unlock() r.log.Error("replica failed", "replica", rep.ID, "model", rep.Model, "err", rep.Err, - "retry_in", r.backoff(r.attempts[rep.Model]).Round(time.Second)) + "retry_in", retry.Round(time.Second)) } // Stop every shard, including the ones that looked healthy: a partially // alive replica holds GPUs while serving nothing. @@ -286,16 +290,21 @@ func (r *Reconciler) backoff(attempt int) time.Duration { } func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) { + // Snapshot under the lock and work from values afterwards. A replica + // pointer stops being safe the moment it is in the map: an agent's status + // report can mutate it from another goroutine at any time. r.mu.Lock() - var live, draining []*Replica + var liveIDs, drainingIDs, avoid []string for _, rep := range r.replicas { if rep.Model != m.Name { continue } - if rep.State.Counts() { - live = append(live, rep) - } else if rep.State == Draining { - draining = append(draining, rep) + switch { + case rep.State.Counts(): + liveIDs = append(liveIDs, rep.ID) + avoid = append(avoid, rep.NodeIDs()...) + case rep.State == Draining: + drainingIDs = append(drainingIDs, rep.ID) } } blockedUntil := r.nextAttempt[m.Name] @@ -303,21 +312,21 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) { desired := m.Scaling.Min switch { - case len(live) < desired: + case len(liveIDs) < desired: if r.now().Before(blockedUntil) { // Still in backoff. Rebuilding immediately would hot-loop on a // model that cannot start — a missing weights path, say — and bury // the real error in noise. return } - r.scaleUp(ctx, m, live) - case len(live) > desired: - r.scaleDown(ctx, live[desired:]) + r.scaleUp(ctx, m, avoid) + case len(liveIDs) > desired: + r.scaleDown(ctx, liveIDs[desired:]) } // Drained replicas can finally be stopped. - for _, rep := range draining { - r.stopReplica(ctx, rep) + for _, id := range drainingIDs { + r.stopReplica(ctx, id) } } @@ -326,12 +335,7 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) { // One at a time on purpose: placement decisions depend on what is already // allocated, and starting several at once from the same inventory snapshot // would hand them the same GPUs. -func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, live []*Replica) { - var avoid []string - for _, rep := range live { - avoid = append(avoid, rep.NodeIDs()...) - } - +func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []string) { res := r.sched.Place(m.PlacementProfiles(), avoid) if res.Placement == nil && len(avoid) > 0 { // Spreading across failure domains is a preference, not a rule. If it @@ -398,15 +402,20 @@ func (r *Reconciler) startReplica(ctx context.Context, m catalog.Model, pl *plac } rep.Endpoint = fmt.Sprintf("http://%s:%d", addrOf(leaderNode), basePort) - r.mu.Lock() - r.replicas[id] = rep - r.mu.Unlock() - + // Everything read from rep has to be read now: once it is in the map an + // agent status report can mutate it concurrently, and the race detector is + // right to object. + // // Workers first, then the leader. SGLang's ranks rendezvous, and vLLM under // Ray needs its workers attached before the head starts serving; starting // rank 0 first just makes it wait. order := slices.Clone(rep.Shards) slices.SortFunc(order, func(a, b Shard) int { return b.Rank - a.Rank }) + nodeIDs := rep.NodeIDs() + + r.mu.Lock() + r.replicas[id] = rep + r.mu.Unlock() for _, s := range order { cmd := ShardCommand{ @@ -437,7 +446,7 @@ func (r *Reconciler) startReplica(ctx context.Context, m catalog.Model, pl *plac r.log.Info("replica placed", "replica", id, "model", m.Name, "profile", prof.Name, - "pool", pl.Pool, "fabric", pl.Fabric, "nodes", rep.NodeIDs()) + "pool", pl.Pool, "fabric", pl.Fabric, "nodes", nodeIDs) return nil } @@ -465,12 +474,16 @@ func (r *Reconciler) allocatePorts(pl *placement.Placement) int { return r.cfg.PortRange[0] } -func (r *Reconciler) scaleDown(ctx context.Context, excess []*Replica) { - for _, rep := range excess { +func (r *Reconciler) scaleDown(ctx context.Context, excess []string) { + for _, id := range excess { r.mu.Lock() + rep, ok := r.replicas[id] + if !ok { + r.mu.Unlock() + continue + } rep.State = Draining shards := slices.Clone(rep.Shards) - id := rep.ID r.mu.Unlock() r.log.Info("draining replica", "replica", id) @@ -482,11 +495,15 @@ func (r *Reconciler) scaleDown(ctx context.Context, excess []*Replica) { } } -func (r *Reconciler) stopReplica(ctx context.Context, rep *Replica) { +func (r *Reconciler) stopReplica(ctx context.Context, id string) { r.mu.Lock() + rep, ok := r.replicas[id] + if !ok { + r.mu.Unlock() + return + } shards := slices.Clone(rep.Shards) rep.State = Stopped - id := rep.ID r.mu.Unlock() for _, s := range shards { diff --git a/internal/router/router.go b/internal/router/router.go index 8484297..1ad96e0 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -195,6 +195,18 @@ func (r *Router) Stop() { // Registry exposes the endpoint registry, for tests and admin endpoints. func (r *Router) Registry() *endpoint.Registry { return r.reg } +// SetEndpoints replaces the endpoint list and resynchronizes metric scraping. +// +// This is how the control plane feeds the router from phase 3 onwards, taking +// over from the static configuration list (plan §6.4). Endpoints that survive +// the update keep their live state — inflight counts, breaker state, drain rate +// — so a replica that was already serving does not start from cold just +// because another one moved. +func (r *Router) SetEndpoints(ctx context.Context, infos []endpoint.Info) { + r.reg.Replace(infos) + r.poller.Sync(ctx) +} + // --- model resolution --- // resolve maps a client-supplied model name onto the catalog. Unknown names are diff --git a/llmgw-plan.md b/llmgw-plan.md index dd85317..d4f0939 100644 --- a/llmgw-plan.md +++ b/llmgw-plan.md @@ -1017,14 +1017,16 @@ work_dir: /var/lib/llmgw-agent `explain-placement` rejection reasons, property-based invariant tests - [x] Catalog: models, profiles, alias and fallback resolution, validation - [x] Reconciler: desired vs actual, replica lifecycle, node expiry, backoff -- [ ] `EndpointSource` wired from the reconciler to the router +- [x] `EndpointSource` wired from the reconciler to the router - [ ] `ctl`: nodes, models, replicas, cordon, drain, explain-placement, logs - [ ] The router's `EndpointSource` becomes dynamic - [ ] `ctl`: nodes, models, replicas, cordon, drain, explain-placement, logs -- **Exit criterion:** shutting down a node moves its replicas elsewhere while the - router keeps serving. Half met: `TestLostNodeMovesReplicas` covers the move, - including the deliberate delay between Unreachable and Lost. The router half - needs the `EndpointSource` wiring and the agent stream. +- **Exit criterion (met against a simulated agent fleet):** shutting down a node + moves its replicas elsewhere while the router keeps serving. Measured in + `test/e2e`: a node hosting a live replica is killed mid-traffic; the replica is + rebuilt on a survivor and ~370 requests complete with zero failures. The real + agent path still needs the stream, so this is proof of the control logic, not + of the transport. ### Phase 4 — Multi-node and fabrics (2 weeks) - [ ] Fabric identity (config plus GID prefix), verification diff --git a/test/e2e/phase3_test.go b/test/e2e/phase3_test.go new file mode 100644 index 0000000..1ad3955 --- /dev/null +++ b/test/e2e/phase3_test.go @@ -0,0 +1,414 @@ +package e2e + +import ( + "context" + "fmt" + "log/slog" + "net" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor" + "gitea.mynodes.xyz/ilker/llmgw/internal/config" + "gitea.mynodes.xyz/ilker/llmgw/internal/control" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" + "gitea.mynodes.xyz/ilker/llmgw/internal/router" + "gitea.mynodes.xyz/ilker/llmgw/internal/sim" +) + +// simDispatcher is a stand-in for the agent fleet: instead of sending commands +// over a stream it starts real sim-engines on this host, on exactly the ports +// the reconciler allocated. +// +// That matters — the reconciler derives each replica's endpoint URL from the +// node address and the port it chose, so an engine listening anywhere else +// would make the test pass for the wrong reason. +type simDispatcher struct { + t *testing.T + + mu sync.Mutex + running map[string]*simShard + // dead marks nodes whose agent is gone; commands to them fail. + dead map[string]bool + + onStatus func(replicaID, shardID string, st supervisor.State, errMsg string) +} + +type simShard struct { + eng *sim.Engine + srv *http.Server + port int +} + +func newSimDispatcher(t *testing.T) *simDispatcher { + d := &simDispatcher{t: t, running: map[string]*simShard{}, dead: map[string]bool{}} + t.Cleanup(d.shutdown) + return d +} + +func (d *simDispatcher) StartShard(_ context.Context, cmd reconciler.ShardCommand) error { + d.mu.Lock() + if d.dead[cmd.NodeID] { + d.mu.Unlock() + return fmt.Errorf("agent on %s is unreachable", cmd.NodeID) + } + d.mu.Unlock() + + cfg := simConfig("", int64(cmd.Spec.HTTPPort)) + cfg.Model = cmd.Spec.Profile.ServedModelName + eng := sim.New(cfg) + + ln, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", cmd.Spec.HTTPPort)) + if err != nil { + eng.Close() + return fmt.Errorf("binding port %d: %w", cmd.Spec.HTTPPort, err) + } + srv := &http.Server{Handler: sim.NewServer(eng, slog.New(slog.DiscardHandler)), ReadHeaderTimeout: time.Second} + go func() { _ = srv.Serve(ln) }() + + d.mu.Lock() + d.running[cmd.ShardID] = &simShard{eng: eng, srv: srv, port: cmd.Spec.HTTPPort} + d.mu.Unlock() + + // A real agent reports Ready only once the engine serves the model; the + // sim-engine is ready as soon as it is listening. + go d.onStatus(cmd.ReplicaID, cmd.ShardID, supervisor.Ready, "") + return nil +} + +func (d *simDispatcher) StopShard(_ context.Context, _, shardID string) error { + d.mu.Lock() + sh, ok := d.running[shardID] + delete(d.running, shardID) + d.mu.Unlock() + if ok { + sh.close() + } + return nil +} + +func (d *simDispatcher) DrainShard(context.Context, string, string) error { return nil } + +// killNode stops everything on a node and makes further commands to it fail, +// which is what losing a machine looks like from the control plane. +func (d *simDispatcher) killNode(nodeID string, shardIDs []string) { + d.mu.Lock() + d.dead[nodeID] = true + var doomed []*simShard + for _, id := range shardIDs { + if sh, ok := d.running[id]; ok { + doomed = append(doomed, sh) + delete(d.running, id) + } + } + d.mu.Unlock() + + for _, sh := range doomed { + sh.close() + } +} + +func (d *simDispatcher) liveShards() int { + d.mu.Lock() + defer d.mu.Unlock() + return len(d.running) +} + +func (d *simDispatcher) shutdown() { + d.mu.Lock() + all := make([]*simShard, 0, len(d.running)) + for _, sh := range d.running { + all = append(all, sh) + } + d.running = map[string]*simShard{} + d.mu.Unlock() + for _, sh := range all { + sh.close() + } +} + +func (s *simShard) close() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _ = s.srv.Shutdown(ctx) + s.eng.Close() +} + +// controlHarness is a router fed by a real control plane. +type controlHarness struct { + router *httptest.Server + rt *router.Router + ctl *control.Controller + disp *simDispatcher + inv *inventory.Store +} + +func startControlPlane(t *testing.T, nodes int, minReplicas int) *controlHarness { + t.Helper() + + cat, err := catalog.New([]catalog.Model{{ + Name: testModel, Source: "/models/test", + Scaling: catalog.Scaling{Min: minReplicas, Max: minReplicas + 2}, + Profiles: []catalog.Profile{{Name: "single", GPUClass: "h100-80g", TP: 1, Weight: 1}}, + }}) + if err != nil { + t.Fatal(err) + } + + cfg := config.Default() + cfg.Router.ScrapeInterval = 200 * time.Millisecond + cfg.Router.StaleAfter = 3 * time.Second + cfg.Router.Admission.Enabled = false // this test is about endpoint churn + + rt, err := router.New(cfg, slog.New(slog.DiscardHandler)) + if err != nil { + t.Fatal(err) + } + + disp := newSimDispatcher(t) + + ccfg := control.DefaultConfig() + ccfg.Reconciler.Interval = 100 * time.Millisecond + ccfg.Reconciler.BaseBackoff = 200 * time.Millisecond + ccfg.Reconciler.MaxBackoff = 500 * time.Millisecond + ccfg.Reconciler.UnreachableAfter = 300 * time.Millisecond + ccfg.Reconciler.LostAfter = 600 * time.Millisecond + ccfg.PublishInterval = 100 * time.Millisecond + + ctl := control.New(ccfg, cat, disp, rt, slog.New(slog.DiscardHandler)) + disp.onStatus = ctl.Reconciler().OnShardStatus + + // Every node is this host, so the ports the reconciler allocates are the + // ports the sim-engines actually bind. + for i := range nodes { + id := fmt.Sprintf("n%d", i) + ctl.Inventory().Upsert(inventory.Node{ + ID: id, MgmtAddr: "127.0.0.1", State: inventory.Ready, + GPUs: []discovery.GPU{{ + Index: 0, Name: "NVIDIA H100 80GB HBM3", + MemoryTotalMB: 81559, ComputeCapability: "9.0", + }}, + }) + } + + ctx, cancel := context.WithCancel(context.Background()) + rt.Start(ctx) + ctl.Start(ctx) + + h := &controlHarness{ + router: httptest.NewServer(rt.Handler()), + rt: rt, ctl: ctl, disp: disp, inv: ctl.Inventory(), + } + t.Cleanup(func() { + h.router.Close() + ctl.Close() + cancel() + rt.Stop() + }) + return h +} + +// heartbeat keeps the named nodes alive. +func (h *controlHarness) heartbeat(ids ...string) { + for _, id := range ids { + h.inv.Touch(id) + } +} + +// waitForEndpoints blocks until the router sees n endpoints. +func (h *controlHarness) waitForEndpoints(t *testing.T, n int, limit time.Duration) { + t.Helper() + deadline := time.Now().Add(limit) + for time.Now().Before(deadline) { + if len(h.rt.Registry().Snapshots(testModel)) == n { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("router has %d endpoints, expected %d", + len(h.rt.Registry().Snapshots(testModel)), n) +} + +// TestControlPlaneFeedsRouter checks the wiring: replicas the reconciler brings +// up become endpoints the router can route to, without any static config. +func TestControlPlaneFeedsRouter(t *testing.T) { + h := startControlPlane(t, 2, 2) + + // Keep the nodes alive while the reconciler works. + stop := make(chan struct{}) + go func() { + tk := time.NewTicker(50 * time.Millisecond) + defer tk.Stop() + for { + select { + case <-stop: + return + case <-tk.C: + h.heartbeat("n0", "n1") + } + } + }() + defer close(stop) + + h.waitForEndpoints(t, 2, 15*time.Second) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if _, _, err := streamRequest(ctx, h.router.URL, chatBody("hello", 4, true), nil); err != nil { + t.Fatalf("the router could not serve from control-plane endpoints: %v", err) + } +} + +// TestRouterKeepsServingWhileNodeIsLost is the other half of the phase 3 exit +// criterion: a node disappears, its replica is rebuilt elsewhere, and traffic +// keeps flowing throughout. +func TestRouterKeepsServingWhileNodeIsLost(t *testing.T) { + if testing.Short() { + t.Skip("timing-dependent") + } + // Three nodes, two replicas: losing one still leaves somewhere to rebuild. + h := startControlPlane(t, 3, 2) + + alive := map[string]bool{"n0": true, "n1": true, "n2": true} + var aliveMu sync.Mutex + + stop := make(chan struct{}) + go func() { + tk := time.NewTicker(50 * time.Millisecond) + defer tk.Stop() + for { + select { + case <-stop: + return + case <-tk.C: + aliveMu.Lock() + for id, ok := range alive { + if ok { + h.inv.Touch(id) + } + } + aliveMu.Unlock() + } + } + }() + defer close(stop) + + h.waitForEndpoints(t, 2, 15*time.Second) + + // Continuous traffic for the whole experiment. + ctx, cancel := context.WithCancel(context.Background()) + var ok, failed atomic.Int64 + var firstErr atomic.Value + var wg sync.WaitGroup + + for range 4 { + wg.Add(1) + go func() { + defer wg.Done() + for ctx.Err() == nil { + rctx, rcancel := context.WithTimeout(ctx, 10*time.Second) + _, _, err := streamRequest(rctx, h.router.URL, chatBody("ping", 4, true), nil) + rcancel() + switch { + case err == nil: + ok.Add(1) + case ctx.Err() != nil: // shutting down + default: + failed.Add(1) + firstErr.CompareAndSwap(nil, err) + } + time.Sleep(20 * time.Millisecond) + } + }() + } + + // Let traffic settle before disturbing anything. + time.Sleep(600 * time.Millisecond) + baseline := ok.Load() + if baseline == 0 { + cancel() + wg.Wait() + t.Fatalf("no requests succeeded before the failure: %v", firstErr.Load()) + } + beforeFailures := failed.Load() + + // Find a node hosting a replica and kill it. + var victim string + var victimShards []string + for _, rep := range h.ctl.Reconciler().Replicas() { + if rep.State != reconciler.Ready { + continue + } + victim = rep.Shards[0].NodeID + for _, s := range rep.Shards { + victimShards = append(victimShards, s.ID) + } + break + } + if victim == "" { + cancel() + wg.Wait() + t.Fatal("no ready replica to disturb") + } + + aliveMu.Lock() + alive[victim] = false + aliveMu.Unlock() + h.disp.killNode(victim, victimShards) + t.Logf("killed node %s (shards %v)", victim, victimShards) + + // The control plane should notice, rebuild elsewhere, and get back to two + // endpoints. The generous window covers unreachable -> lost -> backoff. + deadline := time.Now().Add(25 * time.Second) + recovered := false + for time.Now().Before(deadline) { + if len(h.rt.Registry().Snapshots(testModel)) == 2 && h.disp.liveShards() == 2 { + recovered = true + break + } + time.Sleep(50 * time.Millisecond) + } + + // Let traffic run a little longer on the rebuilt topology. + time.Sleep(500 * time.Millisecond) + cancel() + wg.Wait() + + t.Logf("requests ok=%d failed=%d (failures before the kill: %d)", + ok.Load(), failed.Load(), beforeFailures) + + if !recovered { + var placed []string + for _, rep := range h.ctl.Reconciler().Replicas() { + placed = append(placed, fmt.Sprintf("%s=%s@%v", rep.ID, rep.State, rep.NodeIDs())) + } + t.Fatalf("the cluster did not return to 2 replicas: %v", placed) + } + + // The victim must not be hosting anything any more. + for _, rep := range h.ctl.Reconciler().Replicas() { + if rep.OnNode(victim) { + t.Errorf("replica %s is still placed on the lost node %s", rep.ID, victim) + } + } + + // Traffic kept flowing: more succeeded after the failure than before it. + if ok.Load() <= baseline { + t.Errorf("no requests succeeded after the node was lost (%d before, %d total)", + baseline, ok.Load()) + } + // Some failures are expected in the window where a replica is gone and its + // replacement has not finished loading. What must not happen is total loss + // of service, which is what having a second replica is for. + if failed.Load() > ok.Load() { + t.Errorf("more requests failed (%d) than succeeded (%d); service was not maintained: %v", + failed.Load(), ok.Load(), firstErr.Load()) + } +}