From 6751e39d5d26d02544800658971296d3b7aa299a Mon Sep 17 00:00:00 2001 From: nybarius <76703223+nybarius@users.noreply.github.com> Date: Tue, 20 Oct 2026 20:04:35 -0400 Subject: [PATCH 1/2] test: require priority at the effective free admission width Reproduce a declared head losing the sole free slot while three original reads run, and at a live capacity below the configured width. Preserve the control that a queue fitting all free slots starts without a priority read. --- tests/test_admission_effective_priority.py | 102 +++++++++++++++++++++ 1 file changed, 102 insertions(+) create mode 100644 tests/test_admission_effective_priority.py diff --git a/tests/test_admission_effective_priority.py b/tests/test_admission_effective_priority.py new file mode 100644 index 0000000..e1f2065 --- /dev/null +++ b/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() From 0d13516d5751b14c30e9c162b859d5946df8199c Mon Sep 17 00:00:00 2001 From: nybarius <76703223+nybarius@users.noreply.github.com> Date: Tue, 20 Oct 2026 20:10:16 -0400 Subject: [PATCH 2/2] fix: order admission against currently free slots Apply the existing declared and fallback ordering after the live capacity and pool width are known. Subtract running admissions before deciding whether waiting candidates compete for slots, and do the same before the existing refill failure barrier. Preserve complete snapshot ownership in one-shot mode and avoid a priority command when no new slot can be taken. --- src/svrf/daemon.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/src/svrf/daemon.py b/src/svrf/daemon.py index b2e96af..daa78f0 100644 --- a/src/svrf/daemon.py +++ b/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 []