diff --git a/Directory.Packages.props b/Directory.Packages.props
index 4daa549..af86ee1 100644
--- a/Directory.Packages.props
+++ b/Directory.Packages.props
@@ -22,8 +22,8 @@
IEventDatabase.QueryDeadLetterEventsAsync (dead-letter row drill-in), DeadLetterEvent.TenantId,
and the tenant-aware daemon surface. Polecat implements the dead-letter row query + per-tenant
FetchDeadLetterCountsAsync below. -->
-
-
+
+
-
+
@@ -69,7 +69,7 @@
-
+
diff --git a/src/Polecat.Tests/Daemon/projection_progression_tests.cs b/src/Polecat.Tests/Daemon/projection_progression_tests.cs
index c7eb8d1..3861185 100644
--- a/src/Polecat.Tests/Daemon/projection_progression_tests.cs
+++ b/src/Polecat.Tests/Daemon/projection_progression_tests.cs
@@ -142,6 +142,52 @@ public async Task read_composes_the_tenant_suffix()
global.ShouldBeNull();
}
+ // #333 / jasperfx#529 — the exact ShardName overload looks up the full identity verbatim, no collapsing.
+ [Fact]
+ public async Task read_by_shard_name_reads_the_exact_row()
+ {
+ var shardName = ShardName.Compose("ExactRead");
+ await RecordProgressAsync(shardName, ceiling: 55, upsert: true);
+
+ var row = await theStore.Database.ReadProjectionProgressAsync(shardName, default);
+
+ row.ShouldNotBeNull();
+ row!.ProjectionName.ShouldBe("ExactRead");
+ row.TenantId.ShouldBeNull();
+ row.Sequence.ShouldBe(55);
+ row.AgentStatus.ShouldBeNull();
+ row.LastHeartbeat.ShouldBeNull();
+ }
+
+ [Fact]
+ public async Task read_by_shard_name_returns_null_for_an_unknown_identity()
+ {
+ await RecordProgressAsync(ShardName.Compose("KnownOne"), ceiling: 3, upsert: true);
+
+ var row = await theStore.Database.ReadProjectionProgressAsync(
+ ShardName.Compose("KnownOne", "All", null, 9), default);
+
+ row.ShouldBeNull();
+ }
+
+ [Fact]
+ public async Task read_by_shard_name_targets_the_exact_version_and_tenant()
+ {
+ await RecordProgressAsync(ShardName.Compose("Versioned", "All", null, 2), ceiling: 25, upsert: true);
+ await RecordProgressAsync(ShardName.Compose("Versioned", "All", null, 3), ceiling: 40, upsert: true);
+ await RecordProgressAsync(ShardName.Compose("Versioned", "All", "Red"), ceiling: 7, upsert: true);
+
+ // Exact V2 — not the newest V3.
+ var v2 = await theStore.Database.ReadProjectionProgressAsync(
+ ShardName.Compose("Versioned", "All", null, 2), default);
+ v2!.Sequence.ShouldBe(25);
+
+ var tenant = await theStore.Database.ReadProjectionProgressAsync(
+ ShardName.Compose("Versioned", "All", "Red"), default);
+ tenant!.Sequence.ShouldBe(7);
+ tenant.TenantId.ShouldBe("Red");
+ }
+
private async Task RecordProgressAsync(ShardName shardName, long ceiling, bool upsert)
{
var events = theStore.Database.Events;
diff --git a/src/Polecat/Storage/PolecatDatabase.cs b/src/Polecat/Storage/PolecatDatabase.cs
index 2bce25e..996b037 100644
--- a/src/Polecat/Storage/PolecatDatabase.cs
+++ b/src/Polecat/Storage/PolecatDatabase.cs
@@ -268,6 +268,48 @@ public async Task> AllProjectionProgress(CancellationT
}, (_connectionString, _events, projectionName, tenantId, name), token);
}
+ // #333 / jasperfx#529 — exact per-cell progression read. Unlike the (projectionName, tenantId) overload
+ // above this does no collapsing: it looks up the single pc_event_progression row whose name equals the
+ // full ShardName.Identity verbatim (the daemon writes range.ShardName.Identity), so a blue/green deploy's
+ // versions, a sliced projection's shard keys, and per-tenant partitions each address their own row. A
+ // ShardKey of "All" is the projection's global cell. heartbeat + agent_status are only present under
+ // EnableExtendedProgressionTracking; without it AgentStatus/LastHeartbeat come back null.
+ public async ValueTask ReadProjectionProgressAsync(
+ ShardName name, CancellationToken token)
+ {
+ return await _resilience.ExecuteAsync(static async (state, ct) =>
+ {
+ var (connectionString, events, shard) = state;
+
+ await using var conn = new SqlConnection(connectionString);
+ await conn.OpenAsync(ct);
+
+ await using var cmd = conn.CreateCommand();
+ cmd.CommandText = events.EnableExtendedProgressionTracking
+ ? $"SELECT last_seq_id, heartbeat, agent_status FROM {events.ProgressionTableName} WHERE name = @name;"
+ : $"SELECT last_seq_id FROM {events.ProgressionTableName} WHERE name = @name;";
+ cmd.Parameters.AddWithValue("@name", shard.Identity);
+
+ await using var reader = await cmd.ExecuteReaderAsync(ct);
+ if (!await reader.ReadAsync(ct))
+ {
+ // No row for this identity yet — the meaningful "not observed" answer.
+ return (ProjectionProgressRow?)null;
+ }
+
+ var seq = reader.GetInt64(0);
+ DateTimeOffset? heartbeat = null;
+ string? agentStatus = null;
+ if (events.EnableExtendedProgressionTracking)
+ {
+ if (!reader.IsDBNull(1)) heartbeat = reader.GetDateTimeOffset(1);
+ if (!reader.IsDBNull(2)) agentStatus = reader.GetString(2);
+ }
+
+ return new ProjectionProgressRow(shard.Name, shard.TenantId, seq, agentStatus, heartbeat);
+ }, (_connectionString, _events, name), token);
+ }
+
public async Task FetchHighestEventSequenceNumber(CancellationToken token)
{
return await _resilience.ExecuteAsync(static async (state, ct) =>