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