Conversation
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
ejsmith
marked this pull request as ready for review
April 7, 2026 19:32
niemyjski
self-requested a review
April 11, 2026 00:31
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>
ejsmith
changed the title
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
1 participant
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.
Features and developer experience
Adds background jobs and cross-node pub/sub using the handlers and middleware you already write.
[Queue]and get an acceptance receipt fromEnqueueAsync. Each subscription has its own queue and retry budget by default, with automatic retries, dead letters, and lease renewal.[DistributedNotification]or select it in configuration, then use the samePublishAsynccall. Receiving nodes run their local handlers..UseAws()for SQS/SNS and Redis for shared job state. Run all workers together or deploy selected worker groups from the same build.[QueueLock]coordinates work on shared resources.TrackProgressfor job status, progress, worker identity, heartbeat, and cancellation. Inspect dead letters, retry or purge individual messages or batches, and retain failed-job history.Performance
September 6 measurements: median completed messages/second from three paired runs against MassTransit 8.5.10 on the same LocalStack SQS/SNS transport:
Queues use 8 producers/workers with job tracking off; pub/sub uses 40 producers and one subscriber. These are shared-host LocalStack measurements, not AWS capacity claims. MassTransit won some higher-concurrency queue runs. Foundatio notifications return after local buffering; MassTransit waits for broker acceptance. Full results, limitations, and reproduction steps.
Additional notes