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)

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.

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 {