From c0591b4cf53ea11e214c49c70916c4f6c22a4c07 Mon Sep 17 00:00:00 2001 From: Saksham Gupta Date: Sun, 5 Jul 2026 16:18:53 +0530 Subject: [PATCH] Fix ECR BLOB_UPLOAD_INVALID by forcing new session on push retry --- .../Client/RegistryClient+Push.swift | 128 ++++++++++++------ .../Client/RegistryClient.swift | 6 +- 2 files changed, 87 insertions(+), 47 deletions(-) diff --git a/Sources/ContainerizationOCI/Client/RegistryClient+Push.swift b/Sources/ContainerizationOCI/Client/RegistryClient+Push.swift index dc056baa6..9206ef5fa 100644 --- a/Sources/ContainerizationOCI/Client/RegistryClient+Push.swift +++ b/Sources/ContainerizationOCI/Client/RegistryClient+Push.swift @@ -103,39 +103,7 @@ extension RegistryClient { headers = [ ("Content-Type", mediaType) ] - } else { - // Start upload request for blobs. - components.path = "/v2/\(name)/blobs/uploads/" - try await request(components: components, method: .POST) { response in - switch response.status { - case .ok, .accepted, .noContent: - break - case .created: - throw ContainerizationError(.exists, message: "content already exists \(descriptor.digest)") - default: - let url = components.url?.absoluteString ?? "unknown" - let reason = await ErrorResponse.fromResponseBody(response.body)?.jsonString - throw Error.invalidStatus(url: url, response.status, reason: reason) - } - - // Get the location to upload the blob. - guard let location = response.headers.first(name: "Location") else { - throw ContainerizationError(.invalidArgument, message: "missing required header Location") - } - - guard let urlComponents = URLComponents(string: location) else { - throw ContainerizationError(.invalidArgument, message: "invalid url \(location)") - } - var queryItems = urlComponents.queryItems ?? [] - queryItems.append(URLQueryItem(name: "digest", value: descriptor.digest)) - components.path = urlComponents.path - components.queryItems = queryItems - headers = [ - ("Content-Type", "application/octet-stream"), - ("Content-Length", String(descriptor.size)), - ] - } - } + } // We have to pass a body closure rather than a body to reset the stream when retrying. let bodyClosure = { @@ -144,19 +112,89 @@ extension RegistryClient { return body } - return try await request(components: components, method: .PUT, bodyClosure: bodyClosure, headers: headers) { response in - switch response.status { - case .ok, .created, .noContent: - break - default: - let url = components.url?.absoluteString ?? "unknown" - let reason = await ErrorResponse.fromResponseBody(response.body)?.jsonString - throw Error.invalidStatus(url: url, response.status, reason: reason) - } + // Establish our smart retry limits + let maxRetries = self.retryOptions?.maxRetries ?? 3 + let retryInterval = self.retryOptions?.retryInterval ?? 1_000_000_000 + var retryCount = 0 + + while true { + do { + if !isManifest { + // Start upload request for blobs to get a FRESH session URL. + var postComponents = base + postComponents.path = "/v2/\(name)/blobs/uploads/" + + // Override generic retry: we do NOT want the generic client retrying this POST + try await request( + components: postComponents, + method: .POST, + retryOptionsOverride: RetryOptions(maxRetries: 0, retryInterval: 0) + ) { response in + switch response.status { + case .ok, .accepted, .noContent: + break + case .created: + throw ContainerizationError(.exists, message: "content already exists \(descriptor.digest)") + default: + let url = postComponents.url?.absoluteString ?? "unknown" + let reason = await ErrorResponse.fromResponseBody(response.body)?.jsonString + throw Error.invalidStatus(url: url, response.status, reason: reason) + } + + guard let location = response.headers.first(name: "Location") else { + throw ContainerizationError(.invalidArgument, message: "missing required header Location") + } + + guard let urlComponents = URLComponents(string: location) else { + throw ContainerizationError(.invalidArgument, message: "invalid url \(location)") + } + + var queryItems = urlComponents.queryItems ?? [] + queryItems.append(URLQueryItem(name: "digest", value: descriptor.digest)) + components.path = urlComponents.path + components.queryItems = queryItems + headers = [ + ("Content-Type", "application/octet-stream"), + ("Content-Length", String(descriptor.size)), + ] + } + } + + // Execute the PUT request with generic retries completely disabled + try await request( + components: components, + method: .PUT, + bodyClosure: bodyClosure, + headers: headers, + retryOptionsOverride: RetryOptions(maxRetries: 0, retryInterval: 0) + ) { response in + switch response.status { + case .ok, .created, .noContent: + break + default: + let url = components.url?.absoluteString ?? "unknown" + let reason = await ErrorResponse.fromResponseBody(response.body)?.jsonString + throw Error.invalidStatus(url: url, response.status, reason: reason) + } - guard descriptor.digest == response.headers.first(name: "Docker-Content-Digest") else { - let required = response.headers.first(name: "Docker-Content-Digest") ?? "" - throw ContainerizationError(.internalError, message: "digest mismatch \(descriptor.digest) != \(required)") + guard descriptor.digest == response.headers.first(name: "Docker-Content-Digest") else { + let required = response.headers.first(name: "Docker-Content-Digest") ?? "" + throw ContainerizationError(.internalError, message: "digest mismatch \(descriptor.digest) != \(required)") + } + } + + // If the PUT succeeds without throwing, we break out of our smart retry loop + break + + } catch { + // Intercept the failure. If we haven't hit our max retries, loop again to generate a new POST session. + if retryCount < maxRetries { + retryCount += 1 + try await Task.sleep(nanoseconds: retryInterval) + continue + } else { + throw error + } } } } diff --git a/Sources/ContainerizationOCI/Client/RegistryClient.swift b/Sources/ContainerizationOCI/Client/RegistryClient.swift index 14f3142ca..8862b689a 100644 --- a/Sources/ContainerizationOCI/Client/RegistryClient.swift +++ b/Sources/ContainerizationOCI/Client/RegistryClient.swift @@ -146,6 +146,7 @@ public final class RegistryClient: ContentClient { method: HTTPMethod = .GET, bodyClosure: () throws -> HTTPClientRequest.Body? = { nil }, headers: [(String, String)]? = nil, + retryOptionsOverride: RetryOptions? = nil, closure: (HTTPClientResponse) async throws -> T ) async throws -> T { guard let path = components.url?.absoluteString else { @@ -173,6 +174,7 @@ public final class RegistryClient: ContentClient { var response: HTTPClientResponse? while true { request.body = try bodyClosure() + let activeRetryOptions = retryOptionsOverride ?? self.retryOptions do { let _response = try await client.execute(request, deadline: .distantFuture) response = _response @@ -217,7 +219,7 @@ public final class RegistryClient: ContentClient { retryCount += 1 continue } - guard let retryOptions = self.retryOptions else { + guard let retryOptions = activeRetryOptions else { break } guard retryCount < retryOptions.maxRetries else { @@ -245,7 +247,7 @@ public final class RegistryClient: ContentClient { } } #endif - guard let retryOptions = self.retryOptions, retryCount < retryOptions.maxRetries else { + guard let retryOptions = activeRetryOptions, retryCount < retryOptions.maxRetries else { throw error } retryCount += 1