Retire the previous natural key row when the key changes (#5041)

docs/cSpell.json

@@ -2,6 +2,7 @@
   "version": "0.1",
   "language": "en",
   "words": [
+    "jasperfx",
     "TimescaleDB",
     "timescaledb",
     "hypertable",

docs/events/natural-keys.md

@@ -119,6 +119,26 @@ Every event type that sets or changes the natural key must be declared through t
 
 Events that do not affect the natural key (like `OrderItemAdded` in the example above) do not need any mapping.
 
+### Handler Signature Requirements
+
+The lookup table is written *inline* as events are appended, well before any projection has built the
+aggregate — that is what lets `FetchForWriting` by natural key work even when the snapshot lifecycle is
+`Async`. The key value therefore has to be derivable from the event alone, and a `[NaturalKeySource]`
+method must be one of:
+
+- a static factory or evolve method on the aggregate taking the raw event type, such as
+  `public static Order Create(OrderCreated e)` or `public static Order Apply(OrderRenumbered e, Order current)`
+- an instance `Apply(TEvent)` method on the aggregate whose body sets only the natural key property
+
+::: warning
+Do not write a `[NaturalKeySource]` method whose new key value depends on the *previous* aggregate state,
+and do not rely on any aggregate state other than the natural key inside one. Marten derives the key by
+calling your method with a blank aggregate instance, so anything else on it will be `null` or default.
+
+Signatures taking `IEvent<T>` rather than the raw event type are not currently supported here, and are
+silently ignored rather than reported — see [JasperFx/jasperfx#569](https://github.com/JasperFx/jasperfx/issues/569).
+:::
+
 ## Storage
 
 Marten automatically creates and manages a lookup table for each aggregate type that has a natural key configured. The table maps natural key values to stream ids and is:
@@ -169,6 +189,8 @@ var aggregate = await theSession.Events.FetchLatest<OrderAggregate, OrderNumber>
 
 Natural keys can change over the lifetime of a stream. When an event mapped with `[NaturalKeySource]` is appended, Marten updates the lookup table with the new value. The old key value is replaced, so lookups using the previous key will no longer resolve to that stream.
 
+The retired row is removed rather than left behind, so the previous value immediately becomes available for another stream to claim.
+
 ## Null and Default Keys
 
 If a mapped event produces a `null` or default key value, Marten silently skips writing to the lookup table. This means streams where the natural key has not yet been assigned will not appear in natural key lookups, but will still be accessible by stream id.

src/DaemonTests/Aggregations/Bug_5041_natural_key_source_discovery.cs

@@ -0,0 +1,143 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using DaemonTests.TestingSupport;
+using JasperFx.Events;
+using JasperFx.Events.Aggregation;
+using Marten;
+using Marten.Testing.Harness;
+using Shouldly;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace DaemonTests.Aggregations;
+
+/// <summary>
+/// #5041, from the repro in https://github.com/JasperFx/marten/pull/5042 (thanks @ytqsl).
+///
+/// Both of these hang on [NaturalKeySource] discovery in JasperFx.Events, not on anything Marten
+/// owns — see https://github.com/JasperFx/jasperfx/issues/569:
+///
+///   * a handler whose first parameter is IEvent&lt;T&gt; yields no usable extractor, so
+///     NaturalKeyDefinition.EventMappings never gains an entry for that event type and the
+///     mt_natural_key_X table is silently never written for it;
+///   * an instance Apply(TEvent) handler is invoked reflectively against a fabricated blank
+///     aggregate (Expression.New(TDoc), which also bypasses `required` member enforcement), so a
+///     handler body that touches any other state throws — out of NaturalKeyProjection.ApplyAsync
+///     and out of the caller's SaveChangesAsync.
+///
+/// Unskip both when the JasperFx.Events dependency picks up the fix.
+/// </summary>
+public class Bug_5041_natural_key_source_discovery: DaemonContext
+{
+    private const string schemaName = "bug_5041_natural_key_discovery";
+
+    public Bug_5041_natural_key_source_discovery(ITestOutputHelper output) : base(output)
+    {
+    }
+
+    public sealed record ProductCode(string Value);
+
+    public sealed record ProductRegistered(Guid ProductId, string ProductCode);
+
+    public sealed record ProductCodeChangedByEventWrapper(Guid ProductId, string NewProductCode);
+
+    public sealed record ProductCodeChangedByInstanceMethod(Guid ProductId, string NewProductCode);
+
+    public sealed record Product
+    {
+        public Guid Id { get; set; }
+
+        [NaturalKey]
+        public ProductCode Code { get; set; }
+
+        public required IEnumerable<ProductCode> KnownCodes { get; set; }
+
+        [NaturalKeySource]
+        public static Product Create(ProductRegistered e)
+        {
+            return new Product
+            {
+                Id = e.ProductId,
+                Code = new ProductCode(e.ProductCode),
+                KnownCodes = [new ProductCode(e.ProductCode)]
+            };
+        }
+
+        [NaturalKeySource]
+        public static Product Apply(IEvent<ProductCodeChangedByEventWrapper> e, Product product)
+        {
+            return product with
+            {
+                Code = new ProductCode(e.Data.NewProductCode),
+                KnownCodes = product.KnownCodes
+                    .Where(c => c.Value != e.Data.NewProductCode)
+                    .Append(new ProductCode(e.Data.NewProductCode))
+            };
+        }
+
+        [NaturalKeySource]
+        public void Apply(ProductCodeChangedByInstanceMethod e)
+        {
+            Code = new ProductCode(e.NewProductCode);
+            KnownCodes = KnownCodes
+                .Where(c => c.Value != e.NewProductCode)
+                .Append(new ProductCode(e.NewProductCode));
+        }
+    }
+
+    private static void ConfigureStore(StoreOptions opts)
+    {
+        opts.Connection(ConnectionSource.ConnectionString);
+        opts.DatabaseSchemaName = schemaName;
+        opts.Events.StreamIdentity = StreamIdentity.AsGuid;
+        opts.Events.AppendMode = EventAppendMode.Quick;
+        opts.Projections.Snapshot<Product>(SnapshotLifecycle.Async);
+    }
+
+    [Fact(Skip = "Blocked on JasperFx/jasperfx#569 -- IEvent<T> [NaturalKeySource] handlers yield no extractor")]
+    public async Task natural_key_is_maintained_when_the_handler_takes_IEvent()
+    {
+        await runRenameScenario(streamId => new ProductCodeChangedByEventWrapper(streamId, "PROD-999"));
+    }
+
+    [Fact(Skip = "Blocked on JasperFx/jasperfx#569 -- instance [NaturalKeySource] handlers run against a blank aggregate")]
+    public async Task natural_key_is_maintained_when_the_handler_is_an_instance_method()
+    {
+        await runRenameScenario(streamId => new ProductCodeChangedByInstanceMethod(streamId, "PROD-999"));
+    }
+
+    private async Task runRenameScenario(Func<Guid, object> renameEvent)
+    {
+        StoreOptions(ConfigureStore);
+        await theStore.Storage.ApplyAllConfiguredChangesToDatabaseAsync();
+        await theStore.Advanced.Clean.DeleteAllDocumentsAsync();
+        await theStore.Advanced.Clean.DeleteAllEventDataAsync();
+
+        var streamId = Guid.NewGuid();
+
+        await using (var session = theStore.LightweightSession())
+        {
+            session.Events.StartStream<Product>(streamId, new ProductRegistered(streamId, "PROD-001"));
+            await session.SaveChangesAsync();
+        }
+
+        await using (var session = theStore.LightweightSession())
+        {
+            session.Events.Append(streamId, renameEvent(streamId));
+            await session.SaveChangesAsync();
+        }
+
+        var daemon = await theStore.BuildProjectionDaemonAsync();
+        await daemon.RebuildProjectionAsync<Product>(CancellationToken.None);
+
+        await using var query = theStore.LightweightSession();
+        var product = await query.Events.FetchLatest<Product, ProductCode>(new ProductCode("PROD-999"));
+        product.ShouldNotBeNull();
+        product.Code.Value.ShouldBe("PROD-999");
+        product.KnownCodes.ShouldContain(new ProductCode("PROD-001"));
+        product.KnownCodes.ShouldContain(new ProductCode("PROD-999"));
+    }
+}

src/DaemonTests/Aggregations/rebuild_projection_with_natural_key_update.cs

@@ -1,4 +1,6 @@
 using System;
+using System.Collections.Generic;
+using System.Data.Common;
 using System.Threading;
 using System.Threading.Tasks;
 using DaemonTests.TestingSupport;
@@ -98,4 +100,123 @@ public async Task bug_4966_natural_key_should_be_updated_during_rebuild()
             product.Code.Value.ShouldBe(newCode);
         }
     }
+
+    [Fact]
+    public async Task bug_5041_renaming_the_natural_key_retires_the_previous_row_inline()
+    {
+        StoreOptions(ConfigureStore);
+        await theStore.Storage.ApplyAllConfiguredChangesToDatabaseAsync();
+        await theStore.Advanced.Clean.DeleteAllDocumentsAsync();
+        await theStore.Advanced.Clean.DeleteAllEventDataAsync();
+
+        var streamId = Guid.NewGuid();
+
+        await using (var session = theStore.LightweightSession())
+        {
+            session.Events.StartStream<Product>(streamId, new ProductRegistered(streamId, "PROD-001"));
+            await session.SaveChangesAsync();
+        }
+
+        (await naturalKeysForStream(streamId)).ShouldBe(["PROD-001"]);
+
+        await using (var session = theStore.LightweightSession())
+        {
+            session.Events.Append(streamId, new ProductCodeChanged(streamId, "PROD-999"));
+            await session.SaveChangesAsync();
+        }
+
+        // Before #5041 the retired PROD-001 row survived alongside PROD-999, permanently
+        // squatting on its slot in the natural key table's primary key.
+        (await naturalKeysForStream(streamId)).ShouldBe(["PROD-999"]);
+    }
+
+    [Fact]
+    public async Task bug_5041_the_previous_natural_key_row_is_not_resurrected_by_a_rebuild()
+    {
+        StoreOptions(ConfigureStore);
+        await theStore.Storage.ApplyAllConfiguredChangesToDatabaseAsync();
+        await theStore.Advanced.Clean.DeleteAllDocumentsAsync();
+        await theStore.Advanced.Clean.DeleteAllEventDataAsync();
+
+        var streamId = Guid.NewGuid();
+
+        await using (var session = theStore.LightweightSession())
+        {
+            session.Events.StartStream<Product>(streamId, new ProductRegistered(streamId, "PROD-001"));
+            session.Events.Append(streamId, new ProductCodeChanged(streamId, "PROD-999"));
+            await session.SaveChangesAsync();
+        }
+
+        var daemon = await theStore.BuildProjectionDaemonAsync();
+        await daemon.RebuildProjectionAsync<Product>(CancellationToken.None);
+
+        // The rebuild path (StartProjectionBatchAsync -> QueueUpsertsForEvents) replays both
+        // events through the same upsert builder, so it has to retire PROD-001 too.
+        (await naturalKeysForStream(streamId)).ShouldBe(["PROD-999"]);
+
+        await using var query = theStore.LightweightSession();
+        var product = await query.Events.FetchLatest<Product, ProductCode>(new ProductCode("PROD-999"));
+        product.ShouldNotBeNull();
+        product.Code.Value.ShouldBe("PROD-999");
+    }
+
+    [Fact]
+    public async Task bug_5041_a_retired_natural_key_can_be_claimed_by_another_stream()
+    {
+        StoreOptions(ConfigureStore);
+        await theStore.Storage.ApplyAllConfiguredChangesToDatabaseAsync();
+        await theStore.Advanced.Clean.DeleteAllDocumentsAsync();
+        await theStore.Advanced.Clean.DeleteAllEventDataAsync();
+
+        var first = Guid.NewGuid();
+        var second = Guid.NewGuid();
+
+        await using (var session = theStore.LightweightSession())
+        {
+            session.Events.StartStream<Product>(first, new ProductRegistered(first, "PROD-001"));
+            session.Events.Append(first, new ProductCodeChanged(first, "PROD-999"));
+            await session.SaveChangesAsync();
+        }
+
+        await using (var session = theStore.LightweightSession())
+        {
+            session.Events.StartStream<Product>(second, new ProductRegistered(second, "PROD-001"));
+            await session.SaveChangesAsync();
+        }
+
+        (await naturalKeysForStream(first)).ShouldBe(["PROD-999"]);
+        (await naturalKeysForStream(second)).ShouldBe(["PROD-001"]);
+
+        // The snapshot is Async, so the documents only exist once the daemon has run.
+        var daemon = await theStore.BuildProjectionDaemonAsync();
+        await daemon.RebuildProjectionAsync<Product>(CancellationToken.None);
+
+        await using var query = theStore.LightweightSession();
+        var reused = await query.Events.FetchLatest<Product, ProductCode>(new ProductCode("PROD-001"));
+        reused.ShouldNotBeNull();
+        reused.Id.ShouldBe(second);
+    }
+
+    private async Task<string[]> naturalKeysForStream(Guid streamId)
+    {
+        await using var conn = theStore.Storage.Database.CreateConnection();
+        await conn.OpenAsync();
+
+        await using var cmd = conn.CreateCommand();
+        cmd.CommandText =
+            $"select natural_key_value from {schemaName}.mt_natural_key_product where stream_id = :id order by natural_key_value";
+        var parameter = cmd.CreateParameter();
+        parameter.ParameterName = "id";
+        parameter.Value = streamId;
+        cmd.Parameters.Add(parameter);
+
+        var values = new List<string>();
+        await using var reader = await cmd.ExecuteReaderAsync();
+        while (await reader.ReadAsync())
+        {
+            values.Add(await reader.GetFieldValueAsync<string>(0));
+        }
+
+        return values.ToArray();
+    }
 }

src/Marten/Events/Projections/NaturalKeyProjection.cs

@@ -98,6 +98,27 @@ private void queueUpsertSql(IDocumentOperations operations, Guid streamId, strin
         var streamCol = _isGuid ? "stream_id" : "stream_key";
         object streamIdValue = _isGuid ? (object)streamId : streamKey!;
 
+        // #5041: a stream has exactly one *current* natural key value, but the upsert below only
+        // ever inserts. When an event changes the key, the row carrying the previous value stays
+        // behind pointing at the same stream, so the table accumulates one dead row per rename and
+        // the retired key keeps occupying its slot in the primary key. Retire those rows first —
+        // scoped to this stream (and tenant, when conjoined) so a key legitimately owned by some
+        // other stream is never touched. Queued ahead of the upsert, and QueueSqlCommand preserves
+        // order within the batch, so a Create-then-rename inside a single batch still lands on the
+        // newest value.
+        if (_isConjoined)
+        {
+            operations.QueueSqlCommand(
+                $"DELETE FROM {_tableName} WHERE tenant_id = ? AND {streamCol} = ? AND natural_key_value <> ?",
+                tenantId, streamIdValue, innerValue);
+        }
+        else
+        {
+            operations.QueueSqlCommand(
+                $"DELETE FROM {_tableName} WHERE {streamCol} = ? AND natural_key_value <> ?",
+                streamIdValue, innerValue);
+        }
+
         // When UseArchivedStreamPartitioning is on, is_archived is part of the PK
         // and must be included in the ON CONFLICT clause
         if (_isConjoined && _useArchivedPartitioning)