fix: take lock in PriorityJobQueue.is_done (#100)

* initial implementation of #100

* fix: TOCTOU race

* fix: addressed reviewer's comments

* fix: code clean-up and fix review comments

Author: bradjin8

.github/workflows/ci.yml

@@ -64,6 +64,7 @@ jobs:
   test:
     name: Test (Python ${{ matrix.python-version }})
     runs-on: ubuntu-latest
+    timeout-minutes: 15
     strategy:
       fail-fast: false
       matrix:

CHANGELOG.md

@@ -21,10 +21,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Default patch pipeline** — near-empty `.localci.yml` uses `patches.profile: generic`. Only `container_mounts`, `image_substitution`, and `codecov_skip` are enabled by default; Capy/B2 steps (`b2_source_cache`, `restore_capy_timestamps`, `capy_copy_preservation`, `b2_bootstrap_skip`) are off unless `profile: capy` or explicitly enabled.
 - **`project` defaults** — `repo_full_name` and `native_image_prefix` no longer default to `cppalliance/capy` and `capy-` on the generic path (empty by default). `profile: capy` restores the previous Boost.Capy values unless overridden.
 - **x86 container architecture** — `linux/386` is requested only when no native image prefix is configured or the image tag does not start with `project.native_image_prefix` (empty prefix no longer suppresses 386 for all images).
+- `PriorityJobQueue.is_done` takes `self._lock` while reading job and completion counts, same as `is_empty`.
 
 ### Fixed
 
-- **`derive_image_tag()` prefix** — honours `project.native_image_prefix` instead of hardcoding `capy-`, completing the generic-profile default behaviour documented for Week 30.
+- **`derive_image_tag()` prefix** honours `project.native_image_prefix` instead of hardcoding `capy-`, completing the generic-profile default behaviour documented for Week 30.
+- `PriorityJobQueue`: jobs in `WAITING_DEPS` return to `QUEUED` when their dependencies finish. Same-priority `needs` chains were leaving dependents stuck.
+- `PriorityJobQueue.cancel_all` drops cancelled keys from `_running_keys`. A `READY` job from `next_ready()` no longer counts as running after cancel.
 
 ## [0.1.0] - TBD
 

cli/localci/core/queue.py

@@ -249,15 +249,15 @@ def mark_completed(self, job: QueuedJob, success: bool = True) -> None:
                 self._failed_keys.add(key)
                 self._completed_keys.add(key)
             self._running_keys.discard(key)
-            self._check_priority_advance()
+            self._after_job_terminal_state_change()
 
     def mark_skipped(self, job: QueuedJob) -> None:
         with self._lock:
             key = job.queue_key
             job.status = QueuedJobStatus.SKIPPED
             self._completed_keys.add(key)
             self._running_keys.discard(key)
-            self._check_priority_advance()
+            self._after_job_terminal_state_change()
 
     def mark_running(self, job: QueuedJob) -> None:
         with self._lock:
@@ -283,7 +283,7 @@ def cancel(self, key: str) -> bool:
             job.status = QueuedJobStatus.CANCELLED
             self._completed_keys.add(key)
             self._emit(JobEventType.JOB_CANCELLED, job)
-            self._check_priority_advance()
+            self._after_job_terminal_state_change()
             return True
 
     def cancel_all(self) -> int:
@@ -298,11 +298,26 @@ def cancel_all(self) -> int:
                 ):
                     job.status = QueuedJobStatus.CANCELLED
                     self._completed_keys.add(key)
+                    self._running_keys.discard(key)
                     count += 1
             if count > 0:
-                self._check_priority_advance()
+                self._after_job_terminal_state_change()
         return count
 
+    def _after_job_terminal_state_change(self) -> None:
+        self._promote_waiting_deps()
+        self._check_priority_advance()
+
+    def _promote_waiting_deps(self) -> None:
+        if self._current_priority is None:
+            return
+        for key in self._by_priority.get(self._current_priority, []):
+            job = self._jobs[key]
+            if job.status != QueuedJobStatus.WAITING_DEPS:
+                continue
+            if self._dep_resolver.all_dependencies_met(key, self._completed_keys):
+                job.status = QueuedJobStatus.QUEUED
+
     def _check_priority_advance(self) -> None:
         if self._current_priority is None:
             return
@@ -351,7 +366,8 @@ def is_empty(self) -> bool:
 
     @property
     def is_done(self) -> bool:
-        return len(self._jobs) > 0 and len(self._completed_keys) >= len(self._jobs)
+        with self._lock:
+            return len(self._jobs) > 0 and len(self._completed_keys) >= len(self._jobs)
 
     @property
     def total_jobs(self) -> int:

cli/tests/test_queue.py

@@ -10,6 +10,7 @@
 from localci.core.models import (
     JobEventType,
     QueuedJob,
+    QueuedJobStatus,
 )
 from localci.core.queue import (
     CyclicDependencyError,
@@ -33,6 +34,19 @@
 
 FIXTURES_DIR = Path(__file__).parent / "fixtures"
 FULL_WORKFLOW = FIXTURES_DIR / "sample_workflow.yml"
+_THREAD_JOIN_TIMEOUT = 30.0
+_CONCURRENT_JOB_COUNT = 100
+_CONCURRENT_BATCH_SIZE = _CONCURRENT_JOB_COUNT // 2
+
+
+def _join_threads(threads: list[threading.Thread], label: str) -> None:
+    for thread in threads:
+        thread.join(timeout=_THREAD_JOIN_TIMEOUT)
+        assert not thread.is_alive(), f"{label} thread did not finish"
+
+
+def _consumer_deadline() -> float:
+    return time.monotonic() + _THREAD_JOIN_TIMEOUT
 
 
 # ---------------------------------------------------------------------------
@@ -186,6 +200,70 @@ def test_same_priority_parallel(self):
         second = queue.next_ready()
         assert second is not None
 
+    def test_waiting_deps_promoted_when_dependency_completes(self):
+        queue = PriorityJobQueue()
+        upstream = make_job("Upstream", priority=1, index=0)
+        downstream = make_job(
+            "Downstream", priority=1, index=1, deps=[upstream.queue_key]
+        )
+        queue.enqueue(upstream)
+        queue.enqueue(downstream)
+
+        first = queue.next_ready()
+        assert first is not None
+        assert first.matrix_entry.name == "Upstream"
+        queue.mark_running(first)
+
+        assert queue.next_ready() is None
+        jobs = {j.queue_key: j for j in queue.get_all_jobs()}
+        assert jobs[downstream.queue_key].status == QueuedJobStatus.WAITING_DEPS
+
+        queue.mark_completed(first, success=True)
+
+        second = queue.next_ready()
+        assert second is not None
+        assert second.matrix_entry.name == "Downstream"
+        queue.mark_running(second)
+        queue.mark_completed(second, success=True)
+        assert queue.is_done is True
+
+    def test_is_done_acquires_lock(self):
+        queue = PriorityJobQueue()
+        queue.enqueue(make_job("Job", priority=1))
+
+        lock_held = threading.Event()
+        release_lock = threading.Event()
+        about_to_read = threading.Event()
+
+        def hold_lock() -> None:
+            with queue._lock:
+                lock_held.set()
+                release_lock.wait(timeout=5)
+
+        holder = threading.Thread(target=hold_lock)
+        holder.start()
+        assert lock_held.wait(timeout=5)
+
+        checker_done = threading.Event()
+        is_done_result: list[bool] = []
+
+        def read_is_done() -> None:
+            about_to_read.set()
+            is_done_result.append(queue.is_done)
+            checker_done.set()
+
+        checker = threading.Thread(target=read_is_done)
+        checker.start()
+        assert about_to_read.wait(timeout=5), "checker never reached is_done read"
+        assert not checker_done.is_set(), "is_done must block on self._lock"
+
+        release_lock.set()
+        checker.join(timeout=5)
+        assert not checker.is_alive()
+        holder.join(timeout=5)
+        assert not holder.is_alive()
+        assert is_done_result == [False]
+
     def test_completion_tracking(self):
         queue = PriorityJobQueue()
         job = make_job("Test", priority=1)
@@ -226,6 +304,18 @@ def test_cancel_all(self):
         assert count == 5
         assert queue.is_done is True
 
+    def test_cancel_all_after_next_ready_clears_running_keys(self):
+        queue = PriorityJobQueue()
+        for i in range(3):
+            queue.enqueue(make_job(f"Job {i}", priority=1, index=i))
+        ready = queue.next_ready()
+        assert ready is not None
+        count = queue.cancel_all()
+        assert count == 3
+        assert queue.running_count == 0
+        assert queue.pending_count >= 0
+        assert queue.is_done is True
+
     def test_event_emission(self):
         events = []
         queue = PriorityJobQueue()
@@ -332,23 +422,70 @@ def enqueue_batch(start: int, count: int) -> None:
                 queue.enqueue(make_job(f"Job {i}", priority=1, index=i))
 
         threads = [
-            threading.Thread(target=enqueue_batch, args=(0, 50)),
-            threading.Thread(target=enqueue_batch, args=(50, 50)),
+            threading.Thread(target=enqueue_batch, args=(0, _CONCURRENT_BATCH_SIZE)),
+            threading.Thread(
+                target=enqueue_batch,
+                args=(_CONCURRENT_BATCH_SIZE, _CONCURRENT_BATCH_SIZE),
+            ),
         ]
         for t in threads:
             t.start()
-        for t in threads:
-            t.join()
-        assert queue.total_jobs == 100
+        _join_threads(threads, "producer")
+        assert queue.total_jobs == _CONCURRENT_JOB_COUNT
+
+    def test_concurrent_producers_and_consumers(self):
+        queue = PriorityJobQueue()
+        producers_done = threading.Event()
+        consumer_deadline = _consumer_deadline()
+
+        def enqueue_batch(start: int, count: int) -> None:
+            for i in range(start, start + count):
+                queue.enqueue(make_job(f"Job {i}", priority=1, index=i))
+                time.sleep(0.001)
+
+        def consume() -> None:
+            while not (producers_done.is_set() and queue.is_done):
+                if time.monotonic() >= consumer_deadline:
+                    return
+                job = queue.next_ready()
+                if job is None:
+                    time.sleep(0.001)
+                    continue
+                queue.mark_running(job)
+                queue.mark_completed(job, success=True)
+
+        producers = [
+            threading.Thread(target=enqueue_batch, args=(0, _CONCURRENT_BATCH_SIZE)),
+            threading.Thread(
+                target=enqueue_batch,
+                args=(_CONCURRENT_BATCH_SIZE, _CONCURRENT_BATCH_SIZE),
+            ),
+        ]
+        consumers = [threading.Thread(target=consume) for _ in range(4)]
+
+        for t in producers:
+            t.start()
+        for t in consumers:
+            t.start()
+
+        _join_threads(producers, "producer")
+        producers_done.set()
+        _join_threads(consumers, "consumer")
+
+        assert queue.total_jobs == _CONCURRENT_JOB_COUNT
+        assert queue.passed_count == _CONCURRENT_JOB_COUNT
 
     def test_concurrent_dequeue(self):
         queue = PriorityJobQueue()
         for i in range(20):
             queue.enqueue(make_job(f"Job {i}", priority=1, index=i))
         results: list[QueuedJob] = []
+        consumer_deadline = _consumer_deadline()
 
         def consume() -> None:
             while not queue.is_done:
+                if time.monotonic() >= consumer_deadline:
+                    return
                 job = queue.next_ready()
                 if job is None:
                     time.sleep(0.01)
@@ -360,8 +497,7 @@ def consume() -> None:
         threads = [threading.Thread(target=consume) for _ in range(4)]
         for t in threads:
             t.start()
-        for t in threads:
-            t.join()
+        _join_threads(threads, "consumer")
         assert len(results) == 20
         names = {r.matrix_entry.name for r in results}
         assert len(names) == 20