mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: retry truncated ssh download blocks

Browse Source
This commit is contained in:
Codex
2026-06-02 09:47:55 +00:00
parent 49ac11b829
commit e0e55c74d5
2 changed files with 58 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/ssh-file-transfer.ts
+39 -11
View File
@@ -206,9 +206,10 @@ async function readRemoteFileVerified(
const chunks: Buffer[] = [];
let actualBytes = 0;
let chunkCount = 0;
for (let blockIndex = 0; actualBytes < remote.bytes; blockIndex += 1) {
const encoded = await readRemoteBase64BlockWithRetry(invocation, executor, builders, remotePath, readBlockBytes, blockIndex, remote.bytes, actualBytes);
const chunk = encoded.length === 0 ? Buffer.alloc(0) : Buffer.from(encoded, "base64");
for (let blockIndex = 0; blockIndex * readBlockBytes < remote.bytes; blockIndex += 1) {
const offsetBytes = blockIndex * readBlockBytes;
const expectedChunkBytes = Math.min(readBlockBytes, remote.bytes - offsetBytes);
const chunk = await readRemoteBase64BlockWithRetry(invocation, executor, builders, remotePath, readBlockBytes, blockIndex, expectedChunkBytes, remote.bytes, actualBytes);
chunks.push(chunk);
actualBytes += chunk.length;
chunkCount += 1;
@@ -227,42 +228,62 @@ async function readRemoteBase64BlockWithRetry(
remotePath: string,
readBlockBytes: number,
blockIndex: number,
expectedChunkBytes: number,
expectedBytes: number,
actualBytes: number,
): Promise<string> {
): Promise<Buffer> {
const attemptErrors: Array<Record<string, unknown>> = [];
for (let attempt = 1; attempt <= fileTransferReadBlockMaxAttempts; attempt += 1) {
try {
const read = await checkedFileTransfer(invocation, executor, builders, "read-b64-block", [remotePath, String(blockIndex), String(readBlockBytes)]);
const encoded = read.stdout.replace(/\s+/gu, "");
if (encoded.length > 0) return encoded;
attemptErrors.push({ attempt, exitCode: read.exitCode, stdoutBytes: read.stdout.length, stderrTail: read.stderr.slice(-500) });
const chunk = decodeRemoteBase64Block(encoded);
if (chunk !== null && chunk.length === expectedChunkBytes) return chunk;
const reason = chunk === null ? "invalid-base64" : chunk.length === 0 ? "empty-read" : "short-read";
attemptErrors.push({
attempt,
reason,
exitCode: read.exitCode,
stdoutBytes: read.stdout.length,
encodedBytes: encoded.length,
decodedBytes: chunk?.length ?? null,
expectedChunkBytes,
stderrTail: read.stderr.slice(-500),
});
if (attempt < fileTransferReadBlockMaxAttempts) {
emitEmptyReadRetryProgress(invocation, remotePath, blockIndex, attempt, expectedBytes, actualBytes);
emitReadRetryProgress(invocation, remotePath, blockIndex, attempt, expectedBytes, actualBytes, expectedChunkBytes, chunk?.length ?? null, reason);
await delayMs(fileTransferReadEmptyRetryDelayMs);
}
} catch (error) {
attemptErrors.push({
attempt,
error: error instanceof Error ? error.message : String(error),
expectedChunkBytes,
});
if (attempt < fileTransferReadBlockMaxAttempts) {
emitEmptyReadRetryProgress(invocation, remotePath, blockIndex, attempt, expectedBytes, actualBytes);
emitReadRetryProgress(invocation, remotePath, blockIndex, attempt, expectedBytes, actualBytes, expectedChunkBytes, null, "remote-error");
await delayMs(fileTransferReadEmptyRetryDelayMs);
}
}
}
throw new SshFileTransferError("remote download returned an empty block before EOF after retries", {
throw new SshFileTransferError("remote download returned an invalid block before EOF after retries", {
route: invocation.route.raw,
remotePath,
blockIndex,
attempts: fileTransferReadBlockMaxAttempts,
expectedBytes,
actualBytes,
expectedChunkBytes,
attemptErrors,
});
}
function decodeRemoteBase64Block(encoded: string): Buffer | null {
if (encoded.length === 0) return Buffer.alloc(0);
if (!/^[A-Za-z0-9+/]*={0,2}$/u.test(encoded) || encoded.length % 4 !== 0) return null;
return Buffer.from(encoded, "base64");
}
function emitDownloadProgress(
invocation: ParsedSshInvocation,
remotePath: string,
@@ -286,24 +307,31 @@ function emitDownloadProgress(
})}\n`);
}
function emitEmptyReadRetryProgress(
function emitReadRetryProgress(
invocation: ParsedSshInvocation,
remotePath: string,
blockIndex: number,
attempt: number,
expectedBytes: number,
actualBytes: number,
expectedChunkBytes: number,
actualChunkBytes: number | null,
reason: string,
): void {
const event = reason === "empty-read" ? "unidesk.ssh.download.empty-read-retry" : "unidesk.ssh.download.short-read-retry";
process.stderr.write(`${JSON.stringify({
event: "unidesk.ssh.download.empty-read-retry",
event,
route: invocation.route.raw,
providerId: invocation.providerId,
remotePath,
blockIndex,
attempt,
maxAttempts: fileTransferReadBlockMaxAttempts,
reason,
actualBytes,
expectedBytes,
actualChunkBytes,
expectedChunkBytes,
})}\n`);
}
scripts/ssh-argv-guidance-contract-test.ts
+19 -1
View File
@@ -291,7 +291,7 @@ async function applyPatchV2Fixture(patch: string, files: Record<string, string>)
return { stdout: result.stdout, files: result.files, commands: result.commands };
}
function fileTransferFixture(initial: Record<string, Buffer> = {}, options: { emptyReadOnce?: Record<string, number[]> } = {}): {
function fileTransferFixture(initial: Record<string, Buffer> = {}, options: { emptyReadOnce?: Record<string, number[]>; shortReadOnce?: Record<string, Record<string, number>> } = {}): {
state: Map<string, Buffer>;
commands: Array<{ operation: string; stdin: boolean }>;
executor: SshRemoteCommandExecutor;
@@ -300,6 +300,7 @@ function fileTransferFixture(initial: Record<string, Buffer> = {}, options: { em
const state = new Map(Object.entries(initial));
const pending = new Map<string, string>();
const emptyReadOnce = new Map(Object.entries(options.emptyReadOnce ?? {}).map(([target, blocks]) => [target, new Set(blocks)]));
const shortReadOnce = new Map(Object.entries(options.shortReadOnce ?? {}).map(([target, blocks]) => [target, new Map(Object.entries(blocks).map(([block, bytes]) => [Number(block), bytes]))]));
const commands: Array<{ operation: string; stdin: boolean }> = [];
const builders: SshFileTransferCommandBuilders = {
buildRouteCommand(route, command, options) {
@@ -332,6 +333,12 @@ function fileTransferFixture(initial: Record<string, Buffer> = {}, options: { em
emptyBlocks.delete(blockIndex);
return { exitCode: 0, stdout: "", stderr: "" };
}
const shortBlocks = shortReadOnce.get(target);
const shortBytes = shortBlocks?.get(blockIndex);
if (shortBytes !== undefined) {
shortBlocks?.delete(blockIndex);
return { exitCode: 0, stdout: content.subarray(start, start + Math.max(0, shortBytes)).toString("base64"), stderr: "" };
}
return { exitCode: 0, stdout: content.subarray(start, start + blockSize).toString("base64"), stderr: "" };
}
if (operation === "write-b64-argv" || operation === "write-b64-stdin") {
@@ -473,6 +480,17 @@ export async function runSshArgvGuidanceContract(): Promise<JsonRecord> {
assertCondition(retryReadBlocks.length === Number(retryJson.transfer && typeof retryJson.transfer === "object" ? (retryJson.transfer as JsonRecord).chunks : 0) + 1, "transient empty block should add exactly one repeated read without counting as a chunk", retryTransfer.commands);
assertCondition(retryResult.stderr.includes("unidesk.ssh.download.progress") && retryResult.stderr.includes("unidesk.ssh.download.empty-read-retry"), "download should emit bounded progress and retry events to stderr", retryResult.stderr);
assertCondition(readFileSync(retryDownload).equals(retryPayload), "retry download must preserve complete content after transient empty block", { commands: retryTransfer.commands });
const shortReadDownload = path.join(transferRoot, "downloaded", "short-read-copy.bin");
const shortReadPayload = Buffer.from("fedcba9876543210".repeat(4096), "utf8");
const shortReadTransfer = fileTransferFixture({ "/tmp/short-read-remote.bin": shortReadPayload }, { shortReadOnce: { "/tmp/short-read-remote.bin": { 2: 512 } } });
const shortReadResult = await captureStdout(() => runSshFileTransferOperation(parseSshInvocation("D601", ["download", "--chunk-bytes", "1024", "/tmp/short-read-remote.bin", shortReadDownload]), ["download", "--chunk-bytes", "1024", "/tmp/short-read-remote.bin", shortReadDownload], shortReadTransfer.executor, shortReadTransfer.builders));
const shortReadJson = JSON.parse(shortReadResult.stdout) as JsonRecord;
const shortReadBlocks = shortReadTransfer.commands.filter((item) => item.operation === "read-b64-block");
assertCondition(shortReadResult.exitCode === 0 && shortReadJson.sha256 === sha256BufferHex(shortReadPayload), "download should retry a truncated block and keep sha256 verification", shortReadResult);
assertCondition(shortReadBlocks.length === Number(shortReadJson.transfer && typeof shortReadJson.transfer === "object" ? (shortReadJson.transfer as JsonRecord).chunks : 0) + 1, "short block should add exactly one repeated read without counting as a chunk", shortReadTransfer.commands);
assertCondition(shortReadResult.stderr.includes("unidesk.ssh.download.short-read-retry"), "download should emit short-read retry events to stderr", shortReadResult.stderr);
assertCondition(readFileSync(shortReadDownload).equals(shortReadPayload), "short-read retry download must preserve complete content", { commands: shortReadTransfer.commands });
} finally {
rmSync(transferRoot, { recursive: true, force: true });
}