#!/usr/bin/env python3 """One cold S4 compile, then the bare/full discovery pair. Never retry or clean.""" import datetime import hashlib import json import os from pathlib import Path import shutil import signal import subprocess import sys import tempfile import time ROOT = Path('/home/reavus/projects/spt-core/spt-core') TREE = ROOT / '.worktrees/304-web-helper-rca' TARGET = TREE / 'target' PROOF = ROOT / '.spt/preserved/304-web-helper-rca' FIXTURE_REL = 'crates/spt-daemon/tests/serve_for_discovery_probe.rs' FIXTURE = TREE / FIXTURE_REL SHA = '35d6f7a38f2189455f1a9e1a85a86b1e4bc0481a' HELPER = ROOT / '.spt/preserved/308-registry-process-lock/linux-validation/xtask-b8482445' HELPER_SHA256 = 'bf5e4aa148d8a528a5641377eb785e6eb711c5be562349498c209745acda38fe' GIB = 1024 ** 3 INITIAL_FREE = 96 * GIB FREE_FLOOR = 32 * GIB GROWTH_LIMIT = 64 * GIB BARE_WITNESS = 'BARE_DISCOVERY_REACHED mode=bare reason=DOCS_PORT_DISCOVERY_FAILED' FULL_WITNESS = 'FULL_REGISTERED_BOUND_PORT url=' COMPILE = ['cargo', 'test', '-p', 'spt-daemon', '--test', 'serve_for_discovery_probe', '--no-run', '--message-format=json'] INHERITED = dict(os.environ) SAFE_KEYS = {'PATH', 'USER', 'LOGNAME', 'SHELL', 'LANG', 'TZ', 'TERM', 'SSL_CERT_FILE', 'SSL_CERT_DIR', 'HTTP_PROXY', 'HTTPS_PROXY', 'ALL_PROXY', 'NO_PROXY', 'http_proxy', 'https_proxy', 'all_proxy', 'no_proxy'} BASE_ENV = {k: v for k, v in INHERITED.items() if k in SAFE_KEYS or k.startswith('LC_')} BASE_ENV.update(PATH='/home/reavus/.cargo/bin:' + INHERITED.get('PATH', '/usr/bin:/bin'), CARGO_HOME=INHERITED.get('CARGO_HOME', '/home/reavus/.cargo'), RUSTUP_HOME=INHERITED.get('RUSTUP_HOME', '/home/reavus/.rustup'), CARGO_BUILD_JOBS='2', CARGO_TARGET_DIR=str(TARGET), CARGO_INCREMENTAL='0', CARGO_TERM_COLOR='never', CI='true') PHASES = [] RECEIPTS = [] PRIVATE_DIRS = [] SOURCE_MANIFEST = None FIXTURE_SHA256 = None def utc(): return datetime.datetime.now(datetime.timezone.utc).isoformat() def save(name, value): path = PROOF / name with path.open('x', encoding='utf-8') as stream: json.dump(value, stream, indent=2) stream.write('\n') RECEIPTS.append(path) return path def sha256(path): digest = hashlib.sha256() with Path(path).open('rb') as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b''): digest.update(chunk) return digest.hexdigest() def git(*args): command = ['git', *args] completed = subprocess.run(command, cwd=TREE, env=BASE_ENV, capture_output=True, timeout=60) if completed.returncode: raise RuntimeError('git command failed: ' + repr(command) + ': ' + completed.stderr.decode(errors='replace')) return completed.stdout def source_guard(): head = git('rev-parse', 'HEAD').decode().strip() records = git('status', '--porcelain=v1', '-z', '--untracked-files=all').split(b'\0') # Any tracked status is forbidden (including staged changes and renames). statuses = [r.decode(errors='replace') for r in records if r] forbidden = [r for r in statuses if r != '?? ' + FIXTURE_REL] state = {'head': head, 'status': statuses, 'fixture_sha256': sha256(FIXTURE)} if head != SHA or forbidden: raise RuntimeError('SOURCE_REFUSED ' + json.dumps(state)) if FIXTURE_SHA256 is not None and state['fixture_sha256'] != FIXTURE_SHA256: raise RuntimeError('FIXTURE_CHANGED ' + json.dumps(state)) return state def source_manifest(): files = {} for raw in git('ls-files', '--stage', '-z').split(b'\0'): if not raw: continue metadata, raw_name = raw.split(b'\t', 1) mode, blob, stage = metadata.decode().split() name = os.fsdecode(raw_name) path = TREE / name if stage != '0' or mode == '160000': raise RuntimeError('Unsupported source entry: ' + name) digest = (hashlib.sha256(os.fsencode(os.readlink(path))).hexdigest() if path.is_symlink() else sha256(path)) files[name] = {'git_mode': mode, 'git_blob': blob, 'sha256': digest} return {'head': SHA, 'git_tree': git('rev-parse', 'HEAD^{tree}').decode().strip(), 'tracked_files': files, 'fixture': {'path': FIXTURE_REL, 'sha256': sha256(FIXTURE)}} def path_belongs(value): return value == str(ROOT) or value.startswith(str(ROOT) + '/') def census(): rows = {} unreadable = [] for directory in Path('/proc').iterdir(): if not directory.name.isdigit(): continue pid = int(directory.name) try: stat = (directory / 'stat').read_text() fields = stat.rsplit(') ', 1)[1].split() row = {'pid': pid, 'ppid': int(fields[1]), 'pgid': int(fields[2]), 'session': int(fields[3]), 'start_ticks': fields[19], 'state': fields[0], 'comm': (directory / 'comm').read_text().strip(), 'argv': [os.fsdecode(v) for v in (directory / 'cmdline').read_bytes().split(b'\0') if v]} errors = [] for field in ('cwd', 'exe'): try: row[field] = os.readlink(directory / field) except OSError as error: row[field] = None errors.append(field + ': ' + str(error)) row['inspection_errors'] = errors rows[pid] = row except FileNotFoundError: continue except OSError as error: unreadable.append({'pid': pid, 'error': str(error)}) active, own = [], [] for row in rows.values(): ancestry, seen = [], {row['pid']} parent = row['ppid'] while parent in rows and parent not in seen: seen.add(parent) ancestry.append(parent) parent = rows[parent]['ppid'] row['ancestor_pids'] = ancestry evidence = [] for related in [row] + [rows[pid] for pid in ancestry]: for key in ('cwd', 'exe'): if path_belongs(related[key] or ''): evidence.append({'pid': related['pid'], 'field': key, 'value': related[key]}) for arg in related['argv']: # Also cover --manifest-path=/project/... arguments. if path_belongs(arg.split('=', 1)[-1]): evidence.append({'pid': related['pid'], 'field': 'argv', 'value': arg}) row['project_evidence'] = evidence executable = row['exe'] or '' names = [row['comm'], Path(executable).name] if row['argv']: names.append(Path(row['argv'][0]).name) tool = any(n.startswith(('cargo', 'rustc', 'rust-analyzer', 'nextest')) for n in names) test = ('/deps/' in executable or '/target/' in executable or '--exact' in row['argv'] or any(n in ('test', 'pytest', 'nextest') for n in names)) running = row['state'] != 'Z' if running and (tool or test) and (evidence or row['inspection_errors']): active.append(row['pid']) if running and executable.startswith(str(TARGET) + '/'): own.append(row['pid']) return {'utc': utc(), 'free_bytes': shutil.disk_usage(TREE).free, 'load': os.getloadavg(), 'processes': list(rows.values()), 'active_project_producer_pids': active, 'own_target_survivor_pids': own, 'unreadable_processes': unreadable} def target_size(): allocated, apparent, seen = 0, 0, set() for root, dirs, names in os.walk(TARGET, followlinks=False): for name in dirs + names: try: st = (Path(root) / name).lstat() except FileNotFoundError: continue identity = (st.st_dev, st.st_ino) if identity not in seen: seen.add(identity) allocated += st.st_blocks * 512 apparent += st.st_size return {'allocated_bytes': allocated, 'apparent_bytes': apparent, 'growth_bytes': max(allocated, apparent)} def resources(): return {'utc': utc(), 'free_bytes': shutil.disk_usage(TREE).free, **target_size()} def admit(label): state = {'source': source_guard(), 'census': census(), 'resources': resources()} save(label + '-admission.json', state) current = state['resources'] if state['census']['active_project_producer_pids'] or state['census']['unreadable_processes']: raise RuntimeError('ACTIVE_OR_UNATTRIBUTABLE_PRODUCER_REFUSED ' + label) if current['free_bytes'] < INITIAL_FREE: raise RuntimeError('PRODUCER_ADMISSION_FREE_BELOW_96_GIB ' + label) if current['free_bytes'] <= FREE_FLOOR or current['growth_bytes'] >= GROWTH_LIMIT: raise RuntimeError('RESOURCE_ADMISSION_REFUSED ' + label) def private_environment(label, mode=None): root = Path(tempfile.mkdtemp(prefix='spt-304-pair-' + label + '-', dir='/tmp')) PRIVATE_DIRS.append(str(root)) environment = dict(BASE_ENV) for name in ('home', 'spt-home', 'tmp', 'config', 'cache', 'data', 'runtime'): (root / name).mkdir(mode=0o700) environment.update(HOME=str(root / 'home'), SPT_HOME=str(root / 'spt-home'), TMPDIR=str(root / 'tmp'), TMP=str(root / 'tmp'), TEMP=str(root / 'tmp'), XDG_CONFIG_HOME=str(root / 'config'), XDG_CACHE_HOME=str(root / 'cache'), XDG_DATA_HOME=str(root / 'data'), XDG_RUNTIME_DIR=str(root / 'runtime')) if mode is not None: environment['SPT_WEB_HELPER_REPRO_MODE'] = mode return environment def run(label, argv, timeout, environment, admission=True, disk_guard=True): if admission: admit(label) started = time.monotonic() row = {'label': label, 'argv': argv, 'cwd': str(TREE), 'start_utc': utc(), 'timeout_seconds': timeout, 'native_exit': None, 'forced_terminations': [], 'environment_keys': sorted(environment), 'environment_overrides': {k: environment[k] for k in environment if k not in SAFE_KEYS and not k.startswith('LC_')}, 'resource_samples': [], 'error': None} stdout = PROOF / (label + '.stdout.log') stderr = PROOF / (label + '.stderr.log') process = None interrupted = None def terminate(reason): # Only this child's freshly created process group; never census-discovered PIDs. for sig, grace in ((signal.SIGTERM, 5), (signal.SIGKILL, 5)): try: os.killpg(process.pid, sig) row['forced_terminations'].append({'utc': utc(), 'pgid': process.pid, 'signal': sig.name, 'reason': reason}) except ProcessLookupError: break try: process.wait(timeout=grace) except subprocess.TimeoutExpired: continue # A leader can exit while children remain. Kill only that same owned group. if sig == signal.SIGTERM: time.sleep(0.2) continue break try: with stdout.open('xb') as out, stderr.open('xb') as err: process = subprocess.Popen(argv, cwd=TREE, env=environment, stdout=out, stderr=err, start_new_session=True) row['pid'] = process.pid while process.poll() is None: sample = resources() row['resource_samples'].append(sample) reason = None if time.monotonic() - started >= timeout: reason = 'timeout' elif disk_guard and sample['free_bytes'] <= FREE_FLOOR: reason = 'free_space_at_or_below_32_gib' elif disk_guard and sample['growth_bytes'] >= GROWTH_LIMIT: reason = 'target_growth_at_or_above_64_gib' if reason: terminate(reason) break time.sleep(2) except BaseException as error: row['error'] = repr(error) if process is not None and process.poll() is None: terminate('driver_exception') if isinstance(error, (KeyboardInterrupt, SystemExit)): interrupted = error finally: row['native_exit'] = process.poll() if process is not None else None row['end_utc'] = utc() row['elapsed_seconds'] = round(time.monotonic() - started, 3) for name, path in (('stdout', stdout), ('stderr', stderr)): row[name] = {'path': str(path), 'sha256': sha256(path) if path.exists() else None} if path.exists(): RECEIPTS.append(path) PHASES.append(row) save(label + '-receipt.json', row) if interrupted is not None: raise interrupted return row def completed(row, exit_code=0): return row['native_exit'] == exit_code and not row['forced_terminations'] and not row['error'] def witnessed(row, text): needle = text.encode() for key in ('stdout', 'stderr'): carry = b'' with Path(row[key]['path']).open('rb') as stream: for chunk in iter(lambda: stream.read(65536), b''): data = carry + chunk if needle in data: return True carry = data[-len(needle):] return False def compiled_executable(build): candidates = set() with Path(build['stdout']['path']).open(encoding='utf-8') as stream: for line in stream: try: event = json.loads(line) except json.JSONDecodeError: continue if (event.get('reason') == 'compiler-artifact' and event.get('target', {}).get('name') == 'serve_for_discovery_probe' and event.get('profile', {}).get('test') is True and event.get('executable')): candidates.add(event['executable']) if len(candidates) != 1: raise RuntimeError('Expected exactly one matching test executable: ' + repr(candidates)) executable = Path(candidates.pop()) if (not executable.is_absolute() or executable.is_symlink() or not str(executable.resolve()).startswith(str(TARGET) + '/') or not executable.is_file() or not os.access(executable, os.X_OK)): raise RuntimeError('Executable is not a real executable in own target: ' + str(executable)) return executable def main(): global SOURCE_MANIFEST, FIXTURE_SHA256 if sys.platform != 'linux': raise RuntimeError('Linux only; no Windows producer is authorized') if Path(__file__).resolve().parent != PROOF: raise RuntimeError('Driver must be placed in ' + str(PROOF)) # An exclusive persistent marker refuses reruns, including interrupted attempts. with (PROOF / 'pair-start.json').open('x', encoding='utf-8') as stream: json.dump({'utc': utc(), 'pid': os.getpid(), 'driver_sha256': sha256(__file__)}, stream) RECEIPTS.append(PROOF / 'pair-start.json') result = {'expected': {'bare_exit': 101, 'full_exit': 0, 'bare_substring': BARE_WITNESS, 'full_substring': FULL_WITNESS}, 'source_sha': SHA, 'start_utc': utc(), 'success': False, 'classification': 'preflight_failed', 'phases': PHASES, 'private_directories_retained': PRIVATE_DIRS, 'pool_release': None} claim_attempted = False try: if TREE.resolve() != TREE or not TARGET.is_dir() or TARGET.is_symlink() or TARGET.resolve() != TARGET: raise RuntimeError('Own tree/target must be real directories, not symlinks') foreign = INHERITED.get('CARGO_TARGET_DIR') if foreign is not None and foreign != str(TARGET): raise RuntimeError('Inherited foreign CARGO_TARGET_DIR refused: ' + repr(foreign)) if any(TARGET.iterdir()): raise RuntimeError('Own target must already exist and be empty') admit('initial') if sha256(HELPER) != HELPER_SHA256: raise RuntimeError('Pool helper SHA256 mismatch') FIXTURE_SHA256 = sha256(FIXTURE) SOURCE_MANIFEST = source_manifest() manifest = save('source-manifest.json', SOURCE_MANIFEST) result['source_manifest_sha256'] = sha256(manifest) result['fixture_sha256'] = FIXTURE_SHA256 result['helper'] = {'path': str(HELPER), 'sha256': HELPER_SHA256} save('environment-policy.json', {'policy': 'allowlist; all other inherited variables discarded', 'retained_keys': sorted(BASE_ENV), 'discarded_keys': sorted(set(INHERITED) - set(BASE_ENV)), 'cargo_target_dir': str(TARGET), 'cargo_build_jobs': 2}) control_env = private_environment('pool') result['classification'] = 'pool_claim_failed' admit('pool-claim') # Release even after a partial/failed claim: the helper verifies owner cwd. claim_attempted = True claim = run('pool-claim', [str(HELPER), 'pool-claim', '--pool', str(TARGET), '--label', 'todlando-304-web-helper-rca'], 60, control_env, admission=False) if not completed(claim): return result result['classification'] = 'build_failed_not_probe_red' build = run('compile', COMPILE, 3600, private_environment('compile')) if not completed(build): return result result['classification'] = 'build_artifact_invalid' executable = compiled_executable(build) result['executable'] = {'path': str(executable), 'sha256': sha256(executable)} argv = [str(executable), '--exact', 'serve_for_discovery_probe', '--nocapture'] result['classification'] = 'bare_producer_failed' bare = run('bare', argv, 120, private_environment('bare', 'bare')) result['bare_witness'] = witnessed(bare, BARE_WITNESS) if bare['forced_terminations'] or bare['error'] or bare['native_exit'] is None: return result # An unexpected bare result still gets its one full control, never a retry. result['classification'] = 'full_control_failed' full = run('full', argv, 120, private_environment('full', 'full')) result['full_witness'] = witnessed(full, FULL_WITNESS) if not completed(full) or not result['full_witness']: return result result['classification'] = ('expected_bare101_full0' if completed(bare, 101) and result['bare_witness'] else 'bare_did_not_match_expected_red') result['success'] = result['classification'] == 'expected_bare101_full0' return result except BaseException as error: result['error'] = repr(error) result['success'] = False return result finally: if claim_attempted: try: result['pool_release'] = run('pool-release', [str(HELPER), 'pool-release', '--pool', str(TARGET)], 60, control_env, admission=False, disk_guard=False) if not completed(result['pool_release']): result['success'] = False result['pool_release_failed'] = True except BaseException as error: result['success'] = False result['pool_release_error'] = repr(error) try: final = census() result['final_census'] = str(save('final-census.json', final)) result['final_resources'] = resources() result['final_source'] = source_guard() if SOURCE_MANIFEST is not None and source_manifest() != SOURCE_MANIFEST: raise RuntimeError('Final source content differs from initial hash manifest') if final['active_project_producer_pids'] or final['own_target_survivor_pids'] or final['unreadable_processes']: raise RuntimeError('Final producer census is not clear; no census PID was killed') except BaseException as error: result['success'] = False result['finalization_error'] = repr(error) result['end_utc'] = utc() result['hash_manifest'] = str(save('receipt-hashes.json', {str(p): sha256(p) for p in RECEIPTS})) save('result.json', result) print(json.dumps({'success': result['success'], 'classification': result['classification'], 'result': str(PROOF / 'result.json')}), flush=True) def interrupted(signum, frame): raise KeyboardInterrupt('driver received ' + signal.Signals(signum).name) if __name__ == '__main__': signal.signal(signal.SIGTERM, interrupted) signal.signal(signal.SIGHUP, interrupted) outcome = main() sys.exit(0 if outcome['success'] else 1)