Fix Object Store Rename

src/NATS.Client.ObjectStore/NatsObjStore.cs

@@ -328,6 +328,13 @@ public async ValueTask<ObjectMetadata> UpdateMetaAsync(string key, ObjectMetadat
 
         await PublishMeta(info, cancellationToken);
 
+        // Check if the name changed and purge the old meta record
+        if (key != meta.Name)
+        {
+            var metaSubject = GetMetaSubject(key);
+            await _stream.PurgeAsync(new StreamPurgeRequest { Filter = metaSubject }, cancellationToken).ConfigureAwait(false);
+        }
+
         return info;
     }
 

tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs

@@ -554,4 +554,72 @@ public async Task Put_multiple_times_with_activity()
         await store.PutAsync("my/random/data_2.bin", File.OpenRead(filename), cancellationToken: cancellationToken);
         await store.PutAsync("my/random/data_3.bin", File.OpenRead(filename), cancellationToken: cancellationToken);
     }
+
+    [Fact]
+    public async Task Rename_object_should_perge_old_named_meta()
+    {
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20));
+        var cancellationToken = cts.Token;
+
+        await using var server = await NatsServerProcess.StartAsync();
+        await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+        var js = new NatsJSContext(nats);
+        var ob = new NatsObjContext(js);
+
+        var b1 = await ob.CreateObjectStoreAsync(new NatsObjConfig("b1"), cancellationToken);
+
+        await b1.PutAsync("name1", [1, 2, 3], cancellationToken: cancellationToken);
+
+        var s1 = await js.GetStreamAsync("OBJ_b1", cancellationToken: cancellationToken);
+
+        async Task<List<NatsJSMsg<byte[]>>> GetAllMsgs()
+        {
+            var c = await s1.CreateOrderedConsumerAsync(cancellationToken: cancellationToken);
+            List<NatsJSMsg<byte[]>> msgs = new();
+            await foreach (var msg in c.ConsumeAsync<byte[]>(cancellationToken: cancellationToken))
+            {
+                msgs.Add(msg);
+                if (msg.Metadata?.NumPending == 0)
+                    break;
+            }
+
+            return msgs;
+        }
+
+        // Snapshot the stream before rename
+        var msgs1 = await GetAllMsgs();
+
+        // rename
+        await b1.UpdateMetaAsync("name1", new ObjectMetadata { Name = "name2" }, cancellationToken);
+
+        // Snapshot the stream after rename
+        var msgs2 = await GetAllMsgs();
+
+        Assert.Equal(2, msgs1.Count);
+        Assert.Equal(2, msgs2.Count);
+
+        // Check the sequence numbers
+        Assert.Equal(1, (int)msgs1[0].Metadata?.Sequence.Stream!);
+        Assert.Equal(2, (int)msgs1[1].Metadata?.Sequence.Stream!);
+
+        // after rename, the chunk message is the same, but the meta is new
+        // and the old meta is gone with sequence 2
+        Assert.Equal(1, (int)msgs2[0].Metadata?.Sequence.Stream!);
+        Assert.Equal(3, (int)msgs2[1].Metadata?.Sequence.Stream!);
+
+        // the first message is the chunk with the data, it should be the same
+        Assert.Equal(msgs1[0].Subject, msgs2[0].Subject);
+
+        // the second message is the meta, it should be different
+        // because we updated the name
+        Assert.NotEqual(msgs1[1].Subject, msgs2[1].Subject);
+
+        // Check we can get by the new name as well as the old name fails
+        var n2 = await b1.GetBytesAsync("name2", cancellationToken: cancellationToken);
+        Assert.Equal([1, 2, 3], n2);
+        await Assert.ThrowsAsync<NatsObjNotFoundException>(async () =>
+        {
+            await b1.GetBytesAsync("name1", cancellationToken: cancellationToken);
+        });
+    }
 }