fix.plonds-s3-resumable-publish

PenguinLogisticsOnlineNetworkDistributionSystem/src/Plonds.Core/Publishing/PlondsPublishOptions.cs

@@ -8,4 +8,7 @@ public sealed record PlondsPublishOptions(
     string FilesZipPath,
     string WorkDir,
     string S3KeyPrefix,
-    PlondsS3ClientOptions S3);
+    PlondsS3ClientOptions S3)
+{
+    public int DirectoryUploadConcurrency { get; init; } = 4;
+}

PenguinLogisticsOnlineNetworkDistributionSystem/src/Plonds.Core/Publishing/PlondsPublisher.cs

@@ -62,11 +62,12 @@ public sealed class PlondsPublisher
 
         using var s3 = new PlondsS3Client(options.S3);
 
-        var changedFileCount = await UploadDirectoryAsync(s3, changedExtractRoot, changedFolderKey, cancellationToken).ConfigureAwait(false);
-        var filesFileCount = await UploadDirectoryAsync(s3, filesExtractRoot, filesFolderKey, cancellationToken).ConfigureAwait(false);
+        await UploadArtifactAsync(s3, changedZipPath, changedZipKey, "application/zip", cancellationToken).ConfigureAwait(false);
+        await UploadArtifactAsync(s3, filesZipPath, filesZipKey, "application/zip", cancellationToken).ConfigureAwait(false);
 
-        await s3.UploadFileAsync(new PlondsS3ObjectUpload(changedZipPath, changedZipKey, "application/zip"), cancellationToken).ConfigureAwait(false);
-        await s3.UploadFileAsync(new PlondsS3ObjectUpload(filesZipPath, filesZipKey, "application/zip"), cancellationToken).ConfigureAwait(false);
+        var directoryConcurrency = Math.Max(1, options.DirectoryUploadConcurrency);
+        var changedFileCount = await UploadDirectoryAsync(s3, changedExtractRoot, changedFolderKey, directoryConcurrency, cancellationToken).ConfigureAwait(false);
+        var filesFileCount = await UploadDirectoryAsync(s3, filesExtractRoot, filesFolderKey, directoryConcurrency, cancellationToken).ConfigureAwait(false);
 
         var updatedChecksums = new Dictionary<string, string>(manifest.Checksums, StringComparer.OrdinalIgnoreCase)
         {
@@ -130,18 +131,79 @@ public sealed class PlondsPublisher
         PlondsS3Client s3,
         string sourceDirectory,
         string destinationKeyPrefix,
+        int concurrency,
         CancellationToken cancellationToken)
     {
-        var count = 0;
-        foreach (var filePath in Directory.EnumerateFiles(sourceDirectory, "*", SearchOption.AllDirectories).OrderBy(x => x, StringComparer.OrdinalIgnoreCase))
+        var files = Directory.EnumerateFiles(sourceDirectory, "*", SearchOption.AllDirectories)
+            .Select(filePath =>
+            {
+                var relativePath = PayloadUtilities.NormalizeRelativePath(Path.GetRelativePath(sourceDirectory, filePath));
+                return new DirectoryUploadPlan(
+                    SourcePath: filePath,
+                    ObjectKey: $"{destinationKeyPrefix}/{relativePath}",
+                    ContentType: ResolveContentType(filePath));
+            })
+            .OrderBy(x => x.ObjectKey, StringComparer.OrdinalIgnoreCase)
+            .ToArray();
+
+        if (files.Length == 0)
         {
-            var relativePath = PayloadUtilities.NormalizeRelativePath(Path.GetRelativePath(sourceDirectory, filePath));
-            var objectKey = $"{destinationKeyPrefix}/{relativePath}";
-            await s3.UploadFileAsync(new PlondsS3ObjectUpload(filePath, objectKey, ResolveContentType(filePath)), cancellationToken).ConfigureAwait(false);
-            count++;
+            Console.WriteLine($"No files found under {sourceDirectory}; skipping S3 directory upload to {destinationKeyPrefix}.");
+            return 0;
         }
 
-        return count;
+        Console.WriteLine($"Uploading S3 directory {destinationKeyPrefix}: {files.Length} files with concurrency {concurrency}.");
+
+        var processed = 0;
+        var uploaded = 0;
+        var skipped = 0;
+        await Parallel.ForEachAsync(
+            files,
+            new ParallelOptions
+            {
+                MaxDegreeOfParallelism = concurrency,
+                CancellationToken = cancellationToken
+            },
+            async (file, token) =>
+            {
+                var didUpload = await s3.UploadFileIfChangedAsync(
+                    new PlondsS3ObjectUpload(file.SourcePath, file.ObjectKey, file.ContentType),
+                    token).ConfigureAwait(false);
+
+                if (didUpload)
+                {
+                    Interlocked.Increment(ref uploaded);
+                }
+                else
+                {
+                    Interlocked.Increment(ref skipped);
+                }
+
+                var current = Interlocked.Increment(ref processed);
+                if (current == files.Length || current % 10 == 0)
+                {
+                    Console.WriteLine($"S3 directory progress {destinationKeyPrefix}: {current}/{files.Length} processed ({uploaded} uploaded, {skipped} skipped).");
+                }
+            }).ConfigureAwait(false);
+
+        Console.WriteLine($"Finished S3 directory {destinationKeyPrefix}: {files.Length} files processed ({uploaded} uploaded, {skipped} skipped).");
+        return files.Length;
+    }
+
+    private static async Task UploadArtifactAsync(
+        PlondsS3Client s3,
+        string sourcePath,
+        string objectKey,
+        string contentType,
+        CancellationToken cancellationToken)
+    {
+        var didUpload = await s3.UploadFileIfChangedAsync(
+            new PlondsS3ObjectUpload(sourcePath, objectKey, contentType),
+            cancellationToken).ConfigureAwait(false);
+
+        Console.WriteLine(didUpload
+            ? $"Published S3 artifact {objectKey}."
+            : $"S3 artifact {objectKey} already exists with matching size.");
     }
 
     private static PlondsManifest LoadManifest(string manifestPath)
@@ -201,4 +263,9 @@ public sealed class PlondsPublisher
             ? throw new ArgumentException($"{name} is required.", name)
             : value.Trim();
     }
+
+    private sealed record DirectoryUploadPlan(
+        string SourcePath,
+        string ObjectKey,
+        string ContentType);
 }

PenguinLogisticsOnlineNetworkDistributionSystem/src/Plonds.Core/Publishing/PlondsS3Client.cs

@@ -69,6 +69,29 @@ public sealed class PlondsS3Client : IDisposable
         }
     }
 
+    public async Task<bool> UploadFileIfChangedAsync(PlondsS3ObjectUpload upload, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(upload);
+
+        var sourcePath = Path.GetFullPath(upload.SourcePath);
+        if (!File.Exists(sourcePath))
+        {
+            throw new FileNotFoundException("S3 upload source file not found.", sourcePath);
+        }
+
+        var key = NormalizeKey(upload.Key);
+        var contentLength = new FileInfo(sourcePath).Length;
+        var existing = await TryGetObjectInfoForUploadAsync(key, cancellationToken).ConfigureAwait(false);
+        if (existing?.ContentLength == contentLength)
+        {
+            Console.WriteLine($"Skipping S3 object {key}; existing object has matching size {FormatBytes(contentLength)}.");
+            return false;
+        }
+
+        await UploadFileAsync(upload, cancellationToken).ConfigureAwait(false);
+        return true;
+    }
+
     private async Task UploadFileOnceAsync(
         string sourcePath,
         string key,
@@ -106,6 +129,16 @@ public sealed class PlondsS3Client : IDisposable
     }
 
     public async Task EnsureObjectExistsAsync(string key, CancellationToken cancellationToken = default)
+    {
+        var normalizedKey = NormalizeKey(key);
+        var objectInfo = await TryGetObjectInfoAsync(normalizedKey, cancellationToken).ConfigureAwait(false);
+        if (objectInfo is null)
+        {
+            throw new InvalidOperationException($"S3 object verification failed for {normalizedKey}: object was not found.");
+        }
+    }
+
+    public async Task<PlondsS3ObjectInfo?> TryGetObjectInfoAsync(string key, CancellationToken cancellationToken = default)
     {
         var normalizedKey = NormalizeKey(key);
         var now = DateTimeOffset.UtcNow;
@@ -115,9 +148,32 @@ public sealed class PlondsS3Client : IDisposable
         SignRequest(request, normalizedKey, EmptyPayloadHash, now);
 
         using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
+        if (response.StatusCode is HttpStatusCode.NotFound)
+        {
+            return null;
+        }
+
         if (!response.IsSuccessStatusCode)
         {
-            throw new InvalidOperationException($"S3 object verification failed for {normalizedKey}: HTTP {(int)response.StatusCode} {response.ReasonPhrase}.");
+            throw new InvalidOperationException($"S3 object metadata lookup failed for {normalizedKey}: HTTP {(int)response.StatusCode} {response.ReasonPhrase}.");
+        }
+
+        return new PlondsS3ObjectInfo(
+            Key: normalizedKey,
+            ContentLength: response.Content.Headers.ContentLength,
+            ETag: response.Headers.ETag?.Tag);
+    }
+
+    private async Task<PlondsS3ObjectInfo?> TryGetObjectInfoForUploadAsync(string key, CancellationToken cancellationToken)
+    {
+        try
+        {
+            return await TryGetObjectInfoAsync(key, cancellationToken).ConfigureAwait(false);
+        }
+        catch (Exception ex) when (ex is not OperationCanceledException)
+        {
+            Console.Error.WriteLine($"S3 object metadata lookup for {key} failed; uploading anyway. {ex.Message}");
+            return null;
         }
     }
 

PenguinLogisticsOnlineNetworkDistributionSystem/src/Plonds.Core/Publishing/PlondsS3ObjectInfo.cs

@@ -0,0 +1,6 @@
+namespace Plonds.Core.Publishing;
+
+public sealed record PlondsS3ObjectInfo(
+    string Key,
+    long? ContentLength,
+    string? ETag);

PenguinLogisticsOnlineNetworkDistributionSystem/src/Plonds.Tool/Program.cs

@@ -113,7 +113,10 @@ internal static class PlondsCli
                 AccessKey: Require(options, "s3-access-key"),
                 SecretKey: Require(options, "s3-secret-key"),
                 PublicBaseUrl: Require(options, "s3-public-base-url"),
-                PublicBaseKeyPrefix: Get(options, "s3-public-base-key-prefix", string.Empty) ?? string.Empty))).ConfigureAwait(false);
+                PublicBaseKeyPrefix: Get(options, "s3-public-base-key-prefix", string.Empty) ?? string.Empty))
+        {
+            DirectoryUploadConcurrency = GetInt(options, "directory-upload-concurrency", 4)
+        }).ConfigureAwait(false);
 
         Console.WriteLine($"Published PLONDS release {result.ReleaseTag}:");
         Console.WriteLine($"  Prefix:             {result.VersionPrefix}");
@@ -212,6 +215,19 @@ internal static class PlondsCli
         Console.WriteLine("    --s3-public-base-url <url> Public URL prefix for uploaded keys");
         Console.WriteLine("    [--s3-public-base-key-prefix <prefix>] Key prefix already represented by public URL");
         Console.WriteLine("    [--s3-prefix <prefix>]  Object key prefix (default: lanmountain/update/plonds)");
+        Console.WriteLine("    [--directory-upload-concurrency <n>] Parallel file uploads for expanded directories (default: 4)");
         Console.WriteLine("    [--work-dir <dir>]      Temporary publish work directory");
     }
+
+    private static int GetInt(IReadOnlyDictionary<string, string> options, string key, int defaultValue)
+    {
+        if (!options.TryGetValue(key, out var value) || string.IsNullOrWhiteSpace(value))
+        {
+            return defaultValue;
+        }
+
+        return int.TryParse(value, out var parsed) && parsed > 0
+            ? parsed
+            : throw new InvalidOperationException($"Option --{key} must be a positive integer.");
+    }
 }