Add Debouncer — coalesce rapid workflow calls into one execution (#390)

Implements a debounce mechanism for DBOS workflows analogous to
dbos-transact-py _debouncer.py. Multiple calls with the same key within
a period are collapsed into a single user-workflow execution that runs
with the most recently supplied arguments.

Architecture:
- DebouncerServiceImpl: internal @workflow that runs a recv-loop,
absorbing messages until the debounce period times out or the absolute
debounceTimeout elapses, then starts the user workflow.
- Debouncer<R>: public fluent API. Enqueues the service workflow on
_dbos_internal_queue with a deduplicationId derived from (workflowName,
debounceKey). On DBOSQueueDuplicatedException, forwards a message to the
running debouncer and waits for an ack.
- DBOSExecutor.captureInvocation(): extracted from startWorkflow so
Debouncer can capture a lambda's workflow call without executing it.
- Auto-registration of DebouncerService in DBOS constructor so users
need no boilerplate setup.
- Internal system workflows filtered from getRegisteredWorkflows /
getRegisteredWorkflowInstances to keep public counts clean.

Usage:

```java
var handle = dbos.<String>debouncer()
    .withDebounceTimeout(Duration.ofMinutes(5))
     .debounce("key", Duration.ofSeconds(2), () -> svc.process(arg)); 
String result = handle.getResult();
```
Tests: 6 integration tests via Testcontainers Postgres covering
single-call, multi-call coalescing, absolute timeout, independent keys,
concurrent callers, and queue-based user workflow.

---------

Co-authored-by: Harry Pierson <harrypierson@hotmail.com>
Co-authored-by: Harry Pierson <harry.pierson@dbos.dev>

Author: easmith

transact/src/main/java/dev/dbos/transact/Constants.java

@@ -16,6 +16,13 @@ public class Constants {
 
   public static final String DBOS_INTERNAL_QUEUE = "_dbos_internal_queue";
 
+  public static final String DEBOUNCER_WORKFLOW_NAME = "debouncerWorkflow";
+  public static final String DEBOUNCER_CLASS_NAME = "DBOS.InternalWorkflows";
+  public static final String DEBOUNCER_TOPIC = "_dbos_debouncer_topic";
+  // Event key published by the debouncer-workflow so callers can retrieve the pre-assigned
+  // user workflow id without relying on Jackson deserialization of workflow inputs.
+  public static final String DEBOUNCER_CHILD_ID_KEY = "_dbos_debouncer_child_id";
+
   public static final String SYSTEM_JDBC_URL_ENV_VAR = "DBOS_SYSTEM_JDBC_URL";
 
   public static final int DEFAULT_MAX_RECOVERY_ATTEMPTS = 100;

transact/src/main/java/dev/dbos/transact/DBOS.java

@@ -4,13 +4,15 @@
 import dev.dbos.transact.context.DBOSContext;
 import dev.dbos.transact.execution.DBOSExecutor;
 import dev.dbos.transact.execution.DBOSLifecycleListener;
+import dev.dbos.transact.execution.RegisteredWorkflow;
 import dev.dbos.transact.execution.ThrowingRunnable;
 import dev.dbos.transact.execution.ThrowingSupplier;
 import dev.dbos.transact.internal.DBOSIntegration;
 import dev.dbos.transact.internal.DBOSInvocationHandler;
 import dev.dbos.transact.internal.QueueRegistry;
 import dev.dbos.transact.internal.WorkflowRegistry;
 import dev.dbos.transact.migrations.MigrationManager;
+import dev.dbos.transact.workflow.Debouncer;
 import dev.dbos.transact.workflow.ForkOptions;
 import dev.dbos.transact.workflow.ListWorkflowsInput;
 import dev.dbos.transact.workflow.Queue;
@@ -28,6 +30,7 @@
 import dev.dbos.transact.workflow.WorkflowHandle;
 import dev.dbos.transact.workflow.WorkflowSchedule;
 import dev.dbos.transact.workflow.WorkflowStatus;
+import dev.dbos.transact.workflow.internal.InternalWorkflows;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -65,6 +68,7 @@ public class DBOS implements AutoCloseable {
   private final DBOSConfig config;
   private final AtomicReference<DBOSExecutor> dbosExecutor = new AtomicReference<>();
   private final DBOSIntegration integration;
+  private final RegisteredWorkflow debouncerWorkflow;
 
   private AlertHandler alertHandler;
 
@@ -87,6 +91,16 @@ public DBOS(@NonNull DBOSConfig config) {
     this.integration =
         new DBOSIntegration(
             this.config, this.workflowRegistry, dbosExecutor::get, this::registerLifecycleListener);
+    // Register the built-in debouncer service workflow directly (without a proxy) so callers can
+    // use Debouncer without having to declare and wire the service themselves.
+    var internalWorkflows = new InternalWorkflows(this, dbosExecutor::get);
+    workflowRegistry.registerInternalInstance(internalWorkflows);
+    this.debouncerWorkflow =
+        workflowRegistry.registerInternalWorkflow(
+            Constants.DEBOUNCER_WORKFLOW_NAME,
+            Constants.DEBOUNCER_CLASS_NAME,
+            internalWorkflows,
+            InternalWorkflows.debouncerWorkflowMethod());
   }
 
   /**
@@ -346,6 +360,8 @@ public void launch() {
             new HashSet<>(this.lifecycleRegistry),
             workflowRegistry.getWorkflowSnapshot(),
             workflowRegistry.getInstanceSnapshot(),
+            workflowRegistry.getInternalWorkflowSnapshot(),
+            workflowRegistry.getInternalInstanceSnapshot(),
             queueRegistry.getSnapshot(),
             alertHandler);
       }
@@ -462,6 +478,21 @@ public void sleep(@NonNull Duration duration) {
     return startWorkflow(runnable, null);
   }
 
+  /**
+   * Build a {@link Debouncer} that consolidates a series of calls on the same key into one
+   * execution of the targeted workflow using the most recent arguments.
+   *
+   * <p>The returned debouncer is immutable; configuration helpers like {@link
+   * Debouncer#withQueue(String)} and {@link Debouncer#withDebounceTimeout(java.time.Duration)}
+   * return new instances.
+   *
+   * @param <R> the return type of the debounced workflow (used only for type inference)
+   * @return a fresh debouncer bound to this DBOS instance
+   */
+  public <R> @NonNull Debouncer<R> debouncer() {
+    return new Debouncer<>(this, ensureLaunched("debouncer"), debouncerWorkflow);
+  }
+
   /**
    * Returns the DBOS integration APIs for use by specialized integrations such as AOP aspects and
    * event listeners.

transact/src/main/java/dev/dbos/transact/DBOSClient.java

@@ -815,6 +815,29 @@ public void sendBulk(@NonNull List<SendMessage> messages, @Nullable SendOptions
     return new WorkflowHandleClient<>(workflowId);
   }
 
+  /**
+   * Find the workflow ID of the active workflow with the given queue and deduplication ID.
+   *
+   * @param queueName name of the queue to search
+   * @param deduplicationId deduplication ID to look up
+   * @return the workflow ID, or {@code null} if not found
+   */
+  public @Nullable String findWorkflowIdByDeduplicationId(
+      @NonNull String queueName, @NonNull String deduplicationId) {
+    return systemDatabase.findWorkflowIdByDeduplicationId(queueName, deduplicationId);
+  }
+
+  /**
+   * Create a {@link DebouncerClient} that coalesces repeated calls on the same key into a single
+   * execution of the named workflow.
+   *
+   * @param workflowName name of the workflow function to debounce
+   * @return a new DebouncerClient bound to this client
+   */
+  public <R> @NonNull DebouncerClient<R> debouncer(@NonNull String workflowName) {
+    return new DebouncerClient<>(this, workflowName);
+  }
+
   /**
    * Cancel a worflow
    *