diff --git a/Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxProcessor.cs b/Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxProcessor.cs index 649828c2..d3f7e63c 100644 --- a/Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxProcessor.cs +++ b/Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxProcessor.cs @@ -159,6 +159,15 @@ public async ValueTask PostInBackgroundAsync(CancellationToken cancellationToken return; } + // A timer wakeup can be deferred behind a delivered batch and then + // overwritten by acknowledgement's retry reconciliation. Retain the + // caller's request until a fresh dispatch snapshot includes its work. + drainRequested = true; + if (pendingAcknowledgement is not null) + { + return; + } + EnsureDispatchTimer(TimeSpan.Zero); } @@ -278,10 +287,18 @@ private async Task AcknowledgeInBackgroundAsync(CancellationToken cancellationTo try { + if (acknowledgement is null) + { + // A foreground post may have consumed this batch before its + // queued timer callback ran. Reconciliation here would replace + // that post's requested immediate drain with the retry delay. + return; + } + await RunAsActiveDrainAsync(async () => { await acknowledger.AcknowledgeAsync( - acknowledgement.GetValueOrDefault(), + acknowledgement.Value, cancellationToken); // Keep ownership visible while user callbacks run. An @@ -326,8 +343,6 @@ private async Task DrainOnceAsync(CancellationToken cancellationToken) { await RunAsActiveDrainAsync(async () => { - drainRequested = false; - // A foreground caller can acquire the gate between background // dispatch and its queued acknowledgement turn. Finish that batch // first so already-posted items are acknowledged before taking a @@ -384,6 +399,10 @@ private async Task>> P CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); + // Only taking a fresh snapshot consumes a caller's request. Delivery + // and acknowledgement can accept another request while awaiting work; + // an ordinary retry tick must never create one on their behalf. + drainRequested = false; var pending = GetPendingItems(); if (pending.IsDefaultOrEmpty) diff --git a/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/MessagingTestClusterFixture.cs b/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/MessagingTestClusterFixture.cs index 18189b0a..94e8c069 100644 --- a/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/MessagingTestClusterFixture.cs +++ b/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/MessagingTestClusterFixture.cs @@ -42,6 +42,7 @@ public async ValueTask InitializeAsync() AddStreamProviders(siloBuilder); siloBuilder.ConfigureServices(services => { + GatedOutboxTimerRegistry.Install(services); services.AddDefaultStateManager("Default"); services.AddDefaultStateManager("Payload"); services.AddSingleton(TimeProvider); diff --git a/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/GatedOutboxTimerRegistry.cs b/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/GatedOutboxTimerRegistry.cs new file mode 100644 index 00000000..a82f7112 --- /dev/null +++ b/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/GatedOutboxTimerRegistry.cs @@ -0,0 +1,97 @@ +using Orleans.Timers; + +namespace Egil.Orleans.Messaging.Tests.Outboxes; + +internal sealed class GatedOutboxTimerRegistry(ITimerRegistry timers) : ITimerRegistry +{ + public static void Install(IServiceCollection services) + { + var registration = services.Single(service => service.ServiceType == typeof(ITimerRegistry)); + services.Remove(registration); + services.AddSingleton(provider => new GatedOutboxTimerRegistry( + (ITimerRegistry)ActivatorUtilities.CreateInstance(provider, registration.ImplementationType!))); + } + + [Obsolete("Use RegisterGrainTimer instead.")] + public IDisposable RegisterTimer(IGrainContext grainContext, Func callback, object? state, TimeSpan dueTime, TimeSpan period) => + timers.RegisterTimer(grainContext, callback, state, dueTime, period); + + public IGrainTimer RegisterGrainTimer( + IGrainContext grainContext, + Func callback, + TState state, + GrainTimerCreationOptions options) + { + if (grainContext.GrainInstance is not OutboxDrainRequestGrain grain) + { + return timers.RegisterGrainTimer(grainContext, callback, state, options); + } + + var gate = grain.Gate; + // This test grain registers only the processor's two timers: dispatch + // repeats at RetryDelay, while acknowledgement is a single-shot timer. + // Decorate that runtime boundary without replacing either callback. + var acknowledgementTimer = options.Period == Timeout.InfiniteTimeSpan; + var timer = timers.RegisterGrainTimer(grainContext, async (timerState, cancellationToken) => + { + var requestedDispatchTick = !acknowledgementTimer && gate.DispatchTickRequested.Task.IsCompleted; + if (acknowledgementTimer && grain.HoldAcknowledgementTimer) + { + // Hold before the processor acquires its gate. The real Orleans + // dispatch timer can then hit the pending-batch defer branch, + // rather than merely wait behind an active acknowledgement. + gate.AcknowledgementTimerStarted.TrySetResult(); + await gate.AllowAcknowledgement.Task.WaitAsync(cancellationToken); + } + + await callback(timerState, cancellationToken); + if (acknowledgementTimer) + { + gate.AcknowledgementCallbackCompleted.TrySetResult(); + } + else if (requestedDispatchTick) + { + gate.RequestedDispatchCompleted.TrySetResult(); + } + }, state, options); + + if (!acknowledgementTimer) + { + gate.DispatchTimer = new GatedOutboxDispatchTimer(timer, options); + return gate.DispatchTimer; + } + + return timer; + } +} + +internal sealed class GatedOutboxDispatchTimer(IGrainTimer timer, GrainTimerCreationOptions options) : IGrainTimer +{ + private TimeSpan dueTime = options.DueTime; + private TimeSpan period = options.Period; + private bool paused; + + public void Change(TimeSpan dueTime, TimeSpan period) + { + this.dueTime = dueTime; + this.period = period; + timer.Change(paused ? Timeout.InfiniteTimeSpan : dueTime, period); + } + + public void Pause() + { + // Retain schedule changes while suppressing their runtime wakeup. This + // lets a queued stale acknowledgement finish before the follow-up timer + // fires, without changing the processor or invoking its private methods. + paused = true; + timer.Change(Timeout.InfiniteTimeSpan, period); + } + + public void Resume() + { + paused = false; + timer.Change(dueTime, period); + } + + public void Dispose() => timer.Dispose(); +} diff --git a/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxDrainRequestGrain.cs b/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxDrainRequestGrain.cs new file mode 100644 index 00000000..1bbfa003 --- /dev/null +++ b/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxDrainRequestGrain.cs @@ -0,0 +1,255 @@ +using System.Collections.Concurrent; +using System.Collections.Immutable; +using System.Diagnostics; + +namespace Egil.Orleans.Messaging.Tests.Outboxes; + +public enum OutboxDrainRequestHoldPhase +{ + None, + Delivery, + Acknowledgement, +} + +[GenerateSerializer] +public sealed record OutboxDrainRequestSettings +{ + [Id(0)] + public OutboxDrainRequestHoldPhase HoldPhase { get; init; } + + [Id(1)] + public TimeSpan RetryDelay { get; init; } = TimeSpan.FromMinutes(2); + + [Id(2)] + public bool FailFirstItem { get; init; } + + [Id(3)] + public bool AppendDuringSecondDelivery { get; init; } + + [Id(4)] + public bool HoldAcknowledgementTimer { get; init; } +} + +public interface IOutboxDrainRequestGrain : IGrainWithGuidKey +{ + Task InitializeAsync(OutboxDrainRequestSettings settings); + Task PublishAsync(string value, bool reminder); + Task PublishInForegroundAsync(string value); + Task RequestAsync(bool reminder); + Task RunQueuedDispatchTickAsync(); + Task PauseDispatchTimerAsync(); + Task ResumeDispatchTimerAsync(); + Task StopAsync(); + Task GetStateAsync(); +} + +public sealed class OutboxDrainRequestGrain( + [PersistentState("state", "Default")] IPersistentState state) + : Grain, IOutboxDrainRequestGrain, IOutboxGrain +{ + private OutboxProcessor processor = null!; + private OutboxDrainRequestGate gate = null!; + private OutboxDrainRequestSettings settings = null!; + private bool firstDelivery = true; + + internal OutboxDrainRequestGate Gate => gate; + internal bool HoldAcknowledgementTimer => settings.HoldPhase == OutboxDrainRequestHoldPhase.Acknowledgement || settings.HoldAcknowledgementTimer; + + public Task InitializeAsync(OutboxDrainRequestSettings settings) + { + this.settings = settings; + gate = OutboxDrainRequestGate.Get(this.GetPrimaryKey()); + processor = this.RegisterOutboxProcessor(() => state.State.Outbox ?? [], options => + { + options.AcknowledgePostedAsync = AcknowledgePostedAsync; + options.AcknowledgeFailuresAsync = AcknowledgeFailuresAsync; + options.RetryDelay = settings.RetryDelay; + // Holding an acknowledgement must still allow the test's second + // business write to enter the grain; delivery already interleaves. + options.InterleaveAcknowledgementCallbacks = HoldAcknowledgementTimer; + }).AddPostman(async (message, _, ct) => await DeliverAsync(message, ct)); + return Task.CompletedTask; + } + + public async Task PublishAsync(string value, bool reminder) + { + await AppendAsync(value); + await RequestAsync(reminder); + } + + public async Task PublishInForegroundAsync(string value) + { + await AppendAsync(value); + var post = processor.PostAsync().AsTask(); + gate.ForegroundStarted.TrySetResult(); + await post; + } + + public async Task RequestAsync(bool reminder) + { + if (reminder) + { + await ((IRemindable)this).ReceiveReminder(processor.ReminderName, default); + } + else + { + await processor.PostInBackgroundAsync(); + } + } + + public async Task StopAsync() + { + state.State.Outbox = EnsureOutbox().Clear(); + await state.WriteStateAsync(); + await processor.PostInBackgroundAsync(); + } + + public async Task RunQueuedDispatchTickAsync() + { + gate.DispatchTimer!.Change(TimeSpan.Zero, settings.RetryDelay); + gate.DispatchTickRequested.TrySetResult(); + await gate.RequestedDispatchCompleted.Task.WaitAsync(TimeSpan.FromSeconds(5)); + } + + public Task PauseDispatchTimerAsync() + { + gate.DispatchTimer!.Pause(); + return Task.CompletedTask; + } + + public Task ResumeDispatchTimerAsync() + { + gate.DispatchTimer!.Resume(); + return Task.CompletedTask; + } + + public Task GetStateAsync() => Task.FromResult(state.State); + + private async Task AppendAsync(string value) + { + state.State.Outbox = EnsureOutbox().Add(new OutboxProcessorTestEvent(value)); + await state.WriteStateAsync(); + } + + private async Task DeliverAsync(OutboxProcessorTestEvent message, CancellationToken cancellationToken) + { + if (firstDelivery) + { + firstDelivery = false; + gate.DispatchStarted.TrySetResult(); + if (settings.HoldPhase == OutboxDrainRequestHoldPhase.Delivery) + { + await gate.AllowDispatch.Task.WaitAsync(cancellationToken); + } + } + + if (message.Value == "A" && settings.FailFirstItem) + { + gate.RecordFailure(); + throw new InvalidOperationException("Leave A pending for retry."); + } + + if (message.Value == "B" && settings.AppendDuringSecondDelivery) + { + await AppendAsync("C"); + await processor.PostInBackgroundAsync(cancellationToken); + } + + gate.Delivered.Enqueue(message.Value); + if (message.Value == "B") + { + gate.SecondDelivered.TrySetResult(); + } + else if (message.Value == "C") + { + gate.ThirdDelivered.TrySetResult(); + } + } + + private async ValueTask AcknowledgePostedAsync( + ImmutableArray> items, + CancellationToken cancellationToken) + { + // Remove from the current outbox by ID so a business write made while + // the callback was held is retained for the next dispatch snapshot. + var outbox = EnsureOutbox(); + foreach (var item in items) + { + outbox = outbox.Remove(item.Id); + } + + state.State.Outbox = outbox; + state.State.AcknowledgedCount += items.Length; + await state.WriteStateAsync(cancellationToken); + } + + private async ValueTask AcknowledgeFailuresAsync( + ImmutableArray<(OutboxMessageEnvelope Item, Exception Error, int Attempt)> failures, + CancellationToken cancellationToken) + { + state.State.FailedCount += failures.Length; + await state.WriteStateAsync(cancellationToken); + } + + private Outbox EnsureOutbox() => + state.State.Outbox ??= Outbox.Create(); +} + +internal sealed class OutboxDrainRequestGate : IDisposable +{ + private static readonly ConcurrentDictionary Gates = new(); + private readonly Guid grainKey; + private int failureAttempts; + + private OutboxDrainRequestGate(Guid grainKey) => this.grainKey = grainKey; + + public TaskCompletionSource DispatchStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource AcknowledgementTimerStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource ForegroundStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource AllowDispatch { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource AllowAcknowledgement { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource SecondDelivered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource ThirdDelivered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource RequestedDispatchCompleted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource DispatchTickRequested { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource AcknowledgementCallbackCompleted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public GatedOutboxDispatchTimer? DispatchTimer { get; set; } + public TaskCompletionSource FirstFailure { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource SecondFailure { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource ThirdFailure { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public ConcurrentQueue Delivered { get; } = new(); + + public static OutboxDrainRequestGate Create(Guid grainKey) + { + var gate = new OutboxDrainRequestGate(grainKey); + if (!Gates.TryAdd(grainKey, gate)) + { + throw new InvalidOperationException("A gate already exists for this grain."); + } + + return gate; + } + + public static OutboxDrainRequestGate Get(Guid grainKey) => Gates[grainKey]; + + public void RecordFailure() + { + var signal = Interlocked.Increment(ref failureAttempts) switch + { + 1 => FirstFailure, + 2 => SecondFailure, + _ => ThirdFailure, + }; + signal.TrySetResult(Stopwatch.GetTimestamp()); + } + + public void Dispose() + { + // Release blocked callbacks even when an assertion or watchdog fails. + // Grains retain their gate reference until the cluster tears down. + AllowDispatch.TrySetResult(); + AllowAcknowledgement.TrySetResult(); + DispatchTimer?.Dispose(); + Gates.TryRemove(grainKey, out _); + } +} diff --git a/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxProcessorDrainRequestTests.cs b/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxProcessorDrainRequestTests.cs new file mode 100644 index 00000000..3eddddb5 --- /dev/null +++ b/Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxProcessorDrainRequestTests.cs @@ -0,0 +1,148 @@ +using System.Diagnostics; +using Egil.Orleans.Testing; + +namespace Egil.Orleans.Messaging.Tests.Outboxes; + +public sealed class OutboxProcessorDrainRequestTests(MessagingTestClusterFixture fixture) : IClassFixture +{ + [Theory] + [InlineData(OutboxDrainRequestHoldPhase.Delivery, false)] + [InlineData(OutboxDrainRequestHoldPhase.Delivery, true)] + [InlineData(OutboxDrainRequestHoldPhase.Acknowledgement, false)] + [InlineData(OutboxDrainRequestHoldPhase.Acknowledgement, true)] + public async Task Requests_during_delivery_or_acknowledgement_deliver_new_item_without_waiting_for_retry( + OutboxDrainRequestHoldPhase holdPhase, bool reminder) + { + var grainKey = Guid.NewGuid(); + using var gate = OutboxDrainRequestGate.Create(grainKey); + var grain = fixture.GrainFactory.GetGrain(grainKey); + await grain.InitializeAsync(new OutboxDrainRequestSettings + { + HoldPhase = holdPhase, + }); + + await grain.PublishAsync("A", reminder: false); + var heldPhase = holdPhase == OutboxDrainRequestHoldPhase.Delivery + ? gate.DispatchStarted.Task + : gate.AcknowledgementTimerStarted.Task; + await heldPhase.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + await grain.PublishAsync("B", reminder); + await grain.RequestAsync(reminder); + await grain.RequestAsync(reminder); + + var queuedTick = grain.RunQueuedDispatchTickAsync(); + await gate.DispatchTickRequested.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + gate.AllowDispatch.TrySetResult(); + await queuedTick.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + gate.AllowAcknowledgement.TrySetResult(); + + await gate.AcknowledgementCallbackCompleted.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + await gate.SecondDelivered.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + await fixture.WaitForAssertionAsync(grain, async () => + { + var state = await grain.GetStateAsync(); + Assert.Equal(2, state.AcknowledgedCount); + Assert.Equal(0, state.Outbox?.Count); + Assert.Equal(["A", "B"], gate.Delivered.ToArray()); + }, ct: TestContext.Current.CancellationToken); + } + + [Fact] + public async Task Failed_delivery_without_a_new_request_retries_at_the_configured_delay() + { + var grainKey = Guid.NewGuid(); + using var gate = OutboxDrainRequestGate.Create(grainKey); + var grain = fixture.GrainFactory.GetGrain(grainKey); + var retryDelay = TimeSpan.FromSeconds(1); + await grain.InitializeAsync(new OutboxDrainRequestSettings + { + FailFirstItem = true, + RetryDelay = retryDelay, + }); + + await grain.PublishAsync("A", reminder: false); + var firstAttempt = await gate.FirstFailure.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + var retry = await gate.SecondFailure.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + // Orleans owns the real timer. Allow a small timer-resolution margin, + // while rejecting a request flag that causes an immediate retry loop. + Assert.True(Stopwatch.GetElapsedTime(firstAttempt, retry) >= retryDelay - TimeSpan.FromMilliseconds(100)); + await fixture.WaitForAssertionAsync(grain, async () => + { + var state = await grain.GetStateAsync(); + Assert.Equal(2, state.FailedCount); + Assert.Equal(1, state.Outbox?.Count); + }, ct: TestContext.Current.CancellationToken); + await grain.StopAsync(); + } + + [Fact] + public async Task Explicit_request_during_a_failed_batch_allows_one_prompt_retry_then_returns_to_retry_delay() + { + var grainKey = Guid.NewGuid(); + using var gate = OutboxDrainRequestGate.Create(grainKey); + var grain = fixture.GrainFactory.GetGrain(grainKey); + var retryDelay = TimeSpan.FromSeconds(2); + await grain.InitializeAsync(new OutboxDrainRequestSettings + { + HoldPhase = OutboxDrainRequestHoldPhase.Acknowledgement, + FailFirstItem = true, + RetryDelay = retryDelay, + }); + + await grain.PublishAsync("A", reminder: false); + await gate.AcknowledgementTimerStarted.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + await grain.PublishAsync("B", reminder: false); + await grain.RunQueuedDispatchTickAsync(); + var acknowledgementReleased = Stopwatch.GetTimestamp(); + gate.AllowAcknowledgement.TrySetResult(); + + var promptRetry = await gate.SecondFailure.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + Assert.True(Stopwatch.GetElapsedTime(acknowledgementReleased, promptRetry) < retryDelay / 2); + await fixture.WaitForAssertionAsync(grain, async () => + { + var state = await grain.GetStateAsync(); + Assert.Equal(2, state.FailedCount); + Assert.Equal(2, state.Outbox?.Count); + }, ct: TestContext.Current.CancellationToken); + var scheduledRetry = await gate.ThirdFailure.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + Assert.True(Stopwatch.GetElapsedTime(promptRetry, scheduledRetry) >= retryDelay - TimeSpan.FromMilliseconds(100)); + await grain.StopAsync(); + } + + [Fact] + public async Task Stale_acknowledgement_after_foreground_takeover_preserves_requested_follow_up() + { + var grainKey = Guid.NewGuid(); + using var gate = OutboxDrainRequestGate.Create(grainKey); + var grain = fixture.GrainFactory.GetGrain(grainKey); + await grain.InitializeAsync(new OutboxDrainRequestSettings + { + HoldPhase = OutboxDrainRequestHoldPhase.Delivery, + HoldAcknowledgementTimer = true, + AppendDuringSecondDelivery = true, + }); + + await grain.PublishAsync("A", reminder: false); + await gate.DispatchStarted.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + await grain.PauseDispatchTimerAsync(); + var foreground = grain.PublishInForegroundAsync("B"); + await gate.ForegroundStarted.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + gate.AllowDispatch.TrySetResult(); + await foreground.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + await gate.AcknowledgementTimerStarted.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + gate.AllowAcknowledgement.TrySetResult(); + await gate.AcknowledgementCallbackCompleted.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + await grain.ResumeDispatchTimerAsync(); + + await gate.ThirdDelivered.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + await fixture.WaitForAssertionAsync(grain, async () => + { + var state = await grain.GetStateAsync(); + Assert.Equal(3, state.AcknowledgedCount); + Assert.Equal(0, state.Outbox?.Count); + Assert.Equal(["A", "B", "C"], gate.Delivered.ToArray()); + }, ct: TestContext.Current.CancellationToken); + } +}