diff --git a/Directory.Packages.props b/Directory.Packages.props index f2cbd6590f..43ef228571 100644 --- a/Directory.Packages.props +++ b/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. --> - - - + 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 to Func so an IEvent [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. --> + + + all runtime; build; native; contentfiles; analyzers; buildtransitive - + diff --git a/docs/cSpell.json b/docs/cSpell.json index a15696d61e..9db7e5e444 100644 --- a/docs/cSpell.json +++ b/docs/cSpell.json @@ -38,8 +38,10 @@ "Netstandard", "Guid", "Guids", + "parameterless", "pluggable", "runtimes", + "unbindable", "Travelling", "travelling", "Upcasting", diff --git a/docs/events/natural-keys.md b/docs/events/natural-keys.md index 19844fa831..cbd47df35b 100644 --- a/docs/events/natural-keys.md +++ b/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` — + for example `public static OrderNumber KeyFor(IEvent 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` 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 + +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(SnapshotLifecycle.Async, p => + ((SingleStreamProjection)p).NaturalKeyFor(x => x + // Derive the key from the event body + .SetBy(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(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: diff --git a/docs/events/projections/async-daemon.md b/docs/events/projections/async-daemon.md index 880f73f8c6..9cc7fdc8c5 100644 --- a/docs/events/projections/async-daemon.md +++ b/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 + +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 + +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, diff --git a/src/CoreTests/Exceptions/EventFailureContextTests.cs b/src/CoreTests/Exceptions/EventFailureContextTests.cs new file mode 100644 index 0000000000..da9e3dd97b --- /dev/null +++ b/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 mapping() + { + var graph = new EventGraph(new Marten.StoreOptions()); + return (EventMapping)graph.EventMappingFor(); + } + + [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); + } +} diff --git a/src/DaemonTests/Aggregations/Bug_5041_natural_key_source_discovery.cs b/src/DaemonTests/Aggregations/Bug_5041_natural_key_source_discovery.cs index b3bfe0dd21..898f2b1d07 100644 --- a/src/DaemonTests/Aggregations/Bug_5041_natural_key_source_discovery.cs +++ b/src/DaemonTests/Aggregations/Bug_5041_natural_key_source_discovery.cs @@ -6,7 +6,9 @@ using DaemonTests.TestingSupport; using JasperFx.Events; using JasperFx.Events.Aggregation; +using JasperFx.Events.Projections; using Marten; +using Marten.Events.Aggregation; using Marten.Testing.Harness; using Shouldly; using Xunit; @@ -15,20 +17,26 @@ namespace DaemonTests.Aggregations; /// -/// #5041, from the repro in https://github.com/JasperFx/marten/pull/5042 (thanks @ytqsl). +/// #5041, from the repro in https://github.com/JasperFx/marten/pull/5042 (thanks @ytqsl), closed out +/// by #5052 on JasperFx.Events 2.36.0. /// -/// Both of these hang on [NaturalKeySource] discovery in JasperFx.Events, not on anything Marten -/// owns — see https://github.com/JasperFx/jasperfx/issues/569: +/// Both halves of the original report hung on [NaturalKeySource] discovery in JasperFx.Events rather +/// than on anything Marten owns — see https://github.com/JasperFx/jasperfx/issues/569: /// -/// * a handler whose first parameter is IEvent<T> yields no usable extractor, so -/// NaturalKeyDefinition.EventMappings never gains an entry for that event type and the -/// mt_natural_key_X table is silently never written for it; -/// * an instance Apply(TEvent) handler is invoked reflectively against a fabricated blank +/// * a handler whose first parameter is IEvent<T> yielded no usable extractor, so +/// NaturalKeyDefinition.EventMappings never gained an entry for that event type and the +/// mt_natural_key_X table was silently never written for it; +/// * an instance Apply(TEvent) handler was invoked reflectively against a fabricated blank /// aggregate (Expression.New(TDoc), which also bypasses `required` member enforcement), so a -/// handler body that touches any other state throws — out of NaturalKeyProjection.ApplyAsync -/// and out of the caller's SaveChangesAsync. +/// handler body that touched any other state threw — out of NaturalKeyProjection.ApplyAsync and +/// out of the caller's SaveChangesAsync. /// -/// Unskip both when the JasperFx.Events dependency picks up the fix. +/// jasperfx#571 widened the extraction contract from the event DATA to the whole IEvent, which is what +/// makes an IEvent<T> source bindable at all (see NaturalKeyProjection for Marten's two call +/// sites), and made an unbindable source a loud configuration-time error instead of a mapping that +/// silently never existed. The repro's own aggregate shape — an instance handler on a type with +/// `required` members — is one of those unbindable cases BY DESIGN now, so it is pinned as such here, +/// alongside the two supported ways to express the same rename. /// public class Bug_5041_natural_key_source_discovery: DaemonContext { @@ -46,6 +54,11 @@ public sealed record ProductCodeChangedByEventWrapper(Guid ProductId, string New public sealed record ProductCodeChangedByInstanceMethod(Guid ProductId, string NewProductCode); + /// + /// The reporter's aggregate, verbatim. Every [NaturalKeySource] here other than Create needs a + /// prior aggregate to derive the key, and `required IEnumerable<ProductCode> KnownCodes` means + /// no blank one can be safely fabricated. + /// public sealed record Product { public Guid Id { get; set; } @@ -88,30 +101,175 @@ public void Apply(ProductCodeChangedByInstanceMethod e) } } - private static void ConfigureStore(StoreOptions opts) + /// + /// The same aggregate, with the key derived from the event ALONE — a static [NaturalKeySource] + /// returning the key type and taking IEvent<T>. This is the shape that could not bind before + /// jasperfx#571 and is now the highest-ranked strategy: nothing is fabricated and no user + /// aggregation code runs to work out the key. + /// + public sealed record KeyFromEventProduct + { + public Guid Id { get; set; } + + [NaturalKey] + public ProductCode Code { get; set; } + + public required IEnumerable KnownCodes { get; set; } + + [NaturalKeySource] + public static ProductCode KeyOnRegistration(IEvent e) + => new(e.Data.ProductCode); + + [NaturalKeySource] + public static ProductCode KeyOnRename(IEvent e) + => new(e.Data.NewProductCode); + + public static KeyFromEventProduct Create(ProductRegistered e) + { + return new KeyFromEventProduct + { + Id = e.ProductId, + Code = new ProductCode(e.ProductCode), + KnownCodes = [new ProductCode(e.ProductCode)] + }; + } + + public static KeyFromEventProduct Apply(IEvent e, + KeyFromEventProduct product) + { + return product with + { + Code = new ProductCode(e.Data.NewProductCode), + KnownCodes = product.KnownCodes + .Where(c => c.Value != e.Data.NewProductCode) + .Append(new ProductCode(e.Data.NewProductCode)) + }; + } + } + + private static void configureStore(StoreOptions opts, Action? configureProjection = null) { opts.Connection(ConnectionSource.ConnectionString); opts.DatabaseSchemaName = schemaName; opts.Events.StreamIdentity = StreamIdentity.AsGuid; opts.Events.AppendMode = EventAppendMode.Quick; - opts.Projections.Snapshot(SnapshotLifecycle.Async); + opts.Projections.Snapshot(SnapshotLifecycle.Async, configureProjection!); } - [Fact(Skip = "Blocked on JasperFx/jasperfx#569 -- IEvent [NaturalKeySource] handlers yield no extractor")] + // The original bug was that NOTHING happened: no mapping, no log, no error, and the user found out + // when the natural key lookup returned null at runtime. Silence is the regression to guard against. + [Fact] + public void an_unbindable_natural_key_source_fails_loudly_at_configuration_time() + { + var ex = Should.Throw(() => + { + StoreOptions(opts => configureStore(opts)); + }); + + // Names the offending methods, the reason, and both supported ways out + ex.Message.ShouldContain(nameof(ProductCodeChangedByEventWrapper)); + ex.Message.ShouldContain(nameof(ProductCodeChangedByInstanceMethod)); + ex.Message.ShouldContain("required members"); + ex.Message.ShouldContain("NaturalKeyFor"); + } + + // #5042's failing test. The key source takes IEvent, which yielded no extractor at all before + // the contract widened from the event data to the event. + [Fact] public async Task natural_key_is_maintained_when_the_handler_takes_IEvent() { - await runRenameScenario(streamId => new ProductCodeChangedByEventWrapper(streamId, "PROD-999")); + StoreOptions(opts => + { + opts.Connection(ConnectionSource.ConnectionString); + opts.DatabaseSchemaName = schemaName; + opts.Events.StreamIdentity = StreamIdentity.AsGuid; + opts.Events.AppendMode = EventAppendMode.Quick; + opts.Projections.Snapshot(SnapshotLifecycle.Async); + }); + + var streamId = await appendRenameAsync( + id => new ProductCodeChangedByEventWrapper(id, "PROD-999")); + + var daemon = await theStore.BuildProjectionDaemonAsync(); + await daemon.RebuildProjectionAsync(CancellationToken.None); + + await using var query = theStore.LightweightSession(); + var product = await query.Events.FetchLatest(new ProductCode("PROD-999")); + product.ShouldNotBeNull(); + product.Id.ShouldBe(streamId); + product.Code.Value.ShouldBe("PROD-999"); + product.KnownCodes.ShouldContain(new ProductCode("PROD-001")); + product.KnownCodes.ShouldContain(new ProductCode("PROD-999")); + + // #5041 item 2 on a source shape that could not bind at all before jasperfx#571 + (await naturalKeysForStreamAsync("mt_natural_key_keyfromeventproduct", streamId)) + .ShouldBe(["PROD-999"]); + (await query.Events.FetchLatest(new ProductCode("PROD-001"))) + .ShouldBeNull(); + } + + // The escape hatch for the reporter's own aggregate: NaturalKeyBuilder.SetBy/SetByEvent were dead + // code (internal constructor, nothing ever built one) until jasperfx#571 made them reachable. An + // explicit registration replaces whatever discovery found AND clears the configuration-time error. + [Fact] + public async Task natural_key_is_maintained_through_an_explicit_registration() + { + StoreOptions(opts => configureStore(opts, p => + ((SingleStreamProjection)p).NaturalKeyFor(x => x + .SetBy(e => new ProductCode(e.ProductCode)) + .SetByEvent(e => new ProductCode(e.Data.NewProductCode)) + .SetBy(e => new ProductCode(e.NewProductCode))))); + + var streamId = await appendRenameAsync( + id => new ProductCodeChangedByInstanceMethod(id, "PROD-999")); + + var daemon = await theStore.BuildProjectionDaemonAsync(); + await daemon.RebuildProjectionAsync(CancellationToken.None); + + await using var query = theStore.LightweightSession(); + var product = await query.Events.FetchLatest(new ProductCode("PROD-999")); + product.ShouldNotBeNull(); + product.Id.ShouldBe(streamId); + product.Code.Value.ShouldBe("PROD-999"); + + // #5041 item 2 through an explicitly registered extractor + (await naturalKeysForStreamAsync("mt_natural_key_product", streamId)).ShouldBe(["PROD-999"]); + (await query.Events.FetchLatest(new ProductCode("PROD-001"))) + .ShouldBeNull(); } - [Fact(Skip = "Blocked on JasperFx/jasperfx#569 -- instance [NaturalKeySource] handlers run against a blank aggregate")] - public async Task natural_key_is_maintained_when_the_handler_is_an_instance_method() + /// + /// #5041 item 2 — the retired key must not survive alongside the new one, squatting on its slot in + /// the lookup table's primary key. #5049 fixed and covered that, but only for a source shape that + /// discovery could always bind (a static handler taking the raw event). On the two paths this PR + /// newly enables the question could not even be ASKED before, because item 1 meant nothing was + /// written for those event types at all. + /// + private async Task naturalKeysForStreamAsync(string table, Guid streamId) { - await runRenameScenario(streamId => new ProductCodeChangedByInstanceMethod(streamId, "PROD-999")); + await using var conn = theStore.Storage.Database.CreateConnection(); + await conn.OpenAsync(); + + await using var cmd = conn.CreateCommand(); + cmd.CommandText = + $"select natural_key_value from {schemaName}.{table} where stream_id = :id order by natural_key_value"; + var parameter = cmd.CreateParameter(); + parameter.ParameterName = "id"; + parameter.Value = streamId; + cmd.Parameters.Add(parameter); + + var values = new List(); + await using var reader = await cmd.ExecuteReaderAsync(); + while (await reader.ReadAsync()) + { + values.Add(await reader.GetFieldValueAsync(0)); + } + + return values.ToArray(); } - private async Task runRenameScenario(Func renameEvent) + private async Task appendRenameAsync(Func renameEvent) where T : class { - StoreOptions(ConfigureStore); await theStore.Storage.ApplyAllConfiguredChangesToDatabaseAsync(); await theStore.Advanced.Clean.DeleteAllDocumentsAsync(); await theStore.Advanced.Clean.DeleteAllEventDataAsync(); @@ -120,7 +278,7 @@ private async Task runRenameScenario(Func renameEvent) await using (var session = theStore.LightweightSession()) { - session.Events.StartStream(streamId, new ProductRegistered(streamId, "PROD-001")); + session.Events.StartStream(streamId, new ProductRegistered(streamId, "PROD-001")); await session.SaveChangesAsync(); } @@ -130,14 +288,6 @@ private async Task runRenameScenario(Func renameEvent) await session.SaveChangesAsync(); } - var daemon = await theStore.BuildProjectionDaemonAsync(); - await daemon.RebuildProjectionAsync(CancellationToken.None); - - await using var query = theStore.LightweightSession(); - var product = await query.Events.FetchLatest(new ProductCode("PROD-999")); - product.ShouldNotBeNull(); - product.Code.Value.ShouldBe("PROD-999"); - product.KnownCodes.ShouldContain(new ProductCode("PROD-001")); - product.KnownCodes.ShouldContain(new ProductCode("PROD-999")); + return streamId; } } diff --git a/src/DaemonTests/Bug_5048_shard_failure_progression_columns.cs b/src/DaemonTests/Bug_5048_shard_failure_progression_columns.cs new file mode 100644 index 0000000000..66fa8e90f1 --- /dev/null +++ b/src/DaemonTests/Bug_5048_shard_failure_progression_columns.cs @@ -0,0 +1,314 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using DaemonTests.TestingSupport; +using JasperFx.Core; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Projections; +using Marten.Events.Aggregation; +using Marten.Storage; +using Shouldly; +using Weasel.Core; +using Xunit; +using Xunit.Abstractions; + +namespace DaemonTests; + +public record FailureTelemetryEvent(); + +/// +/// Deserializes only while is off. Both serializers run the +/// parameterless constructor, so flipping the flag corrupts every persisted body of this type on the +/// READ side without touching the rows -- which is precisely the shape of the failure #5048 is about. +/// The flag is private to this test class (whose facts never run concurrently with each other) rather +/// than reusing the process-global FailingEvent.SerializationFails. +/// +public class CorruptibleEvent +{ + public static bool DeserializationFails; + + public CorruptibleEvent() + { + if (DeserializationFails) throw new DivideByZeroException("Boom!"); + } +} + +public class FailureTelemetryStream { public Guid Id { get; set; } } + +public partial class FailureTelemetryProjection: SingleStreamProjection +{ + public void Apply(FailureTelemetryEvent @event, FailureTelemetryStream projection) { } + + // The shard only LOADS event types it is interested in, so the corruptible type has to be part of + // the projection for the read to ever reach it. + public void Apply(CorruptibleEvent @event, FailureTelemetryStream projection) { } +} + +// #5048 / jasperfx#565. ShardState.Failure now rides along on the paused/stopped states the daemon +// publishes; these tests pin the persistence half. A supervisor polling the database -- CritterWatch +// when the publishing node is DOWN, which is exactly when this matters -- must see the same classified +// reason an in-process observer does, and must stop seeing it once the shard recovers. +// +// Follows the #5022 pattern from extended_progression_batch_write: seed the committed progression rows +// directly instead of running a daemon, so the ONLY writer against these rows is the call under test. +public class Bug_5048_shard_failure_progression_columns: DaemonContext +{ + private const string TheShard = "FailureTelemetryStream:All"; + + public Bug_5048_shard_failure_progression_columns(ITestOutputHelper output): base(output) + { + } + + private async Task seedProgressionRowAsync() + { + StoreOptions(x => + { + x.Events.EnableExtendedProgressionTracking = true; + x.Projections.Add(new FailureTelemetryProjection(), ProjectionLifecycle.Async); + }); + + var database = (MartenDatabase)theStore.Storage.Database; + await database.EnsureStorageExistsAsync(typeof(IEvent)); + + await using var session = theStore.LightweightSession(); + session.QueueSqlCommand( + $"select {theStore.Events.DatabaseSchemaName}.mt_mark_event_progression(?, ?)", TheShard, 10L); + await session.SaveChangesAsync(); + } + + private async Task<(object? category, object? sequence, object? eventType, object? tenantId)> readFailureAsync() + { + await using var session = theStore.QuerySession(); + await using var reader = await session.Connection + .CreateCommand( + $"select failure_category, failure_event_sequence, failure_event_type, failure_event_tenant_id from {theStore.Events.DatabaseSchemaName}.mt_event_progression where name = :name") + .With("name", TheShard) + .ExecuteReaderAsync(); + + if (!await reader.ReadAsync()) return (null, null, null, null); + + object? at(int i) => reader.GetValue(i) is DBNull ? null : reader.GetValue(i); + return (at(0), at(1), at(2), at(3)); + } + + private static ShardState paused(ShardFailureCategory category, long sequence, string eventTypeName, + string? tenantId = null) + { + var failure = new ShardFailure + { + Category = category, + ExceptionType = "Marten.Exceptions.EventDeserializationFailureException", + RootExceptionType = "System.DivideByZeroException", + Message = "Boom!", + Detail = "Marten.Exceptions.EventDeserializationFailureException: Boom!\n at Somewhere", + OccurredAt = DateTimeOffset.UtcNow, + Event = new EventFailureDetails + { + Sequence = sequence, EventTypeName = eventTypeName, TenantId = tenantId + } + }; + + return new ShardState(TheShard, 10) + { + Action = ShardAction.Paused, + AgentStatus = "Paused", + PauseReason = failure.Detail, + Failure = failure, + LastHeartbeat = DateTimeOffset.UtcNow + }; + } + + private static ShardState withoutFailure(ShardAction action, string status) + { + return new ShardState(TheShard, 10) + { + Action = action, AgentStatus = status, LastHeartbeat = DateTimeOffset.UtcNow + }; + } + + [Fact] + public async Task persists_the_classified_failure_on_a_paused_shard() + { + await seedProgressionRowAsync(); + var database = (MartenDatabase)theStore.Storage.Database; + + await database.WriteExtendedProgressionAsync( + paused(ShardFailureCategory.EventSerialization, 4815, "failure_telemetry_event", "tenant-a")); + + var row = await readFailureAsync(); + + // The enum NAME, never the ordinal -- reordering ShardFailureCategory must not silently re-label + // rows that were written by an older deployment. + row.category.ShouldBe(nameof(ShardFailureCategory.EventSerialization)); + Convert.ToInt64(row.sequence).ShouldBe(4815); + row.eventType.ShouldBe("failure_telemetry_event"); + row.tenantId.ShouldBe("tenant-a"); + } + + [Fact] + public async Task a_recovered_shard_stops_reporting_the_reason_it_paused() + { + await seedProgressionRowAsync(); + var database = (MartenDatabase)theStore.Storage.Database; + + await database.WriteExtendedProgressionAsync( + paused(ShardFailureCategory.ApplyEvent, 99, "failure_telemetry_event")); + (await readFailureAsync()).category.ShouldNotBeNull(); + + // A restart supersedes whatever paused the agent last. Without this, every supervisor built on + // these columns alerts forever on a failure the operator already fixed. + await database.WriteExtendedProgressionAsync(withoutFailure(ShardAction.Started, "Running")); + + var row = await readFailureAsync(); + row.category.ShouldBeNull(); + row.sequence.ShouldBeNull(); + row.eventType.ShouldBeNull(); + row.tenantId.ShouldBeNull(); + } + + [Fact] + public async Task a_failureless_non_start_publication_leaves_the_reason_alone() + { + await seedProgressionRowAsync(); + var database = (MartenDatabase)theStore.Storage.Database; + + await database.WriteExtendedProgressionAsync( + paused(ShardFailureCategory.EventSerialization, 4815, "failure_telemetry_event")); + + // This is the load-bearing case: SubscriptionAgent publishes a plain Stopped state (no Failure) + // right behind the Paused one, and a heartbeat can arrive with no failure at all. An + // unconditional write would erase the reason microseconds after recording it. + await database.WriteExtendedProgressionAsync(withoutFailure(ShardAction.Stopped, "Stopped")); + await database.WriteExtendedProgressionAsync(withoutFailure(ShardAction.Updated, "Running")); + + var row = await readFailureAsync(); + row.category.ShouldBe(nameof(ShardFailureCategory.EventSerialization)); + Convert.ToInt64(row.sequence).ShouldBe(4815); + } + + [Fact] + public async Task rehydrates_the_failure_on_the_read_side() + { + await seedProgressionRowAsync(); + var database = (MartenDatabase)theStore.Storage.Database; + + await database.WriteExtendedProgressionAsync( + paused(ShardFailureCategory.UnknownEventType, 1623, "trip_started", "tenant-b")); + + // A poller must get the same shape as a live ShardState observer, not just "it's Paused" + var states = await database.AllProjectionProgress(); + var state = states.Single(x => x.ShardName == TheShard); + + state.Failure.ShouldNotBeNull(); + state.Failure.Category.ShouldBe(ShardFailureCategory.UnknownEventType); + state.Failure.Event.ShouldNotBeNull(); + state.Failure.Event.Sequence.ShouldBe(1623); + state.Failure.Event.EventTypeName.ShouldBe("trip_started"); + state.Failure.Event.TenantId.ShouldBe("tenant-b"); + + // ShardFailure.Detail is exactly what PauseReason has always carried, which is why the reason + // text needed no column of its own. + state.Failure.Detail.ShouldBe(state.PauseReason); + } + + [Fact] + public async Task a_healthy_shard_reports_no_failure() + { + await seedProgressionRowAsync(); + var database = (MartenDatabase)theStore.Storage.Database; + + await database.WriteExtendedProgressionAsync(withoutFailure(ShardAction.Started, "Running")); + + var states = await database.AllProjectionProgress(); + states.Single(x => x.ShardName == TheShard).Failure.ShouldBeNull(); + } + + // The acceptance case the issue was written for: before this, a shard paused by a corrupted event + // body classified as ShardFailureCategory.Other with no event details, because the daemon has no + // fallback type-name sniffing -- Marten's exception has to declare its own category. + [Fact] + public async Task a_projection_paused_by_a_corrupted_event_body_reports_the_serialization_failure() + { + CorruptibleEvent.DeserializationFails = false; + + StoreOptions(x => + { + x.Events.EnableExtendedProgressionTracking = true; + x.Projections.Add(new FailureTelemetryProjection(), ProjectionLifecycle.Async); + // Pause on a bad body instead of dead-lettering it and moving on + x.Projections.Errors.SkipSerializationErrors = false; + }, true); + + var streamId = Guid.NewGuid(); + await using (var session = theStore.LightweightSession()) + { + session.Events.StartStream(streamId, new FailureTelemetryEvent(), new CorruptibleEvent()); + await session.SaveChangesAsync(); + } + + long corruptedSequence; + await using (var session = theStore.QuerySession()) + { + var events = await session.Events.FetchStreamAsync(streamId); + corruptedSequence = events.Single(x => x.EventType == typeof(CorruptibleEvent)).Sequence; + } + + var alias = theStore.Events.EventMappingFor().EventTypeName; + + try + { + CorruptibleEvent.DeserializationFails = true; + + using var daemon = await StartDaemon(); + var waiter = daemon.Tracker.WaitForShardCondition(x => x.Failure != null, + "the shard reports a classified failure", 30.Seconds()); + + await daemon.StartAllAsync(); + + var state = await waiter; + + state.Failure.ShouldNotBeNull(); + state.Failure.Category.ShouldBe(ShardFailureCategory.EventSerialization); + state.Failure.Event.ShouldNotBeNull(); + state.Failure.Event.Sequence.ShouldBe(corruptedSequence); + state.Failure.Event.EventTypeName.ShouldBe(alias); + + await daemon.StopAllAsync(); + } + finally + { + CorruptibleEvent.DeserializationFails = false; + } + } + + [Fact] + public async Task never_inserts_a_row_and_never_touches_committed_progression() + { + await seedProgressionRowAsync(); + var database = (MartenDatabase)theStore.Storage.Database; + + await database.WriteExtendedProgressionAsync([ + paused(ShardFailureCategory.ApplyEvent, 7, "failure_telemetry_event"), + // A shard that has never committed progression has nowhere to record a reason: skipped + // silently, exactly like every other extended-progression write. + new ShardState("NoSuchProjection:All:98123456", 10) + { + Action = ShardAction.Paused, AgentStatus = "Paused", LastHeartbeat = DateTimeOffset.UtcNow + } + ]); + + await using var session = theStore.QuerySession(); + var rows = Convert.ToInt64(await session.Connection + .CreateCommand($"select count(*) from {theStore.Events.DatabaseSchemaName}.mt_event_progression") + .ExecuteScalarAsync()); + rows.ShouldBe(1); + + var sequence = Convert.ToInt64(await session.Connection + .CreateCommand( + $"select last_seq_id from {theStore.Events.DatabaseSchemaName}.mt_event_progression where name = :name") + .With("name", TheShard) + .ExecuteScalarAsync()); + sequence.ShouldBe(10); // committed progress untouched + } +} diff --git a/src/Marten/Events/Daemon/Progress/ProjectionProgressStatement.cs b/src/Marten/Events/Daemon/Progress/ProjectionProgressStatement.cs index 5c78faa589..1feceb184e 100644 --- a/src/Marten/Events/Daemon/Progress/ProjectionProgressStatement.cs +++ b/src/Marten/Events/Daemon/Progress/ProjectionProgressStatement.cs @@ -37,9 +37,14 @@ public ProjectionProgressStatement(EventGraph events) protected override void configure(ICommandBuilder builder) { + // #5048 / jasperfx#565: the failure_* columns trail the existing extended block so the ordinals + // ShardStateSelector walks stay stable. + const string extendedColumns = + "heartbeat, agent_status, pause_reason, running_on_node, warning_behind_threshold, critical_behind_threshold, failure_category, failure_event_sequence, failure_event_type, failure_event_tenant_id"; + if (_events.UseOptimizedProjectionRebuilds && _events.EnableExtendedProgressionTracking) { - builder.Append($"select name, last_seq_id, mode, rebuild_threshold, assigned_node, heartbeat, agent_status, pause_reason, running_on_node, warning_behind_threshold, critical_behind_threshold from {_events.DatabaseSchemaName}.mt_event_progression"); + builder.Append($"select name, last_seq_id, mode, rebuild_threshold, assigned_node, {extendedColumns} from {_events.DatabaseSchemaName}.mt_event_progression"); } else if (_events.UseOptimizedProjectionRebuilds) { @@ -47,7 +52,7 @@ protected override void configure(ICommandBuilder builder) } else if (_events.EnableExtendedProgressionTracking) { - builder.Append($"select name, last_seq_id, heartbeat, agent_status, pause_reason, running_on_node, warning_behind_threshold, critical_behind_threshold from {_events.DatabaseSchemaName}.mt_event_progression"); + builder.Append($"select name, last_seq_id, {extendedColumns} from {_events.DatabaseSchemaName}.mt_event_progression"); } else { diff --git a/src/Marten/Events/Daemon/Progress/ShardStateSelector.cs b/src/Marten/Events/Daemon/Progress/ShardStateSelector.cs index 2cedf9106a..35d1eb99cc 100644 --- a/src/Marten/Events/Daemon/Progress/ShardStateSelector.cs +++ b/src/Marten/Events/Daemon/Progress/ShardStateSelector.cs @@ -2,6 +2,8 @@ using System.Data.Common; using System.Threading; using System.Threading.Tasks; +using JasperFx.Core; +using JasperFx.Events.Daemon; using JasperFx.Events.Projections; using Marten.Linq.Selectors; @@ -81,8 +83,88 @@ public async Task ResolveAsync(DbDataReader reader, CancellationToke state.CriticalBehindThreshold = await reader.GetFieldValueAsync(nextIndex, token).ConfigureAwait(false); } nextIndex++; + + // #5048 / jasperfx#565: rehydrate the classified failure so a consumer polling the database + // (CritterWatch when the publishing node is down) gets the same shape as a live ShardState + // observer, rather than only being able to see that the shard is Paused. + string? category = null; + if (!await reader.IsDBNullAsync(nextIndex, token).ConfigureAwait(false)) + { + category = await reader.GetFieldValueAsync(nextIndex, token).ConfigureAwait(false); + } + nextIndex++; + + long? failureSequence = null; + if (!await reader.IsDBNullAsync(nextIndex, token).ConfigureAwait(false)) + { + failureSequence = await reader.GetFieldValueAsync(nextIndex, token).ConfigureAwait(false); + } + nextIndex++; + + string? failureEventType = null; + if (!await reader.IsDBNullAsync(nextIndex, token).ConfigureAwait(false)) + { + failureEventType = await reader.GetFieldValueAsync(nextIndex, token).ConfigureAwait(false); + } + nextIndex++; + + string? failureTenantId = null; + if (!await reader.IsDBNullAsync(nextIndex, token).ConfigureAwait(false)) + { + failureTenantId = await reader.GetFieldValueAsync(nextIndex, token).ConfigureAwait(false); + } + nextIndex++; + + state.Failure = buildFailure(category, failureSequence, failureEventType, failureTenantId, state); } return state; } + + /// + /// #5048 / jasperfx#565: the persisted row is a lossy projection of — by + /// design, since the columns exist to answer "why is this shard down" and not to reconstitute an + /// exception. failure_category is the presence flag: no category means no failure to report. + /// + private static ShardFailure? buildFailure(string? category, long? sequence, string? eventTypeName, + string? tenantId, ShardState state) + { + if (category.IsEmpty() || !Enum.TryParse(category, out var parsed)) + { + return null; + } + + // ShardFailure.Detail is exactly what PauseReason has always carried, which is why the text needed + // no column of its own. Message is the same text here: the short form is a property of the live + // Exception, and that is not persisted. + var detail = state.PauseReason ?? string.Empty; + + return new ShardFailure + { + Category = parsed, + // Not persisted — the reason text in Detail is what an operator acts on, and inventing a type + // name would be worse than admitting we don't have one. + ExceptionType = UnknownExceptionType, + RootExceptionType = UnknownExceptionType, + Message = detail, + Detail = detail, + // The pause/stop publication stamps LastHeartbeat at the moment it classifies the failure, so + // the persisted heartbeat is the closest thing the row has to the failure's timestamp. + OccurredAt = state.LastHeartbeat ?? default, + Event = sequence.HasValue + ? new EventFailureDetails + { + Sequence = sequence.Value, EventTypeName = eventTypeName, TenantId = tenantId + } + : null + }; + } + + /// + /// Placeholder for / + /// on a failure rehydrated from the database, where the exception types were never persisted. Both members + /// are required on the record, so a sentinel is unavoidable; a distinctive one beats an empty string + /// that reads like a real (blank) type name. + /// + internal const string UnknownExceptionType = "Unknown"; } diff --git a/src/Marten/Events/EventDocumentStorage.cs b/src/Marten/Events/EventDocumentStorage.cs index 9e375a2a3f..0a17066f09 100644 --- a/src/Marten/Events/EventDocumentStorage.cs +++ b/src/Marten/Events/EventDocumentStorage.cs @@ -331,7 +331,7 @@ public IEvent Resolve(DbDataReader reader) { var dotnetTypeName = reader.GetFieldValue(2); - mapping = eventMappingForDotNetTypeName(dotnetTypeName, eventTypeName); + mapping = eventMappingForDotNetTypeName(dotnetTypeName, eventTypeName, readSequence(reader)); } // #4680: an upcaster mapping is the authoritative interpretation of the stored // event-type name (it was registered with that name as its SOURCE). Skip the @@ -390,7 +390,8 @@ public async Task ResolveAsync(DbDataReader reader, CancellationToken to { var dotnetTypeName = await reader.GetFieldValueAsync(2, token).ConfigureAwait(false); - mapping = eventMappingForDotNetTypeName(dotnetTypeName, eventTypeName); + mapping = eventMappingForDotNetTypeName(dotnetTypeName, eventTypeName, + await readSequenceAsync(reader, token).ConfigureAwait(false)); } // #4680: see the sync Resolve overload above -- upcaster mappings are authoritative // for their source event-type name and the dotnet_type alt-mapping swap would shadow @@ -435,18 +436,7 @@ public async Task ResolveAsync(DbDataReader reader, CancellationToken to } catch (Exception e) { - // #4515: mt_events.seq_id shifted from ordinal 3 to ordinal 4 after - // bdata's insertion (EventsTable.SelectColumns now pins data, type, - // mt_dotnet_type, bdata, seq_id at 0..4). - long sequence; - try - { - sequence = await reader.GetFieldValueAsync(4, token).ConfigureAwait(false); - } - catch - { - sequence = -1; - } + var sequence = await readSequenceAsync(reader, token).ConfigureAwait(false); throw new EventDeserializationFailureException(sequence, mapping, e); } @@ -459,11 +449,47 @@ public async Task ResolveAsync(DbDataReader reader, CancellationToken to public abstract Task ApplyReaderDataToEventAsync(DbDataReader reader, IEvent e, CancellationToken token); - private EventMapping eventMappingForDotNetTypeName(string dotnetTypeName, string eventTypeName) + /// + /// #4515: mt_events.seq_id shifted from ordinal 3 to ordinal 4 after bdata's insertion + /// (EventsTable.SelectColumns now pins data, type, mt_dotnet_type, bdata, seq_id at 0..4). Reading it + /// is best-effort — this runs on failure paths where the reader may not have the column at all — so a + /// miss degrades to rather than replacing the + /// real failure with an indexing error. + /// + private static long readSequence(DbDataReader reader) + { + try + { + return reader.GetFieldValue(4); + } + catch + { + return UnknownEventTypeException.UnknownSequence; + } + } + + private static async Task readSequenceAsync(DbDataReader reader, CancellationToken token) + { + try + { + return await reader.GetFieldValueAsync(4, token).ConfigureAwait(false); + } + catch + { + return UnknownEventTypeException.UnknownSequence; + } + } + + /// + /// #5048 / jasperfx#565: is threaded down from the row being read so the + /// resulting can name the event that paused the shard, not + /// just its unresolvable alias. + /// + private EventMapping eventMappingForDotNetTypeName(string dotnetTypeName, string eventTypeName, long sequence) { if (dotnetTypeName.IsEmpty()) { - throw new UnknownEventTypeException(eventTypeName); + throw new UnknownEventTypeException(eventTypeName, sequence); } Type type; @@ -473,7 +499,7 @@ private EventMapping eventMappingForDotNetTypeName(string dotnetTypeName, string } catch (ArgumentNullException) { - throw new UnknownEventTypeException(dotnetTypeName); + throw new UnknownEventTypeException(dotnetTypeName, sequence); } return Events.EventMappingFor(type); diff --git a/src/Marten/Events/Projections/NaturalKeyProjection.cs b/src/Marten/Events/Projections/NaturalKeyProjection.cs index 46c7ab3f3b..ddf57f32a2 100644 --- a/src/Marten/Events/Projections/NaturalKeyProjection.cs +++ b/src/Marten/Events/Projections/NaturalKeyProjection.cs @@ -48,7 +48,7 @@ public Task ApplyAsync(IDocumentOperations operations, IEnumerable { if (mapping.EventType.IsAssignableFrom(@event.Data.GetType())) { - var rawValue = mapping.Extractor(@event.Data); + var rawValue = mapping.Extractor(@event); var innerValue = _naturalKey.Unwrap(rawValue); if (innerValue != null) { @@ -81,7 +81,7 @@ internal void QueueUpsertsForEvents(IDocumentOperations operations, IEnumerable< { if (mapping.EventType.IsAssignableFrom(@event.Data.GetType())) { - var rawValue = mapping.Extractor(@event.Data); + var rawValue = mapping.Extractor(@event); var innerValue = _naturalKey.Unwrap(rawValue); if (innerValue != null) { diff --git a/src/Marten/Events/Schema/EventProgressionTable.cs b/src/Marten/Events/Schema/EventProgressionTable.cs index 4087f3fd64..ff1b28cb84 100644 --- a/src/Marten/Events/Schema/EventProgressionTable.cs +++ b/src/Marten/Events/Schema/EventProgressionTable.cs @@ -52,6 +52,17 @@ public EventProgressionTable(EventGraph eventGraph): base(new PostgresqlObjectNa AddColumn("running_on_node", "integer").AllowNulls(); AddColumn("warning_behind_threshold", "bigint").AllowNulls(); AddColumn("critical_behind_threshold", "bigint").AllowNulls(); + + // #5048 / jasperfx#565: the classified reason this shard is paused or stopped, so a consumer + // polling the database (CritterWatch when the publishing node is DOWN, which is exactly when + // it matters) sees the same reason an in-process ShardState observer does. The reason *text* + // needs no new column -- ShardFailure.Detail is precisely what pause_reason has always + // carried. failure_category stores the enum NAME, never the ordinal, so reordering + // ShardFailureCategory can never silently re-label persisted rows. + AddColumn("failure_category", "varchar(50)").AllowNulls(); + AddColumn("failure_event_sequence", "bigint").AllowNulls(); + AddColumn("failure_event_type", "varchar(500)").AllowNulls(); + AddColumn("failure_event_tenant_id", "varchar(500)").AllowNulls(); } PrimaryKeyName = "pk_mt_event_progression"; diff --git a/src/Marten/Exceptions/EventDeserializationFailureException.cs b/src/Marten/Exceptions/EventDeserializationFailureException.cs index 244cc20ce6..01a53f1fff 100644 --- a/src/Marten/Exceptions/EventDeserializationFailureException.cs +++ b/src/Marten/Exceptions/EventDeserializationFailureException.cs @@ -12,20 +12,53 @@ namespace Marten.Exceptions; /// Thrown if Marten encounters an exception while trying to deserialize /// or upcast a persisted event /// -public class EventDeserializationFailureException: MartenException +public class EventDeserializationFailureException: MartenException, IEventFailureContext { public EventDeserializationFailureException(long sequence, IEventType eventType, Exception innerException): base( $"Event deserialization error on sequence = {sequence} for event type {eventType.EventTypeName}" , innerException) { Sequence = sequence; + EventTypeName = eventType.EventTypeName; } public long Sequence { get; } + /// + /// The event store's type alias for the event whose body could not be read (e.g. trip_started). + /// #5048 / jasperfx#565: the constructor has always been handed the and used + /// it only to build the message string. Retaining the alias lets the daemon report the failing event + /// type on rather than leaving it buried in prose. + /// + public string? EventTypeName { get; } + + /// + /// #5048 / jasperfx#565: this exception declares its own failure category, so the daemon never has to + /// sniff exception type names to classify a paused shard. A body Marten could not deserialize or + /// upcast is — a serializer or data problem, + /// governed by SkipSerializationErrors. + /// + public ShardFailureCategory Category => ShardFailureCategory.EventSerialization; + + // Everything below is raised while reading an mt_events row, BEFORE there is an IEvent to inspect, + // so nothing but the sequence and the stored type alias is knowable here. IEventFailureContext makes + // every one of these nullable for exactly this case. + Guid? IEventFailureContext.EventId => null; + Guid? IEventFailureContext.StreamId => null; + string? IEventFailureContext.StreamKey => null; + string? IEventFailureContext.TenantId => null; + long? IEventFailureContext.Version => null; + internal DeadLetterEvent ToDeadLetterEvent(ShardName name) { return new DeadLetterEvent { + // #5048 / jasperfx#565: assign the id here rather than leaving it to document identity + // generation at write time, so the creating process knows the dead letter's id BEFORE the + // (background, retried) write lands and can correlate it with the ShardFailure it reported. + // Marten only generates an id when the value is empty, so pre-assigning changes nothing about + // how the row persists. Version 7 keeps ids time-ordered, matching what jasperfx's + // DeadLetterEvent constructor now does on the ApplyEventException path. + Id = Guid.CreateVersion7(), EventSequence = Sequence, ExceptionMessage = Message, ExceptionType = GetType().FullNameInCode(), diff --git a/src/Marten/Exceptions/UnknownEventTypeException.cs b/src/Marten/Exceptions/UnknownEventTypeException.cs index 3523da3440..21c7b6947c 100644 --- a/src/Marten/Exceptions/UnknownEventTypeException.cs +++ b/src/Marten/Exceptions/UnknownEventTypeException.cs @@ -1,18 +1,57 @@ +using System; using System.Runtime.Serialization; +using JasperFx.Events.Daemon; namespace Marten.Exceptions; -public class UnknownEventTypeException: MartenException +public class UnknownEventTypeException: MartenException, IEventFailureContext { + /// + /// The sequence reported when the throw site had no mt_events row in hand — e.g. resolving a + /// .NET type name outside the event read path. is + /// non-nullable by contract, and -1 is already how Marten's event read path spells "the sequence + /// could not be determined". + /// + public const long UnknownSequence = -1; + public string EventTypeName { get; } - public UnknownEventTypeException(string eventTypeName): base( + public UnknownEventTypeException(string eventTypeName): this(eventTypeName, UnknownSequence) + { + } + + /// + /// #5048 / jasperfx#565: carry the store-wide sequence of the offending mt_events row when the + /// throw site knows it, so a shard paused by an unregistered event type can name the event that + /// stopped it instead of only its alias. + /// + public UnknownEventTypeException(string eventTypeName, long sequence): base( $"Unknown event type name alias '{eventTypeName}.' You may need to register this event type through StoreOptions.Events.AddEventType(type)") { EventTypeName = eventTypeName; + Sequence = sequence; } protected UnknownEventTypeException(SerializationInfo info, StreamingContext context): base(info, context) { + EventTypeName = string.Empty; + Sequence = UnknownSequence; } + + public long Sequence { get; } + + /// + /// #5048 / jasperfx#565: kept deliberately distinct from + /// . An alias that resolves to no known .NET + /// type in this deployment is normally a missing registration or a rollback past the event type's + /// introduction — a deployment fix, not a data fix. + /// + public ShardFailureCategory Category => ShardFailureCategory.UnknownEventType; + + // The type never resolved, so no event was ever materialized to read these from. + Guid? IEventFailureContext.EventId => null; + Guid? IEventFailureContext.StreamId => null; + string? IEventFailureContext.StreamKey => null; + string? IEventFailureContext.TenantId => null; + long? IEventFailureContext.Version => null; } diff --git a/src/Marten/Storage/MartenDatabase.EventStorage.cs b/src/Marten/Storage/MartenDatabase.EventStorage.cs index cd21c090da..e32aa273a0 100644 --- a/src/Marten/Storage/MartenDatabase.EventStorage.cs +++ b/src/Marten/Storage/MartenDatabase.EventStorage.cs @@ -48,34 +48,14 @@ await conn.CreateCommand( /// /// Persist the extended progression telemetry (heartbeat / agent_status / pause_reason / - /// running_on_node) the async daemon publishes for a shard, via the - /// mt_mark_event_progression_extended function. Driven by the JasperFx.Events - /// ExtendedProgressionWriter observer on status transitions and throttled heartbeats - /// (CritterWatch #750). The function updates only the telemetry columns on an existing row, - /// so it never rolls back committed last_seq_id. + /// running_on_node, plus the #5048 classified failure columns) the async daemon publishes for a + /// shard. Driven by the JasperFx.Events ExtendedProgressionWriter observer on status + /// transitions and throttled heartbeats (CritterWatch #750). Update-only, so it never rolls back + /// committed last_seq_id. /// - public async Task WriteExtendedProgressionAsync(ShardState state, CancellationToken token = default) + public Task WriteExtendedProgressionAsync(ShardState state, CancellationToken token = default) { - await EnsureStorageExistsAsync(typeof(IEvent), token).ConfigureAwait(false); - - await using var conn = CreateConnection(); - try - { - await conn.OpenAsync(token).ConfigureAwait(false); - await conn.CreateCommand( - $"select {Options.EventGraph.DatabaseSchemaName}.mt_mark_event_progression_extended(:name, :seq, :heartbeat, :status, :reason, :node)") - .With("name", state.ShardName, NpgsqlDbType.Varchar) - .With("seq", state.Sequence, NpgsqlDbType.Bigint) - .With("heartbeat", (object?)state.LastHeartbeat ?? DBNull.Value, NpgsqlDbType.TimestampTz) - .With("status", (object?)state.AgentStatus ?? DBNull.Value, NpgsqlDbType.Varchar) - .With("reason", (object?)state.PauseReason ?? DBNull.Value, NpgsqlDbType.Text) - .With("node", (object?)state.RunningOnNode ?? DBNull.Value, NpgsqlDbType.Integer) - .ExecuteNonQueryAsync(token).ConfigureAwait(false); - } - finally - { - await conn.CloseAsync().ConfigureAwait(false); - } + return WriteExtendedProgressionAsync([state], token); } /// @@ -84,11 +64,20 @@ await conn.CreateCommand( /// shard's heartbeat on a database into one batch per flush interval and drives this overload, /// because the per-shard single-row write does not scale under per-tenant agent fan-out /// (agents = projections × tenants — jasperfx#553). Deliberately a plain UPDATE ... FROM unnest - /// join instead of a new batched database function, so no schema object is added and deployments - /// running AutoCreate.None pick the batching up without a migration. Semantics match - /// mt_mark_event_progression_extended exactly: update-only telemetry decoration of - /// existing progression rows — never INSERT, never touch last_seq_id / last_updated, - /// shards without a progression row yet are skipped silently. + /// join instead of a database function, so no schema object is added and deployments running + /// AutoCreate.None pick it up without a migration. Semantics match the + /// mt_mark_event_progression_extended function this replaced (the function is still + /// installed for anything calling it directly): update-only telemetry decoration of existing + /// progression rows — never INSERT, never touch last_seq_id / last_updated, shards + /// without a progression row yet are skipped silently. + /// + /// #5048 / jasperfx#565: the four failure_* columns follow a different rule from the rest. + /// They are written when the state carries a , CLEARED when a + /// arrives without one (a recovered shard must stop reporting + /// the reason it paused an hour ago), and otherwise LEFT ALONE. That last case is load-bearing: + /// the Stopped publication that follows a pause carries no failure, and an unconditional + /// write would erase the reason microseconds after recording it. + /// /// public async Task WriteExtendedProgressionAsync(IReadOnlyList states, CancellationToken token = default) { @@ -97,12 +86,6 @@ public async Task WriteExtendedProgressionAsync(IReadOnlyList states return; } - if (states.Count == 1) - { - await WriteExtendedProgressionAsync(states[0], token).ConfigureAwait(false); - return; - } - await EnsureStorageExistsAsync(typeof(IEvent), token).ConfigureAwait(false); var names = new string[states.Count]; @@ -110,14 +93,28 @@ public async Task WriteExtendedProgressionAsync(IReadOnlyList states var statuses = new string?[states.Count]; var reasons = new string?[states.Count]; var nodes = new int?[states.Count]; + var touchFailures = new bool[states.Count]; + var failureCategories = new string?[states.Count]; + var failureSequences = new long?[states.Count]; + var failureEventTypes = new string?[states.Count]; + var failureTenantIds = new string?[states.Count]; for (var i = 0; i < states.Count; i++) { - names[i] = states[i].ShardName; - heartbeats[i] = states[i].LastHeartbeat; - statuses[i] = states[i].AgentStatus; - reasons[i] = states[i].PauseReason; - nodes[i] = states[i].RunningOnNode; + var state = states[i]; + + names[i] = state.ShardName; + heartbeats[i] = state.LastHeartbeat; + statuses[i] = state.AgentStatus; + reasons[i] = state.PauseReason; + nodes[i] = state.RunningOnNode; + + var failure = state.Failure; + touchFailures[i] = failure != null || state.Action == ShardAction.Started; + failureCategories[i] = failure?.Category.ToString(); + failureSequences[i] = failure?.Event?.Sequence; + failureEventTypes[i] = failure?.Event?.EventTypeName; + failureTenantIds[i] = failure?.Event?.TenantId; } await using var conn = CreateConnection(); @@ -129,9 +126,15 @@ await conn.CreateCommand($""" set heartbeat = t.heartbeat, agent_status = t.agent_status, pause_reason = t.pause_reason, - running_on_node = t.running_on_node - from unnest(:names, :heartbeats, :statuses, :reasons, :nodes) - as t(name, heartbeat, agent_status, pause_reason, running_on_node) + running_on_node = t.running_on_node, + failure_category = case when t.touch_failure then t.failure_category else p.failure_category end, + failure_event_sequence = case when t.touch_failure then t.failure_event_sequence else p.failure_event_sequence end, + failure_event_type = case when t.touch_failure then t.failure_event_type else p.failure_event_type end, + failure_event_tenant_id = case when t.touch_failure then t.failure_event_tenant_id else p.failure_event_tenant_id end + from unnest(:names, :heartbeats, :statuses, :reasons, :nodes, :touch_failures, + :failure_categories, :failure_sequences, :failure_event_types, :failure_tenant_ids) + as t(name, heartbeat, agent_status, pause_reason, running_on_node, touch_failure, + failure_category, failure_event_sequence, failure_event_type, failure_event_tenant_id) where p.name = t.name """) .With("names", names, NpgsqlDbType.Array | NpgsqlDbType.Varchar) @@ -139,6 +142,11 @@ as t(name, heartbeat, agent_status, pause_reason, running_on_node) .With("statuses", statuses, NpgsqlDbType.Array | NpgsqlDbType.Varchar) .With("reasons", reasons, NpgsqlDbType.Array | NpgsqlDbType.Text) .With("nodes", nodes, NpgsqlDbType.Array | NpgsqlDbType.Integer) + .With("touch_failures", touchFailures, NpgsqlDbType.Array | NpgsqlDbType.Boolean) + .With("failure_categories", failureCategories, NpgsqlDbType.Array | NpgsqlDbType.Varchar) + .With("failure_sequences", failureSequences, NpgsqlDbType.Array | NpgsqlDbType.Bigint) + .With("failure_event_types", failureEventTypes, NpgsqlDbType.Array | NpgsqlDbType.Varchar) + .With("failure_tenant_ids", failureTenantIds, NpgsqlDbType.Array | NpgsqlDbType.Varchar) .ExecuteNonQueryAsync(token).ConfigureAwait(false); } finally