Sync: upload/delete S3 files concurrently, not one directory call

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<IncrementalDeployService>();
-	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<bool> 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)

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<AwsS3SyncApplyStrategy> _logger = logFactory.CreateLogger<AwsS3SyncApplyStrategy>();
 
-	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<bool> 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<UploadRequest>().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,80 +186,65 @@ 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<string, string>();
-			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);
+				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();
 
-					var stagingDir = tempDir;
-					if (SpecContentType(upload.DestinationPath) is { } specType)
-					{
-						if (!typedDirs.TryGetValue(specType, out stagingDir))
-							typedDirs[specType] = stagingDir = $"{tempDir}-{typedDirs.Count}";
-					}
+				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(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)
+				var request = new TransferUtilityUploadRequest
 				{
-					if (
-						!context.WriteFileSystem.Directory.Exists(directory)
-						|| !context.WriteFileSystem.Directory.EnumerateFiles(directory, "*", SearchOption.AllDirectories).Any()
-					)
-						continue;
-					var directoryRequest = new TransferUtilityUploadDirectoryRequest
-					{
-						BucketName = bucketName,
-						Directory = directory,
-						SearchPattern = "*",
-						SearchOption = SearchOption.AllDirectories,
-						UploadFilesConcurrently = true
-					};
-					if (contentType is not null)
-						directoryRequest.ContentType = contentType;
-					directoryRequest.UploadDirectoryProgressEvent += DisplayProgress;
-					_logger.LogDebug("Starting directory upload from {Directory}", directory);
-					await transferUtility.UploadDirectoryAsync(directoryRequest, ctx);
+					BucketName = bucketName,
+					FilePath = upload.LocalPath,
+					Key = upload.DestinationPath,
+					PartSize = S3EtagCalculator.PartSize
+				};
+				// 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;
+				try
+				{
+					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);
 	}
 
 	/// <summary>
@@ -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)
@@ -288,36 +296,63 @@ 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
 				{
 					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("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<DeleteError>())
+						{
+							_logger.LogError("Failed to delete {Key}: {Message}", error.Key, error.Message);
+							collector.EmitError(error.Key, $"Failed to delete: {error.Message}");
+						}
+					}
+					else
 					{
-						_logger.LogError("Failed to delete {Key}: {Message}", error.Key, error.Message);
-						collector.EmitError(error.Key, $"Failed to delete: {error.Message}");
+						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;
 	}
 }

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);
+	/// <returns>false if any file failed to upload or delete.</returns>
+	Task<bool> Apply(SyncPlan plan, Cancel ctx = default);
 }
 
 public record SyncRequest;