perf(stack): copy snapshot directories with one host command

.github/workflows/test.yml

@@ -149,6 +149,7 @@ jobs:
         run: >-
           pnpm test:integration
           src/Ports.integration.test.ts
+          src/storage/DirectoryCopy.integration.test.ts
           src/State.integration.test.ts
           src/State.process.integration.test.ts
           src/HostProcess.integration.test.ts

packages/stack/src/services/Database.ts

@@ -512,6 +512,7 @@ export const makeDatabase = (
         Effect.provideService(FileSystem.FileSystem, fs),
         Effect.provideService(Path.Path, path),
         Effect.provideService(Crypto.Crypto, crypto),
+        Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
         Effect.mapError((cause) => errorFor("snapshot", cause)),
       );
 

packages/stack/src/services/DatabaseSnapshot.integration.test.ts

@@ -1,6 +1,7 @@
 import { NodeServices } from "@effect/platform-node";
 import { describe, expect, it } from "@effect/vitest";
 import { Clock, Crypto, Effect, Exit, FileSystem, Option, Path, Schema, Scope } from "effect";
+import { ChildProcessSpawner } from "effect/unstable/process";
 import { makeDatabaseSnapshots } from "./DatabaseSnapshot.ts";
 
 const version = "17.6.1.173";
@@ -56,7 +57,15 @@ const entryPath = (path: Path.Path, cache: string, name: string) =>
   path.join(cache, "stack-database-snapshots", "entries", name);
 
 const live = <A, E>(
-  effect: Effect.Effect<A, E, Crypto.Crypto | FileSystem.FileSystem | Path.Path | Scope.Scope>,
+  effect: Effect.Effect<
+    A,
+    E,
+    | ChildProcessSpawner.ChildProcessSpawner
+    | Crypto.Crypto
+    | FileSystem.FileSystem
+    | Path.Path
+    | Scope.Scope
+  >,
 ) => Effect.scoped(effect).pipe(Effect.provide(NodeServices.layer));
 
 describe("managed native database snapshots", () => {

packages/stack/src/services/DatabaseSnapshot.ts

@@ -13,6 +13,7 @@ import {
   Schedule,
   Schema,
 } from "effect";
+import { ChildProcessSpawner } from "effect/unstable/process";
 import { postgresVersion } from "../Artifacts.ts";
 import { copyDirectory } from "../storage/DirectoryCopy.ts";
 import type { DatabaseRuntime } from "./Database.ts";
@@ -67,6 +68,7 @@ export const makeDatabaseSnapshots = Effect.fn("DatabaseSnapshot.make")(function
   const fs = yield* FileSystem.FileSystem;
   const path = yield* Path.Path;
   const crypto = yield* Crypto.Crypto;
+  const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
   const version = postgresVersion(options.version);
   const root = path.join(options.cacheRoot, "stack-database-snapshots");
   const entries = path.join(root, "entries");
@@ -149,6 +151,7 @@ export const makeDatabaseSnapshots = Effect.fn("DatabaseSnapshot.make")(function
     copyDirectory(source, destination).pipe(
       Effect.provideService(FileSystem.FileSystem, fs),
       Effect.provideService(Path.Path, path),
+      Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
       Effect.mapError((cause) => errorFor("copy", cause)),
     );
 

packages/stack/src/storage/DirectoryCopy.integration.test.ts

@@ -1,10 +1,35 @@
 import { NodeServices } from "@effect/platform-node";
+// oxlint-disable-next-line effecttsgo/node-builtin-import -- Windows junctions need Node's junction option.
+import { symlink as nativeSymlink } from "node:fs/promises";
 import { describe, expect, it } from "@effect/vitest";
-import { Effect, Exit, FileSystem, Path, Scope } from "effect";
+import {
+  Cause,
+  Data,
+  Deferred,
+  Effect,
+  Exit,
+  Fiber,
+  FileSystem,
+  Path,
+  Scope,
+  Sink,
+  Stream,
+} from "effect";
+import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
 import { copyDirectory } from "./DirectoryCopy.ts";
 
-const run = <A, E>(effect: Effect.Effect<A, E, FileSystem.FileSystem | Path.Path | Scope.Scope>) =>
-  Effect.scoped(effect).pipe(Effect.provide(NodeServices.layer));
+const run = <A, E>(
+  effect: Effect.Effect<
+    A,
+    E,
+    ChildProcessSpawner.ChildProcessSpawner | FileSystem.FileSystem | Path.Path | Scope.Scope
+  >,
+) => Effect.scoped(effect).pipe(Effect.provide(NodeServices.layer));
+
+class JunctionTestError extends Data.TaggedError("JunctionTestError")<{
+  readonly message: string;
+  readonly cause?: unknown;
+}> {}
 
 describe("copyDirectory", () => {
   it.live("copies nested files and preserves modes", () =>
@@ -78,6 +103,51 @@ describe("copyDirectory", () => {
     ),
   );
 
+  it.live("rejects Windows directory junctions without copying their targets", () =>
+    run(
+      Effect.gen(function* () {
+        if (process.platform !== "win32") return;
+        const fs = yield* FileSystem.FileSystem;
+        const path = yield* Path.Path;
+        const root = yield* fs.makeTempDirectoryScoped({ prefix: "directory-copy-junction-" });
+        const source = path.join(root, "source");
+        const outside = path.join(root, "outside");
+        const destination = path.join(root, "destination");
+        yield* fs.makeDirectory(source);
+        yield* fs.makeDirectory(outside);
+        yield* fs.writeFileString(path.join(outside, "secret.txt"), "outside\n");
+        yield* Effect.tryPromise({
+          try: () => nativeSymlink(outside, path.join(source, "junction"), "junction"),
+          catch: (cause) => new JunctionTestError({ message: "create directory junction", cause }),
+        });
+
+        const result = yield* copyDirectory(source, destination).pipe(Effect.exit);
+
+        expect(Exit.isFailure(result)).toBe(true);
+        expect(yield* fs.exists(destination)).toBe(false);
+        expect(yield* fs.readFileString(path.join(outside, "secret.txt"))).toBe("outside\n");
+      }),
+    ),
+  );
+
+  it.live("rejects a file source and leaves the destination absent", () =>
+    run(
+      Effect.gen(function* () {
+        const fs = yield* FileSystem.FileSystem;
+        const path = yield* Path.Path;
+        const root = yield* fs.makeTempDirectoryScoped({ prefix: "directory-copy-file-" });
+        const source = path.join(root, "source.txt");
+        const destination = path.join(root, "destination");
+        yield* fs.writeFileString(source, "file\n");
+
+        const result = yield* copyDirectory(source, destination).pipe(Effect.exit);
+
+        expect(Exit.isFailure(result)).toBe(true);
+        expect(yield* fs.exists(destination)).toBe(false);
+      }),
+    ),
+  );
+
   it.live("rejects an existing destination", () =>
     run(
       Effect.gen(function* () {
@@ -121,4 +191,157 @@ describe("copyDirectory", () => {
       }),
     ),
   );
+
+  it.live("removes a failed host copy and copies the tree itself", () =>
+    run(
+      Effect.gen(function* () {
+        const fs = yield* FileSystem.FileSystem;
+        const path = yield* Path.Path;
+        const root = yield* fs.makeTempDirectoryScoped({ prefix: "directory-copy-fallback-" });
+        const source = path.join(root, "source");
+        const destination = path.join(root, "destination");
+        yield* fs.makeDirectory(source);
+        yield* fs.writeFileString(path.join(source, "file.txt"), "file\n");
+
+        yield* copyDirectory(source, destination).pipe(
+          Effect.provideService(
+            ChildProcessSpawner.ChildProcessSpawner,
+            failedHostCopy(fs, path, destination),
+          ),
+        );
+
+        expect(yield* fs.readFileString(path.join(destination, "file.txt"))).toBe("file\n");
+        expect(yield* fs.exists(path.join(destination, "nested"))).toBe(false);
+      }),
+    ),
+  );
+
+  it.live("removes a partial copy when the host copy is interrupted", () =>
+    run(
+      Effect.gen(function* () {
+        const fs = yield* FileSystem.FileSystem;
+        const path = yield* Path.Path;
+        const root = yield* fs.makeTempDirectoryScoped({ prefix: "directory-copy-interrupt-" });
+        const source = path.join(root, "source");
+        const destination = path.join(root, "destination");
+        const started = yield* Deferred.make<void>();
+        yield* fs.makeDirectory(source);
+        yield* fs.writeFileString(path.join(source, "file.txt"), "file\n");
+
+        const fiber = yield* copyDirectory(source, destination).pipe(
+          Effect.provideService(
+            ChildProcessSpawner.ChildProcessSpawner,
+            hangingCopy(fs, path, started, destination),
+          ),
+          Effect.forkChild,
+        );
+        yield* Deferred.await(started);
+        yield* Fiber.interrupt(fiber);
+
+        expect(yield* fs.exists(destination)).toBe(false);
+      }),
+    ),
+  );
+
+  it.live("removes a partial copy when failure cleanup is interrupted", () =>
+    run(
+      Effect.gen(function* () {
+        const fs = yield* FileSystem.FileSystem;
+        const path = yield* Path.Path;
+        const root = yield* fs.makeTempDirectoryScoped({ prefix: "directory-copy-cleanup-" });
+        const source = path.join(root, "source");
+        const destination = path.join(root, "destination");
+        const entered = yield* Deferred.make<void>();
+        const release = yield* Deferred.make<void>();
+        yield* fs.makeDirectory(source);
+        yield* fs.writeFileString(path.join(source, "file.txt"), "file\n");
+        const pausing = FileSystem.FileSystem.of({
+          ...fs,
+          chmod: (target, mode) =>
+            Effect.gen(function* () {
+              if (target === destination) {
+                yield* Deferred.succeed(entered, undefined);
+                yield* Deferred.await(release);
+              }
+              yield* fs.chmod(target, mode);
+            }),
+        });
+
+        const fiber = yield* copyDirectory(source, destination).pipe(
+          Effect.provideService(FileSystem.FileSystem, pausing),
+          Effect.provideService(
+            ChildProcessSpawner.ChildProcessSpawner,
+            failedHostCopy(fs, path, destination),
+          ),
+          Effect.forkChild,
+        );
+        yield* Deferred.await(entered);
+        yield* Effect.sync(() => {
+          fiber.interruptUnsafe();
+        });
+        expect(fiber.pollUnsafe()).toBeUndefined();
+        yield* Deferred.succeed(release, undefined);
+        const exit = yield* Fiber.await(fiber);
+
+        expect(Exit.isFailure(exit) && Cause.hasInterrupts(exit.cause)).toBe(true);
+        expect(yield* fs.exists(destination)).toBe(false);
+      }),
+    ),
+  );
 });
+
+const processHandle = (exitCode: Effect.Effect<ChildProcessSpawner.ExitCode>) =>
+  ChildProcessSpawner.makeHandle({
+    pid: ChildProcessSpawner.ProcessId(1),
+    exitCode,
+    isRunning: Effect.succeed(false),
+    kill: () => Effect.void,
+    stdin: Sink.drain,
+    stdout: Stream.empty,
+    stderr: Stream.empty,
+    all: Stream.empty,
+    getInputFd: () => Sink.drain,
+    getOutputFd: () => Stream.empty,
+    unref: Effect.succeed(Effect.void),
+  });
+
+const writePartial = (fs: FileSystem.FileSystem, path: Path.Path, destination: string) =>
+  Effect.gen(function* () {
+    const nested = path.join(destination, "nested");
+    yield* fs.makeDirectory(nested, { recursive: true });
+    yield* fs.writeFileString(path.join(nested, "child.txt"), "partial\n");
+    if (process.platform !== "win32") yield* fs.chmod(nested, 0o555);
+  });
+
+const hostCommand = (command: ChildProcess.Command) => {
+  if (!ChildProcess.isStandardCommand(command)) return Effect.die("unexpected piped command");
+  if (command.command === "find") {
+    return Effect.succeed(processHandle(Effect.succeed(ChildProcessSpawner.ExitCode(0))));
+  }
+  return undefined;
+};
+
+const hangingCopy = (
+  fs: FileSystem.FileSystem,
+  path: Path.Path,
+  started: Deferred.Deferred<void>,
+  destination: string,
+) =>
+  ChildProcessSpawner.make((command) => {
+    const find = hostCommand(command);
+    if (find !== undefined) return find;
+    return writePartial(fs, path, destination).pipe(
+      Effect.as(processHandle(Effect.never)),
+      Effect.tap(() => Deferred.succeed(started, undefined)),
+    );
+  });
+
+// Exit 8 is a real robocopy failure and a non-zero cp status, not a usage error.
+const failedHostCopy = (fs: FileSystem.FileSystem, path: Path.Path, destination: string) =>
+  ChildProcessSpawner.make((command) => {
+    const find = hostCommand(command);
+    if (find !== undefined) return find;
+    return writePartial(fs, path, destination).pipe(
+      Effect.as(processHandle(Effect.succeed(ChildProcessSpawner.ExitCode(8)))),
+    );
+  });