"""CLAIM-01~03: retry known DB outcomes, never repeat rendering or HTTP.""" from collections import Counter from copy import deepcopy import hashlib import json import os from pathlib import Path import threading import time import uuid from app.event_artifact_store import StaleArtifactClaim class ArtifactCompletion: LIMIT = 64 ATTEMPTS = 8 def __init__(self, store): self.store = store self.root = store.config.root / '.completions' self.pending = {} self.busy = set() self.lock = threading.Lock() self.counts = Counter() self.error = None @staticmethod def key(row): return hashlib.sha256((row['event_id']+'\0'+row['claim_token']).encode()).hexdigest() def observe(self, phase, item, exc=None): from contextra_trace import emit try: row = item['row'] emit(direction='AI_TO_WEB', component='artifact', phase=phase, eventId=row['event_id'], cameraId=row['payload'].get('cameraId'), runId=row['payload']['runId'], runGeneration=row['payload']['runGeneration'], claimToken=row['claim_token'], operation=item['operation'], attempt=item['attempts'], error=type(exc).__name__ if exc else None, sqlState=getattr(exc, 'pgcode', None)) except Exception: pass def status(self): with self.lock: return dict(pending=len(self.pending), retry_exhausted=sum( p['attempts'] >= self.ATTEMPTS for p in self.pending.values()), counts=dict(self.counts), error=self.error) def available(self): # Leave room for all eight workers already outside the admission check. with self.lock: return len(self.pending) < self.LIMIT-8 def submit(self, operation, row, *args, **kwargs): if operation not in ('rendered', 'finish_attempt'): raise ValueError('Invalid Artifact completion operation.') key = self.key(row) identity = {k: row[k] for k in ('event_id', 'channel_key', 'claim_token')} identity['payload'] = {k: row['payload'][k] for k in ('runId','runGeneration','cameraId')} item = dict(operation=operation, row=identity, args=deepcopy(args), kwargs=deepcopy(kwargs), attempts=0, next_at=0.) with self.lock: if key in self.pending: raise RuntimeError('Duplicate completion owner.') if len(self.pending) >= self.LIMIT: raise RuntimeError('Artifact completion capacity exhausted.') self.pending[key] = item return self._attempt(key) def _persist(self, key, item): self.root.mkdir(mode=0o700, exist_ok=True) if self.root.is_symlink() or self.root.resolve() != self.root: raise ValueError('Unsafe completion directory.') data = deepcopy({k:item[k] for k in ('operation','row','args','kwargs')}) # Successful responses are already restricted to accepted/duplicated counts. # Raw rejection bodies stay in TRACE, not in a second diagnostic file. if data['operation'] == 'finish_attempt' and data['args'][0] != 'SENT': data['args'] = list(data['args']); data['args'][2] = None encoded = json.dumps(dict(schema=1, **data)).encode() if len(encoded) > 131072: raise ValueError('Completion receipt is too large.') tmp = self.root / (key+'.'+uuid.uuid4().hex+'.tmp') try: fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) with os.fdopen(fd, 'wb') as f: f.write(encoded); f.flush(); os.fsync(f.fileno()) os.replace(tmp, self.root/(key+'.json')) fd = os.open(self.root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: os.fsync(fd) finally: os.close(fd) finally: tmp.unlink(missing_ok=True) def _attempt(self, key): with self.lock: item = self.pending.get(key) if (item is None or key in self.busy or item['attempts'] >= self.ATTEMPTS or item['next_at'] > time.monotonic()): return False self.busy.add(key); item['attempts'] += 1 completed = False try: getattr(self.store, item['operation'])(item['row'], *item['args'], **item['kwargs']) self.observe('ARTIFACT_FINALIZED', item) completed = True self.counts['completed'] += 1 except StaleArtifactClaim as exc: # A prior commit may have succeeded despite a lost connection. Never overwrite # a new claim, or repeat the external action, to resolve that ambiguity. self.observe('ARTIFACT_FINALIZATION_FENCED', item, exc) self.counts['fenced'] += 1 completed = True except Exception as exc: self.error = 'ARTIFACT_FINALIZE_'+type(exc).__name__.upper() self.counts['db_failures'] += 1 self.observe('ARTIFACT_FINALIZATION_DEFERRED', item, exc) try: self._persist(key, item) except Exception as receipt_error: self.error = 'ARTIFACT_COMPLETION_RECEIPT_'+type(receipt_error).__name__.upper() self.counts['receipt_errors'] += 1 self.observe('ARTIFACT_COMPLETION_RECEIPT_FAILED', item, receipt_error) item['next_at'] = time.monotonic()+min(30, 2**item['attempts']) finally: with self.lock: self.busy.discard(key) if completed: self.pending.pop(key, None) if completed: try: (self.root/(key+'.json')).unlink(missing_ok=True) except OSError: self.counts['receipt_cleanup_errors'] += 1 return completed def retry(self, max_jobs=4, budget_seconds=1): until = time.monotonic()+budget_seconds with self.lock: keys = [k for k,v in self.pending.items() if v['next_at'] <= time.monotonic() and v['attempts'] < self.ATTEMPTS and k not in self.busy][:max_jobs] for key in keys: if time.monotonic() >= until: break self._attempt(key) def restore(self): # Called under the parent's exclusive worker lock, before ledger recovery. if not self.root.exists(): return if self.root.is_symlink() or self.root.resolve() != self.root: raise ValueError('Unsafe completion directory.') files = list(self.root.glob('*.json')) if len(files) > self.LIMIT: raise RuntimeError('Completion recovery capacity exceeded.') for p in files: if p.is_symlink() or not p.is_file() or p.stat().st_size > 131072: raise ValueError('Unsafe completion receipt.') data = json.loads(p.read_text()) if data.get('schema') != 1 or p.stem != self.key(data['row']): raise ValueError('Invalid completion receipt identity.') self.submit(data['operation'], data['row'], *data['args'], **data['kwargs']) if self.pending: raise RuntimeError('Known Artifact outcomes could not be recovered.')