How to detect a closed JetStream (due to server restart)?

[barafael]

We are using a WorkQueue consumer with StorageType::Memory and AckPolicy::Explicit. When the server restarts, it forgets the streams which existed before the shutdown. In a consumer, how can we observe that the server has shut down, so we can eventually recreate the stream when the server comes up again?

• We tried relying on the Stream interface, which states that the stream yields None when exhausted, but we do not observe this.
• We can observe the NATS connection state
  • we could lose messages currently queued in the consumer
  • we cannot detect that we have processed all of them, because of the earlier point

We DO observe an error on the stream, around the time of the reconnection:

2024-07-30T15:35:11.045783Z  WARN thing::module::consumer: pull consumer failed. error=pull request failed: request failed: missing field `stream_name` at line 1 column 129

[Jarema]

That looks like a bug. The consumer should recover on its own.
Will take a look.

[Jarema]

@barafael If the stream is gone, being in memory stream, after for example server restart, you should not loose messages, as workqueue stream will remove them only after consumer receives ack.

However, I wonder why you are using memory based consumer if this is a persistent consumer?

[barafael]

@Jarema apologies for the late reply.
Is memory based but persistent consumer not a legitimate use case? We don't want to pummel the disk.

I do think the stream trait implementation is misbehaving here. It's forcing me to have an extra task just to monitor the connection and might be slightly racy because I have to recreate the consumer.

[Jarema]

No worries.
I'm not sure I understand - how you want to achieve R1 in-memory persistence of a consumer?
If you restart the server - every state that was in-memory - is gone.
You have to either use more replicas (3, but of course if all replicas will die at the same time, you will loose data anyway), or file-based.

What do you think is misbehaving? Please elaborate a bit.

[barafael]

Thanks for the insights!

I accept the data loss in case of nats server restart.

Regarding the misbehaviour: see the reproduction above.
Here's how I understand the Stream trait: next() (or really, poll_next()), yields an item once ready (wrapped in a Some(.)). Great. Then we can poll it again. Yields another item. Great. Once it yields None, we know the stream is closed and may be dropped. This is how e.g. the TcpStream from tokio does it and I think it's the intended behaviour. If an error happens on a stream, I believe it may simply be yielded as well, but doesn't signify the end of stream.

From what I can observe here, once the server was disconnected for whatever reason, the jetstream Stream implementation loses the connection (doesn't reconnect). But also it doesn't close. It eventually sends some error though, which we can detect. In that case, we can reopen the connection. But that's overall not that great, I think...

[Jarema]

Stream does not have any concept of connection by itself.
All it does, is poll core nats subscriptions for messages (and in case of pull requests - also send new requests).
Because of that, it does not have to act in any specific way regarding the connections.
NATS Client subscriptions will be automatically resubsribed when reconnect happens.

And as client is meant to handle reconnecitons under the hood, we do not terminate neither Core NATS subscriptions, nor JetStream Consumers.

We had a concept of terminal errors, but with time we realized, that almost never its a good idea to terminale iterator - as so often things can recover on their own, or get fixed/recreted by other apps, operations etc.

And here is the difference between TCP Stream and NATS - NATS automatically handle reconnect, so it should not close streams on reconnections.

I'm happy to help you with specific errors you encounter. I would like to learn why you need a separate task.