Messaging and jobs redesign: one IMessageBus, durable jobs runtime, Redis Streams + AWS SQS/SNS transports

[ejsmith]

Note

Status (2026-07-09): implementation has moved past the original proposal below. The design converged on one IMessageBus — the caller's verb decides delivery (SendAsync = command to exactly one handler, PublishAsync = event fan-out) — and the separate IQueue/IPubSub surfaces described in Part 2 were removed. What has landed on this branch:

• Canonical DestinationAddress identity across the whole transport contract (send/receive/settle/stats/provisioning), with routing config doubling as topology declarations under explicit TopologyMode (Ensure / Validate / None).
• Role-aware TransportCapabilities (queue vs. topic) with core-owned fallbacks — fixes silent delay drops (SQS delay vs. no SNS delay) by routing over-ceiling/unsupported delays through the durable scheduled-dispatch store.
• Core-owned retry/dead-letter (RetryPolicy, RejectAsync, DeadLetterOn<T>, forensics headers), handler delivery intent (Deliveries = Sent | Published | Both), declarative handlers + hosted startup topology.
• Durable jobs runtime: typed payloads (EnqueueAsync<TJob, TArgs>), per-run DI scopes, bounded worker concurrency, supervised lease renewal (cancels the run on lease loss or renewal outage), CRON occurrences with retry/dead-letter budgets, IScheduledDispatchStore split for messaging-only providers.
• Transports: in-memory, Redis Streams, AWS SQS/SNS; cross-provider conformance suites; Foundatio.Testing user-facing test harness.

Current API documentation: docs/guide/messaging-jobs-redesign.md. The sections below are the original proposal, kept for the problem analysis and design rationale.

---

Foundatio Messaging & Jobs — Breaking Redesign

A clean, breaking (major-version) rewrite of Foundatio's queue, pub/sub (IMessageBus), work-item, and CRON-job abstractions into one coherent messaging + durable-jobs core. The current abstractions are single-type, poll-based, allocation-heavy, hard to implement transports for, and lack durable job state and durable scheduling. The new design moves all the hard parts into core, keeps transports tiny (bytes + headers), and unifies queues, pub/sub, and jobs on one substrate. This is still a major-version replacement for the prime Foundatio.Queues / Foundatio.Messaging / Foundatio.Jobs abstractions, but it should be delivered as a gated phased rollout, not a single big rewrite. The public replacement should be cut only after the transport substrate, conformance harness, and at least one real external provider mapping have proven the design.

Approach — expose messaging patterns, not a generic bus

Foundatio deliberately does not try to be a generic, do-everything message bus or broker abstraction. It exposes the messaging patterns the vast majority of apps actually use — work queues (competing consumers; at-least-once; ack/retry/dead-letter) and pub/sub (fan-out) — plus durable jobs and scheduling built on the same substrate, as the first-class abstractions.

This is a deliberate departure from existing messaging libraries. Those tend to be extremely generic and highly configurable — powerful, but the team finds it genuinely hard and confusing to figure out how to implement the super-common patterns most apps need; the flexibility comes at the expense of approachability for the common case. Foundatio inverts that trade-off: the common patterns are the API and are obvious to use, transports stay swappable underneath, and advanced needs are met by clearly-scoped, opt-in capabilities rather than up-front configuration. Everything in this design serves that goal — which is also why the hard parts (serialization, routing, multi-type dispatch, priority, retry/dead-letter, back-pressure, durable scheduling) live in core and transports stay tiny.

---

Part 1 — Shortcomings of the current library

(Verified against src/Foundatio and src/Foundatio.Extensions.Hosting.)

1. One message type per queue/topic

IQueue<T> is generic over a single CLR type; IMessageBus routes by Type with one Topic per bus instance — one queue/topic ⇒ one .NET type. No polymorphic handling: you can't put N message types on one destination and dispatch each to its own handler without N queues/topics or hand-rolled wrappers. No priority support within a destination.

2. Heavyweight transports

Providers inherit fat base classes — QueueBase<T,TOptions> has ~10 abstract overrides and bakes in serializer ownership, CLR-type↔string mapping, the worker loop, metrics, events, behaviors, poison handling, and resilience; MessageBusBase<TOptions> is similar. Transports serialize themselves. Consumption is poll-based (worker loops call DequeueImplAsync; in-memory polls on a 10s timeout) — no push. No back-pressure (unbounded enqueue). Capabilities are implicit — providers silently no-op or throw for unsupported features (e.g. delayed delivery falls back to an in-memory timer that's lost on restart) with no discovery.

3. No batching

No batch enqueue/dequeue or batch publish/subscribe APIs — one round-trip per message caps throughput.

4. Pub/sub conflated as a generic "message bus"

IMessageBus is named like a generic bus but is pub/sub — Foundatio implements messaging patterns (work queues, pub/sub), not a generic bus. Subscribers effectively receive ALL messages and filter by assignable CLR type in-process (task-per-subscriber); topic is a per-instance option, so listening to a subset of topics needs multiple bus instances. No per-message ack/reject; fire-and-forget with no documented delivery guarantee; subscriber errors are swallowed (can't distinguish "no subscribers" from "handler threw").

5. No generic, opt-in job-state tracking

No generic durable/queryable job state. Work-item progress is fire-and-forget WorkItemStatus published over IMessageBus (not stored); CRON history is in-memory, capped ~10, per-node, lost on restart. No cancel-by-id (only host-shutdown token / lock release). Tracking isn't opt-in per job. WorkItemJob is a separate ad-hoc persisted-job system on IQueue<WorkItemData>, not unified with CRON/IJob.

6. Non-durable distributed CRON

Distributed CRON runs the job body in-process under a distributed lock (two locks: a per-scheduled-minute lock held 1h + a per-job-running lock held 15m) and enqueues no durable occurrence — a node crash mid-run loses the run with no retry/dead-letter/monitor. Retries are unimplemented (// TODO set next run time to retry, but need max retry count at ScheduledJobInstance.cs:277 & :296). Coarse per-minute cadence with misfire rules (skip if >1h late, fast-track if <10min out). No node identity for coordination/debugging; manual run has no idempotency/return/completion signal; distributed state sync over IMessageBus is lossy (no durability/replay/resync).

7. Allocation-heavy / no binary path

Values are deep-cloned (QueueEntry.Value = value.DeepClone()); serialization always allocates a fresh byte[]; no ReadOnlyMemory<byte> path. Metrics gauges do sync-over-async (GetAwaiter().GetResult()), risking thread-pool starvation.

8. Other cross-cutting issues

• Assembly-qualified type discriminator ("FullName, Assembly" + a version-stripping fallback in MessageBusBase) — brittle across services/versions.
• Dead-letter access returns typed IEnumerable<T> — can't inspect poison/unknown payloads without the original type.
• Poison messages produce a phantom IQueueEntry<T> with null Value — breaks the type contract.
• Retry policy is global per queue, not per-message; no custom backoff per message.
• No ordering guarantees modeled or documented (FIFO / partitions / sessions).
• Trace/correlation propagation is manual and ad-hoc (CorrelationId doubles as traceparent; a magic "TraceState" property).
• Sync + async disposal on the public interfaces (legacy baggage).
• Metrics/events live in the base class every provider inherits, rather than centralized in core.
• Continuous job loop has no exponential backoff / circuit-break on repeated failures (hot-spins at ~100ms).
• Manual lock renewal required during processing; no auto-renew → orphan / split-brain risk.
• DX/registration friction: triple-generic option builders; no message factory for testing handlers.

---

Part 2 — v2 Design

Two end-user APIs — IQueue (durable competing-consumer work) and IPubSub (fan-out) — share one consume core, one wire model, one capability model, and the jobs layer. Transports are tiny (bytes + headers); core owns serialization, routing, multi-type dispatch, priority lanes, retry/backoff/dead-letter, back-pressure, the due-time substrate, job-state, scheduling, tracing, and metrics.

Namespaces

• Foundatio.Messaging — the single IMessageTransport + wire model (headers, transport messages, capabilities), serializer + content-type registry, type-routing registry, IPubSub.
• Foundatio.Queues — IQueue, the unified IReceivedMessage<T>, worker/consume options.
• Foundatio.Jobs — IJob, JobContext, IJobClient / IJobMonitor for app code, IJobRuntimeStore for provider/runtime coordination, scheduler, monitor, durable CRON.
• Foundatio.TestHarness — reusable transport conformance suite + core-behavior/jobs test bases (referenced by every provider repo); ships the in-memory reference transport that passes it.

Layering

App      ──▶ IQueue / IPubSub / IJob
Core     ──▶ serialize (+content-type registry) · type→destination routing · multi-type/polymorphic dispatch ·
             priority lanes + bounded priority buffer (back-pressure) · Auto/Manual ack · retry/backoff/dead-letter ·
             due-time substrate (delay/backoff/CRON) · expiry · job-state · scheduler + monitor · tracing · metrics
Transport──▶ IMessageTransport  (one contract; bytes + headers; send/consume/complete/abandon + ISupportsX; queue vs topic = destination role)

Wire model & headers (Foundatio.Messaging)

public enum MessagePriority { Low = 0, Normal = 1, High = 2 }
public enum DeliveryGuarantee { AtMostOnce, AtLeastOnce }
public enum OrderingGuarantee { None, Fifo, PerPartition }

public sealed class MessageHeaders : IReadOnlyDictionary<string,string> {        // case-insensitive (Ordinal), immutable, FrozenDictionary
    public static MessageHeaders Empty { get; }
    public static MessageHeaders Create(IEnumerable<KeyValuePair<string,string>> headers);
    public string? GetValueOrDefault(string key);
    public Builder ToBuilder();
    public sealed class Builder { public Builder Add/Set/SetIfMissing(string k, string v); public bool Remove(string k); public MessageHeaders Build(); }
}
public static class KnownHeaders {   // reserved message.* namespace for system keys
    public const string MessageType="message.type", ContentType="message.content_type", CorrelationId="message.correlation_id",
        TraceParent="traceparent", TraceState="tracestate", Priority="message.priority",
        Expiration="message.expiration", Attempts="message.attempts", DeadLetterReason="message.dead_letter.reason";
}

// Transport-facing: PURE body bytes + SEPARATE headers. The transport owns how headers are physically carried.
public sealed record TransportMessage { public required ReadOnlyMemory<byte> Body { get; init; } public MessageHeaders Headers { get; init; } = MessageHeaders.Empty; public string? MessageId { get; init; } }
public sealed record TransportSendOptions { public MessagePriority Priority { get; init; } = MessagePriority.Normal; public DateTimeOffset? DeliverAt { get; init; } public string? DeduplicationId { get; init; } public string? PartitionKey { get; init; } } // Priority honored iff transport is ISupportsPriority; DeliverAt iff ISupportsDelayedDelivery
public sealed record TransportEntry { public required string Id { get; init; } public required string Destination { get; init; } public required ReadOnlyMemory<byte> Body { get; init; } public MessageHeaders Headers { get; init; } = MessageHeaders.Empty; public int DeliveryCount { get; init; } = 1; public DateTimeOffset? EnqueuedUtc { get; init; } public required Receipt Receipt { get; init; } }
public readonly struct Receipt { public object? TransportState { get; init; } }   // opaque, instance/connection-scoped, non-serializable
public sealed record ReceiveRequest { public int MaxMessages { get; init; } = 1; public TimeSpan? MaxWaitTime { get; init; } }
public sealed record SendItemResult { public string? MessageId { get; init; } public required bool Success { get; init; } public string? ErrorCode { get; init; } public bool IsRetryable { get; init; } }
public sealed record SendResult { public required IReadOnlyList<SendItemResult> Items { get; init; } public bool AllSucceeded { get; } }
public sealed class ReceiptExpiredException : Exception { }

// Capabilities are discovered by which interfaces a transport implements — pattern-match `is ISupportsX` — NOT a property on the base contract.
// The few descriptive facts that aren't method-presence are reported via an OPTIONAL interface (absence ⇒ safe defaults: at-most-once, unordered, no advertised limits):
public interface ITransportInfo {
    DeliveryGuarantee DeliveryGuarantee { get; }
    OrderingGuarantee Ordering { get; }
    IReadOnlySet<DestinationRole> SupportedRoles { get; }   // which roles this backend can provision (Queue/Topic/Subscription/…) — validated against the topology at startup
    int? MaxBatchSize { get; }
    long? MaxMessageBytes { get; }
}

Transport contract (provider-facing) — one transport for both patterns

There is a single transport. Queue vs pub/sub is not a different interface — it's the DestinationRole of the destination, set at provisioning: send to a queue = competing-consumer; send to a topic = fan-out to its subscriptions; you consume a queue or a subscription identically. SendAsync covers "publish" (a topic destination fans out). Both end-user APIs (IQueue, IPubSub) run over this one transport.

Required base (produce + ack):

public interface IMessageTransport : IAsyncDisposable {
    Task<SendResult> SendAsync(string destination, IReadOnlyList<TransportMessage> messages, TransportSendOptions options, CancellationToken ct = default); // destination = queue OR topic
    Task CompleteAsync(TransportEntry entry, CancellationToken ct = default);                                          // ack; throws ReceiptExpiredException if invalid; no-op on at-most-once
    Task AbandonAsync(TransportEntry entry, CancellationToken ct = default);                                           // immediate requeue
}

Consume: a transport implements ISupportsPull and/or ISupportsPush (at least one); core adapts both to the unified handler model and prefers push when present. The consume source is a queue or a subscription.
Optional ISupportsX (implement only if native; otherwise core emulates or flags):

ISupportsPull               : IMessageTransport { Task<IReadOnlyList<TransportEntry>> ReceiveAsync(string source, ReceiveRequest request, CancellationToken ct); }                       // batch pull; empty list on timeout (never null/throw)
ISupportsPush               : IMessageTransport { Task<IPushSubscription> SubscribeAsync(string source, Func<TransportEntry,CancellationToken,Task> onMessage, PushOptions options, CancellationToken ct); }   // push delivery — one of pull/push required
ISupportsRedeliveryDelay    : IMessageTransport { Task AbandonAsync(TransportEntry e, TimeSpan redeliveryDelay, CancellationToken ct); }   // else core due-time hold
ISupportsDeadLetter         : IMessageTransport { Task DeadLetterAsync(TransportEntry e, string? reason, CancellationToken ct); }          // else core managed {dest}-deadletter
ISupportsLockRenewal        : IMessageTransport { Task RenewLockAsync(TransportEntry e, TimeSpan? duration, CancellationToken ct); }       // else no-op + log (un-emulable)
ISupportsVisibilityTimeout  : IMessageTransport { Task<IReadOnlyList<TransportEntry>> ReceiveAsync(string source, ReceiveRequest request, TimeSpan visibility, CancellationToken ct); } // un-emulable
ISupportsStats              : IMessageTransport { Task<QueueStats> GetStatsAsync(string destination, CancellationToken ct); }              // un-emulable; absent ⇒ depth unknown
ISupportsPriority           : IMessageTransport { }   // marker: honors TransportSendOptions.Priority + delivers higher-priority first (else core priority lanes)
ISupportsDelayedDelivery    : IMessageTransport { }   // marker: honors TransportSendOptions.DeliverAt (else core due-time substrate)
ISupportsExpiration         : IMessageTransport { }   // marker: honors the message.expiration header / TTL natively (else core enforces on receive)
ISupportsProvisioning       : IMessageTransport { Task EnsureAsync(IReadOnlyList<DestinationDeclaration> declarations, CancellationToken ct); Task DeleteAsync(string name, CancellationToken ct); Task<bool> ExistsAsync(string name, CancellationToken ct); }   // batch ensure (one-item list for dynamic); provisions queues/topics/subscriptions/bindings + related resources per DestinationRole coherently
public interface IPushSubscription : IAsyncDisposable { string Source { get; } }   // named handle so a push subscription can grow (Pause/Resume, Name, IsActive, stats…) without breaking callers

Capability discovery is transport is ISupportsX (pattern-match) — no capabilities property; marker interfaces signal "honor this option/header natively" without adding a method. A transport advertises which DestinationRoles it can provision via ITransportInfo.SupportedRoles, so "this backend can't do topics" is validated at startup rather than expressed as a missing interface. Per-pattern different brokers (e.g. SQS queues + Redis pub/sub) = named/keyed transport registration + a per-destination selector.

Unified received message + ack (Foundatio.Queues)

public enum AckMode { Auto, Manual }
public interface IReceivedMessage<out T> where T : class {
    T Message { get; }
    string Id { get; } MessageHeaders Headers { get; } string? CorrelationId { get; } string? MessageType { get; }
    MessagePriority Priority { get; } int Attempts { get; } bool IsHandled { get; }
    CancellationToken CancellationToken { get; }                          // host shutdown ∪ lock loss ∪ persisted cancel (when tracked)
    Task CompleteAsync(CancellationToken ct = default);
    Task RejectAsync(bool retry = true, string? reason = null, CancellationToken ct = default);  // retry→backoff→DLQ; !retry→DLQ
    Task DeadLetterAsync(string? reason = null, CancellationToken ct = default);
    Task RenewLockAsync(TimeSpan? duration = null, CancellationToken ct = default);              // no-op+log where unsupported
    Task ReportProgressAsync(int? percent = null, string? message = null, CancellationToken ct = default); // logs-only when not tracking
}

Queues (competing-consumer, at-least-once) and pub/sub (fan-out; guarantee per transport) share this exact consume model; Reject/DeadLetter are no-ops + flagged on at-most-once transports.

End-user produce APIs

public sealed record EnqueueOptions { public MessagePriority Priority { get; init; } = MessagePriority.Normal; public TimeSpan? Delay { get; init; } public DateTimeOffset? DeliverAt { get; init; } public TimeSpan? TimeToLive { get; init; } public string? CorrelationId { get; init; } public string? DeduplicationId { get; init; } public string? Destination { get; init; } public MessageHeaders? Headers { get; init; } }
public interface IQueue : IAsyncDisposable {
    Task<string> EnqueueAsync<T>(T message, EnqueueOptions? o = null, CancellationToken ct = default) where T : class;
    Task EnqueueBatchAsync<T>(IEnumerable<T> messages, EnqueueOptions? o = null, CancellationToken ct = default) where T : class;
}
public interface IPubSub : IAsyncDisposable {
    Task PublishAsync<T>(T message, PublishOptions? o = null, CancellationToken ct = default) where T : class;
    Task PublishBatchAsync<T>(IEnumerable<T> messages, PublishOptions? o = null, CancellationToken ct = default) where T : class;
}

Routing, dispatch & workers (core)

• Type → destination: convention (OrderPlaced→order-placed) > [Queue]/[Topic] > fluent Map<T>(...) > custom fn. message.type = logical alias ([MessageType("...")]/registry; N inbound aliases → 1 type).
• Per-type handler registration; core derives the consumer topology — handlers routing to the same destination share ONE worker that dispatches by message.type → most-specific handler (base/interface fallback). Single registered type + missing discriminator ⇒ infer (raw-producer interop). Unknown/unresolvable or deserialize-failure ⇒ poison (default dead-letter, reason in header) + optional HandleUnknown.
• Worker = handler-based (never poll). Auto ack default (return=Complete, throw=Reject(retry)); Manual opt-in. Single-message handler is the default (core auto-batches the fetch); optional batch handler (HandleBatch<T>) with per-message ack (unacked follow batch outcome). Core owns retry/backoff/attempt-cap/dead-letter.

public sealed record WorkerOptions { public AckMode AckMode { get; init; } = AckMode.Auto; public int Prefetch { get; init; } = 1; public int MaxConcurrency { get; init; } = 1; public int BatchSize { get; init; } = 1; public int MaxAttempts { get; init; } = 5; public Func<int,TimeSpan>? RedeliveryBackoff { get; init; } public bool TrackState { get; init; } = false; public PoisonDisposition OnUnknownType { get; init; } = PoisonDisposition.DeadLetter; public PoisonDisposition OnDeserializeFailure { get; init; } = PoisonDisposition.DeadLetter; }
public enum PoisonDisposition { DeadLetter, AbandonWithCap, RouteToFallback, CompleteAndDrop }

Priority lanes & back-pressure (core)

• Core maps (destination, priority) → physical lane (orders, orders$high, …) — the lanes are part of the derived topology (provisioned per the provisioning mode); routes on send by EnqueueOptions.Priority. Native-priority transports (those implementing ISupportsPriority) use one destination + TransportSendOptions.Priority (no lanes).
• Consume fills a bounded priority buffer (size = Prefetch) from the lanes (or native), drains high→low to the handler with MaxConcurrency, with a configurable anti-starvation weight (every Nth pull services a lower lane). The buffer IS the back-pressure: full ⇒ stop pulling / pause push.

Provisioning & topology (core)

Provisioning is a separate optional capability, not on the data-plane contract — ISupportsProvisioning with a batch EnsureAsync(IReadOnlyList<DestinationDeclaration>) (a one-item list for dynamic ensure-on-first-use; the full set on startup), plus DeleteAsync/ExistsAsync — creating queues, topics, subscriptions, and bindings per DestinationRole, related resources coherently in one pass. Auto-creating backends (Redis, in-memory) don't implement it (create-on-send); brokers needing explicit creation do.

public sealed record DestinationOptions {
    public bool PriorityEnabled { get; init; }            // provision priority lanes / native max-priority
    public TimeSpan? DefaultMessageTtl { get; init; }
    public DeadLetterPolicy? DeadLetter { get; init; }    // DLQ target + native redrive/max-receive backstop (the attempt-cap policy stays core-owned)
    public TimeSpan? LockDuration { get; init; }          // default visibility/lock window
    public int? Partitions { get; init; }
    public IReadOnlyDictionary<string,string>? NativeOptions { get; init; }  // escape hatch for provider-specific settings
}
public sealed record DeadLetterPolicy { public string? Destination { get; init; } public int? MaxDeliveries { get; init; } }

public enum DestinationRole { Queue, Topic, Subscription, PriorityLane, DeadLetter }
public sealed record DestinationDeclaration { public required string Name { get; init; } public DestinationRole Role { get; init; } public DestinationOptions Options { get; init; } = new(); public string? TopicBinding { get; init; } } // a subscription/lane/DLQ names what it binds to

The transport maps what it can to native creation params (SQS redrive policy; Azure SB queue properties / lock duration / TTL; RabbitMQ declare args — x-max-priority/message-ttl/DLX); core honors the rest itself.

Core derives the topology (not hand-maintained): from Map<T>, handler registrations, cron jobs, priority lanes (knowable from the fixed Low/Normal/High enum), DLQ targets, and the job/due-time/state destinations → a set of declarations (name, role, DestinationOptions).

A hosted provisioner applies a ProvisioningMode:

• Dynamic (default) — ensure-on-first-use: EnsureAsync([declaration]) (one-item list), lazy + cached.
• OnStartup — provision the entire derived topology in one EnsureAsync(allDeclarations) call, fail-fast (everything exists before traffic); the transport creates related resources/bindings (topic + subscriptions + DLQ + redrive) coherently.
• Validate — don't create; check ExistsAsync for every declaration on startup, fail-fast if missing (locked-down / externally-provisioned environments where the app lacks create permissions).
• None — do nothing; assume pre-provisioned.

If the chosen mode needs provisioning but the transport lacks ISupportsProvisioning (and doesn't auto-create), that's surfaced as a startup configuration error.

Serialization (core)

ISerializer + a content-type registry: a default serializer for send (stamps message.content_type); receive picks the serializer by content-type (mixed formats per destination + external interop). Body = pure bytes; transport owns header encoding (native attributes; pack-into-one under limits like SQS's 10; or fold into payload where it can't carry headers). Buffers owned/non-pooled in v1; the entry/receipt contract is shaped to enable pooling/zero-copy later. Trace context rides in headers — core injects traceparent/tracestate on send and starts a consumer Activity on receive.

Due-time substrate + expiry (core)

• One durable "make visible at T" mechanism backs delayed/scheduled send (Delay/DeliverAt), redelivery-backoff (Reject → reappear later), and CRON occurrences. Native per-message delay used where the transport implements ISupportsDelayedDelivery; else core persists the pending dispatch in the same durable runtime store used for jobs (IJobRuntimeStore), claims due records with a short lease, and materializes them into the destination/lane when due (survives restarts). This intentionally avoids a separate IDueTimeStore; the requirement is one runtime store contract with due-time methods, not two parallel persistence abstractions. A delayed message enters its priority lane when due.
• TTL → absolute message.expiration (enqueue + TTL); enforced on receive (native broker TTL where present); expired ⇒ dead-letter expired (default) or drop.

Job client + runtime store (Foundatio.Jobs)

public enum JobStatus { Queued, Scheduled, Processing, Completed, Failed, Cancelled, DeadLettered }
public enum ScheduledDispatchKind { QueueMessage, PubSubMessage, JobOccurrence }
public sealed record JobState { /* JobId, Name, Status, Progress, ProgressMessage, Attempt, NodeId, timestamps, Error, CancellationRequested, ScheduledForUtc */ }
public sealed record ScheduledDispatchState { /* DispatchId, Kind, Destination, Body, Headers, Options, DueUtc, ClaimOwner, ClaimExpiresUtc, Attempts, JobId? */ }
public sealed record JobStatePatch { /* Status?, Progress?, ProgressMessage?, Error?, AttemptDelta?, NodeId?, LeaseExpiresUtc?, LastUpdatedUtc? */ }

// App-facing read model for dashboards and diagnostics.
public interface IJobMonitor {
    Task<JobState?> GetAsync(string jobId, CancellationToken ct = default);
    Task<IReadOnlyList<JobState>> QueryAsync(JobQuery query, CancellationToken ct = default);
}

// App-facing command surface. Normal users inject this, not the runtime store.
public interface IJobClient : IJobMonitor {
    Task<string> RunAsync<TJob>(RunJobOptions? options = null, CancellationToken ct = default) where TJob : IJob;
    Task<bool> RequestCancellationAsync(string jobId, CancellationToken ct = default);
}

// Runtime/provider-facing SPI. The same concrete implementation can back IJobClient/IJobMonitor,
// but app code should not need these methods.
public interface IJobRuntimeStore : IJobMonitor {                // ALL updates are field-level/partial — never whole-record replace
    Task CreateIfAbsentAsync(JobState initial, CancellationToken ct = default);                 // occurrence anchor + dedup (atomic)
    Task<bool> TryTransitionAsync(string jobId, JobStatus expectedStatus, JobStatus newStatus, JobStatePatch? patch = null, CancellationToken ct = default);
    Task<bool> TryClaimAsync(string jobId, string nodeId, TimeSpan lease, CancellationToken ct = default);
    Task<bool> RenewClaimAsync(string jobId, string nodeId, TimeSpan lease, CancellationToken ct = default);
    Task<bool> ReleaseClaimAsync(string jobId, string nodeId, CancellationToken ct = default);
    Task SetProgressAsync(string jobId, int? percent = null, string? message = null, CancellationToken ct = default);
    Task IncrementAttemptAsync(string jobId, CancellationToken ct = default);
    Task<bool> IsCancellationRequestedAsync(string jobId, CancellationToken ct = default);
    Task ScheduleDispatchAsync(ScheduledDispatchState dispatch, CancellationToken ct = default); // create/dedup pending due-time dispatch
    Task<IReadOnlyList<ScheduledDispatchState>> ClaimDueDispatchesAsync(DateTimeOffset now, int limit, string nodeId, TimeSpan lease, CancellationToken ct = default);
    Task CompleteDispatchAsync(string dispatchId, string nodeId, CancellationToken ct = default);
    Task ReleaseDispatchAsync(string dispatchId, string nodeId, DateTimeOffset nextDueUtc, CancellationToken ct = default);
}

Opt-in per registration (WorkerOptions.TrackState; CRON occurrences are always tracked). When on, core auto-manages the lifecycle (Queued→Processing→Completed/Failed/Cancelled) through conditional transitions, claim ownership, and renewable leases; the handler only adds progress via the context. ReportProgressAsync when untracked ⇒ logs only (never throws). Cancellation is requested through IJobClient and observed cooperatively by the runtime (context token + progress checkpoint). Delayed-send fallback, retry backoff, and CRON occurrence materialization use the scheduled-dispatch methods on IJobRuntimeStore so there is one durable coordination dependency to configure and test, without exposing those methods to normal app code.

Durable CRON (Foundatio.Jobs)

public interface IJob { Task<JobResult> RunAsync(JobContext ctx); }
public sealed class JobContext { /* JobId, Name, TriggerKind, Attempt, ScheduledForUtc, Services, CancellationToken */ public T? GetMessage<T>() where T:class; public Task ReportProgressAsync(int? percent=null, string? message=null, CancellationToken ct=default); public Task RenewLockAsync(TimeSpan? d=null, CancellationToken ct=default); }
public sealed record ScheduledJobDefinition { public required string Name { get; init; } public required string Cron { get; init; } public TimeZoneInfo? TimeZone { get; init; } public ScheduledJobScope Scope { get; init; } = ScheduledJobScope.Global; public OverlapPolicy Overlap { get; init; } = OverlapPolicy.SkipIfRunning; public TimeSpan? MisfireWindow { get; init; } public int MaxRetries { get; init; } = 3; public bool Enabled { get; init; } = true; }
public enum ScheduledJobScope { Global, PerNode }
public interface IJobScheduler { Task ScheduleAsync(ScheduledJobDefinition def, CancellationToken ct = default); Task UnscheduleAsync(string name, CancellationToken ct = default); Task<IReadOnlyList<ScheduledJobDefinition>> GetSchedulesAsync(CancellationToken ct = default); }

Flow: scheduler tick (configurable, ~1s) → due occurrence id {job}:{tick}:{scope} (Global=global, PerNode={nodeId}) → CreateIfAbsentAsync (atomic anchor + dedup; store implementations use native CAS/conditional writes or their own lock internally) → ScheduleDispatchAsync/claim due occurrence → materialize onto IQueue (always tracked) → worker resolves IJob from DI and runs it using TryClaimAsync + renewable ownership. A hosted monitor reconciles IJobRuntimeStore.QueryAsync vs worker liveness (heartbeat via lock-renewal/LastUpdated + node id): created-but-never-picked-up or started-then-stale occurrences are re-enqueued up to MaxRetries, then aborted/dead-lettered (+ optional alert hook). Misfire = skip past the per-schedule window (catch-up within window). NodeId = config > FOUNDATIO_NODE_ID > machine name.

DI / registration

services.AddFoundatio(b => b.UseQueueTransport<InMemoryQueueTransport>().UsePubSubTransport<InMemoryPubSubTransport>().UseSerializer<SystemTextJsonSerializer>().Map<OrderShipped>(queue: "orders"));
services.AddQueueHandler<OrderPlaced>(async (msg, ct) => { … }, o => o with { TrackState = true });   // delegate or class handler
services.AddPubSubHandler<CacheInvalidated>(async (msg, ct) => { … });
services.AddJob<ReportJob>();                                   // one-off / manual via IJobClient
services.AddCronJob<NightlyRollupJob>("0 0 * * *", o => o with { Scope = ScheduledJobScope.Global });

Observability & stats (core)

Stats are split by authoritative vs derived — conflating them is what made this hard before:

• Throughput / rates (sent, received, completed, abandoned, dead-lettered, errors, process-duration, queue-latency) are emitted as OpenTelemetry metrics from the core pipeline — counters/histograms tagged destination / message.type / priority / transport. These are flow metrics; a metrics backend (or an in-proc reader) aggregates them across nodes.
• Current depth (queued / in-flight / dead-lettered) is queried from the transport via optional ISupportsStats — the broker is the source of truth, so out-of-band changes (purges, TTL expiry, other producers/consumers, redelivery) are reflected accurately. Broker depth is inherently cluster-wide (one query returns the whole destination's depth), approximate on some backends, and cached + rate-limited by core. No derived/aggregated depth counter is maintained — it would drift from reality.
• Tracked-work status (in-progress + progress %, completed/failed/cancelled, history) comes from IJobMonitor.QueryAsync — backed by the shared runtime store, so it's authoritative and cluster-wide for tracked jobs/occurrences.
• Observable gauges never block: they read only in-memory counters or the cached transport-depth snapshot — never a synchronous transport call (fixes the old sync-over-async gauge that risked thread-pool starvation).
• Full OpenTelemetry, built in: a Meter per area (Foundatio.Queues / Foundatio.Messaging / Foundatio.Jobs), the shared ActivitySource (producer + consumer spans, context propagated through headers), and structured ILogger. Wire up OTel → metrics + traces + logs with zero extra code.
• High-performance logging patterns: all logging uses source-generated [LoggerMessage] (compile-time, allocation-free, no boxing) with stable event ids and named structured fields — never string interpolation/concatenation in log calls; expensive/diagnostic messages are guarded with ILogger.IsEnabled(level); correlation/message ids flow via ILogger scopes. Logging is cheap enough to leave in the hot path.
• Dashboard: core exposes the data — a per-destination stats snapshot API (core throughput counters + on-demand transport depth) and the IJobMonitor query — plus OTel. No first-party dashboard UI; build your own over the data or point a metrics backend at the OTel output.

Capability model (summary)

Discovery: a transport's capabilities = the interfaces it implements (is ISupportsX) — there is no capabilities property on the contract. Required base: Send, Complete, Abandon, Dispose, plus ≥1 consume capability — ISupportsPull and/or ISupportsPush (core adapts; prefers push when present). Optional ISupportsX, core-emulated when absent: redelivery-delay (→due-time via IJobRuntimeStore), dead-letter (→managed DLQ), priority (→lanes), delayed delivery (→durable due-time via IJobRuntimeStore), expiration (→on-receive). Optional, not emulable (no-op/unavailable): lock-renewal, queue-depth stats (throughput metrics are always-on via core/OTel — only current depth needs the transport), visibility-timeout, provisioning (ISupportsProvisioning — batch EnsureAsync(declarations); auto-creating backends need none — governed by ProvisioningMode). Descriptive facts (delivery guarantee, ordering, supported roles, batch/size limits) come from the optional ITransportInfo interface; absence ⇒ safe defaults. There is one IMessageTransport for both patterns — queue vs pub/sub is the destination's DestinationRole, not a separate interface.

Provider mapping (separate repos)

• In-memory — reference, full surface, exact stats.
• Azure Service Bus — native complete/abandon/DLQ/lock-renewal/scheduled-messages; no per-receive visibility; no native priority → core lanes.
• AWS SQS/SNS — visibility = lock; redelivery-delay via ChangeMessageVisibility; batch/attrs ≤10 (transport packs headers); delay ≤900s else IJobRuntimeStore due-time fallback; no priority → lanes.
• Redis — lists + sorted-sets (ZSET → native priority by score); first-class IJobRuntimeStore (SET NX / ZSET); pub/sub at-most-once, streams durable.
• RabbitMQ — ack/nack/DLX; x-max-priority → native priority; no lock-renewal; delay via plugin/TTL else core.
• Kafka — topic + consumer-group (subscription) roles; at-least-once via offset commit; no per-message complete/abandon, delay, priority, or DLQ (offset-based → core-emulated or unavailable); SupportedRoles excludes queue-role semantics that need per-message ack.

Testing — core tests + a reusable transport conformance harness

A first-class deliverable: shipping a new transport means referencing Foundatio.TestHarness, supplying a factory, and getting a full pass/fail conformance suite — the same model the current library uses (a provider subclasses a base test and overrides a Create… factory).

• Foundatio.TestHarness (NuGet) — a reusable, xUnit-based transport conformance suite. A transport author writes one small subclass and inherits the entire suite:

public class RedisMessageTransportTests : MessageTransportConformanceTests {
    protected override IMessageTransport CreateTransport() => new RedisMessageTransport(/* test connection */);
}

• Capability-aware: the harness inspects which ISupportsX the transport implements (+ ITransportInfo) and runs only the applicable tests — capabilities a transport doesn't claim are skipped, not failed. A minimal transport (send + consume + complete/abandon) passes a small suite; a full-featured one runs everything.
• Conformance coverage: produce + batch send; consume via pull and/or push; complete/abandon; receipt validity + ReceiptExpiredException; redelivery (native + core-emulated); dead-letter (native + core fallback); priority ordering + anti-starvation (lanes or native); expiration; delayed / due-time delivery; provisioning (Ensure/Exists/Delete + roles/topology); back-pressure (bounded in-flight under a slow handler); fan-out vs competing-consumer (topic vs queue role); delivery-guarantee assertions; header + trace round-trip; cancellation + disposal.
• In-memory reference transport ships in core and passes the full conformance suite — it proves the harness, is the default for app dev/testing, and is the baseline the core-behavior + benchmark suites run against.
• Core-behavior suite (transport-agnostic, on in-memory): type→destination routing; multi-type + polymorphic dispatch (most-specific + base/interface fallback); discriminator inference; poison handling; Auto/Manual ack; retry → backoff → dead-letter; batch handler + partial-failure; serializer/content-type registry; priority; trace propagation; observability counters.
• Jobs suite: opt-in tracking + partial state updates; occurrence dedup (create-if-absent) under concurrent schedulers; monitor retry/abort of lost/stuck occurrences; misfire policy; cooperative cancellation; PerNode node-id stability.
• Compat suite: the old-API shims behave correctly over the new core.
• The same transport factory powers the benchmark suite below, so a transport is conformance-tested and benchmarked with one identical plug-in.

Performance & benchmarks

• Binary ReadOnlyMemory<byte> bodies, pooling-ready contract, bounded channels, TimeProvider, minimal hot-path allocation.
• Transport-swappable benchmark suite (BenchmarkDotNet [MemoryDiagnoser] + a sustained throughput/latency rig), parameterized by the same factory as the conformance harness — enqueue/dequeue, priority-drain, dispatch, serialize, fan-out, back-pressure — so transports compare apples-to-apples and regressions are caught in CI.

Migration / compat

Breaking major version. Separate Foundatio.Compat.* packages re-expose the old IQueue<T> / IMessageBus / WorkItemJob / CRON APIs over the new core (namespace/registration swap; not drop-in).

Phased build sequence

The work should be sequenced as vertical slices with explicit exit gates. Keep new public APIs internal/preview until the transport abstraction has passed both the in-memory reference implementation and one real provider. If the real provider exposes a mismatch, revise the abstraction before expanding into jobs, scheduler, or broad provider ports.

0. Compatibility and break map — document old API behaviors that must survive in shims, source/binary breaking changes, provider capability gaps, migration constraints, and success criteria for the major release.
1. Transport foundation — wire model + headers + serializer/content-type registry + capability discovery + minimal IMessageTransport + in-memory reference transport + conformance harness + benchmark baseline.
2. Real-provider vertical slice — port one production transport early, preferably SQS or Azure Service Bus, and validate send/receive, settlement, batching, headers, tracing, delay limits, DLQ, provisioning, and capability reporting against the same conformance harness.
3. Queue core — build IQueue over the proven substrate: routing, dispatch, manual receive API, handler workers, Auto/Manual ack, retry/backoff/DLQ, back-pressure, priority lanes, due-time, expiry, tracing, and core behavior tests.
4. Pub/sub core — build IPubSub over the same substrate with topic/subscription roles, fan-out behavior, at-most-once/at-least-once semantics surfaced by capabilities, and pub/sub-specific conformance coverage.
5. Job client + runtime coordination — add IJobClient / IJobMonitor for app-facing job status, progress, manual run, and cancellation, backed by IJobRuntimeStore for opt-in tracking, ownership/heartbeat, conditional state transitions, and fallback due-time scheduling; do not introduce a separate IDueTimeStore.
6. Durable CRON — add scheduler + monitor on top of IQueue and IJobRuntimeStore, including occurrence deduplication, scheduled-dispatch claim/materialize flow, misfire handling, stale-run recovery, retries, and dead-lettering.
7. Provider ports — port Redis, Azure, AWS, RabbitMQ, Kafka, and other providers one at a time; each provider must pass its advertised conformance matrix and benchmark smoke suite before release.
8. Compat and migration — ship old-API shims, migration guide, upgrade samples, and tests proving IQueue<T>, IMessageBus, WorkItemJob, and CRON compatibility over the new core.

Non-goals

Sagas, routing DSL, exactly-once delivery (impossible in the general case), schema registry, workflow engine, broker-independent topology beyond simple destination/topic provisioning (ISupportsProvisioning + DestinationOptions), IMemoryOwner zero-copy (v1).

Note on exactly-once: at-least-once + idempotent handlers gives effectively-once processing, and the design already ships the dedup primitives to support it — scheduler occurrence CAS (CreateIfAbsentAsync), producer DeduplicationId, and consumer dedup-by-key via the lock/CAS store. True once-per-side-effect still requires the dedup record to commit in the same transaction as the side effect (transactional outbox/inbox); a distributed lock alone can't close the crash window between performing a side effect and acking the message.

[ejsmith]

Pushed two commits addressing the messaging/jobs design-review feedback. All changes are green against the messaging, queue, and jobs suites (plus the conformance harness); full core suite passes.

fix: address messaging/jobs design-review feedback

Durable jobs + CRON

• Hosted runtime driver (JobRuntimeService + AddJobRuntimeService()): pumps occurrence materialization, due-dispatch, recovery, and queued-job execution. The runtime previously never ran end-to-end.
• Lease ownership enforced in TryTransitionAsync (expectedNodeId) so a stale worker can't overwrite the reclaiming node's terminal state.
• Lease renewal heartbeat during execution; the run is cancelled if the lease is lost (RenewClaimAsync was dead code).
• Strong node identity (machine:pid:token), capped exponential retry backoff.
• Replaced the hand-rolled cron parser with the vendored Cronos (moved into core Foundatio.Cronos); materialize every missed occurrence in the misfire window, not just the latest.

Messaging

• Resilient consumer/subscription loops — a poison message or transient receive error no longer silently kills the consumer.
• Fixed an in-memory push-path double-settle (tolerate already-settled receipts in the safety-net abandon).
• In-memory transport now honors visibility timeouts (reaper redelivers unsettled messages) so its advertised at-least-once guarantee is real.
• Log handler exceptions instead of swallowing them; drop the misleading write-only content-type header; add ReceiveDeadLetteredAsync so poison payloads are inspectable.

Conformance harness

• Silent capability skips → Assert.Skip; ordering assertions gated on the declared OrderingGuarantee; added visibility-timeout, competing-consumer, and DLQ-read scenarios.

refactor: hoist shared MessageQueue/PubSub behavior into MessageClientCore (M1, M2)

• Removed the large duplication between MessageQueue and PubSub via an internal MessageClientCore; unified the two handle classes and the two HandleMessageAsync overloads. Pub/sub subscriptions now also honor RedeliveryBackoff (prior drift).
• Enforce ITransportInfo.MaxBatchSize by chunking oversized sends.

Still intentionally not addressed (no external provider on this branch): proving the harness against a real transport before cutting the public API — that's the phased-rollout gate from the plan.