A new stream is told the node's files again
CI / test (push) Successful in 9m31s

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 <noreply@anthropic.com>
This commit is contained in:
Ilker ManapandClaude Opus 5.5 committed 2026-11-08 22:14:52 +01:00
1 parent 5732a5ebda
commit 6320dd5b13
3 files changed
+38 -3

No files matched your search

+5
View File
@@ -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
+13 -3
View File
@@ -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,
+20
View File
@@ -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())
}
}