Oracle: run the durability agent through the shared batching mechanics

Directory.Packages.props

@@ -127,13 +127,13 @@
     <PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="9.0.5" />
     <PackageVersion Include="System.Net.NameResolution" Version="4.3.0" />
     <PackageVersion Include="System.Threading.Tasks.Dataflow" Version="9.0.5" />
-    <PackageVersion Include="Weasel.Core" Version="9.18.1" />
-    <PackageVersion Include="Weasel.EntityFrameworkCore" Version="9.18.1" />
-    <PackageVersion Include="Weasel.MySql" Version="9.18.1" />
-    <PackageVersion Include="Weasel.Oracle" Version="9.18.1" />
-    <PackageVersion Include="Weasel.Postgresql" Version="9.18.1" />
-    <PackageVersion Include="Weasel.SqlServer" Version="9.18.1" />
-    <PackageVersion Include="Weasel.Sqlite" Version="9.18.1" />
+    <PackageVersion Include="Weasel.Core" Version="9.19.0" />
+    <PackageVersion Include="Weasel.EntityFrameworkCore" Version="9.19.0" />
+    <PackageVersion Include="Weasel.MySql" Version="9.19.0" />
+    <PackageVersion Include="Weasel.Oracle" Version="9.19.0" />
+    <PackageVersion Include="Weasel.Postgresql" Version="9.19.0" />
+    <PackageVersion Include="Weasel.SqlServer" Version="9.19.0" />
+    <PackageVersion Include="Weasel.Sqlite" Version="9.19.0" />
     <PackageVersion Include="xunit" Version="2.9.3" />
     <PackageVersion Include="xunit.assemblyfixture" Version="2.2.0" />
     <PackageVersion Include="xunit.runner.visualstudio" Version="2.8.2" />

src/Persistence/Oracle/OracleTests/oracle_durability_agent_recovery.cs

@@ -0,0 +1,145 @@
+using IntegrationTests;
+using JasperFx.Resources;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Oracle.ManagedDataAccess.Client;
+using Shouldly;
+using Wolverine;
+using Wolverine.ComplianceTests;
+using Wolverine.Oracle;
+using Wolverine.Persistence.Durability;
+using Wolverine.RDBMS;
+using Wolverine.RDBMS.Polling;
+using Wolverine.Runtime;
+using Wolverine.Transports;
+
+namespace OracleTests;
+
+/// <summary>
+///     GH-3614, end to end. The durability agent batches its whole recovery operation set into one
+///     command builder and executes it. On Oracle that batch used to arrive as a single command full
+///     of semicolon-separated statements bound with <c>@</c> markers, which ODP.NET rejected outright
+///     with ORA-00933 / ORA-00936 -- so the agent threw on every sweep and nothing persisted in the
+///     inbox or outbox was ever recovered.
+///     <para>
+///     This runs the real <see cref="DurabilityAgent.buildOperationBatch" /> set against a real Oracle
+///     database, which is the assertion that actually would have caught the bug.
+///     </para>
+/// </summary>
+[Collection("oracle")]
+public class oracle_durability_agent_recovery : IAsyncLifetime
+{
+    private IHost theHost = null!;
+
+    public async Task InitializeAsync()
+    {
+        theHost = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.PersistMessagesWithOracle(Servers.OracleConnectionString, "WOLVERINE");
+
+                // Balanced rather than Solo -- ReleaseOrphanedMessagesOperation is only part of the
+                // recovery batch outside Solo mode, and it is one of the two-statement operations
+                opts.Durability.Mode = DurabilityMode.Balanced;
+            })
+            .StartAsync();
+
+        await theHost.ResetResourceState();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await theHost.StopAsync();
+        theHost.Dispose();
+    }
+
+    private (IWolverineRuntime, IMessageDatabase) theRuntimeAndDatabase()
+    {
+        var runtime = theHost.Services.GetRequiredService<IWolverineRuntime>();
+        return (runtime, (IMessageDatabase)runtime.Storage);
+    }
+
+    [Fact]
+    public async Task the_full_recovery_batch_executes_against_oracle()
+    {
+        var (runtime, database) = theRuntimeAndDatabase();
+
+        var operations = new DurabilityAgent(runtime, database).buildOperationBatch();
+        operations.ShouldNotBeEmpty();
+
+        // Before the fix this threw DatabaseBatchCommandException wrapping ORA-00933 / ORA-00936
+        await new DatabaseOperationBatch(database, operations).ExecuteAsync(runtime, CancellationToken.None);
+    }
+
+    [Fact]
+    public async Task the_recovery_batch_releases_messages_owned_by_a_dead_node()
+    {
+        var (runtime, database) = theRuntimeAndDatabase();
+
+        // Persist an incoming envelope owned by a node number that does not exist any more --
+        // exactly the state ReleaseOrphanedMessagesOperation is there to clean up
+        var envelope = ObjectMother.Envelope();
+        envelope.Status = EnvelopeStatus.Incoming;
+        envelope.OwnerId = 8888;
+
+        await database.Inbox.StoreIncomingAsync(envelope);
+        (await ownerOf(envelope.Id)).ShouldBe(8888);
+
+        await new DatabaseOperationBatch(database, new DurabilityAgent(runtime, database).buildOperationBatch())
+            .ExecuteAsync(runtime, CancellationToken.None);
+
+        (await ownerOf(envelope.Id)).ShouldBe(TransportConstants.AnyNode);
+    }
+
+    [Fact]
+    public async Task replayable_dead_letters_are_moved_back_to_the_inbox()
+    {
+        var (runtime, database) = theRuntimeAndDatabase();
+
+        var envelope = ObjectMother.Envelope();
+        envelope.Status = EnvelopeStatus.Incoming;
+
+        await database.Inbox.StoreIncomingAsync(envelope);
+        await database.Inbox.MoveToDeadLetterStorageAsync(envelope, new DivideByZeroException("boom"));
+        await database.DeadLetters.MarkDeadLetterEnvelopesAsReplayableAsync([envelope.Id]);
+
+        // This is the operation that writes two statements -- the insert and the delete -- so it is
+        // the one that proves per-statement splitting works, not just per-operation
+        await new DatabaseOperationBatch(database, new DurabilityAgent(runtime, database).buildOperationBatch())
+            .ExecuteAsync(runtime, CancellationToken.None);
+
+        (await countAsync(
+                $"select count(*) from WOLVERINE.{DatabaseConstants.IncomingTable} where id = :id", envelope.Id))
+            .ShouldBe(1);
+        (await countAsync(
+                $"select count(*) from WOLVERINE.{DatabaseConstants.DeadLetterTable} where id = :id", envelope.Id))
+            .ShouldBe(0);
+    }
+
+    private async Task<int> ownerOf(Guid id)
+    {
+        await using var conn = new OracleConnection(Servers.OracleConnectionString);
+        await conn.OpenAsync();
+
+        await using var command = conn.CreateCommand();
+        command.BindByName = true;
+        command.CommandText =
+            $"select owner_id from WOLVERINE.{DatabaseConstants.IncomingTable} where id = :id";
+        command.Parameters.Add(new OracleParameter("id", OracleDbType.Raw) { Value = id.ToByteArray() });
+
+        return Convert.ToInt32(await command.ExecuteScalarAsync());
+    }
+
+    private async Task<int> countAsync(string sql, Guid id)
+    {
+        await using var conn = new OracleConnection(Servers.OracleConnectionString);
+        await conn.OpenAsync();
+
+        await using var command = conn.CreateCommand();
+        command.BindByName = true;
+        command.CommandText = sql;
+        command.Parameters.Add(new OracleParameter("id", OracleDbType.Raw) { Value = id.ToByteArray() });
+
+        return Convert.ToInt32(await command.ExecuteScalarAsync());
+    }
+}

src/Persistence/Oracle/OracleTests/oracle_durability_command_translation.cs

@@ -0,0 +1,159 @@
+using System.Data.Common;
+using IntegrationTests;
+using JasperFx.Core;
+using Microsoft.Extensions.Logging.Abstractions;
+using Oracle.ManagedDataAccess.Client;
+using Shouldly;
+using Weasel.Core;
+using Wolverine;
+using Wolverine.Oracle;
+using Wolverine.Persistence.Durability;
+using Wolverine.RDBMS;
+using Wolverine.RDBMS.Durability;
+using Wolverine.RDBMS.Polling;
+
+namespace OracleTests;
+
+/// <summary>
+///     GH-3614. The durability agent batches several <see cref="IDatabaseOperation" />s into one
+///     command builder. The generic builder emits <c>@</c> bind markers and concatenates every
+///     statement into a single command, and ODP.NET rejects both -- with ORA-00933 / ORA-00936 --
+///     so no persisted inbox or outbox message was ever recovered on Oracle.
+///     <para>
+///     Oracle now hands back a <c>Weasel.Oracle.OracleDbCommandBuilder</c>, which emits <c>:</c>
+///     markers, types parameters through OracleProvider, and splits at each statement boundary.
+///     These assertions run against the real durability operations, and need no database.
+///     </para>
+/// </summary>
+public class oracle_durability_command_translation
+{
+    private static OracleMessageStore theStore()
+    {
+        return new OracleMessageStore(
+            new DatabaseSettings { SchemaName = "WOLVERINE", Role = MessageStoreRole.Main },
+            new DurabilitySettings(),
+            new OracleDataSource(Servers.OracleConnectionString),
+            NullLogger<OracleMessageStore>.Instance);
+    }
+
+    private static IReadOnlyList<DbCommand> compile(params IDatabaseOperation[] operations)
+    {
+        var builder = theStore().ToCommandBuilder();
+
+        foreach (var operation in operations)
+        {
+            builder.StartNewCommand();
+            operation.ConfigureCommand(builder);
+        }
+
+        return builder.CompileCommands();
+    }
+
+    [Fact]
+    public void the_message_store_hands_back_an_oracle_shaped_builder()
+    {
+        theStore().ToCommandBuilder().ShouldBeOfType<Weasel.Oracle.OracleDbCommandBuilder>();
+    }
+
+    [Fact]
+    public void uses_oracle_bind_markers_rather_than_the_generic_ones()
+    {
+        var commands = compile(
+            new DeleteExpiredEnvelopesOperation(new DbObjectName("WOLVERINE", "wolverine_incoming_envelopes"),
+                DateTimeOffset.UtcNow));
+
+        var sql = commands.Single().CommandText;
+
+        sql.ShouldContain(":p0");
+        sql.ShouldNotContain("@");
+    }
+
+    [Fact]
+    public void separates_batched_operations_into_individual_oracle_statements()
+    {
+        var store = theStore();
+
+        var commands = compile(
+            new DeleteExpiredEnvelopesOperation(new DbObjectName("WOLVERINE", "wolverine_incoming_envelopes"),
+                DateTimeOffset.UtcNow),
+            new MoveReplayableErrorMessagesToIncomingOperation(store),
+            new DeleteOldNodeEventRecords(store, new DurabilitySettings()));
+
+        // One command per *statement*, not per operation -- ODP.NET cannot execute several statements
+        // from one command, and MoveReplayableErrorMessagesToIncomingOperation writes two (the insert
+        // and the delete)
+        commands.Count.ShouldBe(4);
+
+        foreach (var command in commands)
+        {
+            command.ShouldBeOfType<OracleCommand>().BindByName.ShouldBeTrue();
+            command.CommandText.ShouldNotContain(";");
+        }
+    }
+
+    [Fact]
+    public void each_statement_only_carries_its_own_parameters()
+    {
+        var commands = compile(
+            new DeleteExpiredEnvelopesOperation(new DbObjectName("WOLVERINE", "wolverine_incoming_envelopes"),
+                DateTimeOffset.UtcNow),
+            new DeleteExpiredDeadLetterMessagesOperation(theStore(), NullLogger.Instance, DateTimeOffset.UtcNow));
+
+        commands.Count.ShouldBe(2);
+
+        foreach (var command in commands)
+        {
+            foreach (DbParameter parameter in command.Parameters)
+            {
+                command.CommandText.ShouldContain(":" + parameter.ParameterName);
+            }
+        }
+    }
+
+    [Fact]
+    public void converts_boolean_parameters_to_an_oracle_number()
+    {
+        var commands = compile(new MoveReplayableErrorMessagesToIncomingOperation(theStore()));
+
+        // The insert and the delete both bind :replayable, and it is only ever created once, so
+        // both split commands have to carry it
+        commands.Count.ShouldBe(2);
+
+        foreach (var command in commands)
+        {
+            var parameter = command.Parameters
+                .Cast<OracleParameter>()
+                .Single(x => x.ParameterName == "replayable");
+
+            parameter.OracleDbType.ShouldBe(OracleDbType.Int16);
+            parameter.Value.ShouldBe(1);
+        }
+    }
+
+    [Fact]
+    public void converts_guid_parameters_to_oracle_raw()
+    {
+        var id = Guid.NewGuid();
+        var builder = theStore().ToCommandBuilder();
+
+        builder.Append("select 1 from dual where id = ");
+        builder.AppendParameter(id);
+
+        var parameter = (OracleParameter)builder.CompileCommands().Single().Parameters[0];
+
+        parameter.OracleDbType.ShouldBe(OracleDbType.Raw);
+        parameter.Value.ShouldBe(id.ToByteArray());
+    }
+
+    [Fact]
+    public void date_time_offsets_keep_their_oracle_type()
+    {
+        var commands = compile(
+            new DeleteExpiredEnvelopesOperation(new DbObjectName("WOLVERINE", "wolverine_incoming_envelopes"),
+                DateTimeOffset.UtcNow));
+
+        commands.Single().Parameters
+            .Cast<OracleParameter>()
+            .Single().OracleDbType.ShouldBe(OracleDbType.TimeStampTZ);
+    }
+}

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

@@ -387,10 +387,12 @@ public OracleSagaSchema<T, TId> SagaSchemaFor<T, TId>() where T : Saga
     // IMessageDatabase - extra methods
     public Weasel.Core.DbCommandBuilder ToCommandBuilder()
     {
-        // The IMessageDatabase interface requires DbCommandBuilder, but we create an OracleCommandBuilder
-        // internally. Return a DbCommandBuilder that uses Oracle's OracleCommand as the underlying command.
+        // OracleDbCommandBuilder is a DbCommandBuilder, so it satisfies IMessageDatabase, but it emits
+        // Oracle's ':' bind markers instead of the generic '@', types parameters through OracleProvider
+        // (Guid as RAW(16), bool as NUMBER(1)), and -- because ODP.NET cannot execute several statements
+        // from one command -- splits at each StartNewCommand() boundary into one command per statement.
         // Our dead letter methods use ToOracleCommandBuilder() instead.
-        return new Weasel.Core.DbCommandBuilder(CreateConnection());
+        return new Weasel.Oracle.OracleDbCommandBuilder();
     }
 
     internal Weasel.Oracle.CommandBuilder ToOracleCommandBuilder()
@@ -403,7 +405,11 @@ internal Weasel.Oracle.CommandBuilder ToOracleCommandBuilder()
 
     public Task EnqueueAsync(IDatabaseOperation operation)
     {
-        // For Oracle, we execute operations directly since we can't batch
+        // NOTE: this silently drops the operation. OracleMessageStore implements IMessageDatabase
+        // directly rather than deriving from MessageDatabase, so it has no DatabaseBatcher to hand
+        // the operation to. The durability agent does not use this path -- it builds its own
+        // DatabaseOperationBatch -- and the one caller that does, OracleNodePersistence.LogRecordsAsync,
+        // works around it by inserting directly. Tracked separately; see the comment there.
         return Task.CompletedTask;
     }
 

src/Persistence/Oracle/Wolverine.Oracle/OracleNodePersistence.cs

@@ -340,9 +340,8 @@ public async Task LogRecordsAsync(params NodeRecord[] records)
     {
         if (records.Length == 0) return;
 
-        // OracleMessageStore.EnqueueAsync is a no-op (the shared DatabaseBatcher uses
-        // @-prefixed parameter syntax that Oracle rejects), so insert each record
-        // directly using the Oracle command extensions.
+        // OracleMessageStore.EnqueueAsync is a no-op -- it has no DatabaseBatcher to hand the
+        // operation to -- so insert each record directly using the Oracle command extensions.
         await using var conn = await _dataSource.OpenConnectionAsync();
         try
         {

src/Persistence/Wolverine.RDBMS/Durability/MoveReplayableErrorMessagesToIncomingOperation.cs

@@ -24,10 +24,15 @@ public void ConfigureCommand(DbCommandBuilder builder)
         builder.Append(
             $"select {DatabaseConstants.Body}, {DatabaseConstants.Id}, '{EnvelopeStatus.Incoming}', 0, null, 0, {DatabaseConstants.MessageType}, {DatabaseConstants.ReceivedAt}, null ");
         builder.Append(
-            $"from {_database.SchemaName}.{DatabaseConstants.DeadLetterTable} where {DatabaseConstants.Replayable} = @replayable;");
+            $"from {_database.SchemaName}.{DatabaseConstants.DeadLetterTable} where {DatabaseConstants.Replayable} = {builder.ParameterPrefix}replayable;");
         builder.AddNamedParameter("replayable", true);
+
+        // This operation writes two statements, so the boundary has to be explicit -- the trailing
+        // semicolon is enough for the providers that concatenate, but Oracle needs a real split
+        builder.StartNewCommand();
+
         builder.Append(
-            $"delete from {_database.SchemaName}.{DatabaseConstants.DeadLetterTable} where {DatabaseConstants.Replayable} = @replayable;");
+            $"delete from {_database.SchemaName}.{DatabaseConstants.DeadLetterTable} where {DatabaseConstants.Replayable} = {builder.ParameterPrefix}replayable;");
     }
 
     public Task ReadResultsAsync(DbDataReader reader, IList<Exception> exceptions, CancellationToken token)

src/Persistence/Wolverine.RDBMS/Durability/PersistNodeRecord.cs

@@ -23,6 +23,10 @@ public void ConfigureCommand(DbCommandBuilder builder)
 
         foreach (var @event in _events)
         {
+            // One insert per event, so each is its own statement for the providers that cannot
+            // execute several from one command
+            builder.StartNewCommand();
+
             builder.Append("insert into ");
 
             // GH-2940: emit the schema identifier unquoted, matching every other durability SQL