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