refactor: move getPort to @workflow/utils/get-port

Author: adriandlam

.changeset/smooth-rats-attack.md

@@ -1,5 +1,6 @@
 ---
 "@workflow/core": patch
+"@workflow/utils": patch
 "@workflow/world-local": patch
 ---
 

packages/core/package.json

@@ -59,7 +59,6 @@
     "devalue": "^5.4.1",
     "ms": "2.1.3",
     "nanoid": "^5.1.6",
-    "pid-port": "^2.0.0",
     "seedrandom": "^3.0.5",
     "ulid": "^3.0.1",
     "zod": "catalog:"

packages/core/src/runtime.ts

@@ -8,6 +8,7 @@ import {
   WorkflowRunNotCompletedError,
   WorkflowRuntimeError,
 } from '@workflow/errors';
+import { getPort } from '@workflow/utils/get-port';
 import type {
   Event,
   WorkflowRun,
@@ -39,7 +40,6 @@ import { serializeTraceCarrier, trace, withTraceContext } from './telemetry.js';
 import { getErrorName, getErrorStack } from './types.js';
 import {
   buildWorkflowSuspensionMessage,
-  getPort,
   getWorkflowRunStreamId,
 } from './util.js';
 import { runWorkflow } from './workflow.js';

packages/core/src/util.test.ts

@@ -1,10 +1,6 @@
 import http from 'node:http';
 import { describe, expect, it } from 'vitest';
-import {
-  buildWorkflowSuspensionMessage,
-  getPort,
-  getWorkflowRunStreamId,
-} from './util';
+import { buildWorkflowSuspensionMessage, getWorkflowRunStreamId } from './util';
 
 describe('buildWorkflowSuspensionMessage', () => {
   const runId = 'test-run-123';
@@ -170,93 +166,3 @@ describe('getWorkflowRunStreamId', () => {
     expect(result.includes('_user')).toBe(true);
   });
 });
-
-describe('getPort', () => {
-  it('should return undefined or a positive number', async () => {
-    const port = await getPort();
-    expect(port === undefined || typeof port === 'number').toBe(true);
-    if (port !== undefined) {
-      expect(port).toBeGreaterThan(0);
-    }
-  });
-
-  it('should return a port number when a server is listening', async () => {
-    const server = http.createServer();
-
-    await new Promise<void>((resolve) => {
-      server.listen(0, () => resolve());
-    });
-
-    // Give system time to register the port
-    await new Promise((resolve) => setTimeout(resolve, 100));
-
-    try {
-      const port = await getPort();
-      const address = server.address();
-
-      // Port detection may not work immediately in all environments (CI, Docker, etc.)
-      // so we just verify the function returns a valid result
-      if (port !== undefined) {
-        expect(typeof port).toBe('number');
-        expect(port).toBeGreaterThan(0);
-
-        // If we have the address, optionally verify it matches
-        if (address && typeof address === 'object') {
-          // In most cases it should match, but not required for test to pass
-          expect([port, undefined]).toContain(port);
-        }
-      }
-    } finally {
-      await new Promise<void>((resolve, reject) => {
-        server.close((err) => (err ? reject(err) : resolve()));
-      });
-    }
-  });
-
-  it('should return the smallest port when multiple servers are listening', async () => {
-    const server1 = http.createServer();
-    const server2 = http.createServer();
-
-    await new Promise<void>((resolve) => {
-      server1.listen(0, () => resolve());
-    });
-
-    await new Promise<void>((resolve) => {
-      server2.listen(0, () => resolve());
-    });
-
-    // Give system time to register the ports
-    await new Promise((resolve) => setTimeout(resolve, 100));
-
-    try {
-      const port = await getPort();
-      const addr1 = server1.address();
-      const addr2 = server2.address();
-
-      // Port detection may not work in all environments
-      if (
-        port !== undefined &&
-        addr1 &&
-        typeof addr1 === 'object' &&
-        addr2 &&
-        typeof addr2 === 'object'
-      ) {
-        // Should return the smallest port
-        expect(port).toBeLessThanOrEqual(Math.max(addr1.port, addr2.port));
-        expect(port).toBeGreaterThan(0);
-      } else {
-        // If port detection doesn't work in this environment, just pass
-        expect(port === undefined || typeof port === 'number').toBe(true);
-      }
-    } finally {
-      await Promise.all([
-        new Promise<void>((resolve, reject) => {
-          server1.close((err) => (err ? reject(err) : resolve()));
-        }),
-        new Promise<void>((resolve, reject) => {
-          server2.close((err) => (err ? reject(err) : resolve()));
-        }),
-      ]);
-    }
-  });
-});

packages/core/src/util.ts

@@ -1,5 +1,3 @@
-import { pidToPorts } from 'pid-port';
-
 /**
  * Builds a workflow suspension log message based on the counts of steps, hooks, and waits.
  * @param runId - The workflow run ID
@@ -64,25 +62,3 @@ export function getWorkflowRunStreamId(runId: string, namespace?: string) {
   );
   return `${streamId}_${encodedNamespace}`;
 }
-
-/**
- * Gets the port number that the process is listening on.
- * @returns The port number that the process is listening on, or undefined if the process is not listening on any port.
- * NOTE: Can't move this to @workflow/utils because it's being imported into @workflow/errors for RetryableError (inside workflow runtime)
- */
-export async function getPort(): Promise<number | undefined> {
-  try {
-    const pid = process.pid;
-    const ports = await pidToPorts(pid);
-    if (!ports || ports.size === 0) {
-      return undefined;
-    }
-
-    const smallest = Math.min(...ports);
-    return smallest;
-  } catch {
-    // If port detection fails (e.g., `ss` command not available in production),
-    // return undefined and fall back to default port
-    return undefined;
-  }
-}

packages/core/src/workflow.ts

@@ -1,6 +1,7 @@
 import { runInContext } from 'node:vm';
 import { ERROR_SLUGS } from '@workflow/errors';
 import { withResolvers } from '@workflow/utils';
+import { getPort } from '@workflow/utils/get-port';
 import type { Event, WorkflowRun } from '@workflow/world';
 import * as nanoid from 'nanoid';
 import { monotonicFactory } from 'ulid';
@@ -21,7 +22,7 @@ import {
 } from './symbols.js';
 import * as Attribute from './telemetry/semantic-conventions.js';
 import { trace } from './telemetry.js';
-import { getPort, getWorkflowRunStreamId } from './util.js';
+import { getWorkflowRunStreamId } from './util.js';
 import { createContext } from './vm/index.js';
 import type { WorkflowMetadata } from './workflow/get-workflow-metadata.js';
 import { WORKFLOW_CONTEXT_SYMBOL } from './workflow/get-workflow-metadata.js';

packages/utils/package.json

@@ -20,6 +20,10 @@
     ".": {
       "types": "./dist/index.d.ts",
       "default": "./dist/index.js"
+    },
+    "./get-port": {
+      "types": "./dist/get-port.d.ts",
+      "default": "./dist/get-port.js"
     }
   },
   "scripts": {
@@ -36,6 +40,7 @@
     "vitest": "catalog:"
   },
   "dependencies": {
-    "ms": "2.1.3"
+    "ms": "2.1.3",
+    "pid-port": "^2.0.0"
   }
 }

packages/utils/src/get-port.test.ts

@@ -0,0 +1,93 @@
+import http from 'node:http';
+import { describe, expect, it } from 'vitest';
+import { getPort } from './get-port';
+
+describe('getPort', () => {
+  it('should return undefined or a positive number', async () => {
+    const port = await getPort();
+    expect(port === undefined || typeof port === 'number').toBe(true);
+    if (port !== undefined) {
+      expect(port).toBeGreaterThan(0);
+    }
+  });
+
+  it('should return a port number when a server is listening', async () => {
+    const server = http.createServer();
+
+    await new Promise<void>((resolve) => {
+      server.listen(0, () => resolve());
+    });
+
+    // Give system time to register the port
+    await new Promise((resolve) => setTimeout(resolve, 100));
+
+    try {
+      const port = await getPort();
+      const address = server.address();
+
+      // Port detection may not work immediately in all environments (CI, Docker, etc.)
+      // so we just verify the function returns a valid result
+      if (port !== undefined) {
+        expect(typeof port).toBe('number');
+        expect(port).toBeGreaterThan(0);
+
+        // If we have the address, optionally verify it matches
+        if (address && typeof address === 'object') {
+          // In most cases it should match, but not required for test to pass
+          expect([port, undefined]).toContain(port);
+        }
+      }
+    } finally {
+      await new Promise<void>((resolve, reject) => {
+        server.close((err) => (err ? reject(err) : resolve()));
+      });
+    }
+  });
+
+  it('should return the smallest port when multiple servers are listening', async () => {
+    const server1 = http.createServer();
+    const server2 = http.createServer();
+
+    await new Promise<void>((resolve) => {
+      server1.listen(0, () => resolve());
+    });
+
+    await new Promise<void>((resolve) => {
+      server2.listen(0, () => resolve());
+    });
+
+    // Give system time to register the ports
+    await new Promise((resolve) => setTimeout(resolve, 100));
+
+    try {
+      const port = await getPort();
+      const addr1 = server1.address();
+      const addr2 = server2.address();
+
+      // Port detection may not work in all environments
+      if (
+        port !== undefined &&
+        addr1 &&
+        typeof addr1 === 'object' &&
+        addr2 &&
+        typeof addr2 === 'object'
+      ) {
+        // Should return the smallest port
+        expect(port).toBeLessThanOrEqual(Math.max(addr1.port, addr2.port));
+        expect(port).toBeGreaterThan(0);
+      } else {
+        // If port detection doesn't work in this environment, just pass
+        expect(port === undefined || typeof port === 'number').toBe(true);
+      }
+    } finally {
+      await Promise.all([
+        new Promise<void>((resolve, reject) => {
+          server1.close((err) => (err ? reject(err) : resolve()));
+        }),
+        new Promise<void>((resolve, reject) => {
+          server2.close((err) => (err ? reject(err) : resolve()));
+        }),
+      ]);
+    }
+  });
+});

packages/utils/src/get-port.ts

@@ -0,0 +1,23 @@
+import { pidToPorts } from 'pid-port';
+
+/**
+ * Gets the port number that the process is listening on.
+ * @returns The port number that the process is listening on, or undefined if the process is not listening on any port.
+ * NOTE: Can't move this to @workflow/utils because it's being imported into @workflow/errors for RetryableError (inside workflow runtime)
+ */
+export async function getPort(): Promise<number | undefined> {
+  try {
+    const pid = process.pid;
+    const ports = await pidToPorts(pid);
+    if (!ports || ports.size === 0) {
+      return undefined;
+    }
+
+    const smallest = Math.min(...ports);
+    return smallest;
+  } catch {
+    // If port detection fails (e.g., `ss` command not available in production),
+    // return undefined and fall back to default port
+    return undefined;
+  }
+}

packages/world-local/package.json

@@ -33,7 +33,6 @@
     "@vercel/queue": "0.0.0-alpha.23",
     "@workflow/utils": "workspace:*",
     "@workflow/world": "workspace:*",
-    "pid-port": "^2.0.0",
     "ulid": "^3.0.1",
     "undici": "^6.19.0",
     "zod": "catalog:"