LT-5770: [Trading Core] Potential Deduplication Lock Failure

src/MarginTrading.Backend.Services/Infrastructure/StartupDeduplicationService.cs

@@ -4,9 +4,15 @@
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+
+using Lykke.Common.Log;
+
 using MarginTrading.Backend.Core.Settings;
+using MarginTrading.Common.Services;
+
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.Hosting;
+
 using StackExchange.Redis;
 
 namespace MarginTrading.Backend.Services.Infrastructure
@@ -19,17 +25,18 @@ public class StartupDeduplicationService : IDisposable
     {
         private const string LockKey = "TradingEngine:DeduplicationLock";
         private readonly string _lockValue = Environment.MachineName;
+        private readonly WebHostProcessTerminator _processTerminator; 
         private readonly IWebHostEnvironment _hostingEnvironment; 
         private readonly MarginTradingSettings _marginTradingSettings;
         private readonly IDatabase _database;
 
-        private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
-
         public StartupDeduplicationService(
+            WebHostProcessTerminator processTerminator,
             IWebHostEnvironment hostingEnvironment,
             MarginTradingSettings marginTradingSettings,
             IConnectionMultiplexer redis)
         {
+            _processTerminator = processTerminator;
             _hostingEnvironment = hostingEnvironment;
             _marginTradingSettings = marginTradingSettings;
             _database = redis.GetDatabase();
@@ -60,34 +67,38 @@ public void HoldLock()
                 // exception is logged by the global handler
             }
 
-            Exception workerException = null;
-            // ReSharper disable once PossibleNullReferenceException
-            _cancellationTokenSource.Token.Register(() => throw workerException);
-
-            Task.Run(async () =>
-            {
-                try
+            Task.Run(
+                async () =>
                 {
-                    while (true)
+                    var run = true;
+                    while (run)
                     {
-                        // wait and extend lock
-                        await Task.Delay(_marginTradingSettings.DeduplicationLockExtensionPeriod);
+                        try
+                        {
+                            // wait and extend lock
+                            await Task.Delay(_marginTradingSettings.DeduplicationLockExtensionPeriod);
 
-                        await _database.LockExtendAsync(LockKey, _lockValue,
-                            _marginTradingSettings.DeduplicationLockExpiryPeriod);
+                            var extendResult = await _database.LockExtendAsync(
+                                LockKey,
+                                _lockValue,
+                                _marginTradingSettings.DeduplicationLockExpiryPeriod);
+                            if (!extendResult)
+                            {
+                                LogLocator.CommonLog?.Error("DeduplicationService", message: "Lock is taken already.");
+                                _processTerminator.TerminateProcess();
+                                run = false;
+                            }
+                        }
+                        catch (Exception exception)
+                        {
+                            LogLocator.CommonLog?.Warning("DeduplicationService","Failed to extend lock.", exception);
+                        }
                     }
-                }
-                catch (Exception exception)
-                {
-                    workerException = exception;
-                    _cancellationTokenSource.Cancel();
-                }
-            });
+                });
         }
 
         public void Dispose()
         {
-            _cancellationTokenSource.Dispose();
         }
     }
 }

src/MarginTrading.Backend.Services/Infrastructure/WebHostProcessTerminator.cs

@@ -0,0 +1,30 @@
+// Copyright (c) 2019 Lykke Corp.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Threading;
+
+namespace MarginTrading.Backend.Services.Infrastructure
+{
+    public sealed class WebHostProcessTerminator : IDisposable
+    {
+        private readonly CancellationTokenSource _cancellationTokenSource;
+
+        public WebHostProcessTerminator()
+        {
+            _cancellationTokenSource = new CancellationTokenSource();
+        }
+
+        public CancellationToken CancellationToken => _cancellationTokenSource.Token;
+
+        public void TerminateProcess()
+        {
+            _cancellationTokenSource?.Cancel();
+        }
+
+        public void Dispose()
+        {
+            _cancellationTokenSource?.Dispose();
+        }
+    }
+}

src/MarginTrading.Backend/Program.cs

@@ -6,9 +6,12 @@
 using System.Threading.Tasks;
 using Autofac.Extensions.DependencyInjection;
 using JetBrains.Annotations;
+
+using MarginTrading.Backend.Services.Infrastructure;
 using MarginTrading.Common.Services;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.PlatformAbstractions;
 
@@ -20,6 +23,7 @@ namespace MarginTrading.Backend
     public class Program
     {
         internal static IHost AppHost { get; private set; }
+        internal static WebHostProcessTerminator ProcessTerminator { get; private set; }
 
         public static async Task Main(string[] args)
         {
@@ -46,26 +50,29 @@ public static async Task Main(string[] args)
                 {
                     fatalErrorOccured = false;
 
-                    var configuration = new ConfigurationBuilder()
-                        .AddJsonFile("appsettings.json", optional: true)
-                        .AddUserSecrets<Startup>()
-                        .AddEnvironmentVariables()
-                        .Build();
+                    using (ProcessTerminator = new WebHostProcessTerminator())
+                    {
+                        var configuration = new ConfigurationBuilder()
+                            .AddJsonFile("appsettings.json", optional: true)
+                            .AddUserSecrets<Startup>()
+                            .AddEnvironmentVariables()
+                            .Build();
 
-                    AppHost = Host.CreateDefaultBuilder(args)
-                        .UseServiceProviderFactory(new AutofacServiceProviderFactory())
-                        .ConfigureWebHostDefaults(webBuilder =>
-                        {
-                            webBuilder.ConfigureKestrel(serverOptions =>
-                                {
-                                    // Set properties and call methods on options
-                                })
-                                .UseConfiguration(configuration)
-                                .UseStartup<Startup>();
-                        })
-                        .Build();
+                        AppHost = Host.CreateDefaultBuilder(args)
+                            .UseServiceProviderFactory(new AutofacServiceProviderFactory())
+                            .ConfigureWebHostDefaults(webBuilder =>
+                            {
+                                webBuilder.ConfigureKestrel(serverOptions =>
+                                    {
+                                        // Set properties and call methods on options
+                                    })
+                                    .UseConfiguration(configuration)
+                                    .UseStartup<Startup>();
+                            })
+                            .Build();
 
-                    await AppHost.RunAsync();
+                        await AppHost.RunAsync(ProcessTerminator.CancellationToken);   
+                    }
                 }
                 catch (Exception e)
                 {

src/MarginTrading.Backend/Startup.cs

@@ -354,7 +354,9 @@ private void InitializeJobs()
 
         private StartupDeduplicationService RunHealthChecks(IConnectionMultiplexer redis, MarginTradingSettings marginTradingSettings)
         {
-            var deduplicationService = new StartupDeduplicationService(Environment,
+            var deduplicationService = new StartupDeduplicationService(
+                Program.ProcessTerminator,
+                Environment,
                 marginTradingSettings,
                 redis);