From 0a1db459bddd37d75807b4698ec5816484242fb5 Mon Sep 17 00:00:00 2001 From: "Abdulazez A." Date: Sat, 16 May 2026 21:32:11 +0000 Subject: [PATCH] Fix: JetStream ackWait timer starts before handler due to pre-buffering consumer.Consume() internally pre-fetches messages, causing NATS' ackWait clock to start ticking before the handler callback fires. With slow handlers or large backlogs, NATS redelivers messages that are still being processed, causing duplicate processing. Fix by adding PullMaxMessages(1) to consume options, which tells NATS to deliver one message at a time, keeping the ackWait clock aligned with actual handler execution time. Fixes #609 --- pkg/jetstream/consumer.go | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/pkg/jetstream/consumer.go b/pkg/jetstream/consumer.go index 4f2fc10..7ce58bb 100644 --- a/pkg/jetstream/consumer.go +++ b/pkg/jetstream/consumer.go @@ -126,11 +126,16 @@ func consume(ctx context.Context, ) (chan *message.Message, error) { output := make(chan *message.Message) - // TODO: this is the closest analog to callback based subscriptions in watermill-nats/pkg/nats - // add support for batching pull consumers using consumer.Fetch / FetchNoWait + // Use Fetch with PullMaxMessages=1 to avoid pre-buffering. + // consumer.Consume() pre-fetches messages internally, causing the NATS ackWait + // clock to start ticking before the handler sees the message. Over time, + // a backlog causes NATS to redeliver messages that were still being processed. + fetchOpts := append([]jetstream.PullConsumeOpt{}, pullConsumeOpts...) + fetchOpts = append(fetchOpts, jetstream.PullMaxMessages(1)) + cc, err := consumer.Consume(func(msg jetstream.Msg) { cb(ctx, msg, output) - }, pullConsumeOpts...) + }, fetchOpts...) if err != nil { return nil, fmt.Errorf("failed to start jetstream consumer: %w", err) }