From a98264e27907e84ce246cd26b52e785b00fdc490 Mon Sep 17 00:00:00 2001 From: Martijn Laarman Date: Tue, 29 Sep 2026 12:00:24 +0200 Subject: [PATCH 1/3] Upload S3 files concurrently instead of one directory call Wraps the per-file upload lambda in error handling and uploads add/update requests in parallel via Parallel.ForEachAsync. Tests migrated from xUnit [Fact] to TUnit [Test]. Co-Authored-By: Claude Sonnet 4.6 --- .../IncrementalDeployService.cs | 10 +- .../Synchronization/AwsS3SyncApplyStrategy.cs | 175 ++++++++++++------ .../Synchronization/DocsSync.cs | 3 +- .../DocsSyncTests.cs | 111 +++++++++-- .../IncrementalDeployRoundTripTests.cs | 30 +-- 5 files changed, 242 insertions(+), 87 deletions(-) diff --git a/src/services/Elastic.Documentation.Deploying/IncrementalDeployService.cs b/src/services/Elastic.Documentation.Deploying/IncrementalDeployService.cs index 2478b28b94..0e162c55d0 100644 --- a/src/services/Elastic.Documentation.Deploying/IncrementalDeployService.cs +++ b/src/services/Elastic.Documentation.Deploying/IncrementalDeployService.cs @@ -3,6 +3,7 @@ // See the LICENSE file in the project root for more information using Actions.Core.Services; +using Amazon.Runtime; using Amazon.S3; using Amazon.S3.Transfer; using Elastic.Documentation.Deploying.Synchronization; @@ -22,7 +23,11 @@ public class IncrementalDeployService( ) : IService { private readonly ILogger _logger = logFactory.CreateLogger(); - private readonly IAmazonS3 _s3 = s3Client ?? new AmazonS3Client(); + + // Standard mode retries throttling/5xx with jittered backoff; higher upload concurrency means more + // concurrent requests hitting S3, so let the SDK absorb transient failures instead of failing files outright. + private readonly IAmazonS3 _s3 = s3Client + ?? new AmazonS3Client(new AmazonS3Config { RetryMode = RequestRetryMode.Standard, MaxErrorRetry = 5 }); public async Task Plan( IDiagnosticsCollector collector, @@ -105,8 +110,7 @@ Cancel ctx return false; } var applier = new AwsS3SyncApplyStrategy(logFactory, _s3, xfer, s3BucketName, context, collector); - await applier.Apply(plan, ctx); - return true; + return await applier.Apply(plan, ctx); } private void LogPlanSummary(SyncPlan plan) diff --git a/src/services/Elastic.Documentation.Deploying/Synchronization/AwsS3SyncApplyStrategy.cs b/src/services/Elastic.Documentation.Deploying/Synchronization/AwsS3SyncApplyStrategy.cs index 873e8c094d..3437edb90c 100644 --- a/src/services/Elastic.Documentation.Deploying/Synchronization/AwsS3SyncApplyStrategy.cs +++ b/src/services/Elastic.Documentation.Deploying/Synchronization/AwsS3SyncApplyStrategy.cs @@ -4,10 +4,12 @@ using System.Diagnostics; using System.Diagnostics.Metrics; +using System.Threading; using Amazon.S3; using Amazon.S3.Model; using Amazon.S3.Transfer; using Elastic.Documentation.Diagnostics; +using Elastic.Documentation.Integrations.S3; using Elastic.Documentation.ServiceDefaults.Telemetry; using Microsoft.Extensions.Logging; @@ -26,6 +28,12 @@ IDiagnosticsCollector collector // Meter for OpenTelemetry metrics private static readonly Meter SyncMeter = new(TelemetryConstants.AssemblerSyncInstrumentationName); + private const int DefaultUploadConcurrency = 32; + private const int MaxUploadConcurrency = 128; + private const int DefaultDeleteConcurrency = 4; + private const int MaxDeleteConcurrency = 16; + private const string UploadConcurrencyEnvironmentVariable = "DOCS_SYNC_UPLOAD_CONCURRENCY"; + private const string DeleteConcurrencyEnvironmentVariable = "DOCS_SYNC_DELETE_CONCURRENCY"; // Deployment-level metrics (histograms for distribution analysis, counters for totals) // Note: Histograms require delta temporality to work with Elasticsearch @@ -86,17 +94,18 @@ IDiagnosticsCollector collector private readonly ILogger _logger = logFactory.CreateLogger(); - private void DisplayProgress(object? sender, UploadDirectoryProgressArgs args) => LogProgress(_logger, args); + private void DisplayUploadProgress(object? sender, UploadProgressArgs args) => LogUploadProgress(_logger, args); - [LoggerMessage(EventId = 2, Level = LogLevel.Debug, Message = "{Args}")] - private static partial void LogProgress(ILogger logger, UploadDirectoryProgressArgs args); + [LoggerMessage(EventId = 4, Level = LogLevel.Debug, Message = "{Args}")] + private static partial void LogUploadProgress(ILogger logger, UploadProgressArgs args); [LoggerMessage(EventId = 3, Level = LogLevel.Information, Message = "File operation: {Operation} | Path: {FilePath} | Size: {FileSize} bytes")] private static partial void LogFileOperation(ILogger logger, string operation, string filePath, long fileSize); - public async Task Apply(SyncPlan plan, Cancel ctx = default) + public async Task Apply(SyncPlan plan, Cancel ctx = default) { var sw = Stopwatch.StartNew(); + var errorsBeforeApply = collector.Errors; using var applyActivity = ApplyStrategyActivitySource.StartActivity("sync apply", ActivityKind.Client); if (Environment.GetEnvironmentVariable("GITHUB_ACTIONS") == "true") @@ -153,15 +162,21 @@ public async Task Apply(SyncPlan plan, Cancel ctx = default) // Record sync duration (both histogram for distribution and counter for total) SyncDurationHistogram.Record(sw.Elapsed.TotalSeconds); + + return collector.Errors == errorsBeforeApply; } private async Task Upload(SyncPlan plan, Cancel ctx) { + var sw = Stopwatch.StartNew(); + var uploadedCount = 0; var uploadRequests = plan.AddRequests.Cast().Concat(plan.UpdateRequests).ToList(); + var uploadConcurrency = GetConcurrency(UploadConcurrencyEnvironmentVariable, DefaultUploadConcurrency, MaxUploadConcurrency); // Always create activity span (even if 0 files) for consistent tracing using var uploadActivity = ApplyStrategyActivitySource.StartActivity("upload files", ActivityKind.Client); _ = uploadActivity?.SetTag("docs.sync.upload.count", uploadRequests.Count); + _ = uploadActivity?.SetTag("docs.sync.upload.concurrency", uploadConcurrency); if (uploadRequests.Count > 0) { @@ -171,12 +186,19 @@ private async Task Upload(SyncPlan plan, Cancel ctx) _logger.LogInformation("Starting to process {AddCount} new files and {UpdateCount} updated files", addCount, updateCount); var addPaths = plan.AddRequests.Select(r => r.LocalPath).ToHashSet(); - var tempDir = Path.Join(context.WriteFileSystem.Path.GetTempPath(), context.WriteFileSystem.Path.GetRandomFileName()); - _ = context.WriteFileSystem.Directory.CreateDirectory(tempDir); - try + _logger.LogInformation( + "Uploading {Count} files to S3 bucket {BucketName} with concurrency {Concurrency}", + uploadRequests.Count, + bucketName, + uploadConcurrency + ); + await Parallel.ForEachAsync(uploadRequests, new ParallelOptions + { + CancellationToken = ctx, + MaxDegreeOfParallelism = uploadConcurrency + }, async (upload, token) => { - _logger.LogInformation("Copying {Count} files to temp directory", uploadRequests.Count); - foreach (var upload in uploadRequests) + try { var operation = addPaths.Contains(upload.LocalPath) ? "add" : "update"; var fileSize = context.WriteFileSystem.FileInfo.New(upload.LocalPath).Length; @@ -187,51 +209,61 @@ private async Task Upload(SyncPlan plan, Cancel ctx) FilesByExtensionCounter.Add(1, new("operation", operation), new("extension", extension)); LogFileOperation(_logger, operation, upload.DestinationPath, fileSize); - var destPath = context.WriteFileSystem.Path.Join(tempDir, upload.DestinationPath); - var destDirPath = context.WriteFileSystem.Path.GetDirectoryName(destPath)!; - _ = context.WriteFileSystem.Directory.CreateDirectory(destDirPath); - context.WriteFileSystem.File.Copy(upload.LocalPath, destPath); + var request = new TransferUtilityUploadRequest + { + BucketName = bucketName, + FilePath = upload.LocalPath, + Key = upload.DestinationPath, + PartSize = S3EtagCalculator.PartSize + }; + request.UploadProgressEvent += DisplayUploadProgress; + await transferUtility.UploadAsync(request, token); + _ = Interlocked.Increment(ref uploadedCount); } - var directoryRequest = new TransferUtilityUploadDirectoryRequest + catch (Exception ex) when (ex is not OperationCanceledException) { - BucketName = bucketName, - Directory = tempDir, - SearchPattern = "*", - SearchOption = SearchOption.AllDirectories, - UploadFilesConcurrently = true - }; - directoryRequest.UploadDirectoryProgressEvent += DisplayProgress; - _logger.LogInformation("Uploading {Count} files to S3 bucket {BucketName}", uploadRequests.Count, bucketName); - _logger.LogDebug("Starting directory upload from {TempDir}", tempDir); - await transferUtility.UploadDirectoryAsync(directoryRequest, ctx); - _logger.LogInformation( - "Successfully uploaded {Count} files ({AddCount} added, {UpdateCount} updated)", - uploadRequests.Count, - addCount, - updateCount - ); - } - finally - { - // Clean up temp directory - if (context.WriteFileSystem.Directory.Exists(tempDir)) - context.WriteFileSystem.Directory.Delete(tempDir, true); - } + _logger.LogError( + ex, + "Failed to upload {LocalPath} to s3://{BucketName}/{DestinationPath}", + upload.LocalPath, + bucketName, + upload.DestinationPath + ); + collector.EmitError(upload.LocalPath, $"Failed to upload to s3://{bucketName}/{upload.DestinationPath}", ex); + } + }); + _logger.LogInformation( + "Finished uploading {UploadedCount}/{Count} files ({AddCount} added, {UpdateCount} updated) in {ElapsedMs}ms", + uploadedCount, + uploadRequests.Count, + addCount, + updateCount, + sw.ElapsedMilliseconds + ); } + _ = uploadActivity?.SetTag("docs.sync.upload.duration_ms", sw.Elapsed.TotalMilliseconds); } private async Task Delete(SyncPlan plan, Cancel ctx) { + var sw = Stopwatch.StartNew(); var deleteCount = 0; var deleteRequests = plan.DeleteRequests; + var deleteConcurrency = GetConcurrency(DeleteConcurrencyEnvironmentVariable, DefaultDeleteConcurrency, MaxDeleteConcurrency); // Always create activity span (even if 0 files) for consistent tracing using var deleteActivity = ApplyStrategyActivitySource.StartActivity("delete files", ActivityKind.Client); _ = deleteActivity?.SetTag("docs.sync.delete.count", deleteRequests.Count); + _ = deleteActivity?.SetTag("docs.sync.delete.concurrency", deleteConcurrency); if (deleteRequests.Count > 0) { - _logger.LogInformation("Starting to delete {Count} files from S3 bucket {BucketName}", deleteRequests.Count, bucketName); + _logger.LogInformation( + "Starting to delete {Count} files from S3 bucket {BucketName} with concurrency {Concurrency}", + deleteRequests.Count, + bucketName, + deleteConcurrency + ); // Emit file-level metrics (low cardinality) and logs for each file foreach (var delete in deleteRequests) @@ -240,45 +272,70 @@ private async Task Delete(SyncPlan plan, Cancel ctx) // Record by extension (low cardinality) if (!string.IsNullOrEmpty(extension)) - { FilesByExtensionCounter.Add(1, new("operation", "delete"), new("extension", extension)); - } // Log individual file operations for detailed analysis LogFileOperation(_logger, "delete", delete.DestinationPath, 0); } - // Process deletes in batches of 1000 (AWS S3 limit) - foreach (var batch in deleteRequests.Chunk(1000)) + // Process deletes in batches of 1000 (AWS S3 limit), parallelised across batches + await Parallel.ForEachAsync(deleteRequests.Chunk(1000), new ParallelOptions + { + CancellationToken = ctx, + MaxDegreeOfParallelism = deleteConcurrency + }, async (batch, token) => { var deleteObjectsRequest = new DeleteObjectsRequest { BucketName = bucketName, Objects = batch.Select(d => new KeyVersion { Key = d.DestinationPath }).ToList() }; - var response = await s3Client.DeleteObjectsAsync(deleteObjectsRequest, ctx); - if (response.HttpStatusCode != System.Net.HttpStatusCode.OK) + try { - _logger.LogError("Delete batch failed with status code {StatusCode}", response.HttpStatusCode); - foreach (var error in response.DeleteErrors) + var response = await s3Client.DeleteObjectsAsync(deleteObjectsRequest, token); + if (response.HttpStatusCode != System.Net.HttpStatusCode.OK) { - _logger.LogError("Failed to delete {Key}: {Message}", error.Key, error.Message); - collector.EmitError(error.Key, $"Failed to delete: {error.Message}"); + _logger.LogError("Delete batch failed with status code {StatusCode}", response.HttpStatusCode); + if (response.DeleteErrors is null or { Count: 0 }) + collector.EmitGlobalError($"Delete batch failed with status code {response.HttpStatusCode}"); + foreach (var error in response.DeleteErrors ?? Enumerable.Empty()) + { + _logger.LogError("Failed to delete {Key}: {Message}", error.Key, error.Message); + collector.EmitError(error.Key, $"Failed to delete: {error.Message}"); + } + } + else + { + var currentCount = Interlocked.Add(ref deleteCount, batch.Length); + _logger.LogInformation( + "Deleted {BatchCount} files ({CurrentCount}/{TotalCount})", + batch.Length, + currentCount, + deleteRequests.Count + ); } } - else + catch (Exception ex) when (ex is not OperationCanceledException) { - deleteCount += batch.Length; - _logger.LogInformation( - "Deleted {BatchCount} files ({CurrentCount}/{TotalCount})", - batch.Length, - deleteCount, - deleteRequests.Count - ); + _logger.LogError(ex, "Failed to delete batch from S3 bucket {BucketName}", bucketName); + foreach (var delete in batch) + collector.EmitError(delete.DestinationPath, "Failed to delete from S3", ex); } - } - - _logger.LogInformation("Successfully deleted {Count} files", deleteCount); + }); + + _logger.LogInformation( + "Finished deleting {DeletedCount}/{Count} files in {ElapsedMs}ms", + deleteCount, + deleteRequests.Count, + sw.ElapsedMilliseconds + ); } + _ = deleteActivity?.SetTag("docs.sync.delete.duration_ms", sw.Elapsed.TotalMilliseconds); + } + + private static int GetConcurrency(string environmentVariable, int defaultValue, int maxValue) + { + var configured = Environment.GetEnvironmentVariable(environmentVariable); + return int.TryParse(configured, out var parsed) && parsed > 0 ? Math.Min(parsed, maxValue) : defaultValue; } } diff --git a/src/services/Elastic.Documentation.Deploying/Synchronization/DocsSync.cs b/src/services/Elastic.Documentation.Deploying/Synchronization/DocsSync.cs index 4249a3485a..c6714484be 100644 --- a/src/services/Elastic.Documentation.Deploying/Synchronization/DocsSync.cs +++ b/src/services/Elastic.Documentation.Deploying/Synchronization/DocsSync.cs @@ -16,7 +16,8 @@ public record PlanValidationResult(bool Valid, float DeleteRatio, float DeleteTh public interface IDocsSyncApplyStrategy { - Task Apply(SyncPlan plan, Cancel ctx = default); + /// false if any file failed to upload or delete. + Task Apply(SyncPlan plan, Cancel ctx = default); } public record SyncRequest; diff --git a/tests/Elastic.Documentation.Deploying.Tests/DocsSyncTests.cs b/tests/Elastic.Documentation.Deploying.Tests/DocsSyncTests.cs index de9093daf9..eaf16c9bea 100644 --- a/tests/Elastic.Documentation.Deploying.Tests/DocsSyncTests.cs +++ b/tests/Elastic.Documentation.Deploying.Tests/DocsSyncTests.cs @@ -2,8 +2,10 @@ // Elasticsearch B.V licenses this file to you under the Apache 2.0 License. // See the LICENSE file in the project root for more information +using System.Collections.Concurrent; using System.Diagnostics; using System.IO.Abstractions.TestingHelpers; +using Actions.Core.Services; using Amazon.S3; using Amazon.S3.Model; using Amazon.S3.Transfer; @@ -11,6 +13,7 @@ using Elastic.Documentation.Assembler; using Elastic.Documentation.Configuration; using Elastic.Documentation.Configuration.Assembler; +using Elastic.Documentation.Deploying; using Elastic.Documentation.Deploying.Synchronization; using Elastic.Documentation.Diagnostics; using Elastic.Documentation.FileSystems; @@ -280,14 +283,11 @@ public async Task TestApply() { HttpStatusCode = System.Net.HttpStatusCode.OK }); - var transferredFiles = Array.Empty(); - A.CallTo(() => moxTransferUtility.UploadDirectoryAsync(A._, A._)).Invokes(( - TransferUtilityUploadDirectoryRequest request, - Cancel _ - ) => - { - transferredFiles = fileSystem.Directory.GetFiles(request.Directory, request.SearchPattern, request.SearchOption); - }); + var uploadRequests = new ConcurrentBag(); + A + .CallTo(() => moxTransferUtility.UploadAsync(A._, A._)) + .Invokes((TransferUtilityUploadRequest request, Cancel _) => uploadRequests.Add(request)) + .Returns(Task.CompletedTask); // Configure OpenTelemetry to capture telemetry var exportedActivities = new List(); @@ -303,14 +303,18 @@ Cancel _ await applier.Apply(plan, Cancel.None); // Assert - File operations - transferredFiles.Length.Should().Be(4); // 3 add requests + 1 update request - transferredFiles.Should().NotContain("docs/skip.md"); + uploadRequests.Count.Should().Be(4); // 3 add requests + 1 update request + uploadRequests.Should().OnlyContain(r => r.BucketName == "fake"); + uploadRequests.Should().OnlyContain(r => r.PartSize == S3EtagCalculator.PartSize); + uploadRequests.Select(r => r.Key).Should().BeEquivalentTo(["docs/add1.md", "docs/add2.md", "docs/add3.md", "docs/update.md"]); + uploadRequests.Select(r => r.FilePath).Should().NotContain("docs/skip.md"); A.CallTo(() => moxS3Client.DeleteObjectsAsync(A._, A._)).MustHaveHappenedOnceExactly(); + A.CallTo(() => moxTransferUtility.UploadAsync(A._, A._)).MustHaveHappened(); A.CallTo( () => moxTransferUtility.UploadDirectoryAsync(A._, A._) - ).MustHaveHappenedOnceExactly(); + ).MustNotHaveHappened(); // Assert - Telemetry spans are created exportedActivities.Should().Contain(a => a.DisplayName == "sync apply"); @@ -336,4 +340,89 @@ Cancel _ t => t.Key == "docs.sync.files.total" && Convert.ToInt64(t.Value, System.Globalization.CultureInfo.InvariantCulture) == 6 ); } + + [Test] + public async Task TestApply_DeleteRequests_ChunksDeletesAtS3Limit() + { + IReadOnlyCollection diagnosticsOutputs = []; + var collector = new DiagnosticsCollector(diagnosticsOutputs); + var moxS3Client = A.Fake(); + var moxTransferUtility = A.Fake(); + var fileSystem = new MockFileSystem(new MockFileSystemOptions + { + CurrentDirectory = Path.Join(Paths.WorkingDirectoryRoot.FullName, ".artifacts", "assembly"), + }); + var configurationContext = TestHelpers.CreateConfigurationContext(fileSystem); + var config = AssemblyConfiguration.Create(configurationContext.ConfigurationFileProvider); + var assembleFs = CheckoutsFileSystem.FromWorkingDirectory(fileSystem); + var context = new AssembleContext(config, configurationContext, "dev", collector, assembleFs); + var plan = new SyncPlan + { + RemoteListingCompleted = true, + DeleteThresholdDefault = null, + ExcludePatterns = [], + TotalRemoteFiles = 1001, + TotalSourceFiles = 1, + TotalSyncRequests = 1001, + AddRequests = [], + UpdateRequests = [], + SkipRequests = [], + DeleteRequests = Enumerable.Range(0, 1001).Select(i => new DeleteRequest { DestinationPath = $"docs/delete-{i}.md" }).ToArray() + }; + var deleteBatches = new ConcurrentBag(); + A + .CallTo(() => moxS3Client.DeleteObjectsAsync(A._, A._)) + .Invokes((DeleteObjectsRequest request, Cancel _) => deleteBatches.Add(request)) + .Returns(new DeleteObjectsResponse { HttpStatusCode = System.Net.HttpStatusCode.OK }); + + var applier = new AwsS3SyncApplyStrategy(new LoggerFactory(), moxS3Client, moxTransferUtility, "fake", context, collector); + + await applier.Apply(plan, Cancel.None); + + deleteBatches.Count.Should().Be(2); + deleteBatches.Should().OnlyContain(r => r.Objects.Count <= 1000); + } + + [Test] + public async Task Apply_UploadFailure_ReturnsFalse() + { + IReadOnlyCollection diagnosticsOutputs = []; + var collector = new DiagnosticsCollector(diagnosticsOutputs); + var mockS3Client = A.Fake(); + var mockTransferUtility = A.Fake(); + var githubActionsService = A.Fake(); + var outputDirectory = Path.Join(Paths.WorkingDirectoryRoot.FullName, ".artifacts", "assembly"); + var fileSystem = new MockFileSystem( + new Dictionary { { Path.Join(outputDirectory, "docs/add1.md"), new MockFileData("# New Document 1") } }, + new MockFileSystemOptions { CurrentDirectory = outputDirectory, } + ); + var configurationContext = TestHelpers.CreateConfigurationContext(fileSystem); + var config = AssemblyConfiguration.Create(configurationContext.ConfigurationFileProvider); + var assembleFs = CheckoutsFileSystem.FromWorkingDirectory(fileSystem); + var context = new AssembleContext(config, configurationContext, "dev", collector, assembleFs); + var plan = new SyncPlan + { + RemoteListingCompleted = true, + DeleteThresholdDefault = null, + ExcludePatterns = [], + TotalRemoteFiles = 0, + TotalSourceFiles = 1, + TotalSyncRequests = 1, + AddRequests = [new AddRequest { LocalPath = Path.Join(outputDirectory, "docs/add1.md"), DestinationPath = "docs/add1.md" }], + UpdateRequests = [], + SkipRequests = [], + DeleteRequests = [] + }; + var planPath = Path.Join(outputDirectory, "sync-plan.json"); + await fileSystem.File.WriteAllTextAsync(planPath, SyncPlan.Serialize(plan), CancellationToken.None); + A.CallTo(() => mockTransferUtility.UploadAsync(A._, A._)).Throws( + new AmazonS3Exception("Access denied") + ); + var service = new IncrementalDeployService(new LoggerFactory(), githubActionsService, mockS3Client, mockTransferUtility); + + var applyOk = await service.Apply(collector, context, "fake", planPath, Cancel.None); + + applyOk.Should().BeFalse(); + collector.Errors.Should().Be(1); + } } diff --git a/tests/Elastic.Documentation.Deploying.Tests/IncrementalDeployRoundTripTests.cs b/tests/Elastic.Documentation.Deploying.Tests/IncrementalDeployRoundTripTests.cs index b9bd963000..1e3bac343f 100644 --- a/tests/Elastic.Documentation.Deploying.Tests/IncrementalDeployRoundTripTests.cs +++ b/tests/Elastic.Documentation.Deploying.Tests/IncrementalDeployRoundTripTests.cs @@ -2,6 +2,7 @@ // Elasticsearch B.V licenses this file to you under the Apache 2.0 License. // See the LICENSE file in the project root for more information +using System.Collections.Concurrent; using System.IO.Abstractions.TestingHelpers; using Actions.Core.Services; using Amazon.S3; @@ -123,6 +124,7 @@ string outputDir { HttpStatusCode = System.Net.HttpStatusCode.OK }); + A.CallTo(() => xfer.UploadAsync(A._, A._)).Returns(Task.CompletedTask); var svc = new IncrementalDeployService(new LoggerFactory(), gh, s3, xfer, etagCalculator); return (fs, s3, xfer, gh, svc); @@ -138,15 +140,12 @@ private static async Task RunRoundTrip( string outputDir ) { - // Capture the files passed to the upload call - var transferredFiles = Array.Empty(); - A.CallTo(() => xfer.UploadDirectoryAsync(A._, A._)).Invokes(( - TransferUtilityUploadDirectoryRequest request, - Cancel _ - ) => - { - transferredFiles = fs.Directory.GetFiles(request.Directory, request.SearchPattern, request.SearchOption); - }); + // Capture the per-file upload requests + var uploadRequests = new ConcurrentBag(); + A + .CallTo(() => xfer.UploadAsync(A._, A._)) + .Invokes((TransferUtilityUploadRequest request, Cancel _) => uploadRequests.Add(request)) + .Returns(Task.CompletedTask); var planPath = Path.Join(outputDir, "sync-plan.json"); @@ -173,10 +172,14 @@ Cancel _ A.CallTo(() => gh.SetOutputAsync("plan-valid", "true")).MustHaveHappenedOnceExactly(); // Assert — uploads: 3 adds + 1 update; skip.md and remote-only delete.md not uploaded - transferredFiles - .Select(Path.GetFileName) + uploadRequests.Count.Should().Be(4); + uploadRequests.Should().OnlyContain(r => r.BucketName == "fake-bucket"); + uploadRequests.Should().OnlyContain(r => r.PartSize == S3EtagCalculator.PartSize); + uploadRequests + .Select(r => Path.GetFileName(r.FilePath)) .Should() .BeEquivalentTo(["add1.md", "add2.md", "add3.md", "update.md"], "skip.md is unchanged (ETag matches) so it is not re-uploaded"); + uploadRequests.Select(r => r.Key).Should().BeEquivalentTo(["docs/add1.md", "docs/add2.md", "docs/add3.md", "docs/update.md"]); // Assert — deletes: exactly one S3 delete call for docs/delete.md A.CallTo( @@ -186,8 +189,9 @@ Cancel _ ) ).MustHaveHappenedOnceExactly(); - // Assert — uploads called once - A.CallTo(() => xfer.UploadDirectoryAsync(A._, A._)).MustHaveHappenedOnceExactly(); + // Assert — per-file uploads only; no staging directory upload + A.CallTo(() => xfer.UploadAsync(A._, A._)).MustHaveHappened(); + A.CallTo(() => xfer.UploadDirectoryAsync(A._, A._)).MustNotHaveHappened(); } } From 562d5c7d65fb7299c1fd8d53c9b38751dd9bdb4e Mon Sep 17 00:00:00 2001 From: Martijn Laarman Date: Tue, 29 Sep 2026 13:43:49 +0200 Subject: [PATCH 2/3] Tests: prevent OTel activity bleed between concurrent Apply() tests TestApply subscribes the in-memory exporter to the global activity source. Any concurrent test that also calls Apply() emits activities that bleed into TestApply's exported list, causing the tag-count assertions to see the wrong plan's metrics. Add [NotInParallel("deploying-apply")] to all tests in the deploying test projects that exercise Apply() to prevent them running concurrently with each other. Co-Authored-By: Claude Sonnet 4.6 --- tests/Elastic.Documentation.Deploying.Tests/DocsSyncTests.cs | 5 +++++ .../IncrementalDeployRoundTripTests.cs | 3 +++ 2 files changed, 8 insertions(+) diff --git a/tests/Elastic.Documentation.Deploying.Tests/DocsSyncTests.cs b/tests/Elastic.Documentation.Deploying.Tests/DocsSyncTests.cs index eaf16c9bea..dadc761fa1 100644 --- a/tests/Elastic.Documentation.Deploying.Tests/DocsSyncTests.cs +++ b/tests/Elastic.Documentation.Deploying.Tests/DocsSyncTests.cs @@ -237,6 +237,9 @@ bool valid return (validator, planStrategy, plan); } + // NotInParallel: this test subscribes the OTel in-memory exporter to the global activity source; + // concurrent tests that also call Apply() emit activities that bleed into this exporter. + [NotInParallel("deploying-apply")] [Test] public async Task TestApply() { @@ -341,6 +344,7 @@ public async Task TestApply() ); } + [NotInParallel("deploying-apply")] [Test] public async Task TestApply_DeleteRequests_ChunksDeletesAtS3Limit() { @@ -383,6 +387,7 @@ public async Task TestApply_DeleteRequests_ChunksDeletesAtS3Limit() deleteBatches.Should().OnlyContain(r => r.Objects.Count <= 1000); } + [NotInParallel("deploying-apply")] [Test] public async Task Apply_UploadFailure_ReturnsFalse() { diff --git a/tests/Elastic.Documentation.Deploying.Tests/IncrementalDeployRoundTripTests.cs b/tests/Elastic.Documentation.Deploying.Tests/IncrementalDeployRoundTripTests.cs index 1e3bac343f..26dc611348 100644 --- a/tests/Elastic.Documentation.Deploying.Tests/IncrementalDeployRoundTripTests.cs +++ b/tests/Elastic.Documentation.Deploying.Tests/IncrementalDeployRoundTripTests.cs @@ -40,6 +40,7 @@ public class IncrementalDeployRoundTripTests private const string SkipETag = "aaaa0000skip0000etag0000aaaa0000"; private const string AnyOtherETag = "bbbb1111other1111etag1111bbbb1111"; + [NotInParallel("deploying-apply")] [Test] public async Task AssemblerRoundTrip() { @@ -54,6 +55,7 @@ public async Task AssemblerRoundTrip() await RunRoundTrip(fs, s3, xfer, gh, svc, context, outputDir); } + [NotInParallel("deploying-apply")] [Test] public async Task CodexRoundTrip() { @@ -209,6 +211,7 @@ public class IncrementalDeployExcludeTests private const string SkipETag = "aaaa0000skip0000etag0000aaaa0000"; private const string AnyOtherETag = "bbbb1111other1111etag1111bbbb1111"; + [NotInParallel("deploying-apply")] [Test] public async Task ExcludedRemoteObjectsAreNotDeleted() { From 6bf0f6fc773546214143ad2dca250e97718a3daf Mon Sep 17 00:00:00 2001 From: Martijn Laarman Date: Tue, 29 Sep 2026 13:54:07 +0200 Subject: [PATCH 3/3] Deploying: treat S3 per-key delete failures as errors regardless of HTTP status DeleteObjectsAsync returns 200 OK even when individual keys fail; errors are in DeleteErrors. Only counting batch.Length on a 200 response inflated the delete counter and let Apply() return true despite partial failures. Now errors from DeleteErrors are always emitted to the collector, and only successfully deleted keys count toward the progress total. Co-Authored-By: Claude Sonnet 4.6 --- .../Synchronization/AwsS3SyncApplyStrategy.cs | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/src/services/Elastic.Documentation.Deploying/Synchronization/AwsS3SyncApplyStrategy.cs b/src/services/Elastic.Documentation.Deploying/Synchronization/AwsS3SyncApplyStrategy.cs index 3437edb90c..84edb66fc8 100644 --- a/src/services/Elastic.Documentation.Deploying/Synchronization/AwsS3SyncApplyStrategy.cs +++ b/src/services/Elastic.Documentation.Deploying/Synchronization/AwsS3SyncApplyStrategy.cs @@ -293,23 +293,23 @@ private async Task Delete(SyncPlan plan, Cancel ctx) try { var response = await s3Client.DeleteObjectsAsync(deleteObjectsRequest, token); - if (response.HttpStatusCode != System.Net.HttpStatusCode.OK) + // S3 returns 200 even for partial failures; per-key errors live in DeleteErrors. + var errors = response.DeleteErrors ?? []; + foreach (var error in errors) { - _logger.LogError("Delete batch failed with status code {StatusCode}", response.HttpStatusCode); - if (response.DeleteErrors is null or { Count: 0 }) - collector.EmitGlobalError($"Delete batch failed with status code {response.HttpStatusCode}"); - foreach (var error in response.DeleteErrors ?? Enumerable.Empty()) - { - _logger.LogError("Failed to delete {Key}: {Message}", error.Key, error.Message); - collector.EmitError(error.Key, $"Failed to delete: {error.Message}"); - } + _logger.LogError("Failed to delete {Key}: {Message}", error.Key, error.Message); + collector.EmitError(error.Key, $"Failed to delete: {error.Message}"); } - else + if (response.HttpStatusCode != System.Net.HttpStatusCode.OK && errors.Count == 0) + collector.EmitGlobalError($"Delete batch failed with status code {response.HttpStatusCode}"); + + var successCount = batch.Length - errors.Count; + if (successCount > 0) { - var currentCount = Interlocked.Add(ref deleteCount, batch.Length); + var currentCount = Interlocked.Add(ref deleteCount, successCount); _logger.LogInformation( "Deleted {BatchCount} files ({CurrentCount}/{TotalCount})", - batch.Length, + successCount, currentCount, deleteRequests.Count );