Bind the have bucket once per pass, bound by the queue's timeout, and stop the pass at the first error (issue 1132)

internal/link/bus.go

@@ -118,24 +118,38 @@ func (b OverNATS) Alive(ctx context.Context, node string, body []byte) error {
 	return b.Conn.FlushWithContext(flush)
 }
 
-// haveBucket binds to the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132). **Only binds**: a node-engine's
+// HaveBind binds the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132), once for a pass of writes.
-// account reaches no part of the JetStream API beyond what it is granted, and the bucket is the controller's to create
+// **Only binds**: a node-engine's account reaches no part of the JetStream API beyond what it is granted, and the bucket
-// and to configure, so a bucket that is not there is an error said as what it is.
+// is the controller's to create and to configure, so a bucket that is not there is an error said as what it is. Every
-func (b OverNATS) haveBucket() (nats.KeyValue, error) {
+// call on the bucket waits no longer than ctx has left: a refused publish is answered by nothing, and the client's own
-	kv, err := b.JS.KeyValue(delivery.HaveBucket)
+// wait is five seconds.
+func (b OverNATS) HaveBind(ctx context.Context) (HaveBucket, error) {
+	js := b.JS
+	wait := 5 * time.Second
+	if at, ok := ctx.Deadline(); ok {
+		wait = time.Until(at)
+		if wait <= 0 {
+			return nil, context.DeadlineExceeded
+		}
+	}
+	if b.Conn != nil {
+		bounded, err := b.Conn.JetStream(nats.MaxWait(wait))
+		if err != nil {
+			return nil, err
+		}
+		js = bounded
+	}
+	kv, err := js.KeyValue(delivery.HaveBucket)
 	if err != nil {
 		return nil, fmt.Errorf("the have bucket %s: %w", delivery.HaveBucket, err)
 	}
-	return kv, nil
+	return natsHaveBucket{kv}, nil
 }
 
-// HaveGet reads a key of the have bucket with its revision.
+type natsHaveBucket struct{ kv nats.KeyValue }
-func (b OverNATS) HaveGet(_ context.Context, key string) ([]byte, uint64, error) {
+
-	kv, err := b.haveBucket()
+func (n natsHaveBucket) Get(_ context.Context, key string) ([]byte, uint64, error) {
-	if err != nil {
+	entry, err := n.kv.Get(key)
-		return nil, 0, err
-	}
-	entry, err := kv.Get(key)
 	if errors.Is(err, nats.ErrKeyNotFound) {
 		return nil, 0, ErrHaveAbsent
 	}
@@ -145,17 +159,13 @@ func (b OverNATS) HaveGet(_ context.Context, key string) ([]byte, uint64, error)
 	return entry.Value(), entry.Revision(), nil
 }
 
-// HaveSet writes a key of the have bucket over the revision read, or creates it when the revision is zero.
+func (n natsHaveBucket) Set(_ context.Context, key string, value []byte, revision uint64) error {
-func (b OverNATS) HaveSet(_ context.Context, key string, value []byte, revision uint64) error {
+	var err error
-	kv, err := b.haveBucket()
-	if err != nil {
-		return err
-	}
 	if revision == 0 {
-		_, err = kv.Create(key, value)
+		_, err = n.kv.Create(key, value)
 		return err
 	}
-	_, err = kv.Update(key, value, revision)
+	_, err = n.kv.Update(key, value, revision)
 	return err
 }
 

internal/link/have.go

@@ -14,13 +14,20 @@ import (
 // ErrHaveAbsent is what HaveGet returns for a key the bucket does not hold.
 var ErrHaveAbsent = errors.New("link: no such key")
 
-// HaveBus is a link that can read and write the node-engine's own keys in the have bucket (novox/hq ADR 0429 §4). Separate
+// HaveBus is a link that can bind the have bucket (novox/hq ADR 0429 §4). Separate from Bus, as HealthBus is: a link that
-// from Bus, as HealthBus is: a link that cannot is still a link, and the write waits for the next apply or look.
+// cannot is still a link, and the write waits for the next apply or look.
 type HaveBus interface {
-	// HaveGet is a key's value and last revision, or ErrHaveAbsent.
+	// HaveBind binds the bucket the controller made, once for a pass of writes, within ctx. A node whose grant is not there
-	HaveGet(ctx context.Context, key string) (value []byte, revision uint64, err error)
+	// yet gets an error here (or the context's end), and writes nothing.
-	// HaveSet writes a key over the revision read: zero creates it, and a key that moved since is refused.
+	HaveBind(ctx context.Context) (HaveBucket, error)
-	HaveSet(ctx context.Context, key string, value []byte, revision uint64) error
+}
+
+// HaveBucket is the bound bucket.
+type HaveBucket interface {
+	// Get is a key's value and last revision, or ErrHaveAbsent.
+	Get(ctx context.Context, key string) (value []byte, revision uint64, err error)
+	// Set writes a key over the revision read: zero creates it, and a key that moved since is refused.
+	Set(ctx context.Context, key string, value []byte, revision uint64) error
 }
 
 // haveState is what the node-engine last learned of each module of the declaration it applied, and of each module's health
@@ -151,8 +158,15 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) {
 		return
 	}
 	node := q.Membership.Node
+	// The whole pass is bounded by the queue's timeout, and it stops at the first error: a node whose grant is not there
+	// yet would otherwise make the worker wait the bound again for every key (issue 1132).
 	write, cancel := context.WithTimeout(context.WithoutCancel(ctx), q.timeout())
 	defer cancel()
+	bucket, err := hb.HaveBind(write)
+	if err != nil {
+		q.sayHave(fmt.Sprintf("could not bind the have bucket, and wrote nothing there; the next apply or look writes again: %v", err))
+		return
+	}
 	modules := make([]string, 0, len(s.modules))
 	for module := range s.modules {
 		modules = append(modules, module)
@@ -178,10 +192,12 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) {
 			q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err))
 			continue
 		}
-		q.putHave(write, hb, key, raw, func(current []byte) bool {
+		if !q.putHave(write, bucket, key, raw, func(current []byte) bool {
 			c, err := delivery.DecodeHave(current)
 			return err == nil && delivery.SameHave(c, h)
-		})
+		}) {
+			return
+		}
 	}
 	if s.node != nil {
 		key, err := delivery.HaveNodeKey(node)
@@ -189,7 +205,7 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) {
 			var raw []byte
 			if raw, err = delivery.EncodeHaveNode(*s.node); err == nil {
 				want := *s.node
-				q.putHave(write, hb, key, raw, func(current []byte) bool {
+				q.putHave(write, bucket, key, raw, func(current []byte) bool {
 					c, err := delivery.DecodeHaveNode(current)
 					return err == nil && delivery.SameHaveNode(c, want)
 				})
@@ -201,19 +217,26 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) {
 	}
 }
 
-// putHave reads a key and writes it unless same says it already holds the value.
+// putHave reads a key and writes it unless same says it already holds the value. False when the pass should stop: the
-func (q *Queue) putHave(ctx context.Context, hb HaveBus, key string, value []byte, same func([]byte) bool) {
+// bucket failed, the error is said once, and the rest wait for the next pass.
-	current, revision, err := hb.HaveGet(ctx, key)
+func (q *Queue) putHave(ctx context.Context, b HaveBucket, key string, value []byte, same func([]byte) bool) bool {
+	if ctx.Err() != nil {
+		q.sayHave(fmt.Sprintf("the time for writing the have bucket ran out before %s; the next apply or look writes it", key))
+		return false
+	}
+	current, revision, err := b.Get(ctx, key)
 	switch {
 	case errors.Is(err, ErrHaveAbsent):
 		revision = 0
 	case err != nil:
-		q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing there: %v", key, err))
+		q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err))
-		return
+		return false
 	case same(current):
-		return
+		return true
 	}
-	if err := hb.HaveSet(ctx, key, value, revision); err != nil {
+	if err := b.Set(ctx, key, value, revision); err != nil {
 		q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err))
+		return false
 	}
+	return true
 }

internal/link/have_test.go

@@ -18,11 +18,14 @@ import (
 // haveLink is a link with a bucket behind it that keeps each key's revision as the bus does.
 type haveLink struct {
 	*quietLink
-	hmu     sync.Mutex
+	hmu                  sync.Mutex
-	kv      map[string]haveEntry
+	kv                   map[string]haveEntry
-	sets    []string // keys written, in order
+	sets                 []string // keys written, in order
-	events  []string // "report" and "have <key>", in the order they happened
+	events               []string // "report" and "have <key>", in the order they happened
-	failing bool
+	failing              bool
+	binds                int
+	setTries             int
+	bindFails, bindStuck bool
 }
 
 type haveEntry struct {
@@ -44,7 +47,22 @@ func (l *haveLink) Report(ctx context.Context, node string, body []byte) error {
 
 func (l *haveLink) Health(context.Context, string, []byte) error { return nil }
 
-func (l *haveLink) HaveGet(_ context.Context, key string) ([]byte, uint64, error) {
+func (l *haveLink) HaveBind(ctx context.Context) (HaveBucket, error) {
+	l.hmu.Lock()
+	l.binds++
+	stuck, refuse := l.bindStuck, l.bindFails
+	l.hmu.Unlock()
+	if stuck {
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+	if refuse {
+		return nil, errors.New("no grant for the have bucket")
+	}
+	return l, nil
+}
+
+func (l *haveLink) Get(_ context.Context, key string) ([]byte, uint64, error) {
 	l.hmu.Lock()
 	defer l.hmu.Unlock()
 	e, ok := l.kv[key]
@@ -54,9 +72,10 @@ func (l *haveLink) HaveGet(_ context.Context, key string) ([]byte, uint64, error
 	return e.value, e.rev, nil
 }
 
-func (l *haveLink) HaveSet(_ context.Context, key string, value []byte, revision uint64) error {
+func (l *haveLink) Set(_ context.Context, key string, value []byte, revision uint64) error {
 	l.hmu.Lock()
 	defer l.hmu.Unlock()
+	l.setTries++
 	if l.failing {
 		return errors.New("the bucket refused")
 	}
@@ -289,3 +308,34 @@ func TestASilentReconcileStillWritesTheHaves(t *testing.T) {
 		t.Errorf("an unchanged reconcile wrote %d keys", l.setCount()-before)
 	}
 }
+
+// 1132/AC7: a bind that fails, or never answers, writes nothing and returns within the queue's timeout; a write that
+// fails ends the pass, and is said once.
+func TestAFailedBindOrWriteEndsThePassAndReturnsInTime(t *testing.T) {
+	for name, set := range map[string]func(*haveLink){
+		"a bind refused":            func(l *haveLink) { l.bindFails = true },
+		"a bind that never answers": func(l *haveLink) { l.bindStuck = true },
+	} {
+		l := newHaveLink()
+		set(l)
+		q, said := aHaveQueue(l)
+		q.Timeout = 100 * time.Millisecond
+		start := time.Now()
+		if !q.tell(context.Background(), appliedModules(), applied) {
+			t.Fatalf("%s: the report did not go out", name)
+		}
+		if took := time.Since(start); took > time.Second {
+			t.Errorf("%s: the pass took %s", name, took)
+		}
+		if l.binds != 1 || l.setTries != 0 || len(*said) != 1 {
+			t.Errorf("%s: binds %d, writes tried %d, said %v; want one bind, no write, one line", name, l.binds, l.setTries, *said)
+		}
+	}
+	l := newHaveLink()
+	l.failing = true
+	q, said := aHaveQueue(l)
+	q.tell(context.Background(), appliedModules(), applied)
+	if l.binds != 1 || l.setTries != 1 || len(*said) != 1 {
+		t.Errorf("a failed write: binds %d, writes tried %d, said %v; want one of each and one line", l.binds, l.setTries, *said)
+	}
+}

internal/link/hearing_nats.go

@@ -342,12 +342,8 @@ func (l *natsLink) Health(ctx context.Context, node string, body []byte) error {
 	return OverNATS{Conn: l.conn, JS: l.js}.Health(ctx, node, body)
 }
 
-func (l *natsLink) HaveGet(ctx context.Context, key string) ([]byte, uint64, error) {
+func (l *natsLink) HaveBind(ctx context.Context) (HaveBucket, error) {
-	return OverNATS{Conn: l.conn, JS: l.js}.HaveGet(ctx, key)
+	return OverNATS{Conn: l.conn, JS: l.js}.HaveBind(ctx)
-}
-
-func (l *natsLink) HaveSet(ctx context.Context, key string, value []byte, revision uint64) error {
-	return OverNATS{Conn: l.conn, JS: l.js}.HaveSet(ctx, key, value, revision)
 }
 
 // Ask asks a module's tool for a declared tool check (novox/hq ADR 0240, to-be 48 §3). The link the