Break up request graph cache serialisation and run after build completion

packages/core/cache/src/FSCache.js

@@ -105,25 +105,34 @@ export class FSCache implements Cache {
   async setLargeBlob(key: string, contents: Buffer | string): 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),
       );
+    } 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,
+                ),
+          ),
+        );
+      }
+    }
+
+    // If there's already a file following this chunk, it's old and should be removed
+    if (await this.fs.exists(this.#getFilePath(key, chunks))) {
+      writePromises.push(this.fs.unlink(this.#getFilePath(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';
@@ -111,27 +112,47 @@ export class LMDBCache 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;
+      writePromises.push(
+        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},
+          ),
+        );
+      }
     }
 
-    const writePromises: Promise<void>[] = [];
-    for (let i = 0; i < chunks; i += 1) {
+    // If there's already a file following this chunk, it's old and should be removed
+    if (await this.fs.exists(this.#getFilePath(key, chunks))) {
       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.unlink(this.#getFilePath(key, chunks), {
+          signal: options?.signal,
+        }),
       );
     }
 

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) {

packages/core/core/src/RequestTracker.js

@@ -846,6 +846,8 @@ export class RequestGraph extends ContentGraph<
   }
 }
 
+const NODES_PER_BLOB = 2 ** 14;
+
 export default class RequestTracker {
   graph: RequestGraph;
   farm: WorkerFarm;
@@ -1108,55 +1110,88 @@ export default class RequestTracker {
     return {api, subRequestContentKeys};
   }
 
-  async writeToCache() {
-    let cacheKey = getCacheKey(this.options);
-    let requestGraphKey = hashString(`${cacheKey}:requestGraph`);
-    let snapshotKey = hashString(`${cacheKey}:snapshot`);
+  async writeToCache(signal?: AbortSignal) {
+    const cacheKey = getCacheKey(this.options);
+    const requestGraphKey = `requestGraph-${hashString(cacheKey)}`;
+    const snapshotKey = `snapshot-${hashString(cacheKey)}`;
 
     if (this.options.shouldDisableCache) {
       return;
     }
-    let total = 2;
+
+    const serialisedGraph = this.graph.serialize();
+
+    let total = 0;
+    const serialiseAndSet = async (
+      key: string,
+      // $FlowFixMe serialise input is any type
+      contents: any,
+    ): Promise<void> => {
+      if (signal?.aborted) {
+        throw new Error('Serialization was aborted');
+      }
+
+      await this.options.cache.setLargeBlob(
+        key,
+        serialize(contents),
+        signal
+          ? {
+              signal: signal,
+            }
+          : undefined,
+      );
+
+      total += 1;
+
+      report({
+        type: 'cache',
+        phase: 'write',
+        total,
+        size: this.graph.nodes.length,
+      });
+    };
+
+    const promises: Promise<void>[] = [];
+
     report({
       type: 'cache',
       phase: 'start',
       total,
       size: this.graph.nodes.length,
     });
-    let promises = [];
-    for (let node of this.graph.nodes) {
-      if (!node || node.type !== REQUEST) {
-        continue;
-      }
 
-      let resultCacheKey = node.resultCacheKey;
-      if (resultCacheKey != null && node.result != null) {
-        promises.push(
-          this.options.cache.setLargeBlob(
-            resultCacheKey,
-            serialize(node.result),
-          ),
-        );
-        total++;
-        report({
-          type: 'cache',
-          phase: 'write',
-          total,
-          size: this.graph.nodes.length,
-        });
-        delete node.result;
+    // Preallocating a sparse array is faster than pushing when N is high enough
+    const cacheableNodes = new Array(serialisedGraph.nodes.length);
+    for (let i = 0; i < serialisedGraph.nodes.length; i += 1) {
+      const node = serialisedGraph.nodes[i];
+
+      let resultCacheKey = node?.resultCacheKey;
+      if (
+        node?.type === REQUEST &&
+        resultCacheKey != null &&
+        node?.result != null
+      ) {
+        promises.push(serialiseAndSet(resultCacheKey, node.result));
+        // eslint-disable-next-line no-unused-vars
+        const {result: _, ...newNode} = node;
+        cacheableNodes[i] = newNode;
+      } else {
+        cacheableNodes[i] = node;
       }
     }
 
+    for (let i = 0; i * NODES_PER_BLOB < cacheableNodes.length; i += 1) {
+      promises.push(
+        serialiseAndSet(
+          `requestGraph-nodes-${i}-${hashString(cacheKey)}`,
+          cacheableNodes.slice(i * NODES_PER_BLOB, (i + 1) * NODES_PER_BLOB),
+        ),
+      );
+    }
+
     promises.push(
-      this.options.cache.setLargeBlob(requestGraphKey, serialize(this.graph)),
+      serialiseAndSet(requestGraphKey, {...serialisedGraph, nodes: undefined}),
     );
-    report({
-      type: 'cache',
-      phase: 'write',
-      total,
-      size: this.graph.nodes.length,
-    });
 
     let opts = getWatcherOptions(this.options);
     let snapshotPath = path.join(this.options.cacheDir, snapshotKey + '.txt');
@@ -1167,15 +1202,10 @@ export default class RequestTracker {
         opts,
       ),
     );
-    report({
-      type: 'cache',
-      phase: 'write',
-      total,
-      size: this.graph.nodes.length,
-    });
-    report({type: 'cache', phase: 'end', total, size: this.graph.nodes.length});
 
     await Promise.all(promises);
+
+    report({type: 'cache', phase: 'end', total, size: this.graph.nodes.length});
   }
 
   static async init({
@@ -1205,20 +1235,42 @@ async function loadRequestGraph(options): Async<RequestGraph> {
     return new RequestGraph();
   }
 
-  let cacheKey = getCacheKey(options);
-  let requestGraphKey = hashString(`${cacheKey}:requestGraph`);
+  const cacheKey = getCacheKey(options);
+  const hashedCacheKey = hashString(cacheKey);
+  const requestGraphKey = `requestGraph-${hashedCacheKey}`;
   if (await options.cache.hasLargeBlob(requestGraphKey)) {
-    let requestGraph: RequestGraph = deserialize(
-      await options.cache.getLargeBlob(requestGraphKey),
-    );
+    const getAndDeserialize = async (key: string) => {
+      return deserialize(await options.cache.getLargeBlob(key));
+    };
+
+    let i = 0;
+    let nodePromises = [];
+    while (
+      await options.cache.hasLargeBlob(
+        `requestGraph-nodes-${i}-${hashedCacheKey}`,
+      )
+    ) {
+      nodePromises.push(
+        getAndDeserialize(`requestGraph-nodes-${i}-${hashedCacheKey}`),
+      );
+      i += 1;
+    }
+
+    const serializedRequestGraph = await getAndDeserialize(requestGraphKey);
+    const requestGraph = RequestGraph.deserialize({
+      ...serializedRequestGraph,
+      nodes: (await Promise.all(nodePromises)).flatMap(nodeChunk => nodeChunk),
+    });
+
     let opts = getWatcherOptions(options);
-    let snapshotKey = hashString(`${cacheKey}:snapshot`);
+    let snapshotKey = `snapshot-${hashedCacheKey}`;
     let snapshotPath = path.join(options.cacheDir, snapshotKey + '.txt');
     let events = await options.inputFS.getEventsSince(
       options.projectRoot,
       snapshotPath,
       opts,
     );
+
     requestGraph.invalidateUnpredictableNodes();
     requestGraph.invalidateOnBuildNodes();
     requestGraph.invalidateEnvNodes(options.env);