async-nats: Pull consumer issue: "pull request failed: request failed: missing field `stream_name` at line 1 column 129"

[simon-connektica]

Observed behavior

We are creating a Stream, then creating a Consumer and finally listening to messages. We are getting this error in our logs.

{"message":"Nats error: 'pull request failed: request failed: missing field `stream_name` at line 1 column 129'"}

Expected behavior

From the error we are getting, it seems like the server is sometimes not getting the stream name from the request made by the consumer? It feels like it might be a library issue because we just pass the stream name in get_or_create_stream, then use get_or_create_consumer from the stream and use the messages() async iterator.

This error seems to be intermittent, not always occurring. I included in the code sample below the select! block we are doing in case the future being cancelled has anything to do with this.

Server and client version

Client version: async-nats version = "0.33.0" Server version: [1] 2024/01/24 15:07:03.862164 [INF] Version: 2.10.9

Host environment

Linux, Kubernetes on AWS EKS, amd64 arch

Steps to reproduce

The code looks roughly like this:

let jetstream = async_nats::jetstream::new(self.connection.clone());

let stream_name = "worker_logs";
let stream_subject = "*.worker_logs.*";
let log_recorder = "log_recorder";

let stream = jetstream
    .get_or_create_stream(StreamConfig {
        name: stream_name.into(),
        subjects: vec![stream_subject.into()],
        max_messages: 10_000,
        ..Default::default()
    })
    .await?;

let consumer = stream
    .get_or_create_consumer(
        log_recorder,
        PullConfig {
            name: Some(log_recorder.into()),
            durable_name: Some(log_recorder.into()),
            ..Default::default()
        },
    )
    .await?;

let mut stream = consumer.messages().await?;
let mut ticker = tokio::time::interval(Duration::from_secs(1));

loop {
    select! {
        Some(msg) = stream.next() => {
            let msg = msg.map_err(|e| eyre!("log_recorder recv failure: {e}"))?;
            // ...
            msg.ack().await.map_err(|e| eyre!("log_recorder ACK failure {e}"))?;
        }
        _ = ticker.tick() => {}
    }
}

[Jarema]

Thanks for filling the issue.

I was seeing that one too, however I think it was fixed in https://github.com/nats-io/nats.rs/pull/1178/files?diff=split&w=1 I also do not see those errors in last days in chaos testing, however I will double check.

Feel encouraged to test main branch if you can to confirm on your end.

[simon-connektica]

We encountered the error in production so I'd prefer waiting for 0.34 before testing it if you don't mind. Thanks for the fix!

[Jarema]

@simon-connektica the o.34 has been released. Can you confirm your're no longer seeing the issue?