fix.plonds-s3-resumable-publish

.github/workflows/plonds-uploader.yml

@@ -21,11 +21,13 @@ env:
   DOTNET_VERSION: '10.0.x'
   PLONDS_S3_PREFIX: lanmountain/update/plonds
   PLONDS_S3_PUBLIC_BASE_KEY_PREFIX: lanmountain/update
+  PLONDS_S3_DIRECTORY_UPLOAD_CONCURRENCY: '4'
 
 jobs:
   publish:
     if: ${{ github.event_name != 'workflow_run' || github.event.workflow_run.conclusion == 'success' }}
     runs-on: ubuntu-latest
+    timeout-minutes: 45
     permissions:
       contents: write
       actions: read
@@ -116,7 +118,8 @@ jobs:
             --s3-access-key "$S3_ACCESS_KEY" \
             --s3-secret-key "$S3_SECRET_KEY" \
             --s3-public-base-url "$PUBLIC_BASE" \
-            --s3-public-base-key-prefix "$PLONDS_S3_PUBLIC_BASE_KEY_PREFIX"
+            --s3-public-base-key-prefix "$PLONDS_S3_PUBLIC_BASE_KEY_PREFIX" \
+            --directory-upload-concurrency "$PLONDS_S3_DIRECTORY_UPLOAD_CONCURRENCY"
 
           jq -e '.downloads.github.changedZipUrl and .downloads.github.filesZipUrl and .downloads.s3.changedFolderUrl and .downloads.s3.filesFolderUrl' plonds-assets/PLONDS.json >/dev/null
 

.trae/specs/plonds-client-service/spec.md

@@ -116,6 +116,8 @@ Publisher 上传到 S3 的版本目录：
 - `<prefix>/PLONDS.json` 是 S3 的固定 latest manifest 地址，和 GitHub Release latest manifest 一起作为客户端内置初始 source。
 - GitHub Release 仍可保留平台原始文件名，例如 `files-windows-x64.zip`。
 - `PLONDS.json` 的 downloads 字段同时包含 GitHub 与 S3 的增量包、完整包位置。
+- Publisher 必须先完成版本目录内的 `changed.zip`、`Files.zip`、解压目录和版本 `PLONDS.json` 上传，再更新 `<prefix>/PLONDS.json` latest 指针。
+- Publisher 的 S3 目录上传必须支持重跑续传；同 key 且大小一致的对象可以跳过，避免失败后从头上传完整包目录。
 
 ## 7. 建议代码结构
 

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

@@ -27,11 +27,16 @@ public sealed class PlondsS3Client : IDisposable
             AccessKey = Require(options.AccessKey, nameof(options.AccessKey)),
             SecretKey = Require(options.SecretKey, nameof(options.SecretKey)),
             PublicBaseUrl = Require(options.PublicBaseUrl, nameof(options.PublicBaseUrl)).TrimEnd('/'),
-            PublicBaseKeyPrefix = NormalizeOptionalKeyPrefix(options.PublicBaseKeyPrefix)
+            PublicBaseKeyPrefix = NormalizeOptionalKeyPrefix(options.PublicBaseKeyPrefix),
+            RequestTimeout = options.RequestTimeout <= TimeSpan.Zero ? TimeSpan.FromMinutes(30) : options.RequestTimeout,
+            MaxUploadAttempts = Math.Max(1, options.MaxUploadAttempts)
         };
 
-        this.httpClient = httpClient ?? new HttpClient();
         ownsHttpClient = httpClient is null;
+        this.httpClient = httpClient ?? new HttpClient
+        {
+            Timeout = this.options.RequestTimeout
+        };
     }
 
     public async Task UploadFileAsync(PlondsS3ObjectUpload upload, CancellationToken cancellationToken = default)
@@ -47,20 +52,70 @@ public sealed class PlondsS3Client : IDisposable
         var key = NormalizeKey(upload.Key);
         var payloadHash = PayloadUtilities.ComputeSha256(sourcePath);
         var contentLength = new FileInfo(sourcePath).Length;
+
+        for (var attempt = 1; attempt <= options.MaxUploadAttempts; attempt++)
+        {
+            try
+            {
+                await UploadFileOnceAsync(sourcePath, key, upload.ContentType, payloadHash, contentLength, attempt, cancellationToken).ConfigureAwait(false);
+                return;
+            }
+            catch (Exception ex) when (attempt < options.MaxUploadAttempts && IsRetriable(ex))
+            {
+                var delay = TimeSpan.FromSeconds(Math.Min(30, Math.Pow(2, attempt)));
+                Console.Error.WriteLine($"S3 upload retry {attempt + 1}/{options.MaxUploadAttempts} for {key} after {delay.TotalSeconds:0}s: {ex.Message}");
+                await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
+            }
+        }
+    }
+
+    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,
+        string? contentType,
+        string payloadHash,
+        long contentLength,
+        int attempt,
+        CancellationToken cancellationToken = default)
+    {
         var now = DateTimeOffset.UtcNow;
         var requestUri = BuildObjectUri(key);
+        Console.WriteLine($"Uploading S3 object {key} ({FormatBytes(contentLength)}), attempt {attempt}/{options.MaxUploadAttempts}.");
 
-        using var content = new StreamContent(File.OpenRead(sourcePath));
-        content.Headers.ContentType = new MediaTypeHeaderValue(string.IsNullOrWhiteSpace(upload.ContentType)
+        await using var fileStream = File.OpenRead(sourcePath);
+        using var content = new StreamContent(fileStream);
+        content.Headers.ContentType = new MediaTypeHeaderValue(string.IsNullOrWhiteSpace(contentType)
             ? "application/octet-stream"
-            : upload.ContentType);
+            : contentType);
         content.Headers.ContentLength = contentLength;
 
         using var request = new HttpRequestMessage(HttpMethod.Put, requestUri)
         {
             Content = content
         };
-
         SignRequest(request, key, payloadHash, now);
 
         using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
@@ -69,9 +124,21 @@ public sealed class PlondsS3Client : IDisposable
             var body = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
             throw new InvalidOperationException($"S3 upload failed for {key}: HTTP {(int)response.StatusCode} {response.ReasonPhrase}. {Truncate(body, 512)}");
         }
+
+        Console.WriteLine($"Uploaded S3 object {key}.");
     }
 
     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;
@@ -81,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;
         }
     }
 
@@ -257,4 +347,28 @@ public sealed class PlondsS3Client : IDisposable
 
         return value[..maxLength];
     }
+
+    private static bool IsRetriable(Exception exception)
+    {
+        if (exception is TaskCanceledException or TimeoutException or HttpRequestException)
+        {
+            return true;
+        }
+
+        return exception.InnerException is not null && IsRetriable(exception.InnerException);
+    }
+
+    private static string FormatBytes(long bytes)
+    {
+        string[] units = ["B", "KB", "MB", "GB"];
+        double value = bytes;
+        var unit = 0;
+        while (value >= 1024 && unit < units.Length - 1)
+        {
+            value /= 1024;
+            unit++;
+        }
+
+        return $"{value:0.##} {units[unit]}";
+    }
 }

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

@@ -7,4 +7,9 @@ public sealed record PlondsS3ClientOptions(
     string AccessKey,
     string SecretKey,
     string PublicBaseUrl,
-    string PublicBaseKeyPrefix = "");
+    string PublicBaseKeyPrefix = "")
+{
+    public TimeSpan RequestTimeout { get; init; } = TimeSpan.FromMinutes(30);
+
+    public int MaxUploadAttempts { get; init; } = 3;
+}

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.");
+    }
 }