perf(sftp): let small downloads skip the per-session fast slot (#3818)

Files that fit one READ fanout window (64 x 32KB = 2MB) read on the shared
browse channel without taking FAST_DOWNLOAD_CHANNELS_PER_SESSION, so folder
downloads of small files run with the configured transfer concurrency
instead of strictly one file at a time. Large files still serialize (#1507).

Co-authored-by: campaign <campaign@local>
Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: 陈大猫 <binaricat@users.noreply.github.com>
This commit is contained in:
authored and GitHub committed 2026-10-21 17:06:07 +08:00
1 parent 1ec954b424
commit 1559739792
4 files changed
+367 -117

No files matched your search

+171 -117
View File
@@ -27,6 +27,7 @@ const { openBoundedSftpChannel } = require("./boundedSftpOpen.cjs");
const {
DOWNLOAD_TRANSFER_CONCURRENCY,
FAST_DOWNLOAD_CHANNELS_PER_SESSION,
SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES,
SMALL_UPLOAD_SHARED_MAX_BYTES,
SFTP_OPEN_TIMEOUT_MS,
SFTP_REQUEST_TIMEOUT_MS,
@@ -1841,6 +1842,9 @@ function markDisposableUploadChannel(channel) {
return channel;
}
// Mutable only for tests that exercise slot/isolated behaviour with tiny files.
let smallDownloadSlotExemptBytes = SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES;
/**
* ssh2 never answers (or fails) a request issued on an SFTP channel that has
* already ended: cleanupRequests ran at close, later requests are dropped.
@@ -1851,6 +1855,19 @@ function isSftpChannelEnded(sftp) {
return Boolean(sftp) && sftp.readable === false;
}
/**
* Shared-channel failure that main used to recover by opening a fresh SFTP
* subsystem. File status (ENOENT / EACCES) is the same on any channel.
*/
function isStaleOrClosedSharedChannelError(error, sftp) {
if (!error || typeof error !== "object") return false;
const code = error.code;
if (code === "ENOENT" || code === "EACCES") return false;
if (isSftpChannelEnded(sftp) || sftp?.destroyed === true) return true;
if (error.sftpChannelClosed || isSftpChannelLossError(error)) return true;
return /channel closed/i.test(String(error.message || ""));
}
async function openIsolatedSftpChannel(client, signal = null) {
// The dedicated one-channel login already holds its only session channel.
// Opening another SFTP subsystem drops that login, so stay on the browse channel.
@@ -5289,123 +5306,139 @@ async function downloadFile(
// One session-wide fast-path slot for isolated + shared READ fanout (#1507).
// Wait (cancelable) instead of degrading to serial createReadStream (#2719).
// Small files (one fanout window) skip the slot and the isolated channel and
// read on the shared browse channel, so folder downloads run in parallel.
// A stale or closed shared channel retries once on a fresh subsystem.
// ENOENT and EACCES do not: a new channel would get the same file status.
const smallDownload = Number(fileSize) > 0
&& Number(fileSize) <= smallDownloadSlotExemptBytes
&& typeof sftp.open === "function"
&& typeof sftp.read === "function";
// Prefer an isolated SFTP channel so cancellation cannot kill the browse session.
// True when this attempt finished the download (success or unhealthy close
// after a complete body). False when the caller should try the next strategy.
async function tryIsolatedDownload() {
const fastSftp = await acquireIsolatedDownloadChannel(client, transfer);
if (transfer.cancelled) {
if (fastSftp) {
releaseIsolatedDownloadChannel(client, fastSftp, { dispose: true });
}
throw new Error("Transfer cancelled");
}
if (fastSftp && (transfer.resumable || typeof fastSftp.fastGet === "function")) {
try {
if (transfer.resumable) {
transfer.downloadStrategy = "concurrent-isolated";
logTransferDiag(transfer, "strategy", {
strategy: "concurrent-isolated",
fields: {
chunk: formatDiagBytes(TRANSFER_CHUNK_SIZE),
concurrency: DOWNLOAD_TRANSFER_CONCURRENCY,
},
});
await downloadFileResumableFast(
remotePath,
localPath,
fastSftp,
fileSize,
transfer,
sendProgress,
);
await finishSuccessfulDownload();
releaseIsolatedDownloadChannel(client, fastSftp);
return true;
}
transfer.downloadStrategy = "fastGet-isolated";
logTransferDiag(transfer, "strategy", { strategy: "fastGet-isolated" });
await new Promise((resolve, reject) => {
let settled = false;
let onFastSftpError = null;
const onFastSftpClose = () => finish(new Error("SFTP channel closed during download"));
const finish = (err) => {
if (settled) return;
settled = true;
if (transfer.abort === abortFastTransfer) {
transfer.abort = null;
}
if (onFastSftpError) {
try { fastSftp.removeListener("error", onFastSftpError); } catch { }
onFastSftpError = null;
}
fastSftp.removeListener("end", onFastSftpClose);
fastSftp.removeListener("close", onFastSftpClose);
releaseIsolatedDownloadChannel(client, fastSftp, {
dispose: !!err || transfer.cancelled,
});
if (transfer.cancelled) reject(new Error("Transfer cancelled"));
else if (err) reject(err);
else resolve();
};
const abortFastTransfer = () => {
if (settled) return;
transfer.cancelled = true;
finish(new Error("Transfer cancelled"));
};
transfer.abort = abortFastTransfer;
onFastSftpError = (err) => finish(err);
fastSftp.once("error", onFastSftpError);
fastSftp.once("end", onFastSftpClose);
fastSftp.once("close", onFastSftpClose);
if (transfer.cancelled) {
finish(new Error("Transfer cancelled"));
return;
}
fastSftp.fastGet(remotePath, localPath, {
chunkSize: TRANSFER_CHUNK_SIZE,
concurrency: DOWNLOAD_TRANSFER_CONCURRENCY,
step: (transferred, _chunk, total) => {
if (transfer.cancelled) return;
sendProgress(transferred, total || fileSize);
},
}, finish);
});
return true;
} catch (err) {
// Always release before rethrowing cancel — otherwise the channel stays
// in pool.busy and the per-session fast-download budget is exhausted.
releaseIsolatedDownloadChannel(client, fastSftp, { dispose: true });
if (transfer.cancelled) throw err;
if (err?.noTransferFallback) throw err;
if (err?.completedWithUnhealthyChannel) {
await finishSuccessfulDownload();
return true;
}
// Concurrent ranges may leave sparse tails past the contiguous
// checkpoint; truncate before trying the next pipelined strategy.
rememberPipelineError(err);
await prepareDownloadFallbackCheckpoint();
console.warn(
"[transferBridge] isolated download failed, trying next pipelined strategy:",
err?.message || String(err),
);
}
} else if (fastSftp) {
// Acquired a channel but cannot use fast path — return it to the pool.
releaseIsolatedDownloadChannel(client, fastSftp);
}
return false;
}
let holdSessionSlot = false;
try {
await acquireSessionFastDownloadSlot(client, transfer);
holdSessionSlot = true;
// Prefer an isolated SFTP channel so cancellation cannot kill the browse session.
if (!client.__netcattySudoMode) {
const fastSftp = await acquireIsolatedDownloadChannel(client, transfer);
if (transfer.cancelled) {
if (fastSftp) {
releaseIsolatedDownloadChannel(client, fastSftp, { dispose: true });
}
throw new Error("Transfer cancelled");
}
if (fastSftp && (transfer.resumable || typeof fastSftp.fastGet === "function")) {
try {
if (transfer.resumable) {
transfer.downloadStrategy = "concurrent-isolated";
logTransferDiag(transfer, "strategy", {
strategy: "concurrent-isolated",
fields: {
chunk: formatDiagBytes(TRANSFER_CHUNK_SIZE),
concurrency: DOWNLOAD_TRANSFER_CONCURRENCY,
},
});
await downloadFileResumableFast(
remotePath,
localPath,
fastSftp,
fileSize,
transfer,
sendProgress,
);
await finishSuccessfulDownload();
releaseIsolatedDownloadChannel(client, fastSftp);
return;
}
transfer.downloadStrategy = "fastGet-isolated";
logTransferDiag(transfer, "strategy", { strategy: "fastGet-isolated" });
await new Promise((resolve, reject) => {
let settled = false;
let onFastSftpError = null;
const onFastSftpClose = () => finish(new Error("SFTP channel closed during download"));
const finish = (err) => {
if (settled) return;
settled = true;
if (transfer.abort === abortFastTransfer) {
transfer.abort = null;
}
if (onFastSftpError) {
try { fastSftp.removeListener("error", onFastSftpError); } catch { }
onFastSftpError = null;
}
fastSftp.removeListener("end", onFastSftpClose);
fastSftp.removeListener("close", onFastSftpClose);
releaseIsolatedDownloadChannel(client, fastSftp, {
dispose: !!err || transfer.cancelled,
});
if (transfer.cancelled) reject(new Error("Transfer cancelled"));
else if (err) reject(err);
else resolve();
};
const abortFastTransfer = () => {
if (settled) return;
transfer.cancelled = true;
finish(new Error("Transfer cancelled"));
};
transfer.abort = abortFastTransfer;
onFastSftpError = (err) => finish(err);
fastSftp.once("error", onFastSftpError);
fastSftp.once("end", onFastSftpClose);
fastSftp.once("close", onFastSftpClose);
if (transfer.cancelled) {
finish(new Error("Transfer cancelled"));
return;
}
fastSftp.fastGet(remotePath, localPath, {
chunkSize: TRANSFER_CHUNK_SIZE,
concurrency: DOWNLOAD_TRANSFER_CONCURRENCY,
step: (transferred, _chunk, total) => {
if (transfer.cancelled) return;
sendProgress(transferred, total || fileSize);
},
}, finish);
});
return;
} catch (err) {
// Always release before rethrowing cancel — otherwise the channel stays
// in pool.busy and the per-session fast-download budget is exhausted.
releaseIsolatedDownloadChannel(client, fastSftp, { dispose: true });
if (transfer.cancelled) throw err;
if (err?.noTransferFallback) throw err;
if (err?.completedWithUnhealthyChannel) {
await finishSuccessfulDownload();
return;
}
// Concurrent ranges may leave sparse tails past the contiguous
// checkpoint; truncate before trying the next pipelined strategy.
rememberPipelineError(err);
await prepareDownloadFallbackCheckpoint();
console.warn(
"[transferBridge] isolated download failed, trying next pipelined strategy:",
err?.message || String(err),
);
}
} else if (fastSftp) {
// Acquired a channel but cannot use fast path — return it to the pool.
releaseIsolatedDownloadChannel(client, fastSftp);
}
if (!smallDownload) {
await acquireSessionFastDownloadSlot(client, transfer);
holdSessionSlot = true;
}
if (!client.__netcattySudoMode && !smallDownload) {
if (await tryIsolatedDownload()) return;
}
// Pipelined READs on the shared browse session — covers sudo (no isolated
// channel) and non-sudo isolated miss/failure. Same session slot already held.
// channel), small files, and non-sudo isolated miss/failure.
if (typeof sftp.open === "function" && typeof sftp.read === "function") {
try {
transfer.downloadStrategy = "concurrent-shared";
@@ -5431,13 +5464,31 @@ async function downloadFile(
return;
} catch (err) {
if (transfer.cancelled) throw err;
if (err?.noTransferFallback) throw err;
// noTransferFallback means "do not keep using this dead channel".
// A stale shared browse channel can still recover on one fresh
// subsystem, which is the path main took before the small-file skip.
const retryOnFreshChannel = smallDownload
&& !client.__netcattySudoMode
&& isStaleOrClosedSharedChannelError(err, sftp);
if (err?.noTransferFallback && !retryOnFreshChannel) throw err;
rememberPipelineError(err);
await prepareDownloadFallbackCheckpoint();
console.warn(
"[transferBridge] concurrent shared download failed (no serial stream fallback):",
err?.message || String(err),
);
if (!retryOnFreshChannel) {
console.warn(
"[transferBridge] concurrent shared download failed (no serial stream fallback):",
err?.message || String(err),
);
} else {
console.warn(
"[transferBridge] shared download channel failed, trying a fresh SFTP channel:",
err?.message || String(err),
);
if (!holdSessionSlot) {
await acquireSessionFastDownloadSlot(client, transfer);
holdSessionSlot = true;
}
if (await tryIsolatedDownload()) return;
}
}
} else if (!lastPipelineError) {
lastPipelineError = new Error(
@@ -7885,4 +7936,7 @@ module.exports = {
_assertLocalDownloadMatchesRemotePrefixForTests: assertLocalDownloadMatchesRemotePrefix,
_hashRemotePrefixForTests: hashRemotePrefix,
_remoteOpenPathMatchesStagedForTests: remoteOpenPathMatchesStaged,
_setSmallDownloadSlotExemptBytesForTests: (bytes) => {
smallDownloadSlotExemptBytes = bytes == null ? SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES : Math.max(0, Number(bytes) || 0);
},
};
+180
View File
@@ -10,11 +10,16 @@ const { EventEmitter } = require("node:events");
const { PassThrough, Readable, Writable } = require("node:stream");
const transferBridge = require("./transferBridge.cjs");
// Most download tests here drive the session slot / isolated pool with tiny
// fixture files. Keep them on that path; the small-file exemption has its own
// test below that restores the production threshold.
transferBridge._setSmallDownloadSlotExemptBytesForTests(0);
const sftpBridge = require("./sftpBridge.cjs");
const tempDirBridge = require("./tempDirBridge.cjs");
const {
DOWNLOAD_TRANSFER_CONCURRENCY,
TRANSFER_CHUNK_SIZE,
SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES,
SMALL_UPLOAD_SHARED_MAX_BYTES,
UPLOAD_TRANSFER_CONCURRENCY,
} = require("./transferLimits.cjs");
@@ -13832,3 +13837,178 @@ test("single-channel pipelined upload stays on the existing SFTP channel", async
assert.equal(secondChannelOpens, 0);
assert.equal(sharedPuts, 1);
});
test("small downloads skip the session fast slot and run in parallel on the shared channel", async (t) => {
transferBridge._setSmallDownloadSlotExemptBytesForTests(null);
t.after(() => transferBridge._setSmallDownloadSlotExemptBytesForTests(0));
const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-small-download-parallel-"));
t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
const payload = Buffer.alloc(3 * TRANSFER_CHUNK_SIZE, 5);
assert.ok(payload.length <= SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES);
let openHandles = 0;
let maxOpenHandles = 0;
let isolatedOpens = 0;
const pendingReads = [];
const shared = createFastSftp({
open(_p, flags, callback) {
assert.equal(flags, "r");
openHandles += 1;
maxOpenHandles = Math.max(maxOpenHandles, openHandles);
callback(null, Buffer.from("h"));
},
read(_h, buffer, offset, length, position, callback) {
// Hold reads until both downloads have opened their handles.
pendingReads.push(() => {
const n = Math.max(0, Math.min(length, payload.length - position));
payload.copy(buffer, offset, position, position + n);
callback(null, n);
});
},
close(_h, callback) { openHandles -= 1; callback(null); },
});
const client = {
sftp: shared,
stat: async () => ({ size: payload.length, isDirectory: false, mtime: 1 }),
client: { sftp(callback) { isolatedOpens += 1; callback(new Error("isolated channel not expected")); } },
};
transferBridge.init({ sftpClients: new Map([["source", client]]) });
const start = (id) => transferBridge.startTransfer({ sender: createSender() }, {
transferId: id,
sourcePath: `/tmp/${id}.bin`,
targetPath: path.join(tempDir, `${id}.bin`),
sourceType: "sftp",
targetType: "local",
sourceSftpId: "source",
totalBytes: payload.length,
skipAdmission: true,
});
const a = start("small-a");
const b = start("small-b");
assert.ok(await waitUntil(() => maxOpenHandles >= 2, 2000), "both small downloads must be in flight together");
assert.equal(transferBridge._getSessionFastDownloadWaiterCountForTests(client), 0);
const drain = setInterval(() => { while (pendingReads.length) pendingReads.shift()(); }, 1);
t.after(() => clearInterval(drain));
const [ra, rb] = await Promise.all([a, b]);
assert.equal(ra.error, undefined, ra.error);
assert.equal(rb.error, undefined, rb.error);
assert.equal(isolatedOpens, 0);
assert.deepEqual(await fs.promises.readFile(path.join(tempDir, "small-a.bin")), payload);
assert.deepEqual(await fs.promises.readFile(path.join(tempDir, "small-b.bin")), payload);
});
test("small download retries once on a fresh channel when the shared channel has ended", async (t) => {
transferBridge._setSmallDownloadSlotExemptBytesForTests(null);
t.after(() => transferBridge._setSmallDownloadSlotExemptBytesForTests(0));
const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-small-download-fresh-"));
t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
const payload = Buffer.from("fresh-channel-bytes");
assert.ok(payload.length <= SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES);
let sharedOpens = 0;
let isolatedOpens = 0;
const shared = createFastSftp({
// Ended browse channel still exposes open/read, so the small-file path
// selects it. #3814 treats readable === false as a dead channel.
readable: false,
open() {
sharedOpens += 1;
throw new Error("ended shared channel must not accept OPEN");
},
read() {
throw new Error("ended shared channel must not accept READ");
},
});
const fresh = createFastSftp({
fastGet(_remotePath, localPath, _options, done) {
fs.promises.writeFile(localPath, payload).then(() => done(), done);
},
});
const client = {
sftp: shared,
stat: async () => ({ size: payload.length, isDirectory: false, mtime: 1 }),
client: {
sftp(callback) {
isolatedOpens += 1;
callback(null, fresh);
},
},
};
transferBridge.init({ sftpClients: new Map([["source", client]]) });
const targetPath = path.join(tempDir, "fresh.bin");
const result = await transferBridge.startTransfer({ sender: createSender() }, {
transferId: "small-stale-shared",
sourcePath: "/tmp/fresh.bin",
targetPath,
sourceType: "sftp",
targetType: "local",
sourceSftpId: "source",
totalBytes: payload.length,
skipAdmission: true,
});
assert.equal(result.error, undefined, result.error);
assert.equal(sharedOpens, 0);
assert.equal(isolatedOpens, 1);
assert.deepEqual(await fs.promises.readFile(targetPath), payload);
});
test("small download does not open a fresh channel for ENOENT or EACCES", async (t) => {
transferBridge._setSmallDownloadSlotExemptBytesForTests(null);
t.after(() => transferBridge._setSmallDownloadSlotExemptBytesForTests(0));
const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-small-download-file-status-"));
t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
let isolatedOpens = 0;
const client = {
sftp: createFastSftp({
open(_remotePath, _flags, callback) {
const error = new Error("No such file");
error.code = "ENOENT";
callback(error);
},
read() {
throw new Error("read must not run after ENOENT");
},
}),
stat: async () => ({ size: 4, isDirectory: false, mtime: 1 }),
client: {
sftp(callback) {
isolatedOpens += 1;
callback(new Error("fresh channel must not open for ENOENT"));
},
},
};
transferBridge.init({ sftpClients: new Map([["source", client]]) });
const missing = await transferBridge.startTransfer({ sender: createSender() }, {
transferId: "small-enoent",
sourcePath: "/tmp/missing.bin",
targetPath: path.join(tempDir, "missing.bin"),
sourceType: "sftp",
targetType: "local",
sourceSftpId: "source",
totalBytes: 4,
skipAdmission: true,
});
assert.match(missing.error || "", /No such file|ENOENT/);
assert.equal(isolatedOpens, 0);
client.sftp = createFastSftp({
open(_remotePath, _flags, callback) {
const error = new Error("Permission denied");
error.code = "EACCES";
callback(error);
},
read() {
throw new Error("read must not run after EACCES");
},
});
const denied = await transferBridge.startTransfer({ sender: createSender() }, {
transferId: "small-eacces",
sourcePath: "/tmp/denied.bin",
targetPath: path.join(tempDir, "denied.bin"),
sourceType: "sftp",
targetType: "local",
sourceSftpId: "source",
totalBytes: 4,
skipAdmission: true,
});
assert.match(denied.error || "", /Permission denied|EACCES/);
assert.equal(isolatedOpens, 0);
});
+8
View File
@@ -28,6 +28,13 @@ const SFTP_REQUEST_TIMEOUT_MS = 30_000;
// channel pool caps - they must never fall back to serial body streams.
const FAST_DOWNLOAD_CHANNELS_PER_SESSION = 1;
// A file that fits in one READ fanout window (64 x 32KB) cannot hog the
// session the way a multi-GB download can, so it skips the per-session slot
// above and reads on the shared browse channel. Without this, folder downloads
// of small files ran strictly one file at a time regardless of the transfer
// concurrency setting.
const SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES = DOWNLOAD_TRANSFER_CONCURRENCY * TRANSFER_CHUNK_SIZE;
// A file that fits in one fanout window finishes in a couple of round trips.
// Opening a dedicated sftp-server channel for it (channel open + subsystem +
// SFTP init) costs more than the body itself on WAN links and spawns one
@@ -37,6 +44,7 @@ const SMALL_UPLOAD_SHARED_MAX_BYTES = UPLOAD_TRANSFER_CONCURRENCY * TRANSFER_CHU
module.exports = {
DOWNLOAD_TRANSFER_CONCURRENCY,
SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES,
SMALL_UPLOAD_SHARED_MAX_BYTES,
FAST_DOWNLOAD_CHANNELS_PER_SESSION,
SFTP_OPEN_TIMEOUT_MS,
+8
View File
@@ -3,6 +3,7 @@ const assert = require("node:assert/strict");
const {
FAST_DOWNLOAD_CHANNELS_PER_SESSION,
SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES,
DOWNLOAD_TRANSFER_CONCURRENCY,
SMALL_UPLOAD_SHARED_MAX_BYTES,
UPLOAD_TRANSFER_CONCURRENCY,
@@ -27,3 +28,10 @@ test("SFTP transfer limits preserve upload safety and the pre-regression downloa
// after #1533, otherwise high-latency proxy paths collapse to KB/s speeds.
assert.equal(DOWNLOAD_TRANSFER_CONCURRENCY * TRANSFER_CHUNK_SIZE, 2 * 1024 * 1024);
});
test("small downloads (one READ window) are exempt from the per-session fast slot", () => {
// Big files still serialize on FAST_DOWNLOAD_CHANNELS_PER_SESSION (#1507).
assert.equal(FAST_DOWNLOAD_CHANNELS_PER_SESSION, 1);
assert.equal(SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES, DOWNLOAD_TRANSFER_CONCURRENCY * TRANSFER_CHUNK_SIZE);
assert.equal(SMALL_DOWNLOAD_SLOT_EXEMPT_BYTES, 2 * 1024 * 1024);
});