From 5aeba4f0c966c32c992a402f496fa7002ae08935 Mon Sep 17 00:00:00 2001 From: Jan Calanog Date: Mon, 13 Jul 2026 18:12:15 +0200 Subject: [PATCH 1/2] Sync: upload/delete files concurrently instead of one directory call S3 sync is network-bound, not CPU-bound, so uploads and deletes now run in parallel per-file/per-batch (configurable via DOCS_SYNC_UPLOAD_CONCURRENCY and DOCS_SYNC_DELETE_CONCURRENCY) instead of copying every file to a temp directory and issuing a single directory upload sized to CPU count. Failures are now reported per-file/per-batch via the diagnostics collector instead of aborting the whole sync, and Apply returns false if any file failed. The S3 client also configures SDK-level retry (standard mode) to absorb throttling under the higher concurrency. Co-Authored-By: Claude Sonnet 5 --- .../IncrementalDeployService.cs | 13 +- .../Synchronization/AwsS3SyncApplyStrategy.cs | 150 +++++++++++------- .../Synchronization/DocsSync.cs | 3 +- .../DocsSyncTests.cs | 129 ++++++++++++++- .../IncrementalDeployRoundTripTests.cs | 28 ++-- 5 files changed, 246 insertions(+), 77 deletions(-) diff --git a/src/services/Elastic.Documentation.Deploying/IncrementalDeployService.cs b/src/services/Elastic.Documentation.Deploying/IncrementalDeployService.cs index 09493d643b..5725d9ce5d 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,14 @@ 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, IDocsSyncContext context, string s3BucketName, string @out, float? deleteThreshold, string[] excludePatterns, Cancel ctx) { @@ -81,8 +89,7 @@ public async Task Apply(IDiagnosticsCollector collector, IDocsSyncContext 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 54e083d452..9105a11407 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 @@ -77,13 +85,13 @@ 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, + EventId = 4, Level = LogLevel.Debug, Message = "{Args}")] - private static partial void LogProgress(ILogger logger, UploadDirectoryProgressArgs args); + private static partial void LogUploadProgress(ILogger logger, UploadProgressArgs args); [LoggerMessage( EventId = 3, @@ -91,9 +99,10 @@ IDiagnosticsCollector collector 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") @@ -144,15 +153,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) { @@ -162,63 +177,62 @@ 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 { - _logger.LogInformation("Copying {Count} files to temp directory", uploadRequests.Count); - foreach (var upload in uploadRequests) - { - var operation = addPaths.Contains(upload.LocalPath) ? "add" : "update"; - var fileSize = context.WriteFileSystem.FileInfo.New(upload.LocalPath).Length; - var extension = Path.GetExtension(upload.DestinationPath).ToLowerInvariant(); - - FileSizeHistogram.Record(fileSize); - if (!string.IsNullOrEmpty(extension)) - 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 directoryRequest = new TransferUtilityUploadDirectoryRequest + CancellationToken = ctx, + MaxDegreeOfParallelism = uploadConcurrency + }, async (upload, token) => + { + var operation = addPaths.Contains(upload.LocalPath) ? "add" : "update"; + var fileSize = context.WriteFileSystem.FileInfo.New(upload.LocalPath).Length; + var extension = Path.GetExtension(upload.DestinationPath).ToLowerInvariant(); + + FileSizeHistogram.Record(fileSize); + if (!string.IsNullOrEmpty(extension)) + FilesByExtensionCounter.Add(1, new("operation", operation), new("extension", extension)); + LogFileOperation(_logger, operation, upload.DestinationPath, fileSize); + + var request = new TransferUtilityUploadRequest { BucketName = bucketName, - Directory = tempDir, - SearchPattern = "*", - SearchOption = SearchOption.AllDirectories, - UploadFilesConcurrently = true + FilePath = upload.LocalPath, + Key = upload.DestinationPath, + PartSize = S3EtagCalculator.PartSize }; - 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); - } + request.UploadProgressEvent += DisplayUploadProgress; + try + { + await transferUtility.UploadAsync(request, token); + _ = Interlocked.Increment(ref uploadedCount); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _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) @@ -238,7 +252,11 @@ private async Task Delete(SyncPlan plan, Cancel ctx) } // Process deletes in batches of 1000 (AWS S3 limit) - foreach (var batch in deleteRequests.Chunk(1000)) + await Parallel.ForEachAsync(deleteRequests.Chunk(1000), new ParallelOptions + { + CancellationToken = ctx, + MaxDegreeOfParallelism = deleteConcurrency + }, async (batch, token) => { var deleteObjectsRequest = new DeleteObjectsRequest { @@ -248,25 +266,45 @@ private async Task Delete(SyncPlan plan, Cancel ctx) 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 104f65fd6c..adf496c846 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-integration/Elastic.Documentation.IntegrationTests/DocsSyncTests.cs b/tests-integration/Elastic.Documentation.IntegrationTests/DocsSyncTests.cs index 5f90a5d00d..4928bb8862 100644 --- a/tests-integration/Elastic.Documentation.IntegrationTests/DocsSyncTests.cs +++ b/tests-integration/Elastic.Documentation.IntegrationTests/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.Integrations.S3; @@ -275,12 +278,13 @@ public async Task TestApply() { HttpStatusCode = System.Net.HttpStatusCode.OK }); - var transferredFiles = Array.Empty(); - A.CallTo(() => moxTransferUtility.UploadDirectoryAsync(A._, A._)) - .Invokes((TransferUtilityUploadDirectoryRequest request, Cancel _) => + var uploadRequests = new ConcurrentBag(); + A.CallTo(() => moxTransferUtility.UploadAsync(A._, A._)) + .Invokes((TransferUtilityUploadRequest request, Cancel _) => { - transferredFiles = fileSystem.Directory.GetFiles(request.Directory, request.SearchPattern, request.SearchOption); - }); + uploadRequests.Add(request); + }) + .Returns(Task.CompletedTask); // Configure OpenTelemetry to capture telemetry var exportedActivities = new List(); @@ -295,14 +299,20 @@ public async Task TestApply() 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"); @@ -318,4 +328,107 @@ public async Task TestApply() tagObjects.Should().Contain(t => t.Key == "docs.sync.files.skipped" && Convert.ToInt64(t.Value, System.Globalization.CultureInfo.InvariantCulture) == 1); tagObjects.Should().Contain(t => t.Key == "docs.sync.files.total" && Convert.ToInt64(t.Value, System.Globalization.CultureInfo.InvariantCulture) == 6); } + + [Fact] + 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 checkoutDirectory = Path.Join(Paths.WorkingDirectoryRoot.FullName, ".artifacts", "assembly"); + var scopedFs = FileSystemFactory.ScopeCurrentWorkingDirectory(fileSystem); + var scopedWriteFs = FileSystemFactory.ScopeCurrentWorkingDirectoryForWrite(fileSystem); + var context = new AssembleContext(config, configurationContext, "dev", collector, scopedFs, scopedWriteFs, null, checkoutDirectory); + 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); + } + + [Fact] + 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 scopedFs = FileSystemFactory.ScopeCurrentWorkingDirectory(fileSystem); + var scopedWriteFs = FileSystemFactory.ScopeCurrentWorkingDirectoryForWrite(fileSystem); + var context = new AssembleContext(config, configurationContext, "dev", collector, scopedFs, scopedWriteFs, null, outputDirectory); + 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)); + 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-integration/Elastic.Documentation.IntegrationTests/IncrementalDeployRoundTripTests.cs b/tests-integration/Elastic.Documentation.IntegrationTests/IncrementalDeployRoundTripTests.cs index 5facecbe6a..c69c55e8cf 100644 --- a/tests-integration/Elastic.Documentation.IntegrationTests/IncrementalDeployRoundTripTests.cs +++ b/tests-integration/Elastic.Documentation.IntegrationTests/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; @@ -122,6 +123,8 @@ private static (MockFileSystem fs, IAmazonS3 s3, ITransferUtility xfer, ICoreSer A.CallTo(() => s3.DeleteObjectsAsync(A._, A._)) .Returns(new DeleteObjectsResponse { 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); @@ -136,13 +139,13 @@ private static async Task RunRoundTrip( IDocsSyncContext context, string outputDir) { - // Capture the files passed to the upload call - var transferredFiles = Array.Empty(); - A.CallTo(() => xfer.UploadDirectoryAsync(A._, A._)) - .Invokes((TransferUtilityUploadDirectoryRequest request, Cancel _) => + var uploadRequests = new ConcurrentBag(); + A.CallTo(() => xfer.UploadAsync(A._, A._)) + .Invokes((TransferUtilityUploadRequest request, Cancel _) => { - transferredFiles = fs.Directory.GetFiles(request.Directory, request.SearchPattern, request.SearchOption); - }); + uploadRequests.Add(request); + }) + .Returns(Task.CompletedTask); var planPath = Path.Join(outputDir, "sync-plan.json"); @@ -161,9 +164,14 @@ private static async Task RunRoundTrip( 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).Should() + 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(() => s3.DeleteObjectsAsync( @@ -171,9 +179,11 @@ private static async Task RunRoundTrip( A._)) .MustHaveHappenedOnceExactly(); - // Assert — uploads called once + // Assert — per-file uploads only; no staging directory upload + A.CallTo(() => xfer.UploadAsync(A._, A._)) + .MustHaveHappened(); A.CallTo(() => xfer.UploadDirectoryAsync(A._, A._)) - .MustHaveHappenedOnceExactly(); + .MustNotHaveHappened(); } } From 132bc449b85650664920790719dba2800ba77dc8 Mon Sep 17 00:00:00 2001 From: Jan Calanog Date: Mon, 13 Jul 2026 18:16:03 +0200 Subject: [PATCH 2/2] Fix xUnit1051 lint warning in new test Co-Authored-By: Claude Sonnet 5 --- .../Elastic.Documentation.IntegrationTests/DocsSyncTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests-integration/Elastic.Documentation.IntegrationTests/DocsSyncTests.cs b/tests-integration/Elastic.Documentation.IntegrationTests/DocsSyncTests.cs index 4928bb8862..27b9b94a16 100644 --- a/tests-integration/Elastic.Documentation.IntegrationTests/DocsSyncTests.cs +++ b/tests-integration/Elastic.Documentation.IntegrationTests/DocsSyncTests.cs @@ -421,7 +421,7 @@ public async Task Apply_UploadFailure_ReturnsFalse() DeleteRequests = [] }; var planPath = Path.Join(outputDirectory, "sync-plan.json"); - await fileSystem.File.WriteAllTextAsync(planPath, SyncPlan.Serialize(plan)); + await fileSystem.File.WriteAllTextAsync(planPath, SyncPlan.Serialize(plan), TestContext.Current.CancellationToken); A.CallTo(() => mockTransferUtility.UploadAsync(A._, A._)) .Throws(new AmazonS3Exception("Access denied")); var service = new IncrementalDeployService(new LoggerFactory(), githubActionsService, mockS3Client, mockTransferUtility);