Fix race conditions in package linker (#137)

* fix race conditions in fs.copy by performing all operations as a queue - fixes #133, closes #126

* remove optional dependencies that fail compatibility check

* replace fsevents in install script test with flow-bin as it installs properly under linux

* make fs.copy more resillient to changing permissions

* fs.copyBulk should call itself again when unlinking

* unindent requires

* revert bin linking concurrency back to 4

Author: Sebastian McKenzie

src/package-compatibility.js

@@ -17,8 +17,9 @@ import { MessageError } from "./errors.js";
 import map from "./util/map.js";
 import { entries } from "./util/misc.js";
 
+let invariant = require("invariant");
 let semver = require("semver");
-let _      = require("lodash");
+let _ = require("lodash");
 
 function isValid(items: Array<string>, actual: string): boolean {
   let isBlacklist = false;
@@ -71,12 +72,30 @@ export default class PackageCompatibility {
   }
 
   check(info: Manifest) {
-    let didError = false;
-    let human    = `${info.name}@${info.version}`;
+    let didIgnore = false;
+    let didError  = false;
+    let reporter  = this.reporter;
+    let human     = `${info.name}@${info.version}`;
 
     let pushError = (msg) => {
-      didError = true;
-      this.reporter.error(`${human}: ${msg}`);
+      let ref = info.reference;
+      invariant(ref, "expected package reference");
+
+      if (ref.optional) {
+        ref.addIgnore(true);
+
+        reporter.warn(`${human}: ${msg}`);
+        if (!didIgnore) {
+          reporter.info(
+            `${human} is an optional dependency and failed compatibility check. ` +
+            "Excluding it from installation."
+          );
+          didIgnore = true;
+        }
+      } else {
+        reporter.error(`${human}: ${msg}`);
+        didError = true;
+      }
     };
 
     if (Array.isArray(info.os)) {

src/package-linker.js

@@ -280,16 +280,30 @@ export default class PackageLinker {
     });
 
     //
-    let tickCopyModule = this.reporter.progress(flatTree.length);
-    await promise.queue(flatTree, async function ([dest, { pkg, loc: src }]) {
-      pkg.reference.setLocation(dest);
-      await fs.mkdirp(dest);
-
-      let fresh = await fs.copy(src, dest);
-      pkg.reference.setFresh(fresh);
-
-      tickCopyModule(dest);
-    }, 4);
+    let queue = [];
+    for (let [dest, { pkg, loc: src }] of flatTree) {
+      let ref = pkg.reference;
+      invariant(ref, "expected package reference");
+
+      ref.setLocation(dest);
+      queue.push({
+        src,
+        dest,
+        onFresh() {
+          if (ref) ref.setFresh(true);
+        }
+      });
+    }
+    let bar;
+    await fs.copyBulk(queue, {
+      onStart: (num: number) => {
+        bar = this.reporter.progress(num);
+      },
+
+      onProgress(src: string) {
+        if (bar) bar(src);
+      }
+    });
 
     //
     let tickBin = this.reporter.progress(flatTree.length);

src/package-reference.js

@@ -13,6 +13,7 @@ import type Lockfile from "./lockfile/index.js";
 import type Config from "./config.js";
 import type { PackageRemote, Manifest } from "./types.js";
 import type PackageRequest from "./package-request.js";
+import type PackageResolver from "./package-resolver.js";
 import type { RegistryNames } from "./registries/index.js";
 import { entries } from "./util/misc.js";
 import { MessageError } from "./errors.js";
@@ -24,6 +25,7 @@ export default class PackageReference {
     remote: PackageRemote,
     saveForOffline: boolean,
   ) {
+    this.resolver = request.resolver;
     this.lockfile = request.rootLockfile;
     this.requests = [request];
     this.config   = request.config;
@@ -63,6 +65,7 @@ export default class PackageReference {
   remote: PackageRemote;
   registry: RegistryNames;
   location: ?string;
+  resolver: PackageResolver;
 
   async getFolder(): Promise<string> {
     return this.config.registries[this.registry].folder;
@@ -122,12 +125,27 @@ export default class PackageReference {
     }
   }
 
-  addIgnore(ignore: boolean) {
+  addIgnore(ignore: boolean, ancestry?: Set<PackageReference> = new Set) {
     // see comments in addOptional
     if (this.ignore == null) {
       this.ignore = ignore;
     } else if (!ignore) {
       this.ignore = false;
+    } else {
+      // we haven't changed our `ignore` so don't mess with
+      // dependencies
+      return;
+    }
+
+    if (ancestry.has(this)) return;
+    ancestry.add(this);
+
+    // go through and update all transitive dependencies to be ignored
+    for (let pattern of this.dependencies) {
+      let pkg = this.resolver.getResolvedPattern(pattern);
+      if (!pkg) continue;
+
+      pkg.reference.addIgnore(ignore, ancestry);
     }
   }
 }

src/util/fs.js

@@ -10,6 +10,7 @@
  */
 
 import BlockingQueue from "./blocking-queue.js";
+import * as promise from "./promise.js";
 import { promisify } from "./promise.js";
 import map from "./map.js";
 
@@ -31,86 +32,147 @@ export let lstat          = promisify(fs.lstat);
 export let chmod          = promisify(fs.chmod);
 
 let fsSymlink = promisify(fs.symlink);
+let invariant = require("invariant");
 let stripBOM  = require("strip-bom");
 
-export async function copy(src: string, dest: string): Promise<boolean> {
-  let srcStat = await lstat(src);
-  let destFiles = [];
-  let srcFiles = [];
+type CopyQueue = Array<{
+  src: string,
+  dest: string,
+  onFresh?: ?() => void
+}>;
+
+type CopyActions = Array<{
+  src: string,
+  dest: string,
+  atime: number,
+  mtime: number,
+  mode: number
+}>;
+
+async function buildActionsForCopy(queue: CopyQueue): Promise<CopyActions> {
+  let actions: CopyActions = [];
+  await init();
+  return actions;
+
+  // custom concurrency logic as we're always executing stacks of 4 queue items
+  // at a time due to the requirement to push items onto the queue
+  async function init() {
+    let items = queue.splice(0, 4);
+    if (!items.length) return;
+
+    await Promise.all(items.map(build));
+    return init();
+  }
 
-  if (await exists(dest)) {
-    let destStat = await lstat(dest);
+  //
+  async function build(data) {
+    let { src, dest, onFresh } = data;
+    let srcStat = await lstat(src);
+    let srcFiles;
 
-    if (srcStat.mode !== destStat.mode) {
-      // different types
-      await access(dest, srcStat.mode);
+    if (srcStat.isDirectory()) {
+      srcFiles = await readdir(src);
     }
 
-    if (srcStat.isFile() && destStat.isFile() &&
-        srcStat.size === destStat.size && +srcStat.mtime === +destStat.mtime) {
-      // we can safely assume this is the same file
-      return false;
-    }
+    if (await exists(dest)) {
+      let destStat = await lstat(dest);
 
-    if (srcStat.isDirectory() && destStat.isDirectory()) {
-      // remove files that aren't in source
-      [destFiles, srcFiles] = await Promise.all([
-        readdir(dest),
-        readdir(src)
-      ]);
+      let bothFiles   = srcStat.isFile() && destStat.isFile();
+      let bothFolders = !bothFiles && srcStat.isDirectory() && destStat.isDirectory();
 
-      let promises = destFiles.map(async (file) => {
-        if (file !== "node_modules" && srcFiles.indexOf(file) < 0) {
-          await unlink(path.join(dest, file));
+      if (srcStat.mode !== destStat.mode) {
+        if (bothFiles) {
+          await access(dest, srcStat.mode);
+        } else {
+          await unlink(dest);
+          return build(data);
         }
-      });
+      }
 
-      await Promise.all(promises);
-    }
-  }
+      if (bothFiles && srcStat.size === destStat.size && +srcStat.mtime === +destStat.mtime) {
+        // we can safely assume this is the same file
+        return;
+      }
 
-  if (srcStat.isDirectory()) {
-    let anyFresh = false;
+      if (bothFolders) {
+        // remove files that aren't in source
+        let destFiles = await readdir(dest);
+        invariant(srcFiles, "src files not initialised");
 
-    // create dest directory
-    await mkdirp(dest);
+        for (let file of destFiles) {
+          if (file === "node_modules") continue;
 
-    // copy all files from source to dest
-    let promises = srcFiles.map((file) => {
-      return copy(path.join(src, file), path.join(dest, file)).then(function (fresh) {
-        if (fresh) anyFresh = true;
-        return fresh;
+          if (srcFiles.indexOf(file) < 0) {
+            await unlink(path.join(dest, file));
+          }
+        }
+      }
+    }
+
+    if (srcStat.isDirectory()) {
+      await mkdirp(dest);
+
+      // push all files to queue
+      invariant(srcFiles, "src files not initialised");
+      for (let file of srcFiles) {
+        queue.push({
+          onFresh,
+          src: path.join(src, file),
+          dest: path.join(dest, file)
+        });
+      }
+    } else if (srcStat.isFile()) {
+      if (onFresh) onFresh();
+      actions.push({
+        src,
+        dest,
+        atime: srcStat.atime,
+        mtime: srcStat.mtime,
+        mode: srcStat.mode
       });
-    });
+    } else {
+      throw new Error("unsure how to copy this?");
+    }
+  }
+}
 
-    await Promise.all(promises);
+export function copy(src: string, dest: string): Promise<void> {
+  return copyBulk([{ src, dest }]);
+}
 
-    return anyFresh;
-  } else if (srcStat.isFile()) {
-    return new Promise((resolve, reject) => {
-      let readStream = fs.createReadStream(src);
-      let writeStream = fs.createWriteStream(dest, { mode: srcStat.mode });
+export async function copyBulk(
+  queue: CopyQueue,
+  events?: {
+    onProgress: (dest: string) => void,
+    onStart: (num: number) => void
+  }
+): Promise<void> {
+  let actions: CopyActions = await buildActionsForCopy(queue);
 
-      readStream.on("error", reject);
-      writeStream.on("error", reject);
+  if (events) events.onStart(actions.length);
 
-      writeStream.on("open", function () {
-        readStream.pipe(writeStream);
-      });
+  await promise.queue(actions, (data) => new Promise((resolve, reject) => {
+    let readStream = fs.createReadStream(data.src);
+    let writeStream = fs.createWriteStream(data.dest, { mode: data.mode });
 
-      writeStream.once("finish", function () {
-        fs.utimes(dest, srcStat.atime, srcStat.mtime, function (err) {
-          if (err) {
-            reject(err);
-          } else {
-            resolve(true);
-          }
-        });
+    readStream.on("error", reject);
+    writeStream.on("error", reject);
+
+    writeStream.on("open", function () {
+      readStream.pipe(writeStream);
+    });
+
+    writeStream.once("finish", function () {
+      fs.utimes(data.dest, data.atime, data.mtime, function (err) {
+        if (err) {
+          reject(err);
+        } else {
+          if (events) events.onProgress(data.dest);
+          resolve();
+        }
       });
     });
-  } else {
-    throw new Error("unsure how to copy this?");
-  }
+  }), 4);
 }
 
 export async function readFile(loc: string): Promise<string> {