feat(runner): add job poller and result sender

runner/__init__.py

@@ -1,5 +1,7 @@
 """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.
+Registration, heartbeat, HTTP client, active-job tracking, job polling and
+result sending used by the runner process to talk to the backend runner API.
+Job prep reuses the dispatcher's file/testdata helpers; still no docker and
+no Flask in this package.
 """

runner/client.py

@@ -25,6 +25,33 @@ class BackendAuthError(BackendAPIError):
     """The backend rejected our credentials (HTTP 401)."""
 
 
+@dataclass
+class JobPayload:
+    """A job claimed from GET /runners/<rn>/next-job (spec §7.3 wire contract)."""
+    job_id: str
+    submission_id: str
+    problem_id: int
+    language: int
+    code_url: str
+    checker: object
+    tasks: list
+
+    @classmethod
+    def from_dict(cls, body):
+        return cls(
+            job_id=body['job_id'],
+            submission_id=body['submission_id'],
+            # The backend job hash stores fields as strings; normalize here
+            # so consumers never coerce.
+            problem_id=int(body['problem_id']),
+            language=body['language'],
+            code_url=body['code_url'],
+            # The runner does not use the checker yet; keep it optional.
+            checker=body.get('checker'),
+            tasks=body['tasks'],
+        )
+
+
 @dataclass
 class RunnerConfig:
     heartbeat_interval_sec: int
@@ -101,6 +128,47 @@ def heartbeat(self, identity, active_job_ids):
             return
         self._raise_for_status(resp, 'heartbeat')
 
+    def next_job(self, identity):
+        """GET /runners/<rn>/next-job.
+
+        200 -> JobPayload; 204 -> None (no work available); other non-2xx ->
+        BackendAPIError / BackendAuthError. Network errors propagate.
+        """
+        resp = self.session.get(
+            f'{self.base_url}/runners/{identity.runner_id}/next-job',
+            headers={'Authorization': f'Bearer {identity.token}'},
+            timeout=self.timeout,
+        )
+        if resp.status_code == 200:
+            return JobPayload.from_dict(resp.json())
+        if resp.status_code == 204:
+            return None
+        self._raise_for_status(resp, 'next-job')
+
+    def complete(self, identity, job_id, tasks):
+        """PUT /runners/<rn>/jobs/<job_id>/complete. Expect 204."""
+        resp = self.session.put(
+            f'{self.base_url}/runners/{identity.runner_id}/jobs/{job_id}/complete',
+            json={'tasks': tasks},
+            headers={'Authorization': f'Bearer {identity.token}'},
+            timeout=self.timeout,
+        )
+        if resp.status_code == 204:
+            return
+        self._raise_for_status(resp, 'complete')
+
+    def abort(self, identity, job_id, reason):
+        """PUT /runners/<rn>/jobs/<job_id>/abort. Expect 202."""
+        resp = self.session.put(
+            f'{self.base_url}/runners/{identity.runner_id}/jobs/{job_id}/abort',
+            json={'reason': reason},
+            headers={'Authorization': f'Bearer {identity.token}'},
+            timeout=self.timeout,
+        )
+        if resp.status_code == 202:
+            return
+        self._raise_for_status(resp, 'abort')
+
     @staticmethod
     def _raise_for_status(resp, action):
         # Never include tokens in the message: only the status code is logged.

runner/config.py

@@ -27,3 +27,10 @@
 DEFAULT_HEARTBEAT_INTERVAL_SEC = 15
 DEFAULT_POLL_INTERVAL_SEC = 3
 DEFAULT_MAX_CONCURRENT_JOBS = 8
+
+# Local prep attempts before giving up and aborting with prep_failed (spec §10).
+PREP_MAX_ATTEMPTS = 3
+# Backoff (seconds) between prep attempts; len == PREP_MAX_ATTEMPTS - 1.
+PREP_BACKOFF_SCHEDULE = (1, 2)
+# Backoff (seconds) between complete/abort resends; len == max retries (spec §7).
+SEND_RETRY_BACKOFF_SCHEDULE = (1, 2, 4, 8, 16)

runner/poller.py

@@ -0,0 +1,142 @@
+import io
+import logging
+import shutil
+import threading
+import time
+
+import requests
+
+from dispatcher import file_manager, testdata
+from dispatcher.config import SUBMISSION_DIR
+from dispatcher.meta import Meta
+from .client import BackendAPIError
+from .config import REQUEST_TIMEOUT, PREP_MAX_ATTEMPTS, PREP_BACKOFF_SCHEDULE
+from .result_sender import AbortRequest
+
+logger = logging.getLogger(__name__)
+
+
+def prepare_job(payload):
+    """Fetch testdata + source and lay out the local job dir (spec §10).
+
+    Talks to the legacy testdata channel (backend + redis) for now; that
+    coexists until the keystone slice.
+    """
+    testdata.ensure_testdata(payload.problem_id)
+    resp = requests.get(payload.code_url, timeout=REQUEST_TIMEOUT)
+    resp.raise_for_status()
+    job_dir = SUBMISSION_DIR / payload.job_id
+    if job_dir.exists():
+        shutil.rmtree(job_dir)  # leftover from a failed prior attempt
+    file_manager.extract(
+        root_dir=SUBMISSION_DIR,
+        job_id=payload.job_id,
+        meta=Meta.parse_obj({
+            'language': payload.language,
+            'tasks': payload.tasks,
+        }),
+        source=io.BytesIO(resp.content),
+        testdata=testdata.get_problem_root(payload.problem_id),
+    )
+
+
+class PollerThread(threading.Thread):
+    """Claims jobs from the backend and preps them for dispatch (spec §10).
+
+    Only polls when there is spare capacity, adds the claimed job to the
+    tracker before prep so the heartbeat renews the lease while downloading,
+    and retries prep locally before giving up with a prep_failed abort.
+    """
+
+    def __init__(
+        self,
+        client,
+        identity,
+        tracker,
+        result_queue,
+        dispatch,
+        *,
+        prepare=None,
+        poll_interval_sec=None,
+        sleep=time.sleep,
+    ):
+        super().__init__(daemon=True)
+        self._client = client
+        self._identity = identity
+        self._tracker = tracker
+        self._result_queue = result_queue
+        self._dispatch = dispatch
+        self._prepare = prepare if prepare is not None else prepare_job
+        self._poll_interval_sec = (poll_interval_sec
+                                   if poll_interval_sec is not None else
+                                   identity.config.poll_interval_sec)
+        self._sleep = sleep
+        self._stop_event = threading.Event()
+
+    def run(self):
+        while not self._stop_event.is_set():
+            try:
+                idle = self._poll_once()
+            except Exception:
+                # A malformed payload (or any bug) must not kill the poller:
+                # the heartbeat would keep the runner looking alive while it
+                # never claims work again.
+                logger.exception('poller iteration failed')
+                idle = True
+            if idle:
+                # Interruptible wait so stop() takes effect immediately.
+                self._stop_event.wait(self._poll_interval_sec)
+
+    def _poll_once(self):
+        """Claim and prep one job. Returns True when the loop should idle."""
+        # Capacity gate: only GET next-job when there is room (spec §10).
+        if len(self._tracker) >= self._identity.config.max_concurrent_jobs:
+            return True
+        try:
+            payload = self._client.next_job(self._identity)
+        except (BackendAPIError, requests.RequestException) as err:
+            # 401 also just logs; heartbeat owns fail-fast.
+            logger.warning('next-job failed: %s', err)
+            return True
+        if payload is None:
+            return True
+
+        # Add BEFORE prep so the heartbeat renews the lease while we download;
+        # removal is the sender's job once the outcome resolves.
+        self._tracker.add(payload.job_id)
+
+        for i in range(PREP_MAX_ATTEMPTS):
+            try:
+                self._prepare(payload)
+                break
+            except Exception as err:
+                # Heterogeneous causes: network errors, extract ValueError.
+                logger.warning('prep for %s failed (attempt %d/%d): %s',
+                               payload.job_id, i + 1, PREP_MAX_ATTEMPTS, err)
+                if i < len(PREP_BACKOFF_SCHEDULE):
+                    self._sleep(PREP_BACKOFF_SCHEDULE[i])
+        else:
+            logger.error('prep for %s exhausted all attempts; aborting',
+                         payload.job_id)
+            self._abort(payload)
+            return False
+
+        try:
+            self._dispatch(payload.job_id, payload.submission_id)
+        except Exception as err:
+            # One shot only: a failed handle() may have partially enqueued
+            # task entries, and calling it again for the same job_id would
+            # revive them (duplicate execution). Requeue via abort instead;
+            # slice 4 makes handle() atomic and closes this for good.
+            logger.warning('dispatch for %s failed: %s', payload.job_id, err)
+            self._abort(payload)
+        return False
+
+    def _abort(self, payload):
+        self._result_queue.put(
+            AbortRequest(payload.job_id, payload.submission_id, 'prep_failed'))
+        # Do NOT remove from tracker; the sender does that as part of
+        # finalizing the abort.
+
+    def stop(self):
+        self._stop_event.set()