Adopt JasperFx.Events 2.36.0: shard failure classification, drain timeout docs, natural key extraction

Directory.Packages.props

@@ -134,14 +134,32 @@
          AggregationRunner now always invokes the 3-arg overload with slice.Id, so a projection can
          recover the slice identity even when slice.Snapshot is null (e.g. a deleted MultiStreamProjection
          slice) to emit a follow-on event or publish a message. The old 2-arg override still works; the
-         default 3-arg implementation delegates to it. -->
-    <PackageVersion Include="JasperFx" Version="2.35.0" />
-    <PackageVersion Include="JasperFx.Events" Version="2.35.0" />
-    <PackageVersion Include="JasperFx.Events.SourceGenerator" Version="2.35.0">
+         default 3-arg implementation delegates to it.
+         JasperFx 2.36.0: three daemon/projection changes Marten consumes directly.
+         jasperfx#564 (marten#5047) — DaemonSettings.StopAndDrainTimeout bounds how long a single
+         shard's graceful stop-and-drain may take before it is cancelled; the default of 5 seconds is
+         what was hardcoded before, so nothing changes unless configured.
+         jasperfx#565/#567 (marten#5048) — ShardFailure/ShardFailureCategory/EventFailureDetails and the
+         IEventFailureContext seam an exception implements to declare its own failure category and name
+         the event it broke on. ShardState.Failure rides along on the paused/stopped states, and
+         IEventDatabase.WriteExtendedProgressionAsync documents persisting (and clearing) it.
+         jasperfx#569/#571 (marten#5052) — NaturalKeyEventMapping.Extractor widens from
+         Func<object, object?> to Func<IEvent, object?> so an IEvent<T> [NaturalKeySource] handler is
+         bindable, and an unbindable one now fails loudly from AssembleAndAssertValidity().
+         JasperFx 2.36.1: jasperfx#572/#573 (found here) — ShardStateTracker now takes one lock across
+         recording a published state and capturing the listener list, and ShardStatusWatcher subscribes
+         and reads the current-state snapshot through it as a single atomic step. 2.36.0's #568 narrowed
+         the lost-wakeup window between those two paths but left it open, so a watcher could fall between
+         them and see the state on NEITHER — and once a high water agent reaches the head it has nothing
+         left to publish, so the wait could only end in a timeout no matter how generous. That surfaced
+         as HighWaterAgentTests.skips_multiple_gaps_and_keeps_advancing failing ~8 of 10 runs. -->
+    <PackageVersion Include="JasperFx" Version="2.36.1" />
+    <PackageVersion Include="JasperFx.Events" Version="2.36.1" />
+    <PackageVersion Include="JasperFx.Events.SourceGenerator" Version="2.36.1">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageVersion>
-    <PackageVersion Include="JasperFx.SourceGenerator" Version="2.35.0" />
+    <PackageVersion Include="JasperFx.SourceGenerator" Version="2.36.1" />
     <PackageVersion Include="Jil" Version="3.0.0-alpha2" />
     <PackageVersion Include="Lamar" Version="7.1.1" />
     <PackageVersion Include="Lamar.Microsoft.DependencyInjection" Version="15.0.0" />

docs/cSpell.json

@@ -38,8 +38,10 @@
     "Netstandard",
     "Guid",
     "Guids",
+    "parameterless",
     "pluggable",
     "runtimes",
+    "unbindable",
     "Travelling",
     "travelling",
     "Upcasting",

docs/events/natural-keys.md

@@ -123,22 +123,64 @@ Events that do not affect the natural key (like `OrderItemAdded` in the example
 
 The lookup table is written *inline* as events are appended, well before any projection has built the
 aggregate — that is what lets `FetchForWriting` by natural key work even when the snapshot lifecycle is
-`Async`. The key value therefore has to be derivable from the event alone, and a `[NaturalKeySource]`
-method must be one of:
+`Async`. The key value therefore has to be derivable from the event alone.
 
-- a static factory or evolve method on the aggregate taking the raw event type, such as
-  `public static Order Create(OrderCreated e)` or `public static Order Apply(OrderRenumbered e, Order current)`
-- an instance `Apply(TEvent)` method on the aggregate whose body sets only the natural key property
+Marten tries three strategies for a `[NaturalKeySource]` method, in descending order of trustworthiness:
 
-::: warning
-Do not write a `[NaturalKeySource]` method whose new key value depends on the *previous* aggregate state,
-and do not rely on any aggregate state other than the natural key inside one. Marten derives the key by
-calling your method with a blank aggregate instance, so anything else on it will be `null` or default.
+1. **A static method returning the natural key type**, taking either the raw event or `IEvent<TEvent>` —
+   for example `public static OrderNumber KeyFor(IEvent<OrderRenumbered> e) => new(e.Data.NewNumber)`.
+   This is a pure function of the event: nothing is fabricated and none of your aggregation code runs to
+   work the key out. Prefer it.
+2. **A property of the key's type carried on the event body**, when there is exactly one. An event that
+   carries both the old and the new key is ambiguous, so this strategy declines rather than guessing.
+3. **Invoking your method against a blank aggregate** — a static factory or evolve method such as
+   `public static Order Create(OrderCreated e)`, or an instance `Apply(TEvent)` whose body sets the key.
+   The key is read off whatever the method returned.
 
-Signatures taking `IEvent<T>` rather than the raw event type are not currently supported here, and are
-silently ignored rather than reported — see [JasperFx/jasperfx#569](https://github.com/JasperFx/jasperfx/issues/569).
+::: warning
+**A `[NaturalKeySource]` method never sees the current aggregate.** Do not write one whose new key value
+depends on the *previous* aggregate state, and do not read any aggregate state other than the natural key
+inside one. Under strategy 3 Marten calls your method with a *blank* aggregate, so everything else on it
+is `null` or default.
+
+This is a consequence of when the lookup table is written, not an oversight. The table is maintained
+inline at append time — that is the whole reason a natural key lookup works under an `Async` snapshot
+lifecycle, where the aggregate may not have been built yet, and it is why the key has to be a function of
+the event alone. A method that needs the prior state to work out the new key cannot be supported here;
+carry the value you need on the event instead.
+
+Strategy 3 is skipped entirely when the aggregate cannot be safely constructed — most commonly because it
+declares `required` members that a parameterless constructor cannot satisfy. Marten will not hand your
+method an instance that C# itself would not have let you create. Use strategy 1 or the explicit
+registration below for those types.
 :::
 
+If none of the three can bind a method, Marten throws an `InvalidProjectionException` when the projection
+is registered, naming the method and the reason. (Before Marten 9.20 / JasperFx.Events 2.36.0 an
+unbindable method was silently dropped, so the lookup table was simply never written for that event type
+and the first sign of trouble was a natural key lookup returning null at runtime — see
+[JasperFx/jasperfx#569](https://github.com/JasperFx/jasperfx/issues/569).)
+
+### Explicit Registration <Badge type="tip" text="9.20" />
+
+When attribute discovery cannot bind your method — or when you would simply rather be explicit — register
+the mapping directly with `NaturalKeyFor()` on the projection. An explicit registration replaces whatever
+discovery found for the same event type and clears the configuration-time error an unbindable method
+would otherwise raise:
+
+```cs
+opts.Projections.Snapshot<Order>(SnapshotLifecycle.Async, p =>
+    ((SingleStreamProjection<Order, Guid>)p).NaturalKeyFor(x => x
+        // Derive the key from the event body
+        .SetBy<OrderCreated>(e => new OrderNumber(e.Number))
+        // ...or from the whole event, when the key depends on metadata
+        // such as the stream key, timestamp, or headers
+        .SetByEvent<OrderRenumbered>(e => new OrderNumber(e.Data.NewNumber))));
+```
+
+The same `NaturalKeyFor()` method is available on a projection class you register with
+`Projections.Add(...)`; call it from the projection's constructor.
+
 ## Storage
 
 Marten automatically creates and manages a lookup table for each aggregate type that has a natural key configured. The table maps natural key values to stream ids and is:

docs/events/projections/async-daemon.md

@@ -139,6 +139,42 @@ behavior. The governors apply to continuous (running daemon) work only — proje
 by `MaxConcurrentRebuildsPerDatabase`, which derives its default from the Npgsql connection pool size. See
 [Capping Rebuild Concurrency](/events/projections/rebuilding#capping-rebuild-concurrency).
 
+## Graceful Shutdown and the Drain Timeout <Badge type="tip" text="9.20" />
+
+When a projection or subscription shard is stopped, the daemon does not simply cancel it. It first tries to
+*drain* the agent: let the in-flight page of events finish being applied, then flush the shard's progression row
+so the next start picks up exactly where this one left off. `StopAndDrainTimeout` bounds how long the daemon
+waits for that drain on **a single** shard:
+
+```cs
+// The default is 5 seconds
+opts.Projections.StopAndDrainTimeout = 30.Seconds();
+```
+
+The bound applies to every stop path: stopping one agent, stopping all agents (the `SIGTERM`/host shutdown
+path), and the internal stop-if-already-running replacement that happens when an agent is reassigned.
+
+**Why you would raise it.** If the drain is cut off before the progression flush lands, the shard restarts
+against a stale progression row and throws `ProgressionProgressOutOfOrderException` on its next start. Raise
+the timeout when in-flight batches legitimately take longer than five seconds — a large `BatchSize`, expensive
+projection code, heavy rebuild load, or a slow or contended database. This is most visible shutting down a host
+with a large agent universe: a [database-per-tenant](/configuration/multitenancy) deployment with thousands of
+(projection × tenant) shards all draining inside a Kubernetes termination grace window.
+
+::: tip
+A per-shard bound is only useful if the process lives long enough to spend it. Match a raised
+`StopAndDrainTimeout` with the host's own `HostOptions.ShutdownTimeout` and, on Kubernetes, the pod's
+`terminationGracePeriodSeconds`.
+:::
+
+**Why you would lower it.** A deployment that would rather cut a wedged shard loose quickly and take the
+progression replay hit — to keep node failover and reassignment latency low, for instance — can set it below
+the default.
+
+**Opting out.** `Timeout.InfiniteTimeSpan`, or any non-positive value, removes the separate bound so the drain
+is limited only by the daemon's own cancellation. Be aware that this means a genuinely wedged shard can hold up
+shutdown indefinitely.
+
 ## Daemon Logging
 
 The daemon logs through the standard .Net `ILogger` interface service registered in your application's underlying DI container. In the case of the daemon having to skip
@@ -541,13 +577,44 @@ that just emits and update every time that Marten has to "skip" stale events.
 
 ## Extended Progression Tracking
 
-Extended progression tracking adds six monitoring columns (`heartbeat`,
+Extended progression tracking adds ten monitoring columns (`heartbeat`,
 `agent_status`, `pause_reason`, `running_on_node`, `warning_behind_threshold`,
-`critical_behind_threshold`) to `mt_event_progression`. The async daemon writes
+`critical_behind_threshold`, `failure_category`, `failure_event_sequence`,
+`failure_event_type`, `failure_event_tenant_id`) to `mt_event_progression`. The async daemon writes
 them from existing runtime state and the shard-state selector reads them back
 into `ShardState` so monitoring tooling such as CritterWatch can display
 per-shard health.
 
+### Why a shard is down <Badge type="tip" text="9.20" />
+
+The four `failure_*` columns record the *classified* reason a shard paused or stopped, so a consumer
+polling the database — which is exactly what a monitoring tool must fall back to when the node that
+was running the shard is down — sees the same reason an in-process `ShardState` observer does instead
+of only that the shard is `Paused`. They are read back onto `ShardState.Failure`:
+
+```cs
+var states = await store.Storage.Database.AllProjectionProgress();
+foreach (var state in states.Where(x => x.Failure != null))
+{
+    // ApplyEvent, EventSerialization, UnknownEventType, ProgressionOutOfOrder, or Other
+    Console.WriteLine($"{state.ShardName}: {state.Failure!.Category} on {state.Failure.Event}");
+}
+```
+
+`failure_category` stores the enum *name* rather than its ordinal, so reordering
+`ShardFailureCategory` in a future release can never silently re-label rows an older deployment wrote.
+The reason *text* has no column of its own — `ShardFailure.Detail` is exactly what `pause_reason` has
+always carried.
+
+Marten's own read-path exceptions declare their category, so a body that fails to deserialize reports
+`EventSerialization` with the offending event's sequence and type alias, and an event type alias with
+no registered .NET type reports `UnknownEventType`. The two are kept apart deliberately: bad data
+needs a serializer or data fix, while a missing registration is usually a deployment gap or a rollback
+past the event type's introduction.
+
+A shard that recovers clears its failure columns on the next successful start, so a supervisor built on
+them does not keep alerting on a failure that was fixed an hour ago.
+
 **Default: off**. The columns are useful for
 any stuck-shard diagnosis -- not just CritterWatch -- and the write-side cost is
 negligible because they're already-computed daemon-internal values. When enabled,

src/CoreTests/Exceptions/EventFailureContextTests.cs

@@ -0,0 +1,110 @@
+using System;
+using JasperFx.Events.Daemon;
+using Marten.Events;
+using Marten.Exceptions;
+using Shouldly;
+using Xunit;
+
+namespace CoreTests.Exceptions;
+
+// #5048 / jasperfx#565. The daemon deliberately has NO fallback type-name sniffing: a store's exception
+// declares its own ShardFailureCategory through IEventFailureContext, or the failure classifies as
+// "Other" with no event details. These tests pin that contract on Marten's two read-path exceptions.
+public class EventFailureContextTests
+{
+    public record CorruptedEvent(string Name);
+
+    private static EventMapping<CorruptedEvent> mapping()
+    {
+        var graph = new EventGraph(new Marten.StoreOptions());
+        return (EventMapping<CorruptedEvent>)graph.EventMappingFor<CorruptedEvent>();
+    }
+
+    [Fact]
+    public void deserialization_failure_declares_its_category_and_names_the_event()
+    {
+        var eventType = mapping();
+        IEventFailureContext exception =
+            new EventDeserializationFailureException(4815, eventType, new DivideByZeroException("Boom!"));
+
+        exception.Category.ShouldBe(ShardFailureCategory.EventSerialization);
+        exception.Sequence.ShouldBe(4815);
+
+        // The constructor has always been handed the IEventType and used it only to build the message
+        // string. Retaining the alias is the point of the change.
+        exception.EventTypeName.ShouldBe(eventType.EventTypeName);
+
+        // Raised while reading the row, before there is an IEvent, so nothing else is knowable
+        exception.EventId.ShouldBeNull();
+        exception.StreamId.ShouldBeNull();
+        exception.StreamKey.ShouldBeNull();
+        exception.TenantId.ShouldBeNull();
+        exception.Version.ShouldBeNull();
+    }
+
+    [Fact]
+    public void unknown_event_type_is_a_separate_category_from_serialization()
+    {
+        // A missing registration is a deployment fix, not a data fix, so it must not classify as
+        // EventSerialization.
+        IEventFailureContext exception = new UnknownEventTypeException("trip_started", 1623);
+
+        exception.Category.ShouldBe(ShardFailureCategory.UnknownEventType);
+        exception.Sequence.ShouldBe(1623);
+        exception.EventTypeName.ShouldBe("trip_started");
+    }
+
+    [Fact]
+    public void unknown_event_type_reports_an_unknown_sequence_when_the_throw_site_has_no_row()
+    {
+        IEventFailureContext exception = new UnknownEventTypeException("trip_started");
+
+        exception.Sequence.ShouldBe(UnknownEventTypeException.UnknownSequence);
+    }
+
+    [Fact]
+    public void shard_failure_classifies_a_deserialization_failure_through_wrapping()
+    {
+        var eventType = mapping();
+        var inner = new EventDeserializationFailureException(99, eventType, new DivideByZeroException("Boom!"));
+
+        // The per-event exception routinely reaches the daemon wrapped -- ShardStopException around it,
+        // or an AggregateException of a whole batch's failures. ShardFailure.For walks the entire graph.
+        var wrapped = new AggregateException(new InvalidOperationException("unrelated", new Exception("leaf")),
+            new ShardStopException("Trip:All", inner));
+
+        var occurredAt = new DateTimeOffset(2026, 7, 26, 12, 0, 0, TimeSpan.Zero);
+        var failure = ShardFailure.For(wrapped, occurredAt);
+
+        failure.Category.ShouldBe(ShardFailureCategory.EventSerialization);
+        failure.Event.ShouldNotBeNull();
+        failure.Event.Sequence.ShouldBe(99);
+        failure.Event.EventTypeName.ShouldBe(eventType.EventTypeName);
+        failure.OccurredAt.ShouldBe(occurredAt);
+    }
+
+    [Fact]
+    public void shard_failure_classifies_an_unknown_event_type_through_wrapping()
+    {
+        var failure = ShardFailure.For(
+            new ShardStopException("Trip:All", new UnknownEventTypeException("trip_started", 77)),
+            DateTimeOffset.UtcNow);
+
+        failure.Category.ShouldBe(ShardFailureCategory.UnknownEventType);
+        failure.Event!.Sequence.ShouldBe(77);
+        failure.Event.EventTypeName.ShouldBe("trip_started");
+    }
+
+    [Fact]
+    public void dead_letter_event_id_is_assigned_before_the_write()
+    {
+        var exception = new EventDeserializationFailureException(12, mapping(), new DivideByZeroException("Boom!"));
+
+        var deadLetter = exception.ToDeadLetterEvent(new JasperFx.Events.Projections.ShardName("Trip", "All", 1));
+
+        // Previously left Guid.Empty for document identity generation to fill in at write time, which
+        // meant the creating process could not correlate its ShardFailure with the row it produced.
+        deadLetter.Id.ShouldNotBe(Guid.Empty);
+        deadLetter.EventSequence.ShouldBe(12);
+    }
+}