GH-4028: make UseDurableInbox() on a Redis stream a real durable inbox

docs/guide/messaging/transports/redis.md

@@ -360,6 +360,24 @@ public class OurRedisJsonMapper<TMessage> : EnvelopeMapper<StreamEntry, List<Nam
 <sup><a href='https://github.com/JasperFx/wolverine/blob/main/src/Transports/Redis/Wolverine.Redis.Tests/DocumentationSamples.cs#L230-L291' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_ourredisjsonmapper' title='Start of snippet'>anchor</a></sup>
 <!-- endSnippet -->
 
+## Durable Inbox <Badge type="tip" text="6.30" />
+
+`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 <Badge type="tip" text="5.10" />
 
 The Redis transport supports native Redis message scheduling for delayed or scheduled delivery. There's no configuration

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<RedisTransport>();
         var endpoint = transport.StreamEndpoint(streamKey);
 
-        // Verify it implements IDatabaseBackedEndpoint
-        endpoint.ShouldBeAssignableTo<IDatabaseBackedEndpoint>();
+        // 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<IDatabaseBackedEndpoint>();
+        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<IWolverineRuntime>();
         var transport = runtime.Options.Transports.GetOrCreate<RedisTransport>();
-        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<IWolverineRuntime>();
         var transport = runtime.Options.Transports.GetOrCreate<RedisTransport>();
-        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<IWolverineRuntime>();
         var transport = runtime.Options.Transports.GetOrCreate<RedisTransport>();
-        var endpoint = transport.StreamEndpoint(streamKey) as IDatabaseBackedEndpoint;
+        var endpoint = transport.StreamEndpoint(streamKey);
         var database = transport.GetDatabase(database: 0);
         var scheduledKey = transport.StreamEndpoint(streamKey).ScheduledMessagesKey;
 

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<E2EFailingCommand>().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<IDatabaseBackedEndpoint>("Endpoint should implement IDatabaseBackedEndpoint");
+        _endpoint.Mode.ShouldBe(EndpointMode.BufferedInMemory);
+        _endpoint.ShouldNotBeAssignableTo<IDatabaseBackedEndpoint>("GH-4028: that marker skipped the inbox and ACKed on receipt");
 
         _tracker.FailCount = 1; // Fail once, then succeed
 

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<IWolverineRuntime>();
         var transport = runtime.Options.Transports.GetOrCreate<RedisTransport>();
-        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);

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<Exception>()
                 .ScheduleRetry(
                     TimeSpan.FromSeconds(10),

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;
+
+/// <summary>
+/// GH-4028. <c>UseDurableInbox()</c> on a Redis stream used to mean "skip the inbox and XACK on receipt"
+/// because <see cref="RedisStreamEndpoint"/> was <c>IDatabaseBackedEndpoint</c>. These pin the real
+/// durable-inbox contract: the message is written to the inbox before the stream entry is acknowledged,
+/// the row is <c>Incoming</c> while the handler runs, and a scheduled retry is parked in the inbox rather
+/// than in the Redis scheduled sorted set.
+/// </summary>
+[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<BlockingRedisMessage>().ToRedisStream(_streamKey).SendInline();
+                opts.PublishMessage<RetryOnceRedisMessage>().ToRedisStream(_streamKey).SendInline();
+
+                opts.ListenToRedisStream(_streamKey, "durable-real-group")
+                    .UseDurableInbox()
+                    .StartFromBeginning();
+
+                opts.Policies.OnException<InvalidOperationException>().ScheduleRetry(1.Seconds());
+
+                opts.Discovery.IncludeType(typeof(BlockingRedisHandler));
+                opts.Services.AddResourceSetupOnStartup();
+            }).StartAsync();
+
+        _store = _host.Services.GetRequiredService<IMessageStore>();
+        await _store.Admin.ClearAllAsync();
+
+        var transport = _host.Services.GetRequiredService<IWolverineRuntime>().Options.Transports.GetOrCreate<RedisTransport>();
+        _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<Task<bool>> 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<IDatabaseBackedEndpoint>();
+        _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<Guid, int> _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");
+        }
+    }
+}

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);