Add EditAndReplayAsync to Oracle, compliance test, and fix EFCore test setup

src/Http/Wolverine.Http.Tests/Marten/reacting_to_read_aggregate.cs

@@ -44,6 +44,7 @@ async Task IAsyncLifetime.DisposeAsync()
         if (theHost != null)
         {
             await theHost.StopAsync();
+            theHost.Dispose();
         }
     }
 

src/Http/Wolverine.Http.Tests/Persistence/global_entity_defaults_http.cs

@@ -46,6 +46,7 @@ async Task IAsyncLifetime.DisposeAsync()
         if (theHost != null)
         {
             await theHost.StopAsync();
+            theHost.Dispose();
         }
     }
 

src/Http/Wolverine.Http.Tests/Persistence/reacting_to_entity_attributes.cs

@@ -52,6 +52,7 @@ async Task IAsyncLifetime.DisposeAsync()
         if (theHost != null)
         {
             await theHost.StopAsync();
+            theHost.Dispose();
         }
     }
 

src/Http/Wolverine.Http.Tests/Persistence/reacting_to_required_attribute.cs

@@ -46,7 +46,11 @@ public async Task InitializeAsync()
 
     async Task IAsyncLifetime.DisposeAsync()
     {
-        if (theHost != null) await theHost.StopAsync();
+        if (theHost != null)
+        {
+            await theHost.StopAsync();
+            theHost.Dispose();
+        }
     }
 
     [Fact]

src/Persistence/EfCoreTests/Bug_252_codegen_issue.cs

@@ -3,11 +3,13 @@
 using JasperFx.Core;
 using JasperFx.Core.Reflection;
 using JasperFx.Resources;
+using Microsoft.Data.SqlClient;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using SharedPersistenceModels.Items;
 using Shouldly;
+using Weasel.SqlServer;
 using Wolverine;
 using Wolverine.Attributes;
 using Wolverine.EntityFrameworkCore;
@@ -31,6 +33,11 @@ public Bug_252_codegen_issue(ITestOutputHelper output)
     [Fact]
     public async Task use_the_saga_type_to_determine_the_correct_DbContext_type()
     {
+        await using var conn = new SqlConnection(Servers.SqlServerConnectionString);
+        await conn.OpenAsync();
+        await conn.DropSchemaAsync("mt_items");
+        await conn.CloseAsync();
+
         using var host = await Host.CreateDefaultBuilder()
             .UseWolverine(opt =>
             {
@@ -60,6 +67,11 @@ public async Task use_the_saga_type_to_determine_the_correct_DbContext_type()
     [Fact]
     public async Task bug_256_message_bus_should_be_in_outbox_transaction()
     {
+        await using var conn = new SqlConnection(Servers.SqlServerConnectionString);
+        await conn.OpenAsync();
+        await conn.DropSchemaAsync("mt_items");
+        await conn.CloseAsync();
+
         using var host = await Host.CreateDefaultBuilder()
             .UseWolverine(opt =>
             {

src/Persistence/EfCoreTests/end_to_end_efcore_persistence.cs

@@ -8,7 +8,6 @@
 using JasperFx.Resources;
 using SharedPersistenceModels.Items;
 using Shouldly;
-using Wolverine.ComplianceTests;
 using Weasel.Core;
 using Weasel.SqlServer;
 using Weasel.SqlServer.Tables;
@@ -24,11 +23,21 @@
 
 namespace EfCoreTests;
 
-public class EFCorePersistenceContext : BaseContext
+public class EFCorePersistenceContext : IAsyncLifetime
 {
-    public EFCorePersistenceContext() : base(true)
+    public IHost theHost { get; private set; } = null!;
+
+    public async Task InitializeAsync()
     {
-        builder.ConfigureServices((c, services) =>
+        // Drop the schema first so Weasel migrations can cleanly create/alter columns
+        // without conflicting with stale data from previous test runs
+        await using var conn = new SqlConnection(Servers.SqlServerConnectionString);
+        await conn.OpenAsync();
+        await conn.DropSchemaAsync("mt_items");
+        await conn.CloseAsync();
+
+        theHost = await Host.CreateDefaultBuilder()
+            .ConfigureServices((c, services) =>
             {
                 services.AddDbContext<ItemsDbContext>(x => x.UseSqlServer(Servers.SqlServerConnectionString));
                 services.AddDbContext<SampleMappedDbContext>(x => x.UseSqlServer(Servers.SqlServerConnectionString));
@@ -39,16 +48,21 @@ public EFCorePersistenceContext() : base(true)
                 options.PersistMessagesWithSqlServer(Servers.SqlServerConnectionString);
                 options.Services.AddResourceSetupOnStartup(StartupAction.ResetState);
                 options.UseEntityFrameworkCoreTransactions();
-                
+
                 options.Policies.ConfigureConventionalLocalRouting()
                     .CustomizeQueues((_, q) => q.UseDurableInbox());
-                
+
                 options.Policies.AutoApplyTransactions();
-                
+
                 options.UseEntityFrameworkCoreWolverineManagedMigrations();
-            });
+            }).StartAsync();
     }
 
+    public async Task DisposeAsync()
+    {
+        await theHost.StopAsync();
+        theHost.Dispose();
+    }
 }
 
 [Collection("sqlserver")]

src/Persistence/Oracle/OracleTests/OracleMessageStoreTests.cs

@@ -98,4 +98,5 @@ public async Task can_schedule_envelope()
         var counts = await thePersistence.Admin.FetchCountsAsync();
         counts.Scheduled.ShouldBeGreaterThanOrEqualTo(1);
     }
+
 }

src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.DeadLetters.cs

@@ -5,6 +5,7 @@
 using Wolverine.Persistence.Durability;
 using Wolverine.Persistence.Durability.DeadLetterManagement;
 using Wolverine.RDBMS;
+using Wolverine.Runtime.Serialization;
 
 namespace Wolverine.Oracle;
 
@@ -180,6 +181,37 @@ public async Task ReplayAsync(DeadLetterEnvelopeQuery query, CancellationToken t
         }
     }
 
+    public async Task EditAndReplayAsync(Guid envelopeId, byte[] newBody, CancellationToken token)
+    {
+        var deadLetter = await DeadLetterEnvelopeByIdAsync(envelopeId);
+        if (deadLetter == null) return;
+
+        deadLetter.Envelope.Data = newBody;
+        var serialized = EnvelopeSerializer.Serialize(deadLetter.Envelope);
+
+        var builder = ToOracleCommandBuilder();
+        builder.Append(
+            $"UPDATE {SchemaName}.{DatabaseConstants.DeadLetterTable} SET {DatabaseConstants.Body} = ");
+        builder.AppendParameter(serialized);
+        builder.Append($", {DatabaseConstants.Replayable} = ");
+        builder.AppendParameter(1); // Oracle uses NUMBER(1) for bool
+        builder.Append($" WHERE {DatabaseConstants.Id} = ");
+        builder.AppendParameter(envelopeId);
+
+        var cmd = builder.Compile();
+
+        await using var conn = await _dataSource.OpenConnectionAsync(token);
+        try
+        {
+            cmd.Connection = conn;
+            await cmd.ExecuteNonQueryAsync(token);
+        }
+        finally
+        {
+            await conn.CloseAsync();
+        }
+    }
+
     private void writeDeadLetterWhereClause(DeadLetterEnvelopeQuery query, Weasel.Oracle.CommandBuilder builder)
     {
         if (query.Range.From.HasValue)

src/Persistence/Wolverine.CosmosDb/Internals/CosmosDbMessageStore.DeadLetters.cs

@@ -3,6 +3,7 @@
 using Microsoft.Azure.Cosmos;
 using Wolverine.Persistence.Durability;
 using Wolverine.Persistence.Durability.DeadLetterManagement;
+using Wolverine.Runtime.Serialization;
 
 namespace Wolverine.CosmosDb.Internals;
 
@@ -301,6 +302,28 @@ private async Task<List<DeadLetterMessage>> LoadDeadLettersByQuery(DeadLetterEnv
         return messages;
     }
 
+    public async Task EditAndReplayAsync(Guid envelopeId, byte[] newBody, CancellationToken token)
+    {
+        try
+        {
+            var response = await _container.ReadItemAsync<DeadLetterMessage>(DlqId(envelopeId),
+                new PartitionKey(DocumentTypes.DeadLetterPartition), cancellationToken: token);
+            var message = response.Resource;
+
+            // Deserialize the stored envelope, replace its Data with the new message body, re-serialize
+            var envelope = EnvelopeSerializer.Deserialize(message.Body);
+            envelope.Data = newBody;
+            message.Body = EnvelopeSerializer.Serialize(envelope);
+            message.Replayable = true;
+            await _container.ReplaceItemAsync(message, message.Id,
+                new PartitionKey(DocumentTypes.DeadLetterPartition), cancellationToken: token);
+        }
+        catch (CosmosException e) when (e.StatusCode == HttpStatusCode.NotFound)
+        {
+            // Already gone
+        }
+    }
+
     private QueryDefinition RebuildQueryDef(string queryText, DeadLetterEnvelopeQuery query)
     {
         var queryDef = new QueryDefinition(queryText)

src/Persistence/Wolverine.RDBMS/MessageDatabase.DeadLetterAdminService.cs

@@ -4,6 +4,7 @@
 using Weasel.Core;
 using Wolverine.Persistence.Durability.DeadLetterManagement;
 using Wolverine.RDBMS.Durability;
+using Wolverine.Runtime.Serialization;
 
 namespace Wolverine.RDBMS;
 
@@ -186,4 +187,26 @@ public Task ReplayAsync(DeadLetterEnvelopeQuery query, CancellationToken token)
         return executeCommandBatch(builder, token);
     }
 
+    public async Task EditAndReplayAsync(Guid envelopeId, byte[] newBody, CancellationToken token)
+    {
+        // Read the current serialized envelope, replace its Data, and re-serialize
+        var deadLetter = await DeadLetterEnvelopeByIdAsync(envelopeId);
+        if (deadLetter == null) return;
+
+        deadLetter.Envelope.Data = newBody;
+        var serialized = EnvelopeSerializer.Serialize(deadLetter.Envelope);
+
+        var builder = ToCommandBuilder();
+        builder.Append(
+            $"update {SchemaName}.{DatabaseConstants.DeadLetterTable} set {DatabaseConstants.Body} = ");
+        builder.AppendParameter(serialized);
+        builder.Append($", {DatabaseConstants.Replayable} = ");
+        builder.AppendParameter(true);
+        builder.Append($" where {DatabaseConstants.Id} = ");
+        builder.AppendParameter(envelopeId);
+        builder.Append(';');
+
+        await executeCommandBatch(builder, token);
+    }
+
 }

src/Persistence/Wolverine.RavenDb/Internals/RavenDbMessageStore.DeadLetters.cs

@@ -5,6 +5,7 @@
 using Raven.Client.Documents.Queries;
 using Wolverine.Persistence.Durability;
 using Wolverine.Persistence.Durability.DeadLetterManagement;
+using Wolverine.Runtime.Serialization;
 
 namespace Wolverine.RavenDb.Internals;
 
@@ -249,4 +250,18 @@ public async Task MarkDeadLetterEnvelopesAsReplayableAsync(Guid[] ids, string? t
     {
         await ReplayAsync(new DeadLetterEnvelopeQuery { MessageIds = ids }, CancellationToken.None);
     }
+
+    public async Task EditAndReplayAsync(Guid envelopeId, byte[] newBody, CancellationToken token)
+    {
+        using var session = _store.OpenAsyncSession();
+        var message = await session.LoadAsync<DeadLetterMessage>(dlqId(envelopeId), token);
+        if (message is null) return;
+
+        // Deserialize the stored envelope, replace its Data with the new message body, re-serialize
+        var envelope = EnvelopeSerializer.Deserialize(message.Body);
+        envelope.Data = newBody;
+        message.Body = EnvelopeSerializer.Serialize(envelope);
+        message.Replayable = true;
+        await session.SaveChangesAsync(token);
+    }
 }

src/Testing/BackPressureTests/Harness.cs

@@ -99,7 +99,9 @@ public async Task InitializeAsync()
     public async Task DisposeAsync()
     {
         await _sender.StopAsync();
+        _sender.Dispose();
         await _receiver.StopAsync();
+        _receiver.Dispose();
     }
 
     [Fact]

src/Testing/Benchmarks/Driver.cs

@@ -77,6 +77,7 @@ public async Task Teardown()
         if (Host != null)
         {
             await Host.StopAsync();
+            Host.Dispose();
             Host = null;
         }
     }

src/Testing/CoreTests/Acceptance/batch_processing.cs

@@ -50,6 +50,7 @@ public async Task InitializeAsync()
     public async Task DisposeAsync()
     {
         await theHost.StopAsync();
+        theHost.Dispose();
     }
 
     [Fact]

src/Testing/CoreTests/Acceptance/multi_tenancy.cs

@@ -29,6 +29,7 @@ public async Task InitializeAsync()
     public async Task DisposeAsync()
     {
         await _host.StopAsync();
+        _host.Dispose();
     }
 
     [Fact]

src/Testing/CoreTests/Acceptance/remote_invocation.cs

@@ -71,8 +71,11 @@ public async Task InitializeAsync()
     public async Task DisposeAsync()
     {
         await _receiver1.StopAsync();
+        _receiver1.Dispose();
         await _receiver2.StopAsync();
+        _receiver2.Dispose();
         await _sender.StopAsync();
+        _sender.Dispose();
     }
 
     [Fact]

src/Testing/CoreTests/Acceptance/using_with_keyed_services.cs

@@ -27,9 +27,10 @@ public async Task InitializeAsync()
             }).StartAsync();
     }
 
-    public Task DisposeAsync()
+    public async Task DisposeAsync()
     {
-        return _host.StopAsync();
+        await _host.StopAsync();
+        _host.Dispose();
     }
 
     [Fact]
@@ -82,9 +83,10 @@ public async Task InitializeAsync()
             }).StartAsync();
     }
 
-    public Task DisposeAsync()
+    public async Task DisposeAsync()
     {
-        return _host.StopAsync();
+        await _host.StopAsync();
+        _host.Dispose();
     }
 
     [Fact]

src/Testing/CoreTests/Bugs/Bug_2056_saga_code_generation.cs

@@ -30,6 +30,7 @@ public async Task InitializeAsync()
     public async Task DisposeAsync()
     {
         await _host.StopAsync();
+        _host.Dispose();
     }
 
     [Fact]

src/Testing/CoreTests/Bugs/Bug_2073_completed_saga_still_persisted.cs

@@ -26,6 +26,7 @@ public async Task InitializeAsync()
     public async Task DisposeAsync()
     {
         await _host.StopAsync();
+        _host.Dispose();
     }
 
     [Fact]

src/Testing/CoreTests/ErrorHandling/custom_error_action_raises_new_message.cs

@@ -52,7 +52,9 @@ public async Task InitializeAsync()
     public async Task DisposeAsync()
     {
         await theReceiver.StopAsync();
+        theReceiver.Dispose();
         await theSender.StopAsync();
+        theSender.Dispose();
     }
 
     [Fact]
@@ -130,7 +132,9 @@ public async Task InitializeAsync()
     public async Task DisposeAsync()
     {
         await theReceiver.StopAsync();
+        theReceiver.Dispose();
         await theSender.StopAsync();
+        theSender.Dispose();
     }
 
     [Fact]