[leader-election] Update (#423)

* Update

* Small changes

Author: v-federicoar

leader-election/DistributedMutex/BlobDistributedMutex.cs

@@ -7,29 +7,22 @@ namespace DistributedMutex
     using System.Threading;
     using System.Threading.Tasks;
 
-    public class BlobDistributedMutex
+    public class BlobDistributedMutex(BlobSettings blobSettings, Func<CancellationToken, Task> taskToRunWhenLeaseAcquired, Action? onLeaseTimeoutRetry = null)
     {
         private static readonly TimeSpan RenewInterval = TimeSpan.FromSeconds(10);
         private static readonly TimeSpan AcquireAttemptInterval = TimeSpan.FromSeconds(20);
-        private readonly BlobSettings blobSettings;
-        private readonly Func<CancellationToken, Task> taskToRunWhenLeaseAcquired;
-        private readonly Action? onLeaseTimeoutRetry;
+        private readonly BlobSettings blobSettings = blobSettings;
+        private readonly Func<CancellationToken, Task> taskToRunWhenLeaseAcquired = taskToRunWhenLeaseAcquired;
+        private readonly Action? onLeaseTimeoutRetry = onLeaseTimeoutRetry;
 
-        public BlobDistributedMutex(BlobSettings blobSettings, Func<CancellationToken, Task> taskToRunWhenLeaseAcquired, Action? onLeaseTimeoutRetry = null)
-        {
-            this.blobSettings = blobSettings;
-            this.taskToRunWhenLeaseAcquired = taskToRunWhenLeaseAcquired;
-            this.onLeaseTimeoutRetry = onLeaseTimeoutRetry;
-        }
-
-        public async Task RunTaskWhenMutexAcquired(CancellationToken token)
+        public async Task RunTaskWhenMutexAcquiredAsync(CancellationToken token)
         {
             var leaseManager = new BlobLeaseManager(blobSettings);
 
-            await RunTaskWhenBlobLeaseAcquired(leaseManager, token);
+            await RunTaskWhenBlobLeaseAcquiredAsync(leaseManager, token);
         }
 
-        private static async Task CancelAllWhenAnyCompletes(Task leaderTask, Task renewLeaseTask, CancellationTokenSource cts)
+        private static async Task CancelAllWhenAnyCompletesAsync(Task leaderTask, Task renewLeaseTask, CancellationTokenSource cts)
         {
             await Task.WhenAny(leaderTask, renewLeaseTask);
 
@@ -43,53 +36,48 @@ private static async Task CancelAllWhenAnyCompletes(Task leaderTask, Task renewL
             }
             catch (Exception)
             {
-                if (allTasks.Exception != null)
-                {
-                    allTasks.Exception.Handle(ex =>
+                allTasks.Exception?.Handle(ex =>
                     {
-                        if (!(ex is OperationCanceledException))
+                        if (ex is not OperationCanceledException)
                         {
                             Trace.TraceError(ex.Message);
                         }
 
                         return true;
                     });
-                }
             }
         }
 
-        private async Task RunTaskWhenBlobLeaseAcquired(BlobLeaseManager leaseManager, CancellationToken token)
+        private async Task RunTaskWhenBlobLeaseAcquiredAsync(BlobLeaseManager leaseManager, CancellationToken token)
         {
             while (!token.IsCancellationRequested)
             {
                 // Try to acquire the blob lease, otherwise wait for some time before we can try again.
-                string? leaseId = await TryAcquireLeaseOrWait(leaseManager, token);
+                string? leaseId = await TryAcquireLeaseOrWaitAsync(leaseManager, token);
 
                 if (!string.IsNullOrEmpty(leaseId))
                 {
                     // Create a new linked cancellation token source, so if either the
                     // original token is canceled or the lease cannot be renewed,
                     // then the leader task can be canceled.
-                    using (var leaseCts =
-                        CancellationTokenSource.CreateLinkedTokenSource([token]))
-                    {
-                        // Run the leader task.
-                        var leaderTask = taskToRunWhenLeaseAcquired.Invoke(leaseCts.Token);
-
-                        // Keeps renewing the lease in regular intervals.
-                        // If the lease cannot be renewed, then the task completes.
-                        var renewLeaseTask =
-                            KeepRenewingLease(leaseManager, leaseId, leaseCts.Token);
-
-                        // When any task completes (either the leader task or when it could
-                        // not renew the lease) then cancel the other task.
-                        await CancelAllWhenAnyCompletes(leaderTask, renewLeaseTask, leaseCts);
-                    }
+                    using var leaseCts =
+                        CancellationTokenSource.CreateLinkedTokenSource([token]);
+                    // Run the leader task.
+                    var leaderTask = taskToRunWhenLeaseAcquired.Invoke(leaseCts.Token);
+
+                    // Keeps renewing the lease in regular intervals.
+                    // If the lease cannot be renewed, then the task completes.
+                    var renewLeaseTask =
+                        KeepRenewingLeaseAsync(leaseManager, leaseId, leaseCts.Token);
+
+                    // When any task completes (either the leader task or when it could
+                    // not renew the lease) then cancel the other task.
+                    await CancelAllWhenAnyCompletesAsync(leaderTask, renewLeaseTask, leaseCts);
                 }
             }
         }
 
-        private async Task<string?> TryAcquireLeaseOrWait(BlobLeaseManager leaseManager, CancellationToken token)
+        private async Task<string?> TryAcquireLeaseOrWaitAsync(BlobLeaseManager leaseManager, CancellationToken token)
         {
             try
             {
@@ -98,10 +86,7 @@ private async Task RunTaskWhenBlobLeaseAcquired(BlobLeaseManager leaseManager, C
                 {
                     return leaseId;
                 }
-                if (onLeaseTimeoutRetry != null)
-                {
-                    onLeaseTimeoutRetry();
-                }
+                onLeaseTimeoutRetry?.Invoke();
                 await Task.Delay(AcquireAttemptInterval, token);
                 return null;
             }
@@ -111,7 +96,7 @@ private async Task RunTaskWhenBlobLeaseAcquired(BlobLeaseManager leaseManager, C
             }
         }
 
-        private async Task KeepRenewingLease(BlobLeaseManager leaseManager, string leaseId, CancellationToken token)
+        private async Task KeepRenewingLeaseAsync(BlobLeaseManager leaseManager, string leaseId, CancellationToken token)
         {
             var renewOffset = new Stopwatch();
 
@@ -136,7 +121,7 @@ private async Task KeepRenewingLease(BlobLeaseManager leaseManager, string lease
                     // If the adjusted interval is greater than zero wait for that long
                     if (renewIntervalAdjusted > TimeSpan.Zero)
                     {
-                        await Task.Delay(RenewInterval - renewOffset.Elapsed, token);
+                        await Task.Delay(renewIntervalAdjusted, token);
                     }
                 }
                 catch (OperationCanceledException)

leader-election/DistributedMutex/BlobLeaseManager.cs

@@ -38,6 +38,9 @@ internal class BlobLeaseManager
         private readonly BlobContainerClient leaseContainerClient;
         private readonly PageBlobClient leaseBlobClient;
 
+        private const int LeaseAcquireTimeoutSeconds = 15;
+        private const int LeaseAlreadyPresentStatusCode = 412;
+
         public BlobLeaseManager(BlobSettings settings)
             : this(settings.BlobServiceClient, settings.Container, settings.BlobName)
         {
@@ -48,7 +51,15 @@ public BlobLeaseManager(BlobServiceClient blobServiceClient, string leaseContain
             leaseContainerClient = blobServiceClient.GetBlobContainerClient(leaseContainerName);
             leaseBlobClient = leaseContainerClient.GetPageBlobClient(leaseBlobName);
             leaseContainerClient.CreateIfNotExists();
-            leaseBlobClient.CreateIfNotExists(512);
+            try
+            {
+                leaseBlobClient.CreateIfNotExists(512);
+            }
+            catch (RequestFailedException leaseAlreadyPresentException)
+            {
+                // There is currently a lease on the blob and no lease ID was specified in the request. Status=412. It throws an Exception if the lease exists and It is already taken.
+                if (leaseAlreadyPresentException.Status != LeaseAlreadyPresentStatusCode) throw;
+            }
         }
 
         public void ReleaseLease(string leaseId)
@@ -71,15 +82,15 @@ public void ReleaseLease(string leaseId)
             try
             {
                 var leaseClient = leaseBlobClient.GetBlobLeaseClient();
-                var lease = await leaseClient.AcquireAsync(TimeSpan.FromSeconds(15), null, token);
+                var lease = await leaseClient.AcquireAsync(TimeSpan.FromSeconds(LeaseAcquireTimeoutSeconds), null, token);
                 return lease.Value.LeaseId;
             }
             catch (RequestFailedException storageException)
             {
                 Trace.TraceError(storageException.ErrorCode);
 
                 var status = storageException.Status;
-                if (status == (int) HttpStatusCode.NotFound)
+                if (status == (int)HttpStatusCode.NotFound)
                 {
                     blobNotFound = true;
                 }

leader-election/DistributedMutex/DistributedMutex.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFramework>net8.0</TargetFramework>
+    <TargetFramework>net10.0</TargetFramework>
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
   </PropertyGroup>

leader-election/LeaderElectionConsoleWorker/LeaderElectionConsoleWorker.csproj

@@ -2,14 +2,14 @@
 
   <PropertyGroup>
     <OutputType>Exe</OutputType>
-    <TargetFramework>net8.0</TargetFramework>
+    <TargetFramework>net10.0</TargetFramework>
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
   </PropertyGroup>
   <ItemGroup>
     <PackageReference Include="Azure.Identity" Version="1.17.1" />
     <PackageReference Include="Azure.Storage.Blobs" Version="12.26.0" />
-    <PackageReference Include="System.Configuration.ConfigurationManager" Version="9.0.10" />
+    <PackageReference Include="System.Configuration.ConfigurationManager" Version="10.0.1" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\DistributedMutex\DistributedMutex.csproj">

leader-election/LeaderElectionConsoleWorker/Program.cs

@@ -14,7 +14,7 @@ class Program
         static async Task Main(string[] args)
         {
             // Create a new shared cancellation token source
-            CancellationTokenSource source = new CancellationTokenSource();
+            CancellationTokenSource source = new();
             CancellationToken token = source.Token;
 
             // Get the connection string from app settings
@@ -25,13 +25,13 @@ static async Task Main(string[] args)
                 return;
             }
             // Create a BlobSettings object with the connection string and the name of the blob to use for the lease
-            BlobSettings blobSettings = new BlobSettings(
+            BlobSettings blobSettings = new(
                 storageUri,
                 "leases",
                 "leader");
 
             // Get the current process ID for output
-            var pid = System.Diagnostics.Process.GetCurrentProcess().Id;
+            var pid = Environment.ProcessId;
 
             // Start an async task that will wait for a keypress and cancel the token when a key is pressed
             var uiTask = Task.Run(async () =>
@@ -41,32 +41,32 @@ static async Task Main(string[] args)
                     if (Console.KeyAvailable)
                     {
                         Console.ReadKey(true);
-                        Console.WriteLine($"[{DateTime.Now.ToString("HH:mm:ss")}] Requesting shutdown.");
+                        Console.WriteLine($"[{DateTime.Now:HH:mm:ss}] Requesting shutdown.");
                         source.Cancel();
                     }
                     await Task.Delay(500);
                 }
-                Console.WriteLine($"[{DateTime.Now.ToString("HH:mm:ss")}] This process ({pid}) is shutting down.");
+                Console.WriteLine($"[{DateTime.Now:HH:mm:ss}] This process ({pid}) is shutting down.");
 
             });
 
             // Create a new BlobDistributedMutex object with the BlobSettings object and a task
             // to run when the lease is acquired, and an action to run when the lease is not acquired.
-            BlobDistributedMutex distributedMutex = new BlobDistributedMutex(
+            BlobDistributedMutex distributedMutex = new(
                 blobSettings,
-                async (CancellationToken token) =>
+                async token =>
                 {
                     while (!token.IsCancellationRequested)
                     {
-                        Console.WriteLine($"[{DateTime.Now.ToString("HH:mm:ss")}] This process ({pid}) is currently the leader. Press any key to exit.");
-                        await Task.Delay(15000);
+                        Console.WriteLine($"[{DateTime.Now:HH:mm:ss}] This process ({pid}) is currently the leader. Press any key to exit.");
+                        await Task.Delay(15000, token);
                     }
-                }, () => {
-                    Console.WriteLine($"[{DateTime.Now.ToString("HH:mm:ss")}] This process ({pid}) could not acquire lease. Retrying in 20 seconds. Press any key to exit.");
+                },  async () => {
+                    Console.WriteLine($"[{DateTime.Now:HH:mm:ss}] This process ({pid}) could not acquire lease. Retrying in 20 seconds. Press any key to exit.");
                 });
 
             // Wait for completion of the DistributedMutex and the UI task before exiting
-            await distributedMutex.RunTaskWhenMutexAcquired(token);
+            await distributedMutex.RunTaskWhenMutexAcquiredAsync(token);
             await uiTask;
         }
     }

leader-election/Readme.md

@@ -12,7 +12,7 @@ Install the prerequisites and follow the steps to run the example and observe th
 
 ### Prerequisites
 
-- [.NET 8 SDK](https://dotnet.microsoft.com/download/dotnet/8.0)
+- [.NET 10 SDK](https://dotnet.microsoft.com/download/dotnet/10.0)
 - [Azurite emulator for local Azure Storage development](https://learn.microsoft.com/azure/storage/common/storage-use-azurite) or an [Azure Storage Account](https://learn.microsoft.com/azure/storage/common/storage-account-create)
 
 #### Optional
@@ -44,8 +44,8 @@ Install the prerequisites and follow the steps to run the example and observe th
 #### Running with Azurite storage emulator
 
    The included `app.config` file is set up to use a local Azure Storage emulator. Open a new terminal window, navigate to an empty working directory for the Azurite data files, and start the emulator with the command `azurite`, or `npx azurite` if you installed via `npm`.  
-   Azure SDKs by DefaultAzureCredencials needs https, and azurite by default is http. Follow the instructions [here](https://learn.microsoft.com/azure/storage/common/storage-use-azurite?tabs=visual-studio%2Cblob-storage#azure-sdks).  
-   In `LeaderElectionConsoleWorker/app.config` you see `https://127.0.0.1:10000/devstoreaccount1`
+   Azure SDKs by DefaultAzureCredential needs https, and Azurite by default is http. Follow the instructions [here](https://learn.microsoft.com/azure/storage/common/storage-use-azurite?tabs=visual-studio%2Cblob-storage#azure-sdks).  
+   In `LeaderElectionConsoleWorker/app.config` you see `https://127.0.0.1:10000/devstoreaccount1` (This is the Blob service endpoint Azurite exposes for the devstoreaccount1 account.)
 
 #### Running with Azure Storage account