Don't autoclose file

__tests__/restore.test.ts

@@ -255,7 +255,10 @@ test("restore with cache found", async () => {
     expect(getCacheMock).toHaveBeenCalledWith([key]);
     expect(setCacheStateMock).toHaveBeenCalledWith(cacheEntry);
     expect(createTempDirectoryMock).toHaveBeenCalledTimes(1);
-    expect(downloadCacheMock).toHaveBeenCalledWith(cacheEntry, archivePath);
+    expect(downloadCacheMock).toHaveBeenCalledWith(
+        cacheEntry.archiveLocation,
+        archivePath
+    );
     expect(getArchiveFileSizeMock).toHaveBeenCalledWith(archivePath);
     expect(mkdirMock).toHaveBeenCalledWith(cachePath);
 
@@ -333,7 +336,10 @@ test("restore with a pull request event and cache found", async () => {
     expect(getCacheMock).toHaveBeenCalledWith([key]);
     expect(setCacheStateMock).toHaveBeenCalledWith(cacheEntry);
     expect(createTempDirectoryMock).toHaveBeenCalledTimes(1);
-    expect(downloadCacheMock).toHaveBeenCalledWith(cacheEntry, archivePath);
+    expect(downloadCacheMock).toHaveBeenCalledWith(
+        cacheEntry.archiveLocation,
+        archivePath
+    );
     expect(getArchiveFileSizeMock).toHaveBeenCalledWith(archivePath);
     expect(infoMock).toHaveBeenCalledWith(`Cache Size: ~60 MB (62915000 B)`);
     expect(mkdirMock).toHaveBeenCalledWith(cachePath);
@@ -412,7 +418,10 @@ test("restore with cache found for restore key", async () => {
     expect(getCacheMock).toHaveBeenCalledWith([key, restoreKey]);
     expect(setCacheStateMock).toHaveBeenCalledWith(cacheEntry);
     expect(createTempDirectoryMock).toHaveBeenCalledTimes(1);
-    expect(downloadCacheMock).toHaveBeenCalledWith(cacheEntry, archivePath);
+    expect(downloadCacheMock).toHaveBeenCalledWith(
+        cacheEntry.archiveLocation,
+        archivePath
+    );
     expect(getArchiveFileSizeMock).toHaveBeenCalledWith(archivePath);
     expect(infoMock).toHaveBeenCalledWith(`Cache Size: ~0 MB (142 B)`);
     expect(mkdirMock).toHaveBeenCalledWith(cachePath);

__tests__/save.test.ts

@@ -200,7 +200,7 @@ test("save with large cache outputs warning", async () => {
 
     const execMock = jest.spyOn(exec, "exec");
 
-    const cacheSize = 1024 * 1024 * 1024; //~1GB, over the 400MB limit
+    const cacheSize = 4 * 1024 * 1024 * 1024; //~4GB, over the 2GB limit
     jest.spyOn(actionUtils, "getArchiveFileSize").mockImplementationOnce(() => {
         return cacheSize;
     });
@@ -227,7 +227,7 @@ test("save with large cache outputs warning", async () => {
 
     expect(logWarningMock).toHaveBeenCalledTimes(1);
     expect(logWarningMock).toHaveBeenCalledWith(
-        "Cache size of ~1024 MB (1073741824 B) is over the 400MB limit, not saving cache."
+        "Cache size of ~4 GB (4294967296 B) is over the 2GB limit, not saving cache."
     );
 
     expect(failedMock).toHaveBeenCalledTimes(0);

package-lock.json

@@ -4859,9 +4859,9 @@
       "dev": true
     },
     "prettier": {
-      "version": "1.18.2",
+      "version": "1.19.1",
-      "resolved": "https://registry.npmjs.org/prettier/-/prettier-1.18.2.tgz",
+      "resolved": "https://registry.npmjs.org/prettier/-/prettier-1.19.1.tgz",
-      "integrity": "sha512-OeHeMc0JhFE9idD4ZdtNibzY0+TPHSpSSb9h8FqtP+YnoZZ1sl8Vc9b1sasjfymH3SonAF4QcA2+mzHPhMvIiw==",
+      "integrity": "sha512-s7PoyDv/II1ObgQunCbB9PdLmUcBZcnWOcxDh7O0N/UwDEsHyqkW+Qh28jW+mVuCdx7gLB0BotYI1Y6uI9iyew==",
       "dev": true
     },
     "prettier-linter-helpers": {
@@ -5983,9 +5983,9 @@
       }
     },
     "typescript": {
-      "version": "3.6.4",
+      "version": "3.7.3",
-      "resolved": "https://registry.npmjs.org/typescript/-/typescript-3.6.4.tgz",
+      "resolved": "https://registry.npmjs.org/typescript/-/typescript-3.7.3.tgz",
-      "integrity": "sha512-unoCll1+l+YK4i4F8f22TaNVPRHcD9PA3yCuZ8g5e0qGqlVlJ/8FSateOLLSagn+Yg5+ZwuPkL8LFUc0Jcvksg==",
+      "integrity": "sha512-Mcr/Qk7hXqFBXMN7p7Lusj1ktCBydylfQM/FZCk5glCNQJrCUKPkMHdo9R0MTFWsC/4kPFvDS0fDPvukfCkFsw==",
       "dev": true
     },
     "uglify-js": {

package.json

@@ -1,6 +1,6 @@
 {
   "name": "cache",
-  "version": "1.0.3",
+  "version": "1.1.0",
   "private": true,
   "description": "Cache dependencies and build outputs",
   "main": "dist/restore/index.js",
@@ -46,8 +46,8 @@
     "jest": "^24.8.0",
     "jest-circus": "^24.7.1",
     "nock": "^11.7.0",
-    "prettier": "1.18.2",
+    "prettier": "^1.19.1",
     "ts-jest": "^24.0.2",
-    "typescript": "^3.6.4"
+    "typescript": "^3.7.3"
   }
 }

src/cacheHttpClient.ts

@@ -3,24 +3,39 @@ import * as fs from "fs";
 import { BearerCredentialHandler } from "typed-rest-client/Handlers";
 import { HttpClient } from "typed-rest-client/HttpClient";
 import { IHttpClientResponse } from "typed-rest-client/Interfaces";
-import { IRequestOptions, RestClient } from "typed-rest-client/RestClient";
+import {
-import { ArtifactCacheEntry } from "./contracts";
+    IRequestOptions,
+    RestClient,
+    IRestResponse
+} from "typed-rest-client/RestClient";
+import {
+    ArtifactCacheEntry,
+    CommitCacheRequest,
+    ReserveCacheRequest,
+    ReserverCacheResponse
+} from "./contracts";
+import * as utils from "./utils/actionUtils";
 
-function getCacheUrl(): string {
+const MAX_CHUNK_SIZE = 4000000; // 4 MB Chunks
+
+function isSuccessStatusCode(statusCode: number): boolean {
+    return statusCode >= 200 && statusCode < 300;
+}
+function getCacheApiUrl(): string {
     // Ideally we just use ACTIONS_CACHE_URL
-    const cacheUrl: string = (
+    const baseUrl: string = (
         process.env["ACTIONS_CACHE_URL"] ||
         process.env["ACTIONS_RUNTIME_URL"] ||
         ""
     ).replace("pipelines", "artifactcache");
-    if (!cacheUrl) {
+    if (!baseUrl) {
         throw new Error(
             "Cache Service Url not found, unable to restore cache."
         );
     }
 
-    core.debug(`Cache Url: ${cacheUrl}`);
+    core.debug(`Cache Url: ${baseUrl}`);
-    return cacheUrl;
+    return `${baseUrl}_apis/artifactcache/`;
 }
 
 function createAcceptHeader(type: string, apiVersion: string): string {
@@ -29,26 +44,26 @@ function createAcceptHeader(type: string, apiVersion: string): string {
 
 function getRequestOptions(): IRequestOptions {
     const requestOptions: IRequestOptions = {
-        acceptHeader: createAcceptHeader("application/json", "5.2-preview.1")
+        acceptHeader: createAcceptHeader("application/json", "6.0-preview.1")
     };
 
     return requestOptions;
 }
 
-export async function getCacheEntry(
+function createRestClient(): RestClient {
-    keys: string[]
-): Promise<ArtifactCacheEntry | null> {
-    const cacheUrl = getCacheUrl();
     const token = process.env["ACTIONS_RUNTIME_TOKEN"] || "";
     const bearerCredentialHandler = new BearerCredentialHandler(token);
 
-    const resource = `_apis/artifactcache/cache?keys=${encodeURIComponent(
+    return new RestClient("actions/cache", getCacheApiUrl(), [
-        keys.join(",")
-    )}`;
-
-    const restClient = new RestClient("actions/cache", cacheUrl, [
         bearerCredentialHandler
     ]);
+}
+
+export async function getCacheEntry(
+    keys: string[]
+): Promise<ArtifactCacheEntry | null> {
+    const restClient = createRestClient();
+    const resource = `cache?keys=${encodeURIComponent(keys.join(","))}`;
 
     const response = await restClient.get<ArtifactCacheEntry>(
         resource,
@@ -57,14 +72,15 @@ export async function getCacheEntry(
     if (response.statusCode === 204) {
         return null;
     }
-    if (response.statusCode !== 200) {
+    if (!isSuccessStatusCode(response.statusCode)) {
         throw new Error(`Cache service responded with ${response.statusCode}`);
     }
     const cacheResult = response.result;
-    if (!cacheResult || !cacheResult.archiveLocation) {
+    const cacheDownloadUrl = cacheResult?.archiveLocation;
+    if (!cacheDownloadUrl) {
         throw new Error("Cache not found.");
     }
-    core.setSecret(cacheResult.archiveLocation);
+    core.setSecret(cacheDownloadUrl);
     core.debug(`Cache Result:`);
     core.debug(JSON.stringify(cacheResult));
 
@@ -83,46 +99,129 @@ async function pipeResponseToStream(
 }
 
 export async function downloadCache(
-    cacheEntry: ArtifactCacheEntry,
+    archiveLocation: string,
     archivePath: string
 ): Promise<void> {
     const stream = fs.createWriteStream(archivePath);
     const httpClient = new HttpClient("actions/cache");
-    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    const downloadResponse = await httpClient.get(archiveLocation);
-    const downloadResponse = await httpClient.get(cacheEntry.archiveLocation!);
     await pipeResponseToStream(downloadResponse, stream);
 }
 
-export async function saveCache(
+// Reserve Cache
-    key: string,
+export async function reserveCache(
-    archivePath: string
+    key: string
-): Promise<void> {
+): Promise<number> {
-    const stream = fs.createReadStream(archivePath);
+    const restClient = createRestClient();
 
-    const cacheUrl = getCacheUrl();
+    const reserveCacheRequest: ReserveCacheRequest = {
-    const token = process.env["ACTIONS_RUNTIME_TOKEN"] || "";
+        key
-    const bearerCredentialHandler = new BearerCredentialHandler(token);
+    };
+    const response = await restClient.create<ReserverCacheResponse>(
+        "caches",
+        reserveCacheRequest,
+        getRequestOptions()
+    );
 
-    const resource = `_apis/artifactcache/cache/${encodeURIComponent(key)}`;
+    return response?.result?.cacheId ?? -1;
-    const postUrl = cacheUrl + resource;
+}
 
-    const restClient = new RestClient("actions/cache", undefined, [
+function getContentRange(start: number, end: number): string {
-        bearerCredentialHandler
+    // Format: `bytes start-end/filesize
-    ]);
+    // start and end are inclusive
+    // filesize can be *
+    // For a 200 byte chunk starting at byte 0:
+    // Content-Range: bytes 0-199/*
+    return `bytes ${start}-${end}/*`;
+}
 
+// function bufferToStream(buffer: Buffer): NodeJS.ReadableStream {
+//     const stream = new Duplex();
+//     stream.push(buffer);
+//     stream.push(null);
+
+//     return stream;
+// }
+
+async function uploadChunk(
+    restClient: RestClient,
+    resourceUrl: string,
+    data: NodeJS.ReadableStream,
+    start: number,
+    end: number
+): Promise<IRestResponse<void>> {
+    core.debug(`Uploading chunk of size ${end - start + 1} bytes at offset ${start} with content range: ${getContentRange(start, end)}`);
     const requestOptions = getRequestOptions();
     requestOptions.additionalHeaders = {
-        "Content-Type": "application/octet-stream"
+        "Content-Type": "application/octet-stream",
+        "Content-Range": getContentRange(start, end)
     };
 
-    const response = await restClient.uploadStream<void>(
+    return await restClient.uploadStream<void>("PATCH", resourceUrl, data, requestOptions);
-        "POST",
+}
-        postUrl,
+
-        stream,
+async function commitCache(
+    restClient: RestClient,
+    cacheId: number,
+    filesize: number
+): Promise<IRestResponse<void>> {
+    const requestOptions = getRequestOptions();
+    const commitCacheRequest: CommitCacheRequest = { size: filesize };
+    return await restClient.create(
+        `caches/${cacheId.toString()}`,
+        commitCacheRequest,
         requestOptions
     );
-    if (response.statusCode !== 200) {
+}
-        throw new Error(`Cache service responded with ${response.statusCode}`);
+
+export async function saveCache(
+    cacheId: number,
+    archivePath: string
+): Promise<void> {
+    const restClient = createRestClient();
+
+    core.debug("Uploading chunks");
+    // Upload Chunks
+    const fileSize = fs.statSync(archivePath).size;
+    const resourceUrl = getCacheApiUrl() + "caches/" + cacheId.toString();
+    const uploads: IRestResponse<void>[] = [];
+
+    const fd = fs.openSync(archivePath, "r"); // Use the same fd for serial reads? Will this work for parallel too?
+    let offset = 0;
+    while (offset < fileSize) {
+        const chunkSize = offset + MAX_CHUNK_SIZE > fileSize ? fileSize - offset : MAX_CHUNK_SIZE;
+        const end = offset + chunkSize - 1;
+        const chunk = fs.createReadStream(archivePath, { fd, start: offset, end, autoClose: false });
+        uploads.push(await uploadChunk(restClient, resourceUrl, chunk, offset, end)); // Making this serial
+        offset += MAX_CHUNK_SIZE;
+    }
+
+    fs.closeSync(fd);
+
+    core.debug("Awaiting all uploads");
+    //const responses = await Promise.all(uploads);
+
+    const failedResponse = uploads.find(
+        x => !isSuccessStatusCode(x.statusCode)
+    );
+    if (failedResponse) {
+        throw new Error(
+            `Cache service responded with ${failedResponse.statusCode} during chunk upload.`
+        );
+    }
+
+    core.debug("Commiting cache");
+    // Commit Cache
+    const cacheSize = utils.getArchiveFileSize(archivePath);
+    const commitCacheResponse = await commitCache(
+        restClient,
+        cacheId,
+        cacheSize
+    );
+    if (!isSuccessStatusCode(commitCacheResponse.statusCode)) {
+        throw new Error(
+            `Cache service responded with ${commitCacheResponse.statusCode} during commit cache.`
+        );
     }
 
     core.info("Cache saved successfully");

src/contracts.d.ts

@@ -4,3 +4,16 @@ export interface ArtifactCacheEntry {
     creationTime?: string;
     archiveLocation?: string;
 }
+
+export interface CommitCacheRequest {
+    size: number;
+}
+
+export interface ReserveCacheRequest {
+    key: string;
+    version?: string;
+}
+
+export interface ReserverCacheResponse {
+    cacheId: number;
+}

src/restore.ts

@@ -61,7 +61,7 @@ async function run(): Promise<void> {
 
         try {
             const cacheEntry = await cacheHttpClient.getCacheEntry(keys);
-            if (!cacheEntry) {
+            if (!cacheEntry || !cacheEntry?.archiveLocation) {
                 core.info(
                     `Cache not found for input keys: ${keys.join(", ")}.`
                 );
@@ -78,7 +78,11 @@ async function run(): Promise<void> {
             utils.setCacheState(cacheEntry);
 
             // Download the cache from the cache entry
-            await cacheHttpClient.downloadCache(cacheEntry, archivePath);
+            await cacheHttpClient.downloadCache(
+                cacheEntry?.archiveLocation,
+                archivePath
+            );
+            await exec(`md5sum`, [archivePath]);
 
             const archiveFileSize = utils.getArchiveFileSize(archivePath);
             core.info(

src/save.ts

@@ -35,6 +35,13 @@ async function run(): Promise<void> {
             return;
         }
 
+        core.debug("Reserving Cache");
+        const cacheId = await cacheHttpClient.reserveCache(primaryKey);
+        if (cacheId < 0) {
+            core.info(`Unable to reserve cache with key ${primaryKey}, another job may be creating this cache.`);
+            return;
+        }
+        core.debug(`Cache ID: ${cacheId}`);
         const cachePath = utils.resolvePath(
             core.getInput(Inputs.Path, { required: true })
         );
@@ -65,19 +72,21 @@ async function run(): Promise<void> {
         core.debug(`Tar Path: ${tarPath}`);
         await exec(`"${tarPath}"`, args);
 
-        const fileSizeLimit = 400 * 1024 * 1024; // 400MB
+        const fileSizeLimit = 2 * 1024 * 1024 * 1024; // 2GB per repo limit
         const archiveFileSize = utils.getArchiveFileSize(archivePath);
         core.debug(`File Size: ${archiveFileSize}`);
         if (archiveFileSize > fileSizeLimit) {
             utils.logWarning(
                 `Cache size of ~${Math.round(
-                    archiveFileSize / (1024 * 1024)
+                    archiveFileSize / (1024 * 1024 * 1024)
-                )} MB (${archiveFileSize} B) is over the 400MB limit, not saving cache.`
+                )} GB (${archiveFileSize} B) is over the 2GB limit, not saving cache.`
             );
             return;
         }
 
-        await cacheHttpClient.saveCache(primaryKey, archivePath);
+        await exec(`md5sum`, [archivePath]);
+        core.debug("Saving Cache");
+        await cacheHttpClient.saveCache(cacheId, archivePath);
     } catch (error) {
         utils.logWarning(error.message);
     }