Update Debouncer Tests (#460)

Author: kraftp

dbos/_debouncer.py

@@ -232,7 +232,10 @@ def assign_debounce_ids() -> tuple[str, str]:
         while True:
             try:
                 # Attempt to enqueue a debouncer for this workflow.
-                with SetEnqueueOptions(deduplication_id=self.debounce_key):
+                deduplication_id = (
+                    f"{self.options['workflow_name']}-{self.debounce_key}"
+                )
+                with SetEnqueueOptions(deduplication_id=deduplication_id):
                     with SetWorkflowTimeout(None):
                         internal_queue.enqueue(
                             debouncer_workflow,
@@ -249,7 +252,7 @@ def assign_debounce_ids() -> tuple[str, str]:
                 def get_deduplicated_workflow() -> Optional[str]:
                     return dbos._sys_db.get_deduplicated_workflow(
                         queue_name=internal_queue.name,
-                        deduplication_id=self.debounce_key,
+                        deduplication_id=deduplication_id,
                     )
 
                 dedup_wfid = dbos._sys_db.call_function_as_step(
@@ -333,10 +336,13 @@ def debounce(
         while True:
             try:
                 # Attempt to enqueue a debouncer for this workflow.
+                deduplication_id = (
+                    f"{self.debouncer_options['workflow_name']}-{self.debounce_key}"
+                )
                 debouncer_options: EnqueueOptions = {
                     "workflow_name": DEBOUNCER_WORKFLOW_NAME,
                     "queue_name": INTERNAL_QUEUE_NAME,
-                    "deduplication_id": self.debounce_key,
+                    "deduplication_id": deduplication_id,
                 }
                 self.client.enqueue(
                     debouncer_options,
@@ -353,7 +359,7 @@ def debounce(
                 # If there is already a debouncer, send a message to it.
                 dedup_wfid = self.client._sys_db.get_deduplicated_workflow(
                     queue_name=INTERNAL_QUEUE_NAME,
-                    deduplication_id=self.debounce_key,
+                    deduplication_id=deduplication_id,
                 )
                 if dedup_wfid is None:
                     continue

tests/test_debouncer.py

@@ -17,35 +17,33 @@
 from dbos._utils import GlobalParams
 
 
-def workflow(x: int) -> int:
-    return x
-
-
-async def workflow_async(x: int) -> int:
-    return x
-
-
 def test_debouncer(dbos: DBOS) -> None:
 
-    DBOS.workflow()(workflow)
+    @DBOS.workflow()
+    def workflow(x: int) -> int:
+        return x
+
     first_value, second_value, third_value, fourth_value = 0, 1, 2, 3
 
     @DBOS.step()
     def generate_uuid() -> str:
         return str(uuid.uuid4())
 
     def debouncer_test() -> None:
-        debouncer = Debouncer.create(workflow, debounce_key="key")
 
         debounce_period = 2
 
+        debouncer = Debouncer.create(workflow, debounce_key="key")
         first_handle = debouncer.debounce(debounce_period, first_value)
+        debouncer = Debouncer.create(workflow, debounce_key="key")
         second_handle = debouncer.debounce(debounce_period, second_value)
         assert first_handle.workflow_id == second_handle.workflow_id
         assert first_handle.get_result() == second_value
         assert second_handle.get_result() == second_value
 
+        debouncer = Debouncer.create(workflow, debounce_key="key")
         third_handle = debouncer.debounce(debounce_period, third_value)
+        debouncer = Debouncer.create(workflow, debounce_key="key")
         fourth_handle = debouncer.debounce(debounce_period, fourth_value)
         assert third_handle.workflow_id != first_handle.workflow_id
         assert third_handle.workflow_id == fourth_handle.workflow_id
@@ -76,7 +74,10 @@ def debouncer_test() -> None:
 
 def test_debouncer_timeout(dbos: DBOS) -> None:
 
-    DBOS.workflow()(workflow)
+    @DBOS.workflow()
+    def workflow(x: int) -> int:
+        return x
+
     first_value, second_value, third_value, fourth_value = 0, 1, 2, 3
 
     # Set a huge period but small timeout, verify workflows start after the timeout
@@ -115,9 +116,38 @@ def test_debouncer_timeout(dbos: DBOS) -> None:
     assert second_handle.get_result() == second_value
 
 
+def test_multiple_debouncers(dbos: DBOS) -> None:
+
+    @DBOS.workflow()
+    def workflow(x: int) -> int:
+        return x
+
+    first_value, second_value, third_value, fourth_value = 0, 1, 2, 3
+
+    # Set a huge period but small timeout, verify workflows start after the timeout
+    debouncer_one = Debouncer.create(workflow, debounce_key="key_one")
+    debouncer_two = Debouncer.create(workflow, debounce_key="key_two")
+    debounce_period = 2
+
+    first_handle = debouncer_one.debounce(debounce_period, first_value)
+    second_handle = debouncer_one.debounce(debounce_period, second_value)
+    third_handle = debouncer_two.debounce(debounce_period, third_value)
+    fourth_handle = debouncer_two.debounce(debounce_period, fourth_value)
+    assert first_handle.workflow_id == second_handle.workflow_id
+    assert first_handle.workflow_id != third_handle.workflow_id
+    assert third_handle.workflow_id == fourth_handle.workflow_id
+    assert first_handle.get_result() == second_value
+    assert second_handle.get_result() == second_value
+    assert third_handle.get_result() == fourth_value
+    assert fourth_handle.get_result() == fourth_value
+
+
 def test_debouncer_queue(dbos: DBOS) -> None:
 
-    DBOS.workflow()(workflow)
+    @DBOS.workflow()
+    def workflow(x: int) -> int:
+        return x
+
     first_value, second_value, third_value, fourth_value = 0, 1, 2, 3
     queue = Queue("test-queue")
 
@@ -166,7 +196,10 @@ def test_debouncer_queue(dbos: DBOS) -> None:
 @pytest.mark.asyncio
 async def test_debouncer_async(dbos: DBOS) -> None:
 
-    DBOS.workflow()(workflow_async)
+    @DBOS.workflow()
+    async def workflow_async(x: int) -> int:
+        return x
+
     first_value, second_value, third_value, fourth_value = 0, 1, 2, 3
 
     debouncer = Debouncer.create_async(workflow_async, debounce_key="key")
@@ -188,12 +221,15 @@ async def test_debouncer_async(dbos: DBOS) -> None:
 
 def test_debouncer_client(dbos: DBOS, client: DBOSClient) -> None:
 
-    DBOS.workflow()(workflow)
+    @DBOS.workflow()
+    def workflow(x: int) -> int:
+        return x
+
     first_value, second_value, third_value, fourth_value = 0, 1, 2, 3
     queue = Queue("test-queue")
 
     options: EnqueueOptions = {
-        "workflow_name": workflow.__name__,
+        "workflow_name": workflow.__qualname__,
         "queue_name": queue.name,
     }
     debouncer = DebouncerClient(client, options, debounce_key="key")
@@ -230,12 +266,15 @@ def test_debouncer_client(dbos: DBOS, client: DBOSClient) -> None:
 @pytest.mark.asyncio
 async def test_debouncer_client_async(dbos: DBOS, client: DBOSClient) -> None:
 
-    DBOS.workflow()(workflow_async)
+    @DBOS.workflow()
+    async def workflow_async(x: int) -> int:
+        return x
+
     first_value, second_value, third_value, fourth_value = 0, 1, 2, 3
     queue = Queue("test-queue")
 
     options: EnqueueOptions = {
-        "workflow_name": workflow_async.__name__,
+        "workflow_name": workflow_async.__qualname__,
         "queue_name": queue.name,
     }
     debouncer = DebouncerClient(client, options, debounce_key="key")