import hashlib import json import math import os import re import stat import threading import time from collections import deque from datetime import datetime, timezone from runtime_security import ( atomic_write_private_json, durable_replace, durable_unlink, ensure_private_directory, fsync_directory, harden_private_file, private_file_ready, read_private_json, reject_reparse_components, require_private_directory, ) from worker_contracts import ( WORKER_EVENT_SCHEMA, WORKER_EVENT_TYPE, PROGRESS_OUTBOX_RELATIVE_PATH, PROGRESS_OUTBOX_SCHEMA, WorkerEvent, WorkerPhase, decode_diagnostic_envelope, decode_worker_event, diagnostic_material_bytes, encode_diagnostic_envelope, encode_worker_event, validate_utc_timestamp, validate_phase_transition, ) STATUS_SCHEMA = 1 HISTORY_SCHEMA = 2 LEGACY_HISTORY_SCHEMA = 1 DIAGNOSTIC_ARCHIVE_SCHEMA = 1 RETENTION_SCHEMA = 1 MAX_EVENT_LINE_BYTES = 256 * 1024 MAX_HISTORY_LINE_BYTES = 2 * 1024 * 1024 MAX_LOG_LINE_BYTES = 256 * 1024 DEFAULT_LOG_BYTES = 2 * 1024 * 1024 DEFAULT_LOG_FILES = 5 DEFAULT_RETENTION_DAYS = 30 DEFAULT_RETENTION_BYTES = 1024 * 1024 * 1024 DEFAULT_EVENT_SEGMENT_BYTES = 8 * 1024 * 1024 DEFAULT_HISTORY_SEGMENT_BYTES = 8 * 1024 * 1024 EVENT_CURSOR_STRIDE = 128 _EVENT_SEGMENT_RE = re.compile( r'^worker-events\.(0|[1-9][0-9]*)-(0|[1-9][0-9]*)\.jsonl$' ) _HISTORY_SEGMENT_RE = re.compile(r'^worker-history\.(0|[1-9][0-9]*)\.jsonl$') _HISTORY_FIELDS = frozenset({ 'schema', 'history_id', 'instance_id', 'slot_id', 'reservation_id', 'source', 'outcome', 'receipt', 'started_at', 'completed_at', 'duration_seconds', 'first_sequence', 'last_sequence', 'diagnostics', 'timeline', 'phase_durations', }) _TIMELINE_FIELDS = frozenset({ 'sequence', 'timestamp', 'instance_id', 'phase', 'progress', }) _LEGACY_TIMELINE_FIELDS = frozenset({'sequence', 'timestamp', 'phase'}) _DIAGNOSTIC_REFERENCE_FIELDS = frozenset({ 'diagnostic_uid', 'record', 'artifacts', }) _ARTIFACT_FIELDS = frozenset({ 'path', 'sha256', 'original_sha256', 'size', 'original_size', 'encoding', 'truncated', }) _SHA256 = re.compile(r'^[0-9a-f]{64}$') class WorkerLocalStateError(RuntimeError): pass def _progress_cursor_value(path): value = read_private_json(path, max_bytes=64 * 1024) if ( not isinstance(value, dict) or set(value) != {'schema', 'sequence'} or value.get('schema') != PROGRESS_OUTBOX_SCHEMA or type(value.get('sequence')) is not int or value['sequence'] < 0 ): raise WorkerLocalStateError('progress outbox cursor is invalid') return value['sequence'] def prepare_progress_outbox_cursor(state_root, *, create=True): root = os.path.abspath(state_root) ensure_private_directory( os.path.join(root, 'control'), reject_reparse=True, ) old_path = os.path.join(root, 'progress-outbox.json') path = os.path.join(root, *PROGRESS_OUTBOX_RELATIVE_PATH.split('/')) old_exists = os.path.exists(old_path) new_exists = os.path.exists(path) old_sequence = _progress_cursor_value(old_path) if old_exists else None new_sequence = _progress_cursor_value(path) if new_exists else None if old_exists and new_exists: if old_sequence != new_sequence: raise WorkerLocalStateError( 'conflicting progress outbox cursors require operator recovery' ) durable_unlink(old_path) return path, new_sequence if old_exists: atomic_write_private_json(path, { 'schema': PROGRESS_OUTBOX_SCHEMA, 'sequence': old_sequence, }, max_bytes=64 * 1024) durable_unlink(old_path) return path, old_sequence if new_exists: return path, new_sequence if create: atomic_write_private_json(path, { 'schema': PROGRESS_OUTBOX_SCHEMA, 'sequence': 0, }, max_bytes=64 * 1024) return path, 0 return path, None def utc_now(): return datetime.now(timezone.utc).isoformat(timespec='milliseconds').replace('+00:00', 'Z') def _canonical_json(value): return json.dumps( value, ensure_ascii=True, sort_keys=True, separators=(',', ':'), allow_nan=False, ).encode('ascii') def _append_private(path, payload, maximum): if type(payload) is not bytes or not payload.endswith(b'\n') or len(payload) > maximum: raise WorkerLocalStateError('append payload is invalid') parent = ensure_private_directory(os.path.dirname(os.path.abspath(path)), reject_reparse=True) existed = os.path.lexists(path) if existed: reject_reparse_components(path) if not private_file_ready(path): raise WorkerLocalStateError('append target is not private') flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT if hasattr(os, 'O_BINARY'): flags |= os.O_BINARY descriptor = os.open(path, flags, 0o600) try: if not existed: harden_private_file(path) with os.fdopen(descriptor, 'ab', buffering=0) as handle: descriptor = None handle.write(payload) os.fsync(handle.fileno()) fsync_directory(parent) finally: if descriptor is not None: os.close(descriptor) def _write_private_bytes(path, payload, maximum=64 * 1024 * 1024): if type(payload) is not bytes or len(payload) > maximum: raise WorkerLocalStateError('artifact payload is invalid') parent = ensure_private_directory(os.path.dirname(os.path.abspath(path)), reject_reparse=True) temporary = f'{path}.{os.getpid()}.{threading.get_ident()}.tmp' descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(descriptor, 'wb') as handle: descriptor = None handle.write(payload) handle.flush() os.fsync(handle.fileno()) harden_private_file(temporary) durable_replace(temporary, path) finally: if descriptor is not None: os.close(descriptor) try: if os.path.exists(temporary): os.remove(temporary) except OSError: pass if not private_file_ready(path): raise WorkerLocalStateError('artifact is not private after publication') fsync_directory(parent) def _event_value(event): return json.loads(encode_worker_event(event).decode('ascii')) def _relative_path(value, field): if not isinstance(value, str) or not value or '\\' in value or value.startswith('/'): raise WorkerLocalStateError(f'history {field} is invalid') if any(part in ('', '.', '..') for part in value.split('/')): raise WorkerLocalStateError(f'history {field} is invalid') return value def _history_timestamp(value, field, optional=False): if optional and value is None: return None try: return validate_utc_timestamp(value, field) except ValueError as exc: raise WorkerLocalStateError(f'history {field} is invalid') from exc def validate_history_record(value): if not isinstance(value, dict): raise WorkerLocalStateError('terminal history shape is invalid') schema = value.get('schema') legacy_without_instance = _HISTORY_FIELDS - {'instance_id'} fields = frozenset(value) if schema == LEGACY_HISTORY_SCHEMA and fields in { _HISTORY_FIELDS, legacy_without_instance, }: legacy = True value = dict(value) value.setdefault('instance_id', 'legacy-instance-unavailable') elif schema == HISTORY_SCHEMA and fields == _HISTORY_FIELDS: legacy = False value = dict(value) else: raise WorkerLocalStateError('terminal history shape is invalid') if type(schema) is not int: raise WorkerLocalStateError('terminal history schema is invalid') for field in ('history_id', 'instance_id', 'outcome'): if not isinstance(value.get(field), str) or not value[field] or len(value[field]) > 256: raise WorkerLocalStateError(f'terminal history {field} is invalid') if type(value.get('slot_id')) is not int or value['slot_id'] < 0: raise WorkerLocalStateError('terminal history slot identity is invalid') if type(value.get('reservation_id')) is not int or value['reservation_id'] <= 0: raise WorkerLocalStateError('terminal history reservation identity is invalid') source = value.get('source') if source is not None and (not isinstance(source, str) or not source): raise WorkerLocalStateError('terminal history source is invalid') if not isinstance(value.get('receipt'), dict): raise WorkerLocalStateError('terminal history receipt is invalid') try: _canonical_json(value['receipt']) except (TypeError, ValueError) as exc: raise WorkerLocalStateError('terminal history receipt is invalid') from exc started_at = _history_timestamp(value.get('started_at'), 'started_at', optional=True) completed_at = _history_timestamp(value.get('completed_at'), 'completed_at') duration = value.get('duration_seconds') if duration is not None and ( type(duration) not in (int, float) or not math.isfinite(duration) or duration < 0 ): raise WorkerLocalStateError('terminal history duration is invalid') first = value.get('first_sequence') last = value.get('last_sequence') if first is not None and (type(first) is not int or first <= 0): raise WorkerLocalStateError('terminal history first sequence is invalid') if last is not None and (type(last) is not int or last <= 0): raise WorkerLocalStateError('terminal history last sequence is invalid') if first is not None and last is not None and last < first: raise WorkerLocalStateError('terminal history sequence range is invalid') diagnostics = value.get('diagnostics') if not isinstance(diagnostics, list) or len(diagnostics) > 32: raise WorkerLocalStateError('terminal history diagnostics are invalid') normalized_diagnostics = [] for reference in diagnostics: if not isinstance(reference, dict) or set(reference) != _DIAGNOSTIC_REFERENCE_FIELDS: raise WorkerLocalStateError('terminal history diagnostic reference is invalid') uid = reference.get('diagnostic_uid') if not isinstance(uid, str) or _SHA256.fullmatch(uid) is None: raise WorkerLocalStateError('terminal history diagnostic identity is invalid') record_path = _relative_path(reference.get('record'), 'diagnostic record path') artifacts = reference.get('artifacts') if not isinstance(artifacts, dict) or set(artifacts) != {'body', 'stdout', 'stderr'}: raise WorkerLocalStateError('terminal history diagnostic artifacts are invalid') normalized_artifacts = {} for name, artifact in artifacts.items(): if artifact is None: normalized_artifacts[name] = None continue if not isinstance(artifact, dict) or set(artifact) != _ARTIFACT_FIELDS: raise WorkerLocalStateError('terminal history diagnostic artifact is invalid') if ( _SHA256.fullmatch(str(artifact.get('sha256') or '')) is None or _SHA256.fullmatch(str(artifact.get('original_sha256') or '')) is None or type(artifact.get('size')) is not int or artifact['size'] < 0 or type(artifact.get('original_size')) is not int or artifact['original_size'] < artifact['size'] or artifact.get('encoding') not in {'text', 'base64'} or type(artifact.get('truncated')) is not bool ): raise WorkerLocalStateError('terminal history diagnostic artifact fields are invalid') normalized_artifacts[name] = { **artifact, 'path': _relative_path(artifact.get('path'), 'diagnostic artifact path'), } normalized_diagnostics.append({ 'diagnostic_uid': uid, 'record': record_path, 'artifacts': normalized_artifacts, }) timeline = value.get('timeline') if not isinstance(timeline, list) or len(timeline) > 1000: raise WorkerLocalStateError('terminal history timeline is invalid') normalized_timeline = [] previous_sequence = 0 for event in timeline: if not isinstance(event, dict): raise WorkerLocalStateError('terminal history timeline event is invalid') event_fields = frozenset(event) if legacy and event_fields == _LEGACY_TIMELINE_FIELDS: legacy_event = True elif event_fields == _TIMELINE_FIELDS: legacy_event = False else: raise WorkerLocalStateError('terminal history timeline event is invalid') sequence = event.get('sequence') if type(sequence) is not int or sequence <= previous_sequence: raise WorkerLocalStateError('terminal history timeline sequence is invalid') previous_sequence = sequence event_instance = value['instance_id'] if legacy_event else event.get('instance_id') if not isinstance(event_instance, str) or not event_instance: raise WorkerLocalStateError('terminal history timeline instance is invalid') try: phase = WorkerPhase(event.get('phase')).value except (TypeError, ValueError) as exc: raise WorkerLocalStateError('terminal history timeline phase is invalid') from exc event_progress = {} if legacy_event else event.get('progress') if not isinstance(event_progress, dict): raise WorkerLocalStateError('terminal history timeline progress is invalid') try: _canonical_json(event_progress) except (TypeError, ValueError) as exc: raise WorkerLocalStateError('terminal history timeline progress is invalid') from exc normalized_timeline.append({ 'sequence': sequence, 'timestamp': _history_timestamp(event.get('timestamp'), 'timeline timestamp'), 'instance_id': event_instance, 'phase': phase, 'progress': dict(event_progress), }) durations = value.get('phase_durations') if not isinstance(durations, dict): raise WorkerLocalStateError('terminal history phase durations are invalid') normalized_durations = {} for phase, seconds in durations.items(): try: phase = WorkerPhase(phase).value except (TypeError, ValueError) as exc: raise WorkerLocalStateError('terminal history phase duration key is invalid') from exc if type(seconds) not in (int, float) or not math.isfinite(seconds) or seconds < 0: raise WorkerLocalStateError('terminal history phase duration is invalid') normalized_durations[phase] = seconds normalized = dict(value) normalized['schema'] = HISTORY_SCHEMA normalized['started_at'] = started_at normalized['completed_at'] = completed_at normalized['diagnostics'] = normalized_diagnostics normalized['timeline'] = normalized_timeline normalized['phase_durations'] = normalized_durations return normalized class WorkerLocalState: """Append-only worker event authority with rebuildable local views.""" def __init__( self, state_root, *, log_bytes=DEFAULT_LOG_BYTES, log_files=DEFAULT_LOG_FILES, retention_days=DEFAULT_RETENTION_DAYS, retention_bytes=DEFAULT_RETENTION_BYTES, read_only=False, event_segment_bytes=DEFAULT_EVENT_SEGMENT_BYTES, history_segment_bytes=DEFAULT_HISTORY_SEGMENT_BYTES, ): self.read_only = bool(read_only) self.root = os.path.abspath(state_root) directory = require_private_directory if self.read_only else ( lambda path, create=False: ensure_private_directory(path, reject_reparse=True) ) self.root = directory(self.root) self.control_dir = directory(os.path.join(self.root, 'control')) self.events_dir = directory(os.path.join(self.root, 'events')) self.history_dir = directory(os.path.join(self.root, 'history')) self.diagnostics_dir = directory(os.path.join(self.root, 'diagnostics')) self.logs_dir = directory(os.path.join(self.root, 'logs')) self.event_path = os.path.join(self.events_dir, 'worker-events.jsonl') self.history_path = os.path.join(self.history_dir, 'worker-history.jsonl') self.status_path = os.path.join(self.control_dir, 'worker.status.json') self.log_path = os.path.join(self.logs_dir, 'worker.log') self.log_bytes = max(64 * 1024, int(log_bytes)) self.log_files = max(1, min(20, int(log_files))) self.retention_days = max(1, int(retention_days)) self.retention_bytes = max(1024 * 1024, int(retention_bytes)) self.event_segment_bytes = max(1024, int(event_segment_bytes)) self.history_segment_bytes = max(1024, int(history_segment_bytes)) self._lock = threading.RLock() self._sequence = 0 self._events = {} self._event_segments = [] self._active_event_first = None self._active_event_last = None self._active_sparse_offsets = {} self._event_end_offset = 0 self._history_segment_index = 0 self._active_history_records = 0 self._segment_readers = {} self._history_keys = set() self.recover() @property def next_sequence(self): with self._lock: return self._sequence + 1 def _require_writer(self): if self.read_only: raise WorkerLocalStateError('local worker state reader cannot mutate authority') @staticmethod def _truncate_partial_final_line(path, maximum, label): if not os.path.exists(path): return reject_reparse_components(path) if not private_file_ready(path): raise WorkerLocalStateError(f'{label} is not private') with open(path, 'r+b') as handle: handle.seek(0, os.SEEK_END) size = handle.tell() if size == 0: return handle.seek(-1, os.SEEK_END) if handle.read(1) == b'\n': return tail_size = min(size, maximum + 1) handle.seek(size - tail_size) tail = handle.read(tail_size) newline = tail.rfind(b'\n') if newline < 0 and size > maximum: raise WorkerLocalStateError(f'truncated {label} line exceeds its bound') boundary = size - tail_size + newline + 1 handle.seek(boundary) handle.truncate() handle.flush() os.fsync(handle.fileno()) def _closed_event_paths(self): values = [] for name in os.listdir(self.events_dir): match = _EVENT_SEGMENT_RE.fullmatch(name) if match: values.append((int(match.group(1)), int(match.group(2)), os.path.join(self.events_dir, name))) return sorted(values) def _closed_history_paths(self): values = [] for name in os.listdir(self.history_dir): match = _HISTORY_SEGMENT_RE.fullmatch(name) if match: values.append((int(match.group(1)), os.path.join(self.history_dir, name))) return sorted(values) @staticmethod def _read_event_line(handle): raw = handle.readline(MAX_EVENT_LINE_BYTES + 1) if not raw: return None if len(raw) > MAX_EVENT_LINE_BYTES: raise WorkerLocalStateError('event journal line exceeds its bound') if not raw.endswith(b'\n'): return False return raw def recover(self): with self._lock: self._sequence = 0 self._events = {} self._event_segments = [] self._active_event_first = None self._active_event_last = None self._active_sparse_offsets = {} self._event_end_offset = 0 if not self.read_only: self._truncate_partial_final_line( self.event_path, MAX_EVENT_LINE_BYTES, 'event journal', ) phases = {} event_paths = self._closed_event_paths() if os.path.exists(self.event_path): event_paths.append((None, None, self.event_path)) for declared_first, declared_last, event_path in event_paths: observed_first = None observed_last = None active = event_path == self.event_path with open(event_path, 'rb') as handle: while True: offset = handle.tell() raw = self._read_event_line(handle) if raw is None: break if raw is False and active and self.read_only: break if raw is False: raise WorkerLocalStateError('event journal line is invalid') event = decode_worker_event(raw[:-1]) if self._sequence and event.sequence != self._sequence + 1: raise WorkerLocalStateError('event journal sequence is not contiguous') if not self._sequence and event.sequence < 1: raise WorkerLocalStateError('event journal sequence is invalid') key = (event.instance_id, event.slot_id) previous = phases.get(key) if previous is not None: validate_phase_transition(previous, event.phase) phases[key] = event.phase self._events[event.slot_id] = event observed_first = observed_first or event.sequence observed_last = event.sequence if active and ( observed_first == event.sequence or (event.sequence - observed_first) % EVENT_CURSOR_STRIDE == 0 ): self._active_sparse_offsets[event.sequence] = offset if active: self._event_end_offset = handle.tell() self._sequence = event.sequence if active: self._active_event_first = observed_first self._active_event_last = observed_last elif ( observed_first != declared_first or observed_last != declared_last or observed_first is None ): raise WorkerLocalStateError('closed event segment identity is invalid') else: self._event_segments.append({ 'first': observed_first, 'last': observed_last, 'path': event_path, 'bytes': os.path.getsize(event_path), }) if not self.read_only: self._truncate_partial_final_line( self.history_path, MAX_HISTORY_LINE_BYTES, 'worker history', ) self._load_history_keys() if not self.read_only: self._write_projection() return self.snapshot() def _load_history_keys(self): self._history_keys = set() closed = self._closed_history_paths() self._history_segment_index = max((index for index, _path in closed), default=0) paths = [(path, False) for _index, path in closed] if os.path.exists(self.history_path): paths.append((self.history_path, True)) self._active_history_records = 0 for path, active in paths: reject_reparse_components(path) if not private_file_ready(path): raise WorkerLocalStateError('worker history is not private') with open(path, 'rb') as handle: while raw := handle.readline(MAX_HISTORY_LINE_BYTES + 1): if len(raw) > MAX_HISTORY_LINE_BYTES: raise WorkerLocalStateError('worker history line exceeds its bound') if not raw.endswith(b'\n'): if active and self.read_only: break raise WorkerLocalStateError('worker history line is invalid') value = self._decode_history_line(raw) key = value['history_id'] if key in self._history_keys: raise WorkerLocalStateError('worker history identity is duplicated') self._history_keys.add(key) if active: self._active_history_records += 1 @staticmethod def _decode_history_line(raw): try: value = json.loads(raw.decode('ascii')) if _canonical_json(value) != raw[:-1]: raise WorkerLocalStateError('worker history is not canonical') return validate_history_record(value) except WorkerLocalStateError: raise except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError) as exc: raise WorkerLocalStateError('worker history is invalid') from exc def _rotate_event_if_needed(self, incoming): if ( self._active_event_first is None or self._event_end_offset + incoming <= self.event_segment_bytes ): return destination = os.path.join( self.events_dir, f'worker-events.{self._active_event_first}-{self._active_event_last}.jsonl', ) if os.path.exists(destination): raise WorkerLocalStateError('event segment destination already exists') durable_replace(self.event_path, destination) self._event_segments.append({ 'first': self._active_event_first, 'last': self._active_event_last, 'path': destination, 'bytes': self._event_end_offset, }) self._active_event_first = None self._active_event_last = None self._active_sparse_offsets = {} self._event_end_offset = 0 def _rotate_history_if_needed(self, incoming): current = os.path.getsize(self.history_path) if os.path.exists(self.history_path) else 0 if self._active_history_records == 0 or current + incoming <= self.history_segment_bytes: return self._history_segment_index += 1 destination = os.path.join( self.history_dir, f'worker-history.{self._history_segment_index}.jsonl', ) if os.path.exists(destination): raise WorkerLocalStateError('history segment destination already exists') durable_replace(self.history_path, destination) self._active_history_records = 0 def _retain_segment_paths(self, paths): for path in paths: self._segment_readers[path] = self._segment_readers.get(path, 0) + 1 def _release_segment_paths(self, paths): with self._lock: for path in paths: count = self._segment_readers.get(path, 0) - 1 if count > 0: self._segment_readers[path] = count else: self._segment_readers.pop(path, None) def _projection(self): slots = [] for slot_id in sorted(self._events): event = self._events[slot_id] slots.append({ 'slot_id': event.slot_id, 'sequence': event.sequence, 'timestamp': event.timestamp, 'instance_id': event.instance_id, 'reservation_id': event.reservation_id, 'source': event.source, 'phase': event.phase.value, 'phase_started_at': event.phase_started_at, 'scan_deadline_at': event.scan_deadline_at, 'assignment_deadline_at': event.assignment_deadline_at, 'progress': dict(event.progress), }) counts = {} for slot in slots: counts[slot['phase']] = counts.get(slot['phase'], 0) + 1 return { 'schema': STATUS_SCHEMA, 'sequence': self._sequence, 'updated_at': max((item['timestamp'] for item in slots), default=None), 'aggregate': {'slot_count': len(slots), 'phases': counts}, 'slots': slots, } def _write_projection(self): atomic_write_private_json(self.status_path, self._projection()) def snapshot(self): with self._lock: return self._projection() def emit_phase( self, instance_id, slot_id, phase, *, reservation_id=None, source=None, phase_started_at=None, scan_deadline_at=None, assignment_deadline_at=None, progress=None, timestamp=None, ): self._require_writer() with self._lock: phase = WorkerPhase(phase) previous = self._events.get(int(slot_id)) if previous is not None and previous.instance_id == str(instance_id): validate_phase_transition(previous.phase, phase) if timestamp is None: timestamp = utc_now() if ( previous is not None and datetime.fromisoformat(timestamp[:-1] + '+00:00') < datetime.fromisoformat(previous.timestamp[:-1] + '+00:00') ): timestamp = previous.timestamp phase_started_at = ( previous.phase_started_at if previous is not None and previous.instance_id == str(instance_id) and previous.phase == phase else (phase_started_at or timestamp) ) event = WorkerEvent( schema=WORKER_EVENT_SCHEMA, sequence=self._sequence + 1, timestamp=timestamp, instance_id=str(instance_id), slot_id=int(slot_id), reservation_id=(int(reservation_id) if reservation_id else None), source=(str(source) if source else None), type=WORKER_EVENT_TYPE, phase=phase, phase_started_at=phase_started_at, scan_deadline_at=scan_deadline_at, assignment_deadline_at=assignment_deadline_at, progress=dict(progress or {}), ) encoded = encode_worker_event(event) self._rotate_event_if_needed(len(encoded) + 1) offset = self._event_end_offset _append_private(self.event_path, encoded + b'\n', MAX_EVENT_LINE_BYTES) self._sequence = event.sequence self._events[event.slot_id] = event if self._active_event_first is None: self._active_event_first = event.sequence self._active_event_last = event.sequence if ( event.sequence == self._active_event_first or (event.sequence - self._active_event_first) % EVENT_CURSOR_STRIDE == 0 ): self._active_sparse_offsets[event.sequence] = offset self._event_end_offset = offset + len(encoded) + 1 self._write_projection() return _event_value(event) def events_after(self, sequence=0, limit=256): sequence = max(0, int(sequence)) limit = max(1, min(1000, int(limit))) if not os.path.exists(self.event_path): with self._lock: if not self._event_segments: return [] with self._lock: if sequence >= self._sequence: return [] segments = [dict(item) for item in self._event_segments] if self._active_event_first is not None: segments.append({ 'first': self._active_event_first, 'last': self._active_event_last, 'path': self.event_path, 'bytes': self._event_end_offset, 'sparse': dict(self._active_sparse_offsets), }) selected = [item for item in segments if item['last'] > sequence] paths = [item['path'] for item in selected] self._retain_segment_paths(paths) results = [] try: for segment in selected: offset = 0 sparse = segment.get('sparse') or {} if sparse: candidates = [item for item in sparse if item <= sequence + 1] if candidates: offset = sparse[max(candidates)] with open(segment['path'], 'rb') as handle: handle.seek(offset) while handle.tell() < segment['bytes']: remaining = segment['bytes'] - handle.tell() raw = handle.readline(min(MAX_EVENT_LINE_BYTES + 1, remaining + 1)) if len(raw) > MAX_EVENT_LINE_BYTES: raise WorkerLocalStateError('event journal line exceeds its bound') if not raw.endswith(b'\n'): break event = decode_worker_event(raw[:-1]) if event.sequence > sequence: results.append(_event_value(event)) if len(results) >= limit: return results finally: self._release_segment_paths(paths) return results def assignment_timeline(self, slot_id, reservation_id, limit=1000): limit = max(1, min(1000, int(limit))) selected = deque(maxlen=limit) if not os.path.exists(self.event_path): with self._lock: if not self._event_segments: return [] with self._lock: segments = [dict(item) for item in self._event_segments] if self._active_event_first is not None: segments.append({ 'first': self._active_event_first, 'last': self._active_event_last, 'path': self.event_path, 'bytes': self._event_end_offset, }) paths = [item['path'] for item in segments] self._retain_segment_paths(paths) try: for segment in segments: with open(segment['path'], 'rb') as handle: while handle.tell() < segment['bytes']: remaining = segment['bytes'] - handle.tell() raw = handle.readline(min(MAX_EVENT_LINE_BYTES + 1, remaining + 1)) if len(raw) > MAX_EVENT_LINE_BYTES: raise WorkerLocalStateError('event journal line exceeds its bound') if not raw.endswith(b'\n'): break event = decode_worker_event(raw[:-1]) if ( event.slot_id == int(slot_id) and event.reservation_id == int(reservation_id) ): selected.append({ 'sequence': event.sequence, 'timestamp': event.timestamp, 'instance_id': event.instance_id, 'phase': event.phase.value, 'progress': dict(event.progress), }) finally: self._release_segment_paths(paths) return list(selected) def append_history(self, value): self._require_writer() record = validate_history_record({'schema': HISTORY_SCHEMA, **dict(value)}) history_id = record['history_id'] payload = _canonical_json(record) + b'\n' with self._lock: if history_id in self._history_keys: return False self._rotate_history_if_needed(len(payload)) _append_private(self.history_path, payload, MAX_HISTORY_LINE_BYTES) self._history_keys.add(history_id) self._active_history_records += 1 return True def history(self, limit=100, reservation_id=None): limit = max(1, min(1000, int(limit))) selected = deque(maxlen=limit) with self._lock: paths = [path for _index, path in self._closed_history_paths()] if os.path.exists(self.history_path): paths.append(self.history_path) self._retain_segment_paths(paths) try: for path in paths: with open(path, 'rb') as handle: while raw := handle.readline(MAX_HISTORY_LINE_BYTES + 1): if len(raw) > MAX_HISTORY_LINE_BYTES: raise WorkerLocalStateError('worker history line exceeds its bound') if not raw.endswith(b'\n'): break value = self._decode_history_line(raw) if reservation_id is None or int(value.get('reservation_id') or 0) == int(reservation_id): rendered = dict(value) rendered['diagnostics'] = [] for diagnostic in value.get('diagnostics') or []: item = dict(diagnostic) record = item.get('record') item['available'] = bool( record and os.path.isfile(os.path.join(self.root, *str(record).split('/'))) ) artifacts = dict(item.get('artifacts') or {}) item['artifact_availability'] = { name: bool( details and details.get('path') and os.path.isfile( os.path.join(self.root, *str(details['path']).split('/')) ) ) for name, details in artifacts.items() } rendered['diagnostics'].append(item) selected.append(rendered) finally: self._release_segment_paths(paths) return list(selected) def _rotate_log(self, incoming): current = os.path.getsize(self.log_path) if os.path.exists(self.log_path) else 0 if current + incoming <= self.log_bytes: return oldest = f'{self.log_path}.{self.log_files}' if os.path.exists(oldest): durable_unlink(oldest) for index in range(self.log_files - 1, 0, -1): source = self.log_path if index == 1 else f'{self.log_path}.{index - 1}' destination = f'{self.log_path}.{index}' if os.path.exists(source): os.replace(source, destination) fsync_directory(self.logs_dir) def log(self, message, *, timestamp=None): self._require_writer() text = str(message).replace('\r', '\\r').replace('\n', '\\n') payload = f'{timestamp or utc_now()} {text}\n'.encode('utf-8', errors='replace') if len(payload) > MAX_LOG_LINE_BYTES: payload = payload[:MAX_LOG_LINE_BYTES - 1] + b'\n' with self._lock: self._rotate_log(len(payload)) _append_private(self.log_path, payload, MAX_LOG_LINE_BYTES) def log_tail(self, limit=100): limit = max(1, min(10000, int(limit))) lines = deque(maxlen=limit) paths = [f'{self.log_path}.{index}' for index in range(self.log_files, 0, -1)] + [self.log_path] with self._lock: for path in paths: if not os.path.exists(path): continue with open(path, 'rb') as handle: while raw := handle.readline(MAX_LOG_LINE_BYTES + 1): if len(raw) > MAX_LOG_LINE_BYTES: raise WorkerLocalStateError('worker log line exceeds its bound') lines.append(raw.decode('utf-8', errors='replace').rstrip('\r\n')) return list(lines) def _artifact(self, directory, name, material, full_payload=None): if material is None and full_payload is None: return None payload = ( bytes(full_payload) if full_payload is not None else diagnostic_material_bytes(material) ) path = os.path.join(directory, name) _write_private_bytes(path, payload) relative = os.path.relpath(path, self.root).replace(os.sep, '/') digest = hashlib.sha256(payload).hexdigest() if full_payload is not None: try: payload.decode('utf-8', errors='strict') encoding = 'text' except UnicodeDecodeError: encoding = 'base64' else: encoding = material.encoding.value return { 'path': relative, 'sha256': digest, 'original_sha256': digest if full_payload is not None else material.sha256, 'size': len(payload), 'original_size': len(payload) if full_payload is not None else material.original_size, 'encoding': encoding, 'truncated': False if full_payload is not None else material.truncated, } def archive_diagnostic(self, envelope, full_materials=None): self._require_writer() full_materials = dict(full_materials or {}) if set(full_materials) - {'body', 'stdout', 'stderr'}: raise WorkerLocalStateError('diagnostic full material shape is invalid') encoded = encode_diagnostic_envelope(envelope) envelope = decode_diagnostic_envelope(encoded) date = envelope.captured_at[:10] directory = ensure_private_directory( os.path.join(self.diagnostics_dir, date, str(envelope.reservation_id)), reject_reparse=True, ) uid = envelope.diagnostic_uid artifacts = { 'body': self._artifact( directory, f'{uid}.body', envelope.http.body if envelope.http else None, full_materials.get('body'), ), 'stdout': self._artifact( directory, f'{uid}.stdout.log', envelope.process.stdout if envelope.process else None, full_materials.get('stdout'), ), 'stderr': self._artifact( directory, f'{uid}.stderr.log', envelope.process.stderr if envelope.process else None, full_materials.get('stderr'), ), } record = { 'schema': DIAGNOSTIC_ARCHIVE_SCHEMA, 'envelope': json.loads(encoded.decode('ascii')), 'artifacts': artifacts, } path = os.path.join(directory, f'{uid}.json') atomic_write_private_json(path, record, max_bytes=2 * 1024 * 1024) return { 'diagnostic_uid': uid, 'record': os.path.relpath(path, self.root).replace(os.sep, '/'), 'artifacts': artifacts, } def diagnostic_references(self, reservation_id): reservation_name = str(int(reservation_id)) references = [] for date in sorted(os.listdir(self.diagnostics_dir)): directory = os.path.join(self.diagnostics_dir, date, reservation_name) if not os.path.isdir(directory) or os.path.islink(directory): continue for name in sorted(os.listdir(directory)): if not name.endswith('.json'): continue path = os.path.join(directory, name) try: value = read_private_json(path, max_bytes=2 * 1024 * 1024) except OSError: continue if not isinstance(value, dict) or set(value) != { 'schema', 'envelope', 'artifacts', } or value.get('schema') != DIAGNOSTIC_ARCHIVE_SCHEMA: raise WorkerLocalStateError('diagnostic archive record is invalid') envelope = value.get('envelope') artifacts = value.get('artifacts') if not isinstance(envelope, dict) or not isinstance(artifacts, dict): raise WorkerLocalStateError('diagnostic archive record is invalid') try: decoded = decode_diagnostic_envelope(_canonical_json(envelope)) except ValueError as exc: raise WorkerLocalStateError('diagnostic archive envelope is invalid') from exc if decoded.reservation_id != int(reservation_id): raise WorkerLocalStateError('diagnostic archive reservation is invalid') reference = { 'diagnostic_uid': decoded.diagnostic_uid, 'record': os.path.relpath(path, self.root).replace(os.sep, '/'), 'artifacts': artifacts, } validate_history_record({ 'schema': HISTORY_SCHEMA, 'history_id': 'diagnostic-reference-validation', 'instance_id': 'diagnostic-reference-validation', 'slot_id': 0, 'reservation_id': int(reservation_id), 'source': None, 'outcome': 'validation', 'receipt': {}, 'started_at': None, 'completed_at': utc_now(), 'duration_seconds': None, 'first_sequence': None, 'last_sequence': None, 'diagnostics': [reference], 'timeline': [], 'phase_durations': {}, }) references.append(reference) return references def _active_reservations(self): reservations = set() reliable = True with os.scandir(self.root) as entries: for entry in entries: if not re.fullmatch(r'slot-(0|[1-9][0-9]*)\.json', entry.name): continue try: state = read_private_json(entry.path, max_bytes=64 * 1024 * 1024) reservation = dict((state.get('assignment') or {}).get('reservation') or {}) reservation_id = int(reservation.get('reservation_id') or 0) if reservation_id > 0: reservations.add(reservation_id) except (OSError, TypeError, ValueError): reliable = False return reservations, reliable def _diagnostic_is_evictable(self, path, active_reservations, reliable): if not reliable: return False relative = os.path.relpath(path, self.diagnostics_dir).split(os.sep) if len(relative) < 3: return True try: reservation_id = int(relative[1]) except ValueError: return False return reservation_id not in active_reservations def _retained_history_authority(self): retained_paths = set() retained_event_sequences = set() paths = [path for _index, path in self._closed_history_paths()] if os.path.exists(self.history_path): paths.append(self.history_path) for path in paths: with open(path, 'rb') as handle: while raw := handle.readline(MAX_HISTORY_LINE_BYTES + 1): if len(raw) > MAX_HISTORY_LINE_BYTES or not raw.endswith(b'\n'): break value = self._decode_history_line(raw) retained_event_sequences.update( event['sequence'] for event in value['timeline'] ) for reference in value['diagnostics']: retained_paths.add(os.path.abspath(os.path.join( self.root, *reference['record'].split('/'), ))) for artifact in reference['artifacts'].values(): if artifact is not None: retained_paths.add(os.path.abspath(os.path.join( self.root, *artifact['path'].split('/'), ))) return retained_paths, retained_event_sequences def _event_segment_is_evictable( self, path, active_reservations, active_reliable, retained_event_sequences, ): if not active_reliable: return False with open(path, 'rb') as handle: while raw := handle.readline(MAX_EVENT_LINE_BYTES + 1): if len(raw) > MAX_EVENT_LINE_BYTES or not raw.endswith(b'\n'): return False event = decode_worker_event(raw[:-1]) if event.reservation_id in active_reservations: return False if ( event.reservation_id is not None and event.sequence not in retained_event_sequences ): return False return True def _evictable_event_prefix( self, active_reservations, active_reliable, retained_event_sequences, progress_cursor, ): paths = set() for _first, last, path in self._closed_event_paths(): if progress_cursor is not None and last > progress_cursor: break if not self._event_segment_is_evictable( path, active_reservations, active_reliable, retained_event_sequences, ): break paths.add(os.path.abspath(path)) return paths def _progress_outbox_cursor(self): old_path = os.path.join(self.root, 'progress-outbox.json') path = os.path.join( self.root, *PROGRESS_OUTBOX_RELATIVE_PATH.split('/'), ) try: old_sequence = ( _progress_cursor_value(old_path) if os.path.exists(old_path) else None ) new_sequence = ( _progress_cursor_value(path) if os.path.exists(path) else None ) except (OSError, ValueError, WorkerLocalStateError): return 0 if old_sequence is not None and new_sequence is not None: return old_sequence if old_sequence == new_sequence else 0 sequence = new_sequence if new_sequence is not None else old_sequence if sequence is not None: return sequence if os.path.exists(self.event_path) or self._closed_event_paths(): return 0 return None def _effective_event_prefix( self, structural_prefix, total_bytes, cutoff, ): remaining_total = int(total_bytes) effective = set() with self._lock: readers = dict(self._segment_readers) for _first, _last, path in self._closed_event_paths(): absolute = os.path.abspath(path) if absolute not in structural_prefix or readers.get(path, 0): break try: details = os.stat(path, follow_symlinks=False) except OSError: break if details.st_mtime >= cutoff and remaining_total <= self.retention_bytes: break effective.add(absolute) remaining_total -= details.st_size return effective def _retention_snapshot(self, extra_roots=None, now=None): now = time.time() if now is None else float(now) cutoff = now - self.retention_days * 86400 roots = { 'control': self.control_dir, 'events': self.events_dir, 'history': self.history_dir, 'diagnostics': self.diagnostics_dir, 'logs': self.logs_dir, } roots.update(dict(extra_roots or {})) categories = {} total_bytes = 0 total_files = 0 total_evictable_bytes = 0 total_evictable_files = 0 active_reservations, active_reliable = self._active_reservations() retained_diagnostics, retained_event_sequences = self._retained_history_authority() progress_cursor = self._progress_outbox_cursor() structural_event_prefix = self._evictable_event_prefix( active_reservations, active_reliable, retained_event_sequences, progress_cursor, ) for name, root in roots.items(): byte_count = 0 file_count = 0 evictable_bytes = 0 evictable_files = 0 if os.path.isdir(root): for current, directories, files in os.walk(root, followlinks=False): directories[:] = [item for item in directories if not os.path.islink(os.path.join(current, item))] for filename in files: path = os.path.join(current, filename) try: details = os.stat(path, follow_symlinks=False) except OSError: continue if stat.S_ISREG(details.st_mode): byte_count += details.st_size file_count += 1 evictable = ( ( self._diagnostic_is_evictable( path, active_reservations, active_reliable, ) and os.path.abspath(path) not in retained_diagnostics ) if name == 'diagnostics' else ( (name == 'logs' and path != self.log_path) ) ) if evictable: evictable_bytes += details.st_size evictable_files += 1 categories[name] = { 'bytes': byte_count, 'files': file_count, 'evictable_bytes': evictable_bytes, 'evictable_files': evictable_files, 'non_evictable_bytes': byte_count - evictable_bytes, 'non_evictable_files': file_count - evictable_files, } total_bytes += byte_count total_files += file_count total_evictable_bytes += evictable_bytes total_evictable_files += evictable_files state_bytes = 0 state_files = 0 with os.scandir(self.root) as entries: for entry in entries: try: details = entry.stat(follow_symlinks=False) except OSError: continue if stat.S_ISREG(details.st_mode): state_bytes += details.st_size state_files += 1 categories['state'] = { 'bytes': state_bytes, 'files': state_files, 'evictable_bytes': 0, 'evictable_files': 0, 'non_evictable_bytes': state_bytes, 'non_evictable_files': state_files, } total_bytes += state_bytes total_files += state_files effective_event_prefix = self._effective_event_prefix( structural_event_prefix, total_bytes, cutoff, ) event_evictable_bytes = 0 event_evictable_files = 0 for path in effective_event_prefix: try: event_evictable_bytes += os.stat(path, follow_symlinks=False).st_size event_evictable_files += 1 except OSError: continue events = categories['events'] events['evictable_bytes'] = event_evictable_bytes events['evictable_files'] = event_evictable_files events['non_evictable_bytes'] = events['bytes'] - event_evictable_bytes events['non_evictable_files'] = events['files'] - event_evictable_files total_evictable_bytes += event_evictable_bytes total_evictable_files += event_evictable_files document = { 'schema': RETENTION_SCHEMA, 'maximum_age_days': self.retention_days, 'maximum_bytes': self.retention_bytes, 'total_bytes': total_bytes, 'total_files': total_files, 'evictable_bytes': total_evictable_bytes, 'evictable_files': total_evictable_files, 'non_evictable_bytes': total_bytes - total_evictable_bytes, 'non_evictable_files': total_files - total_evictable_files, 'over_limit': total_bytes > self.retention_bytes, 'progress_outbox': { 'cursor_sequence': progress_cursor, 'protected_after_sequence': ( progress_cursor if progress_cursor is not None else None ), }, 'categories': categories, } return document, effective_event_prefix def retention_usage(self, extra_roots=None): document, _event_prefix = self._retention_snapshot(extra_roots) return document def cleanup_retention(self, extra_roots=None): self._require_writer() now = time.time() cutoff = now - self.retention_days * 86400 candidates = [] usage, evictable_event_prefix = self._retention_snapshot( extra_roots, now=now, ) active_reservations, active_reliable = self._active_reservations() retained_diagnostics, retained_event_sequences = self._retained_history_authority() for root in ( self.diagnostics_dir, self.logs_dir, self.events_dir, ): for current, directories, files in os.walk(root, topdown=True, followlinks=False): directories[:] = [item for item in directories if not os.path.islink(os.path.join(current, item))] for filename in files: path = os.path.join(current, filename) if path in {self.log_path, self.event_path} or os.path.islink(path): continue if root == self.events_dir and _EVENT_SEGMENT_RE.fullmatch(filename) is None: continue if root == self.diagnostics_dir and not self._diagnostic_is_evictable( path, active_reservations, active_reliable, ): continue if root == self.diagnostics_dir and os.path.abspath(path) in retained_diagnostics: continue if ( root == self.events_dir and os.path.abspath(path) not in evictable_event_prefix ): continue try: details = os.stat(path, follow_symlinks=False) except OSError: continue if stat.S_ISREG(details.st_mode): event_match = _EVENT_SEGMENT_RE.fullmatch(filename) if root == self.events_dir and event_match: kind = 'events' order = int(event_match.group(1)) sort_key = (0, order) else: kind = None order = 0 sort_key = (1, details.st_mtime) candidates.append(( sort_key, details.st_mtime, details.st_size, path, kind, order, )) removed_bytes = 0 removed_files = 0 total_bytes = usage['total_bytes'] blocked_segments = set() with self._lock: for _sort_key, modified, size, path, kind, _order in sorted(candidates): if kind in blocked_segments: continue if modified >= cutoff and total_bytes <= self.retention_bytes: if kind is not None: blocked_segments.add(kind) continue if self._segment_readers.get(path, 0): if kind is not None: blocked_segments.add(kind) continue try: durable_unlink(path) except FileNotFoundError: continue removed_bytes += size removed_files += 1 total_bytes -= size if root_path := next(( segment for segment in self._event_segments if segment['path'] == path ), None): self._event_segments.remove(root_path) return {'removed_bytes': removed_bytes, 'removed_files': removed_files}