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

.agents/skills/foundatio/SKILL.md

@@ -23,6 +23,19 @@ query-docs(libraryId="/foundatiofx/foundatio", query="How to configure queue ret
 
 Query with specific questions, not single keywords. All provider docs (Redis, Azure, AWS, Kafka, etc.) are included in the main library.
 
+
+## Messaging/Jobs Redesign Notes
+
+- New queue/pub-sub APIs live under `Foundatio.Messaging`: app-facing `IQueue`, `IPubSub`, shared `IReceivedMessage` / `IReceivedMessage<T>`, `QueueMessageOptions`, `QueueReceiveOptions`, `QueueConsumerOptions`, `PubSubMessageOptions`, and `PubSubSubscriptionOptions`.
+- Route resolution is centralized in `IMessageRouter`: operation override > explicit route map > interface/base-type map > `MessageRouteAttribute` > configured default/convention. Configure with `.Messaging.ConfigureRouting(...)`; use `UseDefaultQueue(...)`, `UseDefaultTopic(...)`, `MapQueue(...)`, and `MapTopic(...)` as the normal path. `Destination`, `Source`, `Topic`, and `Subscription` are advanced operation overrides.
+- Routing configuration also declares startup topology. `IMessageTopology.GetDeclarations()` returns configured queues/topics/subscriptions, `EnsureAsync()` creates them through `ISupportsProvisioning`, and `ValidateAsync()` checks that they already exist for apps without create permissions. Operation overrides are not included in topology declarations.
+- Pub/sub topic routing and subscription identity are separate. Topic answers where an event is published; subscription answers the logical service/consumer group. Same subscription across instances means competing consumers; different subscriptions on the same topic fan out. `PubSubSubscriptionOptions.Key` is only the local duplicate-listener key.
+- Raw/envelope paths make grouped/default routes usable: `queue.ReceiveAsync(...)`, `queue.StartConsumerAsync(Func<IReceivedMessage,...>)`, and `pubsub.SubscribeAsync(Func<IReceivedMessage,...>)`. Typed `ReceiveAsync<T>` / `SubscribeAsync<T>` remain the simple path.
+- Listener startup returns handles: `StartConsumerAsync<T>` returns `IMessageConsumer`; `SubscribeAsync<T>` returns `IMessageSubscription`. Same-key duplicate registrations are idempotent only for the same handler/options; conflicting registrations throw. Use `RunConsumerAsync` or `RunSubscriptionAsync` only for blocking lifetime loops.
+- Received-message settlement uses explicit verbs only: `CompleteAsync`, `AbandonAsync`, `DeadLetterAsync`, `RenewLockAsync`, and `ReportProgressAsync`. Unsupported capabilities should throw clearly instead of silently downgrading.
+- New durable job runtime roles are separated: `IJobClient` submits and returns `JobHandle`, `IJobMonitor` queries state, `IJobRuntimeStore` persists runtime state, and `IJobWorker` claims and executes queued jobs. Job types persist stable registry names via `IJobTypeRegistry` / `.Jobs.Register<TJob>(name)`, not assembly-qualified names.
+- In-memory setup for the redesign is `services.AddFoundatio().Messaging.ConfigureRouting(...).UseInMemory().Jobs.UseInMemoryRuntime()`.
+
 ## Core Interfaces
 
 | Interface | Purpose | In-Memory | Production |

src/Foundatio.Extensions.Hosting/Jobs/JobHostExtensions.cs

@@ -61,6 +61,13 @@ public static IServiceCollection AddJob(this IServiceCollection services, string
         return services.AddJob(jobOptionsBuilder.Target);
     }
 
+    /// <remarks>
+    /// Legacy/compat. This registers the in-process <see cref="ScheduledJobService"/>, which runs CRON occurrences
+    /// in-process and does not materialize durable, recoverable occurrences. The forward path in the redesigned runtime
+    /// is the durable scheduler — register it with <c>services.AddFoundatio().Jobs.UseInMemoryRuntime()</c> plus
+    /// <see cref="AddJobRuntimeService"/>, which materializes durable occurrences with retry, recovery, and
+    /// dead-lettering. Routing this default API onto the durable scheduler is a planned follow-up.
+    /// </remarks>
     public static IServiceCollection AddCronJob(this IServiceCollection services, ScheduledJobOptions jobOptions)
     {
         if (jobOptions.JobFactory == null)
@@ -180,6 +187,11 @@ public static IServiceCollection AddDistributedCronJob(this IServiceCollection s
         })));
     }
 
+    /// <remarks>
+    /// Legacy/compat: registers the in-process <see cref="ScheduledJobService"/> CRON scheduler. For durable,
+    /// recoverable CRON occurrences use the redesigned runtime (<c>AddFoundatio().Jobs.UseInMemoryRuntime()</c> +
+    /// <see cref="AddJobRuntimeService"/>) instead.
+    /// </remarks>
     public static IServiceCollection AddJobScheduler(this IServiceCollection services)
     {
         if (!services.Any(s => s.ServiceType == typeof(IHostedService) && s.ImplementationType == typeof(ScheduledJobService)))
@@ -194,6 +206,24 @@ public static IServiceCollection AddJobScheduler(this IServiceCollection service
         return services;
     }
 
+    /// <summary>
+    /// Registers the hosted pump that drives the durable job runtime (<see cref="Foundatio.Jobs.IJobRuntimeStore"/>):
+    /// materializing CRON occurrences, dispatching delayed/scheduled work, recovering stale occurrences, and running
+    /// jobs submitted via <see cref="Foundatio.Jobs.IJobClient"/>. Register the runtime store and job services first
+    /// (e.g. <c>services.AddFoundatio().Jobs.UseInMemoryRuntime()</c>).
+    /// </summary>
+    public static IServiceCollection AddJobRuntimeService(this IServiceCollection services, Action<JobRuntimeServiceOptions>? configure = null)
+    {
+        var options = new JobRuntimeServiceOptions();
+        configure?.Invoke(options);
+        services.AddSingleton(options);
+
+        if (!services.Any(s => s.ServiceType == typeof(IHostedService) && s.ImplementationType == typeof(JobRuntimeService)))
+            services.AddSingleton<IHostedService, JobRuntimeService>();
+
+        return services;
+    }
+
     public static IServiceCollection AddJobLifetimeService(this IServiceCollection services)
     {
         services.AddSingleton<ShutdownHostIfNoJobsRunningService>();

src/Foundatio.Extensions.Hosting/Jobs/JobRuntimeService.cs

@@ -0,0 +1,103 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Foundatio.Jobs;
+using Foundatio.Utility;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace Foundatio.Extensions.Hosting.Jobs;
+
+/// <summary>
+/// Options controlling the cadence and batch size of <see cref="JobRuntimeService"/>.
+/// </summary>
+public class JobRuntimeServiceOptions
+{
+    /// <summary>
+    /// How often the runtime pump materializes CRON occurrences, dispatches due work, and runs queued jobs.
+    /// Defaults to one second so sub-minute CRON schedules and short delays are honored.
+    /// </summary>
+    public TimeSpan PollInterval { get; set; } = TimeSpan.FromSeconds(1);
+
+    /// <summary>
+    /// Maximum number of due dispatches and queued jobs claimed per pump iteration.
+    /// </summary>
+    public int BatchSize { get; set; } = 100;
+
+    /// <summary>
+    /// Maximum number of processing attempts for a durable job before a stale (lease-expired) instance is
+    /// dead-lettered instead of re-queued. Defaults to 3.
+    /// </summary>
+    public int MaxJobAttempts { get; set; } = 3;
+}
+
+/// <summary>
+/// Drives the durable job runtime introduced by <see cref="IJobRuntimeStore"/>. Without this hosted service nothing
+/// materializes CRON occurrences, dispatches delayed/scheduled work, recovers stale (lease-expired) occurrences, or
+/// runs jobs submitted through <see cref="IJobClient"/> — the runtime store would accumulate work that never executes.
+/// </summary>
+public class JobRuntimeService : BackgroundService
+{
+    private readonly JobScheduleProcessor _processor;
+    private readonly IJobWorker _worker;
+    private readonly TimeProvider _timeProvider;
+    private readonly ILogger _logger;
+    private readonly JobRuntimeServiceOptions _options;
+
+    public JobRuntimeService(JobScheduleProcessor processor, IJobWorker worker, TimeProvider? timeProvider = null, ILoggerFactory? loggerFactory = null, JobRuntimeServiceOptions? options = null)
+    {
+        _processor = processor ?? throw new ArgumentNullException(nameof(processor));
+        _worker = worker ?? throw new ArgumentNullException(nameof(worker));
+        _timeProvider = timeProvider ?? TimeProvider.System;
+        _logger = (loggerFactory ?? NullLoggerFactory.Instance).CreateLogger<JobRuntimeService>();
+        _options = options ?? new JobRuntimeServiceOptions();
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        _logger.LogInformation("Job runtime pump starting (poll interval {PollInterval}, batch size {BatchSize})", _options.PollInterval, _options.BatchSize);
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            try
+            {
+                var now = _timeProvider.GetUtcNow();
+
+                // Materialize CRON occurrences due within the misfire window (deduped, idempotent).
+                await _processor.EnqueueDueOccurrencesAsync(now, stoppingToken).AnyContext();
+
+                // Claim and run due dispatches: CRON occurrences plus delayed queue/pub-sub messages. This also
+                // recovers occurrences whose processing lease expired (crash mid-run) and applies retry/dead-letter.
+                await _processor.RunDueOccurrencesAsync(now, _options.BatchSize, lease: null, stoppingToken).AnyContext();
+
+                // Recover plain (non-CRON) jobs whose processing lease expired (a worker crash mid-run): re-queue them
+                // while attempts remain, otherwise dead-letter them. Without this they would strand in Processing.
+                await _worker.RecoverStaleAsync(_options.MaxJobAttempts, _options.BatchSize, stoppingToken).AnyContext();
+
+                // Run jobs submitted via IJobClient that are sitting in the Queued state.
+                await _worker.RunQueuedAsync(_options.BatchSize, stoppingToken).AnyContext();
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                break;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error pumping job runtime: {Message}", ex.Message);
+            }
+
+            try
+            {
+                await _timeProvider.Delay(_options.PollInterval, stoppingToken).AnyContext();
+            }
+            catch (OperationCanceledException)
+            {
+                break;
+            }
+        }
+
+        _logger.LogInformation("Job runtime pump stopped");
+    }
+}

src/Foundatio.Extensions.Hosting/Jobs/ScheduledJobInstance.cs

@@ -5,7 +5,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Foundatio.Caching;
-using Foundatio.Extensions.Hosting.Cronos;
+using Foundatio.Cronos;
 using Foundatio.Jobs;
 using Foundatio.Lock;
 using Foundatio.Messaging;

src/Foundatio.Extensions.Hosting/Jobs/ScheduledJobService.cs

@@ -13,6 +13,11 @@
 
 namespace Foundatio.Extensions.Hosting.Jobs;
 
+/// <summary>
+/// Legacy/compat in-process CRON scheduler used by <see cref="JobHostExtensions.AddCronJob(IServiceCollection, ScheduledJobOptions)"/>.
+/// It runs occurrences in-process and does not materialize durable, recoverable occurrences. The redesigned runtime's
+/// durable scheduler (<c>JobScheduleProcessor</c> driven by <see cref="JobRuntimeService"/>) is the forward path.
+/// </summary>
 public class ScheduledJobService : BackgroundService
 {
     private readonly IServiceProvider _serviceProvider;