diff --git a/docs/guide/messaging/transports/redis.md b/docs/guide/messaging/transports/redis.md index afdfd919d..215ad2e4e 100644 --- a/docs/guide/messaging/transports/redis.md +++ b/docs/guide/messaging/transports/redis.md @@ -360,6 +360,24 @@ public class OurRedisJsonMapper : EnvelopeMappersnippet source | anchor +## Durable Inbox + +`UseDurableInbox()` on a Redis stream listener means exactly what it means on every other transport: +each message is written to your configured message store (`PersistMessagesWithPostgresql()` etc.) **before** +its stream entry is acknowledged, is handled from there, and scheduled retries are parked in the inbox. A +process crash mid-handler therefore replays the message from the inbox instead of losing it. Because the +entry is acknowledged as soon as it is durable, the consumer group's pending list stays short regardless of +how long handlers take. + +::: warning +Before 6.30 the Redis stream endpoint was marked `IDatabaseBackedEndpoint`, which made a "durable" listener +skip the inbox write entirely and acknowledge the entry **on receipt** — effectively at-most-once on a crash, +and scheduled retries went to Redis's own scheduled set rather than the inbox. If you were running +`UseDurableInbox()` on Redis without a message store, that configuration now requires one, the same as for +RabbitMQ, Kafka or SQS. If what you actually wanted was Redis-native scheduled retries without a database, +use the default `BufferedInMemory()` (or `ProcessInline()`) listener: those schedule natively in Redis. +::: + ## Scheduled Messaging The Redis transport supports native Redis message scheduling for delayed or scheduled delivery. There's no configuration diff --git a/src/Transports/Redis/Wolverine.Redis.Tests/DatabaseBackedEndpointTests.cs b/src/Transports/Redis/Wolverine.Redis.Tests/DatabaseBackedEndpointTests.cs index aca089624..6f281e699 100644 --- a/src/Transports/Redis/Wolverine.Redis.Tests/DatabaseBackedEndpointTests.cs +++ b/src/Transports/Redis/Wolverine.Redis.Tests/DatabaseBackedEndpointTests.cs @@ -5,6 +5,7 @@ using StackExchange.Redis; using Wolverine.Configuration; using Wolverine.Redis.Internal; +using Wolverine.Transports; using Wolverine.Runtime; using Wolverine.Runtime.Serialization; using Wolverine.Util; @@ -22,7 +23,7 @@ public DatabaseBackedEndpointTests(ITestOutputHelper output) } [Fact] - public async Task redis_endpoint_implements_idatabase_backed_endpoint() + public async Task redis_endpoint_is_not_database_backed_and_its_listener_schedules_natively_when_not_durable() { var streamKey = $"dbe-test-{Guid.NewGuid():N}"; using var host = await Host.CreateDefaultBuilder() @@ -38,10 +39,13 @@ public async Task redis_endpoint_implements_idatabase_backed_endpoint() var transport = runtime.Options.Transports.GetOrCreate(); var endpoint = transport.StreamEndpoint(streamKey); - // Verify it implements IDatabaseBackedEndpoint - endpoint.ShouldBeAssignableTo(); + // GH-4028: IDatabaseBackedEndpoint made DurableReceiver skip the inbox INSERT and ACK on receipt, + // which is the opposite of what UseDurableInbox() promises. The Redis-native scheduled retry now + // lives on the listener, and only for the non-durable modes. + endpoint.ShouldNotBeAssignableTo(); + typeof(ISupportNativeScheduling).IsAssignableFrom(typeof(RedisStreamListener)).ShouldBeTrue(); - _output.WriteLine("✓ RedisStreamEndpoint implements IDatabaseBackedEndpoint"); + _output.WriteLine("✓ RedisStreamEndpoint is no longer IDatabaseBackedEndpoint; RedisStreamListener is ISupportNativeScheduling"); } [Fact] @@ -59,7 +63,7 @@ public async Task schedule_retry_should_add_message_to_scheduled_set() var runtime = host.Services.GetRequiredService(); var transport = runtime.Options.Transports.GetOrCreate(); - var endpoint = transport.StreamEndpoint(streamKey) as IDatabaseBackedEndpoint; + var endpoint = transport.StreamEndpoint(streamKey); var database = transport.GetDatabase(database: 0); var scheduledKey = transport.StreamEndpoint(streamKey).ScheduledMessagesKey; @@ -123,7 +127,7 @@ public async Task schedule_retry_without_scheduled_time_uses_default_delay() var runtime = host.Services.GetRequiredService(); var transport = runtime.Options.Transports.GetOrCreate(); - var endpoint = transport.StreamEndpoint(streamKey) as IDatabaseBackedEndpoint; + var endpoint = transport.StreamEndpoint(streamKey); var database = transport.GetDatabase(database: 0); var scheduledKey = transport.StreamEndpoint(streamKey).ScheduledMessagesKey; @@ -183,7 +187,7 @@ public async Task scheduled_retry_should_be_picked_up_by_polling() var runtime = host.Services.GetRequiredService(); var transport = runtime.Options.Transports.GetOrCreate(); - var endpoint = transport.StreamEndpoint(streamKey) as IDatabaseBackedEndpoint; + var endpoint = transport.StreamEndpoint(streamKey); var database = transport.GetDatabase(database: 0); var scheduledKey = transport.StreamEndpoint(streamKey).ScheduledMessagesKey; diff --git a/src/Transports/Redis/Wolverine.Redis.Tests/EndToEndRetryTests.cs b/src/Transports/Redis/Wolverine.Redis.Tests/EndToEndRetryTests.cs index 51cf5b49c..6863046c2 100644 --- a/src/Transports/Redis/Wolverine.Redis.Tests/EndToEndRetryTests.cs +++ b/src/Transports/Redis/Wolverine.Redis.Tests/EndToEndRetryTests.cs @@ -40,8 +40,11 @@ public async ValueTask InitializeAsync() // Configure routing to our test stream (without SendInline to ensure durable processing) opts.PublishMessage().ToRedisStream(_streamKey); + // GH-4028: the Redis-native scheduled retry (the sorted set this test inspects) is the + // non-durable path now. A Durable Redis listener uses the real inbox, including for its + // scheduled retries -- see durable_inbox_is_real_4028. opts.ListenToRedisStream(_streamKey, "e2e-retry-group") - .UseDurableInbox() // Use Durable endpoint (for Redis streams BufferedInMemory is default) + .BufferedInMemory() .StartFromBeginning(); // Configure a retry policy @@ -73,8 +76,8 @@ public async ValueTask DisposeAsync() [Fact] public async Task message_with_retry_policy_saves_to_redis_and_retries() { - _endpoint.Mode.ShouldBe(EndpointMode.Durable, "Endpoint should be in Durable mode for retries to work"); - _endpoint.ShouldBeAssignableTo("Endpoint should implement IDatabaseBackedEndpoint"); + _endpoint.Mode.ShouldBe(EndpointMode.BufferedInMemory); + _endpoint.ShouldNotBeAssignableTo("GH-4028: that marker skipped the inbox and ACKed on receipt"); _tracker.FailCount = 1; // Fail once, then succeed diff --git a/src/Transports/Redis/Wolverine.Redis.Tests/NativeSchedulingRetryTests.cs b/src/Transports/Redis/Wolverine.Redis.Tests/NativeSchedulingRetryTests.cs index 49b4d762f..ee749325d 100644 --- a/src/Transports/Redis/Wolverine.Redis.Tests/NativeSchedulingRetryTests.cs +++ b/src/Transports/Redis/Wolverine.Redis.Tests/NativeSchedulingRetryTests.cs @@ -24,17 +24,18 @@ public NativeSchedulingRetryTests(ITestOutputHelper output) } [Fact] - public void redis_endpoint_implements_idatabase_backed_endpoint() + public void redis_endpoint_is_not_database_backed_and_the_listener_owns_native_scheduling() { - // Verify that RedisStreamEndpoint implements IDatabaseBackedEndpoint interface - var endpointType = typeof(RedisStreamEndpoint); - var interfaceType = typeof(IDatabaseBackedEndpoint); - - interfaceType.IsAssignableFrom(endpointType).ShouldBeTrue( - "RedisStreamEndpoint should implement IDatabaseBackedEndpoint"); - - _output.WriteLine("✓ RedisStreamEndpoint implements IDatabaseBackedEndpoint interface"); - _output.WriteLine(" This enables DurableReceiver to call ScheduleRetryAsync() for native retry scheduling"); + // GH-4028: IDatabaseBackedEndpoint on the endpoint made DurableReceiver skip the inbox INSERT and + // complete the delivery on receipt -- "UseDurableInbox()" on a Redis stream never wrote the inbox + // and XACKed before the handler ran. The native scheduled retry is now a listener concern, and a + // Durable listener routes scheduled retries through the inbox like every other transport. + typeof(IDatabaseBackedEndpoint).IsAssignableFrom(typeof(RedisStreamEndpoint)).ShouldBeFalse( + "RedisStreamEndpoint must not be IDatabaseBackedEndpoint -- that marker skips the inbox"); + typeof(ISupportNativeScheduling).IsAssignableFrom(typeof(RedisStreamListener)).ShouldBeTrue( + "RedisStreamListener owns Redis-native scheduled retries for the non-durable modes"); + + _output.WriteLine("✓ RedisStreamEndpoint is not IDatabaseBackedEndpoint; RedisStreamListener is ISupportNativeScheduling"); } [Fact] @@ -49,7 +50,7 @@ public void durable_receiver_implements_native_scheduling() "DurableReceiver should implement ISupportNativeScheduling"); _output.WriteLine("✓ DurableReceiver implements ISupportNativeScheduling"); - _output.WriteLine(" Combined with IDatabaseBackedEndpoint, this enables native retry scheduling"); + _output.WriteLine(" A Durable Redis listener routes scheduled retries through this, i.e. the inbox"); } [Fact] @@ -65,11 +66,11 @@ public async Task endpoint_schedule_retry_async_should_save_to_redis() var runtime = host.Services.GetRequiredService(); var transport = runtime.Options.Transports.GetOrCreate(); - var endpoint = transport.StreamEndpoint(streamKey) as IDatabaseBackedEndpoint; + var endpoint = transport.StreamEndpoint(streamKey); var database = transport.GetDatabase(database: 0); - var scheduledKey = transport.StreamEndpoint(streamKey).ScheduledMessagesKey; + var scheduledKey = endpoint.ScheduledMessagesKey; - endpoint.ShouldNotBeNull("Endpoint should implement IDatabaseBackedEndpoint"); + endpoint.ShouldNotBeNull(); // Clear the scheduled set await database.KeyDeleteAsync(scheduledKey); diff --git a/src/Transports/Redis/Wolverine.Redis.Tests/Samples/RedisTransportWithScheduling.cs b/src/Transports/Redis/Wolverine.Redis.Tests/Samples/RedisTransportWithScheduling.cs index e58430910..381de36ee 100644 --- a/src/Transports/Redis/Wolverine.Redis.Tests/Samples/RedisTransportWithScheduling.cs +++ b/src/Transports/Redis/Wolverine.Redis.Tests/Samples/RedisTransportWithScheduling.cs @@ -30,11 +30,11 @@ public static async Task RunAsync() .SendInline(); opts.ListenToRedisStream("wolverine-messages", "default") - .EnableNativeDeadLetterQueue() // Enable DLQ for failed messages - .UseDurableInbox(); // Use durable inbox so retry messages are persisted + .EnableNativeDeadLetterQueue(); // Enable DLQ for failed messages - // schedule retry delays - // if durable, these will be scheduled natively in Redis + // schedule retry delays. On a Buffered (the default) or Inline Redis listener these are + // parked natively in Redis, in the stream's scheduled sorted set; a Durable listener + // schedules them through its message store's inbox like every other transport opts.OnException() .ScheduleRetry( TimeSpan.FromSeconds(10), diff --git a/src/Transports/Redis/Wolverine.Redis.Tests/durable_inbox_is_real_4028.cs b/src/Transports/Redis/Wolverine.Redis.Tests/durable_inbox_is_real_4028.cs new file mode 100644 index 000000000..9315c1db5 --- /dev/null +++ b/src/Transports/Redis/Wolverine.Redis.Tests/durable_inbox_is_real_4028.cs @@ -0,0 +1,184 @@ +using IntegrationTests; +using JasperFx; +using JasperFx.Core; +using JasperFx.Resources; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Shouldly; +using StackExchange.Redis; +using Wolverine.Configuration; +using Wolverine.ErrorHandling; +using Wolverine.Persistence.Durability; +using Wolverine.Postgresql; +using Wolverine.Redis.Internal; +using Wolverine.Runtime; +using Wolverine.Transports; +using Xunit; + +namespace Wolverine.Redis.Tests; + +/// +/// GH-4028. UseDurableInbox() on a Redis stream used to mean "skip the inbox and XACK on receipt" +/// because was IDatabaseBackedEndpoint. These pin the real +/// durable-inbox contract: the message is written to the inbox before the stream entry is acknowledged, +/// the row is Incoming while the handler runs, and a scheduled retry is parked in the inbox rather +/// than in the Redis scheduled sorted set. +/// +[Collection("durable_inbox_is_real_4028")] +public class durable_inbox_is_real_4028 : IAsyncLifetime +{ + private string _streamKey = null!; + private IHost _host = null!; + private IMessageStore _store = null!; + private RedisStreamEndpoint _endpoint = null!; + private IDatabase _database = null!; + + public async ValueTask InitializeAsync() + { + _streamKey = $"durable-real-{Guid.NewGuid():N}"; + BlockingRedisHandler.Reset(); + + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Durability.Mode = DurabilityMode.Solo; + opts.Durability.ScheduledJobFirstExecution = 100.Milliseconds(); + opts.Durability.ScheduledJobPollingTime = 200.Milliseconds(); + + opts.UseRedisTransport(RedisContainerFixture.ConnectionString).AutoProvision(); + opts.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, "redis_durable_4028"); + + opts.PublishMessage().ToRedisStream(_streamKey).SendInline(); + opts.PublishMessage().ToRedisStream(_streamKey).SendInline(); + + opts.ListenToRedisStream(_streamKey, "durable-real-group") + .UseDurableInbox() + .StartFromBeginning(); + + opts.Policies.OnException().ScheduleRetry(1.Seconds()); + + opts.Discovery.IncludeType(typeof(BlockingRedisHandler)); + opts.Services.AddResourceSetupOnStartup(); + }).StartAsync(); + + _store = _host.Services.GetRequiredService(); + await _store.Admin.ClearAllAsync(); + + var transport = _host.Services.GetRequiredService().Options.Transports.GetOrCreate(); + _endpoint = transport.StreamEndpoint(_streamKey); + _database = transport.GetDatabase(database: _endpoint.DatabaseId); + } + + public async ValueTask DisposeAsync() + { + BlockingRedisHandler.Release(); + await _host.StopAsync(); + _host.Dispose(); + } + + private static async Task waitUntil(Func> condition, TimeSpan timeout) + { + var deadline = DateTimeOffset.UtcNow.Add(timeout); + while (!await condition()) + { + if (DateTimeOffset.UtcNow > deadline) + { + throw new TimeoutException("Condition was not met in time"); + } + + await Task.Delay(100, TestContext.Current.CancellationToken); + } + } + + [Fact] + public void the_endpoint_is_not_database_backed_and_durable_mode_is_really_durable() + { + _endpoint.ShouldNotBeAssignableTo(); + _endpoint.Mode.ShouldBe(EndpointMode.Durable); + } + + [Fact] + public async Task a_message_is_in_the_inbox_and_acked_on_the_stream_before_its_handler_finishes() + { + var id = Guid.NewGuid(); + await _host.MessageBus().PublishAsync(new BlockingRedisMessage(id)); + + // The handler is parked; the message must already be durable + await waitUntil(() => Task.FromResult(BlockingRedisHandler.Started.Task.IsCompleted), 30.Seconds()); + + var counts = await _store.Admin.FetchCountsAsync(); + counts.Incoming.ShouldBe(1, "the envelope must be in the inbox as Incoming while the handler runs"); + + // ...and because it is durable, the stream entry was acknowledged right after the insert, so + // nothing is left pending on the consumer group + var pending = await _database.StreamPendingAsync(_streamKey, _endpoint.ConsumerGroup!); + pending.PendingMessageCount.ShouldBe(0); + + BlockingRedisHandler.Release(); + + await waitUntil(async () => (await _store.Admin.FetchCountsAsync()).Incoming == 0, 30.Seconds()); + BlockingRedisHandler.Executions(id).ShouldBe(1); + } + + [Fact] + public async Task a_scheduled_retry_is_parked_in_the_inbox_not_in_the_redis_scheduled_set() + { + var id = Guid.NewGuid(); + await _host.MessageBus().PublishAsync(new RetryOnceRedisMessage(id)); + + // First attempt fails and schedules a retry 1s out: the inbox holds it as Scheduled + await waitUntil(async () => (await _store.Admin.FetchCountsAsync()).Scheduled >= 1, 30.Seconds()); + + // The Redis-native scheduled sorted set is the NON-durable mechanism and must be untouched here + (await _database.SortedSetLengthAsync(_endpoint.ScheduledMessagesKey)).ShouldBe(0); + + // ...and the retry still happens, from the inbox + await waitUntil(() => Task.FromResult(BlockingRedisHandler.Executions(id) >= 2), 30.Seconds()); + await waitUntil(async () => (await _store.Admin.FetchCountsAsync()).Scheduled == 0, 30.Seconds()); + } +} + +public record BlockingRedisMessage(Guid Id); + +public record RetryOnceRedisMessage(Guid Id); + +public static class BlockingRedisHandler +{ + private static readonly System.Collections.Concurrent.ConcurrentDictionary _executions = new(); + private static TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public static TaskCompletionSource Started { get; private set; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public static int Executions(Guid id) + { + return _executions.GetValueOrDefault(id); + } + + public static void Reset() + { + _executions.Clear(); + _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } + + public static void Release() + { + _gate.TrySetResult(); + } + + public static async Task Handle(BlockingRedisMessage message) + { + _executions.AddOrUpdate(message.Id, 1, (_, n) => n + 1); + Started.TrySetResult(); + await _gate.Task; + } + + public static void Handle(RetryOnceRedisMessage message) + { + var attempt = _executions.AddOrUpdate(message.Id, 1, (_, n) => n + 1); + if (attempt == 1) + { + throw new InvalidOperationException("first attempt fails on purpose"); + } + } +} diff --git a/src/Transports/Redis/Wolverine.Redis.Tests/rate_limiting_end_to_end.cs b/src/Transports/Redis/Wolverine.Redis.Tests/rate_limiting_end_to_end.cs index 19c986523..165f5014b 100644 --- a/src/Transports/Redis/Wolverine.Redis.Tests/rate_limiting_end_to_end.cs +++ b/src/Transports/Redis/Wolverine.Redis.Tests/rate_limiting_end_to_end.cs @@ -35,8 +35,10 @@ public async ValueTask InitializeAsync() opts.UseRedisTransport(RedisContainerFixture.ConnectionString).AutoProvision(); opts.PublishAllMessages().ToRedisStream(streamKey); + // GH-4028: Redis-native scheduling (which this test is about) is the Buffered/Inline path + // now; a Durable listener schedules through the inbox and needs a message store. opts.ListenToRedisStream(streamKey, groupName) - .UseDurableInbox() + .BufferedInMemory() .StartFromBeginning(); opts.RateLimitEndpoint(endpointUri, _limit); diff --git a/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs b/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs index 4d38eb893..345e2f5c2 100644 --- a/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs +++ b/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs @@ -11,7 +11,7 @@ namespace Wolverine.Redis.Internal; -public class RedisStreamEndpoint : Endpoint, IBrokerEndpoint, IBrokerQueue, IDatabaseBackedEndpoint +public class RedisStreamEndpoint : Endpoint, IBrokerEndpoint, IBrokerQueue { private readonly RedisTransport _transport; @@ -402,7 +402,20 @@ public async ValueTask SetupAsync(ILogger logger) } } - // IDatabaseBackedEndpoint implementation + /// + /// Park a message in this stream's scheduled sorted set until , + /// when the listener's polling loop moves it back onto the stream. This is the Redis-native + /// scheduled retry that Buffered and Inline listeners use through + /// on . + /// + /// + /// GH-4028. This endpoint used to implement IDatabaseBackedEndpoint so that DurableReceiver + /// would route scheduled retries here. That marker also told DurableReceiver to skip the inbox INSERT + /// and complete the delivery on receipt -- so "UseDurableInbox()" on a Redis stream ACKed every + /// message before its handler ran and never wrote the inbox at all. A Durable Redis listener now + /// goes through the real inbox like every other transport, including for scheduled retries; + /// Redis-native scheduling is for the non-durable modes. + /// public async Task ScheduleRetryAsync(Envelope envelope, CancellationToken cancellation) { try diff --git a/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamListener.cs b/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamListener.cs index e7c6cfeef..b15f670db 100644 --- a/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamListener.cs +++ b/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamListener.cs @@ -10,7 +10,8 @@ namespace Wolverine.Redis.Internal; -public class RedisStreamListener : IListener, ISupportDeadLetterQueue, IReportConnectionState, IReportReceiveLoopHealth +public class RedisStreamListener : IListener, ISupportDeadLetterQueue, IReportConnectionState, IReportReceiveLoopHealth, + ISupportNativeScheduling { private readonly RedisTransport _transport; private readonly RedisStreamEndpoint _endpoint; @@ -243,6 +244,25 @@ public async ValueTask CompleteAsync(Envelope envelope) } } + /// + /// GH-4028. Redis-native scheduled retries for the non-durable modes: the message is parked in the + /// stream's scheduled sorted set and the polling loop puts it back on the stream when it is due. A + /// Durable listener deliberately reports false here so the scheduled retry goes through the + /// inbox like every other durable transport -- a listener-level reschedule bypasses the inbox + /// entirely, and re-adding the same envelope id to the stream would collide with the inbox row on + /// redelivery and be discarded as a duplicate. + /// + public bool NativeSchedulingEnabled => _endpoint.Mode != EndpointMode.Durable; + + public async Task MoveToScheduledUntilAsync(Envelope envelope, DateTimeOffset time) + { + envelope.ScheduledTime = time; + + // Park the copy first, then settle the original: a crash in between costs a duplicate, not a loss + await _endpoint.ScheduleRetryAsync(envelope, _cancellation.Token); + await CompleteAsync(envelope); + } + public async ValueTask DeferAsync(Envelope envelope) { try