Format and update tests

__tests__/save.test.ts

@@ -212,14 +212,14 @@ test("save with large cache outputs warning", async () => {
     const IS_WINDOWS = process.platform === "win32";
     const args = IS_WINDOWS
         ? [
-              "-cz",
+            "-cz",
-              "--force-local",
+            "--force-local",
-              "-f",
+            "-f",
-              archivePath.replace(/\\/g, "/"),
+            archivePath.replace(/\\/g, "/"),
-              "-C",
+            "-C",
-              cachePath.replace(/\\/g, "/"),
+            cachePath.replace(/\\/g, "/"),
-              "."
+            "."
-          ]
+        ]
         : ["-cz", "-f", archivePath, "-C", cachePath, "."];
 
     expect(execMock).toHaveBeenCalledTimes(1);
@@ -259,6 +259,11 @@ test("save with server error outputs warning", async () => {
     const cachePath = path.resolve(inputPath);
     testUtils.setInput(Inputs.Path, inputPath);
 
+    const cacheId = 4;
+    const reserveCacheMock = jest.spyOn(cacheHttpClient, "reserveCache").mockImplementationOnce(() => {
+        return Promise.resolve(cacheId);
+    });
+
     const execMock = jest.spyOn(exec, "exec");
 
     const saveCacheMock = jest
@@ -269,26 +274,29 @@ test("save with server error outputs warning", async () => {
 
     await run();
 
+    expect(reserveCacheMock).toHaveBeenCalledTimes(1);
+    expect(reserveCacheMock).toHaveBeenCalledWith(primaryKey);
+
     const archivePath = path.join("/foo/bar", "cache.tgz");
 
     const IS_WINDOWS = process.platform === "win32";
     const args = IS_WINDOWS
         ? [
-              "-cz",
+            "-cz",
-              "--force-local",
+            "--force-local",
-              "-f",
+            "-f",
-              archivePath.replace(/\\/g, "/"),
+            archivePath.replace(/\\/g, "/"),
-              "-C",
+            "-C",
-              cachePath.replace(/\\/g, "/"),
+            cachePath.replace(/\\/g, "/"),
-              "."
+            "."
-          ]
+        ]
         : ["-cz", "-f", archivePath, "-C", cachePath, "."];
 
     expect(execMock).toHaveBeenCalledTimes(1);
     expect(execMock).toHaveBeenCalledWith(`"tar"`, args);
 
     expect(saveCacheMock).toHaveBeenCalledTimes(1);
-    expect(saveCacheMock).toHaveBeenCalledWith(primaryKey, archivePath);
+    expect(saveCacheMock).toHaveBeenCalledWith(cacheId, archivePath);
 
     expect(logWarningMock).toHaveBeenCalledTimes(1);
     expect(logWarningMock).toHaveBeenCalledWith("HTTP Error Occurred");
@@ -321,32 +329,40 @@ test("save with valid inputs uploads a cache", async () => {
     const cachePath = path.resolve(inputPath);
     testUtils.setInput(Inputs.Path, inputPath);
 
+    const cacheId = 4;
+    const reserveCacheMock = jest.spyOn(cacheHttpClient, "reserveCache").mockImplementationOnce(() => {
+        return Promise.resolve(cacheId);
+    });
+
     const execMock = jest.spyOn(exec, "exec");
 
     const saveCacheMock = jest.spyOn(cacheHttpClient, "saveCache");
 
     await run();
 
+    expect(reserveCacheMock).toHaveBeenCalledTimes(1);
+    expect(reserveCacheMock).toHaveBeenCalledWith(primaryKey);
+
     const archivePath = path.join("/foo/bar", "cache.tgz");
 
     const IS_WINDOWS = process.platform === "win32";
     const args = IS_WINDOWS
         ? [
-              "-cz",
+            "-cz",
-              "--force-local",
+            "--force-local",
-              "-f",
+            "-f",
-              archivePath.replace(/\\/g, "/"),
+            archivePath.replace(/\\/g, "/"),
-              "-C",
+            "-C",
-              cachePath.replace(/\\/g, "/"),
+            cachePath.replace(/\\/g, "/"),
-              "."
+            "."
-          ]
+        ]
         : ["-cz", "-f", archivePath, "-C", cachePath, "."];
 
     expect(execMock).toHaveBeenCalledTimes(1);
     expect(execMock).toHaveBeenCalledWith(`"tar"`, args);
 
     expect(saveCacheMock).toHaveBeenCalledTimes(1);
-    expect(saveCacheMock).toHaveBeenCalledWith(primaryKey, archivePath);
+    expect(saveCacheMock).toHaveBeenCalledWith(cacheId, archivePath);
 
     expect(failedMock).toHaveBeenCalledTimes(0);
 });

dist/restore/index.js

@@ -1497,7 +1497,6 @@ const Handlers_1 = __webpack_require__(941);
 const HttpClient_1 = __webpack_require__(874);
 const RestClient_1 = __webpack_require__(105);
 const utils = __importStar(__webpack_require__(443));
-const MAX_CHUNK_SIZE = 4000000; // 4 MB Chunks
 function isSuccessStatusCode(statusCode) {
     return statusCode >= 200 && statusCode < 300;
 }
@@ -1615,30 +1614,42 @@ function commitCache(restClient, cacheId, filesize) {
         return yield restClient.create(`caches/${cacheId.toString()}`, commitCacheRequest, requestOptions);
     });
 }
-function saveCache(cacheId, archivePath) {
+function uploadFile(restClient, cacheId, archivePath) {
     return __awaiter(this, void 0, void 0, function* () {
-        const restClient = createRestClient();
-        core.debug("Uploading chunks");
         // Upload Chunks
         const fileSize = fs.statSync(archivePath).size;
         const resourceUrl = getCacheApiUrl() + "caches/" + cacheId.toString();
-        const uploads = [];
+        const responses = [];
-        const fd = fs.openSync(archivePath, "r"); // Use the same fd for serial reads? Will this work for parallel too?
+        const fd = fs.openSync(archivePath, "r");
-        let offset = 0;
+        const concurrency = 16; // # of HTTP requests in parallel
-        while (offset < fileSize) {
+        const MAX_CHUNK_SIZE = 32000000; // 32 MB Chunks
-            const chunkSize = offset + MAX_CHUNK_SIZE > fileSize ? fileSize - offset : MAX_CHUNK_SIZE;
+        core.debug(`Concurrency: ${concurrency} and Chunk Size: ${MAX_CHUNK_SIZE}`);
-            const end = offset + chunkSize - 1;
+        const parallelUploads = [...new Array(concurrency).keys()];
-            const chunk = fs.createReadStream(archivePath, { fd, start: offset, end, autoClose: false });
-            uploads.push(yield 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);
+        let offset = 0;
-        const failedResponse = uploads.find(x => !isSuccessStatusCode(x.statusCode));
+        yield Promise.all(parallelUploads.map(() => __awaiter(this, void 0, void 0, function* () {
+            while (offset < fileSize) {
+                const chunkSize = offset + MAX_CHUNK_SIZE > fileSize ? fileSize - offset : MAX_CHUNK_SIZE;
+                const start = offset;
+                const end = offset + chunkSize - 1;
+                offset += MAX_CHUNK_SIZE;
+                const chunk = fs.createReadStream(archivePath, { fd, start, end, autoClose: false });
+                responses.push(yield uploadChunk(restClient, resourceUrl, chunk, start, end));
+            }
+        })));
+        fs.closeSync(fd);
+        const failedResponse = responses.find(x => !isSuccessStatusCode(x.statusCode));
         if (failedResponse) {
             throw new Error(`Cache service responded with ${failedResponse.statusCode} during chunk upload.`);
         }
+        return;
+    });
+}
+function saveCache(cacheId, archivePath) {
+    return __awaiter(this, void 0, void 0, function* () {
+        const restClient = createRestClient();
+        core.debug("Upload cache");
+        yield uploadFile(restClient, cacheId, archivePath);
         core.debug("Commiting cache");
         // Commit Cache
         const cacheSize = utils.getArchiveFileSize(archivePath);

dist/save/index.js

@@ -1497,7 +1497,6 @@ const Handlers_1 = __webpack_require__(941);
 const HttpClient_1 = __webpack_require__(874);
 const RestClient_1 = __webpack_require__(105);
 const utils = __importStar(__webpack_require__(443));
-const MAX_CHUNK_SIZE = 4000000; // 4 MB Chunks
 function isSuccessStatusCode(statusCode) {
     return statusCode >= 200 && statusCode < 300;
 }
@@ -1615,30 +1614,42 @@ function commitCache(restClient, cacheId, filesize) {
         return yield restClient.create(`caches/${cacheId.toString()}`, commitCacheRequest, requestOptions);
     });
 }
-function saveCache(cacheId, archivePath) {
+function uploadFile(restClient, cacheId, archivePath) {
     return __awaiter(this, void 0, void 0, function* () {
-        const restClient = createRestClient();
-        core.debug("Uploading chunks");
         // Upload Chunks
         const fileSize = fs.statSync(archivePath).size;
         const resourceUrl = getCacheApiUrl() + "caches/" + cacheId.toString();
-        const uploads = [];
+        const responses = [];
-        const fd = fs.openSync(archivePath, "r"); // Use the same fd for serial reads? Will this work for parallel too?
+        const fd = fs.openSync(archivePath, "r");
-        let offset = 0;
+        const concurrency = 16; // # of HTTP requests in parallel
-        while (offset < fileSize) {
+        const MAX_CHUNK_SIZE = 32000000; // 32 MB Chunks
-            const chunkSize = offset + MAX_CHUNK_SIZE > fileSize ? fileSize - offset : MAX_CHUNK_SIZE;
+        core.debug(`Concurrency: ${concurrency} and Chunk Size: ${MAX_CHUNK_SIZE}`);
-            const end = offset + chunkSize - 1;
+        const parallelUploads = [...new Array(concurrency).keys()];
-            const chunk = fs.createReadStream(archivePath, { fd, start: offset, end, autoClose: false });
-            uploads.push(yield 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);
+        let offset = 0;
-        const failedResponse = uploads.find(x => !isSuccessStatusCode(x.statusCode));
+        yield Promise.all(parallelUploads.map(() => __awaiter(this, void 0, void 0, function* () {
+            while (offset < fileSize) {
+                const chunkSize = offset + MAX_CHUNK_SIZE > fileSize ? fileSize - offset : MAX_CHUNK_SIZE;
+                const start = offset;
+                const end = offset + chunkSize - 1;
+                offset += MAX_CHUNK_SIZE;
+                const chunk = fs.createReadStream(archivePath, { fd, start, end, autoClose: false });
+                responses.push(yield uploadChunk(restClient, resourceUrl, chunk, start, end));
+            }
+        })));
+        fs.closeSync(fd);
+        const failedResponse = responses.find(x => !isSuccessStatusCode(x.statusCode));
         if (failedResponse) {
             throw new Error(`Cache service responded with ${failedResponse.statusCode} during chunk upload.`);
         }
+        return;
+    });
+}
+function saveCache(cacheId, archivePath) {
+    return __awaiter(this, void 0, void 0, function* () {
+        const restClient = createRestClient();
+        core.debug("Upload cache");
+        yield uploadFile(restClient, cacheId, archivePath);
         core.debug("Commiting cache");
         // Commit Cache
         const cacheSize = utils.getArchiveFileSize(archivePath);

src/cacheHttpClient.ts

@@ -16,8 +16,6 @@ import {
 } from "./contracts";
 import * as utils from "./utils/actionUtils";
 
-const MAX_CHUNK_SIZE = 4000000; // 4 MB Chunks
-
 function isSuccessStatusCode(statusCode: number): boolean {
     return statusCode >= 200 && statusCode < 300;
 }
@@ -109,9 +107,7 @@ export async function downloadCache(
 }
 
 // Reserve Cache
-export async function reserveCache(
+export async function reserveCache(key: string): Promise<number> {
-    key: string
-): Promise<number> {
     const restClient = createRestClient();
 
     const reserveCacheRequest: ReserveCacheRequest = {
@@ -135,14 +131,6 @@ function getContentRange(start: number, end: number): string {
     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,
@@ -150,14 +138,87 @@ async function uploadChunk(
     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)}`);
+    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-Range": getContentRange(start, end)
     };
 
-    return await restClient.uploadStream<void>("PATCH", resourceUrl, data, requestOptions);
+    return await restClient.uploadStream<void>(
+        "PATCH",
+        resourceUrl,
+        data,
+        requestOptions
+    );
+}
+
+async function uploadFile(
+    restClient: RestClient,
+    cacheId: number,
+    archivePath: string
+): Promise<void> {
+    // Upload Chunks
+    const fileSize = fs.statSync(archivePath).size;
+    const resourceUrl = getCacheApiUrl() + "caches/" + cacheId.toString();
+    const responses: IRestResponse<void>[] = [];
+    const fd = fs.openSync(archivePath, "r");
+
+    const concurrency = 4; // # of HTTP requests in parallel
+    const MAX_CHUNK_SIZE = 32000000; // 32 MB Chunks
+    core.debug(`Concurrency: ${concurrency} and Chunk Size: ${MAX_CHUNK_SIZE}`);
+
+    const parallelUploads = [...new Array(concurrency).keys()];
+    core.debug("Awaiting all uploads");
+    let offset = 0;
+    await Promise.all(
+        parallelUploads.map(async () => {
+            while (offset < fileSize) {
+                const chunkSize =
+                    offset + MAX_CHUNK_SIZE > fileSize
+                        ? fileSize - offset
+                        : MAX_CHUNK_SIZE;
+                const start = offset;
+                const end = offset + chunkSize - 1;
+                offset += MAX_CHUNK_SIZE;
+                const chunk = fs.createReadStream(archivePath, {
+                    fd,
+                    start,
+                    end,
+                    autoClose: false
+                });
+                responses.push(
+                    await uploadChunk(
+                        restClient,
+                        resourceUrl,
+                        chunk,
+                        start,
+                        end
+                    )
+                );
+            }
+        })
+    );
+
+    fs.closeSync(fd);
+
+    const failedResponse = responses.find(
+        x => !isSuccessStatusCode(x.statusCode)
+    );
+    if (failedResponse) {
+        throw new Error(
+            `Cache service responded with ${failedResponse.statusCode} during chunk upload.`
+        );
+    }
+
+    return;
 }
 
 async function commitCache(
@@ -180,38 +241,11 @@ export async function saveCache(
 ): Promise<void> {
     const restClient = createRestClient();
 
-    core.debug("Uploading chunks");
+    core.debug("Upload cache");
-    // Upload Chunks
+    await uploadFile(restClient, cacheId, archivePath);
-    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
+    core.debug("Commiting cache");
     const cacheSize = utils.getArchiveFileSize(archivePath);
     const commitCacheResponse = await commitCache(
         restClient,

src/restore.ts

@@ -82,7 +82,6 @@ async function run(): Promise<void> {
                 cacheEntry?.archiveLocation,
                 archivePath
             );
-            await exec(`md5sum`, [archivePath]);
 
             const archiveFileSize = utils.getArchiveFileSize(archivePath);
             core.info(

src/save.ts

@@ -38,7 +38,9 @@ async function run(): Promise<void> {
         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.`);
+            core.info(
+                `Unable to reserve cache with key ${primaryKey}, another job may be creating this cache.`
+            );
             return;
         }
         core.debug(`Cache ID: ${cacheId}`);
@@ -84,7 +86,6 @@ async function run(): Promise<void> {
             return;
         }
 
-        await exec(`md5sum`, [archivePath]);
         core.debug("Saving Cache");
         await cacheHttpClient.saveCache(cacheId, archivePath);
     } catch (error) {