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 e84e562..6fd8895 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -85,10 +85,12 @@ 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, since a pass -// without it would write wants from half the mesh. -func open(ctx context.Context, js jetstream.JetStream) (buckets, error) { +// open opens every bucket the assigner reads and writes; a bucket it cannot open stops the start, named with its grant, +// since a pass without it would write wants from half the mesh (issue 1945). The Go build module's steps are the one it +// works without: they are watched by each compose ask itself, so one that cannot be opened is said and left. +func open(ctx context.Context, js bucketOpener, logf func(string, ...any)) (buckets, error) { out := buckets{} + steps := delivery.StepsBucket(goBuildModule) for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) { kv, err := js.KeyValue(ctx, name) if name == delivery.AssignerPutsBucket && errors.Is(err, jetstream.ErrBucketNotFound) { @@ -96,8 +98,12 @@ func open(ctx context.Context, js jetstream.JetStream) (buckets, error) { // for want of it. The verb's first put is refused by the runtime until then. continue } + if err != nil && name == steps { + logf("the bucket %s cannot be read (%v): a compose ask's step is read by the ask's own watch, which says so if it cannot be", name, err) + continue + } if err != nil { - return nil, fmt.Errorf("the bucket %s cannot be opened: %w", name, err) + return nil, essentialErr(name, "write a want from the whole mesh", err) } out[name] = kv } @@ -134,7 +140,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - kvs, err := open(ctx, js) + kvs, err := open(ctx, js, logger.Printf) if err != nil { return err } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index f52e1fb..1d5da15 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -2,6 +2,8 @@ package main import ( "context" + "errors" + "fmt" "io" "os" "path/filepath" @@ -10,7 +12,10 @@ import ( "strings" "testing" + "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" ) // 1297/B2a: the buckets the program opens are exactly the manifest's reads and states, as the broker names them @@ -48,6 +53,34 @@ func TestOpensWhatItsManifestNames(t *testing.T) { if got, want := slices.Sorted(slices.Values(kept)), slices.Sorted(slices.Values(ownBuckets)); !reflect.DeepEqual(got, want) { t.Errorf("the program announces %v, keeps %v", got, want) } + // 1945: a bucket the bus refuses stops the start naming its grant, but for the Go build module's steps, which are said. + refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_x") + for _, c := range []struct { + refused string + stops bool + }{{delivery.StepsBucket(goBuildModule), false}, {delivery.ModulesBucket, true}, {delivery.AssignerWantsBucket, true}} { + var said []string + kvs, err := open(context.Background(), refusingOne{name: c.refused, err: refusal}, func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }) + switch { + case c.stops && (err == nil || !strings.Contains(err.Error(), "grant of "+c.refused)): + t.Errorf("%s refused: %v, want the start stopped naming its grant", c.refused, err) + case !c.stops && (err != nil || len(said) != 1 || len(kvs) != len(readBuckets)+len(ownBuckets)-1): + t.Errorf("%s refused: %v, said %v, %d buckets; want it said and the rest opened", c.refused, err, said, len(kvs)) + } + } +} + +// refusingOne is a bus that refuses one bucket with err and opens every other. +type refusingOne struct { + name string + err error +} + +func (r refusingOne) KeyValue(_ context.Context, name string) (jetstream.KeyValue, error) { + if name == r.name { + return nil, r.err + } + return nil, nil } // 1297/B2a: without a credential the program does not start, and says why. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/optional.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/optional.go new file mode 100644 index 0000000..0267144 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/optional.go @@ -0,0 +1,27 @@ +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-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index 852cb29..acc2bfe 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -31,7 +31,8 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, n {Name: build.AskAgainTool, Description: "Ask a waiting build key's asks again now, under new ids the key then names, as the builder does by itself for an ask its build " + "seat's stream gave up on. The builder answers in the reasks state, which " + build.StuckTool + " shows beside the key; a key no ask of " + - "which waits is answered so and nothing is asked.", + "which waits is answered so and nothing is asked. An ask whose step already runs on a node is asked again too: it is then " + + "built twice, the first run's work wasted and its end read by nobody, never a wrong key; builder_stuck shows whether a step runs.", Input: map[string]any{"type": "object", "required": []string{"module", "fp", "why"}, "properties": map[string]any{ "module": map[string]any{"type": "string", "description": "the module's name"}, "fp": map[string]any{"type": "string", "description": "its new-form fingerprint, src2:, as builder_stuck shows it"}, 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 3d867ce..d6b2e9f 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -124,29 +124,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - bucket := func(name string) (jetstream.KeyValue, error) { - kv, err := js.KeyValue(ctx, name) - if err != nil { - return nil, fmt.Errorf("the bucket %s: %w", name, err) - } - return kv, nil - } - builds, err := bucket(delivery.BuilderBuildsBucket) + builds, err := essentialBucket(ctx, js, delivery.BuilderBuildsBucket, "claim or write a build key") if err != nil { return err } - modules, err := bucket(delivery.ModulesBucket) + modules, err := essentialBucket(ctx, js, delivery.ModulesBucket, "know any fingerprint to build") if err != nil { return err } - manifests, err := bucket(delivery.ManifestsBucket) - if err != nil { - return err - } - candidates, err := bucket(delivery.CandidatesBucket) + manifests, err := essentialBucket(ctx, js, 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) bound := defaultStepBound if v := strings.TrimSpace(getenv("MESH_BUILDER_STEP_BOUND")); v != "" { if bound, err = time.ParseDuration(v); err != nil { @@ -181,26 +172,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Stored: storedVia(calls.Call), } // A candidate's preview is merged onto the trunk head the indexer previewed it onto: source.main at the candidate's base - // revision (the gate's source state). Where the bucket is not there yet no candidate's fingerprint is built, and said. - // The source's revisions are watched with their history (a direct get of one revision is not what a read grants): a - // revision heard after the candidate that names it builds that candidate then. - trunks := &sourceHeads{} - var sw jetstream.KeyWatcher - switch kv, err := js.KeyValue(ctx, delivery.SourceBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: no candidate's fingerprint is built until the builder starts again", delivery.SourceBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err) - default: - w, err := kv.Watch(ctx, delivery.SourceMainKey, jetstream.IncludeHistory()) - if err != nil { - return fmt.Errorf("the bucket %s cannot be watched: %w", delivery.SourceBucket, err) - } - if sw, err = settled(ctx, w, func(e jetstream.KeyValueEntry) { trunks.put(e) }); err != nil { - return err - } - b.Trunk = trunks.head - } + // revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision + // is not what a read grants): a revision heard after the candidate that names it builds that candidate then. Where it + // cannot be read the trunk's keys are built and the candidates wait (issue 1945). + errs := make(chan error, 4) + trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: b.TrunkMoved, errs: errs, logf: logger.Printf} + trunks.start() + b.Trunk = trunks.head b.Said = func(e build.AskedAgain) { if err := emitAskedAgain(context.WithoutCancel(ctx), js, cred, e); err != nil { logger.Printf("%v", err) @@ -210,11 +188,8 @@ 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, err := js.KeyValue(ctx, instancesBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: the controller is not told that this builder writes references until it starts again", instancesBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", instancesBucket, err) + switch kv := 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) b.Answered = says.Answered @@ -294,38 +269,26 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro b.ForgetCandidate(e.Key()) } } - cw, err := settle(ctx, candidates, eachCandidate) - if err != nil { - return err + 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) + } 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. A bucket not there yet (the manifest's state not made) is said. - errs := make(chan error, 4) - switch kv, err := js.KeyValue(ctx, build.ReAsksBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: builder_ask_again is not heard until the builder starts again", build.ReAsksBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", build.ReAsksBucket, err) - default: + // 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 { eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, kv, e, logger.Printf) } - rw, err := settle(ctx, kv, eachReAsk) - if err != nil { - return err + 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) + } else { + go func() { errs <- follow(ctx, rw, eachReAsk) }() } - go func() { errs <- follow(ctx, rw, eachReAsk) }() } logger.Printf("started on %s", cred.Node) go func() { errs <- follow(ctx, w, each) }() - go func() { errs <- follow(ctx, cw, eachCandidate) }() - if sw != nil { - go func() { - errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { - if trunks.put(e) { - b.TrunkMoved(ctx) - } - }) - }() - } select { case <-ctx.Done(): b.Wait() diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/optional.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/optional.go new file mode 100644 index 0000000..bd2542e --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/optional.go @@ -0,0 +1,104 @@ +package main + +import ( + "context" + "fmt" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "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 +// 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. +type trunkSource struct { + ctx context.Context + js bucketOpener + heads *sourceHeads + moved func(context.Context) + errs chan<- error + logf func(string, ...any) + + mu sync.Mutex + opened bool +} + +// ensure opens and settles the source's watch once, and follows it; an error leaves it closed for the next ask. +func (t *trunkSource) ensure() error { + t.mu.Lock() + defer t.mu.Unlock() + if t.opened { + return nil + } + kv, err := t.js.KeyValue(t.ctx, delivery.SourceBucket) + if err != nil { + return fmt.Errorf("the bucket %s cannot be read (%v)", delivery.SourceBucket, err) + } + w, err := kv.Watch(t.ctx, delivery.SourceMainKey, jetstream.IncludeHistory()) + if err != nil { + return fmt.Errorf("the bucket %s cannot be watched (%v)", delivery.SourceBucket, err) + } + sw, err := settled(t.ctx, w, func(e jetstream.KeyValueEntry) { t.heads.put(e) }) + if err != nil { + return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err) + } + t.opened = true + go func() { + t.errs <- follow(t.ctx, sw, func(e jetstream.KeyValueEntry) { + if t.heads.put(e) && t.moved != nil { + t.moved(t.ctx) + } + }) + }() + return nil +} + +// start opens the source at start; one that cannot be is said, and the builder goes on. +func (t *trunkSource) start() { + if err := t.ensure(); err != nil { + t.logf("%v: the trunk's keys are built, and a candidate's preview waits until the source can be read, tried again at its next candidate", err) + } +} + +// head is Builder.Trunk: the head of repository at source revision rev, opening the source first if it is not yet. +func (t *trunkSource) head(ctx context.Context, rev uint64, repository string) (string, error) { + if err := t.ensure(); err != nil { + return "", err + } + return t.heads.head(ctx, rev, repository) +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go index 3d6decf..9a06cec 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go @@ -193,7 +193,9 @@ func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) } w, err := settle(ctx, kv, each) if err != nil { - return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err) + // Not fatal either (issue 1945): a watch the bus refuses is the same missing grant as a refused read. + logger.Printf("the bucket %s cannot be watched, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err) + return errs, nil } if gate.Settled() { changed(ctx) 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 28fccc4..ae19091 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 @@ -3,11 +3,17 @@ package main // The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940). import ( + "bytes" + "context" + "errors" "fmt" + "log" "strings" "testing" "time" + "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" ) @@ -89,6 +95,61 @@ func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) { if n := len(set.Entries("typescript")); n != 2 { t.Fatalf("novox and shanks are left; got %d entries", n) } + + // 1945 (a part exits at start for a bucket it works without): a bus that refuses the assignments' watch, or every + // toolchains bucket, leaves the watches unstarted and the part running, said; a refused bucket it cannot work without + // is an error naming the bucket and its grant. + var logs bytes.Buffer + logger := log.New(&logs, "", 0) + refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-controller_assignments")} + if _, err := watchAssignments(context.Background(), refused, newToolchainGate(nil, set.Put, t.Logf), func(context.Context) {}, logger); err != nil { + t.Errorf("a refused watch of the assignments stopped the start: %v", err) + } + if err := newToolchainWatch(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, set, func(context.Context) {}, logger).start(); err != nil { + t.Errorf("a refused toolchains bucket stopped the start: %v", err) + } + 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 || + !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) + } + // The gate's source (the previews' trunk heads): refused at start, the builder goes on, and a candidate asking a trunk + // head is told why and opens it again then. + src := &trunkSource{ctx: context.Background(), js: refusingBus{refuseOpen: true, err: refused.err}, heads: &sourceHeads{}, errs: make(chan error, 1), logf: logger.Printf} + src.start() + if _, err := src.head(context.Background(), 7, "novox/mesh-catalog"); err == nil || !strings.Contains(err.Error(), delivery.SourceBucket) { + t.Errorf("a trunk head with the source refused: %v, want why", err) + } + if src.opened { + t.Error("a refused source counts as opened, so it is never opened again") + } +} + +// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err. +type refusingBus struct { + refuseOpen bool + err error +} + +func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { + if r.refuseOpen { + return nil, r.err + } + return refusedWatch{err: r.err}, nil +} + +type refusedWatch struct { + jetstream.KeyValue + err error +} + +func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return nil, r.err +} +func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return nil, r.err } // A build node's report counts from the first one it writes since its assignment: a newly assigned node that has not 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 69faff2..f574c2d 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go @@ -3,7 +3,6 @@ package main import ( "context" "errors" - "fmt" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" "log" "strings" @@ -62,11 +61,14 @@ func (w *toolchainWatch) ensure(module string) error { w.logger.Printf("the bucket %s does not exist yet: it is watched when %s reports a toolchain", name, module) return nil case err != nil: - return fmt.Errorf("the bucket %s: %w", name, err) + // Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the module's next report. + w.logger.Printf("the bucket %s cannot be read (%v): %s's toolchains are not heard until its next report", name, err, module) + return nil } watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) }) if err != nil { - return err + w.logger.Printf("the bucket %s cannot be watched (%v): %s's toolchains are not heard until its next report", name, err, module) + return nil } w.watched[module] = true go func() { 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 297cfce..7b97f0d 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -152,17 +152,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - bucket := func(name string) (jetstream.KeyValue, error) { - kv, err := js.KeyValue(ctx, name) - if err != nil { - return nil, fmt.Errorf("the bucket %s: %w", name, err) - } - return kv, nil - } + // Its own states, the candidates it checks, and the modules and manifests that say which build modules judge a repository: + // a check without one of them would be wrong or unwritten, so each stops the start, naming its grant (issue 1945). 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 = bucket(name); err != nil { + if buckets[name], err = essentialBucket(ctx, js, name, "judge or keep a check"); err != nil { return err } } @@ -212,12 +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 - switch kv, err := js.KeyValue(ctx, delivery.BuilderBuildsBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: the line says nothing of the builds until the checker starts again", delivery.BuilderBuildsBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", delivery.BuilderBuildsBucket, err) - default: + if kv := 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 @@ -290,47 +280,38 @@ 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. - switch kv, err := js.KeyValue(ctx, delivery.PresenceBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: no node counts as online until the checker starts again", delivery.PresenceBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", delivery.PresenceBucket, err) - default: - pw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { presence.Put(e.Key(), e.Value(), !opOf(e)) }) - if err != nil { - return err + if kv := 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) + } else { + go func() { + errs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) { + if presence.Put(e.Key(), e.Value(), !opOf(e)) { + ch.NodesChanged(ctx) // a node came online or went: a check whose forge was not heard may now be heard + } + }) + }() } - go func() { - errs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) { - if presence.Put(e.Key(), e.Value(), !opOf(e)) { - ch.NodesChanged(ctx) // a node came online or went: a check whose forge was not heard may now be heard - } - }) - }() } // 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. - switch kv, err := js.KeyValue(ctx, delivery.SourceBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: no trunk is known until the checker starts again", delivery.SourceBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err) - default: + if kv := 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 { - return err + 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) + } else { + go func() { + errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { + if e.Key() == delivery.SourceMainKey && trunks.Put(e.Value(), !opOf(e)) { + ch.TrunkMoved(ctx) // the trunk moved: its statuses go pending, and a verdict for the old trunk is judged again + } + }) + }() } - go func() { - errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { - if e.Key() == delivery.SourceMainKey && trunks.Put(e.Value(), !opOf(e)) { - ch.TrunkMoved(ctx) // the trunk moved: its statuses go pending, and a verdict for the old trunk is judged again - } - }) - }() } // The modules say which repository holds a TypeScript module, so the Node build module judges it too. mw, err := settle(ctx, buckets[delivery.ModulesBucket], func(e jetstream.KeyValueEntry) { repos.Put(ctx, e.Key(), e.Value(), !opOf(e)) }) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/optional.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/optional.go new file mode 100644 index 0000000..466cdd0 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/optional.go @@ -0,0 +1,41 @@ +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.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go index 3d6decf..9a06cec 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go @@ -193,7 +193,9 @@ func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) } w, err := settle(ctx, kv, each) if err != nil { - return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err) + // Not fatal either (issue 1945): a watch the bus refuses is the same missing grant as a refused read. + logger.Printf("the bucket %s cannot be watched, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err) + return errs, nil } if gate.Settled() { changed(ctx) 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 28fccc4..c759d3a 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 @@ -3,11 +3,17 @@ package main // The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940). import ( + "bytes" + "context" + "errors" "fmt" + "log" "strings" "testing" "time" + "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" ) @@ -89,6 +95,51 @@ func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) { if n := len(set.Entries("typescript")); n != 2 { t.Fatalf("novox and shanks are left; got %d entries", n) } + + // 1945 (a part exits at start for a bucket it works without): a bus that refuses the assignments' watch, or every + // toolchains bucket, leaves the watches unstarted and the part running, said; a refused bucket it cannot work without + // is an error naming the bucket and its grant. + var logs bytes.Buffer + logger := log.New(&logs, "", 0) + refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-controller_assignments")} + if _, err := watchAssignments(context.Background(), refused, newToolchainGate(nil, set.Put, t.Logf), func(context.Context) {}, logger); err != nil { + t.Errorf("a refused watch of the assignments stopped the start: %v", err) + } + if err := newToolchainWatch(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, set, func(context.Context) {}, logger).start(); err != nil { + t.Errorf("a refused toolchains bucket stopped the start: %v", err) + } + 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 || + !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) + } +} + +// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err. +type refusingBus struct { + refuseOpen bool + err error +} + +func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { + if r.refuseOpen { + return nil, r.err + } + return refusedWatch{err: r.err}, nil +} + +type refusedWatch struct { + jetstream.KeyValue + err error +} + +func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return nil, r.err +} +func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return nil, r.err } // A build node's report counts from the first one it writes since its assignment: a newly assigned node that has not 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 bc848ca..0fe2f41 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go @@ -3,7 +3,6 @@ package main import ( "context" "errors" - "fmt" "log" "strings" "sync" @@ -63,11 +62,14 @@ func (w *toolchainWatch) ensure(module string) error { w.logger.Printf("the bucket %s does not exist yet: it is watched when %s reports a toolchain", name, module) return nil case err != nil: - return fmt.Errorf("the bucket %s: %w", name, err) + // Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the module's next report. + w.logger.Printf("the bucket %s cannot be read (%v): %s's toolchains are not heard until its next report", name, err, module) + return nil } watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) }) if err != nil { - return err + w.logger.Printf("the bucket %s cannot be watched (%v): %s's toolchains are not heard until its next report", name, err, module) + return nil } w.watched[module] = true go func() { 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 a11a3b2..383bedc 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 @@ -188,18 +188,18 @@ 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 := js.KeyValue(ctx, headsBucket) + kv, err := essentialBucket(ctx, js, headsBucket, "keep the heads it wrote on the forge stream") if err != nil { - return fmt.Errorf("the bucket %s: %w", headsBucket, err) + return err } heads := kvHeads{kv} // Everything written on FORGE, an event's and a sync's, keeps the head in the bucket. 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 := js.KeyValue(ctx, performedBucket) + performedKV, err := essentialBucket(ctx, js, performedBucket, "perform an act of the gate only once") if err != nil { - return fmt.Errorf("the bucket %s: %w", performedBucket, err) + return err } perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger} watching := make(chan error, 3) @@ -207,16 +207,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro bucket string react func(ctx context.Context, key string, rev uint64, value []byte) }{{delivery.OwedBucket, perf.owed}, {delivery.MergesBucket, perf.merges}} { - kv, err := js.KeyValue(ctx, w.bucket) - if errors.Is(err, jetstream.ErrBucketNotFound) { - // The gate's buckets come with the gate; the bridge does not wait for them (no timer), and is started again - // by its delivery when the gate arrives. - logger.Printf("the bucket %s does not exist yet: nothing of it is performed until the bridge starts again", w.bucket) + // 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) + if kv == nil { continue } - if err != nil { - return fmt.Errorf("the bucket %s: %w", w.bucket, err) - } go func() { watching <- watch(ctx, kv, func(key string, rev uint64, value []byte) { hctx, cancel := context.WithTimeout(ctx, syncDeadline) @@ -231,7 +227,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go onReconnect(ctx, again, said, logger) cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module) if err != nil { - return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, eventsStream, err) + // Essential: the forge's events are what the bridge carries. + return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, eventsStream, err) } cc, err := cons.Consume(func(m jetstream.Msg) { hear(&turn, func() { 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 c4ab010..0f9cc32 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 @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "log" "slices" "strings" @@ -66,6 +67,16 @@ func body(m map[string]any) []byte { } func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit part; the bus is the integration test) + 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") { + 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") { + t.Errorf("an essential bucket refused: %v, want its grant named", err) + } + }) var logs bytes.Buffer m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{} handle(context.Background(), m, pub, nil, someFiles, log.New(&logs, "", 0)) @@ -230,3 +241,8 @@ func someFiles(_, _, path, _ string) ([]byte, error) { } return nil, bridge.ErrNotFound } + +// refusingBus refuses every bucket with err (issue 1945). +type refusingBus struct{ err error } + +func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.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 new file mode 100644 index 0000000..68df917 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/optional.go @@ -0,0 +1,46 @@ +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-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index fbf6ff3..0e32028 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -240,29 +240,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - bucket := func(name string) (jetstream.KeyValue, error) { - kv, err := js.KeyValue(ctx, name) - if err != nil { - return nil, fmt.Errorf("the bucket %s: %w", name, err) - } - return kv, nil - } - candidates, err := bucket(delivery.CandidatesBucket) + candidates, err := essentialBucket(ctx, js, delivery.CandidatesBucket, "write a candidate's fingerprints") if err != nil { return err } - modules, err := bucket(delivery.ModulesBucket) + modules, err := essentialBucket(ctx, js, delivery.ModulesBucket, "write a module's fingerprint") if err != nil { return err } - manifests, err := bucket(delivery.ManifestsBucket) - if err != nil { - return err - } - builds, err := bucket(delivery.BuildsBucket) + manifests, err := essentialBucket(ctx, js, 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) 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). @@ -273,12 +265,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro known.Delete(e.Key()) } } - w, err := settle(ctx, builds, each) - if err != nil { - return err - } errs := make(chan error, 1) - go func() { errs <- follow(ctx, w, each) }() + 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) + } else { + go func() { errs <- follow(ctx, w, each) }() + } + } // The indexer asks the forge for the objects of a manifest's named paths as its own account (ADR 0499 clause 3). account, err := process.ReadAccount(path) @@ -296,22 +290,17 @@ 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) - switch source, err := js.KeyValue(ctx, delivery.SourceBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: no source revision is read until the indexer starts again", delivery.SourceBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err) - default: + if source := 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) } } - sw, err := settle(ctx, source, each) - if err != nil { - return err + 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) + } else { + go func() { sourceErrs <- follow(ctx, sw, each) }() } - go func() { sourceErrs <- follow(ctx, sw, each) }() } // Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's // assignments are settled first, so the toolchains settle through the gate already knowing them. @@ -381,7 +370,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } cons, err := js.Consumer(ctx, forgeStream, cred.Node+"_"+cred.Module) if err != nil { - return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, forgeStream, err) + // Essential: the forge's merges are what the indexer indexes. + return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the indexer hears no merge without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, forgeStream, err) } cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, ix, logger) }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/optional.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/optional.go new file mode 100644 index 0000000..1f1fb5c --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/optional.go @@ -0,0 +1,41 @@ +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 c42221c..748182b 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go @@ -3,7 +3,6 @@ package main import ( "context" "errors" - "fmt" "log" "strings" "sync" @@ -55,11 +54,14 @@ func (w *previewWatch) ensure() error { w.logger.Printf("the bucket %s does not exist yet: it is watched when the checker says it wrote a preview", delivery.PreviewsBucket) return nil case err != nil: - return fmt.Errorf("the bucket %s: %w", delivery.PreviewsBucket, err) + // Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the checker's next preview. + w.logger.Printf("the bucket %s cannot be read (%v): it is watched when the checker says it wrote a preview", delivery.PreviewsBucket, err) + return nil } watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.ix.PutPreview(e.Key(), e.Value(), !opOf(e)) }) if err != nil { - return err + w.logger.Printf("the bucket %s cannot be watched (%v): it is watched when the checker says it wrote a preview", delivery.PreviewsBucket, err) + return nil } w.watched = true go func() { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go index 3d6decf..9a06cec 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go @@ -193,7 +193,9 @@ func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) } w, err := settle(ctx, kv, each) if err != nil { - return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err) + // Not fatal either (issue 1945): a watch the bus refuses is the same missing grant as a refused read. + logger.Printf("the bucket %s cannot be watched, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err) + return errs, nil } if gate.Settled() { changed(ctx) 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 30892ae..3e4f0cf 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go @@ -3,7 +3,6 @@ package main import ( "context" "errors" - "fmt" "log" "maps" "slices" @@ -73,11 +72,14 @@ func (w *toolchainWatch) ensure(module string) error { w.logger.Printf("the bucket %s does not exist yet: it is watched when %s reports a toolchain", name, module) return nil case err != nil: - return fmt.Errorf("the bucket %s: %w", name, err) + // Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the module's next report. + w.logger.Printf("the bucket %s cannot be read (%v): %s's toolchains are not heard until its next report", name, err, module) + return nil } watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.sink.put(module, e.Key(), e.Value(), !opOf(e)) }) if err != nil { - return err + w.logger.Printf("the bucket %s cannot be watched (%v): %s's toolchains are not heard until its next report", name, err, module) + return nil } w.watched[module] = true go func() { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go index e26bf76..7427efb 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go @@ -3,6 +3,7 @@ package main import ( "bytes" "context" + "errors" "log" "strings" "sync" @@ -21,6 +22,7 @@ type laterBuckets struct { mu sync.Mutex appears map[string]jetstream.KeyValue // buckets that exist now asked map[string]int + refuse error // answered for a bucket that does not exist, when set: a bus that refuses it (issue 1945) } func (l *laterBuckets) KeyValue(_ context.Context, name string) (jetstream.KeyValue, error) { @@ -30,6 +32,9 @@ func (l *laterBuckets) KeyValue(_ context.Context, name string) (jetstream.KeyVa if kv, ok := l.appears[name]; ok { return kv, nil } + if l.refuse != nil { + return nil, l.refuse + } return nil, jetstream.ErrBucketNotFound } @@ -100,6 +105,12 @@ func TestABuildModulesBucketThatAppearsLaterIsWatchedWhenItsEventArrives(t *test recompute: func(context.Context) error { mu.Lock(); recomputes++; mu.Unlock(); return nil }, } var logs bytes.Buffer + // 1945: a bus that refuses both buckets, or does not answer, leaves them unwatched, said, and the indexer running. + refusing := &laterBuckets{appears: map[string]jetstream.KeyValue{}, asked: map[string]int{}, refuse: errors.New("context deadline exceeded")} + if err := newToolchainWatch(ctx, refusing, sink, log.New(&logs, "", 0)).start(); err != nil || !strings.Contains(logs.String(), "cannot be read") { + t.Fatalf("a refused toolchains bucket: start %v, log %q; want it said and the start going on", err, logs.String()) + } + logs.Reset() w := newToolchainWatch(ctx, buckets, sink, log.New(&logs, "", 0)) if err := w.start(); err != nil { // neither bucket exists: nothing is watched, nothing is written, nobody waits t.Fatal(err) 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 f9b9142..8ed2b7b 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -119,12 +119,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil } if err != nil { - return fmt.Errorf("the bucket %s: %w", delivery.AsksBucket, err) + // 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) } + // 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 = js.KeyValue(ctx, name); err != nil { - return fmt.Errorf("the bucket %s: %w", name, err) + if kvs[i], err = essentialBucket(ctx, js, name, "write an answer or remember what it asked"); err != nil { + return err } } state := kvStore{ctx: ctx, kv: kvs[1]} @@ -140,7 +142,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, logger), optionalBucket(ctx, js, delivery.ViewAnswersBucket, logger) + assignerKV, viewKV := optionalBucket(ctx, js, delivery.AssignerAsksBucket, teller, logger.Printf), 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} @@ -172,15 +174,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } } -// optionalBucket is a bucket the teller reads when it exists, and nil, said once, when it does not. -func optionalBucket(ctx context.Context, js jetstream.JetStream, name string, logger *log.Logger) jetstream.KeyValue { - kv, err := js.KeyValue(ctx, name) - if err != nil { - logger.Printf("the bucket %s cannot be read (%v): the asks and answers it holds are not carried until the teller is started again with it", name, err) - return nil - } - return kv -} +// teller is what an optional bucket of the teller leaves out. +const teller = "the asks and answers it holds are not carried until the teller is started again with it" // askedBucket is the teller's state bucket asked, `_`: what it asked of the router and which condition it // raised (manifest state entry asked, novox/hq ADR 0437 §4). diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/optional.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/optional.go new file mode 100644 index 0000000..9f8f3bc --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/optional.go @@ -0,0 +1,46 @@ +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 8986842..ce2cf62 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 @@ -1,8 +1,11 @@ package main import ( + "context" "encoding/json" + "errors" "fmt" + "github.com/nats-io/nats.go/jetstream" "strings" "testing" "time" @@ -236,6 +239,16 @@ func (b *bed) askedOf(subject string) []published { } func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1343/AC1 + 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") { + 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") { + t.Errorf("an essential bucket refused: %v, want its grant named", err) + } + }) b := newBed(t) key, value, ga := gateAsk(t, twoHeads(), twoHeads()) b.gate[key] = value @@ -518,3 +531,8 @@ func TestAnAnswerToAnEarlierAskingDoesNotBlockTheNextAsking(t *testing.T) { // R t.Errorf("writes %d asks %d dropped %v", b.answers.writes, len(b.askedOf(asks.AskSubject(delivery.TellerModule))), b.dropped) } } + +// refusingBus refuses every bucket with err (issue 1945). +type refusingBus struct{ err error } + +func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err } 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 f619da4..e59d27f 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -222,23 +222,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - bucket := func(name string) (jetstream.KeyValue, error) { - kv, err := js.KeyValue(ctx, name) - if err != nil { - return nil, fmt.Errorf("the bucket %s: %w", name, err) - } - return kv, nil - } + // Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants + // and the node-engine's haves only make the pulses of a move: without one the watchman watches presence and the build + // seats' steps, and says that moves are not watched (issue 1945). var kvs struct{ presence, pulse, stuck, wants, haves jetstream.KeyValue } for _, b := range []struct { name string into *jetstream.KeyValue - }{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}, - {delivery.AssignerWantsBucket, &kvs.wants}, {delivery.HaveBucket, &kvs.haves}} { - if *b.into, err = bucket(b.name); err != nil { + }{{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 { 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) + 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 + } w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck}, time.Now) w.Logf(logger.Printf) @@ -283,13 +283,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // What the node agents run, then the assigner's wants (issue 1906, ADR 0539 §19): a want is a pulse owed by its node's agent // until the have names its fingerprint, so the haves are read first and a want already applied makes no pulse. - haveSeen, err := settle(ctx, kvs.haves, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }) - if err != nil { - return err - } - wantSeen, err := settle(ctx, kvs.wants, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) }) - if err != nil { - return err + 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) + } 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) + haveSeen.Stop() + haveSeen = nil + } } // The build seats' steps: a step key `.` written is the ask taken (issue 1887 A8b). A seat whose build module // keeps no steps bucket (yet) is said and left; the asks of its stream are mirrored all the same. @@ -304,7 +306,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro lang := lang sw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) if err != nil { - return err + logger.Printf("the steps of %s cannot be watched (%v): its asks are mirrored and never seen taken", module, err) + continue } taken[lang] = sw } @@ -348,12 +351,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) }) }() - go func() { - errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }) - }() - go func() { - errs <- follow(ctx, wantSeen, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) }) - }() + if haveSeen != nil && wantSeen != nil { + go func() { + errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }) + }() + go func() { + errs <- follow(ctx, wantSeen, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) }) + }() + } for lang, sw := range taken { lang, sw := lang, sw go func() { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) }() 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 6a171c2..bee4b2e 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 @@ -2,11 +2,15 @@ package main import ( "context" + "errors" + "fmt" "io" "os" "path/filepath" "strings" "testing" + + "github.com/nats-io/nats.go/jetstream" ) func TestCredentialIsTheBrokerFile(t *testing.T) { // 1135/AC6 @@ -27,8 +31,23 @@ func TestCredentialIsTheBrokerFile(t *testing.T) { // 1135/AC6 } 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") { + 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") { + t.Errorf("an essential bucket refused: %v, want its grant named", err) + } + }) err := run(context.Background(), func(string) string { return "" }, io.Discard) if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") { t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err) } } + +// refusingBus refuses every bucket with err (issue 1945). +type refusingBus struct{ err error } + +func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.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 new file mode 100644 index 0000000..4d3e2dc --- /dev/null +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/optional.go @@ -0,0 +1,46 @@ +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 +}