From 4297da4f06755b8a2a0e95f960a0bb82546166d3 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 11:39:25 +0000 Subject: [PATCH] scripts: keep the qualification harness within its open-file limit (P1R-C06) The recovery rehearsal on the qualification host ran out of open files under Terminal's default limit of 256: every gateway left its three pipes and its transcript open for the rest of the run, so the 27th functional scenario and every later one failed with "Too many open files", and a journal read at the load interval's drain then ended the run. - The readers of a gateway's output and log close its pipes when it ends, and end_gateway, close and the recovery bench's teardown wait for them. - The fixture authority's reader closes its output; kill and stop close its input and log and wait for the reader. - The evidence writer keeps at most 32 files open, closing the least recently written first and appending again when it reopens one. A recovery interval writes about 17 files at once, so it closes none of them. - The parent's copies of two worker logs are closed after the start. - A journal that cannot be read at the drain makes the interval inconclusive, as the protocol says, instead of ending the run. - Each report records the open-file limit the harness and what it started ran under. Co-Authored-By: Claude Opus 5.5 (1M context) --- scripts/qualify-devguard.py | 44 ++++++---- scripts/qualify_host.py | 101 ++++++++++++++++------ scripts/tests/test_qualify_devguard.py | 115 ++++++++++++++++++++++++- 3 files changed, 215 insertions(+), 45 deletions(-) diff --git a/scripts/qualify-devguard.py b/scripts/qualify-devguard.py index ec1b65b..467e735 100644 --- a/scripts/qualify-devguard.py +++ b/scripts/qualify-devguard.py @@ -39,6 +39,7 @@ from pathlib import Path import platform import re +import resource import secrets import shutil import signal @@ -727,9 +728,10 @@ def scenario_authority_unavailable(bench): def scenario_runner_socket(bench): """`uds` with a worker the gateway did not start: governed executions are unsupported.""" socket = bench.work / 'standalone.sock' - worker = subprocess.Popen([str(bench.args.runtime_bin), str(socket)], stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, stderr=open(bench.evidence.file(f'{bench.mode}/standalone-worker.log'), 'ab'), - env=host.minimal_env(bench.work), start_new_session=True) + with open(bench.evidence.file(f'{bench.mode}/standalone-worker.log'), 'ab') as log: + worker = subprocess.Popen([str(bench.args.runtime_bin), str(socket)], stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=log, env=host.minimal_env(bench.work), + start_new_session=True) bench.live.workers.append(worker) deadline = time.monotonic() + 15 while not socket.exists() and time.monotonic() < deadline and worker.poll() is None: @@ -2258,7 +2260,7 @@ def teardown(self): if gateway.process.poll() is None: gateway.process.kill() gateway.process.wait() - gateway.log_reader.join(timeout=10) + gateway.join_readers() stopped = [runner.stop(limit=20) for runner in self.runners if runner.alive()] killed = kill_leftovers(self.roots.values(), self.began_unix) self.evidence.append(f'{self.prefix}/teardown.jsonl', {'runners_stopped': stopped, 'leftovers_killed': killed}) @@ -2667,10 +2669,10 @@ def recovery_invalid_settings(bench): managed_dir = bench.work / 'mw' managed_dir.mkdir(mode=0o700) managed_socket = managed_dir / 'runner.sock' - worker = subprocess.Popen([str(bench.args.runtime_bin), str(managed_socket)], stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=open(bench.evidence.file(f'{bench.prefix}/managed-worker.log'), 'ab'), - env=host.minimal_env(bench.work), start_new_session=True) + with open(bench.evidence.file(f'{bench.prefix}/managed-worker.log'), 'ab') as log: + worker = subprocess.Popen([str(bench.args.runtime_bin), str(managed_socket)], stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=log, env=host.minimal_env(bench.work), + start_new_session=True) bench.live.workers.append(worker) # A managed worker serves the first connection it accepts and no other, so it is not probed. deadline = time.monotonic() + 15 @@ -4574,14 +4576,18 @@ def drain(self): found = re.findall(r'stopped; (\d+) processes were terminated', self.runner.log()) drain['terminated'] = int(found[-1]) if found else None if self.governed: - drain['retired'], drain['instances'] = wait_retired(self.bench, self.runner.pid) - deadline = time.monotonic() + 60 - while True: - charged = host.journal_charged(self.args.journal)['attempts'].get(host.CONSUMER, 0) - if not charged or time.monotonic() >= deadline: - break - time.sleep(1.0) - drain['charged'] = charged + try: + drain['retired'], drain['instances'] = wait_retired(self.bench, self.runner.pid) + deadline = time.monotonic() + 60 + while True: + charged = host.journal_charged(self.args.journal)['attempts'].get(host.CONSUMER, 0) + if not charged or time.monotonic() >= deadline: + break + time.sleep(1.0) + drain['charged'] = charged + except (OSError, sqlite3.Error) as error: + drain['journal_error'] = f'{type(error).__name__}: {error}' + self.invalid.append(f"the journal could not be read: {drain['journal_error']}") return drain @@ -4952,8 +4958,10 @@ def verify_recovery(directory, problems): def harness(): - """The harness that ran: its files' SHA-256 and the CodeSpace commit.""" - found = {'files': {name: host.sha256(ROOT / 'scripts' / name) for name in HARNESS}} + """The harness that ran: its files' SHA-256, the CodeSpace commit, and the open-file limit it + and every process it started ran under (soft and hard; a macOS terminal's soft limit is 256).""" + found = {'files': {name: host.sha256(ROOT / 'scripts' / name) for name in HARNESS}, + 'open_files_limit': list(resource.getrlimit(resource.RLIMIT_NOFILE))} try: found.update(commit=git('rev-parse', 'HEAD'), dirty=dirty()) except (OSError, subprocess.CalledProcessError): diff --git a/scripts/qualify_host.py b/scripts/qualify_host.py index 11b5ba5..26b7b8b 100644 --- a/scripts/qualify_host.py +++ b/scripts/qualify_host.py @@ -8,6 +8,7 @@ run in a minimal environment (HOME, PATH, TMPDIR, LANG and CodeSpace's own settings) in their own sessions, so a terminal's interrupt reaches only the harness, which then settles what it started. """ +from collections import OrderedDict import hashlib import json import os @@ -40,12 +41,28 @@ NO_ANSWER = 'no_answer' # Codes that are the transport's, not CodeSpace's answer. TRANSPORT_CODES = (TRANSPORT, NO_ANSWER) +# The evidence writer keeps at most this many files open, closing the least recently written +# first, so that a run that writes to many files stays far below the open-file limit (256 by +# default in a macOS terminal). A recovery interval writes about 17 files at once, which all +# stay open. +WRITER_OPEN_FILES = 32 def default_socket(): return Path(f'/private/tmp/devguard-{os.getuid()}/authority.sock') +def close_quietly(handle): + """Close a pipe or file that may be closed already or whose other end is gone; the + descriptor is released either way.""" + if handle is None: + return + try: + handle.close() + except (OSError, ValueError): + pass + + def sha256(path): digest = hashlib.sha256() with open(path, 'rb') as handle: @@ -136,12 +153,13 @@ def verify_manifest(directory): class Writer(threading.Thread): """Appends JSON lines on its own thread, as DevGuard's measure.py does, so no sampler waits on - a disk write: a disk the load saturates delays the evidence, never a sample.""" + a disk write: a disk the load saturates delays the evidence, never a sample. It keeps at most + WRITER_OPEN_FILES files open and opens a closed one again to append to it.""" def __init__(self): super().__init__(daemon=True) self.lines = queue.Queue() - self.files = {} + self.files = OrderedDict() self.errors = [] self.written = 0 self.lag_max = 0.0 @@ -164,10 +182,7 @@ def run(self): batch = [item for item in batch if item is not None] for _, path, line in batch: try: - if path not in self.files: - path.parent.mkdir(parents=True, exist_ok=True) - self.files[path] = open(path, 'a') - self.files[path].write(line) + self._file(path).write(line) self.written += 1 except OSError as error: self.errors.append(f'{path.name}: {error}') @@ -179,6 +194,22 @@ def run(self): if batch: self.lag_max = max(self.lag_max, time.monotonic() - batch[0][0]) + def _file(self, path): + """The open file for `path`, the most recently written last; the least recently written + one is closed when too many are open.""" + if path in self.files: + self.files.move_to_end(path) + return self.files[path] + while len(self.files) >= WRITER_OPEN_FILES: + name, handle = self.files.popitem(last=False) + try: + handle.close() + except OSError as error: + self.errors.append(f'{name.name}: {error}') + path.parent.mkdir(parents=True, exist_ok=True) + self.files[path] = open(path, 'a') + return self.files[path] + def close(self, limit=120): """Write every line handed over and close the files; a disk that does not take them within `limit` seconds is reported, never waited for without end.""" @@ -288,6 +319,11 @@ def _read(self): self.pending.clear() for waiter in waiters: waiter['event'].set() + # The connection is over, so this side of both pipes is closed: a run of many gateways + # must not keep a descriptor of each. + close_quietly(self.process.stdout) + with self.write_lock: + close_quietly(self.process.stdin) def _answer_server(self, message): if 'id' not in message: @@ -461,14 +497,22 @@ def __init__(self, binary, flags, env, log_path, record=None, cwd=None): self.info = None def _log(self): - with open(self.log_path, 'ab') as log: - for line in self.process.stderr: - log.write(line) - log.flush() - entry = launch_line(line.decode('utf-8', 'replace')) - if entry: - with self.lock: - self.executions.setdefault(entry['process_id'], []).append(entry) + try: + with open(self.log_path, 'ab') as log: + for line in self.process.stderr: + log.write(line) + log.flush() + entry = launch_line(line.decode('utf-8', 'replace')) + if entry: + with self.lock: + self.executions.setdefault(entry['process_id'], []).append(entry) + finally: + close_quietly(self.process.stderr) + + def join_readers(self, limit=10): + """Wait for the readers of the gateway's output and log, which close its pipes as they end.""" + self.log_reader.join(timeout=limit) + self.client.reader.join(timeout=limit) def initialize(self, timeout=60): result, latency = self.client.request('initialize', { @@ -599,7 +643,7 @@ def close(self, limit=60): stopped = 'SIGKILL' self.process.kill() self.process.wait() - self.log_reader.join(timeout=10) + self.join_readers() return {'returncode': self.process.returncode, 'stopped': stopped, 'malformed_lines': self.client.malformed} @@ -631,7 +675,7 @@ def end_gateway(gateway, how, limit=60): stopped = 'SIGKILL' gateway.process.kill() gateway.process.wait() - gateway.log_reader.join(timeout=10) + gateway.join_readers() return {'how': how, 'returncode': gateway.process.returncode, 'stopped': stopped, 'exit_ms': round((time.monotonic() - began) * 1000, 3)} @@ -1200,16 +1244,22 @@ def __init__(self, binary, base, env, log_path, limit=2, timeout=60): stdout=subprocess.PIPE, stderr=self.log, env=env, text=True, start_new_session=True) self.lines = queue.Queue() - threading.Thread(target=self._read, daemon=True).start() + self.reader = threading.Thread(target=self._read, daemon=True) + self.reader.start() self.settings = self._answer(timeout) if self.settings is None: self.kill() raise RuntimeError('the fixture authority did not start') def _read(self): - for line in self.process.stdout: - self.lines.put(line) - self.lines.put(None) + try: + for line in self.process.stdout: + self.lines.put(line) + except (OSError, ValueError): + pass + finally: + close_quietly(self.process.stdout) + self.lines.put(None) def _answer(self, timeout): try: @@ -1237,15 +1287,16 @@ def kill(self): if self.process.poll() is None: self.process.kill() self.process.wait() + close_quietly(self.process.stdin) + close_quietly(self.log) + self.reader.join(timeout=10) def stop(self, limit=30): - try: - self.process.stdin.close() - except OSError: - pass + close_quietly(self.process.stdin) try: self.process.wait(timeout=limit) except subprocess.TimeoutExpired: self.kill() - self.log.close() + close_quietly(self.log) + self.reader.join(timeout=10) return self.process.returncode diff --git a/scripts/tests/test_qualify_devguard.py b/scripts/tests/test_qualify_devguard.py index b1c689e..32b8e15 100644 --- a/scripts/tests/test_qualify_devguard.py +++ b/scripts/tests/test_qualify_devguard.py @@ -8,6 +8,7 @@ import math import os from pathlib import Path +import resource import socket import sqlite3 import struct @@ -317,6 +318,21 @@ def test_writer(self): lines = (self.dir / 'w/rows.jsonl').read_text().splitlines() self.assertEqual([json.loads(line)['n'] for line in lines], list(range(100))) + def test_the_writer_keeps_few_files_open(self): + writer = host.Writer() + paths = [self.dir / f'w/{n:02d}.jsonl' for n in range(3 * host.WRITER_OPEN_FILES)] + for number in range(4): + for path in paths: + writer.write(path, {'round': number}) + deadline = time.monotonic() + 30 + while writer.written < 4 * len(paths) and time.monotonic() < deadline: + time.sleep(0.01) + self.assertLessEqual(len(writer.files), host.WRITER_OPEN_FILES) + report = writer.close() + self.assertEqual((report['lines'], report['error_count']), (4 * len(paths), 0)) + for path in paths: + self.assertEqual([json.loads(line)['round'] for line in path.read_text().splitlines()], [0, 1, 2, 3]) + def journal(self): database = self.dir / 'authority.sqlite' connection = sqlite3.connect(database) @@ -410,8 +426,6 @@ def setUp(self): def tearDown(self): self.gateway.close(limit=10) - self.gateway.process.stdout.close() - self.gateway.process.stderr.close() self.scratch.cleanup() def test_answers_errors_and_server_requests(self): @@ -438,6 +452,82 @@ def test_no_answer_and_a_lost_connection(self): self.assertTrue(raised.exception.transport) +FAKE_FIXTURE = textwrap.dedent(""" + import json, sys + print(json.dumps({'base': sys.argv[1]}), flush=True) + for line in sys.stdin: + print(json.dumps({'command': line.strip()}), flush=True) +""") + + +def open_descriptors(expected=None, limit=5.0): + """The test process's open descriptors; with `expected`, once they are that many or `limit` + seconds have passed, so that another test's thread that is still ending cannot make a leak + seem to come or go.""" + deadline = time.monotonic() + limit + while True: + count = len(os.listdir('/dev/fd')) + if expected is None or count == expected or time.monotonic() >= deadline: + return count + time.sleep(0.05) + + +class DescriptorTests(unittest.TestCase): + """Gateways and fixture authorities that ended leave none of their pipes open in the harness, + which starts hundreds of them in one run under a terminal's limit of 256 open files and keeps + each one referenced until the run settles.""" + + def setUp(self): + self.scratch = tempfile.TemporaryDirectory() + self.dir = Path(self.scratch.name) + self.server = self.dir / 'server.py' + self.server.write_text(FAKE_SERVER) + + def tearDown(self): + self.scratch.cleanup() + + def gateway(self, name): + gateway = host.Gateway(sys.executable, [str(self.server)], dict(os.environ), self.dir / f'{name}.log') + gateway.initialize(timeout=30) + return gateway + + def test_ended_gateways_close_their_pipes(self): + started = [self.gateway('first')] + started[0].close(limit=10) + before = open_descriptors() + for _ in range(3): + for how in host.GATEWAY_ENDS: + started.append(self.gateway(how)) + self.assertIsNone(host.end_gateway(started[-1], how, limit=10)['stopped']) + started.append(self.gateway('killed')) + started[-1].process.kill() + started[-1].process.wait() + started[-1].join_readers() + started.append(self.gateway('closed')) + started[-1].close(limit=10) + self.assertEqual(open_descriptors(before), before) + + def test_stopped_fixture_authorities_close_their_pipes(self): + binary = self.dir / 'fixture' + binary.write_text(f'#!{sys.executable}\n' + FAKE_FIXTURE) + binary.chmod(0o755) + + def start(name): + return host.FixtureAuthority(binary, self.dir / name, dict(os.environ), self.dir / f'{name}.log') + started = [start('first')] + started[0].stop() + before = open_descriptors() + for number in range(5): + started.append(start(f'fixture-{number}')) + self.assertEqual(started[-1].settings, {'base': str(self.dir / f'fixture-{number}')}) + self.assertEqual(started[-1].charged(), {'command': 'charged'}) + if number % 2: + started[-1].kill() + else: + self.assertEqual(started[-1].stop(), 0) + self.assertEqual(open_descriptors(before), before) + + class HarnessTests(unittest.TestCase): def setUp(self): self.scratch = tempfile.TemporaryDirectory() @@ -481,6 +571,9 @@ def test_consumer_settings_never_return_a_digest(self): self.assertNotIn('a' * 64, json.dumps(settings)) self.assertEqual(qualify.consumer_settings('[consumers.dev-cli]\nrole = "workload"\n'), {}) + def test_the_harness_records_its_open_file_limit(self): + self.assertEqual(qualify.harness()['open_files_limit'], list(resource.getrlimit(resource.RLIMIT_NOFILE))) + def test_documentation_keeps_the_candidate(self): repo = self.dir / 'repo' (repo / 'docs').mkdir(parents=True) @@ -1152,6 +1245,24 @@ def judged(change): self.assertEqual(analysis.recovery_verdict('incomplete', 'qualified'), 'inconclusive') self.assertEqual(analysis.recovery_verdict('passed', 'failed'), 'failed') + def test_an_unreadable_journal_at_the_drain_is_recorded(self): + interval = qualify.RecoveryInterval.__new__(qualify.RecoveryInterval) + interval.governed, interval.number, interval.invalid = True, 6, [] + interval.args = SimpleNamespace(journal=self.dir / 'missing/authority.sqlite') + + def no_gateway(*args, **kwargs): + raise OSError(24, 'Too many open files') + interval.bench = SimpleNamespace(gateway=no_gateway, socket=self.dir / 'runner.sock', + authority={'consumer': host.CONSUMER}, + database=self.dir / 'missing/authority.sqlite') + interval.runner = SimpleNamespace(pid=4242, wait=lambda limit: 0, + log=lambda: 'stopped; 0 processes were terminated\n') + drain = interval.drain() + self.assertEqual(drain['error'], 'OSError: [Errno 24] Too many open files') + self.assertTrue(drain['journal_error'].startswith('OperationalError: '), drain) + self.assertEqual(interval.invalid, [f"the journal could not be read: {drain['journal_error']}"]) + self.assertEqual((drain['runner_exit'], drain['socket_left'], drain['terminated']), (0, False, 0)) + def test_verify_recovery(self): path, _ = self.recovery_run() result = qualify.verify(path)