ReadProjectionProgressAsync reports the persisted heartbeat and agent status (#5172)

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;
+
+/// <summary>
+/// #5172 — both <c>ReadProjectionProgressAsync</c> overloads hardcoded <c>AgentStatus</c> and
+/// <c>LastHeartbeat</c> 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: <c>ExtendedProgressionWriter</c>
+/// populates both on every flush, and the row-scan path (<c>AllProjectionProgress</c>) 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.
+/// </summary>
+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<ProjectionProgressRow?> read(string projectionName, string? tenantId) =>
+        ((IEventDatabase)theStore.Tenancy.Default.Database)
+        .ReadProjectionProgressAsync(projectionName, tenantId, CancellationToken.None);
+
+    private ValueTask<ProjectionProgressRow?> 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();
+    }
+}

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();
     }

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<IReadOnlyList<ShardState>> AllProjectionProgress(string? tenan
     /// sequence;</item>
     /// <item>no match returns null.</item>
     /// </list>
-    /// <see cref="ProjectionProgressRow.AgentStatus"/> and <see cref="ProjectionProgressRow.LastHeartbeat"/>
-    /// are always null: Marten models the columns but no daemon path writes them (jasperfx#519).
+    /// <para>
+    /// #5172: <see cref="ProjectionProgressRow.AgentStatus"/> and
+    /// <see cref="ProjectionProgressRow.LastHeartbeat"/> carry the real persisted values when
+    /// <see cref="EventGraph.EnableExtendedProgressionTracking"/> 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 <see cref="AllProjectionProgress(string?,CancellationToken)"/>. With
+    /// extended tracking off the columns do not exist on the table and both stay null.
+    /// </para>
     /// </summary>
     public async ValueTask<ProjectionProgressRow?> 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<IReadOnlyList<ShardState>> 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,40 +456,70 @@ public async Task<IReadOnlyList<ShardState>> AllProjectionProgress(string? tenan
                 {
                     best = shard;
                     bestSequence = sequence;
+
+                    if (extended)
+                    {
+                        bestStatus = await readNullableAsync<string>(reader, 2, token).ConfigureAwait(false);
+                        bestHeartbeat = await readNullableStructAsync<DateTimeOffset>(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
         {
             await conn.CloseAsync().ConfigureAwait(false);
         }
     }
 
+    private static async Task<T?> readNullableAsync<T>(DbDataReader reader, int index, CancellationToken token)
+        where T : class
+    {
+        return await reader.IsDBNullAsync(index, token).ConfigureAwait(false)
+            ? null
+            : await reader.GetFieldValueAsync<T>(index, token).ConfigureAwait(false);
+    }
+
+    private static async Task<T?> readNullableStructAsync<T>(DbDataReader reader, int index, CancellationToken token)
+        where T : struct
+    {
+        return await reader.IsDBNullAsync(index, token).ConfigureAwait(false)
+            ? null
+            : await reader.GetFieldValueAsync<T>(index, token).ConfigureAwait(false);
+    }
+
     /// <summary>
     /// #4975 / jasperfx#529 — exact per-cell progression read. Unlike the
     /// <see cref="ReadProjectionProgressAsync(string,string?,CancellationToken)"/> overload this does no
     /// version/shard collapsing: it looks up the single <c>mt_event_progression</c> row whose <c>name</c>
     /// equals <paramref name="name"/>'s <see cref="ShardName.Identity"/> verbatim, so a blue/green deploy's
     /// versions, a sliced projection's shard keys, and per-tenant partitions each address their own row.
     /// A <see cref="ShardName.ShardKey"/> of <c>All</c> is the projection's global cell. Returns null when no
-    /// row exists for that identity; <see cref="ProjectionProgressRow.AgentStatus"/> and
-    /// <see cref="ProjectionProgressRow.LastHeartbeat"/> stay null (jasperfx#519).
+    /// row exists for that identity. <see cref="ProjectionProgressRow.AgentStatus"/> and
+    /// <see cref="ProjectionProgressRow.LastHeartbeat"/> carry the persisted values when
+    /// <see cref="EventGraph.EnableExtendedProgressionTracking"/> is on, and stay null when it is off and
+    /// the columns do not exist (#5172).
     /// </summary>
     public async ValueTask<ProjectionProgressRow?> 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<IReadOnlyList<ShardState>> AllProjectionProgress(string? tenan
             }
 
             var sequence = await reader.GetFieldValueAsync<long>(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<string>(reader, 1, token).ConfigureAwait(false);
+            var heartbeat = await readNullableStructAsync<DateTimeOffset>(reader, 2, token).ConfigureAwait(false);
+
+            return new ProjectionProgressRow(name.Name, name.TenantId, sequence, status, heartbeat);
         }
         finally
         {