From 27ecadda6c249489dea6b609f64b94a6d4f68dd5 Mon Sep 17 00:00:00 2001 From: Yossi T Date: Thu, 25 Jun 2026 15:44:16 -0700 Subject: [PATCH 1/4] test: repro rebuild projection with natural key --- ...rual_key_table_not_populated_on_rebuild.cs | 116 ++++++++++++++++++ 1 file changed, 116 insertions(+) create mode 100644 src/EventSourcingTests/Bugs/Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs diff --git a/src/EventSourcingTests/Bugs/Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs b/src/EventSourcingTests/Bugs/Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs new file mode 100644 index 0000000000..60f55d119a --- /dev/null +++ b/src/EventSourcingTests/Bugs/Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs @@ -0,0 +1,116 @@ +using System; +using System.Data; +using System.Threading; +using System.Threading.Tasks; +using JasperFx.Events; +using JasperFx.Events.Aggregation; +using Marten; +using Marten.Events.Projections; +using Marten.Testing.Harness; +using Shouldly; +using Xunit; + +namespace EventSourcingTests.Bugs; + +public class Bug_xxxx_natrual_key_table_not_populated_on_rebuild: OneOffConfigurationsContext +{ + private const string schemaName = "bug_xxxx"; + + public sealed record OrderNumber(string Value); + + public sealed record OrderPlaced(Guid OrderId, string OrderNumber); + + public sealed record OrderShipped(Guid OrderId, string TrackingNumber); + + public sealed record Order + { + public Guid Id { get; set; } + + [NaturalKey] + public OrderNumber Number { get; set; } + + public string? TrackingNumber { get; set; } + + [NaturalKeySource] + public static Order Create(OrderPlaced e) + { + return new Order + { + Id = e.OrderId, + Number = new OrderNumber(e.OrderNumber) + }; + } + + public static Order Apply(OrderShipped e, Order order) + { + return new Order + { + TrackingNumber = e.TrackingNumber + }; + } + } + + private static void ConfigureStore(StoreOptions opts) + { + opts.Advanced.Migrator.NameDataLength = 100; + opts.Connection(ConnectionSource.ConnectionString); + opts.DatabaseSchemaName = schemaName; + + opts.Events.StreamIdentity = StreamIdentity.AsGuid; + opts.Events.AppendMode = EventAppendMode.Quick; + + opts.Projections.Snapshot(SnapshotLifecycle.Inline); + } + + private async Task ExecuteScalar(string sql) + { + await using var conn = theStore.Storage.Database.CreateConnection(); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = sql; + return await cmd.ExecuteScalarAsync(); + } + + private async Task GetData(string sql) + { + await using var conn = theStore.Storage.Database.CreateConnection(); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = sql; + await using var reader = await cmd.ExecuteReaderAsync(); + var dataTable = new DataTable(); + dataTable.Load(reader); + return dataTable; + } + + [Fact] + public async Task rebuild_snapshot_should_populate_naturalkey_table() + { + StoreOptions(ConfigureStore); + await theStore.Storage.ApplyAllConfiguredChangesToDatabaseAsync(); + + var streamId = Guid.NewGuid(); + await using var session = theStore.LightweightSession(); + session.Events.StartStream(streamId, new OrderPlaced(streamId, "12345")); + await session.SaveChangesAsync(); + + var naturalKeyTableName = $"{schemaName}.mt_natural_key_order"; + var natrualKeys = await GetData($"SELECT * FROM {naturalKeyTableName}"); + natrualKeys.Rows.Count.ShouldBe(1); + + await ExecuteScalar($"DELETE FROM {naturalKeyTableName}"); + natrualKeys = await GetData($"SELECT * FROM {naturalKeyTableName}"); + natrualKeys.Rows.Count.ShouldBe(0); + + // Query pg_indexes to verify the index exists + await using var conn = theStore.Storage.Database.CreateConnection(); + await conn.OpenAsync(); + + var daemon = await SeparateStore(ConfigureStore).BuildProjectionDaemonAsync(); + await daemon.PrepareForRebuildsAsync(); + await daemon.RebuildProjectionAsync($"{nameof(Bug_xxxx_natrual_key_table_not_populated_on_rebuild)}.order", CancellationToken.None); + + natrualKeys = await GetData($"SELECT * FROM {naturalKeyTableName}"); + natrualKeys.Rows.Count.ShouldBe(1); + } +} From 12f16dbef02a0a17791ff3087fe7f01b2c5c23ee Mon Sep 17 00:00:00 2001 From: Yossi T Date: Fri, 26 Jun 2026 08:03:01 -0700 Subject: [PATCH 2/4] fix: query by natural key --- ...atrual_key_table_not_populated_on_rebuild.cs | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/src/EventSourcingTests/Bugs/Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs b/src/EventSourcingTests/Bugs/Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs index 60f55d119a..20638c3b47 100644 --- a/src/EventSourcingTests/Bugs/Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs +++ b/src/EventSourcingTests/Bugs/Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs @@ -12,9 +12,9 @@ namespace EventSourcingTests.Bugs; -public class Bug_xxxx_natrual_key_table_not_populated_on_rebuild: OneOffConfigurationsContext +public class Bug_4788_natrual_key_table_not_populated_on_rebuild: OneOffConfigurationsContext { - private const string schemaName = "bug_xxxx"; + private const string schemaName = "bug_4788"; public sealed record OrderNumber(string Value); @@ -94,6 +94,11 @@ public async Task rebuild_snapshot_should_populate_naturalkey_table() session.Events.StartStream(streamId, new OrderPlaced(streamId, "12345")); await session.SaveChangesAsync(); + var anotherStore = SeparateStore(ConfigureStore); + var anotherSession = anotherStore.LightweightSession(); + var order1 = await anotherSession.Events.FetchLatest(new OrderNumber("12345")); + order1.ShouldNotBeNull(); + var naturalKeyTableName = $"{schemaName}.mt_natural_key_order"; var natrualKeys = await GetData($"SELECT * FROM {naturalKeyTableName}"); natrualKeys.Rows.Count.ShouldBe(1); @@ -102,13 +107,17 @@ public async Task rebuild_snapshot_should_populate_naturalkey_table() natrualKeys = await GetData($"SELECT * FROM {naturalKeyTableName}"); natrualKeys.Rows.Count.ShouldBe(0); - // Query pg_indexes to verify the index exists await using var conn = theStore.Storage.Database.CreateConnection(); await conn.OpenAsync(); var daemon = await SeparateStore(ConfigureStore).BuildProjectionDaemonAsync(); await daemon.PrepareForRebuildsAsync(); - await daemon.RebuildProjectionAsync($"{nameof(Bug_xxxx_natrual_key_table_not_populated_on_rebuild)}.order", CancellationToken.None); + await daemon.RebuildProjectionAsync($"{nameof(Bug_4788_natrual_key_table_not_populated_on_rebuild)}.order", CancellationToken.None); + + var afterRebuildStore = SeparateStore(ConfigureStore); + var afterRebuildSession = afterRebuildStore.LightweightSession(); + var order2 = await afterRebuildSession.Events.FetchLatest(new OrderNumber("12345")); + order2.ShouldNotBeNull(); natrualKeys = await GetData($"SELECT * FROM {naturalKeyTableName}"); natrualKeys.Rows.Count.ShouldBe(1); From 6322a635335107b4035731d5cabd835c40c94722 Mon Sep 17 00:00:00 2001 From: Yossi T Date: Fri, 26 Jun 2026 09:58:02 -0700 Subject: [PATCH 3/4] updated file name --- ....cs => Bug_4788_natrual_key_table_not_populated_on_rebuild.cs} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename src/EventSourcingTests/Bugs/{Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs => Bug_4788_natrual_key_table_not_populated_on_rebuild.cs} (100%) diff --git a/src/EventSourcingTests/Bugs/Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs b/src/EventSourcingTests/Bugs/Bug_4788_natrual_key_table_not_populated_on_rebuild.cs similarity index 100% rename from src/EventSourcingTests/Bugs/Bug_xxxx_natrual_key_table_not_populated_on_rebuild.cs rename to src/EventSourcingTests/Bugs/Bug_4788_natrual_key_table_not_populated_on_rebuild.cs From 0880e5e54daf3143ad93af69522258458320f144 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sat, 27 Jun 2026 09:26:02 -0500 Subject: [PATCH 4/4] Fix #4788: rebuild a [NaturalKey] aggregate also rebuilds the mt_natural_key lookup table MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The natural-keys lookup table (mt_natural_key_{aggregate}) is maintained by NaturalKeyProjection, an auto-registered inline projection whose ApplyAsync iterates the StreamActions queued during the current SaveChanges and writes ON CONFLICT upserts per matching event. That dispatch fires correctly on the normal append path but the async-daemon rebuild path appends no streams — it replays already-persisted events from mt_events through the projection's aggregator and stores the rebuilt snapshots, so the inline projection chain is never given anything to iterate. Net effect: after rebuild the parent projection's documents are repopulated, but the mt_natural_key table stays exactly as teardown left it (empty), and any FetchLatest afterwards misses. Fix — two new hooks in Marten, no JasperFx changes required (the existing abstractions — IAggregateProjection.NaturalKeyDefinition, EventRange.Events on every rebuild page, and ProjectionBatch.SessionForTenant routing writes into the batch's work-tracker — already give Marten everything it needs): 1. teardownProjectionStorage now also wipes the natural-key table when the source projection carries a NaturalKeyDefinition, so the rebuild starts from a clean slate (mirrors the doc-table TRUNCATE). 2. StartProjectionBatchAsync, in Rebuild mode, looks up the source by range.ShardName.Name and — when it has a NaturalKeyDefinition — feeds range.Events through a new NaturalKeyProjection.QueueUpsertsForEvents method per tenant. The IDocumentOperations comes from projectionBatch.SessionForTenant(...) which returns a ProjectionDocumentSession whose ISessionWorkTracker IS the ProjectionUpdateBatch — so the upserts flush inside the same transaction as the rebuilt snapshots, not the bare session's unit-of-work (which the batch's ExecuteAsync never touches). NaturalKeyProjection refactor: extracted queueUpsertSql to take (streamId, streamKey, tenantId, innerValue) instead of a StreamAction so the inline path (StreamAction-driven) and the rebuild path (IEvent-driven) share the SQL builder. ApplyAsync behavior is byte-for-byte unchanged. Test: ytqsl's Bug_4788_natrual_key_table_not_populated_on_rebuild repro (authored in commits 27ecadda6 / 12f16dbef / 6322a6353 on this branch) now passes. Broader regression sweep on net10: 34/34 existing natural-key tests + 8/8 EventSourcingTests rebuild tests + 31/31 DaemonTests rebuild tests all green. Closes #4788 Supersedes #4789 Co-Authored-By: Claude Opus 4.7 (1M context) --- src/Marten/DocumentStore.EventStore.cs | 39 ++++++++++++++++ .../Projections/NaturalKeyProjection.cs | 45 ++++++++++++++++--- 2 files changed, 77 insertions(+), 7 deletions(-) diff --git a/src/Marten/DocumentStore.EventStore.cs b/src/Marten/DocumentStore.EventStore.cs index 691ba7cce1..815c733a07 100644 --- a/src/Marten/DocumentStore.EventStore.cs +++ b/src/Marten/DocumentStore.EventStore.cs @@ -323,6 +323,17 @@ private void teardownProjectionStorage(IProjectionSource(x => x.ProjectionName == source.Name); + + // #4788: a [NaturalKey] aggregate maintains its mt_natural_key_X lookup table via the + // auto-registered NaturalKeyProjection on the inline-append path. Teardown of the parent + // projection must also wipe the natural-key table so the rebuild path repopulates it from + // scratch (the rebuild itself re-emits the upserts via StartProjectionBatchAsync). + if (source is IAggregateProjection aggregateSource && aggregateSource.NaturalKeyDefinition != null) + { + var naturalKeyTable = + $"{Events.DatabaseSchemaName}.mt_natural_key_{aggregateSource.NaturalKeyDefinition.AggregateType.Name.ToLowerInvariant()}"; + session.QueueSqlCommand($"delete from {naturalKeyTable}"); + } } public async ValueTask> StartProjectionBatchAsync( @@ -361,6 +372,34 @@ public async ValueTask> Sta await projectionBatch.RecordProgress(range).ConfigureAwait(false); + // #4788: when rebuilding a snapshot whose aggregate has a [NaturalKey], re-emit the + // natural-key upserts for this page's events. ApplyAsync on the inline path drives off + // newly-appended StreamActions and never fires during rebuild — without this hook the + // mt_natural_key_X table stays empty after teardown. Routing the upsert SQL through + // ProjectionBatch.SessionForTenant returns a ProjectionDocumentSession whose work-tracker + // IS the ProjectionUpdateBatch, so the operations flush alongside the rebuilt snapshots + // inside the same batch transaction. + if (mode == ShardExecutionMode.Rebuild && range.Events.Any()) + { + var naturalKeySource = Options.Projections.All + .FirstOrDefault(s => s.Name.EqualsIgnoreCase(range.ShardName.Name)) + as IAggregateProjection; + if (naturalKeySource?.NaturalKeyDefinition == null) + { + naturalKeySource = null; + } + + if (naturalKeySource != null) + { + var naturalKeyProjection = new NaturalKeyProjection(Options.EventGraph, naturalKeySource.NaturalKeyDefinition!); + foreach (var byTenant in range.Events.GroupBy(e => e.TenantId ?? StorageConstants.DefaultTenantId)) + { + var ops = projectionBatch.SessionForTenant(byTenant.Key); + naturalKeyProjection.QueueUpsertsForEvents(ops, byTenant); + } + } + } + return projectionBatch; } diff --git a/src/Marten/Events/Projections/NaturalKeyProjection.cs b/src/Marten/Events/Projections/NaturalKeyProjection.cs index 4fd5be3dc1..76c35bef48 100644 --- a/src/Marten/Events/Projections/NaturalKeyProjection.cs +++ b/src/Marten/Events/Projections/NaturalKeyProjection.cs @@ -52,7 +52,7 @@ public Task ApplyAsync(IDocumentOperations operations, IEnumerable var innerValue = _naturalKey.Unwrap(rawValue); if (innerValue != null) { - queueUpsertSql(operations, stream, innerValue); + queueUpsertSql(operations, stream.Id, stream.Key, stream.TenantId, innerValue); } } } @@ -62,10 +62,41 @@ public Task ApplyAsync(IDocumentOperations operations, IEnumerable return Task.CompletedTask; } - private void queueUpsertSql(IDocumentOperations operations, StreamAction stream, object innerValue) + /// + /// #4788: rebuild-time counterpart to . The async-daemon rebuild path + /// replays already-persisted events without appending streams, so ApplyAsync's + /// stream-driven dispatch never fires and the mt_natural_key_X table stays empty after + /// teardown. This entry-point feeds raw s straight through the same + /// upsert SQL builder, pulling stream id/key + tenant id off the event itself (events written + /// to mt_events always carry these). Called from StartProjectionBatchAsync per + /// rebuild page, with routed through + /// so the SQL flushes into the projection batch + /// rather than the bare session's unit-of-work. + /// + internal void QueueUpsertsForEvents(IDocumentOperations operations, IEnumerable events) + { + foreach (var @event in events) + { + foreach (var mapping in _naturalKey.EventMappings) + { + if (mapping.EventType.IsAssignableFrom(@event.Data.GetType())) + { + var rawValue = mapping.Extractor(@event.Data); + var innerValue = _naturalKey.Unwrap(rawValue); + if (innerValue != null) + { + queueUpsertSql(operations, @event.StreamId, @event.StreamKey, @event.TenantId, innerValue); + } + } + } + } + } + + private void queueUpsertSql(IDocumentOperations operations, Guid streamId, string? streamKey, + string tenantId, object innerValue) { var streamCol = _isGuid ? "stream_id" : "stream_key"; - object streamId = _isGuid ? (object)stream.Id : stream.Key!; + object streamIdValue = _isGuid ? (object)streamId : streamKey!; // When UseArchivedStreamPartitioning is on, is_archived is part of the PK // and must be included in the ON CONFLICT clause @@ -74,28 +105,28 @@ private void queueUpsertSql(IDocumentOperations operations, StreamAction stream, var sql = $"INSERT INTO {_tableName} (natural_key_value, {streamCol}, tenant_id, is_archived) " + $"VALUES (?, ?, ?, false) " + $"ON CONFLICT (natural_key_value, tenant_id, is_archived) DO UPDATE SET {streamCol} = ?"; - operations.QueueSqlCommand(sql, innerValue, streamId, stream.TenantId, streamId); + operations.QueueSqlCommand(sql, innerValue, streamIdValue, tenantId, streamIdValue); } else if (_isConjoined) { var sql = $"INSERT INTO {_tableName} (natural_key_value, {streamCol}, tenant_id, is_archived) " + $"VALUES (?, ?, ?, false) " + $"ON CONFLICT (natural_key_value, tenant_id) DO UPDATE SET {streamCol} = ?, is_archived = false"; - operations.QueueSqlCommand(sql, innerValue, streamId, stream.TenantId, streamId); + operations.QueueSqlCommand(sql, innerValue, streamIdValue, tenantId, streamIdValue); } else if (_useArchivedPartitioning) { var sql = $"INSERT INTO {_tableName} (natural_key_value, {streamCol}, is_archived) " + $"VALUES (?, ?, false) " + $"ON CONFLICT (natural_key_value, is_archived) DO UPDATE SET {streamCol} = ?"; - operations.QueueSqlCommand(sql, innerValue, streamId, streamId); + operations.QueueSqlCommand(sql, innerValue, streamIdValue, streamIdValue); } else { var sql = $"INSERT INTO {_tableName} (natural_key_value, {streamCol}, is_archived) " + $"VALUES (?, ?, false) " + $"ON CONFLICT (natural_key_value) DO UPDATE SET {streamCol} = ?, is_archived = false"; - operations.QueueSqlCommand(sql, innerValue, streamId, streamId); + operations.QueueSqlCommand(sql, innerValue, streamIdValue, streamIdValue); } }