fix(android): cancel admitted snapshot copies with their operation

Forward caller and timeout cancellation through engine and APK snapshots, release stream handles before removing partial output, and retain the existing root cleanup authority. Verify cancellation after destination bytes appear with a sparse filesystem fixture.

src/android/AndroidTargetSnapshot.ts

@@ -20,10 +20,12 @@ export const snapshotAndroidTarget = async (
   sha256: string,
   root: string,
   operation: AndroidOperation,
+  signal?: AbortSignal,
 ): Promise<string> => {
   const snapshot = join(root, "target.apk");
   try {
-    await copyAndroidSnapshot(path, snapshot, sha256);
+    await copyAndroidSnapshot(path, snapshot, sha256, signal);
+    signal?.throwIfAborted();
     await chmod(snapshot, 0o400);
     return snapshot;
   } catch (cause: unknown) {
@@ -44,10 +46,12 @@ export const snapshotAndroidEngine = async (
   sha256: string,
   root: string,
   operation: AndroidOperation,
+  signal?: AbortSignal,
 ): Promise<{ path: string; sha256: string }> => {
   const path = join(root, "engine.jar");
   try {
-    const observed = await copyAndroidSnapshot(source, path, sha256);
+    const observed = await copyAndroidSnapshot(source, path, sha256, signal);
+    signal?.throwIfAborted();
     await chmod(path, 0o400);
     return { path, sha256: observed };
   } catch (cause: unknown) {
@@ -65,15 +69,22 @@ const copyAndroidSnapshot = async (
   sourcePath: string,
   snapshotPath: string,
   expectedSha256: string,
+  signal?: AbortSignal,
 ): Promise<string> => {
-  const source = await openRegularFile(sourcePath, { symlinks: "follow" });
+  signal?.throwIfAborted();
+  const source = await openRegularFile(sourcePath, {
+    symlinks: "follow",
+    signal,
+  });
   let destination: FileHandle | undefined;
   let inputStream: ReadStream | undefined;
   let outputStream: WriteStream | undefined;
+  let removeSnapshot = false;
   let copiedBytes = 0;
   const hash = createHash("sha256");
   try {
     const initial = await source.stat();
+    signal?.throwIfAborted();
     const digest = new Transform({
       transform(
         chunk: Buffer,
@@ -95,13 +106,15 @@ const copyAndroidSnapshot = async (
     });
     try {
       destination = await open(snapshotPath, "wx", 0o600);
+      signal?.throwIfAborted();
       inputStream = source.createReadStream({
         start: 0,
         end: initial.size,
         autoClose: false,
       });
       outputStream = destination.createWriteStream({ autoClose: false });
-      await pipeline(inputStream, digest, outputStream);
+      await pipeline(inputStream, digest, outputStream, { signal });
+      signal?.throwIfAborted();
       await verifySourceState(source, sourcePath, initial, copiedBytes);
       const observedSha256 = hash.digest("hex");
       if (observedSha256 !== expectedSha256)
@@ -110,8 +123,7 @@ const copyAndroidSnapshot = async (
         );
       return observedSha256;
     } catch (cause: unknown) {
-      if (destination !== undefined)
-        await rm(snapshotPath, { force: true }).catch(() => undefined);
+      removeSnapshot = destination !== undefined;
       throw cause;
     }
   } finally {
@@ -120,7 +132,12 @@ const copyAndroidSnapshot = async (
     // descriptor closure, after the admitted source state has been verified.
     inputStream?.destroy();
     outputStream?.destroy();
-    await Promise.all([source.close(), destination?.close()]);
+    try {
+      await Promise.all([source.close(), destination?.close()]);
+    } finally {
+      if (removeSnapshot)
+        await rm(snapshotPath, { force: true }).catch(() => undefined);
+    }
   }
 };
 

src/android/JadxProvider.ts

@@ -388,12 +388,14 @@ export class JadxProvider implements AndroidAnalysisPort {
           jarHash,
           root.path,
           request.operation,
+          signal,
         );
         const snapshot = await snapshotAndroidTarget(
           target.path,
           target.sha256,
           root.path,
           request.operation,
+          signal,
         );
         const bridge = join(root.path, "ReaJadxBridge.java");
         await copyFile(BRIDGE_SOURCE, bridge);

tests/boundary/android/androidTargetSnapshot.test.ts

@@ -0,0 +1,47 @@
+import { stat, truncate, writeFile } from "node:fs/promises";
+import { join } from "node:path";
+
+import { expect, it } from "vitest";
+
+import { snapshotAndroidTarget } from "../../../src/android/AndroidTargetSnapshot.js";
+import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
+
+it.skipIf(process.platform === "win32")(
+  "cancels an in-flight APK snapshot and removes its partial output",
+  async () => {
+    const root = await createTestTempDirectory("rea-android-snapshot-cancel-");
+    const source = join(root, "large.apk");
+    const snapshot = join(root, "target.apk");
+    await writeFile(source, "");
+    // A sparse extent keeps fixture setup cheap while ensuring the copy cannot
+    // finish before the test observes its first output bytes and aborts it.
+    await truncate(source, 64 * 1024 * 1024 * 1024);
+
+    const controller = new AbortController();
+    const pending = snapshotAndroidTarget(
+      source,
+      "0".repeat(64),
+      root,
+      "inspect_android_package",
+      controller.signal,
+    );
+    try {
+      await expect
+        .poll(
+          () =>
+            stat(snapshot).then(
+              ({ size }) => size > 0,
+              () => false,
+            ),
+          { timeout: 10_000, interval: 10 },
+        )
+        .toBe(true);
+      controller.abort();
+      await expect(pending).rejects.toMatchObject({ name: "AbortError" });
+      await expect(stat(snapshot)).rejects.toMatchObject({ code: "ENOENT" });
+    } finally {
+      controller.abort();
+      await pending.catch(() => undefined);
+    }
+  },
+);