process: a part opens what it cannot work without as essential, the rest as optional (issue 1945)
mesh/delivery merged: source.main moved to novox/mesh-sdk@7ac4493ce9ab
mesh/delivery-group group fix/1945-ask-again-and-why merged: every member is merged
mesh/check pass e0d1de047585 on 3db051be57ca | unit 1642 run, 752 kept, 2 skipped | integration 1710 run, 803 kept, 2 skipped | over bound: unit, inte…

The delivery builder exited twice on 2026-10-25 for a bucket the bus
refused at start, though it could build without it: a grant reaches the
bus with a delivery, so a refusal at start is often only a grant not
there yet. EssentialBucket stops a start naming the bucket and its grant;
OptionalBucket and NotWatched say what is not heard and go on. Here so
the seven delivery parts take one copy, not seven.
This commit is contained in:
2026-10-25 22:12:20 +01:00
parent db26f7e8b6
commit e0d1de0475
2 changed files with 87 additions and 0 deletions
+32
View File
@@ -1,11 +1,16 @@
package process package process
import ( import (
"context"
"encoding/json" "encoding/json"
"errors"
"fmt"
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
"testing" "testing"
"github.com/nats-io/nats.go/jetstream"
) )
// 1171/AC2: given an account file broker-<process id> in the controller's form {url, fingerprint, node, // 1171/AC2: given an account file broker-<process id> in the controller's form {url, fingerprint, node,
@@ -46,6 +51,28 @@ func TestReadAccountToleratesNoClaimsAtAll(t *testing.T) {
// 1171/AC2: a file missing url, user or password is refused, naming the field; the password never appears // 1171/AC2: a file missing url, user or password is refused, naming the field; the password never appears
// in the error even when it is the field present and another is missing. // in the error even when it is the field present and another is missing.
func TestReadAccountRefusesAFileMissingAField(t *testing.T) { func TestReadAccountRefusesAFileMissingAField(t *testing.T) {
t.Run("1945: a refused bucket stops a start only when the part cannot work without it", func(t *testing.T) {
refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_mesh-delivery-gate_source")
for _, c := range []struct {
name string
err error
}{{"refused", refusal}, {"not answered", context.DeadlineExceeded}} {
var said []string
logf := func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
if kv := OptionalBucket(context.Background(), refusingBus{c.err}, "b", "b is not heard", logf); kv != nil {
t.Errorf("%s: an optional bucket answered %v", c.name, kv)
}
NotWatched(logf, "b", c.err, "b is not heard")
if len(said) != 2 || !strings.Contains(said[0], c.err.Error()) || !strings.Contains(said[1], "cannot be watched") {
t.Errorf("%s: said %q", c.name, said)
}
if _, err := EssentialBucket(context.Background(), refusingBus{c.err}, "builder", "b", "build"); err == nil ||
!strings.Contains(err.Error(), "grant of b") || !strings.Contains(err.Error(), "the builder cannot build") {
t.Errorf("%s: an essential bucket: %v", c.name, err)
}
}
})
for _, tc := range []struct { for _, tc := range []struct {
field string field string
values map[string]any values map[string]any
@@ -82,3 +109,8 @@ func writeAccount(t *testing.T, fields map[string]any) string {
} }
return path return path
} }
// refusingBus refuses every bucket with err.
type refusingBus struct{ err error }
func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }
+55
View File
@@ -0,0 +1,55 @@
package process
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 delivery
// builder exited on a refused read of the gate's source and crash-looped). A grant reaches the bus with a delivery, so a
// bucket, a watch or a consumer 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 part opens what it cannot do its job at all without with EssentialBucket, which
// stops its start with an error naming the bucket and its grant, and everything else with OptionalBucket, which says what
// is not heard and goes on; a watch the bus refuses on an optional bucket is said with NotWatched. Here, beside State,
// because this is where a process opens what it keeps and reads on the bus; the parts that open buckets with their own
// JetStream (the delivery parts) take the same three.
// BucketOpener is the part of the bus a start opens buckets with: a jetstream.JetStream, or a test's fake.
type BucketOpener interface {
KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error)
}
// EssentialBucket opens a bucket part cannot work without (it cannot `does` without it); its error is EssentialErr's.
func EssentialBucket(ctx context.Context, js BucketOpener, part, name, does string) (jetstream.KeyValue, error) {
kv, err := js.KeyValue(ctx, name)
if err != nil {
return nil, EssentialErr(part, name, does, err)
}
return kv, nil
}
// EssentialErr is the error of a bucket, consumer or stream named name that part cannot work without: it says what part
// cannot do, that it stops, and the grant to look at.
func EssentialErr(part, name, does string, err error) error {
return fmt.Errorf("%s cannot be read (%v): the %s 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, state or consumes, issued with its credential) has not "+
"reached the bus yet", name, err, part, does, name)
}
// OptionalBucket opens a bucket the part works without: any error is said with what is not heard (without), 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
}
// NotWatched says that an optional bucket opened but its watch was refused or not answered: what is not heard.
func NotWatched(logf func(string, ...any), name string, err error, without string) {
logf("the bucket %s cannot be watched (%v): %s", name, err, without)
}