diff --git a/src/NATS.Client.ObjectStore/NatsObjStore.cs b/src/NATS.Client.ObjectStore/NatsObjStore.cs index 4a5bf0508..759d0f0b1 100644 --- a/src/NATS.Client.ObjectStore/NatsObjStore.cs +++ b/src/NATS.Client.ObjectStore/NatsObjStore.cs @@ -328,6 +328,13 @@ public async ValueTask 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; } diff --git a/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs b/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs index b26d8206e..acbe86ecf 100644 --- a/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs +++ b/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>> GetAllMsgs() + { + var c = await s1.CreateOrderedConsumerAsync(cancellationToken: cancellationToken); + List> msgs = new(); + await foreach (var msg in c.ConsumeAsync(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(async () => + { + await b1.GetBytesAsync("name1", cancellationToken: cancellationToken); + }); + } }