fix: honor priority at effective admission capacity

src/svrf/daemon.py

@@ -339,10 +339,6 @@ def _discover(self, requested, changed, state: dict, summary: dict, *, skip=froz
                 self.stacked(row, heads, state, summary)
             else:
                 summary["skipped"][str(n)] = decision
-        # A backlog (more candidates than admission slots) is read newest first, and a head whose read failed last
-        # tick after the rest, so old or unread heads never hold fresh admissions behind them; every candidate is
-        # still read this tick. Without a backlog every read starts at once and the order is kept.
-        bounded = bool(self.admission_round_reads) and len(candidates) > self.admission_round_reads
         if self.admission_round_reads:
             # every candidate waits from when it is first seen, and again from each read of it; a head that is no
             # longer a candidate (landed, closed, held) leaves the queue
@@ -351,7 +347,17 @@ def _discover(self, requested, changed, state: dict, summary: dict, *, skip=froz
             for n in current:
                 waiting.setdefault(n, self.clock())
             state["waiting_since"] = waiting
-        if bounded or len(candidates) > max(1, self.admission_jobs):
+        return candidates
+
+    def _order_candidates(self, candidates: list[dict], slots: int, state: dict, summary: dict) -> list[dict]:
+        """Order after observing capacity, before an unstarted head takes a free slot."""
+        free = max(0, slots - len(self.inflight))
+        if not free:
+            return candidates  # no new read can start; the next discovery reads priority again
+        bounded = bool(self.admission_round_reads) and len(candidates) > self.admission_round_reads
+        # A backlog reads declared heads first, then the existing age or retry order.
+        # When every candidate fits, keep the list order without another command.
+        if bounded or len(candidates) > free:
             unread = {int(n) for n in (state.get("last_tick") or {}).get("retry_later") or [] if str(n).isdigit()}
             rank: dict[int, int] = {}
             if self.admission_priority_command:
@@ -485,6 +491,7 @@ def _tick_body(self, requested: set[int] | None = None, changed: dict | None = N
                 max_workers=self._pool_slots, thread_name_prefix="admission"
             )
         slots = min(slots, self._pool_slots)
+        candidates = self._order_candidates(candidates, slots, state, summary)
         candidates = self._owned_candidates(candidates, slots, summary)
         if candidates:
             try:
@@ -566,6 +573,7 @@ def _tick_body(self, requested: set[int] | None = None, changed: dict | None = N
                     summary.pop("deferred", None)
                     more = self._discover(requested, changed, state, summary, skip=set(self.inflight), seen=seen,
                                           refill=True)
+                    more = self._order_candidates(more or [], slots, state, summary)
                     # a declared head whose read failed this tick is not read again until the next tick lists: until
                     # then no undeclared head is taken, so the declared lead the next tick instead of a whole backlog
                     declared = (summary.get("admission_priority") or {}).get("read") or []

tests/test_admission_effective_priority.py

@@ -0,0 +1,102 @@
+"""Declared work leads when current admission capacity leaves only one free slot."""
+
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+from fakes import DaemonGitHub, DaemonRepo, FakeGate
+from test_daemon import daemon
+
+
+class EffectivePriority(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.release = threading.Event()
+        self.calls = []
+        self.changed = threading.Condition()
+        self.owner = None
+        self.addCleanup(self.drain)
+
+    def drain(self):
+        self.release.set()
+        if self.owner is not None:
+            while self.owner.inflight:
+                self.owner.tick(drain=True)
+            self.owner.close()
+
+    def start(self, numbers, **options):
+        self.repo = DaemonRepo(numbers)
+        self.gh = DaemonGitHub(self.repo)
+
+        def admission(head, base):
+            with self.changed:
+                self.calls.append(head)
+                self.changed.notify_all()
+            if not self.release.wait(10):
+                raise RuntimeError('test admission was not released')
+            return {'verdict': 'MERGEABLE', 'residuals': []}
+
+        self.owner = daemon(
+            self.repo, self.gh, FakeGate(self.repo), admission, Path(self.temp.name),
+            jobs=2, admission_jobs=4, admission_round_reads=8, tick_seconds=0.05,
+            admission_priority_command='echo 42', **options,
+        )
+
+    def await_calls(self, count):
+        with self.changed:
+            self.assertTrue(self.changed.wait_for(lambda: len(self.calls) >= count, 2))
+
+    def open_head(self, number):
+        sha = f'h{number}'
+        self.repo.commits[sha] = {'lanes': frozenset({0, number}), 'parents': ('c0',)}
+        self.repo.heads[number] = sha
+        self.repo.branches[f'pr-{number}'] = sha
+
+    def test_declared_head_gets_the_slot_left_by_running_admissions(self):
+        self.start([1, 2, 3])
+        first = self.owner.tick()
+        self.await_calls(3)
+        self.assertEqual(first['inflight'], [1, 2, 3])
+        original = {number: item['future'] for number, item in self.owner.inflight.items()}
+        self.open_head(41)
+        self.open_head(42)
+
+        second = self.owner.tick()
+        self.await_calls(4)
+
+        self.assertEqual(self.calls[3], 'h42')
+        self.assertEqual(second['inflight'], [1, 2, 3, 42])
+        self.assertEqual(second['deferred'], [41])
+        self.assertEqual(second['admission_priority']['read'], [42])
+        for number, future in original.items():
+            self.assertIs(self.owner.inflight[number]['future'], future)
+        self.assertEqual(len(self.calls), 4)
+
+    def test_declared_head_leads_at_a_lower_observed_capacity(self):
+        self.start([41, 42], admission_capacity_command='echo 1')
+
+        out = self.owner.tick()
+        self.await_calls(1)
+
+        self.assertEqual(self.calls, ['h42'])
+        self.assertEqual(out['inflight'], [42])
+        self.assertEqual(out['deferred'], [41])
+        self.assertEqual(out['admission_capacity']['slots'], 1)
+        self.assertEqual(out['admission_priority']['read'], [42])
+
+    def test_a_queue_that_fits_starts_without_a_priority_read(self):
+        self.start([41, 42])
+
+        out = self.owner.tick()
+        self.await_calls(2)
+
+        self.assertCountEqual(self.calls, ['h41', 'h42'])
+        self.assertEqual(out['inflight'], [41, 42])
+        self.assertNotIn('admission_priority', out)
+        self.assertNotIn('deferred', out)
+
+
+if __name__ == '__main__':
+    unittest.main()