Acknowledging a message after consumer is deleted does not fail, but the message is not removed from the stream either

[zhulik]

Observed behavior

I'm pretty sure this is some very edge case, but it took me a few hours to figure out what was wrong. I wrote a simple program to reproduce the behavior, see steps to reproduce.

When using a jetstream stream with WorkQueuePolicy retention policy, when a consumer is deleted before a message received from it is acknowledged:

• Ack() does not fail.
• Consumer keeps receiving the same message over and over again when recreated.

The behavior has not been tested with plain NATS or other retention policies.

Expected behavior

One of:

• Ack() fails, so it's clearly visible that something goes off.
• Ack() does not fail, but the message is deleted from the stream.

Server and client version

• github.com/nats-io/nats.go v1.38.0
• nats-server: v2.10.25, Docker image nats:latest, it was latest when pulled anyway.

Host environment

OS: Arch Linux
CPU: AMD Ryzen 7 PRO 5850U and AMD Ryzen 5 5600
Docker:

Client:
 Version:           27.3.1
 API version:       1.47
 Go version:        go1.23.1
 Git commit:        ce1223035a
 Built:             Wed Sep 25 14:43:43 2024
 OS/Arch:           linux/amd64
 Context:           default

Server:
 Engine:
  Version:          27.3.1
  API version:      1.47 (minimum version 1.24)
  Go version:       go1.23.1
  Git commit:       41ca978a0a
  Built:            Wed Sep 25 14:43:43 2024
  OS/Arch:          linux/amd64
  Experimental:     false
 containerd:
  Version:          v2.0.2
  GitCommit:        c507a0257ea6462fbd6f5ba4f5c74facb04021f4.m
 runc:
  Version:          1.2.4
  GitCommit:        
 docker-init:
  Version:          0.19.0
  GitCommit:        de40ad0

Steps to reproduce

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/samber/lo"

	"github.com/google/uuid"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

const (
	stream  = "EVENTS"
	subject = "events"
)

var (
	nc = lo.Must(nats.Connect(nats.DefaultURL))
	js = lo.Must(jetstream.New(nc))

	ctx = context.Background()
)

func consumer() {
	for {
		cons := lo.Must(js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{
			Durable:       "consumer",
			FilterSubject: subject,
		}))

		msg := lo.Must(cons.Next())

		// If consumer is deleted before the message is acknowledged
		// Ack does not fail, but the message is not removed from the stream either.
		lo.Must0(js.DeleteConsumer(ctx, stream, "consumer"))

		fmt.Printf("Received message: %s\n", string(msg.Data()))
		lo.Must0(msg.Ack())
	}
}

func producer() {
	for {
		msg := uuid.NewString()

		lo.Must(js.Publish(context.Background(), subject, []byte(msg)))

		fmt.Printf("Message published %s\n", msg)

		time.Sleep(100 * time.Millisecond)
	}
}

func main() {
	lo.Must(js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
		Name:      stream,
		Retention: jetstream.WorkQueuePolicy,
		Subjects:  []string{subject},
	}))

	go consumer()
        producer()
}

Run the program, observe the consumer to log the same message over and over again, no panics.
Remove lo.Must0(js.DeleteConsumer(ctx, stream, "consumer")), observe the consumer to log different meessages.

[Jarema]

Hey.
Thanks for detailed report.

Ack will not return an error, as it is just a publish.

You should try using AckSync.
In such case, you should get and error if the consumer is gone.