From aa40b2a368dc87d3c54dcd033df5fb07d6f4accf Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 23 Aug 2026 15:05:09 -0500 Subject: [PATCH] GH-4035: select the storage reset by its own marker instead of the durable-inbox one ClearAllWolverineStorageAsync() picked the queues it resets with `.Where(x => x is IDatabaseBackedEndpoint)`, so an endpoint's *inbox* behaviour silently decided whether integration tests could reset it. GH-4028 correctly removed that marker from RedisStreamEndpoint -- and thereby dropped Redis streams out of the reset entirely, with all 38 checks green, because no Redis implementation of the compliance suite existed. IStorageBackedQueue now says what the reset actually means: a queue whose contents live in storage Wolverine provisions rather than in an external broker. It is carried by the five database queues and by RedisStreamEndpoint, and it is independent of IDatabaseBackedEndpoint -- an endpoint may be either, both, or neither. Also fixed, found by the new coverage: * RedisStreamEndpoint.PurgeAsync() deleted the stream key but not the scheduled sorted set, so a scheduled message survived a "reset" and fired into the next test. The sorted set is part of this queue's storage exactly as the scheduled-message table is on the database queues. The compliance suite gains two seams so a non-table transport can implement it honestly rather than by bending its own surface to fit: * queueCountsAsync() is virtual -- RedisStreamEndpoint.GetAttributesAsync() reports streamKey/messageCount/consumerGroup, not the database queues' Count/Scheduled. * TeardownMakesTheQueueUnwritable gates the missing-storage precondition in rebuilds_queue_tables_that_have_been_dropped. A Redis XADD silently recreates a deleted stream key, so only the empties-it half of that scenario is observable there; the rest of the test still runs. Verified the new suite actually catches the regression: with the selector reverted to IDatabaseBackedEndpoint, 3 of its 5 tests fail. wolverine.slnx -c Release -f net9.0 clean. Wolverine.Redis.Tests 152/152. CoreTests 2525 (0 failed, 2 skipped). Sqlite and Postgresql clear_all_wolverine_storage 5/5 each. Co-Authored-By: Claude Opus 5 --- .../Wolverine.MySql/Transport/MySqlQueue.cs | 2 +- .../Wolverine.Oracle/Transport/OracleQueue.cs | 2 +- .../Transport/PostgresqlQueue.cs | 2 +- .../Transport/SqlServerQueue.cs | 2 +- .../Wolverine.Sqlite/Transport/SqliteQueue.cs | 2 +- .../ClearAllWolverineStorageCompliance.cs | 38 +++++++--- .../clear_all_wolverine_storage.cs | 75 +++++++++++++++++++ .../Internal/RedisStreamEndpoint.cs | 10 ++- src/Wolverine/Configuration/Endpoint.cs | 21 ++++++ src/Wolverine/Runtime/StorageExtensions.cs | 17 +++-- 10 files changed, 148 insertions(+), 23 deletions(-) create mode 100644 src/Transports/Redis/Wolverine.Redis.Tests/clear_all_wolverine_storage.cs diff --git a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueue.cs b/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueue.cs index 0d667e040..a24318b52 100644 --- a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueue.cs +++ b/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) { diff --git a/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueue.cs b/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueue.cs index 6f8d6e337..028a7a52f 100644 --- a/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueue.cs +++ b/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) { diff --git a/src/Persistence/Wolverine.Postgresql/Transport/PostgresqlQueue.cs b/src/Persistence/Wolverine.Postgresql/Transport/PostgresqlQueue.cs index 37c2f4f45..3ab4adbb6 100644 --- a/src/Persistence/Wolverine.Postgresql/Transport/PostgresqlQueue.cs +++ b/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) { diff --git a/src/Persistence/Wolverine.SqlServer/Transport/SqlServerQueue.cs b/src/Persistence/Wolverine.SqlServer/Transport/SqlServerQueue.cs index 5f53da58c..00cb600ee 100644 --- a/src/Persistence/Wolverine.SqlServer/Transport/SqlServerQueue.cs +++ b/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) { diff --git a/src/Persistence/Wolverine.Sqlite/Transport/SqliteQueue.cs b/src/Persistence/Wolverine.Sqlite/Transport/SqliteQueue.cs index 4829b3a18..fc014ccfc 100644 --- a/src/Persistence/Wolverine.Sqlite/Transport/SqliteQueue.cs +++ b/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) { diff --git a/src/Testing/Wolverine.ComplianceTests/ClearAllWolverineStorageCompliance.cs b/src/Testing/Wolverine.ComplianceTests/ClearAllWolverineStorageCompliance.cs index 6b49f954b..7724817f1 100644 --- a/src/Testing/Wolverine.ComplianceTests/ClearAllWolverineStorageCompliance.cs +++ b/src/Testing/Wolverine.ComplianceTests/ClearAllWolverineStorageCompliance.cs @@ -75,7 +75,7 @@ public async ValueTask InitializeAsync() theQueue = runtime.Options.Transports .SelectMany(x => x.Endpoints()) .OfType() - .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() + /// + /// How many messages are queued, and how many are parked for later. The database queues all report + /// this through GetAttributesAsync() under the same two keys; a transport that names its + /// diagnostics differently overrides this rather than bending its diagnostic surface to fit. + /// + protected virtual async Task<(long Queued, long Scheduled)> queueCountsAsync() { var attributes = await theQueue.GetAttributesAsync(); return (long.Parse(attributes["Count"]), long.Parse(attributes["Scheduled"])); } + /// + /// Whether writing to this queue after TeardownAsync() 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. + /// + protected virtual bool TeardownMakesTheQueueUnwritable => true; + /// /// One row in the queue table, one in its scheduled-message table. /// @@ -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(); diff --git a/src/Transports/Redis/Wolverine.Redis.Tests/clear_all_wolverine_storage.cs b/src/Transports/Redis/Wolverine.Redis.Tests/clear_all_wolverine_storage.cs new file mode 100644 index 000000000..489f21208 --- /dev/null +++ b/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; + +/// +/// GH-4035. The Redis stream endpoint's storage -- the stream itself plus its scheduled sorted set -- +/// is part of the footprint +/// resets, and there was no coverage of that. GH-4028 removed IDatabaseBackedEndpoint 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. +/// +[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); + } + + /// + /// 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. + /// + protected override bool TeardownMakesTheQueueUnwritable => false; + + /// + /// 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. + /// + protected override async Task<(long Queued, long Scheduled)> queueCountsAsync() + { + var endpoint = (RedisStreamEndpoint)theQueue; + var transport = theHost.GetRuntime().Options.Transports.GetOrCreate(); + 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(); + + // 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; diff --git a/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs b/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs index 345e2f5c2..c025101d0 100644 --- a/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs +++ b/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs @@ -11,7 +11,8 @@ namespace Wolverine.Redis.Internal; -public class RedisStreamEndpoint : Endpoint, IBrokerEndpoint, IBrokerQueue +public class RedisStreamEndpoint : Endpoint, 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) { diff --git a/src/Wolverine/Configuration/Endpoint.cs b/src/Wolverine/Configuration/Endpoint.cs index 661cb8cc0..9948944cf 100644 --- a/src/Wolverine/Configuration/Endpoint.cs +++ b/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 /// +/// +/// GH-4035. This is only about inbox integration: it routes scheduled retries to +/// and it tells DurableReceiver 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 . +/// public interface IDatabaseBackedEndpoint { Task ScheduleRetryAsync(Envelope envelope, CancellationToken cancellation); } +/// +/// 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. +/// builds and empties +/// exactly these. +/// +/// +/// GH-4035. This used to be inferred from , which meant a change +/// to an endpoint's inbox 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. +/// +public interface IStorageBackedQueue; + public enum TenancyBehavior { /// diff --git a/src/Wolverine/Runtime/StorageExtensions.cs b/src/Wolverine/Runtime/StorageExtensions.cs index 74bfb1d4d..ab72b4cf4 100644 --- a/src/Wolverine/Runtime/StorageExtensions.cs +++ b/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() - .Where(x => x is IDatabaseBackedEndpoint) + .Where(x => x is IStorageBackedQueue) .ToArray(); if (queues.Length == 0) continue;