Bound the take-in's two consumers to 16 keys in flight together, 8 each (ADR 0544 §1, its review)

Two consumers of 16 each allowed 32 take-ins on the store at once, twice
issue 1945's bound. A batch takes up to 16. Research 072: the take-ins on a
bounded pull consumer.

cmd/mesh-controller/take_in_consumer.go

@@ -19,12 +19,17 @@ 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 each are unacknowledged: the bound on
+// refused with a delay, and the bus delivers it again. At most startTakeIns keys of the two together are unacknowledged,
-// the store (issues 1945 and 1277: a burst of take-ins once filled it until it refused every connection).
+// 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).
 
-// startTakeIns is how many keys each consumer leaves unacknowledged at most: the take-ins that may wait on the store.
+// 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.
 const startTakeIns = 16
 
+// perConsumer is each consumer's share of startTakeIns.
+const perConsumer = 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.
 const takeInAckWait = 10 * time.Minute
@@ -76,7 +81,7 @@ func takeInConsumers(ctx context.Context, js jetstream.JetStream) []takeInConsum
 		}
 		c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, jetstream.ConsumerConfig{
 			Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver,
-			AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: startTakeIns,
+			AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: perConsumer,
 			Description: "the controller's take-in of done keys (novox/hq ADR 0544)",
 		})
 		if err != nil {
@@ -107,7 +112,7 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 	for _, c := range consumers {
 		bucket := c.bucket
 		read = append(read, c.consumer)
-		it, err := c.consumer.Messages(jetstream.PullMaxMessages(startTakeIns))
+		it, err := c.consumer.Messages(jetstream.PullMaxMessages(perConsumer))
 		if err != nil {
 			fmt.Printf("done keys: the take-in's consumer on %s is not read: %v\n", bucket, err)
 			continue
@@ -164,7 +169,7 @@ func takeInLoop(ctx context.Context, in <-chan takeInMsg, steps takeInSteps,
 		}
 		gather := time.NewTimer(takeInGather)
 	more:
-		for len(batch) < 2*startTakeIns {
+		for len(batch) < startTakeIns {
 			select {
 			case m := <-in:
 				batch = append(batch, m)