From 85e627c08c0d533e771976021ba5b19330810c3f Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Fri, 17 Jul 2026 16:51:16 -0500 Subject: [PATCH] feat(#4962): targeted per-cell ReadProjectionProgressAsync on MartenDatabase MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Override the jasperfx#435 default-throwing IEventDatabase.ReadProjectionProgressAsync so a monitoring tool can poll a single (projection, tenant) progression cell without fetching + filtering every row via AllProjectionProgress. mt_event_progression.name holds a full ShardName.Identity, so one projection name can map to several rows (blue/green versions, custom shard keys, per-tenant partitions). Resolution: candidates are rows whose parsed ShardName.Name equals projectionName and whose parsed TenantId equals tenantId — a null tenantId matches ONLY the bare store-global rows (no tenant suffix); when several match (a blue/green deploy) the NEWEST version wins, tie-broken on the highest sequence; no match returns null. AgentStatus and LastHeartbeat are always null (Marten models the columns but no daemon path writes them — jasperfx#519). The name LIKE 'projectionName:%' probe leans on the ':' delimiter so "Orders" cannot match "OrdersHistory"; final Name/Tenant matching is exact on the parsed ShardName. Callers that already hold a full ShardName (e.g. from AllProjectionProgress) will be able to avoid this collapsing once the exact ShardName overload lands (follow-up). Closes #4962. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01XKKBPsFJ83jd2o4hyGJnj6 --- .../Internals/read_projection_progress.cs | 106 ++++++++++++++++++ .../Storage/MartenDatabase.EventStorage.cs | 70 ++++++++++++ 2 files changed, 176 insertions(+) create mode 100644 src/DaemonTests/Internals/read_projection_progress.cs diff --git a/src/DaemonTests/Internals/read_projection_progress.cs b/src/DaemonTests/Internals/read_projection_progress.cs new file mode 100644 index 0000000000..75d7fe54f6 --- /dev/null +++ b/src/DaemonTests/Internals/read_projection_progress.cs @@ -0,0 +1,106 @@ +using System.Threading; +using System.Threading.Tasks; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Projections; +using Marten; +using Marten.Events.Daemon.Progress; +using Marten.Storage; +using Marten.Testing; +using Marten.Testing.Harness; +using Shouldly; +using Xunit; + +namespace DaemonTests.Internals; + +// #4962 / jasperfx#435 — targeted per-cell progression read. mt_event_progression.name holds a full +// ShardName.Identity, so (projectionName, tenantId) can map to several rows; the resolution rules are: +// candidates match parsed Name + TenantId (null tenant = bare store-global only), newest version wins. +public class read_projection_progress : OneOffConfigurationsContext, IAsyncLifetime +{ + public async Task InitializeAsync() + { + await theStore.Advanced.Clean.DeleteAllEventDataAsync(); + await theStore.EnsureStorageExistsAsync(typeof(IEvent)); + } + + public Task DisposeAsync() + { + Dispose(); + return Task.CompletedTask; + } + + private async Task seedProgression(params (ShardName name, long sequence)[] rows) + { + foreach (var (name, sequence) in rows) + { + theSession.QueueOperation(new InsertProjectionProgress(theStore.Events, new EventRange(name, sequence))); + } + + await theSession.SaveChangesAsync(); + } + + private ValueTask read(string projectionName, string? tenantId) => + ((IEventDatabase)theStore.Tenancy.Default.Database) + .ReadProjectionProgressAsync(projectionName, tenantId, CancellationToken.None); + + [Fact] + public async Task reads_the_store_global_row() + { + await seedProgression((ShardName.Compose("Orders"), 42)); + + var row = await read("Orders", null); + + row.ShouldNotBeNull(); + row.ProjectionName.ShouldBe("Orders"); + row.TenantId.ShouldBeNull(); + row.Sequence.ShouldBe(42); + // Marten models the columns but writes neither — always null (jasperfx#519). + row.AgentStatus.ShouldBeNull(); + row.LastHeartbeat.ShouldBeNull(); + } + + [Fact] + public async Task returns_null_when_no_row_exists() + { + await seedProgression((ShardName.Compose("Orders"), 42)); + + (await read("Unobserved", null)).ShouldBeNull(); + } + + [Fact] + public async Task newest_version_wins_across_a_blue_green_deploy() + { + await seedProgression( + (ShardName.Compose("Orders"), 10), // Orders:All (V1) + (ShardName.Compose("Orders", version: 2), 25), // Orders:V2:All + (ShardName.Compose("Orders", version: 3), 40)); // Orders:V3:All + + var row = await read("Orders", null); + + row.ShouldNotBeNull(); + row.Sequence.ShouldBe(40); + } + + [Fact] + public async Task null_tenant_matches_only_the_bare_store_global_row() + { + await seedProgression( + (ShardName.Compose("Orders"), 10), // Orders:All + (ShardName.Compose("Orders", tenantId: "tenant1"), 99)); // Orders:All:tenant1 + + (await read("Orders", null))!.Sequence.ShouldBe(10); + (await read("Orders", "tenant1"))!.Sequence.ShouldBe(99); + } + + [Fact] + public async Task does_not_match_a_projection_whose_name_is_a_prefix() + { + await seedProgression( + (ShardName.Compose("Orders"), 10), + (ShardName.Compose("OrdersHistory"), 5)); + + (await read("Orders", null))!.Sequence.ShouldBe(10); + (await read("OrdersHistory", null))!.Sequence.ShouldBe(5); + } +} diff --git a/src/Marten/Storage/MartenDatabase.EventStorage.cs b/src/Marten/Storage/MartenDatabase.EventStorage.cs index b6e0cb7912..4747936024 100644 --- a/src/Marten/Storage/MartenDatabase.EventStorage.cs +++ b/src/Marten/Storage/MartenDatabase.EventStorage.cs @@ -276,6 +276,76 @@ public async Task> AllProjectionProgress(string? tenan } } + /// + /// #4962 / jasperfx#435 — targeted per-cell progression read for a single (projection, tenant). + /// The mt_event_progression.name column holds a full , so a + /// projection name can map to several rows (blue/green versions, custom shard keys, per-tenant + /// partitions). Resolution: + /// + /// candidates are the rows whose parsed equals + /// and whose parsed equals + /// — a null matches only the bare + /// store-global rows (no tenant suffix); + /// when several match (a blue/green deploy), the NEWEST version wins, tie-broken on the highest + /// sequence; + /// no match returns null. + /// + /// and + /// are always null: Marten models the columns but no daemon path writes them (jasperfx#519). + /// + public async ValueTask ReadProjectionProgressAsync( + string projectionName, string? tenantId, CancellationToken token) + { + await EnsureStorageExistsAsync(typeof(IEvent), token).ConfigureAwait(false); + + await using var conn = CreateConnection(); + try + { + await conn.OpenAsync(token).ConfigureAwait(false); + + var builder = new CommandBuilder(); + // The trailing ':' guards against a projection whose name is a prefix of another + // (e.g. "Orders" must not match "OrdersHistory:All"). + builder.Append( + $"select name, last_seq_id from {Options.EventGraph.DatabaseSchemaName}.mt_event_progression where name like "); + builder.AppendParameter(projectionName + ":%"); + + ShardName? best = null; + var bestSequence = 0L; + + await using var reader = await conn.ExecuteReaderAsync(builder, token).ConfigureAwait(false); + while (await reader.ReadAsync(token).ConfigureAwait(false)) + { + var name = await reader.GetFieldValueAsync(0, token).ConfigureAwait(false); + var sequence = await reader.GetFieldValueAsync(1, token).ConfigureAwait(false); + + if (!ShardName.TryParse(name, out var shard) || shard is null) + { + continue; + } + + if (shard.Name != projectionName || shard.TenantId != tenantId) + { + continue; + } + + // Newest version wins; tie-break on the furthest-along sequence. + if (best is null || shard.Version > best.Version || + (shard.Version == best.Version && sequence > bestSequence)) + { + best = shard; + bestSequence = sequence; + } + } + + return best is null ? null : new ProjectionProgressRow(projectionName, tenantId, bestSequence, null, null); + } + finally + { + await conn.CloseAsync().ConfigureAwait(false); + } + } + /// /// #4785 / jasperfx#473 — Marten override of the store-agnostic ///