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…
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:
@@ -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 }
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user