From cba489798726a4ceb9813688880c2abf892a3002 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Tue, 3 Nov 2026 10:29:16 +0100 Subject: [PATCH] Deleting unused model files says what went, and sets go whole A sweep counted every file the nodes were asked to delete as freed as soon as it asked, and a node refused a set's directory as not empty, so tens of gigabytes were reported freed and nothing was. The agent removes a set's directory whole (a link only as a link), and the control plane waits for each node to list its files again, saying what it no longer lists and what it kept. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 15 +++ internal/agent/agent.go | 18 +++- internal/agent/removeweights_test.go | 32 ++++++ internal/cli/ctlweights.go | 34 +++--- internal/cli/tui_actions.go | 29 +++-- internal/control/adminapi/purge.go | 17 ++- internal/control/adminapi/weights.go | 123 +++++++++++++++++++--- internal/control/adminapi/weights_test.go | 71 +++++++++++++ 8 files changed, 293 insertions(+), 46 deletions(-) create mode 100644 internal/control/adminapi/weights_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 8052ac8e..4baeff64 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,21 @@ is only released once its **exit criterion** is met, not when its code lands. Until v1.0.0 the minor version is the breaking one: configuration formats and Go APIs change between minors without a deprecation period. +## Unreleased + +### Fixed + +- **Deleting the model files nothing uses said what it was asked to free, + not what it freed.** The console's "Delete model files nothing uses" and + `ctl weights --collect` added up every file the nodes were asked to + delete as soon as the request was sent, and a model of several files — + fetched into a directory of its own — was refused by the node as "not + empty", so a sweep said tens of gigabytes freed and nothing went. The + node now deletes such a directory whole (a link only as a link), and the + sweep waits for the nodes to list their files again: freed is what they + no longer list, and what they kept is said beside, with where to read + why. Removing a model with its files reports the same way. + ## v0.16.3 — 2026-11-03 Replicas counted per profile — seven on cards and four on CPUs under one name, the cards filled first — and every diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 7a90337a..9eef9b7e 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -884,7 +884,23 @@ func (a *Agent) removeWeights(paths []string) { a.log.Warn("refused to remove a file llmgw did not fetch", "path", p) continue } - if err := os.Remove(abs); err != nil && !os.IsNotExist(err) { + // A set — every file of one version, fetched into a directory of its + // own — goes whole: os.Remove refused it as not empty, and nothing + // was freed of what a model of several files left. Lstat, so a link + // is taken away and never what it points at. + info, err := os.Lstat(abs) + switch { + case os.IsNotExist(err): + continue + case err != nil: + a.log.Warn("could not look at cached weights", "path", p, "err", err) + continue + case info.IsDir(): + err = os.RemoveAll(abs) + default: + err = os.Remove(abs) + } + if err != nil { a.log.Warn("could not remove cached weights", "path", p, "err", err) continue } diff --git a/internal/agent/removeweights_test.go b/internal/agent/removeweights_test.go index 107a3dea..33066288 100644 --- a/internal/agent/removeweights_test.go +++ b/internal/agent/removeweights_test.go @@ -94,3 +94,35 @@ func TestWithNoCacheConfiguredNothingIsDeleted(t *testing.T) { t.Errorf("an agent with no cache directory deleted something: %v", err) } } + +// A model of several files is fetched into a directory of its own, and an +// unused one goes whole: it was refused as not empty, and nothing was freed. +func TestAnUnusedSetIsRemovedWhole(t *testing.T) { + dir := t.TempDir() + a := agentWithCache(t, dir) + set := weights.CachePathFor(dir, "https://mirror/coder/v1") + if err := os.MkdirAll(set, 0o755); err != nil { + t.Fatal(err) + } + write(t, filepath.Join(set, "model-00001-of-00002.gguf")) + write(t, filepath.Join(set, ".complete")) + // A link in the cache is taken away, not what it points at. + target := t.TempDir() + write(t, filepath.Join(target, "keep.gguf")) + link := weights.CachePathFor(dir, "https://mirror/linked/v1") + if err := os.Symlink(target, link); err != nil { + t.Fatal(err) + } + + a.removeWeights([]string{set, link}) + + if _, err := os.Stat(set); !os.IsNotExist(err) { + t.Errorf("the set is still there: %v", err) + } + if _, err := os.Lstat(link); !os.IsNotExist(err) { + t.Errorf("the link is still there: %v", err) + } + if _, err := os.Stat(filepath.Join(target, "keep.gguf")); err != nil { + t.Errorf("what the link pointed at was deleted: %v", err) + } +} diff --git a/internal/cli/ctlweights.go b/internal/cli/ctlweights.go index 54e2320c..5a5b8126 100644 --- a/internal/cli/ctlweights.go +++ b/internal/cli/ctlweights.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "strings" + "time" "github.com/spf13/cobra" @@ -87,13 +88,10 @@ func collectOrphans(cmd *cobra.Command, c *ctlClient, node string, asJSON bool) if err != nil { return err } - var results []struct { - Node string `json:"node"` - Files int `json:"files"` - Bytes int64 `json:"bytes"` - Error string `json:"error,omitempty"` - } - if err := c.post(cmd.Context(), "/api/v1/weights/collect", body, &results); err != nil { + var results []adminapi.CollectResult + // The control plane waits for the nodes to list their files without the + // deleted ones: half a minute at most. + if err := c.patient(time.Minute).post(cmd.Context(), "/api/v1/weights/collect", body, &results); err != nil { return err } if asJSON { @@ -103,7 +101,7 @@ func collectOrphans(cmd *cobra.Command, c *ctlClient, node string, asJSON bool) fmt.Fprintln(cmd.OutOrStdout(), "Nothing to reclaim.") return nil } - var freed int64 + var freed, left int64 for _, r := range results { if r.Error != "" { // Reported rather than fatal: the other nodes' space is still worth @@ -111,14 +109,20 @@ func collectOrphans(cmd *cobra.Command, c *ctlClient, node string, asJSON bool) fmt.Fprintf(cmd.OutOrStdout(), "%s: %s\n", r.Node, r.Error) continue } - fmt.Fprintf(cmd.OutOrStdout(), "%s: asked to remove %d files, %s\n", - r.Node, r.Files, fileSize(r.Bytes)) - freed += r.Bytes + line := fmt.Sprintf("%s: removed %d files, %s", r.Node, r.Files, fileSize(r.Bytes)) + if r.Left > 0 { + line += fmt.Sprintf("; %d still there, %s", r.Left, fileSize(r.LeftBytes)) + } + fmt.Fprintln(cmd.OutOrStdout(), line) + freed, left = freed+r.Bytes, left+r.LeftBytes + } + fmt.Fprintf(cmd.OutOrStdout(), "\n%s freed.\n", fileSize(freed)) + if left > 0 { + // Each node decides: one that keeps a path it cannot prove it + // fetched is the feature working. Its agent's log names each. + fmt.Fprintf(cmd.OutOrStdout(), "%s the nodes kept; each node's agent log says why "+ + "(journalctl -u llmgw-agent).\n", fileSize(left)) } - // "asked", not "removed": each node decides, and one that refuses a path it - // cannot prove it fetched is the feature working rather than failing. - fmt.Fprintf(cmd.OutOrStdout(), "\n%s requested. The nodes verify each path before deleting it.\n", - fileSize(freed)) return nil } diff --git a/internal/cli/tui_actions.go b/internal/cli/tui_actions.go index 8aad096d..293f4e6a 100644 --- a/internal/cli/tui_actions.go +++ b/internal/cli/tui_actions.go @@ -643,17 +643,30 @@ func (m consoleModel) nodeMenu(n adminapi.NodeView) *menuDialog { menuItem{"Delete model files nothing uses", func(m *consoleModel) tea.Cmd { return m.call("Unused files on "+n.ID, http.MethodPost, "/api/v1/weights/collect", map[string]string{"node": n.ID}, func(raw json.RawMessage) string { - var rs []struct { - Files int `json:"files"` - Bytes int64 `json:"bytes"` - } + // What the node no longer lists, and what it kept. + var rs []adminapi.CollectResult _ = json.Unmarshal(raw, &rs) - var files int - var b int64 + var files, left int + var b, lb int64 + var errs []string for _, r := range rs { - files, b = files+r.Files, b+r.Bytes + files, b, left, lb = files+r.Files, b+r.Bytes, left+r.Left, lb+r.LeftBytes + if r.Error != "" { + errs = append(errs, r.Error) + } } - return fmt.Sprintf("%d files, %.1f GB freed.", files, float64(b)/(1<<30)) + if len(rs) == 0 { + return "Nothing on " + n.ID + " that no model uses." + } + say := fmt.Sprintf("%d files removed, %.1f GB freed.", files, float64(b)/(1<<30)) + if left > 0 { + say += fmt.Sprintf(" %d files (%.1f GB) are still there: the node kept them, and its agent's log "+ + "says why (journalctl -u llmgw-agent there).", left, float64(lb)/(1<<30)) + } + if len(errs) > 0 { + say += " " + strings.Join(errs, "; ") + } + return say }) }}, menuItem{"Remove from the cluster…", func(m *consoleModel) tea.Cmd { diff --git a/internal/control/adminapi/purge.go b/internal/control/adminapi/purge.go index 67b4a39a..c17e72dd 100644 --- a/internal/control/adminapi/purge.go +++ b/internal/control/adminapi/purge.go @@ -216,7 +216,7 @@ func (a *API) purgeFromNodes(ctx context.Context, sources []string, others []cat names[filepath.Base(s)] = true } wanted := inUse(others) - var out []CollectResult + var rs []removal for _, n := range a.ctl.Inventory().Nodes() { size := map[string]int64{} have := make([]string, 0, len(n.Weights)) @@ -235,15 +235,14 @@ func (a *API) purgeFromNodes(ctx context.Context, sources []string, others []cat if len(paths) == 0 { continue } - res := CollectResult{Node: n.ID, Files: len(paths)} - for _, p := range paths { - res.Bytes += size[p] + rs = append(rs, removal{node: n.ID, paths: paths, size: size}) + } + out := a.removeConfirmed(ctx, rm, rs) + for _, r := range out { + if r.Error != "" || r.Left > 0 { + slog.Warn("a purged model's copies were not all deleted on a node", "node", r.Node, + "left", r.Left, "err", r.Error) } - if err := rm.RemoveWeights(ctx, n.ID, paths); err != nil { - res.Error, res.Files, res.Bytes = err.Error(), 0, 0 - slog.Warn("a purged model's copies could not be deleted on a node", "node", n.ID, "err", err) - } - out = append(out, res) } return out } diff --git a/internal/control/adminapi/weights.go b/internal/control/adminapi/weights.go index f171a882..c097ea30 100644 --- a/internal/control/adminapi/weights.go +++ b/internal/control/adminapi/weights.go @@ -1,11 +1,14 @@ package adminapi import ( + "context" "encoding/json" "net/http" "sort" + "time" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/weights" + "gitea.mynodes.xyz/ilker/llmgw/internal/control" ) // OrphanView is cached weights on one node that no catalog model accounts for. @@ -71,12 +74,100 @@ type CollectRequest struct { Node string `json:"node,omitempty"` } -// CollectResult is what was deleted on one node, or why nothing was. +// CollectResult is what was deleted on one node, or why nothing was: what +// the node no longer lists once asked, not what it was asked to delete. type CollectResult struct { Node string `json:"node"` Files int `json:"files"` Bytes int64 `json:"bytes"` - Error string `json:"error,omitempty"` + // Left and LeftBytes are what the node still lists: refused, or not + // gone by the time the sweep stopped waiting. Its agent's log says why. + Left int `json:"left,omitempty"` + LeftBytes int64 `json:"left_bytes,omitempty"` + Error string `json:"error,omitempty"` +} + +// confirmWait is how long a sweep waits for the nodes to list their files +// without the ones deleted: they scan theirs every heartbeat, five seconds +// by default, and a large set takes a moment to delete. +var confirmWait = 30 * time.Second + +// removal is what one node is asked to delete, with each file's size. +type removal struct { + node string + paths []string + size map[string]int64 +} + +// removeConfirmed asks each node to delete its paths and says what each then +// no longer lists. Counted as asked, a sweep said "25 GB freed" of files the +// node had refused and kept. +func (a *API) removeConfirmed(ctx context.Context, rm control.WeightRemover, rs []removal) []CollectResult { + out := make([]CollectResult, len(rs)) + pending := map[int]bool{} + for i, r := range rs { + out[i].Node = r.node + if err := rm.RemoveWeights(ctx, r.node, r.paths); err != nil { + out[i].Error = err.Error() + for _, p := range r.paths { + out[i].Left++ + out[i].LeftBytes += r.size[p] + } + continue + } + pending[i] = true + } + deadline := time.Now().Add(confirmWait) + for { + for i := range pending { + if left := a.stillListed(rs[i]); len(left) == 0 { + delete(pending, i) + } + } + if len(pending) == 0 || !time.Now().Before(deadline) { + break + } + select { + case <-ctx.Done(): + deadline = time.Now() + case <-time.After(time.Second): + } + } + for i, r := range rs { + if out[i].Error != "" { + continue + } + left := a.stillListed(r) + for _, p := range r.paths { + if left[p] { + out[i].Left++ + out[i].LeftBytes += r.size[p] + } else { + out[i].Files++ + out[i].Bytes += r.size[p] + } + } + } + return out +} + +// stillListed is which of a removal's paths its node still lists. +func (a *API) stillListed(r removal) map[string]bool { + left := map[string]bool{} + n, ok := a.ctl.Inventory().Get(r.node) + if !ok { + return left + } + want := map[string]bool{} + for _, p := range r.paths { + want[p] = true + } + for _, w := range n.Weights { + if want[w.Path] { + left[w.Path] = true + } + } + return left } // collectWeights deletes them. @@ -97,20 +188,26 @@ func (a *API) collectWeights(w http.ResponseWriter, r *http.Request) { // An empty body is the "all nodes" case and must not be an error. _ = json.NewDecoder(r.Body).Decode(&body) - out := make([]CollectResult, 0) + // A node that has gone since the list was built is reported rather than + // failing the whole sweep: the other nodes' space is still worth + // freeing, and this one will be offered again next time. + var rs []removal for _, o := range a.Orphans() { if body.Node != "" && o.Node != body.Node { continue } - res := CollectResult{Node: o.Node, Files: len(o.Paths), Bytes: o.Bytes} - // A node that has gone since the list was built is reported rather than - // failing the whole sweep: the other nodes' space is still worth - // freeing, and this one will be offered again next time. - if err := rm.RemoveWeights(r.Context(), o.Node, o.Paths); err != nil { - res.Error = err.Error() - res.Files, res.Bytes = 0, 0 - } - out = append(out, res) + rs = append(rs, removal{node: o.Node, paths: o.Paths, size: a.sizesOn(o.Node)}) } - writeJSON(w, http.StatusOK, out) + writeJSON(w, http.StatusOK, a.removeConfirmed(r.Context(), rm, rs)) +} + +// sizesOn is each file a node lists, by path, with its size. +func (a *API) sizesOn(node string) map[string]int64 { + out := map[string]int64{} + if n, ok := a.ctl.Inventory().Get(node); ok { + for _, w := range n.Weights { + out[w.Path] = w.SizeBytes + } + } + return out } diff --git a/internal/control/adminapi/weights_test.go b/internal/control/adminapi/weights_test.go new file mode 100644 index 00000000..0b15d998 --- /dev/null +++ b/internal/control/adminapi/weights_test.go @@ -0,0 +1,71 @@ +package adminapi + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "slices" + "strings" + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/weights" + "gitea.mynodes.xyz/ilker/llmgw/internal/control" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" +) + +// deletingDispatcher deletes what it is asked to as a node would, by its +// files dropping out of what the node lists — but for refused, which it +// keeps, as an agent keeps a file it cannot prove it fetched. +type deletingDispatcher struct { + nullDispatcher + inv *inventory.Store + refused string +} + +func (d *deletingDispatcher) RemoveWeights(_ context.Context, node string, paths []string) error { + n, _ := d.inv.Get(node) + n.Weights = slices.DeleteFunc(n.Weights, func(w discovery.LocalWeights) bool { + return w.Path != d.refused && slices.Contains(paths, w.Path) + }) + d.inv.Upsert(n) + return nil +} + +// What a sweep says it freed is what the node no longer lists: a file it +// kept is said to be left, not counted as freed. +func TestASweepSaysWhatWasDeletedNotWhatWasAsked(t *testing.T) { + confirmWait = 1500 * time.Millisecond + t.Cleanup(func() { confirmWait = 30 * time.Second }) + + gone := weights.CachePathFor("/var/lib/llmgw-weights", "http://mirror/old/v1/old.gguf") + set := weights.CachePathFor("/var/lib/llmgw-weights", "http://mirror/big/v1") + kept := weights.CachePathFor("/var/lib/llmgw-weights", "http://mirror/stuck/v1/stuck.gguf") + cat, _ := catalog.New(nil) + d := &deletingDispatcher{refused: kept} + ctl := control.New(control.DefaultConfig(), catalog.NewLive(cat), d, nullSink{}, slog.New(slog.DiscardHandler)) + d.inv = ctl.Inventory() + node := gpuNode("n1", 1) + node.Weights = []discovery.LocalWeights{{Path: gone, SizeBytes: 1 << 30}, {Path: set, SizeBytes: 20 << 30}, + {Path: kept, SizeBytes: 4 << 30}} + ctl.Inventory().Upsert(node) + a := New(ctl) + + rec := httptest.NewRecorder() + a.collectWeights(rec, httptest.NewRequest(http.MethodPost, "/api/v1/weights/collect", strings.NewReader(`{"node":"n1"}`))) + var got []CollectResult + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("%v: %s", err, rec.Body.String()) + } + if len(got) != 1 { + t.Fatalf("results %+v", got) + } + r := got[0] + if r.Files != 2 || r.Bytes != 21<<30 || r.Left != 1 || r.LeftBytes != 4<<30 { + t.Errorf("result %+v: want 2 files and 21 GB freed, 1 file of 4 GB left", r) + } +}