From 9950689a11dbd673313315a7589c2df1c3c6ff6a Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 21:54:08 +0100 Subject: [PATCH 1/4] 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 +} From 462ec9cfb19ec7c70359c15b41d6489060727114 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 21:59:24 +0100 Subject: [PATCH 2/4] The assigner opens the puts' bucket again on a later pass once it is there and granted, so a put works without a restart (ADR 0546, ADR 0201) --- .../integration_test.go | 36 +++++++++++++++++++ .../cmd/mesh-delivery-assigner/main.go | 32 ++++++++++++++++- 2 files changed, 67 insertions(+), 1 deletion(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index ab8d70d..eac7f57 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -182,6 +182,42 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) { t.Errorf("without %s the assigner started", missing) } } + // ADR 0546: without the puts' bucket it starts all the same, and once the bucket is made (by the controller at its start, + // ADR 0201) a pass opens it, so a put is taken further without a restart. + was := PutsReopenEvery + PutsReopenEvery = 0 + defer func() { PutsReopenEvery = was }() + b := startBed(t, delivery.AssignerPutsBucket) + done := b.start(t) + select { + case err := <-done: + t.Fatalf("without %s the assigner stopped: %v", delivery.AssignerPutsBucket, err) + case <-time.After(2 * time.Second): + } + puts, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerPutsBucket, History: 5}) + if err != nil { + t.Fatal(err) + } + pkey, _ := delivery.PutKey("n1", "web") + asked := delivery.Put{FP: delivery.SourcePrefix2 + strings.Repeat("b", 64), Why: "a try", At: time.Now()}.Said(time.Now(), delivery.PutAsked, "asked") + raw, _ := delivery.EncodePut(asked) + if _, err := puts.Put(b.ctx, pkey, raw); err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(15 * time.Second); ; time.Sleep(200 * time.Millisecond) { + // A change of a bucket it watches starts a pass, which opens the puts' bucket. + if _, err := b.buckets[delivery.PresenceBucket].Put(b.ctx, "n1", []byte(time.Now().String())); err != nil { + t.Fatal(err) + } + if e, err := puts.Get(b.ctx, pkey); err == nil { + if got, err := delivery.DecodePut(e.Value()); err == nil && got.Step == delivery.PutRefused && strings.Contains(got.Words, "not assigned") { + break + } + } + if time.Now().After(deadline) { + t.Fatal("the put was not taken further once its bucket was made") + } + } } // putJSON writes a value into one of the bed's buckets. 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 041769d..19e15c0 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -16,6 +16,7 @@ import ( "os/signal" "strings" "syscall" + "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -230,12 +231,37 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a followed := watched if kvs[delivery.AssignerPutsBucket] != nil { if err := watch(delivery.AssignerPutsBucket); err != nil { - logger.Printf("%v: no put is taken further until the assigner starts again", err) + logger.Printf("%v: no put is taken further until it is opened again", err) delete(kvs, delivery.AssignerPutsBucket) } else { followed = append(append([]string{}, watched...), delivery.AssignerPutsBucket) } } + // The puts' bucket is made by the controller from the manifest's states at its start (ADR 0201), and the assigner's + // credential reads it once the bus's user list carries the grant: neither waits for the assigner. So while it is not open, + // a pass started by a change opens it again, at most once a minute, and from then on the puts are watched and taken + // further, without a restart. + var triedPuts time.Time + reopenPuts := func(p *passer) { + if kvs[delivery.AssignerPutsBucket] != nil || time.Since(triedPuts) < PutsReopenEvery { + return + } + triedPuts = time.Now() + kv, err := js.KeyValue(ctx, delivery.AssignerPutsBucket) + if err != nil { + return + } + kvs[delivery.AssignerPutsBucket] = kv + if err := watch(delivery.AssignerPutsBucket); err != nil { + logger.Printf("%v: no put is taken further until it is opened again", err) + delete(kvs, delivery.AssignerPutsBucket) + return + } + if bus != nil { + p.acts = newPutActs(bus, wake, logger.Printf) + } + logger.Printf("the bucket %s is open: puts are taken further", delivery.AssignerPutsBucket) + } for _, name := range watched { var opts []jetstream.WatchOpt if name == delivery.HaveBucket { @@ -271,11 +297,15 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a return err case <-settled: case <-dirty: + reopenPuts(p) p.pass(ctx) } } } +// PutsReopenEvery is how often, at most, a pass opens the puts' bucket again while it is not open. +var PutsReopenEvery = time.Minute + // sayVia publishes a condition on the operator-channel seat's one subject for conditions, under the assigner's own name // (ADR 0437 clause 9). func sayVia(ctx context.Context, js jetstream.JetStream) func(asks.Condition) error { From fa6cdbbe612f793aad8d5df5926dd69b0b6cb093 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 22:06:36 +0100 Subject: [PATCH 3/4] The puts' watch never gates the assigner's first pass: it is not waited for at the start, and one that ends before it handed what stands only stops the puts (ADR 0546) --- .../cmd/mesh-delivery-assigner/main.go | 11 +-- .../cmd/mesh-delivery-assigner/main_test.go | 73 +++++++++++++++++++ 2 files changed, 79 insertions(+), 5 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 19e15c0..0c7341d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -214,7 +214,9 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a return } if e == nil { - settled <- struct{}{} + if !optional { // the puts never gate the first pass, so never the wants (ADR 0546) + settled <- struct{}{} + } continue } var value []byte @@ -228,13 +230,12 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a }() return nil } - followed := watched + // The puts' watch is not waited for before the first pass: a watch of them that ends before it handed what stands only + // stops the puts, never the wants. if kvs[delivery.AssignerPutsBucket] != nil { if err := watch(delivery.AssignerPutsBucket); err != nil { logger.Printf("%v: no put is taken further until it is opened again", err) delete(kvs, delivery.AssignerPutsBucket) - } else { - followed = append(append([]string{}, watched...), delivery.AssignerPutsBucket) } } // The puts' bucket is made by the controller from the manifest's states at its start (ADR 0201), and the assigner's @@ -274,7 +275,7 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil { return err } - for range len(followed) + 1 { + for range len(watched) + 1 { select { case <-ctx.Done(): return nil 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 4f6092c..3c346ab 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 @@ -5,15 +5,18 @@ import ( "errors" "fmt" "io" + "log" "os" "path/filepath" "reflect" "slices" "strings" "testing" + "time" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -74,6 +77,32 @@ func TestOpensWhatItsManifestNames(t *testing.T) { t.Errorf("%s refused with %v: the start answered %v", c.missing, c.err, err) } } + // And a watch of the puts that ends before it handed what stands (its subscription closed, its grant withdrawn) does not + // keep the first pass from being made: the wants are written all the same. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + wrote := make(chan string, 8) + kvs := buckets{} + for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) { + kvs[name] = &watchKV{memKV: &memKV{values: map[string][]byte{}}} + } + ckey, _ := delivery.ComposedNodeKey("ace") + cd, err := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: "sha256:" + strings.Repeat("d", 64), Judgement: delivery.JudgedUnchanged}) + if err != nil { + t.Fatal(err) + } + kvs[delivery.ControllerComposedBucket].(*watchKV).entries = []kvEntry{{key: ckey, value: cd}} + kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true + kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote + go follow(ctx, nil, kvs, func(asks.Condition) error { return nil }, nil, log.New(io.Discard, "", 0)) + select { + case key := <-wrote: + if key != "want.ace.node" { + t.Errorf("the first pass wrote %s", key) + } + case <-ctx.Done(): + t.Error("a puts watch that ended before its first marker kept the first pass from being made: no want was written") + } } // 1297/B2a: without a credential the program does not start, and says why. @@ -97,3 +126,47 @@ func (r refusingJS) KeyValue(_ context.Context, name string) (jetstream.KeyValue } return &memKV{values: map[string][]byte{}}, nil } + +// watchKV is a memKV whose watch hands its entries and then the marker that all stands was handed, or ends at once when +// closed; a put on it is said on wrote. +type watchKV struct { + *memKV + entries []kvEntry + closed bool + wrote chan string +} + +func (w *watchKV) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + ch := make(chan jetstream.KeyValueEntry, len(w.entries)+1) + if w.closed { + close(ch) + return keyWatcher{ch}, nil + } + for _, e := range w.entries { + ch <- e + } + ch <- nil + return keyWatcher{ch}, nil +} + +func (w *watchKV) Put(ctx context.Context, key string, value []byte) (uint64, error) { + if w.wrote != nil { + w.wrote <- key + } + return 1, nil +} + +type keyWatcher struct{ ch chan jetstream.KeyValueEntry } + +func (k keyWatcher) Updates() <-chan jetstream.KeyValueEntry { return k.ch } +func (k keyWatcher) Stop() error { return nil } + +type kvEntry struct { + jetstream.KeyValueEntry + key string + value []byte +} + +func (e kvEntry) Key() string { return e.key } +func (e kvEntry) Value() []byte { return e.value } +func (e kvEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut } From eb65535f45db351897ec6a6454517ec5fcb9a93c Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 22:11:44 +0100 Subject: [PATCH 4/4] A want marked put and held stands while its put is not heard: the rules give it back only on a put heard released, so a start before the puts are open moves no node (ADR 0546 clause 5) --- .../mesh-delivery-assigner/internal/assign/wants.go | 10 +++++++++- .../internal/assign/wants_test.go | 5 +++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/modules/mesh-delivery-assigner/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go index c53f67d..8d0daff 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -226,7 +226,15 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st } before, had := wants[node] var next delivery.Want - if p, put := m.Puts[node]; put { + p, put := m.Puts[node] + if !put && had && before.Put && before.Hold { + // The want is the truth (ADR 0546 clause 5): a want marked put and held stands while its put is not heard (the + // puts not open, or not yet handed after a start); only a put heard released gives it back to the rules. A put + // without hold is not kept this way: its put takes it further once heard. + p, put = delivery.Put{FP: before.FP, Composed: before.Composed, Hold: true, Why: strings.TrimPrefix(before.Words, "put: "), + Step: delivery.PutHolding}, true + } + if put { if want, owns := PutOwns(p); owns { if want != nil { w := *want diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index 7c213b4..69deb19 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -397,10 +397,15 @@ func TestTheBusIsNotMovedButItsUserListRevisionIsWrittenAtOnce(t *testing.T) { {"wanted", delivery.PutWanted, wantedOld(), &delivery.Want{FP: fNew, Composed: dig(fNew), Put: true, Hold: true, Words: "put: the bus's user list"}}, {"holding, written already", delivery.PutHolding, map[string]delivery.Want{"ace": PutWant(put)}, nil}, {"released: the rules again", delivery.PutReleased, map[string]delivery.Want{"ace": PutWant(put)}, &delivery.Want{FP: fNew, Composed: dig(fNew)}}, + {"not heard: a held want at an older fingerprint stands", "", map[string]delivery.Want{"ace": {FP: fOld, Composed: dig(fOld), Put: true, Hold: true, Words: "put: back"}}, nil}, + {"not heard: a held want the rules would rewrite stands", "", map[string]delivery.Want{"ace": PutWant(put)}, nil}, } { p := put p.Step = c.step m.Puts = map[string]delivery.Put{"ace": p} + if c.step == "" { + m.Puts = nil + } out := Wants(m, n, c.wanted, nil) got, wrote := out.Wants["ace"] switch {