diff --git a/go/process/account_test.go b/go/process/account_test.go index 6e57067..c5f9467 100644 --- a/go/process/account_test.go +++ b/go/process/account_test.go @@ -1,11 +1,16 @@ package process import ( + "context" "encoding/json" + "errors" + "fmt" "os" "path/filepath" "strings" "testing" + + "github.com/nats-io/nats.go/jetstream" ) // 1171/AC2: given an account file broker- 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 // in the error even when it is the field present and another is missing. 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 { field string values map[string]any @@ -82,3 +109,8 @@ func writeAccount(t *testing.T, fields map[string]any) string { } 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 } diff --git a/go/process/buckets.go b/go/process/buckets.go new file mode 100644 index 0000000..24651a4 --- /dev/null +++ b/go/process/buckets.go @@ -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) +}