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)

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

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
+}