Remove Temp Workflows for Direct Calls (#430)

Calling a step directly from outside a workflow now always results in it
being called as a normal Python function. Steps are now only retried if
called from within workflows. Steps can still be enqueued.

Author: kraftp

dbos/_core.py

@@ -1156,27 +1156,15 @@ def check_existing_result() -> Union[NoResult, R]:
 
         @wraps(func)
         def wrapper(*args: Any, **kwargs: Any) -> Any:
-            rr: Optional[str] = check_required_roles(func, fi)
-            # Entering step is allowed:
-            #  No DBOS, just call the original function directly
-            #  In a step already, just call the original function directly.
-            #  In a workflow (that is not in a step already)
-            #  Not in a workflow (we will start the single op workflow)
-            if not dbosreg.dbos or not dbosreg.dbos._launched:
-                # Call the original function directly
-                return func(*args, **kwargs)
+            # If the step is called from a workflow, run it as a step.
+            # Otherwise, run it as a normal function.
             ctx = get_local_dbos_context()
-            if ctx and ctx.is_step():
-                # Call the original function directly
-                return func(*args, **kwargs)
-            if ctx and ctx.is_within_workflow():
-                assert ctx.is_workflow(), "Steps must be called from within workflows"
+            if ctx and ctx.is_workflow():
+                rr: Optional[str] = check_required_roles(func, fi)
                 with DBOSAssumeRole(rr):
                     return invoke_step(*args, **kwargs)
             else:
-                tempwf = dbosreg.workflow_info_map.get("<temp>." + step_name)
-                assert tempwf
-                return tempwf(*args, **kwargs)
+                return func(*args, **kwargs)
 
         wrapper = (
             _mark_coroutine(wrapper) if inspect.iscoroutinefunction(func) else wrapper  # type: ignore

tests/test_async.py

@@ -329,30 +329,6 @@ async def test_async_tx() -> None:
     DBOS.destroy(destroy_registry=True)
 
 
-@pytest.mark.asyncio
-async def test_async_step_temp(dbos: DBOS) -> None:
-    step_counter: int = 0
-
-    @DBOS.step()
-    async def test_step(var: str) -> str:
-        await asyncio.sleep(0.1)
-        nonlocal step_counter
-        step_counter += 1
-        DBOS.logger.info("I'm test_step")
-        return var + f"step{step_counter}"
-
-    wfuuid = f"test_async_step_temp-{time.time_ns()}"
-    with SetWorkflowID(wfuuid):
-        result = await test_step("alice")
-        assert result == "alicestep1"
-
-    with SetWorkflowID(wfuuid):
-        result = await test_step("alice")
-        assert result == "alicestep1"
-
-    assert step_counter == 1
-
-
 @pytest.mark.asyncio
 async def test_start_workflow_async(dbos: DBOS) -> None:
     wf_counter: int = 0

tests/test_classdecorators.py

@@ -574,7 +574,7 @@ def step(self, x: int) -> int:
     def call_step() -> None:
         with SetWorkflowID(wfid):
             nonlocal return_value
-            return_value = inst.step(input)
+            return_value = DBOS.start_workflow(inst.step, input).get_result()
 
     thread = threading.Thread(target=call_step)
     thread.start()

tests/test_concurrency.py

@@ -59,10 +59,6 @@ def test_workflow() -> str:
         res = test_step()
         return res
 
-    def test_comm_thread(id: str) -> str:
-        with SetWorkflowID(id):
-            return test_step()
-
     # Need to set isolation level to a lower one, otherwise it gets serialization error instead (we already handle it correctly by automatic retries).
     @DBOS.transaction(isolation_level="REPEATABLE READ")
     def test_transaction() -> str:
@@ -97,15 +93,6 @@ def test_txn_thread(id: str) -> str:
     assert wf_handle1.get_result() == wfuuid
     assert wf_handle2.get_result() == wfuuid
 
-    # Make sure temp workflows can handle conflicts as well.
-    wfuuid = str(uuid.uuid4())
-    with ThreadPoolExecutor(max_workers=2) as executor:
-        future1 = executor.submit(test_comm_thread, wfuuid)
-        future2 = executor.submit(test_comm_thread, wfuuid)
-
-    assert future1.result() == wfuuid
-    assert future2.result() == wfuuid
-
     # Make sure temp transactions can handle conflicts as well.
     wfuuid = str(uuid.uuid4())
     with ThreadPoolExecutor(max_workers=2) as executor:

tests/test_dbos.py

@@ -307,16 +307,12 @@ def call_step(var: str) -> str:
     assert res == "var"
 
     wfs = dbos._sys_db.get_workflows(gwi)
-    assert len(wfs) == 2
+    assert len(wfs) == 1
 
     wfi1 = dbos._sys_db.get_workflow_status(wfs[0].workflow_id)
     assert wfi1
     assert wfi1["name"].startswith("<temp>")
 
-    wfi2 = dbos._sys_db.get_workflow_status(wfs[1].workflow_id)
-    assert wfi2
-    assert wfi2["name"].startswith("<temp>")
-
     assert txn_counter == 1
     assert step_counter == 1
 
@@ -350,7 +346,7 @@ def test_step(var: str) -> str:
     def test_retried_step(var: str) -> str:
         nonlocal retried_step_counter
         retried_step_counter += 1
-        raise Exception(var)
+        raise ValueError(var)
 
     with pytest.raises(Exception) as exc_info:
         test_transaction("tval")
@@ -360,12 +356,12 @@ def test_retried_step(var: str) -> str:
         test_step("cval")
     assert "cval" == str(exc_info.value)
 
-    with pytest.raises(DBOSMaxStepRetriesExceeded) as exc_info:
+    with pytest.raises(ValueError) as exc_info:
         test_retried_step("rval")
 
     assert txn_counter == 1
     assert step_counter == 1
-    assert retried_step_counter == 3
+    assert retried_step_counter == 1
 
 
 def test_recovery_workflow(dbos: DBOS) -> None:
@@ -1102,9 +1098,6 @@ def test_bad_wf4(var: str) -> str:
     with pytest.raises(Exception) as exc_info:
         test_ns_transaction("h")
     assert "data item should not be a function" in str(exc_info.value)
-    with pytest.raises(Exception) as exc_info:
-        test_ns_step("f")
-    assert "data item should not be a function" in str(exc_info.value)
     with pytest.raises(Exception) as exc_info:
         test_ns_wf("g")
     assert "data item should not be a function" in str(exc_info.value)
@@ -1645,22 +1638,14 @@ def workflow(x: int) -> int:
 async def test_step_without_dbos(dbos: DBOS, config: DBOSConfig) -> None:
     DBOS.destroy(destroy_registry=True)
 
-    is_dbos_active = False
-
     @DBOS.step()
     def step(x: int) -> int:
-        if is_dbos_active:
-            assert DBOS.workflow_id is not None
-        else:
-            assert DBOS.workflow_id is None
+        assert DBOS.workflow_id is None
         return x
 
     @DBOS.step()
     async def async_step(x: int) -> int:
-        if is_dbos_active:
-            assert DBOS.workflow_id is not None
-        else:
-            assert DBOS.workflow_id is None
+        assert DBOS.workflow_id is None
         return x
 
     assert step(5) == 5
@@ -1672,7 +1657,30 @@ async def async_step(x: int) -> int:
     assert await async_step(5) == 5
 
     DBOS.launch()
-    is_dbos_active = True
 
     assert step(5) == 5
     assert await async_step(5) == 5
+
+    assert len(DBOS.list_workflows()) == 0
+
+
+def test_nested_steps(dbos: DBOS) -> None:
+
+    @DBOS.step()
+    def outer_step() -> str:
+        return inner_step()
+
+    @DBOS.step()
+    def inner_step() -> str:
+        id = DBOS.workflow_id
+        assert id is not None
+        return id
+
+    @DBOS.workflow()
+    def workflow() -> str:
+        return outer_step()
+
+    id = workflow()
+    steps = DBOS.list_workflow_steps(id)
+    assert len(steps) == 1
+    assert steps[0]["function_name"] == outer_step.__qualname__

tests/test_failures.py

@@ -333,22 +333,20 @@ def enqueue_failing_step() -> None:
     error_message = f"Step {failing_step.__qualname__} has exceeded its maximum of {max_attempts} retries"
 
     # Test calling the step directly
-    with pytest.raises(DBOSMaxStepRetriesExceeded) as excinfo:
+    with pytest.raises(Exception) as excinfo:
         failing_step()
-    assert error_message in str(excinfo.value)
-    assert step_counter == max_attempts
-    assert len(excinfo.value.errors) == max_attempts
-    for error in excinfo.value.errors:
-        assert isinstance(error, Exception)
-        assert error
-        assert "fail" in str(error)
 
     # Test calling the workflow
     step_counter = 0
     with pytest.raises(DBOSMaxStepRetriesExceeded) as excinfo:
         failing_workflow()
     assert error_message in str(excinfo.value)
     assert step_counter == max_attempts
+    assert len(excinfo.value.errors) == max_attempts
+    for error in excinfo.value.errors:
+        assert isinstance(error, Exception)
+        assert error
+        assert "fail" in str(error)
 
     # Test enqueueing the step
     step_counter = 0
@@ -399,7 +397,6 @@ def failing_workflow() -> None:
 
     assert failing_workflow() == None
     step_counter = 0
-    assert failing_step() == None
 
 
 def test_recovery_during_retries(dbos: DBOS) -> None:

tests/test_queue.py

@@ -181,7 +181,8 @@ def test_step(var: str) -> str:
         handle = queue.enqueue(test_step, "abc")
     assert handle.get_result() == "abc1"
     with SetWorkflowID(wfid):
-        assert test_step("abc") == "abc1"
+        handle = queue.enqueue(test_step, "abc")
+    assert handle.get_result() == "abc1"
     assert step_counter == 1