fix: handle condition waiting better in thread_task_executor

The pattern used for the condition waiting was not robust enough
and was susceptible to a race condition.

This prevents that class of race condition by persisting the status
and checking it when the signal is received, rather then relying
on the assumption the notification always happens after all the
relevant waiters are ready to receive it.

JIRA: CQ-2677
risk: low

Author: no23reason

packages/gooddata-flight-server/src/gooddata_flight_server/tasks/thread_task_executor.py

@@ -158,6 +158,7 @@ class _TaskExecution:
         "_result_future",
         "_lock",
         "_completed",
+        "_done",
         "_stats",
     )
 
@@ -189,6 +190,8 @@ def __init__(
         # all these are protected using the lock
         self._result_future: Future[Union[TaskResult, TaskError]] | None = None
         self._completed: threading.Condition = threading.Condition(self._lock)
+        # indicates the task actually finished
+        self._done: bool = False
 
     @property
     def task(self) -> Task:
@@ -235,6 +238,7 @@ def on_result_done(self, fut: Future) -> None:
 
         with self._lock:
             execution_result = self._cb.process_task_result(self, self._result_future)
+            self._done = True
             self._completed.notify_all()
 
         self._complete_execution_span(execution_result)
@@ -291,7 +295,7 @@ def cancel(self) -> bool:
 
     def wait_for_completion(self, timeout: float | None = None) -> None:
         with self._lock:
-            completed = self._completed.wait(timeout=timeout)
+            completed = self._completed.wait_for(lambda: self._done, timeout=timeout)
 
         if not completed:
             raise TaskWaitTimeoutError(task_id=self._task.task_id, cmd=self._task.cmd)