diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 2f2512d..3ba0133 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -86,11 +86,11 @@ var keptStates = []announce.State{ // buckets is every bucket the assigner opened, by name. type buckets map[string]jetstream.KeyValue -// open opens every bucket the assigner reads and writes; a bucket it cannot open stops the start, named with its grant (issue 1945), since a pass -// without it would write wants from half the mesh. The puts' bucket is the exception: whatever keeps it from opening (it +// open opens every bucket the assigner reads and writes; a bucket it cannot open stops the start, named with its grant (issue 1945), +// since a pass without it would write wants from half the mesh. The puts' bucket is the exception: whatever keeps it from opening (it // does not exist yet, the credential is not granted it yet, the bus does not answer in time), the assigner writes wants // without it and takes no put further until it starts again (ADR 0546). -func open(ctx context.Context, js bucketOpener, logf func(string, ...any)) (buckets, error) { +func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any)) (buckets, error) { out := buckets{} for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) { kv, err := js.KeyValue(ctx, name) @@ -99,7 +99,7 @@ func open(ctx context.Context, js bucketOpener, logf func(string, ...any)) (buck continue } if err != nil { - return nil, essentialErr(name, "write a want from the whole mesh", err) + return nil, process.EssentialErr("assigner", name, "write a want from the whole mesh", err) } out[name] = kv } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/optional.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/optional.go deleted file mode 100644 index 0267144..0000000 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/optional.go +++ /dev/null @@ -1,27 +0,0 @@ -package main - -import ( - "context" - "fmt" - - "github.com/nats-io/nats.go/jetstream" -) - -// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the builder of -// mesh-delivery #48 exited on a refused read of the gate's source, and its new reasks bucket would have done the same). A -// grant reaches the bus with a delivery, so a bucket the bus refuses, or times out on, at start is often only not granted -// yet: exiting for it stops the part until a person's hand. The assigner works without none of its buckets but the Go build -// module's steps, so the others stop the start, naming the grant to look at. (The same two functions are in each delivery part; -// they belong in mesh-sdk.) - -// bucketOpener is the part of the bus a start opens buckets with. -type bucketOpener interface { - KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) -} - -// essentialErr is the error of a bucket the part cannot work without: it names the bucket and the grant. -func essentialErr(name, does string, err error) error { - return fmt.Errorf("the bucket %s cannot be read (%v): the assigner cannot %s without it, so it stops; where the bus "+ - "refused it or did not answer, this module's grant of %s (its manifest's reads or state, issued with its credential) "+ - "has not reached the bus yet", name, err, does, name) -} diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index 893c786..df8760f 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-assigner go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum index 2d6d7be..8492faa 100644 --- a/modules/mesh-delivery-assigner/go.sum +++ b/modules/mesh-delivery-assigner/go.sum @@ -6,6 +6,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8 h1:vhYk/iv4 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5 h1:gE80bR0tPME7MTMBLcdP62v+F/9soVXmdcBYNMOye44= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= 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 d6b2e9f..1726dc5 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -13,7 +13,6 @@ import ( "encoding/json" "errors" "fmt" - "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" "io" "log" "maps" @@ -25,6 +24,8 @@ import ( "syscall" "time" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -124,20 +125,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - builds, err := essentialBucket(ctx, js, delivery.BuilderBuildsBucket, "claim or write a build key") + builds, err := process.EssentialBucket(ctx, js, "builder", delivery.BuilderBuildsBucket, "claim or write a build key") if err != nil { return err } - modules, err := essentialBucket(ctx, js, delivery.ModulesBucket, "know any fingerprint to build") + modules, err := process.EssentialBucket(ctx, js, "builder", delivery.ModulesBucket, "know any fingerprint to build") if err != nil { return err } - manifests, err := essentialBucket(ctx, js, delivery.ManifestsBucket, "read what a module builds") + manifests, err := process.EssentialBucket(ctx, js, "builder", delivery.ManifestsBucket, "read what a module builds") if err != nil { return err } // The candidates' fingerprints are built too (issue 1684): a builder that cannot read them builds the trunk's. - candidates := optionalBucket(ctx, js, delivery.CandidatesBucket, "no candidate's fingerprint is built until the builder starts again; the trunk's are", logger.Printf) + candidates := process.OptionalBucket(ctx, js, delivery.CandidatesBucket, "no candidate's fingerprint is built until the builder starts again; the trunk's are", logger.Printf) bound := defaultStepBound if v := strings.TrimSpace(getenv("MESH_BUILDER_STEP_BOUND")); v != "" { if bound, err = time.ParseDuration(v); err != nil { @@ -188,7 +189,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // controller delivers from this builder's keys only while an instance says it writes references. A bucket not there // yet (the manifest's state not made) is said, and the controller's plan waits for the keys, named (no build-agent // is asked since R4g, ADR 0524 §7). - switch kv := optionalBucket(ctx, js, instancesBucket, "the controller is not told that this builder writes references until it starts again", logger.Printf); { + switch kv := process.OptionalBucket(ctx, js, instancesBucket, "the controller is not told that this builder writes references until it starts again", logger.Printf); { case kv == nil: default: says := newSayer(ctx, kv, cred.Node, []string{buildrun.GoBuildModule, buildrun.NodeBuildModule, build.ImageBuildModule}, logger.Printf) @@ -272,17 +273,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if candidates != nil { cw, err := settle(ctx, candidates, eachCandidate) if err != nil { - logger.Printf("the bucket %s cannot be watched (%v): no candidate's fingerprint is built until the builder starts again", delivery.CandidatesBucket, err) + process.NotWatched(logger.Printf, delivery.CandidatesBucket, err, "no candidate's fingerprint is built until the builder starts again") } else { go func() { errs <- follow(ctx, cw, eachCandidate) }() } } // A person asks a key's build again with builder_ask_again, which writes the reasks state (issue 1945): watched, as // every other input. The builder asks again by itself without it. - if kv := optionalBucket(ctx, js, build.ReAsksBucket, "builder_ask_again is not heard until the builder starts again; an ask that can no longer be answered is still asked again by itself", logger.Printf); kv != nil { + if kv := process.OptionalBucket(ctx, js, build.ReAsksBucket, "builder_ask_again is not heard until the builder starts again; an ask that can no longer be answered is still asked again by itself", logger.Printf); kv != nil { eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, kv, e, logger.Printf) } if rw, err := settle(ctx, kv, eachReAsk); err != nil { - logger.Printf("the bucket %s cannot be watched (%v): builder_ask_again is not heard until the builder starts again", build.ReAsksBucket, err) + process.NotWatched(logger.Printf, build.ReAsksBucket, err, "builder_ask_again is not heard until the builder starts again") } else { go func() { errs <- follow(ctx, rw, eachReAsk) }() } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/optional.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/optional.go index bd2542e..02afe10 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/optional.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/optional.go @@ -10,39 +10,13 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" ) -// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the builder of +// **A part does not stop for a bucket it can work without** (mesh-sdk process.EssentialBucket, OptionalBucket) (novox/hq issue 1945, met twice on 2026-10-25: the builder of // mesh-delivery #48 exited on a refused read of the gate's source, and its reasks bucket would have done the same). A grant // reaches the bus with a delivery, so a bucket the bus refuses, or times out on, at start is often only not granted yet: // exiting for it stops every build until a person's hand. A bucket the builder cannot build at all without (its builds, // the indexer's modules and manifests) still stops the start, naming the grant to look at; every other is said and left, // and opened again at the next event that needs it where there is one. -// essentialBucket opens a bucket the builder cannot work without; its error names the bucket and the grant. -func essentialBucket(ctx context.Context, js bucketOpener, name, does string) (jetstream.KeyValue, error) { - kv, err := js.KeyValue(ctx, name) - if err != nil { - return nil, essentialErr(name, does, err) - } - return kv, nil -} - -// essentialErr is the error of a bucket the part cannot work without: it names the bucket and the grant. -func essentialErr(name, does string, err error) error { - return fmt.Errorf("the bucket %s cannot be read (%v): the builder cannot %s without it, so it stops; where the bus "+ - "refused it or did not answer, this module's grant of %s (its manifest's reads or state, issued with its credential) "+ - "has not reached the bus yet", name, err, does, name) -} - -// optionalBucket opens a bucket the builder works without: any error is said with what is not heard, and nil answered. -func optionalBucket(ctx context.Context, js bucketOpener, name, without string, logf func(string, ...any)) jetstream.KeyValue { - kv, err := js.KeyValue(ctx, name) - if err != nil { - logf("the bucket %s cannot be read (%v): %s", name, err, without) - return nil - } - return kv -} - // trunkSource is the gate's source history the candidates' previews are merged onto (sourceHeads), opened at start and, // while it cannot be, again at the next candidate that asks a trunk head: a builder without it builds the trunk's keys, // and a candidate's preview waits, saying why. diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go index ae19091..4a0e7ca 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go @@ -12,6 +12,8 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -111,7 +113,7 @@ func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) { if !strings.Contains(logs.String(), "cannot be watched") || !strings.Contains(logs.String(), "cannot be read") { t.Errorf("the refusals were not said: %q", logs.String()) } - if _, err := essentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "mesh-delivery-indexer_modules", "work"); err == nil || + if _, err := process.EssentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "builder", "mesh-delivery-indexer_modules", "work"); err == nil || !strings.Contains(err.Error(), "grant of mesh-delivery-indexer_modules") { t.Errorf("a refused bucket the part needs: %v, want an error naming its grant", err) } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go index f574c2d..58ae514 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go @@ -3,11 +3,14 @@ package main import ( "context" "errors" - "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" + "fmt" "log" "strings" "sync" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -67,7 +70,7 @@ func (w *toolchainWatch) ensure(module string) error { } watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) }) if err != nil { - w.logger.Printf("the bucket %s cannot be watched (%v): %s's toolchains are not heard until its next report", name, err, module) + process.NotWatched(w.logger.Printf, name, err, fmt.Sprintf("%s's toolchains are not heard until its next report", module)) return nil } w.watched[module] = true diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index 615cce0..fb89b4d 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-builder go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-builder/go.sum b/modules/mesh-delivery-builder/go.sum index d648ca0..2dfa4fd 100644 --- a/modules/mesh-delivery-builder/go.sum +++ b/modules/mesh-delivery-builder/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1 h1:fzOkrz9CSp3mEjeV7NfZx8yoivI1sD1AOtIPtyyF8RI= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 7b97f0d..c46ec4d 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -157,7 +157,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro buckets := map[string]jetstream.KeyValue{} for _, name := range []string{delivery.CheckerVerdictsBucket, delivery.CheckerKeptBucket, delivery.CheckerOwedBucket, delivery.PreviewsBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.ManifestsBucket, check.TriesBucket} { - if buckets[name], err = essentialBucket(ctx, js, name, "judge or keep a check"); err != nil { + if buckets[name], err = process.EssentialBucket(ctx, js, "checker", name, "judge or keep a check"); err != nil { return err } } @@ -207,7 +207,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The builder's builds say how the candidates' fingerprints stand (issue 1684); where its bucket is not there yet the line // says nothing of the builds, until the checker starts again. var builds check.BuildStates - if kv := optionalBucket(ctx, js, delivery.BuilderBuildsBucket, "the line says nothing of the builds until the checker starts again", logger.Printf); kv != nil { + if kv := process.OptionalBucket(ctx, js, delivery.BuilderBuildsBucket, "the line says nothing of the builds until the checker starts again", logger.Printf); kv != nil { builds = buildReader{kvStore{kv}} } // The assigner's compositions (ADR 0528 clause 5, issue 1297): their bucket is looked up at each check until it is @@ -280,9 +280,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // the assigner's own presence says whether a composition is being made (compositionWatch, ADR 0537 §4). A check is no // longer held for a node: its asks wait in the build seats' streams, which redeliver a lost node's ask (ADR 0525 §6, // issue 1912), so the node-engine's have is not watched here. - if kv := optionalBucket(ctx, js, delivery.PresenceBucket, "no node counts as online until the checker starts again", logger.Printf); kv != nil { + if kv := process.OptionalBucket(ctx, js, delivery.PresenceBucket, "no node counts as online until the checker starts again", logger.Printf); kv != nil { if pw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { presence.Put(e.Key(), e.Value(), !opOf(e)) }); err != nil { - logger.Printf("the bucket %s cannot be watched (%v): no node counts as online until the checker starts again", delivery.PresenceBucket, err) + process.NotWatched(logger.Printf, delivery.PresenceBucket, err, "no node counts as online until the checker starts again") } else { go func() { errs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) { @@ -295,14 +295,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // The gate's source says the trunk head of each repository, which a pull request is merged onto. Where it is not there // yet no trunk is known, so nothing is checked. - if kv := optionalBucket(ctx, js, delivery.SourceBucket, "no trunk is known, so nothing is checked until the checker starts again", logger.Printf); kv != nil { + if kv := process.OptionalBucket(ctx, js, delivery.SourceBucket, "no trunk is known, so nothing is checked until the checker starts again", logger.Printf); kv != nil { sw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { if e.Key() == delivery.SourceMainKey { trunks.Put(e.Value(), !opOf(e)) } }) if err != nil { - logger.Printf("the bucket %s cannot be watched (%v): no trunk is known, so nothing is checked until the checker starts again", delivery.SourceBucket, err) + process.NotWatched(logger.Printf, delivery.SourceBucket, err, "no trunk is known, so nothing is checked until the checker starts again") } else { go func() { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/optional.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/optional.go deleted file mode 100644 index 466cdd0..0000000 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/optional.go +++ /dev/null @@ -1,41 +0,0 @@ -package main - -import ( - "context" - "fmt" - - "github.com/nats-io/nats.go/jetstream" -) - -// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the builder of -// mesh-delivery #48 exited on a refused read of the gate's source, and its new reasks bucket would have done the same). A -// grant reaches the bus with a delivery, so a bucket the bus refuses, or times out on, at start is often only not granted -// yet: exiting for it stops the part until a person's hand. A bucket the checker cannot do its job at all without still stops -// the start, naming the grant to look at; every other is said and left. (The same two functions are in each delivery part; -// they belong in mesh-sdk.) - -// essentialBucket opens a bucket the checker cannot work without; its error names the bucket and the grant. -func essentialBucket(ctx context.Context, js bucketOpener, name, does string) (jetstream.KeyValue, error) { - kv, err := js.KeyValue(ctx, name) - if err != nil { - return nil, essentialErr(name, does, err) - } - return kv, nil -} - -// essentialErr is the error of a bucket the part cannot work without: it names the bucket and the grant. -func essentialErr(name, does string, err error) error { - return fmt.Errorf("the bucket %s cannot be read (%v): the checker cannot %s without it, so it stops; where the bus "+ - "refused it or did not answer, this module's grant of %s (its manifest's reads or state, issued with its credential) "+ - "has not reached the bus yet", name, err, does, name) -} - -// optionalBucket opens a bucket the checker works without: any error is said with what is not heard, and nil answered. -func optionalBucket(ctx context.Context, js bucketOpener, name, without string, logf func(string, ...any)) jetstream.KeyValue { - kv, err := js.KeyValue(ctx, name) - if err != nil { - logf("the bucket %s cannot be read (%v): %s", name, err, without) - return nil - } - return kv -} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go index c759d3a..0715aea 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go @@ -12,6 +12,8 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -111,7 +113,7 @@ func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) { if !strings.Contains(logs.String(), "cannot be watched") || !strings.Contains(logs.String(), "cannot be read") { t.Errorf("the refusals were not said: %q", logs.String()) } - if _, err := essentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "mesh-delivery-indexer_modules", "work"); err == nil || + if _, err := process.EssentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "checker", "mesh-delivery-indexer_modules", "work"); err == nil || !strings.Contains(err.Error(), "grant of mesh-delivery-indexer_modules") { t.Errorf("a refused bucket the part needs: %v, want an error naming its grant", err) } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go index 0fe2f41..99c3087 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go @@ -3,10 +3,13 @@ package main import ( "context" "errors" + "fmt" "log" "strings" "sync" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -68,7 +71,7 @@ func (w *toolchainWatch) ensure(module string) error { } watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) }) if err != nil { - w.logger.Printf("the bucket %s cannot be watched (%v): %s's toolchains are not heard until its next report", name, err, module) + process.NotWatched(w.logger.Printf, name, err, fmt.Sprintf("%s's toolchains are not heard until its next report", module)) return nil } w.watched[module] = true diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index e753fa3..5c4d652 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index 744af4c..4e82aa0 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/modules/mesh-delivery-checker/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 h1:FAiFY2KBKEoZSH49OZTt7/zRBIQxyl5g/KhYrpggbN4= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index 383bedc..e069a6c 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -22,6 +22,8 @@ import ( "syscall" "time" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -188,7 +190,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro call := callerOn(conn, cred.User) // No forge event is heard while the open pull requests are read and said (ADR 0443 §5): events and syncs take turns. var turn sync.Mutex - kv, err := essentialBucket(ctx, js, headsBucket, "keep the heads it wrote on the forge stream") + kv, err := process.EssentialBucket(ctx, js, "bridge", headsBucket, "keep the heads it wrote on the forge stream") if err != nil { return err } @@ -197,7 +199,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro rec := &recorder{pub: js, heads: heads, logger: logger} said := func(ctx context.Context) error { return syncOpen(ctx, call, rec, heads, &turn, logger) } // What the gate asks of the forge (ADR 0437 §8): its owed acts and its merges, each performed once per revision. - performedKV, err := essentialBucket(ctx, js, performedBucket, "perform an act of the gate only once") + performedKV, err := process.EssentialBucket(ctx, js, "bridge", performedBucket, "perform an act of the gate only once") if err != nil { return err } @@ -209,7 +211,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro }{{delivery.OwedBucket, perf.owed}, {delivery.MergesBucket, perf.merges}} { // The gate's buckets come with the gate and its grant; the bridge does not wait for them (no timer), and is started // again by its delivery when the gate arrives. A bucket refused or not answered is the same (issue 1945). - kv := optionalBucket(ctx, js, w.bucket, "nothing of it is performed until the bridge starts again", logger.Printf) + kv := process.OptionalBucket(ctx, js, w.bucket, "nothing of it is performed until the bridge starts again", logger.Printf) if kv == nil { continue } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go index 0f9cc32..9a16a64 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go @@ -13,6 +13,8 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -70,10 +72,10 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit t.Run("1945: a refused bucket is said, or stops the start naming its grant", func(t *testing.T) { refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_x") var said []string - if kv := optionalBucket(context.Background(), refusingBus{refusal}, "x", "x is not heard", func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }); kv != nil || len(said) != 1 || !strings.Contains(said[0], "permissions violation") { + if kv := process.OptionalBucket(context.Background(), refusingBus{refusal}, "x", "x is not heard", func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }); kv != nil || len(said) != 1 || !strings.Contains(said[0], "permissions violation") { t.Errorf("an optional bucket refused: %v, said %v", kv, said) } - if _, err := essentialBucket(context.Background(), refusingBus{refusal}, "x", "work"); err == nil || !strings.Contains(err.Error(), "grant of x") { + if _, err := process.EssentialBucket(context.Background(), refusingBus{refusal}, "bridge", "x", "work"); err == nil || !strings.Contains(err.Error(), "grant of x") { t.Errorf("an essential bucket refused: %v, want its grant named", err) } }) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/optional.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/optional.go deleted file mode 100644 index 68df917..0000000 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/optional.go +++ /dev/null @@ -1,46 +0,0 @@ -package main - -import ( - "context" - "fmt" - - "github.com/nats-io/nats.go/jetstream" -) - -// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the builder of -// mesh-delivery #48 exited on a refused read of the gate's source, and its new reasks bucket would have done the same). A -// grant reaches the bus with a delivery, so a bucket the bus refuses, or times out on, at start is often only not granted -// yet: exiting for it stops the part until a person's hand. A bucket the bridge cannot do its job at all without still stops -// the start, naming the grant to look at; every other is said and left. (The same two functions are in each delivery part; -// they belong in mesh-sdk.) - -// bucketOpener is the part of the bus a start opens buckets with. -type bucketOpener interface { - KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) -} - -// essentialBucket opens a bucket the bridge cannot work without; its error names the bucket and the grant. -func essentialBucket(ctx context.Context, js bucketOpener, name, does string) (jetstream.KeyValue, error) { - kv, err := js.KeyValue(ctx, name) - if err != nil { - return nil, essentialErr(name, does, err) - } - return kv, nil -} - -// essentialErr is the error of a bucket the part cannot work without: it names the bucket and the grant. -func essentialErr(name, does string, err error) error { - return fmt.Errorf("the bucket %s cannot be read (%v): the bridge cannot %s without it, so it stops; where the bus "+ - "refused it or did not answer, this module's grant of %s (its manifest's reads or state, issued with its credential) "+ - "has not reached the bus yet", name, err, does, name) -} - -// optionalBucket opens a bucket the bridge works without: any error is said with what is not heard, and nil answered. -func optionalBucket(ctx context.Context, js bucketOpener, name, without string, logf func(string, ...any)) jetstream.KeyValue { - kv, err := js.KeyValue(ctx, name) - if err != nil { - logf("the bucket %s cannot be read (%v): %s", name, err, without) - return nil - } - return kv -} diff --git a/modules/mesh-delivery-forge-bridge/go.mod b/modules/mesh-delivery-forge-bridge/go.mod index 669a04d..6f902ec 100644 --- a/modules/mesh-delivery-forge-bridge/go.mod +++ b/modules/mesh-delivery-forge-bridge/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-forge-bridge/go.sum b/modules/mesh-delivery-forge-bridge/go.sum index db1ea18..bdc8b83 100644 --- a/modules/mesh-delivery-forge-bridge/go.sum +++ b/modules/mesh-delivery-forge-bridge/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 0e32028..a4a515f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -240,21 +240,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - candidates, err := essentialBucket(ctx, js, delivery.CandidatesBucket, "write a candidate's fingerprints") + candidates, err := process.EssentialBucket(ctx, js, "indexer", delivery.CandidatesBucket, "write a candidate's fingerprints") if err != nil { return err } - modules, err := essentialBucket(ctx, js, delivery.ModulesBucket, "write a module's fingerprint") + modules, err := process.EssentialBucket(ctx, js, "indexer", delivery.ModulesBucket, "write a module's fingerprint") if err != nil { return err } - manifests, err := essentialBucket(ctx, js, delivery.ManifestsBucket, "keep a module's manifest") + manifests, err := process.EssentialBucket(ctx, js, "indexer", delivery.ManifestsBucket, "keep a module's manifest") if err != nil { return err } // The controller's build records only name the directory of a module whose key names none (fillPaths): without them // such a module waits for its path, every other is indexed (issue 1945). - builds := optionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf) + builds := process.OptionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf) known := index.NewBuilds() // The controller's build records name the directory of a module whose key names none (fillPaths); no build of any // module moves a fingerprint since no fingerprint takes the builder (ADR 0530 §1). @@ -268,7 +268,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro errs := make(chan error, 1) if builds != nil { if w, err := settle(ctx, builds, each); err != nil { - logger.Printf("the bucket %s cannot be watched (%v): a module whose key names no directory is not given one until the indexer starts again", delivery.BuildsBucket, err) + process.NotWatched(logger.Printf, delivery.BuildsBucket, err, "a module whose key names no directory is not given one until the indexer starts again") } else { go func() { errs <- follow(ctx, w, each) }() } @@ -290,14 +290,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does // not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate. sourceErrs := make(chan error, 1) - if source := optionalBucket(ctx, js, delivery.SourceBucket, "no source revision is read until the indexer starts again", logger.Printf); source != nil { + if source := process.OptionalBucket(ctx, js, delivery.SourceBucket, "no source revision is read until the indexer starts again", logger.Printf); source != nil { each := func(e jetstream.KeyValueEntry) { if opOf(e) { applySource(ctx, ix, e.Key(), e.Revision(), e.Value(), logger) } } if sw, err := settle(ctx, source, each); err != nil { - logger.Printf("the bucket %s cannot be watched (%v): no source revision is read until the indexer starts again", delivery.SourceBucket, err) + process.NotWatched(logger.Printf, delivery.SourceBucket, err, "no source revision is read until the indexer starts again") } else { go func() { sourceErrs <- follow(ctx, sw, each) }() } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/optional.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/optional.go deleted file mode 100644 index 1f1fb5c..0000000 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/optional.go +++ /dev/null @@ -1,41 +0,0 @@ -package main - -import ( - "context" - "fmt" - - "github.com/nats-io/nats.go/jetstream" -) - -// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the builder of -// mesh-delivery #48 exited on a refused read of the gate's source, and its new reasks bucket would have done the same). A -// grant reaches the bus with a delivery, so a bucket the bus refuses, or times out on, at start is often only not granted -// yet: exiting for it stops the part until a person's hand. A bucket the indexer cannot do its job at all without still stops -// the start, naming the grant to look at; every other is said and left. (The same two functions are in each delivery part; -// they belong in mesh-sdk.) - -// essentialBucket opens a bucket the indexer cannot work without; its error names the bucket and the grant. -func essentialBucket(ctx context.Context, js bucketOpener, name, does string) (jetstream.KeyValue, error) { - kv, err := js.KeyValue(ctx, name) - if err != nil { - return nil, essentialErr(name, does, err) - } - return kv, nil -} - -// essentialErr is the error of a bucket the part cannot work without: it names the bucket and the grant. -func essentialErr(name, does string, err error) error { - return fmt.Errorf("the bucket %s cannot be read (%v): the indexer cannot %s without it, so it stops; where the bus "+ - "refused it or did not answer, this module's grant of %s (its manifest's reads or state, issued with its credential) "+ - "has not reached the bus yet", name, err, does, name) -} - -// optionalBucket opens a bucket the indexer works without: any error is said with what is not heard, and nil answered. -func optionalBucket(ctx context.Context, js bucketOpener, name, without string, logf func(string, ...any)) jetstream.KeyValue { - kv, err := js.KeyValue(ctx, name) - if err != nil { - logf("the bucket %s cannot be read (%v): %s", name, err, without) - return nil - } - return kv -} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go index 748182b..3b95a25 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go @@ -3,6 +3,7 @@ package main import ( "context" "errors" + "git.novox.be/novox/mesh-sdk/go/process" "log" "strings" "sync" @@ -60,7 +61,7 @@ func (w *previewWatch) ensure() error { } watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.ix.PutPreview(e.Key(), e.Value(), !opOf(e)) }) if err != nil { - w.logger.Printf("the bucket %s cannot be watched (%v): it is watched when the checker says it wrote a preview", delivery.PreviewsBucket, err) + process.NotWatched(w.logger.Printf, delivery.PreviewsBucket, err, "it is watched when the checker says it wrote a preview") return nil } w.watched = true diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go index 3e4f0cf..92aafa4 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go @@ -3,12 +3,15 @@ package main import ( "context" "errors" + "fmt" "log" "maps" "slices" "strings" "sync" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -78,7 +81,7 @@ func (w *toolchainWatch) ensure(module string) error { } watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.sink.put(module, e.Key(), e.Value(), !opOf(e)) }) if err != nil { - w.logger.Printf("the bucket %s cannot be watched (%v): %s's toolchains are not heard until its next report", name, err, module) + process.NotWatched(w.logger.Printf, name, err, fmt.Sprintf("%s's toolchains are not heard until its next report", module)) return nil } w.watched[module] = true diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 0aaff15..e8728ce 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index 2a3e9f3..8e94e90 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index 8ed2b7b..d73893d 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -22,6 +22,8 @@ import ( "strings" "syscall" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -120,12 +122,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } if err != nil { // Essential: the gate's asks are what the teller tells (issue 1945: the error names the grant). - return essentialErr(delivery.AsksBucket, "tell any of the gate's asks", err) + return process.EssentialErr("teller", delivery.AsksBucket, "tell any of the gate's asks", err) } // The answers it writes and what it asked: its own work, so each stops the start, naming its grant. var kvs [2]jetstream.KeyValue for i, name := range []string{delivery.AnswersBucket, askedBucket} { - if kvs[i], err = essentialBucket(ctx, js, name, "write an answer or remember what it asked"); err != nil { + if kvs[i], err = process.EssentialBucket(ctx, js, "teller", name, "write an answer or remember what it asked"); err != nil { return err } } @@ -142,7 +144,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The view's answers to the assigner's asks (issue 1297 B10): both buckets are the controller's to create, and a bucket not // there is said and leaves the teller without that path rather than without the gate's asks (a wait for it needs a timer). failed := make(chan error, 2) - assignerKV, viewKV := optionalBucket(ctx, js, delivery.AssignerAsksBucket, teller, logger.Printf), optionalBucket(ctx, js, delivery.ViewAnswersBucket, teller, logger.Printf) + assignerKV, viewKV := process.OptionalBucket(ctx, js, delivery.AssignerAsksBucket, teller, logger.Printf), process.OptionalBucket(ctx, js, delivery.ViewAnswersBucket, teller, logger.Printf) // The assigner's asks are raised on the operator channel whether or not the view's answers can be read (issue 1906). if viewKV != nil { t.View = kvGate{ctx: ctx, kv: viewKV} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/optional.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/optional.go deleted file mode 100644 index 9f8f3bc..0000000 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/optional.go +++ /dev/null @@ -1,46 +0,0 @@ -package main - -import ( - "context" - "fmt" - - "github.com/nats-io/nats.go/jetstream" -) - -// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the builder of -// mesh-delivery #48 exited on a refused read of the gate's source, and its new reasks bucket would have done the same). A -// grant reaches the bus with a delivery, so a bucket the bus refuses, or times out on, at start is often only not granted -// yet: exiting for it stops the part until a person's hand. A bucket the teller cannot do its job at all without still stops -// the start, naming the grant to look at; every other is said and left. (The same two functions are in each delivery part; -// they belong in mesh-sdk.) - -// bucketOpener is the part of the bus a start opens buckets with. -type bucketOpener interface { - KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) -} - -// essentialBucket opens a bucket the teller cannot work without; its error names the bucket and the grant. -func essentialBucket(ctx context.Context, js bucketOpener, name, does string) (jetstream.KeyValue, error) { - kv, err := js.KeyValue(ctx, name) - if err != nil { - return nil, essentialErr(name, does, err) - } - return kv, nil -} - -// essentialErr is the error of a bucket the part cannot work without: it names the bucket and the grant. -func essentialErr(name, does string, err error) error { - return fmt.Errorf("the bucket %s cannot be read (%v): the teller cannot %s without it, so it stops; where the bus "+ - "refused it or did not answer, this module's grant of %s (its manifest's reads or state, issued with its credential) "+ - "has not reached the bus yet", name, err, does, name) -} - -// optionalBucket opens a bucket the teller works without: any error is said with what is not heard, and nil answered. -func optionalBucket(ctx context.Context, js bucketOpener, name, without string, logf func(string, ...any)) jetstream.KeyValue { - kv, err := js.KeyValue(ctx, name) - if err != nil { - logf("the bucket %s cannot be read (%v): %s", name, err, without) - return nil - } - return kv -} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go index ce2cf62..ae05ae0 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go @@ -5,11 +5,13 @@ import ( "encoding/json" "errors" "fmt" - "github.com/nats-io/nats.go/jetstream" "strings" "testing" "time" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asker" "git.novox.be/novox/mesh-sdk/go/asks" @@ -242,10 +244,10 @@ func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1 t.Run("1945: a refused bucket is said, or stops the start naming its grant", func(t *testing.T) { refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_x") var said []string - if kv := optionalBucket(context.Background(), refusingBus{refusal}, "x", "x is not heard", func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }); kv != nil || len(said) != 1 || !strings.Contains(said[0], "permissions violation") { + if kv := process.OptionalBucket(context.Background(), refusingBus{refusal}, "x", "x is not heard", func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }); kv != nil || len(said) != 1 || !strings.Contains(said[0], "permissions violation") { t.Errorf("an optional bucket refused: %v, said %v", kv, said) } - if _, err := essentialBucket(context.Background(), refusingBus{refusal}, "x", "work"); err == nil || !strings.Contains(err.Error(), "grant of x") { + if _, err := process.EssentialBucket(context.Background(), refusingBus{refusal}, "teller", "x", "work"); err == nil || !strings.Contains(err.Error(), "grant of x") { t.Errorf("an essential bucket refused: %v, want its grant named", err) } }) diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index db3470d..0937a16 100644 --- a/modules/mesh-delivery-teller/go.mod +++ b/modules/mesh-delivery-teller/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-teller go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-teller/go.sum b/modules/mesh-delivery-teller/go.sum index 97c314d..3baa2a5 100644 --- a/modules/mesh-delivery-teller/go.sum +++ b/modules/mesh-delivery-teller/go.sum @@ -6,6 +6,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8 h1:vhYk/iv4 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5 h1:gE80bR0tPME7MTMBLcdP62v+F/9soVXmdcBYNMOye44= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index 3ffe7d1..d0675db 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -9,8 +9,9 @@ import ( "os" "strings" - stdio "git.novox.be/novox/mesh-sdk/go" "mesh-delivery-view/internal/view" + + stdio "git.novox.be/novox/mesh-sdk/go" ) // Tools are the read tools over a modules state. diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index e59d27f..57994b4 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -22,6 +22,8 @@ import ( "syscall" "time" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -230,12 +232,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro name string into *jetstream.KeyValue }{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}} { - if *b.into, err = essentialBucket(ctx, js, b.name, "say who is online or what stopped moving"); err != nil { + if *b.into, err = process.EssentialBucket(ctx, js, "watchman", b.name, "say who is online or what stopped moving"); err != nil { return err } } - kvs.wants = optionalBucket(ctx, js, delivery.AssignerWantsBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf) - kvs.haves = optionalBucket(ctx, js, delivery.HaveBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf) + kvs.wants = process.OptionalBucket(ctx, js, delivery.AssignerWantsBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf) + kvs.haves = process.OptionalBucket(ctx, js, delivery.HaveBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf) if kvs.wants == nil || kvs.haves == nil { kvs.wants, kvs.haves = nil, nil // a want is a pulse only against the have that ends it } @@ -286,9 +288,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro var haveSeen, wantSeen jetstream.KeyWatcher if kvs.haves != nil { if haveSeen, err = settle(ctx, kvs.haves, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }); err != nil { - logger.Printf("the bucket %s cannot be watched (%v): moves are not watched until the watchman starts again", delivery.HaveBucket, err) + process.NotWatched(logger.Printf, delivery.HaveBucket, err, "moves are not watched until the watchman starts again") } else if wantSeen, err = settle(ctx, kvs.wants, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) }); err != nil { - logger.Printf("the bucket %s cannot be watched (%v): moves are not watched until the watchman starts again", delivery.AssignerWantsBucket, err) + process.NotWatched(logger.Printf, delivery.AssignerWantsBucket, err, "moves are not watched until the watchman starts again") haveSeen.Stop() haveSeen = nil } diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index bee4b2e..58fe0a7 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -10,6 +10,8 @@ import ( "strings" "testing" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" ) @@ -34,10 +36,10 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 t.Run("1945: a refused bucket is said, or stops the start naming its grant", func(t *testing.T) { refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_x") var said []string - if kv := optionalBucket(context.Background(), refusingBus{refusal}, "x", "x is not heard", func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }); kv != nil || len(said) != 1 || !strings.Contains(said[0], "permissions violation") { + if kv := process.OptionalBucket(context.Background(), refusingBus{refusal}, "x", "x is not heard", func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }); kv != nil || len(said) != 1 || !strings.Contains(said[0], "permissions violation") { t.Errorf("an optional bucket refused: %v, said %v", kv, said) } - if _, err := essentialBucket(context.Background(), refusingBus{refusal}, "x", "work"); err == nil || !strings.Contains(err.Error(), "grant of x") { + if _, err := process.EssentialBucket(context.Background(), refusingBus{refusal}, "watchman", "x", "work"); err == nil || !strings.Contains(err.Error(), "grant of x") { t.Errorf("an essential bucket refused: %v, want its grant named", err) } }) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/optional.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/optional.go deleted file mode 100644 index 4d3e2dc..0000000 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/optional.go +++ /dev/null @@ -1,46 +0,0 @@ -package main - -import ( - "context" - "fmt" - - "github.com/nats-io/nats.go/jetstream" -) - -// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the builder of -// mesh-delivery #48 exited on a refused read of the gate's source, and its new reasks bucket would have done the same). A -// grant reaches the bus with a delivery, so a bucket the bus refuses, or times out on, at start is often only not granted -// yet: exiting for it stops the part until a person's hand. A bucket the watchman cannot do its job at all without still stops -// the start, naming the grant to look at; every other is said and left. (The same two functions are in each delivery part; -// they belong in mesh-sdk.) - -// bucketOpener is the part of the bus a start opens buckets with. -type bucketOpener interface { - KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) -} - -// essentialBucket opens a bucket the watchman cannot work without; its error names the bucket and the grant. -func essentialBucket(ctx context.Context, js bucketOpener, name, does string) (jetstream.KeyValue, error) { - kv, err := js.KeyValue(ctx, name) - if err != nil { - return nil, essentialErr(name, does, err) - } - return kv, nil -} - -// essentialErr is the error of a bucket the part cannot work without: it names the bucket and the grant. -func essentialErr(name, does string, err error) error { - return fmt.Errorf("the bucket %s cannot be read (%v): the watchman cannot %s without it, so it stops; where the bus "+ - "refused it or did not answer, this module's grant of %s (its manifest's reads or state, issued with its credential) "+ - "has not reached the bus yet", name, err, does, name) -} - -// optionalBucket opens a bucket the watchman works without: any error is said with what is not heard, and nil answered. -func optionalBucket(ctx context.Context, js bucketOpener, name, without string, logf func(string, ...any)) jetstream.KeyValue { - kv, err := js.KeyValue(ctx, name) - if err != nil { - logf("the bucket %s cannot be read (%v): %s", name, err, without) - return nil - } - return kv -} diff --git a/modules/mesh-delivery-watchman/go.mod b/modules/mesh-delivery-watchman/go.mod index d343e0e..b154f77 100644 --- a/modules/mesh-delivery-watchman/go.mod +++ b/modules/mesh-delivery-watchman/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-watchman go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-watchman/go.sum b/modules/mesh-delivery-watchman/go.sum index bfba8a6..3c78612 100644 --- a/modules/mesh-delivery-watchman/go.sum +++ b/modules/mesh-delivery-watchman/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c h1:pY2YpHSSsfDPBrOEiWTNSIPn0zdnAayRmtmmyQ3hCHY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E= github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=