From e0e55c74d534732de8781462707aa5cfef99bc73 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 2 Jun 2026 09:47:55 +0000 Subject: [PATCH] fix: retry truncated ssh download blocks --- scripts/src/ssh-file-transfer.ts | 50 +++++++++++++++++----- scripts/ssh-argv-guidance-contract-test.ts | 20 ++++++++- 2 files changed, 58 insertions(+), 12 deletions(-) diff --git a/scripts/src/ssh-file-transfer.ts b/scripts/src/ssh-file-transfer.ts index 1c45e082..e5f97684 100644 --- a/scripts/src/ssh-file-transfer.ts +++ b/scripts/src/ssh-file-transfer.ts @@ -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 { +): Promise { const attemptErrors: Array> = []; 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`); } diff --git a/scripts/ssh-argv-guidance-contract-test.ts b/scripts/ssh-argv-guidance-contract-test.ts index cc878f4f..65c7f206 100644 --- a/scripts/ssh-argv-guidance-contract-test.ts +++ b/scripts/ssh-argv-guidance-contract-test.ts @@ -291,7 +291,7 @@ async function applyPatchV2Fixture(patch: string, files: Record) return { stdout: result.stdout, files: result.files, commands: result.commands }; } -function fileTransferFixture(initial: Record = {}, options: { emptyReadOnce?: Record } = {}): { +function fileTransferFixture(initial: Record = {}, options: { emptyReadOnce?: Record; shortReadOnce?: Record> } = {}): { state: Map; commands: Array<{ operation: string; stdin: boolean }>; executor: SshRemoteCommandExecutor; @@ -300,6 +300,7 @@ function fileTransferFixture(initial: Record = {}, options: { em const state = new Map(Object.entries(initial)); const pending = new Map(); 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 = {}, 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 { 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 }); }