From 54014d0018df1ffc3c1911f34d0b8d988dd54af6 Mon Sep 17 00:00:00 2001 From: Glen Date: Thu, 25 Jun 2026 14:55:10 +0200 Subject: [PATCH 1/7] Add in-memory scheduled message store --- .../InMemoryScheduledMessageStore.cs | 113 ++++++++++++++++++ .../InMemoryScheduledMessageStoreTests.cs | 98 +++++++++++++++ 2 files changed, 211 insertions(+) create mode 100644 src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageStore.cs create mode 100644 src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemoryScheduledMessageStoreTests.cs diff --git a/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageStore.cs b/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageStore.cs new file mode 100644 index 00000000000..e361d4efefa --- /dev/null +++ b/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageStore.cs @@ -0,0 +1,113 @@ +using System.Diagnostics.CodeAnalysis; +using Mocha.Middlewares; +using Mocha.Scheduling; + +namespace Mocha.Transport.InMemory; + +/// +/// In-memory implementation of that holds scheduled message +/// envelopes in process memory until they are due for delivery. State is not durable: scheduled +/// messages are lost when the process stops. +/// +public sealed class InMemoryScheduledMessageStore(ISchedulerSignal signal) : IScheduledMessageStore +{ + private const string TokenPrefix = "in-memory-scheduler:"; +#if NET9_0_OR_GREATER + private readonly Lock _lock = new(); +#else + private readonly object _lock = new(); +#endif + private readonly Dictionary _entries = []; + + /// + public ValueTask PersistAsync( + MessageEnvelope envelope, + DateTimeOffset scheduledTime, + CancellationToken cancellationToken) + { + var id = Guid.NewGuid(); + + lock (_lock) + { + _entries[id] = new Entry(envelope, scheduledTime); + } + + signal.Notify(scheduledTime); + + return new ValueTask($"{TokenPrefix}{id:D}"); + } + + /// + public ValueTask CancelAsync(string token, CancellationToken cancellationToken) + { + var value = token.StartsWith(TokenPrefix, StringComparison.Ordinal) + ? token[TokenPrefix.Length..] + : token; + + if (!Guid.TryParse(value, out var id)) + { + return new ValueTask(false); + } + + lock (_lock) + { + return new ValueTask(_entries.Remove(id)); + } + } + + /// + /// Removes and returns the earliest entry whose scheduled time is at or before . + /// + public bool TryTakeDue(DateTimeOffset now, [NotNullWhen(true)] out MessageEnvelope? envelope) + { + lock (_lock) + { + var foundId = Guid.Empty; + var foundTime = DateTimeOffset.MaxValue; + MessageEnvelope? foundEnvelope = null; + + foreach (var (id, entry) in _entries) + { + if (entry.ScheduledTime <= now && entry.ScheduledTime < foundTime) + { + foundId = id; + foundTime = entry.ScheduledTime; + foundEnvelope = entry.Envelope; + } + } + + if (foundEnvelope is null) + { + envelope = null; + return false; + } + + _entries.Remove(foundId); + envelope = foundEnvelope; + return true; + } + } + + /// + /// Returns the earliest scheduled time across all pending entries, or null when empty. + /// + public DateTimeOffset? NextDueTime() + { + lock (_lock) + { + DateTimeOffset? next = null; + + foreach (var entry in _entries.Values) + { + if (next is null || entry.ScheduledTime < next) + { + next = entry.ScheduledTime; + } + } + + return next; + } + } + + private readonly record struct Entry(MessageEnvelope Envelope, DateTimeOffset ScheduledTime); +} diff --git a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemoryScheduledMessageStoreTests.cs b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemoryScheduledMessageStoreTests.cs new file mode 100644 index 00000000000..58716004b66 --- /dev/null +++ b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemoryScheduledMessageStoreTests.cs @@ -0,0 +1,98 @@ +using Mocha.Middlewares; +using Mocha.Scheduling; + +namespace Mocha.Transport.InMemory.Tests.Scheduling; + +public class InMemoryScheduledMessageStoreTests +{ + [Fact] + public async Task PersistAsync_Should_ReturnPrefixedToken_When_Called() + { + // arrange + var store = new InMemoryScheduledMessageStore(new NoopSignal()); + + // act + var token = await store.PersistAsync( + Envelope("a"), + DateTimeOffset.UtcNow.AddMinutes(1), + TestContext.Current.CancellationToken); + + // assert + Assert.StartsWith("in-memory-scheduler:", token); + } + + [Fact] + public async Task TryTakeDue_Should_ReturnEarliestDue_When_MultiplePending() + { + // arrange + var now = DateTimeOffset.UtcNow; + var store = new InMemoryScheduledMessageStore(new NoopSignal()); + await store.PersistAsync(Envelope("late"), now.AddMinutes(10), TestContext.Current.CancellationToken); + await store.PersistAsync(Envelope("early"), now.AddMinutes(1), TestContext.Current.CancellationToken); + + // act + var took = store.TryTakeDue(now.AddMinutes(2), out var envelope); + + // assert + Assert.True(took); + Assert.NotNull(envelope); + Assert.Equal("early", envelope.MessageId); + } + + [Fact] + public async Task TryTakeDue_Should_ReturnFalse_When_NothingDue() + { + // arrange + var now = DateTimeOffset.UtcNow; + var store = new InMemoryScheduledMessageStore(new NoopSignal()); + await store.PersistAsync(Envelope("future"), now.AddMinutes(10), TestContext.Current.CancellationToken); + + // act + var took = store.TryTakeDue(now, out _); + + // assert + Assert.False(took); + } + + [Fact] + public async Task CancelAsync_Should_RemoveEntry_When_TokenValid() + { + // arrange + var now = DateTimeOffset.UtcNow; + var store = new InMemoryScheduledMessageStore(new NoopSignal()); + var token = await store.PersistAsync(Envelope("x"), now.AddMinutes(1), TestContext.Current.CancellationToken); + + // act + var cancelled = await store.CancelAsync(token, TestContext.Current.CancellationToken); + + // assert + Assert.True(cancelled); + Assert.False(store.TryTakeDue(now.AddMinutes(2), out _)); + } + + [Fact] + public async Task CancelAsync_Should_ReturnFalse_When_TokenUnknown() + { + // arrange + var store = new InMemoryScheduledMessageStore(new NoopSignal()); + + // act + var cancelled = await store.CancelAsync( + "in-memory-scheduler:00000000-0000-0000-0000-000000000000", + TestContext.Current.CancellationToken); + + // assert + Assert.False(cancelled); + } + + private static MessageEnvelope Envelope(string id) + => new() { MessageId = id, MessageType = "urn:test", DestinationAddress = "memory://test" }; + + private sealed class NoopSignal : ISchedulerSignal + { + public void Notify(DateTimeOffset scheduledTime) { } + + public Task WaitUntilAsync(DateTimeOffset wakeTime, CancellationToken cancellationToken) + => Task.CompletedTask; + } +} From 79d166dfbb99bd7489d92551928bc9bf42cce02e Mon Sep 17 00:00:00 2001 From: Glen Date: Thu, 25 Jun 2026 16:20:52 +0200 Subject: [PATCH 2/7] Deliver scheduled messages on the in-memory transport --- .../MessageBusBuilderExtensions.cs | 23 ++- .../InMemoryScheduledMessageWorker.cs | 148 ++++++++++++++++++ .../Helpers/MessageBusBuilder.cs | 6 + .../Mocha.Transport.InMemory.Tests.csproj | 4 + .../Scheduling/InMemorySchedulingTests.cs | 104 ++++++++++++ 5 files changed, 284 insertions(+), 1 deletion(-) create mode 100644 src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageWorker.cs create mode 100644 src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemorySchedulingTests.cs diff --git a/src/Mocha/src/Mocha.Transport.InMemory/MessageBusBuilderExtensions.cs b/src/Mocha/src/Mocha.Transport.InMemory/MessageBusBuilderExtensions.cs index 4a31248c25b..60ca8b57918 100644 --- a/src/Mocha/src/Mocha.Transport.InMemory/MessageBusBuilderExtensions.cs +++ b/src/Mocha/src/Mocha.Transport.InMemory/MessageBusBuilderExtensions.cs @@ -1,4 +1,9 @@ -namespace Mocha.Transport.InMemory; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Logging; +using Mocha.Scheduling; + +namespace Mocha.Transport.InMemory; /// /// Extension methods for registering the in-memory messaging transport on an . @@ -19,6 +24,22 @@ public static IMessageBusHostBuilder AddInMemory( busBuilder.ConfigureMessageBus(b => b.AddTransport(transport)); + busBuilder.Services.TryAddSingleton(); + busBuilder.Services.TryAddSingleton( + sp => sp.GetRequiredService()); + busBuilder.Services.TryAddSingleton(sp => new InMemoryScheduledMessageWorker( + sp, + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetService() ?? TimeProvider.System, + sp.GetRequiredService>())); + busBuilder.Services.AddHostedService( + sp => sp.GetRequiredService()); + + busBuilder.UseSchedulerCore(); + return busBuilder; } diff --git a/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageWorker.cs b/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageWorker.cs new file mode 100644 index 00000000000..0d926cdc50b --- /dev/null +++ b/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageWorker.cs @@ -0,0 +1,148 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Mocha.Middlewares; +using Mocha.Scheduling; +using Mocha.Threading; + +namespace Mocha.Transport.InMemory; + +/// +/// Background service that delivers due in-memory scheduled messages. It sleeps on the scheduler +/// signal until the next message is due, then dispatches it through the normal endpoint pipeline. +/// Delivery is best-effort: a message that fails to dispatch is logged and dropped. +/// +internal sealed class InMemoryScheduledMessageWorker( + IServiceProvider services, + IMessagingRuntime runtime, + IMessagingPools pools, + ISchedulerSignal signal, + InMemoryScheduledMessageStore store, + TimeProvider timeProvider, + ILogger logger) : IHostedService, IAsyncDisposable +{ +#if NET9_0_OR_GREATER + private readonly Lock _lock = new(); +#else + private readonly object _lock = new(); +#endif + private ContinuousTask? _task; + + public Task StartAsync(CancellationToken cancellationToken) + { + lock (_lock) + { + _task ??= new ContinuousTask(ProcessAsync); + } + + return Task.CompletedTask; + } + + public async Task StopAsync(CancellationToken cancellationToken) + { + ContinuousTask? task; + + lock (_lock) + { + task = _task; + _task = null; + } + + if (task is not null) + { + await task.DisposeAsync(); + } + } + + public async ValueTask DisposeAsync() + { + await StopAsync(CancellationToken.None); + } + + private async Task ProcessAsync(CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + if (store.TryTakeDue(timeProvider.GetUtcNow(), out var envelope)) + { + await DispatchAsync(envelope, cancellationToken); + continue; + } + + await signal.WaitUntilAsync(store.NextDueTime() ?? DateTimeOffset.MaxValue, cancellationToken); + } + } + + private async Task DispatchAsync(MessageEnvelope envelope, CancellationToken cancellationToken) + { + var messageType = runtime.GetMessageType(envelope.MessageType); + var isReply = envelope.Headers?.IsReply() ?? false; + var endpoint = isReply + ? GetReplyEndpoint(envelope.DestinationAddress) + : GetSendEndpoint(envelope.DestinationAddress); + + if (messageType is null || endpoint is null) + { + logger.CouldNotDispatchScheduledMessage(envelope.MessageId); + return; + } + + var context = pools.DispatchContext.Get(); + try + { + await using var scope = services.CreateAsyncScope(); + context.Initialize(scope.ServiceProvider, endpoint, runtime, messageType, cancellationToken); + context.SkipScheduler(); + context.Envelope = envelope; + await endpoint.ExecuteAsync(context); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.ScheduledMessageDispatchFailed(envelope.MessageId, ex); + } + finally + { + pools.DispatchContext.Return(context); + } + } + + private DispatchEndpoint? GetSendEndpoint(string? address) + { + try + { + return Uri.TryCreate(address, UriKind.Absolute, out var uri) + ? runtime.GetDispatchEndpoint(uri) + : null; + } + catch + { + return null; + } + } + + private DispatchEndpoint? GetReplyEndpoint(string? address) + { + try + { + return Uri.TryCreate(address, UriKind.Absolute, out var uri) + ? runtime.GetTransport(uri)?.ReplyDispatchEndpoint + : null; + } + catch + { + return null; + } + } +} + +internal static partial class InMemoryScheduledMessageWorkerLog +{ + [LoggerMessage(1, LogLevel.Warning, "Could not dispatch scheduled message {MessageId}")] + public static partial void CouldNotDispatchScheduledMessage(this ILogger logger, string? messageId); + + [LoggerMessage(2, LogLevel.Error, "Scheduled message {MessageId} dispatch failed")] + public static partial void ScheduledMessageDispatchFailed( + this ILogger logger, + string? messageId, + Exception exception); +} diff --git a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Helpers/MessageBusBuilder.cs b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Helpers/MessageBusBuilder.cs index 51940aa4ba9..89e8fea2c66 100644 --- a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Helpers/MessageBusBuilder.cs +++ b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Helpers/MessageBusBuilder.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; namespace Mocha.Transport.InMemory.Tests.Helpers; @@ -9,6 +10,11 @@ public static async Task BuildServiceProvider(this IMessageBusH var provider = builder.Services.BuildServiceProvider(); var runtime = (MessagingRuntime)provider.GetRequiredService(); await runtime.StartAsync(CancellationToken.None); + foreach (var svc in provider.GetServices()) + { + await svc.StartAsync(CancellationToken.None); + } + return provider; } diff --git a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Mocha.Transport.InMemory.Tests.csproj b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Mocha.Transport.InMemory.Tests.csproj index ac0279a43ef..69b828d5240 100644 --- a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Mocha.Transport.InMemory.Tests.csproj +++ b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Mocha.Transport.InMemory.Tests.csproj @@ -10,6 +10,10 @@ + + + + diff --git a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemorySchedulingTests.cs b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemorySchedulingTests.cs new file mode 100644 index 00000000000..d7379960775 --- /dev/null +++ b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemorySchedulingTests.cs @@ -0,0 +1,104 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Time.Testing; + +namespace Mocha.Transport.InMemory.Tests.Scheduling; + +public class InMemorySchedulingTests +{ + [Fact] + public async Task ScheduledMessage_Should_NotDeliver_Before_DueTime() + { + // arrange + var ct = TestContext.Current.CancellationToken; + var (provider, time, recorder) = await StartAsync(ct); + await using var _ = provider; + using var scope = provider.CreateScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + + // act - schedule 1 minute out, advance only 30 seconds + await bus.SchedulePublishAsync(new Ping(), time.GetUtcNow().AddMinutes(1), ct); + time.Advance(TimeSpan.FromSeconds(30)); + + // assert - not delivered within a short grace period + Assert.False(await recorder.Signal.WaitAsync(TimeSpan.FromMilliseconds(300), ct)); + } + + [Fact] + public async Task ScheduledMessage_Should_Deliver_After_DueTime() + { + // arrange + var ct = TestContext.Current.CancellationToken; + var (provider, time, recorder) = await StartAsync(ct); + await using var _ = provider; + using var scope = provider.CreateScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + + // act + await bus.SchedulePublishAsync(new Ping(), time.GetUtcNow().AddMinutes(1), ct); + time.Advance(TimeSpan.FromMinutes(2)); + + // assert + Assert.True(await recorder.Signal.WaitAsync(TimeSpan.FromSeconds(10), ct)); + } + + [Fact] + public async Task ScheduledMessage_Should_NotDeliver_When_Cancelled() + { + // arrange + var ct = TestContext.Current.CancellationToken; + var (provider, time, recorder) = await StartAsync(ct); + await using var _ = provider; + using var scope = provider.CreateScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + + // act + var result = await bus.SchedulePublishAsync(new Ping(), time.GetUtcNow().AddMinutes(1), ct); + Assert.NotNull(result.Token); + await bus.CancelScheduledMessageAsync(result.Token, ct); + time.Advance(TimeSpan.FromMinutes(2)); + + // assert + Assert.False(await recorder.Signal.WaitAsync(TimeSpan.FromMilliseconds(300), ct)); + } + + private static async Task<(ServiceProvider provider, FakeTimeProvider time, PingRecorder recorder)> StartAsync( + CancellationToken cancellationToken) + { + var time = new FakeTimeProvider(DateTimeOffset.UtcNow); + var recorder = new PingRecorder(); + + var services = new ServiceCollection(); + services.AddSingleton(time); + services.AddSingleton(recorder); + services.AddMessageBus() + .AddEventHandler() + .AddInMemory(); + + var provider = services.BuildServiceProvider(); + var runtime = (MessagingRuntime)provider.GetRequiredService(); + await runtime.StartAsync(cancellationToken); + foreach (var svc in provider.GetServices()) + { + await svc.StartAsync(cancellationToken); + } + + return (provider, time, recorder); + } + + private sealed record Ping; + + private sealed class PingRecorder + { + public SemaphoreSlim Signal { get; } = new(0); + } + + private sealed class PingHandler(PingRecorder recorder) : IEventHandler + { + public ValueTask HandleAsync(Ping message, CancellationToken cancellationToken) + { + recorder.Signal.Release(); + return default; + } + } +} From 2efb47a86646928fc7732ec2b618b1b7fbe8d6c1 Mon Sep 17 00:00:00 2001 From: Glen Date: Thu, 25 Jun 2026 16:33:21 +0200 Subject: [PATCH 3/7] Start hosted services in saga integration test harness --- .../Mocha.Sagas.Tests/IntegrationTests.cs | 28 +++++++++++++++++-- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs index cdc0396bd15..04ddd4932c4 100644 --- a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs +++ b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs @@ -1,5 +1,6 @@ using System.Collections.Concurrent; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; using Mocha.Transport.InMemory; namespace Mocha.Sagas.Tests; @@ -19,6 +20,12 @@ private static async Task CreateBusAsync(Action(); await runtime.StartAsync(CancellationToken.None); + + foreach (var hostedService in provider.GetServices()) + { + await hostedService.StartAsync(CancellationToken.None); + } + return provider; } @@ -123,13 +130,28 @@ public async Task Saga_Should_TimeoutWithCustomResponse() var recorded = Assert.Single(recorder.Messages); var sagaId = Assert.IsType(recorded).CorrelationId!.Value; + // wait until the saga instance is persisted before sending the timeout, so the + // SagaTimedOutEvent is applied to the stored instance + var runtime = provider.GetRequiredService(); + var sagaName = runtime.Naming.GetSagaName(typeof(TimeoutWithResponseSaga)); + var persistDeadline = DateTime.UtcNow + s_timeout; + while (storage.Load(sagaName, sagaId) is null && DateTime.UtcNow < persistDeadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } + + Assert.NotNull(storage.Load(sagaName, sagaId)); + // simulate timeout by sending SagaTimedOutEvent with the saga ID await bus.SendAsync(new SagaTimedOutEvent(sagaId), CancellationToken.None); - // allow time for final state processing - await Task.Delay(500, TestContext.Current.CancellationToken); + // wait for the saga to reach its final state and be deleted from the store + var deadline = DateTime.UtcNow + s_timeout; + while (storage.Load(sagaName, sagaId) is not null && DateTime.UtcNow < deadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } - // assert - saga should be deleted from store after reaching final state Assert.Equal(0, storage.Count); } From 69d79bb268c0024cd3ae1db76d3315e07d707d3f Mon Sep 17 00:00:00 2001 From: Glen Date: Thu, 25 Jun 2026 16:57:17 +0200 Subject: [PATCH 4/7] Document in-memory transport scheduling and cancellation --- website/src/docs/mocha/v16/sagas.md | 24 +++++++++---------- website/src/docs/mocha/v16/scheduling.md | 8 +++---- .../docs/mocha/v16/transports/in-memory.md | 8 +++++++ 3 files changed, 24 insertions(+), 16 deletions(-) diff --git a/website/src/docs/mocha/v16/sagas.md b/website/src/docs/mocha/v16/sagas.md index 938355d5453..98874954444 100644 --- a/website/src/docs/mocha/v16/sagas.md +++ b/website/src/docs/mocha/v16/sagas.md @@ -623,7 +623,7 @@ A saga that waits for a message that never arrives will stay in its current stat Mocha provides a saga-level `Timeout()` API that sets a single deadline for the entire saga instance. The timeout is scheduled when the saga is created and automatically cancelled when the saga reaches any final state. -> **Prerequisites:** Durable, cancellable timeouts require a scheduling store. Configure `UsePostgresScheduling()` before using `Timeout()` — see [Scheduling: Set up store-based scheduling](/docs/mocha/v16/scheduling#set-up-store-based-scheduling-for-rabbitmq) for setup instructions. Native transport scheduling (InMemory, PostgreSQL) also works but does not support automatic cancellation. +> **Prerequisites:** Durable, cancellable timeouts require a scheduling store. Configure `UsePostgresScheduling()` before using `Timeout()` — see [Scheduling: Set up store-based scheduling](/docs/mocha/v16/scheduling#set-up-store-based-scheduling-for-rabbitmq) for setup instructions. The in-memory transport supports scheduling by default with no extra setup, including cancellation of pending timeouts when the saga reaches a final state. For timeouts that must survive a process restart, use `UsePostgresScheduling()`. ## Configure a saga-level timeout @@ -666,21 +666,21 @@ Use `OnTimeout()` on a state descriptor to define what happens when the timeout ## Key behaviors -| Behavior | Detail | -| ------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Scope | Per-saga instance, not per-state. The deadline covers the entire saga lifetime. | -| Duration | Fixed at configuration time via `TimeSpan`. | -| Auto-cancellation | When the saga reaches any final state (normal completion, error final state, etc.), the pending timeout is cancelled. | -| Late delivery | If the timeout fires after the saga was already deleted, the event is silently dropped. | -| Missing handler | If no `OnTimeout()` handler is configured for the current state, the saga throws an execution error. See [troubleshooting](#timeout-troubleshooting) below. | -| Recommended pattern | `DuringAny().OnTimeout()` handles the timeout regardless of which state the saga is in. | -| Response on timeout | Chain `.Respond()` on the timed-out final state to send a response back to the original requester. | -| Scheduling store | Requires a scheduling store for durable timeouts. Configure `UsePostgresScheduling()` — see [Scheduling](/docs/mocha/v16/scheduling#set-up-store-based-scheduling-for-rabbitmq) for setup. Native transport scheduling (InMemory, PostgreSQL) also works but does not support cancellation. | +| Behavior | Detail | +| ------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Scope | Per-saga instance, not per-state. The deadline covers the entire saga lifetime. | +| Duration | Fixed at configuration time via `TimeSpan`. | +| Auto-cancellation | When the saga reaches any final state (normal completion, error final state, etc.), the pending timeout is cancelled. | +| Late delivery | If the timeout fires after the saga was already deleted, the event is silently dropped. | +| Missing handler | If no `OnTimeout()` handler is configured for the current state, the saga throws an execution error. See [troubleshooting](#timeout-troubleshooting) below. | +| Recommended pattern | `DuringAny().OnTimeout()` handles the timeout regardless of which state the saga is in. | +| Response on timeout | Chain `.Respond()` on the timed-out final state to send a response back to the original requester. | +| Scheduling store | The in-memory transport supports scheduling, including cancellation, by default with no extra setup. For durable timeouts that survive a process restart, configure `UsePostgresScheduling()` (see [Scheduling](/docs/mocha/v16/scheduling#set-up-store-based-scheduling-for-rabbitmq)). | ## Timeout troubleshooting **Timeout never fires.** -Verify that a scheduling provider is configured. For durable, cancellable timeouts, register `UsePostgresScheduling()` with an EF Core DbContext. Native transport scheduling (InMemory, PostgreSQL) works without extra setup but does not cancel timeouts when the saga completes normally. +Verify that a scheduling provider is configured. The in-memory transport supports scheduling and cancellation by default with no extra setup. For durable timeouts that survive a process restart, register `UsePostgresScheduling()` with an EF Core DbContext. **"SagaExecutionException: No transition defined for SagaTimedOutEvent."** You configured `Timeout()` but did not add an `OnTimeout()` handler. Add a catch-all handler: diff --git a/website/src/docs/mocha/v16/scheduling.md b/website/src/docs/mocha/v16/scheduling.md index 10f6e8c24a7..ec7ad4f84af 100644 --- a/website/src/docs/mocha/v16/scheduling.md +++ b/website/src/docs/mocha/v16/scheduling.md @@ -131,7 +131,7 @@ if (result.IsCancellable) | `ScheduledTime` | `DateTimeOffset` | The time at which the message is scheduled for delivery. | | `IsCancellable` | `bool` | `true` when the scheduling infrastructure supports cancellation and a token was assigned. | -`IsCancellable` is `true` when a store-based scheduling provider (like Postgres) is registered. If no store is registered, the message is still scheduled (through the transport's native scheduling), but cancellation is not available. +`IsCancellable` is `true` when a store-based scheduling provider (like Postgres) is registered, or when using the in-memory transport (which supports cancellation by default). If no store is registered and the transport does not support cancellation natively, `IsCancellable` is `false`. ## Real-world example: cancellable reminder @@ -232,11 +232,11 @@ Each transport handles scheduling differently. Mocha adapts automatically based | Transport | Scheduling type | Durability | Cancellation support | Setup required | | ---------- | ------------------------------------- | ---------------------------- | ------------------------------------ | ----------------------------------------- | -| InMemory | Native (in-process scheduler) | Non-durable, lost on restart | No | None | +| InMemory | Native (in-process scheduler) | Non-durable, lost on restart | Yes | None | | PostgreSQL | Native (scheduled_time column) | Durable, survives restarts | No | None | | RabbitMQ | Store-based (via Postgres middleware) | Durable with Postgres store | Yes (with `UsePostgresScheduling()`) | `UsePostgresScheduling()` + EF Core model | -**InMemory:** The transport schedules messages natively using an internal scheduler. Messages scheduled for a time in the past are delivered immediately. Scheduled messages are lost if the process restarts. Cancellation is not supported. +**InMemory:** The transport schedules messages natively using an internal scheduler. Messages scheduled for a time in the past are delivered immediately. Scheduled messages are lost if the process restarts. Cancellation is supported: a pending scheduled message can be cancelled before it is dispatched. **PostgreSQL:** The transport handles scheduling natively. When you set `ScheduledTime`, the transport writes a `scheduled_time` column alongside the message. Messages are only delivered to consumers after the scheduled time has passed. No additional setup is required beyond the standard [PostgreSQL transport configuration](/docs/mocha/v16/transports/postgres). Cancellation is not supported with native scheduling. @@ -327,7 +327,7 @@ This does not happen. The dispatcher uses row-level locking to ensure each messa The message was already dispatched before the cancellation request reached the store. Once the background worker picks up a message and delivers it, the row is deleted and cancellation is no longer possible. If you need a wider cancellation window, schedule messages further in the future or check `SchedulingResult.IsCancellable` to confirm the infrastructure supports cancellation. **`SchedulingResult.IsCancellable` is false.** -No store-based scheduling provider is registered. Cancellation requires a provider like `UsePostgresScheduling()` that persists messages to a store. Transports with native scheduling (InMemory, PostgreSQL) do not support cancellation. If you need cancellation support, configure `UsePostgresScheduling()` with an EF Core DbContext. +No store-based scheduling provider is registered, and the active transport does not support native cancellation. The in-memory transport supports cancellation by default; the PostgreSQL transport's native scheduling does not. If you need cancellation with native PostgreSQL scheduling, configure `UsePostgresScheduling()` with an EF Core DbContext. # Next steps diff --git a/website/src/docs/mocha/v16/transports/in-memory.md b/website/src/docs/mocha/v16/transports/in-memory.md index c51a470ab58..c69279fefae 100644 --- a/website/src/docs/mocha/v16/transports/in-memory.md +++ b/website/src/docs/mocha/v16/transports/in-memory.md @@ -164,6 +164,14 @@ builder.Services // AuditHandler → InMemory (claimed) ``` +# Scheduled messages + +The in-memory transport schedules messages out of the box, with no extra setup. A message dispatched with a future `ScheduledTime` (for example via `ScheduleSendAsync`, `SchedulePublishAsync`, or a saga's `Timeout()`) is held in process and delivered once its time arrives. A message scheduled for a time in the past is delivered immediately. + +Scheduled messages are held in memory only, so they are lost if the process restarts. For scheduled work or timeouts that must survive a restart, use a store-based provider such as `UsePostgresScheduling()`. + +Cancellation is supported: a pending scheduled message can be cancelled before it is dispatched (for example, a saga's timeout is cancelled automatically when the saga reaches a final state). + # Next steps - [RabbitMQ Transport](/docs/mocha/v16/transports/rabbitmq) - Configure the RabbitMQ transport for production deployments. From 4c5a8eee29846efe4330781eea56730ab18d0517 Mon Sep 17 00:00:00 2001 From: Glen Date: Thu, 25 Jun 2026 17:09:39 +0200 Subject: [PATCH 5/7] Document why in-memory scheduler worker omits SkipOutbox --- .../Scheduling/InMemoryScheduledMessageWorker.cs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageWorker.cs b/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageWorker.cs index 0d926cdc50b..0eded6efcf2 100644 --- a/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageWorker.cs +++ b/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageWorker.cs @@ -93,6 +93,10 @@ private async Task DispatchAsync(MessageEnvelope envelope, CancellationToken can await using var scope = services.CreateAsyncScope(); context.Initialize(scope.ServiceProvider, endpoint, runtime, messageType, cancellationToken); context.SkipScheduler(); + + // Unlike the Postgres dispatcher, this worker does not call SkipOutbox(): the in-memory + // transport is not paired with an outbox, and the re-dispatched envelope carries no + // headers, so the outbox middleware passes through. Avoids coupling to Mocha.Outbox. context.Envelope = envelope; await endpoint.ExecuteAsync(context); } From 6a15d10e53789db6c3b7dcc3c52157cd3540a3ff Mon Sep 17 00:00:00 2001 From: Glen Date: Thu, 25 Jun 2026 17:57:21 +0200 Subject: [PATCH 6/7] Deep-copy scheduled message envelope in in-memory store --- .../InMemoryScheduledMessageStore.cs | 4 +- .../InMemoryScheduledMessageStoreTests.cs | 32 +++++++++ .../Scheduling/InMemorySchedulingTests.cs | 71 +++++++++++++++++++ 3 files changed, 106 insertions(+), 1 deletion(-) diff --git a/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageStore.cs b/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageStore.cs index e361d4efefa..807f2629e89 100644 --- a/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageStore.cs +++ b/src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageStore.cs @@ -27,9 +27,11 @@ public ValueTask PersistAsync( { var id = Guid.NewGuid(); + var stored = new MessageEnvelope(envelope) { Body = envelope.Body.ToArray() }; + lock (_lock) { - _entries[id] = new Entry(envelope, scheduledTime); + _entries[id] = new Entry(stored, scheduledTime); } signal.Notify(scheduledTime); diff --git a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemoryScheduledMessageStoreTests.cs b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemoryScheduledMessageStoreTests.cs index 58716004b66..73aee1fda60 100644 --- a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemoryScheduledMessageStoreTests.cs +++ b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemoryScheduledMessageStoreTests.cs @@ -85,6 +85,38 @@ public async Task CancelAsync_Should_ReturnFalse_When_TokenUnknown() Assert.False(cancelled); } + [Fact] + public async Task PersistAsync_Should_PreserveBodyAndHeaders_When_OriginalsMutatedAfterPersist() + { + // arrange + var now = DateTimeOffset.UtcNow; + var store = new InMemoryScheduledMessageStore(new NoopSignal()); + var body = new byte[] { 1, 2, 3 }; + var headers = new Headers(); + headers.Set("k", "v"); + var envelope = new MessageEnvelope + { + MessageType = "urn:test", + DestinationAddress = "memory://test", + Headers = headers, + Body = body + }; + + await store.PersistAsync(envelope, now.AddMinutes(1), TestContext.Current.CancellationToken); + + // act - mutate the originals after persisting + Array.Clear(body); + headers.Set("k", "changed"); + + // assert - stored copy is unaffected + var took = store.TryTakeDue(now.AddMinutes(2), out var taken); + Assert.True(took); + Assert.NotNull(taken); + Assert.Equal(new byte[] { 1, 2, 3 }, taken.Body.ToArray()); + Assert.True(taken.Headers!.TryGetValue("k", out var storedValue)); + Assert.Equal("v", storedValue); + } + private static MessageEnvelope Envelope(string id) => new() { MessageId = id, MessageType = "urn:test", DestinationAddress = "memory://test" }; diff --git a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemorySchedulingTests.cs b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemorySchedulingTests.cs index d7379960775..3930d8a18ff 100644 --- a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemorySchedulingTests.cs +++ b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemorySchedulingTests.cs @@ -62,6 +62,55 @@ public async Task ScheduledMessage_Should_NotDeliver_When_Cancelled() Assert.False(await recorder.Signal.WaitAsync(TimeSpan.FromMilliseconds(300), ct)); } + [Fact] + public async Task ScheduledMessage_Should_PreservePayload_When_PooledContextIsReusedBeforeDueTime() + { + // arrange + var ct = TestContext.Current.CancellationToken; + var time = new FakeTimeProvider(DateTimeOffset.UtcNow); + var greetingResult = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var noiseSent = new SemaphoreSlim(0); + + var services = new ServiceCollection(); + services.AddSingleton(time); + services.AddSingleton(greetingResult); + services.AddSingleton(noiseSent); + services.AddMessageBus() + .AddEventHandler() + .AddEventHandler() + .AddInMemory(); + + var provider = services.BuildServiceProvider(); + await using var _ = provider; + + var runtime = (MessagingRuntime)provider.GetRequiredService(); + await runtime.StartAsync(ct); + foreach (var svc in provider.GetServices()) + { + await svc.StartAsync(ct); + } + + using var scope = provider.CreateScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + + // act - schedule greeting 1 minute out + await bus.SchedulePublishAsync(new Greeting("hello"), time.GetUtcNow().AddMinutes(1), ct); + + // publish several noise messages to reuse the pooled dispatch context and overwrite its writer buffer + for (var i = 0; i < 5; i++) + { + await bus.PublishAsync(new Noise($"overwrite-{i}-padding-padding-padding-padding-padding-padding"), ct); + Assert.True(await noiseSent.WaitAsync(TimeSpan.FromSeconds(10), ct)); + } + + // advance time past the scheduled due time + time.Advance(TimeSpan.FromMinutes(2)); + + // assert - greeting delivered with original payload intact + var received = await greetingResult.Task.WaitAsync(TimeSpan.FromSeconds(30), ct); + Assert.Equal("hello", received); + } + private static async Task<(ServiceProvider provider, FakeTimeProvider time, PingRecorder recorder)> StartAsync( CancellationToken cancellationToken) { @@ -86,8 +135,30 @@ public async Task ScheduledMessage_Should_NotDeliver_When_Cancelled() return (provider, time, recorder); } + private sealed record Greeting(string Text); + + private sealed record Noise(string Content); + private sealed record Ping; + private sealed class GreetingHandler(TaskCompletionSource result) : IEventHandler + { + public ValueTask HandleAsync(Greeting message, CancellationToken cancellationToken) + { + result.TrySetResult(message.Text); + return default; + } + } + + private sealed class NoiseHandler(SemaphoreSlim noiseSent) : IEventHandler + { + public ValueTask HandleAsync(Noise message, CancellationToken cancellationToken) + { + noiseSent.Release(); + return default; + } + } + private sealed class PingRecorder { public SemaphoreSlim Signal { get; } = new(0); From 69f30fc8b2feb710836b8eeea8447e05caa5a9cc Mon Sep 17 00:00:00 2001 From: Glen Date: Fri, 26 Jun 2026 10:42:43 +0200 Subject: [PATCH 7/7] Make UseSchedulerCore idempotent and fix core scheduling tests --- ...chedulerCoreServiceCollectionExtensions.cs | 13 +++ .../SchedulingMiddlewareIntegrationTests.cs | 100 ++---------------- 2 files changed, 19 insertions(+), 94 deletions(-) diff --git a/src/Mocha/src/Mocha/Scheduling/SchedulerCoreServiceCollectionExtensions.cs b/src/Mocha/src/Mocha/Scheduling/SchedulerCoreServiceCollectionExtensions.cs index c31d22082a4..cc265efe4da 100644 --- a/src/Mocha/src/Mocha/Scheduling/SchedulerCoreServiceCollectionExtensions.cs +++ b/src/Mocha/src/Mocha/Scheduling/SchedulerCoreServiceCollectionExtensions.cs @@ -16,6 +16,7 @@ public static class SchedulerCoreServiceCollectionExtensions /// Adds as a singleton and configures the dispatch pipeline /// to persist outgoing messages with a /// through instead of dispatching them directly. + /// Calling this method more than once is safe: the dispatch middleware is installed at most once. /// /// The message bus host builder to configure. /// The same instance for chaining. @@ -24,8 +25,20 @@ public static IMessageBusHostBuilder UseSchedulerCore(this IMessageBusHostBuilde builder.Services.TryAddSingleton(sp => new MessageBusSchedulerSignal(sp.GetService() ?? TimeProvider.System)); + if (builder.Services.Any(d => d.ServiceType == typeof(SchedulerCoreMarker))) + { + return builder; + } + + builder.Services.AddSingleton(); builder.ConfigureMessageBus(x => x.UseDispatch(DispatchSchedulingMiddleware.Create(), after: "Serialization")); return builder; } } + +/// +/// Marker service used to detect whether +/// has already been called, preventing the scheduling dispatch middleware from being installed twice. +/// +internal sealed class SchedulerCoreMarker; diff --git a/src/Mocha/test/Mocha.Tests/Scheduling/SchedulingMiddlewareIntegrationTests.cs b/src/Mocha/test/Mocha.Tests/Scheduling/SchedulingMiddlewareIntegrationTests.cs index 4a64914dfd7..b60075004f5 100644 --- a/src/Mocha/test/Mocha.Tests/Scheduling/SchedulingMiddlewareIntegrationTests.cs +++ b/src/Mocha/test/Mocha.Tests/Scheduling/SchedulingMiddlewareIntegrationTests.cs @@ -226,78 +226,6 @@ public async Task CancelScheduledMessageAsync_Should_ReturnFalse_When_AlreadyCan Assert.False(secondCancel); } - [Fact] - public async Task SchedulePublishAsync_Should_ReturnNonCancellable_When_NoStoreRegistered() - { - // arrange - var timeProvider = new FakeTimeProvider(); - - var services = new ServiceCollection(); - services.AddSingleton(timeProvider); - - var builder = services.AddMessageBus(); - builder.UseSchedulerCore(); - builder.AddInMemory(); - - var provider = services.BuildServiceProvider(); - var runtime = (MessagingRuntime)provider.GetRequiredService(); - await runtime.StartAsync(CancellationToken.None); - - await using (provider) - { - using var scope = provider.CreateScope(); - var bus = scope.ServiceProvider.GetRequiredService(); - - var scheduledTime = timeProvider.GetUtcNow().AddMinutes(10); - - // act - var result = await bus.SchedulePublishAsync( - new SchedulingTestEvent { Payload = "no-store" }, - scheduledTime, - CancellationToken.None); - - // assert - Assert.False(result.IsCancellable); - Assert.Null(result.Token); - } - } - - [Fact] - public async Task Scheduling_Should_PassThrough_When_NoStoreRegistered() - { - // arrange - var timeProvider = new FakeTimeProvider(); - var recorder = new MessageRecorder(); - - var services = new ServiceCollection(); - services.AddSingleton(recorder); - services.AddSingleton(timeProvider); - - var builder = services.AddMessageBus(); - builder.UseSchedulerCore(); - builder.AddEventHandler(); - builder.AddInMemory(); - - var provider = services.BuildServiceProvider(); - var runtime = (MessagingRuntime)provider.GetRequiredService(); - await runtime.StartAsync(CancellationToken.None); - - await using (provider) - { - using var scope = provider.CreateScope(); - var bus = scope.ServiceProvider.GetRequiredService(); - - // act - publish with ScheduledTime but no store registered - await bus.PublishAsync( - new SchedulingTestEvent { Payload = "no-store" }, - new PublishOptions { ScheduledTime = timeProvider.GetUtcNow().AddSeconds(-1) }, - CancellationToken.None); - - // assert - message delivered to handler since middleware was skipped (no store) - Assert.True(await recorder.WaitAsync(s_timeout), "Message should be delivered to handler"); - } - } - private static async Task WaitUntilAsync(Func condition, TimeSpan timeout) { using var cts = new CancellationTokenSource(timeout); @@ -307,20 +235,6 @@ private static async Task WaitUntilAsync(Func condition, TimeSpan timeout) } } - /// - /// Creates a middleware configuration that always inserts the scheduling middleware, - /// bypassing the factory-time IScheduledMessageStore check (which requires bus-internal DI). - /// The middleware itself resolves the store at dispatch time from the scoped host DI. - /// - private static DispatchMiddlewareConfiguration CreateSchedulingMiddleware() - => new( - static (_, next) => - { - var middleware = new DispatchSchedulingMiddleware(); - return ctx => middleware.InvokeAsync(ctx, next); - }, - "Scheduling"); - private static async Task CreateBusWithSchedulingAsync( InMemoryScheduledMessageStore store, Action configure, @@ -337,11 +251,10 @@ private static async Task CreateBusWithSchedulingAsync( var builder = services.AddMessageBus(); - // Register middleware directly (bypassing factory-time DI check) - builder.ConfigureMessageBus(x => x.UseDispatch(CreateSchedulingMiddleware())); - - configure(builder); + // AddInMemory must come before configure so that the "Scheduling" middleware modifier + // is registered before any configure-supplied modifiers that reference it (e.g. before: "Scheduling"). builder.AddInMemory(); + configure(builder); var provider = services.BuildServiceProvider(); var runtime = (MessagingRuntime)provider.GetRequiredService(); @@ -365,11 +278,10 @@ private static async Task CreateBusWithSchedulingAndProviderAsy var builder = services.AddMessageBus(); - // Register middleware directly (bypassing factory-time DI check) - builder.ConfigureMessageBus(x => x.UseDispatch(CreateSchedulingMiddleware())); - - configure(builder); + // AddInMemory must come before configure so that the "Scheduling" middleware modifier + // is registered before any configure-supplied modifiers that reference it (e.g. before: "Scheduling"). builder.AddInMemory(); + configure(builder); var provider = services.BuildServiceProvider(); var runtime = (MessagingRuntime)provider.GetRequiredService();