diff --git a/Directory.Packages.props b/Directory.Packages.props index 50a7d8f03..cf8319548 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -127,13 +127,13 @@ - - - - - - - + + + + + + + diff --git a/src/Persistence/Oracle/OracleTests/oracle_durability_agent_recovery.cs b/src/Persistence/Oracle/OracleTests/oracle_durability_agent_recovery.cs new file mode 100644 index 000000000..531fc6855 --- /dev/null +++ b/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; + +/// +/// 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 @ 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. +/// +/// This runs the real set against a real Oracle +/// database, which is the assertion that actually would have caught the bug. +/// +/// +[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(); + 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 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 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()); + } +} diff --git a/src/Persistence/Oracle/OracleTests/oracle_durability_command_translation.cs b/src/Persistence/Oracle/OracleTests/oracle_durability_command_translation.cs new file mode 100644 index 000000000..7d1e6c800 --- /dev/null +++ b/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; + +/// +/// GH-3614. The durability agent batches several s into one +/// command builder. The generic builder emits @ 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. +/// +/// Oracle now hands back a Weasel.Oracle.OracleDbCommandBuilder, which emits : +/// markers, types parameters through OracleProvider, and splits at each statement boundary. +/// These assertions run against the real durability operations, and need no database. +/// +/// +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.Instance); + } + + private static IReadOnlyList 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(); + } + + [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().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() + .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() + .Single().OracleDbType.ShouldBe(OracleDbType.TimeStampTZ); + } +} diff --git a/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.cs b/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.cs index 167dbbad0..6bd25e8ac 100644 --- a/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.cs +++ b/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.cs @@ -387,10 +387,12 @@ public OracleSagaSchema SagaSchemaFor() 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; } diff --git a/src/Persistence/Oracle/Wolverine.Oracle/OracleNodePersistence.cs b/src/Persistence/Oracle/Wolverine.Oracle/OracleNodePersistence.cs index 7a91996c5..32b8cc616 100644 --- a/src/Persistence/Oracle/Wolverine.Oracle/OracleNodePersistence.cs +++ b/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 { diff --git a/src/Persistence/Wolverine.RDBMS/Durability/MoveReplayableErrorMessagesToIncomingOperation.cs b/src/Persistence/Wolverine.RDBMS/Durability/MoveReplayableErrorMessagesToIncomingOperation.cs index b528928b4..0ad88bd36 100644 --- a/src/Persistence/Wolverine.RDBMS/Durability/MoveReplayableErrorMessagesToIncomingOperation.cs +++ b/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 exceptions, CancellationToken token) diff --git a/src/Persistence/Wolverine.RDBMS/Durability/PersistNodeRecord.cs b/src/Persistence/Wolverine.RDBMS/Durability/PersistNodeRecord.cs index fc6469f22..354fbf5e4 100644 --- a/src/Persistence/Wolverine.RDBMS/Durability/PersistNodeRecord.cs +++ b/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 diff --git a/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesForAncillaryOperation.cs b/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesForAncillaryOperation.cs index 0fab09f6a..6bc606345 100644 --- a/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesForAncillaryOperation.cs +++ b/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesForAncillaryOperation.cs @@ -36,6 +36,11 @@ public void ConfigureCommand(DbCommandBuilder builder) builder.Append( $"update {incomingTable} set {DatabaseConstants.OwnerId} = 0 where {DatabaseConstants.OwnerId} != 0 and {DatabaseConstants.OwnerId} not in ({nodeList});"); + + // Two statements in one operation, so the boundary has to be explicit for the providers + // that cannot execute several statements from one command + builder.StartNewCommand(); + builder.Append( $"update {outgoingTable} set {DatabaseConstants.OwnerId} = 0 where {DatabaseConstants.OwnerId} != 0 and {DatabaseConstants.OwnerId} not in ({nodeList});"); } diff --git a/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesOperation.cs b/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesOperation.cs index 1d2639b9d..0da3bff93 100644 --- a/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesOperation.cs +++ b/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesOperation.cs @@ -30,6 +30,11 @@ public void ConfigureCommand(DbCommandBuilder builder) builder.Append( $"update {incomingTable} set {DatabaseConstants.OwnerId} = 0 where {DatabaseConstants.OwnerId} != 0 and {DatabaseConstants.OwnerId} not in (select {DatabaseConstants.NodeNumber} from {nodesTable});"); + + // Two statements in one operation, so the boundary has to be explicit for the providers + // that cannot execute several statements from one command + builder.StartNewCommand(); + builder.Append( $"update {outgoingTable} set {DatabaseConstants.OwnerId} = 0 where {DatabaseConstants.OwnerId} != 0 and {DatabaseConstants.OwnerId} not in (select {DatabaseConstants.NodeNumber} from {nodesTable});"); } diff --git a/src/Persistence/Wolverine.RDBMS/Polling/DatabaseOperationBatch.cs b/src/Persistence/Wolverine.RDBMS/Polling/DatabaseOperationBatch.cs index c16595f06..e0d71f2a9 100644 --- a/src/Persistence/Wolverine.RDBMS/Polling/DatabaseOperationBatch.cs +++ b/src/Persistence/Wolverine.RDBMS/Polling/DatabaseOperationBatch.cs @@ -48,43 +48,87 @@ public async Task ExecuteAsync(IWolverineRuntime runtime, if (_operations.Length == 0) return AgentCommands.Empty; var builder = _database.ToCommandBuilder(); - foreach (var operation in _operations) operation.ConfigureCommand(builder); - await using var cmd = builder.Compile(); + // Mark a statement boundary before each operation and remember which command each one's + // results will come back on. On every provider whose driver can execute several statements + // from a single command -- which is all of them except Oracle -- StartNewCommand() is a + // no-op, CompileCommands() returns one command holding every statement, and all of this + // collapses to exactly what it has always done. Oracle's builder splits at each boundary. + // + // An operation may contribute more than one statement (ReleaseOrphanedMessagesOperation, + // MoveReplayableErrorMessagesToIncomingOperation, PersistNodeRecord) or none at all + // (ReleaseOrphanedMessagesForAncillaryOperation, when it has no active node numbers). Every + // such operation is IDoNotReturnData -- each of the four that *do* return data appends + // exactly one statement unconditionally -- so associating an operation with the first command + // it produced is enough. Any further commands execute with no callbacks, and a zero-statement + // operation clamps into the last group where ApplyCallbacksAsync skips it as IDoNotReturnData. + var starts = new int[_operations.Length]; + for (var i = 0; i < _operations.Length; i++) + { + builder.StartNewCommand(); + + // Nothing is open at this point, so the builder's count is the index the next command + // it produces will occupy + starts[i] = builder.CommandCount; + _operations[i].ConfigureCommand(builder); + } + + var commands = builder.CompileCommands(); + if (commands.Count == 0) return AgentCommands.Empty; + + var groups = new List[commands.Count]; + for (var i = 0; i < commands.Count; i++) groups[i] = []; + for (var i = 0; i < _operations.Length; i++) + { + groups[Math.Min(starts[i], commands.Count - 1)].Add(_operations[i]); + } await using var conn = await _database.DataSource.OpenConnectionAsync(cancellationToken); - cmd.Connection = conn; var tx = await conn.BeginTransactionAsync(cancellationToken); - cmd.Transaction = tx; + DbCommand? current = null; try { - await using var reader = await cmd.ExecuteReaderAsync(cancellationToken); var exceptions = new List(); - await ApplyCallbacksAsync(_operations, reader, exceptions, cancellationToken); - await reader.CloseAsync(); + + for (var i = 0; i < commands.Count; i++) + { + current = commands[i]; + current.Connection = conn; + current.Transaction = tx; + + await using var reader = await current.ExecuteReaderAsync(cancellationToken); + if (groups[i].Count != 0) + { + await ApplyCallbacksAsync(groups[i], reader, exceptions, cancellationToken); + } + + await reader.CloseAsync(); + } await tx.CommitAsync(cancellationToken); } catch (ObjectDisposedException) { - // The system is shutting down, let this go. + // The system is shutting down, let this go. } catch (Exception e) { await conn.CloseAsync(); - throw new DatabaseBatchCommandException(cmd, _operations, e); + throw new DatabaseBatchCommandException(current!, _operations, e); } - - try + finally { - var commands = new AgentCommands(); - foreach (var operation in _operations) + foreach (var command in commands) { - commands.AddRange(operation.PostProcessingCommands()); + await command.DisposeAsync(); } - return commands; + } + + try + { + return postProcessingCommands(); } finally { @@ -99,6 +143,17 @@ public async Task ExecuteAsync(IWolverineRuntime runtime, } } + private AgentCommands postProcessingCommands() + { + var commands = new AgentCommands(); + foreach (var operation in _operations) + { + commands.AddRange(operation.PostProcessingCommands()); + } + + return commands; + } + public static async Task ApplyCallbacksAsync(IReadOnlyList operations, DbDataReader reader, IList exceptions, CancellationToken token)