Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 1 addition & 1 deletion Directory.Build.props
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@
<Copyright>Copyright © Vulthil</Copyright>
<PackageLicenseExpression>MIT</PackageLicenseExpression>
<PackageProjectUrl>https://github.com/Vulthil/Vulthil.SharedKernel</PackageProjectUrl>
<PackageTags>shared-kernel;ddd;cqrs;result-pattern;dotnet</PackageTags>
<!-- PackageTags are set per project so each package's tags describe its own contents. -->
<RepositoryType>git</RepositoryType>
<RepositoryUrl>https://github.com/Vulthil/Vulthil.SharedKernel</RepositoryUrl>
</PropertyGroup>
Expand Down
2 changes: 1 addition & 1 deletion docs/articles/cqrs-pipeline.md
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ No additional wiring is needed – the validator is picked up automatically from

### Request Logging

`RequestLoggingPipelineBehavior` logs the start and completion of every request, attaching the serialised error to the logging scope when the result is a failure.
`RequestLoggingPipelineBehavior` logs the start and completion of requests whose response type is `Result` (or `Result<T>`), attaching the serialised error to the logging scope when the result is a failure. Requests with any other response type are not covered by this behavior.

### Transactional

Expand Down
15 changes: 12 additions & 3 deletions docs/articles/getting-started.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,17 +12,26 @@ The packages are organised into layers so you can adopt only what you need:
| Domain | `Vulthil.Results` | Railway-oriented `Result<T>` for explicit error handling |
| Application | `Vulthil.SharedKernel.Application` | Commands, queries, handlers, and pipeline behaviors |
| Infrastructure | `Vulthil.SharedKernel.Infrastructure` | EF Core base context, generic repository, and outbox processing |
| Infrastructure | `Vulthil.SharedKernel.Infrastructure.Relational` | Shared relational support and the default relational outbox strategy |
| Infrastructure | `Vulthil.SharedKernel.Outbox` | EF-free transactional outbox engine (relay, dispatchers, `IOutboxStore`) |
| Infrastructure | `Vulthil.SharedKernel.Outbox.EntityFrameworkCore` | EF Core implementation of the outbox engine |
| Infrastructure | `Vulthil.SharedKernel.Infrastructure.Relational` | Shared relational support and the relational outbox store base |
| Infrastructure | `Vulthil.SharedKernel.Infrastructure.Npgsql` | PostgreSQL provider integration (`UseNpgsql`) |
| Infrastructure | `Vulthil.SharedKernel.Infrastructure.MySql` | MySQL provider integration (`UseMySql`) |
| Infrastructure | `Vulthil.SharedKernel.Infrastructure.Cosmos` | Azure Cosmos DB provider integration (`UseCosmosDb`) |
| API | `Vulthil.SharedKernel.Api` | Minimal API endpoint conventions and `Result` → HTTP mapping |
| Hosting | `Vulthil.Extensions.Hosting` | `IRestartableHostedService` marker for cleanly pausable hosted services |
| Messaging | `Vulthil.Messaging.Abstractions` | Transport-agnostic consumer and publisher contracts |
| Messaging | `Vulthil.Messaging` | Queue/consumer registration and hosted service orchestration |
| Messaging | `Vulthil.Messaging` | Queue/consumer registration, hosted orchestration, and the transport SDK |
| Messaging | `Vulthil.Messaging.RabbitMq` | RabbitMQ transport implementation |
| Messaging | `Vulthil.Messaging.Outbox` | Transactional bus-publish outbox (captures publishes into the outbox) |
| Messaging | `Vulthil.Messaging.Inbox` | Idempotent-receiver consume filter for exactly/effectively-once processing |
| Messaging | `Vulthil.Messaging.Inbox.EntityFrameworkCore` | Shared EF Core inbox primitives |
| Messaging | `Vulthil.Messaging.Inbox.Relational` | Relational idempotency store (transactional exactly-once) |
| Messaging | `Vulthil.Messaging.Inbox.Cosmos` | Cosmos idempotency store (effectively-once) |
| Testing | `Vulthil.xUnit` | Base test classes, auto-mocking, and Testcontainers integration |
| Testing | `Vulthil.xUnit.Cosmos` | Cosmos DB emulator fixture for `Vulthil.xUnit` |
| Testing | `Vulthil.Messaging.TestHarness` | In-memory messaging test harness |
| Testing | `Vulthil.Extensions.Testing` | Shared assertion and setup helpers |
| Testing | `Vulthil.Extensions.Testing` | Framework-agnostic polling and HTTP response helpers (no xUnit dependency) |

## Minimal Example

Expand Down
2 changes: 1 addition & 1 deletion docs/articles/inbox-pattern.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ Message delivery is **at-least-once**: a broker can redeliver a message (after a

The inbox pattern makes consumers **idempotent**: it records which messages have already been processed and skips duplicates. `Vulthil.Messaging.Inbox` provides this as a consume filter, with the processed-marker written in the **same transaction** as the consumer's business changes — so on a relational store, processing is exactly-once.

This is an *idempotent receiver*, not a store-and-forward inbox: it rides on top of the transport's existing retry, fault, and dead-letter machinery rather than re-implementing them. Bounding a consumer that keeps failing is therefore **not** the guard's job — on RabbitMQ a poison delivery is retried up to the queue's `MaxRetryCount`, then a `Fault<T>` is published and the message is nacked to the dead-letter exchange. The filter deliberately ignores `IMessageContext.RetryCount` and adds no max-attempts of its own, keeping retry policy in one place (the transport).
This is an *idempotent receiver*, not a store-and-forward inbox: it rides on top of the transport's existing retry, fault, and dead-letter machinery rather than re-implementing them. Bounding a consumer that keeps failing is therefore **not** the guard's job — on RabbitMQ a persistently-failing delivery is retried per the consumer's [retry policy](messaging.md#retries) (its own registration's, falling back to the queue default), then a `Fault<T>` is published and the delivery is dead-lettered when a dead-letter queue is configured. The filter deliberately ignores `IMessageContext.RetryCount` and adds no max-attempts of its own, keeping retry policy in one place (the transport).

## How It Works

Expand Down
27 changes: 20 additions & 7 deletions docs/articles/messaging.md
Original file line number Diff line number Diff line change
Expand Up @@ -305,17 +305,19 @@ Or in code:
m.ConfigureMessagingOptions(opts => opts.ConsumeFilters.EnableLogging = false);
```

The filter stays registered in DI; only its behavior is skipped, so it's still
resolvable in unit tests if you want to assert against it.
The toggle is applied when `AddMessaging` composes the pipeline: a disabled filter is
never registered in DI at all (there is no per-delivery flag check), so flipping the
flag after registration has no effect.

### Idempotent receivers (inbox)

`Vulthil.Messaging.Inbox` ships a consume filter that deduplicates redelivered
messages, giving exactly-once processing on top of at-least-once delivery. It is a
transaction-owning filter: the consumer runs inside an `IIdempotencyStore`
transaction so the processed-marker and the consumer's writes commit atomically.
Opt a message type in with `AddIdempotentInbox<TMessage>()`. See the
[Inbox Pattern](inbox-pattern.md) article for the full design.
messages on top of at-least-once delivery. The consumer invocation is handed to an
`IIdempotencyStore`, which owns the unit: with a transactional (relational) store the
processed-marker and the consumer's writes commit atomically — exactly-once
processing — while a store without cross-partition transactions (e.g. Cosmos) is
effectively-once. Opt a message type in with `AddIdempotentInbox<TMessage>()`. See
the [Inbox Pattern](inbox-pattern.md) article for the full design.

## Ordered Processing (per-aggregate)

Expand Down Expand Up @@ -523,6 +525,17 @@ await publisher.PublishAsync(new OrderCreatedEvent(orderId), ctx =>
Fault publishing is best-effort: a failure to publish the fault is logged and never prevents the
original delivery from being settled (nacked for dead-lettering).

### Poison deliveries

Deliveries that fail **before** a consumer can run bypass the retry and fault machinery entirely, and
no `Fault<T>` is published for them:

- **Unknown message type** (no consumer/plan matches the delivery's type identity): the delivery is
**acked and dropped permanently**, with a warning log — it does not reach the dead-letter queue.
- **Undeserializable body** (malformed JSON, or a body that deserializes to `null`): the delivery is
**nacked without requeue**, with an error log. When the queue has a dead-letter queue configured
(`q.UseDeadLetterQueue()`), the broker moves it there; otherwise it is discarded.

## Routing Keys

Routing keys flow through two distinct configuration sites, one on each side of the wire:
Expand Down
23 changes: 20 additions & 3 deletions docs/articles/outbox-pattern.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,14 +4,19 @@ The transactional outbox pattern guarantees that domain events raised by aggrega

## How It Works

1. During `SaveChangesAsync`, a `SaveChangesInterceptor` serialises every pending domain event into an `OutboxMessage` row in the same database transaction.
1. During `SaveChanges`/`SaveChangesAsync` (both the synchronous and asynchronous paths are intercepted), a `SaveChangesInterceptor` serialises every pending domain event into an `OutboxMessage` row in the same database transaction.
2. The aggregate root's event collection is cleared.
3. A background service (`OutboxBackgroundService`) relays unprocessed outbox messages — woken immediately once the captured rows are durable (on transaction commit, or right after a non-transactional `SaveChanges` that captured domain events) for low latency, and polling on an interval as the backstop.
4. Each message is routed by its `OutboxDestination` to the registered `IOutboxDispatcher` that handles it (in-process domain events by default, or the broker — see below).
5. Successfully relayed messages are marked as processed; failures are retried up to the configured maximum, after which the message is dead-lettered — its `FailedOnUtc` timestamp is set, it is no longer relayed, and an error is logged with the last failure (the `Error` column).

This guarantees **at-least-once delivery** because the event and the business data are committed atomically.

Relay order is best-effort, not a contract: messages are fetched oldest-first (`OccurredOnUtc`, then `Id`), but a
failed message does not block the rest of its batch — the batch keeps dispatching, so later messages overtake the
failure, which is retried on a later cycle (no head-of-line blocking). With `EnableParallelPublishing`, order
within a batch is not preserved at all. Consumers that need ordering must not rely on the outbox for it.

## Configuration

### Enable outbox processing during DbContext registration
Expand Down Expand Up @@ -54,6 +59,11 @@ public sealed class AppDbContext(DbContextOptions<AppDbContext> options)

`BaseDbContext` owns the `OutboxMessages` `DbSet`; apply the provider-optimized mapping in `OnModelCreating` by calling your provider's extension — `ApplyNpgsqlOutbox()`, `ApplyMySqlOutbox()`, or `ApplyCosmosOutbox()` (as shown above). The agnostic `ApplyOutbox()` is available for custom providers.

Because the `DbSet` lives on the base class, EF Core's convention discovery puts the `OutboxMessage` entity into
**every** derived context's model — and its migrations — even when outbox processing is never enabled for that
context. A context that will never use the outbox can opt out with `modelBuilder.Ignore<OutboxMessage>()` in its
`OnModelCreating`; an outbox-enabled context must keep the mapping (the relay store refuses to run without it).

Only one outbox-enabled `DbContext` is supported per host: the relay and retention background services resolve a
single `IOutboxStore`, so calling `EnableOutboxProcessing()` for a second context throws an
`InvalidOperationException` at startup instead of silently leaving the first context's messages unrelayed. The
Expand All @@ -71,7 +81,7 @@ running unprotected.
| `EnableParallelPublishing` | `false` | Publish messages in parallel within a batch (each dispatch runs in its own DI scope) |
| `MaxDegreeOfParallelism` | 4 | Maximum concurrent dispatches when `EnableParallelPublishing` is enabled |
| `OutboxProcessingDelaySeconds` | 2 | Base polling delay between processing cycles |
| `MaxDelaySeconds` | 60 | Maximum back-off delay when no messages are found |
| `MaxDelaySeconds` | 60 | Maximum back-off delay when a cycle relays nothing (no pending messages, or every fetched message failed) |
| `EnableTracing` | `true` | Carry the originating trace identifier when publishing |

## Observability
Expand Down Expand Up @@ -105,7 +115,7 @@ Processed and dead-lettered rows remain in the `OutboxMessages` table after rela
| `SweepInterval` | 1 hour | Delay between sweeps |
| `BatchSize` | 1000 | Rows deleted per batch within a sweep |

The sweep deletes rows whose `ProcessedOnUtc` **or** `FailedOnUtc` is older than `RetentionPeriod`; **pending rows are never touched**. It runs through the registered `IOutboxStore` when it implements `IOutboxRetentionStore` (the EF Core store does) — every store deletes at most `BatchSize` rows per call (relational providers select the oldest eligible keys and delete them set-based with `ExecuteDelete`), so a large backlog is drained in bounded, short-lived batches, and the **same sweep covers Cosmos** (a Cosmos container TTL is not used, because it cannot tell a pending row from a relayed one and could expire an undelivered message).
The sweep deletes rows whose `ProcessedOnUtc` **or** `FailedOnUtc` is older than `RetentionPeriod`; **pending rows are never touched**. It runs through the registered `IOutboxStore` when it implements `IOutboxRetentionStore` (the EF Core store does; a custom store that doesn't gets a one-time warning and the sweep is skipped) — every store deletes at most `BatchSize` rows per call (relational providers select the oldest eligible keys and delete them set-based with `ExecuteDelete`), so a large backlog is drained in bounded, short-lived batches, and the **same sweep covers Cosmos** (a Cosmos container TTL is not used, because it cannot tell a pending row from a relayed one and could expire an undelivered message).

## Custom Outbox Store

Expand Down Expand Up @@ -156,6 +166,13 @@ Capture is relational-only (it enlists in the ambient transaction); the relay wo
the general publish/send **filter pipeline** (`IPublishFilter`, registered via `AddPublishFilter<T>()`), which is the
publish-side counterpart to consume filters and can host other cross-cutting concerns.

Two pipeline consequences to know: capturing **short-circuits** the pipeline, so publish filters registered after
`AddTransactionalOutbox()` never see a captured message; and the relay publishes captured messages straight through
the transport, **bypassing the publish-filter pipeline entirely** (nothing runs twice, but nothing else observes the
relayed publish either). Register any filter that must observe every outgoing message *before*
`AddTransactionalOutbox()`. Request/reply is unaffected — requests never flow through the publish filters and are
never captured (a request awaiting a synchronous reply cannot be deferred to a relay).

The capture preserves the publish faithfully: the message id, correlation id, routing key, send destination, custom
headers, and the current trace context (`TraceParent`/`TraceState`, when tracing is enabled) are stored with the row
and re-applied on relay — the relayed publish runs under an activity linked to the capturing trace. Headers persist
Expand Down
20 changes: 16 additions & 4 deletions docs/articles/packages/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,9 @@ Focused usage guidance for each package in `src`. Pick the page that matches the
## Infrastructure

- [Vulthil.SharedKernel.Infrastructure](vulthil-sharedkernel-infrastructure.md) – `BaseDbContext`, generic repository, and outbox composition entry point.
- [Vulthil.SharedKernel.Infrastructure.Relational](vulthil-sharedkernel-infrastructure-relational.md) – relational outbox strategy and `MigrateAsync` for EF Core providers.
- [Vulthil.SharedKernel.Outbox](vulthil-sharedkernel-outbox.md) – the EF-free transactional outbox engine (relay, dispatchers, `IOutboxStore` seam).
- [Vulthil.SharedKernel.Outbox.EntityFrameworkCore](vulthil-sharedkernel-outbox-entityframeworkcore.md) – the EF Core implementation of the outbox engine.
- [Vulthil.SharedKernel.Infrastructure.Relational](vulthil-sharedkernel-infrastructure-relational.md) – relational outbox store base and `MigrateAsync` for EF Core providers.
- [Vulthil.SharedKernel.Infrastructure.Npgsql](vulthil-sharedkernel-infrastructure-npgsql.md) – `UseNpgsql` provider wiring with PostgreSQL-tuned outbox.
- [Vulthil.SharedKernel.Infrastructure.MySql](vulthil-sharedkernel-infrastructure-mysql.md) – `UseMySql` provider wiring with MySQL-tuned outbox.
- [Vulthil.SharedKernel.Infrastructure.Cosmos](vulthil-sharedkernel-infrastructure-cosmos.md) – `UseCosmosDb` provider wiring with Cosmos-specific outbox.
Expand All @@ -23,14 +25,24 @@ Focused usage guidance for each package in `src`. Pick the page that matches the

- [Vulthil.SharedKernel.Api](vulthil-sharedkernel-api.md) – minimal API endpoint conventions and `Result` → HTTP mapping.

## Hosting

- [Vulthil.Extensions.Hosting](vulthil-extensions-hosting.md) – `IRestartableHostedService`, the marker for hosted services that can be paused and resumed cleanly.

## Messaging

- [Vulthil.Messaging.Abstractions](vulthil-messaging-abstractions.md) – transport-agnostic consumer and publisher contracts.
- [Vulthil.Messaging](vulthil-messaging.md) – queue/consumer registration and hosted processing.
- [Vulthil.Messaging](vulthil-messaging.md) – queue/consumer registration, hosted processing, and the transport-author SDK.
- [Vulthil.Messaging.RabbitMq](vulthil-messaging-rabbitmq.md) – RabbitMQ transport implementation.
- [Vulthil.Messaging.Outbox](vulthil-messaging-outbox.md) – transactional bus-publish outbox bridging the outbox engine to the broker.
- [Vulthil.Messaging.Inbox](vulthil-messaging-inbox.md) – idempotent-receiver consume filter (`IIdempotencyStore` contract).
- [Vulthil.Messaging.Inbox.EntityFrameworkCore](vulthil-messaging-inbox-entityframeworkcore.md) – shared EF Core inbox primitives (`InboxMessage`, `ISaveInboxMessages`).
- [Vulthil.Messaging.Inbox.Relational](vulthil-messaging-inbox-relational.md) – relational idempotency store (transactional exactly-once).
- [Vulthil.Messaging.Inbox.Cosmos](vulthil-messaging-inbox-cosmos.md) – Cosmos idempotency store (effectively-once).

## Testing

- [Vulthil.xUnit](vulthil-xunit.md) – reusable xUnit base classes and auto-mocking.
- [Vulthil.xUnit](vulthil-xunit.md) – reusable xUnit base classes, auto-mocking, containers, and HTTP mocks.
- [Vulthil.xUnit.Cosmos](vulthil-xunit-cosmos.md) – Cosmos DB emulator fixture for `Vulthil.xUnit`.
- [Vulthil.Messaging.TestHarness](vulthil-messaging-testharness.md) – in-memory messaging test harness.
- [Vulthil.Extensions.Testing](vulthil-extensions-testing.md) – shared assertion and setup helpers.
- [Vulthil.Extensions.Testing](vulthil-extensions-testing.md) – framework-agnostic polling and HTTP response helpers.
2 changes: 2 additions & 0 deletions docs/articles/packages/toc.yml
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
href: vulthil-sharedkernel-infrastructure-cosmos.md
- name: Vulthil.SharedKernel.Api
href: vulthil-sharedkernel-api.md
- name: Vulthil.Extensions.Hosting
href: vulthil-extensions-hosting.md
- name: Vulthil.Messaging.Abstractions
href: vulthil-messaging-abstractions.md
- name: Vulthil.Messaging
Expand Down
Loading