From c0a205b6cefa767f91ee727564013c92900fc8cd Mon Sep 17 00:00:00 2001 From: as535364 Date: Tue, 14 Jul 2026 02:05:18 +0800 Subject: [PATCH 1/3] feat(runner): add pull-dispatch coordination layer BackendClient (register/heartbeat), register_with_backoff (1-2-4-8-16-30s, 401 no-retry), HeartbeatThread (interval from register config, carries active_job_ids, 2x consecutive 401 fail-fast via on_fatal) and ActiveJobTracker. Dark slice - nothing wires into main.py yet; unit tests run without docker. Normal-OJ/Normal-OJ#68 (sandbox slice 2/4); spec section 10/13, ADR-0004 --- runner/__init__.py | 5 + runner/active_jobs.py | 31 +++++ runner/client.py | 116 ++++++++++++++++++ runner/config.py | 27 +++++ runner/heartbeat.py | 65 ++++++++++ runner/registration.py | 43 +++++++ tests/test_active_jobs.py | 71 +++++++++++ tests/test_runner_client.py | 186 ++++++++++++++++++++++++++++ tests/test_runner_heartbeat.py | 194 ++++++++++++++++++++++++++++++ tests/test_runner_registration.py | 123 +++++++++++++++++++ 10 files changed, 861 insertions(+) create mode 100644 runner/__init__.py create mode 100644 runner/active_jobs.py create mode 100644 runner/client.py create mode 100644 runner/config.py create mode 100644 runner/heartbeat.py create mode 100644 runner/registration.py create mode 100644 tests/test_active_jobs.py create mode 100644 tests/test_runner_client.py create mode 100644 tests/test_runner_heartbeat.py create mode 100644 tests/test_runner_registration.py diff --git a/runner/__init__.py b/runner/__init__.py new file mode 100644 index 0000000..bb34b8a --- /dev/null +++ b/runner/__init__.py @@ -0,0 +1,5 @@ +"""Pull-based runner coordination layer. + +Registration, heartbeat, HTTP client and active-job tracking used by the +runner process to talk to the backend runner API. No docker, no Flask. +""" diff --git a/runner/active_jobs.py b/runner/active_jobs.py new file mode 100644 index 0000000..eea62dd --- /dev/null +++ b/runner/active_jobs.py @@ -0,0 +1,31 @@ +import threading + + +class ActiveJobTracker: + """Thread-safe set of the job ids this runner currently holds. + + The snapshot is sent with every heartbeat to renew the backend leases + (spec §7.2); jobs absent from the snapshot stop being renewed and become + orphans once their lease expires. + """ + + def __init__(self): + self._lock = threading.Lock() + self._jobs = set() + + def add(self, job_id): + with self._lock: + self._jobs.add(job_id) + + def remove(self, job_id): + # Idempotent: removing an absent id is a no-op. + with self._lock: + self._jobs.discard(job_id) + + def snapshot(self): + with self._lock: + return list(self._jobs) + + def __len__(self): + with self._lock: + return len(self._jobs) diff --git a/runner/client.py b/runner/client.py new file mode 100644 index 0000000..117cbbc --- /dev/null +++ b/runner/client.py @@ -0,0 +1,116 @@ +import logging +from dataclasses import dataclass, field + +import requests + +from .config import ( + REQUEST_TIMEOUT, + DEFAULT_HEARTBEAT_INTERVAL_SEC, + DEFAULT_POLL_INTERVAL_SEC, + DEFAULT_MAX_CONCURRENT_JOBS, +) + +logger = logging.getLogger(__name__) + + +class BackendAPIError(Exception): + """A backend runner API call returned an unexpected non-2xx status.""" + + def __init__(self, message, status_code): + super().__init__(message) + self.status_code = status_code + + +class BackendAuthError(BackendAPIError): + """The backend rejected our credentials (HTTP 401).""" + + +@dataclass +class RunnerConfig: + heartbeat_interval_sec: int + poll_interval_sec: int + max_concurrent_jobs: int + + @classmethod + def from_dict(cls, data): + data = data or {} + return cls( + heartbeat_interval_sec=data.get( + 'heartbeat_interval_sec', + DEFAULT_HEARTBEAT_INTERVAL_SEC, + ), + poll_interval_sec=data.get( + 'poll_interval_sec', + DEFAULT_POLL_INTERVAL_SEC, + ), + max_concurrent_jobs=data.get( + 'max_concurrent_jobs', + DEFAULT_MAX_CONCURRENT_JOBS, + ), + ) + + +@dataclass +class RunnerIdentity: + runner_id: str + # Memory-only credential (ADR-0004); never leak it via repr. + token: str = field(repr=False) + config: RunnerConfig + + +class BackendClient: + """HTTP client for the backend runner API (spec §7).""" + + def __init__(self, base_url, session=None, timeout=REQUEST_TIMEOUT): + self.base_url = base_url.rstrip('/') + self.session = session if session is not None else requests.Session() + self.timeout = timeout + + def register(self, registration_token, name): + """POST /runners/register -> RunnerIdentity. + + 201 -> parsed identity; 401 -> BackendAuthError; other non-2xx -> + BackendAPIError. Network errors propagate as requests.RequestException. + """ + resp = self.session.post( + f'{self.base_url}/runners/register', + json={ + 'registration_token': registration_token, + 'name': name, + }, + timeout=self.timeout, + ) + if resp.status_code == 201: + body = resp.json() + return RunnerIdentity( + runner_id=body['runner_id'], + token=body['token'], + config=RunnerConfig.from_dict(body.get('config')), + ) + self._raise_for_status(resp, 'register') + + def heartbeat(self, identity, active_job_ids): + """POST /runners//heartbeat. Expect 204.""" + resp = self.session.post( + f'{self.base_url}/runners/{identity.runner_id}/heartbeat', + json={'active_job_ids': list(active_job_ids)}, + headers={'Authorization': f'Bearer {identity.token}'}, + timeout=self.timeout, + ) + if resp.status_code == 204: + return + self._raise_for_status(resp, 'heartbeat') + + @staticmethod + def _raise_for_status(resp, action): + # Never include tokens in the message: only the status code is logged. + status = resp.status_code + if status == 401: + raise BackendAuthError( + f'runner {action} rejected with 401', + status_code=status, + ) + raise BackendAPIError( + f'runner {action} failed with status {status}', + status_code=status, + ) diff --git a/runner/config.py b/runner/config.py new file mode 100644 index 0000000..6c23c0c --- /dev/null +++ b/runner/config.py @@ -0,0 +1,27 @@ +import os +import socket + +# Backend base URL (unified name; replaces the legacy BACKEND_API). +BACKEND_URL = os.getenv( + 'BACKEND_URL', + 'http://web:8080', +) +# Shared secret presented to POST /runners/register. +RUNNER_REGISTRATION_TOKEN = os.getenv( + 'RUNNER_REGISTRATION_TOKEN', + '', +) +# Stable, human-readable name for logs / admin display. +RUNNER_NAME = os.getenv('RUNNER_NAME') or socket.gethostname() + +# Registration retry backoff (seconds). After the schedule is exhausted the +# caller stays at the final value (30s) forever. +REGISTRATION_BACKOFF_SCHEDULE = (1, 2, 4, 8, 16, 30) + +# HTTP timeout (seconds) applied to every backend client call. +REQUEST_TIMEOUT = 10 + +# Fallback runner config values used when the register response omits a key. +DEFAULT_HEARTBEAT_INTERVAL_SEC = 15 +DEFAULT_POLL_INTERVAL_SEC = 3 +DEFAULT_MAX_CONCURRENT_JOBS = 8 diff --git a/runner/heartbeat.py b/runner/heartbeat.py new file mode 100644 index 0000000..126ff7f --- /dev/null +++ b/runner/heartbeat.py @@ -0,0 +1,65 @@ +import logging +import threading + +import requests + +from .client import BackendAPIError, BackendAuthError + +logger = logging.getLogger(__name__) + +# Consecutive 401s that trigger fail-fast (spec §10). +_FATAL_AUTH_FAILURES = 2 + + +class HeartbeatThread(threading.Thread): + """Periodically POST /runners//heartbeat to renew leases. + + Sends the first beat immediately, then every ``interval_sec``. Two + consecutive 401s trigger ``on_fatal`` (the future main.py wires this to + process exit) and stop the loop. Any non-401 outcome resets the counter. + """ + + def __init__(self, client, identity, tracker, on_fatal, interval_sec=None): + super().__init__(daemon=True) + self._client = client + self._identity = identity + self._tracker = tracker + self._on_fatal = on_fatal + self._interval_sec = (interval_sec if interval_sec is not None else + identity.config.heartbeat_interval_sec) + self._stop_event = threading.Event() + self._consecutive_auth_failures = 0 + + def run(self): + while not self._stop_event.is_set(): + fatal = self._beat() + if fatal: + break + # Interruptible wait so stop() takes effect immediately. + self._stop_event.wait(self._interval_sec) + + def _beat(self): + """Send one heartbeat. Returns True iff fail-fast was triggered.""" + try: + self._client.heartbeat(self._identity, self._tracker.snapshot()) + except BackendAuthError: + self._consecutive_auth_failures += 1 + if self._consecutive_auth_failures >= _FATAL_AUTH_FAILURES: + logger.error( + 'heartbeat got %d consecutive 401s; failing fast', + self._consecutive_auth_failures, + ) + self._on_fatal() + return True + logger.warning('heartbeat got 401 (%d consecutive)', + self._consecutive_auth_failures) + return False + except (BackendAPIError, requests.RequestException) as err: + self._consecutive_auth_failures = 0 + logger.warning('heartbeat failed: %s', err) + return False + self._consecutive_auth_failures = 0 + return False + + def stop(self): + self._stop_event.set() diff --git a/runner/registration.py b/runner/registration.py new file mode 100644 index 0000000..015aae1 --- /dev/null +++ b/runner/registration.py @@ -0,0 +1,43 @@ +import logging +import time + +import requests + +from .client import BackendAPIError, BackendAuthError +from .config import REGISTRATION_BACKOFF_SCHEDULE + +logger = logging.getLogger(__name__) + + +def register_with_backoff( + client, + registration_token, + name, + *, + schedule=REGISTRATION_BACKOFF_SCHEDULE, + sleep=time.sleep, +): + """Register with the backend, retrying transient failures indefinitely. + + Backoff follows ``schedule`` (1->2->4->8->16->30s) and then stays at the + final value forever. A 401 (BackendAuthError) is fatal per ADR-0004: it is + re-raised immediately so the caller can fail-fast. ``sleep`` is injectable + for tests. + """ + attempt = 0 + while True: + try: + return client.register(registration_token, name) + except BackendAuthError: + # Bad registration token: no amount of retrying will help. + raise + except (BackendAPIError, requests.RequestException) as err: + delay = schedule[min(attempt, len(schedule) - 1)] + logger.warning( + 'runner registration failed (attempt %d): %s; retrying in %ds', + attempt + 1, + err, + delay, + ) + attempt += 1 + sleep(delay) diff --git a/tests/test_active_jobs.py b/tests/test_active_jobs.py new file mode 100644 index 0000000..6529112 --- /dev/null +++ b/tests/test_active_jobs.py @@ -0,0 +1,71 @@ +import threading + +from runner.active_jobs import ActiveJobTracker + + +def test_add_and_snapshot(): + tracker = ActiveJobTracker() + tracker.add('jb_1') + tracker.add('jb_2') + + assert sorted(tracker.snapshot()) == ['jb_1', 'jb_2'] + assert len(tracker) == 2 + + +def test_add_is_idempotent(): + tracker = ActiveJobTracker() + tracker.add('jb_1') + tracker.add('jb_1') + + assert tracker.snapshot() == ['jb_1'] + assert len(tracker) == 1 + + +def test_remove(): + tracker = ActiveJobTracker() + tracker.add('jb_1') + tracker.add('jb_2') + tracker.remove('jb_1') + + assert tracker.snapshot() == ['jb_2'] + assert len(tracker) == 1 + + +def test_remove_absent_is_noop(): + tracker = ActiveJobTracker() + tracker.add('jb_1') + tracker.remove('jb_does_not_exist') + + assert tracker.snapshot() == ['jb_1'] + assert len(tracker) == 1 + + +def test_snapshot_is_a_copy(): + tracker = ActiveJobTracker() + tracker.add('jb_1') + snap = tracker.snapshot() + snap.append('jb_mutated') + + assert tracker.snapshot() == ['jb_1'] + + +def test_thread_safety_smoke(): + tracker = ActiveJobTracker() + n = 200 + + def worker(base): + for i in range(n): + job_id = f'{base}_{i}' + tracker.add(job_id) + tracker.remove(job_id) + + threads = [threading.Thread(target=worker, args=[b]) for b in 'abcd'] + for t in threads: + t.start() + for t in threads: + t.join() + + # Every add was paired with a remove, so the tracker ends empty and did + # not raise (e.g. "set changed size during iteration") under contention. + assert len(tracker) == 0 + assert tracker.snapshot() == [] diff --git a/tests/test_runner_client.py b/tests/test_runner_client.py new file mode 100644 index 0000000..021dece --- /dev/null +++ b/tests/test_runner_client.py @@ -0,0 +1,186 @@ +import pytest +import requests + +from runner.client import ( + BackendClient, + BackendAPIError, + BackendAuthError, + RunnerIdentity, + RunnerConfig, +) + + +class StubResponse: + + def __init__(self, status_code, json_body=None): + self.status_code = status_code + self._json_body = json_body + + def json(self): + return self._json_body + + +class RecordingSession: + """Duck-typed replacement for requests.Session that records calls.""" + + def __init__(self, response=None, exc=None): + self._response = response + self._exc = exc + self.calls = [] + + def post(self, url, json=None, headers=None, timeout=None): + self.calls.append({ + 'url': url, + 'json': json, + 'headers': headers, + 'timeout': timeout, + }) + if self._exc is not None: + raise self._exc + return self._response + + +def test_register_success_parses_identity_and_config(): + session = RecordingSession( + StubResponse( + 201, + { + 'runner_id': 'rn_abc', + 'token': 'rk_secret', + 'config': { + 'heartbeat_interval_sec': 20, + 'poll_interval_sec': 5, + 'max_concurrent_jobs': 4, + }, + }, + )) + client = BackendClient('http://web:8080', session=session, timeout=10) + + identity = client.register('reg-token', 'runner-1') + + assert isinstance(identity, RunnerIdentity) + assert identity.runner_id == 'rn_abc' + assert identity.token == 'rk_secret' + assert identity.config == RunnerConfig(20, 5, 4) + + call = session.calls[0] + assert call['url'] == 'http://web:8080/runners/register' + assert call['json'] == { + 'registration_token': 'reg-token', + 'name': 'runner-1', + } + assert call['timeout'] == 10 + + +def test_register_fills_config_fallback_defaults(): + session = RecordingSession( + StubResponse( + 201, + { + 'runner_id': 'rn_abc', + 'token': 'rk_secret', + 'config': { + 'poll_interval_sec': 7 + }, + }, + )) + client = BackendClient('http://web:8080', session=session) + + identity = client.register('reg-token', 'runner-1') + + # Missing keys use the spec fallback defaults (15 / 3 / 8). + assert identity.config == RunnerConfig(15, 7, 8) + + +def test_register_missing_config_object_uses_all_defaults(): + session = RecordingSession( + StubResponse(201, { + 'runner_id': 'rn_abc', + 'token': 'rk_secret', + })) + client = BackendClient('http://web:8080', session=session) + + identity = client.register('reg-token', 'runner-1') + + assert identity.config == RunnerConfig(15, 3, 8) + + +def test_register_401_raises_auth_error(): + session = RecordingSession(StubResponse(401)) + client = BackendClient('http://web:8080', session=session) + + with pytest.raises(BackendAuthError) as excinfo: + client.register('bad-token', 'runner-1') + assert excinfo.value.status_code == 401 + assert isinstance(excinfo.value, BackendAPIError) + + +def test_register_other_error_raises_api_error(): + session = RecordingSession(StubResponse(500)) + client = BackendClient('http://web:8080', session=session) + + with pytest.raises(BackendAPIError) as excinfo: + client.register('reg-token', 'runner-1') + assert not isinstance(excinfo.value, BackendAuthError) + assert excinfo.value.status_code == 500 + + +def test_register_network_error_propagates(): + session = RecordingSession(exc=requests.ConnectionError('boom')) + client = BackendClient('http://web:8080', session=session) + + with pytest.raises(requests.RequestException): + client.register('reg-token', 'runner-1') + + +def test_heartbeat_sends_url_auth_header_and_body(): + session = RecordingSession(StubResponse(204)) + client = BackendClient('http://web:8080', session=session, timeout=10) + identity = RunnerIdentity('rn_abc', 'rk_secret', RunnerConfig(15, 3, 8)) + + result = client.heartbeat(identity, ['jb_1', 'jb_2']) + + assert result is None + call = session.calls[0] + assert call['url'] == 'http://web:8080/runners/rn_abc/heartbeat' + assert call['headers'] == {'Authorization': 'Bearer rk_secret'} + assert call['json'] == {'active_job_ids': ['jb_1', 'jb_2']} + assert call['timeout'] == 10 + + +def test_heartbeat_401_raises_auth_error(): + session = RecordingSession(StubResponse(401)) + client = BackendClient('http://web:8080', session=session) + identity = RunnerIdentity('rn_abc', 'rk_secret', RunnerConfig(15, 3, 8)) + + with pytest.raises(BackendAuthError): + client.heartbeat(identity, []) + + +def test_heartbeat_other_error_raises_api_error(): + session = RecordingSession(StubResponse(503)) + client = BackendClient('http://web:8080', session=session) + identity = RunnerIdentity('rn_abc', 'rk_secret', RunnerConfig(15, 3, 8)) + + with pytest.raises(BackendAPIError) as excinfo: + client.heartbeat(identity, []) + assert not isinstance(excinfo.value, BackendAuthError) + assert excinfo.value.status_code == 503 + + +def test_base_url_trailing_slash_is_normalized(): + session = RecordingSession(StubResponse(204)) + client = BackendClient('http://web:8080/', session=session) + identity = RunnerIdentity('rn_abc', 'rk_secret', RunnerConfig(15, 3, 8)) + + client.heartbeat(identity, []) + + assert session.calls[0][ + 'url'] == 'http://web:8080/runners/rn_abc/heartbeat' + + +def test_token_absent_from_identity_repr(): + identity = RunnerIdentity('rn_abc', 'rk_super_secret', + RunnerConfig(15, 3, 8)) + assert 'rk_super_secret' not in repr(identity) + assert 'rn_abc' in repr(identity) diff --git a/tests/test_runner_heartbeat.py b/tests/test_runner_heartbeat.py new file mode 100644 index 0000000..3e90a7e --- /dev/null +++ b/tests/test_runner_heartbeat.py @@ -0,0 +1,194 @@ +import threading +import time + +import requests + +from runner.client import ( + BackendAPIError, + BackendAuthError, + RunnerConfig, + RunnerIdentity, +) +from runner.heartbeat import HeartbeatThread +from runner.active_jobs import ActiveJobTracker + + +class ScriptedClient: + """heartbeat() plays back scripted outcomes and records snapshots.""" + + def __init__(self, outcomes): + self._outcomes = list(outcomes) + self.snapshots = [] + self.identities = [] + + def heartbeat(self, identity, active_job_ids): + self.identities.append(identity) + self.snapshots.append(list(active_job_ids)) + if self._outcomes: + outcome = self._outcomes.pop(0) + else: + outcome = None + if isinstance(outcome, Exception): + raise outcome + + +def make_identity(interval=15): + return RunnerIdentity('rn_x', 'rk_tok', RunnerConfig(interval, 3, 8)) + + +class FatalFlag: + + def __init__(self): + self.count = 0 + + def __call__(self): + self.count += 1 + + +def build_thread(outcomes, tracker=None, interval_sec=15): + client = ScriptedClient(outcomes) + identity = make_identity() + tracker = tracker if tracker is not None else ActiveJobTracker() + fatal = FatalFlag() + hb = HeartbeatThread( + client, + identity, + tracker, + on_fatal=fatal, + interval_sec=interval_sec, + ) + return hb, client, fatal + + +def test_beat_sends_tracker_snapshot(): + tracker = ActiveJobTracker() + tracker.add('jb_1') + hb, client, _ = build_thread([None], tracker=tracker) + + fatal_triggered = hb._beat() + + assert fatal_triggered is False + assert client.snapshots == [['jb_1']] + + +def test_single_401_does_not_trigger_fatal(): + hb, client, fatal = build_thread( + [BackendAuthError('nope', status_code=401)]) + + fatal_triggered = hb._beat() + + assert fatal_triggered is False + assert fatal.count == 0 + + +def test_two_consecutive_401_triggers_fatal_once(): + hb, client, fatal = build_thread([ + BackendAuthError('nope', status_code=401), + BackendAuthError('nope', status_code=401), + ]) + + assert hb._beat() is False + assert hb._beat() is True + + assert fatal.count == 1 + + +def test_401_then_success_then_401_does_not_trigger_fatal(): + hb, client, fatal = build_thread([ + BackendAuthError('nope', status_code=401), + None, # success resets the counter + BackendAuthError('nope', status_code=401), + ]) + + assert hb._beat() is False # first 401 + assert hb._beat() is False # success resets + assert hb._beat() is False # isolated 401 again + + assert fatal.count == 0 + + +def test_401_then_non_401_error_resets_counter(): + hb, client, fatal = build_thread([ + BackendAuthError('nope', status_code=401), + BackendAPIError('boom', status_code=500), + BackendAuthError('nope', status_code=401), + ]) + + assert hb._beat() is False + assert hb._beat() is False # non-401 error resets + assert hb._beat() is False + + assert fatal.count == 0 + + +def test_non_401_errors_keep_thread_alive(): + hb, client, fatal = build_thread([ + BackendAPIError('boom', status_code=500), + requests.ConnectionError('down'), + None, + ]) + + assert hb._beat() is False + assert hb._beat() is False + assert hb._beat() is False + + assert fatal.count == 0 + + +def test_real_thread_beats_and_stops_promptly(): + tracker = ActiveJobTracker() + tracker.add('jb_1') + hb, client, fatal = build_thread( + [None] * 50, + tracker=tracker, + interval_sec=0.01, + ) + + hb.start() + # Give it time to emit several beats. + time.sleep(0.1) + t0 = time.time() + hb.stop() + hb.join(timeout=1.0) + elapsed = time.time() - t0 + + assert not hb.is_alive() + assert elapsed < 0.5 # stop() interrupts the wait promptly + assert len(client.snapshots) >= 2 + assert client.snapshots[0] == ['jb_1'] + assert fatal.count == 0 + + +def test_real_thread_fails_fast_on_two_401(): + fatal_event = threading.Event() + + class EventClient: + + def heartbeat(self, identity, active_job_ids): + raise BackendAuthError('nope', status_code=401) + + identity = make_identity() + hb = HeartbeatThread( + EventClient(), + identity, + ActiveJobTracker(), + on_fatal=fatal_event.set, + interval_sec=0.01, + ) + + hb.start() + assert fatal_event.wait(timeout=1.0) + hb.join(timeout=1.0) + assert not hb.is_alive() + + +def test_default_interval_comes_from_identity_config(): + client = ScriptedClient([None]) + identity = RunnerIdentity('rn_x', 'rk_tok', RunnerConfig(42, 3, 8)) + hb = HeartbeatThread( + client, + identity, + ActiveJobTracker(), + on_fatal=FatalFlag(), + ) + assert hb._interval_sec == 42 diff --git a/tests/test_runner_registration.py b/tests/test_runner_registration.py new file mode 100644 index 0000000..3bce5a1 --- /dev/null +++ b/tests/test_runner_registration.py @@ -0,0 +1,123 @@ +import pytest +import requests + +from runner.client import ( + BackendAPIError, + BackendAuthError, + RunnerConfig, + RunnerIdentity, +) +from runner.registration import register_with_backoff + + +class FakeClient: + """register() plays back a scripted list of outcomes.""" + + def __init__(self, outcomes): + self._outcomes = list(outcomes) + self.calls = [] + + def register(self, registration_token, name): + self.calls.append((registration_token, name)) + outcome = self._outcomes.pop(0) + if isinstance(outcome, Exception): + raise outcome + return outcome + + +def make_identity(): + return RunnerIdentity('rn_ok', 'rk_tok', RunnerConfig(15, 3, 8)) + + +def test_success_on_first_try_no_sleep(): + identity = make_identity() + client = FakeClient([identity]) + sleeps = [] + + result = register_with_backoff( + client, + 'reg-token', + 'runner-1', + sleep=sleeps.append, + ) + + assert result is identity + assert sleeps == [] + assert client.calls == [('reg-token', 'runner-1')] + + +def test_backoff_sequence_across_many_failures_then_success(): + identity = make_identity() + # 8 transient failures, then success. Sleep schedule must be + # 1,2,4,8,16,30,30,30 (capped at the final value). + outcomes = [BackendAPIError('fail', status_code=500) for _ in range(8)] + outcomes.append(identity) + client = FakeClient(outcomes) + sleeps = [] + + result = register_with_backoff( + client, + 'reg-token', + 'runner-1', + sleep=sleeps.append, + ) + + assert result is identity + assert sleeps == [1, 2, 4, 8, 16, 30, 30, 30] + + +def test_network_error_is_retried(): + identity = make_identity() + outcomes = [ + requests.ConnectionError('boom'), + requests.Timeout('slow'), + identity, + ] + client = FakeClient(outcomes) + sleeps = [] + + result = register_with_backoff( + client, + 'reg-token', + 'runner-1', + sleep=sleeps.append, + ) + + assert result is identity + assert sleeps == [1, 2] + + +def test_401_raises_immediately_with_zero_sleeps(): + client = FakeClient([BackendAuthError('nope', status_code=401)]) + sleeps = [] + + with pytest.raises(BackendAuthError): + register_with_backoff( + client, + 'bad-token', + 'runner-1', + sleep=sleeps.append, + ) + + assert sleeps == [] + assert len(client.calls) == 1 + + +def test_401_after_transient_failures_stops_retrying(): + outcomes = [ + BackendAPIError('fail', status_code=500), + BackendAuthError('nope', status_code=401), + ] + client = FakeClient(outcomes) + sleeps = [] + + with pytest.raises(BackendAuthError): + register_with_backoff( + client, + 'reg-token', + 'runner-1', + sleep=sleeps.append, + ) + + # Only the transient failure slept; the 401 aborts immediately. + assert sleeps == [1] From 491eabc3e2c14679709977e7792c97cff40885c8 Mon Sep 17 00:00:00 2001 From: as535364 Date: Wed, 22 Jul 2026 22:02:58 +0800 Subject: [PATCH 2/3] fix(runner): keep heartbeat on a fixed cadence Subtract the time the beat itself took from the next wait, so a slow or timed-out request (up to 10s) cannot push the next beat past the 30s lease TTL. Without this, one timeout almost always dropped the lease and forced a needless reclaim. Addresses review feedback. --- runner/heartbeat.py | 8 +++- tests/test_runner_heartbeat.py | 68 ++++++++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+), 1 deletion(-) diff --git a/runner/heartbeat.py b/runner/heartbeat.py index 126ff7f..ffcd1c2 100644 --- a/runner/heartbeat.py +++ b/runner/heartbeat.py @@ -1,5 +1,6 @@ import logging import threading +import time import requests @@ -32,11 +33,16 @@ def __init__(self, client, identity, tracker, on_fatal, interval_sec=None): def run(self): while not self._stop_event.is_set(): + started = time.monotonic() fatal = self._beat() if fatal: break + # Fixed cadence: subtract the time the beat took so a slow or + # timed-out request cannot push the next beat past the lease + # budget (15s interval vs 30s TTL allows exactly one miss). # Interruptible wait so stop() takes effect immediately. - self._stop_event.wait(self._interval_sec) + elapsed = time.monotonic() - started + self._stop_event.wait(max(0.0, self._interval_sec - elapsed)) def _beat(self): """Send one heartbeat. Returns True iff fail-fast was triggered.""" diff --git a/tests/test_runner_heartbeat.py b/tests/test_runner_heartbeat.py index 3e90a7e..db8cb16 100644 --- a/tests/test_runner_heartbeat.py +++ b/tests/test_runner_heartbeat.py @@ -182,6 +182,74 @@ def heartbeat(self, identity, active_job_ids): assert not hb.is_alive() +class FakeClock: + """Stands in for the time module; only monotonic() is used.""" + + def __init__(self): + self.now = 0.0 + + def monotonic(self): + return self.now + + +class RecordingStopEvent: + """Duck-types threading.Event; stops the loop after the first wait.""" + + def __init__(self): + self.waits = [] + self._stopped = False + + def is_set(self): + return self._stopped + + def wait(self, timeout=None): + self.waits.append(timeout) + self._stopped = True + return False + + def set(self): + self._stopped = True + + +class SlowClient: + """heartbeat() advances the fake clock to simulate a slow request.""" + + def __init__(self, clock, cost_sec): + self._clock = clock + self._cost = cost_sec + + def heartbeat(self, identity, active_job_ids): + self._clock.now += self._cost + + +def run_one_cycle(monkeypatch, beat_cost_sec, interval_sec): + clock = FakeClock() + monkeypatch.setattr('runner.heartbeat.time', clock) + hb = HeartbeatThread( + SlowClient(clock, beat_cost_sec), + make_identity(), + ActiveJobTracker(), + on_fatal=FatalFlag(), + interval_sec=interval_sec, + ) + stop_event = RecordingStopEvent() + hb._stop_event = stop_event + hb.run() + return stop_event.waits + + +def test_wait_subtracts_beat_duration(monkeypatch): + # A beat that takes 10s must shrink the following wait to 5s so the + # next beat stays on the fixed 15s cadence (lease-renewal budget). + waits = run_one_cycle(monkeypatch, beat_cost_sec=10, interval_sec=15) + assert waits == [5] + + +def test_wait_clamps_to_zero_when_beat_exceeds_interval(monkeypatch): + waits = run_one_cycle(monkeypatch, beat_cost_sec=40, interval_sec=15) + assert waits == [0.0] + + def test_default_interval_comes_from_identity_config(): client = ScriptedClient([None]) identity = RunnerIdentity('rn_x', 'rk_tok', RunnerConfig(42, 3, 8)) From 5f6b401e207cf51c9d8ce51bbd21bf09bb09c5ac Mon Sep 17 00:00:00 2001 From: as535364 Date: Wed, 22 Jul 2026 22:11:12 +0800 Subject: [PATCH 3/3] docs(runner): clarify BACKEND_URL comment BACKEND_API is not replaced yet; it still serves the old push path until the keystone slice. Addresses review feedback. --- runner/config.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/runner/config.py b/runner/config.py index 6c23c0c..ed7736d 100644 --- a/runner/config.py +++ b/runner/config.py @@ -1,7 +1,9 @@ import os import socket -# Backend base URL (unified name; replaces the legacy BACKEND_API). +# Backend base URL for the pull-based runner API. Coexists with the legacy +# BACKEND_API (dispatcher/config.py) until the keystone slice removes the +# old push path. BACKEND_URL = os.getenv( 'BACKEND_URL', 'http://web:8080',