wehub-resource-sync
70bf21e064
Handle Changesets / Handle Changesets (push) Waiting to run
Semgrep OSS scan / semgrep-oss (push) Waiting to run
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
1454 lines
41 KiB
TypeScript
1454 lines
41 KiB
TypeScript
import { env, runInDurableObject } from "cloudflare:test";
|
|
import { afterEach, describe, it, vi } from "vitest";
|
|
import workerdUnsafe from "workerd:unsafe";
|
|
import { InstanceEvent } from "../src";
|
|
import { computeHash } from "../src/lib/cache";
|
|
import {
|
|
InvalidStepReadableStreamError,
|
|
OversizedStreamChunkError,
|
|
UnsupportedStreamChunkError,
|
|
WorkflowTimeoutError,
|
|
isAbortError,
|
|
} from "../src/lib/errors";
|
|
import {
|
|
STREAMING_STEP_CHUNKS_TABLE,
|
|
getStreamOutputMetaKey,
|
|
StreamOutputState,
|
|
rollbackStreamOutput,
|
|
writeStreamOutput,
|
|
} from "../src/lib/streams";
|
|
import { MODIFIER_KEYS } from "../src/modifier";
|
|
import { runWorkflow, runWorkflowAndAwait } from "./utils";
|
|
import type { Engine, EngineLogs } from "../src/engine";
|
|
import type { StreamOutputMeta } from "../src/lib/streams";
|
|
|
|
afterEach(async () => {
|
|
await workerdUnsafe.abortAllDurableObjects();
|
|
});
|
|
|
|
describe("Context", () => {
|
|
it("should provide attempt count 1 on first successful attempt", async ({
|
|
expect,
|
|
}) => {
|
|
let receivedAttempt: number | undefined;
|
|
|
|
const engineStub = await runWorkflow(
|
|
"MOCK-INSTANCE-ID",
|
|
async (_event, step) => {
|
|
const result = await step.do("a successful step", async (ctx) => {
|
|
receivedAttempt = ctx.attempt;
|
|
return "success";
|
|
});
|
|
return result;
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
expect(receivedAttempt).toBe(1);
|
|
});
|
|
|
|
it("should provide attempt count to callback", async ({ expect }) => {
|
|
const receivedAttempts: number[] = [];
|
|
|
|
const engineStub = await runWorkflow(
|
|
"MOCK-INSTANCE-ID-RETRY",
|
|
async (_event, step) => {
|
|
const result = await step.do(
|
|
"retrying step",
|
|
{
|
|
retries: {
|
|
limit: 2,
|
|
delay: 0,
|
|
},
|
|
},
|
|
async (ctx) => {
|
|
receivedAttempts.push(ctx.attempt);
|
|
throw new Error(`Throwing`);
|
|
}
|
|
);
|
|
return result;
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_FAILURE
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// Should have received attempts 1, 2, and 3
|
|
expect(receivedAttempts).toEqual([1, 2, 3]);
|
|
});
|
|
|
|
it("should wait for retry delays by default", async ({ expect }) => {
|
|
const start = Date.now();
|
|
const engineStub = await runWorkflowAndAwait(
|
|
"MOCK-INSTANCE-RETRY-DELAYS",
|
|
async (_event, step) => {
|
|
const result = await step.do(
|
|
"retrying step with delay",
|
|
{
|
|
retries: {
|
|
limit: 1,
|
|
delay: "1 second",
|
|
backoff: "constant",
|
|
},
|
|
},
|
|
async () => {
|
|
throw new Error("Always fails");
|
|
}
|
|
);
|
|
return result;
|
|
}
|
|
);
|
|
const elapsed = Date.now() - start;
|
|
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
expect(
|
|
logs.logs.filter((val) => val.event === InstanceEvent.ATTEMPT_START)
|
|
).toHaveLength(2);
|
|
// Should have waited at least ~1 second for the retry delay
|
|
expect(elapsed).toBeGreaterThanOrEqual(900);
|
|
});
|
|
|
|
it("should skip retry delays when disableRetryDelays is set", async ({
|
|
expect,
|
|
}) => {
|
|
const engineId = env.ENGINE.idFromName("MOCK-INSTANCE-DISABLE-RETRY");
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
// Set the disableRetryDelays flag before running the workflow
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
await state.storage.put(MODIFIER_KEYS.DISABLE_ALL_RETRY_DELAYS, true);
|
|
});
|
|
|
|
const start = Date.now();
|
|
const stub = await runWorkflowAndAwait(
|
|
"MOCK-INSTANCE-DISABLE-RETRY",
|
|
async (_event, step) => {
|
|
const result = await step.do(
|
|
"retrying step with delay",
|
|
{
|
|
retries: {
|
|
limit: 2,
|
|
delay: "10 seconds",
|
|
backoff: "constant",
|
|
},
|
|
},
|
|
async () => {
|
|
throw new Error("Always fails");
|
|
}
|
|
);
|
|
return result;
|
|
}
|
|
);
|
|
const elapsed = Date.now() - start;
|
|
|
|
const logs = (await stub.readLogs()) as EngineLogs;
|
|
expect(
|
|
logs.logs.filter((val) => val.event === InstanceEvent.ATTEMPT_START)
|
|
).toHaveLength(3);
|
|
// Without disableRetryDelays, this would take 20+ seconds (10s + 10s)
|
|
expect(elapsed).toBeLessThan(5000);
|
|
});
|
|
|
|
// NOTE: The `step` and `config` fields on WorkflowStepContext are new additions
|
|
// that haven't been published to @cloudflare/workers-types yet. The runtime
|
|
// provides them, but TypeScript doesn't know about them, so we capture ctx as
|
|
// `unknown` and assert on the shape.
|
|
// TODO: Remove these workarounds once https://github.com/cloudflare/workerd/pull/6523 lands.
|
|
|
|
it("should provide step name and count in context", async ({ expect }) => {
|
|
let receivedCtx: unknown;
|
|
|
|
const engineStub = await runWorkflow(
|
|
"MOCK-INSTANCE-STEP-CTX",
|
|
async (_event, step) => {
|
|
const result = await step.do("my step", async (ctx) => {
|
|
receivedCtx = ctx;
|
|
return "done";
|
|
});
|
|
return result;
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
expect(receivedCtx).toMatchObject({
|
|
step: { name: "my step", count: 1 },
|
|
});
|
|
});
|
|
|
|
it("should increment step count for steps with the same name", async ({
|
|
expect,
|
|
}) => {
|
|
const receivedContexts: unknown[] = [];
|
|
|
|
const engineStub = await runWorkflow(
|
|
"MOCK-INSTANCE-STEP-COUNT",
|
|
async (_event, step) => {
|
|
await step.do("repeated step", async (ctx) => {
|
|
receivedContexts.push(ctx);
|
|
return "first";
|
|
});
|
|
await step.do("repeated step", async (ctx) => {
|
|
receivedContexts.push(ctx);
|
|
return "second";
|
|
});
|
|
await step.do("different step", async (ctx) => {
|
|
receivedContexts.push(ctx);
|
|
return "third";
|
|
});
|
|
return "done";
|
|
}
|
|
);
|
|
|
|
// Needs extra headroom: 3 sequential step.do calls + cold DO startup on Windows
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 10000 }
|
|
);
|
|
|
|
expect(receivedContexts[0]).toMatchObject({
|
|
step: { name: "repeated step", count: 1 },
|
|
});
|
|
expect(receivedContexts[1]).toMatchObject({
|
|
step: { name: "repeated step", count: 2 },
|
|
});
|
|
expect(receivedContexts[2]).toMatchObject({
|
|
step: { name: "different step", count: 1 },
|
|
});
|
|
});
|
|
|
|
it("should provide resolved config with defaults in context", async ({
|
|
expect,
|
|
}) => {
|
|
let receivedCtx: unknown;
|
|
|
|
const engineStub = await runWorkflow(
|
|
"MOCK-INSTANCE-CTX-CONFIG-DEFAULTS",
|
|
async (_event, step) => {
|
|
// No config provided — ctx.config should have all defaults
|
|
const result = await step.do("default config step", async (ctx) => {
|
|
receivedCtx = ctx;
|
|
return "done";
|
|
});
|
|
return result;
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
expect(receivedCtx).toMatchObject({
|
|
config: {
|
|
retries: {
|
|
limit: 5,
|
|
delay: 1000,
|
|
backoff: "exponential",
|
|
},
|
|
timeout: "10 minutes",
|
|
},
|
|
});
|
|
});
|
|
|
|
it("should provide resolved config with user overrides merged in context", async ({
|
|
expect,
|
|
}) => {
|
|
let receivedCtx: unknown;
|
|
|
|
const engineStub = await runWorkflow(
|
|
"MOCK-INSTANCE-CTX-CONFIG-OVERRIDES",
|
|
async (_event, step) => {
|
|
const result = await step.do(
|
|
"custom config step",
|
|
{
|
|
retries: {
|
|
limit: 3,
|
|
delay: 500,
|
|
},
|
|
timeout: "5 minutes",
|
|
},
|
|
async (ctx) => {
|
|
receivedCtx = ctx;
|
|
return "done";
|
|
}
|
|
);
|
|
return result;
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// User overrides should be merged with defaults
|
|
expect(receivedCtx).toMatchObject({
|
|
config: {
|
|
retries: {
|
|
limit: 3,
|
|
delay: 500,
|
|
backoff: "exponential",
|
|
},
|
|
timeout: "5 minutes",
|
|
},
|
|
});
|
|
});
|
|
|
|
it("should not allow user callback to mutate engine retry config", async ({
|
|
expect,
|
|
}) => {
|
|
const receivedAttempts: number[] = [];
|
|
|
|
const engineStub = await runWorkflow(
|
|
"MOCK-INSTANCE-CTX-CONFIG-MUTATION",
|
|
async (_event, step) => {
|
|
const result = await step.do(
|
|
"mutating step",
|
|
{
|
|
retries: {
|
|
limit: 1,
|
|
delay: 0,
|
|
},
|
|
},
|
|
async (ctx) => {
|
|
receivedAttempts.push(ctx.attempt);
|
|
// Attempt to escalate retries from 1 to 100
|
|
(
|
|
ctx as unknown as { config: { retries: { limit: number } } }
|
|
).config.retries.limit = 100;
|
|
throw new Error("retry me");
|
|
}
|
|
);
|
|
return result;
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_FAILURE
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// With limit: 1, the engine should execute attempt 1 + 1 retry = 2 attempts total.
|
|
// If the mutation leaked, this would keep going far beyond 2.
|
|
expect(receivedAttempts).toEqual([1, 2]);
|
|
});
|
|
});
|
|
|
|
// ── Helpers for stream tests ────────────────────────────────────────────────
|
|
|
|
function encodeUtf8(str: string): Uint8Array {
|
|
return new TextEncoder().encode(str);
|
|
}
|
|
|
|
function decodeUtf8(bytes: Uint8Array): string {
|
|
return new TextDecoder().decode(bytes);
|
|
}
|
|
|
|
async function readStreamBytes(
|
|
stream: ReadableStream<Uint8Array>
|
|
): Promise<Uint8Array> {
|
|
const chunks: Uint8Array[] = [];
|
|
const reader = stream.getReader();
|
|
while (true) {
|
|
const { done, value } = await reader.read();
|
|
if (done) {
|
|
break;
|
|
}
|
|
chunks.push(value);
|
|
}
|
|
let totalLength = 0;
|
|
for (const chunk of chunks) {
|
|
totalLength += chunk.byteLength;
|
|
}
|
|
const result = new Uint8Array(totalLength);
|
|
let offset = 0;
|
|
for (const chunk of chunks) {
|
|
result.set(chunk, offset);
|
|
offset += chunk.byteLength;
|
|
}
|
|
return result;
|
|
}
|
|
|
|
function countStreamOutputChunks(
|
|
state: DurableObjectState,
|
|
cacheKey: string
|
|
): number {
|
|
const row = state.storage.sql
|
|
.exec<{ cnt: number }>(
|
|
`SELECT COUNT(*) AS cnt FROM streaming_step_chunks WHERE cache_key = ?`,
|
|
cacheKey
|
|
)
|
|
.one();
|
|
return row?.cnt ?? 0;
|
|
}
|
|
|
|
describe("Context - ReadableStream step outputs", () => {
|
|
it("should persist a readable stream output and replay from cache", async ({
|
|
expect,
|
|
}) => {
|
|
const payload = "hello from a readable stream ".repeat(500); // ~14KB
|
|
const payloadBytes = encodeUtf8(payload);
|
|
let callCount = 0;
|
|
|
|
const engineStub = await runWorkflow(
|
|
"STREAM-BASIC",
|
|
async (_event, step) => {
|
|
const stream = await step.do("stream step", async () => {
|
|
callCount++;
|
|
return new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
// Enqueue in two chunks
|
|
const mid = Math.floor(payloadBytes.length / 2);
|
|
controller.enqueue(payloadBytes.slice(0, mid));
|
|
controller.enqueue(payloadBytes.slice(mid));
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
|
|
// The result should be a ReadableStream we can read
|
|
const bytes = await readStreamBytes(
|
|
stream as ReadableStream<Uint8Array>
|
|
);
|
|
return decodeUtf8(bytes);
|
|
}
|
|
);
|
|
|
|
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 successLog = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
expect(successLog?.metadata.result).toBe(payload);
|
|
|
|
// The closure should have been called exactly once (cache hit on replay)
|
|
expect(callCount).toBe(1);
|
|
|
|
// Verify the stream metadata is stored and marked complete
|
|
const hash = await computeHash("stream step");
|
|
const cacheKey = `${hash}-1`;
|
|
const metaKey = getStreamOutputMetaKey(cacheKey);
|
|
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
const meta = (await state.storage.get(metaKey)) as StreamOutputMeta;
|
|
expect(meta).toBeDefined();
|
|
expect(meta.state).toBe(StreamOutputState.Complete);
|
|
expect(meta.totalBytes).toBe(payloadBytes.byteLength);
|
|
expect(meta.chunkCount).toBeGreaterThanOrEqual(1);
|
|
});
|
|
});
|
|
|
|
it("should persist an empty readable stream", async ({ expect }) => {
|
|
const engineStub = await runWorkflow(
|
|
"STREAM-EMPTY",
|
|
async (_event, step) => {
|
|
const stream = await step.do("empty stream step", async () => {
|
|
return new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
|
|
// Should be a readable stream that yields no data
|
|
const bytes = await readStreamBytes(
|
|
stream as ReadableStream<Uint8Array>
|
|
);
|
|
return bytes.byteLength;
|
|
}
|
|
);
|
|
|
|
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 successLog = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
expect(successLog?.metadata.result).toBe(0);
|
|
|
|
// Verify stream metadata
|
|
const hash = await computeHash("empty stream step");
|
|
const cacheKey = `${hash}-1`;
|
|
const metaKey = getStreamOutputMetaKey(cacheKey);
|
|
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
const meta = (await state.storage.get(metaKey)) as StreamOutputMeta;
|
|
expect(meta).toBeDefined();
|
|
expect(meta.state).toBe(StreamOutputState.Complete);
|
|
expect(meta.totalBytes).toBe(0);
|
|
expect(meta.chunkCount).toBe(0);
|
|
});
|
|
});
|
|
|
|
// ── Tests that exercise stream error paths directly inside the DO ────
|
|
//
|
|
// The tests below call writeStreamOutput() inside runInDurableObject()
|
|
// rather than running a full workflow through step.do().
|
|
//
|
|
// Why: In the test harness the user callback runs in a separate
|
|
// WorkerEntrypoint (TestWorkflow) and communicates with the Engine
|
|
// Durable Object over RPC. Returning a problematic ReadableStream from
|
|
// that callback fails at the *RPC transfer* layer with native workerd
|
|
// errors that are different from the custom errors the engine raises:
|
|
//
|
|
// • Locked stream → TypeError: "The ReadableStream has been locked
|
|
// to a reader."
|
|
// • Non-byte chunks → TypeError: "This ReadableStream did not return
|
|
// bytes."
|
|
// • Very large chunks → Error: "Network connection lost." (the RPC
|
|
// pipe disconnects under load)
|
|
//
|
|
// In production the callback executes in the *same* isolate as the
|
|
// engine, so the stream reaches writeStreamOutput() directly and the
|
|
// engine's own validation (iterateStreamChunks, normalizeChunkToUint8Array,
|
|
// chunk-size check) surfaces the correct custom error types.
|
|
//
|
|
// By calling writeStreamOutput() inside the DO we replicate the
|
|
// production code-path and avoid the RPC transfer artefacts.
|
|
|
|
it("should surface a locked readable stream as a fatal error", async ({
|
|
expect,
|
|
}) => {
|
|
const engineId = env.ENGINE.idFromName("STREAM-LOCKED");
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
// Create the streaming table (normally done during engine init)
|
|
// eslint-disable-next-line workers-sdk/no-unsafe-command-execution -- DO SQL exec, not child_process
|
|
state.storage.sql.exec(`
|
|
CREATE TABLE IF NOT EXISTS ${STREAMING_STEP_CHUNKS_TABLE} (
|
|
cache_key TEXT NOT NULL,
|
|
attempt INTEGER NOT NULL,
|
|
chunk_index INTEGER NOT NULL,
|
|
chunk BLOB NOT NULL,
|
|
PRIMARY KEY (cache_key, attempt, chunk_index)
|
|
) WITHOUT ROWID
|
|
`);
|
|
|
|
const stream = new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(encodeUtf8("data"));
|
|
controller.close();
|
|
},
|
|
});
|
|
// Lock the stream by acquiring a reader
|
|
stream.getReader();
|
|
|
|
try {
|
|
await writeStreamOutput({
|
|
storage: state.storage,
|
|
cacheKey: "locked-stream-test",
|
|
attempt: 1,
|
|
stream,
|
|
});
|
|
expect.unreachable("writeStreamOutput should have thrown");
|
|
} catch (e) {
|
|
expect(e).toBeInstanceOf(InvalidStepReadableStreamError);
|
|
}
|
|
});
|
|
});
|
|
|
|
it("should surface an unsupported chunk type as a fatal error", async ({
|
|
expect,
|
|
}) => {
|
|
const engineId = env.ENGINE.idFromName("STREAM-UNSUPPORTED-CHUNK");
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
// eslint-disable-next-line workers-sdk/no-unsafe-command-execution -- DO SQL exec, not child_process
|
|
state.storage.sql.exec(`
|
|
CREATE TABLE IF NOT EXISTS ${STREAMING_STEP_CHUNKS_TABLE} (
|
|
cache_key TEXT NOT NULL,
|
|
attempt INTEGER NOT NULL,
|
|
chunk_index INTEGER NOT NULL,
|
|
chunk BLOB NOT NULL,
|
|
PRIMARY KEY (cache_key, attempt, chunk_index)
|
|
) WITHOUT ROWID
|
|
`);
|
|
|
|
const stream = new ReadableStream({
|
|
start(controller) {
|
|
// Enqueue a string -- not supported (only ArrayBuffer / TypedArray)
|
|
controller.enqueue("this is a string, not bytes");
|
|
controller.close();
|
|
},
|
|
});
|
|
|
|
try {
|
|
await writeStreamOutput({
|
|
storage: state.storage,
|
|
cacheKey: "unsupported-chunk-test",
|
|
attempt: 1,
|
|
stream,
|
|
});
|
|
expect.unreachable("writeStreamOutput should have thrown");
|
|
} catch (e) {
|
|
expect(e).toBeInstanceOf(UnsupportedStreamChunkError);
|
|
}
|
|
});
|
|
});
|
|
|
|
it("should surface an oversized stream chunk as a fatal error", async ({
|
|
expect,
|
|
}) => {
|
|
const engineId = env.ENGINE.idFromName("STREAM-OVERSIZED-CHUNK");
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
// eslint-disable-next-line workers-sdk/no-unsafe-command-execution -- DO SQL exec, not child_process
|
|
state.storage.sql.exec(`
|
|
CREATE TABLE IF NOT EXISTS ${STREAMING_STEP_CHUNKS_TABLE} (
|
|
cache_key TEXT NOT NULL,
|
|
attempt INTEGER NOT NULL,
|
|
chunk_index INTEGER NOT NULL,
|
|
chunk BLOB NOT NULL,
|
|
PRIMARY KEY (cache_key, attempt, chunk_index)
|
|
) WITHOUT ROWID
|
|
`);
|
|
|
|
const stream = new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
// 17 MiB chunk -- exceeds the 16 MiB per-chunk limit
|
|
controller.enqueue(new Uint8Array(17 * 1024 * 1024));
|
|
controller.close();
|
|
},
|
|
});
|
|
|
|
try {
|
|
await writeStreamOutput({
|
|
storage: state.storage,
|
|
cacheKey: "oversized-chunk-test",
|
|
attempt: 1,
|
|
stream,
|
|
});
|
|
expect.unreachable("writeStreamOutput should have thrown");
|
|
} catch (e) {
|
|
expect(e).toBeInstanceOf(OversizedStreamChunkError);
|
|
}
|
|
});
|
|
});
|
|
|
|
it("should clean up streaming_step_chunks on restart", async ({ expect }) => {
|
|
const instanceId = "STREAM-RESTART-CLEANUP";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
|
|
const engineStub = await runWorkflowAndAwait(
|
|
instanceId,
|
|
async (_event, step) => {
|
|
const stream = await step.do("stream before restart", async () => {
|
|
return new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(encodeUtf8("data for restart test"));
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
const bytes = await readStreamBytes(
|
|
stream as ReadableStream<Uint8Array>
|
|
);
|
|
return decodeUtf8(bytes);
|
|
}
|
|
);
|
|
|
|
// Verify chunks exist before restart
|
|
const hash = await computeHash("stream before restart");
|
|
const cacheKey = `${hash}-1`;
|
|
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
const chunkCount = countStreamOutputChunks(state, cacheKey);
|
|
expect(chunkCount).toBeGreaterThanOrEqual(1);
|
|
});
|
|
|
|
// Trigger restart
|
|
try {
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
await engine.changeInstanceStatus("restart");
|
|
});
|
|
} catch (e) {
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
const restartedStub: DurableObjectStub<Engine> = env.ENGINE.get(engineId);
|
|
await runInDurableObject(restartedStub, async (engine, state) => {
|
|
await engine.attemptRestart();
|
|
|
|
// Check immediately after attemptRestart returns, before the
|
|
// fire-and-forget init() call (engine.ts:892) has a chance to
|
|
// re-execute the workflow and recreate the chunks.
|
|
const chunkCount = countStreamOutputChunks(state, cacheKey);
|
|
expect(chunkCount).toBe(0);
|
|
});
|
|
});
|
|
|
|
it("should preserve mock stream chunks across restart", async ({
|
|
expect,
|
|
}) => {
|
|
const mockPayload = "mock stream survives restart";
|
|
const mockPayloadBytes = encodeUtf8(mockPayload);
|
|
|
|
const instanceId = "STREAM-MOCK-SURVIVES-RESTART";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
// Set up a mocked stream step via the modifier
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
const modifier = engine.getInstanceModifier();
|
|
await modifier.mockStepResult(
|
|
{ name: "mocked restart stream" },
|
|
new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(mockPayloadBytes);
|
|
controller.close();
|
|
},
|
|
})
|
|
);
|
|
});
|
|
|
|
const hash = await computeHash("mocked restart stream");
|
|
const baseCacheKey = `${hash}-1`;
|
|
|
|
// Run the workflow once so the engine is initialised and can be restarted
|
|
const stub = await runWorkflow(instanceId, async (_event, step) => {
|
|
const stream = await step.do("mocked restart stream", async () => {
|
|
return new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(encodeUtf8("WRONG - real step ran"));
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
const bytes = await readStreamBytes(stream as ReadableStream<Uint8Array>);
|
|
return decodeUtf8(bytes);
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await stub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
// Trigger restart
|
|
try {
|
|
await runInDurableObject(stub, async (engine) => {
|
|
await engine.changeInstanceStatus("restart");
|
|
});
|
|
} catch (e) {
|
|
if (!isAbortError(e)) {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
// After restart, mock stream chunks (attempt=0) must still exist
|
|
const restartedStub: DurableObjectStub<Engine> = env.ENGINE.get(engineId);
|
|
await runInDurableObject(restartedStub, async (engine, state) => {
|
|
await engine.attemptRestart();
|
|
|
|
const chunkCount = countStreamOutputChunks(state, baseCacheKey);
|
|
expect(chunkCount).toBeGreaterThanOrEqual(1);
|
|
});
|
|
|
|
// Run the workflow again — the mock sentinel (KV) and its SQL chunks
|
|
// should both be present, allowing the mocked stream to replay correctly.
|
|
const stub2 = await runWorkflow(instanceId, async (_event, step) => {
|
|
const stream = await step.do("mocked restart stream", async () => {
|
|
return new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(
|
|
encodeUtf8("WRONG - real step ran after restart")
|
|
);
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
const bytes = await readStreamBytes(stream as ReadableStream<Uint8Array>);
|
|
return decodeUtf8(bytes);
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await stub2.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logs = (await stub2.readLogs()) as EngineLogs;
|
|
const successLog = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
expect(successLog?.metadata.result).toBe(mockPayload);
|
|
});
|
|
|
|
it("should normalize TypedArray chunks to Uint8Array on replay", async ({
|
|
expect,
|
|
}) => {
|
|
// Return Int16Array chunks -- they should be stored as raw bytes
|
|
// and replayed as Uint8Array
|
|
const int16Data = new Int16Array([1, 2, 3, 256, -1]);
|
|
const expectedBytes = new Uint8Array(
|
|
int16Data.buffer,
|
|
int16Data.byteOffset,
|
|
int16Data.byteLength
|
|
);
|
|
|
|
const engineStub = await runWorkflow(
|
|
"STREAM-TYPED-ARRAY",
|
|
async (_event, step) => {
|
|
const stream = await step.do("typed array step", async () => {
|
|
return new ReadableStream({
|
|
start(controller) {
|
|
controller.enqueue(int16Data);
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
|
|
const bytes = await readStreamBytes(
|
|
stream as ReadableStream<Uint8Array>
|
|
);
|
|
return Array.from(bytes);
|
|
}
|
|
);
|
|
|
|
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 successLog = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
expect(successLog?.metadata.result).toEqual(Array.from(expectedBytes));
|
|
});
|
|
|
|
it("should return a replay ReadableStream from waitForStepResult", async ({
|
|
expect,
|
|
}) => {
|
|
const payload = "stream content for waitForStepResult";
|
|
const payloadBytes = encodeUtf8(payload);
|
|
|
|
const engineStub = await runWorkflow(
|
|
"STREAM-WAIT-FOR-STEP",
|
|
async (_event, step) => {
|
|
const stream = await step.do("stream step", async () => {
|
|
return new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(payloadBytes);
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
const bytes = await readStreamBytes(
|
|
stream as ReadableStream<Uint8Array>
|
|
);
|
|
return decodeUtf8(bytes);
|
|
}
|
|
);
|
|
|
|
// Use engine's waitForStepResult to get the stream
|
|
const stepResult = await engineStub.waitForStepResult("stream step");
|
|
|
|
// Should be a ReadableStream
|
|
expect(stepResult).toBeInstanceOf(ReadableStream);
|
|
|
|
const replayBytes = await readStreamBytes(
|
|
stepResult as ReadableStream<Uint8Array>
|
|
);
|
|
expect(decodeUtf8(replayBytes)).toBe(payload);
|
|
|
|
// Wait for workflow to finish
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
});
|
|
|
|
it("should mock a step result with a ReadableStream via modifier", async ({
|
|
expect,
|
|
}) => {
|
|
const mockPayload = "mocked stream content from modifier";
|
|
const mockPayloadBytes = encodeUtf8(mockPayload);
|
|
|
|
const instanceId = "STREAM-MOCK-STEP-RESULT";
|
|
const engineId = env.ENGINE.idFromName(instanceId);
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
// Set up the mock stream result via the modifier before running the workflow
|
|
await runInDurableObject(engineStub, async (engine) => {
|
|
const modifier = engine.getInstanceModifier();
|
|
await modifier.mockStepResult(
|
|
{ name: "mocked stream step" },
|
|
new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(mockPayloadBytes);
|
|
controller.close();
|
|
},
|
|
})
|
|
);
|
|
});
|
|
|
|
const hash = await computeHash("mocked stream step");
|
|
const baseCacheKey = `${hash}-1`;
|
|
const valueKey = `${baseCacheKey}-value`;
|
|
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
// normal non-mocked key should not exist
|
|
expect(
|
|
await state.storage.get(getStreamOutputMetaKey(baseCacheKey))
|
|
).toBeUndefined();
|
|
|
|
const replaceResult = await state.storage.get<{
|
|
__mockStreamOutput: true;
|
|
cacheKey: string;
|
|
meta: StreamOutputMeta;
|
|
}>(`${MODIFIER_KEYS.REPLACE_RESULT}${valueKey}`);
|
|
|
|
expect(replaceResult?.__mockStreamOutput).toBe(true);
|
|
expect(replaceResult?.cacheKey).toBe(baseCacheKey);
|
|
expect(replaceResult?.meta.state).toBe(StreamOutputState.Complete);
|
|
expect(replaceResult?.meta.attempt).toBe(0);
|
|
expect(
|
|
countStreamOutputChunks(state, baseCacheKey)
|
|
).toBeGreaterThanOrEqual(1);
|
|
});
|
|
|
|
// Run a workflow that uses the mocked step
|
|
const stub = await runWorkflow(instanceId, async (_event, step) => {
|
|
const stream = await step.do("mocked stream step", async () => {
|
|
// This should NOT be called -- the mock replaces it
|
|
return new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(encodeUtf8("WRONG - real step ran"));
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
const bytes = await readStreamBytes(stream as ReadableStream<Uint8Array>);
|
|
return decodeUtf8(bytes);
|
|
});
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await stub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 5000 }
|
|
);
|
|
|
|
const logs = (await stub.readLogs()) as EngineLogs;
|
|
const successLog = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
// The workflow should have received the mocked stream content
|
|
expect(successLog?.metadata.result).toBe(mockPayload);
|
|
});
|
|
|
|
it("should resolve stream output to a text preview in readLogs", async ({
|
|
expect,
|
|
}) => {
|
|
const payload = "hello from stream preview test";
|
|
const payloadBytes = encodeUtf8(payload);
|
|
|
|
const engineStub = await runWorkflow(
|
|
"STREAM-PREVIEW-TEXT",
|
|
async (_event, step) => {
|
|
const stream = await step.do("preview step", async () => {
|
|
return new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(payloadBytes);
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
const bytes = await readStreamBytes(
|
|
stream as ReadableStream<Uint8Array>
|
|
);
|
|
return decodeUtf8(bytes);
|
|
}
|
|
);
|
|
|
|
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 stepLog = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.STEP_SUCCESS
|
|
);
|
|
// readLogs() should resolve the stream output metadata to the preview text
|
|
expect(stepLog?.metadata.result).toBe(payload);
|
|
});
|
|
|
|
it("should truncate a long stream preview in readLogs", async ({
|
|
expect,
|
|
}) => {
|
|
// Generate a payload longer than the 1024-char preview limit
|
|
const payload = "A".repeat(2048);
|
|
const payloadBytes = encodeUtf8(payload);
|
|
|
|
const engineStub = await runWorkflow(
|
|
"STREAM-PREVIEW-TRUNCATED",
|
|
async (_event, step) => {
|
|
const stream = await step.do("long preview step", async () => {
|
|
return new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(payloadBytes);
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
const bytes = await readStreamBytes(
|
|
stream as ReadableStream<Uint8Array>
|
|
);
|
|
return decodeUtf8(bytes);
|
|
}
|
|
);
|
|
|
|
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 stepLog = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.STEP_SUCCESS
|
|
);
|
|
// readLogs() should truncate the preview to 1024 chars
|
|
expect(stepLog?.metadata.result).toBe(
|
|
"A".repeat(1024) + "[truncated output]"
|
|
);
|
|
});
|
|
|
|
it("should resolve non-UTF-8 stream output to a binary summary in readLogs", async ({
|
|
expect,
|
|
}) => {
|
|
// Write raw bytes that are not valid UTF-8
|
|
const invalidUtf8 = new Uint8Array([0xff, 0xfe, 0x80, 0x81]);
|
|
|
|
const engineStub = await runWorkflow(
|
|
"STREAM-PREVIEW-BINARY",
|
|
async (_event, step) => {
|
|
const stream = await step.do("binary step", async () => {
|
|
return new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(invalidUtf8);
|
|
controller.close();
|
|
},
|
|
});
|
|
});
|
|
const bytes = await readStreamBytes(
|
|
stream as ReadableStream<Uint8Array>
|
|
);
|
|
return bytes.byteLength;
|
|
}
|
|
);
|
|
|
|
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 stepLog = logs.logs.find(
|
|
(val) => val.event === InstanceEvent.STEP_SUCCESS
|
|
);
|
|
// readLogs() should fall back to a binary size summary
|
|
expect(stepLog?.metadata.result).toBe(
|
|
`[ReadableStream (binary): ${invalidUtf8.byteLength} bytes]`
|
|
);
|
|
});
|
|
|
|
it("should time out during stream write and roll back", async ({
|
|
expect,
|
|
}) => {
|
|
// In the full workflow path a stream-write timeout triggers retries
|
|
// (default limit: 5, exponential backoff from 1 s), so the test would
|
|
// exceed its timeout before WORKFLOW_FAILURE is ever logged. Testing
|
|
// writeStreamOutput() directly lets us verify timeout + rollback
|
|
// behaviour without retry overhead.
|
|
|
|
const engineId = env.ENGINE.idFromName("STREAM-TIMEOUT");
|
|
const engineStub = env.ENGINE.get(engineId);
|
|
|
|
const cacheKey = "timeout-stream-test";
|
|
|
|
await runInDurableObject(engineStub, async (_engine, state) => {
|
|
// eslint-disable-next-line workers-sdk/no-unsafe-command-execution -- DO SQL exec, not child_process
|
|
state.storage.sql.exec(`
|
|
CREATE TABLE IF NOT EXISTS ${STREAMING_STEP_CHUNKS_TABLE} (
|
|
cache_key TEXT NOT NULL,
|
|
attempt INTEGER NOT NULL,
|
|
chunk_index INTEGER NOT NULL,
|
|
chunk BLOB NOT NULL,
|
|
PRIMARY KEY (cache_key, attempt, chunk_index)
|
|
) WITHOUT ROWID
|
|
`);
|
|
|
|
const abortController = new AbortController();
|
|
|
|
// Create a timeout promise that rejects after 1 second, mirroring
|
|
// the timeoutPromise() used by Context.do() in production.
|
|
const timeoutTask = new Promise<never>((_, reject) => {
|
|
setTimeout(() => {
|
|
const error = new WorkflowTimeoutError(
|
|
"Execution timed out after 1000ms"
|
|
);
|
|
abortController.abort(error);
|
|
reject(error);
|
|
}, 1000);
|
|
});
|
|
|
|
// Stream that never closes — each pull emits a small chunk then
|
|
// waits longer than the timeout.
|
|
const stream = new ReadableStream<Uint8Array>({
|
|
async pull(controller) {
|
|
controller.enqueue(encodeUtf8("chunk "));
|
|
await new Promise((resolve) => setTimeout(resolve, 2000));
|
|
},
|
|
});
|
|
|
|
try {
|
|
await writeStreamOutput({
|
|
storage: state.storage,
|
|
cacheKey,
|
|
attempt: 1,
|
|
stream,
|
|
signal: abortController.signal,
|
|
timeoutTask,
|
|
});
|
|
expect.unreachable("writeStreamOutput should have thrown");
|
|
} catch (e) {
|
|
expect(e).toBeInstanceOf(WorkflowTimeoutError);
|
|
}
|
|
|
|
// Mimic the rollback that context.ts performs in the outer catch
|
|
// (line ~704) when a stream write times out.
|
|
await rollbackStreamOutput(state.storage, cacheKey, 1);
|
|
|
|
// Verify that all chunks and metadata were cleaned up
|
|
const chunkCount = countStreamOutputChunks(state, cacheKey);
|
|
expect(chunkCount).toBe(0);
|
|
|
|
const metaKey = getStreamOutputMetaKey(cacheKey);
|
|
const meta = await state.storage.get(metaKey);
|
|
expect(meta).toBeUndefined();
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("Context - typed-array step outputs (issue #14101)", () => {
|
|
it("should persist a 200KB Uint8Array step output on a tight backing buffer", async ({
|
|
expect,
|
|
}) => {
|
|
// Baseline: a 200KB view sized exactly to its backing buffer must
|
|
// succeed. This case worked on `main` too — anchors the regression test
|
|
// suite to "small enough to fit, regardless of normalisation".
|
|
const engineStub = await runWorkflow(
|
|
"UINT8-TIGHT-200K",
|
|
async (_event, step) => {
|
|
return await step.do("emit-tight-bytes", async () => {
|
|
return new Uint8Array(200_000);
|
|
});
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 10000 }
|
|
);
|
|
|
|
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 persist a 200KB Uint8Array view sliced from an 800KB backing buffer (regression #14101)", async ({
|
|
expect,
|
|
}) => {
|
|
// The exact reporter's repro: a Uint8Array view sized 200KB but
|
|
// sliced from a much larger backing ArrayBuffer (800KB here). Prior
|
|
// to the fix, workerd's v8::ValueSerializer would serialise the
|
|
// entire 800KB backing buffer along with the view, blowing past the
|
|
// 1MiB SQL blob limit and surfacing a raw `SQLITE_TOOBIG` error.
|
|
const engineStub = await runWorkflow(
|
|
"UINT8-SLICED-200K-OF-800K",
|
|
async (_event, step) => {
|
|
return await step.do("emit-sliced-bytes", async () => {
|
|
const backing = new ArrayBuffer(800_000);
|
|
return new Uint8Array(backing, 0, 200_000);
|
|
});
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 10000 }
|
|
);
|
|
|
|
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 return the original Uint8Array to the next step in the live execution path", async ({
|
|
expect,
|
|
}) => {
|
|
// The persisted form preserves the view type (`Uint8Array` stays
|
|
// `Uint8Array`) so the live execution path and cached replays observe
|
|
// the same shape — downstream user code that branches on
|
|
// `value instanceof Uint8Array` works either way.
|
|
let observedInNext: unknown = "not-seen";
|
|
const engineStub = await runWorkflow(
|
|
"UINT8-ROUND-TRIP",
|
|
async (_event, step) => {
|
|
const bytes = await step.do("emit-bytes", async () => {
|
|
return new Uint8Array([1, 2, 3, 4, 5]);
|
|
});
|
|
await step.do("read-bytes", async () => {
|
|
observedInNext = bytes;
|
|
return "ok";
|
|
});
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 10000 }
|
|
);
|
|
|
|
expect(observedInNext).toBeInstanceOf(Uint8Array);
|
|
expect(Array.from(observedInNext as Uint8Array)).toEqual([1, 2, 3, 4, 5]);
|
|
});
|
|
|
|
it("should preserve typed-array view types other than Uint8Array on the live execution path", async ({
|
|
expect,
|
|
}) => {
|
|
// Persisting must preserve the constructor type (`Int16Array` stays
|
|
// `Int16Array`, etc.) so downstream `instanceof` checks behave the
|
|
// same whether the step ran live or via cached replay.
|
|
let observedInNext: unknown = "not-seen";
|
|
const engineStub = await runWorkflow(
|
|
"INT16-ROUND-TRIP",
|
|
async (_event, step) => {
|
|
const bytes = await step.do("emit-int16", async () => {
|
|
return new Int16Array([1, 2, 3, 4, 5]);
|
|
});
|
|
await step.do("read-int16", async () => {
|
|
observedInNext = bytes;
|
|
return "ok";
|
|
});
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 10000 }
|
|
);
|
|
|
|
expect(observedInNext).toBeInstanceOf(Int16Array);
|
|
expect(Array.from(observedInNext as Int16Array)).toEqual([1, 2, 3, 4, 5]);
|
|
});
|
|
|
|
it("should compact typed-array views nested inside an object (deep-walk regression)", async ({
|
|
expect,
|
|
}) => {
|
|
// A view sliced from a much larger backing buffer would bloat past
|
|
// SQLITE_TOOBIG if nested-level normalisation were missing. Recurses
|
|
// into plain objects.
|
|
const engineStub = await runWorkflow(
|
|
"NESTED-UINT8-SLICED",
|
|
async (_event, step) => {
|
|
return await step.do("emit-nested-sliced-bytes", async () => {
|
|
const backing = new ArrayBuffer(800_000);
|
|
return {
|
|
image: new Uint8Array(backing, 0, 200_000),
|
|
meta: { width: 100, height: 100 },
|
|
};
|
|
});
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 10000 }
|
|
);
|
|
|
|
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 compact typed-array views nested deep inside arrays of objects", async ({
|
|
expect,
|
|
}) => {
|
|
// Two levels of nesting (array of object of view). Confirms the
|
|
// walker descends through both arrays and plain objects.
|
|
const engineStub = await runWorkflow(
|
|
"ARRAY-OF-OBJECTS-WITH-SLICED-VIEWS",
|
|
async (_event, step) => {
|
|
return await step.do("emit-array-of-objects", async () => {
|
|
const backing = new ArrayBuffer(800_000);
|
|
return [
|
|
{ data: new Uint8Array(backing, 0, 100_000) },
|
|
{ data: new Uint8Array(backing, 100_000, 100_000) },
|
|
];
|
|
});
|
|
}
|
|
);
|
|
|
|
await vi.waitUntil(
|
|
async () => {
|
|
const logs = (await engineStub.readLogs()) as EngineLogs;
|
|
return logs.logs.some(
|
|
(val) => val.event === InstanceEvent.WORKFLOW_SUCCESS
|
|
);
|
|
},
|
|
{ timeout: 10000 }
|
|
);
|
|
|
|
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);
|
|
});
|
|
});
|