LM-2622 Inconsistent order state (#8)

* - Improved tx hash projection logging - Added dlx for the ExecutionQueueReader * LM-2622 Incosistent order state - Improved tx hash projection logging
LykkeCity · Dec 2, 2022 · 8016123 · 8016123
1 parent f2ffbbd
commit 8016123
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 10 deletions.
diff --git a/src/Antares.Job.History/Workflow/ExecutionProcessing/BaseBatchQueueReader.cs b/src/Antares.Job.History/Workflow/ExecutionProcessing/BaseBatchQueueReader.cs
@@ -104,7 +104,18 @@ private async Task StartQueueReader()
             {
                 channel.BasicQos(0, (ushort)_prefetchCount, false);
 
-                channel.QueueDeclare(QueueName, true, false, false);
+                // see https://www.rabbitmq.com/dlx.html
+                var dlxName = $"{QueueName}-dlx";
+                var poisonQueueName = $"{QueueName}-poison";
+                var args = new Dictionary<string, object>
+                {
+                    { "x-dead-letter-exchange", dlxName }
+                };
+                channel.ExchangeDeclare(dlxName, type: "direct", durable: true);
+                channel.QueueDeclare(poisonQueueName, durable: true, exclusive: false, autoDelete: false);
+                channel.QueueBind(poisonQueueName, exchange: $"{QueueName}-dlx", routingKey: string.Empty);
+
+                channel.QueueDeclare(QueueName, durable: true, exclusive: false, autoDelete: false);
 
                 foreach (var key in RoutingKeys)
                     channel.QueueBind(QueueName, ExchangeName, key);

diff --git a/src/Antares.Job.History/Workflow/Projections/TransactionHashProjection.cs b/src/Antares.Job.History/Workflow/Projections/TransactionHashProjection.cs
@@ -29,7 +29,7 @@ public async Task<CommandHandlingResult> Handle(CashinCompletedEvent @event)
         {
             if (!await _historyRecordsRepository.UpdateBlockchainHashAsync(@event.OperationId, @event.TxHash))
             {
-                _logger.Warning($"Transaction hash was not set", context: new
+                _logger.Warning($"Bitcoin cashin hash was not set", context: new
                 {
                     id = @event.OperationId,
                     hash = @event.TxHash
@@ -50,7 +50,7 @@ public async Task<CommandHandlingResult> Handle(CashoutCompletedEvent @event)
         {
             if (!await _historyRecordsRepository.UpdateBlockchainHashAsync(@event.OperationId, @event.TxHash))
             {
-                _logger.Warning($"Transaction hash was not set", context: new
+                _logger.Warning($"Bitcoin cashout hash was not set", context: new
                 {
                     id = @event.OperationId,
                     hash = @event.TxHash
@@ -72,7 +72,7 @@ public async Task<CommandHandlingResult> Handle(
         {
             if (!await _historyRecordsRepository.UpdateBlockchainHashAsync(@event.OperationId, @event.TransactionHash))
             {
-                _logger.Warning($"Transaction hash was not set", context: new
+                _logger.Warning($"BIL cashin hash was not set", context: new
                 {
                     id = @event.OperationId,
                     hash = @event.TransactionHash
@@ -94,7 +94,7 @@ public async Task<CommandHandlingResult> Handle(
         {
             if (!await _historyRecordsRepository.UpdateBlockchainHashAsync(@event.OperationId, @event.TransactionHash))
             {
-                _logger.Warning($"Transaction hash was not set", context: new
+                _logger.Warning($"Sirius cashin hash was not set", context: new
                 {
                     id = @event.OperationId,
                     hash = @event.TransactionHash
@@ -116,7 +116,7 @@ public async Task<CommandHandlingResult> Handle(
         {
             if (!await _historyRecordsRepository.UpdateBlockchainHashAsync(@event.OperationId, @event.TransactionHash))
             {
-                _logger.Warning($"Transaction hash was not set", context: new
+                _logger.Warning($"BIL cashout hash was not set", context: new
                 {
                     id = @event.OperationId,
                     hash = @event.TransactionHash
@@ -138,7 +138,7 @@ public async Task<CommandHandlingResult> Handle(
         {
             if (!await _historyRecordsRepository.UpdateBlockchainHashAsync(@event.OperationId, @event.TransactionHash))
             {
-                _logger.Warning($"Transaction hash was not set", context: new
+                _logger.Warning($"Sirius cashout hash was not set", context: new
                 {
                     id = @event.OperationId,
                     hash = @event.TransactionHash
@@ -170,7 +170,7 @@ public async Task<CommandHandlingResult> Handle(
                 if (!await _historyRecordsRepository.UpdateBlockchainHashAsync(cashout.OperationId,
                     @event.TransactionHash))
                 {
-                    _logger.Warning($"Transaction hash was not set, BIL cashout", context: new
+                    _logger.Warning($"BIL batch transaction hash was not set", context: new
                     {
                         id = cashout.OperationId,
                         hash = @event.TransactionHash
@@ -194,7 +194,7 @@ public async Task<CommandHandlingResult> Handle(
             if (!await _historyRecordsRepository.UpdateBlockchainHashAsync(@event.OperationId,
                 _crossClientTransactionHashSubstituition))
             {
-                _logger.Warning($"Transaction hash was not set. " +
+                _logger.Warning($"BIL cross-client cashout hash was not set. " +
                                 $"OperationId: {@event.OperationId}, " +
                                 $"TxHash: {_crossClientTransactionHashSubstituition}", context: new
                                 {
@@ -208,7 +208,7 @@ public async Task<CommandHandlingResult> Handle(
             if (!await _historyRecordsRepository.UpdateBlockchainHashAsync(@event.CashinOperationId,
                 _crossClientTransactionHashSubstituition))
             {
-                _logger.Warning($"Transaction cashin hash was not set. " +
+                _logger.Warning($"BIL cross-client cashin hash was not set. " +
                                 $"OperationId: {@event.OperationId}, " +
                                 $"TxHash: {_crossClientTransactionHashSubstituition}", context: new
                 {