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

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)
 	})
 }
 

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 {

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))
 }
 

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()