diff --git a/BACKLOG.md b/BACKLOG.md index c3bd760..add3dc1 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -44,7 +44,7 @@ providers, of L5 cache-write and audio prices. L6 and L7 are llmgw-80's; L8 (A2A **Models we can trust** (section M, 2026-10-25): M1 done (pickles refused; the explorer lists only what this cluster's hardware holds, with CPU and GPU tok/s). Next M2 (every download pinned and checked against Hugging -Face), then M5 (engines without the internet), M3, M4, M6. +Face, done 2026-10-30), then M5 (engines without the internet), M3, M4, M6. **Governance.** 7. **C1**: a tenant-admin role, and forms hidden from viewers. @@ -1823,7 +1823,14 @@ pickle files. `allow_pickle: true` on a model lets it through, on purpose. On th PoC, `triposr` (`hf:stabilityai/TripoSR/model.ckpt`) is the one model that needs it. -### M2. Every download pinned and checked against Hugging Face (M) +### M2. Every download pinned and checked against Hugging Face (M) — done 2026-10-30 +Done: what is added from Hugging Face is pinned (the file's commit URL, or +`--revision` for vLLM/SGLang) with each file's sha256 in the origin; the node +refuses and deletes a copy that is not it, naming both, and keeps the +checksum beside a cached file; the mirror checks what `ctl model serve` +fetches, `mirror add` a Hugging Face URL. Left: models declared in the +settings are not pinned, the admin page's and the console's mirror fetches +send no sha256, and adapters by hub id are not pinned. - `hf:org/repo/file` follows `main` and is checked by nothing; a bare `org/repo` given to vLLM is downloaded by vLLM, of whatever `main` is. Both: resolve to the commit when the model is added, keep it in the diff --git a/CHANGELOG.md b/CHANGELOG.md index 30bc2bc..591628d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,12 @@ APIs change between minors without a deprecation period. waiting out a busy engine rather than failing the document. About a kilobyte a passage leaves the machine instead of a vector down and up again, some thirty times more. `llmgw-kb` does so with a gateway that says it can (`server_embeds` in a collection's answer); `--embed-here` keeps the old way. +- **Weights pinned to a commit, and checked against what Hugging Face published.** A model added from Hugging Face + (`hf:`, a resolve URL on a branch, or a repository vLLM or SGLang downloads) is pinned to the commit it is at + then — the file's URL at that commit, or `--revision` — and each file's published sha256 goes in its origin. A node + refuses and deletes a copy that is not it, naming both checksums; the mirror does the same for what `ctl model + serve` fetches, and `llmgw mirror add` for a Hugging Face URL (`--sha256` for others). Nodes need this version to + check; older ones fetch as before. ### Fixed diff --git a/api/gen/llmgw/v1/agent.pb.go b/api/gen/llmgw/v1/agent.pb.go index fc3419a..1834f37 100644 --- a/api/gen/llmgw/v1/agent.pb.go +++ b/api/gen/llmgw/v1/agent.pb.go @@ -3849,7 +3849,11 @@ type EngineProfile struct { KvEvents bool `protobuf:"varint,13,opt,name=kv_events,json=kvEvents,proto3" json:"kv_events,omitempty"` // allow_pickle lets the node fetch weights in a pickle format, which run // code when loaded: the catalog's allow_pickle, set on purpose. - AllowPickle bool `protobuf:"varint,14,opt,name=allow_pickle,json=allowPickle,proto3" json:"allow_pickle,omitempty"` + AllowPickle bool `protobuf:"varint,14,opt,name=allow_pickle,json=allowPickle,proto3" json:"allow_pickle,omitempty"` + // model_sha256 is the sha256 the registry published for the file + // model_path names, from the catalog's origin: the node refuses a copy + // that is not it. Empty when nothing is known to check against. + ModelSha256 string `protobuf:"bytes,15,opt,name=model_sha256,json=modelSha256,proto3" json:"model_sha256,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -3982,6 +3986,13 @@ func (x *EngineProfile) GetAllowPickle() bool { return false } +func (x *EngineProfile) GetModelSha256() string { + if x != nil { + return x.ModelSha256 + } + return "" +} + // RemoveWeights deletes cached model files a node no longer needs. // // The paths are computed by the control plane from the catalog, but the agent @@ -5536,7 +5547,7 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "profile_fp\x18\x0f \x01(\tR\tprofileFp\x12\x17\n" + "\aapi_key\x18\x10 \x01(\tR\x06apiKey\x12\x1f\n" + "\vcpu_threads\x18\x11 \x01(\x05R\n" + - "cpuThreads\"\xf0\x03\n" + + "cpuThreads\"\x93\x04\n" + "\rEngineProfile\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x16\n" + "\x06engine\x18\x02 \x01(\tR\x06engine\x12\x1b\n" + @@ -5553,7 +5564,8 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "\x11served_model_name\x18\v \x01(\tR\x0fservedModelName\x121\n" + "\badapters\x18\f \x03(\v2\x15.llmgw.v1.LoraAdapterR\badapters\x12\x1b\n" + "\tkv_events\x18\r \x01(\bR\bkvEvents\x12!\n" + - "\fallow_pickle\x18\x0e \x01(\bR\vallowPickle\x1a6\n" + + "\fallow_pickle\x18\x0e \x01(\bR\vallowPickle\x12!\n" + + "\fmodel_sha256\x18\x0f \x01(\tR\vmodelSha256\x1a6\n" + "\bEnvEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"%\n" + diff --git a/api/proto/llmgw/v1/agent.proto b/api/proto/llmgw/v1/agent.proto index 2d89258..91ab189 100644 --- a/api/proto/llmgw/v1/agent.proto +++ b/api/proto/llmgw/v1/agent.proto @@ -767,6 +767,10 @@ message EngineProfile { // allow_pickle lets the node fetch weights in a pickle format, which run // code when loaded: the catalog's allow_pickle, set on purpose. bool allow_pickle = 14; + // model_sha256 is the sha256 the registry published for the file + // model_path names, from the catalog's origin: the node refuses a copy + // that is not it. Empty when nothing is known to check against. + string model_sha256 = 15; } // RemoveWeights deletes cached model files a node no longer needs. diff --git a/docs/install.md b/docs/install.md index af8b824..45e5dbc 100644 --- a/docs/install.md +++ b/docs/install.md @@ -897,6 +897,24 @@ the node — an adapter given by its id, a TensorRT-LLM model's — is refused when the model is added, unless it says `allow_pickle`: mirror its safetensors and give that instead. +**Weights are pinned to a commit and checked.** A model added with +`ctl model add`, the API or the console, whose weights come from Hugging Face, +is pinned to the commit the repository is at when it is added: + +- `hf:org/repo/file` and a `…/resolve/main/…` URL become that commit's URL; +- a repository vLLM or SGLang downloads itself is given `--revision`; +- the sha256 Hugging Face published for each file goes in the model's origin + (its definition, `GET /api/v1/models/NAME`, shows it). + +A node then refuses a copy of the file that is not it, naming both checksums, +and deletes it; a cached copy is checked once and its checksum kept beside it. +The mirror does the same for a model `ctl model serve` fetches, and +`llmgw mirror add` for a Hugging Face URL (`--sha256` for any other). A new +upload to the repository reaches the cluster only through `ctl model update`, +which says what changed. Not pinned: models declared in the cluster's settings, +which are not added this way, and a vLLM model whose arguments already say +`--revision` keeps that one. + Put them where the node can read them. `scp`, a shared filesystem, or the engine's own downloader are all valid, and which you want depends on how many nodes you have. diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 7565252..1540dca 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -878,6 +878,7 @@ func (a *Agent) removeWeights(paths []string) { a.log.Warn("could not remove cached weights", "path", p, "err", err) continue } + _ = os.Remove(weights.ChecksumPath(abs)) a.log.Info("removed cached weights no model asks for", "path", p) } } @@ -1133,7 +1134,7 @@ func (a *Agent) startShard(ctx context.Context, m *llmgwv1.StartShard, out chan< }, }) }) - if path, err := a.weights().Ensure(ctx, spec.Profile.ModelPath, spec.Profile.AllowPickle); err != nil { + if path, err := a.weights().EnsureChecked(ctx, spec.Profile.ModelPath, spec.Profile.AllowPickle, spec.Profile.ModelSHA256); err != nil { stop() report(supervisor.Failed, err.Error(), nil) return diff --git a/internal/agent/engines/engine.go b/internal/agent/engines/engine.go index 160e96c..5a86470 100644 --- a/internal/agent/engines/engine.go +++ b/internal/agent/engines/engine.go @@ -50,6 +50,10 @@ type Profile struct { // AllowPickle lets the node fetch weights in a pickle format, which run // code when loaded; the catalog's allow_pickle. AllowPickle bool `json:"allow_pickle,omitempty"` + // ModelSHA256 is the sha256 the registry published for the file + // ModelPath names; the node refuses a copy that is not it. Empty when + // nothing is known to check against. + ModelSHA256 string `json:"model_sha256,omitempty"` } // ShardSpec is one shard of one replica on this node. diff --git a/internal/agent/weights/fetch.go b/internal/agent/weights/fetch.go index 962d205..2c39b0e 100644 --- a/internal/agent/weights/fetch.go +++ b/internal/agent/weights/fetch.go @@ -94,6 +94,13 @@ var ErrNotLocal = errors.New("the weights are not on this node and the source sa // A source in a pickle format, or a set with such a file in it, is refused // before anything is fetched unless allowPickle; see Pickled. func (f *Fetcher) Ensure(ctx context.Context, source string, allowPickle bool) (path string, err error) { + return f.EnsureChecked(ctx, source, allowPickle, "") +} + +// EnsureChecked is Ensure for a file whose sha256 the registry published: a +// copy that is not it, downloaded or cached, is refused rather than served. +// An empty sha256 checks nothing. +func (f *Fetcher) EnsureChecked(ctx context.Context, source string, allowPickle bool, sha256 string) (path string, err error) { ctx, span := obs.Tracer(tracerName).Start(ctx, "fetch weights", trace.WithAttributes(attribute.String("llmgw.weights.source", redact(source)))) defer func() { @@ -103,7 +110,7 @@ func (f *Fetcher) Ensure(ctx context.Context, source string, allowPickle bool) ( if !allowPickle && !IsSet(source) && Pickled(source) { return "", refusePickle(source) } - return f.ensure(ctx, source, allowPickle) + return f.ensure(ctx, source, allowPickle, strings.ToLower(sha256)) } // tracerName names the spans this package opens. @@ -125,7 +132,7 @@ func redact(source string) string { return u.String() } -func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool) (string, error) { +func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool, want string) (string, error) { if source == "" { return "", fmt.Errorf("no source given") } @@ -156,8 +163,13 @@ func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool) ( return "", err } if _, err := os.Stat(target); err == nil { - outcome(ctx, "cached") - return target, nil + if ok, err := f.cachedIs(target, want); err != nil { + return "", err + } else if ok { + outcome(ctx, "cached") + return target, nil + } + // Not the file the catalog names: fetched again below, and checked. } f.mu.Lock() @@ -179,7 +191,7 @@ func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool) ( f.mu.Unlock() outcome(ctx, "downloaded") - call.path, call.err = f.fetch(ctx, source, target) + call.path, call.err = f.fetchChecked(ctx, source, target, want) close(call.done) f.mu.Lock() @@ -193,6 +205,72 @@ func (f *Fetcher) fetch(ctx context.Context, source, target string) (string, err return f.fetchAs(ctx, source, source, target) } +// fetchChecked is fetch for a file whose sha256 is known: a download that +// is not it is deleted and refused, naming both, and one that is has its +// checksum kept beside it so a cached copy is not read again at each start. +func (f *Fetcher) fetchChecked(ctx context.Context, source, target, want string) (string, error) { + got, err := f.fetch(ctx, source, target) + if err != nil || want == "" { + return got, err + } + sum, err := fileSHA256(target) + if err != nil { + return "", err + } + if sum != want { + _ = os.Remove(target) + return "", fmt.Errorf("%s arrived with sha256 %s, and the registry published %s for it: refused and deleted", + redact(source), sum, want) + } + _ = writeChecksum(target, sum) + return target, nil +} + +// cachedIs says whether a cached file is the one whose sha256 is want: by +// the checksum kept beside it, or worked out once and then kept. One that +// is not is deleted, to be fetched again. Any file is, when want is empty. +func (f *Fetcher) cachedIs(target, want string) (bool, error) { + if want == "" || readChecksum(target) == want { + return true, nil + } + sum, err := fileSHA256(target) + if err != nil { + return false, err + } + if sum == want { + _ = writeChecksum(target, sum) + return true, nil + } + if f.Log != nil { + f.Log.Warn("a cached file is not the one the catalog names; fetching it again", + "path", target, "sha256", sum, "want", want) + } + _ = os.Remove(checksumPath(target)) + return false, os.Remove(target) +} + +// checksumPath is where a cached file's sha256 is kept: beside it, hidden, +// so no scan takes it for weights. +func checksumPath(file string) string { + return filepath.Join(filepath.Dir(file), "."+filepath.Base(file)+".sha256") +} + +func writeChecksum(file, sum string) error { + return os.WriteFile(checksumPath(file), []byte(sum+"\n"), 0o644) //nolint:gosec // a checksum, for anyone to read +} + +func readChecksum(file string) string { + b, err := os.ReadFile(checksumPath(file)) //nolint:gosec // beside a file this fetcher wrote + if err != nil { + return "" + } + return strings.TrimSpace(string(b)) +} + +// ChecksumPath is where the sha256 of a cached file is kept, for whoever +// deletes the file to delete with it. +func ChecksumPath(file string) string { return checksumPath(file) } + // fetchAs downloads one URL but reports progress under another name. // // The two differ for a set: the shards waiting on it asked for the version, not diff --git a/internal/agent/weights/fetch_test.go b/internal/agent/weights/fetch_test.go index ed17277..e69837a 100644 --- a/internal/agent/weights/fetch_test.go +++ b/internal/agent/weights/fetch_test.go @@ -1,6 +1,8 @@ package weights import ( + "crypto/sha256" + "encoding/hex" "errors" "net/http" "net/http/httptest" @@ -287,3 +289,53 @@ func TestAnAbsolutePathThatIsMissingIsStillAnError(t *testing.T) { t.Errorf("err = %v, want ErrNotLocal", err) } } + +// A file whose sha256 the registry published is served only as that file: +// changed upstream after the model was added, it is refused with both +// checksums named, and a cached copy that is not it is fetched again. +func TestAFileThatIsNotWhatTheRegistryPublishedIsRefused(t *testing.T) { + body := "the weights as published" + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(body)) + })) + defer srv.Close() + sum := sha256.Sum256([]byte(body)) + want := hex.EncodeToString(sum[:]) + f := &Fetcher{CacheDir: t.TempDir()} + src := srv.URL + "/model.gguf" + + path, err := f.EnsureChecked(t.Context(), src, false, strings.ToUpper(want)) + if err != nil { + t.Fatal(err) + } + if readChecksum(path) != want { + t.Errorf("the checksum was not kept beside it: %q", readChecksum(path)) + } + + // The cached copy changed under it: fetched again, and the right one kept. + if err := os.WriteFile(path, []byte("tampered"), 0o644); err != nil { + t.Fatal(err) + } + _ = os.Remove(checksumPath(path)) + if _, err := f.EnsureChecked(t.Context(), src, false, want); err != nil { + t.Fatal(err) + } + if b, _ := os.ReadFile(path); string(b) != body { + t.Errorf("the cached copy that was not it was served: %q", b) + } + + // Upstream changed after the model was added. + body = "weights someone replaced" + f2 := &Fetcher{CacheDir: t.TempDir()} + _, err = f2.EnsureChecked(t.Context(), src, false, want) + if err == nil || !strings.Contains(err.Error(), want) || !strings.Contains(err.Error(), "sha256") { + t.Fatalf("not refused, or without both checksums: %v", err) + } + if _, err := os.Stat(CachePathFor(f2.CacheDir, src)); !os.IsNotExist(err) { + t.Error("the file that was not it was kept") + } + // With nothing to check against, it is fetched as before. + if _, err := f2.Ensure(t.Context(), src, false); err != nil { + t.Error(err) + } +} diff --git a/internal/cli/mirror.go b/internal/cli/mirror.go index 35f4a22..f517197 100644 --- a/internal/cli/mirror.go +++ b/internal/cli/mirror.go @@ -16,6 +16,7 @@ import ( "github.com/spf13/cobra" "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/hf" "gitea.mynodes.xyz/ilker/llmgw/internal/mirror" "gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf" ) @@ -57,7 +58,7 @@ func newMirrorCmd() *cobra.Command { } func newMirrorAddCmd(dir *string) *cobra.Command { - var model, version, as string + var model, version, as, sha, registry string c := &cobra.Command{ Use: "add ...", @@ -79,7 +80,7 @@ func newMirrorAddCmd(dir *string) *cobra.Command { if model == "" || version == "" { return fmt.Errorf("--model and --version are required: they are the layout") } - if as != "" && len(args) > 1 { + if (as != "" || sha != "") && len(args) > 1 { return fmt.Errorf("--as names one file; with %d URLs each keeps its own name", len(args)) } out := cmd.OutOrStdout() @@ -90,6 +91,21 @@ func newMirrorAddCmd(dir *string) *cobra.Command { file = mirror.LastPathElement(source) } e := mirror.Entry{Model: model, Version: version, File: file} + // From the registry: what it published for the file, to + // compare with what arrives. + want := strings.ToLower(sha) + if id, rev, path, ok := hf.FileSource(source, registry); ok && want == "" { + repo, err := hf.New(registry, 30*time.Second).RepoAt(cmd.Context(), id, rev) + if err != nil { + return fmt.Errorf("looking up %s in the registry, for the sha256 it published: %w "+ + "(--sha256 gives it instead)", id, err) + } + for _, f := range repo.Files { + if f.Path == path { + want = f.SHA256 + } + } + } fmt.Fprintf(out, "fetching %s\n", source) added, err := mirror.Add(cmd.Context(), *dir, source, e, func(done, total int64) { if total > 0 { @@ -102,6 +118,14 @@ func newMirrorAddCmd(dir *string) *cobra.Command { if err != nil { return err } + if want != "" && added.SHA256 != want { + _ = mirror.Remove(*dir, e) + return fmt.Errorf("%s arrived with sha256 %s, and %s was published for it: deleted", + source, added.SHA256, want) + } + if want != "" { + fmt.Fprintln(out, " sha256 matches the one published") + } fmt.Fprintf(out, "%s/%s/%s %s\n sha256 %s\n", added.Entry.Model, added.Entry.Version, added.Entry.File, fileSize(added.Entry.Bytes), added.SHA256) @@ -123,6 +147,8 @@ func newMirrorAddCmd(dir *string) *cobra.Command { c.Flags().StringVar(&model, "model", "", "the model's name, one directory level") c.Flags().StringVar(&version, "version", "", "the version, one directory level; a date works well") c.Flags().StringVar(&as, "as", "", "the filename to save as, for one URL; taken from the URL when omitted") + c.Flags().StringVar(&sha, "sha256", "", "the sha256 the file must have, for one URL; looked up for a Hugging Face URL") + c.Flags().StringVar(®istry, "registry", hf.DefaultEndpoint, "the registry a resolve URL is looked up in") return c } diff --git a/internal/cli/server.go b/internal/cli/server.go index 8f9d2bd..511bb01 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -453,6 +453,9 @@ func newServerCmd() *cobra.Command { if st := rt.Store(); st != nil { svc := modelsvc.New(cfg.Models, st, admin, log) svc.SetCheck(servableCheck(cfg.WantedEngines()["llamacpp"])) + // What is added from the registry is pinned to the + // commit it is at, with each file's sha256 for the nodes. + svc.SetRegistry(hf.New(cfg.Mirror.Registry, 30*time.Second)) // Other llmgw clusters' models, offered as PEER/MODEL: // read before the catalog is built, so a model that falls // back to one is whole at startup. diff --git a/internal/control/agentsvc/convert.go b/internal/control/agentsvc/convert.go index d4f8466..5c32539 100644 --- a/internal/control/agentsvc/convert.go +++ b/internal/control/agentsvc/convert.go @@ -236,6 +236,7 @@ func StartShardFromCommand(cmd reconciler.ShardCommand) *llmgwv1.StartShard { Adapters: adaptersToWire(p.Adapters), KvEvents: p.KVEvents, AllowPickle: p.AllowPickle, + ModelSha256: p.ModelSHA256, }, StartupTimeoutMs: cmd.StartupTimeout.Milliseconds(), Pool: cmd.Pool, @@ -274,6 +275,7 @@ func ShardSpecFromStart(m *llmgwv1.StartShard) (engines.ShardSpec, time.Duration Adapters: adaptersFromWire(p.GetAdapters()), KVEvents: p.GetKvEvents(), AllowPickle: p.GetAllowPickle(), + ModelSHA256: p.GetModelSha256(), }, } return spec, time.Duration(m.GetStartupTimeoutMs()) * time.Millisecond diff --git a/internal/control/catalog/catalog.go b/internal/control/catalog/catalog.go index 23177b0..b38fa62 100644 --- a/internal/control/catalog/catalog.go +++ b/internal/control/catalog/catalog.go @@ -1327,6 +1327,26 @@ func (m Model) SourceOf(p Profile) string { return m.Source } +// OriginSHA256 is the sha256 the registry published for the one file a +// source names, from the model's origin: "" for a set, a repository, or a +// file the origin does not list. +func (m Model) OriginSHA256(source string) string { + if m.Origin == nil || strings.HasSuffix(source, "/") { + return "" + } + p := source + if u, err := url.Parse(source); err == nil && u.Path != "" { + p = u.Path + } + name := path.Base(p) + for _, f := range m.Origin.Files { + if f.SHA256 != "" && path.Base(f.Path) == name { + return f.SHA256 + } + } + return "" +} + // WeightSources are every source a model's replicas may load: the model's // and each profile's of its own, once each. What a node must not have // collected from under it. @@ -1402,6 +1422,7 @@ func (m Model) EngineProfile(name string) (engines.Profile, bool) { Adapters: m.engineAdapters(), KVEvents: p.KVEvents, AllowPickle: m.AllowPickle, + ModelSHA256: m.OriginSHA256(m.SourceOf(p)), }.WithRuntimeLoRA(m.MaxLoadedAdapters), true } return engines.Profile{}, false diff --git a/internal/control/catalog/origin_sha_test.go b/internal/control/catalog/origin_sha_test.go new file mode 100644 index 0000000..47f6980 --- /dev/null +++ b/internal/control/catalog/origin_sha_test.go @@ -0,0 +1,27 @@ +package catalog + +import "testing" + +func TestTheFileASourceNamesIsGivenItsPublishedChecksum(t *testing.T) { + m := Model{Name: "q", Source: "http://10.0.0.146:8081/q/abc/q4.gguf", Origin: &Origin{Repo: "o/r", Revision: "abc", + Files: []OriginFile{{Path: "sub/q4.gguf", SHA256: "aaaa"}, {Path: "config.json"}}}} + for src, want := range map[string]string{ + m.Source: "aaaa", + "https://huggingface.co/o/r/resolve/abc/sub/q4.gguf": "aaaa", + "http://10.0.0.146:8081/q/abc/": "", // a set + "o/r": "", + "http://10.0.0.146:8081/q/abc/config.json": "", // no checksum published + } { + if got := m.OriginSHA256(src); got != want { + t.Errorf("%s: %q, want %q", src, got, want) + } + } + m.Profiles = []Profile{{Name: "cpu", Engine: "llamacpp"}} + if ep, ok := m.EngineProfile("cpu"); !ok || ep.ModelSHA256 != "aaaa" { + t.Errorf("the node is not told it: %+v", ep) + } + m.Origin = nil + if m.OriginSHA256(m.Source) != "" { + t.Error("a checksum with no origin") + } +} diff --git a/internal/control/modelserve/modelserve.go b/internal/control/modelserve/modelserve.go index 9dc12bf..43c2a18 100644 --- a/internal/control/modelserve/modelserve.go +++ b/internal/control/modelserve/modelserve.go @@ -848,6 +848,9 @@ func (s *Server) mirror(ctx context.Context, j *Job, repo hf.Repo, set hf.FileSe } resp, err := s.deps.Mirror.Fetch(ctx, mirror.FetchRequest{ Source: repo.DownloadURL("", f.Path), Model: name, Version: version, File: file, + // The mirror deletes a copy that is not what the registry + // published, rather than holding it for the nodes. + SHA256: f.SHA256, }) // Already there is not a failure: the same commit fetched before. if err != nil && !strings.Contains(err.Error(), "already") { diff --git a/internal/control/modelsvc/modelsvc.go b/internal/control/modelsvc/modelsvc.go index 9496bbf..ecdc3a2 100644 --- a/internal/control/modelsvc/modelsvc.go +++ b/internal/control/modelsvc/modelsvc.go @@ -51,6 +51,8 @@ type Service struct { // check refuses a model that cannot run before it is added; see // SetCheck. check func(context.Context, catalog.Model) error + // registry pins what is added from it to a commit; nil pins nothing. + registry Registry // federated are the models federated clusters offer; see SetFederated. federated func() []catalog.Model @@ -133,6 +135,11 @@ func (s *Service) add(ctx context.Context, m catalog.Model, by string, onlyNew b "placement could not tell when a node is full", m.Name, p.Name) } } + if s.registry != nil { + if err := pin(ctx, s.registry, &m); err != nil { + return err + } + } if s.check != nil { if err := s.check(ctx, m); err != nil { return err diff --git a/internal/control/modelsvc/pin.go b/internal/control/modelsvc/pin.go new file mode 100644 index 0000000..46f85de --- /dev/null +++ b/internal/control/modelsvc/pin.go @@ -0,0 +1,133 @@ +package modelsvc + +import ( + "context" + "fmt" + "regexp" + "slices" + "strings" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/hf" +) + +// A model added with weights from the registry is pinned to the commit they +// were at: a file named by hf: or by a resolve URL on a branch becomes that +// commit's URL, and a repository an engine downloads itself is given +// --revision. The registry's sha256 of each file goes in the model's origin, +// for the nodes to check what they fetch against. Without it, a new upload +// to the repository reached the nodes that fetched after it, unseen, beside +// nodes that still had the old one. + +// Registry is where a repository's commit and files are looked up. +type Registry interface { + RepoAt(ctx context.Context, id, rev string) (hf.Repo, error) + Base() string +} + +// SetRegistry has every model added from now on pinned against it. +func (s *Service) SetRegistry(r Registry) { s.registry = r } + +// engineDownloads are the engines that download a repository named by its id +// themselves, and take --revision. +var engineDownloads = map[string]bool{"vllm": true, "sglang": true} + +var repoID = regexp.MustCompile(`^[A-Za-z0-9][\w.-]*/[\w.-]+$`) + +// pin pins a model's registry sources, in place. One already given an origin +// — served through the mirror, which records it — is left as it is. +func pin(ctx context.Context, reg Registry, m *catalog.Model) error { + if m.Origin != nil { + return nil + } + repos := map[string]hf.Repo{} + lookup := func(id, rev string) (hf.Repo, error) { + key := id + "@" + rev + if r, ok := repos[key]; ok { + return r, nil + } + r, err := reg.RepoAt(ctx, id, rev) + if err != nil { + return r, fmt.Errorf("model %q: looking up %s in the registry, to pin it to a commit: %w", m.Name, id, err) + } + repos[key] = r + return r, nil + } + record := func(r hf.Repo, files ...hf.File) { + if m.Origin == nil { + m.Origin = &catalog.Origin{Repo: r.ID, Revision: r.Revision} + } + if m.Origin.Repo != r.ID || m.Origin.Revision != r.Revision { + return // one origin a model; a second repository is pinned, not recorded + } + for _, f := range files { + m.Origin.Files = append(m.Origin.Files, catalog.OriginFile{Path: f.Path, SHA256: f.SHA256, Bytes: f.Bytes}) + } + } + + // One file: its commit's URL. + pinned := map[string]string{} + for _, src := range m.WeightSources() { + id, rev, file, ok := hf.FileSource(src, reg.Base()) + if !ok { + continue + } + r, err := lookup(id, rev) + if err != nil { + return err + } + i := slices.IndexFunc(r.Files, func(f hf.File) bool { return f.Path == file }) + if i < 0 { + return fmt.Errorf("model %q: %s has no file %s at %s", m.Name, id, file, r.ShortRevision()) + } + pinned[src] = r.DownloadURL(reg.Base(), file) + record(r, r.Files[i]) + } + if to, ok := pinned[m.Source]; ok { + m.Source = to + } + for i := range m.Profiles { + if to, ok := pinned[m.Profiles[i].Source]; ok { + m.Profiles[i].Source = to + } + } + + // A repository an engine downloads: --revision. + for i := range m.Profiles { + p := &m.Profiles[i] + src := m.SourceOf(*p) + if !engineDownloads[p.Engine] || !repoID.MatchString(src) { + continue + } + r, err := lookup(src, argValue(p.Args, "--revision")) + if err != nil { + return err + } + if argValue(p.Args, "--revision") == "" { + p.Args = append(slices.Clone(p.Args), "--revision", r.Revision) + } + var weights []hf.File + for _, f := range r.Files { + if f.IsWeights() { + weights = append(weights, f) + } + } + if m.Origin == nil || m.Origin.Repo != r.ID || len(m.Origin.Files) == 0 { + record(r, weights...) + } + } + return nil +} + +// argValue is a flag's value in arguments, "" when absent. +func argValue(args []string, flag string) string { + for i, a := range args { + if a == flag && i+1 < len(args) { + return args[i+1] + } + if v, ok := strings.CutPrefix(a, flag+"="); ok { + return v + } + } + return "" +} diff --git a/internal/control/modelsvc/pin_test.go b/internal/control/modelsvc/pin_test.go new file mode 100644 index 0000000..de4da12 --- /dev/null +++ b/internal/control/modelsvc/pin_test.go @@ -0,0 +1,88 @@ +package modelsvc + +import ( + "context" + "errors" + "slices" + "strings" + "testing" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/hf" +) + +const commit = "0123456789abcdef0123456789abcdef01234567" + +type fakeRegistry struct{ asked []string } + +func (f *fakeRegistry) Base() string { return "https://huggingface.co" } + +func (f *fakeRegistry) RepoAt(_ context.Context, id, rev string) (hf.Repo, error) { + f.asked = append(f.asked, id+"@"+rev) + if id == "gone/repo" { + return hf.Repo{}, errors.New("404") + } + return hf.Repo{ID: id, Revision: commit, Files: []hf.File{ + {Path: "q4.gguf", Bytes: 10, SHA256: "aaaa"}, + {Path: "model-00001-of-00002.safetensors", Bytes: 20, SHA256: "bbbb"}, + {Path: "config.json", Bytes: 1}, + }}, nil +} + +func TestAFileFromTheRegistryIsPinnedToItsCommit(t *testing.T) { + for _, src := range []string{"hf:o/r/q4.gguf", "https://huggingface.co/o/r/resolve/main/q4.gguf"} { + m := catalog.Model{Name: "q", Source: src, Profiles: []catalog.Profile{{Name: "cpu", Engine: "llamacpp"}}} + if err := pin(t.Context(), &fakeRegistry{}, &m); err != nil { + t.Fatal(err) + } + if m.Source != "https://huggingface.co/o/r/resolve/"+commit+"/q4.gguf" { + t.Errorf("%s became %s", src, m.Source) + } + if m.Origin == nil || m.Origin.Revision != commit || m.OriginSHA256(m.Source) != "aaaa" { + t.Errorf("%s: origin %+v", src, m.Origin) + } + } +} + +func TestARepositoryAnEngineDownloadsIsGivenItsRevision(t *testing.T) { + m := catalog.Model{Name: "q", Source: "o/r", Profiles: []catalog.Profile{ + {Name: "gpu", Engine: "vllm", Args: []string{"--max-model-len", "4096"}}, + {Name: "cpu", Engine: "llamacpp", Source: "hf:o/r/q4.gguf"}}} + if err := pin(t.Context(), &fakeRegistry{}, &m); err != nil { + t.Fatal(err) + } + if a := m.Profiles[0].Args; !slices.Equal(a, []string{"--max-model-len", "4096", "--revision", commit}) { + t.Errorf("vllm args %v", a) + } + if !strings.Contains(m.Profiles[1].Source, commit) { + t.Errorf("the llama.cpp profile's file: %s", m.Profiles[1].Source) + } + // A revision given is the one looked up, and kept. + reg := &fakeRegistry{} + m = catalog.Model{Name: "q", Source: "o/r", Profiles: []catalog.Profile{{Name: "gpu", Engine: "vllm", Args: []string{"--revision", "v2"}}}} + if err := pin(t.Context(), reg, &m); err != nil { + t.Fatal(err) + } + if !slices.Contains(reg.asked, "o/r@v2") || argValue(m.Profiles[0].Args, "--revision") != "v2" { + t.Errorf("asked %v, args %v", reg.asked, m.Profiles[0].Args) + } +} + +func TestWhatIsNotTheRegistrysIsLeftAndWhatCannotBePinnedIsRefused(t *testing.T) { + m := catalog.Model{Name: "q", Source: "http://10.0.0.146:8081/q/abc/q4.gguf", Profiles: []catalog.Profile{{Name: "cpu", Engine: "llamacpp"}}} + reg := &fakeRegistry{} + if err := pin(t.Context(), reg, &m); err != nil || len(reg.asked) != 0 || m.Origin != nil { + t.Errorf("a mirror's file: %v %v %+v", err, reg.asked, m.Origin) + } + // Served through the mirror, with its origin: left alone. + m = catalog.Model{Name: "q", Source: "hf:o/r/q4.gguf", Origin: &catalog.Origin{Repo: "o/r"}} + if err := pin(t.Context(), reg, &m); err != nil || m.Source != "hf:o/r/q4.gguf" { + t.Errorf("an origin given was not kept: %v %s", err, m.Source) + } + for _, src := range []string{"hf:gone/repo/q4.gguf", "hf:o/r/missing.gguf"} { + m = catalog.Model{Name: "q", Source: src} + if err := pin(t.Context(), reg, &m); err == nil { + t.Errorf("%s was not refused", src) + } + } +} diff --git a/internal/hf/hf.go b/internal/hf/hf.go index bed8791..a9c42ad 100644 --- a/internal/hf/hf.go +++ b/internal/hf/hf.go @@ -715,3 +715,35 @@ func ModelName(repoID string) string { name = strings.TrimSuffix(name, "-gguf") return strings.ToLower(name) } + +// FileSource reads a source naming one file of a registry repository: hf: +// with the file, or a resolve URL on the registry. A resolve URL already at +// a commit is still looked up, for the file's sha256. +func FileSource(src, base string) (id, rev, file string, ok bool) { + if rest, found := strings.CutPrefix(src, "hf:"); found { + parts := strings.Split(rest, "/") + if len(parts) < 3 { + return "", "", "", false + } + return parts[0] + "/" + parts[1], "", strings.Join(parts[2:], "/"), true + } + u, err := url.Parse(src) + if err != nil || (u.Scheme != "https" && u.Scheme != "http") || !isRegistryHost(u.Host, base) { + return "", "", "", false + } + parts := strings.Split(strings.Trim(u.Path, "/"), "/") + if len(parts) < 5 || parts[2] != "resolve" { + return "", "", "", false + } + return parts[0] + "/" + parts[1], parts[3], strings.Join(parts[4:], "/"), true +} + +// isRegistryHost says whether a host is the registry's: Hugging Face's own, or +// the endpoint the cluster names instead. +func isRegistryHost(host, base string) bool { + if host == "huggingface.co" { + return true + } + b, err := url.Parse(base) + return err == nil && b.Host != "" && b.Host == host +} diff --git a/internal/mirror/api.go b/internal/mirror/api.go index 20cbf97..c2715ec 100644 --- a/internal/mirror/api.go +++ b/internal/mirror/api.go @@ -75,6 +75,9 @@ type FetchRequest struct { // empty, which is right almost always and wrong visibly rather than // quietly when it is not. File string `json:"file"` + // SHA256 is what the registry published for the file: a download that + // is not it is deleted and the fetch fails. Empty checks nothing. + SHA256 string `json:"sha256,omitempty"` } // FetchResponse reports where a download stands, and where it will be. @@ -102,7 +105,7 @@ func (a *API) fetch(w http.ResponseWriter, r *http.Request) { } e := Entry{Model: req.Model, Version: req.Version, File: req.File} - st, err := a.m.Start(req.Source, e) + st, err := a.m.StartChecked(req.Source, e, strings.ToLower(req.SHA256)) switch { case errors.Is(err, ErrAlreadyFetching): // Not an error the caller needs to handle: a control plane that asks diff --git a/internal/mirror/manager.go b/internal/mirror/manager.go index c48357c..2092392 100644 --- a/internal/mirror/manager.go +++ b/internal/mirror/manager.go @@ -112,6 +112,12 @@ func (m *Manager) Start(source string, e Entry) (Status, error) { return m.start(source, e, "", m.client) } +// StartChecked is Start for a file whose sha256 the registry published: one +// that arrives otherwise is deleted, and the fetch fails saying both. +func (m *Manager) StartChecked(source string, e Entry, sha string) (Status, error) { + return m.start(source, e, sha, m.client) +} + // startCopy fetches a file a peer mirror holds, from a private address, // and keeps it only if its checksum is the peer's. func (m *Manager) startCopy(source string, e Entry, sha string) (Status, error) { @@ -163,9 +169,10 @@ func (m *Manager) run(source string, e Entry, key, expect string, client *http.C }) if err == nil && expect != "" && added.SHA256 != expect { - // Not what the peer holds: a file changed or cut short on the way. + // Not what the registry published, or what the peer holds: a file + // changed upstream, or on the way. _ = Remove(m.root, e) - err = fmt.Errorf("the copy's sha256 %s is not the peer's %s", added.SHA256, expect) + err = fmt.Errorf("%s arrived with sha256 %s, not the %s expected of it: deleted", e.File, added.SHA256, expect) } if err == nil { // Added again after it was removed: the new one stands. diff --git a/internal/mirror/peer_test.go b/internal/mirror/peer_test.go index 4e25970..aac4a81 100644 --- a/internal/mirror/peer_test.go +++ b/internal/mirror/peer_test.go @@ -5,6 +5,7 @@ import ( "net/http/httptest" "os" "path/filepath" + "strings" "testing" "time" @@ -243,3 +244,30 @@ func TestAFileRemovedByHandIsNotReportedAndItsRemovalIsCarried(t *testing.T) { t.Error("the peer kept it") } } + +// A fetch given the registry's sha256 keeps only that file. +func TestAFetchThatIsNotWhatTheRegistryPublishedIsDeleted(t *testing.T) { + served := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte("replaced upstream")) + })) + defer served.Close() + tm := newTestMirror(t) + e := Entry{Model: "qwen", Version: "v1", File: "m.gguf"} + if _, err := tm.m.StartChecked(served.URL+"/m.gguf", e, "aaaa"); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(5 * time.Second) + var st Status + for time.Now().Before(deadline) { + if st, _ = tm.m.Status("qwen", "v1"); st.State == Failed { + break + } + time.Sleep(20 * time.Millisecond) + } + if st.State != Failed || !strings.Contains(st.Err, "aaaa") { + t.Fatalf("%+v", st) + } + if _, err := os.Stat(e.Path(tm.root)); !os.IsNotExist(err) { + t.Error("kept") + } +}