From d683a78d9caf63c9b3797382b21fe1b649dde61d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 13:24:40 +0100 Subject: [PATCH] CAT-D3, D2: one mirror copy, said once; a refusal at the cap stays loud until the holder takes a publish again A second tools process following mirrors would copy again and say published twice. The ask is claimed by compare-and-set before it is copied, and its outcome carries the ask's id as the event id, so a redelivery is one message. The cap's publish-refused finding stands on incoming, not on one file, and clears with the next publish taken. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/distribution-tools/main.go | 19 +++++++-- .../internal/holder/holder_test.go | 40 +++++++++++++++++++ .../distribution/internal/holder/transfer.go | 14 ++++++- modules/distribution/internal/holder/verbs.go | 37 ++++++++++++++++- 4 files changed, 103 insertions(+), 7 deletions(-) diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index a196ba17e..948631ce7 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/modules/distribution/cmd/distribution-tools/main.go @@ -6,6 +6,8 @@ package main import ( + "crypto/rand" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -87,7 +89,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, eventBody(event, body), "") + return stdio.SeatPublish("mesh.seat."+seat+".event."+event, eventBody(event, body), body.ID) } func finding(kind, subject, words string, raised bool) { @@ -219,11 +221,22 @@ func (mirrors) Ask(key string, r holder.MirrorRequest) error { } return nil } +func (mirrors) Claim(key string, r holder.MirrorRequest, revision uint64) (bool, error) { + _, err := stdio.State(holder.MirrorsState).Update(key, r, revision) + if errors.Is(err, stdio.ErrChanged) { + return false, nil + } + return err == nil, err +} 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. +// restart cut) and each new one, one at a time. Each is claimed first (Verbs.Take), so a second follower copies none +// a first one took on. func followMirrors(v *holder.Verbs) { + id := make([]byte, 8) + _, _ = rand.Read(id) + self := "distribution-tools-" + hex.EncodeToString(id) <-stdio.Served() stdio.State(holder.MirrorsState).Follow("", func(ch stdio.StateChange) error { if ch.Op != stdio.OpPut { @@ -234,7 +247,7 @@ func followMirrors(v *holder.Verbs) { 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) + return v.Take(ch.Key, r, ch.Revision, ch.Current, self) }) } diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index 0866cdeb2..1e7403371 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/modules/distribution/internal/holder/holder_test.go @@ -129,8 +129,23 @@ func (m memMirrors) Ask(k string, r MirrorRequest) error { } return nil } +func (m memMirrors) Claim(k string, r MirrorRequest, rev uint64) (bool, error) { + if cur, ok := m[k]; !ok || mirrorRev(cur) != rev { + return false, nil + } + m[k] = r + return true, nil +} func (m memMirrors) Delete(k string) error { delete(m, k); return nil } +// mirrorRev is a kept ask's revision in memMirrors: 1 unclaimed, 2 once claimed. +func mirrorRev(r MirrorRequest) uint64 { + if r.Claim == "" { + return 1 + } + return 2 +} + type memHolds map[string]Hold func (m memHolds) Get(k string) (Hold, bool, error) { h, ok := m[k]; return h, ok, nil } @@ -301,6 +316,17 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) { if b.findings["incoming-full incoming"] { t.Error("the finding stayed after the transfer was dropped") } + // A refusal for want of room stays loud on incoming until the holder takes a publish again. + b.h.Cap = 100 + _, _ = b.h.Publish(ArtifactSeat, builder, req("t4")) + if !b.findings["publish-refused incoming"] { + t.Errorf("a refusal at the cap raised %v, want publish-refused on incoming", b.findings) + } + b.h.Cap = 1 << 20 + b.store.has[part.Digest] = true + if _, err := b.h.Publish(ArtifactSeat, builder, req("t5")); err != nil || b.findings["publish-refused incoming"] { + t.Errorf("the next publish taken = %v, findings %v: want the refusal at the cap cleared", err, b.findings) + } }) t.Run("the sweep's removal, hold and release", func(t *testing.T) { b := newBed(true) @@ -447,6 +473,20 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) { 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) } + // One copy, said once: two followers on the same ask; the second sees it claimed and passes, and the outcome + // carries the ask's id as its event id. + _, _ = b.v.Mirror(ImageSeat, "novox.mesh-build-image", MirrorRequest{Source: "nats", Digest: d}) + ask := asked[key] + copies := len(b.said) + for _, follower := range []string{"one", "two"} { + if err := b.v.Take(key, ask, 1, false, follower); err != nil { + t.Fatalf("Take by %s: %v", follower, err) + } + } + if len(b.said) != copies+1 || ask.ID == "" || b.said[copies].body.ID != ask.ID || len(asked) != 0 { + t.Fatalf("two followers on one ask said %+v (ask id %q), asks %v: want one copy, said once under the ask's id", b.said[copies:], ask.ID, asked) + } + b.said = b.said[:copies] 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 { diff --git a/modules/distribution/internal/holder/transfer.go b/modules/distribution/internal/holder/transfer.go index a9c331f8c..84dafb239 100644 --- a/modules/distribution/internal/holder/transfer.go +++ b/modules/distribution/internal/holder/transfer.go @@ -99,6 +99,9 @@ type Event struct { Reason string `json:"reason,omitempty"` // By is who asked a removal, hold or release: the caller the tool runner proved. By string `json:"by,omitempty"` + // ID is the event's id on the bus when the outcome has one of its own (a mirror ask's): the same outcome said + // twice, by a redelivery or a second follower, is one message. Empty: the runtime makes one. + ID string `json:"-"` } // The events of the seats. @@ -197,6 +200,8 @@ func (h *Holder) receiver() *parts.Receiver[Opening] { return errors.Join(errors.New("refused: "+reason), h.refused(t, reason)) } h.Finding("publish-refused", t.Shape.Name+"@"+t.Shape.Version, "", false) + // Room was found again: a refusal for want of it clears with the first publish the holder takes. + h.Finding("publish-refused", IncomingObjects, "", false) return h.Say(t.Shape.Seat, EventPublished, h.eventOf(t.Shape.Seat, t.Shape.PublishRequest, ref, "")) }, Refuse: h.refused, @@ -205,9 +210,14 @@ func (h *Holder) receiver() *parts.Receiver[Opening] { }} } -// refused writes the finding and says refused on the transfer's seat. +// refused writes the finding and says refused on the transfer's seat. A refusal for want of room is the holder's, +// not the file's: its finding stands on incoming and clears when the holder takes a publish again. func (h *Holder) refused(t Transfer, reason string) error { - h.Finding("publish-refused", t.Shape.Name+"@"+t.Shape.Version, reason, true) + subject := t.Shape.Name + "@" + t.Shape.Version + if parts.RefusedAtCap(reason) { + subject = IncomingObjects + } + h.Finding("publish-refused", subject, reason, true) return h.Say(t.Shape.Seat, EventRefused, h.eventOf(t.Shape.Seat, t.Shape.PublishRequest, "", reason)) } diff --git a/modules/distribution/internal/holder/verbs.go b/modules/distribution/internal/holder/verbs.go index 5f8bdbdbc..312be40df 100644 --- a/modules/distribution/internal/holder/verbs.go +++ b/modules/distribution/internal/holder/verbs.go @@ -1,6 +1,8 @@ package holder import ( + "crypto/rand" + "encoding/hex" "errors" "fmt" "slices" @@ -32,6 +34,9 @@ const MirrorsState = "mirrors" 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 + // Claim writes r (its Claim set) under key at revision, compare-and-set: false, and no error, when another + // writer wrote the key first. + Claim(key string, r MirrorRequest, revision uint64) (bool, error) Delete(key string) error } @@ -225,6 +230,10 @@ type MirrorRequest struct { Source string `json:"source"` Digest string `json:"digest"` By string `json:"by,omitempty"` + // ID is the ask's own id, made when it is asked: the id of the event that says its end. + ID string `json:"id,omitempty"` + // Claim is the follower that took the copy on: none until one claims it. + Claim string `json:"claim,omitempty"` } // MirrorsKey is an ask's key in the state mirrors: the copy's name and digest, in a key's characters. @@ -250,7 +259,12 @@ func (v *Verbs) Mirror(seat, caller string, r MirrorRequest) (string, error) { if err != nil { return "", err } - r.By = caller + r.By, r.Claim = caller, "" + if r.ID == "" { + id := make([]byte, 8) + _, _ = rand.Read(id) + r.ID = "mirror-" + hex.EncodeToString(id) + } 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) } @@ -265,7 +279,7 @@ func (v *Verbs) Mirrored(r MirrorRequest) error { if err != nil { return err } - e := Event{Name: src.Name(), Kind: KindUpstream, Digest: r.Digest, Holder: Self, By: r.By} + e := Event{Name: src.Name(), Kind: KindUpstream, Digest: r.Digest, Holder: Self, By: r.By, ID: r.ID} 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) @@ -276,6 +290,25 @@ func (v *Verbs) Mirrored(r MirrorRequest) error { return v.Mirrors.Delete(MirrorsKey(src.Name(), r.Digest)) } +// Take decides one change of the state mirrors for the follower self: an ask no follower claimed, or one claimed +// before this follower's watch began (its claimant may have been cut by a restart), is claimed by compare-and-set at +// revision and copied by the follower that won; an ask claimed since, by self or by another, is passed. Two followers +// that both start on the same cut ask can both win in turn; the outcome then carries the ask's id, so it is said once. +func (v *Verbs) Take(key string, r MirrorRequest, revision uint64, current bool, self string) error { + if r.Claim != "" && (r.Claim == self || !current) { + return nil + } + r.Claim = self + won, err := v.Mirrors.Claim(key, r, revision) + if err != nil { + return fmt.Errorf("mirror: the ask %s cannot be claimed: %w", key, err) + } + if !won { + return nil + } + return v.Mirrored(r) +} + // HeldNow lists the holds, for registry_holds. func (v *Verbs) HeldNow() ([]Hold, error) { all, err := v.Holds.List()