jschoubben
e0d1de0475
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.
56 lines
2.7 KiB
Go
56 lines
2.7 KiB
Go
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)
|
|
}
|