[POC] Prototype language-initiated Workflow cache eviction

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.

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<Worker>,
+    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]

packages/core-bridge/ts/native.ts

@@ -203,6 +203,8 @@ export declare function workerPollWorkflowActivation(worker: Worker): Promise<Bu
 
 export declare function workerCompleteWorkflowActivation(worker: Worker, result: Buffer): Promise<void>;
 
+export declare function workerRequestWorkflowEviction(worker: Worker, runId: string): void;
+
 export declare function workerPollActivityTask(worker: Worker): Promise<Buffer>;
 
 export declare function workerCompleteActivityTask(worker: Worker, result: Buffer): Promise<void>;

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<NativeWorkerLike> {
     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<Buffer> {
     // 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');
       },

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<number> {
+    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']);
+});

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();
     }
   });