How to resume fetching messages after restart

[dvaldivia]

Hi, I'm creating a durable stream and then a Pull Consumer that fetches messages every X seconds, the first time the stream gets created, everything works fine, however, if I restart the consumer code, no message gets delivered, and I can see in the ConsumerInfo that some messages are waiting to be pulled but the consumer itself does nothing, unless I restart the nats-server

How I'm adding the stream

// check if stream exists
streamInfo, err := js.StreamInfo("audit-stream")

if err != nil {
	if errors.Is(err, nats.ErrStreamNotFound) {
		// Create a test stream if it doesn't exist
		streamInfo, err = js.AddStream(&nats.StreamConfig{
			Name:      "audit-stream",
			Storage:   nats.FileStorage,
			Retention: nats.WorkQueuePolicy,
			Subjects:  []string{"audit", "logs"},
		})
		if err != nil {
			log.Println(err)
		}
		fmt.Println("New Stream created")
	} else {
		log.Println(err)
		return
	}
} else {
	fmt.Println("Stream already exists")
}

How I'm consuming the stream

sub, err := js.PullSubscribe(topic, "audit-stream")
if err != nil {
	log.Println(topic, ":", err)
	return
}
// main loop, should pause for X minutes at a time (where x is the window size)
for {
	for {
		batch, err := sub.Fetch(1000)
		if err != nil {
			log.Println(err)
			continue;
		}
		// ... doing stuff with the message
		if err := m.Ack(); err != nil {
			log.Println(topic, ":", err)
		}
	}

stuck messages

➜ nats consumer info audit-stream logs-stream
Information for Consumer audit-stream > logs-stream created 2023-01-27T11:08:31-08:00

Configuration:

                Name: logs-stream
           Pull Mode: true
      Filter Subject: logs
      Deliver Policy: All
          Ack Policy: Explicit
            Ack Wait: 30s
       Replay Policy: Instant
     Max Ack Pending: 65,536
   Max Waiting Pulls: 512

State:

   Last Delivered Message: Consumer sequence: 8 Stream sequence: 8
     Acknowledgment floor: Consumer sequence: 8 Stream sequence: 8
         Outstanding Acks: 0 out of maximum 65,536
     Redelivered Messages: 0
     Unprocessed Messages: 21
            Waiting Pulls: 0 of maximum 512

at this point, this consumer doesn't receive any messages after restart.

How can I have this code resume fetching messages? I expect this code to have multiple concurrent consumers on the same topic, so I assume this could be a problem with more consumers.

Additional question: do I need to bind the stream when creating the consumer like

sub, err := js.PullSubscribe(topic, "audit-stream", nats.BindStream("audit-stream"))

[piotrpio]

Hello @dvaldivia

In general, when you use js.PullSubscribe() with durable consumer set (the second parameter), the consumer is either created (if it does not exist), or it should resume receiving messages from the stream starting after the last message acknowledged - or in your case (because you are setting your stream retention policy to WorkQueuePolicy), this would be the first matching message from the stream (with WorkQueuePolicy, each acknowledged message is removed from the stream after Ack).

I went through the example you provided an I noticed one inconsistency - when creating the subscription, you are setting the durable consumer name to audit-stream (same as stream name):

sub, err := js.PullSubscribe(topic, "audit-stream")

While this is certainly possible, I am wondering if that is not a mistake, since later on I see that you are checking a different consumer using nats cli (logs-stream):

nats consumer info audit-stream logs-stream

Also, since you are using WorkQueuePolicy as retention policy, if there is any other consumer receiving messages from logs subject when your audit-stream consumer is not fetching, those messages will be removed on acknowledgements (i.e. only one consumer receives messages from the stream with this retention policy).

I tried reproducing your issue with this simple program - https://gist.github.com/piotrpio/9f023c71a317d446acfc817352cc7e6a

After starting it, I tried publishing some messages on logs subject, stopping the program at various points, publishing some more messages and restarting the program. In no scenario was I able to reproduce the bug (get the audit-stream consumer stuck).

If your issue persists, it would be good if you provide your client and server versions (I am using [email protected] and [email protected]).

As to your other question:

Additional question: do I need to bind the stream when creating the consumer like

No, you don't need to use BindStream() - PullSubscribe() will do the stream lookup based on subject provided. However, if you want to skip the lookup (it involves an additional round trip to the server) and you know which stream has interest in your subject, you can use BindStream().

Let me know if it works for you.

[dvaldivia]

Thanks for the reply @piotrpio

the original Stream didn't had the WorkQueuePolicy, I added it while trying to experiment in trying to recover the messages, I already have a topic to which apps keep publishing messages, but I cannot get the stream to resume consuming the messages the new messages, if the original stream did not had WorkQueuePolicy would that cause a problem?

the stream name mismatch was because I original was trying to setup one stream per topic but then I realized I could ask for multiple topics under one stream and then have two separate js.PullScribe() one for audit-stream and one for log-stream, I assume this would be more efficient if it's not please let me know so I can create separare streams for each topic.

I'm using the latest and greatest versions of nats-server and the nats.go client, I was able to replicate the problem locally, which is what I used to submit this discussion.

After starting the code, I get that the stream already exist and when getting the consumer information

linfo, err := lib.GetInstance().Js.ConsumerInfo("audit-stream", "logs-stream")

I still see 21 pending messages, even if I comment the Retention: nats.WorkQueuePolicy that I added after the initial stream was created

 {
  "stream_name": "audit-stream",
  "name": "logs-stream",
  "created": "2023-01-27T19:08:31.104258Z",
  "config": {
   "durable_name": "logs-stream",
   "name": "logs-stream",
   "deliver_policy": "all",
   "ack_policy": "explicit",
   "ack_wait": 30000000000,
   "max_deliver": -1,
   "filter_subject": "logs",
   "replay_policy": "instant",
   "max_waiting": 512,
   "max_ack_pending": 65536,
   "num_replicas": 0
  },
  "delivered": {
   "consumer_seq": 8,
   "stream_seq": 8
  },
  "ack_floor": {
   "consumer_seq": 8,
   "stream_seq": 8
  },
  "num_ack_pending": 0,
  "num_redelivered": 0,
  "num_waiting": 0,
  "num_pending": 21
 }

[piotrpio]

Did you take a look at the gist I added to verify whether what I did is what you are trying to do? That would be very helpful, because I am not able to reproduce your issue.

In general, here's more or less what I did, step by step:

1. Create stream called audit-logs with file storage, with interest in 2 subjects: audit and logs (if it does not exist).
2. Run js.PullSubscribe("logs", "logs-consumer"). This will do 2 things:

• Create a consumer with the name logs-consumer (if it does not exist), which is tied to audit-logs stream created earlier. This is done implicitly, by looking up stream by subject logs (subjects cannot overlap between multiple streams, that guarantees that only one stream is available for a given subject). If the consumer already exists for this stream, it will be re-used.
• Create a core NATS subscription on a subject on which the messages will be delivered. In case of pull subscription, new messages will be pushed to this subject whenever you execute Fetch(). The name of this subject does not matter and is randomized by the client library.

3. Create an endless loop, which continuously calls Fetch(1000) on the previously created subscription. Each Fetch() will send a request to nats-server and return up to 1000 messages from the stream. Then print each message and send Ack().
4. Publish some messages to logs subject using nats CLI: nats pub logs "some msg"
5. Terminate the application. Server detects that the client has disconnected and terminates the client connection. This is visible in server logs as something like this:

[16941] 2023/01/31 13:06:58.612080 [DBG] 127.0.0.1:57319 - cid:65 - "v1.23.0:go" - Client connection closed: Client Closed

6. Publish some more messages to logs subject via nats cli.
7. Restart the application. It should skip creating both stream and consumer, but because I'm using js.PullSubscribe(), a brand new subscription to logs-consumer should be created.
8. Start fetching in a loop on the newly created subscription. Because all messages received previously were acknowledged, I am receiving only the messages which were published after the application was killed (and all messages published since).

It is important to distinguish streams and consumers:

• Stream is used to store messages published on given subjects (audit and logs). Using client library, stream has to be created with AddStream().
• Consumer is used to retrieve messages from the stream. There can be multiple consumer listening for the same stream, each of them will receive the same messages. Consuming messages from a stream does not affect the stream itself (unless you are using WorkQueuePolicy, but that does not seem to be required in your case).
• Consumers may be durable (their state is stored on the server and survives server restart) or ephemeral (deleted when not used). Because we supply the name of the consumer in PullSubscribe(), in our case it is durable.

Because a consumer it generally tied to a stream, not a subject, you could even have one consumer to retrieve messages from both logs and audit subjects. This would be achieved by having empty subject and BindStream() in PullSubscribe():

sub, err := js.PullSubscribe("", "logs-consumer-2", nats.BindStream("audit-stream"))

If it still does not work for you, it would be great if you could provide a more detailed instructions of how to reproduce. Additionally, server logs would be very helpful. To start nats-server with debug and trace logs, use -DV flags:

nats-server -DV -js

[zerlok]

Here is a similar code with the new jetstream API to test the stream / consumer behavior during restarts.

package main

import (
	"context"
	"fmt"
	"github.com/nats-io/nats.go/jetstream"
	"log"
	"time"

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

func main() {
	ctx := context.Background()

	nc, err := nats.Connect("127.0.0.1:4222")
	if err != nil {
		log.Fatal(err)
	}
	defer nc.Close()

	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatal(err)
	}

	_, err = js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
		Name:     "audit-stream",
		Subjects: []string{"audit.*"},
	})
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("Stream created or updated")

	sub, err := js.CreateOrUpdateConsumer(ctx, "audit-stream", jetstream.ConsumerConfig{
		Durable:        "audit-logs-consumer",
		FilterSubjects: []string{"audit.logs"},
	})
	if err != nil {
		log.Fatal(err)
	}

	for {
		batch, err := sub.Fetch(100, jetstream.FetchMaxWait(1*time.Second))
		if err != nil {
			log.Print(err)
			continue
		}

		for m := range batch.Messages() {
			// ... doing stuff with the message
			fmt.Println("msg: ", string(m.Data()))

			if err := m.Ack(); err != nil {
				log.Print(err)
				continue
			}
		}

		// pause for some time between fetches
		time.Sleep(10 * time.Second)
	}
}

[zerlok]

Faced the same issue and found the cause: I didn't mark messages as acknowledged msg.Ack() and I also set consumer config options MaxAckPending: 10 and AckWait: 1*time.Hour. Thus in a simple test with 10 messages my worker didn't receive next 10 messages after restarts.