When the ackwait is exceeded, the stream still displays messages, but consumers cannot retrieve them again

[birdycn]

Observed behavior

Expected behavior

how can i get the messages again？

Server and client version

nats-server: v2.11.4-RC.1

Host environment

No response

Steps to reproduce

No response

[birdycn]

    "account_details": [
        {
            "name": "$G",
            "id": "$G",
            "memory": 0,
            "storage": 657,
            "reserved_memory": 18446744073709552000,
            "reserved_storage": 18446744073709552000,
            "accounts": 0,
            "ha_assets": 0,
            "api": {
                "level": 0,
                "total": 58,
                "errors": 4
            },
            "stream_detail": [
                {
                    "name": "testQueueS",
                    "created": "2025-05-19T12:28:50.7387314Z",
                    "cluster": {
                        "leader": "NB36THQKG4GHVCTCNQ4QRC22ZG72OCGXGYH3CILQWTJE6U2FOHAP4RQJ"
                    },
                    "state": {
                        "messages": 13,
                        "bytes": 657,
                        "first_seq": 23,
                        "first_ts": "2025-05-19T12:29:24.6426466Z",
                        "last_seq": 82,
                        "last_ts": "2025-05-19T12:40:46.9248638Z",
                        "num_subjects": 1,
                        "num_deleted": 47,
                        "consumer_count": 1
                    }
                }
            ]
        }
    ],
    "total": 1

[neilalexander]

You will likely want to configure redeliveries if you are using AckWait. What is your consumer config?

[birdycn]

You will likely want to configure redeliveries if you are using AckWait. What is your consumer config?

func DefaultJetStreamConfig() *streamConfig {
	return &streamConfig{
		Retention: jetstream.WorkQueuePolicy,
		Discard:   jetstream.DiscardOld,
		Storage:   jetstream.FileStorage,
		MaxAge:    48 * time.Hour,
		MaxMsgs:   10240,
	}
}
func DefaultJetConsumerConfig() *consumerConfig {
	return &consumerConfig{
		AckPolicy:  jetstream.AckExplicitPolicy,
		MaxDeliver: 5,
		AckWait:    2 * time.Minute,
	}
}

func TestQueue(t *testing.T) {
	js, err := NewJetStreamClient("127.0.0.1:4222")
	assert.Equal(t, nil, err)

	const stream = "testQueueS"
	const consumer = "testQueueC"
	const topic = "test.queue"

	cfg := DefaultJetStreamConfig()
	cfg.Retention = jetstream.WorkQueuePolicy
	s, err := js.CreateStream(context.Background(), stream, []string{topic}, cfg)
	assert.Equal(t, nil, err)

	consumerCfg := DefaultJetConsumerConfig()
	consumerCfg.AckWait = 1 * time.Second

	c, err := s.CreateConsumer(context.Background(), consumer, consumerCfg)
	assert.Equal(t, nil, err)

	c.Subscribe(
		context.Background(),
		func(msg Message) error {
			time.Sleep(2*time.Second)
			msg.Ack()
			t.Logf("stream:%s,consumer:%s,topic:%s,msg:%s",
				stream, consumer, topic, string(msg.Data()))
			return nil
		},
		WithHandleError(func(err error) {
			t.Log(err)
		}))

	var i int
	for {
		time.Sleep(500 * time.Millisecond)

		if err := c.Publish(
			context.Background(), topic,
			[]byte(fmt.Sprintf("sequence:%d", i))); err != nil {
			t.Log(err)
			if errors.Is(err, nats.ErrConnectionClosed) {
				break
			}
		}
		i++
	}
}

[souravagrawal]

AFAIK, even after the consumer application reaches the maximum number of redelivery attempts, if it still fails to acknowledge the message successfully , the message will no longer be retried. However, it will continue to persist in the stream unless explicitly removed.
To prevent such unacknowledged messages from lingering indefinitely, it's important to configure a Time-To-Live (TTL) on the stream. This ensures stale or unprocessable messages are automatically cleaned up over time.