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 e38e908bb3..eab677ee43 100644 --- a/packages/core/src/runtime/get-port-lazy.ts +++ b/packages/core/src/runtime/get-port-lazy.ts @@ -1,11 +1,11 @@ import { getPort } from '@workflow/utils/get-port'; -// 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 aefac8aee3..13476ab0be 100644 --- a/packages/core/src/runtime/step-handler.test.ts +++ b/packages/core/src/runtime/step-handler.test.ts @@ -134,23 +134,33 @@ vi.mock('../private.js', () => ({ getStepFunction: vi.fn().mockReturnValue(mockStepFn), })); -// Mock get-port -vi.mock('@workflow/utils/get-port', () => ({ - getPort: vi.fn().mockResolvedValue(3000), -})); - // Import the module AFTER all mocks are set up - this triggers createQueueHandler // which populates capturedHandlerRef import './step-handler.js'; -import { MAX_QUEUE_DELIVERIES } from './constants.js'; import { getStepFunction } from '../private.js'; import { getErrorName, getErrorStack, normalizeUnknownError, } from '../types.js'; +import { MAX_QUEUE_DELIVERIES } from './constants.js'; +import { + resetPortCacheForTesting, + setPortResolverForTesting, +} from './get-port-lazy.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 } @@ -228,6 +238,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 440612826d..c0efefd524 100644 --- a/packages/core/src/runtime/step-handler.ts +++ b/packages/core/src/runtime/step-handler.ts @@ -10,7 +10,6 @@ import { WorkflowWorldError, } from '@workflow/errors'; import { createWorkflowBaseUrl, pluralize } from '@workflow/utils'; -import { getPort } from '@workflow/utils/get-port'; import { getQueueTopicPrefix, resolveQueueNamespace, @@ -41,6 +40,7 @@ import { } from '../types.js'; import { MAX_QUEUE_DELIVERIES } from './constants.js'; +import { getPortLazy } from './get-port-lazy.js'; import { getQueueOverhead, getWorkflowQueueName, @@ -158,7 +158,7 @@ const stepHandler = createQueueHandler( // Resolve local async values concurrently before entering the trace span const [port, spanKind] = await Promise.all([ - isVercel ? undefined : getPort(), + isVercel ? undefined : getPortLazy(), getSpanKind('CONSUMER'), ]); diff --git a/packages/world-postgres/src/queue.test.ts b/packages/world-postgres/src/queue.test.ts index 9a48b0d37b..2c2079d5d1 100644 --- a/packages/world-postgres/src/queue.test.ts +++ b/packages/world-postgres/src/queue.test.ts @@ -29,6 +29,9 @@ vi.mock('graphile-worker', () => ({ vi.mock('@workflow/utils/get-port', () => ({ getWorkflowPort: vi.fn(), + // stepEntrypoint (imported from the core dist bundle below) calls + // getPortLazy(), which imports getPort from this same module. + getPort: vi.fn().mockResolvedValue(undefined), })); vi.mock('@workflow/world-local', async (importOriginal) => {