From 462ec9cfb19ec7c70359c15b41d6489060727114 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 21:59:24 +0100 Subject: [PATCH] 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 {