diff --git a/docs/events/projections/healthchecks.md b/docs/events/projections/healthchecks.md index 8ed95a1e38..4911bfc402 100644 --- a/docs/events/projections/healthchecks.md +++ b/docs/events/projections/healthchecks.md @@ -113,9 +113,9 @@ Services.AddHealthChecks().AddMartenHighWaterHealthCheck( databaseFilter: db => LocallyOwnedDatabaseIdentifiers.Contains(db.Identifier), // Assert even under DaemonMode.ExternallyManaged (Wolverine-managed distribution). In this - // mode only the liveness heartbeat signal is used (never the sequence-gap fallback), because - // an external owner can legitimately pause the mark — so this requires - // Events.EnableExtendedProgressionTracking to be turned on. + // mode only the per-tenant poll-cycle signal is used (never the store-global sequence-gap + // heuristic), because an external owner can legitimately pause the mark — so a non-partitioned + // store has nothing to assert on there. includeExternallyManaged: true); ``` @@ -148,11 +148,22 @@ Both overloads register the settings through a factory, so replacing you need to reach further into DI than the predicate allows. Under `UseTenantPartitionedEvents` the high-water mark is tracked **per tenant** as -`HighWaterMark:` progression rows rather than a single store-global `HighWaterMark`. The -check evaluates those per-tenant rows too, using the liveness heartbeat (the sequence-gap -fallback is store-global and cannot be applied per tenant). Enable -`Events.EnableExtendedProgressionTracking` so the per-tenant heartbeats are persisted, otherwise -a stalled per-tenant high-water agent cannot be detected. +`HighWaterMark:` progression rows rather than a single store-global `HighWaterMark`, and +those are the rows the check evaluates. It uses a different — and better — signal there: every +vectorized per-tenant poll re-stamps the row's `last_updated` column whether or not the mark +advances, so its age proves the poll loop is still *cycling*, independent of whether new events +exist. A quiet tenant never false-positives, and the sequence-gap heuristic (which is inherently +store-global) is never applied per tenant. No extra configuration is needed — +`EnableExtendedProgressionTracking` is not involved +(see [marten#5174](https://github.com/JasperFx/marten/issues/5174)). + +::: warning +On the **store-global** `HighWaterMark` row `last_updated` only moves when the mark *advances*, so +it says nothing about liveness and the sequence-gap heuristic is the only signal available from the +database. That means a store-global poll loop that has wedged while fully caught up is not +detectable from the progression table. For in-process detection of that case use +`IProjectionDaemon.HighWaterAgent.IsStale` / `LastPolledAt` on the node hosting the daemon. +::: ::: tip INFO This health check does **not** force the high-water mark forward — it is detection only. diff --git a/src/DaemonTests.ManualOnly/HealthChecks/HighWaterHealthCheckTests.cs b/src/DaemonTests.ManualOnly/HealthChecks/HighWaterHealthCheckTests.cs index 9028f916d2..d30261aa45 100644 --- a/src/DaemonTests.ManualOnly/HealthChecks/HighWaterHealthCheckTests.cs +++ b/src/DaemonTests.ManualOnly/HealthChecks/HighWaterHealthCheckTests.cs @@ -73,38 +73,19 @@ private async Task appendEventsAsync(int count) await session.SaveChangesAsync(); } - private async Task seedHighWaterMarkAsync(long sequence) - { - var sql = - $"insert into {theStore.Events.DatabaseSchemaName}.mt_event_progression (name, last_seq_id, last_updated) " + - $"values ('HighWaterMark', {sequence}, now()) " + - "on conflict (name) do update set last_seq_id = excluded.last_seq_id"; - await theSession.ExecuteAsync(new NpgsqlCommand(sql)); - } - - // Seeds the HighWaterMark row's liveness heartbeat (jasperfx#539). Requires - // EnableExtendedProgressionTracking so the `heartbeat` column exists. - private async Task seedHighWaterHeartbeatAsync(long sequence, DateTimeOffset heartbeat) - { - var sql = - $"insert into {theStore.Events.DatabaseSchemaName}.mt_event_progression (name, last_seq_id, last_updated, heartbeat) " + - $"values ('HighWaterMark', {sequence}, now(), '{heartbeat:O}'::timestamptz) " + - "on conflict (name) do update set last_seq_id = excluded.last_seq_id, heartbeat = excluded.heartbeat"; - await theSession.ExecuteAsync(new NpgsqlCommand(sql)); - } + private Task seedHighWaterMarkAsync(long sequence) => seedProgressionRowAsync("HighWaterMark", sequence); // Seeds an arbitrary progression row (store-global "HighWaterMark" or a per-tenant - // "HighWaterMark:" row) with an optional liveness heartbeat. The heartbeat column only - // exists when EnableExtendedProgressionTracking is on, so pass a heartbeat only in that case. - private async Task seedProgressionRowAsync(string name, long sequence, DateTimeOffset? heartbeat = null) + // "HighWaterMark:" row). marten#5174: last_updated is the per-cycle liveness signal for + // per-tenant rows -- mt_mark_event_progression stamps it on EVERY vectorized poll, mark advance or + // not -- so it is settable here to simulate a poll loop that has stopped cycling. + private async Task seedProgressionRowAsync(string name, long sequence, DateTimeOffset? lastUpdated = null) { - var sql = heartbeat is { } hb - ? $"insert into {theStore.Events.DatabaseSchemaName}.mt_event_progression (name, last_seq_id, last_updated, heartbeat) " + - $"values ('{name}', {sequence}, now(), '{hb:O}'::timestamptz) " + - "on conflict (name) do update set last_seq_id = excluded.last_seq_id, heartbeat = excluded.heartbeat" - : $"insert into {theStore.Events.DatabaseSchemaName}.mt_event_progression (name, last_seq_id, last_updated) " + - $"values ('{name}', {sequence}, now()) " + - "on conflict (name) do update set last_seq_id = excluded.last_seq_id"; + var stamp = lastUpdated is { } at ? $"'{at:O}'::timestamptz" : "now()"; + var sql = + $"insert into {theStore.Events.DatabaseSchemaName}.mt_event_progression (name, last_seq_id, last_updated) " + + $"values ('{name}', {sequence}, {stamp}) " + + "on conflict (name) do update set last_seq_id = excluded.last_seq_id, last_updated = excluded.last_updated"; await theSession.ExecuteAsync(new NpgsqlCommand(sql)); } @@ -241,22 +222,25 @@ public async Task healthy_when_mark_advances_before_threshold() result.Status.ShouldBe(HealthStatus.Healthy); } - // ---- heartbeat primary signal (marten#4986) ------------------------------------------ + // ---- per-tenant poll-cycle signal (marten#4986, revised by marten#5174) --------------- + + // marten#5174: the ExtendedProgression `heartbeat` column is never written for high-water rows + // (ExtendedProgressionWriter.OnNext drops HighWaterMark states outright), so the check no longer + // reads it. For per-tenant rows the real per-cycle signal is last_updated, which + // mt_mark_event_progression stamps on every vectorized poll whether or not the mark advances. [Fact] - public async Task healthy_when_heartbeat_is_fresh_even_though_mark_is_behind() + public async Task healthy_when_the_per_tenant_poll_is_fresh_even_though_the_mark_is_behind() { - // ExtendedProgression on -> the heartbeat is the primary signal. A fresh heartbeat means the - // agent is cycling, so a mark sitting behind the latest event is NOT unhealthy (unlike the gap - // heuristic, which would trip here). + // A fresh poll cycle means the agent is alive, so a per-tenant mark sitting behind the latest + // event is NOT unhealthy. StoreOptions(x => { x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); x.Projections.AsyncMode = DaemonMode.Solo; - x.Events.EnableExtendedProgressionTracking = true; }); await appendEventsAsync(20); - await seedHighWaterHeartbeatAsync(1, _now.AddSeconds(-5)); // mark stuck at 1, but heartbeat is 5s old + await seedProgressionRowAsync("HighWaterMark:acme", 1, _now.AddSeconds(-5)); var result = await buildCheck(30.Seconds()).CheckHealthAsync(new HealthCheckContext(), TestContext.Current.CancellationToken); @@ -264,25 +248,52 @@ public async Task healthy_when_heartbeat_is_fresh_even_though_mark_is_behind() } [Fact] - public async Task unhealthy_when_heartbeat_is_stale_even_though_mark_is_caught_up() + public async Task unhealthy_when_the_per_tenant_poll_is_stale_even_though_the_mark_is_caught_up() { - // A stale heartbeat means the loop stopped cycling. This trips even when the mark is fully caught - // up (gap == 0) — the case the gap heuristic is blind to. + // A stale poll cycle means the loop stopped cycling. This trips even when the mark is fully + // caught up — the case the store-global gap heuristic is blind to. StoreOptions(x => { x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); x.Projections.AsyncMode = DaemonMode.Solo; - x.Events.EnableExtendedProgressionTracking = true; }); await appendEventsAsync(20); var stats = await theStore.Advanced.FetchEventStoreStatistics(token: TestContext.Current.CancellationToken); - await seedHighWaterHeartbeatAsync(stats.EventSequenceNumber, _now.AddSeconds(-90)); // caught up, heartbeat 90s old + await seedProgressionRowAsync("HighWaterMark:acme", stats.EventSequenceNumber, _now.AddSeconds(-90)); var result = await buildCheck(30.Seconds()).CheckHealthAsync(new HealthCheckContext(), TestContext.Current.CancellationToken); result.Status.ShouldBe(HealthStatus.Unhealthy); } + [Fact] + public async Task the_extended_progression_heartbeat_column_is_never_consulted() + { + // The pin for marten#5174. With ExtendedProgression on and a heartbeat that is *fresh*, a + // store-global mark stuck behind the latest event must still be caught by the gap heuristic -- + // proving the check no longer defers to a column the daemon does not write. (Before this change + // the fresh heartbeat short-circuited the whole assessment.) + StoreOptions(x => + { + x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); + x.Projections.AsyncMode = DaemonMode.Solo; + x.Events.EnableExtendedProgressionTracking = true; + }); + await appendEventsAsync(20); + await seedProgressionRowAsync("HighWaterMark", 1); + await theSession.ExecuteAsync(new NpgsqlCommand( + $"update {theStore.Events.DatabaseSchemaName}.mt_event_progression set heartbeat = '{_now:O}'::timestamptz where name = 'HighWaterMark'")); + + var check = buildCheck(30.Seconds()); + + (await check.CheckHealthAsync(new HealthCheckContext(), TestContext.Current.CancellationToken)).Status + .ShouldBe(HealthStatus.Healthy); + + _timeProvider.GetUtcNow().Returns(_now.AddSeconds(60)); + (await check.CheckHealthAsync(new HealthCheckContext(), TestContext.Current.CancellationToken)).Status + .ShouldBe(HealthStatus.Unhealthy); + } + // ---- autoRestart remediation (marten#4986) ------------------------------------------- [Fact] @@ -292,11 +303,10 @@ public async Task autorestart_triggers_a_restart_once_and_still_reports_unhealth { x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); x.Projections.AsyncMode = DaemonMode.Solo; - x.Events.EnableExtendedProgressionTracking = true; }); await appendEventsAsync(20); var stats = await theStore.Advanced.FetchEventStoreStatistics(token: TestContext.Current.CancellationToken); - await seedHighWaterHeartbeatAsync(stats.EventSequenceNumber, _now.AddSeconds(-90)); + await seedProgressionRowAsync("HighWaterMark:acme", stats.EventSequenceNumber, _now.AddSeconds(-90)); var daemon = Substitute.For(); var coordinator = new FakeCoordinator(daemon); @@ -319,11 +329,10 @@ public async Task without_autorestart_no_restart_is_attempted() { x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); x.Projections.AsyncMode = DaemonMode.Solo; - x.Events.EnableExtendedProgressionTracking = true; }); await appendEventsAsync(20); var stats = await theStore.Advanced.FetchEventStoreStatistics(token: TestContext.Current.CancellationToken); - await seedHighWaterHeartbeatAsync(stats.EventSequenceNumber, _now.AddSeconds(-90)); + await seedProgressionRowAsync("HighWaterMark:acme", stats.EventSequenceNumber, _now.AddSeconds(-90)); var daemon = Substitute.For(); var coordinator = new FakeCoordinator(daemon); @@ -475,19 +484,18 @@ public void scoped_overload_registers_the_settings_through_a_factory_carrying_th // ---- ExternallyManaged gate (marten#4991) -------------------------------------------- [Fact] - public async Task healthy_under_externally_managed_by_default_even_if_heartbeat_stale() + public async Task healthy_under_externally_managed_by_default_even_if_per_tenant_poll_stale() { // Default: ExternallyManaged (e.g. Wolverine-managed distribution) hosts no local daemon, so the - // check stays a no-op and a stale heartbeat is not asserted. + // check stays a no-op and a stalled poll is not asserted. StoreOptions(x => { x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); x.Projections.AsyncMode = DaemonMode.ExternallyManaged; - x.Events.EnableExtendedProgressionTracking = true; }); await appendEventsAsync(20); var stats = await theStore.Advanced.FetchEventStoreStatistics(token: TestContext.Current.CancellationToken); - await seedHighWaterHeartbeatAsync(stats.EventSequenceNumber, _now.AddSeconds(-90)); + await seedProgressionRowAsync("HighWaterMark:acme", stats.EventSequenceNumber, _now.AddSeconds(-90)); var result = await buildCheck(30.Seconds()).CheckHealthAsync(new HealthCheckContext(), TestContext.Current.CancellationToken); @@ -495,18 +503,17 @@ public async Task healthy_under_externally_managed_by_default_even_if_heartbeat_ } [Fact] - public async Task unhealthy_under_externally_managed_when_opted_in_and_heartbeat_stale() + public async Task unhealthy_under_externally_managed_when_opted_in_and_per_tenant_poll_stale() { - // Opted in: assert under ExternallyManaged too — via the heartbeat signal. + // Opted in: assert under ExternallyManaged too — via the per-tenant poll-cycle signal. StoreOptions(x => { x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); x.Projections.AsyncMode = DaemonMode.ExternallyManaged; - x.Events.EnableExtendedProgressionTracking = true; }); await appendEventsAsync(20); var stats = await theStore.Advanced.FetchEventStoreStatistics(token: TestContext.Current.CancellationToken); - await seedHighWaterHeartbeatAsync(stats.EventSequenceNumber, _now.AddSeconds(-90)); + await seedProgressionRowAsync("HighWaterMark:acme", stats.EventSequenceNumber, _now.AddSeconds(-90)); var result = await buildCheck(30.Seconds(), includeExternallyManaged: true) .CheckHealthAsync(new HealthCheckContext(), TestContext.Current.CancellationToken); @@ -515,10 +522,10 @@ public async Task unhealthy_under_externally_managed_when_opted_in_and_heartbeat } [Fact] - public async Task externally_managed_opted_in_does_not_use_gap_fallback_without_heartbeat() + public async Task externally_managed_opted_in_does_not_use_the_store_global_gap_heuristic() { - // Opted in but ExtendedProgression off -> no heartbeat. The gap fallback must be suppressed under - // ExternallyManaged because an external owner can legitimately pause the mark; it stays Healthy. + // Opted in, store-global row only. The gap heuristic must be suppressed under ExternallyManaged + // because an external owner can legitimately pause the mark; it stays Healthy. StoreOptions(x => { x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); @@ -538,16 +545,16 @@ public async Task externally_managed_opted_in_does_not_use_gap_fallback_without_ // ---- per-tenant high water (marten#4991) --------------------------------------------- [Fact] - public async Task detects_stale_per_tenant_high_water_via_heartbeat() + public async Task detects_a_stale_per_tenant_high_water_row() { // UseTenantPartitionedEvents persists HighWaterMark: rows rather than a single // store-global HighWaterMark. The original check matched only "HighWaterMark" and was blind to - // a stalled per-tenant agent; now a stale per-tenant heartbeat is detected. + // a stalled per-tenant agent; now a per-tenant row whose poll has stopped cycling is detected -- + // with no dependency on ExtendedProgression (marten#5174). StoreOptions(x => { x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); x.Projections.AsyncMode = DaemonMode.Solo; - x.Events.EnableExtendedProgressionTracking = true; }); await appendEventsAsync(20); await seedProgressionRowAsync("HighWaterMark:acme", 5, _now.AddSeconds(-90)); @@ -558,13 +565,12 @@ public async Task detects_stale_per_tenant_high_water_via_heartbeat() } [Fact] - public async Task healthy_when_per_tenant_high_water_heartbeat_is_fresh() + public async Task healthy_when_the_per_tenant_high_water_row_was_just_polled() { StoreOptions(x => { x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); x.Projections.AsyncMode = DaemonMode.Solo; - x.Events.EnableExtendedProgressionTracking = true; }); await appendEventsAsync(20); await seedProgressionRowAsync("HighWaterMark:acme", 5, _now.AddSeconds(-5)); @@ -575,24 +581,27 @@ public async Task healthy_when_per_tenant_high_water_heartbeat_is_fresh() } [Fact] - public async Task per_tenant_high_water_without_heartbeat_is_not_gap_assessed() + public async Task a_per_tenant_row_is_never_gap_assessed() { - // ExtendedProgression off -> a per-tenant row has no heartbeat, and there is no per-tenant - // highest-sequence to compute a meaningful gap (FetchHighestEventSequenceNumber is store-global), - // so the gap fallback must NOT run for it — otherwise a tenant with no new events false-positives. + // There is no per-tenant highest-sequence to compute a meaningful gap against + // (FetchHighestEventSequenceNumber is store-global), so the gap heuristic must never run for a + // per-tenant row -- otherwise a tenant with no new events false-positives. Its poll cycle is + // fresh, so it stays Healthy no matter how far behind the store-global sequence it sits. StoreOptions(x => { x.Projections.Add(new HwFakeProjection(), ProjectionLifecycle.Async); x.Projections.AsyncMode = DaemonMode.Solo; }); await appendEventsAsync(20); - await seedProgressionRowAsync("HighWaterMark:acme", 1); var check = buildCheck(30.Seconds()); + await seedProgressionRowAsync("HighWaterMark:acme", 1, _now); (await check.CheckHealthAsync(new HealthCheckContext(), TestContext.Current.CancellationToken)).Status.ShouldBe(HealthStatus.Healthy); + // The poll keeps cycling at the same mark; still healthy a full window later. _timeProvider.GetUtcNow().Returns(_now.AddSeconds(60)); + await seedProgressionRowAsync("HighWaterMark:acme", 1, _now.AddSeconds(60)); (await check.CheckHealthAsync(new HealthCheckContext(), TestContext.Current.CancellationToken)).Status.ShouldBe(HealthStatus.Healthy); } diff --git a/src/Marten.AspNetCore/Daemon/HighWaterHealthCheckExtensions.cs b/src/Marten.AspNetCore/Daemon/HighWaterHealthCheckExtensions.cs index 83eca6ab7a..2891a588d0 100644 --- a/src/Marten.AspNetCore/Daemon/HighWaterHealthCheckExtensions.cs +++ b/src/Marten.AspNetCore/Daemon/HighWaterHealthCheckExtensions.cs @@ -4,13 +4,16 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; +using JasperFx.Events; using JasperFx.Events.Daemon; using JasperFx.Events.Projections; using Marten.Events.Daemon.HighWater; using Marten.Storage; +using Weasel.Postgresql; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Diagnostics.HealthChecks; +using NpgsqlTypes; namespace Marten.Events.Daemon; @@ -22,21 +25,29 @@ namespace Marten.Events.Daemon; /// reports Healthy. This check instead detects that the high-water agent has stopped and, /// optionally, restarts it. /// -/// Two staleness signals, best first (marten#4986): +/// Two staleness signals, each used where it is actually valid (marten#4986, revised in +/// marten#5174): /// /// -/// Heartbeat age (primary). When ExtendedProgression is enabled, the -/// high-water agent stamps a liveness heartbeat on the HighWaterMark -/// progression row on every completed poll cycle (JasperFx/jasperfx#539). Its age -/// is a direct signal that the loop is cycling, independent of whether the -/// mark advances — so a quiet store with no new events never trips it. +/// Poll-cycle age — per-tenant rows. Under +/// UseTenantPartitionedEvents every vectorized per-tenant poll re-stamps the +/// HighWaterMark:<tenant> row through mt_mark_event_progression, +/// which always sets last_updated — even when the mark does not move. Its age +/// is therefore a direct signal that the loop is cycling, independent of +/// whether the mark advances, so a quiet tenant never trips it. No extra +/// write, and no dependency on ExtendedProgression. /// /// -/// Sequence gap (fallback). When ExtendedProgression is off, no heartbeat is -/// persisted, so the check falls back to the original heuristic: the store-global -/// mark sitting unchanged while later events pile up past it. +/// Sequence gap — the store-global row. There last_updated only moves +/// when the mark advances, so it says nothing about liveness and the original +/// heuristic is the honest one: the mark sitting unchanged while later events pile +/// up past it. /// /// +/// The ExtendedProgression heartbeat column is deliberately not consulted: +/// it is never written for high-water rows (ExtendedProgressionWriter.OnNext drops +/// HighWaterMark states outright), so reading it only made the check look like it had +/// a signal it did not have — marten#5174. /// /// /// marten#4991: on a multi-database (sharded / MultiTenantedWithShardedDatabases) @@ -45,26 +56,25 @@ namespace Marten.Events.Daemon; /// for — otherwise a probe fans a connection out across all N databases and (with /// autoRestart) would try to restart agents this node does not own. Under /// UseTenantPartitionedEvents the high-water mark is tracked per tenant -/// (HighWaterMark:<tenant> rows), and those are evaluated too — via the -/// heartbeat signal, which is the only reliable per-tenant staleness signal. +/// (HighWaterMark:<tenant> rows), and those are the rows evaluated. /// /// public static class HighWaterHealthCheckExtensions { /// /// Adds a health check that reports when the - /// high-water agent has stopped for at least — via its - /// liveness heartbeat where available, otherwise via the sequence-gap heuristic - /// (marten#4961 / marten#4986). + /// high-water agent has stopped for at least — via the age + /// of its last completed poll cycle on per-tenant rows, otherwise via the sequence-gap + /// heuristic (marten#4961 / marten#4986 / marten#5174). /// /// /// - /// How long the high-water agent may go without a heartbeat (or, on the fallback path, how - /// long the mark may sit unchanged while behind the latest event sequence) before the store - /// is considered unhealthy. Defaults to 30 seconds. + /// How long the high-water agent may go without completing a poll cycle (or, on the + /// store-global gap path, how long the mark may sit unchanged while behind the latest event + /// sequence) before the store is considered unhealthy. Defaults to 30 seconds. /// /// - /// Fallback path only: the gap (highest event sequence minus high-water mark) that is + /// Store-global gap path only: the gap (highest event sequence minus high-water mark) that is /// treated as "caught up" and never trips the check, absorbing the normal safe-harbor lag. /// Defaults to 1. /// @@ -95,8 +105,9 @@ public static class HighWaterHealthCheckExtensions /// , because in /// this store hosts no daemon and a frozen mark is legitimate. Set this to true to /// also assert under (e.g. Wolverine-managed - /// distribution) — in which case only the heartbeat signal is used, never the gap - /// fallback, since an external owner can legitimately pause the mark. Combine with + /// distribution) — in which case only the per-tenant poll-cycle signal is used, never the + /// store-global gap heuristic, since an external owner can legitimately pause the mark. + /// A non-partitioned store therefore has nothing to assert on there. Combine with /// so the check only asserts on databases the local node /// actually owns. Defaults to false. /// @@ -267,18 +278,18 @@ public async Task CheckHealthAsync( // Solo / HotCold host the daemon here, so use every available signal. ExternallyManaged // (e.g. Wolverine-managed distribution, marten#4991) hosts no local daemon — opt in with - // includeExternallyManaged to still assert, but only via the heartbeat signal (the gap - // fallback would false-positive when an external owner legitimately pauses the mark). - // Disabled and everything else stays a no-op. + // includeExternallyManaged to still assert, but only via the per-tenant poll-cycle signal + // (the store-global gap heuristic would false-positive when an external owner legitimately + // pauses the mark). Disabled and everything else stays a no-op. var mode = projections.AsyncMode; - bool heartbeatOnly; + bool skipGapHeuristic; if (mode is DaemonMode.Solo or DaemonMode.HotCold) { - heartbeatOnly = false; + skipGapHeuristic = false; } else if (mode == DaemonMode.ExternallyManaged && _includeExternallyManaged) { - heartbeatOnly = true; + skipGapHeuristic = true; } else { @@ -313,7 +324,7 @@ public async Task CheckHealthAsync( foreach (var database in scoped) { - var result = await checkDatabaseAsync(database, heartbeatOnly, perTenantHighWater, + var result = await checkDatabaseAsync(database, skipGapHeuristic, perTenantHighWater, cancellationToken) .ConfigureAwait(false); if (result.Status != HealthStatus.Healthy) @@ -330,15 +341,11 @@ public async Task CheckHealthAsync( } } - private async Task checkDatabaseAsync(IMartenDatabase database, bool heartbeatOnly, + private async Task checkDatabaseAsync(IMartenDatabase database, bool skipGapHeuristic, bool perTenantHighWater, CancellationToken token) { - var allProgress = await database.AllProjectionProgress(token).ConfigureAwait(false); - - var allHighWater = allProgress - .Where(x => string.Equals(x.ShardName, HighWaterMarkShard, StringComparison.Ordinal) - || x.ShardName.StartsWith(PerTenantHighWaterPrefix, StringComparison.Ordinal)) - .ToArray(); + var allHighWater = await readHighWaterRowsAsync(database, _store.Options.Events.DatabaseSchemaName, token) + .ConfigureAwait(false); // marten#4991: under UseTenantPartitionedEvents the authoritative rows are the per-tenant // HighWaterMark: rows (the original check matched only the exact "HighWaterMark" @@ -347,17 +354,9 @@ private async Task checkDatabaseAsync(IMartenDatabase databas // per-tenant rows, and evaluate only the authoritative set — so the store-global row (which is // intentionally frozen under partitioning, the daemon skips its loop) is never gap-assessed // there and can't false-positive. - var perTenantMode = perTenantHighWater || - allHighWater.Any(x => - x.ShardName.StartsWith(PerTenantHighWaterPrefix, StringComparison.Ordinal)); - - var highWaterRows = perTenantMode - ? allHighWater - .Where(x => x.ShardName.StartsWith(PerTenantHighWaterPrefix, StringComparison.Ordinal)) - .ToArray() - : allHighWater - .Where(x => string.Equals(x.ShardName, HighWaterMarkShard, StringComparison.Ordinal)) - .ToArray(); + var perTenantMode = perTenantHighWater || allHighWater.Any(x => x.IsPerTenant); + + var highWaterRows = allHighWater.Where(x => x.IsPerTenant == perTenantMode).ToArray(); // No HighWaterMark progression row yet -> the daemon has not started here. Nothing to assert. if (highWaterRows.Length == 0) @@ -371,21 +370,31 @@ private async Task checkDatabaseAsync(IMartenDatabase databas foreach (var row in highWaterRows) { - var isPerTenant = - !string.Equals(row.ShardName, HighWaterMarkShard, StringComparison.Ordinal); var key = trackingKey(database.Identifier, row.ShardName); - // Primary signal (marten#4986): the liveness heartbeat (jasperfx#539), present only when - // ExtendedProgression is enabled. Heartbeat age proves the poll loop is *cycling* - // independent of whether the mark *advances*, so a quiet store never trips it — a strictly - // better signal than the gap heuristic, and the only reliable per-tenant signal. Use it - // whenever it is available. - if (row.LastHeartbeat is { } lastHeartbeat) + // Primary signal, per-tenant rows only (marten#5174). Every vectorized per-tenant poll + // re-stamps HighWaterMark: through mt_mark_event_progression, which always sets + // last_updated = transaction_timestamp() — even when the mark does not move. So its age + // proves the poll loop is *cycling* independent of whether the mark *advances*, which is + // what a liveness check needs and what no other persisted column offers here. It costs + // nothing extra: the write already happens on every cycle. + // + // This replaces the ExtendedProgression `heartbeat` column, which this check used to read + // as its primary signal. That column is never written for high-water rows — + // ExtendedProgressionWriter.OnNext drops HighWaterMark states outright (pinned by + // skips_high_water_mark_and_all_projections_states) — so the branch was unreachable in any + // real deployment and the check silently degraded to the gap heuristic. + if (row.IsPerTenant) { - // The gap tracker is only for the fallback path; keep it clear while on the heartbeat path. + // The gap tracker is only for the store-global fallback path; keep it clear here. _tracker.Readings.TryRemove(key, out _); - var age = now - lastHeartbeat; + if (row.LastUpdated is not { } lastUpdated) + { + continue; + } + + var age = now - lastUpdated; if (age < _staleThreshold) { continue; @@ -393,16 +402,14 @@ private async Task checkDatabaseAsync(IMartenDatabase databas var restartNote = await tryAutoRestartAsync(database.Identifier, now, token).ConfigureAwait(false); return HealthCheckResult.Unhealthy( - $"Unhealthy: the high-water agent for '{shardDescription(database, row)}' last reported a liveness heartbeat {age.TotalSeconds:F0}s ago (at {lastHeartbeat:O}), exceeding the {_staleThreshold} staleness threshold. Its poll loop has stopped cycling (see JasperFx/jasperfx#539 / marten#4961).{restartNote}"); + $"Unhealthy: the high-water agent for '{shardDescription(database, row)}' last completed a poll cycle {age.TotalSeconds:F0}s ago (at {lastUpdated:O}), exceeding the {_staleThreshold} staleness threshold. Its poll loop has stopped cycling (see marten#4961).{restartNote}"); } - // No heartbeat. The gap fallback is only reliable for the store-global mark under a mode - // this store actually hosts: - // - heartbeatOnly (ExternallyManaged) -> an external owner may legitimately pause the mark. - // - per-tenant -> there is no per-tenant highest-sequence to compute a meaningful gap - // (FetchHighestEventSequenceNumber is store-global), so a tenant with no new events would - // look permanently "behind". Enable ExtendedProgression for per-tenant staleness. - if (heartbeatOnly || isPerTenant) + // Store-global row. last_updated only moves when the mark ADVANCES here + // (HighWaterDetector.persistDetectedMarkAsync returns early when nothing changed), so it is + // not a liveness signal and the gap heuristic is the only honest one. It is unusable under + // skipGapHeuristic (ExternallyManaged), where an external owner may legitimately pause the mark. + if (skipGapHeuristic) { _tracker.Readings.TryRemove(key, out _); continue; @@ -442,11 +449,63 @@ private async Task checkDatabaseAsync(IMartenDatabase databas return HealthCheckResult.Healthy("Healthy"); } - private static string shardDescription(IMartenDatabase database, JasperFx.Events.Projections.ShardState row) + /// + /// marten#5174: read just the high-water progression rows, with last_updated. This used + /// to go through and + /// filter in memory, which pulled every projection × tenant row on every probe and still could + /// not see last_updated — has no + /// field for it. Two high-water rows out of hundreds is the whole working set here. + /// + private static async Task> readHighWaterRowsAsync(IMartenDatabase database, + string schema, CancellationToken token) + { + await database.EnsureStorageExistsAsync(typeof(IEvent), token).ConfigureAwait(false); + + var rows = new List(); + + await using var conn = database.CreateConnection(); + await conn.OpenAsync(token).ConfigureAwait(false); + try + { + // Both operands are compile-time constants from HighWaterShardIdentity, so there is no + // pattern grammar reaching this from user input. + await using var reader = await conn + .CreateCommand( + $"select name, last_seq_id, last_updated from {schema}.mt_event_progression where name = :global or name like :prefix") + .With("global", HighWaterMarkShard, NpgsqlDbType.Varchar) + .With("prefix", PerTenantHighWaterPrefix + "%", NpgsqlDbType.Varchar) + .ExecuteReaderAsync(token).ConfigureAwait(false); + + while (await reader.ReadAsync(token).ConfigureAwait(false)) + { + var name = await reader.GetFieldValueAsync(0, token).ConfigureAwait(false); + var sequence = await reader.GetFieldValueAsync(1, token).ConfigureAwait(false); + DateTimeOffset? lastUpdated = await reader.IsDBNullAsync(2, token).ConfigureAwait(false) + ? null + : await reader.GetFieldValueAsync(2, token).ConfigureAwait(false); + + rows.Add(new HighWaterRow(name, sequence, lastUpdated)); + } + } + finally + { + await conn.CloseAsync().ConfigureAwait(false); + } + + return rows; + } + + private record HighWaterRow(string ShardName, long Sequence, DateTimeOffset? LastUpdated) + { + public bool IsPerTenant { get; } = + !string.Equals(ShardName, HighWaterMarkShard, StringComparison.Ordinal); + } + + private static string shardDescription(IMartenDatabase database, HighWaterRow row) { - return string.Equals(row.ShardName, HighWaterMarkShard, StringComparison.Ordinal) - ? $"database '{database.Identifier}'" - : $"database '{database.Identifier}', shard '{row.ShardName}'"; + return row.IsPerTenant + ? $"database '{database.Identifier}', shard '{row.ShardName}'" + : $"database '{database.Identifier}'"; } private static string trackingKey(string databaseIdentifier, string shardName) =>