diff --git a/src/EventTests/Daemon/ReadProjectionProgressDefaultsTests.cs b/src/EventTests/Daemon/ReadProjectionProgressDefaultsTests.cs new file mode 100644 index 00000000..82aae1c5 --- /dev/null +++ b/src/EventTests/Daemon/ReadProjectionProgressDefaultsTests.cs @@ -0,0 +1,143 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Projections; +using Shouldly; +using Xunit; + +namespace EventTests.Daemon; + +/// +/// jasperfx#435 — ReadProjectionProgressAsync ships as a default interface implementation so the +/// out-of-tree IEventDatabase implementors (Marten, Polecat) keep compiling on the 1.x line. These +/// pin the contract that matters to a consumer: "not implemented" and "no row yet" are different +/// answers, and the default must not conflate them. +/// +public class ReadProjectionProgressDefaultsTests +{ + // A bare IEventDatabase that does NOT override ReadProjectionProgressAsync, so the call below + // exercises the default. Everything else throws — those members are never invoked here. + private sealed class BareEventDatabase : IEventDatabase + { + public string Identifier => throw new NotImplementedException(); + public Uri DatabaseUri => throw new NotImplementedException(); + public ShardStateTracker Tracker => throw new NotImplementedException(); + public string StorageIdentifier => throw new NotImplementedException(); + + public Task StoreDeadLetterEventAsync(object storage, DeadLetterEvent deadLetterEvent, CancellationToken token) + => throw new NotImplementedException(); + + public Task EnsureStorageExistsAsync(Type storageType, CancellationToken token) + => throw new NotImplementedException(); + + public Task WaitForNonStaleProjectionDataAsync(TimeSpan timeout) + => throw new NotImplementedException(); + + public Task ProjectionProgressFor(ShardName name, CancellationToken token = default) + => throw new NotImplementedException(); + + public Task FindEventStoreFloorAtTimeAsync(DateTimeOffset timestamp, CancellationToken token) + => throw new NotImplementedException(); + + public Task FetchHighestEventSequenceNumber(CancellationToken token) + => throw new NotImplementedException(); + + public Task> AllProjectionProgress(CancellationToken token = default) + => throw new NotImplementedException(); + } + + // An IEventDatabase that DOES implement the query, standing in for Marten/Polecat. Proves the + // member is overridable and that a real implementation's null survives as "no row". + private sealed class ProgressionEventDatabase : BareDatabaseBase + { + public override ValueTask ReadProjectionProgressAsync( + string projectionName, string? tenantId, CancellationToken token) + => projectionName == "Orders" + ? new ValueTask( + new ProjectionProgressRow("Orders", tenantId, 42, "Running", null)) + : new ValueTask((ProjectionProgressRow?)null); + } + + // Split out so the overriding stub does not have to restate every unrelated member. + private abstract class BareDatabaseBase : IEventDatabase + { + public virtual ValueTask ReadProjectionProgressAsync( + string projectionName, string? tenantId, CancellationToken token) + => throw new NotImplementedException(); + + public string Identifier => throw new NotImplementedException(); + public Uri DatabaseUri => throw new NotImplementedException(); + public ShardStateTracker Tracker => throw new NotImplementedException(); + public string StorageIdentifier => throw new NotImplementedException(); + + public Task StoreDeadLetterEventAsync(object storage, DeadLetterEvent deadLetterEvent, CancellationToken token) + => throw new NotImplementedException(); + + public Task EnsureStorageExistsAsync(Type storageType, CancellationToken token) + => throw new NotImplementedException(); + + public Task WaitForNonStaleProjectionDataAsync(TimeSpan timeout) + => throw new NotImplementedException(); + + public Task ProjectionProgressFor(ShardName name, CancellationToken token = default) + => throw new NotImplementedException(); + + public Task FindEventStoreFloorAtTimeAsync(DateTimeOffset timestamp, CancellationToken token) + => throw new NotImplementedException(); + + public Task FetchHighestEventSequenceNumber(CancellationToken token) + => throw new NotImplementedException(); + + public Task> AllProjectionProgress(CancellationToken token = default) + => throw new NotImplementedException(); + } + + private readonly IEventDatabase theBareDatabase = new BareEventDatabase(); + private readonly IEventDatabase theProgressionDatabase = new ProgressionEventDatabase(); + + [Fact] + public async Task default_throws_rather_than_reporting_a_live_cell_as_absent() + { + // The critical distinction: null is the documented "no row for this pair yet" answer, so an + // unimplemented store must not return it. A monitoring UI would render a running projection + // as having no progress at all. + await Should.ThrowAsync(async () => + await theBareDatabase.ReadProjectionProgressAsync("Orders", null, CancellationToken.None)); + } + + [Fact] + public async Task default_throws_for_a_tenant_scoped_read_too() + { + await Should.ThrowAsync(async () => + await theBareDatabase.ReadProjectionProgressAsync("Orders", "tenant-1", CancellationToken.None)); + } + + [Fact] + public async Task an_implementing_store_can_return_a_row() + { + var row = await theProgressionDatabase.ReadProjectionProgressAsync("Orders", "tenant-1", CancellationToken.None); + + row.ShouldNotBeNull(); + row.ProjectionName.ShouldBe("Orders"); + row.TenantId.ShouldBe("tenant-1"); + row.Sequence.ShouldBe(42); + row.AgentStatus.ShouldBe("Running"); + row.LastHeartbeat.ShouldBeNull(); + } + + [Fact] + public async Task an_implementing_store_returns_null_when_no_row_exists_for_the_pair() + { + var row = await theProgressionDatabase.ReadProjectionProgressAsync("Unobserved", null, CancellationToken.None); + row.ShouldBeNull(); + } + + [Fact] + public void null_tenant_id_means_store_global_or_default_tenant() + { + new ProjectionProgressRow("Orders", null, 1, "Running", null).TenantId.ShouldBeNull(); + } +} diff --git a/src/JasperFx.Events/IEventDatabase.cs b/src/JasperFx.Events/IEventDatabase.cs index cf0591d3..ad6b9f72 100644 --- a/src/JasperFx.Events/IEventDatabase.cs +++ b/src/JasperFx.Events/IEventDatabase.cs @@ -174,4 +174,31 @@ Task> FetchDeadLetterCountsAsync(string? ten ? FetchDeadLetterCountsAsync(token) : throw new NotSupportedException( "Per-tenant FetchDeadLetterCountsAsync is not implemented on this IEventDatabase. Use an event store that implements per-tenant partitioning."); + + /// + /// Read the current sequence + lifecycle state for a single (projection, tenant) cell directly + /// from the progression table, without spinning up an . + /// Returns null when no row exists for the pair yet — e.g. the daemon has not yet observed this + /// projection for this tenant. A null means store-global on a + /// non-tenanted store, or the default-tenant row on a tenanted store. + /// + /// This is the targeted counterpart to , + /// which returns every row across every projection × tenant pair and leaves the caller to filter. + /// For a per-cell UI polling loop (e.g. 1Hz against a single visible batch) the targeted query is + /// materially cheaper. + /// + /// The default implementation throws rather than returning + /// null: null is the meaningful "no row yet" answer, so a store that simply has not implemented + /// this must not borrow it and report a live cell as absent. Event stores (Marten, Polecat) + /// override this against their progression table. See jasperfx#435. + /// + /// Name of the projection whose cell to read. + /// Tenant owning the cell. Null means store-global / default tenant. + /// + ValueTask ReadProjectionProgressAsync( + string projectionName, + string? tenantId, + CancellationToken token) + => throw new NotSupportedException( + "ReadProjectionProgressAsync is not implemented on this IEventDatabase. Use an event store (Marten or Polecat) that supports reading a single projection progression row."); } \ No newline at end of file diff --git a/src/JasperFx.Events/ProjectionProgressRow.cs b/src/JasperFx.Events/ProjectionProgressRow.cs new file mode 100644 index 00000000..db074969 --- /dev/null +++ b/src/JasperFx.Events/ProjectionProgressRow.cs @@ -0,0 +1,29 @@ +namespace JasperFx.Events; + +/// +/// Point in time state of a single (projection, tenant) progression cell, read straight from the +/// event store's progression table. This is the targeted, per-cell counterpart to +/// — it exists so a monitoring tool polling one +/// visible cell does not have to fetch and filter every projection × tenant row on each tick. +/// See jasperfx#435. +/// +/// Name of the projection this cell tracks. +/// +/// Tenant the cell belongs to. Null means store-global on a non-tenanted store, or the +/// default-tenant row on a tenanted store. +/// +/// Event sequence number this cell has processed through. +/// +/// Lifecycle state of the agent driving this cell. Left as a string rather than the +/// enum on purpose: this is a diagnostic read of whatever the +/// store persisted, and a store may report a state outside the enum's Running/Stopped/Paused. +/// +/// +/// Timestamp the cell last reported progress; null when the store does not track a heartbeat for it. +/// +public record ProjectionProgressRow( + string ProjectionName, + string? TenantId, + long Sequence, + string AgentStatus, + DateTimeOffset? LastHeartbeat);