Keep refused keys from holding the take-in's slots, and judge its stall by acknowledgements (issue 1945, ADR 0544, review)

A refused key stays unacknowledged until its delay ends, so with 8 a consumer
a handful refused for long (a manifest word this controller does not read
yet, a node that keeps failing) stopped that consumer for good, the build that
reads the word among what it held back. Each consumer now leaves up to 1000
unacknowledged; the store's bound is the one loop's batch of 16. A refusal
waits 30 s, doubling each time up to 10 minutes.

The stall watch judged movement by deliveries, which rise with every
redelivery: it now counts the keys this process acknowledged. A consumer that
cannot be made raises take-in-stalled instead of a log line. On stop the
gathered batch and a key in hand are refused with no delay, so the next
controller has them at once. A trunk value with no fingerprint of the new form
is said once, as every wait for a write.
Research 072: the take-ins on a bounded pull consumer.

cmd/mesh-controller/composed_keys.go

@@ -57,6 +57,8 @@ type takeInSteps struct {
 type delivered struct {
 	module string
 	again  bool
+	// deliveries is how often the bus delivered it, this time included: what a refusal's delay grows with.
+	deliveries uint64
 }
 
 // verdict is what becomes of a module's delivered keys.
@@ -123,7 +125,18 @@ func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[strin
 			continue
 		}
 		trunk, err := s.trunk(ctx, m)
-		if errors.Is(err, builderkeys.ErrNoValue) || (err == nil && !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2)) {
+		if errors.Is(err, builderkeys.ErrNoValue) {
+			out = append(out, r)
+			continue
+		}
+		if err == nil && !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) {
+			// The indexer's next write names one: said once, as every wait for a write (ADR 0544 §4).
+			why := trunk.None2
+			if why == "" {
+				why = "the indexer wrote no fingerprint of the new form"
+			}
+			r.waits = fmt.Sprintf("%v: %s at %.8s has no fingerprint to take a build by: %s", errWaitsForAWrite, m,
+				trunk.Commit, why)
 			out = append(out, r)
 			continue
 		}
@@ -253,9 +266,23 @@ func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[strin
 // doneKeysSay says a line of the take-ins; a variable so a test may hear it.
 var doneKeysSay = func(s string) { fmt.Println(s) }
 
-// doneKeyRetry is how long the bus waits before it delivers again a key whose take-in was refused.
+// doneKeyRetry is how long the bus waits before it delivers again a key whose take-in was refused the first time; each
+// refusal after doubles it, up to doneKeyRetryMost, so a key refused for long (a manifest word this controller does not
+// read yet, a node whose composition keeps failing) costs a take-in every few minutes, not every half minute.
 var doneKeyRetry = 30 * time.Second
 
+// doneKeyRetryMost is the longest a refused key waits before it is delivered again.
+const doneKeyRetryMost = 10 * time.Minute
+
+// refusalDelay is how long the bus waits before it delivers again a key refused on its deliveries-th delivery.
+func refusalDelay(deliveries uint64) time.Duration {
+	d := doneKeyRetry
+	for i := uint64(1); i < deliveries && d < doneKeyRetryMost; i++ {
+		d *= 2
+	}
+	return min(d, doneKeyRetryMost)
+}
+
 // kindNotComposed is the condition of a module's new fingerprint whose compositions or grant are not written: until
 // they are, no want can name it.
 const kindNotComposed = "fingerprint-not-composed"

cmd/mesh-controller/composed_keys_test.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/novox/mesh-controller/internal/builderkeys"
 )
 
@@ -199,9 +200,9 @@ func TestATakeInBatch(t *testing.T) {
 			nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}}, takeInErr: map[string]error{"b": waitsOnARead}}
 		var acked, naked []string
 		msg := func(module string) takeInMsg {
-			return takeInMsg{delivered: delivered{module: module}, ack: func() { acked = append(acked, module) },
+			return takeInMsg{delivered: delivered{module: module, deliveries: 3}, ack: func() { acked = append(acked, module) },
 				nak: func(d time.Duration) {
-					if d != doneKeyRetry {
+					if d != refusalDelay(3) {
 						t.Errorf("refused with a delay of %s", d)
 					}
 					naked = append(naked, module)
@@ -216,30 +217,50 @@ func TestATakeInBatch(t *testing.T) {
 			t.Fatalf("said %q", said)
 		}
 	})
+
+	// A key refused for long holds no slot the others need: each consumer leaves far more keys unacknowledged than a
+	// batch takes, so 8 keys refused on each stop no ninth (the review of ADR 0544: with 8 a consumer, they stopped it
+	// for good), and a refusal waits longer each time, up to ten minutes.
+	t.Run("refused keys stop no other", func(t *testing.T) {
+		for _, bucket := range []string{sdkdelivery.ModulesBucket, sdkdelivery.BuilderBuildsBucket} {
+			if c := takeInConfig(bucket); c.MaxAckPending < 8+startTakeIns || c.AckPolicy != jetstream.AckExplicitPolicy {
+				t.Errorf("%s: %d unacknowledged at most (%v): 8 refused keys and a batch do not fit", bucket,
+					c.MaxAckPending, c.AckPolicy)
+			}
+		}
+		was := time.Duration(0)
+		for n := uint64(1); n <= 8; n++ {
+			d := refusalDelay(n)
+			if d < was || d > doneKeyRetryMost || (n == 1 && d != doneKeyRetry) {
+				t.Fatalf("the %d. refusal waits %s (the one before %s)", n, d, was)
+			}
+			was = d
+		}
+		if was != doneKeyRetryMost {
+			t.Fatalf("the 8. refusal waits %s, want %s", was, doneKeyRetryMost)
+		}
+	})
 }
 
-// A backlog that stands above zero with nothing delivered for five minutes is a stall; one that moves, or empties, is
-// not (ADR 0544 §6).
+// A backlog that stands above zero with no key acknowledged for five minutes is a stall, however often its keys are
+// delivered again; one that takes keys in, or empties, is not (ADR 0544 §6 and its review).
 func TestATakeInStallsWhenItsBacklogStandsStill(t *testing.T) {
 	at := time.Date(2026, 10, 25, 17, 36, 0, 0, time.UTC)
 	w := &stallWatch{}
-	read := func(pending uint64, delivered ...uint64) takeInBacklog {
-		return takeInBacklog{Pending: pending, delivered: delivered}
-	}
 	steps := []struct {
 		after            time.Duration
-		backlog          takeInBacklog
+		backlog, acked   uint64
 		stalled, changed bool
 	}{
-		{0, read(190, 10, 3), false, false},
-		{4 * time.Minute, read(190, 10, 3), false, false},
-		{5 * time.Minute, read(190, 10, 3), true, true},
-		{6 * time.Minute, read(190, 10, 3), true, false},
-		{7 * time.Minute, read(180, 11, 3), false, true},
-		{20 * time.Minute, read(0, 11, 3), false, false},
+		{0, 190, 10, false, false},
+		{4 * time.Minute, 190, 10, false, false},
+		{5 * time.Minute, 190, 10, true, true}, // its keys delivered again all along, none acknowledged
+		{6 * time.Minute, 190, 10, true, false},
+		{7 * time.Minute, 180, 20, false, true},
+		{20 * time.Minute, 0, 20, false, false},
 	}
 	for i, s := range steps {
-		stalled, changed := w.look(s.backlog, at.Add(s.after))
+		stalled, changed := w.look(takeInBacklog{Taking: s.backlog}, s.acked, at.Add(s.after))
 		if stalled != s.stalled || changed != s.changed {
 			t.Fatalf("reading %d: stalled %v changed %v, want %v %v", i, stalled, changed, s.stalled, s.changed)
 		}

cmd/mesh-controller/take_in_consumer.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
@@ -19,16 +20,23 @@ import (
 // of a controller whose consumer is new, and nothing at a restart but what the controller before it left
 // unacknowledged. On the builder's builds bucket it delivers each write from when it is made. A key is acknowledged once
 // its module is taken in, has nothing new, or waits for a write that will be delivered; any other wait or failure is
-// refused with a delay, and the bus delivers it again. At most startTakeIns keys of the two together are unacknowledged,
-// half on each: the bound on the store (issues 1945 and 1277: a burst of take-ins once filled it until it refused every
-// connection).
+// refused with a delay that grows with each refusal, and the bus delivers it again.
+//
+// **The bound on the store is the one take-in loop**, a batch of at most startTakeIns keys at a time (issues 1945 and
+// 1277: a burst of take-ins once filled the store until it refused every connection). The consumers' own limit on
+// unacknowledged keys is far above it, because a refused key stays unacknowledged until its delay ends: with a limit of
+// a batch, a handful of keys refused for long (a manifest word this controller does not read yet) would stop a consumer
+// for good, the build that reads the word among what it holds back (the review of ADR 0544).
 
-// startTakeIns is how many keys the two consumers leave unacknowledged at most together, and the most a batch takes: the
-// take-ins that may wait on the store.
+// startTakeIns is the most a batch takes: the take-ins that may wait on the store at once.
 const startTakeIns = 16
 
-// perConsumer is each consumer's share of startTakeIns.
-const perConsumer = startTakeIns / 2
+// takeInPending is how many keys each consumer may leave unacknowledged: delivered and being taken in, or refused and
+// waiting for their delay. Far above a batch, so refused keys never stop the others.
+const takeInPending = 1000
+
+// takeInBuffered is how many delivered keys each consumer's reader holds ahead of the loop.
+const takeInBuffered = startTakeIns / 2
 
 // takeInAckWait is how long a delivered key may stay unacknowledged before the bus delivers it again: longer than a batch
 // takes, its two composer passes among it.
@@ -57,11 +65,11 @@ func takeInKeyOf(bucket string, m jetstream.Msg) (delivered, bool) {
 	if module == "" {
 		return delivered{}, false
 	}
-	again := false
+	var deliveries uint64 = 1
 	if meta, err := m.Metadata(); err == nil {
-		again = meta.NumDelivered > 1
+		deliveries = meta.NumDelivered
 	}
-	return delivered{module: module, again: again}, true
+	return delivered{module: module, again: deliveries > 1, deliveries: deliveries}, true
 }
 
 // takeInConsumer is one bucket's consumer of the take-in.
@@ -70,23 +78,34 @@ type takeInConsumer struct {
 	consumer jetstream.Consumer
 }
 
+// takeInConfig is the take-in's consumer on bucket's stream.
+func takeInConfig(bucket string) jetstream.ConsumerConfig {
+	deliver := jetstream.DeliverNewPolicy
+	if bucket == sdkdelivery.ModulesBucket {
+		deliver = jetstream.DeliverLastPerSubjectPolicy // the start's look: every module's trunk value once
+	}
+	return jetstream.ConsumerConfig{
+		Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver,
+		AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: takeInPending,
+		Description: "the controller's take-in of done keys (novox/hq ADR 0544)",
+	}
+}
+
 // takeInConsumers makes or finds the take-in's consumer on each bucket's stream; one that cannot be made is said, and
 // the other is still read.
 func takeInConsumers(ctx context.Context, js jetstream.JetStream) []takeInConsumer {
 	var out []takeInConsumer
 	for _, bucket := range broker.TakeInBuckets {
-		deliver := jetstream.DeliverNewPolicy
-		if bucket == sdkdelivery.ModulesBucket {
-			deliver = jetstream.DeliverLastPerSubjectPolicy // the start's look: every module's trunk value once
-		}
-		c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, jetstream.ConsumerConfig{
-			Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver,
-			AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: perConsumer,
-			Description: "the controller's take-in of done keys (novox/hq ADR 0544)",
-		})
+		c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, takeInConfig(bucket))
 		if err != nil {
-			fmt.Printf("done keys: the take-in's consumer on %s is not made, so its keys are not taken in until the "+
-				"next start: %v\n", bucket, err)
+			// Loud: a consumer whose deliver policy differs cannot be updated, and nothing of its bucket is taken in.
+			o := takeInStalledObservation(takeInBacklog{}, time.Now())
+			o.Summary = fmt.Sprintf("the take-in's consumer on %s is not made, so its keys are not taken in until the "+
+				"next start: %v", bucket, err)
+			doneKeysSay("done keys: " + o.Summary)
+			if keeper := conditionsFrom; keeper != nil {
+				_, _ = keeper.Observe(ctx, o)
+			}
 			continue
 		}
 		out = append(out, takeInConsumer{bucket: bucket, consumer: c})
@@ -108,11 +127,14 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 		return
 	}
 	in := make(chan takeInMsg)
+	// acked counts the keys this process acknowledged: what tells a take-in that moves from one whose keys are only
+	// delivered again (the stall watch, ADR 0544 §6).
+	acked := &atomic.Uint64{}
 	read := make([]jetstream.Consumer, 0, len(consumers))
 	for _, c := range consumers {
 		bucket := c.bucket
 		read = append(read, c.consumer)
-		it, err := c.consumer.Messages(jetstream.PullMaxMessages(perConsumer))
+		it, err := c.consumer.Messages(jetstream.PullMaxMessages(takeInBuffered))
 		if err != nil {
 			fmt.Printf("done keys: the take-in's consumer on %s is not read: %v\n", bucket, err)
 			continue
@@ -135,11 +157,16 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 					_ = m.Ack()
 					continue
 				}
-				msg := takeInMsg{delivered: d, ack: func() { _ = m.Ack() },
+				msg := takeInMsg{delivered: d, ack: func() {
+					if m.Ack() == nil {
+						acked.Add(1)
+					}
+				},
 					nak: func(delay time.Duration) { _ = m.NakWithDelay(delay) }}
 				select {
 				case in <- msg:
 				case <-ctx.Done():
+					_ = m.Nak() // delivered again at once to the controller after this one, not after the ack wait
 					return
 				}
 			}
@@ -148,7 +175,7 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 	backlog := func(ctx context.Context) (takeInBacklog, error) { return backlogOf(ctx, read) }
 	steps := liveTakeInSteps(open, keys, keys.JS)
 	goServing(ctx, func() { takeInLoop(ctx, in, steps, backlog) })
-	goServing(ctx, func() { watchTakeInStall(ctx, backlog) })
+	goServing(ctx, func() { watchTakeInStall(ctx, backlog, acked.Load) })
 	goServing(ctx, func() {
 		<-ctx.Done()
 		closeKeys()
@@ -177,6 +204,9 @@ func takeInLoop(ctx context.Context, in <-chan takeInMsg, steps takeInSteps,
 				break more
 			case <-ctx.Done():
 				gather.Stop()
+				for _, m := range batch {
+					m.nak(0)
+				}
 				return
 			}
 		}
@@ -245,7 +275,7 @@ func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeI
 	}
 	for _, m := range batch {
 		if verdicts[m.module] == refused {
-			m.nak(doneKeyRetry)
+			m.nak(refusalDelay(m.deliveries))
 		} else {
 			m.ack()
 		}
@@ -254,12 +284,11 @@ func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeI
 		len(batch), len(settled), taken, nothing, waiting, failed, passes)
 }
 
-// takeInBacklog is the take-in's backlog on the bus: keys not delivered yet, keys delivered and not acknowledged, and
-// the last delivered sequence of each consumer, which moves when anything is delivered.
+// takeInBacklog is the take-in's backlog on the bus: keys not delivered yet, and keys delivered and not acknowledged
+// (being taken in, or refused and waiting for their delay).
 type takeInBacklog struct {
-	Pending   uint64 `json:"pending"`
-	Taking    uint64 `json:"taking"`
-	delivered []uint64
+	Pending uint64 `json:"pending"`
+	Taking  uint64 `json:"taking"`
 }
 
 // Total is every key the take-in has not acknowledged.
@@ -275,7 +304,6 @@ func backlogOf(ctx context.Context, consumers []jetstream.Consumer) (takeInBackl
 		}
 		b.Pending += info.NumPending
 		b.Taking += uint64(info.NumAckPending)
-		b.delivered = append(b.delivered, info.Delivered.Consumer)
 	}
 	return b, nil
 }
@@ -308,39 +336,35 @@ func takeInStalledObservation(b takeInBacklog, since time.Time) conditions.Obser
 			b.Total(), since.UTC().Format(time.RFC3339))}
 }
 
-// stallWatch judges the backlog read at each look: stalled once it stood above zero, with nothing delivered, for
-// takeInStall.
+// stallWatch judges the backlog read at each look: stalled once it stood above zero, with no key acknowledged, for
+// takeInStall. Acknowledgements, not deliveries: a key refused and delivered again moves the deliveries and takes
+// nothing in (the review of ADR 0544).
 type stallWatch struct {
-	mu        sync.Mutex
-	last      []uint64
-	movedAt   time.Time
-	stalled   bool
-	lastTotal uint64
+	mu      sync.Mutex
+	acked   uint64
+	movedAt time.Time
+	stalled bool
 }
 
-// look takes a reading at now and answers whether the take-in stalls and whether that changed.
-func (w *stallWatch) look(b takeInBacklog, now time.Time) (stalled, changed bool) {
+// look takes a reading at now, with how many keys this process has acknowledged, and answers whether the take-in
+// stalls and whether that changed.
+func (w *stallWatch) look(b takeInBacklog, acked uint64, now time.Time) (stalled, changed bool) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	moved := b.Total() == 0 || len(w.last) != len(b.delivered)
-	for i := range b.delivered {
-		if !moved && b.delivered[i] != w.last[i] {
-			moved = true
-		}
-	}
-	if moved || w.movedAt.IsZero() {
+	if b.Total() == 0 || acked != w.acked || w.movedAt.IsZero() {
 		w.movedAt = now
 	}
-	w.last, w.lastTotal = b.delivered, b.Total()
+	w.acked = acked
 	nowStalled := b.Total() > 0 && now.Sub(w.movedAt) >= takeInStall
 	changed = nowStalled != w.stalled
 	w.stalled = nowStalled
 	return nowStalled, changed
 }
 
-// watchTakeInStall reads the backlog every minute and raises take-in-stalled while it stands still, clearing it when it
-// moves: a watchdog's timer, as research 072 §2 keeps them until the health part leaves the controller.
-func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeInBacklog, error)) {
+// watchTakeInStall reads the backlog every minute and raises take-in-stalled while it stands with nothing acknowledged,
+// clearing it when it moves: a watchdog's timer, as research 072 §2 keeps them until the health part leaves the
+// controller.
+func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeInBacklog, error), acked func() uint64) {
 	w := &stallWatch{}
 	tick := time.NewTicker(time.Minute)
 	defer tick.Stop()
@@ -354,12 +378,15 @@ func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeIn
 		if err != nil {
 			continue
 		}
-		stalled, changed := w.look(b, time.Now())
+		stalled, changed := w.look(b, acked(), time.Now())
 		keeper := conditionsFrom
 		if !changed || keeper == nil {
 			continue
 		}
-		o := takeInStalledObservation(b, w.movedAt)
+		w.mu.Lock()
+		since := w.movedAt
+		w.mu.Unlock()
+		o := takeInStalledObservation(b, since)
 		if stalled {
 			doneKeysSay("done keys: " + o.Summary)
 			_, _ = keeper.Observe(ctx, o)