diff --git a/test/e2e/docs/jetson-dispatch.md b/test/e2e/docs/jetson-dispatch.md index b109ce8b30..7e52068030 100644 --- a/test/e2e/docs/jetson-dispatch.md +++ b/test/e2e/docs/jetson-dispatch.md @@ -139,8 +139,12 @@ workflow run ID and attempt, selectors, event, and hardware opt-in decisions. The Jetson controller writes private files under the target artifact directory: -- `jetson-dispatch.json` contains the validated completed status and bounded - log. It excludes the base64 archive payload. +- `jetson-dispatch.json` records the validated request and derived job ID before + submission begins. It records the cancellation reason and final outcome. If + cancellation reports that the job is absent after submission may have reached + the dispatcher, the controller records one follow-up request. A completed + artifact replaces this recovery state with the validated status and bounded + log. The file excludes the base64 archive payload. - `jetson-e2e-artifacts.tar.gz` contains the decoded target evidence when the service returns an archive. @@ -149,6 +153,13 @@ failure. A successful proof requires the exact candidate request, a conclusion of `success`, `cleanup: "succeeded"`, a device identity, and the artifact archive. +If a workflow fails after submission begins, inspect `jetson-dispatch.json` +before another dispatch. Use its job ID to inspect the operator-service job, +even when the receipt has no `cancellation` record. If artifact upload failed +and the file is unavailable, use the job ID from the workflow error or logs. +Cancel the job or confirm completion before another dispatch, regardless of +whether the cancellation outcome is absent, pending, succeeded, or failed. + ## Live Target `test/e2e/live/jetson-nvmap-gpu.test.ts` runs the Jetson hardware target for the diff --git a/test/e2e/support/jetson-dispatch-client.test.ts b/test/e2e/support/jetson-dispatch-client.test.ts index 5a0e85b05b..aa53c4c47a 100644 --- a/test/e2e/support/jetson-dispatch-client.test.ts +++ b/test/e2e/support/jetson-dispatch-client.test.ts @@ -3,15 +3,18 @@ import { createHash } from "node:crypto"; import fs from "node:fs"; +import os from "node:os"; import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import { createGitHubOidcTokenProvider, + createJetsonCancellation, dispatcherBaseUrl, dispatcherRequest, jetsonDispatchRequestFromEnvironment, pollJetsonDispatch, + submitJetsonDispatch, } from "../../../tools/e2e/jetson-dispatch-client.mts"; import { JETSON_DISPATCH_AUDIENCE, @@ -62,9 +65,23 @@ const invalidV2RequestErrors: Record = { "noncanonical managed-image revision": "managedImageRevision must be a lowercase 40-character commit SHA", }; +const temporaryDirectories: string[] = []; + +function temporaryReceiptFile(): string { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), "nemoclaw-jetson-dispatch-")); + temporaryDirectories.push(directory); + return path.join(directory, "jetson-dispatch.json"); +} + +function readReceipt(receiptFile: string): Record { + return JSON.parse(fs.readFileSync(receiptFile, "utf8")) as Record; +} afterEach(() => { vi.restoreAllMocks(); + for (const directory of temporaryDirectories.splice(0)) { + fs.rmSync(directory, { recursive: true }); + } }); describe("Jetson dispatch static HTTP contract", () => { @@ -421,6 +438,7 @@ describe("Jetson dispatch GitHub controller", () => { it("retries status transport failures and validates the completed status (#8142)", async () => { vi.spyOn(console, "warn").mockImplementation(() => {}); + const receiptFile = temporaryReceiptFile(); const requestImpl = vi .fn() .mockRejectedValueOnce(new Error("transport reset")) @@ -431,12 +449,68 @@ describe("Jetson dispatch GitHub controller", () => { baseUrl: new URL("https://dispatch.test/"), deadlineMs: 10_000, initialStatus: queuedStatus, - jobId: queuedStatus.jobId, now: () => 0, + receiptFile, request: requestImpl, wait: async () => {}, }), ).resolves.toEqual(completedStatus); + expect(readReceipt(receiptFile)).toEqual({ + schemaVersion: 1, + jobId: queuedStatus.jobId, + request: queuedStatus.request, + }); + }); + + it("cancels an accepted job when stopping was requested during submission (#8142)", async () => { + const receiptFile = temporaryReceiptFile(); + const requestImpl = vi.fn(async () => ({ job: queuedStatus })); + + await expect( + pollJetsonDispatch({ + baseUrl: new URL("https://dispatch.test/"), + deadlineMs: 10_000, + initialStatus: queuedStatus, + now: () => 0, + receiptFile, + request: requestImpl, + stopping: () => true, + wait: async () => {}, + }), + ).rejects.toThrow( + `Jetson dispatch ${queuedStatus.jobId} cancellation requested; cancellation request succeeded`, + ); + expect(requestImpl).toHaveBeenCalledOnce(); + expect(requestImpl).toHaveBeenCalledWith( + expect.objectContaining({ method: "DELETE", path: `v1/jobs/${queuedStatus.jobId}` }), + ); + expect(readReceipt(receiptFile)).toMatchObject({ + cancellation: { outcome: "succeeded", reason: "signal" }, + }); + }); + + it("cancels an accepted job when the initial receipt cannot be written (#8142)", async () => { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), "nemoclaw-jetson-dispatch-")); + temporaryDirectories.push(directory); + const requestImpl = vi.fn(async () => ({ job: queuedStatus })); + + await expect( + pollJetsonDispatch({ + baseUrl: new URL("https://dispatch.test/"), + deadlineMs: 10_000, + initialStatus: queuedStatus, + now: () => 0, + receiptFile: directory, + request: requestImpl, + wait: async () => {}, + }), + ).rejects.toThrow( + `Jetson dispatch ${queuedStatus.jobId} was accepted but its recovery receipt could not be written; cancellation request succeeded; recovery receipt update failed`, + ); + expect(requestImpl).toHaveBeenCalledOnce(); + expect(requestImpl).toHaveBeenCalledWith( + expect.objectContaining({ method: "DELETE", path: `v1/jobs/${queuedStatus.jobId}` }), + ); }); it("requests cancellation when the controller deadline expires (#8142)", async () => { @@ -447,17 +521,337 @@ describe("Jetson dispatch GitHub controller", () => { baseUrl: new URL("https://dispatch.test/"), deadlineMs: 10_000, initialStatus: queuedStatus, - jobId: queuedStatus.jobId, now: () => 10_000, + receiptFile: temporaryReceiptFile(), request: requestImpl, wait: async () => {}, }), - ).rejects.toThrow("Jetson dispatcher did not complete before the controller deadline"); + ).rejects.toThrow( + `Jetson dispatch ${queuedStatus.jobId} did not complete before the controller deadline; cancellation request succeeded`, + ); expect(requestImpl).toHaveBeenCalledWith( expect.objectContaining({ method: "DELETE", path: `v1/jobs/${queuedStatus.jobId}` }), ); }); + it("records a rejected deadline cancellation for operator recovery (#8142)", async () => { + const receiptFile = temporaryReceiptFile(); + const requestImpl = vi.fn(async () => { + throw new Error("untrusted cancellation response text"); + }); + + await expect( + pollJetsonDispatch({ + baseUrl: new URL("https://dispatch.test/"), + deadlineMs: 10_000, + initialStatus: queuedStatus, + now: () => 10_000, + receiptFile, + request: requestImpl, + wait: async () => {}, + }), + ).rejects.toThrow( + `Jetson dispatch ${queuedStatus.jobId} did not complete before the controller deadline; cancellation request failed (transport-error)`, + ); + expect(readReceipt(receiptFile)).toEqual({ + schemaVersion: 1, + jobId: queuedStatus.jobId, + request: queuedStatus.request, + cancellation: { + outcome: "failed", + reason: "controller-deadline", + failure: "transport-error", + }, + }); + expect(fs.statSync(receiptFile).mode & 0o777).toBe(0o600); + }); + + it("classifies a non-object cancellation error as an invalid response (#8142)", async () => { + const receiptFile = temporaryReceiptFile(); + const requestImpl = vi.fn(async () => { + throw new Error("Jetson dispatcher error must be an object"); + }); + + await expect( + pollJetsonDispatch({ + baseUrl: new URL("https://dispatch.test/"), + deadlineMs: 10_000, + initialStatus: queuedStatus, + now: () => 10_000, + receiptFile, + request: requestImpl, + wait: async () => {}, + }), + ).rejects.toThrow("cancellation request failed (invalid-response)"); + expect(readReceipt(receiptFile)).toMatchObject({ + cancellation: { failure: "invalid-response" }, + }); + }); + + it("accepts one empty successful response for concurrent cancellation callers (#8142)", async () => { + const receiptFile = temporaryReceiptFile(); + const fetchImpl = vi.fn(async () => new Response(null, { status: 204 })); + const requestImpl: typeof dispatcherRequest = async (options) => + dispatcherRequest({ + ...options, + fetchImpl, + tokenProvider: async () => "oidc-token", + }); + const cancel = createJetsonCancellation({ + baseUrl: new URL("https://dispatch.test/"), + dispatch: queuedStatus, + receiptFile, + request: requestImpl, + }); + + const deadlineCancellation = cancel("controller-deadline"); + const signalCancellation = cancel("signal"); + + await expect(Promise.all([deadlineCancellation, signalCancellation])).resolves.toEqual([ + { outcome: "succeeded", receiptWritten: true }, + { outcome: "succeeded", receiptWritten: true }, + ]); + expect(fetchImpl).toHaveBeenCalledOnce(); + expect(fetchImpl).toHaveBeenCalledWith( + new URL(`https://dispatch.test/v1/jobs/${queuedStatus.jobId}`), + expect.objectContaining({ method: "DELETE" }), + ); + expect(readReceipt(receiptFile)).toMatchObject({ + cancellation: { outcome: "succeeded", reason: "controller-deadline" }, + }); + }); + + it("records and cancels a job when submission times out after acceptance (#8142)", async () => { + const receiptFile = temporaryReceiptFile(); + const requestImpl = vi + .fn() + .mockImplementationOnce(async () => { + expect(readReceipt(receiptFile)).toEqual({ + schemaVersion: 1, + jobId: queuedStatusV2.jobId, + request: requestV2, + }); + throw Object.assign(new Error("submission response timed out"), { name: "TimeoutError" }); + }) + .mockResolvedValueOnce(undefined); + + await expect( + submitJetsonDispatch({ + baseUrl: new URL("https://dispatch.test/"), + dispatchRequest: requestV2, + receiptFile, + request: requestImpl, + }), + ).rejects.toThrow( + `Jetson dispatch ${queuedStatusV2.jobId} submission outcome was not confirmed; cancellation request succeeded`, + ); + expect(requestImpl).toHaveBeenCalledTimes(2); + expect(requestImpl).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ method: "POST", path: "v1/jobs", body: requestV2 }), + ); + expect(requestImpl).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ method: "DELETE", path: `v1/jobs/${queuedStatusV2.jobId}` }), + ); + expect(readReceipt(receiptFile)).toEqual({ + schemaVersion: 1, + jobId: queuedStatusV2.jobId, + request: requestV2, + cancellation: { outcome: "succeeded", reason: "submission-outcome-unknown" }, + }); + }); + + it("retries a missing job cancellation after an unconfirmed submission (#8142)", async () => { + const receiptFile = temporaryReceiptFile(); + const requestImpl = vi + .fn() + .mockRejectedValueOnce( + Object.assign(new Error("submission response timed out"), { name: "TimeoutError" }), + ) + .mockRejectedValueOnce(new Error("Jetson dispatcher returned HTTP 404: request failed")) + .mockResolvedValueOnce(undefined); + + await expect( + submitJetsonDispatch({ + baseUrl: new URL("https://dispatch.test/"), + dispatchRequest: requestV2, + receiptFile, + request: requestImpl, + }), + ).rejects.toThrow( + `Jetson dispatch ${queuedStatusV2.jobId} submission outcome was not confirmed; cancellation request succeeded`, + ); + expect(requestImpl).toHaveBeenCalledTimes(3); + expect(requestImpl).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ method: "POST", path: "v1/jobs", body: requestV2 }), + ); + expect(requestImpl).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ method: "DELETE", path: `v1/jobs/${queuedStatusV2.jobId}` }), + ); + expect(requestImpl).toHaveBeenNthCalledWith( + 3, + expect.objectContaining({ method: "DELETE", path: `v1/jobs/${queuedStatusV2.jobId}` }), + ); + expect(readReceipt(receiptFile)).toEqual({ + schemaVersion: 1, + jobId: queuedStatusV2.jobId, + request: requestV2, + cancellation: { outcome: "succeeded", reason: "submission-outcome-unknown" }, + }); + }); + + it("cancels an accepted job when a signal arrives before the submission response (#8142)", async () => { + const receiptFile = temporaryReceiptFile(); + let stopping = false; + let markPostStarted!: () => void; + const postStarted = new Promise((resolve) => { + markPostStarted = resolve; + }); + let releasePost!: () => void; + const postRelease = new Promise((resolve) => { + releasePost = resolve; + }); + const requestImpl = vi + .fn() + .mockImplementationOnce(async () => { + markPostStarted(); + await postRelease; + return { job: queuedStatusV2 }; + }) + .mockResolvedValueOnce(undefined); + const cancel = createJetsonCancellation({ + baseUrl: new URL("https://dispatch.test/"), + dispatch: queuedStatusV2, + receiptFile, + request: requestImpl, + }); + + const submission = submitJetsonDispatch({ + baseUrl: new URL("https://dispatch.test/"), + cancel, + dispatchRequest: requestV2, + receiptFile, + request: requestImpl, + stopping: () => stopping, + }); + await postStarted; + stopping = true; + await expect(cancel("signal")).resolves.toEqual({ + outcome: "succeeded", + receiptWritten: true, + }); + expect(requestImpl).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ method: "DELETE", path: `v1/jobs/${queuedStatusV2.jobId}` }), + ); + releasePost(); + + await expect(submission).rejects.toThrow( + `Jetson dispatch ${queuedStatusV2.jobId} cancellation requested; cancellation request succeeded`, + ); + expect(requestImpl).toHaveBeenCalledTimes(2); + expect(requestImpl).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ method: "DELETE", path: `v1/jobs/${queuedStatusV2.jobId}` }), + ); + expect(readReceipt(receiptFile)).toMatchObject({ + cancellation: { outcome: "succeeded", reason: "signal" }, + }); + }); + + it("retries an early job-not-found cancellation after submission settles (#8142)", async () => { + const receiptFile = temporaryReceiptFile(); + let stopping = false; + let markPostStarted!: () => void; + const postStarted = new Promise((resolve) => { + markPostStarted = resolve; + }); + let releasePost!: () => void; + const postRelease = new Promise((resolve) => { + releasePost = resolve; + }); + const requestImpl = vi + .fn() + .mockImplementationOnce(async () => { + markPostStarted(); + await postRelease; + return { job: queuedStatusV2 }; + }) + .mockRejectedValueOnce(new Error("Jetson dispatcher returned HTTP 404: request failed")) + .mockResolvedValueOnce(undefined); + const cancel = createJetsonCancellation({ + baseUrl: new URL("https://dispatch.test/"), + dispatch: queuedStatusV2, + receiptFile, + request: requestImpl, + }); + + const submission = submitJetsonDispatch({ + baseUrl: new URL("https://dispatch.test/"), + cancel, + dispatchRequest: requestV2, + receiptFile, + request: requestImpl, + stopping: () => stopping, + }); + await postStarted; + stopping = true; + await expect(cancel("signal")).resolves.toEqual({ + failure: "job-not-found", + outcome: "failed", + receiptWritten: true, + }); + releasePost(); + + await expect(submission).rejects.toThrow( + `Jetson dispatch ${queuedStatusV2.jobId} cancellation requested; cancellation request succeeded`, + ); + expect(requestImpl).toHaveBeenCalledTimes(3); + expect(requestImpl).toHaveBeenNthCalledWith( + 3, + expect.objectContaining({ method: "DELETE", path: `v1/jobs/${queuedStatusV2.jobId}` }), + ); + expect(readReceipt(receiptFile)).toMatchObject({ + cancellation: { outcome: "succeeded", reason: "signal" }, + }); + }); + + it("records a rejected cancellation after repeated status failures (#8142)", async () => { + vi.spyOn(console, "warn").mockImplementation(() => {}); + const receiptFile = temporaryReceiptFile(); + const requestImpl = vi.fn(async () => { + throw new Error("status transport reset"); + }); + + const failure = pollJetsonDispatch({ + baseUrl: new URL("https://dispatch.test/"), + deadlineMs: 10_000, + initialStatus: queuedStatus, + now: () => 0, + receiptFile, + request: requestImpl, + wait: async () => {}, + }); + await expect(failure).rejects.toThrow( + `Jetson dispatch ${queuedStatus.jobId} status failed 3 consecutive times; cancellation request failed (transport-error)`, + ); + await expect(failure).rejects.not.toHaveProperty("cause"); + expect(requestImpl).toHaveBeenCalledTimes(4); + expect(readReceipt(receiptFile)).toEqual({ + schemaVersion: 1, + jobId: queuedStatus.jobId, + request: queuedStatus.request, + cancellation: { + outcome: "failed", + reason: "status-request-failures", + failure: "transport-error", + }, + }); + }); + it("keeps the published response types compatible with the controller (#8142)", () => { const typedRequest: JetsonDispatchRequest = request; const typedStatus: JetsonDispatchStatus = completedStatus; diff --git a/tools/e2e/jetson-dispatch-client.mts b/tools/e2e/jetson-dispatch-client.mts index a097005fc7..875a7f82a7 100644 --- a/tools/e2e/jetson-dispatch-client.mts +++ b/tools/e2e/jetson-dispatch-client.mts @@ -9,6 +9,7 @@ import { decodeJetsonArtifactArchive, JETSON_DISPATCH_AUDIENCE, JETSON_DISPATCH_TARGET, + jetsonDispatchJobId, type JetsonDispatchArtifact, type JetsonDispatchRequest, type JetsonDispatchStatus, @@ -25,6 +26,23 @@ const MAX_WAIT_MS = 54 * 60_000; const MAX_CONSECUTIVE_POLL_FAILURES = 3; const OIDC_TOKEN_CACHE_MS = 4 * 60_000; +type JetsonCancellationFailure = + | "authorization-failed" + | "dispatcher-http-error" + | "invalid-response" + | "job-not-found" + | "request-timeout" + | "transport-error"; +type JetsonCancellationReason = + | "controller-deadline" + | "recovery-receipt-failure" + | "signal" + | "submission-outcome-unknown" + | "status-request-failures"; +type JetsonCancellationResult = + | { outcome: "failed"; failure: JetsonCancellationFailure; receiptWritten: boolean } + | { outcome: "succeeded"; receiptWritten: boolean }; + function record(value: unknown, name: string): Record { if (!value || typeof value !== "object" || Array.isArray(value)) { throw new Error(`${name} must be an object`); @@ -131,7 +149,10 @@ export async function dispatcherRequest(options: { ) { throw new Error("Jetson dispatcher response is too large"); } - if (!response.body) throw new Error("Jetson dispatcher returned an empty response"); + if (!response.body) { + if (response.ok && options.method === "DELETE") return undefined; + throw new Error("Jetson dispatcher returned an empty response"); + } const reader = response.body.getReader(); const chunks: Uint8Array[] = []; let responseBytes = 0; @@ -146,6 +167,10 @@ export async function dispatcherRequest(options: { chunks.push(value); } const bytes = Buffer.concat(chunks, responseBytes); + if (responseBytes === 0) { + if (response.ok && options.method === "DELETE") return undefined; + throw new Error("Jetson dispatcher returned an empty response"); + } let payload: unknown; try { payload = JSON.parse(bytes.toString("utf8")); @@ -165,34 +190,214 @@ function delay(milliseconds: number): Promise { return new Promise((resolve) => setTimeout(resolve, milliseconds)); } +function writeJetsonRecoveryReceipt( + receiptFile: string, + dispatch: Pick, + cancellation?: { + failure?: JetsonCancellationFailure; + outcome: "failed" | "pending" | "succeeded"; + reason: JetsonCancellationReason; + }, +): void { + writePrivateRegularFile( + receiptFile, + `${JSON.stringify( + { + schemaVersion: 1, + jobId: dispatch.jobId, + request: dispatch.request, + ...(cancellation === undefined ? {} : { cancellation }), + }, + null, + 2, + )}\n`, + ); +} + +function classifyCancellationFailure(error: unknown): JetsonCancellationFailure { + const message = error instanceof Error ? error.message : ""; + if (error instanceof Error && ["AbortError", "TimeoutError"].includes(error.name)) { + return "request-timeout"; + } + if (/returned HTTP (?:401|403)(?::|$)/u.test(message)) return "authorization-failed"; + if (/returned HTTP 404(?::|$)/u.test(message)) return "job-not-found"; + if (/returned HTTP [0-9]{3}(?::|$)/u.test(message)) return "dispatcher-http-error"; + if (/empty response|invalid JSON|response is too large|must be an object/u.test(message)) { + return "invalid-response"; + } + return "transport-error"; +} + +async function cancelJetsonDispatch(options: { + baseUrl: URL; + dispatch: Pick; + reason: JetsonCancellationReason; + receiptFile: string; + request: typeof dispatcherRequest; +}): Promise { + const cancellation = { outcome: "pending", reason: options.reason } as const; + try { + writeJetsonRecoveryReceipt(options.receiptFile, options.dispatch, cancellation); + } catch { + // The cancellation request must continue when the local recovery receipt cannot be updated. + } + + let result: { outcome: "failed"; failure: JetsonCancellationFailure } | { outcome: "succeeded" }; + try { + await options.request({ + baseUrl: options.baseUrl, + method: "DELETE", + path: `v1/jobs/${options.dispatch.jobId}`, + maxBytes: MAX_STATUS_BYTES, + }); + result = { outcome: "succeeded" }; + } catch (error) { + result = { outcome: "failed", failure: classifyCancellationFailure(error) }; + } + + let receiptWritten = true; + try { + writeJetsonRecoveryReceipt(options.receiptFile, options.dispatch, { + outcome: result.outcome, + reason: options.reason, + ...(result.outcome === "failed" ? { failure: result.failure } : {}), + }); + } catch { + receiptWritten = false; + } + return { ...result, receiptWritten } as JetsonCancellationResult; +} + +function cancellationResultMessage(result: JetsonCancellationResult): string { + const outcome = + result.outcome === "succeeded" + ? "cancellation request succeeded" + : `cancellation request failed (${result.failure})`; + return result.receiptWritten ? outcome : `${outcome}; recovery receipt update failed`; +} + +export type CancelJetsonDispatch = ( + reason: JetsonCancellationReason, + options?: { retryJobNotFound?: boolean }, +) => Promise; + +export function createJetsonCancellation(options: { + baseUrl: URL; + dispatch: Pick; + receiptFile: string; + request: typeof dispatcherRequest; +}): CancelJetsonDispatch { + let inFlight: Promise | undefined; + let retry: Promise | undefined; + return (reason, callOptions) => { + inFlight ??= cancelJetsonDispatch({ ...options, reason }); + if (!callOptions?.retryJobNotFound) return inFlight; + retry ??= inFlight.then((result) => + result.outcome === "failed" && result.failure === "job-not-found" + ? cancelJetsonDispatch({ ...options, reason }) + : result, + ); + return retry; + }; +} + +export async function submitJetsonDispatch(options: { + baseUrl: URL; + cancel?: CancelJetsonDispatch; + dispatchRequest: JetsonDispatchRequest; + receiptFile: string; + request?: typeof dispatcherRequest; + stopping?: () => boolean; +}): Promise<{ cancel: CancelJetsonDispatch; status: JetsonDispatchStatus }> { + const jobId = jetsonDispatchJobId(options.dispatchRequest); + const dispatch = { jobId, request: options.dispatchRequest }; + writeJetsonRecoveryReceipt(options.receiptFile, dispatch); + if (options.stopping?.()) { + throw new Error(`Jetson dispatch ${jobId} stopped before submission`); + } + const request = options.request ?? dispatcherRequest; + const cancel = + options.cancel ?? + createJetsonCancellation({ + baseUrl: options.baseUrl, + dispatch, + receiptFile: options.receiptFile, + request, + }); + + let status: JetsonDispatchStatus; + try { + status = parseJetsonDispatchStatusResponse( + await request({ + baseUrl: options.baseUrl, + method: "POST", + path: "v1/jobs", + body: options.dispatchRequest, + maxBytes: MAX_STATUS_BYTES, + }), + options.dispatchRequest, + ); + } catch { + const reason = options.stopping?.() ? "signal" : "submission-outcome-unknown"; + const cancellation = await cancel(reason, { retryJobNotFound: true }); + throw new Error( + `Jetson dispatch ${jobId} submission outcome was not confirmed; ${cancellationResultMessage(cancellation)}`, + ); + } + if (options.stopping?.()) { + const cancellation = await cancel("signal", { retryJobNotFound: true }); + throw new Error( + `Jetson dispatch ${jobId} cancellation requested; ${cancellationResultMessage(cancellation)}`, + ); + } + return { cancel, status }; +} + export async function pollJetsonDispatch(options: { baseUrl: URL; + cancel?: CancelJetsonDispatch; deadlineMs: number; initialStatus: JetsonDispatchStatus; - jobId: string; now?: () => number; + receiptFile: string; request?: typeof dispatcherRequest; stopping?: () => boolean; wait?: typeof delay; }): Promise { const now = options.now ?? Date.now; const request = options.request ?? dispatcherRequest; + const jobId = options.initialStatus.jobId; + const cancel = + options.cancel ?? + createJetsonCancellation({ + baseUrl: options.baseUrl, + dispatch: options.initialStatus, + receiptFile: options.receiptFile, + request, + }); const wait = options.wait ?? delay; let consecutiveFailures = 0; let status = options.initialStatus; - if (status.jobId !== options.jobId) { - throw new Error("Jetson dispatcher status does not match the accepted job"); + try { + writeJetsonRecoveryReceipt(options.receiptFile, status); + } catch { + const cancellation = await cancel("recovery-receipt-failure"); + throw new Error( + `Jetson dispatch ${jobId} was accepted but its recovery receipt could not be written; ${cancellationResultMessage(cancellation)}`, + ); } while (status.state !== "completed") { - if (options.stopping?.()) throw new Error("Jetson dispatch cancellation requested"); + if (options.stopping?.()) { + const cancellation = await cancel("signal"); + throw new Error( + `Jetson dispatch ${jobId} cancellation requested; ${cancellationResultMessage(cancellation)}`, + ); + } if (now() >= options.deadlineMs) { - await request({ - baseUrl: options.baseUrl, - method: "DELETE", - path: `v1/jobs/${options.jobId}`, - maxBytes: MAX_STATUS_BYTES, - }).catch(() => undefined); - throw new Error("Jetson dispatcher did not complete before the controller deadline"); + const cancellation = await cancel("controller-deadline"); + throw new Error( + `Jetson dispatch ${jobId} did not complete before the controller deadline; ${cancellationResultMessage(cancellation)}`, + ); } await wait(POLL_INTERVAL_MS); try { @@ -200,12 +405,12 @@ export async function pollJetsonDispatch(options: { await request({ baseUrl: options.baseUrl, method: "GET", - path: `v1/jobs/${options.jobId}`, + path: `v1/jobs/${jobId}`, maxBytes: MAX_STATUS_BYTES, }), options.initialStatus.request, ); - if (status.jobId !== options.jobId) { + if (status.jobId !== jobId) { throw new Error("Jetson dispatcher status does not match the accepted job"); } consecutiveFailures = 0; @@ -213,13 +418,10 @@ export async function pollJetsonDispatch(options: { } catch (error) { consecutiveFailures += 1; if (consecutiveFailures >= MAX_CONSECUTIVE_POLL_FAILURES) { - await request({ - baseUrl: options.baseUrl, - method: "DELETE", - path: `v1/jobs/${options.jobId}`, - maxBytes: MAX_STATUS_BYTES, - }).catch(() => undefined); - throw error; + const cancellation = await cancel("status-request-failures"); + throw new Error( + `Jetson dispatch ${jobId} status failed ${MAX_CONSECUTIVE_POLL_FAILURES} consecutive times; ${cancellationResultMessage(cancellation)}`, + ); } console.warn("Jetson dispatch status request failed; retrying"); } @@ -247,46 +449,48 @@ async function main(): Promise { if (!path.isAbsolute(artifactDirectory)) throw new Error("E2E_ARTIFACT_DIR must be absolute"); fs.mkdirSync(artifactDirectory, { recursive: true, mode: 0o700 }); fs.chmodSync(artifactDirectory, 0o700); + const receiptFile = path.join(artifactDirectory, "jetson-dispatch.json"); - let jobId: string | undefined; + const jobId = jetsonDispatchJobId(request); + const cancelDispatch = createJetsonCancellation({ + baseUrl, + dispatch: { jobId, request }, + receiptFile, + request: dispatcherRequest, + }); + let submissionStarted = false; let stopping = false; const cancel = (): void => { if (stopping) return; stopping = true; - if (!jobId) { + if (!submissionStarted) { process.exitCode = 1; return; } - void dispatcherRequest({ - baseUrl, - method: "DELETE", - path: `v1/jobs/${jobId}`, - maxBytes: MAX_STATUS_BYTES, - }).finally(() => { + void cancelDispatch("signal").finally(() => { process.exitCode = 1; }); }; process.on("SIGINT", cancel); process.on("SIGTERM", cancel); - const dispatched = parseJetsonDispatchStatusResponse( - await dispatcherRequest({ - baseUrl, - method: "POST", - path: "v1/jobs", - body: request, - maxBytes: MAX_STATUS_BYTES, - }), - request, - ); - jobId = dispatched.jobId; + submissionStarted = true; + const submission = await submitJetsonDispatch({ + baseUrl, + cancel: cancelDispatch, + dispatchRequest: request, + receiptFile, + stopping: () => stopping, + }); + const dispatched = submission.status; console.log(`Jetson dispatch accepted as ${jobId}`); const deadline = Date.now() + MAX_WAIT_MS; await pollJetsonDispatch({ baseUrl, + cancel: cancelDispatch, deadlineMs: deadline, initialStatus: dispatched, - jobId, + receiptFile, stopping: () => stopping, }); @@ -298,10 +502,7 @@ async function main(): Promise { }); const artifact: JetsonDispatchArtifact = parseJetsonDispatchArtifact(artifactValue, jobId); const { artifactArchiveBase64, ...artifactReceipt } = artifact; - writePrivateRegularFile( - path.join(artifactDirectory, "jetson-dispatch.json"), - `${JSON.stringify(artifactReceipt, null, 2)}\n`, - ); + writePrivateRegularFile(receiptFile, `${JSON.stringify(artifactReceipt, null, 2)}\n`); if (artifactArchiveBase64 !== undefined) { writePrivateRegularFile( path.join(artifactDirectory, "jetson-e2e-artifacts.tar.gz"), diff --git a/tools/e2e/jetson-dispatch-contract.mts b/tools/e2e/jetson-dispatch-contract.mts index 58d883ad75..575347f8e2 100644 --- a/tools/e2e/jetson-dispatch-contract.mts +++ b/tools/e2e/jetson-dispatch-contract.mts @@ -168,7 +168,7 @@ export function parseJetsonDispatchRequest(value: unknown): JetsonDispatchReques }; } -function expectedJobId(request: JetsonDispatchRequest): string { +export function jetsonDispatchJobId(request: JetsonDispatchRequest): string { const managedImageRevision = request.schemaVersion === 2 ? `:${request.managedImageRevision}` : ""; return createHash("sha256") @@ -226,7 +226,7 @@ export function parseJetsonDispatchStatus(value: unknown): JetsonDispatchStatus status.schemaVersion !== request.schemaVersion || typeof status.jobId !== "string" || !JOB_ID_PATTERN.test(status.jobId) || - status.jobId !== expectedJobId(request) + status.jobId !== jetsonDispatchJobId(request) ) { throw new Error("Jetson dispatch status does not match its request and job ID"); } @@ -332,7 +332,7 @@ export function parseJetsonDispatchStatusResponse( const response = record(value, "Jetson dispatcher response"); requireFields(response, "Jetson dispatcher response", ["job"]); const status = parseJetsonDispatchStatus(response.job); - if (expectedRequest !== undefined && status.jobId !== expectedJobId(expectedRequest)) { + if (expectedRequest !== undefined && status.jobId !== jetsonDispatchJobId(expectedRequest)) { throw new Error("Jetson dispatcher response does not match the submitted request"); } return status;