KV empty keys fix (#399)

Author: mtmk

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

@@ -50,6 +50,7 @@ internal class NatsKVWatcher<T> : IAsyncDisposable
     private ulong _sequenceConsumer;
     private string _consumer;
     private volatile NatsKVWatchSub<T>? _sub;
+    private INatsJSConsumer? _initialConsumer;
 
     public NatsKVWatcher(
         NatsJSContext context,
@@ -114,6 +115,12 @@ public NatsKVWatcher(
 
     public ChannelReader<NatsKVEntry<T>> Entries => _entryChannel.Reader;
 
+    internal INatsJSConsumer InitialConsumer
+    {
+        get => _initialConsumer ?? throw new InvalidOperationException("Consumer not initialized");
+        private set => _initialConsumer = value;
+    }
+
     internal string Consumer
     {
         get => Volatile.Read(ref _consumer);
@@ -136,10 +143,10 @@ public async ValueTask DisposeAsync()
         await _commandTask;
     }
 
-    internal ValueTask InitAsync()
+    internal async ValueTask InitAsync()
     {
         Consumer = NewNuid();
-        return CreatePushConsumer("init");
+        InitialConsumer = await CreatePushConsumer("init");
     }
 
     private ValueTask OnDisconnected(object? sender, NatsEventArgs args)
@@ -308,7 +315,7 @@ private async Task ConsumerCreateLoop()
         }
     }
 
-    private async ValueTask CreatePushConsumer(string origin)
+    private async ValueTask<INatsJSConsumer> CreatePushConsumer(string origin)
     {
         if (_debug)
         {
@@ -375,7 +382,7 @@ private async ValueTask CreatePushConsumer(string origin)
             config.OptStartSeq = sequence + 1;
         }
 
-        await _context.CreateOrUpdateConsumerAsync(
+        var consumer = await _context.CreateOrUpdateConsumerAsync(
             _stream,
             config,
             cancellationToken: _cancellationToken);
@@ -384,6 +391,8 @@ await _context.CreateOrUpdateConsumerAsync(
         {
             _logger.LogDebug(NatsKVLogEvents.NewConsumerCreated, "Created new consumer {Consumer} from {Origin}", Consumer, origin);
         }
+
+        return consumer;
     }
 
     private string NewNuid()

src/NATS.Client.KeyValueStore/NatsKVStore.cs

@@ -430,6 +430,9 @@ public async IAsyncEnumerable<string> GetKeysAsync(NatsKVWatchOpts? opts = defau
         // Type doesn't matter here, we're just using the watcher to get the keys
         await using var watcher = await WatchInternalAsync<int>(">", serializer: default, opts, cancellationToken);
 
+        if (watcher.InitialConsumer.Info.NumPending == 0)
+            yield break;
+
         while (await watcher.Entries.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
         {
             while (watcher.Entries.TryRead(out var entry))

tests/NATS.Client.KeyValueStore.Tests/GetKeysTest.cs

@@ -45,4 +45,39 @@ public async Task Get_keys_should_not_hang_when_there_are_deleted_keys()
 
         Assert.Equal(new List<string> { "k1", "k3" }, ks2);
     }
+
+    [Fact]
+    public async Task Get_keys_when_empty()
+    {
+        const string bucket = "b1";
+        var config = new NatsKVConfig(bucket) { History = 10 };
+
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+        var cancellationToken = cts.Token;
+
+        await using var server = NatsServer.StartJS();
+        await using var nats1 = server.CreateClientConnection();
+        var js1 = new NatsJSContext(nats1);
+        var kv1 = new NatsKVContext(js1);
+        var store1 = await kv1.CreateStoreAsync(config, cancellationToken: cancellationToken);
+
+        var count = 0;
+        await foreach (var k in store1.GetKeysAsync(cancellationToken: cancellationToken))
+        {
+            count++;
+        }
+
+        Assert.Equal(0, count);
+
+        await store1.PutAsync("k1", 1, cancellationToken: cancellationToken);
+        await store1.PutAsync("k2", 2, cancellationToken: cancellationToken);
+        await store1.PutAsync("k3", 3, cancellationToken: cancellationToken);
+
+        await foreach (var k in store1.GetKeysAsync(cancellationToken: cancellationToken))
+        {
+            count++;
+        }
+
+        Assert.Equal(3, count);
+    }
 }