Limit concurrency on audit queries

DFE-Digital · Jan 15, 2025 · 06afbe7 · 06afbe7
1 parent 6faf191
commit 06afbe7
Showing 1 changed file with 31 additions and 5 deletions.
diff --git a/TeachingRecordSystem/src/TeachingRecordSystem.Core/Services/TrsDataSync/TrsDataSyncHelper.cs b/TeachingRecordSystem/src/TeachingRecordSystem.Core/Services/TrsDataSync/TrsDataSyncHelper.cs
@@ -667,7 +667,27 @@ private async Task<IReadOnlyDictionary<Guid, AuditDetailCollection>> GetAuditRec
         IEnumerable<Guid> ids,
         CancellationToken cancellationToken)
     {
-        return (IsFakeXrm ? new Dictionary<Guid, AuditDetailCollection>() : (await Task.WhenAll(ids
+        if (IsFakeXrm)
+        {
+            return new Dictionary<Guid, AuditDetailCollection>();
+        }
+
+        // Throttle the amount of concurrent requests
+        using var requestThrottle = new SemaphoreSlim(20, 20);
+
+        // Keep track of the last seen 'retry-after' value
+        var retryDelayUpdateLock = new object();
+        var retryDelay = Task.Delay(0, cancellationToken);
+
+        void UpdateRetryDelay(TimeSpan ts)
+        {
+            lock (retryDelayUpdateLock)
+            {
+                retryDelay = Task.Delay(ts, cancellationToken);
+            }
+        }
+
+        return (await Task.WhenAll(ids
             .Chunk(MaxAuditRequestsPerBatch)
             .Select(async chunk =>
             {
@@ -687,16 +707,22 @@ private async Task<IReadOnlyDictionary<Guid, AuditDetailCollection>> GetAuditRec
                 ExecuteMultipleResponse response;
                 while (true)
                 {
+                    await retryDelay;
+                    await requestThrottle.WaitAsync(cancellationToken);
                     try
                     {
                         response = (ExecuteMultipleResponse)await organizationService.ExecuteAsync(request, cancellationToken);
                     }
                     catch (FaultException fex) when (fex.IsCrmRateLimitException(out var retryAfter))
                     {
                         logger.LogWarning("Hit CRM service limits getting {entityLogicalName} audit records; Fault exception. Retrying after {retryAfter} seconds.", entityLogicalName, retryAfter.TotalSeconds);
-                        await Task.Delay(retryAfter, cancellationToken);
+                        UpdateRetryDelay(retryAfter);
                         continue;
                     }
+                    finally
+                    {
+                        requestThrottle.Release();
+                    }
 
                     if (response.IsFaulted)
                     {
@@ -705,13 +731,13 @@ private async Task<IReadOnlyDictionary<Guid, AuditDetailCollection>> GetAuditRec
                         if (firstFault.IsCrmRateLimitFault(out var retryAfter))
                         {
                             logger.LogWarning("Hit CRM service limits getting {entityLogicalName} audit records; CRM rate limit fault. Retrying after {retryAfter} seconds.", entityLogicalName, retryAfter.TotalSeconds);
-                            await Task.Delay(retryAfter, cancellationToken);
+                            UpdateRetryDelay(retryAfter);
                             continue;
                         }
                         else if (firstFault.Message.Contains("The HTTP status code of the response was not expected (429)"))
                         {
                             logger.LogWarning("Hit CRM service limits getting {entityLogicalName} audit records; 429 too many requests", entityLogicalName);
-                            await Task.Delay(TimeSpan.FromMinutes(2), cancellationToken);
+                            UpdateRetryDelay(TimeSpan.FromMinutes(2));
                             continue;
                         }
 
@@ -726,7 +752,7 @@ private async Task<IReadOnlyDictionary<Guid, AuditDetailCollection>> GetAuditRec
                     (r, e) => (Id: e, ((RetrieveRecordChangeHistoryResponse)r.Response).AuditDetailCollection));
             })))
             .SelectMany(b => b)
-            .ToDictionary(t => t.Id, t => t.AuditDetailCollection));
+            .ToDictionary(t => t.Id, t => t.AuditDetailCollection);
     }
 
     private async Task<IReadOnlyDictionary<Guid, AuditDetailCollection>> GetAuditRecordsFromAuditRepositoryAsync(