Merge pull request 'The assigner writes wants whatever keeps its puts from opening, and never moves a held put's want (ADR 0546, issue 1945)' (#50) from fix/assigner-starts-without-puts into main

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"
@@ -86,14 +87,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.
+// without it would write wants from half the mesh. The puts' bucket is the exception: whatever keeps it from opening (it
-func open(ctx context.Context, js jetstream.JetStream) (buckets, error) {
+// 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) {
+		if name == delivery.AssignerPutsBucket && err != nil {
-			// The puts (ADR 0546) are taken further only once their bucket exists: an assigner never stops writing wants
+			logf("the bucket %s cannot be opened (%v): no put is taken further until the assigner starts again", name, err)
-			// for want of it. The verb's first put is refused by the runtime until then.
 			continue
 		}
 		if err != nil {
@@ -134,13 +136,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 +195,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,11 +206,17 @@ 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
 					}
 					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
@@ -223,11 +230,40 @@ 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 {
-		followed = append(append([]string{}, watched...), delivery.AssignerPutsBucket)
+		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)
+		}
 	}
-	for _, name := range followed {
+	// 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 {
 			opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back
@@ -239,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
@@ -262,11 +298,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 {

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go

@@ -2,15 +2,23 @@ package main
 
 import (
 	"context"
+	"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"
 )
 
 // 1297/B2a: the buckets the program opens are exactly the manifest's reads and states, as the broker names them
@@ -48,6 +56,53 @@ 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)
+		}
+	}
+	// 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.
@@ -57,3 +112,61 @@ 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
+}
+
+// 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 }

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

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 {