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)

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

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 }