GH-4035: select the storage reset by its own marker instead of the durable-inbox one

src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueue.cs

@@ -11,7 +11,7 @@
 
 namespace Wolverine.MySql.Transport;
 
-public class MySqlQueue : Endpoint, IBrokerQueue, IDatabaseBackedEndpoint
+public class MySqlQueue : Endpoint, IBrokerQueue, IDatabaseBackedEndpoint, IStorageBackedQueue
 {
     internal static Uri ToUri(string name, string? databaseName)
     {

src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueue.cs

@@ -11,7 +11,7 @@
 
 namespace Wolverine.Oracle.Transport;
 
-public class OracleQueue : Endpoint, IBrokerQueue, IDatabaseBackedEndpoint
+public class OracleQueue : Endpoint, IBrokerQueue, IDatabaseBackedEndpoint, IStorageBackedQueue
 {
     internal static Uri ToUri(string name, string? databaseName)
     {

src/Persistence/Wolverine.Postgresql/Transport/PostgresqlQueue.cs

@@ -11,7 +11,7 @@
 
 namespace Wolverine.Postgresql.Transport;
 
-public class PostgresqlQueue : Endpoint, IBrokerQueue, IDatabaseBackedEndpoint
+public class PostgresqlQueue : Endpoint, IBrokerQueue, IDatabaseBackedEndpoint, IStorageBackedQueue
 {
     internal static Uri ToUri(string name, string? databaseName)
     {

src/Persistence/Wolverine.SqlServer/Transport/SqlServerQueue.cs

@@ -15,7 +15,7 @@
 
 namespace Wolverine.SqlServer.Transport;
 
-public class SqlServerQueue : Endpoint, IBrokerQueue, IDatabaseBackedEndpoint
+public class SqlServerQueue : Endpoint, IBrokerQueue, IDatabaseBackedEndpoint, IStorageBackedQueue
 {
     internal static Uri ToUri(string name, string? databaseName)
     {

src/Persistence/Wolverine.Sqlite/Transport/SqliteQueue.cs

@@ -15,7 +15,7 @@
 
 namespace Wolverine.Sqlite.Transport;
 
-public class SqliteQueue : Endpoint, IBrokerQueue, IDatabaseBackedEndpoint
+public class SqliteQueue : Endpoint, IBrokerQueue, IDatabaseBackedEndpoint, IStorageBackedQueue
 {
     internal static Uri ToUri(string name, string? databaseName)
     {

src/Testing/Wolverine.ComplianceTests/ClearAllWolverineStorageCompliance.cs

@@ -75,7 +75,7 @@ public async ValueTask InitializeAsync()
         theQueue = runtime.Options.Transports
             .SelectMany(x => x.Endpoints())
             .OfType<IBrokerQueue>()
-            .Single(x => x is IDatabaseBackedEndpoint && ((Endpoint)x).EndpointName == QueueName);
+            .Single(x => x is IStorageBackedQueue && ((Endpoint)x).EndpointName == QueueName);
 
         theMessageStore = runtime.Storage;
 
@@ -92,12 +92,25 @@ public virtual async ValueTask DisposeAsync()
         theHost.Dispose();
     }
 
-    protected async Task<(long Queued, long Scheduled)> queueCountsAsync()
+    /// <summary>
+    /// How many messages are queued, and how many are parked for later. The database queues all report
+    /// this through <c>GetAttributesAsync()</c> under the same two keys; a transport that names its
+    /// diagnostics differently overrides this rather than bending its diagnostic surface to fit.
+    /// </summary>
+    protected virtual async Task<(long Queued, long Scheduled)> queueCountsAsync()
     {
         var attributes = await theQueue.GetAttributesAsync();
         return (long.Parse(attributes["Count"]), long.Parse(attributes["Scheduled"]));
     }
 
+    /// <summary>
+    /// Whether writing to this queue after <c>TeardownAsync()</c> fails. True for the database queues,
+    /// whose tables really are gone. False where the storage is implicitly recreated by the write --
+    /// a Redis XADD silently recreates a deleted stream key, so there is no missing "table" to observe
+    /// and only the empties-it half of the rebuild scenario is meaningful.
+    /// </summary>
+    protected virtual bool TeardownMakesTheQueueUnwritable => true;
+
     /// <summary>
     /// One row in the queue table, one in its scheduled-message table.
     /// </summary>
@@ -188,17 +201,20 @@ public async Task rebuilds_queue_tables_that_have_been_dropped()
         // writing rather than through CheckAsync() or a count: Weasel's schema diff throws an NRE
         // against a table that is entirely absent rather than reporting a difference, and some
         // providers' CountAsync swallows the missing-table error and reports zero.
-        var missing = false;
-        try
+        if (TeardownMakesTheQueueUnwritable)
         {
-            await sendToQueueAsync(ObjectMother.Envelope());
-        }
-        catch (Exception)
-        {
-            missing = true;
-        }
+            var missing = false;
+            try
+            {
+                await sendToQueueAsync(ObjectMother.Envelope());
+            }
+            catch (Exception)
+            {
+                missing = true;
+            }
 
-        missing.ShouldBeTrue();
+            missing.ShouldBeTrue();
+        }
 
         await theHost.ClearAllWolverineStorageAsync();
 

src/Transports/Redis/Wolverine.Redis.Tests/clear_all_wolverine_storage.cs

@@ -0,0 +1,75 @@
+using IntegrationTests;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using StackExchange.Redis;
+using Wolverine.ComplianceTests;
+using Wolverine.Postgresql;
+using Wolverine.Redis.Internal;
+using Wolverine.Runtime;
+using Wolverine.Tracking;
+using Xunit;
+
+namespace Wolverine.Redis.Tests;
+
+/// <summary>
+/// GH-4035. The Redis stream endpoint's storage -- the stream itself plus its scheduled sorted set --
+/// is part of the footprint <see cref="Wolverine.Runtime.StorageExtensions.ClearAllWolverineStorageAsync"/>
+/// resets, and there was no coverage of that. GH-4028 removed <c>IDatabaseBackedEndpoint</c> from the
+/// endpoint for good reasons, which silently dropped Redis out of the reset because that marker doubled
+/// as the selector; 38/38 checks stayed green. This suite is what would have failed.
+/// </summary>
+[Collection("ClearAllWolverineStorageRedis4035")]
+public class clear_all_wolverine_storage : ClearAllWolverineStorageCompliance
+{
+    private readonly string _streamKey = $"reset-{Guid.NewGuid():N}";
+
+    protected override void ConfigureStorage(WolverineOptions options)
+    {
+        options.UseRedisTransport(RedisContainerFixture.ConnectionString).AutoProvision();
+        options.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, "redis_reset_4035");
+
+        // Subscriber only -- a listener would drain the stream out from under the assertions before
+        // the reset ever runs. Named so the compliance suite's endpoint lookup finds it.
+        options.PublishAllMessages().ToRedisStream(_streamKey).Named(QueueName);
+    }
+
+    /// <summary>
+    /// A Redis XADD recreates a deleted stream key silently, so there is no missing "table" to observe
+    /// after TeardownAsync(). Only the empties-it half of the rebuild scenario means anything here.
+    /// </summary>
+    protected override bool TeardownMakesTheQueueUnwritable => false;
+
+    /// <summary>
+    /// RedisStreamEndpoint.GetAttributesAsync() reports streamKey/messageCount/consumerGroup rather than
+    /// the database queues' Count/Scheduled, so read the two keys directly instead of reshaping a
+    /// diagnostic surface other things depend on.
+    /// </summary>
+    protected override async Task<(long Queued, long Scheduled)> queueCountsAsync()
+    {
+        var endpoint = (RedisStreamEndpoint)theQueue;
+        var transport = theHost.GetRuntime().Options.Transports.GetOrCreate<RedisTransport>();
+        var database = transport.GetDatabase(database: endpoint.DatabaseId);
+
+        var queued = await database.KeyExistsAsync(endpoint.StreamKey)
+            ? await database.StreamLengthAsync(endpoint.StreamKey)
+            : 0L;
+
+        var scheduled = await database.SortedSetLengthAsync(endpoint.ScheduledMessagesKey);
+
+        return (queued, scheduled);
+    }
+
+    protected override ValueTask sendToQueueAsync(Envelope envelope)
+    {
+        var endpoint = (RedisStreamEndpoint)theQueue;
+        var runtime = theHost.GetRuntime();
+        var transport = runtime.Options.Transports.GetOrCreate<RedisTransport>();
+
+        // The endpoint has no SendAsync(Envelope) of its own; the inline sender is the seam that puts
+        // an immediate message on the stream and a scheduled one in the sorted set.
+        return new InlineRedisStreamSender(transport, endpoint, runtime).SendAsync(envelope);
+    }
+}
+
+[CollectionDefinition("ClearAllWolverineStorageRedis4035", DisableParallelization = true)]
+public class ClearAllWolverineStorageRedis4035Collection;

src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs

@@ -11,7 +11,8 @@
 
 namespace Wolverine.Redis.Internal;
 
-public class RedisStreamEndpoint : Endpoint<IRedisEnvelopeMapper, RedisEnvelopeMapper>, IBrokerEndpoint, IBrokerQueue
+public class RedisStreamEndpoint : Endpoint<IRedisEnvelopeMapper, RedisEnvelopeMapper>, IBrokerEndpoint, IBrokerQueue,
+    IStorageBackedQueue
 {
     private readonly RedisTransport _transport;
 
@@ -331,6 +332,13 @@ public async ValueTask PurgeAsync(ILogger logger)
             var db = _transport.GetDatabase(database: DatabaseId);
             if (await db.KeyDeleteAsync(StreamKey))
                 logger.LogInformation("Purged Redis stream {StreamKey}", StreamKey);
+
+            // GH-4035. The scheduled sorted set is part of this queue's storage just as the
+            // scheduled-message *table* is on the database queues, and PurgeAsync() is expected to
+            // leave a queue completely empty -- IHost.ClearAllWolverineStorageAsync() relies on it.
+            // Leaving it behind meant a scheduled message survived a "reset" and fired into the next test.
+            if (await db.KeyDeleteAsync(ScheduledMessagesKey))
+                logger.LogInformation("Purged scheduled messages for Redis stream {StreamKey}", StreamKey);
         }
         catch (Exception e)
         {

src/Wolverine/Configuration/Endpoint.cs

@@ -31,11 +31,32 @@ public enum PartitionSlots
 /// Marker interface that tells Wolverine internals that this endpoint directly
 /// integrates with the active transactional inbox
 /// </summary>
+/// <remarks>
+/// GH-4035. This is <i>only</i> about inbox integration: it routes scheduled retries to
+/// <see cref="ScheduleRetryAsync"/> and it tells <c>DurableReceiver</c> that the endpoint persists
+/// incoming messages itself, so the arrival INSERT is skipped and the delivery is completed on receipt.
+/// Do not reuse it to mean "this queue has storage of its own" -- see <see cref="IStorageBackedQueue"/>.
+/// </remarks>
 public interface IDatabaseBackedEndpoint
 {
     Task ScheduleRetryAsync(Envelope envelope, CancellationToken cancellation);
 }
 
+/// <summary>
+/// Marker for a queue whose contents live in storage that Wolverine itself provisions -- the database
+/// queue tables, or a Redis stream and its scheduled sorted set -- rather than in an external broker.
+/// <see cref="Wolverine.Runtime.StorageExtensions.ClearAllWolverineStorageAsync"/> builds and empties
+/// exactly these.
+/// </summary>
+/// <remarks>
+/// GH-4035. This used to be inferred from <see cref="IDatabaseBackedEndpoint"/>, which meant a change
+/// to an endpoint's <i>inbox</i> behaviour silently changed whether integration tests could reset it.
+/// Removing that marker from the Redis stream endpoint in GH-4028 dropped Redis out of the reset with
+/// nothing in CI able to see it. The two concerns are separate now: an endpoint may be either, both, or
+/// neither.
+/// </remarks>
+public interface IStorageBackedQueue;
+
 public enum TenancyBehavior
 {
     /// <summary>

src/Wolverine/Runtime/StorageExtensions.cs

@@ -31,16 +31,21 @@ public static async Task ClearAllWolverineStorageAsync(this IHost host)
             await store.Admin.RebuildAsync();
         }
 
-        // Then the database-backed queue transports. SetupAsync() builds the queue table and its
-        // scheduled-message table if they are missing, PurgeAsync() empties both -- and each fans out
-        // across every tenant database on a multi-tenanted transport. That pair exists on every
-        // database queue transport (PostgreSQL, SQL Server, MySQL, Oracle, SQLite, and Redis streams),
-        // so this needs no provider-specific code.
+        // Then the queue transports whose contents live in storage Wolverine provisions. SetupAsync()
+        // builds the queue's storage if it is missing, PurgeAsync() empties it -- and each fans out
+        // across every tenant database on a multi-tenanted transport. That pair exists on every such
+        // transport (PostgreSQL, SQL Server, MySQL, Oracle, SQLite, and Redis streams), so this needs
+        // no provider-specific code.
+        //
+        // GH-4035: selected by IStorageBackedQueue, NOT by IDatabaseBackedEndpoint. The latter is about
+        // inbox integration, and using it here meant GH-4028's (correct) removal of it from the Redis
+        // stream endpoint silently dropped Redis streams out of this reset -- with no test able to see
+        // it, because no Redis implementation of ClearAllWolverineStorageCompliance existed.
         foreach (var transport in runtime.Options.Transports)
         {
             var queues = transport.Endpoints()
                 .OfType<IBrokerQueue>()
-                .Where(x => x is IDatabaseBackedEndpoint)
+                .Where(x => x is IStorageBackedQueue)
                 .ToArray();
 
             if (queues.Length == 0) continue;