Add configurable polling intervals for wait APIs (#1265)

## Summary
- add per-call pollingIntervalMs options for DBOS result, waitFirst,
recv, and getEvent APIs
- thread the interval through system database polling and durable sleep
waits while preserving numeric timeout compatibility
- add coverage for DBOS, workflow handle, and DBOSClient polling options
plus validation errors

---------

Co-authored-by: Peter Kraft <petereliaskraft@gmail.com>

Author: jlalmes

src/client.ts

@@ -23,8 +23,11 @@ import {
 import { cancellableSleep } from './utils';
 import {
   type GetEventOptions,
+  type PollingOptions,
+  type WaitFirstOptions,
   type SetWorkflowDelayOptions,
   type CancelWorkflowsOptions,
+  resolvePollingIntervalMs,
   resolveTimeoutSeconds,
   resolveDelayEpochMS,
 } from './dbos';
@@ -173,8 +176,15 @@ export class ClientHandle<R> implements WorkflowHandle<R> {
     return status ? toWorkflowStatus(status, this.systemDatabase.getSerializer()) : null;
   }
 
-  async getResult(): Promise<R> {
-    const res = await this.systemDatabase.awaitWorkflowResult(this.workflowID);
+  async getResult(options?: PollingOptions): Promise<R> {
+    const pollingIntervalMs = resolvePollingIntervalMs(options);
+    const res = await this.systemDatabase.awaitWorkflowResult(
+      this.workflowID,
+      undefined,
+      undefined,
+      undefined,
+      pollingIntervalMs,
+    );
     if (res?.cancelled) {
       throw new DBOSAwaitedWorkflowCancelledError(this.workflowID);
     }
@@ -479,7 +489,8 @@ export class DBOSClient {
    */
   async getEvent<T>(workflowID: string, key: string, options?: number | GetEventOptions): Promise<T | null> {
     const timeoutSeconds = resolveTimeoutSeconds(options);
-    const evt = await this.systemDatabase.getEvent(workflowID, key, timeoutSeconds ?? 60);
+    const pollingIntervalMs = resolvePollingIntervalMs(options);
+    const evt = await this.systemDatabase.getEvent(workflowID, key, timeoutSeconds ?? 60, undefined, pollingIntervalMs);
     return (await deserializeValue(evt.serializedValue, evt.serialization, this.serializer)) as T;
   }
 
@@ -556,7 +567,8 @@ export class DBOSClient {
     return listWorkflowSteps(this.systemDatabase, workflowID, true, options);
   }
 
-  async waitFirst(handles: WorkflowHandle<unknown>[]): Promise<WorkflowHandle<unknown>> {
+  async waitFirst(handles: WorkflowHandle<unknown>[], options?: WaitFirstOptions): Promise<WorkflowHandle<unknown>> {
+    const pollingIntervalMs = resolvePollingIntervalMs(options);
     if (handles.length === 0) {
       throw new Error('handles must not be empty');
     }
@@ -567,7 +579,11 @@ export class DBOSClient {
       }
       handleMap.set(handle.workflowID, handle);
     }
-    const completedId = await this.systemDatabase.awaitFirstWorkflowId([...handleMap.keys()]);
+    const completedId = await this.systemDatabase.awaitFirstWorkflowId(
+      [...handleMap.keys()],
+      undefined,
+      pollingIntervalMs,
+    );
     return handleMap.get(completedId)!;
   }
 

src/dbos-executor.ts

@@ -921,8 +921,9 @@ export class DBOSExecutor {
     workflowUUID: string,
     key: string,
     timeoutSeconds: number = DBOSExecutor.defaultNotificationTimeoutSec,
+    pollingIntervalMs?: number,
   ): Promise<T | null> {
-    const evt = await this.systemDatabase.getEvent(workflowUUID, key, timeoutSeconds);
+    const evt = await this.systemDatabase.getEvent(workflowUUID, key, timeoutSeconds, undefined, pollingIntervalMs);
     return (await deserializeValue(evt.serializedValue, evt.serialization, this.serializer)) as T;
   }
 

src/dbos.ts

@@ -189,10 +189,38 @@ export interface WriteStreamOptions {
   serializationType?: WorkflowSerializationFormat;
 }
 
+/**
+ * Options for DBOS operations that poll the system database while waiting.
+ */
+export interface PollingOptions {
+  /**
+   * Milliseconds to wait between system database polls. Must be positive and finite.
+   *
+   * This affects DB-backed polling waits. Live in-process workflow handles that await
+   * an already-running promise do not poll.
+   */
+  pollingIntervalMs?: number;
+}
+
+/**
+ * Options for `DBOS.getResult` and workflow handle `getResult`.
+ */
+export interface GetResultOptions extends PollingOptions {
+  /** Timeout in seconds; if the workflow does not complete before the timeout, `null` will be returned */
+  timeoutSeconds?: number;
+  /** Absolute deadline as a UNIX epoch timestamp in milliseconds; if the workflow does not complete before this time, `null` will be returned */
+  deadlineEpochMS?: number;
+}
+
+/**
+ * Options for `DBOS.waitFirst`.
+ */
+export type WaitFirstOptions = PollingOptions;
+
 /**
  * Options for `DBOS.recv`
  */
-export interface RecvOptions {
+export interface RecvOptions extends PollingOptions {
   /** Timeout in seconds; if no message is received before the timeout (default 60 seconds), `null` will be returned */
   timeoutSeconds?: number;
   /** Absolute deadline as a UNIX epoch timestamp in milliseconds; if no message is received before this time, `null` will be returned */
@@ -202,7 +230,7 @@ export interface RecvOptions {
 /**
  * Options for `DBOS.getEvent`
  */
-export interface GetEventOptions {
+export interface GetEventOptions extends PollingOptions {
   /** Timeout in seconds; if no event is received before the timeout (default 60 seconds), `null` will be returned */
   timeoutSeconds?: number;
   /** Absolute deadline as a UNIX epoch timestamp in milliseconds; if no event is received before this time, `null` will be returned */
@@ -238,7 +266,9 @@ export interface SetEventOptions {
   serializationType?: WorkflowSerializationFormat;
 }
 
-export function resolveTimeoutSeconds(options?: number | RecvOptions | GetEventOptions): number | undefined {
+export function resolveTimeoutSeconds(
+  options?: number | RecvOptions | GetEventOptions | GetResultOptions,
+): number | undefined {
   if (options === undefined) return undefined;
   if (typeof options === 'number') return options;
   if (options.deadlineEpochMS !== undefined) {
@@ -247,6 +277,16 @@ export function resolveTimeoutSeconds(options?: number | RecvOptions | GetEventO
   return options.timeoutSeconds;
 }
 
+export function resolvePollingIntervalMs(options?: number | PollingOptions): number | undefined {
+  if (options === undefined || typeof options === 'number') return undefined;
+  const interval = options.pollingIntervalMs;
+  if (interval === undefined) return undefined;
+  if (!Number.isFinite(interval) || interval <= 0) {
+    throw new DBOSError('pollingIntervalMs must be a positive finite number');
+  }
+  return interval;
+}
+
 export function resolveDelayEpochMS(options: number | SetWorkflowDelayOptions): number {
   if (typeof options === 'number') {
     if (options <= 0) throw new DBOSError('delaySeconds must be greater than 0');
@@ -774,21 +814,24 @@ export class DBOS {
    * @param timeoutSeconds - Maximum time to wait for result; if not provided, the operation does not time out
    * @returns The return value of the workflow, or throws the exception thrown by the workflow, or `null` if times out
    */
-  static async getResult<T>(workflowID: string, timeoutSeconds?: number): Promise<T | null> {
+  static async getResult<T>(workflowID: string, options?: number | GetResultOptions): Promise<T | null> {
     ensureDBOSIsLaunched('getResult');
+    const timeoutSeconds = resolveTimeoutSeconds(options);
+    const pollingIntervalMs = resolvePollingIntervalMs(options);
     let timerFuncID: number | undefined = undefined;
     if (DBOS.isWithinWorkflow() && timeoutSeconds !== undefined) {
       // Reserve the function ID synchronously, before any await.
       timerFuncID = functionIDGetIncrement();
     }
-    return await DBOS.getResultInternal(workflowID, timeoutSeconds, timerFuncID, undefined);
+    return await DBOS.getResultInternal(workflowID, timeoutSeconds, timerFuncID, undefined, pollingIntervalMs);
   }
 
   static async getResultInternal<T>(
     workflowID: string,
     timeoutSeconds?: number,
     timerFuncID?: number,
     assignedFuncID?: number,
+    pollingIntervalMs?: number,
   ): Promise<T | null> {
     return await runInternalStep(
       async () => {
@@ -797,6 +840,7 @@ export class DBOS {
           timeoutSeconds,
           DBOS.workflowID,
           timerFuncID,
+          pollingIntervalMs,
         );
         if (!rres) return null;
         if (rres?.cancelled) {
@@ -820,8 +864,12 @@ export class DBOS {
    * @param handles - Non-empty array of workflow handles to wait on
    * @returns The first handle whose workflow has completed
    */
-  static async waitFirst(handles: WorkflowHandle<unknown>[]): Promise<WorkflowHandle<unknown>> {
+  static async waitFirst(
+    handles: WorkflowHandle<unknown>[],
+    options?: WaitFirstOptions,
+  ): Promise<WorkflowHandle<unknown>> {
     ensureDBOSIsLaunched('waitFirst');
+    const pollingIntervalMs = resolvePollingIntervalMs(options);
     if (handles.length === 0) {
       throw new Error('handles must not be empty');
     }
@@ -836,7 +884,11 @@ export class DBOS {
     const workflowIds = [...handleMap.keys()];
 
     const completedId = await runInternalStep(async () => {
-      return await DBOSExecutor.globalInstance!.systemDatabase.awaitFirstWorkflowId(workflowIds, DBOS.workflowID);
+      return await DBOSExecutor.globalInstance!.systemDatabase.awaitFirstWorkflowId(
+        workflowIds,
+        DBOS.workflowID,
+        pollingIntervalMs,
+      );
     }, 'DBOS.waitFirst');
 
     return handleMap.get(completedId)!;
@@ -1362,12 +1414,14 @@ export class DBOS {
       const functionID: number = functionIDGetIncrement();
       const timeoutFunctionID: number = functionIDGetIncrement();
       const timeoutSeconds = resolveTimeoutSeconds(options);
+      const pollingIntervalMs = resolvePollingIntervalMs(options);
       const msg = await DBOSExecutor.globalInstance!.systemDatabase.recv(
         DBOS.workflowID!,
         functionID,
         timeoutFunctionID,
         topic,
         timeoutSeconds,
+        pollingIntervalMs,
       );
 
       return (await deserializeValue(msg.serializedValue, msg.serialization, DBOS.#executor.serializer)) as T;
@@ -1426,6 +1480,7 @@ export class DBOS {
   static async getEvent<T>(workflowID: string, key: string, options?: number | GetEventOptions): Promise<T | null> {
     ensureDBOSIsLaunched('getEvent');
     const timeoutSeconds = resolveTimeoutSeconds(options);
+    const pollingIntervalMs = resolvePollingIntervalMs(options);
     if (DBOS.isWithinWorkflow()) {
       if (!DBOS.isInWorkflow()) {
         throw new DBOSInvalidWorkflowTransitionError(
@@ -1445,10 +1500,11 @@ export class DBOS {
         key,
         timeoutSeconds ?? DBOSExecutor.defaultNotificationTimeoutSec,
         params,
+        pollingIntervalMs,
       );
       return (await deserializeValue(evt.serializedValue, evt.serialization, DBOS.#executor.serializer)) as T;
     }
-    return DBOS.#executor.getEvent(workflowID, key, timeoutSeconds);
+    return DBOS.#executor.getEvent(workflowID, key, timeoutSeconds, pollingIntervalMs);
   }
 
   /**

src/index.ts

@@ -1,4 +1,12 @@
-export { DBOS, RecvOptions, GetEventOptions, SetWorkflowDelayOptions } from './dbos';
+export {
+  DBOS,
+  RecvOptions,
+  GetEventOptions,
+  GetResultOptions,
+  PollingOptions,
+  WaitFirstOptions,
+  SetWorkflowDelayOptions,
+} from './dbos';
 
 export { DBOSClient } from './client';
 

src/system_database.ts

@@ -1703,9 +1703,11 @@ export class SystemDatabase {
     timeoutSeconds?: number,
     callerID?: string,
     timerFuncID?: number,
+    pollingIntervalMs?: number,
   ): Promise<SystemDatabaseStoredResult | undefined> {
     const timeoutms = timeoutSeconds !== undefined ? timeoutSeconds * 1000 : undefined;
     let finishTime = timeoutms !== undefined ? Date.now() + timeoutms : undefined;
+    const pollIntervalMs = pollingIntervalMs ?? this.dbPollingIntervalResultMs;
 
     while (true) {
       let resolveNotification: () => void;
@@ -1758,14 +1760,14 @@ export class SystemDatabase {
             callerID,
             timerFuncID,
             timeoutms,
-            this.dbPollingIntervalResultMs,
+            pollIntervalMs,
           );
           finishTime = endTime;
           timeoutPromise = promise;
           timeoutCancel = cancel;
         } else {
-          let poll = finishTime ? finishTime - ct : this.dbPollingIntervalResultMs;
-          poll = Math.min(this.dbPollingIntervalResultMs, poll);
+          let poll = finishTime ? finishTime - ct : pollIntervalMs;
+          poll = Math.min(pollIntervalMs, poll);
           const { promise, cancel } = cancellableSleep(poll);
           timeoutPromise = promise;
           timeoutCancel = cancel;
@@ -1784,8 +1786,9 @@ export class SystemDatabase {
   }
 
   @dbRetry()
-  async awaitFirstWorkflowId(workflowIds: string[], callerID?: string): Promise<string> {
+  async awaitFirstWorkflowId(workflowIds: string[], callerID?: string, pollingIntervalMs?: number): Promise<string> {
     const placeholders = workflowIds.map((_, i) => `$${i + 1}`).join(', ');
+    const pollIntervalMs = pollingIntervalMs ?? this.dbPollingIntervalResultMs;
 
     while (true) {
       let resolveNotification: () => void;
@@ -1816,7 +1819,7 @@ export class SystemDatabase {
           return rows[0].workflow_uuid;
         }
 
-        const { promise: sleepPromise, cancel: sleepCancel } = cancellableSleep(this.dbPollingIntervalResultMs);
+        const { promise: sleepPromise, cancel: sleepCancel } = cancellableSleep(pollIntervalMs);
         try {
           await Promise.race([wakeupPromise, sleepPromise]);
         } finally {
@@ -1939,6 +1942,7 @@ export class SystemDatabase {
     timeoutFunctionID: number,
     topic?: string,
     timeoutSeconds: number = DBOSExecutor.defaultNotificationTimeoutSec,
+    pollingIntervalMs?: number,
   ): Promise<{ serializedValue: string | null; serialization: string | null }> {
     topic = topic ?? this.nullTopic;
     const startTime = Date.now();
@@ -1953,6 +1957,7 @@ export class SystemDatabase {
 
     const timeoutms = timeoutSeconds !== undefined ? timeoutSeconds * 1000 : undefined;
     let finishTime = timeoutms !== undefined ? Date.now() + timeoutms : undefined;
+    const pollIntervalMs = pollingIntervalMs ?? this.dbPollingIntervalEventMs;
 
     while (true) {
       // register the key with the global notifications listener.
@@ -1989,14 +1994,14 @@ export class SystemDatabase {
             workflowID,
             timeoutFunctionID,
             timeoutms,
-            this.dbPollingIntervalEventMs,
+            pollIntervalMs,
           );
           timeoutPromise = promise;
           timeoutCancel = cancel;
           finishTime = endTime;
         } else {
-          let poll = finishTime ? finishTime - ct : this.dbPollingIntervalEventMs;
-          poll = Math.min(this.dbPollingIntervalEventMs, poll);
+          let poll = finishTime ? finishTime - ct : pollIntervalMs;
+          poll = Math.min(pollIntervalMs, poll);
           const { promise, cancel } = cancellableSleep(poll);
           timeoutPromise = promise;
           timeoutCancel = cancel;
@@ -2121,6 +2126,7 @@ export class SystemDatabase {
       functionID: number;
       timeoutFunctionID: number;
     },
+    pollingIntervalMs?: number,
   ): Promise<{ serializedValue: string | null; serialization: string | null }> {
     const startTime = Date.now();
     // Check if the operation has been done before for OAOO (only do this inside a workflow).
@@ -2148,6 +2154,7 @@ export class SystemDatabase {
     const payloadKey = `${workflowID}::${key}`;
     const timeoutms = timeoutSeconds !== undefined ? timeoutSeconds * 1000 : undefined;
     let finishTime = timeoutms !== undefined ? Date.now() + timeoutms : undefined;
+    const pollIntervalMs = pollingIntervalMs ?? this.dbPollingIntervalEventMs;
 
     // Register the key with the global notifications listener first... we do not want to look in the DB first
     //  or that would cause a timing hole.
@@ -2192,14 +2199,14 @@ export class SystemDatabase {
             callerWorkflow.workflowID,
             callerWorkflow.timeoutFunctionID ?? -1,
             timeoutms,
-            this.dbPollingIntervalEventMs,
+            pollIntervalMs,
           );
           timeoutPromise = promise;
           timeoutCancel = cancel;
           finishTime = endTime;
         } else {
-          let poll = finishTime ? finishTime - ct : this.dbPollingIntervalEventMs;
-          poll = Math.min(this.dbPollingIntervalEventMs, poll);
+          let poll = finishTime ? finishTime - ct : pollIntervalMs;
+          poll = Math.min(pollIntervalMs, poll);
           const { promise, cancel } = cancellableSleep(poll);
           timeoutPromise = promise;
           timeoutCancel = cancel;