diff --git a/src/arcadedb/async_executor.py b/src/arcadedb/async_executor.py index 98267362..8377b95e 100644 --- a/src/arcadedb/async_executor.py +++ b/src/arcadedb/async_executor.py @@ -826,6 +826,12 @@ def wait_completion(self, timeout_ms: Optional[int] = None): argument gives: work a running task schedules after the snapshot is not covered by a successful poll. + Returning from ``wait_completion()`` means every operation queued before + the call has run. It does not make a following ``is_pending()`` False at + once: since engine 26.11.1-SNAPSHOT a worker can still report itself as + executing for an instant after the wait is released. To wait for an idle + executor, poll ``is_pending()`` until it is False. + Raises: TimeoutError: If timeout is reached before completion ValueError: If timeout_ms is negative diff --git a/tests/test_async_executor.py b/tests/test_async_executor.py index 8b4b5d30..cff0b6d0 100644 --- a/tests/test_async_executor.py +++ b/tests/test_async_executor.py @@ -13,6 +13,23 @@ LOSS_REPRO_ROWS = 9_742 +def _wait_until_idle(async_exec, timeout=30.0): + """Wait, with a deadline, for ``is_processing()`` to turn False. + + ``wait_completion()`` returns once every task queued before the call has run. It does + not promise that the executor is quiet at that instant: from engine 26.11.1-SNAPSHOT + (ArcadeData/arcadedb#11191) a worker still flags itself as executing for a moment after + the last marker has released the waiter, so ``is_processing()`` asked right after the + return can answer True once, with nothing left to run (ArcadeData/arcadedb-python#210). + Asserting "idle now" straight after ``wait_completion()`` therefore races; this waits + for the documented condition instead. + """ + deadline = time.monotonic() + timeout + while async_exec.is_processing(): + assert time.monotonic() < deadline, "the executor never went idle" + time.sleep(0.001) + + def test_async_executor_sql_command_insert_is_exact_at_parallel_one(tmp_path): """Every command submitted at parallel level 1 becomes a row. @@ -198,6 +215,7 @@ def hold(_result): release.set() async_exec.wait_completion() + _wait_until_idle(async_exec) assert not async_exec.is_pending() assert not async_exec.is_processing() assert db.count_type("Msg") == 1 @@ -266,6 +284,7 @@ def test_async_executor_is_pending_true_while_queued(temp_db): ] async_exec.wait_completion() + _wait_until_idle(async_exec) assert async_exec.is_pending() is False async_exec.close() @@ -435,6 +454,7 @@ def test_async_executor_wait_completion_zero_returns_when_idle(temp_db): async_exec = temp_db.async_executor() try: async_exec.wait_completion() + _wait_until_idle(async_exec) assert async_exec.wait_completion(0) is None finally: @@ -535,6 +555,7 @@ def test_wait_completion_commits_everything_queued_including_the_tail(temp_db): ex.create_record(doc, callback=callbacks.append) ex.wait_completion() + _wait_until_idle(ex) assert not ex.is_pending() assert not ex.is_processing()