Debouncing (#455)

Workflows can now be debounced. Debouncing delays workflow execution
until some time has passed since the workflow has last been called. This
is useful for preventing wasted work when a workflow may be triggered
multiple times in quick succession.
For example, if a user is editing a form, we can debounce every change
to the form and perform a synchronization workflow only after they
haven't edited the form for a certain period of time.

### Debouncer.create

```python
Debouncer.create(
    workflow: Callable[P, R],
    *,
    debounce_key: str,
    debounce_timeout_sec: Optional[float] = None,
    queue: Optional[Queue] = None,
) -> Debouncer[P, R]
```

**Parameters:**
- `workflow`: The workflow to debounce.
- `debounce_key`: The **unique** debounce key for this debouncer. Used
to group workflows that will be debounced.
- `debounce_timeout_sec`: After this time elapses since the first time a
workflow is submitted from this debouncer, the workflow is started
regardless of the debounce period.
- `queue`: When starting a workflow after debouncing, enqueue it on this
queue instead of executing it directly.

### debounce

```python
debouncer.debounce(
    debounce_period_sec: float,
    *args: P.args,
    **kwargs: P.kwargs,
) -> WorkflowHandle[R]
```

Submit a workflow for execution but delay it by `debounce_period_sec`.
Returns a handle to the workflow.
The workflow may be debounced again, which further delays its execution
(up to `debounce_timeout_sec`).
When the workflow eventually executes, it uses the **last** set of
inputs passed into `debounce`.

After the workflow begins execution, the next call to `debounce` starts
the debouncing process again for a new workflow execution.

**Example Syntax**:

```python
@DBOS.workflow()
def process_input(user_input):
    ...

# Each time a user submits a new input, debounce the process_input workflow.
# The workflow will wait until 60 seconds after the user stops submitting new inputs,
# then process the last input submitted.
def on_user_input_submit(user_id, user_input):
    debounce_period_sec = 60
    debouncer = Debouncer.create(process_input, debounce_key=user_id)
    debouncer.debounce(debounce_period_sec, user_input)
```

Author: kraftp

.github/workflows/unit-test.yml

@@ -87,6 +87,7 @@ jobs:
     - name: Run Unit Tests
       run: pdm run pytest tests
       working-directory: ./
+      timeout-minutes: 30
       env:
         PGPASSWORD: a!b@c$d()e*_,/:;=?@ff[]22
         GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

dbos/__init__.py

@@ -9,6 +9,7 @@
 )
 from ._dbos import DBOS, DBOSConfiguredInstance, WorkflowHandle, WorkflowHandleAsync
 from ._dbos_config import DBOSConfig
+from ._debouncer import Debouncer, DebouncerClient
 from ._kafka_message import KafkaMessage
 from ._queue import Queue
 from ._sys_db import GetWorkflowsInput, WorkflowStatus, WorkflowStatusString
@@ -32,4 +33,6 @@
     "WorkflowStatusString",
     "error",
     "Queue",
+    "Debouncer",
+    "DebouncerClient",
 ]

dbos/_client.py

@@ -3,6 +3,7 @@
 import time
 import uuid
 from typing import (
+    TYPE_CHECKING,
     Any,
     AsyncGenerator,
     Generator,
@@ -24,7 +25,10 @@
     from typing import NotRequired
 
 from dbos import _serialization
-from dbos._dbos import WorkflowHandle, WorkflowHandleAsync
+
+if TYPE_CHECKING:
+    from dbos._dbos import WorkflowHandle, WorkflowHandleAsync
+
 from dbos._dbos_config import (
     get_application_database_url,
     get_system_database_url,
@@ -224,23 +228,25 @@ def _enqueue(self, options: EnqueueOptions, *args: Any, **kwargs: Any) -> str:
 
     def enqueue(
         self, options: EnqueueOptions, *args: Any, **kwargs: Any
-    ) -> WorkflowHandle[R]:
+    ) -> "WorkflowHandle[R]":
         workflow_id = self._enqueue(options, *args, **kwargs)
         return WorkflowHandleClientPolling[R](workflow_id, self._sys_db)
 
     async def enqueue_async(
         self, options: EnqueueOptions, *args: Any, **kwargs: Any
-    ) -> WorkflowHandleAsync[R]:
+    ) -> "WorkflowHandleAsync[R]":
         workflow_id = await asyncio.to_thread(self._enqueue, options, *args, **kwargs)
         return WorkflowHandleClientAsyncPolling[R](workflow_id, self._sys_db)
 
-    def retrieve_workflow(self, workflow_id: str) -> WorkflowHandle[R]:
+    def retrieve_workflow(self, workflow_id: str) -> "WorkflowHandle[R]":
         status = get_workflow(self._sys_db, workflow_id)
         if status is None:
             raise DBOSNonExistentWorkflowError(workflow_id)
         return WorkflowHandleClientPolling[R](workflow_id, self._sys_db)
 
-    async def retrieve_workflow_async(self, workflow_id: str) -> WorkflowHandleAsync[R]:
+    async def retrieve_workflow_async(
+        self, workflow_id: str
+    ) -> "WorkflowHandleAsync[R]":
         status = await asyncio.to_thread(get_workflow, self._sys_db, workflow_id)
         if status is None:
             raise DBOSNonExistentWorkflowError(workflow_id)
@@ -311,11 +317,13 @@ def cancel_workflow(self, workflow_id: str) -> None:
     async def cancel_workflow_async(self, workflow_id: str) -> None:
         await asyncio.to_thread(self.cancel_workflow, workflow_id)
 
-    def resume_workflow(self, workflow_id: str) -> WorkflowHandle[Any]:
+    def resume_workflow(self, workflow_id: str) -> "WorkflowHandle[Any]":
         self._sys_db.resume_workflow(workflow_id)
         return WorkflowHandleClientPolling[Any](workflow_id, self._sys_db)
 
-    async def resume_workflow_async(self, workflow_id: str) -> WorkflowHandleAsync[Any]:
+    async def resume_workflow_async(
+        self, workflow_id: str
+    ) -> "WorkflowHandleAsync[Any]":
         await asyncio.to_thread(self.resume_workflow, workflow_id)
         return WorkflowHandleClientAsyncPolling[Any](workflow_id, self._sys_db)
 
@@ -451,7 +459,7 @@ def fork_workflow(
         start_step: int,
         *,
         application_version: Optional[str] = None,
-    ) -> WorkflowHandle[Any]:
+    ) -> "WorkflowHandle[Any]":
         forked_workflow_id = fork_workflow(
             self._sys_db,
             self._app_db,
@@ -467,7 +475,7 @@ async def fork_workflow_async(
         start_step: int,
         *,
         application_version: Optional[str] = None,
-    ) -> WorkflowHandleAsync[Any]:
+    ) -> "WorkflowHandleAsync[Any]":
         forked_workflow_id = await asyncio.to_thread(
             fork_workflow,
             self._sys_db,

dbos/_context.py

@@ -524,6 +524,7 @@ def __init__(self, attributes: TracedAttributes) -> None:
         self.saved_workflow_timeout: Optional[int] = None
         self.saved_deduplication_id: Optional[str] = None
         self.saved_priority: Optional[int] = None
+        self.saved_is_within_set_workflow_id_block: bool = False
 
     def __enter__(self) -> DBOSContext:
         # Code to create a basic context
@@ -533,6 +534,9 @@ def __enter__(self) -> DBOSContext:
             ctx = DBOSContext()
             _set_local_dbos_context(ctx)
         assert not ctx.is_within_workflow()
+        # Unset is_within_set_workflow_id_block as the workflow is not within a block
+        self.saved_is_within_set_workflow_id_block = ctx.is_within_set_workflow_id_block
+        ctx.is_within_set_workflow_id_block = False
         # Unset the workflow_timeout_ms context var so it is not applied to this
         # workflow's children (instead we propagate the deadline)
         self.saved_workflow_timeout = ctx.workflow_timeout_ms
@@ -557,6 +561,8 @@ def __exit__(
         ctx = assert_current_dbos_context()
         assert ctx.is_within_workflow()
         ctx.end_workflow(exc_value)
+        # Restore is_within_set_workflow_id_block
+        ctx.is_within_set_workflow_id_block = self.saved_is_within_set_workflow_id_block
         # Restore the saved workflow timeout
         ctx.workflow_timeout_ms = self.saved_workflow_timeout
         # Clear any propagating timeout

dbos/_core.py

@@ -50,6 +50,7 @@
     DBOSException,
     DBOSMaxStepRetriesExceeded,
     DBOSNonExistentWorkflowError,
+    DBOSQueueDeduplicatedError,
     DBOSRecoveryError,
     DBOSUnexpectedStepError,
     DBOSWorkflowCancelledError,
@@ -95,6 +96,7 @@
 F = TypeVar("F", bound=Callable[..., Any])
 
 TEMP_SEND_WF_NAME = "<temp>.temp_send_workflow"
+DEBOUNCER_WORKFLOW_NAME = "_dbos_debouncer_workflow"
 
 
 def check_is_in_coroutine() -> bool:
@@ -310,10 +312,22 @@ def _init_workflow(
     }
 
     # Synchronously record the status and inputs for workflows
-    wf_status, workflow_deadline_epoch_ms = dbos._sys_db.init_workflow(
-        status,
-        max_recovery_attempts=max_recovery_attempts,
-    )
+    try:
+        wf_status, workflow_deadline_epoch_ms = dbos._sys_db.init_workflow(
+            status,
+            max_recovery_attempts=max_recovery_attempts,
+        )
+    except DBOSQueueDeduplicatedError as e:
+        if ctx.has_parent():
+            result: OperationResultInternal = {
+                "workflow_uuid": ctx.parent_workflow_id,
+                "function_id": ctx.parent_workflow_fid,
+                "function_name": wf_name,
+                "output": None,
+                "error": _serialization.serialize_exception(e),
+            }
+            dbos._sys_db.record_operation_result(result)
+        raise
 
     if workflow_deadline_epoch_ms is not None:
         evt = threading.Event()

dbos/_dbos.py

@@ -32,12 +32,14 @@
 from rich import print
 
 from dbos._conductor.conductor import ConductorWebsocket
+from dbos._debouncer import debouncer_workflow
 from dbos._sys_db import SystemDatabase, WorkflowStatus
 from dbos._utils import INTERNAL_QUEUE_NAME, GlobalParams
 from dbos._workflow_commands import fork_workflow, list_queued_workflows, list_workflows
 
 from ._classproperty import classproperty
 from ._core import (
+    DEBOUNCER_WORKFLOW_NAME,
     TEMP_SEND_WF_NAME,
     WorkflowHandleAsyncPolling,
     WorkflowHandlePolling,
@@ -390,11 +392,12 @@ def send_temp_workflow(
         ) -> None:
             self.send(destination_id, message, topic)
 
-        temp_send_wf = workflow_wrapper(self._registry, send_temp_workflow)
-        set_dbos_func_name(send_temp_workflow, TEMP_SEND_WF_NAME)
-        set_dbos_func_name(temp_send_wf, TEMP_SEND_WF_NAME)
-        set_temp_workflow_type(send_temp_workflow, "send")
-        self._registry.register_wf_function(TEMP_SEND_WF_NAME, temp_send_wf, "send")
+        decorate_workflow(self._registry, TEMP_SEND_WF_NAME, None)(send_temp_workflow)
+
+        # Register the debouncer workflow
+        decorate_workflow(self._registry, DEBOUNCER_WORKFLOW_NAME, None)(
+            debouncer_workflow
+        )
 
         for handler in dbos_logger.handlers:
             handler.flush()