From 277f3f4b1508a5ca670dd9280f94c2ea8e96f5d9 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Tue, 4 Aug 2026 06:59:33 -0500 Subject: [PATCH] feat: supported per-tenant projection lag read (#619) "How far behind is projection X, at its current version, for tenant T on this database?" had no supported answer, so the correlation had been reinvented three times -- the daemon's blue/green side-effect gate, Marten's WaitForNonStaleDataAsync, and application code in the field -- each rediscovering the same traps (marten#4761, marten#5161, marten#4797, marten#5170). - `ProjectionLag`: one value per (shard, tenant) cell, carrying the whole `ShardName` (so a sliced projection doesn't collapse to one row), the database identifier (so a 512-database fan-out stays attributable), and `HasProgressionRow` as a real field rather than a `Sequence == 0` sentinel. - `ProjectionLagCalculator`: the provider-neutral correlation, anchored on registered sources at their current version. A missing row is fully behind, not caught up; prior-version rows are never borrowed; non-shard bookkeeping rows are excluded; each tenant is measured against its own high-water mark; a store-global `:All` agent under a tenanted store keeps its own cell. - `IEventDatabase.FetchProjectionLagAsync` (+ a ShardName-scoped overload) as default interface methods over the single AllProjectionProgress round trip that already exists -- no new SQL -- and `IEventStore<,>` overloads that supply `AllShards()` as the registry. The one thing the API cannot fix is documented on it: the tenant set is discovered from the HighWaterMark:{tenant} rows, so a tenant that has never had one written is invisible rather than "fully behind". Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Fuk1GybEEmohFmboJuM4Po --- src/EventTests/Daemon/ProjectionLagTests.cs | 313 ++++++++++++++++++ src/JasperFx.Events/Daemon/ProjectionLag.cs | 66 ++++ .../Daemon/ProjectionLagCalculator.cs | 148 +++++++++ src/JasperFx.Events/IEventDatabase.cs | 54 +++ src/JasperFx.Events/IEventStore.cs | 24 +- 5 files changed, 604 insertions(+), 1 deletion(-) create mode 100644 src/EventTests/Daemon/ProjectionLagTests.cs create mode 100644 src/JasperFx.Events/Daemon/ProjectionLag.cs create mode 100644 src/JasperFx.Events/Daemon/ProjectionLagCalculator.cs diff --git a/src/EventTests/Daemon/ProjectionLagTests.cs b/src/EventTests/Daemon/ProjectionLagTests.cs new file mode 100644 index 0000000..8355c31 --- /dev/null +++ b/src/EventTests/Daemon/ProjectionLagTests.cs @@ -0,0 +1,313 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Projections; +using Shouldly; + +namespace EventTests.Daemon; + +/// +/// jasperfx#619 — the supported per-tenant projection lag read. Every rule pinned here exists +/// because a real deployment broke on its absence; the issue references are the incidents. +/// +public class ProjectionLagTests +{ + private static ShardState row(string identity, long sequence) => new(identity, sequence); + + private static IReadOnlyList calculate(IEnumerable shards, + params ShardState[] progress) + => ProjectionLagCalculator.Calculate(shards, progress, "db1"); + + [Fact] + public void lag_is_the_distance_to_the_mark() + { + var lag = new ProjectionLag(ShardName.Compose("Trips"), "db1", 40, 100, true); + lag.Lag.ShouldBe(60); + lag.IsCaughtUp.ShouldBeFalse(); + } + + [Fact] + public void lag_never_goes_negative_when_a_shard_reads_past_a_stale_mark() + { + new ProjectionLag(ShardName.Compose("Trips"), "db1", 120, 100, true).Lag.ShouldBe(0); + } + + [Fact] + public void a_cell_with_no_row_is_never_caught_up() + { + // The ambiguity the HasProgressionRow field exists to kill: a readiness probe that + // conflates "never started" with "at zero" latches green during a version bump. + var lag = new ProjectionLag(ShardName.Compose("Trips"), "db1", 0, 0, false); + lag.Lag.ShouldBe(0); + lag.IsCaughtUp.ShouldBeFalse(); + } + + [Fact] + public void store_global_store_correlates_each_registered_shard_against_the_global_mark() + { + var lags = calculate([ShardName.Compose("Trips"), ShardName.Compose("Orders")], + row(ShardState.HighWaterMark, 100), + row("Trips:All", 90), + row("Orders:All", 100)); + + lags.Count.ShouldBe(2); + lags.Single(x => x.Shard.Name == "Trips").Lag.ShouldBe(10); + lags.Single(x => x.Shard.Name == "Orders").IsCaughtUp.ShouldBeTrue(); + lags.ShouldAllBe(x => x.DatabaseIdentifier == "db1"); + } + + [Fact] + public void a_registered_shard_with_no_row_at_all_is_fully_behind() + { + var lags = calculate([ShardName.Compose("Trips")], row(ShardState.HighWaterMark, 100)); + + var lag = lags.ShouldHaveSingleItem(); + lag.HasProgressionRow.ShouldBeFalse(); + lag.Sequence.ShouldBe(0); + lag.HighWaterMark.ShouldBe(100); + lag.Lag.ShouldBe(100); + lag.IsCaughtUp.ShouldBeFalse(); + } + + [Fact] + public void a_prior_versions_row_is_never_borrowed_by_the_current_version() + { + // The blue/green trap: V2's row still sits at the mark, but V3 is what is registered. + var lags = calculate([ShardName.Compose("Trips", version: 3)], + row(ShardState.HighWaterMark, 100), + row("Trips:V2:All", 100)); + + var lag = lags.ShouldHaveSingleItem(); + lag.Shard.Version.ShouldBe(3u); + lag.HasProgressionRow.ShouldBeFalse(); + lag.IsCaughtUp.ShouldBeFalse(); + lag.Lag.ShouldBe(100); + } + + [Fact] + public void non_shard_bookkeeping_rows_are_excluded() + { + // marten#5161: rows that are not projection shards never advance, and reporting them as + // projections that are permanently behind is what broke the reporter's status page. + var lags = calculate([ShardName.Compose("Trips")], + row(ShardState.HighWaterMark, 100), + row("Trips:All", 100), + row("some_bookkeeping_row", 0)); + + lags.ShouldHaveSingleItem().Shard.Name.ShouldBe("Trips"); + } + + [Fact] + public void each_tenant_is_measured_against_its_own_high_water_mark() + { + // marten#4761: under per-tenant event partitioning every tenant draws its own sequence, + // so a store-global bar attributes one tenant's height to all of them. + var lags = calculate([ShardName.Compose("Trips")], + row(ShardState.HighWaterMark, 5000), + row("HighWaterMark:acme", 100), + row("HighWaterMark:zeta", 20), + row("Trips:All:acme", 100), + row("Trips:All:zeta", 5)); + + lags.Count.ShouldBe(2); + + var acme = lags.Single(x => x.Shard.TenantId == "acme"); + acme.HighWaterMark.ShouldBe(100); + acme.IsCaughtUp.ShouldBeTrue(); + + var zeta = lags.Single(x => x.Shard.TenantId == "zeta"); + zeta.HighWaterMark.ShouldBe(20); + zeta.Lag.ShouldBe(15); + } + + [Fact] + public void a_tenant_with_no_row_for_a_registered_shard_is_fully_behind_not_missing() + { + var lags = calculate([ShardName.Compose("Trips")], + row("HighWaterMark:acme", 100), + row("HighWaterMark:zeta", 60), + row("Trips:All:acme", 100)); + + lags.Count.ShouldBe(2); + var zeta = lags.Single(x => x.Shard.TenantId == "zeta"); + zeta.HasProgressionRow.ShouldBeFalse(); + zeta.Lag.ShouldBe(60); + } + + [Fact] + public void a_store_global_agent_under_a_tenanted_store_keeps_its_own_cell() + { + // The marten#4761 follow-up: a single :All agent records no per-tenant rows at all. Without + // its own cell it disappears from the report entirely. + var lags = calculate([ShardName.Compose("Trips"), ShardName.Compose("Orders")], + row(ShardState.HighWaterMark, 120), + row("HighWaterMark:acme", 100), + row("Trips:All:acme", 100), + row("Orders:All", 60)); + + var orders = lags.Where(x => x.Shard.Name == "Orders").ToArray(); + orders.Length.ShouldBe(2); + + var global = orders.Single(x => x.Shard.TenantId == null); + global.HasProgressionRow.ShouldBeTrue(); + global.Sequence.ShouldBe(60); + global.HighWaterMark.ShouldBe(120); + + // ...and the tenant cell for the same shard is still reported as behind + orders.Single(x => x.Shard.TenantId == "acme").HasProgressionRow.ShouldBeFalse(); + + // The per-tenant projection does NOT gain a spurious store-global cell + lags.Count(x => x.Shard.Name == "Trips").ShouldBe(1); + } + + [Fact] + public void a_sliced_projection_reports_one_cell_per_slice() + { + // Flattening to (name, version, tenant) would collapse these into one indistinguishable row. + var lags = calculate( + [ShardName.Compose("Trips", "One"), ShardName.Compose("Trips", "Two")], + row(ShardState.HighWaterMark, 100), + row("Trips:One", 100), + row("Trips:Two", 40)); + + lags.Count.ShouldBe(2); + lags.Single(x => x.Shard.ShardKey == "One").IsCaughtUp.ShouldBeTrue(); + lags.Single(x => x.Shard.ShardKey == "Two").Lag.ShouldBe(60); + } + + [Fact] + public void an_already_fanned_out_registry_does_not_multiply_cells() + { + var lags = calculate( + [ShardName.Compose("Trips", "All", "acme"), ShardName.Compose("Trips", "All", "zeta")], + row("HighWaterMark:acme", 100), + row("HighWaterMark:zeta", 100), + row("Trips:All:acme", 100), + row("Trips:All:zeta", 100)); + + lags.Count.ShouldBe(2); + } + + [Fact] + public void filter_by_projection_name_spans_slices_and_tenants() + { + var lags = calculate( + [ShardName.Compose("Trips", "One"), ShardName.Compose("Trips", "Two"), ShardName.Compose("Orders")], + row(ShardState.HighWaterMark, 100), + row("Trips:One", 100), + row("Trips:Two", 40), + row("Orders:All", 100)); + + ProjectionLagCalculator.Filter(lags, ShardName.Compose("Trips")).Count.ShouldBe(2); + ProjectionLagCalculator.Filter(lags, ShardName.Compose("Trips", "Two")) + .ShouldHaveSingleItem().Lag.ShouldBe(60); + } + + [Fact] + public void filter_by_tenant_qualified_name() + { + var lags = calculate([ShardName.Compose("Trips")], + row("HighWaterMark:acme", 100), + row("HighWaterMark:zeta", 100), + row("Trips:All:acme", 90), + row("Trips:All:zeta", 10)); + + var filtered = ProjectionLagCalculator.Filter(lags, ShardName.Compose("Trips", "All", "zeta")); + filtered.ShouldHaveSingleItem().Lag.ShouldBe(90); + } + + [Fact] + public void filter_ignores_the_version_of_the_argument() + { + // The registry only ever holds the current version, so a caller must not have to know it + var lags = calculate([ShardName.Compose("Trips", version: 4)], + row(ShardState.HighWaterMark, 100), + row("Trips:V4:All", 100)); + + ProjectionLagCalculator.Filter(lags, ShardName.Compose("Trips")).ShouldHaveSingleItem() + .Shard.Version.ShouldBe(4u); + } + + [Fact] + public async Task database_default_correlates_over_one_all_projection_progress_round_trip() + { + var recording = new LagDatabase([ + row(ShardState.HighWaterMark, 100), + row("HighWaterMark:acme", 80), + row("Trips:All:acme", 30) + ]); + + // Default interface members are reached through the interface -- which is the point: + // a store that has not implemented anything new gets this read for free + IEventDatabase database = recording; + + var lags = await database.FetchProjectionLagAsync([ShardName.Compose("Trips")], + TestContext.Current.CancellationToken); + + recording.Reads.ShouldBe(1); + var lag = lags.ShouldHaveSingleItem(); + lag.DatabaseIdentifier.ShouldBe("lag-db"); + lag.Shard.TenantId.ShouldBe("acme"); + lag.Lag.ShouldBe(50); + } + + [Fact] + public async Task database_default_filters_to_the_named_shard() + { + IEventDatabase database = new LagDatabase([ + row(ShardState.HighWaterMark, 100), + row("Trips:All", 90), + row("Orders:All", 10) + ]); + + var lags = await database.FetchProjectionLagAsync( + [ShardName.Compose("Trips"), ShardName.Compose("Orders")], + ShardName.Compose("Orders"), + TestContext.Current.CancellationToken); + + lags.ShouldHaveSingleItem().Lag.ShouldBe(90); + } + + // A bare IEventDatabase that overrides nothing but the progression read, so the calls above + // exercise the default interface implementations added for jasperfx#619 + private sealed class LagDatabase : IEventDatabase + { + private readonly IReadOnlyList _progress; + + public LagDatabase(IReadOnlyList progress) => _progress = progress; + + public int Reads { get; private set; } + + public Task> AllProjectionProgress(CancellationToken token = default) + { + Reads++; + return Task.FromResult(_progress); + } + + public string Identifier => "lag-db"; + 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(); + } +} diff --git a/src/JasperFx.Events/Daemon/ProjectionLag.cs b/src/JasperFx.Events/Daemon/ProjectionLag.cs new file mode 100644 index 0000000..dbba4bb --- /dev/null +++ b/src/JasperFx.Events/Daemon/ProjectionLag.cs @@ -0,0 +1,66 @@ +using JasperFx.Events.Projections; + +namespace JasperFx.Events.Daemon; + +/// +/// How far behind the event stream a single registered projection/subscription cell is, at the +/// version that is registered right now. One value per (shard, tenant) cell. +/// +/// jasperfx#619: the "anchor on registered sources at their current version, and treat a missing +/// row as fully behind rather than caught up" semantic had grown three independent +/// implementations (the daemon's blue/green side-effect gate, Marten's +/// WaitForNonStaleDataAsync, and application code in the field), each of which had to +/// rediscover the same four traps: a store-global high-water bar is wrong when each tenant draws +/// its own sequence, a store-global :All agent records no per-tenant rows at all, +/// non-shard bookkeeping rows never advance and must be excluded, and a sliced projection fans +/// out across shard keys. +/// +/// +/// +/// The cell this lag describes. Carries , +/// , and +/// , so a projection sliced across custom shard keys reports one +/// value per slice instead of collapsing them. +/// +/// +/// this reading came from. A fan-out across many +/// databases is otherwise an unattributable list — every database publishes the same shard names. +/// +/// +/// The persisted progression for this cell at its CURRENT version, or 0 when there is no row for +/// it (see ). A prior version's row is never borrowed: the +/// version is baked into the progression-row identity, so a version bump reads as "no progress +/// yet", which is what it is. +/// +/// +/// The high-water mark this cell advances against — that tenant's own mark under per-tenant event +/// partitioning, not a store-global one. +/// +/// +/// Whether a progression row exists for this cell at its current version. A real field rather than +/// a Sequence == 0 sentinel: conflating "never started" with "at zero" is exactly what +/// latches a readiness probe green during a version bump while the previous version's row still +/// sits at the old mark. +/// +public readonly record struct ProjectionLag( + ShardName Shard, + string? DatabaseIdentifier, + long Sequence, + long HighWaterMark, + bool HasProgressionRow) +{ + /// + /// Events this cell still has to process. Never negative — a shard can legitimately read + /// ahead of a stale high-water snapshot. + /// + public long Lag => Math.Max(0, HighWaterMark - Sequence); + + /// + /// True only when this cell actually has a progression row AND that row has reached the mark. + /// A cell with no row is NOT caught up, however small the lag arithmetic makes it look. + /// + public bool IsCaughtUp => HasProgressionRow && Sequence >= HighWaterMark; + + public override string ToString() + => $"{Shard.Identity} @ {Sequence}/{HighWaterMark} (lag {Lag}{(HasProgressionRow ? "" : ", no row")})"; +} diff --git a/src/JasperFx.Events/Daemon/ProjectionLagCalculator.cs b/src/JasperFx.Events/Daemon/ProjectionLagCalculator.cs new file mode 100644 index 0000000..7013bc1 --- /dev/null +++ b/src/JasperFx.Events/Daemon/ProjectionLagCalculator.cs @@ -0,0 +1,148 @@ +using JasperFx.Events.Projections; + +namespace JasperFx.Events.Daemon; + +/// +/// The provider-neutral correlation behind — +/// registered shards (the current versions) × the progression rows of one database × the +/// high-water rows, with no database access of its own. Exposed publicly so a caller that already +/// holds both halves (a poller that fetched AllProjectionProgress for its own reasons, a +/// test, a store with a cheaper read) can run the same correlation without a second round trip. +/// See jasperfx#619. +/// +public static class ProjectionLagCalculator +{ + /// + /// Correlate registered shards against one database's progression rows. + /// + /// The rules, all of which exist because a real deployment broke on their absence: + /// + /// Anchor on , not on the rows. A row is only ever + /// consulted for a shard that is registered right now at the version it is registered at, so + /// a prior version's row can never be mistaken for current progress and non-shard bookkeeping + /// rows (marten#5161) can never masquerade as a projection that never advances. + /// A registered cell with no row is reported with HasProgressionRow == false and + /// Sequence == 0 — fully behind, not caught up. + /// Tenants are discovered from the HighWaterMark:{tenant} rows. When any exist, + /// each registered shard reports one cell per tenant against THAT tenant's mark (marten#4761: + /// under per-tenant event partitioning every tenant draws its own sequence, so a store-global + /// bar is meaningless). + /// A registered shard that is still running store-global under a tenanted store — a + /// single :All agent, which records no per-tenant rows at all — keeps its own cell, + /// measured against the store-global mark, falling back to the highest tenant mark when the + /// store-global row is absent. Without this that agent is invisible. + /// + /// + /// + /// + /// The shard names registered in the running application, at their current versions — i.e. + /// IEventStore<,>.AllShards() / ProjectionGraph.AllShards() projected onto + /// Name. Tenant-qualified names are accepted and reduced to their store-global form + /// before the tenant fan-out, so passing an already-fanned-out list does not multiply cells. + /// + /// Every progression row of one database, as returned by AllProjectionProgress. + /// Stamped onto every result so a multi-database fan-out stays attributable. + public static IReadOnlyList Calculate( + IEnumerable registeredShards, + IReadOnlyList progress, + string? databaseIdentifier = null) + { + var tenantMarks = new Dictionary(); + var sequences = new Dictionary(); + long globalMark = 0; + + foreach (var state in progress) + { + // Anything that isn't a shard identity we understand is bookkeeping, and bookkeeping rows + // never advance. Dropping them here is what keeps them from being reported as a projection + // that is permanently behind (marten#5161). + if (!ShardName.TryParse(state.ShardName, out var parsed) || parsed == null) continue; + + if (parsed.IsHighWaterMark) + { + if (parsed.TenantId == null) + { + globalMark = Math.Max(globalMark, state.Sequence); + } + else + { + tenantMarks[parsed.TenantId] = + Math.Max(tenantMarks.TryGetValue(parsed.TenantId, out var current) ? current : 0, + state.Sequence); + } + + continue; + } + + sequences[parsed.Identity] = + Math.Max(sequences.TryGetValue(parsed.Identity, out var existing) ? existing : 0, state.Sequence); + } + + // Deterministic ordering so a status endpoint's output doesn't churn between polls + var tenants = tenantMarks.Keys.OrderBy(x => x, StringComparer.Ordinal).ToArray(); + var globalCeiling = globalMark > 0 ? globalMark : tenantMarks.Values.DefaultIfEmpty(0).Max(); + + var results = new List(); + var seen = new HashSet(); + + foreach (var shard in registeredShards) + { + // A caller may hand us either form; reduce to the store-global identity so the tenant + // fan-out below is the ONLY thing that produces tenant cells + var name = shard.TenantId == null ? shard : shard.ForTenant(null); + if (!seen.Add(name.Identity)) continue; + + if (tenants.Length == 0) + { + results.Add(lagFor(name, globalMark)); + continue; + } + + var hasTenantRows = false; + foreach (var tenantId in tenants) + { + var cell = name.ForTenant(tenantId); + hasTenantRows |= sequences.ContainsKey(cell.Identity); + results.Add(lagFor(cell, tenantMarks[tenantId])); + } + + // A store-global agent under a tenanted store: it owns no per-tenant rows, so without its + // own cell it would vanish from the report entirely. Only emitted when its row exists — + // a projection that has simply never run is already fully described by the tenant cells. + if (!hasTenantRows && sequences.ContainsKey(name.Identity)) + { + results.Add(lagFor(name, globalCeiling)); + } + } + + return results; + + ProjectionLag lagFor(ShardName name, long mark) + { + var has = sequences.TryGetValue(name.Identity, out var sequence); + return new ProjectionLag(name, databaseIdentifier, has ? sequence : 0, mark, has); + } + } + + /// + /// Narrow a calculated set to the cells addressed by . + /// + /// Matching is on the projection always; on + /// only when the caller supplied something other than + /// (so the store-global name of a sliced projection means "every + /// slice"); and on only when the caller supplied one (so a + /// tenant-less name means "every tenant"). is deliberately + /// NOT matched: the read is anchored on the registry, which only ever holds the current + /// version, so a caller does not have to know what that version is to ask the question. + /// + /// + public static IReadOnlyList Filter(IEnumerable lags, ShardName name) + { + return lags.Where(x => + string.Equals(x.Shard.Name, name.Name, StringComparison.OrdinalIgnoreCase) + && (name.ShardKey == ShardName.All || + string.Equals(x.Shard.ShardKey, name.ShardKey, StringComparison.OrdinalIgnoreCase)) + && (name.TenantId == null || string.Equals(x.Shard.TenantId, name.TenantId))) + .ToList(); + } +} diff --git a/src/JasperFx.Events/IEventDatabase.cs b/src/JasperFx.Events/IEventDatabase.cs index 053aeeb..4c58f57 100644 --- a/src/JasperFx.Events/IEventDatabase.cs +++ b/src/JasperFx.Events/IEventDatabase.cs @@ -111,6 +111,60 @@ Task> AllProjectionProgress(string? tenantId, Cancella : throw new NotSupportedException( "Per-tenant AllProjectionProgress is not implemented on this IEventDatabase. Use an event store that implements per-tenant partitioning."); + /// + /// How far behind each registered projection/subscription cell is on this database, at the + /// version registered right now. This is the supported read behind "how far behind is + /// projection X, at its current version, for tenant T?" — the correlation every readiness + /// probe, status endpoint and lag explorer has otherwise had to re-derive from raw + /// progression-row names. See jasperfx#619, marten#5170. + /// + /// No new SQL: this is an in-memory correlation over the single + /// round trip plus the registry the + /// caller passes in. Rules — anchored on the registry so prior-version rows and non-shard + /// bookkeeping rows can never be mistaken for progress; a registered cell with NO row reports + /// HasProgressionRow == false (fully behind, not caught up); tenants are discovered + /// from the HighWaterMark:{tenant} rows and each cell is measured against its own + /// tenant's mark. See for the full set. + /// + /// + /// Documented limitation. The tenant set is discovered from the high-water rows + /// themselves. A tenant that exists but has never had a high-water row written produces no + /// row, and therefore no result — invisible rather than "fully behind". For a readiness probe + /// that is the same failure mode this API exists to eliminate, just relocated, so cross-check + /// the results against a tenant list from the store when you have one. + /// + /// + /// + /// The shards registered in the running application at their current versions — i.e. the + /// s of IEventStore<,>.AllShards(). + /// + /// + async Task> FetchProjectionLagAsync( + IReadOnlyList registeredShards, CancellationToken token = default) + { + var progress = await AllProjectionProgress(token).ConfigureAwait(false); + return ProjectionLagCalculator.Calculate(registeredShards, progress, Identifier); + } + + /// + /// The lag of the cells addressed by . The overload takes a + /// rather than a string? tenantId because a tenant-scoped read + /// is just a tenant-qualified shard name, and because suffix-matching a tenant onto a shard + /// identity is how marten#5171 happened. + /// + /// A of means every slice of a + /// sliced projection, a null means every tenant, and + /// is not matched at all — the read is anchored on the + /// registry, which only holds the current version. + /// + /// + async Task> FetchProjectionLagAsync( + IReadOnlyList registeredShards, ShardName name, CancellationToken token = default) + { + var all = await FetchProjectionLagAsync(registeredShards, token).ConfigureAwait(false); + return ProjectionLagCalculator.Filter(all, name); + } + /// /// Delete a single projection-progression row by its raw shard-name /// (e.g. claim_lines:V9:All), independent of whether a diff --git a/src/JasperFx.Events/IEventStore.cs b/src/JasperFx.Events/IEventStore.cs index 447dd39..31f5c38 100644 --- a/src/JasperFx.Events/IEventStore.cs +++ b/src/JasperFx.Events/IEventStore.cs @@ -398,7 +398,29 @@ public interface IEventStore : IEventStore where TOp ErrorHandlingOptions RebuildErrors { get; } IReadOnlyList> AllShards(); - + + /// + /// jasperfx#619: how far behind each registered projection/subscription cell is on + /// , at the version registered right now. Supplies + /// — "registered sources at their current version" — to + /// , + /// which is where the correlation and its caveats are documented. + /// + Task> FetchProjectionLagAsync(IEventDatabase database, CancellationToken token = default) + => database.FetchProjectionLagAsync(registeredShardNames(), token); + + /// + /// jasperfx#619: the lag of the cells addressed by on + /// . A tenant-scoped read is a tenant-qualified shard name. + /// + Task> FetchProjectionLagAsync(IEventDatabase database, ShardName name, + CancellationToken token = default) + => database.FetchProjectionLagAsync(registeredShardNames(), name, token); + + private IReadOnlyList registeredShardNames() + => AllShards().Select(x => x.Name).ToList(); + + /// /// TimeProvider used for event timestamping metadata. Replace for controlling the timestamps /// in testing