From 5c751b997e994b5773b1cdf69a29958e94b14563 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Tue, 4 Aug 2026 09:16:58 -0500 Subject: [PATCH] ReadProjectionProgressAsync reports the persisted heartbeat and agent status (#5172) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both overloads hardcoded `LastHeartbeat` and `AgentStatus` to null and never selected the columns, on the rationale that Marten modelled them but no daemon path wrote them (jasperfx#519). That stopped being true at jasperfx#537 — `ExtendedProgressionWriter` populates both on every flush, and `AllProjectionProgress` reads them back correctly. So the targeted per-cell read that jasperfx#435 added *specifically* as the monitoring alternative to scanning every row returned NULL for exactly the two fields a monitor calls it for, with nothing to distinguish a placeholder from a fact — while the expensive path it was meant to replace returned the truth. Both queries now select `agent_status` and `heartbeat` when `EnableExtendedProgressionTracking` is on and hydrate them into the record. With extended tracking off the columns are not on the table, so the narrow column list is kept and both fields stay null. In the version-collapsing overload the telemetry follows the winning row rather than whichever candidate was read first. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017CTtw2kVRSZKp1p5RTxgAy --- ...Bug_5172_read_progress_extended_columns.cs | 162 ++++++++++++++++++ .../Internals/read_projection_progress.cs | 3 +- .../Storage/MartenDatabase.EventStorage.cs | 67 +++++++- 3 files changed, 223 insertions(+), 9 deletions(-) create mode 100644 src/DaemonTests/Bugs/Bug_5172_read_progress_extended_columns.cs diff --git a/src/DaemonTests/Bugs/Bug_5172_read_progress_extended_columns.cs b/src/DaemonTests/Bugs/Bug_5172_read_progress_extended_columns.cs new file mode 100644 index 0000000000..5f0e55ac47 --- /dev/null +++ b/src/DaemonTests/Bugs/Bug_5172_read_progress_extended_columns.cs @@ -0,0 +1,162 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Projections; +using Marten.Events.Daemon.Progress; +using Marten.Storage; +using Marten.Testing; +using Marten.Testing.Harness; +using Shouldly; +using Xunit; + +namespace DaemonTests.Bugs; + +/// +/// #5172 — both ReadProjectionProgressAsync overloads hardcoded AgentStatus and +/// LastHeartbeat to null and never selected the columns, on the rationale (jasperfx#519) that no +/// daemon path wrote them. That stopped being true at jasperfx#537: ExtendedProgressionWriter +/// populates both on every flush, and the row-scan path (AllProjectionProgress) reads them back +/// correctly. Only these targeted per-cell reads — the ones a monitor is meant to use instead of pulling +/// every row — were left returning a placeholder that reads exactly like a fact. +/// +public class Bug_5172_read_progress_extended_columns: OneOffConfigurationsContext, IAsyncLifetime +{ + private static readonly DateTimeOffset TheHeartbeat = + new(2026, 8, 4, 12, 30, 15, TimeSpan.Zero); + + public override ValueTask DisposeAsync() + { + Dispose(); + return base.DisposeAsync(); + } + + private async Task withExtendedTracking(bool enabled) + { + StoreOptions(x => x.Events.EnableExtendedProgressionTracking = enabled); + await theStore.Advanced.Clean.DeleteAllEventDataAsync(); + await theStore.EnsureStorageExistsAsync(typeof(IEvent)); + } + + private async Task seedProgression(params (ShardName name, long sequence)[] rows) + { + await using var session = theStore.LightweightSession(); + foreach (var (name, sequence) in rows) + { + session.QueueOperation(new InsertProjectionProgress(theStore.Events, new EventRange(name, sequence))); + } + + await session.SaveChangesAsync(); + } + + private Task writeTelemetry(ShardName name, string status, DateTimeOffset heartbeat) + { + var state = new ShardState(name, 0) { AgentStatus = status, LastHeartbeat = heartbeat }; + return ((IEventDatabase)theStore.Tenancy.Default.Database) + .WriteExtendedProgressionAsync([state], CancellationToken.None); + } + + private ValueTask read(string projectionName, string? tenantId) => + ((IEventDatabase)theStore.Tenancy.Default.Database) + .ReadProjectionProgressAsync(projectionName, tenantId, CancellationToken.None); + + private ValueTask read(ShardName name) => + ((IEventDatabase)theStore.Tenancy.Default.Database) + .ReadProjectionProgressAsync(name, CancellationToken.None); + + [Fact] + public async Task the_projection_name_overload_reads_the_persisted_status_and_heartbeat() + { + await withExtendedTracking(true); + + var shard = ShardName.Compose("Orders"); + await seedProgression((shard, 42)); + await writeTelemetry(shard, "Running", TheHeartbeat); + + var row = await read("Orders", null); + + row.ShouldNotBeNull(); + row.Sequence.ShouldBe(42); + row.AgentStatus.ShouldBe("Running"); + row.LastHeartbeat.ShouldNotBeNull(); + row.LastHeartbeat.Value.ToUniversalTime().ShouldBe(TheHeartbeat); + } + + [Fact] + public async Task the_exact_shard_name_overload_reads_the_persisted_status_and_heartbeat() + { + await withExtendedTracking(true); + + var shard = ShardName.Compose("Orders", tenantId: "tenant1"); + await seedProgression((shard, 17)); + await writeTelemetry(shard, "Paused", TheHeartbeat); + + var row = await read(shard); + + row.ShouldNotBeNull(); + row.Sequence.ShouldBe(17); + row.TenantId.ShouldBe("tenant1"); + row.AgentStatus.ShouldBe("Paused"); + row.LastHeartbeat!.Value.ToUniversalTime().ShouldBe(TheHeartbeat); + } + + // The version-collapsing overload picks the newest version's row -- the telemetry it reports has to + // come from that same row, not from whichever candidate happened to be read first. + [Fact] + public async Task the_winning_version_supplies_the_telemetry() + { + await withExtendedTracking(true); + + var v1 = ShardName.Compose("Orders"); + var v3 = ShardName.Compose("Orders", version: 3); + await seedProgression((v1, 10), (v3, 40)); + + await writeTelemetry(v1, "Stopped", TheHeartbeat.AddHours(-5)); + await writeTelemetry(v3, "Running", TheHeartbeat); + + var row = await read("Orders", null); + + row!.Sequence.ShouldBe(40); + row.AgentStatus.ShouldBe("Running"); + row.LastHeartbeat!.Value.ToUniversalTime().ShouldBe(TheHeartbeat); + } + + // A row that exists but has never been decorated still reads back cleanly -- the columns are simply + // NULL, which is a fact ("nothing reported yet") rather than the old unconditional placeholder. + [Fact] + public async Task an_undecorated_row_reports_nulls() + { + await withExtendedTracking(true); + + var shard = ShardName.Compose("Orders"); + await seedProgression((shard, 42)); + + var row = await read("Orders", null); + + row!.Sequence.ShouldBe(42); + row.AgentStatus.ShouldBeNull(); + row.LastHeartbeat.ShouldBeNull(); + } + + // With extended tracking off the columns are not on the table at all, so both overloads must keep + // selecting the narrow list rather than failing on an undefined column. + [Fact] + public async Task without_extended_tracking_both_overloads_still_work_and_report_nulls() + { + await withExtendedTracking(false); + + var shard = ShardName.Compose("Orders"); + await seedProgression((shard, 42)); + + var byName = await read("Orders", null); + byName!.Sequence.ShouldBe(42); + byName.AgentStatus.ShouldBeNull(); + byName.LastHeartbeat.ShouldBeNull(); + + var byShard = await read(shard); + byShard!.Sequence.ShouldBe(42); + byShard.AgentStatus.ShouldBeNull(); + byShard.LastHeartbeat.ShouldBeNull(); + } +} diff --git a/src/DaemonTests/Internals/read_projection_progress.cs b/src/DaemonTests/Internals/read_projection_progress.cs index 25d4f1eb28..1dca907a64 100644 --- a/src/DaemonTests/Internals/read_projection_progress.cs +++ b/src/DaemonTests/Internals/read_projection_progress.cs @@ -59,7 +59,8 @@ public async Task reads_the_store_global_row() row.ProjectionName.ShouldBe("Orders"); row.TenantId.ShouldBeNull(); row.Sequence.ShouldBe(42); - // Marten models the columns but writes neither — always null (jasperfx#519). + // This store leaves EnableExtendedProgressionTracking off, so the columns don't exist and there + // is nothing to report. See Bug_5172 for the extended-tracking half. row.AgentStatus.ShouldBeNull(); row.LastHeartbeat.ShouldBeNull(); } diff --git a/src/Marten/Storage/MartenDatabase.EventStorage.cs b/src/Marten/Storage/MartenDatabase.EventStorage.cs index e32aa273a0..7a627e02fe 100644 --- a/src/Marten/Storage/MartenDatabase.EventStorage.cs +++ b/src/Marten/Storage/MartenDatabase.EventStorage.cs @@ -1,6 +1,7 @@ #nullable enable using System; using System.Collections.Generic; +using System.Data.Common; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -399,14 +400,22 @@ public async Task> AllProjectionProgress(string? tenan /// sequence; /// no match returns null. /// - /// and - /// are always null: Marten models the columns but no daemon path writes them (jasperfx#519). + /// + /// #5172: and + /// carry the real persisted values when + /// is on — the daemon has written those + /// columns since jasperfx#537, and this targeted read is exactly the call a monitor makes instead of + /// pulling every row through . With + /// extended tracking off the columns do not exist on the table and both stay null. + /// /// public async ValueTask ReadProjectionProgressAsync( string projectionName, string? tenantId, CancellationToken token) { await EnsureStorageExistsAsync(typeof(IEvent), token).ConfigureAwait(false); + var extended = Options.EventGraph.EnableExtendedProgressionTracking; + await using var conn = CreateConnection(); try { @@ -415,12 +424,15 @@ public async Task> AllProjectionProgress(string? tenan 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"). + var columns = extended ? "name, last_seq_id, agent_status, heartbeat" : "name, last_seq_id"; builder.Append( - $"select name, last_seq_id from {Options.EventGraph.DatabaseSchemaName}.mt_event_progression where name like "); + $"select {columns} from {Options.EventGraph.DatabaseSchemaName}.mt_event_progression where name like "); builder.AppendParameter(projectionName + ":%"); ShardName? best = null; var bestSequence = 0L; + string? bestStatus = null; + DateTimeOffset? bestHeartbeat = null; await using var reader = await conn.ExecuteReaderAsync(builder, token).ConfigureAwait(false); while (await reader.ReadAsync(token).ConfigureAwait(false)) @@ -444,10 +456,19 @@ public async Task> AllProjectionProgress(string? tenan { best = shard; bestSequence = sequence; + + if (extended) + { + bestStatus = await readNullableAsync(reader, 2, token).ConfigureAwait(false); + bestHeartbeat = await readNullableStructAsync(reader, 3, token) + .ConfigureAwait(false); + } } } - return best is null ? null : new ProjectionProgressRow(projectionName, tenantId, bestSequence, null, null); + return best is null + ? null + : new ProjectionProgressRow(projectionName, tenantId, bestSequence, bestStatus, bestHeartbeat); } finally { @@ -455,6 +476,22 @@ public async Task> AllProjectionProgress(string? tenan } } + private static async Task readNullableAsync(DbDataReader reader, int index, CancellationToken token) + where T : class + { + return await reader.IsDBNullAsync(index, token).ConfigureAwait(false) + ? null + : await reader.GetFieldValueAsync(index, token).ConfigureAwait(false); + } + + private static async Task readNullableStructAsync(DbDataReader reader, int index, CancellationToken token) + where T : struct + { + return await reader.IsDBNullAsync(index, token).ConfigureAwait(false) + ? null + : await reader.GetFieldValueAsync(index, token).ConfigureAwait(false); + } + /// /// #4975 / jasperfx#529 — exact per-cell progression read. Unlike the /// overload this does no @@ -462,22 +499,27 @@ public async Task> AllProjectionProgress(string? tenan /// equals 's verbatim, so a blue/green deploy's /// versions, a sliced projection's shard keys, and per-tenant partitions each address their own row. /// A of All is the projection's global cell. Returns null when no - /// row exists for that identity; and - /// stay null (jasperfx#519). + /// row exists for that identity. and + /// carry the persisted values when + /// is on, and stay null when it is off and + /// the columns do not exist (#5172). /// public async ValueTask ReadProjectionProgressAsync( ShardName name, CancellationToken token) { await EnsureStorageExistsAsync(typeof(IEvent), token).ConfigureAwait(false); + var extended = Options.EventGraph.EnableExtendedProgressionTracking; + await using var conn = CreateConnection(); try { await conn.OpenAsync(token).ConfigureAwait(false); var builder = new CommandBuilder(); + var columns = extended ? "last_seq_id, agent_status, heartbeat" : "last_seq_id"; builder.Append( - $"select last_seq_id from {Options.EventGraph.DatabaseSchemaName}.mt_event_progression where name = "); + $"select {columns} from {Options.EventGraph.DatabaseSchemaName}.mt_event_progression where name = "); builder.AppendParameter(name.Identity); await using var reader = await conn.ExecuteReaderAsync(builder, token).ConfigureAwait(false); @@ -487,7 +529,16 @@ public async Task> AllProjectionProgress(string? tenan } var sequence = await reader.GetFieldValueAsync(0, token).ConfigureAwait(false); - return new ProjectionProgressRow(name.Name, name.TenantId, sequence, null, null); + + if (!extended) + { + return new ProjectionProgressRow(name.Name, name.TenantId, sequence, null, null); + } + + var status = await readNullableAsync(reader, 1, token).ConfigureAwait(false); + var heartbeat = await readNullableStructAsync(reader, 2, token).ConfigureAwait(false); + + return new ProjectionProgressRow(name.Name, name.TenantId, sequence, status, heartbeat); } finally {