From 0c2ef4c2c6e639952b3e4f0caf90ac1cbf23f31a Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Wed, 27 May 2026 10:28:48 +0530
Subject: [PATCH] =?UTF-8?q?fix:=20close=20FAILED=E2=86=92COMPENSATING=20ra?=
 =?UTF-8?q?ce=20in=20saga=20startup?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Move _mark_run_compensating earlier in start_compensation — right
after the cheap in-memory eligibility check instead of after the
storage-heavy _fetch_compensable_nodes call. On Windows with SQLite
under contention, the storage I/O gap let wait()'s poll see the
transient FAILED state as terminal.
---
 py_src/taskito/workflows/saga/orchestrator.py | 35 ++++++++++++++-----
 1 file changed, 26 insertions(+), 9 deletions(-)

diff --git a/py_src/taskito/workflows/saga/orchestrator.py b/py_src/taskito/workflows/saga/orchestrator.py
index 21ea8eae..d6ef8aa1 100644
--- a/py_src/taskito/workflows/saga/orchestrator.py
+++ b/py_src/taskito/workflows/saga/orchestrator.py
@@ -155,26 +155,33 @@ def start_compensation(
             if not (has_explicit or has_subwf_propagation):
                 return False
 
+            # Transition to COMPENSATING immediately after the cheap
+            # eligibility check. The storage call in _fetch_compensable_nodes
+            # can be slow (especially SQLite on Windows under contention), and
+            # wait()'s poll loop would see the transient FAILED state as
+            # terminal if we delay this transition.
+            self._mark_run_compensating(run_id)
+
+            self._queue._emit_event(
+                EventType.WORKFLOW_COMPENSATING,
+                {"workflow_run_id": run_id},
+            )
+
             completed = self._fetch_compensable_nodes(run_id, run_config)
             if not completed:
-                return False
+                self._finalize_locked(run_id, succeeded=True)
+                return True
 
             waves = _reverse_topo_waves(steps, restrict_to=completed)
             if not waves:
-                return False
+                self._finalize_locked(run_id, succeeded=True)
+                return True
 
             self._run_waves[run_id] = waves
             self._run_inflight[run_id] = set()
             self._run_any_failed[run_id] = False
             self._run_depth[run_id] = depth
 
-            self._mark_run_compensating(run_id)
-
-            self._queue._emit_event(
-                EventType.WORKFLOW_COMPENSATING,
-                {"workflow_run_id": run_id},
-            )
-
         self._dispatch_next_wave(run_id, run_config)
         return True
 
@@ -502,6 +509,16 @@ def _propagate_to_child_saga(self, parent_run_id: str, parent_node_name: str) ->
                         parent_run_id,
                         parent_node_name,
                     )
+                    # If start_compensation finalized immediately (vacuously
+                    # compensated — no nodes to undo), the child saga already
+                    # completed but _pending_parent_after_child was set after
+                    # the fact. Detect and resolve the parent node inline.
+                    already_done = (
+                        child_run_id not in self._run_waves
+                        and child_run_id not in self._run_inflight
+                    )
+                if already_done:
+                    self._on_child_saga_terminal(child_run_id, True, None)
                 propagated_any = True
 
         return propagated_any