Publicerad källa

CH-J Server Manager

Bläddra i kataloger och filer för en specifik programutgåva.

Ladda ner källa ZIP

CH-J / Server Manager 377 filer

core-20261019T111616Z-160a00bfcddc Tillagd: 19 okt. 2026 22:11 · Byggd: 19 okt. 2026 13:16

transferManager.js Tillagd: Datum inte noterat 65,3 KB · 1259 linjer

Kod Ladda ner fil

1"use strict";

3// Central SFTP Transfer Manager (Main process). Every file is one job with its own SFTP channel,

4// a bounded pipeline of positioned reads/writes (backpressure: at most PIPELINE chunks in flight)

5// and a confirmed offset that only advances over contiguous acknowledged chunks. Pause and cancel

6// stop issuing requests on the channel itself; resume reopens the partial file, verifies the tail

7// against the source and continues from that offset. Completed uploads are published by the shared

8// remote helper (verified hash, original metadata, atomic rename, privileged when needed); completed

9// downloads are hashed locally and renamed into place. Job state (never secrets) survives restarts.

10const crypto = require("node:crypto");

11const fs = require("node:fs");

12const path = require("node:path");

13const { EventEmitter } = require("node:events");

14const { atomicWriteJson } = require("../storage/atomicFile");

15const { fileError, normalizeRemotePath, safeEntryName } = require("./remoteFileService");

const STATES = Object.freeze(["waiting", "connecting", "transferring", "paused", "interrupted", "reconnecting", "verifying", "finalizing", "completed", "failed", "cancelled"]);

18const RUNNING = new Set(["connecting", "transferring", "verifying", "finalizing"]);

19const FINAL = new Set(["completed", "failed", "cancelled"]);

20const CHUNK = 256 * 1024;

21const PIPELINE = 8;

22const TAIL_CHECK = 1024 * 1024;

23const STALL_MS = 60000;

24const STOP_GRACE_MS = 5000;

25const EMIT_MS = 250;

26const PERSIST_MS = 1000;

27const RETRY_DELAYS_MS = [2000, 4000, 8000, 16000, 32000];

28const MAX_TOTAL_RETRIES = 50;

29const MAX_HISTORY = 500;

30const MAX_JOBS = 20000;

31const MAX_PLAN_ENTRIES = 10000;

32const MAX_PLAN_BYTES = 4 * 1024 * 1024 * 1024 * 1024;

33const PLAN_TTL_MS = 10 * 60 * 1000;

34const ORPHAN_AGE_MS = 24 * 3600 * 1000;

35const HASH_RATE = 10 * 1024 * 1024;

36const DEFAULT_SETTINGS = Object.freeze({ maxConcurrent: 2, autoRetry: true, verify: true });

const CONNECTION_CODES = new Set(["SESSION_NOT_CONNECTED", "SFTP_OPEN_FAILED", "TRANSFER_STALLED", "REMOTE_COMMAND_TIMEOUT", "REMOTE_COMMAND_FAILED",

38 "FILE_EXEC_FAILED", "ECONNRESET", "ETIMEDOUT", "EPIPE"]);

39const NO_HELPER_CODES = new Set(["FILE_DEPENDENCY_UNAVAILABLE", "FEATURE_PLATFORM_UNSUPPORTED"]);

const AUTH_CODES = new Set(["SUDO_PASSWORD_REQUIRED", "SUDO_PASSWORD_INCORRECT", "FILE_PERMISSION_DENIED", "FILE_METADATA_PERMISSION"]);

42class StopSignal extends Error {

43 constructor(kind) { super(`Transfer ${kind}`); this.stop = kind; }

44}

46function hex(bytes = 16) { return crypto.randomBytes(bytes).toString("hex"); }

47function call(sftp, method, ...args) {

  return new Promise((resolve, reject) => sftp[method](...args, (error, ...results) => error ? reject(error) : resolve(results.length > 1 ? results : results[0])));

49}

50function errorCode(error) {

51 if (error?.code === 6 || error?.code === 7) return "SESSION_NOT_CONNECTED";

52 if (typeof error?.code === "string") return error.code;

  if (/no response|not connected|connection (lost|closed)|channel|ECONNRESET|socket/i.test(String(error?.message || ""))) return "SESSION_NOT_CONNECTED";

54 if (error?.code === 3) return "FILE_PERMISSION_DENIED";

55 if (error?.code === 2) return "FILE_NOT_FOUND";

56 return "TRANSFER_FAILED";

57}

function localIdentity(stats) { return { size: stats.size, mtimeMs: Math.trunc(stats.mtimeMs), ino: Number(stats.ino) || 0, dev: Number(stats.dev) || 0 }; }

function sameIdentity(left, right) { return Boolean(left && right) && ["size", "mtimeMs", "ino", "dev"].every((key) => left[key] === right[key]); }

60function percent(transferred, size) {

61 if (!Number.isFinite(size) || size < 0) return null;

62 if (size === 0) return null;

63 return Math.min(100, Math.floor((Math.min(transferred, size) / size) * 1000) / 10);

64}

65// A dead channel never answers: cleanup calls must not block the job forever.

66function within(promise, ms = 3000) {

67 let timer;

  return Promise.race([Promise.resolve(promise).catch(() => {}), new Promise((resolve) => { timer = setTimeout(resolve, ms); timer.unref?.(); })]).finally(() => clearTimeout(timer));

69}

70function partName(name, id) { return `.${String(name).slice(0, 120)}.chj-part-${id}`; }

72// Aggregate view: percentages are byte based; an unknown size yields an indeterminate total.

73function summarize(jobs) {

74 const counted = jobs.filter((job) => job.state !== "cancelled");

75 const sizesKnown = counted.every((job) => Number.isFinite(job.size) && job.size >= 0);

76 const totalBytes = counted.reduce((sum, job) => sum + (Number.isFinite(job.size) ? job.size : 0), 0);

  const transferredBytes = counted.reduce((sum, job) => sum + Math.min(job.transferred || 0, Number.isFinite(job.size) ? job.size : job.transferred || 0), 0);

78 const completed = counted.filter((job) => job.state === "completed").length;

79 const active = jobs.filter((job) => RUNNING.has(job.state));

80 const speed = active.reduce((sum, job) => sum + (job.speed || 0), 0);

  const remaining = counted.filter((job) => !FINAL.has(job.state)).reduce((sum, job) => sum + Math.max(0, (job.size || 0) - (job.transferred || 0)), 0);

82 const etaReliable = active.length > 0 && active.every((job) => job.state !== "transferring" || job.eta !== null);

83 let overall = null;

84 if (sizesKnown && totalBytes > 0) overall = Math.floor((transferredBytes / totalBytes) * 1000) / 10;

85 else if (sizesKnown && counted.length && completed === counted.length) overall = 100;

86 if (overall !== null && overall >= 100 && completed !== counted.length) overall = 99.9;

87 return {

88 total: jobs.length, completed, active: active.length,

89 waiting: jobs.filter((job) => job.state === "waiting" || job.state === "reconnecting").length,

90 paused: jobs.filter((job) => job.state === "paused").length,

91 interrupted: jobs.filter((job) => job.state === "interrupted").length,

92 failed: jobs.filter((job) => job.state === "failed").length,

93 cancelled: jobs.length - counted.length,

94 totalBytes, transferredBytes, sizesKnown, percent: overall, speed,

    eta: speed > 0 && etaReliable && remaining > 0 ? Math.ceil(remaining / speed) : remaining === 0 && completed === counted.length ? 0 : null

96 };

97}

99class TransferManager extends EventEmitter {

100 constructor(options) {

101 super();

102 this.sessionManager = options.sessionManager;

103 this.files = options.remoteFileService;

104 this.selectUploadPath = options.selectUploadPath || options.remoteFileService?.selectUploadPath;

105 this.selectDownloadPath = options.selectDownloadPath || options.remoteFileService?.selectDownloadPath;

106

    this.selectDownloadDirectory = options.selectDownloadDirectory || options.remoteFileService?.selectDownloadDirectory;

107 this.logger = options.logger;

108 this.now = options.now || (() => Date.now());

109 this.stopGraceMs = Number.isFinite(options.stopGraceMs) ? options.stopGraceMs : STOP_GRACE_MS;

110

    this.retryDelays = Array.isArray(options.retryDelaysMs) && options.retryDelaysMs.length ? options.retryDelaysMs : RETRY_DELAYS_MS;

111 this.file = options.storageRoot ? path.join(options.storageRoot, "transfers.json") : null;

112 this.settings = { ...DEFAULT_SETTINGS };

113 this.jobs = new Map();

114 this.runtime = new Map();

115 this.secrets = new Map(); // jobId -> sudo password, memory only

116 this.plans = new Map();

117 this.changed = new Set();

118 this.cleanedIdentities = new Set();

119 this.order = 0;

120 this._load();

121 this._onState = (status) => this._sessionState(status);

122 this.sessionManager?.on?.("state", this._onState);

123 }

124

125 // ---- public API (owner = plugin id) ---------------------------------------------------------

126

127 list(owner) {

128 const jobs = this._owned(owner).map((job) => this._public(job));

129 this._cleanupOrphansLater();

130 return { jobs, summary: summarize(jobs), settings: { ...this.settings } };

131 }

132

133 updateSettings(_owner, patch = {}) {

134 if (patch.maxConcurrent !== undefined) {

135 const value = Number(patch.maxConcurrent);

136

      if (!Number.isInteger(value) || value < 1 || value > 8) throw fileError("TRANSFER_INVALID_SETTINGS", "Concurrent transfers must be between 1 and 8.");

137 this.settings.maxConcurrent = value;

138 }

139 if (patch.autoRetry !== undefined) this.settings.autoRetry = patch.autoRetry === true;

140 if (patch.verify !== undefined) this.settings.verify = patch.verify === true;

141 this._persist(true);

142 this._schedule();

143 return { ...this.settings };

144 }

145

146 async planUpload(owner, sessionId, remoteDirectory, options = {}) {

147 const directory = normalizeRemotePath(remoteDirectory || "/");

148 const identity = this._identity(sessionId);

149

    if (typeof this.selectUploadPath !== "function") throw fileError("TRANSFER_DIALOG_UNAVAILABLE", "Upload dialog is unavailable.");

150 const selected = await this.selectUploadPath();

151

    const localPaths = selected && !selected.canceled ? [...new Set((selected.paths || (selected.path ? [selected.path] : [])).map(String).filter(Boolean))] : [];

152 if (!localPaths.length) return { canceled: true };

153

    const plan = { id: hex(), owner, direction: "upload", sessionId, identity, directory, files: [], directories: [], skipped: [], bytes: 0, createdAt: this.now() };

154

    for (const localPath of localPaths) await this._walkLocal(plan, localPath, path.posix.join(directory, safeEntryName(path.basename(localPath))));

155 await this.files._withSftp(sessionId, async (sftp) => {

156 plan.mode = await this._directoryMode(sftp, sessionId, directory);

157 await this._remoteConflicts(sftp, plan);

158 });

159 if (plan.mode === "privileged") plan.privilege = await this.files.privileges(sessionId);

160 if (plan.mode === "privileged" && plan.privilege.mode === "none") {

161

      throw fileError(plan.privilege.sudoInstalled ? "SUDO_NOT_PERMITTED" : "SUDO_UNAVAILABLE", "The destination is not writable and the SSH account cannot use sudo.");

162 }

163 return this._storePlan(plan);

164 }

165

166 async planDownload(owner, sessionId, remotePaths, options = {}) {

167 const identity = this._identity(sessionId);

168

    const targets = [...new Set((Array.isArray(remotePaths) ? remotePaths : [remotePaths]).filter((value) => value !== undefined && value !== null).map(normalizeRemotePath))];

169

    if (!targets.length || targets.length > 100) throw fileError("TRANSFER_INVALID_SELECTION", "Select between 1 and 100 remote entries.");

170

    if (targets.includes("/")) throw fileError("TRANSFER_INVALID_SELECTION", "Downloading the entire remote root is not allowed in one batch.");

171

    const plan = { id: hex(), owner, direction: "download", sessionId, identity, files: [], directories: [], skipped: [], bytes: 0, createdAt: this.now(),

172 mode: options.sudo === true ? "privileged-allowed" : "direct" };

173 await this.files._withSftp(sessionId, async (sftp) => {

174 const first = targets.length === 1 ? await this._remoteStat(sftp, targets[0], options, sessionId) : null;

175 if (first && first.type === "file") {

176

        if (typeof this.selectDownloadPath !== "function") throw fileError("TRANSFER_DIALOG_UNAVAILABLE", "Download dialog is unavailable.");

177 const chosen = await this.selectDownloadPath(path.posix.basename(targets[0]));

178 if (!chosen || chosen.canceled || !chosen.path) { plan.canceled = true; return; }

179 plan.localDirectory = path.dirname(chosen.path);

180 this._addDownload(plan, targets[0], chosen.path, first);

181 plan.files[0].overwriteConfirmed = true; // The native save dialog already confirmed replacing.

182 return;

183 }

184

      if (typeof this.selectDownloadDirectory !== "function") throw fileError("TRANSFER_DIALOG_UNAVAILABLE", "Download directory dialog is unavailable.");

185 const chosen = await this.selectDownloadDirectory();

186 if (!chosen || chosen.canceled || !chosen.path) { plan.canceled = true; return; }

187 plan.localDirectory = chosen.path;

188

      for (const target of targets) await this._walkRemote(sftp, sessionId, plan, target, path.join(chosen.path, safeEntryName(path.posix.basename(target))), options, true);

189 });

190 if (plan.canceled) return { canceled: true };

191 for (const file of plan.files) {

192

      if (!file.overwriteConfirmed && await fs.promises.lstat(file.localPath).then(() => true, () => false)) file.conflict = true;

193 }

194 return this._storePlan(plan);

195 }

196

197 async start(owner, planId, options = {}) {

198 const plan = this.plans.get(String(planId || ""));

199

    if (!plan || plan.owner !== owner || this.now() - plan.createdAt > PLAN_TTL_MS) throw fileError("TRANSFER_PLAN_EXPIRED", "The transfer plan expired; select the files again.");

200 const conflict = ["overwrite", "skip"].includes(options.conflict) ? options.conflict : "fail";

201 const conflicts = plan.files.filter((file) => file.conflict);

202

    if (conflicts.length && conflict === "fail") throw fileError("FILE_EXISTS", `${conflicts.length} destination file(s) already exist.`);

203

    if (this.jobs.size + plan.files.length > MAX_JOBS) throw fileError("TRANSFER_QUEUE_FULL", "Remove finished transfers before starting more.");

204 const password = options.sudoPassword ? String(options.sudoPassword) : "";

205

    const privileged = plan.mode === "privileged" || (plan.direction === "download" && (plan.mode === "privileged-allowed" || options.sudo === true));

206 if (plan.mode === "privileged") {

207 // Verifies the sudo authorization once before any data moves (and caches it on request).

208

      await this.files.helper(plan.sessionId, { operation: "stat", path: plan.directory }, { sudo: true, sudoPassword: password, rememberSudo: options.rememberSudo === true });

209 }

210 await this._createDirectories(plan, privileged, password);

211 const created = [];

212 for (const file of plan.files) {

213 if (file.conflict && conflict === "skip") continue;

214 const job = {

215

        id: hex(), owner, batchId: plan.id, direction: plan.direction, identity: plan.identity, sessionId: plan.sessionId, order: ++this.order,

216

        name: file.name, localPath: file.localPath, remotePath: file.remotePath, size: file.size, transferred: 0, movedBytes: 0, activeMs: 0,

217 state: "waiting", mode: plan.direction === "upload" ? plan.mode : privileged ? "privileged-allowed" : "direct",

218

        overwrite: conflict === "overwrite" || file.overwriteConfirmed === true, localIdentity: file.localIdentity || null, remoteIdentity: file.remoteIdentity || null, remoteMode: file.mode || null,

219

        finalize: { state: "pending" }, attempts: 0, totalRetries: 0, error: null, createdAt: this.now(), updatedAt: this.now()

220 };

221 this.jobs.set(job.id, job);

222 if (password && privileged) this.secrets.set(job.id, password);

223 created.push(job.id);

224 this._mark(job);

225 }

226 this.plans.delete(plan.id);

227 this._trimHistory();

228 this._persist(true);

229

    this.logger?.info("Transfer batch queued.", { direction: plan.direction, files: created.length, bytes: plan.bytes, mode: plan.mode, sessionId: plan.sessionId });

230 this._schedule();

231 return { batchId: plan.id, jobs: created.length, skipped: plan.files.length - created.length };

232 }

233

234 pause(owner, ids) {

235 let count = 0;

236 for (const job of this._select(owner, ids)) {

237 if (job.state === "waiting" || job.state === "reconnecting" || job.state === "interrupted") {

238 this._clearRetry(job);

239 this._set(job, "paused");

240 count += 1;

241 } else if (RUNNING.has(job.state) && job.state !== "finalizing") {

242 this._stop(job, "pause");

243 count += 1;

244 }

245 }

246 this._persist(true);

247 return { paused: count };

248 }

249

250 resume(owner, ids, options = {}) {

251 let count = 0;

252 for (const job of this._select(owner, ids)) {

253

      const resumable = ["paused", "interrupted", "reconnecting"].includes(job.state) || (job.state === "failed" && job.resumable);

254 if (!resumable) continue;

255 this._authorize(job, options);

256 if (options.overwrite === true && ["FILE_EXISTS", "FILE_CONFLICT"].includes(job.error?.code)) {

257 job.overwrite = true;

258 job.targetBaseline = undefined;

259 job.finalize = { state: "pending" };

260 }

261 this._clearRetry(job);

262 job.attempts = 0;

263 job.authRequired = false;

264 this._set(job, "waiting");

265 count += 1;

266 }

267 this._persist(true);

268 this._schedule();

269 return { resumed: count };

270 }

271

272 // Restart from offset zero (used when the partial data cannot be safely continued).

273 async retry(owner, ids, options = {}) {

274 let count = 0;

275 for (const job of this._select(owner, ids)) {

276 if (!["failed", "interrupted", "paused", "cancelled"].includes(job.state)) continue;

277 // The deterministic partial file (same job id) is truncated and rewritten from offset zero.

278 if (job.localPart) await fs.promises.unlink(job.localPart).catch(() => {});

279 this._authorize(job, options);

280

      Object.assign(job, { transferred: 0, staged: null, targetBaseline: undefined, hash: null, verification: null, finalize: { state: "pending" },

281 attempts: 0, authRequired: false, error: null, resumable: false, finalizeWithSudo: false });

282 if (job.direction === "upload") {

283 const stats = await fs.promises.stat(job.localPath).catch(() => null);

284 if (stats) { job.localIdentity = localIdentity(stats); job.size = stats.size; }

285 } else job.remoteIdentity = null;

286 if (options.overwrite === true) job.overwrite = true;

287 this._set(job, "waiting");

288 count += 1;

289 }

290 this._persist(true);

291 this._schedule();

292 return { retried: count };

293 }

294

295 async cancel(owner, ids) {

296 let count = 0;

297 for (const job of this._select(owner, ids)) {

298 if (FINAL.has(job.state)) continue;

299 count += 1;

300 if (RUNNING.has(job.state)) {

301 if (job.state === "finalizing") continue; // The remote replacement must finish or fail on its own.

302 this._stop(job, "cancel");

303 continue;

304 }

305 this._clearRetry(job);

306 await this._finishCancel(job);

307 }

308 this._persist(true);

309 return { cancelled: count };

310 }

311

312 remove(owner, ids) {

313 let count = 0;

314 for (const job of this._select(owner, ids)) {

315 if (!FINAL.has(job.state)) continue;

316 if (job.cleanupPending) continue; // Kept until its temporary data is removed.

317 this.jobs.delete(job.id);

318 this.secrets.delete(job.id);

319 this.runtime.delete(job.id);

320 this.changed.add(job.id);

321 count += 1;

322 }

323 this._persist(true);

324 this._emitSoon(owner, true);

325 return { removed: count };

326 }

327

328 shutdown() {

329 if (this.closed) return;

330 clearTimeout(this.emitTimer);

331 this.sessionManager?.removeListener?.("state", this._onState);

332 for (const job of this.jobs.values()) {

333 this._clearRetry(job);

334 if (RUNNING.has(job.state)) this._stop(job, "shutdown");

335 }

336 this._persist(true, true);

337 this.closed = true; // Late completions of stopped jobs must not overwrite the saved state.

338 this.secrets.clear();

339 }

340

341 // ---- planning helpers -----------------------------------------------------------------------

342

343 _storePlan(plan) {

344

    if (plan.files.length + plan.directories.length > MAX_PLAN_ENTRIES) throw fileError("TRANSFER_TOO_MANY_ENTRIES", `A transfer can contain at most ${MAX_PLAN_ENTRIES} entries.`);

345

    if (plan.bytes > MAX_PLAN_BYTES) throw fileError("TRANSFER_TOO_LARGE", "The selection exceeds the transfer size limit.");

346 for (const [id, other] of this.plans) if (this.now() - other.createdAt > PLAN_TTL_MS) this.plans.delete(id);

347 this.plans.set(plan.id, plan);

348 const conflicts = plan.files.filter((file) => file.conflict);

349 return {

350

      canceled: false, planId: plan.id, direction: plan.direction, files: plan.files.length, directories: plan.directories.length, bytes: plan.bytes,

351 mode: plan.mode, requiresSudo: plan.mode === "privileged" && plan.privilege?.mode !== "root",

352

      sudoPasswordRequired: plan.mode === "privileged" && plan.privilege?.mode === "sudo-password" && !plan.privilege?.sudoCached,

353

      conflicts: conflicts.slice(0, 50).map((file) => file.direction === "upload" ? file.remotePath : file.localPath), conflictCount: conflicts.length,

354

      skipped: plan.skipped.slice(0, 50), skippedCount: plan.skipped.length, destination: plan.direction === "upload" ? plan.directory : plan.localDirectory

355 };

356 }

357

358 async _walkLocal(plan, localPath, remotePath, depth = 0) {

359

    if (plan.files.length + plan.directories.length > MAX_PLAN_ENTRIES) throw fileError("TRANSFER_TOO_MANY_ENTRIES", `A transfer can contain at most ${MAX_PLAN_ENTRIES} entries.`);

360 const stats = await fs.promises.lstat(localPath);

361 if (stats.isSymbolicLink()) { plan.skipped.push({ path: localPath, reason: "symlink" }); return; }

362 if (stats.isDirectory()) {

363 plan.directories.push(remotePath);

364 if (depth > 64) throw fileError("TRANSFER_TOO_DEEP", "The selected directory tree is too deep.");

365 for (const entry of await fs.promises.readdir(localPath, { withFileTypes: true })) {

366 let name;

367

        try { name = safeEntryName(entry.name); } catch { plan.skipped.push({ path: path.join(localPath, entry.name), reason: "name" }); continue; }

368 await this._walkLocal(plan, path.join(localPath, name), path.posix.join(remotePath, name), depth + 1);

369 }

370 return;

371 }

372 if (!stats.isFile()) { plan.skipped.push({ path: localPath, reason: "type" }); return; }

373 plan.bytes += stats.size;

374

    plan.files.push({ direction: "upload", name: path.basename(localPath), localPath, remotePath, size: stats.size, localIdentity: localIdentity(stats) });

375 }

376

377 // Writable by the SSH account: direct SFTP; otherwise the privileged staging route.

378 async _directoryMode(sftp, sessionId, directory) {

379

    const attrs = await call(sftp, "stat", directory).catch((error) => { throw fileError(errorCode(error) === "FILE_NOT_FOUND" ? "FILE_NOT_FOUND" : errorCode(error), `${directory} is not accessible.`); });

380

    if (typeof attrs?.isDirectory === "function" && !attrs.isDirectory()) throw fileError("FILE_NOT_DIRECTORY", `${directory} is not a directory.`);

381 const probe = path.posix.join(directory, `.chj-probe-${hex(6)}`);

382 try {

383 const handle = await call(sftp, "open", probe, "wx", { mode: 0o600 });

384 await call(sftp, "close", handle).catch(() => {});

385 await call(sftp, "unlink", probe).catch(() => {});

386 return "direct";

387 } catch (error) {

388 if (errorCode(error) === "FILE_PERMISSION_DENIED") return "privileged";

389

      throw fileError(errorCode(error) === "TRANSFER_FAILED" ? "FILE_REMOTE_IO_FAILED" : errorCode(error), `${directory} is not writable.`);

390 }

391 }

392

393 async _remoteConflicts(sftp, plan) {

394 const queue = [...plan.files];

395 const worker = async () => {

396 for (let file = queue.shift(); file; file = queue.shift()) {

397 const attrs = await call(sftp, "lstat", file.remotePath).catch(() => null);

398 if (!attrs) continue;

399

        if (typeof attrs.isDirectory === "function" && attrs.isDirectory()) { file.blocked = true; plan.skipped.push({ path: file.remotePath, reason: "directory-exists" }); }

400 else file.conflict = true;

401 }

402 };

403 await Promise.all(Array.from({ length: 16 }, worker));

404 plan.files = plan.files.filter((file) => !file.blocked);

405 }

406

407 async _remoteStat(sftp, target, options, sessionId) {

408 try {

409 const attrs = await call(sftp, "stat", target);

410

      return { type: typeof attrs.isDirectory === "function" && attrs.isDirectory() ? "directory" : "file", size: Number(attrs.size) || 0, mtime: Number(attrs.mtime) || 0, mode: Number(attrs.mode) || 0 };

411 } catch (error) {

412

      if (errorCode(error) !== "FILE_PERMISSION_DENIED" || options.sudo !== true) throw fileError(errorCode(error), `${target} is not accessible.`);

413 const value = await this.files.helper(sessionId, { operation: "stat", path: target }, { ...options, sudo: true });

414 if (!value.exists) throw fileError("FILE_NOT_FOUND", `${target} does not exist.`);

415

      return { type: value.type, size: value.baseline.size, mtime: Math.trunc(Number(value.baseline.mtimeNs) / 1e9), mode: value.baseline.mode };

416 }

417 }

418

419 _addDownload(plan, remotePath, localPath, info) {

420 plan.bytes += info.size;

421

    plan.files.push({ direction: "download", name: path.posix.basename(remotePath), remotePath, localPath, size: info.size,

422 remoteIdentity: { size: info.size, mtime: info.mtime }, mode: info.mode & 0o777 });

423 }

424

425 async _walkRemote(sftp, sessionId, plan, remotePath, localPath, options, top = false, depth = 0) {

426

    if (plan.files.length + plan.directories.length > MAX_PLAN_ENTRIES) throw fileError("TRANSFER_TOO_MANY_ENTRIES", `A transfer can contain at most ${MAX_PLAN_ENTRIES} entries.`);

427 let info;

428 if (top) info = await this._remoteStat(sftp, remotePath, options, sessionId);

429 else {

430

      const attrs = await call(sftp, "lstat", remotePath).catch((error) => { throw fileError(errorCode(error), `${remotePath} is not accessible.`); });

431

      if (typeof attrs.isSymbolicLink === "function" && attrs.isSymbolicLink()) { plan.skipped.push({ path: remotePath, reason: "symlink" }); return; }

432

      info = { type: typeof attrs.isDirectory === "function" && attrs.isDirectory() ? "directory" : typeof attrs.isFile === "function" && !attrs.isFile() ? "other" : "file",

433 size: Number(attrs.size) || 0, mtime: Number(attrs.mtime) || 0, mode: Number(attrs.mode) || 0 };

434 }

435 if (info.type === "other") { plan.skipped.push({ path: remotePath, reason: "type" }); return; }

436 if (info.type !== "directory") { this._addDownload(plan, remotePath, localPath, info); return; }

437 if (depth > 64) throw fileError("TRANSFER_TOO_DEEP", "The selected directory tree is too deep.");

438 plan.directories.push(localPath);

439 let names;

440 try { names = (await call(sftp, "readdir", remotePath) || []).map((entry) => String(entry.filename)); }

441 catch (error) {

442

      if (errorCode(error) !== "FILE_PERMISSION_DENIED" || options.sudo !== true) throw fileError(errorCode(error), `Directory ${remotePath} cannot be listed.`);

443

      const listed = await this.files.helper(sessionId, { operation: "listdir", path: remotePath }, { ...options, sudo: true });

444 for (const entry of listed.entries || []) {

445 let name;

446

        try { name = safeEntryName(entry.name); } catch { plan.skipped.push({ path: path.posix.join(remotePath, String(entry.name)), reason: "name" }); continue; }

447 const child = path.posix.join(remotePath, name);

448 if (entry.type === "symlink" || entry.type === "other") plan.skipped.push({ path: child, reason: entry.type });

449

        else if (entry.type === "directory") await this._walkRemote(sftp, sessionId, plan, child, path.join(localPath, name), options, true, depth + 1);

450

        else this._addDownload(plan, child, path.join(localPath, name), { size: entry.size, mtime: entry.mtime, mode: entry.mode });

451 }

452 return;

453 }

454 for (const raw of names) {

455 if (raw === "." || raw === "..") continue;

456 let name;

457

      try { name = safeEntryName(raw); } catch { plan.skipped.push({ path: path.posix.join(remotePath, raw), reason: "name" }); continue; }

458

      await this._walkRemote(sftp, sessionId, plan, path.posix.join(remotePath, name), path.join(localPath, name), options, false, depth + 1);

459 }

460 }

461

462 async _createDirectories(plan, privileged, password) {

463 if (plan.direction === "download") {

464 for (const directory of plan.directories) await fs.promises.mkdir(directory, { recursive: true, mode: 0o700 });

465 return;

466 }

467 if (!plan.directories.length) return;

468 if (privileged) {

469 for (const directory of plan.directories) {

470

        await this.files.helper(plan.sessionId, { operation: "mkdir", path: directory, parents: true }, { sudo: true, sudoPassword: password })

471 .catch((error) => { if (error.code !== "FILE_EXISTS") throw error; });

472 }

473 return;

474 }

475 await this.files._withSftp(plan.sessionId, async (sftp) => {

476 for (const directory of plan.directories) {

477 try { await call(sftp, "mkdir", directory, { mode: 0o755 }); }

478 catch (error) {

479 const existing = await call(sftp, "stat", directory).catch(() => null);

480

          if (!existing || typeof existing.isDirectory !== "function" || !existing.isDirectory()) throw fileError(errorCode(error), `Directory ${directory} could not be created.`);

481 }

482 }

483 });

484 }

485

486 // ---- scheduling ----------------------------------------------------------------------------

487

488 _schedule() {

489 if (this.scheduling || this.closed) return;

490 this.scheduling = true;

491 setImmediate(() => {

492 this.scheduling = false;

493 let running = [...this.runtime.values()].filter((rt) => rt.running).length;

494

      const waiting = [...this.jobs.values()].filter((job) => job.state === "waiting").sort((a, b) => a.order - b.order);

495 for (const job of waiting) {

496 if (running >= this.settings.maxConcurrent) break;

497 running += 1;

498 void this._run(job);

499 }

500 });

501 }

502

503 _rt(job) {

504 let rt = this.runtime.get(job.id);

505

    if (!rt) { rt = { running: false, stop: null, speed: 0, samples: 0, lastBytes: job.transferred, lastTime: this.now() }; this.runtime.set(job.id, rt); }

506 return rt;

507 }

508

509 async _run(job) {

510 const rt = this._rt(job);

511 if (rt.running) return;

512

    Object.assign(rt, { running: true, stop: null, speed: 0, samples: 0, lastBytes: job.transferred, lastTime: this.now(), abort: null, progressSinceResume: 0,

513 generation: (rt.generation || 0) + 1 });

514 job.error = null;

515 job.resumable = false;

516 try {

517 this._set(job, "connecting");

518 const sessionId = this._sessionFor(job);

519 job.sessionId = sessionId;

520 if (job.direction === "upload") await this._runUpload(job, rt, sessionId);

521 else await this._runDownload(job, rt, sessionId);

522 } catch (error) {

523 await this._failure(job, rt, error);

524 } finally {

525 rt.running = false;

526 rt.abort = null;

527 clearTimeout(rt.stopTimer);

528 rt.stopTimer = null;

529 this._persist(true);

530 this._schedule();

531 }

532 }

533

534 async _failure(job, rt, error) {

535 if (error instanceof StopSignal || rt.stop) {

536 const kind = error.stop || rt.stop;

537 if (kind === "cancel") await this._finishCancel(job);

538 else if (kind === "pause") this._set(job, "paused");

539

      else this._set(job, "interrupted", { code: "APP_RESTARTED", message: "The application was closed during the transfer." });

540 return;

541 }

542 const code = errorCode(error);

543 const message = String(error?.message || code).replace(/^[A-Z_]+: /, "").slice(0, 400);

544

    this.logger?.warn("Transfer interrupted.", { jobId: job.id, direction: job.direction, state: job.state, code, transferred: job.transferred, size: job.size });

545 if (job.state === "finalizing" && (code === "FILE_RESULT_UNKNOWN" || CONNECTION_CODES.has(code))) {

546 job.finalize = { ...job.finalize, state: "unknown" };

547 }

548 if (AUTH_CODES.has(code)) {

549 job.authRequired = true;

550 this._set(job, "interrupted", { code, message });

551 return;

552 }

553 if (CONNECTION_CODES.has(code) || code === "FILE_RESULT_UNKNOWN") {

554 this._set(job, "interrupted", { code, message });

555 if (this.settings.autoRetry && job.totalRetries < MAX_TOTAL_RETRIES) this._scheduleRetry(job);

556 return;

557 }

558

    job.resumable = ["FILE_NO_SPACE", "FILE_EXISTS", "FILE_CONFLICT", "LOCAL_NO_SPACE", "ENOSPC", "FILE_READ_ONLY_FILESYSTEM", "FILE_BUSY"].includes(code);

559 this._set(job, "failed", { code: code === "ENOSPC" ? "LOCAL_NO_SPACE" : code, message });

560 }

561

562 _scheduleRetry(job) {

563 this._clearRetry(job);

564 if (job.attempts >= this.retryDelays.length) return; // Manual resume from here on: no endless loop.

565 const delay = this.retryDelays[job.attempts];

566 job.attempts += 1;

567 job.totalRetries += 1;

568 job.nextRetryAt = this.now() + delay;

569 this._set(job, "reconnecting", job.error);

570 const timer = setTimeout(() => {

571 this._rt(job).retryTimer = null;

572 if (job.state !== "reconnecting") return;

573 if (this._findSession(job)) { this._set(job, "waiting"); this._schedule(); }

574 else if (job.attempts < this.retryDelays.length) this._scheduleRetry(job);

575 else this._set(job, "interrupted", job.error);

576 }, delay);

577 timer.unref?.();

578 this._rt(job).retryTimer = timer;

579 }

580

581 _clearRetry(job) {

582 const rt = this.runtime.get(job.id);

583 if (rt?.retryTimer) { clearTimeout(rt.retryTimer); rt.retryTimer = null; }

584 job.nextRetryAt = null;

585 }

586

587 _sessionState(status) {

588 if (status && status.state === "disconnected") {

589 // A dropped SSH connection does not answer outstanding SFTP requests: end those jobs now.

590 for (const job of this.jobs.values()) {

591 const rt = this.runtime.get(job.id);

592 if (!rt?.running || job.sessionId !== status.sessionId) continue;

593 rt.abort?.();

594 rt.kill?.(fileError("SESSION_NOT_CONNECTED", "The SSH connection was lost."));

595 }

596 return;

597 }

598 if (!status || status.state !== "connected") return;

599 const identity = JSON.stringify([status.profileId, status.host, status.port, status.username]);

600 for (const job of this.jobs.values()) {

601

      if (JSON.stringify([job.identity?.profileId, job.identity?.host, job.identity?.port, job.identity?.username]) !== identity) continue;

602

      if (job.state === "reconnecting" || (job.state === "interrupted" && this.settings.autoRetry && CONNECTION_CODES.has(job.error?.code) && job.totalRetries < MAX_TOTAL_RETRIES)) {

603 this._clearRetry(job);

604 this._set(job, "waiting");

605 }

606 }

607 this.cleanedIdentities.delete(identity);

608 this._cleanupOrphansLater();

609 this._schedule();

610 }

611

612 _stop(job, kind) {

613 const rt = this._rt(job);

614 rt.stop = kind;

615 rt.wake?.();

616 // Requests already on the wire normally settle quickly; a dead link is cut after a grace period.

617 // The timer belongs to this run only: a resumed run must never be stopped by an old pause.

618 const generation = rt.generation;

619 clearTimeout(rt.stopTimer);

620 rt.stopTimer = setTimeout(() => {

621 if (rt.generation !== generation || !rt.running) return;

622 rt.abort?.();

623 rt.kill?.(new StopSignal(kind));

624 }, kind === "shutdown" ? 0 : this.stopGraceMs);

625 rt.stopTimer.unref?.();

626 }

627

628 // ---- transfer engine -----------------------------------------------------------------------

629

630 async _openChannel(job, rt, sessionId) {

631 const sftp = await this.sessionManager.openSftp(sessionId);

632 rt.abort = () => { try { sftp.end?.(); } catch {} };

633 if (rt.stop) { rt.abort(); throw new StopSignal(rt.stop); }

634 return sftp;

635 }

636

637 _helperAuth(job, internal = {}) {

638 const privileged = job.mode === "privileged";

639

    return privileged ? [{ sudo: true, sudoPassword: this.secrets.get(job.id) || "" }, { useCachedSudo: true, ...internal }] : [{}, internal];

640 }

641

642 async _runUpload(job, rt, sessionId) {

643

    const stats = await fs.promises.lstat(job.localPath).catch((error) => { throw fileError(error.code === "ENOENT" ? "LOCAL_NOT_FOUND" : "LOCAL_READ_FAILED", `${job.localPath} cannot be read.`); });

644 if (!stats.isFile()) throw fileError("LOCAL_NOT_FILE", `${job.localPath} is no longer a regular file.`);

645 if (!sameIdentity(localIdentity(stats), job.localIdentity)) {

646

      if (job.transferred > 0) throw fileError("FILE_SOURCE_CHANGED", "The local file changed since the transfer started; restart it from the beginning.");

647 job.localIdentity = localIdentity(stats);

648 job.size = stats.size;

649 }

650

    if (job.finalize?.state === "unknown" || (job.hash && job.transferred === job.size && job.partPath)) return this._finalizeUpload(job, rt, sessionId);

651 await this._prepareUploadTarget(job, sessionId);

652 const sftp = await this._openChannel(job, rt, sessionId);

653 let local;

654 let handle;

655 try {

656 local = await fs.promises.open(job.localPath, "r");

657 handle = await this._openRemotePart(sftp, job);

658 if (job.transferred > 0) {

659 const remote = await call(sftp, "fstat", handle);

660 job.transferred = Math.min(job.transferred, Number(remote.size) || 0);

661

        await this._verifyTail(job, (buffer, position) => local.read(buffer, 0, buffer.length, position).then((r) => r.bytesRead),

662 (buffer, position) => this._remoteRead(sftp, handle, buffer, position));

663 }

664 this._set(job, "transferring");

665 await this._pump(job, rt, {

666 produce: async (position, length) => {

667 const buffer = Buffer.allocUnsafe(length);

668 const { bytesRead } = await local.read(buffer, 0, length, position);

669

          if (bytesRead !== length) throw fileError("FILE_SOURCE_CHANGED", "The local file shrank during the transfer.");

670 return buffer;

671 },

672 consume: (buffer, position) => call(sftp, "write", handle, buffer, 0, buffer.length, position)

673 });

674 const remote = await call(sftp, "fstat", handle);

675 if ((Number(remote.size) || 0) > job.size) await call(sftp, "fsetstat", handle, { size: job.size });

676 if (typeof sftp.ext_openssh_fsync === "function") await call(sftp, "ext_openssh_fsync", handle).catch(() => {});

677 await call(sftp, "close", handle).catch(() => {});

678 handle = null;

679

      if (!sameIdentity(localIdentity(await local.stat()), job.localIdentity)) throw fileError("FILE_SOURCE_CHANGED", "The local file changed during the transfer.");

680 this._set(job, "verifying");

681 job.hash = await this._localHash(job.localPath, rt);

682 job.verification = { method: "sha256", local: job.hash };

683 } finally {

684 if (handle) await within(call(sftp, "close", handle));

685 await local?.close().catch(() => {});

686 try { sftp.end?.(); } catch {}

687 rt.abort = null;

688 }

689 return this._finalizeUpload(job, rt, sessionId);

690 }

691

692 async _prepareUploadTarget(job, sessionId) {

693 if (job.resolvedPath && job.partPath && job.targetBaseline !== undefined) return;

694 const [auth, internal] = this._helperAuth(job);

695 let stat;

696 try {

697 stat = await this.files.helper(sessionId, { operation: "stat", path: job.remotePath }, auth, internal);

698 job.helper = true;

699 } catch (error) {

700 if (!NO_HELPER_CODES.has(error.code) || job.mode === "privileged") throw error;

701 // No python3 / Windows server: plain SFTP is enough for new files; existing ones need the

702 // metadata-preserving Linux helper.

703 job.helper = false;

704

      const exists = await this.files._withSftp(sessionId, (sftp) => call(sftp, "lstat", job.remotePath).then(() => true, () => false));

705 if (exists) throw fileError(error.code, error.code === "FEATURE_PLATFORM_UNSUPPORTED"

706

        ? "Replacing an existing file on this server is not supported (no metadata-preserving helper); remove it first or choose another name."

707 : "Replacing an existing file requires python3 on the server to preserve its owner, permissions and ACLs.");

708 stat = { exists: false, resolvedPath: job.remotePath };

709 }

710

    if (stat.exists && stat.type !== "file") throw fileError("FILE_UNSAFE_TYPE_OR_LINK", `${job.remotePath} is not a regular file.`);

711

    if (stat.exists && stat.baseline.nlink > 1) throw fileError("FILE_HARDLINK_UNSUPPORTED", `${job.remotePath} has several hard links.`);

712 if (stat.exists && !job.overwrite) throw fileError("FILE_EXISTS", `${job.remotePath} already exists.`);

713 job.resolvedPath = stat.resolvedPath || job.remotePath;

714 job.targetBaseline = stat.exists ? stat.baseline : null;

715 if (job.mode === "privileged") {

716 const workspace = await this.files.helper(sessionId, { operation: "transfer-workspace" });

717 job.ownerUid = workspace.uid;

718 job.partPath = path.posix.join(workspace.path, `${job.id}.part`);

719 } else {

720 if (job.helper) job.ownerUid = (await this.files.helper(sessionId, { operation: "transfer-workspace" })).uid;

721

      job.partPath = path.posix.join(path.posix.dirname(job.resolvedPath), partName(path.posix.basename(job.resolvedPath), job.id));

722 }

723 this._persist(true);

724 }

725

726 async _openRemotePart(sftp, job) {

727 if (job.transferred > 0) {

728 const attrs = await call(sftp, "lstat", job.partPath).catch(() => null);

729 if (!attrs || (typeof attrs.isFile === "function" && !attrs.isFile())) {

730 job.transferred = 0; // The partial file disappeared; start this file again.

731 } else {

732 return call(sftp, "open", job.partPath, "r+");

733 }

734 }

735 try { return await call(sftp, "open", job.partPath, "wx", { mode: 0o600 }); }

736 catch (error) {

737 // Only our own leftover (same job id) can exist here; remove the entry itself, never follow it.

738

      if (errorCode(error) === "FILE_PERMISSION_DENIED") throw fileError("FILE_PERMISSION_DENIED", `${path.posix.dirname(job.partPath)} is not writable.`);

739 await call(sftp, "unlink", job.partPath).catch(() => {});

740 return call(sftp, "open", job.partPath, "wx", { mode: 0o600 });

741 }

742 }

743

744 async _finalizeUpload(job, rt, sessionId) {

745 if (rt.stop) throw new StopSignal(rt.stop);

746 this._set(job, "finalizing");

747 const retry = job.finalize?.state === "unknown" || job.finalize?.state === "running";

748 job.finalize = { state: "running", startedAt: this.now() };

749 this._persist(true);

750 if (job.helper === false) return this._publishWithoutHelper(job, sessionId);

751

    const [auth, internal] = this._helperAuth(job.finalizeWithSudo ? { ...job, mode: "privileged" } : job, { timeoutMs: 120000 + Math.ceil((job.size / HASH_RATE) * 1000) });

752 const result = await this.files.helper(sessionId, {

753

      operation: "publish", source: job.partPath, uid: job.ownerUid, path: job.remotePath, hash: job.hash, size: job.size,

754

      baseline: job.targetBaseline || null, overwrite: job.overwrite === true, strategy: job.mode === "privileged" ? "copy" : "rename",

755 parents: true, retry, mode: 0o644

756 }, auth, internal);

757

    job.finalize = { state: "done", resolvedPath: result.resolvedPath, created: result.created === true, metadataPreserved: result.metadataPreserved === true,

758 alreadyPublished: result.alreadyPublished === true, finishedAt: this.now() };

759 job.verification = { ...(job.verification || {}), remote: result.hash, method: "sha256" };

760

    if (result.alreadyPublished && job.mode === "privileged") await this.files.helper(sessionId, { operation: "transfer-remove", id: job.id }).catch(() => {});

761 this.secrets.delete(job.id);

762 this._set(job, "completed");

763

    this.logger?.info("Transfer completed.", { jobId: job.id, direction: "upload", bytes: job.size, mode: job.mode, path: job.remotePath });

764 }

765

766 async _publishWithoutHelper(job, sessionId) {

767 const remoteHash = await this.sessionManager.remoteChecksum?.(sessionId, job.partPath).catch(() => null);

768 if (remoteHash && remoteHash !== job.hash) {

769 await this._discardPartial(job).catch(() => {});

770 job.transferred = 0;

771 throw fileError("FILE_VERIFY_FAILED", "The uploaded data does not match the local file.");

772 }

773

    job.verification = { ...(job.verification || {}), remote: remoteHash || null, method: remoteHash ? "sha256" : "size" };

774 await this.files._withSftp(sessionId, async (sftp) => {

775 const attrs = await call(sftp, "stat", job.partPath);

776

      if (Number(attrs.size) !== job.size) throw fileError("FILE_VERIFY_FAILED", "The uploaded size does not match the local file.");

777

      if (await call(sftp, "lstat", job.remotePath).then(() => true, () => false)) throw fileError("FILE_EXISTS", `${job.remotePath} appeared during the transfer.`);

778 await call(sftp, "rename", job.partPath, job.remotePath);

779 });

780

    job.finalize = { state: "done", resolvedPath: job.remotePath, created: true, metadataPreserved: false, finishedAt: this.now() };

781 this._set(job, "completed");

782 }

783

784 async _runDownload(job, rt, sessionId) {

785 const sftp = await this._openChannel(job, rt, sessionId);

786 let handle;

787 let local;

788 try {

789 let source = job.remotePath;

790 if (job.mode === "privileged") source = await this._ensureStaged(job, sftp, sessionId);

791 else {

792

        const attrs = await call(sftp, "stat", job.remotePath).catch((error) => { throw fileError(errorCode(error), `${job.remotePath} is not accessible.`); });

793 const current = { size: Number(attrs.size) || 0, mtime: Number(attrs.mtime) || 0 };

794

        if (job.remoteIdentity && (current.size !== job.remoteIdentity.size || current.mtime !== job.remoteIdentity.mtime)) {

795

          if (job.transferred > 0) throw fileError("FILE_SOURCE_CHANGED", "The remote file changed since the transfer started; restart it from the beginning.");

796 }

797 job.remoteIdentity = current;

798 job.size = current.size;

799 job.remoteMode = (Number(attrs.mode) || 0) & 0o777;

800 }

801 try { handle = await call(sftp, "open", source, "r"); }

802 catch (error) {

803 if (errorCode(error) === "FILE_PERMISSION_DENIED" && job.mode === "privileged-allowed") {

804 job.mode = "privileged";

805 source = await this._ensureStaged(job, sftp, sessionId);

806 handle = await call(sftp, "open", source, "r");

807 } else throw fileError(errorCode(error), `${job.remotePath} cannot be read.`);

808 }

809 job.localPart ||= path.join(path.dirname(job.localPath), partName(path.basename(job.localPath), job.id));

810 local = await this._openLocalPart(job);

811 if (job.transferred > 0) {

812

        await this._verifyTail(job, (buffer, position) => local.read(buffer, 0, buffer.length, position).then((r) => r.bytesRead),

813 (buffer, position) => this._remoteRead(sftp, handle, buffer, position));

814 }

815 this._set(job, "transferring");

816 await this._pump(job, rt, {

817 produce: async (position, length) => {

818 const buffer = Buffer.allocUnsafe(length);

819 const read = await this._remoteRead(sftp, handle, buffer, position);

820 if (read !== length) throw fileError("FILE_SOURCE_CHANGED", "The remote file shrank during the transfer.");

821 return buffer;

822 },

823 consume: (buffer, position) => local.write(buffer, 0, buffer.length, position)

824 });

825 await local.truncate(job.size);

826 await local.sync();

827 await local.close();

828 local = null;

829 await call(sftp, "close", handle).catch(() => {});

830 handle = null;

831 this._set(job, "verifying");

832 job.hash = await this._localHash(job.localPart, rt);

833 job.verification = { method: "size", local: job.hash };

834 if (job.mode === "privileged") job.verification = { method: "sha256", local: job.hash, remote: job.staged.hash };

835 else {

836 const attrs = await call(sftp, "stat", job.remotePath);

837 if (Number(attrs.size) !== job.remoteIdentity.size || Number(attrs.mtime) !== job.remoteIdentity.mtime) {

838

          throw fileError("FILE_SOURCE_CHANGED", "The remote file changed during the transfer; restart it from the beginning.");

839 }

840 if (this.settings.verify) {

841

          const remote = await this.files.helper(sessionId, { operation: "hash", path: job.remotePath }, {}, { timeoutMs: 120000 + Math.ceil((job.size / HASH_RATE) * 1000) })

842

            .then((value) => value.hash, () => this.sessionManager.remoteChecksum?.(sessionId, job.remotePath).catch(() => null));

843 if (remote) job.verification = { method: "sha256", local: job.hash, remote };

844 }

845 }

846 if (job.verification.remote && job.verification.remote !== job.hash) {

847 await this._discardPartial(job).catch(() => {});

848 job.transferred = 0;

849 throw fileError("FILE_VERIFY_FAILED", "The downloaded data does not match the remote file.");

850 }

851 } finally {

852 if (handle) await within(call(sftp, "close", handle));

853 await local?.close().catch(() => {});

854 try { sftp.end?.(); } catch {}

855 rt.abort = null;

856 }

857 if (rt.stop) throw new StopSignal(rt.stop);

858 this._set(job, "finalizing");

859 const exists = await fs.promises.lstat(job.localPath).then(() => true, () => false);

860 if (exists && !job.overwrite) throw fileError("FILE_EXISTS", `${job.localPath} already exists.`);

861

    await fs.promises.chmod(job.localPart, job.mode === "privileged" ? 0o600 : ((job.remoteMode ?? 0o644) & 0o755) | 0o600).catch(() => {});

862 await fs.promises.rename(job.localPart, job.localPath);

863

    if (job.mode === "privileged") await this.files.helper(sessionId, { operation: "transfer-remove", id: job.id }).catch(() => { job.cleanupPending = true; });

864 job.finalize = { state: "done", finishedAt: this.now() };

865 this.secrets.delete(job.id);

866 this._set(job, "completed");

867 this.logger?.info("Transfer completed.", { jobId: job.id, direction: "download", bytes: job.size, mode: job.mode });

868 }

869

870 async _ensureStaged(job, sftp, sessionId) {

871 if (job.staged) {

872 const attrs = await call(sftp, "stat", job.staged.path).catch(() => null);

873 if (attrs && Number(attrs.size) === job.staged.size) return job.staged.path;

874 job.transferred = 0;

875 }

876 this._set(job, "connecting");

877 const workspace = await this.files.helper(sessionId, { operation: "transfer-workspace" });

878

    const [auth, internal] = this._helperAuth({ ...job, mode: "privileged" }, { timeoutMs: 120000 + Math.ceil((job.size / HASH_RATE) * 1000) });

879

    const staged = await this.files.helper(sessionId, { operation: "stage-out", path: job.remotePath, id: job.id, uid: workspace.uid }, auth, internal);

880 job.staged = { path: staged.path, size: staged.size, hash: staged.hash, mtimeNs: staged.mtimeNs };

881 job.size = staged.size;

882 job.transferred = 0;

883 this._persist(true);

884 return staged.path;

885 }

886

887 async _openLocalPart(job) {

888 if (job.transferred > 0) {

889 const stats = await fs.promises.lstat(job.localPart).catch(() => null);

890 if (stats?.isFile()) {

891 job.transferred = Math.min(job.transferred, stats.size);

892 return fs.promises.open(job.localPart, "r+");

893 }

894 job.transferred = 0;

895 }

896 await fs.promises.unlink(job.localPart).catch(() => {});

897 return fs.promises.open(job.localPart, "wx", 0o600);

898 }

899

900 async _remoteRead(sftp, handle, buffer, position) {

901 let total = 0;

902 while (total < buffer.length) {

903 const result = await call(sftp, "read", handle, buffer, total, buffer.length - total, position + total);

904 const read = Array.isArray(result) ? result[0] : result;

905 if (!read) break;

906 total += read;

907 }

908 return total;

909 }

910

911 // Resume only when the last confirmed bytes of the partial file equal the source.

912 async _verifyTail(job, readLocal, readRemote) {

913 const window = Math.min(TAIL_CHECK, job.transferred);

914 if (!window) return;

915 const position = job.transferred - window;

916 const left = Buffer.alloc(window);

917 const right = Buffer.alloc(window);

918 const [leftRead, rightRead] = await Promise.all([readLocal(left, position), readRemote(right, position)]);

919 if (leftRead !== window || rightRead !== window || !left.equals(right)) {

920

      throw fileError("FILE_RESUME_MISMATCH", "The partial data does not match the source; restart the transfer from the beginning.");

921 }

922 }

923

924 _pump(job, rt, { produce, consume }) {

925 return new Promise((resolve, reject) => {

926 let next = job.transferred;

927 let confirmed = job.transferred;

928 const completed = new Map();

929 let inflight = 0;

930 let failure = null;

931 let done = false;

932 let lastProgress = this.now();

933 const finish = (callback) => {

934 if (done) return;

935 done = true;

936 clearInterval(stall);

937 rt.wake = null;

938 rt.kill = null;

939 callback();

940 };

941 // Ends the run even while requests are still outstanding on a dead channel; late replies are ignored.

942 rt.kill = (error) => { if (!failure) failure = error; finish(() => reject(failure)); };

943 const stall = setInterval(() => {

944

        if (this.now() - lastProgress > STALL_MS && !failure) { rt.abort?.(); rt.kill(fileError("TRANSFER_STALLED", "No data was confirmed for 60 seconds.")); }

945 }, 5000);

946 stall.unref?.();

947 const settle = () => {

948 if (inflight || done) return;

949 if (failure) finish(() => reject(failure));

950 else if (rt.stop) finish(() => reject(new StopSignal(rt.stop)));

951 else if (confirmed >= job.size) finish(resolve);

952 };

953 const issue = () => {

954 while (!done && !failure && !rt.stop && inflight < PIPELINE && next < job.size) {

955 const position = next;

956 const length = Math.min(CHUNK, job.size - position);

957 next += length;

958 inflight += 1;

959

          Promise.resolve().then(() => produce(position, length)).then((buffer) => consume(buffer, position)).then(() => {

960 inflight -= 1;

961 if (done) return;

962 completed.set(position, length);

963 while (completed.has(confirmed)) {

964 const size = completed.get(confirmed);

965 completed.delete(confirmed);

966 confirmed += size;

967 }

968 const moved = confirmed - job.transferred;

969 job.transferred = confirmed;

970 job.movedBytes = (job.movedBytes || 0) + moved;

971 rt.progressSinceResume += moved;

972 if (rt.progressSinceResume > 1024 * 1024) job.attempts = 0;

973 lastProgress = this.now();

974 this._progress(job, rt);

975 issue();

976 settle();

977 }, (error) => {

978 inflight -= 1;

979 if (done) return;

980 if (!failure) failure = error;

981 settle();

982 });

983 }

984 settle();

985 };

986 rt.wake = settle;

987 issue();

988 });

989 }

990

991 async _localHash(file, rt) {

992 const digest = crypto.createHash("sha256");

993 const stream = fs.createReadStream(file, { highWaterMark: 1024 * 1024 });

994 rt.abort = () => stream.destroy(new StopSignal(rt.stop || "pause"));

995 for await (const chunk of stream) {

996 if (rt.stop) { stream.destroy(); throw new StopSignal(rt.stop); }

997 digest.update(chunk);

998 }

999 return digest.digest("hex");

1000 }

1001

1002 // ---- cancel / cleanup ------------------------------------------------------------------------

1003

1004 async _finishCancel(job) {

1005 this.secrets.delete(job.id);

1006 try { await this._discardPartial(job); job.cleanupPending = false; }

1007 catch { job.cleanupPending = true; }

1008 this._set(job, "cancelled");

1009 }

1010

1011 // Removes only temporary data this job created; existing destination files are never touched.

1012 async _discardPartial(job) {

1013 if (job.direction === "download") {

1014

      if (job.localPart) await fs.promises.unlink(job.localPart).catch((error) => { if (error.code !== "ENOENT") throw error; });

1015 if (job.staged) {

1016 const sessionId = this._findSession(job);

1017 if (!sessionId) throw fileError("SESSION_NOT_CONNECTED", "Staged data is removed after reconnecting.");

1018 await this.files.helper(sessionId, { operation: "transfer-remove", id: job.id });

1019 job.staged = null;

1020 }

1021 return;

1022 }

1023 if (!job.partPath) return;

1024 const sessionId = this._findSession(job);

1025 if (!sessionId) throw fileError("SESSION_NOT_CONNECTED", "Partial data is removed after reconnecting.");

1026 if (job.mode === "privileged") await this.files.helper(sessionId, { operation: "transfer-remove", id: job.id });

1027 else {

1028 await this.files._withSftp(sessionId, async (sftp) => {

1029 const attrs = await call(sftp, "lstat", job.partPath).catch(() => null);

1030

        if (attrs && !(typeof attrs.isDirectory === "function" && attrs.isDirectory())) await call(sftp, "unlink", job.partPath);

1031 });

1032 }

1033 job.partPath = null;

1034 }

1035

1036 _cleanupOrphansLater() {

1037 if (this.cleanupScheduled) return;

1038 this.cleanupScheduled = true;

1039 setImmediate(() => { this.cleanupScheduled = false; void this._cleanupOrphans().catch(() => {}); });

1040 }

1041

1042 // After a restart: delete temporary transfer data that no resumable job owns any more.

1043 async _cleanupOrphans() {

1044 for (const session of this.sessionManager?.list?.() || []) {

1045 if (session.state !== "connected") continue;

1046 const identity = JSON.stringify([session.profileId, session.host, session.port, session.username]);

1047 if (this.cleanedIdentities.has(identity)) continue;

1048 this.cleanedIdentities.add(identity);

1049

      const mine = [...this.jobs.values()].filter((job) => JSON.stringify([job.identity?.profileId, job.identity?.host, job.identity?.port, job.identity?.username]) === identity);

1050 for (const job of mine.filter((entry) => entry.cleanupPending)) {

1051 await this._discardPartial(job).then(() => { job.cleanupPending = false; this._mark(job); }, () => {});

1052 }

1053 const keep = new Set(mine.filter((job) => !FINAL.has(job.state)).map((job) => job.id));

1054 let listed;

1055 try { listed = await this.files.helper(session.sessionId, { operation: "transfer-list" }); }

1056 catch { continue; }

1057 for (const item of listed.items || []) {

1058 const known = this.jobs.get(item.id);

1059 const abandoned = known ? FINAL.has(known.state) : this.now() - Number(item.mtime) * 1000 > ORPHAN_AGE_MS;

1060 if (keep.has(item.id) || !abandoned) continue;

1061 await this.files.helper(session.sessionId, { operation: "transfer-remove", id: item.id }).catch(() => {});

1062

        this.logger?.info("Abandoned transfer data removed.", { sessionId: session.sessionId, id: item.id, bytes: item.size });

1063 }

1064 this._persist();

1065 }

1066 }

1067

1068 // ---- state, events, persistence ----------------------------------------------------------------

1069

1070 _authorize(job, options) {

1071 if (options.sudo === true || options.sudoPassword) {

1072 // Data is already next to the target: only the metadata-preserving replacement needs root.

1073

      if (job.direction === "upload" && job.mode === "direct" && job.transferred > 0 && job.transferred === job.size) job.finalizeWithSudo = true;

1074 if (job.direction === "download" && job.mode === "direct") job.mode = "privileged-allowed";

1075

      if (job.direction === "upload" && job.mode === "direct" && job.transferred === 0 && AUTH_CODES.has(job.error?.code)) {

1076 job.mode = "privileged";

1077 job.partPath = null;

1078 job.targetBaseline = undefined;

1079 }

1080 }

1081 if (options.sudoPassword) this.secrets.set(job.id, String(options.sudoPassword));

1082 }

1083

1084 _set(job, state, error = null) {

1085 if (!STATES.includes(state)) throw new Error(`Unknown transfer state ${state}`);

1086 const rt = this.runtime.get(job.id);

1087 const now = this.now();

1088

    if (job.state === "transferring" && rt?.activeSince) { job.activeMs = (job.activeMs || 0) + (now - rt.activeSince); rt.activeSince = null; }

1089 if (state === "transferring" && rt) { rt.activeSince = now; rt.lastTime = now; rt.lastBytes = job.transferred; }

1090 job.state = state;

1091 job.error = error;

1092 job.updatedAt = now;

1093 if (state === "transferring" && !job.startedAt) job.startedAt = now;

1094 if (FINAL.has(state)) job.finishedAt = now;

1095 this._mark(job);

1096 this._persist(RUNNING.has(state) ? false : true);

1097 }

1098

1099 _progress(job, rt) {

1100 const now = this.now();

1101 const elapsed = now - rt.lastTime;

1102 if (elapsed >= 500) {

1103 const instant = ((job.transferred - rt.lastBytes) * 1000) / elapsed;

1104 rt.speed = rt.samples ? 0.3 * instant + 0.7 * rt.speed : instant;

1105 rt.samples += 1;

1106 rt.lastTime = now;

1107 rt.lastBytes = job.transferred;

1108 }

1109 job.updatedAt = now;

1110 this._mark(job);

1111 this._persist();

1112 }

1113

1114 _public(job) {

1115 const rt = this.runtime.get(job.id);

1116 const now = this.now();

1117 let speed = 0;

1118 let eta = null;

1119 if (job.state === "transferring" && rt) {

1120 const idle = now - rt.lastTime;

1121 speed = idle > 2000 ? (rt.speed * 2000) / idle : rt.speed;

1122

      if (rt.samples >= 3 && speed > 1 && Number.isFinite(job.size)) eta = Math.ceil(Math.max(0, job.size - job.transferred) / speed);

1123 }

1124 const activeMs = (job.activeMs || 0) + (job.state === "transferring" && rt?.activeSince ? now - rt.activeSince : 0);

1125 const state = job.state;

1126 return {

1127 id: job.id, batchId: job.batchId, direction: job.direction, name: job.name, state,

1128

      source: job.direction === "upload" ? job.localPath : job.remotePath, target: job.direction === "upload" ? job.remotePath : job.localPath,

1129

      resolvedTarget: job.finalize?.resolvedPath || job.resolvedPath || null, mode: job.mode === "privileged" ? "privileged" : "direct",

1130

      size: Number.isFinite(job.size) ? job.size : null, transferred: job.transferred || 0, percent: percent(job.transferred || 0, job.size),

1131

      speed: Math.round(speed), averageSpeed: activeMs > 1000 ? Math.round(((job.movedBytes || 0) * 1000) / activeMs) : null, eta,

1132

      finalize: { state: job.finalize?.state || "pending", metadataPreserved: job.finalize?.metadataPreserved === true, created: job.finalize?.created === true },

1133

      verification: job.verification ? { method: job.verification.method, matched: Boolean(job.verification.remote && job.verification.remote === job.verification.local) } : null,

1134

      error: job.error ? { code: job.error.code, message: job.error.message } : null, authRequired: job.authRequired === true,

1135 attempts: job.attempts || 0, nextRetryAt: job.nextRetryAt || null, cleanupPending: job.cleanupPending === true,

1136 actions: {

1137 pause: ["waiting", "reconnecting", "interrupted", "connecting", "transferring", "verifying"].includes(state),

1138

        resume: ["paused", "interrupted", "reconnecting"].includes(state) || (state === "failed" && job.resumable === true),

1139 retry: ["failed", "interrupted", "paused", "cancelled"].includes(state),

1140 cancel: !FINAL.has(state) && state !== "finalizing",

1141

        overwrite: ["FILE_EXISTS", "FILE_CONFLICT"].includes(job.error?.code) && ["failed", "interrupted"].includes(state),

1142 remove: FINAL.has(state) && !job.cleanupPending

1143 },

1144 createdAt: job.createdAt, updatedAt: job.updatedAt, finishedAt: job.finishedAt || null

1145 };

1146 }

1147

1148 _mark(job) {

1149 this.changed.add(job.id);

1150 this._emitSoon(job.owner);

1151 }

1152

1153 _emitSoon(owner, full = false) {

1154 this.pendingOwners ||= new Set();

1155 this.pendingOwners.add(owner);

1156 if (full) this.fullOwners = new Set([...(this.fullOwners || []), owner]);

1157 if (this.emitTimer) return;

1158 this.emitTimer = setTimeout(() => {

1159 this.emitTimer = null;

1160 const owners = this.pendingOwners;

1161 const full = this.fullOwners || new Set();

1162 this.pendingOwners = new Set();

1163 this.fullOwners = new Set();

1164 const changed = this.changed;

1165 this.changed = new Set();

1166 for (const owner of owners) {

1167 const all = this._owned(owner).map((job) => this._public(job));

1168 const jobs = full.has(owner) ? all : all.filter((job) => changed.has(job.id) || RUNNING.has(job.state));

1169 const removed = [...changed].filter((id) => !this.jobs.has(id));

1170

        this.emit("update", { owner, jobs, removed, full: full.has(owner), summary: summarize(all), settings: { ...this.settings } });

1171 }

1172 }, EMIT_MS);

1173 this.emitTimer.unref?.();

1174 }

1175

1176 _owned(owner) {

1177 return [...this.jobs.values()].filter((job) => job.owner === owner).sort((a, b) => a.order - b.order);

1178 }

1179

1180 _select(owner, ids) {

1181 const owned = this._owned(owner);

1182 if (ids === "all") return owned;

1183 const wanted = new Set((Array.isArray(ids) ? ids : [ids]).map(String));

1184 return owned.filter((job) => wanted.has(job.id));

1185 }

1186

1187 _identity(sessionId) {

1188

    const session = this.sessionManager.list().find((entry) => entry.sessionId === sessionId && entry.state === "connected");

1189 if (!session) throw fileError("SESSION_NOT_CONNECTED", "The SSH session is not connected.");

1190 return { profileId: session.profileId, host: session.host, port: session.port, username: session.username };

1191 }

1192

1193 _findSession(job) {

1194

    const sessions = (this.sessionManager?.list?.() || []).filter((entry) => entry.state === "connected" && entry.profileId === job.identity?.profileId

1195

      && entry.host === job.identity?.host && entry.port === job.identity?.port && entry.username === job.identity?.username);

1196 return (sessions.find((entry) => entry.sessionId === job.sessionId) || sessions[0])?.sessionId || null;

1197 }

1198

1199 _sessionFor(job) {

1200 const sessionId = this._findSession(job);

1201

    if (!sessionId) throw fileError("SESSION_NOT_CONNECTED", `Connect to ${job.identity?.username}@${job.identity?.host} to continue.`);

1202 return sessionId;

1203 }

1204

1205 _trimHistory() {

1206

    const finished = [...this.jobs.values()].filter((job) => FINAL.has(job.state) && !job.cleanupPending).sort((a, b) => (a.finishedAt || 0) - (b.finishedAt || 0));

1207 while (finished.length > MAX_HISTORY) {

1208 const job = finished.shift();

1209 this.jobs.delete(job.id);

1210 this.changed.add(job.id);

1211 }

1212 }

1213

1214 _persist(soon = false, sync = false) {

1215 if (!this.file || this.closed) return;

1216 const write = () => {

1217 this.persistTimer = null;

1218 this._trimHistory();

1219 try {

1220 // Persisted state never contains passwords, keys or runtime handles.

1221 const jobs = [...this.jobs.values()].map(({ ...job }) => job);

1222 atomicWriteJson(this.file, { version: 1, settings: this.settings, order: this.order, jobs });

1223 } catch (error) {

1224 this.logger?.warn("Transfer state could not be saved.", { message: error.message });

1225 }

1226 };

1227 if (sync) { clearTimeout(this.persistTimer); write(); return; }

1228 if (this.persistTimer && !soon) return;

1229 clearTimeout(this.persistTimer);

1230 this.persistTimer = setTimeout(write, soon ? 50 : PERSIST_MS);

1231 this.persistTimer.unref?.();

1232 }

1233

1234 _load() {

1235 if (!this.file) return;

1236 let data;

1237 try { data = JSON.parse(fs.readFileSync(this.file, "utf8")); }

1238 catch { return; }

1239 if (!data || data.version !== 1 || !Array.isArray(data.jobs)) return;

1240

    for (const key of Object.keys(DEFAULT_SETTINGS)) if (data.settings?.[key] !== undefined) this.settings[key] = data.settings[key];

1241

    if (!(Number.isInteger(this.settings.maxConcurrent) && this.settings.maxConcurrent >= 1 && this.settings.maxConcurrent <= 8)) this.settings.maxConcurrent = DEFAULT_SETTINGS.maxConcurrent;

1242 this.order = Number(data.order) || 0;

1243 for (const job of data.jobs.slice(-MAX_JOBS)) {

1244 if (!job || !/^[a-f0-9]{32}$/.test(String(job.id)) || !STATES.includes(job.state)) continue;

1245 // Nothing restarts on its own after the application was closed: the user is offered to resume.

1246 if (RUNNING.has(job.state) || job.state === "waiting" || job.state === "reconnecting") {

1247

        if (job.state === "finalizing" || job.finalize?.state === "running") job.finalize = { ...job.finalize, state: "unknown" };

1248 job.state = "interrupted";

1249 job.error = { code: "APP_RESTARTED", message: "The application was closed during the transfer." };

1250 }

1251 job.nextRetryAt = null;

1252 job.attempts = 0;

1253 this.jobs.set(job.id, job);

1254 this.order = Math.max(this.order, Number(job.order) || 0);

1255 }

1256 }

1257}

1258

1259module.exports = { CHUNK, PIPELINE, STATES, TransferManager, summarize };

1260

SHA-256

15f33c128d0270dea5b8519e21b7187f46a8da73f4b6daf3b6ed863968340c82

CH-J Proprietary Software License 1.14 · Källan finns under CH-J Proprietary Software License 1.14. Dess tillgänglighet ändrar inte licensvillkoren eller ger ytterligare rättigheter.

Arkiv SHA-256

a9605c0870a223b6a8542bb0705edd51d8b04f3069582653417a8397fabcc09a