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)