P3: transfer engine with queue, progress, cancel + transfer bar UI
- TransferEngine: FIFO queue, provider-generic stream pipeline (covers all four panel combos), throttled transferUpdate push, cancel via stream destroy, self-copy guards, failSession hook for dead SSH sessions - TransfersService + TransferBar: per-job rows, progress bars, cancel/dismiss, auto-dismiss completed, dst panel refresh - F5 + Copy button wired to active-panel selection - verified: 30MB nested tree copy (UTF-8 names, empty dirs), self-copy rejection, mid-transfer cancel leaving partial file Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jokin
@@ -0,0 +1,227 @@
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import type { Readable } from "node:stream";
|
||||
import type { EntryType, TransferSnapshot, TransferState } from "../../shared/types";
|
||||
import type { FileSystemProvider } from "../fs/provider";
|
||||
import type { SessionRegistry } from "../fs/registry";
|
||||
import { walkTree } from "./walk";
|
||||
|
||||
// One at a time keeps remote servers happy; bump for parallel jobs.
|
||||
const TRANSFER_CONCURRENCY = 1;
|
||||
const PROGRESS_INTERVAL_MS = 100;
|
||||
|
||||
interface Job {
|
||||
snapshot: TransferSnapshot;
|
||||
type: EntryType;
|
||||
cancelled: boolean;
|
||||
currentStream: Readable | null;
|
||||
}
|
||||
|
||||
const TERMINAL: TransferState[] = ["completed", "failed", "cancelled"];
|
||||
|
||||
async function ensureDir(fs: FileSystemProvider, path: string): Promise<void> {
|
||||
try {
|
||||
const st = await fs.stat(path);
|
||||
if (st.type === "dir") return;
|
||||
throw new Error(`"${path}" exists and is not a directory`);
|
||||
} catch {
|
||||
await fs.mkdir(path);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* FIFO queue of copy jobs. Because both providers speak node streams, one
|
||||
* pipeline() uniformly covers local->local, upload, download and
|
||||
* remote->remote (piped through this process).
|
||||
*/
|
||||
export class TransferEngine {
|
||||
onUpdate: (snapshot: TransferSnapshot) => void = () => {};
|
||||
|
||||
private jobs = new Map<string, Job>();
|
||||
private queue: Job[] = [];
|
||||
private running = 0;
|
||||
private counter = 0;
|
||||
private lastEmit = new Map<string, number>();
|
||||
|
||||
constructor(private registry: SessionRegistry) {}
|
||||
|
||||
enqueue(params: {
|
||||
srcSessionId: string;
|
||||
items: { path: string; type: EntryType }[];
|
||||
dstSessionId: string;
|
||||
dstDir: string;
|
||||
}): string[] {
|
||||
const src = this.registry.get(params.srcSessionId);
|
||||
const jobIds: string[] = [];
|
||||
for (const item of params.items) {
|
||||
const id = `job-${++this.counter}`;
|
||||
const job: Job = {
|
||||
type: item.type,
|
||||
cancelled: false,
|
||||
currentStream: null,
|
||||
snapshot: {
|
||||
id,
|
||||
name: src.basename(item.path),
|
||||
srcSessionId: params.srcSessionId,
|
||||
srcPath: item.path,
|
||||
dstSessionId: params.dstSessionId,
|
||||
dstDir: params.dstDir,
|
||||
state: "queued",
|
||||
totalFiles: 0,
|
||||
totalBytes: 0,
|
||||
doneFiles: 0,
|
||||
doneBytes: 0,
|
||||
currentFile: "",
|
||||
currentFileBytes: 0,
|
||||
currentFileTotal: 0,
|
||||
},
|
||||
};
|
||||
this.jobs.set(id, job);
|
||||
this.queue.push(job);
|
||||
jobIds.push(id);
|
||||
this.emit(job, true);
|
||||
}
|
||||
this.pump();
|
||||
return jobIds;
|
||||
}
|
||||
|
||||
cancel(jobId: string): void {
|
||||
const job = this.jobs.get(jobId);
|
||||
if (!job || TERMINAL.includes(job.snapshot.state)) return;
|
||||
job.cancelled = true;
|
||||
if (job.snapshot.state === "queued") {
|
||||
this.queue = this.queue.filter((j) => j !== job);
|
||||
this.finish(job, "cancelled");
|
||||
} else {
|
||||
job.currentStream?.destroy(new Error("transfer cancelled"));
|
||||
}
|
||||
}
|
||||
|
||||
dismiss(jobId: string): void {
|
||||
const job = this.jobs.get(jobId);
|
||||
if (job && TERMINAL.includes(job.snapshot.state)) this.jobs.delete(jobId);
|
||||
}
|
||||
|
||||
/** Fail every non-terminal job that touches a dead session. */
|
||||
failSession(sessionId: string, reason: string): void {
|
||||
for (const job of this.jobs.values()) {
|
||||
const { srcSessionId, dstSessionId, state } = job.snapshot;
|
||||
if (TERMINAL.includes(state)) continue;
|
||||
if (srcSessionId !== sessionId && dstSessionId !== sessionId) continue;
|
||||
job.cancelled = true;
|
||||
job.snapshot.error = reason;
|
||||
if (state === "queued") {
|
||||
this.queue = this.queue.filter((j) => j !== job);
|
||||
this.finish(job, "failed");
|
||||
} else {
|
||||
job.currentStream?.destroy(new Error(reason));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private pump(): void {
|
||||
while (this.running < TRANSFER_CONCURRENCY && this.queue.length) {
|
||||
const job = this.queue.shift()!;
|
||||
this.running++;
|
||||
this.run(job)
|
||||
.catch(() => {})
|
||||
.finally(() => {
|
||||
this.running--;
|
||||
this.pump();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async run(job: Job): Promise<void> {
|
||||
const snap = job.snapshot;
|
||||
try {
|
||||
const src = this.registry.get(snap.srcSessionId);
|
||||
const dst = this.registry.get(snap.dstSessionId);
|
||||
|
||||
snap.state = "scanning";
|
||||
this.emit(job, true);
|
||||
|
||||
// collect the work list
|
||||
let files: { srcPath: string; dstPath: string; size: number; name: string }[];
|
||||
if (job.type === "dir") {
|
||||
const dstRoot = dst.join(snap.dstDir, snap.name);
|
||||
this.guardAgainstSelfCopy(snap, dstRoot);
|
||||
const tree = await walkTree(src, snap.srcPath);
|
||||
for (const segments of tree.dirs) {
|
||||
if (job.cancelled) throw new Error("transfer cancelled");
|
||||
await ensureDir(dst, dst.join(dstRoot, ...segments));
|
||||
}
|
||||
files = tree.files.map((f) => ({
|
||||
srcPath: f.entry.path,
|
||||
dstPath: dst.join(dstRoot, ...f.dirSegments, f.entry.name),
|
||||
size: f.entry.size,
|
||||
name: f.entry.name,
|
||||
}));
|
||||
} else {
|
||||
const dstPath = dst.join(snap.dstDir, snap.name);
|
||||
this.guardAgainstSelfCopy(snap, dstPath);
|
||||
const st = await src.stat(snap.srcPath);
|
||||
files = [{ srcPath: snap.srcPath, dstPath, size: st.size, name: snap.name }];
|
||||
}
|
||||
|
||||
snap.totalFiles = files.length;
|
||||
snap.totalBytes = files.reduce((sum, f) => sum + f.size, 0);
|
||||
snap.state = "running";
|
||||
this.emit(job, true);
|
||||
|
||||
for (const file of files) {
|
||||
if (job.cancelled) throw new Error("transfer cancelled");
|
||||
snap.currentFile = file.name;
|
||||
snap.currentFileBytes = 0;
|
||||
snap.currentFileTotal = file.size;
|
||||
this.emit(job, true);
|
||||
|
||||
const readStream = src.createReadStream(file.srcPath);
|
||||
job.currentStream = 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;
|
||||
snap.doneFiles++;
|
||||
this.emit(job, true);
|
||||
}
|
||||
|
||||
this.finish(job, "completed");
|
||||
} catch (error) {
|
||||
job.currentStream = null;
|
||||
if (!job.cancelled && !snap.error) {
|
||||
snap.error = error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
this.finish(job, job.cancelled && !snap.error ? "cancelled" : "failed");
|
||||
}
|
||||
}
|
||||
|
||||
/** Copying onto itself truncates the source; into a descendant never ends. */
|
||||
private guardAgainstSelfCopy(snap: TransferSnapshot, dstPath: string): void {
|
||||
if (snap.srcSessionId !== snap.dstSessionId) return;
|
||||
if (dstPath === snap.srcPath) {
|
||||
throw new Error("Source and destination are the same");
|
||||
}
|
||||
if (dstPath.startsWith(snap.srcPath.endsWith("/") ? snap.srcPath : `${snap.srcPath}/`)) {
|
||||
throw new Error("Cannot copy a folder into itself");
|
||||
}
|
||||
}
|
||||
|
||||
private finish(job: Job, state: TransferState): void {
|
||||
if (TERMINAL.includes(job.snapshot.state)) return;
|
||||
job.snapshot.state = state;
|
||||
this.emit(job, true);
|
||||
}
|
||||
|
||||
private emit(job: Job, force = false): void {
|
||||
const now = Date.now();
|
||||
if (!force && now - (this.lastEmit.get(job.snapshot.id) ?? 0) < PROGRESS_INTERVAL_MS) {
|
||||
return;
|
||||
}
|
||||
this.lastEmit.set(job.snapshot.id, now);
|
||||
this.onUpdate({ ...job.snapshot });
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user