modularize build and file browser and add more precautions for downloading from file transfer.
This commit is contained in:
@@ -254,6 +254,126 @@ describe("file upload route flow", () => {
|
||||
await removeTempDataDir(dataDir);
|
||||
}
|
||||
});
|
||||
|
||||
test("keeps staged upload payload available for agent range retry after interrupted pull", async () => {
|
||||
const auth = await createAdminToken();
|
||||
const clientId = `client-${Date.now().toString(36)}`;
|
||||
const dataDir = await createTempDataDir();
|
||||
const deps = makeRouteDeps(dataDir);
|
||||
const payload = new Uint8Array(1024 * 1024 + 17);
|
||||
crypto.getRandomValues(payload.subarray(0, Math.min(payload.length, 65536)));
|
||||
|
||||
clientManager.addClient(clientId, {
|
||||
id: clientId,
|
||||
lastSeen: Date.now(),
|
||||
role: "client",
|
||||
ws: { send: () => {} },
|
||||
});
|
||||
|
||||
const prevDisableAgentAuth = process.env.OVERLORD_DISABLE_AGENT_AUTH;
|
||||
process.env.OVERLORD_DISABLE_AGENT_AUTH = "true";
|
||||
|
||||
try {
|
||||
const requestUrl = new URL("https://localhost/api/file/upload/request");
|
||||
const requestRes = await handleFileDownloadRoutes(
|
||||
new Request(requestUrl, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
Authorization: `Bearer ${auth.token}`,
|
||||
},
|
||||
body: JSON.stringify({
|
||||
clientId,
|
||||
path: "C:\\Games\\retry.bin",
|
||||
fileName: "retry.bin",
|
||||
}),
|
||||
}),
|
||||
requestUrl,
|
||||
mockServer,
|
||||
deps,
|
||||
);
|
||||
const requestJson = await requestRes!.json() as any;
|
||||
|
||||
const stageUrl = new URL(`https://localhost${requestJson.uploadUrl}`);
|
||||
const stageRes = await handleFileDownloadRoutes(
|
||||
new Request(stageUrl, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/octet-stream",
|
||||
Authorization: `Bearer ${auth.token}`,
|
||||
},
|
||||
body: payload,
|
||||
}),
|
||||
stageUrl,
|
||||
mockServer,
|
||||
deps,
|
||||
);
|
||||
const stageJson = await stageRes!.json() as any;
|
||||
const pullUrl = new URL(stageJson.pullUrl);
|
||||
|
||||
const firstPullRes = await handleFileDownloadRoutes(
|
||||
new Request(pullUrl, {
|
||||
method: "GET",
|
||||
headers: { "x-overlord-client-id": clientId },
|
||||
}),
|
||||
pullUrl,
|
||||
mockServer,
|
||||
deps,
|
||||
);
|
||||
expect(firstPullRes).not.toBeNull();
|
||||
expect(firstPullRes!.status).toBe(200);
|
||||
const reader = firstPullRes!.body!.getReader();
|
||||
const firstRead = await reader.read();
|
||||
expect(firstRead.done).toBe(false);
|
||||
const firstChunk = firstRead.value!;
|
||||
expect(firstChunk.byteLength).toBeGreaterThan(0);
|
||||
await reader.cancel();
|
||||
|
||||
const retryStart = firstChunk.byteLength;
|
||||
const retryRes = await handleFileDownloadRoutes(
|
||||
new Request(pullUrl, {
|
||||
method: "GET",
|
||||
headers: {
|
||||
"x-overlord-client-id": clientId,
|
||||
Range: `bytes=${retryStart}-`,
|
||||
},
|
||||
}),
|
||||
pullUrl,
|
||||
mockServer,
|
||||
deps,
|
||||
);
|
||||
expect(retryRes).not.toBeNull();
|
||||
expect(retryRes!.status).toBe(206);
|
||||
expect(retryRes!.headers.get("Content-Range")).toBe(
|
||||
`bytes ${retryStart}-${payload.length - 1}/${payload.length}`,
|
||||
);
|
||||
const retried = new Uint8Array(await retryRes!.arrayBuffer());
|
||||
expect(retried.length).toBe(payload.length - retryStart);
|
||||
expect(retried[0]).toBe(payload[retryStart]);
|
||||
expect(retried[retried.length - 1]).toBe(payload[payload.length - 1]);
|
||||
|
||||
const pullAgainRes = await handleFileDownloadRoutes(
|
||||
new Request(pullUrl, {
|
||||
method: "GET",
|
||||
headers: { "x-overlord-client-id": clientId },
|
||||
}),
|
||||
pullUrl,
|
||||
mockServer,
|
||||
deps,
|
||||
);
|
||||
expect(pullAgainRes).not.toBeNull();
|
||||
expect(pullAgainRes!.status).toBe(404);
|
||||
} finally {
|
||||
if (prevDisableAgentAuth === undefined) {
|
||||
delete process.env.OVERLORD_DISABLE_AGENT_AUTH;
|
||||
} else {
|
||||
process.env.OVERLORD_DISABLE_AGENT_AUTH = prevDisableAgentAuth;
|
||||
}
|
||||
clientManager.deleteClient(clientId);
|
||||
deleteUser(auth.userId);
|
||||
await removeTempDataDir(dataDir);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
async function createOperatorToken(opts: { fileBrowserAllowed: boolean; clientScope?: "none" | "all" } = { fileBrowserAllowed: true, clientScope: "all" }) {
|
||||
|
||||
@@ -16,7 +16,7 @@ import {
|
||||
isSafeRemotePath,
|
||||
makeIncrementalPullStream,
|
||||
notifyPullWaiters,
|
||||
streamFileAndDelete,
|
||||
streamFileRangeWithCleanup,
|
||||
uploadIntents,
|
||||
uploadPulls,
|
||||
waitForPullProgress,
|
||||
@@ -557,15 +557,21 @@ export async function handleFileDownloadRoutes(
|
||||
};
|
||||
|
||||
if (pull.deleteFile) {
|
||||
uploadPulls.delete(pullId);
|
||||
clearTimeout(pull.timeout);
|
||||
const cleanupPull = async () => {
|
||||
const current = uploadPulls.get(pullId);
|
||||
if (current === pull) {
|
||||
uploadPulls.delete(pullId);
|
||||
clearTimeout(pull.timeout);
|
||||
}
|
||||
await fs.unlink(pull.tmpPath).catch(() => {});
|
||||
};
|
||||
|
||||
if (pull.state && !rangeHeader) {
|
||||
const expected = pull.expectedTotal ?? pull.size;
|
||||
const stream = makeIncrementalPullStream(pull);
|
||||
const cleanupOnDone = stream.pipeThrough(new TransformStream({
|
||||
flush: async () => {
|
||||
await fs.unlink(pull.tmpPath).catch(() => {});
|
||||
await cleanupPull();
|
||||
},
|
||||
}));
|
||||
return new Response(cleanupOnDone, {
|
||||
@@ -578,13 +584,53 @@ export async function handleFileDownloadRoutes(
|
||||
await waitForPullProgress(pull.state);
|
||||
}
|
||||
if (pull.state.error) {
|
||||
await fs.unlink(pull.tmpPath).catch(() => {});
|
||||
await cleanupPull();
|
||||
return new Response("Upload streaming failed", { status: 500 });
|
||||
}
|
||||
}
|
||||
return new Response(streamFileAndDelete(pull.tmpPath), {
|
||||
headers: { ...baseHeaders, "Content-Length": String(pull.size) },
|
||||
});
|
||||
|
||||
if (rangeHeader) {
|
||||
const match = /^bytes=(\d+)-(\d*)$/.exec(rangeHeader.trim());
|
||||
if (!match) {
|
||||
return new Response("Range Not Satisfiable", {
|
||||
status: 416,
|
||||
headers: { ...baseHeaders, "Content-Range": `bytes */${pull.size}` },
|
||||
});
|
||||
}
|
||||
const start = Number(match[1]);
|
||||
const end = match[2] === "" ? pull.size - 1 : Number(match[2]);
|
||||
if (
|
||||
!Number.isFinite(start) ||
|
||||
!Number.isFinite(end) ||
|
||||
start < 0 ||
|
||||
end >= pull.size ||
|
||||
start > end
|
||||
) {
|
||||
return new Response("Range Not Satisfiable", {
|
||||
status: 416,
|
||||
headers: { ...baseHeaders, "Content-Range": `bytes */${pull.size}` },
|
||||
});
|
||||
}
|
||||
const length = end - start + 1;
|
||||
return new Response(
|
||||
streamFileRangeWithCleanup(pull.tmpPath, start, end, cleanupPull),
|
||||
{
|
||||
status: 206,
|
||||
headers: {
|
||||
...baseHeaders,
|
||||
"Content-Range": `bytes ${start}-${end}/${pull.size}`,
|
||||
"Content-Length": String(length),
|
||||
},
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
return new Response(
|
||||
streamFileRangeWithCleanup(pull.tmpPath, 0, Math.max(0, pull.size - 1), cleanupPull),
|
||||
{
|
||||
headers: { ...baseHeaders, "Content-Length": String(pull.size) },
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
if (rangeHeader) {
|
||||
|
||||
Reference in New Issue
Block a user