Issue with binary deserialization of reply for NatsConnection.RequestAsync

[robertmircea]

I am trying a simple scenario of request/reply using MemoryPack binary serialization. Unfortunately, the library throws while trying to deserialize NATS (not Jetstream) response. It seems some kind of malformed packet deserialization. Nats Jetstream deserialization works fine with MemoryPack in 2.0.1.

Versions:

• Server: 2.10.6
• Client:
  <PackageReference Include="MemoryPack" Version="1.10.0" />
<PackageReference Include="NATS.Net" Version="2.0.1" />

The code to reproduce:

using System.Buffers;
using MemoryPack;
using NATS.Client.Core;

var opts = NatsOpts.Default with { SerializerRegistry = new NatsMemoryPackContextSerializerRegistry()};
await using var nats = new NatsConnection(opts);

var subscription = Task.Run(async () =>
{
    await foreach (var msg in nats.SubscribeAsync<Request>("tst"))
    {
        await msg.ReplyAsync(new Response() { Status = "123" });
        break;
    }
});

var response = await nats.RequestAsync<Request, Response>("tst", new Request("123"));
Console.WriteLine("Reply received: {0}", response.Data?.Status);
await subscription;


[MemoryPackable(SerializeLayout.Explicit)]
public partial class Request(string id)
{
    [MemoryPackOrder(0)] public string Id { get; set; } = id;
}

[MemoryPackable(SerializeLayout.Explicit)]
public partial class Response
{
    [MemoryPackOrder(0)] public string Status { get; set; }
}
public sealed class NatsMemoryPackContextSerializerRegistry : INatsSerializerRegistry
{
    public INatsSerialize<T> GetSerializer<T>() => NatsMemoryPackContextSerializer<T>.Default;
    public INatsDeserialize<T> GetDeserializer<T>() => NatsMemoryPackContextSerializer<T>.Default;
}
public sealed class NatsMemoryPackContextSerializer<T> : INatsSerializer<T>
{
    public static readonly INatsSerializer<T> Default = new NatsMemoryPackContextSerializer<T>();
    
    public void Serialize(IBufferWriter<byte> bufferWriter, T value) => MemoryPackSerializer.Serialize(bufferWriter, value);

    public T? Deserialize(in ReadOnlySequence<byte> buffer) => MemoryPackSerializer.Deserialize<T>(buffer);
}

Exception thrown:

Unhandled exception. MemoryPack.MemoryPackSerializationException: Response property count is 1 but binary's header maked as 123, can't deserialize about versioning.
   at MemoryPack.MemoryPackSerializationException.ThrowInvalidPropertyCount(Type type, Byte expected, Byte actual)
   at Response.MemoryPack.IMemoryPackable<Response>.Deserialize(MemoryPackReader& reader, Response& value) in /Users/robert/Sandbox/RequestAsyncBug/MemoryPack.Generator/MemoryPack.Generator.MemoryPackGenerator/Response.MemoryPackFormatter.g.cs:line 108
   at MemoryPack.Formatters.MemoryPackableFormatter`1.Deserialize(MemoryPackReader& reader, T& value)
   at MemoryPack.MemoryPackSerializer.Deserialize[T](ReadOnlySequence`1& buffer, T& value, MemoryPackSerializerOptions options)
   at MemoryPack.MemoryPackSerializer.Deserialize[T](ReadOnlySequence`1& buffer, MemoryPackSerializerOptions options)
   at NatsMemoryPackContextSerializer`1.Deserialize(ReadOnlySequence`1& buffer) in /Users/robert/Sandbox/RequestAsyncBug/Program.cs:line 47
   at NatsMemoryPackContextSerializer`1.NATS.Client.Core.INatsDeserialize<T>.Deserialize(ReadOnlySequence`1& buffer)
   at NATS.Client.Core.NatsMsg`1.Build(String subject, String replyTo, Nullable`1& headersBuffer, ReadOnlySequence`1& payloadBuffer, INatsConnection connection, NatsHeaderParser headerParser, INatsDeserialize`1 serializer)
   at NATS.Client.Core.NatsSub`1.ReceiveInternalAsync(String subject, String replyTo, Nullable`1 headersBuffer, ReadOnlySequence`1 payloadBuffer)
   at NATS.Client.Core.NatsSubBase.ReceiveAsync(String subject, String replyTo, Nullable`1 headersBuffer, ReadOnlySequence`1 payloadBuffer)
   at NATS.Client.Core.NatsSubBase.DisposeAsync()
   at NATS.Client.Core.NatsConnection.RequestAsync[TRequest,TReply](String subject, TRequest data, NatsHeaders headers, INatsSerialize`1 requestSerializer, INatsDeserialize`1 replySerializer, NatsPubOpts requestOpts, NatsSubOpts replyOpts, CancellationToken cancellationToken)
   at Program.<Main>$(String[] args) in /Users/robert/Sandbox/RequestAsyncBug/Program.cs:line 20
   at Program.<Main>$(String[] args) in /Users/robert/Sandbox/RequestAsyncBug/Program.cs:line 22
   at Program.<Main>(String[] args)

Process finished with exit code 134.

Nats server trace log:

[53888] 2023/12/10 14:17:58.089813 [TRC] [::1]:53074 - cid:289 - <<- [CONNECT {"echo":true,"verbose":false,"pedantic":false,"tls_required":false,"user":"user","pass":"[REDACTED]","name":"NATS .Net Client","lang":"C#","version":"1.0.0","protocol":1,"new_account":null,"headers":true,"no_responders":false}]
[53888] 2023/12/10 14:17:58.089932 [TRC] [::1]:53074 - cid:289 - <<- [PING]
[53888] 2023/12/10 14:17:58.089936 [TRC] [::1]:53074 - cid:289 - ->> [PONG]
[53888] 2023/12/10 14:17:58.091833 [TRC] [::1]:53074 - cid:289 - <<- [PING]
[53888] 2023/12/10 14:17:58.091844 [TRC] [::1]:53074 - cid:289 - ->> [PONG]
[53888] 2023/12/10 14:17:58.095355 [TRC] [::1]:53074 - cid:289 - <<- [SUB tst 1]
[53888] 2023/12/10 14:17:58.095386 [TRC] [::1]:53074 - cid:289 - <<- [SUB _INBOX.rF8IXS6NqsN72s9i1zzRsR.* 2]
[53888] 2023/12/10 14:17:58.101094 [TRC] [::1]:53074 - cid:289 - <<- [PUB tst _INBOX.rF8IXS6NqsN72s9i1zzRsR.rF8IXS6NqsN7Uu9i1zzRsR 12]
[53888] 2023/12/10 14:17:58.101110 [TRC] [::1]:53074 - cid:289 - <<- MSG_PAYLOAD: ["\x01\xfc\xff\xff\xff\x03\x00\x00\x00123"]
[53888] 2023/12/10 14:17:58.101133 [TRC] [::1]:53074 - cid:289 - ->> [MSG tst 1 _INBOX.rF8IXS6NqsN72s9i1zzRsR.rF8IXS6NqsN7Uu9i1zzRsR 12]
[53888] 2023/12/10 14:17:58.101368 [TRC] [::1]:53074 - cid:289 - ->> [MSG _INBOX.rF8IXS6NqsN72s9i1zzRsR.rF8IXS6NqsN7Uu9i1zzRsR 2 39]
[53888] 2023/12/10 14:17:58.111674 [TRC] [::1]:53074 - cid:289 - <<- [PUB _INBOX.rF8IXS6NqsN72s9i1zzRsR.rF8IXS6NqsN7Uu9i1zzRsR 12]
[53888] 2023/12/10 14:17:58.111690 [TRC] [::1]:53074 - cid:289 - <<- MSG_PAYLOAD: ["\x01\xfc\xff\xff\xff\x03\x00\x00\x00123"]
[53888] 2023/12/10 14:17:58.111703 [TRC] [::1]:53074 - cid:289 - ->> [MSG _INBOX.rF8IXS6NqsN72s9i1zzRsR.rF8IXS6NqsN7Uu9i1zzRsR 2 12]
[53888] 2023/12/10 14:17:58.112577 [TRC] [::1]:53074 - cid:289 - <<- [UNSUB 1]
[53888] 2023/12/10 14:17:58.112584 [TRC] [::1]:53074 - cid:289 - <-> [DELSUB 1]
[53888] 2023/12/10 14:17:58.114625 [DBG] [::1]:53074 - cid:289 - Client connection closed: Read Error
[53888] 2023/12/10 14:17:58.114653 [TRC] [::1]:53074 - cid:289 - <-> [DELSUB 2]

[mtmk]

The example above works for me. Are there any project settings I need to use?

[robertmircea]

Not specifically... It should be pretty straightforward. Please let me know if you can reproduce it.

[mtmk]

here's what I'm trying and it's working for me:

https://github.com/mtmk/natsbug1

feel free to add anything I missed, added you to that repo.

(also just to note I'm on windows11 + net-8.0 sdk, not that it should make any difference)

[mtmk]

fyi, ran on a mac that worked too.

[robertmircea]

I thing I found the issue: Is failing because I already have a subject tst part of a Jetstream stream and it seems I cannot use the same subject for plain Nats messages. If I change the subject name to something different than tst it works on my side too.

❯ nats stream info bsp-plugin
Information for Stream bsp-plugin created 2023-12-10 00:10:18

              Subjects: tst
              Replicas: 1
               Storage: File

Options:

             Retention: WorkQueue
       Acknowledgments: true
        Discard Policy: Old
      Duplicate Window: 2m0s
            Direct Get: true
     Allows Msg Delete: true
          Allows Purge: true
        Allows Rollups: false

Limits:

      Maximum Messages: unlimited
   Maximum Per Subject: unlimited
         Maximum Bytes: unlimited
           Maximum Age: unlimited
  Maximum Message Size: unlimited
     Maximum Consumers: unlimited

State:

              Messages: 2
                 Bytes: 124 B
        First Sequence: 117 @ 2023-12-11 16:28:14 UTC
         Last Sequence: 118 @ 2023-12-11 16:30:25 UTC
      Active Consumers: 2
    Number of Subjects: 1