From 2d14918b99afeb49d3a4ba1b6ec770a31972af46 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Wed, 7 Oct 2026 19:43:13 +0200 Subject: [PATCH] An adopted engine that turns ready after registration is reported The registration carries each shard's state as it is then. An adopted engine is health-checked after that, and nothing reported the ready that followed, so when the agent registered first the control plane kept the replica starting while it served. After llmgw restart on the PoC two replicas stayed so, one the only replica of qwen2.5-coder-14b. The session's watcher now reports every shard the registration called not ready once it is. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 6 ++++ internal/agent/agent.go | 40 +++++++++++++++++++++++-- internal/agent/watchexits_test.go | 50 ++++++++++++++++++++++++++++++- 3 files changed, 93 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f84a54..5960dc8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,12 @@ What was left of the field report from v0.10.6. ### Fixed +- **A replica no longer stays "starting" while it serves after a restart**: + an agent that registered before its adopted engines had answered their + health check never said they had, so the control plane kept them starting + and routed nothing to them — on the reference cluster, after `llmgw + restart`, the only replica of a model. The agent now reports each shard + the registration called not ready once it is. - **A failover no longer replaces healthy replicas**: a standby that took over placed at once, before the agents had reconnected and said what runs, so it started a second copy of replicas it did not know of and then drained diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 0f5cde4..c12986e 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -417,6 +417,17 @@ func (a *Agent) weights() *weights.Fetcher { } // runningShards describes what this agent is supervising, for registration. +// unready are the shards a registration reported as not yet ready. +func unready(running []*llmgwv1.RunningShard) map[string]bool { + out := map[string]bool{} + for _, r := range running { + if r.GetState() != agentsvc.StateToWire(supervisor.Ready) { + out[r.GetShardId()] = true + } + } + return out +} + func (a *Agent) runningShards() []*llmgwv1.RunningShard { var out []*llmgwv1.RunningShard for _, st := range a.sup.Statuses() { @@ -571,7 +582,7 @@ func (a *Agent) session(ctx context.Context, server string) error { a.sendLoop(ctx, stream, out) }() defer wg.Wait() - go a.watchExits(ctx, out) + go a.watchExits(ctx, out, unready(reg.RunningShards)) go a.forwardKV(ctx, out) go a.reportCompute(ctx, out) @@ -1191,7 +1202,14 @@ func (a *Agent) startShard(ctx context.Context, m *llmgwv1.StartShard, out chan< // reported the engine dying; and an adopted engine had no watcher at all — // on node06 one was killed and its replica stayed "ready" for the control // plane, routed to and failing, indefinitely. -func (a *Agent) watchExits(ctx context.Context, out chan<- *llmgwv1.AgentMessage) { +// +// It also reports the shards the registration said were not ready yet once +// they are. An adopted engine is health-checked after the agent registers, +// and a shard started on the stream before is reported on that stream; so +// when the registration came first nothing ever told the control plane, and +// the replica stayed "starting" while it served — on node05, after a +// control plane restart, the only replica of a model. +func (a *Agent) watchExits(ctx context.Context, out chan<- *llmgwv1.AgentMessage, pending map[string]bool) { t := time.NewTicker(time.Second) defer t.Stop() ready := map[string]bool{} @@ -1200,6 +1218,24 @@ func (a *Agent) watchExits(ctx context.Context, out chan<- *llmgwv1.AgentMessage switch st.State { case supervisor.Ready: ready[st.ID] = true + if !pending[st.ID] { + continue + } + delete(pending, st.ID) + sh, err := a.sup.Get(st.ID) + if err != nil { + continue + } + msg := &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Shard{Shard: &llmgwv1.ShardStatus{ + NodeId: a.cfg.NodeID, ReplicaId: sh.Spec.ReplicaID, ShardId: st.ID, + State: agentsvc.StateToWire(st.State), + Compute: computeToWire(a.sup.ComputeOf(st.ID)), + }}} + select { + case out <- msg: + case <-ctx.Done(): + return + } case supervisor.Failed: if !ready[st.ID] { continue // never ready here: its start reported it diff --git a/internal/agent/watchexits_test.go b/internal/agent/watchexits_test.go index 9343fe9..7eae9fb 100644 --- a/internal/agent/watchexits_test.go +++ b/internal/agent/watchexits_test.go @@ -10,6 +10,7 @@ import ( llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc" ) type alwaysReady struct{} @@ -40,7 +41,7 @@ func TestAnAdoptedEngineThatDiesIsReported(t *testing.T) { out := make(chan *llmgwv1.AgentMessage, 4) ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go a.watchExits(ctx, out) + go a.watchExits(ctx, out, nil) time.Sleep(1500 * time.Millisecond) // seen ready _ = cmd.Process.Kill() @@ -54,3 +55,50 @@ func TestAnAdoptedEngineThatDiesIsReported(t *testing.T) { t.Fatal("the adopted engine died and nothing was reported") } } + +// Registered while its health check still ran: the ready that follows is +// reported, or the control plane keeps the replica starting while it serves. +func TestAShardRegisteredStartingIsReportedReady(t *testing.T) { + sup, err := supervisor.New(supervisor.Config{WorkDir: t.TempDir(), RunAs: "-"}, alwaysReady{}, slog.New(slog.DiscardHandler)) + if err != nil { + t.Fatal(err) + } + cmd := exec.Command("sleep", "60") + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + defer func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }() + sh, err := sup.Adopt(supervisor.Orphan{ShardID: "m-r0", ReplicaID: "m", PID: cmd.Process.Pid, + Spec: engines.ShardSpec{ReplicaID: "m", ShardID: "m-r0", HTTPPort: 1}, Alive: true}) + if err != nil { + t.Fatal(err) + } + + a := &Agent{sup: sup, log: slog.New(slog.DiscardHandler)} + pending := unready(a.runningShards()) + if !pending["m-r0"] { + t.Fatalf("registration reported it ready already: %v", pending) + } + out := make(chan *llmgwv1.AgentMessage, 4) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go a.watchExits(ctx, out, pending) + if err := sup.WaitReady(t.Context(), sh, "http://127.0.0.1:1", "m", 5*time.Second); err != nil { + t.Fatal(err) + } + + select { + case m := <-out: + s := m.GetShard() + if s.GetShardId() != "m-r0" || s.GetState() != agentsvc.StateToWire(supervisor.Ready) { + t.Errorf("reported %+v", s) + } + case <-time.After(10 * time.Second): + t.Fatal("it became ready and nothing was reported") + } + select { + case m := <-out: + t.Errorf("reported twice: %+v", m.GetShard()) + case <-time.After(2500 * time.Millisecond): + } +}