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
This commit is contained in:
@@ -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}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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("<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) {
|
||||
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("<module>/<artifact>")}, "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/<host>/<path> (every platform, only the parts it lacks) and answer its reference; says published (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) {
|
||||
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{}),
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user