Add scheduling support to the Mocha in-memory transport#9987
Draft
glen-84 wants to merge 9 commits into
Draft
Conversation
github-actions
Bot
added
📚 documentation
Contributor
There was a problem hiding this comment.
Pull request overview
Adds first-class scheduled delivery semantics to Mocha’s in-memory transport by wiring it into the shared scheduling middleware, backed by an in-process scheduled message store + hosted background worker, and updates docs/tests accordingly.
Changes:
- Register an in-memory
IScheduledMessageStore+IHostedServicedispatcher inAddInMemory()and enable scheduler core by default. - Add unit/integration coverage for due-time delivery, cancellation, and payload/header preservation under pooled context reuse.
- Update Mocha v16 docs (scheduling, sagas, in-memory transport) to reflect in-memory scheduling + cancellation support and non-durability.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| website/src/docs/mocha/v16/transports/in-memory.md | Documents in-memory scheduled delivery behavior, non-durability, and cancellation. |
| website/src/docs/mocha/v16/scheduling.md | Updates cancellation semantics and transport capability matrix for InMemory. |
| website/src/docs/mocha/v16/sagas.md | Updates saga Timeout() prerequisites/behavior notes for in-memory scheduling + cancellation. |
| src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemorySchedulingTests.cs | New end-to-end tests for scheduled delivery timing, cancellation, and payload preservation. |
| src/Mocha/test/Mocha.Transport.InMemory.Tests/Scheduling/InMemoryScheduledMessageStoreTests.cs | New unit tests for store token format, ordering, cancellation, and deep-copy behavior. |
| src/Mocha/test/Mocha.Transport.InMemory.Tests/Mocha.Transport.InMemory.Tests.csproj | Adds Microsoft.Extensions.TimeProvider.Testing dependency for FakeTimeProvider. |
| src/Mocha/test/Mocha.Transport.InMemory.Tests/Helpers/MessageBusBuilder.cs | Starts hosted services in the in-memory transport test harness so scheduling worker runs. |
| src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs | Starts hosted services and adjusts timeout test flow to better synchronize with persistence/finalization. |
| src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageWorker.cs | New hosted worker that wakes on scheduler signal and dispatches due scheduled envelopes. |
| src/Mocha/src/Mocha.Transport.InMemory/Scheduling/InMemoryScheduledMessageStore.cs | New in-memory scheduled message store that deep-copies envelopes and supports cancellation. |
| src/Mocha/src/Mocha.Transport.InMemory/MessageBusBuilderExtensions.cs | Wires scheduling store/worker into AddInMemory() and enables scheduler core. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+64
to
+73
| while (!cancellationToken.IsCancellationRequested) | ||
| { | ||
| if (store.TryTakeDue(timeProvider.GetUtcNow(), out var envelope)) | ||
| { | ||
| await DispatchAsync(envelope, cancellationToken); | ||
| continue; | ||
| } | ||
|
|
||
| await signal.WaitUntilAsync(store.NextDueTime() ?? DateTimeOffset.MaxValue, cancellationToken); | ||
| } |
Comment on lines
+63
to
+90
| 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; | ||
| } |
Comment on lines
+27
to
+42
| busBuilder.Services.TryAddSingleton<InMemoryScheduledMessageStore>(); | ||
| busBuilder.Services.TryAddSingleton<IScheduledMessageStore>( | ||
| sp => sp.GetRequiredService<InMemoryScheduledMessageStore>()); | ||
| busBuilder.Services.TryAddSingleton(sp => new InMemoryScheduledMessageWorker( | ||
| sp, | ||
| sp.GetRequiredService<IMessagingRuntime>(), | ||
| sp.GetRequiredService<IMessagingPools>(), | ||
| sp.GetRequiredService<ISchedulerSignal>(), | ||
| sp.GetRequiredService<InMemoryScheduledMessageStore>(), | ||
| sp.GetService<TimeProvider>() ?? TimeProvider.System, | ||
| sp.GetRequiredService<ILogger<InMemoryScheduledMessageWorker>>())); | ||
| busBuilder.Services.AddHostedService( | ||
| sp => sp.GetRequiredService<InMemoryScheduledMessageWorker>()); | ||
|
|
||
| busBuilder.UseSchedulerCore(); | ||
|
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
ScheduledTime(viaScheduleSendAsync/SchedulePublishAsync, a saga'sTimeout(), or delayed redelivery) is delivered at its scheduled time instead of immediately.IScheduledMessageStoreand a backgroundIHostedServiceworker, wired intoAddInMemorythrough Mocha's generic scheduling middleware (UseSchedulerCore). Cancellation is supported: a pending timeout is cancelled when a saga reaches a final state. Scheduled messages are held in process (non-durable); for durability useUsePostgresScheduling().Timeout()sagas are driven by real scheduling.Test plan
FakeTimeProvider): a scheduled message is delivered after its due time and not before, cancellation prevents delivery, and a scheduled message's body and headers are preserved until delivery.