fix: guard wait() against transient FAILED during saga startup

py_src/taskito/workflows/run.py

@@ -68,6 +68,27 @@ def wait(self, timeout: float | None = None, poll_interval: float = 0.1) -> Work
             while True:
                 snapshot = self.status()
                 if snapshot.state.is_terminal():
+                    # FAILED and COMPLETED_WITH_FAILURES can transition to
+                    # COMPENSATING when a saga is about to start.  Between
+                    # mark_workflow_node_result (sets FAILED in Rust) and
+                    # start_compensation (sets COMPENSATING from Python)
+                    # there is a window where the poll sees the transient
+                    # terminal state.  When the tracker owns an event for
+                    # this run, defer to the event — it is set only after
+                    # the saga decision is made — instead of returning the
+                    # possibly-transient state.
+                    if (
+                        snapshot.state
+                        in (WorkflowState.FAILED, WorkflowState.COMPLETED_WITH_FAILURES)
+                        and event is not None
+                        and not event.is_set()
+                    ):
+                        grace = poll_interval
+                        if deadline is not None:
+                            grace = min(grace, max(0.0, deadline - time.monotonic()))
+                        if grace > 0:
+                            event.wait(timeout=grace)
+                            continue
                     return snapshot
 
                 remaining: float

py_src/taskito/workflows/saga/orchestrator.py

@@ -601,11 +601,21 @@ def _record_node_outcome(
         )
 
     def _mark_run_compensating(self, run_id: str) -> None:
-        """Move the run from ``Running`` to ``Compensating`` in storage."""
-        try:
-            self._queue._inner.set_workflow_run_compensating(run_id)
-        except Exception:
-            logger.exception("saga: failed to mark run %s compensating", run_id)
+        """Move the run to ``Compensating`` in storage.
+
+        Retries briefly on transient SQLite "database is locked" errors
+        so that contention from a concurrent result-handler write does
+        not silently leave the run in ``Failed``.
+        """
+        for attempt in range(5):
+            try:
+                self._queue._inner.set_workflow_run_compensating(run_id)
+                return
+            except Exception as exc:
+                if "database is locked" not in str(exc) or attempt == 4:
+                    logger.exception("saga: failed to mark run %s compensating", run_id)
+                    return
+                time.sleep(0.05 * (attempt + 1))
 
     def _finalize_locked(self, run_id: str, *, succeeded: bool) -> None:
         """Cleanly finalize a saga. Must be called with ``self._lock``."""