From 6320dd5b13d473266d9bf31811b6e5eacff81cc4 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sun, 8 Nov 2026 22:14:52 +0100 Subject: [PATCH] A new stream is told the node's files again Registration carries what the agent's startup scan found, and a heartbeat sends the list only when it changed since last sent; so a file deleted after the agent started came back on a control plane that restarted. The fingerprint is now forgotten with the provisioning one on every stream. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 +++++ internal/agent/agent.go | 16 +++++++++++++--- internal/agent/agent_test.go | 20 ++++++++++++++++++++ 3 files changed, 38 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 32e99ef..68a1af0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -64,6 +64,11 @@ APIs change between minors without a deprecation period. four, as the room left holds, and its memory figure is for that pool. A plan says what the model was trained for (`trained`). Strata stays at its 32768; vLLM's form is unchanged. +- **Deleted model files came back after a control plane restart.** An agent + registers with the files its startup scan found, and sends the list again + only when it changes; a file deleted after the agent started was listed + again by a control plane that restarted, until the next change. A new + stream now gets the list on its first heartbeat. - **`ctl weights` offered models in use for deletion.** A model of several files (a set, its address ending in `/`) is fetched into a directory named from the address without the slash; the sweep named it from the address diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 9eef9b7..63b8241 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -530,6 +530,18 @@ func (a *Agent) scan(ctx context.Context) error { return err } +// forgetSent is a new stream: a control plane that knows nothing of where +// installs stand, nor which files are here beyond what registration says — +// and registration says what the startup scan found. So the next heartbeat +// tells it both again. Without it, files deleted after the agent started came +// back on a control plane that restarted, and stayed until the next change. +func (a *Agent) forgetSent() { + a.mu.Lock() + defer a.mu.Unlock() + a.provSent = "" + a.lastWeights = "" +} + // session runs one connection until it breaks. func (a *Agent) session(ctx context.Context, server string) error { client := llmgwv1connect.NewAgentServiceClient(a.http, server) @@ -570,10 +582,8 @@ func (a *Agent) session(ctx context.Context, server string) error { reg.RunningShards = a.runningShards() a.mu.Lock() h := a.prov.Host - // A new stream is a control plane that knows nothing of where installs - // stand, so the next heartbeat tells it again. - a.provSent = "" a.mu.Unlock() + a.forgetSent() reg.Host = &llmgwv1.Host{ OsId: h.OSID, OsVersion: h.OSVersion, OsCodename: h.OSCodename, Arch: h.Arch, CpuFlags: h.CPUFlags, NvidiaDriver: h.NVIDIADriver, diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index c7ff2b6..41dcee8 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -925,3 +925,23 @@ func TestABandwidthTestIsRunOnTheAdapterAskedFor(t *testing.T) { t.Errorf("reply %+v, ran %+v", r, got) } } + +func TestANewStreamIsToldTheFilesAgain(t *testing.T) { + // Registration carries what the startup scan found; a file deleted since + // came back on a control plane that restarted, because the next + // heartbeat saw the list unchanged since it was last sent. + dir := t.TempDir() + h := start(t, func(c *Config) { c.ModelDirs = []string{dir} }) + if err := os.WriteFile(filepath.Join(dir, "0123456789abcdef-m.gguf"), []byte("x"), 0o644); err != nil { + t.Fatal(err) + } + h.ag.heartbeat() + if hb := h.ag.heartbeat().GetHeartbeat(); hb.GetWeightsChanged() { + t.Fatal("an unchanged list should not be sent again on the same stream") + } + h.ag.forgetSent() + hb := h.ag.heartbeat().GetHeartbeat() + if !hb.GetWeightsChanged() || len(hb.GetWeights()) != 1 { + t.Errorf("a new stream should be told the files: changed=%v %v", hb.GetWeightsChanged(), hb.GetWeights()) + } +}