Make Helix Job Monitor uploads restart-safe

src/Microsoft.DotNet.Helix/AzureDevOpsTestPublisher/AzureDevOpsResultPublisher.cs

@@ -14,6 +14,7 @@ namespace Microsoft.DotNet.Helix.AzureDevOpsTestPublisher;
 
 public sealed class AzureDevOpsResultPublisher : IDisposable
 {
+    private const int DefaultRetryCount = 10;
     private const int TestListBuckets = 32;
     private static readonly TimeSpan s_httpClientTimeout = TimeSpan.FromMinutes(5);
     private static readonly JsonSerializerOptions s_serializerOptions = new(JsonSerializerDefaults.Web)
@@ -190,6 +191,7 @@ await SendWithRetryAsync(
             HttpMethod.Post,
             $"{_azdoParameters.TeamProject}/_apis/test/runs/{_azdoParameters.TestRunId}/attachments?api-version=7.1-preview.1",
             new TestRunAttachmentRequest(fileNameBase, base64Data),
+            _azdoParameters.RetryWrites ? DefaultRetryCount : 0,
             cancellationToken);
 
         string metadataUrl = await _uploadClient.UploadAsync(compressedBytes, fileNameBase, "application/gzip", cancellationToken);
@@ -216,6 +218,7 @@ private async Task<IReadOnlyList<PublishedTestCase>> PublishResultsAsync(
             HttpMethod.Post,
             $"{_azdoParameters.TeamProject}/_apis/test/runs/{_azdoParameters.TestRunId}/results?api-version=7.1-preview.6",
             testCaseResults,
+            _azdoParameters.RetryWrites ? DefaultRetryCount : 0,
             cancellationToken);
 
         IReadOnlyList<PublishedTestCaseResultReference> publishedResults = await ReadPublishedResultsAsync(response, cancellationToken);
@@ -294,7 +297,12 @@ private async Task SendAttachmentAsync(
             ? $"{_azdoParameters.TeamProject}/_apis/test/runs/{_azdoParameters.TestRunId}/results/{testId}/attachments?testSubResultId={subId}&api-version=7.1-preview.1"
             : $"{_azdoParameters.TeamProject}/_apis/test/runs/{_azdoParameters.TestRunId}/results/{testId}/attachments?api-version=7.1-preview.1";
 
-        using HttpResponseMessage response = await SendWithRetryAsync(HttpMethod.Post, path, request, cancellationToken);
+        using HttpResponseMessage response = await SendWithRetryAsync(
+            HttpMethod.Post,
+            path,
+            request,
+            _azdoParameters.RetryWrites ? DefaultRetryCount : 0,
+            cancellationToken);
         _ = response;
     }
 
@@ -483,9 +491,10 @@ private async Task<HttpResponseMessage> SendWithRetryAsync(
         HttpMethod method,
         string relativePath,
         object? payload,
+        int retryCount,
         CancellationToken cancellationToken)
     {
-        int triesLeft = 10;
+        int triesLeft = retryCount;
         string? body = payload is null ? null : JsonSerializer.Serialize(payload, s_serializerOptions);
         if (!string.IsNullOrEmpty(body))
         {

src/Microsoft.DotNet.Helix/AzureDevOpsTestPublisher/Model/AzureDevOpsReportingParameters.cs

@@ -1,11 +1,40 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
+using System.Text.Json.Serialization;
+
 namespace Microsoft.DotNet.Helix.AzureDevOpsTestPublisher.Model;
 
 public sealed record AzureDevOpsReportingParameters(
     Uri CollectionUri,
     string TeamProject,
     string TestRunId,
-    string? AccessToken = null,
-    bool UseFullyQualifiedTestName = false);
+    string? AccessToken,
+    bool UseFullyQualifiedTestName,
+    bool RetryWrites)
+{
+    [JsonConstructor]
+    public AzureDevOpsReportingParameters(
+        Uri CollectionUri,
+        string TeamProject,
+        string TestRunId,
+        string? AccessToken = null,
+        bool UseFullyQualifiedTestName = false)
+        : this(CollectionUri, TeamProject, TestRunId, AccessToken, UseFullyQualifiedTestName, RetryWrites: true)
+    {
+    }
+
+    public void Deconstruct(
+        out Uri collectionUri,
+        out string teamProject,
+        out string testRunId,
+        out string? accessToken,
+        out bool useFullyQualifiedTestName)
+    {
+        collectionUri = CollectionUri;
+        teamProject = TeamProject;
+        testRunId = TestRunId;
+        accessToken = AccessToken;
+        useFullyQualifiedTestName = UseFullyQualifiedTestName;
+    }
+}

src/Microsoft.DotNet.Helix/AzureDevOpsTestPublisher/RetryHelper.cs

@@ -6,23 +6,41 @@ namespace Microsoft.DotNet.Helix.AzureDevOpsTestPublisher;
 public class RetryHelper
 {
     public static async Task<T> RetryAsync<T>(Func<Task<T>> action, CancellationToken cancellationToken)
+        => await RetryAsync(
+            action,
+            retryCount: 4,
+            static _ => true,
+            onRetry: null,
+            cancellationToken);
+
+    public static async Task<T> RetryAsync<T>(
+        Func<Task<T>> action,
+        int retryCount,
+        Func<Exception, bool> isRetryable,
+        Action<Exception, int>? onRetry,
+        CancellationToken cancellationToken,
+        Func<TimeSpan, CancellationToken, Task>? delayAsync = null)
     {
-        Exception? last = null;
-        for (int attempt = 0; attempt < 5; attempt++)
+        ArgumentOutOfRangeException.ThrowIfNegative(retryCount);
+        delayAsync ??= Task.Delay;
+
+        for (int retry = 0; ; retry++)
         {
             cancellationToken.ThrowIfCancellationRequested();
 
             try
             {
                 return await action();
             }
-            catch (Exception ex) when (attempt < 4)
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
             {
-                last = ex;
-                await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, attempt + 1)), cancellationToken);
+                throw;
+            }
+            catch (Exception ex) when (retry < retryCount && isRetryable(ex))
+            {
+                onRetry?.Invoke(ex, retry + 1);
+                await delayAsync(TimeSpan.FromSeconds(Math.Pow(2, retry + 1)), cancellationToken);
             }
         }
-
-        throw last ?? new InvalidOperationException("Retry failed without capturing an exception.");
     }
 }

src/Microsoft.DotNet.Helix/JobMonitor/JobMonitorRunner.Design.md

@@ -198,7 +198,11 @@ Upload is restart-resilient but logically independent from retry.
 3. Uploads happen in lineage order — oldest incarnation first. If both an
    original job and its resubmission have completed and neither has been
    uploaded, the original uploads first.
-4. Upload failures are logged but never affect pass/fail.
+4. Upload failures are logged as warnings but never affect pass/fail. Read-only
+   download failures are retried a bounded number of times. State-changing
+   operations are not replayed by the queue after an ambiguous failure because
+   they may have partially succeeded. A failed upload remains untagged so a
+   later monitor invocation may replay it in a new run.
 5. A failed original Helix job may be resubmitted on entry and still have
    its original test results uploaded during the same invocation if those
    results were not uploaded earlier.
@@ -281,7 +285,9 @@ behaviorally; method names are illustrative.
 - **List work items for a job** — return all work-item summaries.
 - **Download test results** — given a job and a set of work-item names,
   download recognized result files into a working directory. Individual
-  per-work-item failures must not abort the batch.
+  per-file failures must not prevent the remaining files from being attempted.
+  After the batch, transient failures cause the read-only download phase to be
+  retried; permanent failures are logged and omitted.
 - **Cancel a job** — best-effort cancellation.
 - **Resubmit failed work items** — given the original job and a set of
   failed (or unfinished) work items, submit a new Helix job that contains only
@@ -449,12 +455,21 @@ lines are plain logger output.
 
 ### 5.9 Test-result upload pipeline
 
-Uploads are asynchronous tasks tracked for normal completion:
+Uploads are asynchronous tasks tracked for normal completion. Their in-memory
+lifecycle distinguishes queued, in-progress, durably completed, and failed
+uploads; only a completed, tagged test run is considered durable.
 
 - Each upload is queued asynchronously and tracked. Multiple uploads may
   proceed concurrently.
-- An upload retries indefinitely on transient errors; only cancellation
-  exits the retry loop.
+- Test results are downloaded before the AzDO test run is created. Transient
+  download failures are safe to retry and use a bounded retry budget.
+- The queue invokes each state-changing phase — create the run, publish results,
+  and complete/tag the run — once. The monitor's adapters also disable automatic
+  retries for those writes because an error response may arrive after the
+  service has partially committed the request. Replaying it could duplicate
+  test runs, results, or attachments.
+- Permanent failures, exhausted safe retries, and ambiguous write failures are
+  logged as warnings and stop the upload task without affecting pass/fail.
 - The normal-termination path waits for queued uploads to drain before exiting.
 - The cancellation path does not wait for pending or in-flight uploads. If an
   upload has not completed and applied its Helix-job tag, it remains untagged;

src/Microsoft.DotNet.Helix/JobMonitor/JobMonitorRunner.cs

@@ -80,7 +80,7 @@ internal JobMonitorRunner(
                 _options.SourceBranch);
 
             _reporter = new StatusReporter(_logger, _options, _helix, _state);
-            _uploads = new TestResultUploadQueue(_logger, _options, _azdo, _helix, _state, Delay);
+            _uploads = new TestResultUploadQueue(_logger, _options, _azdo, _helix, _state, _delayFunc);
         }
 
         public async Task<int> RunAsync(CancellationToken cancellationToken)
@@ -304,7 +304,6 @@ await _azdo.GetTimelineRecordsAsync(cancellationToken),
             foreach (HelixJobInfo job in completedJobs.Where(j => !_state.IsHelixJobProcessed(j.JobName)))
             {
                 await ReconcileCompletedJobAsync(job, queueUpload: true, cancellationToken);
-                _state.TryMarkHelixJobProcessed(job.JobName);
             }
 
             // Second pass: ensure outcomes for every completed job (any attempt) are reflected in
@@ -395,7 +394,10 @@ private async Task ReconcileCompletedJobAsync(
 
             if (queueUpload && !alreadyUploadedByPriorAttempt)
             {
-                _uploads.Enqueue(helixJob, workItems, cancellationToken);
+                if (_state.TryQueueHelixJobUpload(helixJob.JobName))
+                {
+                    _uploads.Enqueue(helixJob, workItems, cancellationToken);
+                }
             }
 
             if (!alreadyUploadedByPriorAttempt)

src/Microsoft.DotNet.Helix/JobMonitor/MonitorState.cs

@@ -11,6 +11,14 @@
 
 namespace Microsoft.DotNet.Helix.JobMonitor
 {
+    internal enum TestResultUploadState
+    {
+        Queued,
+        InProgress,
+        DurablyCompleted,
+        Failed,
+    }
+
     /// <summary>
     /// Thread-safe container for every piece of runtime state the monitor accumulates while
     /// observing a single invocation. Mutations from the main poll loop and from background
@@ -28,9 +36,11 @@ internal sealed class MonitorState
         // by GetSubmitterChainKey to walk back through PreviousHelixJobName links across polls.
         private readonly Dictionary<string, HelixJobInfo> _associatedJobs = new(StringComparer.OrdinalIgnoreCase);
 
-        // Helix jobs whose results have been uploaded to Azure DevOps in this or a prior
-        // monitor invocation. Seeded on entry from the AzDO test-run tags.
-        private readonly HashSet<string> _processedHelixJobs = new(StringComparer.OrdinalIgnoreCase);
+        // Upload lifecycle for each Helix job observed by this invocation. Jobs discovered from
+        // Azure DevOps completion tags are seeded as DurablyCompleted. A queued/in-progress job
+        // must not be treated as durable: cancellation may abandon it, in which case the absent
+        // tag causes a later invocation to replay the upload.
+        private readonly Dictionary<string, TestResultUploadState> _testResultUploadStates = new(StringComparer.OrdinalIgnoreCase);
 
         // Tracks the latest outcome for each logical work item, keyed by
         // (SubmitterChainKey, WorkItemName). See GetSubmitterChainKey for the keying rationale.
@@ -156,21 +166,54 @@ public void AddProcessedHelixJobs(IEnumerable<string> jobNames)
             {
                 foreach (string jobName in jobNames)
                 {
-                    _processedHelixJobs.Add(jobName);
+                    _testResultUploadStates[jobName] = TestResultUploadState.DurablyCompleted;
                 }
             }
         }
 
+        public bool TryQueueHelixJobUpload(string jobName)
+        {
+            lock (_sync)
+            {
+                if (_testResultUploadStates.ContainsKey(jobName))
+                {
+                    return false;
+                }
+
+                _testResultUploadStates[jobName] = TestResultUploadState.Queued;
+                return true;
+            }
+        }
+
+        public void MarkHelixJobUploadInProgress(string jobName)
+        {
+            lock (_sync)
+            {
+                _testResultUploadStates[jobName] = TestResultUploadState.InProgress;
+            }
+        }
+
+        public void MarkHelixJobUploadFailed(string jobName)
+        {
+            lock (_sync)
+            {
+                _testResultUploadStates[jobName] = TestResultUploadState.Failed;
+            }
+        }
+
         /// <summary>
-        /// Marks the given Helix job as processed and increments <see cref="ProcessedJobCount"/>.
-        /// Returns true if this is the first time the job was marked.
+        /// Marks the given Helix job as durably completed and increments
+        /// <see cref="ProcessedJobCount"/>. This must only be called after Azure DevOps has
+        /// completed and tagged the test run.
         /// </summary>
         public bool TryMarkHelixJobProcessed(string jobName)
         {
             lock (_sync)
             {
-                if (_processedHelixJobs.Add(jobName))
+                if (!_testResultUploadStates.TryGetValue(jobName, out TestResultUploadState state)
+                    || state != TestResultUploadState.DurablyCompleted)
                 {
+                    _testResultUploadStates[jobName] = TestResultUploadState.DurablyCompleted;
                     _processedJobCount++;
                     return true;
                 }
@@ -183,7 +226,8 @@ public bool IsHelixJobProcessed(string jobName)
         {
             lock (_sync)
             {
-                return _processedHelixJobs.Contains(jobName);
+                return _testResultUploadStates.TryGetValue(jobName, out TestResultUploadState state)
+                    && state == TestResultUploadState.DurablyCompleted;
             }
         }