From 9950689a11dbd673313315a7589c2df1c3c6ff6a Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 21:54:08 +0100 Subject: [PATCH] The assigner starts and writes wants whatever keeps the puts' bucket from opening or being watched (not there, not granted, no answer): it takes no put further until it starts again, said in its journal (ADR 0546) --- .../cmd/mesh-delivery-assigner/main.go | 31 +++++++++----- .../cmd/mesh-delivery-assigner/main_test.go | 40 +++++++++++++++++++ 2 files changed, 60 insertions(+), 11 deletions(-) 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..041769d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -86,14 +86,15 @@ var keptStates = []announce.State{ 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) { +// without it would write wants from half the mesh. The puts' bucket is the exception: whatever keeps it from opening (it +// does not exist yet, the credential is not granted it yet, the bus does not answer in time), the assigner writes wants +// without it and takes no put further until it starts again (ADR 0546). +func open(ctx context.Context, js jetstream.JetStream, logf func(string, ...any)) (buckets, error) { out := buckets{} for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) { kv, err := js.KeyValue(ctx, name) - if name == delivery.AssignerPutsBucket && errors.Is(err, jetstream.ErrBucketNotFound) { - // The puts (ADR 0546) are taken further only once their bucket exists: an assigner never stops writing wants - // for want of it. The verb's first put is refused by the runtime until then. + if name == delivery.AssignerPutsBucket && err != nil { + logf("the bucket %s cannot be opened (%v): no put is taken further until the assigner starts again", name, err) continue } if err != nil { @@ -134,13 +135,10 @@ 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 } - if kvs[delivery.AssignerPutsBucket] == nil { - logger.Printf("the bucket %s does not exist yet, so no put is taken further until the assigner starts again", delivery.AssignerPutsBucket) - } // What the assigner keeps, said on the bus (ADR 0495 §2-§3): a reader is granted from it. Not fatal when it cannot // be said now: process.Conn says it again at every reconnect. said, err := process.Connect(account, nats.Name(account.Node+"."+account.Module+".keeps"), nats.RetryOnFailedConnect(true)) @@ -196,6 +194,8 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a if err != nil { return fmt.Errorf("the bucket %s cannot be watched: %w", name, err) } + // The puts' watch never stops the wants (ADR 0546): when it ends, the puts are no longer heard, said in the journal. + optional := name == delivery.AssignerPutsBucket go func() { defer kw.Stop() for { @@ -205,6 +205,10 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a return case e, ok := <-kw.Updates(): if !ok { + if optional { + logger.Printf("the watch of %s ended: no put is taken further until the assigner starts again", name) + return + } errs <- fmt.Errorf("the watch of %s ended", name) return } @@ -225,9 +229,14 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a } followed := watched if kvs[delivery.AssignerPutsBucket] != nil { - followed = append(append([]string{}, watched...), delivery.AssignerPutsBucket) + if err := watch(delivery.AssignerPutsBucket); err != nil { + logger.Printf("%v: no put is taken further until the assigner starts again", err) + delete(kvs, delivery.AssignerPutsBucket) + } else { + followed = append(append([]string{}, watched...), delivery.AssignerPutsBucket) + } } - for _, name := range followed { + for _, name := range watched { var opts []jetstream.WatchOpt if name == delivery.HaveBucket { opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back 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..4f6092c 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,27 @@ 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) } + // ADR 0546: whatever keeps the puts' bucket from opening (not there, not granted, no answer in time), the assigner opens + // the rest and starts, writing wants; any other bucket it cannot open stops the start, named. + for _, c := range []struct { + missing string + err error + starts bool + }{ + {delivery.AssignerPutsBucket, jetstream.ErrBucketNotFound, true}, + {delivery.AssignerPutsBucket, errors.New("nats: permissions violation for subscription to \"$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_puts\""), true}, + {delivery.AssignerPutsBucket, context.DeadlineExceeded, true}, + {delivery.AssignerWantsBucket, errors.New("nats: permissions violation"), false}, + } { + var logged string + kvs, err := open(context.Background(), refusingJS{missing: c.missing, err: c.err}, func(f string, a ...any) { logged = fmt.Sprintf(f, a...) }) + switch { + case c.starts && (err != nil || kvs[c.missing] != nil || len(kvs) != len(readBuckets)+len(ownBuckets)-1 || !strings.Contains(logged, "no put is taken further")): + t.Errorf("%s refused with %v: the start answered %v, %d buckets, logged %q", c.missing, c.err, err, len(kvs), logged) + case !c.starts && (err == nil || !strings.Contains(err.Error(), c.missing)): + t.Errorf("%s refused with %v: the start answered %v", c.missing, c.err, err) + } + } } // 1297/B2a: without a credential the program does not start, and says why. @@ -57,3 +83,17 @@ func TestNoCredential(t *testing.T) { t.Errorf("run without a credential = %v", err) } } + +// refusingJS opens every bucket but one, which it refuses with err. +type refusingJS struct { + jetstream.JetStream + missing string + err error +} + +func (r refusingJS) KeyValue(_ context.Context, name string) (jetstream.KeyValue, error) { + if name == r.missing { + return nil, r.err + } + return &memKV{values: map[string][]byte{}}, nil +}