Is it safe to ack/nack messages in a different thread? #249
-
|
Is it safe to acknowledge or negatively acknowledge (ack/nack) messages in a different thread than the one running the await foreach loop that consumes messages from a NATS server? The below provided demo code appears to function correctly, but in my production code, I encounter an issue with message acknowledgment. The consumer loop makes a call to an external service and does not wait for result, and when the result of processing is received (in a different thread through a different endpoint), the application attempts to acknowledge the NATS message using |
Beta Was this translation helpful? Give feedback.
Replies: 2 comments 11 replies
-
|
It should work. Another issue might be caused by messages buffered by the consume call, currently 1000. You could try reducing it e.g. |
Beta Was this translation helpful? Give feedback.
-
|
|
Beta Was this translation helpful? Give feedback.
-
|
any luck reproducing it? your code snippet above worked for me with a million messages. can you also see if you can publish anything when in the consume loop? (like a simple publish to a random subject and see if you receive the message) edit: Can you reproduce the issue locally? If there aren't huge amount of messages flying you can run You should see the ack as something like |
Beta Was this translation helpful? Give feedback.
-
|
I've managed to reproduce the issue in a demo app which I attach here. I've extracted the relevant classes from the project and simulated everything with MemoryPack custom serializer. Maybe is worth mentioning: the stream is created with retention=WorkQueue, Duplicate Window=2m, Storage=File and Subjects="any-test.outgoing.>"
The trace log from Nats server: |
Beta Was this translation helpful? Give feedback.
-
|
I think I found it. It's a serialization bug in the library. sorry about that. Problem is here: In @robertmircea thank you so much for the repro! edit: // <PackageReference Include="NATS.Net" Version="2.0.0" />
// <PackageReference Include="MemoryPack" Version="1.10.0" />
using System.Buffers;
using MemoryPack;
using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
var nats = new NatsConnection(new NatsOpts { SerializerRegistry = new NatsMessagePackContextSerializerRegistry() });
var js = new NatsJSContext(nats);
try { await js.DeleteStreamAsync("S1"); } catch { /*ignore*/ }
await js.CreateStreamAsync(new StreamConfig("S1", new[]{ "s1" }));
await js.PublishAsync("s1", new GatewayMessage(777));
var c = await js.CreateConsumerAsync("S1", new ConsumerConfig("C1"));
await (await c.NextAsync<GatewayMessage>())!.Value.AckAsync();
await c.RefreshAsync();
Console.WriteLine(c.Info.NumAckPending == 0 ? "PASS" : "FAIL");
[MemoryPackable(SerializeLayout.Explicit)]
public partial class GatewayMessage(int messageId)
{
[MemoryPackOrder(0)] public int MessageId { get; set; } = messageId;
}
public sealed class NatsMessagePackContextSerializer<T> : INatsSerializer<T>
{
public void Serialize(IBufferWriter<byte> bufferWriter, T value) => MemoryPackSerializer.Serialize(bufferWriter, value);
public T? Deserialize(in ReadOnlySequence<byte> buffer) => MemoryPackSerializer.Deserialize<T>(buffer);
}
public sealed class NatsMessagePackContextSerializerRegistry : INatsSerializerRegistry
{
public INatsSerialize<T> GetSerializer<T>() => new NatsMessagePackContextSerializer<T>();
public INatsDeserialize<T> GetDeserializer<T>() => new NatsMessagePackContextSerializer<T>();
} |
Beta Was this translation helpful? Give feedback.
-
|
Great that you found the issue! Is there an upcoming release of a new NuGet package that includes the necessary fix? |
Beta Was this translation helpful? Give feedback.
-
|
Yes, I'm working on 2.0.1 it should be out soon. |
Beta Was this translation helpful? Give feedback.
-
|
@robertmircea 2.0.1 is released |
Beta Was this translation helpful? Give feedback.
-
|
Great... I've already tested it. Works flawlessly until now! Thanks a lot for quick fixing. |
Beta Was this translation helpful? Give feedback.
I think I found it. It's a serialization bug in the library. sorry about that.
Problem is here:
In
msg.AckAsync()we're still (wrongly) using the custom serializer serializing+ACKinstead we should use default.@robertmircea thank you so much for the repro!
edit:
minimal example: