"""Training-only settings/CAS, bounded durable jobs and owned-bundle retention.""" from contextlib import contextmanager from datetime import datetime, timezone, timedelta import fcntl import hashlib import json import os from pathlib import Path import shutil import sqlite3 import time import uuid DEFAULTS = dict(enabled=True, minimum_persons=8, interval_seconds=900, width=1280, height=720, max_bytes=200_000_000_000, min_free_bytes=100_000_000_000) # Compressed cache (32 MiB) plus two maximum-size RGB/JPEG images and bounded JSON. RESERVE = 96 * 1024**2 MARKER = 'contextra-training-v1' def root_path(): value = os.environ.get('CONTEXTRA_TRAINING_ROOT') return Path(value) if value else None def revision(value): return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest() def validate(value): if set(value) != set(DEFAULTS) or type(value['enabled']) is not bool: raise ValueError('Invalid training settings.') bounds = dict(minimum_persons=(1, 300), interval_seconds=(1, 86400), width=(64, 3840), height=(64, 2160), max_bytes=(1024, 10_000_000_000_000), min_free_bytes=(0, 10_000_000_000_000)) for key, (lo, hi) in bounds.items(): if type(value[key]) is not int or not lo <= value[key] <= hi: raise ValueError('Invalid training setting: ' + key) return value class TrainingStore: def __init__(self, root): self.root = Path(root) if not self.root.is_absolute() or self.root.is_symlink() or self.root.resolve() != self.root: raise ValueError('Training root must be an absolute non-symlink path.') self.root.mkdir(mode=0o700, parents=True, exist_ok=True) with self.lock(): if not (self.root/'settings.json').exists(): self._write_settings(DEFAULTS) with self.db() as db: db.execute('CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, camera TEXT, state TEXT, ' 'created REAL, finished REAL, path TEXT, error TEXT)') db.execute('CREATE TABLE IF NOT EXISTS cameras (camera TEXT PRIMARY KEY, saved REAL, attempted REAL)') def safe(self, relative): path = self.root / relative if path == self.root or self.root not in path.parents or path.resolve() != path: raise ValueError('Unsafe training path.') return path @contextmanager def lock(self): fd = os.open(self.root/'.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) try: fcntl.flock(fd, fcntl.LOCK_EX) yield finally: os.close(fd) @contextmanager def db(self): path = self.safe('index.sqlite') db = sqlite3.connect(path, timeout=1) try: yield db db.commit() finally: db.close() def settings(self): value = validate(json.loads(self.safe('settings.json').read_text())) effective = dict(value) from app.storage_policy import load policy = load() if policy: row = policy['training'] effective.update(max_bytes=row['max_bytes'], min_free_bytes=row['min_free_bytes'], cleanup_target_bytes=row['target_bytes'], retention_enabled=row['auto_cleanup']) return dict(revision=revision(value), configured=value, effective=effective, apply='subsequent_jobs', refresh_seconds=1, units=dict(bytes='decimal', interval='seconds')) def _write_settings(self, value): target = self.safe('settings.new') with target.open('w') as f: json.dump(value, f); f.flush(); os.fsync(f.fileno()) target.chmod(0o600) os.replace(target, self.safe('settings.json')) def patch(self, expected, changes): if type(changes) is not dict or not changes or set(changes)-set(DEFAULTS): raise ValueError('Invalid training settings patch.') with self.lock(): current = self.settings() if expected != current['revision']: raise RuntimeError('Training revision conflict.') candidate = validate(dict(current['configured'], **changes)) if candidate != current['configured']: self._write_settings(candidate) return self.settings() def usage(self): total = 0 for parent, dirs, files in os.walk(self.root, followlinks=False): for name in dirs + files: path = Path(parent)/name if path.is_symlink(): raise ValueError('Symlink in training storage; retention refused.') total += sum((Path(parent)/name).stat().st_size for name in files) return total def _space(self, db, cfg, reserve): used, free = self.usage(), shutil.disk_usage(self.root).free # Shared disk admission keeps headroom for the other owner's bounded in-flight work. external = os.environ.get('CONTEXTRA_TRAINING_EXTERNAL_RESERVE_BYTES', '0') if not external.isascii() or not external.isdecimal() or not 0 <= int(external) <= 10_000_000_000_000: raise ValueError('Invalid training external reservation.') free -= int(external) excess = used + reserve > cfg['max_bytes'] target = cfg.get('cleanup_target_bytes', cfg['max_bytes']*9//10) if excess else cfg['max_bytes'] if not cfg.get('retention_enabled', True): return used + reserve <= cfg['max_bytes'] and free - reserve >= cfg['min_free_bytes'] rows = db.execute("SELECT id,path FROM jobs WHERE state='READY' ORDER BY created").fetchall() for job_id, relative in rows: if used + reserve <= target and free - reserve >= cfg['min_free_bytes']: break path = self.safe(relative) if not path.is_dir(): continue marker = path/'metadata.json' if marker.is_symlink() or not marker.is_file(): continue manifest = json.loads(marker.read_text()) if manifest.get('owner') != MARKER or manifest.get('jobId') != job_id: continue size = sum(p.stat().st_size for p in path.rglob('*') if p.is_file()) shutil.rmtree(path) db.execute("UPDATE jobs SET state='DELETED' WHERE id=?", (job_id,)) used -= size free = shutil.disk_usage(self.root).free-int(external) return used + reserve <= target and free - reserve >= cfg['min_free_bytes'] def admission_wait(self, camera, cfg, *, now=None): """Cheap scheduling hint only; reserve remains the authoritative transaction.""" now = time.time() if now is None else now with self.db() as db: row = db.execute('SELECT saved,attempted FROM cameras WHERE camera=?', (str(camera),)).fetchone() pending = db.execute("SELECT 1 FROM jobs WHERE camera=? AND state IN ('CAPTURING','QUEUED','RUNNING') LIMIT 1", (str(camera),)).fetchone() if pending: return 1.0 if row: return max(0., row[0]+cfg['interval_seconds']-now, row[1]+min(30,cfg['interval_seconds'])-now) return 0. def reserve(self, camera, cfg, *, now=None, reserve=RESERVE): now = time.time() if now is None else now camera = str(camera) with self.lock(), self.db() as db: if not cfg['enabled']: return None, 'disabled' row = db.execute('SELECT saved,attempted FROM cameras WHERE camera=?', (camera,)).fetchone() if row and (now-row[0] < cfg['interval_seconds'] or now-row[1] < min(30, cfg['interval_seconds'])): return None, 'interval' pending = db.execute("SELECT camera FROM jobs WHERE state IN ('CAPTURING','QUEUED','RUNNING')").fetchall() if (camera,) in pending or len(pending) >= 4: return None, 'busy' if not self._space(db, cfg, reserve*(len(pending)+1)): return None, 'capacity' job = uuid.uuid4().hex stage = self.safe('.staging/'+job) stage.mkdir(mode=0o700, parents=True) db.execute('INSERT INTO jobs VALUES (?,?,?,?,?,?,?)', (job,camera,'CAPTURING',now,None,str(stage.relative_to(self.root)),None)) db.execute('INSERT INTO cameras VALUES (?,?,?) ON CONFLICT(camera) DO UPDATE SET attempted=excluded.attempted', (camera, row[0] if row else 0, now)) return job, None def finish(self, job, error=None): with self.lock(), self.db() as db: row = db.execute('SELECT camera,path,state FROM jobs WHERE id=?', (job,)).fetchone() if not row or row[2] not in ('CAPTURING','QUEUED','RUNNING'): return stage = self.safe(row[1]) if error: # Only service-created staging is removed, never original Event captures. if stage.parent == self.safe('.staging') and stage.name == job and stage.is_dir(): self.usage() # Refuse cleanup if any symlink is present. shutil.rmtree(stage) db.execute("UPDATE jobs SET state='FAILED',finished=?,error=? WHERE id=?", (time.time(),error,job)) return meta = json.loads((stage/'metadata.json').read_text()) if sum(p.stat().st_size for p in stage.iterdir() if p.is_file()) > RESERVE: raise ValueError('Training reservation exceeded.') stamp = datetime.fromtimestamp(meta['savedAt']/1000, timezone(timedelta(hours=9))) camera = row[0] if row[0].isdigit() else hashlib.sha256(row[0].encode()).hexdigest()[:16] rel = (f'{stamp:%Y-%m-%d}/camera-{camera}/' f'{stamp:%Y%m%dT%H%M%S}+0900_{meta["providerKey"]}_person-{meta["personCount"]:02d}_{job}') destination = self.safe(rel) destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True) for name in ('capture.bin','job.json'): (stage/name).unlink() stage.rename(destination) db.execute("UPDATE jobs SET state='READY',finished=?,path=? WHERE id=?", (time.time(),rel,job)) db.execute('UPDATE cameras SET saved=? WHERE camera=?', (time.time(),row[0])) def summary(self): with self.db() as db: return dict(counts=dict(db.execute('SELECT state,count(*) FROM jobs GROUP BY state')), recent=[dict(zip(('id','camera','state','created','finished','error'),r)) for r in db.execute('SELECT id,camera,state,created,finished,error FROM jobs ORDER BY created DESC LIMIT 10')])