Use zlib for compression

packages/cache/__tests__/saveCache.test.ts

@@ -2,10 +2,14 @@ import * as core from '@actions/core'
 import * as path from 'path'
 import {saveCache} from '../src/cache'
 import * as cacheHttpClient from '../src/internal/cacheHttpClient'
+import * as cacheTwirpClient from '../src/internal/cacheTwirpClient'
+import {GetCacheBlobUploadURLResponse} from '../src/generated/results/api/v1/blobcache'
+import {BlobCacheServiceClientJSON} from '../src/generated/results/api/v1/blobcache.twirp'
 import * as cacheUtils from '../src/internal/cacheUtils'
 import {CacheFilename, CompressionMethod} from '../src/internal/constants'
 import * as tar from '../src/internal/tar'
 import {TypedResponse} from '@actions/http-client/lib/interfaces'
+import * as uploadCache from '../src/internal/v2/upload-cache'
 import {
   ReserveCacheResponse,
   ITypedResponseWithError
@@ -327,3 +331,74 @@ test('save with non existing path should not save cache', async () => {
     `Path Validation Error: Path(s) specified in the action for caching do(es) not exist, hence no cache is being saved.`
   )
 })
+
+test('throwaway test', async () => {
+  const filePath = 'node_modules'
+  const primaryKey = 'Linux-node-bb828da54c148048dd17899ba9fda624811cfb43'
+  const cachePaths = [path.resolve(filePath)]
+
+  const cacheSignedURL = 'https://container.blob.core.windows.net/cache/${primaryKey}?sig=1234'
+  const getCacheBlobUploadURL: GetCacheBlobUploadURLResponse = {
+    urls: [
+      {
+        key: primaryKey,
+        url: cacheSignedURL,
+      },
+    ]
+  }
+
+  const cacheId = 4
+  const reserveCacheMock = jest
+    .spyOn(cacheHttpClient, 'reserveCache')
+    .mockImplementation(async () => {
+      const response: TypedResponse<ReserveCacheResponse> = {
+        statusCode: 500,
+        result: {cacheId},
+        headers: {}
+      }
+      return response
+    })
+
+  const getCacheBlobUploadURLMock = jest
+  .spyOn(BlobCacheServiceClientJSON.prototype, 'GetCacheBlobUploadURL')
+  .mockResolvedValue(getCacheBlobUploadURL)
+
+  const uploadCacheMock = jest
+  .spyOn(uploadCache, 'UploadCacheFile')
+  .mockImplementation(async () => {
+    return {
+      status: 200    
+    }
+  })
+
+  const createTarMock = jest.spyOn(tar, 'createTar')
+
+  const saveCacheMock = jest.spyOn(cacheHttpClient, 'saveCache')
+  const compression = CompressionMethod.Zstd
+  const getCompressionMock = jest
+    .spyOn(cacheUtils, 'getCompressionMethod')
+    .mockReturnValue(Promise.resolve(compression))
+  
+  await uploadCache.UploadCacheFile(getCacheBlobUploadURL, cachePaths[0])
+  await saveCache([filePath], primaryKey)
+
+  expect(reserveCacheMock).toHaveBeenCalledTimes(1)
+  expect(reserveCacheMock).toHaveBeenCalledWith(primaryKey, [filePath], {
+    cacheSize: undefined,
+    compressionMethod: compression,
+    enableCrossOsArchive: false
+  })
+  expect (getCacheBlobUploadURLMock).toHaveBeenCalledTimes(1)
+  const archiveFolder = '/foo/bar'
+  const archiveFile = path.join(archiveFolder, CacheFilename.Zstd)
+  expect(createTarMock).toHaveBeenCalledTimes(1)
+  expect(createTarMock).toHaveBeenCalledWith(
+    archiveFolder,
+    cachePaths,
+    compression
+  )
+  expect(uploadCacheMock).toHaveBeenCalledTimes(2)
+  expect(saveCacheMock).toHaveBeenCalledTimes(1)
+  expect(saveCacheMock).toHaveBeenCalledWith(cacheId, archiveFile, undefined)
+  expect(getCompressionMock).toHaveBeenCalledTimes(1)
+})

packages/cache/package.json

@@ -38,6 +38,7 @@
   },
   "dependencies": {
     "@actions/core": "^1.10.0",
+    "@actions/artifact": "^2.1.7",
     "@actions/exec": "^1.0.1",
     "@actions/glob": "^0.1.0",
     "@actions/http-client": "^2.1.1",

packages/cache/src/cache.ts

@@ -1,13 +1,18 @@
 import * as core from '@actions/core'
 import * as path from 'path'
 import * as utils from './internal/cacheUtils'
-import {CacheUrl} from './internal/constants'
+import {CacheServiceVersion, CacheUrl} from './internal/constants'
 import * as cacheHttpClient from './internal/cacheHttpClient'
 import * as cacheTwirpClient from './internal/cacheTwirpClient'
 import {createTar, extractTar, listTar} from './internal/tar'
 import {DownloadOptions, UploadOptions} from './options'
 import {GetCacheBlobUploadURLRequest, GetCacheBlobUploadURLResponse} from './generated/results/api/v1/blobcache'
-import {UploadCache} from './internal/v2/upload/upload-cache'
+import {UploadCacheStream} from './internal/v2/upload-cache'
+import {
+  UploadZipSpecification,
+  getUploadZipSpecification
+} from '@actions/artifact/lib/internal/upload/upload-zip-specification'
+import {createZipUploadStream} from '@actions/artifact/lib/internal/upload/zip'
 
 export class ValidationError extends Error {
   constructor(message: string) {
@@ -175,16 +180,22 @@ export async function saveCache(
   checkPaths(paths)
   checkKey(key)
   
-  // TODO: REMOVE ME
-  // Making a call to the service
-  const twirpClient = cacheTwirpClient.internalBlobCacheTwirpClient()
-  const getSignedUploadURL: GetCacheBlobUploadURLRequest = {
-    organization: "github",
-    keys: [key],
+  console.debug(`Cache Service Version: ${CacheServiceVersion}`)
+  switch (CacheServiceVersion) {
+    case "v2":
+      return await saveCachev1(paths, key, options, enableCrossOsArchive)
+    case "v1":
+    default:
+      return await saveCachev2(paths, key, options, enableCrossOsArchive)
   }
-  const signedUploadURL: GetCacheBlobUploadURLResponse = await twirpClient.GetCacheBlobUploadURL(getSignedUploadURL)
-  core.info(`GetCacheBlobUploadURLResponse: ${JSON.stringify(signedUploadURL)}`)
+}
 
+async function saveCachev1(
+  paths: string[],
+  key: string,
+  options?: UploadOptions,
+  enableCrossOsArchive = false
+): Promise<number> {
   const compressionMethod = await utils.getCompressionMethod()
   let cacheId = -1
 
@@ -224,15 +235,6 @@ export async function saveCache(
       )
     }
 
-
-    // Cache v2 upload
-    // inputs:
-    // - getSignedUploadURL
-    // - archivePath
-    core.info(`Saving Cache v2: ${archivePath}`)
-    await UploadCache(signedUploadURL, archivePath)
-
-
     core.debug('Reserving Cache')
     const reserveCacheResponse = await cacheHttpClient.reserveCache(
       key,
@@ -281,3 +283,47 @@ export async function saveCache(
 
   return cacheId
 }
+
+async function saveCachev2(
+  paths: string[],
+  key: string,
+  options?: UploadOptions,
+  enableCrossOsArchive = false
+): Promise<number> {
+  const twirpClient = cacheTwirpClient.internalBlobCacheTwirpClient()
+  const getSignedUploadURL: GetCacheBlobUploadURLRequest = {
+    organization: "github",
+    keys: [key],
+  }
+  const signedUploadURL: GetCacheBlobUploadURLResponse = await twirpClient.GetCacheBlobUploadURL(getSignedUploadURL)
+  core.info(`GetCacheBlobUploadURLResponse: ${JSON.stringify(signedUploadURL)}`)
+
+  // Archive
+  // We're going to handle 1 path fow now. This needs to be fixed to handle all 
+  // paths passed in.
+  const rootDir = path.dirname(paths[0])
+  const zipSpecs: UploadZipSpecification[] = getUploadZipSpecification(paths, rootDir)
+  if (zipSpecs.length === 0) {
+    throw new Error(
+      `Error with zip specs: ${zipSpecs.flatMap(s => (s.sourcePath ? [s.sourcePath] : [])).join(', ')}`
+    )
+  }
+
+  // 0: No compression
+  // 1: Best speed
+  // 6: Default compression (same as GNU Gzip)
+  // 9: Best compression Higher levels will result in better compression, but will take longer to complete. For large files that are not easily compressed, a value of 0 is recommended for significantly faster uploads.
+  const zipUploadStream = await createZipUploadStream(
+    zipSpecs,
+    6
+  )
+
+  // Cache v2 upload
+  // inputs:
+  // - getSignedUploadURL
+  // - archivePath
+  core.info(`Saving Cache v2: ${paths[0]}`)
+  await UploadCacheStream(signedUploadURL.urls[0].url, zipUploadStream)
+
+  return 0
+}

packages/cache/src/internal/constants.ts

@@ -37,5 +37,6 @@ export const TarFilename = 'cache.tar'
 
 export const ManifestFilename = 'manifest.txt'
 
-// Cache URL
-export const CacheUrl = `${process.env['ACTIONS_CACHE_URL_NEXT']}`
+// Cache Service Metadata
+export const CacheUrl = `${process.env['ACTIONS_CACHE_URL_NEXT']} || ${process.env['ACTIONS_CACHE_URL']}`
+export const CacheServiceVersion = `${process.env['ACTIONS_CACHE_URL_NEXT']} ? 'v2' : 'v1'`

packages/cache/src/internal/v2/upload-cache.ts

@@ -0,0 +1,130 @@
+import * as core from '@actions/core'
+import {GetCacheBlobUploadURLResponse} from '../../generated/results/api/v1/blobcache'
+import {ZipUploadStream} from '@actions/artifact/lib/internal/upload/zip'
+import {NetworkError} from '@actions/artifact/'
+import {TransferProgressEvent} from '@azure/core-http'
+import * as stream from 'stream'
+import * as crypto from 'crypto'
+import {
+  BlobClient, 
+  BlockBlobClient, 
+  BlockBlobUploadStreamOptions,
+  BlockBlobParallelUploadOptions
+} from '@azure/storage-blob'
+
+export async function UploadCacheStream(
+  signedUploadURL: string,
+  zipUploadStream: ZipUploadStream
+): Promise<{}> {
+  let uploadByteCount = 0
+  let lastProgressTime = Date.now()
+  let timeoutId: NodeJS.Timeout | undefined
+
+  const chunkTimer = (timeout: number): NodeJS.Timeout => {
+    // clear the previous timeout
+    if (timeoutId) {
+      clearTimeout(timeoutId)
+    }
+
+    timeoutId = setTimeout(() => {
+      const now = Date.now()
+      // if there's been more than 30 seconds since the
+      // last progress event, then we'll consider the upload stalled
+      if (now - lastProgressTime > timeout) {
+        throw new Error('Upload progress stalled.')
+      }
+    }, timeout)
+    return timeoutId
+  }
+
+  const maxConcurrency = 32
+  const bufferSize = 8 * 1024 * 1024 // 8 MB Chunks
+  const blobClient = new BlobClient(signedUploadURL)
+  const blockBlobClient = blobClient.getBlockBlobClient()
+  const timeoutDuration = 300000 // 30 seconds
+
+  core.debug(
+    `Uploading cache zip to blob storage with maxConcurrency: ${maxConcurrency}, bufferSize: ${bufferSize}`
+  )
+
+  const uploadCallback = (progress: TransferProgressEvent): void => {
+    core.info(`Uploaded bytes ${progress.loadedBytes}`)
+    uploadByteCount = progress.loadedBytes
+    chunkTimer(timeoutDuration)
+    lastProgressTime = Date.now()
+  }
+
+  const options: BlockBlobUploadStreamOptions = {
+    blobHTTPHeaders: {blobContentType: 'zip'},
+    onProgress: uploadCallback
+  }
+
+  let sha256Hash: string | undefined = undefined
+  const uploadStream = new stream.PassThrough()
+  const hashStream = crypto.createHash('sha256')
+
+  zipUploadStream.pipe(uploadStream) // This stream is used for the upload
+  zipUploadStream.pipe(hashStream).setEncoding('hex') // This stream is used to compute a hash of the zip content that gets used. Integrity check
+
+  core.info('Beginning upload of cache to blob storage')
+  try {
+    // Start the chunk timer
+    timeoutId = chunkTimer(timeoutDuration)
+    await blockBlobClient.uploadStream(
+      uploadStream,
+      bufferSize,
+      maxConcurrency,
+      options
+    )
+  } catch (error) {
+    if (NetworkError.isNetworkErrorCode(error?.code)) {
+      throw new NetworkError(error?.code)
+    }
+    throw error
+  } finally {
+    // clear the timeout whether or not the upload completes
+    if (timeoutId) {
+      clearTimeout(timeoutId)
+    }
+  }
+
+  core.info('Finished uploading cache content to blob storage!')
+  
+  hashStream.end()
+  sha256Hash = hashStream.read() as string
+  core.info(`SHA256 hash of uploaded artifact zip is ${sha256Hash}`)
+  core.info(`Uploaded: ${uploadByteCount} bytes`)
+
+  if (uploadByteCount === 0) {
+    core.error(
+      `No data was uploaded to blob storage. Reported upload byte count is 0.`
+    )
+  }
+  return {
+    uploadSize: uploadByteCount,
+    sha256Hash
+  }
+}
+
+export async function UploadCacheFile(
+  uploadURL: GetCacheBlobUploadURLResponse,
+  archivePath: string,
+): Promise<{}> {
+  core.info(`Uploading ${archivePath} to: ${JSON.stringify(uploadURL)}`)
+  
+  // Specify data transfer options
+  const uploadOptions: BlockBlobParallelUploadOptions = {
+    blockSize: 4 * 1024 * 1024, // 4 MiB max block size
+    concurrency: 2, // maximum number of parallel transfer workers
+    maxSingleShotSize: 8 * 1024 * 1024, // 8 MiB initial transfer size
+  };
+
+  // const blobClient: BlobClient = new BlobClient(uploadURL.urls[0])
+  const blobClient: BlobClient = new BlobClient(uploadURL.urls[0].url)
+  const blockBlobClient: BlockBlobClient = blobClient.getBlockBlobClient()
+
+  core.info(`BlobClient: ${JSON.stringify(blobClient)}`)
+  core.info(`blockBlobClient: ${JSON.stringify(blockBlobClient)}`)
+
+  return blockBlobClient.uploadFile(archivePath, uploadOptions);
+}

packages/cache/src/internal/v2/upload/upload-cache.ts

@@ -1,26 +0,0 @@
-import * as core from '@actions/core'
-import {GetCacheBlobUploadURLResponse} from '../../../generated/results/api/v1/blobcache'
-import {BlobClient, BlockBlobClient, BlockBlobParallelUploadOptions} from '@azure/storage-blob'
-
-export async function UploadCache(
-  uploadURL: GetCacheBlobUploadURLResponse,
-  archivePath: string,
-): Promise<{}> {
-  core.info(`Uploading ${archivePath} to: ${JSON.stringify(uploadURL)}`)
-  
-  // Specify data transfer options
-  const uploadOptions: BlockBlobParallelUploadOptions = {
-    blockSize: 4 * 1024 * 1024, // 4 MiB max block size
-    concurrency: 2, // maximum number of parallel transfer workers
-    maxSingleShotSize: 8 * 1024 * 1024, // 8 MiB initial transfer size
-  };
-
-  // const blobClient: BlobClient = new BlobClient(uploadURL.urls[0])
-  const blobClient: BlobClient = new BlobClient(uploadURL.urls[0].url)
-  const blockBlobClient: BlockBlobClient = blobClient.getBlockBlobClient()
-
-  core.info(`BlobClient: ${JSON.stringify(blobClient)}`)
-  core.info(`blockBlobClient: ${JSON.stringify(blockBlobClient)}`)
-
-  return blockBlobClient.uploadFile(archivePath, uploadOptions);
-}