From 9f01b9e79ae65f0b9c4eb24c8f0167fce1efd160 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 13:19:46 +0100 Subject: [PATCH] CAT-D3: mirror answers at once and the holder says the copy's end on the seat's events A large base could outlast a call's bound. The ask is kept in the state mirrors and the copy runs on that state's watch, so a copy a restart cut is taken on again; published (or refused, with why) on container-image-registry's events is the end the caller waits on, the place the seat's contract already gives mirror. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/distribution-tools/main.go | 33 +++++++++++- .../internal/holder/holder_test.go | 49 +++++++++++++++-- .../distribution/internal/holder/mirror.go | 5 +- modules/distribution/internal/holder/verbs.go | 54 ++++++++++++++++--- modules/distribution/module.json | 5 ++ 5 files changed, 133 insertions(+), 13 deletions(-) diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index 2f0bd5a1..a196ba17 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/modules/distribution/cmd/distribution-tools/main.go @@ -210,6 +210,34 @@ func learnInto(state casState, module, fp string, attempts int) (bool, error) { return false, fmt.Errorf("%s was written by another writer first %d times: %w", holder.PublishingKey, attempts, stdio.ErrChanged) } +// mirrors is the state mirrors: an ask is created once; one already kept is the same copy, taken on already. +type mirrors struct{} + +func (mirrors) Ask(key string, r holder.MirrorRequest) error { + if _, err := stdio.State(holder.MirrorsState).Create(key, r); err != nil && !errors.Is(err, stdio.ErrChanged) { + return err + } + return nil +} +func (mirrors) Delete(key string) error { return stdio.State(holder.MirrorsState).Delete(key) } + +// followMirrors runs every copy `mirror` took on, on the state's watch: the asks kept when the bundle starts (a copy a +// restart cut) and each new one, one at a time. +func followMirrors(v *holder.Verbs) { + <-stdio.Served() + stdio.State(holder.MirrorsState).Follow("", func(ch stdio.StateChange) error { + if ch.Op != stdio.OpPut { + return nil + } + var r holder.MirrorRequest + if err := json.Unmarshal(ch.Value, &r); err != nil { + logf("the mirror ask %s is no ask (%v): dropped", ch.Key, err) + return stdio.State(holder.MirrorsState).Delete(ch.Key) + } + return v.Mirrored(r) + }) +} + func arg[T any](args map[string]any) (T, error) { var v T clean := map[string]any{} @@ -319,7 +347,7 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { } 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).", + 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): 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("/"), "digest": s("sha256:…")}, "source", "digest"), Run: func(args map[string]any) (any, error) { in, err := arg[reg.MirrorInput](args) @@ -409,7 +437,8 @@ func main() { store := &holder.OCI{Base: base, Registry: os.Getenv("MESH_REGISTRY_REACH"), Upstream: &holder.Upstream{}} 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.ID }}} - v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}} + v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}, Mirrors: mirrors{}} + go followMirrors(v) windows := kept[holder.Window]{state: "windows", key: func(w holder.Window) string { return w.Name }} closeBootstrap = bootstrapCloser(windows) if err := stdio.Serve("", tools(v, windows)); err != nil { diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index 70cf7c68..0866cdeb 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/modules/distribution/internal/holder/holder_test.go @@ -84,8 +84,9 @@ type bed struct { } type fakeCatalog struct { - tags map[string]string // name:version -> digest - removed []string + tags map[string]string // name:version -> digest + removed []string + mirrorErr error } func (c *fakeCatalog) Versions(name string) ([]string, error) { @@ -110,8 +111,25 @@ func (c *fakeCatalog) RemoveDigest(name, digest string) error { return nil } func (c *fakeCatalog) Mirror(source, digest string) (string, error) { - return "registry.mesh/upstream/" + source + "@" + digest, nil + if c.mirrorErr != nil { + return "", c.mirrorErr + } + src, _ := ParseSource(source) + return c.Reference(src.Name(), digest), nil } +func (c *fakeCatalog) Reference(name, digest string) string { + return "registry.mesh/" + name + "@" + digest +} + +type memMirrors map[string]MirrorRequest + +func (m memMirrors) Ask(k string, r MirrorRequest) error { + if _, ok := m[k]; !ok { + m[k] = r + } + return nil +} +func (m memMirrors) Delete(k string) error { delete(m, k); return nil } type memHolds map[string]Hold @@ -135,7 +153,7 @@ func newBed(builderOK bool) *bed { return nil }, Finding: func(kind, subject, _ string, raised bool) { b.findings[kind+" "+subject] = raised }} - b.v = &Verbs{Holder: b.h, Catalog: b.cat, Holds: memHolds{}} + b.v = &Verbs{Holder: b.h, Catalog: b.cat, Holds: memHolds{}, Mirrors: memMirrors{}} return b } @@ -412,6 +430,29 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) { } } }) + t.Run("mirror answers at once and says the copy's end on the seat's events", func(t *testing.T) { + b := newBed(true) + d := "sha256:" + strings.Repeat("c", 64) + ref, err := b.v.Mirror(ImageSeat, "novox.mesh-build-image", MirrorRequest{Source: "docker.io/library/nats:2", Digest: d}) + want := "registry.mesh/upstream/docker.io/library/nats@" + d + key := MirrorsKey("upstream/docker.io/library/nats", d) + asked := b.v.Mirrors.(memMirrors) + if err != nil || ref != want || len(b.said) != 0 || asked[key].By != "novox.mesh-build-image" { + t.Fatalf("mirror = %q, %v, said %v, asks %v: want the reference at once, the copy asked, nothing said yet", ref, err, b.said, asked) + } + if strings.ContainsAny(key, ":@") { + t.Errorf("the ask's key %q holds characters a key may not", key) + } + if err := b.v.Mirrored(asked[key]); err != nil || len(b.said) != 1 || b.said[0].event != EventPublished || + b.said[0].body.Reference != want || b.said[0].body.Kind != KindUpstream || len(asked) != 0 { + t.Fatalf("the copy's end = %v, said %+v, asks %v: want published with the reference and the ask gone", err, b.said, asked) + } + b.cat.mirrorErr = errors.New("the upstream said 404") + _, _ = b.v.Mirror(ImageSeat, "novox.mesh-build-image", MirrorRequest{Source: "nats", Digest: d}) + if err := b.v.Mirrored(asked[key]); err != nil || b.said[1].event != EventRefused || !strings.Contains(b.said[1].body.Reason, "404") || len(asked) != 0 { + t.Errorf("a failed copy = %v, said %+v: want refused with why", err, b.said) + } + }) t.Run("the windows", func(t *testing.T) { nodes := []NodeRead{ {Node: "ace", Serves: true, Stated: []string{NewNamesCapability}, NamesArtifact: true, BuildModules: map[string]bool{"mesh-build-go": true}}, diff --git a/modules/distribution/internal/holder/mirror.go b/modules/distribution/internal/holder/mirror.go index 720dfca6..cf12b710 100644 --- a/modules/distribution/internal/holder/mirror.go +++ b/modules/distribution/internal/holder/mirror.go @@ -295,5 +295,8 @@ func (o *OCI) Mirror(source, digest string) (string, error) { if err := o.copyManifest(o.Upstream, src, name, digest, 0); err != nil { return "", err } - return o.Registry + "/" + name + "@" + digest, nil + return o.Reference(name, digest), nil } + +// Reference is the seat's answer for name at digest: `/@`. +func (o *OCI) Reference(name, digest string) string { return o.Registry + "/" + name + "@" + digest } diff --git a/modules/distribution/internal/holder/verbs.go b/modules/distribution/internal/holder/verbs.go index 746cec75..5f8bdbdb 100644 --- a/modules/distribution/internal/holder/verbs.go +++ b/modules/distribution/internal/holder/verbs.go @@ -20,6 +20,19 @@ type Catalog interface { RemoveDigest(name, digest string) error // Mirror copies an outside image, pinned by digest, into upstream//, answering the reference. Mirror(source, digest string) (string, error) + // Reference is the seat's answer for name at digest, known before anything is copied. + Reference(name, digest string) string +} + +// MirrorsState names the state of the copies `mirror` took on. +const MirrorsState = "mirrors" + +// Mirrors is the state mirrors: each copy `mirror` took on, kept until its outcome is said, so a copy cut by a +// restart is taken on again from the state's watch. +type Mirrors interface { + // Ask keeps r under its key; an ask already kept is no fault (the copy is taken on once). + Ask(key string, r MirrorRequest) error + Delete(key string) error } // HoldState is distribution's own state holds (bus state, template state; ADR 0617 §2). @@ -51,6 +64,7 @@ type Verbs struct { *Holder Catalog Catalog Holds HoldState + Mirrors Mirrors } // VersionsAnswer is `versions`: the versions the seat holds of name. @@ -206,13 +220,22 @@ func (v *Verbs) Tags(seat, caller, name string) (VersionsAnswer, error) { return VersionsAnswer{Seat: seat, Name: name, Versions: got}, nil } -// MirrorRequest is `mirror`: an outside image by its pinned digest. +// MirrorRequest is `mirror`: an outside image by its pinned digest; By is the proved caller, kept with the ask. type MirrorRequest struct { Source string `json:"source"` Digest string `json:"digest"` + By string `json:"by,omitempty"` } -// Mirror is `mirror` on the image seat: mesh-build-image alone; an image pinned by no digest is refused. +// MirrorsKey is an ask's key in the state mirrors: the copy's name and digest, in a key's characters. +func MirrorsKey(name, digest string) string { + return strings.NewReplacer(":", "_", "@", "_").Replace(name + "/" + digest) +} + +// Mirror is `mirror` on the image seat: mesh-build-image alone; an image pinned by no digest is refused. It answers at +// once with the reference the copy will have and keeps the ask in the state mirrors; the copy runs on that state's +// watch (Mirrored), which says published or refused on the seat's events when it ends. The caller waits on that +// event, not on the call, so a large base never outlasts a call's bound. func (v *Verbs) Mirror(seat, caller string, r MirrorRequest) (string, error) { if seat != ImageSeat { return "", fmt.Errorf("mirror: %s serves no mirror", seat) @@ -227,11 +250,30 @@ func (v *Verbs) Mirror(seat, caller string, r MirrorRequest) (string, error) { if err != nil { return "", err } - ref, err := v.Catalog.Mirror(r.Source, r.Digest) - if err != nil { - return "", fmt.Errorf("mirror: %s@%s was not copied: %w", r.Source, r.Digest, err) + r.By = caller + if err := v.Mirrors.Ask(MirrorsKey(src.Name(), r.Digest), r); err != nil { + return "", fmt.Errorf("mirror: the copy of %s@%s cannot be taken on: %w", r.Source, r.Digest, err) } - return ref, v.Say(seat, EventPublished, Event{Name: src.Name(), Kind: KindUpstream, Digest: r.Digest, Holder: Self, Reference: ref, By: caller}) + return v.Catalog.Reference(src.Name(), r.Digest), nil +} + +// Mirrored runs one asked copy and says its outcome on the image seat's events: published with the reference, or +// refused with the reason; the ask leaves the state either way. An outcome that cannot be said keeps the ask, so the +// next watch takes it on again. +func (v *Verbs) Mirrored(r MirrorRequest) error { + src, err := ParseSource(r.Source) + if err != nil { + return err + } + e := Event{Name: src.Name(), Kind: KindUpstream, Digest: r.Digest, Holder: Self, By: r.By} + event := EventPublished + if e.Reference, err = v.Catalog.Mirror(r.Source, r.Digest); err != nil { + event, e.Reference, e.Reason = EventRefused, "", fmt.Sprintf("%s@%s was not copied: %v", r.Source, r.Digest, err) + } + if err := v.Say(ImageSeat, event, e); err != nil { + return fmt.Errorf("mirror: the outcome of %s@%s cannot be said: %w", r.Source, r.Digest, err) + } + return v.Mirrors.Delete(MirrorsKey(src.Name(), r.Digest)) } // HeldNow lists the holds, for registry_holds. diff --git a/modules/distribution/module.json b/modules/distribution/module.json index 29e3038c..7a6f2179 100644 --- a/modules/distribution/module.json +++ b/modules/distribution/module.json @@ -96,6 +96,11 @@ "storage": "key-value", "template": "state" }, + { + "name": "mirrors", + "storage": "key-value", + "template": "state" + }, { "name": "findings", "storage": "key-value",