From 78fc3ffa359755250832c2b2e7305a68379edee7 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 16:38:17 +0100 Subject: [PATCH] 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 --- .../cmd/mesh-delivery-builder/main.go | 15 ++- .../mesh-delivery-builder/publishbounds.go | 101 ++++++++++++++++++ .../cmd/mesh-delivery-builder/refused_test.go | 63 +++++++++++ .../cmd/mesh-delivery-builder/seats.go | 14 +++ .../internal/versions/claim.go | 14 ++- .../internal/versions/publish.go | 24 +++++ .../internal/versions/versions_test.go | 5 + .../mesh-delivery-builder/manifest_test.go | 2 +- modules/mesh-delivery-builder/module.json | 1 + 9 files changed, 234 insertions(+), 5 deletions(-) create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go create mode 100644 modules/mesh-delivery-builder/internal/versions/publish.go diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index d58246d3..fe855e4c 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/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 { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go new file mode 100644 index 00000000..b3dac3e2 --- /dev/null +++ b/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."}) +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go index a7305b5c..7e4282ae 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go +++ b/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 } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go index 58ca0421..098abc3e 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go +++ b/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) diff --git a/modules/mesh-delivery-builder/internal/versions/claim.go b/modules/mesh-delivery-builder/internal/versions/claim.go index bc87d79b..a9990b02 100644 --- a/modules/mesh-delivery-builder/internal/versions/claim.go +++ b/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 { diff --git a/modules/mesh-delivery-builder/internal/versions/publish.go b/modules/mesh-delivery-builder/internal/versions/publish.go new file mode 100644 index 00000000..f9236720 --- /dev/null +++ b/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" +) diff --git a/modules/mesh-delivery-builder/internal/versions/versions_test.go b/modules/mesh-delivery-builder/internal/versions/versions_test.go index b31f0208..baedbd95 100644 --- a/modules/mesh-delivery-builder/internal/versions/versions_test.go +++ b/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) diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index aba1e8fe..d7362cb0 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/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 diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 54217918..71a1b159 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/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 },