update supervisor and schema

This commit is contained in:
nicktrn
2025-05-06 14:44:46 +01:00
parent 52a57a449e
commit 16b344d67c
5 changed files with 49 additions and 8 deletions
@@ -17,6 +17,7 @@ import {
WorkloadRunAttemptStartRequestBody,
type WorkloadRunAttemptStartResponseBody,
type WorkloadRunLatestSnapshotResponseBody,
WorkloadRunSnapshotsSinceResponseBody,
type WorkloadServerToClientEvents,
type WorkloadSuspendRunResponseBody,
} from "@trigger.dev/core/v3/workers";
@@ -341,6 +342,31 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
} satisfies WorkloadRunLatestSnapshotResponseBody);
},
})
.route(
"/api/v1/workload-actions/runs/:runFriendlyId/snapshots/since/:snapshotFriendlyId",
"GET",
{
paramsSchema: WorkloadActionParams,
handler: async ({ req, reply, params }) => {
const sinceSnapshotResponse = await this.workerClient.getSnapshotsSince(
params.runFriendlyId,
params.snapshotFriendlyId,
this.runnerIdFromRequest(req)
);
if (!sinceSnapshotResponse.success) {
console.error("Failed to get snapshots since", {
runId: params.runFriendlyId,
error: sinceSnapshotResponse.error,
});
reply.empty(500);
return;
}
reply.json(sinceSnapshotResponse.data satisfies WorkloadRunSnapshotsSinceResponseBody);
},
}
)
.route("/api/v1/workload-actions/runs/:runFriendlyId/logs/debug", "POST", {
paramsSchema: WorkloadActionParams.pick({ runFriendlyId: true }),
bodySchema: WorkloadDebugLogRequestBody,
@@ -16,15 +16,15 @@ export const loader = createLoaderWorkerApiRoute(
}): Promise<TypedResponse<WorkerApiRunSnapshotsSinceResponseBody>> => {
const { runFriendlyId, snapshotId } = params;
const executions = await authenticatedWorker.getSnapshotsSince({
const snapshots = await authenticatedWorker.getSnapshotsSince({
runFriendlyId,
snapshotId,
});
if (!executions) {
if (!snapshots) {
throw new Error("Failed to retrieve snapshots since given snapshot");
}
return json({ executions });
return json({ snapshots });
}
);
@@ -1105,22 +1105,22 @@ export class RunExecution {
return;
}
const { executions } = response.data;
const { snapshots } = response.data;
if (!executions.length) {
if (!snapshots.length) {
this.sendDebugLog(`fetchAndProcessSnapshotChanges: no new snapshots`, { source });
return;
}
// Only act on the last snapshot
const lastSnapshot = executions[executions.length - 1];
const lastSnapshot = snapshots[snapshots.length - 1];
if (!lastSnapshot) {
this.sendDebugLog(`fetchAndProcessSnapshotChanges: no last snapshot`, { source });
return;
}
const previousSnapshots = executions.slice(0, -1);
const previousSnapshots = snapshots.slice(0, -1);
// If any previous snapshot is QUEUED or SUSPENDED, deprecate this worker
const deprecatedStatus: TaskRunExecutionStatus[] = ["QUEUED", "SUSPENDED"];
@@ -17,6 +17,7 @@ import {
WorkerApiDebugLogBody,
WorkerApiSuspendRunRequestBody,
WorkerApiSuspendRunResponseBody,
WorkerApiRunSnapshotsSinceResponseBody,
} from "./schemas.js";
import { SupervisorClientCommonOptions } from "./types.js";
import { getDefaultWorkerHeaders } from "./util.js";
@@ -185,6 +186,20 @@ export class SupervisorHttpClient {
);
}
async getSnapshotsSince(runId: string, snapshotId: string, runnerId?: string) {
return wrapZodFetch(
WorkerApiRunSnapshotsSinceResponseBody,
`${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/since/${snapshotId}`,
{
method: "GET",
headers: {
...this.defaultHeaders,
...this.runnerIdHeader(runnerId),
},
}
);
}
async sendDebugLog(runId: string, body: WorkerApiDebugLogBody, runnerId?: string): Promise<void> {
try {
const res = await wrapZodFetch(
@@ -164,7 +164,7 @@ export type WorkerApiSuspendCompletionResponseBody = z.infer<
>;
export const WorkerApiRunSnapshotsSinceResponseBody = z.object({
executions: z.array(RunExecutionData),
snapshots: z.array(RunExecutionData),
});
export type WorkerApiRunSnapshotsSinceResponseBody = z.infer<
typeof WorkerApiRunSnapshotsSinceResponseBody