Async executor tests: wait for idle after wait_completion() (#210)

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

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()