novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'feat/47-G-fix11-c3d' into feat/47-G-join11

Browse Source 
# Conflicts:
#	modules/distribution/go.mod
#	modules/distribution/internal/holder/holder_test.go
This commit is contained in:
jschoubben
2026-11-12 10:26:55 +01:00
parent 9e005317dd 8d73b5ba1b
commit 2dc2c7e431
17 changed files with 273 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/distribution/cmd/distribution-tools/contract_test.go
+1 -24
View File
@@ -3,7 +3,6 @@ package main
import (
"bytes"
"encoding/json"
"errors"
"reflect"
"strings"
"testing"
@@ -31,7 +30,7 @@ func TestTheVerbsSpeakTheSeatsContract(t *testing.T) {
reg.VerbVersions: reg.VersionsInput{}, reg.VerbRemove: reg.RemoveInput{}, reg.VerbHold: reg.HoldInput{},
reg.VerbRelease: reg.ReleaseInput{}, reg.VerbMirror: reg.MirrorInput{}, reg.VerbTags: reg.TagsInput{}}
seen := 0
for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}, func() (holder.Callers, error) { return holder.DefaultCallers(), nil }) {
for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}) {
seat, verb, ok := strings.Cut(tool.Name, ".")
in, isVerb := inputs[verb]
if !ok || !isVerb || (seat != holder.ImageSeat && seat != holder.ArtifactSeat) {
@@ -57,28 +56,6 @@ func TestTheVerbsSpeakTheSeatsContract(t *testing.T) {
t.Errorf("%s's body %s is not the contract's: %v", event, b, err)
}
}
// The caller list is read held: one never read answers no verb, and registry_callers says it as read.
unread := func() (holder.Callers, error) { return nil, errors.New("config refused") }
for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}, unread) {
if tool.Name == holder.ImageSeat+"."+holder.VerbVersions {
if _, err := tool.Run(map[string]any{"name": "m/a"}); err == nil || !strings.Contains(err.Error(), "caller list") {
t.Errorf("versions with an unread caller list = %v, want refused naming the caller list", err)
}
}
}
answered := false
for _, tool := range tools(&holder.Verbs{}, kept[holder.Window]{}, func() (holder.Callers, error) { return holder.Callers{holder.VerbPublish: {"mesh-build-go"}}, nil }) {
if tool.Name == holder.CallersTool {
answered = true
got, err := tool.Run(map[string]any{})
if b, _ := json.Marshal(got); err != nil || string(b) != `{"callers":{"publish":["mesh-build-go"]}}` {
t.Errorf("%s = %s, %v", holder.CallersTool, b, err)
}
}
}
if !answered {
t.Errorf("no tool %s: the program could not read the caller list", holder.CallersTool)
}
if a := publishAnswer(holder.PublishAnswer{Transfer: "t", Missing: []string{"sha256:bb"}}); !reflect.DeepEqual(a.Lacks, []string{"sha256:bb"}) || a.Held {
t.Errorf("publish answers %+v", a)
}
modules/distribution/cmd/distribution-tools/main.go
+5 -65
View File
@@ -23,6 +23,7 @@ import (
reg "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/containerimageregistry"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/parts"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/seatlist"
"git.novox.be/novox/mesh-sdk/go/views"
"distribution/internal/holder"
@@ -232,16 +233,8 @@ func arg[T any](args map[string]any) (T, error) {
func str(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
// tools are distribution's tools over v; each call that checks a caller reads the caller list held (callers:
// the config value as mesh-config resolves it now), so an operator's change of it holds from the next call on.
func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.Callers, error)) []stdio.Tool {
now := func() (*holder.Verbs, error) {
c, err := callers()
if err != nil {
return nil, fmt.Errorf("the caller list cannot be read, so no verb is answered: %w", err)
}
return v.WithCallers(c), nil
}
// tools are distribution's tools over v.
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}
}
@@ -252,10 +245,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
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("<module>/<artifact>"), "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) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.PublishInput](args)
if err != nil {
return nil, err
@@ -269,10 +258,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
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"), "part": s("the part's digest"), "sequence": map[string]any{"type": "integer"}, "last": map[string]any{"type": "boolean"}, "data": s("base64")}, "transfer", "part", "sequence", "last", "data"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.PublishPartInput](args)
if err != nil {
return nil, err
@@ -286,10 +271,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
}},
stdio.Tool{Name: seat + "." + holder.VerbVersions, Description: "The versions the seat holds of a name.", Input: obj(map[string]any{"name": s("<module>/<artifact>")}, "name"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
a, err := v.Versions(seat, stdio.CallerOf(args), str(args, "name"))
if err != nil {
return nil, err
@@ -299,10 +280,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
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; an index with the platform manifests no tagged or held index names, and never a manifest such an index names); says removed.",
Input: obj(map[string]any{"name": s("name"), "version": s("a tagged version"), "tag": s("its tag"), "digest": s("for retention"), "reason": s("why")}, "name", "reason"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.RemoveInput](args)
if err != nil {
return nil, err
@@ -317,10 +294,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
stdio.Tool{Name: seat + "." + holder.VerbList, Description: "What the store holds of one repository, or of every repository the seat holds: each manifest's digest, whether it is an index and the manifests it names, the tags naming it, its size and whether it is held (ADR 0632 §3). Read from the store's own files; changes nothing.",
Input: obj(map[string]any{"name": s("<module>/<artifact>; every repository when not given")}),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
repos, err := v.List(context.Background(), seat, stdio.CallerOf(args), str(args, "name"))
if err != nil {
return nil, err
@@ -330,10 +303,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
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"), "digest": s("sha256:…"), "reason": s("which of the sweep's rules keeps it")}, "name", "digest", "reason"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.HoldInput](args)
if err != nil {
return nil, err
@@ -346,10 +315,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
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:…"), "reason": s("why it is let go")}, "name", "digest", "reason"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.ReleaseInput](args)
if err != nil {
return nil, err
@@ -365,10 +330,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
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) {
v, err := now()
if err != nil {
return nil, err
}
a, err := v.Tags(holder.ImageSeat, stdio.CallerOf(args), str(args, "name"))
if err != nil {
return nil, err
@@ -378,10 +339,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbMirror, Description: "Copy an outside image, pinned by its digest, into the store as upstream/<host>/<path> (every platform, only the parts it lacks): answers at once with the reference the copy will have, and says published (or refused, with why) on the seat's events when the copy ends; wait on that event (mesh-build-image alone).",
Input: obj(map[string]any{"source": s("<host>/<path>"), "digest": s("sha256:…")}, "source", "digest"),
Run: func(args map[string]any) (any, error) {
v, err := now()
if err != nil {
return nil, err
}
in, err := arg[reg.MirrorInput](args)
if err != nil {
return nil, err
@@ -392,15 +349,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
}
return reg.MirrorAnswer{Digest: in.Digest, Reference: ref}, nil
}},
stdio.Tool{Name: holder.CallersTool, Description: "Who may call which verb of distribution's seats: the trusted config value callers as mesh-config resolves it on this machine now (design 51's list where unset). The program asks it at each wake of the windows, as it has no config of its own.",
Input: obj(map[string]any{}),
Run: func(map[string]any) (any, error) {
c, err := callers()
if err != nil {
return nil, err
}
return map[string]any{"callers": c}, nil
}},
stdio.Tool{Name: "registry_health", Description: "distribution's health check, run by the runtime each minute: versions of each seat asked through the store's read side (the program's port 5100); fails naming the program when it does not answer.",
Input: obj(map[string]any{}),
Run: func(map[string]any) (any, error) {
@@ -410,10 +358,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
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) {
v, err := now()
if err != nil {
return nil, err
}
idle, _ := v.DropIdle()
if id := str(args, "drop"); id != "" {
t, err := v.DropByHand(id, stdio.CallerOf(args), str(args, "why"))
@@ -456,10 +400,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
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) {
v, err := now()
if err != nil {
return nil, err
}
if name := str(args, "release"); name != "" {
if str(args, "why") == "" {
return nil, errors.New("a release by hand is a hand act: say why")
@@ -532,7 +472,7 @@ func main() {
Transfers: parts.BusTransfers[holder.Opening](holder.TransfersState), Cap: parts.BusCap(holder.IncomingObjects), Record: recordAct, Idle: parts.BusArm(holder.IdleState)}
files := &holder.Files{Container: or(os.Getenv("MESH_REGISTRY_CONTAINER"), "mesh-registry"), Run: holder.ExecRunner, UID: os.Getuid()}
v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}, Mirrors: mirrors{},
Files: files}
Files: files, SweepHolder: func() (string, error) { return seatlist.HolderOf(stdio.Ask, holder.SweepSeat) }}
go followMirrors(v)
go parts.FollowIdle(holder.IdleState, parts.IDs(h.Transfers), h.Expired, func(id string, dropped bool, err error) {
if err != nil {
@@ -542,7 +482,7 @@ func main() {
logf("transfer %s had no part for %s: dropped with its parts", id, holder.TransferIdle)
})
windows := kept[holder.Window]{state: "windows", key: func(w holder.Window) string { return w.Name }}
if err := stdio.Serve("", tools(v, windows, holder.ConfiguredCallers)); err != nil {
if err := stdio.Serve("", tools(v, windows)); err != nil {
logf("%v", err)
os.Exit(1)
}