From 85fd46b4cc1e1cb70b0554d9ede9b5b55e1ed3a1 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 13:07:14 +0100 Subject: [PATCH] CAT-D2: distribution's verbs and events speak the two seats' contracts from mesh-contracts The verbs answered the holder's own shapes (missing and ended, a part named digest, versions as bare strings, held as the publish's event) while the seat definitions, now generated into mesh-contracts, name lacks and held, part, versions of artifacts, tags with digests and a hold's own body. A build module using mesh-sdk's parts helper would have read no lacking part and sent nothing. The bundle now takes and answers the contract's types and says each event in its body; an image's manifest is the part whose digest is the publish's, as the contract names no media type, and a hold on the image seat is of a copied base. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/distribution-tools/contract.go | 71 +++++++++++++++++++ .../cmd/distribution-tools/contract_test.go | 65 +++++++++++++++++ .../cmd/distribution-tools/main.go | 69 ++++++++++++------ modules/distribution/go.mod | 1 + modules/distribution/internal/holder/oci.go | 18 ++++- modules/distribution/internal/holder/verbs.go | 10 ++- 6 files changed, 208 insertions(+), 26 deletions(-) create mode 100644 modules/distribution/cmd/distribution-tools/contract.go create mode 100644 modules/distribution/cmd/distribution-tools/contract_test.go diff --git a/modules/distribution/cmd/distribution-tools/contract.go b/modules/distribution/cmd/distribution-tools/contract.go new file mode 100644 index 000000000..8e724dd21 --- /dev/null +++ b/modules/distribution/cmd/distribution-tools/contract.go @@ -0,0 +1,71 @@ +package main + +// The seats' contracts (mesh-contracts, generated from the seat definitions; ADR 0620): every verb's input and answer, +// and every event's body, as the two seats distribution holds define them. The verbs common to both seats have one +// shape in every registry seat, so containerimageregistry's types serve oci-artifact-registry's too. The holder's own +// types stay inside; what crosses the bus is the contract's. + +import ( + "encoding/base64" + "fmt" + "time" + + reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry" + + "distribution/internal/holder" +) + +func publishOf(in reg.PublishInput) holder.PublishRequest { + r := holder.PublishRequest{Transfer: in.Transfer, Name: in.Name, Version: in.Version, Tag: in.Tag, Kind: in.Kind, + Fingerprint: in.Fingerprint, Commit: in.Commit, Digest: in.Digest} + for _, p := range in.Parts { + r.Parts = append(r.Parts, holder.Part{Digest: p.Digest, Size: p.Size}) + } + return r +} + +func publishAnswer(a holder.PublishAnswer) reg.PublishAnswer { + return reg.PublishAnswer{Transfer: a.Transfer, Lacks: a.Missing, Held: a.Ended && len(a.Missing) == 0} +} + +func partOf(in reg.PublishPartInput) (holder.PartRequest, error) { + data, err := base64.StdEncoding.DecodeString(in.Data) + if err != nil { + return holder.PartRequest{}, fmt.Errorf("publish-part: data is not base64: %w", err) + } + return holder.PartRequest{Transfer: in.Transfer, Digest: in.Part, Sequence: int(in.Sequence), Last: in.Last, Data: data}, nil +} + +func partAnswer(in reg.PublishPartInput, a holder.PublishAnswer) reg.PublishPartAnswer { + return reg.PublishPartAnswer{Transfer: in.Transfer, Part: in.Part, Received: in.Sequence + 1, Complete: a.Ended} +} + +// versionsAnswer is `versions`: each version with the digest the store holds under it. +func versionsAnswer(v *holder.Verbs, kind string, a holder.VersionsAnswer) reg.VersionsAnswer { + out := reg.VersionsAnswer{Name: a.Name, Versions: []reg.Artifact{}} + for _, ver := range a.Versions { + d, _ := v.Catalog.Digest(a.Name, ver) + out.Versions = append(out.Versions, reg.Artifact{Name: a.Name, Version: ver, Digest: d, Kind: kind}) + } + return out +} + +func tagsAnswer(v *holder.Verbs, a holder.VersionsAnswer) reg.TagsAnswer { + out := reg.TagsAnswer{Name: a.Name, Tags: []reg.TagsAnswerTagsItem{}} + for _, t := range a.Versions { + d, _ := v.Catalog.Digest(a.Name, t) + out.Tags = append(out.Tags, reg.TagsAnswerTagsItem{Tag: t, Digest: d}) + } + return out +} + +// eventBody is an event's body as the seat's contract defines it: an Outcome for published, refused and removed, a +// Hold for held and released. +func eventBody(event string, e holder.Event) any { + switch event { + case holder.EventHeld, holder.EventReleased: + return reg.Hold{Name: e.Name, Digest: e.Digest, Holder: e.Holder, By: e.By, Reason: e.Reason, At: time.Now().UTC().Format(time.RFC3339)} + } + return reg.Outcome{Name: e.Name, Version: e.Version, Kind: e.Kind, Tag: e.Tag, Digest: e.Digest, Fingerprint: e.Fingerprint, + Commit: e.Commit, Holder: e.Holder, Transfer: e.Transfer, Reference: e.Reference, Reason: e.Reason} +} diff --git a/modules/distribution/cmd/distribution-tools/contract_test.go b/modules/distribution/cmd/distribution-tools/contract_test.go new file mode 100644 index 000000000..e76eecf1c --- /dev/null +++ b/modules/distribution/cmd/distribution-tools/contract_test.go @@ -0,0 +1,65 @@ +package main + +import ( + "bytes" + "encoding/json" + "reflect" + "strings" + "testing" + + reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry" + + "distribution/internal/holder" +) + +func jsonFields(v any) map[string]bool { + out := map[string]bool{} + t := reflect.TypeOf(v) + for i := 0; i < t.NumField(); i++ { + name, _, _ := strings.Cut(t.Field(i).Tag.Get("json"), ",") + out[name] = true + } + return out +} + +// TestTheVerbsSpeakTheSeatsContract holds distribution's seat verbs to the contracts mesh-contracts generates from the +// two seats' definitions (ADR 0620 *How it is checked*): every argument a verb's tool takes is a field of the +// contract's input, and every event body decodes into the contract's type with no field left over. +func TestTheVerbsSpeakTheSeatsContract(t *testing.T) { + inputs := map[string]any{reg.VerbPublish: reg.PublishInput{}, reg.VerbPublishPart: reg.PublishPartInput{}, + 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]{}) { + seat, verb, ok := strings.Cut(tool.Name, ".") + in, isVerb := inputs[verb] + if !ok || !isVerb || (seat != holder.ImageSeat && seat != holder.ArtifactSeat) { + continue + } + seen++ + fields := jsonFields(in) + props, _ := tool.Input["properties"].(map[string]any) + for p := range props { + if !fields[p] { + t.Errorf("%s takes %q, which the contract's input does not name", tool.Name, p) + } + } + } + if seen != 14 { + t.Errorf("%d seat verbs served, want 14 (8 on the image seat, 6 on the artifact seat)", seen) + } + for event, want := range map[string]any{holder.EventPublished: ®.Outcome{}, holder.EventHeld: ®.Hold{}, holder.EventReleased: ®.Hold{}} { + b, _ := json.Marshal(eventBody(event, holder.Event{Name: "m/a", Kind: holder.KindArchive, Digest: "sha256:aa", Holder: holder.Self, By: "x.artifact-store-tools"})) + dec := json.NewDecoder(bytes.NewReader(b)) + dec.DisallowUnknownFields() + if err := dec.Decode(want); err != nil { + t.Errorf("%s's body %s is not the contract's: %v", event, b, err) + } + } + 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) + } + if r, err := partOf(reg.PublishPartInput{Transfer: "t", Part: "sha256:bb", Data: "aGk=", Last: true}); err != nil || string(r.Data) != "hi" || r.Digest != "sha256:bb" { + t.Errorf("publish-part read %+v, %v", r, err) + } +} diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index dd67b04f7..9fa9423af 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/modules/distribution/cmd/distribution-tools/main.go @@ -15,6 +15,7 @@ import ( "sync" "time" + reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/health" @@ -86,7 +87,7 @@ func (builderVersions) Recorded(name, version, digest string) (bool, string, err } func say(seat, event string, body holder.Event) error { - return stdio.SeatPublish("mesh.seat."+seat+".event."+event, body, "") + return stdio.SeatPublish("mesh.seat."+seat+".event."+event, eventBody(event, body), "") } func finding(kind, subject, words string, raised bool) { @@ -195,76 +196,102 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { 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) + in, err := arg[reg.PublishInput](args) if err != nil { return nil, err } _, _ = v.DropIdle() - a, err := v.Publish(seat, stdio.CallerOf(args), r) + a, err := v.Publish(seat, stdio.CallerOf(args), publishOf(in)) if err == nil && closeBootstrap != nil { closeBootstrap() } if err == nil { learnPublishing(windows, stdio.CallerOf(args)) } - return a, err + return publishAnswer(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"), + 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) { - r, err := arg[holder.PartRequest](args) + in, err := arg[reg.PublishPartInput](args) if err != nil { return nil, err } - return v.PublishPart(seat, stdio.CallerOf(args), r) + r, err := partOf(in) + if err != nil { + return nil, err + } + a, err := v.PublishPart(seat, stdio.CallerOf(args), r) + return partAnswer(in, a), err }}, 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")) + a, err := v.Versions(seat, stdio.CallerOf(args), str(args, "name")) + if err != nil { + return nil, err + } + return versionsAnswer(v, str(args, "kind"), a), nil }}, 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"), + 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) { - r, err := arg[holder.RemoveRequest](args) + in, err := arg[reg.RemoveInput](args) if err != nil { return nil, err } - return map[string]any{"removed": true}, v.Remove(seat, stdio.CallerOf(args), r) + r := holder.RemoveRequest{Name: in.Name, Version: in.Version, Digest: in.Digest, Reason: in.Reason} + if err := v.Remove(seat, stdio.CallerOf(args), r); err != nil { + return nil, err + } + return reg.RemoveAnswer{Removed: true, Artifact: ®.Artifact{Name: in.Name, Version: in.Version, Tag: in.Tag, Digest: in.Digest}}, nil }}, 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"), + 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) { - r, err := arg[holder.HoldRequest](args) + in, err := arg[reg.HoldInput](args) if err != nil { return nil, err } - return map[string]any{"held": true}, v.HoldOne(seat, stdio.CallerOf(args), r) + if err := v.HoldOne(seat, stdio.CallerOf(args), holder.HoldRequest{Name: in.Name, Digest: in.Digest, Why: in.Reason}); err != nil { + return nil, err + } + return reg.HoldAnswer{Held: true, Since: time.Now().UTC().Format(time.RFC3339)}, nil }}, 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"), + 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) { - r, err := arg[holder.HoldRequest](args) + in, err := arg[reg.ReleaseInput](args) if err != nil { return nil, err } - return map[string]any{"released": true}, v.Release(seat, stdio.CallerOf(args), r) + if err := v.Release(seat, stdio.CallerOf(args), holder.HoldRequest{Name: in.Name, Digest: in.Digest, Why: in.Reason}); err != nil { + return nil, err + } + return reg.ReleaseAnswer{Released: true}, nil }}, ) } 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")) + a, err := v.Tags(holder.ImageSeat, stdio.CallerOf(args), str(args, "name")) + if err != nil { + return nil, err + } + return tagsAnswer(v, a), nil }}, stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbMirror, Description: "Copy an outside image, pinned by its digest, into the store as upstream// (every platform, only the parts it lacks) and answer its reference; says published (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) + in, err := arg[reg.MirrorInput](args) if err != nil { return nil, err } - ref, err := v.Mirror(holder.ImageSeat, stdio.CallerOf(args), r) - return map[string]any{"reference": ref}, err + ref, err := v.Mirror(holder.ImageSeat, stdio.CallerOf(args), holder.MirrorRequest{Source: in.Source, Digest: in.Digest}) + if err != nil { + return nil, err + } + return reg.MirrorAnswer{Digest: in.Digest, Reference: ref}, 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 5000); fails naming the program when it does not answer.", Input: obj(map[string]any{}), diff --git a/modules/distribution/go.mod b/modules/distribution/go.mod index e8d93c522..a54291c73 100644 --- a/modules/distribution/go.mod +++ b/modules/distribution/go.mod @@ -3,6 +3,7 @@ module distribution go 1.26.0 require ( + git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/distribution/internal/holder/oci.go b/modules/distribution/internal/holder/oci.go index 9753d1346..23c1a5852 100644 --- a/modules/distribution/internal/holder/oci.go +++ b/modules/distribution/internal/holder/oci.go @@ -118,7 +118,9 @@ func (o *OCI) Put(t Transfer, parts map[string][]byte) (string, error) { } sort.Strings(digests) for i, p := range t.Parts { - if strings.Contains(p.MediaType, "manifest") { + // An image's manifest is the part whose digest is the publish's (the seat's contract names no media type); a + // part naming a manifest's media type is one too. + if (t.Kind == KindImage || t.Kind == KindUpstream) && (p.Digest == t.Digest || strings.Contains(p.MediaType, "manifest")) { manifest = &t.Parts[i] continue } @@ -137,7 +139,7 @@ func (o *OCI) Put(t Transfer, parts map[string][]byte) (string, error) { data, ok := parts[manifest.Digest] if !ok { // Held already: fetch it to tag it under the version. - resp, b, err := o.do(http.MethodGet, "/v2/"+t.Name+"/manifests/"+manifest.Digest, "", nil, manifest.MediaType) + resp, b, err := o.do(http.MethodGet, "/v2/"+t.Name+"/manifests/"+manifest.Digest, "", nil, accepted...) if err != nil || resp.StatusCode != http.StatusOK { return "", fmt.Errorf("the held manifest %s cannot be read: %v", manifest.Digest, err) } @@ -146,7 +148,17 @@ func (o *OCI) Put(t Transfer, parts map[string][]byte) (string, error) { if ref == "" { ref = manifest.Digest } - d, err := o.putManifest(t.Name, ref, manifest.MediaType, data) + mt := manifest.MediaType + var probe struct { + MediaType string `json:"mediaType"` + } + if mt == "" && json.Unmarshal(data, &probe) == nil { + mt = probe.MediaType + } + if mt == "" { + mt = mediaManifest + } + d, err := o.putManifest(t.Name, ref, mt, data) if err != nil { return "", err } diff --git a/modules/distribution/internal/holder/verbs.go b/modules/distribution/internal/holder/verbs.go index f8cd3f4dd..746cec757 100644 --- a/modules/distribution/internal/holder/verbs.go +++ b/modules/distribution/internal/holder/verbs.go @@ -147,8 +147,14 @@ func (v *Verbs) HoldOne(seat, caller string, r HoldRequest) error { if err := v.Callers.Allowed(VerbHold, caller); err != nil { return err } - switch r.Kind { - case KindBundle, KindArchive, KindUpstream: + // The seat's contract names no kind: on the image seat a hold is of a copied base (upstream/…), never an image of + // its own; on the artifact seat of a bundle or an archive. + if r.Kind == "" && seat == ImageSeat && strings.HasPrefix(r.Name, "upstream/") { + r.Kind = KindUpstream + } + switch { + case r.Kind == "" && seat == ArtifactSeat: + case r.Kind == KindBundle, r.Kind == KindArchive, r.Kind == KindUpstream: default: return fmt.Errorf("hold: refused: a %s is never held; only an untagged bundle, archive or copied base", r.Kind) }