From c9b98477b7081756c65d07135002faee3f0b8743 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 11:56:48 +0100 Subject: [PATCH] CAT-D1, D2, D3b, D4, D6 (wiring): distribution's program, its tools bundle and its manifest The program serves the store's port 5000 in front of the registry container, now on the node's loopback (127.0.0.1:5001), listed before the program so the port is free when it starts (E14). The tools bundle serves each seat's verbs as . and registry_incoming, registry_holds, registry_old_name and registry_direct_writes, on distribution's own account, with the caller the tool runner proved. The manifest claims container-image-registry and oci-artifact-registry with their verbs, keeps mesh-artifact-store and provides artifact-store until the old name closes (E15), asks its bus state in requires (incoming, transfers, holds, windows, findings, events) and declares its health outside the program. The bootstrap's one-off credential is discarded at the first publish answered. Not yet: the runtime's object store helper for incoming (mesh-sdk), the windows' driver on delivered and manifest-reads-changed, the read-side health probe calling versions per seat. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/distribution-tools/main.go | 280 ++++++++++++++++++ modules/distribution/cmd/distribution/main.go | 117 ++++++++ modules/distribution/go.mod | 13 + modules/distribution/go.sum | 26 ++ .../internal/holder/holder_test.go | 49 +++ modules/distribution/module.json | 157 +++++++++- 6 files changed, 636 insertions(+), 6 deletions(-) create mode 100644 modules/distribution/cmd/distribution-tools/main.go create mode 100644 modules/distribution/cmd/distribution/main.go diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go new file mode 100644 index 00000000..9d9ed628 --- /dev/null +++ b/modules/distribution/cmd/distribution-tools/main.go @@ -0,0 +1,280 @@ +// Command distribution-tools serves distribution's seat verbs and tools (novox/hq design 51): the verbs of +// container-image-registry and oci-artifact-registry, each as `.`, and registry_incoming, registry_holds, +// registry_old_name and registry_direct_writes. The runtime launches it on distribution's own account; the caller of +// every verb is the one the tool runner proved (stdio.CallerOf), never a name in the arguments. stdout is the MCP +// channel; what it says, it says on stderr. +package main + +import ( + "encoding/json" + "errors" + "fmt" + "net/url" + "os" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/health" + + "distribution/internal/holder" +) + +func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, "[distribution-tools] "+format+"\n", a...) } + +// kept is one state of distribution's, through the runtime. +type kept[T any] struct { + state string + key func(T) string +} + +func (k kept[T]) Get(key string) (T, bool, error) { + var v T + e, err := stdio.State(k.state).Get(key) + if err != nil || e == nil { + return v, false, err + } + return v, true, json.Unmarshal(e.Value, &v) +} +func (k kept[T]) Put(v T) error { + _, err := stdio.State(k.state).Put(k.key(v), v) + return err +} +func (k kept[T]) Delete(key string) error { return stdio.State(k.state).Delete(key) } +func (k kept[T]) List() ([]T, error) { + keys, err := stdio.State(k.state).Keys() + if err != nil { + return nil, err + } + var out []T + for _, key := range keys { + if v, ok, err := k.Get(key); err == nil && ok { + out = append(out, v) + } + } + return out, nil +} + +// errNoObjects is said while the runtime offers a bundle no object store: the helper is mesh-sdk's (ADR 0611 §7, the +// parts helper in mesh-sdk's runtime), asked of the milestone's mesh-sdk work. Until then a part is refused loudly. +var errNoObjects = errors.New("incoming cannot be written: the runtime's object store helper (mesh-sdk, ADR 0611 §7) is not in this build") + +type objects struct{} + +func (objects) Put(string, []byte) error { return errNoObjects } +func (objects) Get(string) ([]byte, error) { return nil, errNoObjects } +func (objects) Delete(string) error { return nil } +func (objects) Used() (uint64, error) { return 0, nil } + +// builderVersions asks the builder's builder_versions (ADR 0605 §4). +type builderVersions struct{} + +func (builderVersions) Recorded(name, version, digest string) (bool, string, error) { + raw, err := stdio.Ask("mesh-delivery-builder.builder_versions", map[string]any{"name": name, "version": version, "digest": digest}) + if err != nil { + return false, "", err + } + var a struct { + Recorded *bool `json:"recorded"` + Why string `json:"why"` + } + if err := json.Unmarshal(raw, &a); err != nil || a.Recorded == nil { + return false, "", fmt.Errorf("builder_versions answered no recorded: %s", raw) + } + return *a.Recorded, a.Why, nil +} + +func say(seat, event string, body holder.Event) error { + return stdio.SeatPublish("mesh.seat."+seat+".event."+event, body, "") +} + +func finding(kind, subject, words string, raised bool) { + key := health.FindingKey(health.ScopeSeat, "distribution", kind+"."+strings.NewReplacer(".", "_", "@", "_", "/", "_").Replace(subject)) + if !raised { + _ = stdio.State(health.FindingsState).Delete(key) + return + } + f := health.Finding{V: health.FindingVersion, Key: key, Kind: kind, Ref: key[strings.LastIndex(key, ".")+1:], + Subject: health.Subject{Scope: health.ScopeSeat, ID: "distribution"}, Severity: health.Warning, + Summary: words, Headline: words, Explanation: words, Resolved: "cleared once it no longer holds", + Source: kind, Observer: "distribution", At: time.Now().UTC()} + if _, err := stdio.State(health.FindingsState).Put(key, f); err != nil { + logf("the finding %s could not be written: %v", key, err) + } +} + +// removeBootstrap discards the bootstrap's one-off credential once a publish is answered (design 51 *On a fresh mesh*). +var removeBootstrap = sync.OnceFunc(func() { + if err := os.Remove(holder.BootstrapFile); err == nil { + logf("the bootstrap's one-off credential is discarded: the holders serve publish") + } +}) + +func arg[T any](args map[string]any) (T, error) { + var v T + clean := map[string]any{} + for k, x := range args { + if k != "_meta" { + clean[k] = x + } + } + b, _ := json.Marshal(clean) + return v, json.Unmarshal(b, &v) +} + +func str(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + +func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { + obj := func(props map[string]any, req ...string) map[string]any { + return map[string]any{"type": "object", "required": req, "properties": props} + } + s := func(d string) map[string]any { return map[string]any{"type": "string", "description": d} } + var out []stdio.Tool + for _, seat := range holder.Seats { + out = append(out, + stdio.Tool{Name: seat + "." + holder.VerbPublish, Description: "Open a transfer of a publish (ADR 0611): its name, version, kind, digest and parts; answers the parts the holder lacks.", + Input: obj(map[string]any{"transfer": s("the transfer's id"), "name": s("/"), "version": s("the builder's version"), "kind": s("image, bundle, archive or upstream"), "digest": s("sha256:…"), "parts": map[string]any{"type": "array"}}, "transfer", "name", "kind", "digest", "parts"), + Run: func(args map[string]any) (any, error) { + r, err := arg[holder.PublishRequest](args) + if err != nil { + return nil, err + } + _, _ = v.DropIdle() + a, err := v.Publish(seat, stdio.CallerOf(args), r) + if err == nil { + removeBootstrap() + } + return a, err + }}, + stdio.Tool{Name: seat + "." + holder.VerbPublishPart, Description: "Send one chunk (at most 1 MiB) of one part of an open transfer, from the caller that opened it.", + Input: obj(map[string]any{"transfer": s("its id"), "digest": s("the part's digest"), "sequence": map[string]any{"type": "integer"}, "last": map[string]any{"type": "boolean"}, "data": s("base64")}, "transfer", "digest", "data"), + Run: func(args map[string]any) (any, error) { + r, err := arg[holder.PartRequest](args) + if err != nil { + return nil, err + } + return v.PublishPart(seat, stdio.CallerOf(args), r) + }}, + stdio.Tool{Name: seat + "." + holder.VerbVersions, Description: "The versions the seat holds of a name.", Input: obj(map[string]any{"name": s("/")}, "name"), + Run: func(args map[string]any) (any, error) { + return v.Versions(seat, stdio.CallerOf(args), str(args, "name")) + }}, + stdio.Tool{Name: seat + "." + holder.VerbRemove, Description: "Remove a tagged version (the builder), or for retention an untagged one by its digest (the artifact sweep); says removed.", + Input: obj(map[string]any{"name": s("name"), "version": s("a tagged version"), "digest": s("for retention"), "kind": s("kind"), "reason": s("why")}, "name", "reason"), + Run: func(args map[string]any) (any, error) { + r, err := arg[holder.RemoveRequest](args) + if err != nil { + return nil, err + } + return map[string]any{"removed": true}, v.Remove(seat, stdio.CallerOf(args), r) + }}, + stdio.Tool{Name: seat + "." + holder.VerbHold, Description: "Keep one untagged bundle, archive or copied base from retention (the artifact sweep alone); says held.", + Input: obj(map[string]any{"name": s("name"), "kind": s("bundle, archive or upstream"), "digest": s("sha256:…"), "why": s("why")}, "name", "kind", "digest"), + Run: func(args map[string]any) (any, error) { + r, err := arg[holder.HoldRequest](args) + if err != nil { + return nil, err + } + return map[string]any{"held": true}, v.HoldOne(seat, stdio.CallerOf(args), r) + }}, + stdio.Tool{Name: seat + "." + holder.VerbRelease, Description: "Let a hold go (the artifact sweep, or the operator's session through registry_holds); says released.", + Input: obj(map[string]any{"name": s("name"), "digest": s("sha256:…"), "why": s("why")}, "name", "digest"), + Run: func(args map[string]any) (any, error) { + r, err := arg[holder.HoldRequest](args) + if err != nil { + return nil, err + } + return map[string]any{"released": true}, v.Release(seat, stdio.CallerOf(args), r) + }}, + ) + } + out = append(out, + stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbTags, Description: "The tags of an image.", Input: obj(map[string]any{"name": s("name")}, "name"), + Run: func(args map[string]any) (any, error) { + return v.Tags(holder.ImageSeat, stdio.CallerOf(args), str(args, "name")) + }}, + stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbMirror, Description: "Where the store holds a copy of an outside image, by its pinned digest (mesh-build-image alone).", + Input: obj(map[string]any{"source": s("/"), "digest": s("sha256:…")}, "source", "digest"), + Run: func(args map[string]any) (any, error) { + r, err := arg[holder.MirrorRequest](args) + if err != nil { + return nil, err + } + ref, err := v.Mirror(holder.ImageSeat, stdio.CallerOf(args), r) + return map[string]any{"reference": ref}, err + }}, + stdio.Tool{Name: "registry_incoming", Description: "Open transfers of incoming and its fill; with drop, drops one by hand (a hand act, why required).", + Input: obj(map[string]any{"drop": s("a transfer id to drop"), "why": s("why it is dropped")}), + Run: func(args map[string]any) (any, error) { + idle, _ := v.DropIdle() + if id := str(args, "drop"); id != "" { + if str(args, "why") == "" { + return nil, errors.New("a drop is a hand act: say why") + } + t, err := v.Drop(id) + logf("transfer %s dropped by hand by %s: %s", id, stdio.CallerOf(args), str(args, "why")) + return map[string]any{"dropped": t.Transfer}, err + } + all, err := v.Transfers.List() + used, _ := v.Objects.Used() + return map[string]any{"transfers": all, "used": used, "cap": holder.IncomingCap, "dropped_idle": idle}, err + }}, + stdio.Tool{Name: "registry_holds", Description: "What is held, since when and for whom; with release, lets one go through the seat's release as the operator's session (a hand act).", + Input: obj(map[string]any{"seat": s("the seat"), "release": s("the name to release"), "digest": s("its digest"), "why": s("why")}), + Run: func(args map[string]any) (any, error) { + if name := str(args, "release"); name != "" { + if str(args, "why") == "" { + return nil, errors.New("a release by hand is a hand act: say why") + } + // The caller proved on this call (the operator's session) is carried on; distribution's own never. + return map[string]any{"released": name}, v.Release(str(args, "seat"), stdio.CallerOf(args), holder.HoldRequest{Name: name, Digest: str(args, "digest"), Why: str(args, "why")}) + } + return v.HeldNow() + }}, + ) + for _, w := range []string{holder.OldNameWindow, holder.DirectWritesWindow} { + name := "registry_old_name" + if w == holder.DirectWritesWindow { + name = "registry_direct_writes" + } + out = append(out, stdio.Tool{Name: name, Description: "Whether the " + w + " window is open and the nodes holding it, since when; with drop, drops a retired node from it by hand (a hand act, why required).", + Input: obj(map[string]any{"drop": s("a retired node"), "why": s("why")}), + Run: func(args map[string]any) (any, error) { + cur, _, err := windows.Get(w) + if err != nil { + return nil, err + } + cur.Name = w + if node := str(args, "drop"); node != "" { + if cur, err = cur.Drop(node, stdio.CallerOf(args), str(args, "why"), time.Now().UTC()); err != nil { + return nil, err + } + if err := windows.Put(cur); err != nil { + return nil, err + } + } + _, bootstrap := os.Stat(holder.BootstrapFile) + return map[string]any{"window": cur, "bootstrap_credential_present": bootstrap == nil}, nil + }}) + } + return out +} + +func main() { + callers, err := holder.ParseCallers(os.Getenv("MESH_CONFIG_CALLERS")) + if err != nil { + logf("%v", err) + os.Exit(1) + } + base, _ := url.Parse("http://127.0.0.1:5001") + store := &holder.OCI{Base: base, Registry: os.Getenv("MESH_REGISTRY_REACH")} + h := &holder.Holder{Callers: callers, Objects: objects{}, Builder: builderVersions{}, Store: store, Say: say, Finding: finding, + Transfers: kept[holder.Transfer]{state: holder.TransfersState, key: func(t holder.Transfer) string { return t.Transfer }}} + v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}} + windows := kept[holder.Window]{state: "windows", key: func(w holder.Window) string { return w.Name }} + if err := stdio.Serve("", tools(v, windows)); err != nil { + logf("%v", err) + os.Exit(1) + } +} diff --git a/modules/distribution/cmd/distribution/main.go b/modules/distribution/cmd/distribution/main.go new file mode 100644 index 00000000..ccf51512 --- /dev/null +++ b/modules/distribution/cmd/distribution/main.go @@ -0,0 +1,117 @@ +// Command distribution is distribution's program (novox/hq design 51): it serves the artifact store's port in front of +// the registry container, which listens on its node's loopback only. Reads (GET, HEAD) go through to every node's +// container runtime; a write is refused but while the direct-write window is open, never over a tag the seat holds +// published, or with the bootstrap's one-off credential (holder.Front). The window is distribution's state windows, +// written by its tools bundle; the program reads it on each write and holds no timer. +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log" + "net/http" + "net/url" + "os" + "os/signal" + "strings" + "sync/atomic" + "syscall" + "time" + + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" + + "distribution/internal/holder" +) + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + if err := run(ctx, os.Getenv); err != nil { + fmt.Fprintln(os.Stderr, "distribution:", err) + os.Exit(1) + } +} + +func run(ctx context.Context, getenv func(string) string) error { + logger := log.New(os.Stderr, "distribution: ", log.LstdFlags) + store, err := url.Parse(or(getenv("MESH_REGISTRY_LOOPBACK"), "http://127.0.0.1:5001")) + if err != nil { + return err + } + oci := &holder.OCI{Base: store} + var open atomic.Bool + open.Store(true) // until the window's state says it closed: no wider than today's store + cred, err := process.ReadBrokerAccount(getenv) + if err != nil { + return err + } + conn, js, err := cred.Dial() + if err != nil { + return err + } + defer conn.Close() + // The direct-write window's record, written by the tools bundle on distribution's own account. + windows := process.OptionalBucket(ctx, js, "distribution_windows", "the direct-write window is read as open until the program starts again", logger.Printf) + if windows != nil { + go follow(ctx, windows, &open, logger) + } + front := &holder.Front{Store: store, Window: &open, Logf: logger.Printf, + Published: func(name, tag string) (bool, error) { + d, err := oci.Digest(name, tag) + return d != "", err + }, + Bootstrap: func() (string, bool) { + b, err := os.ReadFile(holder.BootstrapFile) + if err != nil { + return "", false + } + return strings.TrimSpace(string(b)), true + }} + srv := &http.Server{Addr: or(getenv("MESH_REGISTRY_LISTEN"), ":5000"), Handler: front, ReadHeaderTimeout: 30 * time.Second} + errs := make(chan error, 1) + go func() { errs <- srv.ListenAndServe() }() + logger.Printf("serving the store's port %s in front of %s", srv.Addr, store) + select { + case <-ctx.Done(): + shut, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + return srv.Shutdown(shut) + case err := <-errs: + if errors.Is(err, http.ErrServerClosed) { + return nil + } + return err + } +} + +// follow keeps open as the direct-write window's record says, on every change of it (a watch, no timer). +func follow(ctx context.Context, kv jetstream.KeyValue, open *atomic.Bool, logger *log.Logger) { + w, err := kv.Watch(ctx, holder.DirectWritesWindow) + if err != nil { + logger.Printf("the direct-write window cannot be watched (%v): read as open", err) + return + } + for e := range w.Updates() { + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + var win holder.Window + if json.Unmarshal(e.Value(), &win) != nil { + continue + } + closed := !win.Open && !win.Closed.IsZero() + if closed && open.Swap(false) { + logger.Printf("the direct-write window is closed: every write but distribution's own is refused, naming publish") + } + } +} + +func or(v, d string) string { + if strings.TrimSpace(v) == "" { + return d + } + return v +} diff --git a/modules/distribution/go.mod b/modules/distribution/go.mod index be2ed40e..b7e3cf30 100644 --- a/modules/distribution/go.mod +++ b/modules/distribution/go.mod @@ -1,3 +1,16 @@ module distribution go 1.26.0 + +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 + github.com/nats-io/nats.go v1.54.0 +) + +require ( + github.com/klauspost/compress v1.20.0 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sys v0.48.0 // indirect +) diff --git a/modules/distribution/go.sum b/modules/distribution/go.sum index e69de29b..d72bd4ff 100644 --- a/modules/distribution/go.sum +++ b/modules/distribution/go.sum @@ -0,0 +1,26 @@ +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 h1:XEbIVImp6yGjA91rp0iZBoDdL5KfamjOuDLwWvqh0hc= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= +github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index 2b1a6824..97835e87 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/modules/distribution/internal/holder/holder_test.go @@ -3,10 +3,12 @@ package holder import ( "crypto/sha256" "encoding/hex" + "encoding/json" "errors" "net/http" "net/http/httptest" "net/url" + "os" "reflect" "strings" "sync/atomic" @@ -383,4 +385,51 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) { t.Error("a closed window closed again") } }) + t.Run("the manifest", func(t *testing.T) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var m struct { + Provides []struct{ Name string } `json:"provides"` + Claims []struct { + Name string `json:"name"` + Serves []string `json:"serves"` + } `json:"claims"` + Resources []map[string]any `json:"resources"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + // E15: artifact-store is still offered until the old name closes, so a revert needs no edge. + if len(m.Provides) != 1 || m.Provides[0].Name != "artifact-store" { + t.Errorf("provides = %+v, want artifact-store kept", m.Provides) + } + claims := map[string][]string{} + for _, c := range m.Claims { + claims[c.Name] = c.Serves + } + if !reflect.DeepEqual(claims[ImageSeat], VerbsOf(ImageSeat)) || !reflect.DeepEqual(claims[ArtifactSeat], VerbsOf(ArtifactSeat)) { + t.Errorf("claims = %v, want each seat serving its verbs", claims) + } + if _, ok := claims[OldSeat]; !ok { + t.Error("the old seat mesh-artifact-store is no longer claimed while its window is open") + } + // E14: the registry container, on the loopback, is listed before the program, so the port is free when it starts. + store, program := -1, -1 + for i, r := range m.Resources { + switch r["id"] { + case "store": + store = i + if ports, _ := r["ports"].([]any); len(ports) != 1 || ports[0] != "127.0.0.1:5001:5000" { + t.Errorf("the store's ports = %v, want the loopback only", r["ports"]) + } + case "program": + program = i + } + } + if store < 0 || program < 0 || store > program { + t.Errorf("store at %d, program at %d: want the store listed first", store, program) + } + }) } diff --git a/modules/distribution/module.json b/modules/distribution/module.json index 5037f429..5f3fcf44 100644 --- a/modules/distribution/module.json +++ b/modules/distribution/module.json @@ -1,6 +1,6 @@ { "module": "distribution", - "version": "1", + "version": "2", "provides": [ { "name": "artifact-store", @@ -11,17 +11,111 @@ } ], "claims": [ + { + "name": "container-image-registry", + "scope": "mesh", + "serves": [ + "hold", + "mirror", + "publish", + "publish-part", + "release", + "remove", + "tags", + "versions" + ] + }, + { + "name": "oci-artifact-registry", + "scope": "mesh", + "serves": [ + "hold", + "publish", + "publish-part", + "release", + "remove", + "versions" + ] + }, { "name": "mesh-artifact-store", "scope": "mesh" } ], + "tools": [ + "registry_direct_writes", + "registry_holds", + "registry_incoming", + "registry_old_name" + ], + "invokes": [ + "mesh-delivery-builder.builder_versions", + "mesh-delivery-view.node_haves", + "seat:node-engine.manifest-reads" + ], + "consumes": [ + "mesh-delivery-view.delivered" + ], + "emits": [ + "old-name-closed", + "direct-writes-closed" + ], + "config": { + "callers": { + "type": "string", + "trusted": true, + "description": "per verb, the modules that may call it (JSON object of verb to modules; operator for the operator's session); empty is design 51's list" + } + }, + "requires": [ + { + "name": "incoming", + "storage": "object-store", + "template": "objects", + "options": { + "max-mib": 8192 + } + }, + { + "name": "transfers", + "storage": "key-value", + "template": "state", + "options": { + "ttl-seconds": 86400 + } + }, + { + "name": "holds", + "storage": "key-value", + "template": "state" + }, + { + "name": "windows", + "storage": "key-value", + "template": "state" + }, + { + "name": "findings", + "storage": "key-value", + "template": "state" + }, + { + "name": "events", + "storage": "stream", + "template": "events", + "options": { + "max-age-days": 7, + "max-mib": 16 + } + } + ], "capabilities": [ "container-runtime" ], "own-secrets": { "broker": "/var/lib/mesh/registry/broker" }, + "secrets-owner": "distribution", "serves": { "artifact-store": { "port": 5000 @@ -33,7 +127,7 @@ "port": 5000, "protocol": "tcp", "from": "mesh", - "why": "every machine pulls images and artifacts from here" + "why": "every machine pulls images and artifacts from here; distribution's program serves it in front of the store" } ], "data": { @@ -49,19 +143,28 @@ ] }, "resources": [ + { + "id": "account", + "type": "user", + "name": "distribution", + "shell": "/usr/bin/nologin", + "home": "/var/lib/mesh/registry" + }, { "id": "state", "type": "directory", "path": "/var/lib/mesh/registry", "mode": "0700", - "unknown": "program" + "unknown": "program", + "owner": "distribution" }, { "id": "registry-data", "type": "directory", "path": "/var/lib/mesh-registry", "mode": "0700", - "unknown": "program" + "unknown": "program", + "owner": "distribution" }, { "id": "store", @@ -69,7 +172,7 @@ "name": "mesh-registry", "image": "registry@sha256:a3d8aaa63ed8681a604f1dea0aa03f100d5895b6a58ace528858a7b332415373", "ports": [ - "5000:5000" + "127.0.0.1:5001:5000" ], "volumes": [ "/var/lib/mesh-registry:/var/lib/registry" @@ -78,6 +181,25 @@ "REGISTRY_STORAGE_DELETE_ENABLED": "true" } }, + { + "id": "program", + "type": "process", + "name": "distribution", + "artifact": "program", + "run": [ + "./distribution" + ], + "env": { + "MESH_BROKER_FILE": "/var/lib/mesh/registry/broker", + "MESH_REGISTRY_LOOPBACK": "http://127.0.0.1:5001", + "MESH_REGISTRY_LISTEN": ":5000" + }, + "user": "distribution", + "health": { + "kind": "http", + "url": "http://127.0.0.1:5000/v2/" + } + }, { "id": "collect", "type": "container", @@ -95,5 +217,28 @@ "store" ] } - ] + ], + "build": { + "artifacts": [ + { + "name": "program", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/distribution", + "binary": "distribution" + }, + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/distribution-tools", + "binary": "distribution-tools", + "loads": [ + "distribution-tools" + ] + } + ] + } }