Conversation
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
force-pushed
the
codex/core-distributed-alternative
branch
from
September 12, 2026 23:55
673102f to
da703ea
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds queued handlers and cluster notifications using Foundatio #533. The existing Mediator runtime, source generators, and core tests match
mainexactly. This is an alternative to #149 built through existing extension points.Developer experience
Configure the infrastructure once:
Add
[Queue]to an ordinary handler:mediator.EnqueueAsync(new ExportReport("monthly"), ct)returnsResult<QueueReceipt>confirming broker acceptance.[Queue]configures retries/concurrency;[QueueLock]adds resource locking.AddDistributedNotifications(o => o.Include<OrderChanged>()), thenmediator.PublishAsync(...). Typed subscriptions exclude unrelated local traffic; overlapping selections publish once. No local handler is required.Performance
Median jobs/sec, system .NET 10.0.12 Release, concurrency 64:
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
56a88f0aon #533; publishing waits for those APIs to ship.