Skip to content

Distributed queues and cluster events through Mediator extension points - #335

Draft
ejsmith wants to merge 63 commits into
mainfrom
codex/core-distributed-alternative
Draft

ejsmith wants to merge 63 commits into
mainfrom
codex/core-distributed-alternative

Conversation

@ejsmith

@ejsmith ejsmith commented Sep 8, 2026

Copy link
Copy Markdown
Contributor

Adds queued handlers and cluster notifications using Foundatio #533. The existing Mediator runtime, source generators, and core tests match main exactly. This is an alternative to #149 built through existing extension points.

Developer experience

Configure the infrastructure once:

var foundatio = builder.Services.AddFoundatio();
foundatio.Messaging.UseAws();
foundatio.Jobs.UseRedis();
foundatio.Locking.UseRedis();
builder.Services.AddMediator().AddDistributedQueues();

Add [Queue] to an ordinary handler:

[Queue(TrackProgress = true, Concurrency = 4, MaxAttempts = 3)]
public class ExportReportHandler
{
    public async Task<Result> HandleAsync(ExportReport message,
        IReportService reports, MessageProcessingContext context, CancellationToken ct)
    {
        await reports.ExportAsync(message.ReportId, ct);
        await context.ReportProgressAsync(100, "Report ready", ct);
        return Result.Ok();
    }
}
  • Enqueue: mediator.EnqueueAsync(new ExportReport("monthly"), ct) returns Result<QueueReceipt> confirming broker acceptance.
  • Validate and track: ordinary middleware runs before enqueue; native Foundatio APIs provide progress and cancellation. [Queue] configures retries/concurrency; [QueueLock] adds resource locking.
  • Broadcast: AddDistributedNotifications(o => o.Include<OrderChanged>()), then mediator.PublishAsync(...). Typed subscriptions exclude unrelated local traffic; overlapping selections publish once. No local handler is required.
  • Operate and scale: select workers by queue/group, inspect/replay/delete dead letters, and follow jobs/live events in the console and Clean Architecture samples. Processing jobs appear first in the dashboard.

Performance

Median jobs/sec, system .NET 10.0.12 Release, concurrency 64:

Scenario #149 This PR
In memory 181,195 100,346
In memory, tracked 36,658 34,111
SQS / LocalStack 2,788 2,968
In memory + Redis tracking 10,619 7,600
SQS / LocalStack + Redis tracking 2,702 2,582

This pass improves untracked memory throughput 5–8% with about 5% fewer allocations. Longer tracked-memory runs were level; Redis was 4% slower, with 5% less CPU. #149 remains leaner in memory; LocalStack is not production AWS capacity. Method, layer isolation, latency and raw results.

Additional details

  • Rebased on main. One integration package; no distributed changes to Mediator core. Foundatio is pinned to 56a88f0a on #533; publishing waits for those APIs to ship.
  • Queues are at least once; cluster broadcasts are best effort. Receipts confirm acceptance, not completed business work.
  • Release builds now preserve native dependency configuration; CI rejects unoptimized benchmark assemblies.
  • Passed: 756 .NET tests, 23 browser scenarios, console workflow and docs. 7.52M comparison jobs had no missing/duplicate deliveries. A ten-minute 69,354-job run verified queued/running cancellation, forced crash recovery and graceful worker restart, with no pending or failed jobs.

@ejsmith ejsmith changed the title Alternative distributed Mediator using native Foundatio messaging Distributed queues and cluster events through Mediator extension points Sep 8, 2026
ejsmith and others added 27 commits September 12, 2026 17:46
New packages:
- Foundatio.Mediator.Distributed: IQueueClient / IPubSubClient /
  IQueueJobStateStore abstractions, [Queue] attribute + QueueMiddleware
  routing, QueueWorker hosted service, DistributedNotificationWorker,
  QueueContext (ack/abandon/renew/progress), QueueRetryDelay, in-memory
  implementations, worker registry and queue dashboard handlers.
- Foundatio.Mediator.Distributed.Aws: SqsQueueClient and SNS/SQS pub/sub.
- Foundatio.Mediator.Distributed.Redis: RedisQueueJobStateStore.

Sample: CleanArchitectureSample gains Aspire AppHost + ServiceDefaults,
Redis job state, queue dashboard UI, SSE live events, worker/api role
flags. Docs: four distributed guide pages.

Squashed from the 29 commits on queues (previous head fa998b4)
and rebased onto main daaab7a. Conflict resolutions: took main's
dependency versions everywhere; kept main's RequestDurationFilter and
"Orders"/"Products" endpoint groups; folded main's rate limiter into the
API-only block of Program.cs; replaced the VitePress nav entries with
Lume front matter on the new guide pages; merged the landing-page tile
changes into main's reformatted index.md.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Foundatio.Xunit.v3 13.0.4 (was a branch preview build), xunit.v3.mtp-v2
4.0.0, xunit.runner.visualstudio 4.0.0, Microsoft.NET.Test.Sdk 18.9.0,
Microsoft.Extensions.DependencyInjection 10.0.11, GitHubActionsTestLogger
3.0.5, matching tests/Foundatio.Mediator.Tests.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Handlers that share a queue now share one message: the designated handler's
middleware enqueues once and the worker dispatches to every registration whose
message type accepts the concrete type carried in fm-message-type, so handlers
declared on an interface receive concrete events. Conflicting [Queue] settings
on a shared queue fail at registration.

Also in this change:
- QueueMiddleware is a singleton, returns without running when the message is
  the inbound notification being re-published, waits for infrastructure once
  the host has started, writes a correlation id, runs IQueueHeaderProvider
  hooks, and starts a producer span.
- QueueWorker links consumer spans to the producer, resolves types through the
  allowlist, dead-letters undeserializable messages immediately, survives
  exceptions in the consumer loop, keeps renewing after a failed renewal,
  heartbeats the job state store, drains for ShutdownTimeout on stop, and never
  acknowledges finished work with the stopping token. RateLimited is retryable.
- IQueueClient.ReceiveAsync takes the visibility timeout; SQS applies it.
  QueueDefinition carries visibility, retention, and max attempts. Dead-letter
  receive and replay have default implementations.
- InMemoryQueueClient models leases: received messages are invisible until
  completed, abandoned, or expired, and delayed abandons are scheduled.
- QueueRetryPolicy.Schedule with RetryDelays = "5s,1m,15m,30m".
- DistributedNotificationOptions.IncludeAssignableTo/Exclude; the worker logs
  the resolved types and retries a failed subscribe.
- QueueJobState.Metadata via DistributedQueueOptions.JobMetadataProvider,
  IQueueJobStateStore.HeartbeatAsync.
- Meter Foundatio.Mediator.Distributed with counters, duration, in-flight and
  polled depth gauges.
- AddDistributedQueues fails fast when workers are disabled or filtered and no
  transport is registered.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
DistributedQueueOptions.Workers replaces WorkersEnabled, Group and Queues.
WorkerSelection.Parse("all" | "none" | "exports,imports,!ProcessGridExport")
makes the same build an API node, a full worker, or a chosen set of groups or
queues from a config value, and the sample reads it from --workers or
Distributed:Workers.

[QueueLock] serializes a queued handler on an IQueueLockProvider keyed by a
fixed key, the message's IHaveLockKey, or queue plus message id; a held lock
completes the message without running it. InMemoryQueueLockProvider is
registered automatically only with the in-memory queue client.

Tests cover shared-queue fan-out with one message, interface-typed dispatch,
one execution per inbound notification across two nodes, poison and unknown
type dead-lettering, header providers, job metadata, the enqueue-only guard,
worker selection, retry schedules, renewal after a failed renew, and drain on
stop. The in-memory client test for delayed abandon now asserts lease
semantics.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Every job write is now one conditional MULTI/EXEC keyed on the current Status field (retried with jitter on conflict, bounded), so a job is a member of exactly one status set even under concurrent updates. Status indexes are trimmed by creation time on write and dangling members are removed when a listing finds their hash gone; index TTLs are only ever extended so a short-lived write cannot expire an index that still holds live members. Non-terminal writes are floored at the new NonTerminalExpiry option (7 days) so queued and processing jobs do not vanish mid-flight, and cancellation keys share the job's TTL. Metadata is persisted as meta:* hash fields, HeartbeatAsync updates LastUpdatedUtc/LastHeartbeatUtc without touching Status, Attempt is now persisted by SetJobStateAsync, and the Redis 7-only EXPIRE NX on counter buckets is replaced with an unconditional EXPIRE in the same transaction.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…nters

Adds a 50-way concurrent status update test, expired-member listing and count tests, write-side trim coverage, real TTL assertions through the multiplexer (replacing the placeholder GetHashCode key lookup), per-field UpdateJobStatusAsync/UpdateJobProgressAsync coverage, metadata round-trips, heartbeat behavior, cancellation TTL parity, and counter bucketing across an hour boundary via a FakeTimeProvider.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
EnsureQueuesAsync now creates (or validates) each queue with the
VisibilityTimeout and MessageRetentionPeriod from its QueueDefinition,
creates the dead-letter queue, and sets a RedrivePolicy whose
maxReceiveCount sits above the worker's MaxAttempts so the library's own
dead-lettering runs first. Definitions SQS cannot honor (visibility over
12 hours, retention outside 1 minute to 14 days) are rejected up front.

SqsQueueClientOptions.AutoCreateQueues becomes the SqsProvisioningMode
enum (Create, Validate, None); the bool stays as an [Obsolete] shim.
Validate reports every missing queue and attribute mismatch in a single
exception, and Validate/None never call CreateQueue from URL lookups.
ProvisionAsync always creates or updates so a one-off command with
elevated IAM rights can run it for an app in Validate mode.

Bodies are sent as UTF-8 JSON text instead of base64, messages over
256 KiB (body plus attributes) are rejected with the queue, size and
message type in the error, and batches split on total size as well as
the 10-entry limit. The dead-letter negative cache is only populated on
QueueDoesNotExistException and expires after one minute.

UseAws registers SDK clients from the default credential chain when no
ServiceUrl is set instead of requiring pre-registration, and the option
docs list the IAM actions each role needs.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Per-node subscription queues are created with a short retention
(SubscriptionQueueRetention, default 5 minutes) and tagged fm:role,
fm:host and fm:heartbeat; the client refreshes the heartbeat every
HeartbeatInterval (default 2 minutes). On EnsureTopicsAsync or the first
subscribe, each client sweeps queues under its prefix whose heartbeat is
older than StaleSubscriptionAge (default 10 minutes), removing their SNS
subscriptions and the queue.

The SNS-to-SQS queue policy is now applied on the SubscribeAsync path as
well and is built from every topic ARN the client knows about. A stable
HostId that restarts inside SQS's 60-second re-creation window falls back
to a timestamped queue name instead of failing; disposal tolerates queues
and subscriptions that were already swept.

Bodies travel as UTF-8 text with headers as SNS message attributes (raw
delivery) instead of a base64 JSON envelope, publishing uses
PublishBatch in chunks of 10, and the poll loop deletes a batch before
invoking handlers concurrently, matching the at-most-once contract.
The shared queue tuple becomes a sealed record to avoid torn reads, and
the unused SqsPubSubClientOptions.TopicName is removed.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The fixture becomes a collection fixture with a container-runtime probe
that skips every test with a reason when Docker is unavailable, and
enables SQS_DELAY_RECENTLY_DELETED so the stable-host restart path is
exercised. New tests cover provisioning attributes and redrive policy,
Validate/None modes, visibility timeout locking, the dead-letter negative
cache expiry, raw JSON bodies and oversize rejection, stale-queue
sweeping, fan-out to two live hosts, heartbeat refresh and the policy
set by SubscribeAsync alone.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
QueueAdministrationHandler ships mediator handlers with no HTTP surface of
their own: queue overview and detail (depth, workers, counters, handlers),
tracked job listing, lookup and cancellation, and dead-letter peek, replay and
purge. Replay ignores messages dead-lettered after the operation began so a
poison message cannot be replayed in a loop. The library now runs the source
generator (interceptors disabled) so hosts discover these handlers by
referencing the package.

Foundatio.Mediator.Distributed.Testing adds RecordingQueueClient, which wraps
any queue client, exposes typed Sent<T>() assertions and DrainAsync(), and
InMemoryTransport, one queue/pub-sub/lock trio that several service providers
share to prove cross-node behaviour in one process.

QueueJobState.LastHeartbeatUtc is populated by the Redis store.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Corrects the retry table, the job id location, and the transport option
names; documents worker selection, shared queues, interface-typed handlers,
retry schedules, [QueueLock], header providers, job metadata, graceful
shutdown, provisioning modes and IAM, subscription-queue cleanup, the Redis
key layout, the administration handlers, metrics and traces, and the testing
package. Adds Scaling Out, Operations, and Testing pages.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…meouts

distributed.props no longer pins 0.1.0-beta1 with MinVerSkip; branch builds
publish a new preview of every Distributed package again. The AWS and Redis
packages get their own descriptions and tags.

QueueWorker acknowledgements and state writes run under a disposed
per-operation timeout source instead of leaking a timer per call.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…n rule

A node that receives a bus message for a type it did not register at startup
now loads the type and accepts it when its own notification rules would have
distributed it. Handlers declared on an interface receive concrete events
published elsewhere, and no node loads a type its rules do not cover.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
… key and Redis batches

Aspire runs Vite, so the Api project no longer needs SpaProxy or the Web.esproj
reference. Aspire moves to 13.5.3 and OpenTelemetry to 1.18.0 (Redis
instrumentation stays on its latest prerelease). The LocalStack image is pinned
so the dev environment stops drifting. CachingMiddleware keyed entries on
GetHashCode(), which differs per process and defeats the shared L2 cache; it now
keys on type name plus the message JSON. The Redis repositories awaited nothing
after Execute(), so a failed batch write was silently lost. The Web lockfile is
brought back in sync with package.json.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…tureSample

Order events now share an IOrderEvent interface: OrderAuditHandler is declared on
the interface and receives OrderCreated, OrderUpdated, OrderShipped, and
OrderDeleted through one "order-events" queue. OrderConfirmationHandler
(Common.Module) and OrderFulfillmentHandler (Orders.Module) share the
"order-created" queue, so one OrderCreated message runs both; fulfillment
publishes OrderShipped from the worker, which takes the same path as an
API-side publish.

New handlers cover the remaining scenarios: FlakyWebhookHandler fails a
configurable number of attempts on a 1s/3s retry schedule and dead-letters after
three (a malformed URL dead-letters at once via Result.Invalid);
GenerateBankFileHandler is [QueueLock]-protected on a per-bank key backed by a
Redis SET NX/compare-and-delete lock provider, so two files for one bank run once;
ImportProductCatalogHandler gives the "imports" group its own tracked job.
DemoExportJob moves to the exports group with a 60s visibility timeout and
concurrency 2, reports the processing host in its progress, and publishes
DemoJobCompleted so the event feed shows the spread across replicas.

TenantHeaderProvider copies X-Tenant and the user into message headers on
enqueue and restores a TenantContext on the worker (CallContext parameter plus
ambient value for follow-on messages); JobMetadataProvider records the same on
tracked jobs. Notifications switch to an explicit rule set:
IncludeNotificationsFromAssemblyOf<IOrderEvent>() with ProductStockChanged
excluded, since only queued handlers consume it.

QueueDashboardHandler delegates to the library's administration messages
(overview, jobs, cancel, dead-letter list/replay/purge) instead of reading the
transport itself; reads stay anonymous, mutations require the Admin role. The
enqueue endpoints read the job id from Result.Location, which is where the queue
middleware puts it. ServiceDefaults registers the Foundatio.Mediator.Distributed
meter so queue.* metrics reach the Aspire dashboard, and the AppHost offers two
topologies from one project: SAMPLE_TOPOLOGY=single runs everything in one
process; the default split runs 2 API replicas with Workers=none plus
worker-exports (x2), worker-imports, and worker-events.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…, host visibility

The Queues page gains admin-only controls for each scenario (export batch,
import, flaky webhook with URL and failure count, two bank files for one bank),
a dead-letter tab per queue with reason, attempts, correlation id, body preview,
and Replay/Purge, job metadata chips (tenant, user) and heartbeat age on tracked
jobs, a badge naming the API replica that answered each poll, and a tally of
DemoJobCompleted events by host so the spread across worker replicas is visible.
The header gets a tenant selector that the fetch client sends as X-Tenant. The
event-stream store is now generic over event type and records the publishing
host, so worker-side events (DemoJobCompleted, BankFileGenerated,
WebhookDelivered, OrderShipped) appear on the Live Events page without per-type
wiring.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…demo script

The README maps each scenario to its handler and UI control, documents the
split and single topologies, and explains worker selection, tracked jobs,
interface-typed and shared queues, notification rules, [QueueLock], retry
schedules with dead-letter replay, tenant propagation, and per-host
completions. The video demo script moves out of the docs site to
samples/CleanArchitectureSample/DEMO.md with its stale references fixed.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The in-memory default and the workers-without-transport guard ran at
registration time, so `.AddDistributedQueues().UseAws()` — the order the
sample and docs show — threw on every enqueue-only node. The default is
now a TryAdd factory that DI only resolves when nothing else registered
IQueueClient, the guard fires inside that factory, and the in-memory lock
fallback moves into QueueLockMiddleware where the resolved client is known.

Also pins the sample's LocalStack image to 3.8.1, matching the tests.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Live run against LocalStack found two SQS-specific bugs:

- Administration peeks cancelled a 20 s long poll client-side. SQS keeps
  that receive open, so it grabbed the message the peek had just released
  and the next peek, and any replay, came back empty. IQueueClient gains a
  ReceiveDeadLettersAsync overload with a server-side wait; SqsQueueClient
  maps it to WaitTimeSeconds and the administration handler polls for 1 s.

- SQS allows ten message attributes, and a traced, tracked message already
  carries seven framework headers, so the dead-letter headers pushed it to
  eleven and DeadLetterAsync threw; a replayed poison message could never be
  dead-lettered again. Headers beyond ten now travel packed in one
  `fm-headers` JSON attribute on both SQS and SNS and are unpacked on read.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
@ejsmith
ejsmith force-pushed the codex/core-distributed-alternative branch from 673102f to da703ea Compare September 12, 2026 23:55
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant