From 94bb74966fd4c26ddbe0d4750a9c4d82ff5cc31c Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:01:18 +0100 Subject: [PATCH] =?UTF-8?q?Bound=20the=20take-in's=20two=20consumers=20to?= =?UTF-8?q?=2016=20keys=20in=20flight=20together,=208=20each=20(ADR=200544?= =?UTF-8?q?=20=C2=A71,=20its=20review)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/cmd/mesh-controller/take_in_consumer.go b/cmd/mesh-controller/take_in_consumer.go index 7592521b..767356cd 100644 --- a/cmd/mesh-controller/take_in_consumer.go +++ b/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 -// the store (issues 1945 and 1277: a burst of take-ins once filled it until it refused every connection). +// 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). -// 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)