In-memory AdmissionController (idle/reserved/running, FIFO tickets keyed by
the session cookie, reservation and heartbeat expiry, promotion after
cleanup, EMA wait estimate) behind POST /api/v1/queue/join,
GET /api/v1/queue/status and DELETE /api/v1/queue/ticket. POST /api/v1/query
claims the session's reservation first and joins implicitly when idle.
Settings: CONFLUENCE_WEB_QUEUE_{RESERVATION_SECONDS,HEARTBEAT_SECONDS,MAX_LENGTH}.
297 lines
8.4 KiB
Python
297 lines
8.4 KiB
Python
"""Unit tests for backend.admission.AdmissionController (docs/QUEUE_SPECIFICATION.md)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from backend.admission import AdmissionController
|
|
from backend.errors import QueueFullError, TicketNotFoundError
|
|
|
|
|
|
class FakeClock:
|
|
"""Injectable monotonic clock for deterministic queue tests."""
|
|
|
|
def __init__(self, start: float = 0.0):
|
|
self._now = start
|
|
|
|
def __call__(self) -> float:
|
|
return self._now
|
|
|
|
def advance(self, seconds: float) -> None:
|
|
self._now += seconds
|
|
|
|
|
|
def make_controller(reservation_seconds=45.0, heartbeat_seconds=15.0, max_queue_length=20, clock=None):
|
|
clock = clock or FakeClock(0.0)
|
|
return AdmissionController(
|
|
reservation_seconds=reservation_seconds,
|
|
heartbeat_seconds=heartbeat_seconds,
|
|
max_queue_length=max_queue_length,
|
|
clock=clock,
|
|
), clock
|
|
|
|
|
|
def run_cycle(ctl: AdmissionController, session_id: str, duration: float, cancelled: bool = False) -> None:
|
|
"""Join (idle), claim, and end a run for session_id. Assumes the runner is idle."""
|
|
view = ctl.join(session_id)
|
|
assert view["status"] == "ready"
|
|
assert ctl.claim(session_id) is True
|
|
ctl.run_end(session_id, duration_seconds=duration, cancelled=cancelled)
|
|
|
|
|
|
# -- Join / idle / concurrent queueing --------------------------------------
|
|
|
|
|
|
def test_join_while_idle_reserves_directly():
|
|
ctl, clock = make_controller(reservation_seconds=45.0)
|
|
view = ctl.join("s1")
|
|
assert view["status"] == "ready"
|
|
assert view["position"] == 1
|
|
assert view["ahead"] == 0
|
|
assert view["runner"] == "reserved"
|
|
assert view["reservation_expires_in_seconds"] == 45
|
|
assert view["eta_seconds"] is None
|
|
assert view["ticket_id"].startswith("q_")
|
|
|
|
|
|
def test_concurrent_second_join_is_queued():
|
|
ctl, clock = make_controller()
|
|
ctl.join("s1")
|
|
view2 = ctl.join("s2")
|
|
assert view2["status"] == "queued"
|
|
assert view2["position"] == 2
|
|
assert view2["ahead"] == 1
|
|
assert view2["runner"] == "reserved"
|
|
assert view2["reservation_expires_in_seconds"] is None
|
|
|
|
|
|
def test_join_is_idempotent_per_session():
|
|
ctl, clock = make_controller()
|
|
ctl.join("s1")
|
|
view_a = ctl.join("s2")
|
|
view_b = ctl.join("s2")
|
|
assert view_a["ticket_id"] == view_b["ticket_id"]
|
|
assert view_b["status"] == "queued"
|
|
|
|
|
|
# -- FIFO order and promotion ------------------------------------------------
|
|
|
|
|
|
def test_fifo_order_and_promotion_on_completion():
|
|
ctl, clock = make_controller()
|
|
ctl.join("a")
|
|
ctl.join("b")
|
|
ctl.join("c")
|
|
|
|
assert ctl.claim("a") is True
|
|
# "a" is now running, not reserved; it does not occupy a front-list slot.
|
|
assert ctl.status("b")["position"] == 1
|
|
assert ctl.status("c")["position"] == 2
|
|
|
|
ctl.run_end("a", duration_seconds=10.0, cancelled=False)
|
|
|
|
status_b = ctl.status("b")
|
|
assert status_b["status"] == "ready"
|
|
assert status_b["position"] == 1
|
|
assert status_b["ahead"] == 0
|
|
|
|
status_c = ctl.status("c")
|
|
assert status_c["status"] == "queued"
|
|
assert status_c["position"] == 2
|
|
assert status_c["ahead"] == 1
|
|
|
|
with pytest.raises(TicketNotFoundError):
|
|
ctl.status("a")
|
|
|
|
|
|
def test_promotion_on_failure_and_timeout_paths():
|
|
# run_end is called identically for success, failure, and timeout by app.py;
|
|
# the controller has no notion of outcome beyond "cancelled".
|
|
ctl, clock = make_controller()
|
|
ctl.join("a")
|
|
ctl.join("b")
|
|
ctl.claim("a")
|
|
ctl.run_end("a", duration_seconds=5.0, cancelled=False)
|
|
assert ctl.status("b")["status"] == "ready"
|
|
|
|
|
|
def test_promotion_on_cancellation():
|
|
ctl, clock = make_controller()
|
|
ctl.join("a")
|
|
ctl.join("b")
|
|
ctl.claim("a")
|
|
ctl.run_end("a", duration_seconds=1.0, cancelled=True)
|
|
assert ctl.status("b")["status"] == "ready"
|
|
|
|
|
|
def test_promotion_on_leave():
|
|
ctl, clock = make_controller()
|
|
ctl.join("a")
|
|
ctl.join("b")
|
|
ctl.leave("a")
|
|
assert ctl.status("b")["status"] == "ready"
|
|
|
|
|
|
def test_leave_is_idempotent():
|
|
ctl, clock = make_controller()
|
|
ctl.leave("never-joined") # no error
|
|
ctl.join("a")
|
|
ctl.leave("a")
|
|
with pytest.raises(TicketNotFoundError):
|
|
ctl.status("a")
|
|
ctl.leave("a") # still no error
|
|
|
|
|
|
# -- Reservation and heartbeat expiry ----------------------------------------
|
|
|
|
|
|
def test_reservation_expiry_drops_holder_and_promotes_next():
|
|
clock = FakeClock(0.0)
|
|
ctl, _ = make_controller(reservation_seconds=30.0, heartbeat_seconds=15.0, clock=clock)
|
|
ctl.join("holder")
|
|
ctl.join("waiter")
|
|
|
|
clock.advance(31.0)
|
|
|
|
# The dropped session's query attempt gets refused.
|
|
assert ctl.claim("holder") is False
|
|
with pytest.raises(TicketNotFoundError):
|
|
ctl.status("holder")
|
|
|
|
status_w = ctl.status("waiter")
|
|
assert status_w["status"] == "ready"
|
|
|
|
|
|
def test_heartbeat_expiry_drops_only_queued_tickets():
|
|
clock = FakeClock(0.0)
|
|
ctl, _ = make_controller(reservation_seconds=100.0, heartbeat_seconds=10.0, clock=clock)
|
|
ctl.join("ready_holder")
|
|
ctl.join("queued_one")
|
|
|
|
clock.advance(11.0)
|
|
|
|
# Triggers lazy expiry; the ready ticket's reservation window (100s) has not elapsed.
|
|
view = ctl.status("ready_holder")
|
|
assert view["status"] == "ready"
|
|
|
|
with pytest.raises(TicketNotFoundError):
|
|
ctl.status("queued_one")
|
|
|
|
|
|
def test_heartbeat_poll_keeps_queued_ticket_alive():
|
|
clock = FakeClock(0.0)
|
|
ctl, _ = make_controller(reservation_seconds=100.0, heartbeat_seconds=10.0, clock=clock)
|
|
ctl.join("ready_holder")
|
|
ctl.join("queued_one")
|
|
|
|
clock.advance(6.0)
|
|
ctl.status("queued_one") # refresh heartbeat
|
|
clock.advance(6.0) # 6s since refresh, under the 10s window
|
|
|
|
view = ctl.status("queued_one")
|
|
assert view["status"] == "queued"
|
|
|
|
|
|
# -- Capacity -----------------------------------------------------------------
|
|
|
|
|
|
def test_queue_full_at_max_length_excludes_reserved():
|
|
ctl, clock = make_controller(max_queue_length=2)
|
|
ctl.join("reserved_session") # does not count against max_length
|
|
ctl.join("q1")
|
|
ctl.join("q2")
|
|
with pytest.raises(QueueFullError):
|
|
ctl.join("q3")
|
|
|
|
|
|
# -- Estimate -------------------------------------------------------------------
|
|
|
|
|
|
def test_eta_null_before_history():
|
|
ctl, clock = make_controller()
|
|
view = ctl.join("s1")
|
|
assert view["eta_seconds"] is None
|
|
|
|
|
|
def test_eta_ema_update_over_multiple_runs():
|
|
ctl, clock = make_controller()
|
|
|
|
run_cycle(ctl, "w1", 10.0)
|
|
probe = ctl.join("probe1")
|
|
assert probe["eta_seconds"] == 10
|
|
ctl.leave("probe1")
|
|
|
|
run_cycle(ctl, "w2", 20.0)
|
|
probe = ctl.join("probe2")
|
|
# EMA: 0.3*20 + 0.7*10 = 13
|
|
assert probe["eta_seconds"] == 13
|
|
ctl.leave("probe2")
|
|
|
|
run_cycle(ctl, "w3", 30.0)
|
|
probe = ctl.join("probe3")
|
|
# EMA: 0.3*30 + 0.7*13 = 18.1 -> rounds to 18
|
|
assert probe["eta_seconds"] == 18
|
|
|
|
|
|
def test_ahead_counting_excludes_running_session():
|
|
ctl, clock = make_controller()
|
|
ctl.join("running_session")
|
|
ctl.claim("running_session")
|
|
|
|
view = ctl.join("q1")
|
|
assert view["status"] == "queued"
|
|
assert view["position"] == 1
|
|
assert view["ahead"] == 0
|
|
assert view["runner"] == "running"
|
|
|
|
|
|
def test_eta_clamps_at_zero_when_elapsed_exceeds_estimate():
|
|
clock = FakeClock(0.0)
|
|
ctl, _ = make_controller(clock=clock)
|
|
|
|
run_cycle(ctl, "w1", 10.0) # seeds mean = 10
|
|
|
|
ctl.join("w2")
|
|
ctl.claim("w2")
|
|
clock.advance(50.0) # far beyond the 10s estimate
|
|
|
|
view = ctl.status("w2")
|
|
assert view["eta_seconds"] == 0
|
|
|
|
|
|
def test_cancelled_runs_under_five_seconds_do_not_update_estimate():
|
|
ctl, clock = make_controller()
|
|
run_cycle(ctl, "w1", 3.0, cancelled=True)
|
|
view = ctl.join("probe")
|
|
assert view["eta_seconds"] is None
|
|
|
|
|
|
def test_cancelled_runs_at_or_over_five_seconds_update_estimate():
|
|
ctl, clock = make_controller()
|
|
run_cycle(ctl, "w1", 7.0, cancelled=True)
|
|
view = ctl.join("probe")
|
|
assert view["eta_seconds"] == 7
|
|
|
|
|
|
# -- Claim admission table ----------------------------------------------------
|
|
|
|
|
|
def test_claim_refused_for_other_session_while_reserved():
|
|
ctl, clock = make_controller()
|
|
ctl.join("holder")
|
|
assert ctl.claim("intruder") is False
|
|
# Queue/ticket state untouched by the refusal.
|
|
assert ctl.status("holder")["status"] == "ready"
|
|
|
|
|
|
def test_claim_refused_while_running():
|
|
ctl, clock = make_controller()
|
|
ctl.join("holder")
|
|
ctl.claim("holder")
|
|
assert ctl.claim("someone_else") is False
|
|
|
|
|
|
def test_implicit_join_admits_while_idle():
|
|
ctl, clock = make_controller()
|
|
assert ctl.claim("walk_up") is True
|