"""CHANNELPROC-01: Redis writer ownership is fenced across process lifetimes.""" from collections import Counter from dataclasses import replace import json import threading import time from uuid import uuid4 from app.metadata_publisher import MetadataPublisher, RedisSnapshotSink, camera_key CHANNEL_BYTES = 16 * 1024 * 1024 CONTROL_BYTES = 16 * 1024 class StaleWriter(RuntimeError): pass class FencedSink(RedisSnapshotSink): def __init__(self, settings, token, client=None, *, frames_only=False, preserve_frames=False): super().__init__(settings, client) self.token = token self.inflight_bytes = 0 self.frames_only = frames_only self.preserve_frames = preserve_frames def owner_key(self, key): return f'{self.settings.namespace}:camera:{key}:writer' def claim(self, key, expected, token, *, bootstrap=False): self._connect() with self.client.pipeline(transaction=True) as pipe: owner = self.owner_key(key) pipe.watch(owner) actual = pipe.get(owner) if not bootstrap and actual != expected: raise StaleWriter('Metadata writer ownership changed.') pipe.multi() pipe.set(owner, token) if token.startswith('run-'): pipe.delete(f'{self.settings.namespace}:camera:{key}:frames') pipe.execute() def _commands(self, pipe, key, status, payload): if self.frames_only: if payload is not None: stream = f'{self.settings.namespace}:camera:{key}:frames' pipe.xadd(stream, {'json': payload}, maxlen=self.settings.stream_length, approximate=False) pipe.expire(stream, self.settings.stream_ttl) return if self.preserve_frames and status['state'] in ('STARTING', 'RUNNING'): # The run owner cleared once at claim; provider writers may already be active. stream = f'{self.settings.namespace}:camera:{key}:frames' pipe.hset(f'{self.settings.namespace}:channels', key, stream) pipe.set(f'{self.settings.namespace}:camera:{key}:status', json.dumps(dict(status, streamKey=stream), allow_nan=False), ex=self.settings.stream_ttl) pipe.expire(f'{self.settings.namespace}:channels', self.settings.stream_ttl) return super()._commands(pipe, key, status, payload) def write(self, key, status, payload): self._connect() self.inflight_bytes = len(payload.encode()) if payload else 0 try: with self.client.pipeline(transaction=True) as pipe: owner = self.owner_key(key) pipe.watch(owner) if pipe.get(owner) != self.token: raise StaleWriter('Stale Metadata writer.') pipe.multi() self._commands(pipe, key, status, payload) pipe.execute() self.runs[key] = status['runId'] finally: self.inflight_bytes = 0 class ChildPublisher(MetadataPublisher): """One channel, one persistent transmission thread; terminal write is acknowledged.""" def __init__(self, settings, source_id, run_id, generation, token, sink=None, *, frames_only=False, preserve_frames=False): settings = replace(settings, worker_mode='fixed', workers=1, max_channels=1, active_worker_limit=1, max_bytes=CHANNEL_BYTES) self.identity = source_id, run_id, generation self.terminal = threading.Event() self.terminal_ok = False target = sink or FencedSink(settings, token, frames_only=frames_only, preserve_frames=preserve_frames) publisher = self class TerminalSink: def write(self, key, status, payload): terminal = status['state'] in ('STOPPED', 'ERROR') try: target.write(key, status, payload) if terminal: publisher.terminal_ok = True finally: if terminal: publisher.terminal.set() def close(self): target.close() self.target = target super().__init__(settings, sink=TerminalSink()) def channel(self, camera_id, run_id, generation, state): if (camera_id, run_id, generation) != self.identity: return False return super().channel(camera_id, run_id, generation, state) def wait_channel(self, camera_id, run_id, generation, timeout): return ((camera_id, run_id, generation) == self.identity and self.terminal.wait(timeout) and self.terminal_ok) def metrics(self, include_timeline=False): result = super().metrics(include_timeline) with self._cv: result['lastSeq'] = max((v['lastSeq'] for v in self._channels.values()), default=-1) result.update(workerMode='process-channel', inflightBytes=getattr(self.target, 'inflight_bytes', 0), inflightByteLimit=self.settings.max_message_bytes) return result def note_provider_sequence(self, sequence): if type(sequence) is not int or sequence < 0: return with self._cv: state = self._channels.get(camera_key(self.identity[0])) if state and state['state'] == 'RUNNING' and sequence > state['lastSeq']: state['lastSeq'] = sequence self._queue(camera_key(self.identity[0]), None, 'state') class ProcessMetadataControl: """Parent owns only inactive statuses; active Frame payloads never cross here.""" def __init__(self, settings, *, sink_factory=FencedSink): if settings.max_message_bytes > 10 * 1024 * 1024: raise ValueError('Process publisher supports at most 10 MiB per Frame.') # Each child has one <=10MiB slot. Inactive control snapshots share the global budget. bound = 16 * settings.max_message_bytes + settings.max_channels * CONTROL_BYTES if bound > settings.max_bytes: raise ValueError('Global Metadata pending budget is insufficient.') self.settings, self._factory = settings, sink_factory self._entries, self._stats = {}, {} self._lock = threading.RLock() self._closed = threading.Event() self._thread = None self._metrics = Counter() self.pending_bound = bound def start(self): if self._thread is None: self._thread = threading.Thread(target=self._run, name='metadata-inactive', daemon=True) self._thread.start() def channel(self, camera_id, run_id, generation, state): if state not in ('STOPPED', 'ERROR'): raise ValueError('Parent cannot publish active-channel Metadata.') key = camera_key(camera_id) with self._lock: if key in self._entries: return True if self._closed.is_set() or len(self._entries) >= self.settings.max_channels: return False token = 'idle-' + uuid4().hex sink = self._factory(self.settings, token) try: sink.claim(key, None, token, bootstrap=True) finally: sink.close() status = dict(cameraId=camera_id, runId=run_id, runGeneration=generation, state=state, lastSeq=-1) self._entries[key] = dict(token=token, active=False, status=status) return True def claim(self, source_id, run_id, generation): key = camera_key(source_id) with self._lock: entry = self._entries[key] if entry['active']: raise RuntimeError('Channel writer is already active.') token = 'run-' + uuid4().hex sink = self._factory(self.settings, token) try: sink.claim(key, entry['token'], token) finally: sink.close() self._entries[key] = dict(token=token, active=True, status=dict(cameraId=source_id, runId=run_id, runGeneration=generation, state='STARTING', lastSeq=-1)) counters = ('sent','sent_frames','sent_status','replaced','expired','send_error', 'byte_limit','message_limit','inactive','stale_generation_or_seq','shutdown_drop','close_error') for name in counters: self._metrics[name] += self._stats.get(key, {}).get(name, 0) self._stats[key] = {} return token def retire(self, source_id, token, run_id, generation, failed=False): key = camera_key(source_id) with self._lock: entry = self._entries[key] if entry['token'] != token: raise StaleWriter('Run changed before writer retirement.') new = 'idle-' + uuid4().hex sink = self._factory(self.settings, new) try: sink.claim(key, token, new) finally: sink.close() self._entries[key] = dict(token=new, active=False, status=dict(cameraId=source_id, runId=run_id, runGeneration=generation, state='ERROR' if failed else 'STOPPED', lastSeq=-1)) def update_stats(self, source_id, token, stats): key = camera_key(source_id) with self._lock: if self._entries.get(key, {}).get('token') == token: self._stats[key] = {k:v for k,v in stats.items() if type(v) in (int,float)} def _run(self): while not self._closed.is_set(): with self._lock: pending = [(k, dict(v)) for k,v in self._entries.items() if not v['active']] for key, entry in pending: sink = self._factory(self.settings, entry['token']) try: if len(json.dumps(entry['status']).encode()) > CONTROL_BYTES: raise ValueError('Control status exceeds bound.') sink.write(key, entry['status'], None) with self._lock: self._metrics['sent_status'] += 1 except StaleWriter: with self._lock: self._metrics['fenced_control'] += 1 except Exception: with self._lock: self._metrics['send_error'] += 1 finally: sink.close() self._closed.wait(self.settings.stream_ttl / 3) def metrics(self, include_timeline=False): with self._lock: result = Counter(self._metrics) for stats in self._stats.values(): result.update(stats) values = dict(result) active = sum(v['active'] for v in self._entries.values()) values.update(workerMode='process-channel', workers=active, workerLimit=16, controlWorkers=int(self._thread is not None and self._thread.is_alive()), channels=len(self._entries), pendingByteLimit=self.settings.max_bytes, pendingByteBound=self.pending_bound, childReservedBytes=CHANNEL_BYTES, inflightByteLimit=16*self.settings.max_message_bytes, profile=dict(enabled=False, supported=False, reason='Use per-stream bounded metrics.')) values['sent'] = values.get('sent_frames',0) + values.get('sent_status',0) return values def configure_metrics(self, enabled, seconds=30): if enabled: raise RuntimeError('Process publisher aggregate profiling is unsupported.') return dict(enabled=False, supported=False) def close(self, timeout=1): self._closed.set() if self._thread is not None: self._thread.join(timeout) return self._thread is None or not self._thread.is_alive()