diff --git a/.changeset/cache-step-port-discovery.md b/.changeset/cache-step-port-discovery.md new file mode 100644 index 0000000000..28903e56df --- /dev/null +++ b/.changeset/cache-step-port-discovery.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Cache self-hosted step port discovery per process. diff --git a/packages/core/src/runtime/get-port-lazy.ts b/packages/core/src/runtime/get-port-lazy.ts index 605e39ad68..dbba7651f9 100644 --- a/packages/core/src/runtime/get-port-lazy.ts +++ b/packages/core/src/runtime/get-port-lazy.ts @@ -12,12 +12,12 @@ import { pathToFileURL } from 'node:url'; let _getPort: (() => Promise) | undefined; -// Per-process cache of the resolved port. The workflow dev server listens on a +// Per-process cache of the resolved port. The workflow server listens on a // stable port for the lifetime of the process, but `getPort()` rediscovers it // on every call by querying the OS for the process's listening sockets — on // macOS that shells out to `lsof` (~60ms), which the runtime pays on EVERY -// workflow replay (the non-Vercel branch of `runWorkflow`). Since the port does -// not change within a process, resolve it once and reuse it. `_inFlight` +// workflow replay or step invocation. Since the port does not change within a +// process, resolve it once and reuse it. `_inFlight` // dedupes concurrent first calls so discovery never runs more than once. // // The first concrete port is pinned for the lifetime of the process — there is diff --git a/packages/core/src/runtime/step-handler.test.ts b/packages/core/src/runtime/step-handler.test.ts index 1304cce25c..ac3920d0a6 100644 --- a/packages/core/src/runtime/step-handler.test.ts +++ b/packages/core/src/runtime/step-handler.test.ts @@ -178,11 +178,6 @@ vi.mock('../private.js', () => ({ getStepFunction: vi.fn().mockReturnValue(mockStepFn), })); -// Mock get-port -vi.mock('@workflow/utils/get-port', () => ({ - getPort: vi.fn().mockResolvedValue(3000), -})); - import { getStepFunction } from '../private.js'; import { cancelAbortReaders, dehydrateStepError } from '../serialization.js'; import { @@ -191,6 +186,10 @@ import { normalizeUnknownError, } from '../types.js'; import { MAX_QUEUE_DELIVERIES } from './constants.js'; +import { + resetPortCacheForTesting, + setPortResolverForTesting, +} from './get-port-lazy.js'; import { executeStep } from './step-executor.js'; // Import the module AFTER all mocks are set up // Since getWorldHandlers is now async, we need to call stepEntrypoint @@ -198,6 +197,17 @@ import { executeStep } from './step-executor.js'; import { stepEntrypoint } from './step-handler.js'; import { getWorld } from './world.js'; +const mockPortResolver = vi.fn(async () => 3000); + +beforeEach(() => { + mockPortResolver.mockReset().mockResolvedValue(3000); + setPortResolverForTesting(mockPortResolver); +}); + +afterEach(() => { + resetPortCacheForTesting(); +}); + function capturedHandler( message: unknown, metadata: { queueName: string; messageId: string; attempt: number } @@ -281,6 +291,13 @@ describe('step-handler 409 handling', () => { vi.restoreAllMocks(); }); + it('reuses cached port discovery across step invocations', async () => { + await capturedHandler(createMessage(), createMetadata('myStep')); + await capturedHandler(createMessage(), createMetadata('myStep')); + + expect(mockPortResolver).toHaveBeenCalledOnce(); + }); + describe('step_completed 409', () => { it('should warn and return when step_completed gets a 409', async () => { // step_started succeeds, step function succeeds, step_completed returns 409 diff --git a/packages/core/src/runtime/step-handler.ts b/packages/core/src/runtime/step-handler.ts index adbcd6e495..e5460ee2ac 100644 --- a/packages/core/src/runtime/step-handler.ts +++ b/packages/core/src/runtime/step-handler.ts @@ -15,7 +15,6 @@ import { pluralize, stepDisplayName, } from '@workflow/utils'; -import { getPort } from '@workflow/utils/get-port'; import { getQueueTopicPrefix, resolveQueueNamespace, @@ -53,6 +52,7 @@ import { } from '../types.js'; import { getMaxQueueDeliveries } from './constants.js'; +import { getPortLazy } from './get-port-lazy.js'; import { getQueueOverhead, getWorkflowQueueName, @@ -224,7 +224,7 @@ function createStepHandler(namespace?: string) { // Resolve local async values concurrently before entering the trace span const [port, spanKind] = await Promise.all([ - isVercel ? undefined : getPort(), + isVercel ? undefined : getPortLazy(), getSpanKind('CONSUMER'), ]);