Folder transfers parallelize within the shared 'At once' budget

The concurrency setting now caps total concurrent file STREAMS, not just jobs:
a resizable counting semaphore is shared engine-wide, and each job copies its
files through a small worker pool drawing from it. One folder job can use the
whole budget; several jobs share it fairly; Single stays strictly serial.
Cancel/session-death tear down every in-flight stream of the job.

Verified headless: 6-file folder at budget 3 -> exactly 3 parallel streams;
budget 1 -> serial; two folder jobs never exceed the combined budget; cancel
mid-parallel leaves no straggler streams. Also demoed a real remote error
surfacing in a failed job (upload into /proc -> server message on the row).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/bun/transfers/engine.ts

@@ -6,12 +6,48 @@ import type { SessionRegistry } from "../fs/registry";
 import { walkTree } from "./walk";
 
 const PROGRESS_INTERVAL_MS = 100;
+// workers per job; the shared slot budget is the real limit
+const MAX_WORKERS_PER_JOB = 8;
 
 interface Job {
 	snapshot: TransferSnapshot;
 	type: EntryType;
 	cancelled: boolean;
-	currentStream: Readable | null;
+	activeStreams: Set<Readable>;
+}
+
+/** Counting semaphore with live resize (concurrency setting changes mid-flight). */
+class Semaphore {
+	private inUse = 0;
+	private waiters: (() => void)[] = [];
+
+	constructor(private capacity: number) {}
+
+	async acquire(): Promise<void> {
+		if (this.inUse < this.capacity) {
+			this.inUse++;
+			return;
+		}
+		// wake() reserves the slot on our behalf before resolving
+		await new Promise<void>((resolve) => this.waiters.push(resolve));
+	}
+
+	release(): void {
+		this.inUse--;
+		this.wake();
+	}
+
+	resize(capacity: number): void {
+		this.capacity = capacity;
+		this.wake();
+	}
+
+	private wake(): void {
+		while (this.waiters.length && this.inUse < this.capacity) {
+			this.inUse++;
+			this.waiters.shift()!();
+		}
+	}
 }
 
 const TERMINAL: TransferState[] = ["completed", "failed", "cancelled"];
@@ -39,13 +75,17 @@ export class TransferEngine {
 	private running = 0;
 	private counter = 0;
 	private lastEmit = new Map<string, number>();
-	// serial by default; the transfers setting raises it (jobs are independent)
+	// serial by default; the transfers setting raises it. The same number limits
+	// active jobs AND total concurrent file streams (folder jobs parallelize
+	// internally out of this shared budget).
 	private concurrency = 1;
+	private slots = new Semaphore(1);
 
 	constructor(private registry: SessionRegistry) {}
 
 	setConcurrency(n: number): void {
 		this.concurrency = Math.max(1, Math.min(8, Math.round(n) || 1));
+		this.slots.resize(this.concurrency);
 		this.pump(); // a raised limit starts queued jobs immediately
 	}
 
@@ -62,7 +102,7 @@ export class TransferEngine {
 			const job: Job = {
 				type: item.type,
 				cancelled: false,
-				currentStream: null,
+				activeStreams: new Set(),
 				snapshot: {
 					id,
 					name: src.basename(item.path),
@@ -97,7 +137,7 @@ export class TransferEngine {
 			this.queue = this.queue.filter((j) => j !== job);
 			this.finish(job, "cancelled");
 		} else {
-			job.currentStream?.destroy(new Error("transfer cancelled"));
+			this.destroyStreams(job, new Error("transfer cancelled"));
 		}
 	}
 
@@ -118,7 +158,7 @@ export class TransferEngine {
 				this.queue = this.queue.filter((j) => j !== job);
 				this.finish(job, "failed");
 			} else {
-				job.currentStream?.destroy(new Error(reason));
+				this.destroyStreams(job, new Error(reason));
 			}
 		}
 	}
@@ -173,30 +213,59 @@ export class TransferEngine {
 			snap.state = "running";
 			this.emit(job, true);
 
-			for (const file of files) {
-				if (job.cancelled) throw new Error("transfer cancelled");
+			// copy with a small worker pool drawing from the SHARED stream budget:
+			// one folder job can use several slots, several jobs share them fairly
+			let nextIndex = 0;
+			let firstError: unknown = null;
+
+			const copyOne = async (file: (typeof files)[number]): Promise<void> => {
 				snap.currentFile = file.name;
 				snap.currentFileBytes = 0;
 				snap.currentFileTotal = file.size;
 				this.emit(job, true);
 
 				const readStream = src.createReadStream(file.srcPath);
-				job.currentStream = readStream;
+				job.activeStreams.add(readStream);
 				readStream.on("data", (chunk: Buffer | string) => {
 					const n = typeof chunk === "string" ? Buffer.byteLength(chunk) : chunk.length;
-					snap.currentFileBytes += n;
 					snap.doneBytes += n;
 					this.emit(job);
 				});
-				await pipeline(readStream, dst.createWriteStream(file.dstPath));
-				job.currentStream = null;
+				try {
+					await pipeline(readStream, dst.createWriteStream(file.dstPath));
+				} finally {
+					job.activeStreams.delete(readStream);
+				}
 				snap.doneFiles++;
 				this.emit(job, true);
-			}
+			};
+
+			const worker = async (): Promise<void> => {
+				while (!job.cancelled && !firstError && nextIndex < files.length) {
+					const file = files[nextIndex++];
+					await this.slots.acquire();
+					try {
+						if (job.cancelled || firstError) return;
+						await copyOne(file);
+					} catch (error) {
+						firstError ??= error;
+						this.destroyStreams(
+							job,
+							error instanceof Error ? error : new Error(String(error)),
+						);
+					} finally {
+						this.slots.release();
+					}
+				}
+			};
+
+			const workerCount = Math.max(1, Math.min(files.length, MAX_WORKERS_PER_JOB));
+			await Promise.all(Array.from({ length: workerCount }, worker));
+			if (job.cancelled) throw new Error("transfer cancelled");
+			if (firstError) throw firstError;
 
 			this.finish(job, "completed");
 		} catch (error) {
-			job.currentStream = null;
 			if (!job.cancelled && !snap.error) {
 				snap.error = error instanceof Error ? error.message : String(error);
 			}
@@ -204,6 +273,14 @@ export class TransferEngine {
 		}
 	}
 
+	/** Tear down every in-flight stream of a job (cancel, session death, first error). */
+	private destroyStreams(job: Job, reason: Error): void {
+		for (const stream of job.activeStreams) {
+			stream.destroy(reason);
+		}
+		job.activeStreams.clear();
+	}
+
 	/** Copying onto itself truncates the source; into a descendant never ends. */
 	private guardAgainstSelfCopy(snap: TransferSnapshot, dstPath: string): void {
 		if (snap.srcSessionId !== snap.dstSessionId) return;