Weights from Hugging Face pinned to a commit and checked against the sha256 it published
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
29d7f54cfd
commit
b8ed368e31
22 files changed
+578
-16
No files matched your search
+9
-2
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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" +
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
+28
-2
@@ -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 <url>...",
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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") {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 ""
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user