Delivery parts: take the essential and optional bucket opens from mesh-sdk (issue 1945)
mesh/delivery merged: source.main moved to novox/mesh-delivery@db53ec975f01
mesh/delivery-group group fix/1945-ask-again-and-why merged: every member is merged
mesh/check pass 4628ba26b41a on 73be3df28cd4 | unit 167 run, 443 kept | integration 177 run, 509 kept | composition passed on 3 machines

The open helpers were copied into seven parts; code shared between
modules lives in mesh-sdk, so they are mesh-sdk process's now and the
parts are pinned at fix/1945-optional-buckets (e0d1de04), moved to its
main once it merges, one delivery group with mesh-sdk.
This commit is contained in:
2026-10-25 22:30:22 +01:00
parent 20027e53ff
commit 4628ba26b4
38 changed files with 102 additions and 333 deletions
@@ -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
}
@@ -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)
}
+1 -1
View File
@@ -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
)
+2
View File
@@ -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=
@@ -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) }()
}
@@ -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.
@@ -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)
}
@@ -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
+1 -1
View File
@@ -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
)
+2
View File
@@ -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=
@@ -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) {
@@ -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
}
@@ -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)
}
@@ -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
+1 -1
View File
@@ -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
)
+2
View File
@@ -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=
@@ -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
}
@@ -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)
}
})
@@ -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
}
+1 -1
View File
@@ -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
)
@@ -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=
@@ -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) }()
}
@@ -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
}
@@ -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
@@ -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
+1 -1
View File
@@ -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
)
+2
View File
@@ -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=
@@ -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}
@@ -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
}
@@ -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)
}
})
+1 -1
View File
@@ -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
)
+2
View File
@@ -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=
@@ -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.
@@ -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
}
@@ -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)
}
})
@@ -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
}
+1 -1
View File
@@ -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
)
+2
View File
@@ -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=