From 1e900c140c0d36c62ba8dbeef3ea246f353fffb5 Mon Sep 17 00:00:00 2001 From: James Watkins-Harvey Date: Fri, 4 Sep 2026 18:42:15 -0400 Subject: [PATCH] Prototype language-initiated workflow cache eviction --- CHANGELOG.md | 5 + packages/core-bridge/sdk-core | 2 +- packages/core-bridge/src/worker.rs | 15 + packages/core-bridge/ts/native.ts | 2 + packages/test/src/mock-native-worker.ts | 9 +- packages/test/src/test-threaded-vm.ts | 158 +++++++++ .../test-worker-lifecycle.cloud-pending.ts | 19 +- packages/test/src/test-worker-lifecycle.ts | 225 ++++++++++++ packages/test/src/test-worker-options.ts | 29 ++ .../src/test-workflow-thread-heap-policy.ts | 17 + packages/test/src/test-workflows.ts | 104 +++++- packages/worker/src/worker-options.ts | 20 ++ packages/worker/src/worker.ts | 129 +++++-- packages/worker/src/workflow/interface.ts | 16 + .../worker/src/workflow/threaded-vm-errors.ts | 26 ++ packages/worker/src/workflow/threaded-vm.ts | 332 +++++++++++++----- .../workflow/workflow-thread-heap-policy.ts | 17 + .../src/workflow/workflow-worker-thread.ts | 77 +++- .../workflow/workflow-worker-thread/input.ts | 17 +- .../workflow/workflow-worker-thread/output.ts | 15 +- 20 files changed, 1102 insertions(+), 132 deletions(-) create mode 100644 packages/test/src/test-threaded-vm.ts create mode 100644 packages/test/src/test-workflow-thread-heap-policy.ts create mode 100644 packages/worker/src/workflow/threaded-vm-errors.ts create mode 100644 packages/worker/src/workflow/workflow-thread-heap-policy.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index f0ec522610..60b7e02d35 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -62,6 +62,9 @@ to docs, or any other relevant information. ### Added +- Workers can bound each Workflow worker thread's heap with `maxWorkflowThreadHeapMiB`. Under heap pressure, the + Worker evicts least-recently-used idle Workflows; if a Workflow thread exits unexpectedly, it is replaced and its + cached Workflows are evicted from Core. - **Experimental**: `@temporalio/openai-agents` can run OpenAI Agents `SandboxAgent`s as Temporal Workflows. SandboxAgent operations are Activities; hosted tool credentials and sandbox environment values that reference allowlisted Worker environment variables are resolved on Worker so their values are not recorded in Workflow history. @@ -117,6 +120,8 @@ to docs, or any other relevant information. ### Fixed +- Workflow activation failures now retain Workflow state until Core eviction, preventing premature + execution-context disposal after converter or codec errors. - Nexus handlers now report uncaught Workflow and standalone Activity already-started errors as non-retryable `INTERNAL` Handler Errors, preventing retries when ID reuse or conflict policies reject duplicate execution IDs. diff --git a/packages/core-bridge/sdk-core b/packages/core-bridge/sdk-core index db635f0f21..5c958888d0 160000 --- a/packages/core-bridge/sdk-core +++ b/packages/core-bridge/sdk-core @@ -1 +1 @@ -Subproject commit db635f0f21b0168a01f993e0f1b6c0f1983ec2bc +Subproject commit 5c958888d017225f4f8108f1b4eeafa519ff2009 diff --git a/packages/core-bridge/src/worker.rs b/packages/core-bridge/src/worker.rs index b7996361bb..b1a0d8b88e 100644 --- a/packages/core-bridge/src/worker.rs +++ b/packages/core-bridge/src/worker.rs @@ -45,6 +45,10 @@ pub fn init(cx: &mut ModuleContext) -> NeonResult<()> { "workerCompleteWorkflowActivation", worker_complete_workflow_activation, )?; + cx.export_function( + "workerRequestWorkflowEviction", + worker_request_workflow_eviction, + )?; cx.export_function("workerPollActivityTask", worker_poll_activity_task)?; cx.export_function("workerCompleteActivityTask", worker_complete_activity_task)?; @@ -197,6 +201,17 @@ pub fn worker_complete_workflow_activation( }) } +/// Ask Core to evict a cached Workflow by run ID. +#[js_function] +pub fn worker_request_workflow_eviction( + worker: OpaqueInboundHandle, + run_id: String, +) -> BridgeResult<()> { + let worker_ref = worker.borrow()?; + worker_ref.core_worker.request_workflow_eviction(&run_id); + Ok(()) +} + /// Initiate a single activity task poll request. /// There should be only one concurrent poll request for this type. #[js_function] diff --git a/packages/core-bridge/ts/native.ts b/packages/core-bridge/ts/native.ts index 9b78d9f9f4..e9c742c8af 100644 --- a/packages/core-bridge/ts/native.ts +++ b/packages/core-bridge/ts/native.ts @@ -203,6 +203,8 @@ export declare function workerPollWorkflowActivation(worker: Worker): Promise; +export declare function workerRequestWorkflowEviction(worker: Worker, runId: string): void; + export declare function workerPollActivityTask(worker: Worker): Promise; export declare function workerCompleteActivityTask(worker: Worker, result: Buffer): Promise; diff --git a/packages/test/src/mock-native-worker.ts b/packages/test/src/mock-native-worker.ts index 9feaa947b2..2896f61a30 100644 --- a/packages/test/src/mock-native-worker.ts +++ b/packages/test/src/mock-native-worker.ts @@ -49,6 +49,7 @@ export class MockNativeWorker implements NativeWorkerLike { reject?: (err: Error) => void; namespace = 'mock'; logger = new DefaultLogger('DEBUG'); + requestedWorkflowEvictions: string[] = []; public static async create(): Promise { return new this(); @@ -96,6 +97,10 @@ export class MockNativeWorker implements NativeWorkerLike { this.workflowCompletionCallback = undefined; } + public requestWorkflowEviction(runId: string): void { + this.requestedWorkflowEvictions.push(runId); + } + public async pollNexusTask(): Promise { // Not implementing this in the mock worker, testing with real worker instead. throw new Error('not implemented'); @@ -197,7 +202,7 @@ export const defaultOptions: WorkerOptions = { taskQueue: 'test', }; -export function isolateFreeWorker(options: WorkerOptions = defaultOptions): Worker { +export function isolateFreeWorker(options: WorkerOptions = defaultOptions, workflowCreator?: WorkflowCreator): Worker { const runtime = Runtime.instance(); const logger = LoggerWithComposedMetadata.compose(runtime.logger, { sdkComponent: SdkComponent.worker, @@ -208,7 +213,7 @@ export function isolateFreeWorker(options: WorkerOptions = defaultOptions): Work taskQueue: options.taskQueue ?? 'default', }); return new Worker( - { + workflowCreator ?? { async createWorkflow() { throw new Error('Not implemented'); }, diff --git a/packages/test/src/test-threaded-vm.ts b/packages/test/src/test-threaded-vm.ts new file mode 100644 index 0000000000..e46466eec7 --- /dev/null +++ b/packages/test/src/test-threaded-vm.ts @@ -0,0 +1,158 @@ +import { EventEmitter } from 'node:events'; +import test from 'ava'; +import { DefaultLogger, Runtime } from '@temporalio/worker'; +import { compileWorkerOptions } from '@temporalio/worker/lib/worker-options'; +import { WorkerThreadClient } from '@temporalio/worker/lib/workflow/threaded-vm'; +import { WorkflowThreadLostError } from '@temporalio/worker/lib/workflow/threaded-vm-errors'; +import type { + WorkflowCreateOptions, + WorkflowCreator, + WorkflowThreadEvictionEvent, +} from '@temporalio/worker/lib/workflow/interface'; +import { defaultOptions, Worker as MockWorker } from './mock-native-worker'; + +class FakeWorkerThread extends EventEmitter { + public terminateCount = 0; + + postMessage(): void { + // Responses are emitted explicitly by each test. + } + + async terminate(): Promise { + this.terminateCount++; + return 1; + } +} + +function workflowOptions(runId: string): WorkflowCreateOptions { + return { info: { runId } } as WorkflowCreateOptions; +} + +test('WorkerThreadClient tracks proactive local evictions', async (t) => { + const workerThread = new FakeWorkerThread(); + const notifications: string[][] = []; + const client = new WorkerThreadClient(workerThread as never, new DefaultLogger('ERROR'), undefined, ({ runIds }) => + notifications.push(runIds) + ); + + const created = client.send({ type: 'create-workflow', options: workflowOptions('run-1') }); + workerThread.emit('message', { requestId: 0n, result: { type: 'ok' } }); + await created; + t.is(client.getActiveWorkflowCount(), 1); + + workerThread.emit('message', { + type: 'workflow-evictions', + runIds: ['run-1'], + usedHeapSize: 800, + heapSizeLimit: 1000, + }); + t.is(client.getActiveWorkflowCount(), 0); + t.deepEqual(notifications, [['run-1']]); +}); + +test('WorkerThreadClient reports owned runs and rejects pending work when its thread exits', async (t) => { + const workerThread = new FakeWorkerThread(); + let lostRunIds: string[] | undefined; + const client = new WorkerThreadClient( + workerThread as never, + new DefaultLogger('ERROR'), + undefined, + undefined, + (_client, runIds) => { + lostRunIds = runIds; + } + ); + + const created = client.send({ type: 'create-workflow', options: workflowOptions('run-1') }); + workerThread.emit('exit', 1); + + await t.throwsAsync(created, { instanceOf: WorkflowThreadLostError }); + t.deepEqual(lostRunIds, ['run-1']); + t.is(client.getActiveWorkflowCount(), 0); +}); + +test('WorkerThreadClient includes a create request that races with thread exit', async (t) => { + const workerThread = new FakeWorkerThread(); + let lostRunIds: string[] | undefined; + const client = new WorkerThreadClient( + workerThread as never, + new DefaultLogger('ERROR'), + undefined, + undefined, + (_client, runIds) => { + lostRunIds = runIds; + } + ); + + workerThread.emit('error', new Error('simulated thread failure')); + await t.throwsAsync(client.send({ type: 'create-workflow', options: workflowOptions('racing-run') }), { + instanceOf: WorkflowThreadLostError, + }); + workerThread.emit('exit', 1); + + t.deepEqual(lostRunIds, ['racing-run']); +}); + +test('WorkerThreadClient replaces a thread when heap-pressure disposal fails', async (t) => { + const workerThread = new FakeWorkerThread(); + let lostRunIds: string[] | undefined; + const lifecycleOrder: string[] = []; + const client = new WorkerThreadClient( + workerThread as never, + new DefaultLogger('ERROR'), + undefined, + undefined, + (_client, runIds) => { + lifecycleOrder.push('eviction-requested'); + lostRunIds = runIds; + } + ); + + let created = client.send({ type: 'create-workflow', options: workflowOptions('run-1') }); + workerThread.emit('message', { requestId: 0n, result: { type: 'ok' } }); + await created; + created = client.send({ type: 'create-workflow', options: workflowOptions('run-2') }); + workerThread.emit('message', { requestId: 1n, result: { type: 'ok' } }); + await created; + + const idle = client.send({ type: 'mark-workflow-idle', runId: 'run-1' }).catch((error) => { + lifecycleOrder.push('activation-rejected'); + throw error; + }); + workerThread.emit('message', { + requestId: 2n, + result: { + type: 'error', + name: 'WorkflowThreadDisposalError', + message: 'Failed to dispose Workflow run-1 under heap pressure', + }, + }); + t.is(workerThread.terminateCount, 1); + + workerThread.emit('exit', 1); + await t.throwsAsync(idle, { instanceOf: WorkflowThreadLostError }); + t.deepEqual(lostRunIds, ['run-1', 'run-2']); + t.deepEqual(lifecycleOrder, ['eviction-requested', 'activation-rejected']); +}); + +test('Worker forwards language-side eviction requests to Core', (t) => { + let evictionHandler: ((event: WorkflowThreadEvictionEvent) => void) | undefined; + const workflowCreator: WorkflowCreator = { + async createWorkflow() { + throw new Error('not implemented'); + }, + async destroy() {}, + setLifecycleHandlers(handler) { + evictionHandler = handler; + }, + }; + const runtime = Runtime.instance(); + const worker = new MockWorker( + workflowCreator, + compileWorkerOptions(defaultOptions, runtime.logger, runtime.metricMeter) + ); + + evictionHandler!({ runIds: ['run-1', 'run-2'], reason: 'heap-pressure' }); + + t.deepEqual(worker.native.requestedWorkflowEvictions, ['run-1', 'run-2']); +}); diff --git a/packages/test/src/test-worker-lifecycle.cloud-pending.ts b/packages/test/src/test-worker-lifecycle.cloud-pending.ts index 42e05ce854..26369908f0 100644 --- a/packages/test/src/test-worker-lifecycle.cloud-pending.ts +++ b/packages/test/src/test-worker-lifecycle.cloud-pending.ts @@ -4,7 +4,7 @@ import { setTimeout } from 'timers/promises'; import test from 'ava'; import { Client } from '@temporalio/client'; import { PromiseCompletionTimeoutError, Runtime } from '@temporalio/worker'; -import { TransportError, UnexpectedError } from '@temporalio/worker/lib/errors'; +import { TransportError } from '@temporalio/worker/lib/errors'; import { isBun, RUN_INTEGRATION_TESTS, Worker } from './helpers'; import { defaultOptions } from './mock-native-worker'; import { fillMemory } from './workflows'; @@ -65,11 +65,11 @@ if (RUN_INTEGRATION_TESTS) { ); }); - (isBun ? test.skip : test.serial)('Threaded VM gracely stops and fails on ERR_WORKER_OUT_OF_MEMORY', async (t) => { + (isBun ? test.skip : test.serial)('Threaded VM replaces a thread after ERR_WORKER_OUT_OF_MEMORY', async (t) => { t.timeout(30_000); const taskQueue = t.title.replace(/ /g, '_'); const client = new Client(); - const worker = await Worker.create({ ...defaultOptions, taskQueue }); + const worker = await Worker.create({ ...defaultOptions, taskQueue, maxWorkflowThreadHeapMiB: 128 }); client.workflow .start(fillMemory, { @@ -82,21 +82,12 @@ if (RUN_INTEGRATION_TESTS) { const workerRun = worker.run(); try { await Promise.race([setTimeout(10_000), workerRun]); + t.is(worker.getState(), 'RUNNING'); + } finally { if (worker.getState() === 'RUNNING') { worker.shutdown(); await workerRun; } - t.log('Non-conclusive result: Worker did not fail as expected'); - t.pass(); - } catch (err) { - t.is((err as Error).name, UnexpectedError.name); - t.is( - (err as Error).message, - 'Workflow Worker Thread exited prematurely: Error [ERR_WORKER_OUT_OF_MEMORY]: ' + - 'Worker terminated due to reaching memory limit: JS heap out of memory' - ); - t.is(worker.getState(), 'FAILED'); - } finally { if (Runtime._instance) await Runtime._instance.shutdown(); } }); diff --git a/packages/test/src/test-worker-lifecycle.ts b/packages/test/src/test-worker-lifecycle.ts index 957e7a6c64..e0ccffcbc8 100644 --- a/packages/test/src/test-worker-lifecycle.ts +++ b/packages/test/src/test-worker-lifecycle.ts @@ -6,8 +6,14 @@ */ import { randomUUID } from 'crypto'; import test from 'ava'; +import Long from 'long'; +import { createPayloadValidationError, defaultPayloadConverter, type PayloadCodec } from '@temporalio/common'; +import { msToTs } from '@temporalio/common/lib/time'; import type { LogEntry, NativeConnection } from '@temporalio/worker'; import { DefaultLogger, MetricsBuffer, Runtime } from '@temporalio/worker'; +import { UnexpectedError } from '@temporalio/worker/lib/errors'; +import { WorkflowLocallyEvictedError } from '@temporalio/worker/lib/workflow/threaded-vm-errors'; +import type { WorkflowThreadEvictionEvent } from '@temporalio/worker/lib/workflow/interface'; import { isolateFreeWorker, Worker as MockWorker } from './mock-native-worker'; test.serial('Worker.create debug log options are JSON serializable with buffered metrics and connection', async (t) => { @@ -65,6 +71,225 @@ test.serial('Mocked run shuts down gracefully', async (t) => { } }); +test('Worker retains a failed Workflow until Core eviction', async (t) => { + const invalidPayload = defaultPayloadConverter.toPayload('invalid-payload'); + const codec: PayloadCodec = { + async encode(payloads) { + if (payloads.some((payload) => defaultPayloadConverter.fromPayload(payload) === 'invalid-payload')) { + throw createPayloadValidationError({ field: 'nexus-input' }); + } + return payloads; + }, + async decode(payloads) { + return payloads; + }, + }; + let disposeCount = 0; + const worker = isolateFreeWorker( + { + taskQueue: t.title.replace(/ /g, '_'), + activities: {}, + dataConverter: { payloadCodecs: [codec] }, + }, + { + async createWorkflow() { + return { + async activate() { + return { + successful: { + commands: [{ scheduleNexusOperation: { seq: 1, input: invalidPayload } }], + }, + }; + }, + async getAndResetSinkCalls() { + return []; + }, + async dispose() { + disposeCount++; + }, + }; + }, + async destroy() {}, + } + ); + const runId = randomUUID(); + const now = msToTs(Date.now()); + const run = worker.run(); + try { + const failed = await worker.native.runWorkflowActivation({ + runId, + timestamp: now, + jobs: [ + { + initializeWorkflow: { + workflowId: 'workflow-id', + workflowType: 'test', + randomnessSeed: Long.ONE, + firstExecutionRunId: runId, + originalExecutionRunId: runId, + attempt: 1, + startTime: now, + workflowTaskTimeout: msToTs('10 seconds'), + }, + }, + ], + }); + + t.is(failed.failed?.failure?.applicationFailureInfo?.type, 'PayloadValidationError'); + t.is(disposeCount, 0); + + const evicted = await worker.native.runWorkflowActivation({ + runId, + jobs: [{ removeFromCache: {} }], + }); + + t.truthy(evicted.successful); + t.is(disposeCount, 1); + } finally { + worker.shutdown(); + await run; + } +}); + +test('Worker fails an outstanding activation after local state loss and waits for Core eviction', async (t) => { + let evictionHandler: ((event: WorkflowThreadEvictionEvent) => void) | undefined; + let disposeCount = 0; + const worker = isolateFreeWorker( + { + taskQueue: t.title.replace(/ /g, '_'), + activities: {}, + }, + { + async createWorkflow() { + return { + async activate() { + throw new WorkflowLocallyEvictedError('Workflow state was discarded under heap pressure'); + }, + async getAndResetSinkCalls() { + return []; + }, + async dispose() { + disposeCount++; + }, + }; + }, + async destroy() {}, + setLifecycleHandlers(handler) { + evictionHandler = handler; + }, + } + ); + const runId = randomUUID(); + const now = msToTs(Date.now()); + const run = worker.run(); + try { + evictionHandler!({ runIds: [runId], reason: 'heap-pressure' }); + const failed = await worker.native.runWorkflowActivation({ + runId, + timestamp: now, + jobs: [ + { + initializeWorkflow: { + workflowId: 'workflow-id', + workflowType: 'test', + randomnessSeed: Long.ONE, + firstExecutionRunId: runId, + originalExecutionRunId: runId, + attempt: 1, + startTime: now, + workflowTaskTimeout: msToTs('10 seconds'), + }, + }, + ], + }); + + t.deepEqual(worker.native.requestedWorkflowEvictions, [runId]); + t.is(failed.failed?.failure?.applicationFailureInfo?.type, 'WorkflowLocallyEvictedError'); + t.is(disposeCount, 0); + + const evicted = await worker.native.runWorkflowActivation({ + runId, + jobs: [{ removeFromCache: {} }], + }); + t.truthy(evicted.successful); + t.is(disposeCount, 1); + } finally { + worker.shutdown(); + await run; + } +}); + +test('Worker closes an evicted Workflow when disposal fails', async (t) => { + const disposeFailure = new Error('dispose failed'); + let disposeCount = 0; + const worker = isolateFreeWorker( + { + taskQueue: t.title.replace(/ /g, '_'), + activities: {}, + }, + { + async createWorkflow() { + return { + async activate() { + return { successful: {} }; + }, + async getAndResetSinkCalls() { + return []; + }, + async dispose() { + disposeCount++; + throw disposeFailure; + }, + }; + }, + async destroy() {}, + } + ); + const runId = randomUUID(); + const now = msToTs(Date.now()); + const run = worker.run(); + try { + const started = await worker.native.runWorkflowActivation({ + runId, + timestamp: now, + jobs: [ + { + initializeWorkflow: { + workflowId: 'workflow-id', + workflowType: 'test', + randomnessSeed: Long.ONE, + firstExecutionRunId: runId, + originalExecutionRunId: runId, + attempt: 1, + startTime: now, + workflowTaskTimeout: msToTs('10 seconds'), + }, + }, + ], + }); + t.truthy(started.successful); + t.is(worker.getStatus().numCachedWorkflows, 1); + + const evicted = await worker.native.runWorkflowActivation({ + runId, + jobs: [{ removeFromCache: {} }], + }); + + t.truthy(evicted.successful); + t.is(disposeCount, 1); + t.is(worker.getStatus().numCachedWorkflows, 0); + + const error = await t.throwsAsync(run); + if (error === undefined) return; + t.assert(error instanceof UnexpectedError); + t.is(error.cause, disposeFailure); + t.is(worker.getState(), 'FAILED'); + } finally { + if (worker.getState() === 'RUNNING') worker.shutdown(); + await run.catch(() => undefined); + } +}); + test.serial('Mocked run shuts down gracefully if interrupted before running', async (t) => { try { const worker = isolateFreeWorker({ diff --git a/packages/test/src/test-worker-options.ts b/packages/test/src/test-worker-options.ts index 8cdee9b91e..ae8dcff319 100644 --- a/packages/test/src/test-worker-options.ts +++ b/packages/test/src/test-worker-options.ts @@ -42,3 +42,32 @@ for (const value of [-1, 1.5, Number.NaN]) { ); }); } + +test('forwards maxWorkflowThreadHeapMiB to the compiled Worker options', (t) => { + const runtime = Runtime.instance(); + const compiled = compileWorkerOptions( + { ...defaultOptions, maxWorkflowThreadHeapMiB: 512 }, + runtime.logger, + runtime.metricMeter + ); + + t.is(compiled.maxWorkflowThreadHeapMiB, 512); +}); + +for (const value of [0, -1, Number.NaN, Number.POSITIVE_INFINITY]) { + test(`rejects invalid Workflow thread heap limit ${value}`, (t) => { + const runtime = Runtime.instance(); + t.throws( + () => + compileWorkerOptions( + { ...defaultOptions, maxWorkflowThreadHeapMiB: value }, + runtime.logger, + runtime.metricMeter + ), + { + instanceOf: TypeError, + message: 'maxWorkflowThreadHeapMiB must be a positive number', + } + ); + }); +} diff --git a/packages/test/src/test-workflow-thread-heap-policy.ts b/packages/test/src/test-workflow-thread-heap-policy.ts new file mode 100644 index 0000000000..572cc1502c --- /dev/null +++ b/packages/test/src/test-workflow-thread-heap-policy.ts @@ -0,0 +1,17 @@ +import test from 'ava'; +import { getWorkflowHeapEvictionBatchSize } from '@temporalio/worker/lib/workflow/workflow-thread-heap-policy'; + +test('does not evict below the high watermark', (t) => { + t.is(getWorkflowHeapEvictionBatchSize(799, 1000, 100), 0); +}); + +test('evicts a bounded batch toward the low watermark', (t) => { + t.is(getWorkflowHeapEvictionBatchSize(800, 1000, 100), 13); + t.is(getWorkflowHeapEvictionBatchSize(900, 1000, 100), 23); +}); + +test('evicts at least one idle Workflow and never more than are idle', (t) => { + t.is(getWorkflowHeapEvictionBatchSize(800, 1000, 1), 1); + t.is(getWorkflowHeapEvictionBatchSize(2000, 1000, 10), 7); + t.is(getWorkflowHeapEvictionBatchSize(2000, 1000, 0), 0); +}); diff --git a/packages/test/src/test-workflows.ts b/packages/test/src/test-workflows.ts index eb27cda8ce..e837acec14 100644 --- a/packages/test/src/test-workflows.ts +++ b/packages/test/src/test-workflows.ts @@ -1,5 +1,6 @@ import path from 'node:path'; import type vm from 'node:vm'; +import { Worker as NodeWorker } from 'node:worker_threads'; import type { ExecutionContext, TestFn } from 'ava'; import anyTest from 'ava'; import dedent from 'dedent'; @@ -19,12 +20,17 @@ import { sleep as workflowSleep, type WorkflowInfo } from '@temporalio/workflow' import { DefaultLogger, LogTimestamp, type LogEntry } from '@temporalio/worker'; import { WorkflowCodeBundler } from '@temporalio/worker/lib/workflow/bundler'; import { invokePatchActivationCallback } from '@temporalio/worker/lib/workflow/patch-activation-callback'; -import { ThreadedVMWorkflowCreator } from '@temporalio/worker/lib/workflow/threaded-vm'; +import { + ThreadedVMWorkflowCreator, + VMWorkflowThreadProxy, + WorkerThreadClient, +} from '@temporalio/worker/lib/workflow/threaded-vm'; +import { WorkflowLocallyEvictedError } from '@temporalio/worker/lib/workflow/threaded-vm-errors'; import type { WorkflowBundleWithSourceMapAndFilename } from '@temporalio/worker/lib/workflow/workflow-worker-thread/input'; import type { PatchActivationCallback, PatchActivationInput } from '@temporalio/worker'; import type { VMWorkflow } from '@temporalio/worker/lib/workflow/vm'; import { VMWorkflowCreator } from '@temporalio/worker/lib/workflow/vm'; -import type { WorkflowCreator } from '@temporalio/worker/lib/workflow/interface'; +import type { WorkflowCreator, WorkflowThreadEvictionEvent } from '@temporalio/worker/lib/workflow/interface'; import type { SdkFlag } from '@temporalio/workflow/lib/flags'; import { SdkFlags } from '@temporalio/workflow/lib/flags'; import { createUnsafeRandomSource } from '@temporalio/workflow/lib/random-helpers'; @@ -559,6 +565,100 @@ test('successString', async (t) => { compareCompletion(t, req, makeSuccess([makeCompleteWorkflowExecution(defaultPayloadConverter.toPayload('success'))])); }); +test('thread heap pressure locally evicts an idle Workflow successString', async (t) => { + const runId = `${t.context.runId}-threaded`; + const evictedRunIds: string[] = []; + const client = new WorkerThreadClient( + new NodeWorker(require.resolve('@temporalio/worker/lib/workflow/workflow-worker-thread')), + new DefaultLogger('ERROR'), + undefined, + ({ runIds }) => evictedRunIds.push(...runIds) + ); + await client.send({ + type: 'init', + workflowBundle: { + ...t.context.workflowBundle, + filename: `${t.context.workflowBundle.filename}-${runId}`, + }, + isolateExecutionTimeoutMs: 400, + reuseV8Context: REUSE_V8_CONTEXT, + registeredActivityNames: new Set(), + hasPatchActivationCallback: false, + // Force the soft-limit path without constraining the actual Worker Thread heap. + heapSizeLimitBytes: 1, + }); + const workflow = await VMWorkflowThreadProxy.create(client, { + info: makeWorkflowInfo('successString', runId), + randomnessSeed: Long.fromInt(1337).toBytes(), + now: Date.now(), + showStackTraceSources: true, + }); + + try { + await workflow.activate(makeStartWorkflow('successString')); + await workflow.getAndResetSinkCalls(); + await workflow.activationCompletionAccepted(); + + t.deepEqual(evictedRunIds, [runId]); + await t.throwsAsync(workflow.activate({ runId, jobs: [] }), { instanceOf: WorkflowLocallyEvictedError }); + } finally { + await workflow.dispose(); + await client.destroy(); + } +}); + +test('thread replacement accepts new Workflows after an unexpected exit successString', async (t) => { + const firstRunId = `${t.context.runId}-first`; + const secondRunId = `${t.context.runId}-second`; + const creator = await ThreadedVMWorkflowCreator.create({ + workflowBundle: { + ...t.context.workflowBundle, + filename: `${t.context.workflowBundle.filename}-${firstRunId}`, + }, + threadPoolSize: 1, + isolateExecutionTimeoutMs: 400, + reuseV8Context: REUSE_V8_CONTEXT, + registeredActivityNames: new Set(), + logger: new DefaultLogger('ERROR'), + }); + let resolveExit!: (event: WorkflowThreadEvictionEvent) => void; + const exitEvent = new Promise((resolve) => { + resolveExit = resolve; + }); + creator.setLifecycleHandlers( + (event) => resolveExit(event), + (error) => t.fail(`Thread replacement failed: ${error}`) + ); + const firstWorkflow = await createWorkflow('successString', firstRunId, Date.now(), creator); + + try { + const [client] = ( + creator as unknown as { + workerThreadClients: Array<{ workerThread: NodeWorker }>; + } + ).workerThreadClients; + await client!.workerThread.terminate(); + t.deepEqual(await exitEvent, { runIds: [firstRunId], reason: 'thread-exit' }); + + const secondWorkflow = await createWorkflow('successString', secondRunId, Date.now(), creator); + try { + const completion = await secondWorkflow.activate(makeStartWorkflow('successString')); + t.deepEqual( + coresdk.workflow_completion.WorkflowActivationCompletion.create(completion).toJSON(), + coresdk.workflow_completion.WorkflowActivationCompletion.create({ + ...makeSuccess([makeCompleteWorkflowExecution(defaultPayloadConverter.toPayload('success'))]), + runId: secondRunId, + }).toJSON() + ); + } finally { + await secondWorkflow.dispose(); + } + } finally { + await firstWorkflow.dispose(); + await creator.destroy(); + } +}); + test('continueAsNewSuggested', async (t) => { const { workflowType } = t.context; const activation = makeStartWorkflow(workflowType); diff --git a/packages/worker/src/worker-options.ts b/packages/worker/src/worker-options.ts index 9e56b66a12..770beb7250 100644 --- a/packages/worker/src/worker-options.ts +++ b/packages/worker/src/worker-options.ts @@ -436,6 +436,20 @@ export interface WorkerOptions { */ maxCachedWorkflows?: number; + /** + * Maximum old-generation heap size, in MiB, for each Workflow worker thread. + * + * The Workflow cache monitors heap usage inside each thread and proactively evicts least-recently-used idle + * Workflows as the thread approaches this limit. The same value is also installed as the thread's V8 hard limit, + * allowing the Worker to replace the thread if proactive eviction cannot prevent an out-of-memory termination. + * + * This is a per-thread limit. {@link maxCachedWorkflows} remains an independent, Worker-wide count limit. + * Ignored when {@link debugMode} is enabled. + * + * @default The V8 default heap limit. + */ + maxWorkflowThreadHeapMiB?: number; + /** * Controls the number of threads to be created for executing Workflow Tasks. * @@ -1110,6 +1124,12 @@ export function compileWorkerOptions( } const opts = addDefaultWorkerOptions(rawOpts, logger, metricMeter); + if ( + opts.maxWorkflowThreadHeapMiB !== undefined && + (!Number.isFinite(opts.maxWorkflowThreadHeapMiB) || opts.maxWorkflowThreadHeapMiB <= 0) + ) { + throw new TypeError('maxWorkflowThreadHeapMiB must be a positive number'); + } if ( opts.maxEagerActivityReservationsPerWorkflowTask !== undefined && (!Number.isSafeInteger(opts.maxEagerActivityReservationsPerWorkflowTask) || diff --git a/packages/worker/src/worker.ts b/packages/worker/src/worker.ts index 84bdf76765..3b0bea31cd 100644 --- a/packages/worker/src/worker.ts +++ b/packages/worker/src/worker.ts @@ -101,6 +101,7 @@ import { isBunPre1_4 } from './workflow/bun'; import type { Workflow, WorkflowCreator } from './workflow/interface'; import { ReusableVMWorkflowCreator } from './workflow/reusable-vm'; import { ThreadedVMWorkflowCreator } from './workflow/threaded-vm'; +import { WorkflowLocallyEvictedError, WorkflowThreadLostError } from './workflow/threaded-vm-errors'; import { VMWorkflowCreator } from './workflow/vm'; import { invokePatchActivationCallback } from './workflow/patch-activation-callback'; import type { WorkflowBundleWithSourceMapAndFilename } from './workflow/workflow-worker-thread/input'; @@ -162,6 +163,7 @@ export interface NativeWorkerLike { pollActivityTask: OmitFirstParam; pollNexusTask: OmitFirstParam; completeWorkflowActivation: OmitFirstParam; + requestWorkflowEviction: OmitFirstParam; completeActivityTask: OmitFirstParam; completeNexusTask: OmitFirstParam; recordActivityHeartbeat: OmitFirstParam; @@ -189,6 +191,8 @@ interface WorkflowWithLogAttributes { info: WorkflowInfo; } +class WorkflowDisposeError extends UnexpectedError {} + function addBuildIdIfMissing(options: CompiledWorkerOptions, bundleCode?: string): CompiledWorkerOptionsWithBuildId { const bid = options.buildId; if (bid != null) { @@ -205,6 +209,7 @@ export class NativeWorker implements NativeWorkerLike { public readonly pollActivityTask: OmitFirstParam; public readonly pollNexusTask: OmitFirstParam; public readonly completeWorkflowActivation: OmitFirstParam; + public readonly requestWorkflowEviction: OmitFirstParam; public readonly completeActivityTask: OmitFirstParam; public readonly completeNexusTask: OmitFirstParam; public readonly recordActivityHeartbeat: OmitFirstParam; @@ -239,6 +244,7 @@ export class NativeWorker implements NativeWorkerLike { this.pollActivityTask = native.workerPollActivityTask.bind(undefined, nativeWorker); this.pollNexusTask = native.workerPollNexusTask.bind(undefined, nativeWorker); this.completeWorkflowActivation = native.workerCompleteWorkflowActivation.bind(undefined, nativeWorker); + this.requestWorkflowEviction = native.workerRequestWorkflowEviction.bind(undefined, nativeWorker); this.completeActivityTask = native.workerCompleteActivityTask.bind(undefined, nativeWorker); this.completeNexusTask = native.workerCompleteNexusTask.bind(undefined, nativeWorker); this.recordActivityHeartbeat = native.workerRecordActivityHeartbeat.bind(undefined, nativeWorker); @@ -621,6 +627,7 @@ export class Worker { registeredActivityNames, logger, patchActivationCallback: compiledOptions.patchActivationCallback, + maxWorkflowThreadHeapMiB: compiledOptions.maxWorkflowThreadHeapMiB, }); } } @@ -850,7 +857,26 @@ export class Worker { protected readonly plugins: WorkerPlugin[], protected _connection?: NativeConnection, protected readonly isReplayWorker: boolean = false - ) {} + ) { + if (workflowCreator?.setLifecycleHandlers !== undefined) { + workflowCreator.setLifecycleHandlers( + ({ runIds, reason, usedHeapSize, heapSizeLimit }) => { + this.logger.info('Workflow Worker Thread evicted cached Workflows', { + reason, + evictedWorkflowCount: runIds.length, + usedHeapSize, + heapSizeLimit, + }); + try { + for (const runId of runIds) this.nativeWorker.requestWorkflowEviction(runId); + } catch (error) { + this.unexpectedErrorSubject.error(error); + } + }, + (error) => this.unexpectedErrorSubject.error(error) + ); + } + } /** * An Observable which emits each time the number of in flight activations changes @@ -1473,7 +1499,7 @@ export class Worker { */ protected handleWorkflowActivations( activations$: CloseableGroupedObservable - ): Observable { + ): Observable { const syntheticEvictionActivations$ = this.workflowPollerStateSubject.pipe( // Core has indicated that it will not return any more poll results; evict all cached WFs. filter((state) => state !== 'POLLING'), @@ -1492,17 +1518,45 @@ export class Worker { tap(() => { this.numInFlightActivationsSubject.next(this.numInFlightActivationsSubject.value + 1); }), - mergeMapWithState(this.handleActivation.bind(this), undefined), - tap(({ close }) => { - this.numInFlightActivationsSubject.next(this.numInFlightActivationsSubject.value - 1); + mergeMapWithState< + WorkflowWithLogAttributes | undefined, + { activation: coresdk.workflow_activation.WorkflowActivation; synthetic: boolean }, + { completion?: Uint8Array; close: boolean; fatalError?: Error } + >(async (state, input) => { + const result = await this.handleActivation(state, input); + const { completion, close } = result.output; if (close) { activations$.close(); this.numCachedWorkflowsSubject.next(this.numCachedWorkflowsSubject.value - 1); } + if (completion !== undefined) { + try { + await this.nativeWorker.completeWorkflowActivation(Buffer.from(completion, completion.byteOffset)); + } catch (error) { + this.logger.error('Core reported failure in completeWorkflowActivation(). Initiating Worker shutdown.', { + error, + }); + this.unexpectedErrorSubject.error(error); + return result; + } + if (!close) { + try { + await result.state?.workflow.activationCompletionAccepted?.(); + } catch (error) { + if (!(error instanceof WorkflowThreadLostError || error instanceof WorkflowLocallyEvictedError)) { + this.unexpectedErrorSubject.error(error); + } + } + } + } + return result; + }, undefined), + tap(({ fatalError }) => { + this.numInFlightActivationsSubject.next(this.numInFlightActivationsSubject.value - 1); + if (fatalError !== undefined) this.unexpectedErrorSubject.error(fatalError); }), takeWhile(({ close }) => !close, true /* inclusive */), - map(({ completion }) => completion), - filter((result): result is Uint8Array => result !== undefined) + map(() => undefined) ); } @@ -1514,7 +1568,7 @@ export class Worker { { activation, synthetic }: { activation: coresdk.workflow_activation.WorkflowActivation; synthetic: boolean } ): Promise<{ state: WorkflowWithLogAttributes | undefined; - output: { completion?: Uint8Array; close: boolean }; + output: { completion?: Uint8Array; close: boolean; fatalError?: Error }; }> { try { const removeFromCacheIx = activation.jobs.findIndex(({ removeFromCache }) => removeFromCache); @@ -1532,7 +1586,11 @@ export class Worker { activation.jobs = jobs; if (jobs.length === 0) { this.logger.trace('Disposing workflow', workflow ? workflow.logAttributes : { runId: activation.runId }); - await workflow?.workflow.dispose(); + try { + await workflow?.workflow.dispose(); + } catch (cause) { + throw new WorkflowDisposeError('Failed to dispose Workflow during eviction', cause); + } if (!close) { throw new IllegalStateError('Got a Workflow activation with no jobs'); } @@ -1617,31 +1675,56 @@ export class Worker { this.logger.trace('Completed activation', workflow.logAttributes); } } catch (error) { + const workflowStateLost = + error instanceof WorkflowLocallyEvictedError || error instanceof WorkflowThreadLostError; + if (workflowStateLost) { + this.logger.warn('Workflow Activation failed after its local Workflow state was lost', { + runId: activation.runId, + ...workflow?.logAttributes, + error, + }); + } + + if (error instanceof WorkflowDisposeError) { + const completion = synthetic + ? undefined + : coresdk.workflow_completion.WorkflowActivationCompletion.encodeDelimited({ + runId: activation.runId, + successful: {}, + }).finish(); + return { state: undefined, output: { close: true, completion, fatalError: error } }; + } + let logMessage = 'Failed to process Workflow Activation'; - if (error instanceof UnexpectedError) { + if (error instanceof UnexpectedError && !workflowStateLost) { // Something went wrong in the workflow; we'll do our best to shut the Worker // down gracefully, but then we'll need to terminate the Worker ASAP. logMessage = 'An unexpected error occurred while processing Workflow Activation. Initiating Worker shutdown.'; this.unexpectedErrorSubject.error(error); } - this.logger.error(logMessage, { - runId: activation.runId, - ...workflow?.logAttributes, - error, - workflowExists: workflow !== undefined, - }); + if (!workflowStateLost) { + this.logger.error(logMessage, { + runId: activation.runId, + ...workflow?.logAttributes, + error, + workflowExists: workflow !== undefined, + }); + } const completion = coresdk.workflow_completion.WorkflowActivationCompletion.encodeDelimited({ runId: activation.runId, failed: { - failure: await encodeErrorToFailure(this.options.loadedDataConverter, error), + failure: await encodeErrorToFailure( + this.options.loadedDataConverter, + workflowStateLost ? ApplicationFailure.fromError(error, { type: error.name }) : error + ), }, }).finish(); // We do not dispose of the Workflow yet, wait to be evicted from Core. // This is done to simplify the Workflow lifecycle so Core is the sole driver. - return { state: undefined, output: { close: true, completion } }; + return { state: workflow, output: { close: false, completion } }; } } @@ -1991,16 +2074,6 @@ export class Worker { return this.workflowPoll$().pipe( closeableGroupBy((activation) => activation.runId), mergeMap(this.handleWorkflowActivations.bind(this)), - mergeMap(async (completion) => { - try { - await this.nativeWorker.completeWorkflowActivation(Buffer.from(completion, completion.byteOffset)); - } catch (error) { - this.logger.error('Core reported failure in completeWorkflowActivation(). Initiating Worker shutdown.', { - error, - }); - this.unexpectedErrorSubject.error(error); - } - }), tap({ complete: () => { this.logger.debug('Workflow Worker terminated'); diff --git a/packages/worker/src/workflow/interface.ts b/packages/worker/src/workflow/interface.ts index b89807930c..43bef1e7f4 100644 --- a/packages/worker/src/workflow/interface.ts +++ b/packages/worker/src/workflow/interface.ts @@ -4,6 +4,13 @@ import { type WorkflowCreateOptions } from '@temporalio/workflow/lib/interfaces' export { WorkflowCreateOptions }; +export interface WorkflowThreadEvictionEvent { + runIds: string[]; + reason: 'heap-pressure' | 'thread-exit'; + usedHeapSize?: number; + heapSizeLimit?: number; +} + export interface Workflow { /** * Activate the Workflow. @@ -23,6 +30,9 @@ export interface Workflow { */ getAndResetSinkCalls(): Promise; + /** Notify the Workflow that Core accepted its latest activation completion. */ + activationCompletionAccepted?(): Promise; + /** * Dispose this instance, and release its resources. * @@ -43,4 +53,10 @@ export interface WorkflowCreator { * Destroy and cleanup any resources */ destroy(): Promise; + + /** Connect lifecycle events for Workflow creators that manage replaceable execution threads. */ + setLifecycleHandlers?( + evictionHandler: (event: WorkflowThreadEvictionEvent) => void, + fatalErrorHandler: (error: Error) => void + ): void; } diff --git a/packages/worker/src/workflow/threaded-vm-errors.ts b/packages/worker/src/workflow/threaded-vm-errors.ts new file mode 100644 index 0000000000..2b7c65f228 --- /dev/null +++ b/packages/worker/src/workflow/threaded-vm-errors.ts @@ -0,0 +1,26 @@ +export class WorkflowLocallyEvictedError extends Error { + public override readonly name = 'WorkflowLocallyEvictedError'; +} + +export class WorkflowThreadLostError extends Error { + public override readonly name = 'WorkflowThreadLostError'; + + constructor( + message: string, + public readonly cause?: unknown + ) { + super(message); + } +} + +/** Signals that Workflow cleanup failed and the owning Worker Thread must be discarded. */ +export class WorkflowThreadDisposalError extends Error { + public override readonly name = 'WorkflowThreadDisposalError'; + + constructor( + message: string, + public readonly cause?: unknown + ) { + super(message); + } +} diff --git a/packages/worker/src/workflow/threaded-vm.ts b/packages/worker/src/workflow/threaded-vm.ts index 898748befe..8ce32c4059 100644 --- a/packages/worker/src/workflow/threaded-vm.ts +++ b/packages/worker/src/workflow/threaded-vm.ts @@ -17,15 +17,26 @@ import { createUnsafeRandomSource } from '@temporalio/workflow/lib/random-helper import type { Logger } from '@temporalio/common'; import type { PatchActivationCallback } from '../worker-options'; import { UnexpectedError } from '../errors'; +import { MiB } from '../utils'; import type { + Init, PatchActivationCallbackRequest, WorkflowBundleWithSourceMapAndFilename, WorkerThreadInput, WorkerThreadRequest, } from './workflow-worker-thread/input'; -import type { Workflow, WorkflowCreateOptions, WorkflowCreator } from './interface'; -import type { WorkerThreadOutput, WorkerThreadResponse } from './workflow-worker-thread/output'; +import type { Workflow, WorkflowCreateOptions, WorkflowCreator, WorkflowThreadEvictionEvent } from './interface'; +import type { + WorkerThreadOutput, + WorkerThreadResponse, + WorkflowEvictionNotification, +} from './workflow-worker-thread/output'; import { isBunPre1_4 } from './bun'; +import { + WorkflowLocallyEvictedError, + WorkflowThreadDisposalError, + WorkflowThreadLostError, +} from './threaded-vm-errors'; import { completePatchActivationCallback, invokePatchActivationCallbackWithSnapshot, @@ -38,6 +49,7 @@ import { export const TERMINATED_EXIT_CODE = isBunPre1_4 ? 0 : 1; interface Completion { + input: WorkerThreadInput; resolve(value: T): void; reject(error: any): void; } @@ -68,47 +80,83 @@ export class WorkerThreadClient { private requestIdToCompletion = new Map>(); private shutDownRequested = false; private workerExited = false; - private activeWorkflowCount = 0; - private exitError: Error | undefined; + private exitError: WorkflowThreadLostError | undefined; + private readonly workflowRunIds = new Set(); constructor( protected workerThread: NodeWorker, protected logger: Logger, - protected patchActivationCallback?: PatchActivationCallback + protected patchActivationCallback?: PatchActivationCallback, + protected onEviction?: (notification: WorkflowEvictionNotification) => void, + protected onUnexpectedExit?: (client: WorkerThreadClient, runIds: string[], error: WorkflowThreadLostError) => void ) { - workerThread.on('message', (message: WorkerThreadResponse | PatchActivationCallbackRequest) => { - if (!('requestId' in message)) { - this.handlePatchActivationCallback(message); - return; - } - const { requestId, result } = message; - const completion = this.requestIdToCompletion.get(requestId); - if (completion === undefined) { - throw new IllegalStateError(`Got completion for unknown requestId ${requestId}`); - } - this.requestIdToCompletion.delete(requestId); - if (result.type === 'error') { - const ctor = errorNameToClass(result.name); - const err = new ctor(result.message); - err.stack = result.stack; - completion.reject(err); - return; - } + workerThread.on( + 'message', + (message: WorkerThreadResponse | PatchActivationCallbackRequest | WorkflowEvictionNotification) => { + if (!('requestId' in message)) { + if (message.type === 'patch-activation-callback') { + this.handlePatchActivationCallback(message); + } else { + for (const runId of message.runIds) this.workflowRunIds.delete(runId); + this.onEviction?.(message); + } + return; + } + const { requestId, result } = message; + const completion = this.requestIdToCompletion.get(requestId); + if (completion === undefined) { + throw new IllegalStateError(`Got completion for unknown requestId ${requestId}`); + } + if (result.type === 'error') { + if (result.name === 'WorkflowThreadDisposalError') { + const disposalError = new WorkflowThreadDisposalError(result.message); + disposalError.stack = result.stack; + this.exitError = new WorkflowThreadLostError( + 'Workflow Worker Thread failed to dispose a Workflow and will be replaced', + disposalError + ); + this.logger.warn(this.exitError.message, { error: disposalError }); + // Keep the completion pending. The exit handler rejects every outstanding request only after the + // creator has synchronously requested Core eviction for all Workflows owned by this thread. + void this.workerThread.terminate().catch((error) => { + this.logger.error('Failed to terminate Workflow Worker Thread after a disposal failure', { error }); + }); + return; + } + this.requestIdToCompletion.delete(requestId); + if (completion.input.type === 'create-workflow' || completion.input.type === 'dispose-workflow') { + const runId = + completion.input.type === 'create-workflow' + ? completion.input.options.info.runId + : completion.input.runId; + this.workflowRunIds.delete(runId); + } + const ctor = errorNameToClass(result.name); + const err = new ctor(result.message); + err.stack = result.stack; + completion.reject(err); + return; + } - completion.resolve(result.output); - }); + this.requestIdToCompletion.delete(requestId); + if (completion.input.type === 'dispose-workflow') { + this.workflowRunIds.delete(completion.input.runId); + } + completion.resolve(result.output); + } + ); workerThread.on('error', (err) => { - logger.error(`Workflow Worker Thread failed: ${err}`, err); - this.exitError = new UnexpectedError(`Workflow Worker Thread exited prematurely: ${err}`, err); + logger.warn(`Workflow Worker Thread failed and will be replaced: ${err}`, { error: err }); + this.exitError = new WorkflowThreadLostError(`Workflow Worker Thread exited prematurely: ${err}`, err); // Node will automatically terminate the Worker Thread, immediately after this event. }); workerThread.on('exit', (exitCode) => { logger.trace(`Workflow Worker Thread exited with code ${exitCode}`, { exitError: this.exitError }); this.workerExited = true; - const error = + const error: WorkflowThreadLostError = this.exitError ?? - new UnexpectedError('Workflow Worker Thread exited while there were still pending completions', { + new WorkflowThreadLostError('Workflow Worker Thread exited while there were still pending completions', { shutDownRequested: this.shutDownRequested, }); @@ -117,6 +165,9 @@ export class WorkerThreadClient { for (const completion of completions) { completion.reject(error); } + const runIds = Array.from(this.workflowRunIds); + this.workflowRunIds.clear(); + if (!this.shutDownRequested) this.onUnexpectedExit?.(this, runIds, error); }); } @@ -149,25 +200,34 @@ export class WorkerThreadClient { * Send input to Worker thread and await for output */ async send(input: WorkerThreadInput): Promise { + // Reserve new runs before checking exitError so the imminent exit event includes an init activation that raced + // with the thread's error event. Core must invalidate that activation too, even though it was never posted. + if (input.type === 'create-workflow' && !this.workerExited) { + this.workflowRunIds.add(input.options.info.runId); + } if (this.exitError || this.workerExited) { - throw this.exitError ?? new UnexpectedError('Received request after worker thread exited'); + throw this.exitError ?? new WorkflowThreadLostError('Received request after worker thread exited'); } const requestId = this.requestIdx++; const request: WorkerThreadRequest = { requestId, input }; - if (request.input.type === 'create-workflow') { - this.activeWorkflowCount++; - } else if (request.input.type === 'dispose-workflow') { - this.activeWorkflowCount--; - } - // Transfer ownership of activation buffer for zero-copy transfer - if (request.input.type === 'activate-workflow' && request.input.activation instanceof Uint8Array) { - this.workerThread.postMessage(request, [request.input.activation.buffer]); - } else { - this.workerThread.postMessage(request); - } - return new Promise((resolve, reject) => { - this.requestIdToCompletion.set(requestId, { resolve, reject }); + const result = new Promise((resolve, reject) => { + this.requestIdToCompletion.set(requestId, { input, resolve, reject }); }); + try { + // Transfer ownership of activation buffer for zero-copy transfer + if (request.input.type === 'activate-workflow' && request.input.activation instanceof Uint8Array) { + this.workerThread.postMessage(request, [request.input.activation.buffer]); + } else { + this.workerThread.postMessage(request); + } + } catch (err) { + this.requestIdToCompletion.delete(requestId); + if (request.input.type === 'create-workflow') { + this.workflowRunIds.delete(request.input.options.info.runId); + } + throw err; + } + return result; } /** @@ -204,7 +264,7 @@ export class WorkerThreadClient { } public getActiveWorkflowCount(): number { - return this.activeWorkflowCount; + return this.workflowRunIds.size; } } @@ -216,6 +276,7 @@ export interface ThreadedVMWorkflowCreatorOptions { registeredActivityNames: Set; logger: Logger; patchActivationCallback?: PatchActivationCallback; + maxWorkflowThreadHeapMiB?: number; } /** @@ -227,57 +288,156 @@ export class ThreadedVMWorkflowCreator implements WorkflowCreator { * * This method creates and initializes the workflow-worker-thread instances. */ - static async create({ - threadPoolSize, - workflowBundle, - isolateExecutionTimeoutMs, - reuseV8Context, - registeredActivityNames, - logger, - patchActivationCallback, - }: ThreadedVMWorkflowCreatorOptions): Promise { - const workerThreadClients = Array(threadPoolSize) - .fill(0) - .map( - () => - new WorkerThreadClient( - new NodeWorker(require.resolve('./workflow-worker-thread')), - logger, - patchActivationCallback - ) - ); - await Promise.all( - workerThreadClients.map((client) => - client.send({ - type: 'init', - workflowBundle, - isolateExecutionTimeoutMs, - reuseV8Context, - registeredActivityNames, - hasPatchActivationCallback: patchActivationCallback !== undefined, - }) - ) + static async create(options: ThreadedVMWorkflowCreatorOptions): Promise { + const creator = new this(options); + try { + await creator.initialize(); + return creator; + } catch (err) { + await creator.destroy(); + throw err; + } + } + + protected readonly workerThreadClients: Array; + private readonly initializingClients = new Set(); + private readonly replacementPromises = new Map>(); + private readonly pendingEvictionEvents: WorkflowThreadEvictionEvent[] = []; + private readonly pendingFatalErrors: Error[] = []; + private destroyed = false; + private evictionHandler?: (event: WorkflowThreadEvictionEvent) => void; + private fatalErrorHandler?: (error: Error) => void; + + protected constructor(protected readonly options: ThreadedVMWorkflowCreatorOptions) { + this.workerThreadClients = new Array(options.threadPoolSize); + } + + private async initialize(): Promise { + const results = await Promise.allSettled( + Array.from({ length: this.options.threadPoolSize }, (_, index) => this.spawnWorkerThread(index)) ); - return new this(workerThreadClients); + const failure = results.find((result): result is PromiseRejectedResult => result.status === 'rejected'); + if (failure !== undefined) throw failure.reason; } - constructor(protected readonly workerThreadClients: WorkerThreadClient[]) {} + private async spawnWorkerThread(index: number): Promise { + const { logger, patchActivationCallback, maxWorkflowThreadHeapMiB } = this.options; + const workerThread = new NodeWorker( + require.resolve('./workflow-worker-thread'), + maxWorkflowThreadHeapMiB === undefined + ? undefined + : { resourceLimits: { maxOldGenerationSizeMb: maxWorkflowThreadHeapMiB } } + ); + const client = new WorkerThreadClient( + workerThread, + logger, + patchActivationCallback, + (notification) => this.handleHeapEvictions(notification), + (exitedClient, runIds, error) => this.handleUnexpectedExit(index, exitedClient, runIds, error) + ); + this.initializingClients.add(client); + try { + const init: Init = { + type: 'init', + workflowBundle: this.options.workflowBundle, + isolateExecutionTimeoutMs: this.options.isolateExecutionTimeoutMs, + reuseV8Context: this.options.reuseV8Context, + registeredActivityNames: this.options.registeredActivityNames, + hasPatchActivationCallback: patchActivationCallback !== undefined, + heapSizeLimitBytes: maxWorkflowThreadHeapMiB === undefined ? undefined : maxWorkflowThreadHeapMiB * MiB, + }; + await client.send(init); + if (this.destroyed) { + await client.destroy(); + } else { + this.workerThreadClients[index] = client; + } + } catch (err) { + await client.destroy().catch(() => undefined); + throw err; + } finally { + this.initializingClients.delete(client); + } + } + + private handleHeapEvictions(notification: WorkflowEvictionNotification): void { + this.emitEvictions({ + runIds: notification.runIds, + reason: 'heap-pressure', + usedHeapSize: notification.usedHeapSize, + heapSizeLimit: notification.heapSizeLimit, + }); + } + + private handleUnexpectedExit( + index: number, + client: WorkerThreadClient, + runIds: string[], + error: WorkflowThreadLostError + ): void { + if (this.destroyed || this.initializingClients.has(client) || this.workerThreadClients[index] !== client) return; + + this.workerThreadClients[index] = undefined; + if (runIds.length > 0) this.emitEvictions({ runIds, reason: 'thread-exit' }); + this.options.logger.warn('Replacing failed Workflow Worker Thread', { + error, + affectedWorkflowCount: runIds.length, + }); + + const replacement = this.spawnWorkerThread(index) + .catch((replacementError) => { + const error = new UnexpectedError('Failed to replace Workflow Worker Thread', replacementError); + if (this.fatalErrorHandler === undefined) this.pendingFatalErrors.push(error); + else this.fatalErrorHandler(error); + }) + .finally(() => this.replacementPromises.delete(index)); + this.replacementPromises.set(index, replacement); + } + + private emitEvictions(event: WorkflowThreadEvictionEvent): void { + if (this.evictionHandler === undefined) this.pendingEvictionEvents.push(event); + else this.evictionHandler(event); + } + + /** Connect thread-local lifecycle events after the native Core Worker has been constructed. */ + setLifecycleHandlers( + evictionHandler: (event: WorkflowThreadEvictionEvent) => void, + fatalErrorHandler: (error: Error) => void + ): void { + this.evictionHandler = evictionHandler; + this.fatalErrorHandler = fatalErrorHandler; + for (const event of this.pendingEvictionEvents.splice(0)) evictionHandler(event); + for (const error of this.pendingFatalErrors.splice(0)) fatalErrorHandler(error); + } /** * Create a workflow with given options */ async createWorkflow(options: WorkflowCreateOptions): Promise { - const workerThreadClient = this.workerThreadClients.reduce((prev, curr) => - prev.getActiveWorkflowCount() < curr.getActiveWorkflowCount() ? prev : curr - ); - return await VMWorkflowThreadProxy.create(workerThreadClient, options); + for (;;) { + const availableClients = this.workerThreadClients.filter( + (client): client is WorkerThreadClient => client !== undefined + ); + if (availableClients.length > 0) { + const workerThreadClient = availableClients.reduce((prev, curr) => + prev.getActiveWorkflowCount() < curr.getActiveWorkflowCount() ? prev : curr + ); + return await VMWorkflowThreadProxy.create(workerThreadClient, options); + } + if (this.replacementPromises.size === 0) { + throw new UnexpectedError('No Workflow Worker Threads are available'); + } + await Promise.race(this.replacementPromises.values()); + } } /** * Destroy and terminate all threads created by this instance */ async destroy(): Promise { - await Promise.all(this.workerThreadClients.map((client) => client.destroy())); + this.destroyed = true; + await Promise.all(this.replacementPromises.values()); + await Promise.all(this.workerThreadClients.map((client) => client?.destroy())); } } @@ -314,6 +474,9 @@ export class VMWorkflowThreadProxy implements Workflow { type: 'extract-sink-calls', runId: this.runId, }); + if (output?.type === 'workflow-locally-evicted') { + throw new WorkflowLocallyEvictedError(`Workflow ${this.runId} was evicted by its Worker Thread`); + } if (output?.type !== 'sink-calls') { throw new TypeError(`Got invalid response output from Workflow Worker thread ${output}`); } @@ -339,6 +502,9 @@ export class VMWorkflowThreadProxy implements Workflow { activation: isBunPre1_4 ? coresdk.workflow_activation.WorkflowActivation.encode(activation).finish() : activation, runId: this.runId, }); + if (output?.type === 'workflow-locally-evicted') { + throw new WorkflowLocallyEvictedError(`Workflow ${this.runId} was evicted by its Worker Thread`); + } if (output?.type !== 'activation-completion') { throw new TypeError(`Got invalid response output from Workflow Worker thread ${output}`); } @@ -348,6 +514,10 @@ export class VMWorkflowThreadProxy implements Workflow { return output.completion; } + async activationCompletionAccepted(): Promise { + await this.workerThreadClient.send({ type: 'mark-workflow-idle', runId: this.runId }); + } + /** * Proxy request to the VMWorkflow instance */ diff --git a/packages/worker/src/workflow/workflow-thread-heap-policy.ts b/packages/worker/src/workflow/workflow-thread-heap-policy.ts new file mode 100644 index 0000000000..6e336f7dda --- /dev/null +++ b/packages/worker/src/workflow/workflow-thread-heap-policy.ts @@ -0,0 +1,17 @@ +export const WORKFLOW_HEAP_HIGH_WATERMARK = 0.8; +export const WORKFLOW_HEAP_LOW_WATERMARK = 0.7; + +/** Return the size of the bounded LRU batch to evict at this heap-usage sample. */ +export function getWorkflowHeapEvictionBatchSize( + usedHeapSize: number, + heapSizeLimit: number, + idleWorkflowCount: number +): number { + if (idleWorkflowCount === 0 || usedHeapSize < heapSizeLimit * WORKFLOW_HEAP_HIGH_WATERMARK) return 0; + + const fractionToEvict = Math.min( + 1, + Math.max(1 / idleWorkflowCount, (usedHeapSize - heapSizeLimit * WORKFLOW_HEAP_LOW_WATERMARK) / usedHeapSize) + ); + return Math.max(1, Math.ceil(idleWorkflowCount * fractionToEvict)); +} diff --git a/packages/worker/src/workflow/workflow-worker-thread.ts b/packages/worker/src/workflow/workflow-worker-thread.ts index 646de9d4c6..833f319a96 100644 --- a/packages/worker/src/workflow/workflow-worker-thread.ts +++ b/packages/worker/src/workflow/workflow-worker-thread.ts @@ -1,4 +1,5 @@ import { isMainThread, parentPort as parentPortOrNull } from 'node:worker_threads'; +import * as v8 from 'node:v8'; import { IllegalStateError } from '@temporalio/common'; import { coresdk } from '@temporalio/proto'; import type { WorkflowInfo } from '@temporalio/workflow'; @@ -6,13 +7,15 @@ import type { Workflow, WorkflowCreator } from './interface'; import { ReusableVMWorkflowCreator } from './reusable-vm'; import { VMWorkflowCreator } from './vm'; import type { PatchActivationCallbackRequest, WorkerThreadRequest } from './workflow-worker-thread/input'; -import type { WorkerThreadResponse } from './workflow-worker-thread/output'; +import type { WorkerThreadResponse, WorkflowEvictionNotification } from './workflow-worker-thread/output'; import { isBun, isBunPre1_4 } from './bun'; import { makePatchActivationWorkflowInfoSnapshot, PATCH_ACTIVATION_CALLBACK_BUFFER_SIZE, waitForPatchActivationCallbackResult, } from './patch-activation-callback'; +import { getWorkflowHeapEvictionBatchSize } from './workflow-thread-heap-policy'; +import { WorkflowThreadDisposalError } from './threaded-vm-errors'; if (isMainThread) { throw new IllegalStateError(`Imported ${__filename} from main thread`); @@ -31,6 +34,50 @@ function ok(requestId: bigint): WorkerThreadResponse { let workflowCreator: WorkflowCreator | undefined; let workflowGetter: (runId: string) => Workflow | undefined; +let heapSizeLimit = 0; +const idleWorkflows = new Map(); +const locallyEvictedWorkflows = new Set(); + +function locallyEvicted(requestId: bigint): WorkerThreadResponse { + return { requestId, result: { type: 'ok', output: { type: 'workflow-locally-evicted' } } }; +} + +/** + * Discard a bounded LRU batch when the thread is under heap pressure. + * + * Dereferencing a Workflow does not synchronously reduce V8's used-heap counter, so attempting to loop until the + * low watermark would often discard the entire cache. Instead, each safe point removes the proportion of idle + * Workflows needed to move from the observed usage toward the low watermark. Subsequent safe points remeasure. + */ +async function evictWorkflowsUnderHeapPressure(): Promise { + if (idleWorkflows.size === 0) return; + + const { used_heap_size: usedHeapSize, heap_size_limit: v8HeapSizeLimit } = v8.getHeapStatistics(); + const effectiveHeapSizeLimit = heapSizeLimit || v8HeapSizeLimit; + const evictionCount = getWorkflowHeapEvictionBatchSize(usedHeapSize, effectiveHeapSizeLimit, idleWorkflows.size); + if (evictionCount === 0) return; + const runIds = Array.from(idleWorkflows.keys()).slice(0, evictionCount); + + for (const runId of runIds) { + idleWorkflows.delete(runId); + locallyEvictedWorkflows.add(runId); + try { + await workflowGetter(runId)?.dispose(); + } catch (cause) { + // A failed dispose can leave the Workflow strongly referenced by the VM implementation. Discard the whole + // isolate so the parent can recreate it and request Core eviction for every Workflow the thread owned. + throw new WorkflowThreadDisposalError(`Failed to dispose Workflow ${runId} under heap pressure`, cause); + } + } + + const notification: WorkflowEvictionNotification = { + type: 'workflow-evictions', + runIds, + usedHeapSize, + heapSizeLimit: effectiveHeapSizeLimit, + }; + parentPort.postMessage(notification); +} function requestPatchActivation(workflowInfo: WorkflowInfo, patchId: string): boolean { const resultBuffer = new SharedArrayBuffer(PATCH_ACTIVATION_CALLBACK_BUFFER_SIZE); @@ -50,6 +97,7 @@ function requestPatchActivation(workflowInfo: WorkflowInfo, patchId: string): bo async function handleRequest({ requestId, input }: WorkerThreadRequest): Promise { switch (input.type) { case 'init': + heapSizeLimit = input.heapSizeLimitBytes ?? v8.getHeapStatistics().heap_size_limit; if (input.reuseV8Context) { workflowCreator = await ReusableVMWorkflowCreator.create( input.workflowBundle, @@ -75,10 +123,18 @@ async function handleRequest({ requestId, input }: WorkerThreadRequest): Promise if (workflowCreator === undefined) { throw new IllegalStateError('No WorkflowCreator in Worker thread'); } + if (locallyEvictedWorkflows.has(input.options.info.runId)) { + throw new IllegalStateError( + `Tried to recreate locally evicted workflow with runId: ${input.options.info.runId}` + ); + } + await evictWorkflowsUnderHeapPressure(); await workflowCreator.createWorkflow(input.options); return ok(requestId); } case 'activate-workflow': { + idleWorkflows.delete(input.runId); + if (locallyEvictedWorkflows.has(input.runId)) return locallyEvicted(requestId); const workflow = workflowGetter(input.runId); if (workflow === undefined) { throw new IllegalStateError(`Tried to activate non running workflow with runId: ${input.runId}`); @@ -107,6 +163,7 @@ async function handleRequest({ requestId, input }: WorkerThreadRequest): Promise }; } case 'extract-sink-calls': { + if (locallyEvictedWorkflows.has(input.runId)) return locallyEvicted(requestId); const workflow = workflowGetter(input.runId); if (workflow === undefined) { throw new IllegalStateError(`Tried to activate non running workflow with runId: ${input.runId}`); @@ -131,11 +188,27 @@ async function handleRequest({ requestId, input }: WorkerThreadRequest): Promise }; } case 'dispose-workflow': { + idleWorkflows.delete(input.runId); + if (locallyEvictedWorkflows.delete(input.runId)) return ok(requestId); const workflow = workflowGetter(input.runId); if (workflow === undefined) { throw new IllegalStateError(`Tried to dispose non running workflow with runId: ${input.runId}`); } - await workflow.dispose(); + try { + await workflow.dispose(); + } catch (cause) { + throw new WorkflowThreadDisposalError(`Failed to dispose Workflow ${input.runId} during Core eviction`, cause); + } + return ok(requestId); + } + case 'mark-workflow-idle': { + if (locallyEvictedWorkflows.has(input.runId) || workflowGetter(input.runId) === undefined) { + return ok(requestId); + } + // Reinsertion updates Map iteration order, giving us a compact LRU queue. + idleWorkflows.delete(input.runId); + idleWorkflows.set(input.runId, undefined); + await evictWorkflowsUnderHeapPressure(); return ok(requestId); } } diff --git a/packages/worker/src/workflow/workflow-worker-thread/input.ts b/packages/worker/src/workflow/workflow-worker-thread/input.ts index 6fbc70b6bc..7aa748f629 100644 --- a/packages/worker/src/workflow/workflow-worker-thread/input.ts +++ b/packages/worker/src/workflow/workflow-worker-thread/input.ts @@ -19,6 +19,8 @@ export interface Init { registeredActivityNames: Set; reuseV8Context: boolean; hasPatchActivationCallback: boolean; + /** Soft-limit basis. When absent, the thread uses V8's configured heap limit. */ + heapSizeLimitBytes?: number; } /** @@ -61,7 +63,20 @@ export interface DisposeWorkflow { runId: string; } -export type WorkerThreadInput = Init | Destroy | CreateWorkflow | ActivateWorkflow | ExtractSinkCalls | DisposeWorkflow; +/** Mark a Workflow as safe to evict after Core accepted its latest activation completion. */ +export interface MarkWorkflowIdle { + type: 'mark-workflow-idle'; + runId: string; +} + +export type WorkerThreadInput = + | Init + | Destroy + | CreateWorkflow + | ActivateWorkflow + | ExtractSinkCalls + | DisposeWorkflow + | MarkWorkflowIdle; /** * Request including a unique ID and input. diff --git a/packages/worker/src/workflow/workflow-worker-thread/output.ts b/packages/worker/src/workflow/workflow-worker-thread/output.ts index 5e50b2b711..97aec78c47 100644 --- a/packages/worker/src/workflow/workflow-worker-thread/output.ts +++ b/packages/worker/src/workflow/workflow-worker-thread/output.ts @@ -19,7 +19,12 @@ export interface SinkCallList { calls: SinkCall[]; } -export type WorkerThreadOutput = ActivationCompletion | SinkCallList | undefined; +/** The requested Workflow was already discarded locally while awaiting Core's eviction marker. */ +export interface WorkflowLocallyEvicted { + type: 'workflow-locally-evicted'; +} + +export type WorkerThreadOutput = ActivationCompletion | SinkCallList | WorkflowLocallyEvicted | undefined; /** * Successful result for a given request @@ -51,3 +56,11 @@ export interface WorkerThreadResponse { result: WorkerThreadOkResult | WorkflowThreadErrorResult; } + +/** Unsolicited notification that the thread discarded idle Workflows due to heap pressure. */ +export interface WorkflowEvictionNotification { + type: 'workflow-evictions'; + runIds: string[]; + usedHeapSize: number; + heapSizeLimit: number; +}