""" AUTH.3E — job_store.py ====================== SQLite Inbox für den Executor Command Channel. Eigenschaften: * Getrennte DB-Dateien pro Executor (c5a_save.db / c5a_delete.db) * Atomarer Claim (READY -> CLAIMED) via SQLite-Transaktion (kein TOCTOU) * Lease (lease_until) für Crash-Recovery * Idempotenz (job_id / idempotency_key UNIQUE) * Immutable fields after claim (UPDATE verboten für sicherheitskritische Felder) * FAIL CLOSED bei DB-Fehler Isoliert implementiert (KEIN produktiver Container). Nutzt job_schema + job_state_machine. """ from __future__ import annotations import json import sqlite3 import time import uuid from typing import Any, Dict, List, Optional from job_schema import ( JOB_TYPE_DELETE, JOB_TYPE_SAVE, JobRejectedError, validate_job, ) from job_state_machine import ( ST_CLAIMED, ST_CREATED, ST_EXECUTING, ST_FAILED, ST_OUTCOME_UNKNOWN, ST_READY, ST_REJECTED, ST_SUCCEEDED, InvalidTransitionError, transition, ) from job_claim import ( JobAlreadyClaimedError, JobNotFoundError, atomic_claim, recover_stale_claims as _recover_stale_claims, renew_lease as _renew_lease, ) # --------------------------------------------------------------------------- # Immutable fields after claim (UPDATE verboten) # --------------------------------------------------------------------------- IMMUTABLE_FIELDS = frozenset({ "job_type", "object_id", "vault_path", "approval_id", "expected_commit", "expected_provenance_hash", "delete_request_id", "source_commit", "provenance_hash", "mission_id", }) class JobStoreError(Exception): """Basis-Fehler für JobStore.""" class JobImmutableFieldError(JobStoreError): pass class JobStore: """ SQLite Inbox. Ein Store pro Executor (SAVE oder DELETE). worker_scope: "SAVE" oder "DELETE" — bestimmt, welche Job-Types claimbar sind. """ def __init__(self, db_path: str, worker_scope: str): self.db_path = db_path self.worker_scope = worker_scope.upper() if self.worker_scope not in ("SAVE", "DELETE"): raise JobStoreError(f"invalid worker_scope: {worker_scope!r}") self._conn = sqlite3.connect(db_path) self._conn.row_factory = sqlite3.Row self._init_schema() def _init_schema(self) -> None: self._conn.execute(""" CREATE TABLE IF NOT EXISTS jobs ( job_id TEXT PRIMARY KEY, job_version INTEGER NOT NULL, mission_id TEXT NOT NULL, job_type TEXT NOT NULL, object_id TEXT NOT NULL, vault_path TEXT NOT NULL, payload TEXT NOT NULL, -- vollständiger Job (JSON) state TEXT NOT NULL, idempotency_key TEXT NOT NULL UNIQUE, worker_id TEXT, claim_id TEXT, claimed_at INTEGER, lease_until INTEGER, attempt_count INTEGER NOT NULL DEFAULT 0, result_code TEXT, created_at TEXT NOT NULL, updated_at INTEGER NOT NULL ) """) self._conn.execute(""" CREATE INDEX IF NOT EXISTS idx_jobs_state ON jobs(state) """) self._conn.execute(""" CREATE INDEX IF NOT EXISTS idx_jobs_type ON jobs(job_type) """) self._conn.commit() # -- Job-Type-Scope ----------------------------------------------------- def _job_type_allowed(self, job_type: str) -> bool: """SAVE-Executor claimt nur SAVE-Jobs; DELETE-Executor nur DELETE-Jobs.""" if self.worker_scope == "SAVE": return job_type == JOB_TYPE_SAVE return job_type == JOB_TYPE_DELETE # -- Erzeugen (RQ-Seite) ------------------------------------------------ def create_job(self, job: Dict[str, Any]) -> Dict[str, Any]: """ Erzeugt einen Job (Status CREATED). Idempotent per job_id. Wirft JobRejectedError bei Schema-Verletzung. """ validated = validate_job(job) job_id = validated["job_id"] now = int(time.time() * 1000) # Idempotenz: gleiche job_id -> bestehenden Job zurückgeben (kein Fehler). existing = self.get_job(job_id) if existing is not None: return existing try: self._conn.execute( """ INSERT INTO jobs (job_id, job_version, mission_id, job_type, object_id, vault_path, payload, state, idempotency_key, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( job_id, validated["job_version"], validated["mission_id"], validated["job_type"], validated["object_id"], validated["vault_path"], json.dumps(validated), ST_CREATED, validated["idempotency_key"], validated["created_at"], now, ), ) self._conn.commit() except sqlite3.IntegrityError as e: # UNIQUE-Verletzung (idempotency_key bereits vergeben) -> Anomalie. raise JobStoreError(f"create_job failed (idempotency_key collision): {e}") from e return self.get_job(job_id) def get_job(self, job_id: str) -> Optional[Dict[str, Any]]: row = self._conn.execute( "SELECT * FROM jobs WHERE job_id = ?", (job_id,) ).fetchone() return self._row_to_dict(row) if row else None def get_job_by_idempotency_key(self, idempotency_key: str) -> Optional[Dict[str, Any]]: row = self._conn.execute( "SELECT * FROM jobs WHERE idempotency_key = ?", (idempotency_key,) ).fetchone() return self._row_to_dict(row) if row else None def list_jobs(self, state: Optional[str] = None) -> List[Dict[str, Any]]: if state: rows = self._conn.execute( "SELECT * FROM jobs WHERE state = ? ORDER BY created_at", (state,) ).fetchall() else: rows = self._conn.execute( "SELECT * FROM jobs ORDER BY created_at" ).fetchall() return [self._row_to_dict(r) for r in rows] # -- State-Transition (mit Immutable-Fields-Schutz) --------------------- def _transition(self, job_id: str, to_state: str, *, result_code: Optional[str] = None, worker_id: Optional[str] = None) -> Dict[str, Any]: """ Führt eine State-Transition aus. Wirft InvalidTransitionError bei ungültiger Transition (FAIL CLOSED). """ job = self.get_job(job_id) if job is None: raise JobNotFoundError(f"job not found: {job_id}") from_state = job["state"] transition(from_state, to_state) # wirft bei ungültiger Transition now = int(time.time() * 1000) self._conn.execute( """ UPDATE jobs SET state = ?, result_code = ?, updated_at = ?, worker_id = COALESCE(?, worker_id) WHERE job_id = ? """, (to_state, result_code, now, worker_id, job_id), ) self._conn.commit() return self.get_job(job_id) def mark_ready(self, job_id: str) -> Dict[str, Any]: return self._transition(job_id, ST_READY) def mark_rejected(self, job_id: str, result_code: str) -> Dict[str, Any]: return self._transition(job_id, ST_REJECTED, result_code=result_code) def mark_succeeded(self, job_id: str, worker_id: str) -> Dict[str, Any]: return self._transition(job_id, ST_SUCCEEDED, worker_id=worker_id) def mark_failed(self, job_id: str, result_code: str, worker_id: str) -> Dict[str, Any]: return self._transition(job_id, ST_FAILED, result_code=result_code, worker_id=worker_id) def mark_outcome_unknown(self, job_id: str, worker_id: str) -> Dict[str, Any]: return self._transition(job_id, ST_OUTCOME_UNKNOWN, worker_id=worker_id) # -- Claim / Lease ------------------------------------------------------ def claim_job(self, job_id: str, worker_id: str, lease_seconds: int = 60) -> Dict[str, Any]: """ Atomarer Claim: READY -> CLAIMED, nur wenn Lease abgelaufen oder nie gesetzt. Kein TOCTOU (SQLite-Transaktion mit Status-Bedingung). Delegiert an job_claim. """ claimed = atomic_claim(self._conn, job_id, worker_id, lease_seconds) return claimed def begin_execution(self, job_id: str, worker_id: str) -> Dict[str, Any]: """CLAIMED -> EXECUTING (Claim bestätigt).""" return self._transition(job_id, ST_EXECUTING, worker_id=worker_id) def renew_lease(self, job_id: str, lease_seconds: int = 60) -> Dict[str, Any]: """Verlängert die Lease eines CLAIMED/EXECUTING-Jobs.""" _renew_lease(self._conn, job_id, lease_seconds) return self.get_job(job_id) # -- Crash-Recovery ----------------------------------------------------- def recover_stale_claims(self, worker_id: str, lease_seconds: int = 60) -> List[Dict[str, Any]]: """ Findet stale CLAIMED-Jobs (Lease abgelaufen) und überführt sie: * SAVE: zurück zu READY (wieder claimbar, RETRYABLE) * DELETE: zu OUTCOME_UNKNOWN (NICHT blind wiederholen) """ # job_claim.recover_stale_claims überführt SAVE->READY, DELETE->OUTCOME_UNKNOWN # anhand des job_type. Wir rufen es pro Job-Type auf (Scope-getrennt). recovered = [] for job_type in (JOB_TYPE_SAVE, JOB_TYPE_DELETE): if self._job_type_allowed(job_type): _recover_stale_claims(self._conn, job_type) # Re-read recovered jobs for row in self._conn.execute( "SELECT job_id FROM jobs WHERE state IN (?, ?)", (ST_READY, ST_OUTCOME_UNKNOWN), ).fetchall(): job = self.get_job(row["job_id"]) if job is not None: recovered.append(job) return recovered # -- Idempotenz --------------------------------------------------------- def is_duplicate(self, job_id: str, idempotency_key: str) -> bool: """True, wenn job_id ODER idempotency_key bereits existiert.""" row = self._conn.execute( "SELECT 1 FROM jobs WHERE job_id = ? OR idempotency_key = ? LIMIT 1", (job_id, idempotency_key), ).fetchone() return row is not None # -- Audit -------------------------------------------------------------- def audit_trail(self, job_id: str) -> List[Dict[str, Any]]: """Gibt den Audit-Trail eines Jobs zurück (aus der jobs-Tabelle).""" job = self.get_job(job_id) if job is None: return [] return [{ "job_id": job["job_id"], "mission_id": job["mission_id"], "job_type": job["job_type"], "object_id": job["object_id"], "state": job["state"], "worker_id": job["worker_id"], "attempt_count": job["attempt_count"], "result_code": job["result_code"], "created_at": job["created_at"], "updated_at": job["updated_at"], }] # -- Helpers ------------------------------------------------------------ def _row_to_dict(self, row: sqlite3.Row) -> Dict[str, Any]: d = dict(row) payload = json.loads(d["payload"]) # Merge Payload-Felder in das Top-Level-Dict, damit Executor-Cores # auf approval_id/expected_commit/source_commit etc. zugreifen können. d.update(payload) d["payload"] = payload return d def close(self) -> None: self._conn.close()