diff --git a/dispatch/job.py b/dispatch/job.py index f02312b..2520f75 100644 --- a/dispatch/job.py +++ b/dispatch/job.py @@ -1,27 +1,66 @@ -"""Job lifecycle: enqueue / claim / renew / reclaim (spec §7.2, §7.3, §8, §9). +"""Job lifecycle: enqueue / claim / renew / reclaim / complete / abort / JE land +(spec §7.2–§7.5, §8, §9). This is the Python skin over the Lua state transitions in ``scripts.py``. Every transition is atomic in Redis; this layer only picks the ``now`` clock (Python ``time.time()``, passed into every script via ARGV — never a server-side TIME), -walks ``jobs:leased`` for the time-gated orphan scan, and shapes payloads. - -Slice 2 of the delivery plan (§15.2) — dark: nothing calls these yet. The -``complete``/``abort``/JE-landing transitions are slice 3; presigned ``code_url`` -signing on the payload is slice 4. Both are called out at their seams below. +walks ``jobs:leased`` for the time-gated orphan scan, holds the per-submission +lock around a landing, and shapes payloads. + +Slices 2 and 3 of the delivery plan (§15.2) — dark: nothing calls these yet. +The Mongo side is not wired either: result/JE landing is done through injected +callables so this layer stays payload- and storage-agnostic (the keystone slice +closes over ``process_result`` / a JE marker). The remaining seam is the +presigned ``code_url`` signing on the payload plus HTTP status mapping, both +slice 4; they are called out at their seams below. """ +import contextlib +import enum +import logging +import threading import time -from typing import Dict, Optional +from typing import Callable, Dict, Optional from ulid import ULID +import redis.exceptions from mongo.utils import RedisCache from . import params from . import redis_keys from . import scripts +logger = logging.getLogger(__name__) + JOB_ID_PREFIX = 'jb_' + +class CompleteOutcome(enum.Enum): + """Result of a complete attempt; slice 4 maps these to HTTP status. + + LANDED / STALE_DROPPED → 204 (a stale drop MUST be logged — INV4); + NOT_OWNER → 409; GONE → 404; BUSY / LAND_FAILED → retryable 5xx. + """ + LANDED = enum.auto() + STALE_DROPPED = enum.auto() + NOT_OWNER = enum.auto() + GONE = enum.auto() + BUSY = enum.auto() + LAND_FAILED = enum.auto() + + +class AbortOutcome(enum.Enum): + """Result of an abort attempt; slice 4 maps these to HTTP status. + + REQUEUED / STALE_DROPPED / CONVERGED → 202; NOT_OWNER → 409; GONE → 404. + """ + REQUEUED = enum.auto() + STALE_DROPPED = enum.auto() + CONVERGED = enum.auto() + NOT_OWNER = enum.auto() + GONE = enum.auto() + + # One shared RedisCache instance, mirroring runner.py: under fakeredis each # RedisCache owns an isolated dataset, so enqueue/claim/renew must share one to # stay coherent; in production they all share the pooled real Redis anyway. @@ -45,6 +84,85 @@ def _decode(value) -> Optional[str]: return value.decode() if isinstance(value, bytes) else value +@contextlib.contextmanager +def _submission_lock(client, submission_id: str): + """Non-blocking per-submission landing lock (INV3); yields the acquired flag. + + Landing callers must bail out on False rather than wait: complete answers + BUSY (the runner retries), the JE sweep leaves the job for the next window. + A Mongo landing has no completion bound, so a fixed TTL alone cannot hold + INV3 — a watchdog thread extends the lock every TTL/3 for as long as the + landing runs, and the TTL only bounds crash recovery. The watchdog is + stopped and joined before release so an extend can never race the release. + """ + lock = client.lock( + redis_keys.submission_job_lock(submission_id), + timeout=params.JOB_LOCK_TTL_SEC, + blocking=False, + # The token must live on the Lock instance, not in thread-local + # storage: the watchdog extends from its own thread, and each Lock + # object is private to one landing call anyway. + thread_local=False, + ) + acquired = lock.acquire() + if not acquired: + yield False + return + stop = threading.Event() + watchdog = threading.Thread( + target=_keep_lock_alive, + args=(lock, stop, submission_id), + daemon=True, + ) + watchdog.start() + try: + yield True + finally: + stop.set() + watchdog.join() + try: + lock.release() + except redis.exceptions.LockError: + pass + + +def _keep_lock_alive(lock, stop: threading.Event, submission_id: str) -> None: + """Extend the landing lock until told to stop (the INV3 watchdog). + + An extend that fails means exclusivity is already lost (the key evaporated — + Redis data loss, or a pathological stall past the TTL): nothing mid-flight + can be safely undone, so log loudly and let delete-after-write plus the + landing contract (re-run tolerance, spec §17.2) absorb the fallout. Closing + this residual for real takes a fencing token / conditional write on the + Mongo side — a pre-keystone task, out of this layer's reach. + """ + while not stop.wait(params.JOB_LOCK_TTL_SEC / 3.0): + try: + lock.extend(params.JOB_LOCK_TTL_SEC, replace_ttl=True) + except redis.exceptions.LockError: + logger.error('landing lock lost mid-write: submission %s', + submission_id) + return + + +def _sid_or_reap(client, job_id: str) -> Optional[str]: + """Read a job's submission_id (the lock name), reaping unidentifiable jobs. + + A vanished hash leaves at most a ghost index entry; a hash that exists but + lost submission_id can never pass a currency check again and the hash has no + TTL — destroy both (same policy as the Lua side) so neither lingers + unindexed forever. + """ + sid = _decode(client.hget(redis_keys.job(job_id), 'submission_id')) + if sid is not None: + return sid + pipe = client.pipeline() + pipe.delete(redis_keys.job(job_id)) + pipe.srem(redis_keys.JOBS_LEASED, job_id) + pipe.execute() + return None + + def enqueue_job( submission_id: str, problem_id: str, @@ -91,17 +209,24 @@ def enqueue_job( return job_id -def claim_next_job(runner_id: str) -> Optional[Dict]: +def claim_next_job( + runner_id: str, + land_je: Optional[Callable[[str], None]] = None, +) -> Optional[Dict]: """Hand ``runner_id`` its next job, or None when there is nothing to do. Order matches spec §7.3: first a time-gated orphan scan (at most one runner per ``ORPHAN_SCAN_INTERVAL_SEC`` window), whose first successful reclaim goes straight to this caller; otherwise fall through to the pending queue. + + ``land_je`` is the injected JE marker (keystone wires it to Mongo); the scan + uses it to converge attempts-exhausted orphans. None keeps the scan dark — + exhausted orphans are left in place for a later, wired call to sweep. """ now = _now() client = _redis() - reclaimed = _orphan_scan(client, runner_id, now) + reclaimed = _orphan_scan(client, runner_id, now, land_je) if reclaimed is not None: return _payload(client, reclaimed) @@ -128,14 +253,132 @@ def renew_lease(runner_id: str, job_id: str) -> bool: return result == 1 -def _orphan_scan(client, runner_id: str, now: float) -> Optional[str]: +def complete_job( + runner_id: str, + job_id: str, + land_result: Callable[[str], None], +) -> CompleteOutcome: + """Land ``runner_id``'s result for ``job_id`` (spec §7.4; INV1/INV3/INV4). + + ``land_result(submission_id)`` performs the Mongo landing and raises on + failure (the keystone closes over the request's tasks payload; this layer + stays payload-agnostic). The whole landing runs under the per-submission lock + (INV3): mark completing, land, then delete-after-write. A lease expiring + during a slow write stays deliberately reclaimable — the lock plus the + delete-after-write ordering, not lease extension, is what bounds double + writes (spec §9). + + Contract: this layer guarantees at-least-once invocation, never + exactly-once. ``land_result`` must therefore tolerate re-invocation both as + a full re-run after a reclaim and as a retry after its own partial write + (LAND_FAILED → the runner resends). Verifying/fixing ``process_result`` for + this is the keystone's pre-wiring task (spec §17.2, ADR-0003 Consequences). + """ + client = _redis() + scr = scripts.load(client) + + sid = _sid_or_reap(client, job_id) + if sid is None: + return CompleteOutcome.GONE + + with _submission_lock(client, sid) as acquired: + if not acquired: + return CompleteOutcome.BUSY + begin = scr.complete_begin( + keys=[redis_keys.JOBS_LEASED, + redis_keys.job(job_id)], + args=[job_id, runner_id], + ) + if begin == -2: + return CompleteOutcome.GONE + if begin == 0: + return CompleteOutcome.NOT_OWNER + if begin == -1: + logger.warning( + 'stale result dropped: job %s submission %s runner %s', job_id, + sid, runner_id) + return CompleteOutcome.STALE_DROPPED + + try: + land_result(sid) + except Exception as exc: + logger.exception('result landing failed: job %s submission %s', + job_id, sid) + scr.complete_restore( + keys=[redis_keys.job(job_id)], + args=[runner_id, str(exc)[:200]], + ) + return CompleteOutcome.LAND_FAILED + + scr.complete_finish( + keys=[redis_keys.JOBS_LEASED, + redis_keys.job(job_id)], + args=[job_id], + ) + return CompleteOutcome.LANDED + + +def abort_job( + runner_id: str, + job_id: str, + reason: str, + land_je: Optional[Callable[[str], None]] = None, +) -> AbortOutcome: + """Requeue or converge ``job_id`` on a runner-reported abort (spec §7.5, INV5). + + ``drain`` never counts against attempts; ``prep_failed`` / ``rejected`` count + toward the poison ceiling. When counting tips the job to MAX_ATTEMPTS it + converges to JE: the abort is accepted regardless of whether the JE landing + succeeds — a failed landing leaves the job in place for the orphan-scan sweep + to retry (INV1). + """ + if reason not in ('drain', 'prep_failed', 'rejected'): + raise ValueError(f'invalid abort reason: {reason!r}') + + client = _redis() + + sid = _sid_or_reap(client, job_id) + if sid is None: + return AbortOutcome.GONE + + result = scripts.load(client).abort_requeue( + keys=[ + redis_keys.JOBS_LEASED, + redis_keys.JOBS_PENDING, + redis_keys.job(job_id), + ], + args=[job_id, runner_id, reason, params.MAX_ATTEMPTS], + ) + if result == 1: + return AbortOutcome.REQUEUED + if result == -1: + logger.warning('stale abort dropped: job %s submission %s runner %s', + job_id, sid, runner_id) + return AbortOutcome.STALE_DROPPED + if result == -2: + return AbortOutcome.GONE + if result == 0: + return AbortOutcome.NOT_OWNER + + _try_land_je(sid, job_id, land_je) + return AbortOutcome.CONVERGED + + +def _orphan_scan( + client, + runner_id: str, + now: float, + land_je: Optional[Callable[[str], None]] = None, +) -> Optional[str]: """Time-gated sweep of ``jobs:leased``; return an id reclaimed for the caller. The ``SET NX EX`` gate on ``dispatch:last_recovery`` lets at most one scan run per window (spec §7.3 step 1). Within a winning scan, each expired candidate is offered to ``reclaim_expired``: the first ``1`` (reclaimed to this caller) - is returned immediately; ``-1`` (attempts exhausted) candidates are skipped - and left in place — the JE landing sweep that removes them is slice 3. + is returned immediately; a ``-1`` (attempts exhausted) candidate is converged + via ``_try_land_je`` and the scan keeps looking for reclaimable work for the + caller. Iteration is over a materialized snapshot, so a landing that unindexes + a member mid-scan is harmless. """ acquired = client.set( redis_keys.DISPATCH_LAST_RECOVERY, @@ -162,9 +405,62 @@ def _orphan_scan(client, runner_id: str, now: float) -> Optional[str]: ) if result == 1: return job_id + if result == -1: + sid = _decode(client.hget(redis_keys.job(job_id), 'submission_id')) + if sid is not None: + _try_land_je(sid, job_id, land_je) return None +def _try_land_je( + submission_id: str, + job_id: str, + land_je: Optional[Callable[[str], None]], +) -> bool: + """Land a JE marker for an attempts-exhausted job, or leave it for a retry. + + Runs under the same per-submission lock as ``complete_job``. If a rejudge has + moved current_job past this job it is a superseded orphan: destroy it (pointer + untouched). Otherwise land the JE; on landing failure the job is left in place + — still in ``jobs:leased`` with an expired lease, so the next time-gated scan + retries the landing (INV1). ``land_je`` None (not wired) touches nothing. + """ + if land_je is None: + logger.debug('JE landing skipped (not wired): job %s submission %s', + job_id, submission_id) + return False + + client = _redis() + with _submission_lock(client, submission_id) as acquired: + if not acquired: + return False + current = _decode( + client.get(redis_keys.submission_current_job(submission_id))) + if current != job_id: + pipe = client.pipeline() + pipe.delete(redis_keys.job(job_id)) + pipe.srem(redis_keys.JOBS_LEASED, job_id) + pipe.execute() + logger.warning('stale JE job destroyed: job %s submission %s', + job_id, submission_id) + return True + + try: + land_je(submission_id) + except Exception: + logger.exception('JE landing failed: job %s submission %s', job_id, + submission_id) + return False + + scripts.load(client).complete_finish( + keys=[redis_keys.JOBS_LEASED, + redis_keys.job(job_id)], + args=[job_id], + ) + logger.info('JE landed: job %s submission %s', job_id, submission_id) + return True + + def _payload(client, job_id: str) -> Optional[Dict]: """Shape a job hash into a claim payload (spec §7.3), or None if it vanished. diff --git a/dispatch/params.py b/dispatch/params.py index a61f66b..87153c6 100644 --- a/dispatch/params.py +++ b/dispatch/params.py @@ -6,6 +6,11 @@ HEARTBEAT_INTERVAL_SEC = 15 LEASE_TTL_SEC = 30 +# Per-submission landing lock (INV3). Auto-extended by a watchdog while a +# landing is in flight — a Mongo landing has no completion bound, so the TTL +# only bounds crash recovery (how long a dead holder blocks the submission), +# never the write duration. +JOB_LOCK_TTL_SEC = 60 POLL_INTERVAL_SEC = 3 ORPHAN_SCAN_INTERVAL_SEC = 15 MAX_ATTEMPTS = 3 diff --git a/dispatch/scripts.py b/dispatch/scripts.py index a6a5c04..3612f31 100644 --- a/dispatch/scripts.py +++ b/dispatch/scripts.py @@ -5,17 +5,19 @@ Every script takes ``now`` via ARGV — never ``redis.call('TIME')`` — so tests are deterministic and fakeredis (which has no server clock for scripts) works. -The scripts here cover slice 2 of the delivery plan (§15.2): ``claim_pending``, -``renew_lease``, ``reclaim_expired``. ``abort_requeue`` belongs to slice 3 and -is deliberately absent. +The scripts here cover slices 2 and 3 of the delivery plan (§15.2): +``claim_pending``, ``renew_lease``, ``reclaim_expired`` (slice 2) and +``complete_begin``, ``complete_finish``, ``complete_restore``, ``abort_requeue`` +(slice 3 — result landing / abort / JE convergence). Key handling: static keys (``jobs:pending``, ``jobs:leased``, a specific ``job:`` hash) are passed in as KEYS by the Python layer via redis_keys. -``claim_pending`` is the sole exception — it discovers a job id by RPOP and so -must build that job's ``job:`` hash key and its -``submission::current_job`` pointer key from inside Lua; those two literal -templates mirror ``redis_keys.job`` / ``redis_keys.submission_current_job`` and -are the only place key names are spelled outside redis_keys. +The ``submission::current_job`` pointer key is the exception — several +scripts learn ``sid`` only after reading it out of the job hash and so must +build that pointer key from inside Lua (``claim_pending`` discovers the job id +by RPOP and builds both keys the same way). Those literal templates mirror +``redis_keys.job`` / ``redis_keys.submission_current_job`` and are the only +place key names are spelled outside redis_keys. Corrupted state (partial Redis data loss — never produced by normal operation) is reaped lazily, in passing: the job hash is the single source of truth and @@ -148,12 +150,165 @@ return 1 ''' +# complete_begin — ownership + currency gate, then mark completing (INV3/INV4). +# +# KEYS[1] = jobs:leased (SET) KEYS[2] = job: (HASH) +# ARGV[1] = job_id ARGV[2] = runner_id +# +# Entry gate of the single-writer landing (INV3). Ownership is checked before +# currency so a reclaimed-away or never-owned job answers 409, not a stale drop. +# Currency (INV4, result side) destroys the job hash and unindexes it but never +# touches the pointer — the pointer belongs to whoever is current now. A hash +# that lost its submission_id is corrupted beyond a currency check and is treated +# as a stale drop (same policy as claim_pending). The lease_deadline is left +# alone on purpose: a lease expiring during a slow Mongo write is deliberately +# reclaimable (spec §9), and the per-submission lock + delete-after-write is what +# bounds double writes — not lease extension. Returns 1 proceed / 0 not-owner / +# -1 stale-or-corrupt destroyed / -2 gone. +COMPLETE_BEGIN = ''' +if redis.call('EXISTS', KEYS[2]) == 0 then + redis.call('SREM', KEYS[1], ARGV[1]) + return -2 +end +if redis.call('HGET', KEYS[2], 'leased_by') ~= ARGV[2] then + return 0 +end +local sid = redis.call('HGET', KEYS[2], 'submission_id') +if not sid then + redis.call('DEL', KEYS[2]) + redis.call('SREM', KEYS[1], ARGV[1]) + return -1 +end +if redis.call('GET', 'submission:' .. sid .. ':current_job') ~= ARGV[1] then + redis.call('DEL', KEYS[2]) + redis.call('SREM', KEYS[1], ARGV[1]) + return -1 +end +redis.call('HSET', KEYS[2], 'state', 'completing') +return 1 +''' + +# complete_finish — post-Mongo-success cleanup, the sole INV1 exit for a result. +# +# KEYS[1] = jobs:leased (SET) KEYS[2] = job: (HASH) +# ARGV[1] = job_id +# +# The result has landed, so the job is done: DEL the hash and SREM it +# unconditionally (a reclaim that re-leased it mid-write does not matter — the +# re-run runner's later complete answers -2/404 and drops). The pointer is +# deleted only if it still points at this job (compare-and-delete): a rejudge may +# have advanced current_job to a newer job while we wrote, and that pointer is +# not ours to clear. Returns 1, or 0 if the hash was already gone (still SREMs). +COMPLETE_FINISH = ''' +local sid = redis.call('HGET', KEYS[2], 'submission_id') +local existed = redis.call('DEL', KEYS[2]) +redis.call('SREM', KEYS[1], ARGV[1]) +if sid and redis.call('GET', 'submission:' .. sid .. ':current_job') == ARGV[1] then + redis.call('DEL', 'submission:' .. sid .. ':current_job') +end +if existed == 1 then + return 1 +end +return 0 +''' + +# complete_restore — undo the completing mark after a failed Mongo write. +# +# KEYS[1] = job: (HASH) +# ARGV[1] = runner_id ARGV[2] = last_error (pre-truncated to <=200 chars) +# +# Lets the same runner's retry re-enter complete_begin. Only acts when the hash +# still exists, is owned by the same runner, and is still in completing: if a +# reclaim already handed the job to a new owner, leased_by differs and restore +# must NOT stomp the new owner's state. lease_deadline is untouched. Anything +# else → 0, no write. +COMPLETE_RESTORE = ''' +if redis.call('EXISTS', KEYS[1]) == 0 then + return 0 +end +if redis.call('HGET', KEYS[1], 'leased_by') ~= ARGV[1] then + return 0 +end +if redis.call('HGET', KEYS[1], 'state') ~= 'completing' then + return 0 +end +redis.call('HSET', KEYS[1], 'state', 'leased', 'last_error', ARGV[2]) +return 1 +''' + +# abort_requeue — clear lease, requeue at tail, count by reason (spec §7.5, INV5). +# +# KEYS[1] = jobs:leased (SET) KEYS[2] = jobs:pending (LIST) KEYS[3] = job: +# ARGV[1] = job_id ARGV[2] = runner_id ARGV[3] = reason +# ARGV[4] = max_attempts +# +# Ownership then currency, same as complete_begin: evaporated hash → -2 (404), +# wrong owner → 0 (409), lost submission_id / superseded job → destroy + -1 +# (requeueing a stale job is pointless; the pointer is never touched). ``drain`` +# must not cost the job an attempt (rolling restarts must not shorten a job's +# life, spec §7.5): claim counts on every claim, but a drained execution never +# ran to a verdict, so the claim+drain round-trip must net zero — the drain +# branch REFUNDS one attempt (floored at 0) as it resets to pending and LPUSHes +# onto the queue tail (same side as enqueue; RPOP serves the head); otherwise a +# rolling restart would erode the retry budget. +# ``prep_failed`` / ``rejected`` both count: attempts+1, and at the ceiling the +# job is left IN jobs:leased with an immediately-expired lease (leased_by +# unchanged) — the exact orphan shape reclaim_expired answers -1 to, so a JE +# landing sweep can still find it if the caller's inline landing fails (INV1). +# Below the ceiling it requeues like drain. Reason is validated in Python; the +# script assumes one of the three. Returns 1 requeued / 2 converged / -1 stale / +# -2 gone / 0 not-owner. +ABORT_REQUEUE = ''' +if redis.call('EXISTS', KEYS[3]) == 0 then + redis.call('SREM', KEYS[1], ARGV[1]) + return -2 +end +if redis.call('HGET', KEYS[3], 'leased_by') ~= ARGV[2] then + return 0 +end +local sid = redis.call('HGET', KEYS[3], 'submission_id') +if not sid then + redis.call('DEL', KEYS[3]) + redis.call('SREM', KEYS[1], ARGV[1]) + return -1 +end +if redis.call('GET', 'submission:' .. sid .. ':current_job') ~= ARGV[1] then + redis.call('DEL', KEYS[3]) + redis.call('SREM', KEYS[1], ARGV[1]) + return -1 +end +if ARGV[3] == 'drain' then + local refunded = math.max(0, (tonumber(redis.call('HGET', KEYS[3], 'attempts')) or 0) - 1) + redis.call('HSET', KEYS[3], + 'state', 'pending', 'leased_by', '', 'lease_deadline', 0, + 'attempts', refunded, 'last_error', 'abort:drain') + redis.call('SREM', KEYS[1], ARGV[1]) + redis.call('LPUSH', KEYS[2], ARGV[1]) + return 1 +end +local attempts = (tonumber(redis.call('HGET', KEYS[3], 'attempts')) or 0) + 1 +redis.call('HSET', KEYS[3], 'attempts', attempts, 'last_error', 'abort:' .. ARGV[3]) +if attempts >= tonumber(ARGV[4]) then + redis.call('HSET', KEYS[3], 'lease_deadline', 0) + return 2 +end +redis.call('HSET', KEYS[3], + 'state', 'pending', 'leased_by', '', 'lease_deadline', 0) +redis.call('SREM', KEYS[1], ARGV[1]) +redis.call('LPUSH', KEYS[2], ARGV[1]) +return 1 +''' + @dataclass(frozen=True) class Scripts: claim_pending: Any renew_lease: Any reclaim_expired: Any + complete_begin: Any + complete_finish: Any + complete_restore: Any + abort_requeue: Any def load(client) -> Scripts: @@ -167,4 +322,8 @@ def load(client) -> Scripts: claim_pending=client.register_script(CLAIM_PENDING), renew_lease=client.register_script(RENEW_LEASE), reclaim_expired=client.register_script(RECLAIM_EXPIRED), + complete_begin=client.register_script(COMPLETE_BEGIN), + complete_finish=client.register_script(COMPLETE_FINISH), + complete_restore=client.register_script(COMPLETE_RESTORE), + abort_requeue=client.register_script(ABORT_REQUEUE), ) diff --git a/tests/test_dispatch_landing.py b/tests/test_dispatch_landing.py new file mode 100644 index 0000000..bac7c21 --- /dev/null +++ b/tests/test_dispatch_landing.py @@ -0,0 +1,674 @@ +import logging +import threading +import time + +import pytest + +from config import settings +from mongo.utils import RedisCache +from dispatch import params, redis_keys, scripts +from dispatch import job as job_mod +from dispatch.job import CompleteOutcome, AbortOutcome + + +@pytest.fixture(autouse=True, scope='session') +def setup_minio(): + # Shadow conftest's Docker/MinIO session fixture: these landing unit tests + # touch only fakeredis, so they must not require a container engine. + yield + + +@pytest.fixture(autouse=True) +def fresh_fakeredis(monkeypatch): + # Force each test onto a fresh FakeStrictRedis: RedisCache caches its + # connection pool on the class and dispatch caches one RedisCache instance. + # REDIS_PORT must be None in settings so RedisCache falls back to fakeredis. + monkeypatch.setattr(settings, 'REDIS_HOST', None) + monkeypatch.setattr(settings, 'REDIS_PORT', None) + RedisCache.POOL = None + job_mod._cache = None + yield + RedisCache.POOL = None + job_mod._cache = None + + +def _enqueue(submission_id='sn_1', problem_id='pb_1', language=2): + return job_mod.enqueue_job( + submission_id=submission_id, + problem_id=problem_id, + language=language, + code_minio_path=f'{submission_id}/main.py', + checker='diff', + tasks_meta_json='[]', + ) + + +def _client(): + return job_mod._redis() + + +def _hash(job_id): + raw = _client().hgetall(redis_keys.job(job_id)) + return {k.decode(): v.decode() for k, v in raw.items()} + + +def _reclaim(job_id, runner, now): + return scripts.load(_client()).reclaim_expired( + keys=[redis_keys.JOBS_LEASED, + redis_keys.job(job_id)], + args=[job_id, runner, now, params.LEASE_TTL_SEC, params.MAX_ATTEMPTS], + ) + + +def _expire(job_id, deadline='0'): + _client().hset(redis_keys.job(job_id), 'lease_deadline', deadline) + + +def _leased(job_id): + return _client().sismember(redis_keys.JOBS_LEASED, job_id) + + +def _pending(): + return [ + m.decode() for m in _client().lrange(redis_keys.JOBS_PENDING, 0, -1) + ] + + +def _current(submission_id): + v = _client().get(redis_keys.submission_current_job(submission_id)) + return v.decode() if v is not None else None + + +class Writer: + """Injected landing callable; records the sids it lands. + + ``side_effect`` runs mid-write (to simulate a concurrent reclaim / rejudge); + ``raises`` fires afterwards to model a Mongo write failure; ``sleep`` widens + the write window so a racing caller is guaranteed to collide on the lock. + """ + + def __init__(self, side_effect=None, raises=None, sleep=0.0): + self.calls = [] + self._side_effect = side_effect + self._raises = raises + self._sleep = sleep + + def __call__(self, submission_id): + self.calls.append(submission_id) + if self._sleep: + time.sleep(self._sleep) + if self._side_effect is not None: + self._side_effect() + if self._raises is not None: + raise self._raises + + +# --- INV3 complete: happy path ------------------------------------------ + + +def test_complete_happy_path_lands_once_and_cleans_up(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + + writer = Writer() + assert job_mod.complete_job('rn_a', job_id, + writer) == CompleteOutcome.LANDED + + assert writer.calls == ['sn_1'] # landed exactly once, with the sid + assert _client().exists(redis_keys.job(job_id)) == 0 + assert not _leased(job_id) + assert _current('sn_1') is None + + +def test_second_complete_after_landing_is_gone(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + + first = Writer() + assert job_mod.complete_job('rn_a', job_id, + first) == CompleteOutcome.LANDED + + second = Writer() + assert job_mod.complete_job('rn_a', job_id, second) == CompleteOutcome.GONE + assert second.calls == [] + + +def test_concurrent_completes_exactly_one_lands(monkeypatch): + # Two completes race under the per-submission lock; the writer sleeps so the + # window is wide. Exactly one LANDED, the loser BUSY (non-blocking acquire + # failed) or GONE (winner finished first); the writer runs exactly once. + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + + writer = Writer(sleep=0.2) + barrier = threading.Barrier(2) + results = {} + + def racer(name): + barrier.wait() + results[name] = job_mod.complete_job('rn_a', job_id, writer) + + threads = [ + threading.Thread(target=racer, args=(n, )) for n in ('t1', 't2') + ] + for t in threads: + t.start() + for t in threads: + t.join() + + landed = [o for o in results.values() if o == CompleteOutcome.LANDED] + assert len(landed) == 1 + loser = next(o for o in results.values() if o != CompleteOutcome.LANDED) + assert loser in (CompleteOutcome.BUSY, CompleteOutcome.GONE) + assert len(writer.calls) == 1 + + +def test_complete_lock_pre_held_returns_busy(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + + _client().set(redis_keys.submission_job_lock('sn_1'), + 'held') # someone owns it + + writer = Writer() + assert job_mod.complete_job('rn_a', job_id, writer) == CompleteOutcome.BUSY + assert writer.calls == [] + assert _hash(job_id)['state'] == 'leased' # untouched + + +def test_complete_lock_renewal_outlives_slow_write(monkeypatch): + # A Mongo landing has no completion bound, so a fixed TTL alone cannot hold + # INV3: once the lock expired mid-write a rival complete could acquire it + # and land the same result twice. The watchdog must keep extending the lock + # for the whole write — every rival attempt during the write stays locked + # out, and exactly one landing happens overall. + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + monkeypatch.setattr(params, 'JOB_LOCK_TTL_SEC', 0.3) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + + writer = Writer(sleep=1.0) # spans several lock TTLs + started = threading.Event() + outcome = {} + + def slow_complete(): + started.set() + outcome['first'] = job_mod.complete_job('rn_a', job_id, writer) + + t = threading.Thread(target=slow_complete) + t.start() + started.wait() + time.sleep(0.05) # let the slow caller take the lock + + rival_writers, rivals = [], [] + while t.is_alive(): + w = Writer() + rival_writers.append(w) + rivals.append(job_mod.complete_job('rn_a', job_id, w)) + time.sleep(0.05) + t.join() + + assert outcome['first'] == CompleteOutcome.LANDED + assert writer.calls == ['sn_1'] # landed exactly once + # rivals during the write are BUSY; one racing the cleanup may see GONE + assert set(rivals) <= {CompleteOutcome.BUSY, CompleteOutcome.GONE} + assert all(w.calls == [] for w in rival_writers) + + +def test_complete_reclaim_during_write_still_single_writer(monkeypatch): + # The lease expires and the job is reclaimed to a new runner mid-write, then + # the write succeeds: the result still lands and the job is destroyed. The + # new runner's later complete finds nothing — exactly one landing overall. + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + + def reclaim_away(): + _expire(job_id) + assert _reclaim(job_id, 'rn_new', now=5000.0) == 1 + + writer = Writer(side_effect=reclaim_away) + assert job_mod.complete_job('rn_a', job_id, + writer) == CompleteOutcome.LANDED + assert _client().exists(redis_keys.job(job_id)) == 0 + + rerun = Writer() + assert job_mod.complete_job('rn_new', job_id, + rerun) == CompleteOutcome.GONE + assert rerun.calls == [] + + +# --- INV4 currency ------------------------------------------------------ + + +def test_complete_stale_is_dropped_not_landed(monkeypatch, caplog): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job1 = _enqueue(submission_id='sn_x') + job_mod.claim_next_job('rn_a') + job2 = _enqueue(submission_id='sn_x') # rejudge moves the pointer + + writer = Writer() + with caplog.at_level(logging.WARNING): + outcome = job_mod.complete_job('rn_a', job1, writer) + + assert outcome == CompleteOutcome.STALE_DROPPED + assert writer.calls == [] + assert _client().exists(redis_keys.job(job1)) == 0 # destroyed + assert not _leased(job1) + assert _current('sn_x') == job2 # pointer untouched + assert 'stale result dropped' in caplog.text + assert job1 in caplog.text + + +def test_complete_rejudge_during_write_does_not_delete_new_pointer( + monkeypatch): + # A rejudge fires while Mongo is written: current_job advances to a new job. + # The write succeeds, the old job is deleted, but the compare-and-delete must + # leave the NEW pointer (and its job) intact. + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job1 = _enqueue(submission_id='sn_x') + job_mod.claim_next_job('rn_a') + + new_job = {} + + def rejudge(): + new_job['id'] = _enqueue(submission_id='sn_x') + + writer = Writer(side_effect=rejudge) + assert job_mod.complete_job('rn_a', job1, writer) == CompleteOutcome.LANDED + + job2 = new_job['id'] + assert _client().exists(redis_keys.job(job1)) == 0 + assert _current('sn_x') == job2 # new pointer survived + assert _client().exists(redis_keys.job(job2)) == 1 + assert job2 in _pending() + + +# --- INV1 complete failure ---------------------------------------------- + + +def test_complete_land_failure_restores_and_retries(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + + failing = Writer(raises=RuntimeError('mongo down')) + assert job_mod.complete_job('rn_a', job_id, + failing) == CompleteOutcome.LAND_FAILED + + h = _hash(job_id) + assert h['state'] == 'leased' # restored + assert h['last_error'] == 'mongo down' + assert _leased(job_id) + assert _current('sn_1') == job_id # pointer intact + + ok = Writer() + assert job_mod.complete_job('rn_a', job_id, ok) == CompleteOutcome.LANDED + assert ok.calls == ['sn_1'] + assert _client().exists(redis_keys.job(job_id)) == 0 + assert not _leased(job_id) + assert _current('sn_1') is None + + +def test_complete_retry_after_partial_write_reinvokes_in_full(monkeypatch): + # The landing contract is at-least-once, never exactly-once: a writer that + # dies after partially applying its effects is re-invoked IN FULL on the + # runner's retry — the duplicate side effect is real, and absorbing it is + # the mongo side's obligation (spec §17.2), pinned here at the seam. + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + + calls, side_effects = [], [] + + def flaky_writer(sid): + calls.append(sid) + side_effects.append('user_counter_bumped') # applied before the crash + if len(calls) == 1: + raise RuntimeError('save failed after partial apply') + + assert job_mod.complete_job('rn_a', job_id, + flaky_writer) == CompleteOutcome.LAND_FAILED + assert _hash(job_id)['state'] == 'leased' # INV1: restored, retryable + + assert job_mod.complete_job('rn_a', job_id, + flaky_writer) == CompleteOutcome.LANDED + assert calls == ['sn_1', 'sn_1'] # full re-invocation, not resumption + assert len(side_effects) == 2 # duplicated — mongo must tolerate this + assert _client().exists(redis_keys.job(job_id)) == 0 + + +def test_complete_restore_refuses_to_stomp_new_owner(monkeypatch): + # The lease expires and the job is reclaimed to rn_new before the write + # raises: restore must NOT overwrite the new owner's state. + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') # attempts = 1 + + def reclaim_then_boom(): + _expire(job_id) + assert _reclaim(job_id, 'rn_new', now=5000.0) == 1 # attempts = 2 + + failing = Writer(side_effect=reclaim_then_boom, + raises=RuntimeError('boom')) + assert job_mod.complete_job('rn_a', job_id, + failing) == CompleteOutcome.LAND_FAILED + + h = _hash(job_id) + assert h['leased_by'] == 'rn_new' # restore left the new owner alone + assert h['state'] == 'leased' + assert h['attempts'] == '2' # the reclaim bump stands + + +# --- ownership / existence ---------------------------------------------- + + +def test_complete_non_owner_is_refused(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + + writer = Writer() + assert job_mod.complete_job('rn_b', job_id, + writer) == CompleteOutcome.NOT_OWNER + assert writer.calls == [] + assert _hash(job_id)['leased_by'] == 'rn_a' # untouched + + +def test_complete_unknown_job_is_gone_and_resurrects_nothing(): + writer = Writer() + assert job_mod.complete_job('rn_a', 'jb_missing', + writer) == CompleteOutcome.GONE + assert writer.calls == [] + assert _client().exists(redis_keys.job('jb_missing')) == 0 + assert not _leased('jb_missing') + + +def test_complete_corrupt_hash_without_sid_is_reaped(monkeypatch): + # A leased hash that lost submission_id can never pass a currency check and + # has no TTL: the pre-lock read must destroy it (hash AND index), not just + # unindex it and leave the hash to linger forever. + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + _client().hdel(redis_keys.job(job_id), 'submission_id') + + writer = Writer() + assert job_mod.complete_job('rn_a', job_id, writer) == CompleteOutcome.GONE + assert writer.calls == [] + assert _client().exists(redis_keys.job(job_id)) == 0 + assert not _leased(job_id) + + +def test_abort_corrupt_hash_without_sid_is_reaped(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + _client().hdel(redis_keys.job(job_id), 'submission_id') + + assert job_mod.abort_job('rn_a', job_id, 'drain') == AbortOutcome.GONE + assert _client().exists(redis_keys.job(job_id)) == 0 + assert not _leased(job_id) + assert _pending() == [] # reaped, not requeued + + +# --- abort matrix ------------------------------------------------------- + + +def test_abort_drain_requeues_without_counting(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') # attempts = 1 + assert _hash(job_id)['attempts'] == '1' + + assert job_mod.abort_job('rn_a', job_id, 'drain') == AbortOutcome.REQUEUED + + h = _hash(job_id) + # claim counted (attempts 0->1); drain refunds it — the drained execution + # never ran a verdict, so the round-trip nets zero (spec §7.5). + assert h['attempts'] == '0' + assert h['state'] == 'pending' + assert h['leased_by'] == '' + assert h['lease_deadline'] == '0' + assert h['last_error'] == 'abort:drain' + assert job_id in _pending() + assert not _leased(job_id) + + +def test_abort_drain_never_converges(monkeypatch): + # Draining the same job forever must never converge to JE and must never + # erode the retry budget: attempts is '1' while leased and refunded to '0' + # after each drain requeue — bounded, never accumulating. + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + land_je = Writer() + + for _ in range(5): + assert job_mod.claim_next_job( + 'rn_a')['job_id'] == job_id # still claimable + assert _hash(job_id)['attempts'] == '1' # claim bump + assert job_mod.abort_job('rn_a', job_id, 'drain', + land_je) == AbortOutcome.REQUEUED + assert _hash(job_id)['attempts'] == '0' # refunded + + assert land_je.calls == [] + + +def test_drain_preserves_full_budget_for_later_failures(monkeypatch): + # After any number of drains, a genuinely failing execution path still gets + # its full retry budget: the JE converges at exactly the same boundary a + # never-drained job would (claim=1, reclaim=2, reclaim=3, then -1). + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + land_je = Writer() + + for _ in range( + 3): # rolling restarts churn the job without spending budget + job_mod.claim_next_job('rn_a') + assert job_mod.abort_job('rn_a', job_id, + 'drain') == AbortOutcome.REQUEUED + assert _hash(job_id)['attempts'] == '0' + + # now the real failure path, from a clean budget + job_mod.claim_next_job('rn_a') # attempts = 1 + assert _reclaim(job_id, 'rn_b', now=2000.0) == 1 # attempts = 2 + assert _reclaim(job_id, 'rn_c', now=3000.0) == 1 # attempts = 3 + # still one real attempt short of the ceiling: no JE yet + assert land_je.calls == [] + _expire(job_id) + _client().delete(redis_keys.DISPATCH_LAST_RECOVERY) + + monkeypatch.setattr(job_mod, '_now', lambda: 5000.0) + assert job_mod.claim_next_job('rn_z', land_je=land_je) is None + assert land_je.calls == ['sn_1'] # converged at the same boundary + assert _client().exists(redis_keys.job(job_id)) == 0 + + +@pytest.mark.parametrize('reason', ['prep_failed', 'rejected']) +def test_abort_counting_reason_requeues_below_max(monkeypatch, reason): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') # attempts = 1 + + assert job_mod.abort_job('rn_a', job_id, reason) == AbortOutcome.REQUEUED + h = _hash(job_id) + assert h['attempts'] == '2' # counted, still below MAX_ATTEMPTS + assert h['last_error'] == f'abort:{reason}' + assert job_id in _pending() + + +def test_abort_countable_exhaustion_converges(monkeypatch): + assert params.MAX_ATTEMPTS == 3 + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + land_je = Writer() + + job_mod.claim_next_job('rn_a') # attempts = 1 + assert job_mod.abort_job('rn_a', job_id, 'prep_failed', + land_je) == AbortOutcome.REQUEUED # attempts = 2 + job_mod.claim_next_job('rn_a') # attempts = 3 + assert job_mod.abort_job('rn_a', job_id, 'prep_failed', + land_je) == AbortOutcome.CONVERGED # attempts = 4 + + assert land_je.calls == ['sn_1'] + assert _client().exists(redis_keys.job(job_id)) == 0 + assert not _leased(job_id) + assert job_id not in _pending() + assert _current('sn_1') is None + + +def test_abort_converge_boundary_at_max(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + _client().hset(redis_keys.job(job_id), 'attempts', + '2') # one below the ceiling + + land_je = Writer() + assert job_mod.abort_job('rn_a', job_id, 'rejected', + land_je) == AbortOutcome.CONVERGED + assert land_je.calls == ['sn_1'] + assert _client().exists(redis_keys.job(job_id)) == 0 + + +def test_abort_converge_je_failure_leaves_job_for_sweep(monkeypatch): + # Exhausted, but the JE landing raises (Mongo down): the abort is still + # accepted (CONVERGED) and the job REMAINS as an immediately-expired orphan; + # a later wired claim sweeps it once Mongo recovers (INV1 retry path). + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + _client().hset(redis_keys.job(job_id), 'attempts', '2') + + failing_je = Writer(raises=RuntimeError('mongo down')) + assert job_mod.abort_job('rn_a', job_id, 'rejected', + failing_je) == AbortOutcome.CONVERGED + assert _leased(job_id) # job stays discoverable + h = _hash(job_id) + assert h['lease_deadline'] == '0' + assert int(h['attempts']) >= params.MAX_ATTEMPTS + + # recovery: clear the time gate, then a wired claim sweeps the orphan + _client().delete(redis_keys.DISPATCH_LAST_RECOVERY) + monkeypatch.setattr(job_mod, '_now', lambda: 5000.0) + ok_je = Writer() + assert job_mod.claim_next_job('rn_x', land_je=ok_je) is None + assert ok_je.calls == ['sn_1'] + assert _client().exists(redis_keys.job(job_id)) == 0 + assert not _leased(job_id) + assert _current('sn_1') is None + + +def test_abort_stale_is_dropped_not_requeued(monkeypatch, caplog): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job1 = _enqueue(submission_id='sn_x') + job_mod.claim_next_job('rn_a') + job2 = _enqueue(submission_id='sn_x') # rejudge moves the pointer + + land_je = Writer() + with caplog.at_level(logging.WARNING): + outcome = job_mod.abort_job('rn_a', job1, 'prep_failed', land_je) + + assert outcome == AbortOutcome.STALE_DROPPED + assert land_je.calls == [] + assert _client().exists(redis_keys.job(job1)) == 0 # destroyed + assert job1 not in _pending() + assert _current('sn_x') == job2 # pointer untouched + assert 'stale abort dropped' in caplog.text + + +def test_abort_ownership_existence_and_validation(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + + # non-owner → refused, untouched + assert job_mod.abort_job('rn_b', job_id, 'drain') == AbortOutcome.NOT_OWNER + assert _hash(job_id)['leased_by'] == 'rn_a' + assert not _pending() + + # unknown job → gone + assert job_mod.abort_job('rn_a', 'jb_missing', + 'drain') == AbortOutcome.GONE + + # invalid reason → ValueError before any Redis effect + with pytest.raises(ValueError): + job_mod.abort_job('rn_a', job_id, 'bogus') + assert _hash(job_id)['state'] == 'leased' # nothing happened + + +# --- JE sweep via orphan scan (reclaim exhaustion) ---------------------- + + +def _exhaust_to_expired_orphan(job_id): + """Drive a claimed job to MAX_ATTEMPTS via reclaims, then expire its lease.""" + assert _reclaim(job_id, 'rn_b', now=2000.0) == 1 # attempts = 2 + assert _reclaim(job_id, 'rn_c', now=3000.0) == 1 # attempts = 3 + _expire(job_id) + _client().delete(redis_keys.DISPATCH_LAST_RECOVERY) + + +def test_scan_sweeps_exhausted_orphan_to_je(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') # attempts = 1 + _exhaust_to_expired_orphan(job_id) + + monkeypatch.setattr(job_mod, '_now', lambda: 5000.0) + land_je = Writer() + assert job_mod.claim_next_job('rn_z', land_je=land_je) is None + + assert land_je.calls == ['sn_1'] + assert _client().exists(redis_keys.job(job_id)) == 0 + assert not _leased(job_id) + assert _current('sn_1') is None + + +def test_scan_destroys_superseded_orphan_without_je(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job1 = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') # attempts = 1 + _exhaust_to_expired_orphan(job1) + job2 = _enqueue(submission_id='sn_1') # rejudge: fresh, claimable + + monkeypatch.setattr(job_mod, '_now', lambda: 5000.0) + land_je = Writer() + # scan destroys the stale orphan without a JE, then hands out the fresh job + assert job_mod.claim_next_job('rn_z', land_je=land_je)['job_id'] == job2 + + assert land_je.calls == [] + assert _client().exists(redis_keys.job(job1)) == 0 + assert not _leased(job1) + + +def test_scan_dark_default_leaves_orphan_untouched(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + _exhaust_to_expired_orphan(job_id) + + monkeypatch.setattr(job_mod, '_now', lambda: 5000.0) + # land_je unset (dark): the exhausted orphan is left exactly as slice 2 left it + assert job_mod.claim_next_job('rn_z') is None + assert _client().exists(redis_keys.job(job_id)) == 1 + assert _leased(job_id) + + +def test_scan_lock_held_skips_sweep(monkeypatch): + monkeypatch.setattr(job_mod, '_now', lambda: 1000.0) + job_id = _enqueue(submission_id='sn_1') + job_mod.claim_next_job('rn_a') + _exhaust_to_expired_orphan(job_id) + _client().set(redis_keys.submission_job_lock('sn_1'), 'held') + + monkeypatch.setattr(job_mod, '_now', lambda: 5000.0) + land_je = Writer() + assert job_mod.claim_next_job('rn_z', land_je=land_je) is None + + assert land_je.calls == [] # next window retries + assert _client().exists(redis_keys.job(job_id)) == 1 + assert _leased(job_id)