Skip to content

Fix ECR BLOB_UPLOAD_INVALID by forcing new session on push retry #792

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
am-saksham wants to merge 1 commit into
base: main
Choose a base branch
from
+87 −47
Open

Fix ECR BLOB_UPLOAD_INVALID by forcing new session on push retry #792

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 83 additions & 45 deletions Sources/ContainerizationOCI/Client/RegistryClient+Push.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = {
Expand All @@ -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
}
}
}
}
Expand Down
6 changes: 4 additions & 2 deletions Sources/ContainerizationOCI/Client/RegistryClient.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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
Expand Down