Make Helix Job Monitor uploads restart-safe

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

@@ -1,10 +1,8 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
-using System.IO.Compression;
 using System.Net;
 using System.Net.Http.Headers;
-using System.Security.Cryptography;
 using System.Text;
 using System.Text.Json;
 using Microsoft.DotNet.Helix.AzureDevOpsTestPublisher.Model;
@@ -14,7 +12,7 @@ namespace Microsoft.DotNet.Helix.AzureDevOpsTestPublisher;
 
 public sealed class AzureDevOpsResultPublisher : IDisposable
 {
-    private const int TestListBuckets = 32;
+    private const int DefaultRetryCount = 10;
     private static readonly TimeSpan s_httpClientTimeout = TimeSpan.FromMinutes(5);
     private static readonly JsonSerializerOptions s_serializerOptions = new(JsonSerializerDefaults.Web)
     {
@@ -95,8 +93,6 @@ public async Task<long> UploadTestResultsWithCountAsync(IEnumerable<AggregatedRe
 
             _logger.LogDebug("Uploaded {Count} results", publishedTestCount);
 
-            // TODO - Do we need this?
-            // await SendMetadataAsync(resultList, cancellationToken);
             return publishedTestCount;
         }
         catch (TerminalError ex)
@@ -106,105 +102,6 @@ public async Task<long> UploadTestResultsWithCountAsync(IEnumerable<AggregatedRe
         }
     }
 
-    private static async Task SendMetadataAsync(
-        IEnumerable<AggregatedResult> allTestResults,
-        CancellationToken cancellationToken)
-    {
-        var partitionedResults = new Dictionary<int, List<TestListRow>>();
-        var resultCounts = new Dictionary<string, int>(StringComparer.Ordinal);
-
-        void ProcessResultForMetadata(AggregatedResult result)
-        {
-            resultCounts[result.Result] = resultCounts.TryGetValue(result.Result, out int count) ? count + 1 : 1;
-            if (!string.Equals(result.Result, "Passed", StringComparison.Ordinal))
-            {
-                return;
-            }
-
-            string name = result.Name;
-            string? argumentHash = null;
-            string partitionKey = name;
-            int parenthesisIndex = name.IndexOf('(');
-            if (parenthesisIndex >= 0)
-            {
-                string argumentList = name[(parenthesisIndex + 1)..].TrimEnd(')');
-                name = name[..parenthesisIndex];
-                argumentHash = Convert.ToBase64String(SHA1.HashData(Encoding.UTF8.GetBytes(argumentList)));
-                partitionKey = name + argumentHash;
-            }
-
-            int bucket = SHA1.HashData(Encoding.UTF8.GetBytes(partitionKey))[0] % TestListBuckets;
-            if (!partitionedResults.TryGetValue(bucket, out List<TestListRow>? testNames))
-            {
-                testNames = [];
-                partitionedResults[bucket] = testNames;
-            }
-
-            testNames.Add(new TestListRow(name, argumentHash));
-        }
-
-        void ProcessTestForMetadata(AggregatedResult result)
-        {
-            if (result.AggregationType == AggregationType.DataDriven && result.SubResults.Count > 0)
-            {
-                foreach (AggregatedResult subResult in result.SubResults)
-                {
-                    ProcessTestForMetadata(subResult);
-                }
-            }
-            else if (result.AggregationType == AggregationType.Single)
-            {
-                ProcessResultForMetadata(result);
-            }
-        }
-
-        foreach (AggregatedResult result in allTestResults)
-        {
-            cancellationToken.ThrowIfCancellationRequested();
-            ProcessTestForMetadata(result);
-        }
-        /*
-        var uploadedUrls = new Dictionary<int, string>();
-        foreach ((int key, List<TestListRow>? testNames) in partitionedResults)
-        {
-            byte[] csvBytes = CreateCompressedCsv(testNames);
-            string fileName = $"{Guid.NewGuid():N}.csv.gz";
-            uploadedUrls[key] = await _uploadClient.UploadAsync(csvBytes, fileName, "application/gzip", cancellationToken);
-        }
-
-        var dataModel = new
-        {
-            version = 2,
-            rerun_tests = backChannelCases,
-            test_lists = uploadedUrls,
-            partitions = TestListBuckets,
-            result_counts = resultCounts,
-        };
-
-        byte[] rawBytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(dataModel, s_serializerOptions));
-        byte[] compressedBytes = Compress(rawBytes);
-        string base64Data = Convert.ToBase64String(compressedBytes);
-        string fileNameBase = $"__helix_metadata_{Guid.NewGuid():N}.json.gz";
-
-        await SendWithRetryAsync(
-            HttpMethod.Post,
-            $"{_azdoParameters.TeamProject}/_apis/test/runs/{_azdoParameters.TestRunId}/attachments?api-version=7.1-preview.1",
-            new TestRunAttachmentRequest(fileNameBase, base64Data),
-            cancellationToken);
-
-        string metadataUrl = await _uploadClient.UploadAsync(compressedBytes, fileNameBase, "application/gzip", cancellationToken);
-        await _eventClient.SendAsync(
-            new
-            {
-                Type = "AzureDevOpsTestRunMetadata",
-                TestRunProject = _azdoParameters.TeamProject,
-                TestRunId = _azdoParameters.TestRunId,
-                Url = metadataUrl,
-            },
-            cancellationToken);
-        */
-    }
-
     private async Task<IReadOnlyList<PublishedTestCase>> PublishResultsAsync(
         IReadOnlyList<ConvertedResult> converted,
         CancellationToken cancellationToken)
@@ -216,6 +113,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 +192,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 +386,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))
         {
@@ -629,52 +533,10 @@ private static PublishedSubResultReference ParsePublishedSubResult(JsonElement e
             subResults);
     }
 
-    private static byte[] CreateCompressedCsv(IEnumerable<TestListRow> rows)
-    {
-        var builder = new StringBuilder();
-        foreach (TestListRow row in rows)
-        {
-            builder.Append(EscapeCsv(row.TestName));
-            builder.Append(',');
-            builder.Append(EscapeCsv(row.ArgumentHash));
-            builder.AppendLine();
-        }
-
-        return Compress(Encoding.UTF8.GetBytes(builder.ToString()));
-    }
-
-    private static string EscapeCsv(string? value)
-    {
-        if (string.IsNullOrEmpty(value))
-        {
-            return string.Empty;
-        }
-
-        if (!value.Contains('"') && !value.Contains(',') && !value.Contains('\n') && !value.Contains('\r'))
-        {
-            return value;
-        }
-
-        return $"\"{value.Replace("\"", "\"\"")}\"";
-    }
-
-    private static byte[] Compress(ReadOnlySpan<byte> rawBytes)
-    {
-        using var target = new MemoryStream();
-        using (var gzip = new GZipStream(target, CompressionLevel.SmallestSize, leaveOpen: true))
-        {
-            gzip.Write(rawBytes);
-        }
-
-        return target.ToArray();
-    }
-
     private sealed record ConvertedResult(PublishedTestCase Converted, AggregatedResult Aggregated);
 
     private sealed record ChunkPair(PublishedSubResult Converted, AggregatedResult Aggregated);
 
-    private sealed record TestListRow(string TestName, string? ArgumentHash);
-
     private sealed record TestRunAttachmentRequest(string FileName, string Stream);
 
     private sealed record CustomField(string FieldName, object Value);

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

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

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