mirror of
https://github.com/binaricat/Netcatty.git
synced 2026-11-10 16:21:22 +00:00
[codex] Fix rz upload completion wait (#1531)
* fix rz upload completion wait * Address zmodem timeout recovery review
This commit is contained in:
3 files changed
+264
-12
No files matched your search
@@ -71,7 +71,7 @@ export const ZmodemProgressIndicator: React.FC<ZmodemProgressIndicatorProps> = (
|
||||
/>
|
||||
</div>
|
||||
<div className="text-[10px] opacity-50 mt-0.5">
|
||||
{formatBytes(transferred)} / {formatBytes(total)}
|
||||
{finalizing ? label : `${formatBytes(transferred)} / ${formatBytes(total)}`}
|
||||
</div>
|
||||
</div>
|
||||
<Tooltip>
|
||||
|
||||
@@ -105,6 +105,7 @@ function createZmodemSentry(opts) {
|
||||
let active = false;
|
||||
let currentZSession = null;
|
||||
let _needsDrain = false;
|
||||
let _sawUploadBackpressure = false;
|
||||
const pendingEchoes = [];
|
||||
let pendingTerminalSuppression = null;
|
||||
let cancelInterruptTimer = null;
|
||||
@@ -385,7 +386,10 @@ function createZmodemSentry(opts) {
|
||||
const ok = writeToRemote(Buffer.from(octets));
|
||||
// Track backpressure: if stream.write() returned false, the
|
||||
// kernel TCP buffer is full. The upload loop should pause.
|
||||
if (ok === false) _needsDrain = true;
|
||||
if (ok === false) {
|
||||
_needsDrain = true;
|
||||
_sawUploadBackpressure = true;
|
||||
}
|
||||
},
|
||||
|
||||
on_detect(detection) {
|
||||
@@ -425,6 +429,14 @@ function createZmodemSentry(opts) {
|
||||
getDragDropUpload: () => dragDropUpload,
|
||||
takeDragDropUpload,
|
||||
clearDragDropUpload,
|
||||
hasUploadBackpressure: () => _sawUploadBackpressure,
|
||||
resetUploadBackpressure: () => {
|
||||
_sawUploadBackpressure = false;
|
||||
},
|
||||
onUploadTimeout: () => {
|
||||
ignoreDetectionUntil = Date.now() + 1000;
|
||||
cooldownUntil = Date.now() + COOLDOWN_MS;
|
||||
},
|
||||
waitForDrain: () => {
|
||||
if (!_needsDrain) return Promise.resolve();
|
||||
_needsDrain = false;
|
||||
@@ -629,22 +641,37 @@ function createZmodemSentry(opts) {
|
||||
// Shared helpers (module-level, usable from handleUpload / handleDownload)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const UPLOAD_FILE_END_TIMEOUT_MS = 45000;
|
||||
const UPLOAD_BACKPRESSURE_FILE_END_TIMEOUT_MS = 120000;
|
||||
const UPLOAD_SESSION_CLOSE_TIMEOUT_MS = 15000;
|
||||
|
||||
function resolveTimeoutMs(value, fallback) {
|
||||
return Number.isFinite(value) && value >= 0 ? value : fallback;
|
||||
}
|
||||
|
||||
/**
|
||||
* Race a promise against a timeout. If the promise doesn't settle within
|
||||
* `ms`, resolve with undefined instead of hanging forever. This prevents
|
||||
* zmodem.js internal promises (xfer.end, zsession.close) from blocking
|
||||
* indefinitely after cancel/abort.
|
||||
* `ms`, reject instead of hanging forever. This prevents zmodem.js internal
|
||||
* promises (xfer.end, zsession.close) from blocking indefinitely.
|
||||
*/
|
||||
function withTimeout(promise, ms) {
|
||||
function withTimeout(promise, ms, message = "ZMODEM handshake timeout") {
|
||||
let timer;
|
||||
return Promise.race([
|
||||
promise,
|
||||
Promise.resolve(promise),
|
||||
new Promise((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error("ZMODEM handshake timeout")), ms);
|
||||
timer = setTimeout(() => {
|
||||
const err = new Error(message);
|
||||
err.code = "NETCATTY_ZMODEM_TIMEOUT";
|
||||
reject(err);
|
||||
}, ms);
|
||||
}),
|
||||
]).finally(() => clearTimeout(timer));
|
||||
}
|
||||
|
||||
function isZmodemTimeoutError(err) {
|
||||
return err && err.code === "NETCATTY_ZMODEM_TIMEOUT";
|
||||
}
|
||||
|
||||
/**
|
||||
* Send CAN bytes + delayed Ctrl-C to kill the remote rz/sz process.
|
||||
* Used from dialog-cancel paths that run outside the sentry closure.
|
||||
@@ -656,6 +683,33 @@ function abortRemoteProcess(writeToRemote) {
|
||||
}, 150);
|
||||
}
|
||||
|
||||
function resolveUploadFileEndTimeoutMs(opts) {
|
||||
const normalTimeout = resolveTimeoutMs(
|
||||
opts.uploadFileEndTimeoutMs,
|
||||
UPLOAD_FILE_END_TIMEOUT_MS,
|
||||
);
|
||||
const slowTimeout = resolveTimeoutMs(
|
||||
opts.slowUploadFileEndTimeoutMs,
|
||||
UPLOAD_BACKPRESSURE_FILE_END_TIMEOUT_MS,
|
||||
);
|
||||
|
||||
return opts.hasUploadBackpressure?.()
|
||||
? Math.max(normalTimeout, slowTimeout)
|
||||
: normalTimeout;
|
||||
}
|
||||
|
||||
async function waitForUploadHandshake(promise, ms, message, opts) {
|
||||
try {
|
||||
return await withTimeout(promise, ms, message);
|
||||
} catch (err) {
|
||||
if (isZmodemTimeoutError(err)) {
|
||||
try { opts.onUploadTimeout?.(); } catch { /* ignore */ }
|
||||
abortRemoteProcess(opts.writeToRemote);
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Transfer handlers
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -676,6 +730,10 @@ async function handleUpload(zsession, opts) {
|
||||
const contents = getWebContents();
|
||||
const { BrowserWindow, dialog } = getElectron();
|
||||
const yieldToIO = () => new Promise((resolve) => setImmediate(resolve));
|
||||
const uploadSessionCloseTimeoutMs = resolveTimeoutMs(
|
||||
opts.uploadSessionCloseTimeoutMs,
|
||||
UPLOAD_SESSION_CLOSE_TIMEOUT_MS,
|
||||
);
|
||||
|
||||
const dragDrop = opts.takeDragDropUpload?.() ?? opts.getDragDropUpload?.();
|
||||
let filePaths;
|
||||
@@ -745,6 +803,7 @@ async function handleUpload(zsession, opts) {
|
||||
|
||||
for (let i = 0; i < offers.length; i++) {
|
||||
const { filePath, stat, name } = offers[i];
|
||||
opts.resetUploadBackpressure?.();
|
||||
|
||||
safeSend(contents, "netcatty:zmodem:progress", {
|
||||
sessionId,
|
||||
@@ -818,13 +877,23 @@ async function handleUpload(zsession, opts) {
|
||||
transferType: "upload",
|
||||
finalizing: true,
|
||||
});
|
||||
await withTimeout(xfer.end(), 120000);
|
||||
await waitForUploadHandshake(
|
||||
xfer.end(),
|
||||
resolveUploadFileEndTimeoutMs(opts),
|
||||
`Remote did not confirm receiving ${name}. The upload was stopped so the terminal can recover.`,
|
||||
opts,
|
||||
);
|
||||
} finally {
|
||||
fs.closeSync(fd);
|
||||
}
|
||||
}
|
||||
|
||||
await withTimeout(zsession.close(), 120000);
|
||||
await waitForUploadHandshake(
|
||||
zsession.close(),
|
||||
uploadSessionCloseTimeoutMs,
|
||||
"Remote did not finish the ZMODEM upload session in time. The upload was stopped so the terminal can recover.",
|
||||
opts,
|
||||
);
|
||||
|
||||
// rz re-creates overwritten files with the remote umask, dropping their
|
||||
// original permission bits. Now that everything is on disk, restore them
|
||||
@@ -1034,4 +1103,4 @@ function safeSend(contents, channel, data) {
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { createZmodemSentry, buildUploadPlan, buildModeRestores };
|
||||
module.exports = { createZmodemSentry, buildUploadPlan, buildModeRestores, handleUpload };
|
||||
@@ -3,7 +3,7 @@ const assert = require("node:assert/strict");
|
||||
const fs = require("node:fs");
|
||||
const os = require("node:os");
|
||||
const path = require("node:path");
|
||||
const { createZmodemSentry, buildUploadPlan, buildModeRestores } = require("./zmodemHelper.cjs");
|
||||
const { createZmodemSentry, buildUploadPlan, buildModeRestores, handleUpload } = require("./zmodemHelper.cjs");
|
||||
|
||||
const never = () => { throw new Error("resolver should not be called"); };
|
||||
|
||||
@@ -225,3 +225,186 @@ test("queued drag-drop upload cleans temp files when command write fails", () =>
|
||||
assert.equal(fs.existsSync(secondTempPath), false);
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
test("handleUpload completes when the remote confirms after progress reaches 100 percent", async () => {
|
||||
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-"));
|
||||
const filePath = path.join(tempDir, "upload.txt");
|
||||
fs.writeFileSync(filePath, "payload");
|
||||
const events = [];
|
||||
let closed = false;
|
||||
let endCalled = false;
|
||||
|
||||
const zsession = {
|
||||
async send_offer() {
|
||||
return {
|
||||
send() {},
|
||||
async end() {
|
||||
endCalled = true;
|
||||
},
|
||||
};
|
||||
},
|
||||
async close() {
|
||||
closed = true;
|
||||
},
|
||||
};
|
||||
|
||||
await handleUpload(zsession, {
|
||||
sessionId: "session-1",
|
||||
getWebContents: () => ({
|
||||
isDestroyed: () => false,
|
||||
send: (channel, data) => events.push({ channel, data }),
|
||||
}),
|
||||
takeDragDropUpload: () => ({
|
||||
filePaths: [filePath],
|
||||
remoteNames: ["upload.txt"],
|
||||
}),
|
||||
});
|
||||
|
||||
assert.equal(endCalled, true);
|
||||
assert.equal(closed, true);
|
||||
assert.equal(
|
||||
events.some((event) => event.channel === "netcatty:zmodem:progress" && event.data.finalizing === true),
|
||||
true,
|
||||
);
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
test("handleUpload times out when the remote never confirms after progress reaches 100 percent", async () => {
|
||||
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-"));
|
||||
const filePath = path.join(tempDir, "upload.txt");
|
||||
fs.writeFileSync(filePath, "payload");
|
||||
let releaseEnd;
|
||||
const writes = [];
|
||||
let timeoutNotified = false;
|
||||
|
||||
const zsession = {
|
||||
async send_offer() {
|
||||
return {
|
||||
send() {},
|
||||
end() {
|
||||
return new Promise((resolve) => {
|
||||
releaseEnd = resolve;
|
||||
});
|
||||
},
|
||||
};
|
||||
},
|
||||
async close() {},
|
||||
};
|
||||
|
||||
const uploadPromise = handleUpload(zsession, {
|
||||
sessionId: "session-1",
|
||||
getWebContents: () => null,
|
||||
writeToRemote: (buf) => {
|
||||
writes.push(Buffer.from(buf));
|
||||
return true;
|
||||
},
|
||||
takeDragDropUpload: () => ({
|
||||
filePaths: [filePath],
|
||||
remoteNames: ["upload.txt"],
|
||||
}),
|
||||
uploadFileEndTimeoutMs: 10,
|
||||
uploadSessionCloseTimeoutMs: 10,
|
||||
onUploadTimeout: () => {
|
||||
timeoutNotified = true;
|
||||
},
|
||||
});
|
||||
|
||||
const outcome = await Promise.race([
|
||||
uploadPromise.then(
|
||||
() => "resolved",
|
||||
(err) => String(err.message || err),
|
||||
),
|
||||
new Promise((resolve) => setTimeout(() => resolve("still waiting"), 50)),
|
||||
]);
|
||||
|
||||
assert.match(outcome, /Remote did not confirm receiving upload\.txt/);
|
||||
assert.equal(timeoutNotified, true);
|
||||
assert.deepEqual([...writes[0]], [0x18, 0x18, 0x18, 0x18, 0x18, 0x18, 0x18, 0x18]);
|
||||
releaseEnd();
|
||||
await uploadPromise.catch(() => {});
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
test("handleUpload does not run timeout recovery when the remote rejects the final confirmation", async () => {
|
||||
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-"));
|
||||
const filePath = path.join(tempDir, "upload.txt");
|
||||
fs.writeFileSync(filePath, "payload");
|
||||
const writes = [];
|
||||
let timeoutNotified = false;
|
||||
|
||||
const zsession = {
|
||||
async send_offer() {
|
||||
return {
|
||||
send() {},
|
||||
async end() {
|
||||
throw new Error("remote cancelled upload");
|
||||
},
|
||||
};
|
||||
},
|
||||
async close() {},
|
||||
};
|
||||
|
||||
await assert.rejects(
|
||||
handleUpload(zsession, {
|
||||
sessionId: "session-1",
|
||||
getWebContents: () => null,
|
||||
writeToRemote: (buf) => {
|
||||
writes.push(Buffer.from(buf));
|
||||
return true;
|
||||
},
|
||||
takeDragDropUpload: () => ({
|
||||
filePaths: [filePath],
|
||||
remoteNames: ["upload.txt"],
|
||||
}),
|
||||
uploadFileEndTimeoutMs: 50,
|
||||
uploadSessionCloseTimeoutMs: 50,
|
||||
onUploadTimeout: () => {
|
||||
timeoutNotified = true;
|
||||
},
|
||||
}),
|
||||
/remote cancelled upload/,
|
||||
);
|
||||
|
||||
assert.equal(timeoutNotified, false);
|
||||
assert.equal(writes.length, 0);
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
test("handleUpload allows a longer final wait after upload backpressure", async () => {
|
||||
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-"));
|
||||
const filePath = path.join(tempDir, "upload.txt");
|
||||
fs.writeFileSync(filePath, "payload");
|
||||
let closed = false;
|
||||
|
||||
const zsession = {
|
||||
async send_offer() {
|
||||
return {
|
||||
send() {},
|
||||
end() {
|
||||
return new Promise((resolve) => setTimeout(resolve, 30));
|
||||
},
|
||||
};
|
||||
},
|
||||
async close() {
|
||||
closed = true;
|
||||
},
|
||||
};
|
||||
|
||||
await handleUpload(zsession, {
|
||||
sessionId: "session-1",
|
||||
getWebContents: () => null,
|
||||
writeToRemote: () => true,
|
||||
takeDragDropUpload: () => ({
|
||||
filePaths: [filePath],
|
||||
remoteNames: ["upload.txt"],
|
||||
}),
|
||||
hasUploadBackpressure: () => true,
|
||||
resetUploadBackpressure: () => {},
|
||||
uploadFileEndTimeoutMs: 10,
|
||||
slowUploadFileEndTimeoutMs: 80,
|
||||
uploadSessionCloseTimeoutMs: 10,
|
||||
});
|
||||
|
||||
assert.equal(closed, true);
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
});
|
||||
Reference in new issue
Block a user