wehub-resource-sync
70bf21e064
Deploy (to testing) and Test Playground Preview Worker / Deploy Playground Preview Worker (testing) (push) Has been skipped
Deploy Workers Shared Staging / Deploy Workers Shared Staging (push) Failing after 0s
Prerelease / build (push) Has been skipped
Handle Changesets / Handle Changesets (push) Has been cancelled
Semgrep OSS scan / semgrep-oss (push) Has been cancelled
2308 lines
61 KiB
TypeScript
2308 lines
61 KiB
TypeScript
import { runInDurableObject } from "cloudflare:test";
|
|
import { env } from "cloudflare:workers";
|
|
import { NonRetryableError } from "cloudflare:workflows";
|
|
import { afterEach, describe, it, vi } from "vitest";
|
|
import workerdUnsafe from "workerd:unsafe";
|
|
import { DEFAULT_STEP_LIMIT, InstanceEvent, InstanceStatus } from "../src";
|
|
import { ABORT_REASONS, isAbortError } from "../src/lib/errors";
|
|
import { setTestWorkflowCallback } from "./test-entry";
|
|
import { runWorkflow, runWorkflowAndAwait } from "./utils";
|
|
import type {
|
|
DatabaseInstance,
|
|
DatabaseVersion,
|
|
DatabaseWorkflow,
|
|
EngineLogs,
|
|
} from "../src/engine";
|
|
import type {
|
|
RollbackContext,
|
|
RollbackFn,
|
|
WorkflowStepRollbackOptions,
|
|
} from "../src/lib/rollback";
|
|
import type {
|
|
WorkflowDelayFunction,
|
|
WorkflowStep,
|
|
WorkflowStepConfig,
|
|
WorkflowStepContext,
|
|
} from "cloudflare:workers";
|
|
|
|
afterEach(async () => {
|
|
await workerdUnsafe.abortAllDurableObjects();
|
|
});
|
|
|
|
describe("Engine", () => {
|
|
it("should not retry after NonRetryableError is thrown", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "NON-RETRYABLE-ERROR";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
|
|
await runWorkflow(instanceId, async (_event, step) => {
|
|
await step.do("should only have one retry", async () => {
|
|
throw new NonRetryableError("Should only retry once");
|
|
});
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
const logs = (await env.ENGINE.get(
|
|
engineId
|
|
).readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_FAILURE
|
|
);
|
|
} catch (e) {
|
|
// DO may still be aborting — retry
|
|
if (isAbortError(e)) {
|
|
return false;
|
|
}
|
|
throw e;
|
|
}
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logs = (await env.ENGINE.get(engineId).readLogs()) as EngineLogs;
|
|
|
|
expect(
|
|
logs.logs.filter((val) => val.event === InstanceEvent.ATTEMPT_START)
|
|
).toHaveLength(1);
|
|
});
|
|
|
|
it("should preserve NonRetryableError message when compat flag is enabled", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "NON-RETRYABLE-PRESERVE-MSG";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
vi.stubGlobal("Cloudflare", {
|
|
compatibilityFlags: {
|
|
workflows_preserve_non_retryable_error_message: true,
|
|
},
|
|
});
|
|
|
|
try {
|
|
setTestWorkflowCallback(async (_event, step) => {
|
|
await step.do("failing-step", async () => {
|
|
throw new NonRetryableError("my custom error message");
|
|
});
|
|
});
|
|
|
|
await engineStub
|
|
.init(
|
|
12346,
|
|
{} as DatabaseWorkflow,
|
|
{} as DatabaseVersion,
|
|
{ id: instanceId } as DatabaseInstance,
|
|
{ payload: {}, timestamp: new Date(), instanceId, workflowName: "" }
|
|
)
|
|
.catch(() => {});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
const logs = (await env.ENGINE.get(
|
|
engineId
|
|
).readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_FAILURE
|
|
);
|
|
} catch (e) {
|
|
if (isAbortError(e)) {
|
|
return false;
|
|
}
|
|
throw e;
|
|
}
|
|
},
|
|
{ timeout: 3000 }
|
|
);
|
|
|
|
const logs = (await env.ENGINE.get(engineId).readLogs()) as EngineLogs;
|
|
|
|
const workflowFailure = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_FAILURE
|
|
);
|
|
expect(workflowFailure?.metadata.error).toEqual({
|
|
name: "NonRetryableError",
|
|
message: "my custom error message",
|
|
});
|
|
|
|
const attemptFailure = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.ATTEMPT_FAILURE
|
|
);
|
|
expect(attemptFailure?.metadata.error).toEqual({
|
|
name: "NonRetryableError",
|
|
message: "my custom error message",
|
|
});
|
|
} finally {
|
|
vi.unstubAllGlobals();
|
|
}
|
|
});
|
|
|
|
it("should not error out if step fails but is try-catched", async ({
|
|
expect,
|
|
}) => {
|
|
const engineStub = await runWorkflow(
|
|
"MOCK-INSTANCE-ID-TRY-CATCH",
|
|
async (_event, step) => {
|
|
try {
|
|
await step.do(
|
|
"always errors out",
|
|
{
|
|
retries: {
|
|
limit: 0,
|
|
delay: 1000,
|
|
},
|
|
},
|
|
async () => {
|
|
throw new Error("Step errors out");
|
|
}
|
|
);
|
|
} catch {}
|
|
return "finished";
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event == InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
|
|
expect(
|
|
logs.logs.some((val) => val.event == InstanceEvent.WORKFLOW_SUCCESS)
|
|
).toBe(true);
|
|
|
|
expect(
|
|
logs.logs.filter((val) => val.event == InstanceEvent.ATTEMPT_FAILURE)
|
|
).toHaveLength(1);
|
|
});
|
|
|
|
it("applies a dynamic delay function and succeeds after retries", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "DYNAMIC-DELAY-APPLIED";
|
|
const dynamicDelay: WorkflowDelayFunction = () => 50;
|
|
let attempts = 0;
|
|
|
|
const engineStub = await runWorkflow(instanceId, async (_event, step) => {
|
|
await step.do(
|
|
"dynamic-delay-step",
|
|
{
|
|
retries: { limit: 3, delay: dynamicDelay, backoff: "constant" },
|
|
},
|
|
async () => {
|
|
attempts++;
|
|
if (attempts < 2) {
|
|
throw new Error("transient");
|
|
}
|
|
return "ok";
|
|
}
|
|
);
|
|
return "done";
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
const attemptFailures = logs.logs.filter(
|
|
(val) => val.event === InstanceEvent.ATTEMPT_FAILURE
|
|
);
|
|
|
|
expect(attemptFailures.length).toBeGreaterThanOrEqual(1);
|
|
expect(attemptFailures[0]?.metadata).toMatchObject({ retryDelayMs: 50 });
|
|
});
|
|
|
|
it("fails non-retryably when the delay function throws", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "DYNAMIC-DELAY-BROKEN";
|
|
const throwingDelay: WorkflowDelayFunction = () => {
|
|
throw new Error("delay boom");
|
|
};
|
|
|
|
const engineStub = await runWorkflow(instanceId, async (_event, step) => {
|
|
await step.do(
|
|
"broken-delay-step",
|
|
{
|
|
retries: { limit: 5, delay: throwingDelay, backoff: "constant" },
|
|
},
|
|
async () => {
|
|
throw new Error("step failed");
|
|
}
|
|
);
|
|
return "done";
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_FAILURE
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
|
|
// The broken delay function turns the retry into a non-retryable failure,
|
|
// so the step is only attempted once.
|
|
expect(
|
|
logs.logs.filter((val) => val.event === InstanceEvent.ATTEMPT_START)
|
|
).toHaveLength(1);
|
|
|
|
// The NonRetryableDelayError crosses the engine<->workflow RPC boundary,
|
|
// which collapses the custom error class to name "Error" and folds the
|
|
// original class name into the message.
|
|
const workflowFailure = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_FAILURE
|
|
);
|
|
expect(workflowFailure?.metadata.error.message).toContain(
|
|
"NonRetryableDelayError"
|
|
);
|
|
expect(workflowFailure?.metadata.error.message).toContain("delay function");
|
|
expect(workflowFailure?.metadata.error.message).toContain("delay boom");
|
|
});
|
|
|
|
it("omits delay from ctx.config for a dynamic delay but includes it for a static delay", async ({
|
|
expect,
|
|
}) => {
|
|
const dynamicDelay: WorkflowDelayFunction = () => 10;
|
|
let dynamicConfig:
|
|
| WorkflowStepContext<WorkflowDelayFunction>["config"]
|
|
| undefined;
|
|
let staticConfig: WorkflowStepContext<number>["config"] | undefined;
|
|
|
|
const dynamicStub = await runWorkflow(
|
|
"DYNAMIC-DELAY-CTX-HIDDEN",
|
|
async (_event, step) => {
|
|
await step.do(
|
|
"dynamic-config-step",
|
|
{
|
|
retries: { limit: 1, delay: dynamicDelay, backoff: "constant" },
|
|
},
|
|
async (ctx) => {
|
|
dynamicConfig = ctx.config;
|
|
return "ok";
|
|
}
|
|
);
|
|
return "done";
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await dynamicStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const staticStub = await runWorkflow(
|
|
"STATIC-DELAY-CTX-VISIBLE",
|
|
async (_event, step) => {
|
|
await step.do(
|
|
"static-config-step",
|
|
{
|
|
retries: { limit: 1, delay: 1234, backoff: "constant" },
|
|
},
|
|
async (ctx) => {
|
|
staticConfig = ctx.config;
|
|
return "ok";
|
|
}
|
|
);
|
|
return "done";
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await staticStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const dynamicRetries = dynamicConfig?.retries;
|
|
expect(dynamicRetries).toBeDefined();
|
|
expect(dynamicRetries && "delay" in dynamicRetries).toBe(false);
|
|
expect(staticConfig?.retries?.delay).toBe(1234);
|
|
});
|
|
|
|
it("waitForEvent should receive events while active", async ({ expect }) => {
|
|
const engineStub = await runWorkflow(
|
|
"MOCK-INSTANCE-ID-WAIT-FOR-EVENT",
|
|
async (_, step) => {
|
|
return await step.waitForEvent("i'm a event!", {
|
|
type: "event-type-1",
|
|
timeout: "10 seconds",
|
|
});
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some((val) => val.event === InstanceEvent.WAIT_START);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
await engineStub.receiveEvent({
|
|
type: "event-type-1",
|
|
timestamp: new Date(),
|
|
payload: {},
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
expect(logs.logs.some((v) => v.event === InstanceEvent.WAIT_START)).toBe(
|
|
true
|
|
);
|
|
expect(
|
|
logs.logs.some((v) => v.event === InstanceEvent.WORKFLOW_SUCCESS)
|
|
).toBe(true);
|
|
});
|
|
|
|
it("waitForEvent should receive events even if not active", async ({
|
|
expect,
|
|
}) => {
|
|
const engineStub = await runWorkflow(
|
|
"MOCK-INSTANCE-ID-WAIT-FOR-EVENT-NOT-ACTIVE",
|
|
async (_, step) => {
|
|
return await step.waitForEvent("i'm a event!", {
|
|
type: "event-type-1",
|
|
timeout: "10 seconds",
|
|
});
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some((val) => val.event === InstanceEvent.WAIT_START);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.abort("kabooom");
|
|
});
|
|
} catch {
|
|
// supposed to error out
|
|
}
|
|
|
|
// Get a new stub since we've just aborted the durable object
|
|
const newStub = env.ENGINE.get(
|
|
env.ENGINE.idFromName("MOCK-INSTANCE-ID-WAIT-FOR-EVENT-NOT-ACTIVE")
|
|
);
|
|
|
|
await newStub.receiveEvent({
|
|
type: "event-type-1",
|
|
timestamp: new Date(),
|
|
payload: {},
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await newStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logs = (await newStub.readLogs()) as EngineLogs;
|
|
expect(logs.logs.some((v) => v.event === InstanceEvent.WAIT_START)).toBe(
|
|
true
|
|
);
|
|
expect(
|
|
logs.logs.some((v) => v.event === InstanceEvent.WORKFLOW_SUCCESS)
|
|
).toBe(true);
|
|
});
|
|
|
|
it("waitForEvent should not deliver events to timed-out events with the same type", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "WAIT-FOR-EVENT-STALE-WAITER";
|
|
const engineStub = await runWorkflow(instanceId, async (_, step) => {
|
|
const results: Array<{
|
|
iteration: number;
|
|
received: boolean;
|
|
}> = [];
|
|
|
|
for (let i = 1; i <= 3; i++) {
|
|
try {
|
|
await step.waitForEvent(`my-event-waiter-${i}`, {
|
|
type: "my-event",
|
|
timeout: 500,
|
|
});
|
|
results.push({ iteration: i, received: true });
|
|
} catch {
|
|
results.push({ iteration: i, received: false });
|
|
}
|
|
}
|
|
|
|
return { results };
|
|
});
|
|
|
|
// 1st waitForEvent iteration - should receive event
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return (
|
|
logs.logs.filter((val) => val.event === InstanceEvent.WAIT_START)
|
|
.length >= 1
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
await engineStub.receiveEvent({
|
|
type: "my-event",
|
|
timestamp: new Date(),
|
|
payload: { iteration: 1 },
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return (
|
|
logs.logs.filter((val) => val.event === InstanceEvent.WAIT_START)
|
|
.length >= 2
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// 2nd waitForEvent iteration - should timeout (500ms)
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WAIT_TIMED_OUT
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// 3rd waitForEvent iteration - should receive event
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return (
|
|
logs.logs.filter((val) => val.event === InstanceEvent.WAIT_START)
|
|
.length >= 3
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
await engineStub.receiveEvent({
|
|
type: "my-event",
|
|
timestamp: new Date(),
|
|
payload: { iteration: 3 },
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
// Iterations 1 and 3 received events; iteration 2 timed out
|
|
expect(
|
|
logs.logs.filter((val) => val.event === InstanceEvent.WAIT_COMPLETE)
|
|
).toHaveLength(2);
|
|
expect(
|
|
logs.logs.filter((val) => val.event === InstanceEvent.WAIT_TIMED_OUT)
|
|
).toHaveLength(1);
|
|
});
|
|
|
|
it("should restore state from storage when accountId is undefined", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "RESTORE-TEST-INSTANCE";
|
|
const accountId = 12345;
|
|
const workflow: DatabaseWorkflow = {
|
|
name: "test-workflow",
|
|
id: "workflow-123",
|
|
created_on: new Date().toISOString(),
|
|
modified_on: new Date().toISOString(),
|
|
script_name: "test-script",
|
|
class_name: "TestWorkflow",
|
|
triggered_on: null,
|
|
};
|
|
const version: DatabaseVersion = {
|
|
id: "version-123",
|
|
class_name: "TestWorkflow",
|
|
created_on: new Date().toISOString(),
|
|
modified_on: new Date().toISOString(),
|
|
workflow_id: workflow.id,
|
|
mutable_pipeline_id: "pipeline-123",
|
|
};
|
|
const instance: DatabaseInstance = {
|
|
id: instanceId,
|
|
created_on: new Date().toISOString(),
|
|
modified_on: new Date().toISOString(),
|
|
workflow_id: workflow.id,
|
|
version_id: version.id,
|
|
status: InstanceStatus.Running,
|
|
started_on: new Date().toISOString(),
|
|
ended_on: null,
|
|
};
|
|
const event = {
|
|
payload: {},
|
|
timestamp: new Date(),
|
|
instanceId: instanceId,
|
|
workflowName: "",
|
|
};
|
|
|
|
const engineStub = await runWorkflow(instanceId, async () => {
|
|
return "test";
|
|
});
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.init(accountId, workflow, version, instance, event);
|
|
await engine.setStatus(accountId, instanceId, InstanceStatus.Running);
|
|
await engine.abort(ABORT_REASONS.GRACE_PERIOD_COMPLETE);
|
|
});
|
|
} catch (e) {
|
|
// Expected - abort throws to break the DO
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const restartedStub = env.ENGINE.get(engineId);
|
|
|
|
const status = await runInDurableObject(restartedStub, (engine) => {
|
|
return engine.getStatus();
|
|
});
|
|
|
|
expect(status).toBe(InstanceStatus.Running);
|
|
|
|
const logs = (await restartedStub.readLogs()) as EngineLogs;
|
|
expect(
|
|
logs.logs.some((log) => log.event === InstanceEvent.WORKFLOW_START)
|
|
).toBe(true);
|
|
});
|
|
|
|
it("should complete a step that returns a large Uint8Array without SQLITE_TOOBIG", async ({
|
|
expect,
|
|
}) => {
|
|
// Regression: JSON.stringify(Uint8Array) encodes each byte as a numeric key,
|
|
// producing a string far larger than byteLength. A 200 KB Uint8Array → ~2 MB JSON
|
|
// → SQLITE_TOOBIG. writeLog must use a replacer to sanitize TypedArrays.
|
|
const instanceId = "LARGE-UINT8ARRAY-RESULT";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
await runWorkflowAndAwait(instanceId, async (_event, step) => {
|
|
await step.do("large-binary-step", async () => {
|
|
return new Uint8Array(200_000); // ~200 KB, triggers SQLITE_TOOBIG without fix
|
|
});
|
|
});
|
|
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
expect(
|
|
logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS)
|
|
).toBe(true);
|
|
expect(
|
|
logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE)
|
|
).toBe(false);
|
|
});
|
|
|
|
it("should complete a step that returns an object containing a large Uint8Array without SQLITE_TOOBIG", async ({
|
|
expect,
|
|
}) => {
|
|
// Regression: nested TypedArrays inside objects also cause SQLITE_TOOBIG without
|
|
// the JSON.stringify replacer, since sanitization must be recursive.
|
|
const instanceId = "NESTED-UINT8ARRAY-RESULT";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
await runWorkflowAndAwait(instanceId, async (_event, step) => {
|
|
await step.do("nested-binary-step", async () => {
|
|
return { payload: new Uint8Array(200_000), label: "test" };
|
|
});
|
|
});
|
|
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
expect(
|
|
logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS)
|
|
).toBe(true);
|
|
expect(
|
|
logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE)
|
|
).toBe(false);
|
|
});
|
|
|
|
describe("step limits", () => {
|
|
it("should enforce step limit when exceeded", async ({ expect }) => {
|
|
const stepLimit = 3;
|
|
const instanceId = "STEP-LIMIT-EXCEEDED";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
await runInDurableObject(engineStub, (engine) => {
|
|
engine.stepLimit = stepLimit;
|
|
});
|
|
|
|
setTestWorkflowCallback(async (_event, step) => {
|
|
for (let i = 0; i < stepLimit + 1; i++) {
|
|
await step.do(`step-${i}`, async () => `result-${i}`);
|
|
}
|
|
});
|
|
|
|
await engineStub.init(
|
|
12346,
|
|
{} as DatabaseWorkflow,
|
|
{} as DatabaseVersion,
|
|
{ id: instanceId } as DatabaseInstance,
|
|
{ payload: {}, timestamp: new Date(), instanceId, workflowName: "" }
|
|
);
|
|
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
|
|
expect(
|
|
logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE)
|
|
).toBe(true);
|
|
});
|
|
|
|
it("should succeed when steps are exactly at the limit", async ({
|
|
expect,
|
|
}) => {
|
|
const stepLimit = 3;
|
|
const instanceId = "STEP-LIMIT-AT-LIMIT";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
await runInDurableObject(engineStub, (engine) => {
|
|
engine.stepLimit = stepLimit;
|
|
});
|
|
|
|
setTestWorkflowCallback(async (_event, step) => {
|
|
for (let i = 0; i < stepLimit; i++) {
|
|
await step.do(`step-${i}`, async () => `result-${i}`);
|
|
}
|
|
return "done";
|
|
});
|
|
|
|
await engineStub.init(
|
|
12346,
|
|
{} as DatabaseWorkflow,
|
|
{} as DatabaseVersion,
|
|
{ id: instanceId } as DatabaseInstance,
|
|
{ payload: {}, timestamp: new Date(), instanceId, workflowName: "" }
|
|
);
|
|
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
|
|
expect(
|
|
logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS)
|
|
).toBe(true);
|
|
expect(
|
|
logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE)
|
|
).toBe(false);
|
|
});
|
|
|
|
it("should use DEFAULT_STEP_LIMIT when no limit is configured", async ({
|
|
expect,
|
|
}) => {
|
|
const engineId = env.ENGINE.idFromName("STEP-LIMIT-DEFAULT");
|
|
const freshStub = env.ENGINE.get(engineId);
|
|
|
|
const stepLimit = await runInDurableObject(
|
|
freshStub,
|
|
(engine) => engine.stepLimit
|
|
);
|
|
|
|
expect(stepLimit).toBe(DEFAULT_STEP_LIMIT);
|
|
});
|
|
});
|
|
|
|
describe("lifecycle methods", () => {
|
|
it.for([
|
|
InstanceStatus.Complete,
|
|
InstanceStatus.Errored,
|
|
InstanceStatus.Terminated,
|
|
])(
|
|
"should throw when calling terminate on instance in finite state: %s",
|
|
async (finiteStatus, { expect }) => {
|
|
const engineStub = await runWorkflowAndAwait(
|
|
`TERMINATE-${finiteStatus}-INSTANCE`,
|
|
async () => "done"
|
|
);
|
|
|
|
// If not Complete, manually set the status
|
|
if (finiteStatus !== InstanceStatus.Complete) {
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
await state.storage.put("ENGINE_STATUS", finiteStatus);
|
|
});
|
|
}
|
|
|
|
await expect(
|
|
runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("terminate");
|
|
})
|
|
).rejects.toThrow(
|
|
"Cannot terminate instance since its on a finite state"
|
|
);
|
|
}
|
|
);
|
|
|
|
it.for([
|
|
InstanceStatus.Complete,
|
|
InstanceStatus.Errored,
|
|
InstanceStatus.Terminated,
|
|
InstanceStatus.Running,
|
|
InstanceStatus.Paused,
|
|
])(
|
|
"should restart workflow from status: %s",
|
|
async (initialStatus, { expect }) => {
|
|
const instanceId = `RESTART-${initialStatus}-INSTANCE`;
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
|
|
const engineStub = await runWorkflowAndAwait(
|
|
instanceId,
|
|
async (_event: unknown, step: WorkflowStep) => {
|
|
await step.do("test-step", async () => "step-result");
|
|
return "done";
|
|
}
|
|
);
|
|
|
|
// Set the status to initialStatus
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
await state.storage.put("ENGINE_STATUS", initialStatus);
|
|
});
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("restart");
|
|
});
|
|
} catch (e) {
|
|
// Expected - abort throws to break the DO
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
const restartedStub = env.ENGINE.get(engineId);
|
|
|
|
await runInDurableObject(restartedStub, async (engine) => {
|
|
await engine.attemptRestart();
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const status = await runInDurableObject(restartedStub, (engine) =>
|
|
engine.getStatus()
|
|
);
|
|
return status === InstanceStatus.Complete;
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// Verify the workflow ran again by checking logs
|
|
const logs = (await restartedStub.readLogs()) as EngineLogs;
|
|
|
|
expect(
|
|
logs.logs.some((log) => log.event === InstanceEvent.WORKFLOW_START)
|
|
).toBe(true);
|
|
|
|
expect(
|
|
logs.logs.some((log) => log.event === InstanceEvent.STEP_START)
|
|
).toBe(true);
|
|
|
|
expect(
|
|
logs.logs.some((log) => log.event === InstanceEvent.WORKFLOW_SUCCESS)
|
|
).toBe(true);
|
|
}
|
|
);
|
|
|
|
it("should restart from a specific step and preserve earlier results", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "RESTART-FROM-STEP";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
|
|
const engineStub = await runWorkflowAndAwait(
|
|
instanceId,
|
|
async (_event: unknown, step: WorkflowStep) => {
|
|
const a = await step.do("step-a", async () => "result-a");
|
|
const b = await step.do("step-b", async () => "result-b");
|
|
return { a, b };
|
|
}
|
|
);
|
|
|
|
const statusBefore = await runInDurableObject(engineStub, (engine) =>
|
|
engine.getStatus()
|
|
);
|
|
expect(statusBefore).toBe(InstanceStatus.Complete);
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("restart", {
|
|
name: "step-b",
|
|
});
|
|
});
|
|
} catch (e) {
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
const restartedStub = env.ENGINE.get(engineId);
|
|
|
|
await runInDurableObject(restartedStub, async (engine) => {
|
|
await engine.attemptRestart();
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const status = await runInDurableObject(restartedStub, (engine) =>
|
|
engine.getStatus()
|
|
);
|
|
return status === InstanceStatus.Complete;
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logs = (await restartedStub.readLogs()) as EngineLogs;
|
|
|
|
// step-a should still have its cached result from the first run
|
|
const stepSuccessLogs = logs.logs.filter(
|
|
(log) => log.event === InstanceEvent.STEP_SUCCESS
|
|
);
|
|
expect(stepSuccessLogs.length).toBeGreaterThanOrEqual(2);
|
|
|
|
expect(
|
|
logs.logs.some((log) => log.event === InstanceEvent.WORKFLOW_SUCCESS)
|
|
).toBe(true);
|
|
});
|
|
|
|
it("should default restart from step type to step.do", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "RESTART-FROM-STEP-DEFAULT-DO";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
|
|
const engineStub = await runWorkflowAndAwait(
|
|
instanceId,
|
|
async (_event: unknown, step: WorkflowStep) => {
|
|
const setup = await step.do("setup", async () => crypto.randomUUID());
|
|
await step.sleep("checkpoint", 1);
|
|
const between = await step.do("between", async () =>
|
|
crypto.randomUUID()
|
|
);
|
|
const checkpoint = await step.do("checkpoint", async () =>
|
|
crypto.randomUUID()
|
|
);
|
|
const after = await step.do("after", async () => crypto.randomUUID());
|
|
return { setup, between, checkpoint, after };
|
|
}
|
|
);
|
|
|
|
const logsBefore = (await engineStub.readLogs()) as EngineLogs;
|
|
const stepResultsBefore = logsBefore.logs
|
|
.filter((log) => log.event === InstanceEvent.STEP_SUCCESS)
|
|
.map((log) => log.metadata.result);
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("restart", {
|
|
name: "checkpoint",
|
|
});
|
|
});
|
|
} catch (e) {
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
const restartedStub = env.ENGINE.get(engineId);
|
|
|
|
await runInDurableObject(restartedStub, async (engine) => {
|
|
await engine.attemptRestart();
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const status = await runInDurableObject(restartedStub, (engine) =>
|
|
engine.getStatus()
|
|
);
|
|
return status === InstanceStatus.Complete;
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logsAfter = (await restartedStub.readLogs()) as EngineLogs;
|
|
const stepResultsAfter = logsAfter.logs
|
|
.filter((log) => log.event === InstanceEvent.STEP_SUCCESS)
|
|
.map((log) => log.metadata.result);
|
|
|
|
expect(stepResultsAfter[0]).toEqual(stepResultsBefore[0]);
|
|
expect(stepResultsAfter[1]).toEqual(stepResultsBefore[1]);
|
|
expect(stepResultsAfter[2]).not.toEqual(stepResultsBefore[2]);
|
|
expect(stepResultsAfter[3]).not.toEqual(stepResultsBefore[3]);
|
|
});
|
|
|
|
it("should throw when restarting from a non-existing step", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "RESTART-FROM-BAD-STEP";
|
|
|
|
const engineStub = await runWorkflowAndAwait(
|
|
instanceId,
|
|
async (_event: unknown, step: WorkflowStep) => {
|
|
await step.do("step-a", async () => "result-a");
|
|
return "done";
|
|
}
|
|
);
|
|
|
|
await expect(
|
|
runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("restart", {
|
|
name: "nonexistent-step",
|
|
});
|
|
})
|
|
).rejects.toThrow("not found in execution history");
|
|
});
|
|
|
|
it("should pause after in-flight step.do finishes", async ({ expect }) => {
|
|
const instanceId = "PAUSE-AFTER-DO";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
|
|
const engineStub = await runWorkflow(instanceId, async (_event, step) => {
|
|
await step.do("long-step", async () => {
|
|
await scheduler.wait(500);
|
|
return "first";
|
|
});
|
|
// step-2 should never run because pause takes effect after long-step
|
|
await step.do("step-2", async () => "second");
|
|
return "done";
|
|
});
|
|
|
|
// Wait for long-step to start
|
|
await vi.waitUntil(
|
|
async () => {
|
|
return await runInDurableObject(engineStub, (engine) => {
|
|
const logs = engine.readLogs() as unknown as EngineLogs;
|
|
return logs.logs.some(
|
|
(log) => log.event === InstanceEvent.STEP_START
|
|
);
|
|
});
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// Request pause while long-step is in flight
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("pause");
|
|
});
|
|
} catch (e) {
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
return await runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
async (engine) =>
|
|
(await engine.getStatus()) === InstanceStatus.Paused
|
|
);
|
|
} catch (e) {
|
|
if (isAbortError(e)) {
|
|
return false;
|
|
}
|
|
throw e;
|
|
}
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const freshStub = env.ENGINE.get(engineId);
|
|
const finalStatus = await runInDurableObject(freshStub, (engine) =>
|
|
engine.getStatus()
|
|
);
|
|
expect(finalStatus).toBe(InstanceStatus.Paused);
|
|
|
|
// Verify long-step completed but step-2 never ran
|
|
const logs = await runInDurableObject(freshStub, (engine) => {
|
|
return engine.readLogs() as unknown as EngineLogs;
|
|
});
|
|
const stepSuccesses = logs.logs.filter(
|
|
(log) => log.event === InstanceEvent.STEP_SUCCESS
|
|
);
|
|
expect(stepSuccesses).toHaveLength(1);
|
|
});
|
|
|
|
it("should pause after multiple concurrent in-flight step.dos finish", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "PAUSE-AFTER-CONCURRENT-DOS";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
|
|
const engineStub = await runWorkflow(instanceId, async (_event, step) => {
|
|
const [resultA, resultB] = await Promise.all([
|
|
step.do("slow-step-a", async () => {
|
|
await scheduler.wait(500);
|
|
return "a-done";
|
|
}),
|
|
step.do("slow-step-b", async () => {
|
|
await scheduler.wait(500);
|
|
return "b-done";
|
|
}),
|
|
]);
|
|
|
|
// This step should never run
|
|
await step.do("step-after-pause", async () => "should-not-run");
|
|
return { resultA, resultB };
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
return await runInDurableObject(engineStub, (engine) => {
|
|
const logs = engine.readLogs() as unknown as EngineLogs;
|
|
return (
|
|
logs.logs.filter((log) => log.event === InstanceEvent.STEP_START)
|
|
.length >= 2
|
|
);
|
|
});
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// Request pause while both slow steps are in flight
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("pause");
|
|
});
|
|
} catch (e) {
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
return await runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
async (engine) =>
|
|
(await engine.getStatus()) === InstanceStatus.Paused
|
|
);
|
|
} catch (e) {
|
|
if (isAbortError(e)) {
|
|
return false;
|
|
}
|
|
throw e;
|
|
}
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const freshStub = env.ENGINE.get(engineId);
|
|
const finalStatus = await runInDurableObject(freshStub, (engine) =>
|
|
engine.getStatus()
|
|
);
|
|
expect(finalStatus).toBe(InstanceStatus.Paused);
|
|
|
|
// Both concurrent steps should have completed, but step-after-pause should not
|
|
const logs = await runInDurableObject(freshStub, (engine) => {
|
|
return engine.readLogs() as unknown as EngineLogs;
|
|
});
|
|
const stepSuccesses = logs.logs.filter(
|
|
(log) => log.event === InstanceEvent.STEP_SUCCESS
|
|
);
|
|
expect(stepSuccesses).toHaveLength(2);
|
|
});
|
|
|
|
it("should unblock concurrent steps blocked when resume cancels pending pause", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "RESUME-UNBLOCKS-WAITING-STEPS";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
|
|
const engineStub = await runWorkflow(instanceId, async (_event, step) => {
|
|
const [slowResult, fastResult] = await Promise.all([
|
|
step.do("slow-step", async () => {
|
|
await scheduler.wait(1000);
|
|
return "slow-done";
|
|
}),
|
|
step.do("fast-step", async () => {
|
|
return "fast-done";
|
|
}),
|
|
]);
|
|
|
|
await step.do("step-after-resume", async () => "after-resume");
|
|
return { slowResult, fastResult };
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
return await runInDurableObject(engineStub, (engine) => {
|
|
const logs = engine.readLogs() as unknown as EngineLogs;
|
|
return logs.logs.some(
|
|
(log) => log.event === InstanceEvent.STEP_START
|
|
);
|
|
});
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("pause");
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () =>
|
|
runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
async (engine) =>
|
|
(await engine.getStatus()) === InstanceStatus.WaitingForPause
|
|
),
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// Resume before slow-step finishes
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("resume");
|
|
});
|
|
|
|
// Verify status goes back to Running
|
|
const statusAfterResume = await runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
(engine) => engine.getStatus()
|
|
);
|
|
expect(statusAfterResume).toBe(InstanceStatus.Running);
|
|
|
|
await vi.waitUntil(
|
|
async () =>
|
|
runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
async (engine) =>
|
|
(await engine.getStatus()) === InstanceStatus.Complete
|
|
),
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const freshStub = env.ENGINE.get(engineId);
|
|
const finalStatus = await runInDurableObject(freshStub, (engine) =>
|
|
engine.getStatus()
|
|
);
|
|
expect(finalStatus).toBe(InstanceStatus.Complete);
|
|
|
|
// All three steps should have completed
|
|
const logs = await runInDurableObject(freshStub, (engine) => {
|
|
return engine.readLogs() as unknown as EngineLogs;
|
|
});
|
|
const stepSuccesses = logs.logs.filter(
|
|
(log) => log.event === InstanceEvent.STEP_SUCCESS
|
|
);
|
|
expect(stepSuccesses).toHaveLength(3);
|
|
});
|
|
|
|
it("should pause during a step.sleep, then resume and complete", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "PAUSE-DURING-SLEEP-RESUME";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
|
|
const engineStub = await runWorkflow(instanceId, async (_event, step) => {
|
|
await step.do("first-step", async () => "first-done");
|
|
|
|
await step.sleep("short-sleep", "1 second");
|
|
|
|
await step.do("after-sleep", async () => "after-sleep-done");
|
|
return "done";
|
|
});
|
|
|
|
// Wait for the first step to complete (workflow is now in the sleep)
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
return await runInDurableObject(engineStub, (engine) => {
|
|
const logs = engine.readLogs() as unknown as EngineLogs;
|
|
return logs.logs.some(
|
|
(log) => log.event === InstanceEvent.STEP_SUCCESS
|
|
);
|
|
});
|
|
} catch (e) {
|
|
if (isAbortError(e)) {
|
|
return false;
|
|
}
|
|
throw e;
|
|
}
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// Request pause while in step.sleep — should pause immediately
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("pause");
|
|
});
|
|
} catch (e) {
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
return await runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
async (engine) =>
|
|
(await engine.getStatus()) === InstanceStatus.Paused
|
|
);
|
|
} catch (e) {
|
|
if (isAbortError(e)) {
|
|
return false;
|
|
}
|
|
throw e;
|
|
}
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
expect(
|
|
await runInDurableObject(env.ENGINE.get(engineId), (engine) =>
|
|
engine.getStatus()
|
|
)
|
|
).toBe(InstanceStatus.Paused);
|
|
|
|
// Only the first step should have succeeded — sleep was interrupted
|
|
const logsBeforeResume = await runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
(engine) => engine.readLogs() as unknown as EngineLogs
|
|
);
|
|
expect(
|
|
logsBeforeResume.logs.filter(
|
|
(log) => log.event === InstanceEvent.STEP_SUCCESS
|
|
)
|
|
).toHaveLength(1);
|
|
|
|
// Resume the workflow
|
|
try {
|
|
await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
|
|
await engine.changeInstanceStatus("resume");
|
|
});
|
|
} catch (e) {
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
// Wait for the workflow to complete — sleep should finish and after-sleep step should run
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
return await runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
async (engine) =>
|
|
(await engine.getStatus()) === InstanceStatus.Complete
|
|
);
|
|
} catch (e) {
|
|
if (isAbortError(e)) {
|
|
return false;
|
|
}
|
|
throw e;
|
|
}
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const finalLogs = await runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
(engine) => engine.readLogs() as unknown as EngineLogs
|
|
);
|
|
const stepSuccesses = finalLogs.logs.filter(
|
|
(log) => log.event === InstanceEvent.STEP_SUCCESS
|
|
);
|
|
// first-step + after-sleep = 2 step successes
|
|
expect(stepSuccesses).toHaveLength(2);
|
|
});
|
|
|
|
it("should pause during a waitForEvent, then resume and complete", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "PAUSE-DURING-WAIT-FOR-EVENT";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
|
|
const engineStub = await runWorkflow(instanceId, async (_event, step) => {
|
|
await step.do("first-step", async () => "first-done");
|
|
|
|
await step.waitForEvent("wait-for-signal", {
|
|
type: "continue",
|
|
timeout: "30 seconds",
|
|
});
|
|
|
|
await step.do("after-event", async () => "after-event-done");
|
|
return "done";
|
|
});
|
|
|
|
// Wait for the first step to complete (workflow is now waiting for event)
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
return await runInDurableObject(engineStub, (engine) => {
|
|
const logs = engine.readLogs() as unknown as EngineLogs;
|
|
return logs.logs.some(
|
|
(log) => log.event === InstanceEvent.WAIT_START
|
|
);
|
|
});
|
|
} catch (e) {
|
|
if (isAbortError(e)) {
|
|
return false;
|
|
}
|
|
throw e;
|
|
}
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// Request pause while in waitForEvent
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("pause");
|
|
});
|
|
} catch (e) {
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
return await runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
async (engine) =>
|
|
(await engine.getStatus()) === InstanceStatus.Paused
|
|
);
|
|
} catch (e) {
|
|
if (isAbortError(e)) {
|
|
return false;
|
|
}
|
|
throw e;
|
|
}
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
expect(
|
|
await runInDurableObject(env.ENGINE.get(engineId), (engine) =>
|
|
engine.getStatus()
|
|
)
|
|
).toBe(InstanceStatus.Paused);
|
|
|
|
// Resume the workflow
|
|
try {
|
|
await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
|
|
await engine.changeInstanceStatus("resume");
|
|
});
|
|
} catch (e) {
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
// Send the event after resume
|
|
await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
|
|
await engine.receiveEvent({
|
|
timestamp: new Date(),
|
|
payload: { done: true },
|
|
type: "continue",
|
|
});
|
|
});
|
|
|
|
// Wait for the workflow to complete
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
return await runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
async (engine) =>
|
|
(await engine.getStatus()) === InstanceStatus.Complete
|
|
);
|
|
} catch (e) {
|
|
if (isAbortError(e)) {
|
|
return false;
|
|
}
|
|
throw e;
|
|
}
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const finalLogs = await runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
(engine) => engine.readLogs() as unknown as EngineLogs
|
|
);
|
|
const stepSuccesses = finalLogs.logs.filter(
|
|
(log) => log.event === InstanceEvent.STEP_SUCCESS
|
|
);
|
|
// first-step + after-event = 2 step successes
|
|
expect(stepSuccesses).toHaveLength(2);
|
|
});
|
|
|
|
it("should transition WaitingForPause to Paused on init() entry", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "WAITING-FOR-PAUSE-INIT";
|
|
|
|
const engineStub = await runWorkflowAndAwait(
|
|
instanceId,
|
|
async () => "done"
|
|
);
|
|
|
|
// Manually set status to WaitingForPause (simulating a DO restart scenario)
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
await state.storage.put(
|
|
"ENGINE_STATUS",
|
|
InstanceStatus.WaitingForPause
|
|
);
|
|
});
|
|
|
|
// Now call init() — it should detect WaitingForPause and transition to Paused
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
// Reset isRunning so init() doesn't short-circuit
|
|
engine.isRunning = false;
|
|
await engine.init(
|
|
12346,
|
|
{} as DatabaseWorkflow,
|
|
{} as DatabaseVersion,
|
|
{ id: instanceId } as DatabaseInstance,
|
|
{
|
|
payload: {},
|
|
timestamp: new Date(),
|
|
instanceId,
|
|
workflowName: "",
|
|
}
|
|
);
|
|
});
|
|
|
|
const status = await runInDurableObject(engineStub, (engine) =>
|
|
engine.getStatus()
|
|
);
|
|
expect(status).toBe(InstanceStatus.Paused);
|
|
});
|
|
});
|
|
});
|
|
|
|
// RPC-stubbed rollback fns can't reliably mutate test-scope closures —
|
|
// assert via engine log events instead.
|
|
describe("Rollback", () => {
|
|
async function readLogsAfter(
|
|
stub: { readLogs(): Promise<EngineLogs> | EngineLogs },
|
|
predicate: (logs: EngineLogs) => boolean,
|
|
timeout = 5000
|
|
): Promise<EngineLogs> {
|
|
await vi.waitUntil(
|
|
async () => predicate((await stub.readLogs()) as EngineLogs),
|
|
{ timeout }
|
|
);
|
|
return (await stub.readLogs()) as EngineLogs;
|
|
}
|
|
|
|
function targetsOf(
|
|
logs: EngineLogs,
|
|
event: InstanceEvent
|
|
): (string | null)[] {
|
|
return logs.logs.filter((l) => l.event === event).map((l) => l.target);
|
|
}
|
|
|
|
function countOf(logs: EngineLogs, event: InstanceEvent): number {
|
|
return logs.logs.filter((l) => l.event === event).length;
|
|
}
|
|
|
|
function doWithRollback<T>(
|
|
step: WorkflowStep,
|
|
name: string,
|
|
callback: (ctx: unknown) => Promise<T>,
|
|
options: WorkflowStepRollbackOptions
|
|
): Promise<unknown>;
|
|
function doWithRollback<T>(
|
|
step: WorkflowStep,
|
|
name: string,
|
|
config: WorkflowStepConfig,
|
|
callback: (ctx: unknown) => Promise<T>,
|
|
options: WorkflowStepRollbackOptions
|
|
): Promise<unknown>;
|
|
function doWithRollback<T>(
|
|
step: WorkflowStep,
|
|
name: string,
|
|
configOrCallback: WorkflowStepConfig | ((ctx: unknown) => Promise<T>),
|
|
callbackOrOptions:
|
|
| ((ctx: unknown) => Promise<T>)
|
|
| WorkflowStepRollbackOptions,
|
|
options?: WorkflowStepRollbackOptions
|
|
): Promise<unknown> {
|
|
if (typeof configOrCallback === "function") {
|
|
// @ts-expect-error -- rollback options are not in workers-types yet
|
|
return step.do(name, configOrCallback, callbackOrOptions);
|
|
}
|
|
// @ts-expect-error -- rollback options are not in workers-types yet
|
|
return step.do(name, configOrCallback, callbackOrOptions, options);
|
|
}
|
|
|
|
async function noopRollback(_ctx: RollbackContext): Promise<void> {}
|
|
|
|
function rollbackOptions(
|
|
fn: RollbackFn = noopRollback
|
|
): WorkflowStepRollbackOptions {
|
|
return { rollback: fn };
|
|
}
|
|
|
|
function rollbackOptionsWithConfig(
|
|
rollbackConfig: WorkflowStepConfig,
|
|
fn: RollbackFn = noopRollback
|
|
): WorkflowStepRollbackOptions {
|
|
return { rollback: fn, rollbackConfig };
|
|
}
|
|
|
|
it("runs rollback fns in LIFO order on workflow failure", async ({
|
|
expect,
|
|
}) => {
|
|
const stub = await runWorkflowAndAwait("RB-LIFO", async (_e, step) => {
|
|
await doWithRollback(
|
|
step,
|
|
"step-1",
|
|
async () => "out-1",
|
|
rollbackOptions()
|
|
);
|
|
await doWithRollback(
|
|
step,
|
|
"step-2",
|
|
async () => "out-2",
|
|
rollbackOptions()
|
|
);
|
|
await doWithRollback(
|
|
step,
|
|
"step-3",
|
|
async () => "out-3",
|
|
rollbackOptions()
|
|
);
|
|
throw new Error("boom");
|
|
});
|
|
const logs = await readLogsAfter(stub, (l) =>
|
|
l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE)
|
|
);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"step-3-1",
|
|
"step-2-1",
|
|
"step-1-1",
|
|
]);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_ATTEMPT_SUCCESS)).toEqual([
|
|
"step-3-1",
|
|
"step-2-1",
|
|
"step-1-1",
|
|
]);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_FAILED)).toBe(0);
|
|
});
|
|
|
|
it("runs parallel rollbacks in reverse step start order", async ({
|
|
expect,
|
|
}) => {
|
|
let firstStarted: () => void = () => {};
|
|
let firstMayFinish: () => void = () => {};
|
|
const firstStartedPromise = new Promise<void>((resolve) => {
|
|
firstStarted = resolve;
|
|
});
|
|
const firstMayFinishPromise = new Promise<void>((resolve) => {
|
|
firstMayFinish = resolve;
|
|
});
|
|
|
|
const stub = await runWorkflowAndAwait("RB-PARALLEL", async (_e, step) => {
|
|
const first = doWithRollback(
|
|
step,
|
|
"first",
|
|
async () => {
|
|
firstStarted();
|
|
await firstMayFinishPromise;
|
|
await scheduler.wait(50);
|
|
return "out-1";
|
|
},
|
|
rollbackOptions()
|
|
);
|
|
await firstStartedPromise;
|
|
|
|
const second = doWithRollback(
|
|
step,
|
|
"second",
|
|
async () => {
|
|
firstMayFinish();
|
|
return "out-2";
|
|
},
|
|
rollbackOptions()
|
|
);
|
|
|
|
await Promise.all([first, second]);
|
|
throw new Error("boom");
|
|
});
|
|
const logs = await readLogsAfter(stub, (l) =>
|
|
l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE)
|
|
);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"second-1",
|
|
"first-1",
|
|
]);
|
|
});
|
|
|
|
it("uses rollbackConfig when executing rollback", async ({ expect }) => {
|
|
const rollbackConfig = {
|
|
retries: { limit: 0, delay: 0, backoff: "constant" },
|
|
timeout: "30 seconds",
|
|
} satisfies WorkflowStepConfig;
|
|
const stub = await runWorkflowAndAwait("RB-CONFIG", async (_e, step) => {
|
|
await doWithRollback(
|
|
step,
|
|
"configured-step",
|
|
async () => "out",
|
|
rollbackOptionsWithConfig(rollbackConfig)
|
|
);
|
|
throw new Error("boom");
|
|
});
|
|
const logs = await readLogsAfter(stub, (l) =>
|
|
l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE)
|
|
);
|
|
expect(
|
|
logs.logs.find((l) => l.event === InstanceEvent.ROLLBACK_STEP_START)
|
|
?.metadata
|
|
).toMatchObject({ config: rollbackConfig });
|
|
});
|
|
|
|
it("passes rollback context to rollback fn", async ({ expect }) => {
|
|
const stub = await runWorkflowAndAwait("RB-CONTEXT", async (_e, step) => {
|
|
await doWithRollback(
|
|
step,
|
|
"ctx-step",
|
|
async () => "out",
|
|
rollbackOptions(async (ctx) => {
|
|
if (
|
|
ctx.error.name !== "Error" ||
|
|
ctx.error.message !== "boom" ||
|
|
ctx.output !== "out" ||
|
|
ctx.ctx.step.name !== "ctx-step" ||
|
|
ctx.ctx.step.count !== 1 ||
|
|
ctx.stepName !== "ctx-step-1"
|
|
) {
|
|
throw new Error("unexpected rollback context");
|
|
}
|
|
})
|
|
);
|
|
throw new Error("boom");
|
|
});
|
|
const logs = await readLogsAfter(stub, (l) =>
|
|
l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE)
|
|
);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"ctx-step-1",
|
|
]);
|
|
});
|
|
|
|
it("runs rollback for failed step", async ({ expect }) => {
|
|
const stub = await runWorkflowAndAwait(
|
|
"RB-FAILED-STEP",
|
|
async (_e, step) => {
|
|
await doWithRollback(
|
|
step,
|
|
"failed-step",
|
|
{ retries: { limit: 0, delay: 0, backoff: "constant" } },
|
|
async () => {
|
|
throw new Error("step-boom");
|
|
},
|
|
rollbackOptions(async (ctx) => {
|
|
if (
|
|
ctx.error.message !== "step-boom" ||
|
|
ctx.output !== undefined ||
|
|
ctx.ctx.step.name !== "failed-step" ||
|
|
ctx.ctx.step.count !== 1 ||
|
|
ctx.stepName !== "failed-step-1"
|
|
) {
|
|
throw new Error("unexpected failed-step rollback context");
|
|
}
|
|
})
|
|
);
|
|
}
|
|
);
|
|
const logs = await readLogsAfter(stub, (l) =>
|
|
l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE)
|
|
);
|
|
const rollbackStepTargets = targetsOf(
|
|
logs,
|
|
InstanceEvent.ROLLBACK_STEP_SUCCESS
|
|
);
|
|
const stepFailureTargets = targetsOf(logs, InstanceEvent.STEP_FAILURE);
|
|
|
|
expect(rollbackStepTargets).toEqual(["failed-step-1"]);
|
|
expect(stepFailureTargets).toContain("failed-step-1");
|
|
});
|
|
|
|
it("only runs rollbacks for steps with a registered fn", async ({
|
|
expect,
|
|
}) => {
|
|
const stub = await runWorkflowAndAwait("RB-PARTIAL", async (_e, step) => {
|
|
await step.do("plain-step", async () => "v1");
|
|
await doWithRollback(
|
|
step,
|
|
"step-with-rollback",
|
|
async () => "v2",
|
|
rollbackOptions()
|
|
);
|
|
await step.do("plain-step-after", async () => "v3");
|
|
throw new Error("boom");
|
|
});
|
|
const logs = await readLogsAfter(stub, (l) =>
|
|
l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE)
|
|
);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"step-with-rollback-1",
|
|
]);
|
|
expect(
|
|
logs.logs.find((l) => l.event === InstanceEvent.ROLLBACK_START)?.metadata
|
|
).toMatchObject({ totalSteps: 1 });
|
|
});
|
|
|
|
it("stops at the first failing rollback and logs ROLLBACK_FAILED", async ({
|
|
expect,
|
|
}) => {
|
|
const stub = await runWorkflowAndAwait("RB-FAILS", async (_e, step) => {
|
|
await doWithRollback(step, "step-1", async () => "v1", rollbackOptions());
|
|
await doWithRollback(
|
|
step,
|
|
"step-2",
|
|
async () => "v2",
|
|
rollbackOptionsWithConfig(
|
|
{ retries: { limit: 0, delay: 0, backoff: "constant" } },
|
|
async () => {
|
|
throw new Error("rollback-boom");
|
|
}
|
|
)
|
|
);
|
|
await doWithRollback(step, "step-3", async () => "v3", rollbackOptions());
|
|
throw new Error("boom");
|
|
});
|
|
const logs = await readLogsAfter(stub, (l) =>
|
|
l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_FAILED)
|
|
);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"step-3-1",
|
|
]);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_FAILURE)).toEqual([
|
|
"step-2-1",
|
|
]);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_ATTEMPT_FAILURE)).toEqual([
|
|
"step-2-1",
|
|
]);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(0);
|
|
});
|
|
|
|
it("runs rollback when terminate requests it", async ({ expect }) => {
|
|
const instanceId = "RB-TERMINATE";
|
|
const engineStub = await runWorkflow(instanceId, async (_e, step) => {
|
|
await doWithRollback(
|
|
step,
|
|
"setup-resource",
|
|
async () => "resource-id",
|
|
rollbackOptions()
|
|
);
|
|
await step.sleep("wait-forever", "1 hour");
|
|
});
|
|
|
|
await readLogsAfter(engineStub, (currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) =>
|
|
log.event === InstanceEvent.STEP_SUCCESS &&
|
|
log.target === "setup-resource-1"
|
|
)
|
|
);
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("terminate", undefined, {
|
|
rollback: true,
|
|
});
|
|
});
|
|
} catch (error) {
|
|
if (
|
|
!(error instanceof Error) ||
|
|
!error.message.startsWith("Aborting engine:")
|
|
) {
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
const logs = await readLogsAfter(
|
|
env.ENGINE.get(env.ENGINE.idFromName(instanceId)),
|
|
(currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) => log.event === InstanceEvent.WORKFLOW_TERMINATED
|
|
)
|
|
);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"setup-resource-1",
|
|
]);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
|
|
});
|
|
|
|
it("replays cached steps before terminate rollback when registry is empty", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "RB-TERMINATE-REPLAY";
|
|
const engineStub = await runWorkflow(instanceId, async (_e, step) => {
|
|
await doWithRollback(
|
|
step,
|
|
"setup-resource",
|
|
async () => "resource-id",
|
|
rollbackOptions()
|
|
);
|
|
await step.sleep("wait-forever", "1 hour");
|
|
});
|
|
|
|
await readLogsAfter(engineStub, (currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) =>
|
|
log.event === InstanceEvent.STEP_SUCCESS &&
|
|
log.target === "setup-resource-1"
|
|
)
|
|
);
|
|
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
engine.rollbackRegistry.clear();
|
|
});
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("terminate", undefined, {
|
|
rollback: true,
|
|
});
|
|
});
|
|
} catch (error) {
|
|
if (
|
|
!(error instanceof Error) ||
|
|
!error.message.startsWith("Aborting engine:")
|
|
) {
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
const logs = await readLogsAfter(
|
|
env.ENGINE.get(env.ENGINE.idFromName(instanceId)),
|
|
(currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) => log.event === InstanceEvent.WORKFLOW_TERMINATED
|
|
)
|
|
);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"setup-resource-1",
|
|
]);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
|
|
});
|
|
|
|
it("replays started unfinished steps before terminate rollback when registry is empty", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = `RB-TERMINATE-STARTED-REPLAY-${crypto.randomUUID()}`;
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const engineStub = await runWorkflow(instanceId, async (_e, step) => {
|
|
await doWithRollback(
|
|
step,
|
|
"started-setup",
|
|
async () => {
|
|
await scheduler.wait(30_000);
|
|
return "late-resource-id";
|
|
},
|
|
rollbackOptions()
|
|
);
|
|
});
|
|
|
|
await readLogsAfter(engineStub, (currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) =>
|
|
log.event === InstanceEvent.ATTEMPT_START &&
|
|
log.target === "started-setup-1"
|
|
)
|
|
);
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.abort("test restart before step completion");
|
|
});
|
|
} catch (error) {
|
|
if (
|
|
!(error instanceof Error) ||
|
|
error.message !== "test restart before step completion"
|
|
) {
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
const restartedStub = env.ENGINE.get(engineId);
|
|
await runInDurableObject(restartedStub, async (engine) => {
|
|
engine.rollbackRegistry.clear();
|
|
});
|
|
|
|
try {
|
|
await runInDurableObject(restartedStub, async (engine) => {
|
|
await engine.changeInstanceStatus("terminate", undefined, {
|
|
rollback: true,
|
|
});
|
|
});
|
|
} catch (error) {
|
|
if (!isAbortError(error)) {
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
let logs: EngineLogs | undefined;
|
|
await vi.waitUntil(
|
|
async () => {
|
|
try {
|
|
logs = (await env.ENGINE.get(engineId).readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(log) => log.event === InstanceEvent.WORKFLOW_TERMINATED
|
|
);
|
|
} catch (error) {
|
|
if (isAbortError(error)) {
|
|
return false;
|
|
}
|
|
throw error;
|
|
}
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
if (logs === undefined) {
|
|
throw new Error("Expected workflow logs after terminate");
|
|
}
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"started-setup-1",
|
|
]);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
|
|
});
|
|
|
|
it("fails terminate rollback when replay cannot rebuild an eligible handler", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = `RB-TERMINATE-MISSING-REPLAY-${crypto.randomUUID()}`;
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const engineStub = await runWorkflow(instanceId, async (_e, step) => {
|
|
await doWithRollback(
|
|
step,
|
|
"branchy-setup",
|
|
async () => "resource-id",
|
|
rollbackOptions()
|
|
);
|
|
await step.sleep("wait-forever", "1 hour");
|
|
});
|
|
|
|
await readLogsAfter(engineStub, (currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) =>
|
|
log.event === InstanceEvent.STEP_SUCCESS &&
|
|
log.target === "branchy-setup-1"
|
|
)
|
|
);
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.abort("test restart before replay branch changes");
|
|
});
|
|
} catch (error) {
|
|
if (
|
|
!(error instanceof Error) ||
|
|
error.message !== "test restart before replay branch changes"
|
|
) {
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
setTestWorkflowCallback(async () => {
|
|
// Simulate nondeterministic replay control flow that no longer reaches
|
|
// the persisted eligible step. Production treats this as rollback failure.
|
|
});
|
|
|
|
try {
|
|
await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
|
|
await engine.changeInstanceStatus("terminate", undefined, {
|
|
rollback: true,
|
|
});
|
|
});
|
|
} catch (error) {
|
|
if (!isAbortError(error)) {
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
const logs = await readLogsAfter(env.ENGINE.get(engineId), (currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) => log.event === InstanceEvent.WORKFLOW_TERMINATED
|
|
)
|
|
);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_FAILURE)).toEqual([
|
|
"branchy-setup-1",
|
|
]);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_FAILED)).toBe(1);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(0);
|
|
});
|
|
|
|
it("replays cached failed steps before terminate rollback when registry is empty", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "RB-TERMINATE-FAILED-REPLAY";
|
|
const engineStub = await runWorkflow(instanceId, async (_e, step) => {
|
|
try {
|
|
await doWithRollback(
|
|
step,
|
|
"failed-setup",
|
|
{ retries: { limit: 0, delay: 0, backoff: "constant" } },
|
|
async () => {
|
|
throw new Error("step-boom");
|
|
},
|
|
rollbackOptions()
|
|
);
|
|
} catch {
|
|
// Keep the workflow running so terminate rollback can replay the cached
|
|
// failed step and re-register its rollback handler.
|
|
}
|
|
await step.sleep("wait-forever", "1 hour");
|
|
});
|
|
|
|
await readLogsAfter(engineStub, (currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) =>
|
|
log.event === InstanceEvent.STEP_FAILURE &&
|
|
log.target === "failed-setup-1"
|
|
)
|
|
);
|
|
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
engine.rollbackRegistry.clear();
|
|
});
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("terminate", undefined, {
|
|
rollback: true,
|
|
});
|
|
});
|
|
} catch (error) {
|
|
if (
|
|
!(error instanceof Error) ||
|
|
!error.message.startsWith("Aborting engine:")
|
|
) {
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
const logs = await readLogsAfter(
|
|
env.ENGINE.get(env.ENGINE.idFromName(instanceId)),
|
|
(currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) => log.event === InstanceEvent.WORKFLOW_TERMINATED
|
|
)
|
|
);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"failed-setup-1",
|
|
]);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
|
|
});
|
|
|
|
it("omits dynamic delay from rollback ctx.config on cached-error replay", async ({
|
|
expect,
|
|
}) => {
|
|
const dynamicDelay: WorkflowDelayFunction = () => 1000;
|
|
const instanceId = "RB-TERMINATE-FAILED-REPLAY-DYNAMIC-DELAY";
|
|
const engineStub = await runWorkflow(instanceId, async (_e, step) => {
|
|
try {
|
|
await doWithRollback(
|
|
step,
|
|
"failed-setup",
|
|
{ retries: { limit: 0, delay: dynamicDelay, backoff: "constant" } },
|
|
async () => {
|
|
throw new Error("step-boom");
|
|
},
|
|
rollbackOptions(async (ctx) => {
|
|
// A dynamic delay must be omitted from the user-facing config,
|
|
// never surfaced as the internal "[dynamic]" marker string.
|
|
if ("delay" in ctx.ctx.config.retries) {
|
|
throw new Error(
|
|
"dynamic delay marker leaked into rollback ctx.config"
|
|
);
|
|
}
|
|
})
|
|
);
|
|
} catch {
|
|
// Keep the workflow running so terminate rollback can replay the cached
|
|
// failed step and re-register its rollback handler.
|
|
}
|
|
await step.sleep("wait-forever", "1 hour");
|
|
});
|
|
|
|
await readLogsAfter(engineStub, (currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) =>
|
|
log.event === InstanceEvent.STEP_FAILURE &&
|
|
log.target === "failed-setup-1"
|
|
)
|
|
);
|
|
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
engine.rollbackRegistry.clear();
|
|
});
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("terminate", undefined, {
|
|
rollback: true,
|
|
});
|
|
});
|
|
} catch (error) {
|
|
if (
|
|
!(error instanceof Error) ||
|
|
!error.message.startsWith("Aborting engine:")
|
|
) {
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
const logs = await readLogsAfter(
|
|
env.ENGINE.get(env.ENGINE.idFromName(instanceId)),
|
|
(currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) => log.event === InstanceEvent.WORKFLOW_TERMINATED
|
|
)
|
|
);
|
|
// ROLLBACK_STEP_SUCCESS only fires if the rollback fn did not throw, i.e.
|
|
// the leaked marker was stripped by toEngineStepConfig on the replay path.
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"failed-setup-1",
|
|
]);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
|
|
});
|
|
|
|
it("runs terminate rollback while paused with an empty registry", async ({
|
|
expect,
|
|
}) => {
|
|
const instanceId = "RB-TERMINATE-PAUSED";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const engineStub = await runWorkflow(instanceId, async (_e, step) => {
|
|
await doWithRollback(
|
|
step,
|
|
"setup-resource",
|
|
async () => "resource-id",
|
|
rollbackOptions()
|
|
);
|
|
await step.sleep("wait-forever", "1 hour");
|
|
});
|
|
|
|
await readLogsAfter(engineStub, (currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) =>
|
|
log.event === InstanceEvent.STEP_SUCCESS &&
|
|
log.target === "setup-resource-1"
|
|
)
|
|
);
|
|
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("pause");
|
|
});
|
|
} catch (error) {
|
|
if (!isAbortError(error)) {
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
await vi.waitUntil(
|
|
async () =>
|
|
runInDurableObject(
|
|
env.ENGINE.get(engineId),
|
|
async (engine) => (await engine.getStatus()) === InstanceStatus.Paused
|
|
),
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
|
|
engine.rollbackRegistry.clear();
|
|
});
|
|
|
|
try {
|
|
await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
|
|
await engine.changeInstanceStatus("terminate", undefined, {
|
|
rollback: true,
|
|
});
|
|
});
|
|
} catch (error) {
|
|
if (!isAbortError(error)) {
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
const logs = await readLogsAfter(env.ENGINE.get(engineId), (currentLogs) =>
|
|
currentLogs.logs.some(
|
|
(log) => log.event === InstanceEvent.WORKFLOW_TERMINATED
|
|
)
|
|
);
|
|
expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
|
|
"setup-resource-1",
|
|
]);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
|
|
});
|
|
|
|
it("does not run rollback when workflow succeeds", async ({ expect }) => {
|
|
const stub = await runWorkflowAndAwait("RB-NOOP", async (_e, step) => {
|
|
await doWithRollback(step, "a", async () => "ok", rollbackOptions());
|
|
return "done";
|
|
});
|
|
const logs = await readLogsAfter(stub, (l) =>
|
|
l.logs.some((r) => r.event === InstanceEvent.WORKFLOW_SUCCESS)
|
|
);
|
|
expect(countOf(logs, InstanceEvent.ROLLBACK_START)).toBe(0);
|
|
});
|
|
});
|