Key/Value Store fixes

sandbox/Example.KeyValueStore.Watcher/Program.cs

@@ -12,7 +12,7 @@
 
 var store = await kv.CreateStoreAsync("e1");
 
-await foreach (var entry in store.WatchAllAsync<int>())
+await foreach (var entry in store.WatchAsync<int>())
 {
     Console.WriteLine($"[RCV] {entry}");
 }

src/NATS.Client.JetStream/Models/PubAckResponse.cs

@@ -24,7 +24,7 @@ public record PubAckResponse
     [System.Text.Json.Serialization.JsonPropertyName("seq")]
     [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)]
     [System.ComponentModel.DataAnnotations.Range(0D, 18446744073709552000D)]
-    public long Seq { get; set; } = default!;
+    public ulong Seq { get; set; } = default!;
 
     /// <summary>
     /// Indicates that the message was not stored due to the Nats-Msg-Id header and duplicate tracking

src/NATS.Client.JetStream/Models/StoredMessage.cs

@@ -16,7 +16,7 @@ public record StoredMessage
     [System.Text.Json.Serialization.JsonPropertyName("seq")]
     [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.Never)]
     [System.ComponentModel.DataAnnotations.Range(0D, 18446744073709552000D)]
-    public long Seq { get; set; } = default!;
+    public ulong Seq { get; set; } = default!;
 
     /// <summary>
     /// The base64 encoded payload of the message body

src/NATS.Client.JetStream/Models/StreamMsgGetRequest.cs

@@ -11,7 +11,7 @@ public record StreamMsgGetRequest
     /// </summary>
     [System.Text.Json.Serialization.JsonPropertyName("seq")]
     [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)]
-    public int Seq { get; set; } = default!;
+    public ulong Seq { get; set; } = default!;
 
     /// <summary>
     /// Retrieves the last message for a given subject, cannot be combined with seq

src/NATS.Client.JetStream/NatsJSContext.cs

@@ -106,6 +106,14 @@ public async ValueTask<PubAckResponse> PublishAsync<T>(
         throw new NatsJSException("No response received");
     }
 
+    public ValueTask<PubAckResponse> PublishAsync(
+        string subject,
+        string? msgId = default,
+        NatsHeaders? headers = default,
+        NatsPubOpts? opts = default,
+        CancellationToken cancellationToken = default) =>
+        PublishAsync<object?>(subject, default, msgId, headers, opts, cancellationToken);
+
     internal string NewInbox() => NatsConnection.NewInbox(Connection.Opts.InboxPrefix);
 
     internal async ValueTask<TResponse> JSRequestResponseAsync<TRequest, TResponse>(

src/NATS.Client.JetStream/NatsJSException.cs

@@ -62,14 +62,14 @@ public class NatsJSDuplicateMessageException : NatsJSException
     /// Create JetStream duplicate message exception.
     /// </summary>
     /// <param name="sequence">The duplicate sequence number.</param>
-    public NatsJSDuplicateMessageException(long sequence)
+    public NatsJSDuplicateMessageException(ulong sequence)
         : base($"Duplicate of {sequence}") =>
         Sequence = sequence;
 
     /// <summary>
     /// The duplicate sequence number.
     /// </summary>
-    public long Sequence { get; }
+    public ulong Sequence { get; }
 }
 
 /// <summary>

src/NATS.Client.JetStream/NatsJSStream.cs

@@ -1,4 +1,5 @@
 using NATS.Client.Core;
+using NATS.Client.JetStream.Internal;
 using NATS.Client.JetStream.Models;
 
 namespace NATS.Client.JetStream;
@@ -168,7 +169,7 @@ public async ValueTask RefreshAsync(CancellationToken cancellationToken = defaul
             request: null,
             cancellationToken).ConfigureAwait(false);
 
-    public ValueTask<NatsMsg<T?>?> GetDirectAsync<T>(string subject, INatsSerializer? serializer = default, CancellationToken cancellationToken = default)
+    public ValueTask<NatsMsg<T?>?> GetDirectAsync<T>(StreamMsgGetRequest request, INatsSerializer? serializer = default, CancellationToken cancellationToken = default)
     {
         NatsSubOpts? subOpts;
         if (serializer != null)
@@ -180,10 +181,12 @@ public async ValueTask RefreshAsync(CancellationToken cancellationToken = defaul
             subOpts = default;
         }
 
-        return _context.Connection.RequestAsync<object, T>(
-            subject: $"{_context.Opts.Prefix}.DIRECT.GET.{_name}.{subject}",
-            data: default,
-            requestOpts: default,
+        var requestSubject = $"{_context.Opts.Prefix}.DIRECT.GET.{_name}";
+
+        return _context.Connection.RequestAsync<StreamMsgGetRequest, T>(
+            subject: requestSubject,
+            data: request,
+            requestOpts: new NatsPubOpts { Serializer = NatsJSJsonSerializer.Default },
             replyOpts: subOpts,
             cancellationToken: cancellationToken);
     }

src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs

@@ -25,7 +25,7 @@ public NatsKVWatchCommandMsg()
     public NatsJSMsg<T?> Msg { get; init; } = default;
 }
 
-internal class NatsKVWatcher<T> : INatsKVWatcher<T>
+internal class NatsKVWatcher<T> : IAsyncDisposable
 {
     private readonly ILogger _logger;
     private readonly bool _debug;
@@ -38,7 +38,7 @@ internal class NatsKVWatcher<T> : INatsKVWatcher<T>
     private readonly string _filter;
     private readonly NatsConnection _nats;
     private readonly Channel<NatsKVWatchCommandMsg<T>> _commandChannel;
-    private readonly Channel<NatsKVEntry<T?>> _entryChannel;
+    private readonly Channel<NatsKVEntry<T>> _entryChannel;
     private readonly Channel<string> _consumerCreateChannel;
     private readonly Timer _timer;
     private readonly int _hbTimeout;
@@ -96,11 +96,10 @@ public NatsKVWatcher(
             Timeout.Infinite,
             Timeout.Infinite);
 
-        // Channel size 1 is enough because we want backpressure to go all the way to the subscription
-        // so that we get most accurate view of the stream. We can keep them as 1 until we find a case
-        // where it's not enough due to performance for example.
-        _commandChannel = Channel.CreateBounded<NatsKVWatchCommandMsg<T>>(1);
-        _entryChannel = Channel.CreateBounded<NatsKVEntry<T?>>(1);
+        // Keep the channel size large enough to avoid blocking the connection
+        // TCP receiver thread in case other operations are in-flight.
+        _commandChannel = Channel.CreateBounded<NatsKVWatchCommandMsg<T>>(1000);
+        _entryChannel = Channel.CreateBounded<NatsKVEntry<T>>(1000);
 
         // A single request to create the consumer is enough because we don't want to create a new consumer
         // back to back in case the consumer is being recreated due to a timeout and a mismatch in consumer
@@ -115,7 +114,7 @@ public NatsKVWatcher(
         _commandTask = Task.Run(CommandLoop);
     }
 
-    public ChannelReader<NatsKVEntry<T?>> Entries => _entryChannel.Reader;
+    public ChannelReader<NatsKVEntry<T>> Entries => _entryChannel.Reader;
 
     internal string Consumer
     {
@@ -126,6 +125,12 @@ internal string Consumer
     public async ValueTask DisposeAsync()
     {
         _nats.ConnectionDisconnected -= OnDisconnected;
+
+        if (_sub != null)
+        {
+            await _sub.DisposeAsync();
+        }
+
         _consumerCreateChannel.Writer.TryComplete();
         _commandChannel.Writer.TryComplete();
         _entryChannel.Writer.TryComplete();
@@ -174,7 +179,7 @@ private async Task CommandLoop()
 
                                     operation = operationValues[0] switch
                                     {
-                                        "DEL" => NatsKVOperation.Delete,
+                                        "DEL" => NatsKVOperation.Del,
                                         "PURGE" => NatsKVOperation.Purge,
                                         _ => operation,
                                     };
@@ -219,17 +224,17 @@ private async Task CommandLoop()
                                         continue;
                                     }
 
-                                    if (_opts.IgnoreDeletes && operation is NatsKVOperation.Delete or NatsKVOperation.Purge)
+                                    if (_opts.IgnoreDeletes && operation is NatsKVOperation.Del or NatsKVOperation.Purge)
                                     {
                                         continue;
                                     }
 
                                     var delta = (long)metadata.NumPending;
 
-                                    var entry = new NatsKVEntry<T?>(_bucket, key)
+                                    var entry = new NatsKVEntry<T>(_bucket, key)
                                     {
                                         Value = msg.Data,
-                                        Revision = (long)metadata.Sequence.Stream,
+                                        Revision = metadata.Sequence.Stream,
                                         Operation = operation,
                                         Created = metadata.Timestamp,
                                         Delta = delta,

src/NATS.Client.KeyValueStore/NatsKVEntry.cs

@@ -26,7 +26,7 @@ public readonly record struct NatsKVEntry<T>(string Bucket, string Key)
     /// <summary>
     /// A unique sequence for this value.
     /// </summary>
-    public long Revision { get; init; } = default;
+    public ulong Revision { get; init; } = default;
 
     /// <summary>
     /// Distance from the latest value.

src/NATS.Client.KeyValueStore/NatsKVException.cs

@@ -26,3 +26,36 @@ public NatsKVException(string message, Exception exception)
     {
     }
 }
+
+public class NatsKVKeyDeletedException : NatsKVException
+{
+    public NatsKVKeyDeletedException(ulong revision)
+        : base("Key was deleted") =>
+        Revision = revision;
+
+    public ulong Revision { get; }
+}
+
+public class NatsKVWrongLastRevisionException : NatsKVException
+{
+    public NatsKVWrongLastRevisionException()
+        : base("Wrong last revision")
+    {
+    }
+}
+
+public class NatsKVCreateException : NatsKVException
+{
+    public NatsKVCreateException()
+        : base("Can't create entry")
+    {
+    }
+}
+
+public class NatsKVKeyNotFoundException : NatsKVException
+{
+    public NatsKVKeyNotFoundException()
+        : base("Key not found")
+    {
+    }
+}

src/NATS.Client.KeyValueStore/NatsKVOpts.cs

@@ -48,3 +48,17 @@ public record NatsKVWatchOpts
     /// </summary>
     public bool MetaOnly { get; init; } = false;
 }
+
+public record NatsKVDeleteOpts
+{
+    public bool Purge { get; init; }
+
+    public ulong Revision { get; init; }
+}
+
+public record NatsKVPurgeOpts
+{
+    public static readonly NatsKVPurgeOpts Default = new() { DeleteMarkersThreshold = TimeSpan.FromMinutes(30) };
+
+    public TimeSpan DeleteMarkersThreshold { get; init; }
+}