Break up request graph cache serialisation and run after build completion

packages/core/cache/src/FSCache.js

@@ -5,14 +5,14 @@ import type {FilePath} from '@parcel/types';
 import type {FileSystem} from '@parcel/fs';
 import type {Cache} from './types';
 
+import type {AbortSignal} from 'abortcontroller-polyfill/dist/cjs-ponyfill';
 import stream from 'stream';
 import path from 'path';
 import {promisify} from 'util';
 import logger from '@parcel/logger';
 import {serialize, deserialize, registerSerializableClass} from '@parcel/core';
 // flowlint-next-line untyped-import:off
 import packageJson from '../package.json';
-
 import {WRITE_LIMIT_CHUNK} from './constants';
 
 const pipeline: (Readable, Writable) => Promise<void> = promisify(
@@ -87,6 +87,15 @@ export class FSCache implements Cache {
     return path.join(this.dir, `${key}-${index}`);
   }
 
+  async #unlinkChunks(key: string, index: number): Promise<void> {
+    try {
+      await this.fs.unlink(this.#getFilePath(key, index));
+      await this.#unlinkChunks(key, index + 1);
+    } catch (err) {
+      // If there's an error, no more chunks are left to delete
+    }
+  }
+
   hasLargeBlob(key: string): Promise<boolean> {
     return this.fs.exists(this.#getFilePath(key, 0));
   }
@@ -102,30 +111,44 @@ export class FSCache implements Cache {
     return Buffer.concat(await Promise.all(buffers));
   }
 
-  async setLargeBlob(key: string, contents: Buffer | string): Promise<void> {
+  async setLargeBlob(
+    key: string,
+    contents: Buffer | string,
+    options?: {|signal?: AbortSignal|},
+  ): Promise<void> {
     const chunks = Math.ceil(contents.length / WRITE_LIMIT_CHUNK);
 
+    const writePromises: Promise<void>[] = [];
     if (chunks === 1) {
       // If there's one chunk, don't slice the content
-      await this.fs.writeFile(this.#getFilePath(key, 0), contents);
-      return;
-    }
-
-    const writePromises: Promise<void>[] = [];
-    for (let i = 0; i < chunks; i += 1) {
       writePromises.push(
-        this.fs.writeFile(
-          this.#getFilePath(key, i),
-          typeof contents === 'string'
-            ? contents.slice(i * WRITE_LIMIT_CHUNK, (i + 1) * WRITE_LIMIT_CHUNK)
-            : contents.subarray(
-                i * WRITE_LIMIT_CHUNK,
-                (i + 1) * WRITE_LIMIT_CHUNK,
-              ),
-        ),
+        this.fs.writeFile(this.#getFilePath(key, 0), contents, {
+          signal: options?.signal,
+        }),
       );
+    } else {
+      for (let i = 0; i < chunks; i += 1) {
+        writePromises.push(
+          this.fs.writeFile(
+            this.#getFilePath(key, i),
+            typeof contents === 'string'
+              ? contents.slice(
+                  i * WRITE_LIMIT_CHUNK,
+                  (i + 1) * WRITE_LIMIT_CHUNK,
+                )
+              : contents.subarray(
+                  i * WRITE_LIMIT_CHUNK,
+                  (i + 1) * WRITE_LIMIT_CHUNK,
+                ),
+            {signal: options?.signal},
+          ),
+        );
+      }
     }
 
+    // If there's already a files following this chunk, it's old and should be removed
+    writePromises.push(this.#unlinkChunks(key, chunks));
+
     await Promise.all(writePromises);
   }
 

packages/core/cache/src/LMDBCache.js

@@ -3,6 +3,7 @@ import type {FilePath} from '@parcel/types';
 import type {Cache} from './types';
 import type {Readable, Writable} from 'stream';
 
+import type {AbortSignal} from 'abortcontroller-polyfill/dist/cjs-ponyfill';
 import stream from 'stream';
 import path from 'path';
 import {promisify} from 'util';
@@ -12,7 +13,8 @@ import {NodeFS} from '@parcel/fs';
 import packageJson from '../package.json';
 // $FlowFixMe
 import lmdb from 'lmdb';
-import {WRITE_LIMIT_CHUNK} from './constants';
+
+import {FSCache} from './FSCache';
 
 const pipeline: (Readable, Writable) => Promise<void> = promisify(
   stream.pipeline,
@@ -23,10 +25,12 @@ export class LMDBCache implements Cache {
   dir: FilePath;
   // $FlowFixMe
   store: any;
+  fsCache: FSCache;
 
   constructor(cacheDir: FilePath) {
     this.fs = new NodeFS();
     this.dir = cacheDir;
+    this.fsCache = new FSCache(this.fs, cacheDir);
 
     this.store = lmdb.open(cacheDir, {
       name: 'parcel-cache',
@@ -100,42 +104,18 @@ export class LMDBCache implements Cache {
     return this.fs.exists(this.#getFilePath(key, 0));
   }
 
+  // eslint-disable-next-line require-await
   async getLargeBlob(key: string): Promise<Buffer> {
-    const buffers: Promise<Buffer>[] = [];
-    for (let i = 0; await this.fs.exists(this.#getFilePath(key, i)); i += 1) {
-      const file: Promise<Buffer> = this.fs.readFile(this.#getFilePath(key, i));
-
-      buffers.push(file);
-    }
-
-    return Buffer.concat(await Promise.all(buffers));
+    return this.fsCache.getLargeBlob(key);
   }
 
-  async setLargeBlob(key: string, contents: Buffer | string): Promise<void> {
-    const chunks = Math.ceil(contents.length / WRITE_LIMIT_CHUNK);
-
-    if (chunks === 1) {
-      // If there's one chunk, don't slice the content
-      await this.fs.writeFile(this.#getFilePath(key, 0), contents);
-      return;
-    }
-
-    const writePromises: Promise<void>[] = [];
-    for (let i = 0; i < chunks; i += 1) {
-      writePromises.push(
-        this.fs.writeFile(
-          this.#getFilePath(key, i),
-          typeof contents === 'string'
-            ? contents.slice(i * WRITE_LIMIT_CHUNK, (i + 1) * WRITE_LIMIT_CHUNK)
-            : contents.subarray(
-                i * WRITE_LIMIT_CHUNK,
-                (i + 1) * WRITE_LIMIT_CHUNK,
-              ),
-        ),
-      );
-    }
-
-    await Promise.all(writePromises);
+  // eslint-disable-next-line require-await
+  async setLargeBlob(
+    key: string,
+    contents: Buffer | string,
+    options?: {|signal?: AbortSignal|},
+  ): Promise<void> {
+    return this.fsCache.setLargeBlob(key, contents, options);
   }
 
   refresh(): void {

packages/core/cache/src/types.js

@@ -1,5 +1,6 @@
 // @flow
 import type {Readable} from 'stream';
+import type {AbortSignal} from 'abortcontroller-polyfill/dist/cjs-ponyfill';
 
 export interface Cache {
   ensure(): Promise<void>;
@@ -12,7 +13,11 @@ export interface Cache {
   setBlob(key: string, contents: Buffer | string): Promise<void>;
   hasLargeBlob(key: string): Promise<boolean>;
   getLargeBlob(key: string): Promise<Buffer>;
-  setLargeBlob(key: string, contents: Buffer | string): Promise<void>;
+  setLargeBlob(
+    key: string,
+    contents: Buffer | string,
+    options?: {|signal?: AbortSignal|},
+  ): Promise<void>;
   getBuffer(key: string): Promise<?Buffer>;
   /**
    * In a multi-threaded environment, where there are potentially multiple Cache

packages/core/core/src/Parcel.js

@@ -163,6 +163,8 @@ export default class Parcel {
     }
 
     let result = await this._build({startTime});
+
+    await this.#requestTracker.writeToCache();
     await this._end();
 
     if (result.type === 'buildFailure') {
@@ -175,10 +177,31 @@ export default class Parcel {
   async _end(): Promise<void> {
     this.#initialized = false;
 
-    await this.#requestTracker.writeToCache();
     await this.#disposable.dispose();
   }
 
+  async writeRequestTrackerToCache(): Promise<void> {
+    if (this.#watchQueue.getNumWaiting() === 0) {
+      // If there's no queued events, we are safe to write the request graph to disk
+      const abortController = new AbortController();
+
+      const unsubscribe = this.#watchQueue.subscribeToAdd(() => {
+        abortController.abort();
+      });
+
+      try {
+        await this.#requestTracker.writeToCache(abortController.signal);
+      } catch (err) {
+        if (!abortController.signal.aborted) {
+          // We expect abort errors if we interrupt the cache write
+          throw err;
+        }
+      }
+
+      unsubscribe();
+    }
+  }
+
   async _startNextBuild(): Promise<?BuildEvent> {
     this.#watchAbortController = new AbortController();
     await this.#farm.callAllWorkers('clearConfigCache', []);
@@ -198,6 +221,9 @@ export default class Parcel {
       if (!(err instanceof BuildAbortError)) {
         throw err;
       }
+    } finally {
+      // If the build passes or fails, we want to cache the request graph
+      await this.writeRequestTrackerToCache();
     }
   }
 
@@ -372,7 +398,6 @@ export default class Parcel {
       };
 
       await this.#reporterRunner.report(event);
-
       return event;
     } finally {
       if (this.isProfiling) {