Add timeout wehn Read&Write Parititon Table (#1200)

* initial commit

* make this timeout configurable

Author: nytian

src/DurableTask.ApplicationInsights/DurableTask.ApplicationInsights.csproj

@@ -11,7 +11,7 @@
     <!-- Version Info -->
     <PropertyGroup>
         <MajorVersion>0</MajorVersion>
-        <MinorVersion>2</MinorVersion>
+        <MinorVersion>3</MinorVersion>
         <PatchVersion>0</PatchVersion>
         <VersionPrefix>$(MajorVersion).$(MinorVersion).$(PatchVersion)</VersionPrefix>
         <FileVersion>$(VersionPrefix).0</FileVersion>

src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs

@@ -170,6 +170,14 @@ public class AzureStorageOrchestrationServiceSettings
         /// </remarks>
         public bool AllowReplayingTerminalInstances { get; set; } = false;
 
+        /// <summary>
+        /// Specifies the timeout (in seconds) for read and write operations on the partition table in partition manager V3 (table partition manager).
+        /// This helps detect and recover from potential silent hangs caused by Azure Storage client's internal retries. 
+        /// If the operation exceeds the timeout, a PartitionManagerWarning is logged and the operation is retried.
+        /// The default time is 2 seconds.
+        /// </summary>
+        public TimeSpan PartitionTableOperationTimeout { get; set; } = TimeSpan.FromSeconds(2);
+
         /// <summary>
         /// If UseAppLease is true, gets or sets the AppLeaseOptions used for acquiring the lease to start the application.
         /// </summary>

src/DurableTask.AzureStorage/DurableTask.AzureStorage.csproj

@@ -21,8 +21,8 @@
   <!-- Version Info -->
   <PropertyGroup>
     <MajorVersion>2</MajorVersion>
-    <MinorVersion>0</MinorVersion>
-    <PatchVersion>2</PatchVersion>
+    <MinorVersion>1</MinorVersion>
+    <PatchVersion>0</PatchVersion>
     <VersionPrefix>$(MajorVersion).$(MinorVersion).$(PatchVersion)</VersionPrefix>
     <FileVersion>$(VersionPrefix).0</FileVersion>
     <!-- FileVersionRevision is expected to be set by the CI. This is useful for distinguishing between multiple builds of the same version. -->

src/DurableTask.AzureStorage/Partitioning/TablePartitionManager.cs

@@ -118,7 +118,10 @@ async Task PartitionManagerLoop(CancellationToken gracefulShutdownToken, Cancell
 
                 try
                 {
-                    ReadTableReponse response = await this.tableLeaseManager.ReadAndWriteTableAsync(isShuttingDown, forcefulShutdownToken);
+                    using var timeoutCts = new CancellationTokenSource(this.settings.PartitionTableOperationTimeout);
+                    using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(forcefulShutdownToken, timeoutCts.Token);
+
+                    ReadTableReponse response = await this.tableLeaseManager.ReadAndWriteTableAsync(isShuttingDown, linkedCts.Token);
 
                     // If shutdown is requested and already released all ownership leases, then break the loop. 
                     if (isShuttingDown && response.ReleasedAllLeases)
@@ -147,6 +150,20 @@ async Task PartitionManagerLoop(CancellationToken gracefulShutdownToken, Cancell
                 {
                     consecutiveFailureCount++;
                 }
+                // ReadAndWriteTableAsync exceeded the set timeout.
+                // This may indicate a transient storage or network issue.
+                // The operation will be retried immediately unless it fails more than 10 consecutive times.
+                catch (OperationCanceledException) when (!forcefulShutdownToken.IsCancellationRequested)
+                {
+                    this.settings.Logger.PartitionManagerWarning(
+                        this.storageAccountName,
+                        this.settings.TaskHubName,
+                        this.settings.WorkerId,
+                        partitionId: NotApplicable,
+                        details: "Operation to read and write the partition table exceeded the 2-second timeout.");
+
+                    consecutiveFailureCount++;
+                }
                 // Eat any unexpected exceptions.
                 catch (Exception exception)
                 {