Debounce With Delay

schemas/system_db_schema.ts

@@ -36,6 +36,8 @@ export interface workflow_status {
   completed_at?: number | null;
   attributes?: Record<string, unknown> | null; // Custom key-value attributes attached at creation, stored as JSONB.
   schedule_name?: string | null; // If enqueued by a named schedule, that schedule's name.
+  debounce_deadline_epoch_ms?: number | null; // Absolute cap (epoch ms) beyond which bounces may not extend the delay.
+  is_debounced?: boolean; // True if the deduplication ID is a debounce key cleared on the DELAYED->ENQUEUED transition.
 }
 
 export interface notifications {

src/client.ts

@@ -4,6 +4,8 @@ import {
   type WorkflowStatusInternal,
   type WorkflowScheduleInternal,
   type VersionInfo,
+  type DebounceParams,
+  type DebounceResult,
   DBOS_STREAM_CLOSED_SENTINEL,
   DEFAULT_POOL_SIZE,
 } from './system_database';
@@ -302,6 +304,20 @@ export class DBOSClient {
     options: ClientEnqueueOptions,
     ...args: Parameters<T>
   ): Promise<WorkflowHandle<Awaited<ReturnType<T>>>> {
+    const internalStatus = await this.#buildEnqueueStatus(options, args);
+
+    let finalID: string;
+    if (options.duplicationPolicy === 'return-existing') {
+      finalID = await this.#initSingletonWorkflow(internalStatus, options);
+    } else {
+      await this.systemDatabase.initWorkflowStatus(internalStatus, null);
+      finalID = internalStatus.workflowUUID;
+    }
+
+    return new ClientHandle<Awaited<ReturnType<T>>>(this.systemDatabase, finalID);
+  }
+
+  async #buildEnqueueStatus(options: ClientEnqueueOptions, args: unknown[]): Promise<WorkflowStatusInternal> {
     validateWorkflowAttributes(options.attributes);
     const { workflowName, workflowClassName, workflowConfigName, queueName, appVersion } = options;
     const workflowUUID = options.workflowID ?? randomUUID();
@@ -311,7 +327,7 @@ export class DBOSClient {
       options.delaySeconds !== undefined && options.delaySeconds > 0
         ? Date.now() + options.delaySeconds * 1000
         : undefined;
-    const internalStatus: WorkflowStatusInternal = {
+    return {
       workflowUUID: workflowUUID,
       status: delayUntilEpochMS !== undefined ? StatusString.DELAYED : StatusString.ENQUEUED,
       workflowName: workflowName,
@@ -338,16 +354,39 @@ export class DBOSClient {
       delayUntilEpochMS,
       attributes: options.attributes,
     };
+  }
 
-    let finalID: string;
-    if (options.duplicationPolicy === 'return-existing') {
-      finalID = await this.#initSingletonWorkflow(internalStatus, options);
-    } else {
-      await this.systemDatabase.initWorkflowStatus(internalStatus, null);
-      finalID = internalStatus.workflowUUID;
+  /**
+   * Enqueue a debounced workflow for `DebouncerClient`.
+   * The debounce fields are stamped onto the built status here because they are
+   * not part of the public enqueue API.
+   * @internal
+   */
+  async enqueueDebounced(
+    options: ClientEnqueueOptions,
+    debounceDeadlineEpochMS: number | undefined,
+    args: unknown[],
+  ): Promise<string> {
+    const internalStatus = await this.#buildEnqueueStatus(options, args);
+    internalStatus.debounceDeadlineEpochMS = debounceDeadlineEpochMS;
+    internalStatus.isDebounced = true;
+    if (
+      internalStatus.delayUntilEpochMS !== undefined &&
+      debounceDeadlineEpochMS !== undefined &&
+      debounceDeadlineEpochMS < internalStatus.delayUntilEpochMS
+    ) {
+      internalStatus.delayUntilEpochMS = debounceDeadlineEpochMS;
     }
+    await this.systemDatabase.initWorkflowStatus(internalStatus, null);
+    return internalStatus.workflowUUID;
+  }
 
-    return new ClientHandle<Awaited<ReturnType<T>>>(this.systemDatabase, finalID);
+  /**
+   * Extend an existing debounced DELAYED workflow, for `DebouncerClient`.
+   * @internal
+   */
+  async debounceDelayedWorkflow(params: DebounceParams): Promise<DebounceResult> {
+    return await this.systemDatabase.debounceDelayedWorkflow(params);
   }
 
   /**

src/dbos-executor.ts

@@ -56,7 +56,7 @@ import {
   runWithTopContext,
 } from './context';
 import { deserializeError, serializeError } from 'serialize-error';
-import { globalParams, sleepms, INTERNAL_QUEUE_NAME, DEBOUNCER_WORKLOW_NAME as DEBOUNCER_WORKLOW_NAME } from './utils';
+import { globalParams, sleepms, INTERNAL_QUEUE_NAME } from './utils';
 import {
   DBOSPortableJSON,
   DBOSSerializer,
@@ -69,7 +69,7 @@ import {
   serializeResErrorWithSerializer,
   serializeValue,
 } from './serialization';
-import { DBOS, GetWorkflowsInput } from '.';
+import { GetWorkflowsInput } from '.';
 
 import { wfQueueRunner, WorkflowQueue } from './wfqueue';
 import { debugTriggerPoint, DEBUG_TRIGGER_WORKFLOW_ENQUEUE } from './debugpoint';
@@ -84,7 +84,6 @@ import {
   toWorkflowStatus,
 } from './workflow_management';
 import { maskDatabaseUrl } from './database_utils';
-import { debouncerWorkflowFunction } from './debouncer';
 import { Pool } from 'pg';
 
 // eslint-disable-next-line @typescript-eslint/no-empty-object-type
@@ -485,10 +484,18 @@ export class DBOSExecutor {
         : (funcArgs.deserialized as T);
 
     const delaySeconds = params.enqueueOptions?.delaySeconds;
-    const delayUntilEpochMS =
+    let delayUntilEpochMS =
       params.queueName !== undefined && delaySeconds !== undefined && delaySeconds > 0
         ? Date.now() + delaySeconds * 1000
         : undefined;
+    const debounceDeadlineEpochMS = params.enqueueOptions?.debounceDeadlineEpochMS;
+    if (
+      delayUntilEpochMS !== undefined &&
+      debounceDeadlineEpochMS !== undefined &&
+      debounceDeadlineEpochMS < delayUntilEpochMS
+    ) {
+      delayUntilEpochMS = debounceDeadlineEpochMS;
+    }
     const internalStatus: WorkflowStatusInternal = {
       workflowUUID: workflowID,
       status:
@@ -521,6 +528,8 @@ export class DBOSExecutor {
       serialization: funcArgs.sername,
       delayUntilEpochMS,
       attributes: params.workflowAttributes,
+      debounceDeadlineEpochMS,
+      isDebounced: params.enqueueOptions?.isDebounced ?? false,
     };
 
     if (isTempWorkflow) {
@@ -1355,15 +1364,4 @@ export class DBOSExecutor {
     }
     DBOSExecutor.internalQueue = new WorkflowQueue(INTERNAL_QUEUE_NAME, {});
   }
-
-  static debouncerWorkflow: UntypedAsyncFunction | undefined = undefined;
-
-  static createDebouncerWorkflow() {
-    if (DBOSExecutor.debouncerWorkflow !== undefined) {
-      return;
-    }
-    DBOSExecutor.debouncerWorkflow = DBOS.registerWorkflow(debouncerWorkflowFunction, {
-      name: DEBOUNCER_WORKLOW_NAME,
-    }) as UntypedAsyncFunction;
-  }
 }

src/dbos.ts

@@ -352,7 +352,7 @@ export function runInternalStep<T>(
  * methods so they participate in the same transaction.
  * Outside a workflow, the callback is called directly with `undefined` as the client.
  */
-async function runTransactionalInternalStep<T>(
+export async function runTransactionalInternalStep<T>(
   callback: (client: PoolClient | undefined) => Promise<T>,
   funcName: string,
 ): Promise<T> {
@@ -457,7 +457,6 @@ export class DBOS {
       globalParams.executorID = randomUUID();
     }
 
-    DBOSExecutor.createDebouncerWorkflow();
     DBOSExecutor.createInternalQueue();
     DBOSExecutor.globalInstance = new DBOSExecutor(internalConfig);
 
@@ -619,7 +618,6 @@ export class DBOS {
     assert(!DBOS.isInitialized(), 'Cannot call DBOS.clearRegistry after DBOS.launch');
     clearAllRegistrations();
     wfQueueRunner.clearRegistrations();
-    DBOSExecutor.debouncerWorkflow = undefined;
     DBOSExecutor.internalQueue = undefined;
   }