feat(#4962): targeted per-cell ReadProjectionProgressAsync on MartenDatabase

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<ProjectionProgressRow?> 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);
+    }
+}

src/Marten/Storage/MartenDatabase.EventStorage.cs

@@ -276,6 +276,76 @@ public async Task<IReadOnlyList<ShardState>> AllProjectionProgress(string? tenan
         }
     }
 
+    /// <summary>
+    /// #4962 / jasperfx#435 — targeted per-cell progression read for a single (projection, tenant).
+    /// The <c>mt_event_progression.name</c> column holds a full <see cref="ShardName.Identity"/>, so a
+    /// projection name can map to several rows (blue/green versions, custom shard keys, per-tenant
+    /// partitions). Resolution:
+    /// <list type="bullet">
+    /// <item>candidates are the rows whose parsed <see cref="ShardName.Name"/> equals
+    /// <paramref name="projectionName"/> and whose parsed <see cref="ShardName.TenantId"/> equals
+    /// <paramref name="tenantId"/> — a null <paramref name="tenantId"/> matches only the bare
+    /// store-global rows (no tenant suffix);</item>
+    /// <item>when several match (a blue/green deploy), the NEWEST version wins, tie-broken on the highest
+    /// 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).
+    /// </summary>
+    public async ValueTask<ProjectionProgressRow?> 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<string>(0, token).ConfigureAwait(false);
+                var sequence = await reader.GetFieldValueAsync<long>(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);
+        }
+    }
+
     /// <summary>
     /// #4785 / jasperfx#473 — Marten override of the store-agnostic
     /// <see cref="IEventDatabase.DeleteProjectionProgressByShardNameAsync"/>