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 93d3a943..0289ec8f 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -381,11 +381,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } pubs.ended(ctx, e) } - if pw, err := settle(ctx, kv, each); err != nil { - process.NotWatched(logger.Printf, versions.BoundsBucket, err, "a publish no holder answers, and a node holding the old form seven days, are not raised until the builder starts again") - } else { - go func() { errs <- follow(ctx, pw, each) }() - } + watchBounds(ctx, kv, each, found.notHeard(ctx), errs, logger.Printf) }) boundsKV.Ready(func(kv jetstream.KeyValue) { each := func(e jetstream.KeyValueEntry) { bounds.ended(ctx, e) } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go index 7f4340df..d3b6a9c9 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go @@ -9,6 +9,9 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/process" + + "git.novox.be/novox/mesh-delivery/sdk/heard" "mesh-delivery-builder/internal/versions" ) @@ -24,16 +27,31 @@ type publishes struct { logf func(string, ...any) } -// arm writes the version's bound when none stands, so a standing bound keeps its time. +// arm writes the version's bound when none stands, so a standing bound keeps its time. A bound that cannot be written +// would never end, so the version is publish-unanswered at once (review round 2): the finding says why. func (p *publishes) arm(ctx context.Context, e versions.Entry) { kv := p.bounds() + var err error 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.BoundsBucket) + err = fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket) + } else if _, err = kv.Create(ctx, versions.PublishBoundKey(e.Name, e.Version), versions.EncodeBound("the publish of "+e.Name+"@"+e.Version+" to "+e.Seat, time.Now().Add(p.bound)), jetstream.KeyTTL(p.bound)); errors.Is(err, jetstream.ErrKeyExists) { + err = nil + } + if err == nil { return } - if _, err := kv.Create(ctx, versions.PublishBoundKey(e.Name, e.Version), versions.EncodeBound("the publish of "+e.Name+"@"+e.Version+" to "+e.Seat, time.Now().Add(p.bound)), 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) - } + p.logf("the publish bound of %s@%s is not written, so it is raised unanswered now: %v", e.Name, e.Version, err) + p.unanswered(ctx, e, fmt.Sprintf("its bound could not be written (%v), so the builder cannot wait on the holder's word", err)) +} + +// unanswered raises publish-unanswered about e's name, saying why. +func (p *publishes) unanswered(ctx context.Context, v versions.Entry, why string) { + 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: %s", v.Seat, v.Name, v.Version, why), + 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; no word of its holder is known, 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."}) } // heard is told each holder's word the numberer took in: published clears the name's findings, refused raises @@ -93,10 +111,19 @@ func (p *publishes) ended(ctx context.Context, e jetstream.KeyValueEntry) { 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."}) + p.unanswered(ctx, v, fmt.Sprintf("nothing within %s of its build", p.bound)) +} + +// watchBounds watches the state bounds, handing each change to each: a watch that cannot start loses every bound's end, +// so it is the finding family-not-heard naming the state until a later start watches it (review round 2). +func watchBounds(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyValueEntry), w heard.Writer, errs chan<- error, logf func(string, ...any)) { + const consequence = "a publish no holder answers, and a node holding the old form seven days, are not raised until the builder starts again" + pw, err := settle(ctx, kv, each) + if err != nil { + process.NotWatched(logf, versions.BoundsBucket, err, consequence) + w.NotHeard(versions.BoundsState, consequence, err) + return + } + w.Heard(versions.BoundsState) + go func() { errs <- follow(ctx, pw, each) }() } 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 1494f272..0e53e2f2 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 @@ -366,6 +366,18 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { 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) } + // Review round 2: a bound that cannot be written is publish-unanswered at once; a watch of bounds that cannot start + // is family-not-heard naming the state. + delete(st.vals, unansweredKey) + pubs.arm(context.Background(), versions.Entry{Name: "m/program", Version: "0.1.4", Seat: "oci-artifact-registry"}) + if got, err := health.DecodeFinding(st.vals[unansweredKey]); err != nil || !strings.Contains(got.Summary, "its bound could not be written") { + t.Fatalf("a bound not written: the finding = %+v, %v", got, err) + } + watchBounds(context.Background(), refusedWatch{err: errors.New("permissions violation")}, func(jetstream.KeyValueEntry) {}, f.notHeard(context.Background()), nil, t.Logf) + if got, err := health.DecodeFinding(st.vals[heard.Key(delivery.BuilderModule, versions.BoundsState)]); err != nil || !strings.Contains(got.Summary, "permissions violation") { + t.Fatalf("bounds not watched: the finding = %+v, %v", got, err) + } + delete(st.vals, heard.Key(delivery.BuilderModule, versions.BoundsState)) 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 {