diff --git a/src/services/Elastic.Documentation.Deploying/IncrementalDeployService.cs b/src/services/Elastic.Documentation.Deploying/IncrementalDeployService.cs index 5d6a09b0cf..a4ee687a58 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 76309dda2a..ffb4daaaa7 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,16 +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); - // Files with an explicit Content-Type are staged in sibling directories, one per type, so each - // directory upload carries its own type. The SDK shares one Headers collection across all per-file - // requests of a directory upload, so setting ContentType per file would leak it onto other files. - var typedDirs = new Dictionary(); - 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) + CancellationToken = ctx, + MaxDegreeOfParallelism = uploadConcurrency + }, async (upload, token) => + { + try { var operation = addPaths.Contains(upload.LocalPath) ? "add" : "update"; var fileSize = context.WriteFileSystem.FileInfo.New(upload.LocalPath).Length; @@ -191,60 +209,42 @@ private async Task Upload(SyncPlan plan, Cancel ctx) FilesByExtensionCounter.Add(1, new("operation", operation), new("extension", extension)); LogFileOperation(_logger, operation, upload.DestinationPath, fileSize); - var stagingDir = tempDir; - if (SpecContentType(upload.DestinationPath) is { } specType) - { - if (!typedDirs.TryGetValue(specType, out stagingDir)) - typedDirs[specType] = stagingDir = $"{tempDir}-{typedDirs.Count}"; - } - - var destPath = context.WriteFileSystem.Path.Join(stagingDir, upload.DestinationPath); - var destDirPath = context.WriteFileSystem.Path.GetDirectoryName(destPath)!; - _ = context.WriteFileSystem.Directory.CreateDirectory(destDirPath); - context.WriteFileSystem.File.Copy(upload.LocalPath, destPath); - } - _logger.LogInformation("Uploading {Count} files to S3 bucket {BucketName}", uploadRequests.Count, bucketName); - var uploads = typedDirs.Select(kv => (Directory: kv.Value, ContentType: (string?)kv.Key)).Prepend( - (Directory: tempDir, ContentType: null) - ); - foreach (var (directory, contentType) in uploads) - { - if ( - !context.WriteFileSystem.Directory.Exists(directory) - || !context.WriteFileSystem.Directory.EnumerateFiles(directory, "*", SearchOption.AllDirectories).Any() - ) - continue; - var directoryRequest = new TransferUtilityUploadDirectoryRequest + var request = new TransferUtilityUploadRequest { BucketName = bucketName, - Directory = directory, - SearchPattern = "*", - SearchOption = SearchOption.AllDirectories, - UploadFilesConcurrently = true + FilePath = upload.LocalPath, + Key = upload.DestinationPath, + PartSize = S3EtagCalculator.PartSize }; - if (contentType is not null) - directoryRequest.ContentType = contentType; - directoryRequest.UploadDirectoryProgressEvent += DisplayProgress; - _logger.LogDebug("Starting directory upload from {Directory}", directory); - await transferUtility.UploadDirectoryAsync(directoryRequest, ctx); + // Per-file requests own their headers, so the type cannot leak onto other uploads. + if (SpecContentType(upload.DestinationPath) is { } specType) + request.ContentType = specType; + request.UploadProgressEvent += DisplayUploadProgress; + await transferUtility.UploadAsync(request, token); + _ = Interlocked.Increment(ref uploadedCount); } - _logger.LogInformation( - "Successfully uploaded {Count} files ({AddCount} added, {UpdateCount} updated)", - uploadRequests.Count, - addCount, - updateCount - ); - } - finally - { - // Clean up temp directories - foreach (var directory in typedDirs.Values.Prepend(tempDir)) + catch (Exception ex) when (ex is not OperationCanceledException) { - if (context.WriteFileSystem.Directory.Exists(directory)) - context.WriteFileSystem.Directory.Delete(directory, 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); } /// @@ -261,16 +261,24 @@ private async Task Upload(SyncPlan plan, Cancel ctx) 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) @@ -279,45 +287,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); + // S3 returns 200 even for partial failures; per-key errors live in DeleteErrors. + var errors = response.DeleteErrors ?? []; + foreach (var error in errors) { _logger.LogError("Failed to delete {Key}: {Message}", error.Key, error.Message); collector.EmitError(error.Key, $"Failed to delete: {error.Message}"); } + 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, successCount); + _logger.LogInformation( + "Deleted {BatchCount} files ({CurrentCount}/{TotalCount})", + successCount, + 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 32a3c8638d..012125575c 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 3d0de1f5d1..5ffdb6ce7d 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; @@ -279,6 +282,8 @@ 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. [Test] [NotInParallel(nameof(AwsS3SyncApplyStrategy))] // TestApply asserts on spans from a process-wide ActivitySource public async Task TestApply() @@ -326,14 +331,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(); @@ -349,14 +351,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"); @@ -383,6 +389,93 @@ Cancel _ ); } + [Test] + [NotInParallel(nameof(AwsS3SyncApplyStrategy))] + 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] + [NotInParallel(nameof(AwsS3SyncApplyStrategy))] + 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); + } + [Test] [NotInParallel(nameof(AwsS3SyncApplyStrategy))] public async Task Apply_SetsContentTypeOnSpecUploads() @@ -427,14 +520,11 @@ public async Task Apply_SetsContentTypeOnSpecUploads() SkipRequests = [], DeleteRequests = [] }; - // The SDK shares one Headers collection across the per-file requests of a directory upload, - // so a typed file must never share a directory upload with files of another type. - var uploads = new List<(string? ContentType, string[] Files)>(); - A.CallTo(() => moxTransferUtility.UploadDirectoryAsync(A._, A._)).Invokes( - (TransferUtilityUploadDirectoryRequest request, Cancel _) => uploads.Add( - (request.ContentType, fileSystem.Directory.GetFiles(request.Directory, request.SearchPattern, request.SearchOption)) - ) - ); + var uploads = new ConcurrentBag(); + A + .CallTo(() => moxTransferUtility.UploadAsync(A._, A._)) + .Invokes((TransferUtilityUploadRequest request, Cancel _) => uploads.Add(request)) + .Returns(Task.CompletedTask); var applier = new AwsS3SyncApplyStrategy(new LoggerFactory(), A.Fake(), moxTransferUtility, "fake", context, collector); // Act @@ -442,9 +532,9 @@ public async Task Apply_SetsContentTypeOnSpecUploads() // Assert uploads.Should().HaveCount(3); - var byType = uploads.ToDictionary(u => u.ContentType ?? "", u => u.Files.Select(f => Path.GetFileName(f)).ToArray()); - byType["text/plain; charset=utf-8"].Should().BeEquivalentTo(["es.yaml"]); - byType["application/json"].Should().BeEquivalentTo(["es.json"]); - byType[""].Should().BeEquivalentTo(["index.md"]); + var byKey = uploads.ToDictionary(u => u.Key, u => u.ContentType); + byKey["docs/api/doc/es.yaml"].Should().Be("text/plain; charset=utf-8"); + byKey["docs/api/doc/es.json"].Should().Be("application/json"); + byKey["docs/index.md"].Should().BeNull(); } } diff --git a/tests/Elastic.Documentation.Deploying.Tests/IncrementalDeployRoundTripTests.cs b/tests/Elastic.Documentation.Deploying.Tests/IncrementalDeployRoundTripTests.cs index 94a1ceca56..16dd142d3c 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; @@ -126,6 +127,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); @@ -141,15 +143,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"); @@ -176,10 +175,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( @@ -189,8 +192,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(); } }