diff --git a/Directory.Packages.props b/Directory.Packages.props
index 50a7d8f03..d8c7c7c39 100644
--- a/Directory.Packages.props
+++ b/Directory.Packages.props
@@ -36,20 +36,20 @@
-
-
-
+
+
+
-
+
-
+
-
+
-
-
+
+
diff --git a/docs/guide/durability/marten/distribution.md b/docs/guide/durability/marten/distribution.md
index 41094ff71..f469adbfb 100644
--- a/docs/guide/durability/marten/distribution.md
+++ b/docs/guide/durability/marten/distribution.md
@@ -185,3 +185,53 @@ var host = await Host.CreateDefaultBuilder()
snippet source | anchor
+
+## When a Projection Fails
+
+A projection or subscription shard that throws while applying an event is *paused* by the Marten/Polecat
+daemon rather than skipped, unless you have opted into skipping through `ErrorHandlingOptions`
+(`SkipApplyErrors` and friends). A paused shard makes no further progress, and Wolverine deliberately does
+not restart it — restarting would fail on the exact same event, so the shard would thrash instead of
+advance.
+
+Wolverine surfaces the paused shard so it does not simply go quiet:
+
+* The agent's health check reports the failure **category**, the sequence number and type of the event it
+ died on, and the root exception type — enough to act on without going to dig through logs.
+* `IWolverineObserver.AgentPaused(Uri agentUri, ShardFailure? failure)` fires once per transition into the
+ failed state (and again if the shard recovers and later fails anew). Implement it on a custom observer to
+ raise your own alert; [CritterWatch](https://critterwatch.io) uses this hook.
+* A `NodeRecordType.AgentPaused` record is written to the node-record log with the classified reason, so
+ the failure is readable after the fact and from another process.
+* `IEventSubscriptionAgent.Failure` exposes the same `ShardFailure` value directly. It is a plain,
+ serializable record — category, the failing event, the exception message and full detail — not an
+ `Exception`, so it survives being shipped to a monitoring UI.
+
+The category tells you what to do about it:
+
+| Category | What it means |
+|----------|---------------|
+| `ApplyEvent` | Your projection code threw on an event — the classic "poison pill". Needs a code fix, or `SkipApplyErrors`. |
+| `EventSerialization` | The store could not deserialize or upcast a stored event body. Needs a serializer or data fix. |
+| `UnknownEventType` | A stored event alias resolves to no known .NET type in *this* deployment — usually a missing registration or a rollback. |
+| `ProgressionOutOfOrder` | The shard's progression row moved underneath it, which almost always means two processes are running the same shard. |
+| `Other` | A database outage, a timeout, or a bug. No single event can be blamed. |
+
+Only `Other` is treated as potentially self-healing, so it is the only category Wolverine's stall detector
+will auto-restart. The rest are left alone until you resolve the underlying problem, at which point
+restarting or rewinding the agent picks it back up.
+
+## Agent Start Retries
+
+An agent's very first assignment can race the subsystems it depends on coming up — an event-subscription
+shard evaluated before its store's high-water detection is running, for instance, which on a multi-store
+host could leave a different shard idle on every boot. Wolverine retries a failed agent start locally a
+couple of times before leaving it to the next assignment reevaluation:
+
+```csharp
+opts.Durability.AgentStartRetryAttempts = 2; // default
+opts.Durability.AgentStartRetryDelay = TimeSpan.FromMilliseconds(250); // default, multiplied by attempt number
+```
+
+Set `AgentStartRetryAttempts` to `0` to disable the local retry entirely. A failure that outlives the
+retries is logged and picked up again on the next `CheckAssignmentPeriod` tick, exactly as before.
diff --git a/docs/guide/durability/polecat/distribution.md b/docs/guide/durability/polecat/distribution.md
index 6ff859b4a..fdd5bb335 100644
--- a/docs/guide/durability/polecat/distribution.md
+++ b/docs/guide/durability/polecat/distribution.md
@@ -80,3 +80,11 @@ Other requirements:
runs — just all of them on the single node. `Serverless` and `MediatorOnly` start no agents at all.
* In `Balanced` mode you cannot disable external transports with `StubAllExternalTransports()`, because the nodes
need the control queue to communicate
+
+## When a Projection Fails
+
+The failure handling for a paused projection or subscription shard — the classified `ShardFailure` on
+`IEventSubscriptionAgent`, the `IWolverineObserver.AgentPaused` hook, the `NodeRecordType.AgentPaused`
+record, and the rule that only a self-healing failure is auto-restarted — is shared by both event store
+integrations. See [When a Projection Fails](/guide/durability/marten/distribution#when-a-projection-fails)
+on the Marten page for the details; everything there applies identically to Polecat.
diff --git a/src/Testing/CoreTests/Runtime/Agents/agent_start_retry_on_startup_race.cs b/src/Testing/CoreTests/Runtime/Agents/agent_start_retry_on_startup_race.cs
new file mode 100644
index 000000000..73b64d198
--- /dev/null
+++ b/src/Testing/CoreTests/Runtime/Agents/agent_start_retry_on_startup_race.cs
@@ -0,0 +1,180 @@
+using JasperFx;
+using JasperFx.Core;
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using Shouldly;
+using Wolverine.Runtime;
+using Wolverine.Runtime.Agents;
+using Xunit;
+
+namespace CoreTests.Runtime.Agents;
+
+///
+/// Regression coverage for GH-3519. On a multi-store Marten host, one event-subscription agent — a
+/// different one on every boot — failed its very first assignment start because it was evaluated before
+/// its store's high-water detection was up, and then sat wedged for the life of the process. The daemon
+/// side is fixed in JasperFx.Events 2.36.x: the start failure now arrives as a ShardStartException
+/// that names its cause (jasperfx#534) and the half-started shard is released rather than orphaned
+/// (jasperfx#540), so the next attempt succeeds. What was left on this side was WHEN that next attempt
+/// happens — the node retried only on the next assignment reevaluation, so the loser of a sub-second
+/// startup race idled for a full CheckAssignmentPeriod (30s by default) while its high-water climbed.
+///
+public class agent_start_retry_on_startup_race
+{
+ private readonly WolverineOptions _options;
+ private readonly IWolverineRuntime _runtime;
+ private readonly CancellationTokenSource _cancellation = new();
+
+ public agent_start_retry_on_startup_race()
+ {
+ _options = new WolverineOptions { ApplicationAssembly = GetType().Assembly };
+ _options.Durability.Mode = DurabilityMode.Solo;
+ _options.Durability.DurabilityAgentEnabled = false;
+ _options.Durability.CheckAssignmentPeriod = 1.Hours();
+
+ // Keep the test fast; the retry COUNT is what's under test, not the pacing.
+ _options.Durability.AgentStartRetryDelay = 1.Milliseconds();
+
+ _runtime = Substitute.For();
+ _runtime.Options.Returns(_options);
+ _runtime.DurabilitySettings.Returns(_options.Durability);
+ _runtime.Observer.Returns(Substitute.For());
+ }
+
+ private NodeAgentController controllerFor(params FlakyAgent[] agents)
+ {
+ var family = new FlakyAgentFamily("event-subscriptions");
+ foreach (var agent in agents)
+ {
+ family.Add(agent);
+ }
+
+ return new NodeAgentController(_runtime, Substitute.For(), [family],
+ NullLogger.Instance, _cancellation.Token);
+ }
+
+ [Fact]
+ public async Task recovers_from_a_start_that_loses_the_first_assignment_race()
+ {
+ var uri = new Uri("event-subscriptions://marten/iincidentsstore/localhost.postgres/incident/all");
+
+ // Exactly the reported shape: high-water detection isn't up yet on the first attempt and is by
+ // the second.
+ var agent = new FlakyAgent(uri, failuresBeforeSuccess: 1);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+
+ agent.AttemptCount.ShouldBe(2);
+ agent.Status.ShouldBe(AgentStatus.Running);
+ controller.Agents.ContainsKey(uri).ShouldBeTrue();
+ }
+
+ [Fact]
+ public async Task gives_up_after_the_configured_attempts_and_preserves_the_daemon_s_reason()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FlakyAgent(uri, failuresBeforeSuccess: int.MaxValue);
+ var controller = controllerFor(agent);
+
+ var ex = await Should.ThrowAsync(() => controller.StartAgentAsync(uri));
+
+ // Default is 2 retries on top of the initial attempt. A failure that outlives them is left to
+ // the next assignment reevaluation rather than retried harder here.
+ agent.AttemptCount.ShouldBe(3);
+
+ // The daemon's reason has to survive the wrapping, or we are back to the causeless "Unable to
+ // start a subscription agent" that made this issue undiagnosable in the first place.
+ ex.InnerException.ShouldNotBeNull();
+ ex.InnerException.Message.ShouldContain("Incident:All");
+ ex.InnerException.Message.ShouldContain("High-water detection is not running yet");
+
+ controller.Agents.ContainsKey(uri).ShouldBeFalse();
+ }
+
+ [Fact]
+ public async Task retries_can_be_turned_off()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ _options.Durability.AgentStartRetryAttempts = 0;
+
+ var agent = new FlakyAgent(uri, failuresBeforeSuccess: 1);
+ var controller = controllerFor(agent);
+
+ await Should.ThrowAsync(() => controller.StartAgentAsync(uri));
+
+ agent.AttemptCount.ShouldBe(1);
+ }
+
+ [Fact]
+ public async Task a_healthy_agent_still_starts_on_the_first_attempt()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FlakyAgent(uri, failuresBeforeSuccess: 0);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+
+ agent.AttemptCount.ShouldBe(1);
+ }
+
+ private class FlakyAgentFamily : IAgentFamily
+ {
+ private readonly Dictionary _agents = new();
+
+ public FlakyAgentFamily(string scheme) => Scheme = scheme;
+
+ public void Add(FlakyAgent agent) => _agents[agent.Uri] = agent;
+
+ public string Scheme { get; }
+
+ public ValueTask> AllKnownAgentsAsync()
+ => ValueTask.FromResult>(_agents.Keys.ToList());
+
+ public ValueTask BuildAgentAsync(Uri uri, IWolverineRuntime wolverineRuntime)
+ => ValueTask.FromResult(_agents[uri]);
+
+ public ValueTask> SupportedAgentsAsync()
+ => ValueTask.FromResult>(_agents.Keys.ToList());
+
+ public ValueTask EvaluateAssignmentsAsync(AssignmentGrid assignments) => ValueTask.CompletedTask;
+ }
+
+ private class FlakyAgent : IAgent
+ {
+ private readonly int _failuresBeforeSuccess;
+
+ public FlakyAgent(Uri uri, int failuresBeforeSuccess)
+ {
+ Uri = uri;
+ _failuresBeforeSuccess = failuresBeforeSuccess;
+ }
+
+ public int AttemptCount { get; private set; }
+
+ public Uri Uri { get; }
+ public AgentStatus Status { get; private set; } = AgentStatus.Stopped;
+
+ public Task StartAsync(CancellationToken cancellationToken)
+ {
+ AttemptCount++;
+ if (AttemptCount <= _failuresBeforeSuccess)
+ {
+ // Stands in for the ShardStartException JasperFxAsyncDaemon.StartAgentAsync(ShardName)
+ // now throws instead of a bare, causeless Exception (its constructors are internal to
+ // JasperFx.Events, so this reproduces the message shape rather than the type).
+ throw new Exception(
+ "Unable to start a subscription agent for 'Incident:All'. High-water detection is not running yet, so the shard could not be positioned.");
+ }
+
+ Status = AgentStatus.Running;
+ return Task.CompletedTask;
+ }
+
+ public Task StopAsync(CancellationToken cancellationToken)
+ {
+ Status = AgentStatus.Stopped;
+ return Task.CompletedTask;
+ }
+ }
+}
diff --git a/src/Testing/CoreTests/Runtime/Agents/paused_shard_failure_surfacing.cs b/src/Testing/CoreTests/Runtime/Agents/paused_shard_failure_surfacing.cs
new file mode 100644
index 000000000..54875dc2b
--- /dev/null
+++ b/src/Testing/CoreTests/Runtime/Agents/paused_shard_failure_surfacing.cs
@@ -0,0 +1,455 @@
+using JasperFx;
+using JasperFx.Core;
+using JasperFx.Events.Daemon;
+using JasperFx.Events.Projections;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using Shouldly;
+using Wolverine.Runtime;
+using Wolverine.Runtime.Agents;
+using Xunit;
+using IProjectionDaemon = JasperFx.Events.Daemon.IProjectionDaemon;
+using JasperFxSubscriptionAgent = JasperFx.Events.Daemon.ISubscriptionAgent;
+
+namespace CoreTests.Runtime.Agents;
+
+///
+/// Coverage for GH-3637 / GH-3638 (WO-8), the Wolverine half of JasperFx/jasperfx#565. When a shard hits
+/// an ApplyEventException under a non-skipping error policy the daemon pauses it, and until now the
+/// assignment plane could see only — never why. The anti-thrash guards
+/// meant the shard wasn't restart-looped, but nothing surfaced it either: progress silently flatlined and
+/// the only trace was a log line on one node. Now the classified rides through
+/// the wrapper into the health check and the observer plane, and a failure that will recur on the same
+/// event is no longer swept back up by the GH-3519 wedge recovery.
+///
+public class paused_shard_failure_surfacing
+{
+ private static ShardFailure FailureOf(ShardFailureCategory category, long? sequence = 42)
+ {
+ return new ShardFailure
+ {
+ Category = category,
+ ExceptionType = "JasperFx.Events.Daemon.ApplyEventException",
+ RootExceptionType = "System.NullReferenceException",
+ Message = "Object reference not set to an instance of an object.",
+ Detail = "JasperFx.Events.Daemon.ApplyEventException: ... ---> System.NullReferenceException: ...",
+ OccurredAt = DateTimeOffset.UtcNow,
+ Event = sequence == null
+ ? null
+ : new EventFailureDetails { Sequence = sequence.Value, EventTypeName = "trip_started" }
+ };
+ }
+
+ public class wrapper_delegation
+ {
+ private readonly IProjectionDaemon _daemon = Substitute.For();
+ private readonly JasperFxSubscriptionAgent _inner = Substitute.For();
+ private readonly EventSubscriptionAgent _agent;
+
+ public wrapper_delegation()
+ {
+ _daemon.StartAgentAsync(Arg.Any(), Arg.Any()).Returns(_inner);
+ _agent = new EventSubscriptionAgent(
+ new Uri("event-subscriptions://marten/incident/all"), new ShardName("Incident"), _daemon);
+ }
+
+ [Fact]
+ public void reports_no_failure_before_it_has_ever_started()
+ {
+ // Nothing to delegate to yet. This is the shape every non-tracking implementation reads as,
+ // and it must never throw.
+ _agent.Failure.ShouldBeNull();
+ }
+
+ [Fact]
+ public async Task delegates_the_failure_to_the_live_inner_agent()
+ {
+ var failure = FailureOf(ShardFailureCategory.ApplyEvent);
+ _inner.Failure.Returns(failure);
+
+ await _agent.StartAsync(CancellationToken.None);
+
+ _agent.Failure.ShouldBeSameAs(failure);
+ }
+
+ [Fact]
+ public async Task a_recovered_shard_stops_reporting_the_old_failure()
+ {
+ _inner.Failure.Returns(FailureOf(ShardFailureCategory.ApplyEvent));
+ await _agent.StartAsync(CancellationToken.None);
+
+ // The daemon clears Failure when the agent starts or replays. Because the wrapper reads
+ // through rather than caching, that clearing is visible here immediately — a cached copy
+ // would have kept alerting on a shard that had already recovered.
+ _inner.Failure.Returns((ShardFailure?)null);
+
+ _agent.Failure.ShouldBeNull();
+ }
+ }
+
+ public class health_check_reporting
+ {
+ private readonly IProjectionDaemon _daemon = Substitute.For();
+ private readonly JasperFxSubscriptionAgent _inner = Substitute.For();
+ private readonly EventSubscriptionAgent _agent;
+
+ public health_check_reporting()
+ {
+ _daemon.StartAgentAsync(Arg.Any(), Arg.Any()).Returns(_inner);
+ _agent = new EventSubscriptionAgent(
+ new Uri("event-subscriptions://marten/incident/all"), new ShardName("Incident"), _daemon);
+ }
+
+ private Task checkAsync()
+ => _agent.CheckHealthAsync(new HealthCheckContext(), CancellationToken.None);
+
+ [Fact]
+ public async Task a_paused_shard_reports_the_category_the_failing_event_and_the_root_exception()
+ {
+ _inner.Status.Returns(AgentStatus.Paused);
+ _inner.Failure.Returns(FailureOf(ShardFailureCategory.ApplyEvent));
+ await _agent.StartAsync(CancellationToken.None);
+
+ var result = await checkAsync();
+
+ result.Status.ShouldBe(HealthStatus.Unhealthy);
+ var description = result.Description!;
+
+ // The whole point of WO-8: the difference between an alert an operator can act on and one
+ // they have to go dig for. Category says what kind of problem, the sequence says exactly
+ // where it stopped, and the ROOT exception type names the actual fault rather than the
+ // ApplyEventException wrapper around it.
+ description.ShouldContain("ApplyEvent");
+ description.ShouldContain("42");
+ description.ShouldContain("trip_started");
+ description.ShouldContain("System.NullReferenceException");
+ }
+
+ [Fact]
+ public async Task a_shard_the_daemon_stopped_out_of_order_reports_its_reason_too()
+ {
+ // ProgressionOutOfOrder is the one category where the daemon STOPS rather than pauses, so
+ // the reason has to survive the Stopped branch as well.
+ _inner.Status.Returns(AgentStatus.Stopped);
+ _inner.Failure.Returns(FailureOf(ShardFailureCategory.ProgressionOutOfOrder, sequence: null));
+ await _agent.StartAsync(CancellationToken.None);
+
+ var result = await checkAsync();
+
+ result.Status.ShouldBe(HealthStatus.Unhealthy);
+ result.Description!.ShouldContain("ProgressionOutOfOrder");
+ }
+
+ [Fact]
+ public async Task a_paused_shard_with_no_reported_reason_keeps_the_old_message()
+ {
+ // An inner agent that doesn't track failures reads exactly as it did before this change.
+ _inner.Status.Returns(AgentStatus.Paused);
+ _inner.Failure.Returns((ShardFailure?)null);
+ await _agent.StartAsync(CancellationToken.None);
+
+ var result = await checkAsync();
+
+ result.Status.ShouldBe(HealthStatus.Unhealthy);
+ result.Description!.ShouldContain("paused due to errors");
+ }
+ }
+
+ public class restart_suppression
+ {
+ private readonly WolverineOptions _options;
+ private readonly IWolverineRuntime _runtime;
+ private readonly IWolverineObserver _observer = Substitute.For();
+ private readonly CancellationTokenSource _cancellation = new();
+
+ public restart_suppression()
+ {
+ _options = new WolverineOptions { ApplicationAssembly = GetType().Assembly };
+ _options.Durability.Mode = DurabilityMode.Solo;
+ _options.Durability.DurabilityAgentEnabled = false;
+ _options.Durability.CheckAssignmentPeriod = 1.Hours();
+
+ _runtime = Substitute.For();
+ _runtime.Options.Returns(_options);
+ _runtime.DurabilitySettings.Returns(_options.Durability);
+ _runtime.Observer.Returns(_observer);
+ }
+
+ private NodeAgentController controllerFor(params FakeSubscriptionAgent[] agents)
+ {
+ var family = new FakeSubscriptionAgentFamily("event-subscriptions");
+ foreach (var agent in agents)
+ {
+ family.Add(agent);
+ }
+
+ return new NodeAgentController(_runtime, Substitute.For(), [family],
+ NullLogger.Instance, _cancellation.Token);
+ }
+
+ [Fact]
+ public async Task does_not_restart_a_shard_stopped_on_a_failure_that_will_recur()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FakeSubscriptionAgent(uri);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+ agent.StartCount.ShouldBe(1);
+
+ // The daemon stopped this shard on a progression row two processes are fighting over.
+ // Restarting adds a third contender; the GH-3519 wedge recovery must stand down here.
+ agent.SimulateFailure(AgentStatus.Stopped, FailureOf(ShardFailureCategory.ProgressionOutOfOrder));
+
+ await controller.StartAgentAsync(uri);
+
+ agent.StartCount.ShouldBe(1);
+ agent.StopCount.ShouldBe(0);
+ await _observer.Received(1).AgentPaused(uri, Arg.Any());
+ }
+
+ [Fact]
+ public async Task keeps_reporting_at_most_once_per_transition_into_failure()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FakeSubscriptionAgent(uri);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+ agent.SimulateFailure(AgentStatus.Stopped, FailureOf(ShardFailureCategory.ApplyEvent));
+
+ // The leader re-issues AssignAgent on every 30s reevaluation for as long as the grid sees
+ // this agent as not running. One alert per failure, not one every tick.
+ await controller.StartAgentAsync(uri);
+ await controller.StartAgentAsync(uri);
+ await controller.StartAgentAsync(uri);
+
+ await _observer.Received(1).AgentPaused(uri, Arg.Any());
+ }
+
+ [Fact]
+ public async Task still_restarts_a_shard_whose_failure_could_clear_on_its_own()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FakeSubscriptionAgent(uri);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+
+ // A database outage or a timeout is exactly what restart-on-failure exists for.
+ agent.SimulateFailure(AgentStatus.Stopped, FailureOf(ShardFailureCategory.Other, sequence: null));
+
+ await controller.StartAgentAsync(uri);
+
+ agent.StartCount.ShouldBe(2);
+ }
+
+ [Fact]
+ public async Task still_restarts_a_wedged_shard_that_reports_no_failure()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FakeSubscriptionAgent(uri);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+
+ // GH-3519's case, unchanged: the shard died with nothing to report, so the wedge recovery
+ // still owns it.
+ agent.SimulateFailure(AgentStatus.Stopped, null);
+
+ await controller.StartAgentAsync(uri);
+
+ agent.StartCount.ShouldBe(2);
+ }
+ }
+
+ public class local_sweep
+ {
+ private readonly WolverineOptions _options;
+ private readonly IWolverineRuntime _runtime;
+ private readonly IWolverineObserver _observer = Substitute.For();
+ private readonly CancellationTokenSource _cancellation = new();
+
+ public local_sweep()
+ {
+ _options = new WolverineOptions { ApplicationAssembly = GetType().Assembly };
+ _options.Durability.Mode = DurabilityMode.Solo;
+ _options.Durability.DurabilityAgentEnabled = false;
+ _options.Durability.CheckAssignmentPeriod = 1.Hours();
+
+ _runtime = Substitute.For();
+ _runtime.Options.Returns(_options);
+ _runtime.DurabilitySettings.Returns(_options.Durability);
+ _runtime.Observer.Returns(_observer);
+ }
+
+ private NodeAgentController controllerFor(params FakeSubscriptionAgent[] agents)
+ {
+ var family = new FakeSubscriptionAgentFamily("event-subscriptions");
+ foreach (var agent in agents)
+ {
+ family.Add(agent);
+ }
+
+ return new NodeAgentController(_runtime, Substitute.For(), [family],
+ NullLogger.Instance, _cancellation.Token);
+ }
+
+ [Fact]
+ public async Task surfaces_a_locally_owned_shard_the_daemon_paused()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FakeSubscriptionAgent(uri);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+
+ // Paused is the ApplyEventException shape, and it never reaches the restart path at all --
+ // StartAgentAsync treats any non-Stopped agent as running and returns early. Without this
+ // sweep nothing in the process would ever have mentioned it again.
+ agent.SimulateFailure(AgentStatus.Paused, FailureOf(ShardFailureCategory.ApplyEvent));
+
+ await controller.ReportFailedLocalAgentsAsync();
+
+ await _observer.Received(1)
+ .AgentPaused(uri, Arg.Is(f => f!.Category == ShardFailureCategory.ApplyEvent));
+ }
+
+ [Fact]
+ public async Task reports_once_per_transition_not_once_per_tick()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FakeSubscriptionAgent(uri);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+ agent.SimulateFailure(AgentStatus.Paused, FailureOf(ShardFailureCategory.ApplyEvent));
+
+ await controller.ReportFailedLocalAgentsAsync();
+ await controller.ReportFailedLocalAgentsAsync();
+ await controller.ReportFailedLocalAgentsAsync();
+
+ await _observer.Received(1).AgentPaused(uri, Arg.Any());
+ }
+
+ [Fact]
+ public async Task re_arms_after_the_shard_recovers()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FakeSubscriptionAgent(uri);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+ agent.SimulateFailure(AgentStatus.Paused, FailureOf(ShardFailureCategory.ApplyEvent));
+ await controller.ReportFailedLocalAgentsAsync();
+
+ // Operator skips the poison event and the shard runs again...
+ agent.SimulateRecovered();
+ await controller.ReportFailedLocalAgentsAsync();
+
+ // ...and a NEW failure has to alert again rather than being swallowed as a duplicate.
+ agent.SimulateFailure(AgentStatus.Paused, FailureOf(ShardFailureCategory.EventSerialization));
+ await controller.ReportFailedLocalAgentsAsync();
+
+ await _observer.Received(2).AgentPaused(uri, Arg.Any());
+ }
+
+ [Fact]
+ public async Task says_nothing_about_a_healthy_shard()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FakeSubscriptionAgent(uri);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+
+ await controller.ReportFailedLocalAgentsAsync();
+
+ await _observer.DidNotReceive().AgentPaused(Arg.Any(), Arg.Any());
+ }
+
+ [Fact]
+ public async Task says_nothing_about_a_shard_stopped_with_no_reason_to_report()
+ {
+ var uri = new Uri("event-subscriptions://marten/incident/all");
+ var agent = new FakeSubscriptionAgent(uri);
+ var controller = controllerFor(agent);
+
+ await controller.StartAgentAsync(uri);
+
+ // An ordinary stop, or a wedged shard GH-3519's recovery owns. Reporting it here would fire
+ // on every deliberate stop and drown the signal this whole sweep exists to raise.
+ agent.SimulateFailure(AgentStatus.Stopped, null);
+
+ await controller.ReportFailedLocalAgentsAsync();
+
+ await _observer.DidNotReceive().AgentPaused(Arg.Any(), Arg.Any());
+ }
+ }
+
+ private class FakeSubscriptionAgentFamily : IAgentFamily
+ {
+ private readonly Dictionary _agents = new();
+
+ public FakeSubscriptionAgentFamily(string scheme) => Scheme = scheme;
+
+ public void Add(FakeSubscriptionAgent agent) => _agents[agent.Uri] = agent;
+
+ public string Scheme { get; }
+
+ public ValueTask> AllKnownAgentsAsync()
+ => ValueTask.FromResult>(_agents.Keys.ToList());
+
+ public ValueTask BuildAgentAsync(Uri uri, IWolverineRuntime wolverineRuntime)
+ => ValueTask.FromResult(_agents[uri]);
+
+ public ValueTask> SupportedAgentsAsync()
+ => ValueTask.FromResult>(_agents.Keys.ToList());
+
+ public ValueTask EvaluateAssignmentsAsync(AssignmentGrid assignments) => ValueTask.CompletedTask;
+ }
+
+ private class FakeSubscriptionAgent : IEventSubscriptionAgent
+ {
+ public FakeSubscriptionAgent(Uri uri) => Uri = uri;
+
+ public int StartCount { get; private set; }
+ public int StopCount { get; private set; }
+
+ public Uri Uri { get; }
+ public AgentStatus Status { get; private set; } = AgentStatus.Stopped;
+ public ShardFailure? Failure { get; private set; }
+
+ // Mimic the daemon flipping status and reason together underneath the wrapper, without going
+ // through the controller's registration.
+ public void SimulateFailure(AgentStatus status, ShardFailure? failure)
+ {
+ Status = status;
+ Failure = failure;
+ }
+
+ public void SimulateRecovered()
+ {
+ Status = AgentStatus.Running;
+ Failure = null;
+ }
+
+ public Task StartAsync(CancellationToken cancellationToken)
+ {
+ StartCount++;
+ Status = AgentStatus.Running;
+ Failure = null;
+ return Task.CompletedTask;
+ }
+
+ public Task StopAsync(CancellationToken cancellationToken)
+ {
+ StopCount++;
+ Status = AgentStatus.Stopped;
+ return Task.CompletedTask;
+ }
+
+ public Task RebuildAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+ }
+}
diff --git a/src/Wolverine/DurabilitySettings.cs b/src/Wolverine/DurabilitySettings.cs
index c20c786b0..3ef413004 100644
--- a/src/Wolverine/DurabilitySettings.cs
+++ b/src/Wolverine/DurabilitySettings.cs
@@ -279,6 +279,26 @@ internal set
///
public int MaxAgentStopParallelism { get; set; } = 10;
+ ///
+ /// GH-3519: how many extra times this node immediately re-tries an agent that failed to start,
+ /// before giving up and leaving it to the next assignment reevaluation. A first-assignment start
+ /// races the subsystems the agent depends on — an event-subscription shard evaluated before its
+ /// store's high-water detection is up is the reported case, and on a multi-store host a different
+ /// shard lost that race on every boot. Without a local retry the loser waited a full
+ /// (30s by default) doing nothing while its high-water mark
+ /// climbed. Set to 0 to restore the old single-attempt behavior. Default 2.
+ ///
+ public int AgentStartRetryAttempts { get; set; } = 2;
+
+ ///
+ /// GH-3519: how long this node waits before each of the
+ /// immediate re-tries of a failed agent start, multiplied by the attempt number so the second
+ /// retry waits twice as long as the first. Sized for a startup race that resolves in well under a
+ /// second, not for an outage — a failure that outlives these attempts is left to the next
+ /// assignment reevaluation rather than retried harder here. Default 250ms.
+ ///
+ public TimeSpan AgentStartRetryDelay { get; set; } = 250.Milliseconds();
+
///
/// Opt-in switch for the dynamic listener registry: persisted listener URIs that
/// are activated at runtime in addition to the listeners declared statically
diff --git a/src/Wolverine/Runtime/Agents/EventSubscriptionAgent.cs b/src/Wolverine/Runtime/Agents/EventSubscriptionAgent.cs
index 59603ed51..6e1c784a7 100644
--- a/src/Wolverine/Runtime/Agents/EventSubscriptionAgent.cs
+++ b/src/Wolverine/Runtime/Agents/EventSubscriptionAgent.cs
@@ -142,17 +142,61 @@ public AgentStatus Status
private set => _status = value;
}
+ ///
+ // GH-3638: the reason rides alongside the status for exactly the same reason the status itself
+ // delegates (GH-3519) -- the daemon owns both, and it sets Failure at the same moment it flips Status
+ // to Paused/Stopped and clears it on start/replay. Reading a cached copy here would let the wrapper
+ // report a stale reason for a shard that has since recovered.
+ public ShardFailure? Failure => _innerAgent?.Failure;
+
+ ///
+ /// The failure categories that will fail again on the exact same event no matter how many times the
+ /// shard is restarted, so an automated restart is pure churn: it burns the daemon's start path, resets
+ /// the operator's view of the failure, and re-pauses on the same sequence.
+ /// is worse than useless — it means two
+ /// processes are on the same shard, and restarting adds a third contender. Only
+ /// (a database outage, a timeout) is what auto-restart is
+ /// actually for. See GH-3638.
+ ///
+ private static bool canSelfHeal(ShardFailure? failure)
+ => failure == null || failure.Category == ShardFailureCategory.Other;
+
+ // GH-3638: turn "paused due to errors" into something an operator can act on without going to dig
+ // through logs -- what kind of failure, which event it died on, and the exception type that actually
+ // names the fault (the root, not the ApplyEventException/ShardStopException wrapper around it).
+ private static string describeFailure(Uri uri, AgentStatus status, ShardFailure failure)
+ {
+ var where = failure.Event == null
+ ? string.Empty
+ : $" at event sequence {failure.Event.Sequence}" +
+ (failure.Event.EventTypeName == null ? string.Empty : $" ({failure.Event.EventTypeName})") +
+ (failure.Event.TenantId == null ? string.Empty : $" for tenant '{failure.Event.TenantId}'");
+
+ return $"Projection {uri} is {status} on a {failure.Category} failure{where}: " +
+ $"{failure.RootExceptionType}: {failure.Message}";
+ }
+
public Task CheckHealthAsync(HealthCheckContext context,
CancellationToken cancellationToken = default)
{
- if (Status == AgentStatus.Paused)
+ var status = Status;
+ if (status is AgentStatus.Paused or AgentStatus.Stopped)
+ {
+ var failure = Failure;
+ if (failure != null)
+ {
+ return Task.FromResult(HealthCheckResult.Unhealthy(describeFailure(Uri, status, failure)));
+ }
+ }
+
+ if (status == AgentStatus.Paused)
{
return Task.FromResult(HealthCheckResult.Unhealthy($"Projection {Uri} paused due to errors"));
}
- if (Status != AgentStatus.Running)
+ if (status != AgentStatus.Running)
{
- return Task.FromResult(HealthCheckResult.Unhealthy($"Agent {Uri} is {Status}"));
+ return Task.FromResult(HealthCheckResult.Unhealthy($"Agent {Uri} is {status}"));
}
// Load thresholds on first health check
@@ -208,6 +252,18 @@ public Task CheckHealthAsync(HealthCheckContext context,
if (_consecutiveStallCount >= MaxConsecutiveStallsBeforeRestart)
{
+ // GH-3638: a shard stalled behind a per-event failure will die on that same event on every
+ // restart, so restarting it is churn that also resets the operator's view every time the
+ // stall detector fires. Surface the failure and leave it alone; only a self-healing
+ // category (a database blip, a timeout -- what auto-restart actually exists for) is retried.
+ var failure = Failure;
+ if (!canSelfHeal(failure))
+ {
+ return Task.FromResult(HealthCheckResult.Unhealthy(
+ describeFailure(Uri, Status, failure!) +
+ $" -- stalled for {_consecutiveStallCount} consecutive health checks. Auto-restart suppressed: this failure will recur on the same event until it is resolved."));
+ }
+
// Trigger auto-restart
_ = Task.Run(() => AttemptAutoRestartAsync(cancellationToken), cancellationToken);
diff --git a/src/Wolverine/Runtime/Agents/IEventSubscriptionAgent.cs b/src/Wolverine/Runtime/Agents/IEventSubscriptionAgent.cs
index c2bd5de8e..7550a4d07 100644
--- a/src/Wolverine/Runtime/Agents/IEventSubscriptionAgent.cs
+++ b/src/Wolverine/Runtime/Agents/IEventSubscriptionAgent.cs
@@ -1,3 +1,5 @@
+using JasperFx.Events.Daemon;
+
namespace Wolverine.Runtime.Agents;
///
@@ -26,4 +28,19 @@ public interface IEventSubscriptionAgent : IAgent
/// Optional point-in-time to rewind to; replays events on/after this time.
Task RewindAsync(long? sequenceFloor, DateTimeOffset? timestamp, CancellationToken cancellationToken)
=> throw new NotSupportedException("This event-subscription agent does not support rewind.");
+
+ ///
+ /// WHY this shard was paused or stopped, when it was. alone says a shard
+ /// is but never what an operator should do about it, so
+ /// progress could flatline with nothing actionable to alert on. The category distinguishes a poison
+ /// event (needs a code fix or a skip) from a serialization fault, an unregistered event type, two
+ /// processes racing the same shard, or a transient infrastructure blip — and the failing event's
+ /// sequence names exactly where it stopped.
+ ///
+ /// A plain serializable value rather than an , so it survives the hop to
+ /// the assignment plane, a persisted progression row, or a monitoring UI. Null whenever the shard is
+ /// not reporting a failure. Default null so existing implementations are unaffected.
+ /// See GH-3637 / GH-3638 and JasperFx/jasperfx#565.
+ ///
+ ShardFailure? Failure => null;
}
diff --git a/src/Wolverine/Runtime/Agents/IWolverineObserver.cs b/src/Wolverine/Runtime/Agents/IWolverineObserver.cs
index 699e5f3d6..dd1604c2e 100644
--- a/src/Wolverine/Runtime/Agents/IWolverineObserver.cs
+++ b/src/Wolverine/Runtime/Agents/IWolverineObserver.cs
@@ -1,4 +1,5 @@
using JasperFx.Events;
+using JasperFx.Events.Daemon;
using Wolverine.Configuration;
using Wolverine.ErrorHandling;
using Wolverine.Logging;
@@ -26,6 +27,21 @@ public interface IWolverineObserver
Task AgentStarted(Uri agentUri);
Task AgentStopped(Uri agentUri);
+ ///
+ /// A locally-owned agent stopped running because of a failure it reported, not because it was asked
+ /// to stop. The case this exists for is an event-subscription shard the daemon paused on an
+ /// ApplyEventException under a non-skipping error policy: the anti-thrash guards already keep
+ /// it from being restarted in a loop, but nothing surfaced it, so the projection's progress simply
+ /// flatlined and the only trace was a log line on one node.
+ ///
+ /// Fires once per transition into the failed state — not on every health-check tick — and again
+ /// if the agent recovers and fails anew. is the classified reason
+ /// (category, the failing event, the root exception type) when the agent can report one, and null
+ /// when all that is known is that it stopped. Default no-op so existing observers are unaffected.
+ /// See GH-3637 / GH-3638.
+ ///
+ Task AgentPaused(Uri agentUri, ShardFailure? failure) => Task.CompletedTask;
+
// Loop through and decide what you want here.
Task AssignmentsChanged(AssignmentGrid grid, AgentCommands commands);
@@ -84,6 +100,12 @@ void ConnectionBudget(ConnectionBudgetSnapshot snapshot)
internal class PersistenceWolverineObserver : IWolverineObserver
{
+ // The narrowest wolverine_node_records.description column any store provisions.
+ private const int DescriptionLength = 500;
+
+ private static string truncate(string text, int max)
+ => text.Length <= max ? text : text[..(max - 3)] + "...";
+
private readonly IWolverineRuntime _runtime;
public PersistenceWolverineObserver(IWolverineRuntime runtime)
@@ -180,6 +202,33 @@ await _runtime.Storage.Nodes.LogRecordsAsync(NodeRecord.For(_runtime.Options, No
agentUri));
}
+ public async Task AgentPaused(Uri agentUri, ShardFailure? failure)
+ {
+ var record = NodeRecord.For(_runtime.Options, NodeRecordType.AgentPaused, agentUri);
+
+ // NodeRecord.For seeds Description with the agent URI, which the record already carries in
+ // AgentUri; the reason is the whole point of this record, so it takes the slot when we have one.
+ // Deliberately ShardFailure.ToString() (category + failing event + message) rather than Detail:
+ // the full exception text with stack traces belongs in the log line the caller writes, not in a
+ // node-record column every store has to hold. Truncated to the narrowest description column any
+ // store provisions (varchar(500) on SQL Server and Oracle) -- an exception message long enough to
+ // overflow it must not turn this diagnostic record into a failed insert.
+ if (failure != null)
+ {
+ record.Description = truncate(failure.ToString(), DescriptionLength);
+ }
+
+ try
+ {
+ await _runtime.Storage.Nodes.LogRecordsAsync(record);
+ }
+ catch (NotSupportedException)
+ {
+ // NullMessageStore does not support node persistence; a storeless Solo node can still run
+ // event-subscription agents, and losing the record must not break the health-check sweep.
+ }
+ }
+
public async Task StaleNodes(IReadOnlyList staleNodes)
{
var records = staleNodes.Select(x => new NodeRecord
diff --git a/src/Wolverine/Runtime/Agents/NodeAgentController.HeartBeat.cs b/src/Wolverine/Runtime/Agents/NodeAgentController.HeartBeat.cs
index 7a00d4931..63b6102c4 100644
--- a/src/Wolverine/Runtime/Agents/NodeAgentController.HeartBeat.cs
+++ b/src/Wolverine/Runtime/Agents/NodeAgentController.HeartBeat.cs
@@ -142,6 +142,20 @@ private async Task DoHealthChecksInternalAsync()
// conflict.
await ensureLocalNodeRegisteredAsync(_cancellation.Token);
+ // GH-3637 / GH-3638: before any leadership work, surface this node's OWN agents that stopped or
+ // paused on a failure. Deliberately here rather than in the leader-only branch below -- the daemon
+ // pauses a shard on whichever node owns it, and a follower has to be able to report its own.
+ // Wrapped so a reporting fault can never cost this node its heartbeat or its leadership lease.
+ try
+ {
+ await ReportFailedLocalAgentsAsync();
+ }
+ catch (Exception e)
+ {
+ _logger.LogError(e, "Error trying to report failed agents on node {NodeNumber}",
+ _runtime.Options.Durability.AssignedNodeNumber);
+ }
+
var (nodes, restrictions) = await _persistence.LoadNodeAgentStateAsync(_cancellation.Token);
diff --git a/src/Wolverine/Runtime/Agents/NodeAgentController.cs b/src/Wolverine/Runtime/Agents/NodeAgentController.cs
index b28a6cf5e..2abc1708e 100644
--- a/src/Wolverine/Runtime/Agents/NodeAgentController.cs
+++ b/src/Wolverine/Runtime/Agents/NodeAgentController.cs
@@ -1,5 +1,6 @@
using System.Collections.Concurrent;
using JasperFx;
+using JasperFx.Events.Daemon;
using Microsoft.Extensions.Logging;
using Wolverine.Transports;
@@ -31,6 +32,12 @@ private readonly Dictionary
// capability-matched distribution, which silently shrinks the cluster.
private IReadOnlyList _capabilities = Array.Empty();
+ // GH-3638: agents already reported as failed, so the sweep below and the restart suppression in
+ // StartAgentAsync fire once per transition into failure rather than on every 30s tick. An entry is
+ // dropped the moment the agent is seen running again, so a shard that recovers and later fails anew
+ // reports the new failure.
+ private readonly ConcurrentDictionary _reportedFailures = new();
+
// 0=free, 1=busy; guards against concurrent DoHealthChecksAsync calls
// from the heartbeat loop and a CheckAgentHealth message arriving
// simultaneously. Prevents a race on _lastLockIndex / _lastLockETag in
@@ -207,6 +214,18 @@ public async Task StartAgentAsync(Uri agentUri)
return;
}
+ // GH-3638: a shard the daemon stopped on a failure that will recur on the exact same event --
+ // a poison event, a body it cannot deserialize, an event type this deployment doesn't know, or
+ // a progression row two processes are fighting over -- must NOT be swept back up by the
+ // GH-3519 wedge recovery below. Restarting it re-runs the identical failure every
+ // reevaluation, and each restart resets the operator's view of when and where it broke. Leave
+ // it stopped, and surface the reason instead so somebody can act on it.
+ if (existing is IEventSubscriptionAgent subscription && !canSelfHeal(subscription.Failure))
+ {
+ await reportAgentPausedAsync(agentUri, subscription.Failure);
+ return;
+ }
+
// GH-3519: the agent is still registered on this node but its underlying shard has stopped
// (e.g. an event-subscription shard that lost a first-assignment startup race and wedged, or
// whose daemon execution loop faulted). The old blanket ContainsKey short-circuit treated any
@@ -229,25 +248,89 @@ public async Task StartAgentAsync(Uri agentUri)
}
}
- var agent = await findAgentAsync(agentUri);
- try
- {
- await agent.StartAsync(_cancellation.Token);
- await _observer.AgentStarted(agentUri);
-
- _logger.LogInformation("Successfully started agent {AgentUri} on Node {NodeNumber}", agentUri,
- _runtime.Options.Durability.AssignedNodeNumber);
- }
- catch (Exception e)
- {
- throw new AgentStartingException(agentUri, _runtime.Options.UniqueNodeId, e);
- }
+ var agent = await startWithRetriesAsync(agentUri);
Agents[agentUri] = agent;
+ // GH-3638: a start that succeeded supersedes whatever failure was last reported for this agent, so
+ // a later failure alerts again instead of being swallowed as a duplicate of the old one.
+ _reportedFailures.TryRemove(agentUri, out _);
+
await upsertAssignmentAsync(agentUri);
}
+ ///
+ /// Start an agent, retrying a failure a bounded number of times before giving up on this tick.
+ ///
+ /// GH-3519: a first-assignment start races whatever the agent depends on coming up. The reported
+ /// shape is a multi-store Marten host where one event-subscription shard — a different one on every
+ /// boot — was evaluated before its store's high-water detection was running and failed; the daemon now
+ /// says so in as many words (ShardStartException, JasperFx/jasperfx#534) and releases the
+ /// half-started shard (jasperfx#540), so the very next attempt succeeds. Without a local retry that
+ /// attempt only came on the next assignment reevaluation, so the loser of the race sat idle for a full
+ /// CheckAssignmentPeriod while its high-water mark climbed — the "permanent 30-second retry loop" in
+ /// the report.
+ ///
+ /// Each attempt goes back through : a faulted start may have left
+ /// the family's agent object unusable, and the family owns whether a rebuild is a fresh object or the
+ /// same one. The exception thrown after the last attempt is the LAST failure with its cause intact —
+ /// the daemon's reason for the final attempt is what an operator needs, not the first one's.
+ ///
+ private async Task startWithRetriesAsync(Uri agentUri)
+ {
+ var maxAttempts = Math.Max(1, _runtime.Options.Durability.AgentStartRetryAttempts + 1);
+
+ for (var attempt = 1; ; attempt++)
+ {
+ var agent = await findAgentAsync(agentUri);
+ try
+ {
+ await agent.StartAsync(_cancellation.Token);
+ await _observer.AgentStarted(agentUri);
+
+ if (attempt > 1)
+ {
+ _logger.LogInformation(
+ "Successfully started agent {AgentUri} on Node {NodeNumber} on attempt {Attempt}",
+ agentUri, _runtime.Options.Durability.AssignedNodeNumber, attempt);
+ }
+ else
+ {
+ _logger.LogInformation("Successfully started agent {AgentUri} on Node {NodeNumber}", agentUri,
+ _runtime.Options.Durability.AssignedNodeNumber);
+ }
+
+ return agent;
+ }
+ catch (Exception e)
+ {
+ if (attempt >= maxAttempts || _cancellation.IsCancellationRequested)
+ {
+ throw new AgentStartingException(agentUri, _runtime.Options.UniqueNodeId, e);
+ }
+
+ var delay = _runtime.Options.Durability.AgentStartRetryDelay * attempt;
+ _logger.LogWarning(e,
+ "Attempt {Attempt} of {MaxAttempts} to start agent {AgentUri} on node {NodeNumber} failed; retrying in {Delay}",
+ attempt, maxAttempts, agentUri, _runtime.Options.Durability.AssignedNodeNumber, delay);
+
+ if (delay > TimeSpan.Zero)
+ {
+ try
+ {
+ await Task.Delay(delay, _cancellation.Token);
+ }
+ catch (OperationCanceledException)
+ {
+ // Shutting down mid-retry. Report the start failure the caller was actually
+ // waiting on rather than a cancellation that says nothing about why it failed.
+ throw new AgentStartingException(agentUri, _runtime.Options.UniqueNodeId, e);
+ }
+ }
+ }
+ }
+ }
+
// Persist that this node owns agentUri. AddAssignmentAsync is an upsert, so this is safe to call for a
// freshly started agent or an already-running one whose assignment row may have been lost.
// ensureLocalNodeRegisteredAsync first side-steps FK problems and timing issues (the assignment row
@@ -303,6 +386,80 @@ public Uri[] AllRunningAgentUris()
return Agents.Where(x => x.Value.Status != AgentStatus.Stopped).Select(x => x.Key).ToArray();
}
+ ///
+ /// Whether an event-subscription agent's failure is one an automated restart could plausibly clear.
+ /// A transient infrastructure fault () is exactly what
+ /// restart-on-stall exists for; every other category is bound to a specific event or to two processes
+ /// racing one shard, and will reproduce identically on the next start. A null failure means the agent
+ /// isn't reporting one — the ordinary wedged-shard case GH-3519 recovers — so it stays restartable.
+ ///
+ private static bool canSelfHeal(ShardFailure? failure)
+ => failure == null || failure.Category == ShardFailureCategory.Other;
+
+ ///
+ /// Surface a locally-owned agent that stopped or paused on a failure: log it with the classified
+ /// reason and notify observers, once per transition into the failed state. See GH-3637 / GH-3638.
+ ///
+ private async Task reportAgentPausedAsync(Uri agentUri, ShardFailure? failure)
+ {
+ if (!_reportedFailures.TryAdd(agentUri, 0))
+ {
+ return;
+ }
+
+ _logger.LogError(
+ "Agent {AgentUri} on node {NodeNumber} is not running because of a failure it cannot recover from by restarting: {Failure}. It will be left alone until the underlying problem is resolved.{Detail}",
+ agentUri, _runtime.Options.Durability.AssignedNodeNumber,
+ failure?.ToString() ?? "no reason reported",
+ failure == null ? string.Empty : Environment.NewLine + failure.Detail);
+
+ try
+ {
+ await _observer.AgentPaused(agentUri, failure);
+ }
+ catch (Exception e)
+ {
+ _logger.LogError(e, "Error notifying observers that agent {AgentUri} paused", agentUri);
+ }
+ }
+
+ ///
+ /// Sweep this node's own agents for any that have stopped or paused underneath us on a reported
+ /// failure. Runs on every node on every health-check tick, independently of leadership: the daemon
+ /// pauses a shard on the node that owns it, and before this nothing in the assignment plane ever
+ /// distinguished a paused shard from a running one — the anti-thrash guards kept it from being
+ /// restarted in a loop, so its progress just silently flatlined. See GH-3637 / GH-3638.
+ ///
+ internal async Task ReportFailedLocalAgentsAsync()
+ {
+ foreach (var entry in Agents.ToArray())
+ {
+ if (entry.Value is not IEventSubscriptionAgent subscription)
+ {
+ continue;
+ }
+
+ // Status is read once: it delegates to the live daemon shard (GH-3519), so two reads in one
+ // pass can legitimately disagree.
+ var status = subscription.Status;
+ if (status == AgentStatus.Running)
+ {
+ _reportedFailures.TryRemove(entry.Key, out _);
+ continue;
+ }
+
+ var failure = subscription.Failure;
+ if (failure == null)
+ {
+ // Stopped with nothing to report is the ordinary wedged-shard case; GH-3519's recovery in
+ // StartAgentAsync owns that one, and reporting it here would fire on every stop.
+ continue;
+ }
+
+ await reportAgentPausedAsync(entry.Key, failure);
+ }
+ }
+
///
/// THIS IS STRICTLY FOR TESTING
///
diff --git a/src/Wolverine/Runtime/Agents/NodeRecordType.cs b/src/Wolverine/Runtime/Agents/NodeRecordType.cs
index 59b12316a..9a75a6ff6 100644
--- a/src/Wolverine/Runtime/Agents/NodeRecordType.cs
+++ b/src/Wolverine/Runtime/Agents/NodeRecordType.cs
@@ -20,7 +20,17 @@ public enum NodeRecordType
/// pg_terminate_backend, AlwaysOn failover, etc.) and stepped down so a
/// new leadership election could happen. See GH-2602.
///
- LeadershipLost
+ LeadershipLost,
+
+ ///
+ /// A locally-owned agent stopped running because of a failure it reported rather
+ /// than because anything asked it to stop — an event-subscription shard the daemon
+ /// paused on an ApplyEventException being the case this exists for. The
+ /// record's Description carries the classified reason (category, failing event,
+ /// root exception) so the failure is readable after the fact, and from another
+ /// process, instead of only in this node's logs. See GH-3637 / GH-3638.
+ ///
+ AgentPaused
}
// This is marked as ISerializable so that it can go to CritterWatch w/o