Make Helix Job Monitor uploads restart-safe

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

@@ -476,7 +476,7 @@ private async Task<HttpResponseMessage> SendWithRetryAsync(
         object? payload,
         CancellationToken cancellationToken)
     {
-        int triesLeft = 10;
+        int triesLeft = GetRetryCount(method, _azdoParameters.RetryWrites);
         string? body = payload is null ? null : JsonSerializer.Serialize(payload, s_serializerOptions);
         if (!string.IsNullOrEmpty(body))
         {
@@ -549,6 +549,9 @@ await _uploadClient.UploadAsync(
         }
     }
 
+    internal static int GetRetryCount(HttpMethod method, bool retryWrites)
+        => retryWrites || method == HttpMethod.Get ? 10 : 0;
+
     private static TimeSpan? GetRetryDelay(HttpResponseMessage response)
     {
         RetryConditionHeaderValue? retryAfter = response.Headers.RetryAfter;

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

@@ -8,4 +8,7 @@ public sealed record AzureDevOpsReportingParameters(
     string TeamProject,
     string TestRunId,
     string? AccessToken = null,
-    bool UseFullyQualifiedTestName = false);
+    bool UseFullyQualifiedTestName = false)
+{
+    public bool RetryWrites { get; init; } = true;
+}

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

@@ -186,29 +186,23 @@ addresses each:
    wait" scheme. → Resubmission-not-possible is treated as an actionable hard
    failure so the invocation fails fast instead of hanging (§2.3.3).
 
-Each of these cases is pinned by a pipeline-emulating test in
-`JobMonitorRunnerTests` (the `AttemptScoped_*` suite): `RetryOnlyMonitor`
-(case 1), `StrandedWaitingPreviousWork` (cases 2/3, mirroring #17156),
-`FastRerun_CurrentIncarnationExists` (case 4),
-`UnlinkedRerunDuplicates_HigherAttemptWinsOutcome` (case 5), and
-`UnresubmittablePreviousWork_FailsFast` (case 6). The `MultiAttempt_*` tests
-additionally exercise the end-to-end lineage across five stage attempts —
-including an attempt in which the monitor never ran — verifying that only
-still-unfinished streams are carried forward and that a stream which has passed
-in any prior attempt is never resubmitted again.
-
 ### 2.4 Upload invariants
 
 Upload is restart-resilient but logically independent from retry.
 
-1. The same Helix job's test results are never uploaded twice. The durable
+1. A Helix job has at most one durably completed, tagged test run. The durable
    deduplication signal is the Helix-job-name tag on completed AzDO test runs.
-2. For every completed Helix job not already uploaded, all available test
-   results are uploaded.
+   An interrupted, untagged upload is intentionally replayed.
+2. For every completed Helix job without a completed, tagged upload, all
+   available test results are uploaded.
 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.
@@ -235,15 +229,17 @@ Exit code is `0` only when both checks pass; otherwise `1`. Cancellation
 The runner must be safe to re-run after any abrupt termination. In
 particular:
 
-- Partial uploads must not cause duplicate uploads on the next run.
+- Completed, tagged uploads must not be repeated. Interrupted, untagged uploads
+  are intentionally replayed on the next run.
 - Retry candidates must be rediscovered from Helix job properties, not from
   prior in-memory state.
-- Cancellation must drain in-flight uploads before exiting so partially
-  uploaded results are not lost.
-- On cancellation, in-flight Helix jobs should receive a best-effort cancel
-  request even though the runner's own cancellation token has already
-  fired. This requires a fresh, short-lived cancellation budget for the
-  cleanup path.
+- Cancellation must not wait for pending or in-flight uploads. An upload that
+  does not complete remains untagged and is rediscovered and re-uploaded by a
+  later invocation.
+- On cancellation, immediately emit the timeout report, then best-effort cancel
+  the latest in-flight Helix job incarnation in each lineage even though the
+  runner's own cancellation token has already fired. This cleanup uses a fresh,
+  bounded cancellation token so it cannot extend shutdown indefinitely.
 
 Re-attach spans stage attempts. Within the same attempt (a monitor job-retry or
 a crashed-and-restarted monitor process) the runner re-attaches to the same
@@ -289,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
@@ -326,8 +324,10 @@ or the runner will silently fail to see its own jobs.
 3. Perform the one-shot retry pass (§5.3).
 4. Enter the poll loop (§5.4) until the build finishes or cancellation
    fires.
-5. On cancellation (timeout), drain pending uploads, emit a timeout report
-   (§5.6), best-effort cancel in-flight Helix jobs (§2.6), and exit `1`.
+5. On cancellation (timeout), do not wait for pending uploads. Immediately emit
+   a timeout report (§5.6), use a fresh bounded token to best-effort cancel the
+   latest in-flight Helix jobs (§2.6), and exit `1`. Incomplete uploads remain
+   untagged and are retried by a later invocation.
 6. On normal completion, emit the final summary and exit per §2.5.
 
 ### 5.3 Retry pass
@@ -455,16 +455,25 @@ lines are plain logger output.
 
 ### 5.9 Test-result upload pipeline
 
-Uploads are fire-and-forget tasks tracked for later draining:
+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.
-- Both the normal-termination and cancellation paths wait for queued
-  uploads to drain before exiting. The cancellation drain uses a fresh
-  cancellation budget so uploads in progress when the runner token fires
-  are not abandoned.
+- 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;
+  durable-state discovery causes a later invocation to upload it again.
 
 The upload sequence per job is: create (or reuse) a test run with the plain
 `{TestRunName}`, download results, upload them, complete the test run and tag
@@ -483,8 +492,8 @@ least one work item), or `Waiting` (no work items observed yet).
 
 ## 6. Externally observable formats
 
-These shapes are observed by other tools, downstream parsers, or tests and
-must be preserved:
+These shapes are observed by other tools or downstream parsers and must be
+preserved:
 
 - AzDO test-run tag: `helixjob<guid-without-dashes>`, applied to a completed
   test run as an object-form tag (`{ "name": "..." }`).

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

@@ -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;
             }
         }
 

src/Microsoft.DotNet.Helix/JobMonitor/Services/AzureDevOpsService.cs

@@ -318,6 +318,7 @@ public async Task<int> CreateTestRunAsync(string name, CancellationToken cancell
                     ["name"] = name,
                     ["state"] = "InProgress",
                 },
+                retryTransientFailures: false,
                 cancellationToken: cancellationToken);
             return result?["id"]?.ToObject<int>() ?? 0;
         }
@@ -360,6 +361,7 @@ public async Task CompleteTestRunAsync(
             await SendAsync(new HttpMethod("PATCH"),
                 $"{_options.CollectionUri}{_options.TeamProject}/_apis/test/runs/{testRunId}?api-version=7.1",
                 body,
+                retryTransientFailures: false,
                 cancellationToken: cancellationToken);
         }
 
@@ -391,6 +393,7 @@ await SendAsync(
                 HttpMethod.Post,
                 $"{_options.CollectionUri}{_options.TeamProject}/_apis/test/Runs/{testRunId}/attachments?api-version=7.1",
                 body,
+                retryTransientFailures: false,
                 cancellationToken: cancellationToken);
         }
 
@@ -399,13 +402,17 @@ await SendAsync(
             IReadOnlyList<WorkItemTestResults> results,
             CancellationToken cancellationToken)
         {
-            using var publisher = new AzureDevOpsResultPublisher(
-                new AzureDevOpsReportingParameters(
+            var reportingParameters = new AzureDevOpsReportingParameters(
                     new Uri(_options.CollectionUri, UriKind.Absolute),
                     _options.TeamProject,
                     testRunId.ToString(CultureInfo.InvariantCulture),
                     _options.SystemAccessToken,
-                    _options.UseFullyQualifiedTestName),
+                    _options.UseFullyQualifiedTestName)
+            {
+                RetryWrites = false,
+            };
+            using var publisher = new AzureDevOpsResultPublisher(
+                reportingParameters,
                 _logger);
 
             async Task<TestResultUploadSummary> UploadWorkItemAsync(WorkItemTestResults workItem)
@@ -441,18 +448,28 @@ async Task<TestResultUploadSummary> UploadWorkItemAsync(WorkItemTestResults work
             return testSummaries.ToDictionary(t => (t.WorkItem.JobName, t.WorkItem.WorkItemName), t => t.Summary);
         }
 
-        private async Task<JObject> SendAsync(HttpMethod method, string requestUri, JToken body = null, CancellationToken cancellationToken = default)
+        private async Task<JObject> SendAsync(
+            HttpMethod method,
+            string requestUri,
+            JToken body = null,
+            bool retryTransientFailures = true,
+            CancellationToken cancellationToken = default)
         {
-            string content = await SendForStringAsync(method, requestUri, body, cancellationToken);
+            string content = await SendForStringAsync(method, requestUri, body, retryTransientFailures, cancellationToken);
             return string.IsNullOrWhiteSpace(content) ? [] : JObject.Parse(content);
         }
 
         // Sends a request and returns the raw response body as a string. Used for endpoints
         // (such as test-run attachment downloads) that do not return JSON, where SendAsync's
         // JObject parsing would fail or discard the payload.
-        private async Task<string> SendForStringAsync(HttpMethod method, string requestUri, JToken body = null, CancellationToken cancellationToken = default)
+        private async Task<string> SendForStringAsync(
+            HttpMethod method,
+            string requestUri,
+            JToken body = null,
+            bool retryTransientFailures = true,
+            CancellationToken cancellationToken = default)
         {
-            return await RetryHelper.RetryAsync(async () =>
+            async Task<string> SendOnceAsync()
             {
                 using var request = new HttpRequestMessage(method, requestUri);
                 if (body != null)
@@ -465,12 +482,19 @@ private async Task<string> SendForStringAsync(HttpMethod method, string requestU
                 if (!response.IsSuccessStatusCode)
                 {
                     await HonorRateLimitAsync(response, requestUri, cancellationToken);
-                    throw new HttpRequestException($"Request to {requestUri} failed with {(int)response.StatusCode} {response.ReasonPhrase}. {content}");
+                    throw new HttpRequestException(
+                        $"Request to {requestUri} failed with {(int)response.StatusCode} {response.ReasonPhrase}. {content}",
+                        null,
+                        response.StatusCode);
                 }
 
                 await HonorRateLimitAsync(response, requestUri, cancellationToken);
                 return content;
-            }, cancellationToken);
+            }
+
+            return retryTransientFailures
+                ? await RetryHelper.RetryAsync(SendOnceAsync, cancellationToken)
+                : await SendOnceAsync();
         }
 
         // Honors Azure DevOps rate limiting guidance: