add retries and fix up tests

2024-08-23 13:17:07 +01:00
parent 72b670f356
commit 1b9faf628d
7 changed files with 654 additions and 419 deletions
@@ -1,210 +1,310 @@
 import * as core from '@actions/core'
 import * as ociContainer from './oci-container'

-export async function uploadOCIImageManifest(
-  token: string,
-  registry: URL,
-  repository: string,
-  manifest: ociContainer.OCIImageManifest,
-  blobs: Map<string, Buffer>,
-  tag?: string
-): Promise<string> {
-  const b64Token = Buffer.from(token).toString('base64')
-  const manifestSHA = ociContainer.sha256Digest(manifest)
+const defaultRetries = 5
+const defaultBackoff = 1000
+const retryableStatusCodes = [408, 429, 500, 502, 503, 504]

-  if (tag) {
-    core.info(
-      `Uploading manifest ${manifestSHA} with tag ${tag} to ${repository}.`
-    )
-  } else {
-    core.info(`Uploading manifest ${manifestSHA} to ${repository}.`)
-  }
+export interface RetryOptions {
+  retries: number
+  backoff: number
+}

-  // We must also upload the config layer
-  const layersToUpload = manifest.layers.concat(manifest.config)
+export class Client {
+  private _b64Token: string
+  private _registry: URL
+  private _retryOptions: RetryOptions

-  const layerUploads: Promise<void>[] = layersToUpload.map(async layer => {
-    const blob = blobs.get(layer.digest)
-    if (!blob) {
-      throw new Error(`Blob for layer ${layer.digest} not found`)
+  constructor(
+    token: string,
+    registry: URL,
+    retryOptions: RetryOptions = {
+      retries: defaultRetries,
+      backoff: defaultBackoff
    }
-    return uploadLayer(layer, blob, registry, repository, b64Token)
-  })
-
-  await Promise.all(layerUploads)
-
-  const publishedDigest = await uploadManifest(
-    JSON.stringify(manifest),
-    manifest.mediaType,
-    registry,
-    repository,
-    tag || manifestSHA,
-    b64Token
-  )
-
-  if (publishedDigest !== manifestSHA) {
-    throw new Error(
-      `Digest mismatch. Expected ${manifestSHA}, got ${publishedDigest}.`
-    )
+  ) {
+    this._b64Token = Buffer.from(token).toString('base64')
+    this._registry = registry
+    this._retryOptions = retryOptions
  }

-  return manifestSHA
-}
+  async uploadOCIImageManifest(
+    repository: string,
+    manifest: ociContainer.OCIImageManifest,
+    blobs: Map<string, Buffer>,
+    tag?: string
+  ): Promise<string> {
+    const manifestSHA = ociContainer.sha256Digest(manifest)

-export async function uploadOCIIndexManifest(
-  token: string,
-  registry: URL,
-  repository: string,
-  manifest: ociContainer.OCIIndexManifest,
-  tag: string
-): Promise<string> {
-  const b64Token = Buffer.from(token).toString('base64')
-  const manifestSHA = ociContainer.sha256Digest(manifest)
+    if (tag) {
+      core.info(
+        `Uploading manifest ${manifestSHA} with tag ${tag} to ${repository}.`
+      )
+    } else {
+      core.info(`Uploading manifest ${manifestSHA} to ${repository}.`)
+    }

-  core.info(
-    `Uploading index manifest ${manifestSHA} with tag ${tag} to ${repository}.`
-  )
+    // We must also upload the config layer
+    const layersToUpload = manifest.layers.concat(manifest.config)

-  const publishedDigest = await uploadManifest(
-    JSON.stringify(manifest),
-    manifest.mediaType,
-    registry,
-    repository,
-    tag,
-    b64Token
-  )
+    const layerUploads: Promise<void>[] = layersToUpload.map(async layer => {
+      const blob = blobs.get(layer.digest)
+      if (!blob) {
+        throw new Error(`Blob for layer ${layer.digest} not found`)
+      }
+      return this.uploadLayer(layer, blob, repository)
+    })

-  if (publishedDigest !== manifestSHA) {
-    throw new Error(
-      `Digest mismatch. Expected ${manifestSHA}, got ${publishedDigest}.`
+    await Promise.all(layerUploads)
+
+    const publishedDigest = await this.uploadManifest(
+      JSON.stringify(manifest),
+      manifest.mediaType,
+      repository,
+      tag || manifestSHA
    )
+
+    if (publishedDigest !== manifestSHA) {
+      throw new Error(
+        `Digest mismatch. Expected ${manifestSHA}, got ${publishedDigest}.`
+      )
+    }
+
+    return manifestSHA
  }

-  return manifestSHA
-}
+  async uploadOCIIndexManifest(
+    repository: string,
+    manifest: ociContainer.OCIIndexManifest,
+    tag: string
+  ): Promise<string> {
+    const manifestSHA = ociContainer.sha256Digest(manifest)

-async function uploadLayer(
-  layer: ociContainer.Descriptor,
-  data: Buffer,
-  registryURL: URL,
-  repository: string,
-  b64Token: string
-): Promise<void> {
-  const checkExistsResponse = await fetchWithDebug(
-    checkBlobEndpoint(registryURL, repository, layer.digest),
-    {
-      method: 'HEAD',
+    core.info(
+      `Uploading index manifest ${manifestSHA} with tag ${tag} to ${repository}.`
+    )
+
+    const publishedDigest = await this.uploadManifest(
+      JSON.stringify(manifest),
+      manifest.mediaType,
+      repository,
+      tag
+    )
+
+    if (publishedDigest !== manifestSHA) {
+      throw new Error(
+        `Digest mismatch. Expected ${manifestSHA}, got ${publishedDigest}.`
+      )
+    }
+
+    return manifestSHA
+  }
+
+  private async uploadLayer(
+    layer: ociContainer.Descriptor,
+    data: Buffer,
+    repository: string
+  ): Promise<void> {
+    const checkExistsResponse = await this.fetchWithRetries(
+      this.checkBlobEndpoint(repository, layer.digest),
+      {
+        method: 'HEAD',
+        headers: {
+          Authorization: `Bearer ${this._b64Token}`
+        }
+      }
+    )
+
+    if (
+      checkExistsResponse.status === 200 ||
+      checkExistsResponse.status === 202
+    ) {
+      core.info(`Layer ${layer.digest} already exists. Skipping upload.`)
+      return
+    }
+
+    if (checkExistsResponse.status !== 404) {
+      throw new Error(
+        await errorMessageForFailedRequest(
+          `check blob (${layer.digest}) exists`,
+          checkExistsResponse
+        )
+      )
+    }
+
+    core.info(`Uploading layer ${layer.digest}.`)
+
+    const initiateUploadBlobURL = this.uploadBlobEndpoint(repository)
+
+    const initiateUploadResponse = await this.fetchWithRetries(
+      initiateUploadBlobURL,
+      {
+        method: 'POST',
+        headers: {
+          Authorization: `Bearer ${this._b64Token}`
+        },
+        body: JSON.stringify(layer)
+      }
+    )
+
+    if (initiateUploadResponse.status !== 202) {
+      throw new Error(
+        await errorMessageForFailedRequest(
+          `initiate layer upload`,
+          initiateUploadResponse
+        )
+      )
+    }
+
+    const locationResponseHeader =
+      initiateUploadResponse.headers.get('location')
+    if (locationResponseHeader === undefined) {
+      throw new Error(
+        `No location header in response from upload post ${initiateUploadBlobURL} for layer ${layer.digest}`
+      )
+    }
+
+    const pathname = `${locationResponseHeader}?digest=${layer.digest}`
+    const uploadBlobUrl = new URL(pathname, this._registry).toString()
+
+    const putResponse = await this.fetchWithRetries(uploadBlobUrl, {
+      method: 'PUT',
      headers: {
-        Authorization: `Bearer ${b64Token}`
+        Authorization: `Bearer ${this._b64Token}`,
+        'Content-Type': 'application/octet-stream',
+        'Accept-Encoding': 'gzip',
+        'Content-Length': layer.size.toString()
+      },
+      body: data
+    })
+
+    if (putResponse.status !== 201) {
+      throw new Error(
+        await errorMessageForFailedRequest(
+          `layer (${layer.digest}) upload`,
+          putResponse
+        )
+      )
+    }
+  }
+
+  // Uploads the manifest and returns the digest returned by GHCR
+  private async uploadManifest(
+    manifestJSON: string,
+    manifestMediaType: string,
+    repository: string,
+    version: string
+  ): Promise<string> {
+    const manifestUrl = this.manifestEndpoint(repository, version)
+
+    core.info(`Uploading manifest to ${manifestUrl}.`)
+
+    const putResponse = await this.fetchWithRetries(manifestUrl, {
+      method: 'PUT',
+      headers: {
+        Authorization: `Bearer ${this._b64Token}`,
+        'Content-Type': manifestMediaType
+      },
+      body: manifestJSON
+    })
+
+    if (putResponse.status !== 201) {
+      throw new Error(
+        await errorMessageForFailedRequest(`manifest upload`, putResponse)
+      )
+    }
+
+    const digestResponseHeader =
+      putResponse.headers.get('docker-content-digest') || ''
+
+    return digestResponseHeader
+  }
+
+  private checkBlobEndpoint(repository: string, digest: string): string {
+    return new URL(
+      `v2/${repository}/blobs/${digest}`,
+      this._registry
+    ).toString()
+  }
+
+  private uploadBlobEndpoint(repository: string): string {
+    return new URL(`v2/${repository}/blobs/uploads/`, this._registry).toString()
+  }
+
+  private manifestEndpoint(repository: string, version: string): string {
+    return new URL(
+      `v2/${repository}/manifests/${version}`,
+      this._registry
+    ).toString()
+  }
+
+  // TODO: Add retries with backoff
+  private async fetchWithDebug(
+    url: string,
+    config: RequestInit = {}
+  ): Promise<Response> {
+    core.debug(`Request from ${url} with config: ${JSON.stringify(config)}`)
+    try {
+      const response = await fetch(url, config)
+      core.debug(`Response with ${JSON.stringify(response)}`)
+      return response
+    } catch (error) {
+      core.debug(`Error with ${error}`)
+      throw error
+    }
+  }
+
+  private async fetchWithRetries(
+    url: string,
+    config: RequestInit = {}
+  ): Promise<Response> {
+    const allowedAttempts = this._retryOptions.retries + 1 // Initial attempt + retries
+
+    for (
+      let attemptNumber = 1;
+      attemptNumber <= allowedAttempts;
+      attemptNumber++
+    ) {
+      let backoff = this._retryOptions.backoff
+
+      try {
+        const response = await this.fetchWithDebug(url, config)
+
+        // If this is the last attempt, just return it
+        if (attemptNumber === allowedAttempts) {
+          return response
+        }
+
+        // If the response is retryable, backoff and retry
+        if (retryableStatusCodes.includes(response.status)) {
+          const retryAfter = response.headers.get('retry-after')
+          if (retryAfter) {
+            backoff = parseInt(retryAfter) * 1000 // convert to ms
+          }
+
+          core.info(
+            `Received ${response.status} response. Retrying after ${backoff}ms...`
+          )
+          await new Promise(resolve => setTimeout(resolve, backoff))
+          continue
+        }
+
+        // Otherwise, just return the response
+        return response
+      } catch (error) {
+        // If this is the last attempt, throw the error
+        if (attemptNumber === allowedAttempts) {
+          throw error
+        }
+
+        core.info(`Encountered error: ${error}. Retrying after ${backoff}ms...`)
+        await new Promise(resolve => setTimeout(resolve, backoff))
      }
    }
-  )

-  if (
-    checkExistsResponse.status === 200 ||
-    checkExistsResponse.status === 202
-  ) {
-    core.info(`Layer ${layer.digest} already exists. Skipping upload.`)
-    return
+    // Should be unreachable
+    throw new Error('Exhausted retries without a successful response')
  }
-
-  if (checkExistsResponse.status !== 404) {
-    throw new Error(
-      await errorMessageForFailedRequest(
-        `check blob (${layer.digest}) exists`,
-        checkExistsResponse
-      )
-    )
-  }
-
-  core.info(`Uploading layer ${layer.digest}.`)
-
-  const initiateUploadBlobURL = uploadBlobEndpoint(registryURL, repository)
-
-  const initiateUploadResponse = await fetchWithDebug(initiateUploadBlobURL, {
-    method: 'POST',
-    headers: {
-      Authorization: `Bearer ${b64Token}`
-    },
-    body: JSON.stringify(layer)
-  })
-
-  if (initiateUploadResponse.status !== 202) {
-    throw new Error(
-      await errorMessageForFailedRequest(
-        `initiate layer upload`,
-        initiateUploadResponse
-      )
-    )
-  }
-
-  const locationResponseHeader = initiateUploadResponse.headers.get('location')
-  if (locationResponseHeader === undefined) {
-    throw new Error(
-      `No location header in response from upload post ${initiateUploadBlobURL} for layer ${layer.digest}`
-    )
-  }
-
-  const pathname = `${locationResponseHeader}?digest=${layer.digest}`
-  const uploadBlobUrl = new URL(pathname, registryURL).toString()
-
-  const putResponse = await fetchWithDebug(uploadBlobUrl, {
-    method: 'PUT',
-    headers: {
-      Authorization: `Bearer ${b64Token}`,
-      'Content-Type': 'application/octet-stream',
-      'Accept-Encoding': 'gzip',
-      'Content-Length': layer.size.toString()
-    },
-    body: data
-  })
-
-  if (putResponse.status !== 201) {
-    throw new Error(
-      await errorMessageForFailedRequest(
-        `layer (${layer.digest}) upload`,
-        putResponse
-      )
-    )
-  }
-}
-
-// Uploads the manifest and returns the digest returned by GHCR
-async function uploadManifest(
-  manifestJSON: string,
-  manifestMediaType: string,
-  registry: URL,
-  repository: string,
-  version: string,
-  b64Token: string
-): Promise<string> {
-  const manifestUrl = manifestEndpoint(registry, repository, version)
-
-  core.info(`Uploading manifest to ${manifestUrl}.`)
-
-  const putResponse = await fetchWithDebug(manifestUrl, {
-    method: 'PUT',
-    headers: {
-      Authorization: `Bearer ${b64Token}`,
-      'Content-Type': manifestMediaType
-    },
-    body: manifestJSON
-  })
-
-  if (putResponse.status !== 201) {
-    throw new Error(
-      await errorMessageForFailedRequest(`manifest upload`, putResponse)
-    )
-  }
-
-  const digestResponseHeader = putResponse.headers.get('docker-content-digest')
-  if (digestResponseHeader === undefined || digestResponseHeader === null) {
-    throw new Error(
-      `No digest header in response from PUT manifest ${manifestUrl}`
-    )
-  }
-
-  return digestResponseHeader
 }

 interface ghcrError {
@@ -257,39 +357,3 @@ function isGHCRError(obj: unknown): boolean {
    typeof (obj as { message: unknown }).message === 'string'
  )
 }
-
-function checkBlobEndpoint(
-  registry: URL,
-  repository: string,
-  digest: string
-): string {
-  return new URL(`v2/${repository}/blobs/${digest}`, registry).toString()
-}
-
-function uploadBlobEndpoint(registry: URL, repository: string): string {
-  return new URL(`v2/${repository}/blobs/uploads/`, registry).toString()
-}
-
-function manifestEndpoint(
-  registry: URL,
-  repository: string,
-  version: string
-): string {
-  return new URL(`v2/${repository}/manifests/${version}`, registry).toString()
-}
-
-// TODO: Add retries with backoff
-const fetchWithDebug = async (
-  url: string,
-  config: RequestInit = {}
-): Promise<Response> => {
-  core.debug(`Request from ${url} with config: ${JSON.stringify(config)}`)
-  try {
-    const response = await fetch(url, config)
-    core.debug(`Response with ${JSON.stringify(response)}`)
-    return response
-  } catch (error) {
-    core.debug(`Error with ${error}`)
-    throw error
-  }
-}
@@ -53,6 +53,11 @@ export async function run(): Promise<void> {

    const manifestDigest = ociContainer.sha256Digest(manifest)

+    const ghcrClient = new ghcr.Client(
+      options.token,
+      options.containerRegistryUrl
+    )
+
    // Attestations are not supported in GHES.
    if (!options.isEnterprise) {
      const { bundle, bundleDigest } = await generateAttestation(
@@ -77,7 +82,8 @@ export async function run(): Promise<void> {
      )

      const { attestationSHA, referrerIndexSHA } = await publishAttestation(
-        options,
+        ghcrClient,
+        options.nameWithOwner,
        bundle,
        bundleDigest,
        manifest,
@@ -96,19 +102,14 @@ export async function run(): Promise<void> {
    }

    const publishedDigest = await publishImmutableActionVersion(
-      options,
+      ghcrClient,
+      options.nameWithOwner,
      semverTag.raw,
      archives.zipFile,
      archives.tarFile,
      manifest
    )

-    if (manifestDigest !== publishedDigest) {
-      throw new Error(
-        `Unexpected digest returned for manifest. Expected ${manifestDigest}, got ${publishedDigest}`
-      )
-    }
-
    core.setOutput('package-manifest-sha', publishedDigest)
  } catch (error) {
    // Fail the workflow run if an error occurs
@@ -138,7 +139,8 @@ function parseSemverTagFromRef(opts: cfg.PublishActionOptions): semver.SemVer {
 }

 async function publishImmutableActionVersion(
-  options: cfg.PublishActionOptions,
+  client: ghcr.Client,
+  nameWithOwner: string,
  semverTag: string,
  zipFile: fsHelper.FileMetadata,
  tarFile: fsHelper.FileMetadata,
@@ -155,10 +157,8 @@ async function publishImmutableActionVersion(
  files.set(tarFile.sha256, fsHelper.readFileContents(tarFile.path))
  files.set(ociContainer.emptyConfigSha, Buffer.from('{}'))

-  return await ghcr.uploadOCIImageManifest(
-    options.token,
-    options.containerRegistryUrl,
-    options.nameWithOwner,
+  return await client.uploadOCIImageManifest(
+    nameWithOwner,
    manifest,
    files,
    semverTag
@@ -166,7 +166,8 @@ async function publishImmutableActionVersion(
 }

 async function publishAttestation(
-  options: cfg.PublishActionOptions,
+  client: ghcr.Client,
+  nameWithOwner: string,
  bundle: Buffer,
  bundleDigest: string,
  subjectManifest: ociContainer.OCIImageManifest,
@@ -191,10 +192,8 @@ async function publishAttestation(
  files.set(ociContainer.emptyConfigSha, Buffer.from('{}'))
  files.set(bundleDigest, bundle)

-  const attestationSHA = await ghcr.uploadOCIImageManifest(
-    options.token,
-    options.containerRegistryUrl,
-    options.nameWithOwner,
+  const attestationSHA = await client.uploadOCIImageManifest(
+    nameWithOwner,
    attestationManifest,
    files
  )
@@ -206,10 +205,8 @@ async function publishAttestation(
    `Publishing referrer index ${referrerIndexManifestDigest} with tag ${referrerTag} for attestation ${attestationManifestDigest} and subject ${subjectManifestDigest}.`
  )

-  const referrerIndexSHA = await ghcr.uploadOCIIndexManifest(
-    options.token,
-    options.containerRegistryUrl,
-    options.nameWithOwner,
+  const referrerIndexSHA = await client.uploadOCIIndexManifest(
+    nameWithOwner,
    referrerIndexManifest,
    referrerTag
  )