Skip to content

Polecat support and DCB workflows with Marten and Polecat #2271

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jeremydmiller merged 1 commit into from
Mar 9, 2026
Merged

Polecat support and DCB workflows with Marten and Polecat #2271

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 12 additions & 11 deletions Directory.Packages.props
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,14 @@
<PackageVersion Include="Grpc.Core" Version="2.46.6" />
<PackageVersion Include="Grpc.Tools" Version="2.72.0" />
<PackageVersion Include="HtmlTags" Version="9.0.0" />
<PackageVersion Include="JasperFx" Version="1.20.0" />
<PackageVersion Include="JasperFx.Events" Version="1.22.0" />
<PackageVersion Include="JasperFx" Version="1.21.0" />
<PackageVersion Include="JasperFx.Events" Version="1.23.1" />
<PackageVersion Include="JasperFx.RuntimeCompiler" Version="4.4.0" />
<PackageVersion Include="Lamar.Microsoft.DependencyInjection" Version="15.0.1" />
<PackageVersion Include="Marten" Version="8.22.2" />
<PackageVersion Include="Marten" Version="8.23.0" />
<PackageVersion Include="Polecat" Version="0.9.0" />
<PackageVersion Include="Microsoft.Azure.Cosmos" Version="3.46.1" />
<PackageVersion Include="Marten.AspNetCore" Version="8.22.2" />
<PackageVersion Include="Marten.AspNetCore" Version="8.23.0" />
<PackageVersion Include="MemoryPack" Version="1.21.3" />
<PackageVersion Include="MessagePack" Version="3.1.3" />
<PackageVersion Include="Meziantou.Extensions.Logging.Xunit" Version="1.0.15" />
Expand Down Expand Up @@ -79,13 +80,13 @@
<PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="9.0.5" />
<PackageVersion Include="System.Net.NameResolution" Version="4.3.0" />
<PackageVersion Include="System.Threading.Tasks.Dataflow" Version="9.0.5" />
<PackageVersion Include="Weasel.Core" Version="8.8.1" />
<PackageVersion Include="Weasel.EntityFrameworkCore" Version="8.8.1" />
<PackageVersion Include="Weasel.MySql" Version="8.8.1" />
<PackageVersion Include="Weasel.Oracle" Version="8.8.1" />
<PackageVersion Include="Weasel.Postgresql" Version="8.8.1" />
<PackageVersion Include="Weasel.SqlServer" Version="8.8.1" />
<PackageVersion Include="Weasel.Sqlite" Version="8.8.1" />
<PackageVersion Include="Weasel.Core" Version="8.9.0" />
<PackageVersion Include="Weasel.EntityFrameworkCore" Version="8.9.0" />
<PackageVersion Include="Weasel.MySql" Version="8.9.0" />
<PackageVersion Include="Weasel.Oracle" Version="8.9.0" />
<PackageVersion Include="Weasel.Postgresql" Version="8.9.0" />
<PackageVersion Include="Weasel.SqlServer" Version="8.9.0" />
<PackageVersion Include="Weasel.Sqlite" Version="8.9.0" />
<PackageVersion Include="xunit" Version="2.9.3" />
<PackageVersion Include="xunit.assemblyfixture" Version="2.2.0" />
<PackageVersion Include="xunit.runner.visualstudio" Version="2.8.2" />
Expand Down
22 changes: 11 additions & 11 deletions build/build.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,7 +212,7 @@ class Build : NukeBuild
.Executes(() =>
{
DotNetTest(c => c
.SetProjectFile(Solution.Persistence.SqliteTests)
.SetProjectFile(Solution.Persistence.Sqlite.SqliteTests)
.SetConfiguration(Configuration)
.EnableNoBuild()
.EnableNoRestore()
Expand Down Expand Up @@ -329,13 +329,13 @@ class Build : NukeBuild
Solution.Transports.Pulsar.Wolverine_Pulsar,
Solution.Transports.GCP.Wolverine_Pubsub,
Solution.Persistence.Wolverine_RDBMS,
Solution.Persistence.Wolverine_Postgresql,
Solution.Persistence.Wolverine_Marten,
Solution.Persistence.Wolverine_RavenDb,
Solution.Persistence.Wolverine_SqlServer,
Solution.Persistence.Wolverine_MySql,
Solution.Persistence.Wolverine_Oracle,
Solution.Persistence.Wolverine_Sqlite,
Solution.Persistence.PostgreSQL.Wolverine_Postgresql,
Solution.Persistence.Marten.Wolverine_Marten,
Solution.Persistence.RavenDb.Wolverine_RavenDb,
Solution.Persistence.SqlServer.Wolverine_SqlServer,
Solution.Persistence.MySql.Wolverine_MySql,
Solution.Persistence.Oracle.Wolverine_Oracle,
Solution.Persistence.Sqlite.Wolverine_Sqlite,
Solution.Persistence.CosmosDb.Wolverine_CosmosDb,
Solution.Extensions.Wolverine_FluentValidation,
Solution.Extensions.Wolverine_MemoryPack,
Expand Down Expand Up @@ -478,10 +478,10 @@ private IEnumerable<NugetToProjectReference> nugetReferences()
{
yield return new(Solution.Wolverine, ["JasperFx", "JasperFx.RuntimeCompiler", "JasperFx.Events"]);

yield return new(Solution.Persistence.Wolverine_Postgresql, ["Weasel.Postgresql"]);
yield return new(Solution.Persistence.PostgreSQL.Wolverine_Postgresql, ["Weasel.Postgresql"]);
yield return new(Solution.Persistence.Wolverine_RDBMS, ["Weasel.Core"]);
yield return new(Solution.Persistence.Wolverine_SqlServer, ["Weasel.SqlServer"]);
yield return new(Solution.Persistence.Wolverine_Marten, ["Marten"]);
yield return new(Solution.Persistence.SqlServer.Wolverine_SqlServer, ["Weasel.SqlServer"]);
yield return new(Solution.Persistence.Marten.Wolverine_Marten, ["Marten"]);
}

Target Attach => _ => _.Executes(() =>
Expand Down
14 changes: 14 additions & 0 deletions docs/.vitepress/config.mts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ const config: UserConfig<DefaultTheme.Config> = {
{text: 'Vertical Slice Architecture', link: '/tutorials/vertical-slice-architecture'},
{text: 'Modular Monoliths', link: '/tutorials/modular-monolith'},
{text: 'Event Sourcing and CQRS with Marten', link: '/tutorials/cqrs-with-marten'},
{text: 'Event Sourcing and CQRS with Polecat', link: '/tutorials/cqrs-with-polecat'},
{text: 'Railway Programming with Wolverine', link: '/tutorials/railway-programming'},
{text: 'Interoperability with Non-Wolverine Systems', link: '/tutorials/interop'},
{text: 'Leader Election and Agents', link: '/tutorials/leader-election'},
Expand Down Expand Up @@ -232,6 +233,7 @@ const config: UserConfig<DefaultTheme.Config> = {
{text: 'Uploading Files', link: '/guide/http/files'},
{text: 'Integration with Sagas', link: '/guide/http/sagas'},
{text: 'Integration with Marten', link: '/guide/http/marten'},
{text: 'Integration with Polecat', link: '/guide/http/polecat'},
{text: 'Validation', link: '/guide/http/validation'},
{text: 'Fluent Validation', link: '/guide/http/fluentvalidation'},
{text: 'Problem Details', link: '/guide/http/problemdetails'},
Expand Down Expand Up @@ -259,6 +261,18 @@ const config: UserConfig<DefaultTheme.Config> = {
{text: 'Multi-Tenancy and Marten', link: '/guide/durability/marten/multi-tenancy'},
{text: 'Ancillary Marten Stores', link: '/guide/durability/marten/ancillary-stores'},
]},
{text: 'Polecat Integration', link: '/guide/durability/polecat/', collapsed: true, items: [
{text: 'Transactional Middleware', link: '/guide/durability/polecat/transactional-middleware'},
{text: 'Transactional Outbox Support', link: '/guide/durability/polecat/outbox'},
{text: 'Transactional Inbox Support', link: '/guide/durability/polecat/inbox'},
{text: 'Operation Side Effects', link: '/guide/durability/polecat/operations'},
{text: 'Aggregate Handlers and Event Sourcing', link: '/guide/durability/polecat/event-sourcing'},
{text: 'Event Forwarding to Wolverine', link: '/guide/durability/polecat/event-forwarding'},
{text: 'Event Subscriptions', link: '/guide/durability/polecat/subscriptions'},
{text: 'Subscription/Projection Distribution', link: '/guide/durability/polecat/distribution'},
{text: 'Sagas', link: '/guide/durability/polecat/sagas'},
{text: 'Multi-Tenancy and Polecat', link: '/guide/durability/polecat/multi-tenancy'},
]},
{text: 'Sql Server Integration', link: '/guide/durability/sqlserver'},
{text: 'PostgreSQL Integration', link: '/guide/durability/postgresql'},
{text: 'MySQL Integration', link: '/guide/durability/mysql'},
Expand Down
227 changes: 226 additions & 1 deletion docs/guide/durability/marten/event-sourcing.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -905,6 +905,202 @@ public class when_transfering_money
<sup><a href='https://github.com/JasperFx/wolverine/blob/main/src/Http/Wolverine.Http.Tests/Marten/working_against_multiple_streams.cs#L163-L190' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_when_transfering_money' title='Start of snippet'>anchor</a></sup>
<!-- endSnippet -->

### Finer-Grained Optimistic Concurrency in Multi-Stream Operations <Badge type="tip" text="5.17" />

When a handler uses multiple `[WriteAggregate]` parameters, Wolverine automatically applies version discovery only
to the **first** aggregate parameter. Secondary aggregate parameters will **not** automatically look for a `Version`
variable, preventing them from accidentally sharing the same version source.

To opt a secondary stream into optimistic concurrency checking, use the `VersionSource` property to explicitly point
it at a different member:

```cs
public record TransferMoney(Guid FromId, Guid ToId, decimal Amount,
long FromVersion, long ToVersion);

public static class TransferMoneyHandler
{
public static void Handle(
TransferMoney command,

// First parameter: discovers "Version" automatically, or use
// VersionSource for an explicit member name
[WriteAggregate(nameof(TransferMoney.FromId),
VersionSource = nameof(TransferMoney.FromVersion))]
IEventStream<Account> fromAccount,

// Secondary parameter: only gets version checking if VersionSource is set
[WriteAggregate(nameof(TransferMoney.ToId),
VersionSource = nameof(TransferMoney.ToVersion))]
IEventStream<Account> toAccount)
{
if (fromAccount.Aggregate.Amount >= command.Amount)
{
fromAccount.AppendOne(new Withdrawn(command.Amount));
toAccount.AppendOne(new Debited(command.Amount));
}
}
}
```

You can also use `AlwaysEnforceConsistency` on individual streams within a multi-stream operation to ensure a
concurrency check even when no events are appended to that stream:

```cs
public static void Handle(
TransferMoney command,
[WriteAggregate(nameof(TransferMoney.FromId),
AlwaysEnforceConsistency = true)]
IEventStream<Account> fromAccount,
[WriteAggregate(nameof(TransferMoney.ToId))]
IEventStream<Account> toAccount)
{
// Even if insufficient funds cause no events to be appended
// to fromAccount, Marten will still verify its version hasn't changed
if (fromAccount.Aggregate.Amount >= command.Amount)
{
fromAccount.AppendOne(new Withdrawn(command.Amount));
toAccount.AppendOne(new Debited(command.Amount));
}
}
```

## Enforcing Consistency Without New Events <Badge type="tip" text="5.17" />

In some cases, your command handler may decide not to emit any new events after evaluating the current aggregate state.
By default, Marten will silently succeed in this case without checking whether the stream has been modified since it was fetched.
This can be problematic for cross-stream operations where you need to guarantee that all referenced aggregates are still in the
state you expect at commit time.

The `AlwaysEnforceConsistency` option tells Marten to perform an optimistic concurrency check on the stream even if no events
are appended. If another session has written to the stream between your `FetchForWriting()` and `SaveChangesAsync()`, Marten
will throw a `ConcurrencyException`.

### Using the property on `[AggregateHandler]`

You can set `AlwaysEnforceConsistency = true` on the `[AggregateHandler]` attribute:

```cs
[AggregateHandler(AlwaysEnforceConsistency = true)]
public static class MyAggregateHandler
{
public static void Handle(DoSomething command, IEventStream<MyAggregate> stream)
{
// Even if no events are appended, Marten will verify
// the stream version hasn't changed since it was fetched
}
}
```

### Using `[ConsistentAggregateHandler]`

For convenience, there is a `[ConsistentAggregateHandler]` attribute that automatically sets `AlwaysEnforceConsistency = true`:

```cs
[ConsistentAggregateHandler]
public static class MyAggregateHandler
{
public static void Handle(DoSomething command, IEventStream<MyAggregate> stream)
{
// AlwaysEnforceConsistency is automatically true
}
}
```

### Parameter-level usage with `[ConsistentAggregate]`

When using parameter-level attributes (the `[Aggregate]` pattern), you can use `[ConsistentAggregate]` instead:

```cs
public static class MyHandler
{
public static void Handle(DoSomething command,
[ConsistentAggregate] IEventStream<MyAggregate> stream)
{
// AlwaysEnforceConsistency is automatically true
}
}
```

Or set the property directly on `[Aggregate]`:

```cs
public static class MyHandler
{
public static void Handle(DoSomething command,
[Aggregate(AlwaysEnforceConsistency = true)] IEventStream<MyAggregate> stream)
{
// Explicitly opt into consistency enforcement
}
}
```

## Overriding Version Discovery <Badge type="tip" text="5.17" />

By default, Wolverine discovers a version member on your command type by looking for a property or field named `Version`
of type `int` or `long`. In multi-stream operations where each stream needs its own version source, this convention
breaks down because you can't have multiple properties all named "Version".

The `VersionSource` property lets you explicitly specify which member supplies the expected stream version for
optimistic concurrency checks.

### On `[AggregateHandler]`

```cs
public record TransferMoney(Guid FromId, Guid ToId, decimal Amount, long FromVersion);

[AggregateHandler(VersionSource = nameof(TransferMoney.FromVersion))]
public static class TransferMoneyHandler
{
public static IEnumerable<object> Handle(TransferMoney command, Account account)
{
// FromVersion will be checked against the "from" account's stream version
yield return new Withdrawn(command.Amount);
}
}
```

### On `[WriteAggregate]` / `[Aggregate]`

This is particularly useful for multi-stream operations where each stream needs independent version tracking:

```cs
public record TransferMoney(Guid FromId, Guid ToId, decimal Amount,
long FromVersion, long ToVersion);

public static class TransferMoneyHandler
{
public static void Handle(
TransferMoney command,
[WriteAggregate(nameof(TransferMoney.FromId),
VersionSource = nameof(TransferMoney.FromVersion))]
IEventStream<Account> fromAccount,
[WriteAggregate(nameof(TransferMoney.ToId),
VersionSource = nameof(TransferMoney.ToVersion))]
IEventStream<Account> toAccount)
{
if (fromAccount.Aggregate.Amount >= command.Amount)
{
fromAccount.AppendOne(new Withdrawn(command.Amount));
toAccount.AppendOne(new Debited(command.Amount));
}
}
}
```

For HTTP endpoints, `VersionSource` can resolve from route arguments, query string parameters, or request body members:

```cs
[WolverinePost("/orders/{orderId}/ship/{expectedVersion}")]
[EmptyResponse]
public static OrderShipped Ship(
ShipOrder command,
[Aggregate(VersionSource = "expectedVersion")] Order order)
{
return new OrderShipped();
}
```

## Strong Typed Identifiers <Badge type="tip" text="5.0" />

If you're so inclined, you can use strong typed identifiers from tools like [Vogen](https://github.com/SteveDunn/Vogen) and [StronglyTypedId](https://github.com/andrewlock/StronglyTypedId)
Expand Down Expand Up @@ -1024,4 +1220,33 @@ public static StrongLetterAggregate Handle(


tools do this for you, and value types generated by these tools are
legal route argument variables for Wolverine.HTTP now.
legal route argument variables for Wolverine.HTTP now.

## Natural Keys

Marten supports [natural keys](/events/natural-keys) on aggregates, allowing you to look up event streams by a domain-meaningful identifier (like an order number) instead of the internal stream id. Wolverine's aggregate handler workflow fully supports natural keys, letting you route commands to the correct aggregate using a business identifier.

### Defining the Aggregate with a Natural Key

First, define your aggregate with a `[NaturalKey]` property and mark the methods that set the key with `[NaturalKeySource]`:

<!-- snippet: sample_wolverine_marten_natural_key_aggregate -->
<!-- endSnippet -->

### Using Natural Keys in Command Handlers

When your command carries the natural key value instead of a stream id, Wolverine can resolve it automatically. The command property should match the aggregate's natural key type:

<!-- snippet: sample_wolverine_marten_natural_key_commands -->
<!-- endSnippet -->

Wolverine uses the natural key type on the command property to call `FetchForWriting<TAggregate, TNaturalKey>()` under the covers, resolving the stream by the natural key in a single database round-trip.

### Handler Examples

Here are the handlers that process those commands, using `[WriteAggregate]` and `IEventStream<T>`:

<!-- snippet: sample_wolverine_marten_natural_key_handlers -->
<!-- endSnippet -->

For more details on how natural keys work at the Marten level, see the [Marten natural keys documentation](https://martendb.io/events/natural-keys).
Loading
Loading