F-delivery 5: the builder raises publish-refused, publish-unanswered and seat-outcome-unread

A holder's refused, and a version no holder answered, raised no finding outside the floor, so a build said done
while its file was not in the registry; a seat's outcome that does not read was only logged (design 51 Health
and conditions). The bound is a key of the new state publish-bounds with its own time to live, armed when a
built version is still numbered; a later published of the name clears both findings.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go

@@ -158,6 +158,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 				logger.Printf("%v", err)
 			}
 		}}
+	// A publish that ends badly is a finding (design 51 *Health and conditions*): a holder's refused, a version no holder
+	// answered within its bound (a key of publish-bounds with its own time to live), a seat's outcome that does not read.
+	pubBoundsKV := process.AwaitBucket(ctx, js, conn, versions.PublishBoundsBucket, "a publish no holder answers is not raised", logger.Printf)
+	pubs := &publishes{bounds: pubBoundsKV.KV, versions: numberer.Versions, findings: found, bound: versions.PublishBound, logf: logger.Printf}
+	numberer.Publishing = pubs.arm
 	b := &build.Builder{
 		Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
 		Asks:    buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound},
@@ -322,7 +327,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// condition keeper's condition-cleared.
 	floor := floorWake{b: b, waits: floorWaits, logf: logger.Printf}
 	// And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version.
-	seats := seatHearer{n: numberer, b: b, logf: logger.Printf}
+	seats := seatHearer{n: numberer, b: b, logf: logger.Printf, pub: pubs}
 	hear(eventsStream, delivery.ArtifactsGoneEvent+", what wakes a floor, the registry seats' words and manifest-reads-changed are not heard until the builder starts again", func(m jetstream.Msg) {
 		if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4)
 			return
@@ -365,6 +370,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			go func() { errs <- follow(ctx, tw, each) }()
 		}
 	})
+	pubBoundsKV.Ready(func(kv jetstream.KeyValue) {
+		each := func(e jetstream.KeyValueEntry) { pubs.ended(ctx, e) }
+		if pw, err := settle(ctx, kv, each); err != nil {
+			process.NotWatched(logger.Printf, versions.PublishBoundsBucket, err, "a publish no holder answers is not raised until the builder starts again")
+		} else {
+			go func() { errs <- follow(ctx, pw, each) }()
+		}
+	})
 	boundsKV.Ready(func(kv jetstream.KeyValue) {
 		each := func(e jetstream.KeyValueEntry) { bounds.ended(ctx, e) }
 		if bw, err := settle(ctx, kv, each); err != nil {

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go

@@ -0,0 +1,101 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/health"
+
+	"mesh-delivery-builder/internal/versions"
+)
+
+// publishes raises and clears the builder's findings of a publish that ended badly (versions/publish.go): a holder's
+// refused, a version unanswered past its bound, kept as a key of publish-bounds with its own time to live, and a seat's
+// outcome that does not read.
+type publishes struct {
+	bounds   func() jetstream.KeyValue
+	versions versions.Versions
+	findings *findings
+	bound    time.Duration
+	logf     func(string, ...any)
+}
+
+// arm writes the version's bound when none stands, so a standing bound keeps its time.
+func (p *publishes) arm(ctx context.Context, e versions.Entry) {
+	kv := p.bounds()
+	if kv == nil {
+		p.logf("the publish bound of %s@%s is not written (%s is not on the bus yet), so it is not raised if no holder answers", e.Name, e.Version, versions.PublishBoundsBucket)
+		return
+	}
+	if _, err := kv.Create(ctx, versions.Key(e.Name, e.Version), []byte(`{}`), jetstream.KeyTTL(p.bound)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) {
+		p.logf("the publish bound of %s@%s is not written, so it is not raised if no holder answers: %v", e.Name, e.Version, err)
+	}
+}
+
+// heard is told each holder's word the numberer took in: published clears the name's findings, refused raises
+// publish-refused with the holder's reason, and either ends the version's bound.
+func (p *publishes) heard(ctx context.Context, seat, event string, o versions.Outcome) {
+	if o.Version != "" {
+		if kv := p.bounds(); kv != nil {
+			if err := kv.Delete(ctx, versions.Key(o.Name, o.Version)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+				p.logf("the publish bound of %s@%s is not ended: %v", o.Name, o.Version, err)
+			}
+		}
+	}
+	switch event {
+	case versions.HeardPublished:
+		p.findings.clear(ctx, versions.PublishRefused, health.ScopeModule, o.Name)
+		p.findings.clear(ctx, versions.PublishUnanswered, health.ScopeModule, o.Name)
+	case versions.HeardRefused:
+		reason := o.Reason
+		if reason == "" {
+			reason = "it gave no reason"
+		}
+		p.findings.raise(ctx, finding{kind: versions.PublishRefused, scope: health.ScopeModule, id: o.Name,
+			summary:     fmt.Sprintf("%s refused to publish %s@%s (digest %s): %s", seat, o.Name, o.Version, o.Digest, reason),
+			headline:    "A registry seat refused a publish",
+			explanation: "The builder built and numbered " + o.Name + " " + o.Version + ", and the holder of " + seat + " refused to store it, so that version is not in the registry.",
+			resolved:    "That version is published, or a later version of " + o.Name + " is.",
+			needs:       "builder_versions shows the version and why; builder_publish_again publishes it again once the cause is mended."})
+	}
+}
+
+// unread raises seat-outcome-unread about seat: what it said on subject is no outcome. read clears it.
+func (p *publishes) unread(ctx context.Context, seat, subject string, why error) {
+	p.findings.raise(ctx, finding{kind: versions.SeatOutcomeUnread, scope: health.ScopeModule, id: seat,
+		summary:     fmt.Sprintf("%s said on %s what is no outcome of a publish: %v", seat, subject, why),
+		headline:    "A registry seat said what the builder cannot read",
+		explanation: "The builder hears each registry seat's published, refused and removed to move a numbered version; this one does not read, so the version it was about does not move.",
+		resolved:    "The next outcome of " + seat + " reads.",
+		needs:       "a look at the seat's holder: the outcome's body is its contract's (mesh-contracts)."})
+}
+
+func (p *publishes) read(ctx context.Context, seat string) {
+	p.findings.clear(ctx, versions.SeatOutcomeUnread, health.ScopeModule, seat)
+}
+
+// ended is told each change of publish-bounds: a bound whose time ended (the bus's marker) on a version still numbered
+// raises publish-unanswered.
+func (p *publishes) ended(ctx context.Context, e jetstream.KeyValueEntry) {
+	if e.Operation() != jetstream.KeyValuePurge {
+		return
+	}
+	raw, _, found, err := p.versions.Store.Get(ctx, e.Key())
+	if err != nil || !found {
+		return
+	}
+	v, err := versions.Decode(raw)
+	if err != nil || v.State != versions.Numbered {
+		return
+	}
+	p.findings.raise(ctx, finding{kind: versions.PublishUnanswered, scope: health.ScopeModule, id: v.Name,
+		summary:     fmt.Sprintf("%s said neither published nor refused of %s@%s within %s of its build", v.Seat, v.Name, v.Version, p.bound),
+		headline:    "A registry seat did not answer a publish",
+		explanation: "The builder built and numbered " + v.Name + " " + v.Version + " and asked " + v.Seat + " to publish it; its holder said nothing within the bound, so the build says done while the version may not be in the registry.",
+		resolved:    "That version is published, or a later version of " + v.Name + " is.",
+		needs:       "a look at the holder of " + v.Seat + " (its health and log); builder_publish_again asks the publish again."})
+}

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go

@@ -340,6 +340,41 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) {
 	if _, ok := st.vals[floorKey]; ok {
 		t.Error("a known floor leaves floor-silent")
 	}
+	// Design 51 *Health and conditions* (review round 1): a holder's refused is publish-refused with its reason, a version
+	// no holder answered within its bound publish-unanswered, a seat's outcome that does not read seat-outcome-unread;
+	// a later published of the name clears the first two, the seat's next outcome that reads the third.
+	vs := versions.Versions{Store: memVersions{}}
+	for _, v := range []string{"0.1.3", "0.1.4"} {
+		raw, err := versions.Encode(versions.Entry{Name: "m/program", Kind: versions.KindBundle, Series: "0.1", Version: v, Seat: "oci-artifact-registry", State: versions.Numbered})
+		if err != nil {
+			t.Fatal(err)
+		}
+		vs.Store.(memVersions)[versions.Key("m/program", v)] = raw
+	}
+	pubs := &publishes{bounds: func() jetstream.KeyValue { return nil }, versions: vs, findings: f, bound: versions.PublishBound, logf: t.Logf}
+	h := &seatHearer{n: &versions.Numberer{Versions: vs}, b: &build.Builder{}, logf: t.Logf, pub: pubs}
+	refusedKey := health.FindingKey(health.ScopeModule, "m/program", versions.PublishRefused)
+	refused := &wakeMsg{subject: "mesh.seat.oci-artifact-registry.event.refused", data: []byte(`{"name":"m/program","version":"0.1.3","digest":"sha256:bb","reason":"bytes differ"}`)}
+	if !h.heard(context.Background(), refused) || !refused.acked {
+		t.Fatalf("a refusal: acked %v", refused.acked)
+	}
+	if got, err := health.DecodeFinding(st.vals[refusedKey]); err != nil || !strings.Contains(got.Summary, "refused to publish m/program@0.1.3") || !strings.Contains(got.Summary, "bytes differ") {
+		t.Fatalf("a holder's refused: the finding = %+v, %v", got, err)
+	}
+	pubs.ended(context.Background(), purged{key: versions.Key("m/program", "0.1.4")})
+	unansweredKey := health.FindingKey(health.ScopeModule, "m/program", versions.PublishUnanswered)
+	if got, err := health.DecodeFinding(st.vals[unansweredKey]); err != nil || !strings.Contains(got.Summary, "neither published nor refused of m/program@0.1.4") {
+		t.Fatalf("a bound ended unanswered: the finding = %+v, %v", got, err)
+	}
+	malformed := &wakeMsg{subject: "mesh.seat.oci-artifact-registry.event.published", data: []byte(`{"version":"0.1.4"}`)}
+	unreadKey := health.FindingKey(health.ScopeModule, "oci-artifact-registry", versions.SeatOutcomeUnread)
+	if !h.heard(context.Background(), malformed) || !malformed.acked || st.vals[unreadKey] == nil {
+		t.Fatalf("an outcome that does not read: acked %v, findings %v", malformed.acked, st.vals)
+	}
+	published := &wakeMsg{subject: "mesh.seat.oci-artifact-registry.event.published", data: []byte(`{"name":"m/program","version":"0.1.4","digest":"sha256:cc"}`)}
+	if !h.heard(context.Background(), published) || st.vals[refusedKey] != nil || st.vals[unansweredKey] != nil || st.vals[unreadKey] != nil {
+		t.Fatalf("a later published leaves the findings: %v", st.vals)
+	}
 	// ADR 0615 §5 (review round 1): a consumer not there yet is family-not-heard naming its stream, cleared when it binds.
 	notHeard := f.notHeard(context.Background())
 	unheardKey := heard.Key(delivery.BuilderModule, "FORGE")
@@ -394,6 +429,34 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) {
 	}
 }
 
+// memVersions is a state versions in memory.
+type memVersions map[string][]byte
+
+func (m memVersions) Get(_ context.Context, key string) ([]byte, uint64, bool, error) {
+	v, ok := m[key]
+	return v, 1, ok, nil
+}
+func (m memVersions) Create(_ context.Context, key string, v []byte) (uint64, error) {
+	m[key] = v
+	return 1, nil
+}
+func (m memVersions) Update(_ context.Context, key string, v []byte, _ uint64) (uint64, error) {
+	m[key] = v
+	return 2, nil
+}
+func (m memVersions) KeysOf(context.Context, string) ([]string, error) {
+	return slices.Collect(maps.Keys(m)), nil
+}
+
+// purged is the bus's marker of a key whose time to live ended.
+type purged struct {
+	jetstream.KeyValueEntry
+	key string
+}
+
+func (p purged) Key() string                     { return p.key }
+func (p purged) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePurge }
+
 // memFindings is a findings bucket in memory.
 type memFindings struct{ vals map[string][]byte }
 

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go

@@ -44,6 +44,8 @@ type seatHearer struct {
 	n    *versions.Numberer
 	b    *build.Builder
 	logf func(string, ...any)
+	// pub raises and clears the findings of a publish that ended badly; nil raises none.
+	pub *publishes
 }
 
 // heard takes a message of the events consumer in when it is a registry seat's outcome, and answers whether it was.
@@ -56,10 +58,19 @@ func (h *seatHearer) heard(ctx context.Context, m jetstream.Msg) bool {
 	}
 	var o versions.Outcome
 	if err := json.Unmarshal(m.Data(), &o); err != nil || o.Name == "" {
+		if err == nil {
+			err = errors.New("it names no name")
+		}
 		h.logf("%s holds what is no outcome of %s, left: %v", m.Subject(), se.seat, err)
+		if h.pub != nil {
+			h.pub.unread(ctx, se.seat, m.Subject(), err)
+		}
 		h.ack(m)
 		return true
 	}
+	if h.pub != nil {
+		h.pub.read(ctx, se.seat)
+	}
 	id := ""
 	if m.Headers() != nil {
 		id = m.Headers().Get("x-event-id")
@@ -72,6 +83,9 @@ func (h *seatHearer) heard(ctx context.Context, m jetstream.Msg) bool {
 		}
 		return true
 	}
+	if h.pub != nil {
+		h.pub.heard(ctx, se.seat, se.event, o)
+	}
 	h.ack(m)
 	if len(wake) > 0 {
 		h.logf("the floor of %s is asked again: %s said %s", strings.Join(wake, ", "), se.seat, se.event)

modules/mesh-delivery-builder/internal/versions/claim.go

@@ -142,7 +142,10 @@ type Numberer struct {
 	Said func(delivery.VersionNumbered)
 	// Names answers a package's name in its ecosystem; nil numbers a package as NameOf names it.
 	Names PackageNames
-	Now   func() time.Time
+	// Publishing is told each version Built recorded that is still numbered, its holder's word not heard yet: the
+	// builder arms its publish bound (publish.go); nil arms none.
+	Publishing func(ctx context.Context, e Entry)
+	Now        func() time.Time
 }
 
 // Number numbers every numbered artifact of c. An error is a store that cannot be read or written: the claim is then
@@ -237,14 +240,19 @@ func (n *Numberer) Built(ctx context.Context, c BuildClaim, got Numbering, diges
 		if d == "" {
 			continue
 		}
-		if _, err := n.Versions.Mark(ctx, got.NameOf(c.Module, artifact), version, func(e *Entry) {
+		e, err := n.Versions.Mark(ctx, got.NameOf(c.Module, artifact), version, func(e *Entry) {
 			// The new digest replaces the recorded one only while no published was said for the number (design 51
 			// *Tools of the builder*): a rebuild of a gone file under the same number.
 			if e.Digest == "" || e.State == Numbered || e.State == Refused {
 				e.Digest = d
 			}
-		}); err != nil {
+		})
+		if err != nil {
 			errs = append(errs, err.Error())
+			continue
+		}
+		if e.State == Numbered && n.Publishing != nil {
+			n.Publishing(ctx, e)
 		}
 	}
 	if len(errs) > 0 {

modules/mesh-delivery-builder/internal/versions/publish.go

@@ -0,0 +1,24 @@
+package versions
+
+import "time"
+
+// **A publish that ends badly is loud** (design 51 *Health and conditions*; review round 1): the builder raises the
+// finding publish-refused when a holder says refused, and publish-unanswered when a version it built and numbered is
+// neither published nor refused within PublishBound. Each is about the name, so a later published of any version of it
+// clears both. A seat's outcome that does not read is the finding seat-outcome-unread about the seat, cleared by its
+// next outcome that reads.
+//
+// The bound is a key of the state publish-bounds per version (Key), written with its own time to live when the build
+// is done and the version is still numbered, deleted at the holder's word; the bus's marker of its end, heard on the
+// watch, is what raises publish-unanswered while the version is still numbered. No timer.
+const (
+	PublishBoundsState  = "publish-bounds"
+	PublishBoundsBucket = "mesh-delivery-builder_publish-bounds"
+	// PublishBound is the build module's own wait for the holder's word (design 51 *How a publish ends*: at most
+	// 10 minutes), so a version unanswered past it is one the holder never answered.
+	PublishBound = 10 * time.Minute
+
+	PublishRefused    = "publish-refused"
+	PublishUnanswered = "publish-unanswered"
+	SeatOutcomeUnread = "seat-outcome-unread"
+)

modules/mesh-delivery-builder/internal/versions/versions_test.go

@@ -421,9 +421,14 @@ func TestTheBuildersVersions(t *testing.T) {
 		}
 		said, asked = saidBefore, askedBefore
 		got, _ = n.Number(ctx, c)
+		var publishing []string // review round 1: a version built and still numbered arms its publish bound
+		n.Publishing = func(_ context.Context, e Entry) { publishing = append(publishing, e.Name+"@"+e.Version) }
 		if err := n.Built(ctx, c, got, map[string]string{"program": "sha256:aa"}); err != nil {
 			t.Fatal(err)
 		}
+		if n.Publishing = nil; !reflect.DeepEqual(publishing, []string{"m/program@0.1.14"}) {
+			t.Fatalf("the publish bounds armed: %v, want m/program@0.1.14", publishing)
+		}
 		again, _ := n.Number(ctx, c)
 		if again.Versions["program"] != "0.1.14" || len(said) != 1 || len(asked) != 3 {
 			t.Fatalf("again: %+v, said %d, asked %v; want the number kept, nothing said, the floor not asked", again, len(said), asked)

modules/mesh-delivery-builder/manifest_test.go

@@ -94,7 +94,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	if m.Module != "mesh-delivery-builder" {
 		t.Errorf("module = %q", m.Module)
 	}
-	if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
+	if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "publish-bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and

modules/mesh-delivery-builder/module.json

@@ -81,6 +81,7 @@
     { "name": "hearing", "history": 1 },
     { "name": "instances", "history": 1 },
     { "name": "old-form", "history": 1 },
+    { "name": "publish-bounds", "history": 1, "key-ttl": true },
     { "name": "queues", "history": 1 },
     { "name": "reasks", "history": 1 },
     { "name": "tag-ask-bounds", "history": 1, "key-ttl": true },