1#!/usr/bin/env python3
2"""Gates a revision (main by default) on formatting, Clippy, the tests, the Python suite and
3every platform's build, checked out in the jj workspace workspaces/ci so that edits in
4progress elsewhere never reach it. See tools/TESTING.md."""
5import argparse
6from concurrent.futures import ThreadPoolExecutor
7from datetime import datetime
8import fcntl
9import json
10import os
11from pathlib import Path
12import re
13import shutil
14import signal
15import subprocess
16import sys
17import threading
18import time
19
20ROOT = Path(__file__).resolve().parents[1]
21# The checkout holding the jj repository, which a nested workspace's `.jj/repo` file names.
22STORE = ROOT / '.jj/repo'
23MAIN = (STORE.parent / STORE.read_text()).resolve().parents[1] if STORE.is_file() else ROOT
24CI = MAIN / 'workspaces/ci'
25TARGET = CI / 'target'
26RUNS = TARGET / 'ci'
27# Changes here rebuild every package.
28EVERYTHING = ('Cargo.toml', 'Cargo.lock', 'vendor/', 'rust-toolchain', '.cargo/')
29JSON = '--message-format=json'
30
31
32def cargo(*args):
33 return ['cargo', args[0], '--locked', JSON, *args[1:]]
34
35
36def cross(script, arch, *args):
37 return ['sh', script, arch, args[0], '--locked', JSON, *args[1:]]
38
39
40def lanes():
41 """Each lane: its commands, the packages whose dependencies (None for all) and the paths
42 that select it under --changed, extra environment, and why it can't run here, if so."""
43 clippy = ['clippy', '--workspace', '--all-targets', '--all-features', '--', '-D', 'warnings']
44 # What ships: mobile is iOS's alone, and the tests and lab examples run only on macOS.
45 cross_clippy = ['clippy', '--workspace', '--exclude', 'mobile', '--lib', '--bins', '--all-features',
46 '--', '-D', 'warnings']
47 windows = CI / 'platform/windows/cargo.sh'
48 linux = ROOT / 'platform/linux/cargo.sh'
49 mingw = os.environ.get('LLVM_MINGW') or MAIN / 'target/windows/llvm-mingw'
50 sdk = Path(os.environ.get('SNOW_LEOPARD_SDK') or MAIN / 'target/snow-leopard/MacOSX10.6.sdk')
51 nightly = subprocess.run(['rustup', 'component', 'list', '--installed', '--toolchain', 'nightly'],
52 capture_output=True, text=True).stdout.split()
53 result = [
54 dict(name='test', minutes=45, packages=None, paths=()),
55 dict(name='clippy', minutes=30, packages=None, paths=(), commands=[
56 cargo(*clippy),
57 cargo('clippy', '-p', 'snowbound', '--no-default-features', '--', '-D', 'warnings')]),
58 dict(name='fmt', minutes=5, packages=None, paths=(), commands=[['cargo', 'fmt', '--all', '--check']]),
59 dict(name='python', minutes=30, packages=None, paths=('tools/', 'corpus/'), commands=[
60 # The suite runs this checkout's examples from target/debug.
61 cargo('build', '--workspace', '--all-features', '--examples', '--bins'),
62 ['uv', 'run', '--no-project', '--python', '3.12', '--with', 'pillow', '--with', 'pdfplumber',
63 'python', '-m', 'unittest', 'discover', '-s', 'tools', '-p', 'test_*.py']],
64 environment={'PYTHONPATH': str(CI / 'tools')},
65 missing=None if shutil.which('uv') else 'needs uv'),
66 ]
67 for arch in ('x86_64', 'aarch64'):
68 result.append(dict(
69 name=f'windows-{arch}', minutes=45, packages=['snowbound'], paths=('platform/windows/',),
70 # x86_64's nightly Clippy would lint std, which it builds, and nightly's own new lints.
71 commands=[*([] if arch == 'x86_64' else [cross(windows, arch, *cross_clippy)]),
72 cross(windows, arch, 'build', '-p', 'snowbound')],
73 environment={'LLVM_MINGW': str(mingw)},
74 missing=None if arch == 'aarch64' or 'rust-src' in nightly else 'needs nightly with rust-src'))
75 for arch in ('x86_64', 'aarch64'):
76 result.append(dict(
77 name=f'linux-{arch}', minutes=30, packages=['snowbound'], paths=('platform/linux/',),
78 commands=[cross(linux, arch, *cross_clippy), cross(linux, arch, 'build', '-p', 'snowbound')],
79 environment={'CARGO_TARGET_DIR': str(TARGET / 'linux')},
80 missing=None if shutil.which('zig') else 'needs zig'))
81 targets = subprocess.run(['rustup', 'target', 'list', '--installed'], capture_output=True, text=True).stdout
82 wasm = web_environment()
83 result.append(dict(
84 name='web', minutes=20, packages=['snowbound'], paths=('crates/snowbound/web/', 'tools/release_web.py'),
85 # Clippy builds it; the module itself is linked by release_web.py.
86 commands=[cargo('clippy', '-p', 'snowbound', '--target', 'wasm32-unknown-unknown', '--no-default-features', '--features', 'wgpu,live', '--', '-D', 'warnings')],
87 environment={**(wasm or {}), 'CARGO_TARGET_DIR': str(TARGET / 'wasm')},
88 missing='needs the wasm32-unknown-unknown target' if 'wasm32-unknown-unknown' not in targets.split()
89 else None if wasm else 'needs nix for a clang that builds for wasm32'))
90 result.append(dict(
91 name='ios', minutes=30, packages=['mobile'], paths=('apps/ios/',),
92 # build-rust.sh builds the Rust half into target/ios.
93 commands=[['xcodebuild', '-quiet', '-project', 'apps/ios/Snowbound.xcodeproj', '-scheme', 'Snowbound',
94 '-configuration', 'Debug', '-destination', 'generic/platform=iOS Simulator',
95 '-derivedDataPath', TARGET / 'ios-xcode', 'CODE_SIGNING_ALLOWED=NO', 'build']],
96 missing=None if shutil.which('xcodebuild') else 'needs Xcode'))
97 result.append(dict(
98 name='macos-10.6', minutes=45, packages=['snowbound'], paths=('platform/snow-leopard/',),
99 commands=[['sh', CI / 'platform/snow-leopard/cargo.sh', 'build', '--locked', JSON, '-p', 'snowbound',
100 '--no-default-features']],
101 environment={'SNOW_LEOPARD_SDK': str(sdk)},
102 missing=('needs nightly with rust-src' if 'rust-src' not in nightly else
103 None if sdk.exists() else 'needs the 10.6 SDK from platform/snow-leopard/remote.sh sdk')))
104 return result
105
106
107def web_environment():
108 """A clang and llvm-ar that build SQLite for wasm32, as release_web.py finds them; none
109 where it finds none."""
110 import release_web
111 try:
112 environment = release_web.environment()
113 except SystemExit:
114 return None
115 return {key: environment[key] for key in ('CC_wasm32_unknown_unknown', 'AR_wasm32_unknown_unknown')}
116
117
118def jj(*args, cwd=ROOT):
119 return subprocess.run(['jj', *args], cwd=cwd, check=True, capture_output=True, text=True).stdout
120
121
122def checkout(rev):
123 """Points the CI workspace's own commit, a child of main, at `rev`'s files."""
124 commits = jj('log', '--no-graph', '-r', rev, '-T', 'commit_id ++ "\\n"').split()
125 if len(commits) != 1:
126 sys.exit(f'{rev} names {len(commits)} revisions, not one.')
127 if not CI.exists():
128 CI.parent.mkdir(exist_ok=True)
129 jj('workspace', 'add', '--name', 'ci', '-r', 'main', str(CI))
130 jj('workspace', 'update-stale', cwd=CI)
131 jj('rebase', '-r', '@', '-o', 'main', cwd=CI)
132 jj('restore', '--from', commits[0], cwd=CI)
133 return commits[0]
134
135
136class Run:
137 def __init__(self, folder, environment):
138 self.folder = folder
139 self.environment = environment
140 self.processes = set()
141 self.lock = threading.Lock()
142 self.stopping = False
143
144 def execute(self, command, log, deadline, environment=(), cwd=CI):
145 """The command's exit status, or None where it outlived `deadline`."""
146 with open(log, 'a') as stream:
147 stream.write(f'$ {" ".join(map(str, command))}\n')
148 stream.flush()
149 try:
150 process = subprocess.Popen(list(map(str, command)), cwd=cwd, stdout=stream,
151 stderr=subprocess.STDOUT, start_new_session=True,
152 env={**self.environment, **dict(environment)})
153 except OSError as error:
154 stream.write(f'error: {error}\n')
155 return 127
156 with self.lock:
157 self.processes.add(process)
158 if self.stopping:
159 stop(process)
160 try:
161 return process.wait(timeout=max(deadline - time.monotonic(), 0))
162 except subprocess.TimeoutExpired:
163 stop(process)
164 stream.write('\n(timed out)\n')
165 return None
166 finally:
167 with self.lock:
168 self.processes.discard(process)
169
170 def stop_all(self):
171 with self.lock:
172 self.stopping = True
173 for process in self.processes:
174 stop(process)
175
176
177def stop(process):
178 for sent in (signal.SIGTERM, signal.SIGKILL):
179 try:
180 os.killpg(process.pid, sent)
181 process.wait(timeout=5)
182 return
183 except (ProcessLookupError, subprocess.TimeoutExpired):
184 pass
185
186
187def status(code):
188 return 'passed' if code == 0 else 'timeout' if code is None else 'failed'
189
190
191def run_lane(run, lane, deadline):
192 log = run.folder / f'{lane["name"]}.log'
193 for command in lane['commands']:
194 code = run.execute(command, log, deadline, lane.get('environment', {}))
195 if code != 0:
196 return status(code), [log]
197 return 'passed', [log]
198
199
200def run_tests(run, deadline, packages, jobs):
201 """Builds every test target, then runs those of `packages` (all where None) side by side,
202 each from its package's folder as cargo would, beside the doctests."""
203 log = run.folder / 'test.log'
204 code = run.execute(cargo('test', '--workspace', '--all-features', '--no-run'), log, deadline)
205 if code != 0:
206 return status(code), [log], []
207 binaries = []
208 for line in log.read_text(errors='replace').splitlines():
209 message = json.loads(line) if line.startswith('{') else {}
210 if message.get('reason') == 'compiler-artifact' and message['profile']['test'] and message['executable']:
211 folder = Path(message['manifest_path']).parent
212 if packages is None or folder.name in packages:
213 binaries.append((f'{folder.name}.{message["target"]["name"]}', [message['executable']], folder))
214 binaries.append(('doctests', ['cargo', 'test', '--locked', '--workspace', '--all-features', '--doc'], CI))
215 # The slowest last time start first, so that none of them starts last.
216 timings = RUNS / 'test-seconds.json'
217 seconds = json.loads(timings.read_text()) if timings.exists() else {}
218 binaries.sort(key=lambda binary: -seconds.get(binary[0], float('inf')))
219 (run.folder / 'test').mkdir()
220
221 def one(binary):
222 name, command, cwd = binary
223 started = time.monotonic()
224 part_log = run.folder / 'test' / f'{name}.log'
225 code = run.execute(command, part_log, deadline, cwd=cwd)
226 return {'name': name, 'status': status(code), 'seconds': round(time.monotonic() - started, 1),
227 'log': str(part_log)}
228
229 with ThreadPoolExecutor(jobs) as pool:
230 parts = list(pool.map(one, binaries))
231 timings.write_text(json.dumps({**seconds, **{part['name']: part['seconds'] for part in parts}}, indent=1))
232 failed = [part for part in parts if part['status'] != 'passed']
233 result = 'timeout' if any(part['status'] == 'timeout' for part in failed) else 'failed' if failed else 'passed'
234 return result, [log, *(Path(part['log']) for part in failed)], sorted(parts, key=lambda part: -part['seconds'])
235
236
237def relative(path):
238 return path.removeprefix(f'{CI}/')
239
240
241def diagnose(log):
242 """The errors a log reports, each with the file and line it names where it names one."""
243 errors, text = [], []
244 lines = log.read_text(errors='replace').splitlines()
245 test = None
246 for index, line in enumerate(lines):
247 following = lines[index + 1] if index + 1 < len(lines) else ''
248 if line.startswith('{'):
249 try:
250 message = json.loads(line).get('message') or {}
251 except json.JSONDecodeError:
252 continue
253 spans = [span for span in message.get('spans', []) if span.get('is_primary')]
254 if message.get('level') == 'error' and spans:
255 errors.append({'error': message['message'],
256 'file': f'{relative(spans[0]["file_name"])}:{spans[0]["line_start"]}'})
257 elif match := re.match(r'---- (\S+) stdout ----', line):
258 test = match[1]
259 elif match := re.search(r'panicked at (\S+?):(\d+):\d+:$', line):
260 errors.append({'error': f'{test or "test"} panicked: {following.strip()}',
261 'file': f'{relative(match[1])}:{match[2]}'})
262 elif match := re.match(r'Diff in (\S+?):(\d+):', line):
263 errors.append({'error': 'not formatted (cargo fmt)', 'file': f'{relative(match[1])}:{match[2]}'})
264 elif match := re.match(r'(FAIL|ERROR): (\S+) \((\S+)\)', line):
265 block = lines[index + 2:]
266 end = next((i for i, rest in enumerate(block) if rest.startswith(('=' * 20, '-' * 20))), len(block))
267 block = [rest for rest in block[:end] if rest.strip()]
268 frames = [frame for frame in block if f'File "{CI}/' in frame]
269 where = re.search(r'File "(\S+)", line (\d+)', frames[-1]) if frames else None
270 errors.append({'error': f'{match[3]}: {block[-1].strip() if block else match[1]}',
271 'file': f'{relative(where[1])}:{where[2]}' if where else None})
272 elif match := re.match(r'(/\S+?):(\d+):\d+: (?:fatal )?error: (.*)', line):
273 errors.append({'error': match[3], 'file': f'{relative(match[1])}:{match[2]}'})
274 elif (match := re.match(r'error(?:\[\w+\])?: (.*)', line)) and (
275 where := re.match(r'\s*--> (\S+?):(\d+):\d+', following)):
276 errors.append({'error': match[1], 'file': f'{relative(where[1])}:{where[2]}'})
277 elif re.match(r'(error|test \S+ \.\.\. FAILED)', line):
278 text.append({'error': line.strip(), 'file': None})
279 if not errors and not text:
280 text = [{'error': line, 'file': None} for line in lines if line.strip() and not line.startswith('{')][-3:]
281 return list({(error['error'], error['file']): error for error in errors or text}.values())
282
283
284def affected(changed, metadata):
285 """The workspace packages that changed files reach, through what depends on them."""
286 folders = {package['name']: Path(package['manifest_path']).parent.relative_to(CI).as_posix()
287 for package in metadata['packages']}
288 if any(path.startswith(EVERYTHING) for path in changed):
289 return set(folders)
290 reached = {name for name, folder in folders.items() if any(path.startswith(folder + '/') for path in changed)}
291 users = {package['name']: {dependency['name'] for dependency in package['dependencies']
292 if dependency['name'] in folders} for package in metadata['packages']}
293 while more := {name for name, uses in users.items() if uses & reached} - reached:
294 reached |= more
295 return reached
296
297
298def dependencies(roots, metadata):
299 uses = {package['name']: {dependency['name'] for dependency in package['dependencies']}
300 for package in metadata['packages']}
301 closure, pending = set(), list(roots)
302 while pending:
303 name = pending.pop()
304 if name in uses and name not in closure:
305 closure.add(name)
306 pending.extend(uses[name])
307 return closure
308
309
310def size(path):
311 if not path.is_dir() or path.is_symlink():
312 return path.lstat().st_blocks * 512
313 return sum(file.lstat().st_blocks * 512 for file in path.rglob('*'))
314
315
316def prune(budget, started):
317 """Deletes, least recently used first, the build units this run didn't use until the
318 target folder fits `budget` bytes. Cargo reads a unit's fingerprint whenever it checks
319 it, so the fingerprint's access time is when a build last used the unit."""
320 total = disk_usage()
321 if total <= budget:
322 return 0
323 units = {}
324 for fingerprints in [*TARGET.glob('*/.fingerprint'), *TARGET.glob('*/*/.fingerprint'),
325 *TARGET.glob('*/*/*/.fingerprint')]:
326 profile = fingerprints.parent
327 for kind in ('deps', 'build', '.fingerprint', 'incremental', 'examples'):
328 for entry in os.scandir(profile / kind) if (profile / kind).is_dir() else ():
329 if match := re.search(r'-([0-9a-f]{16})(?:\.|$)', entry.name):
330 units.setdefault((profile, match[1]), []).append(Path(entry.path))
331 used = {}
332 for unit, paths in units.items():
333 stamps = [file.stat().st_atime for path in paths if path.parent.name == '.fingerprint'
334 for file in path.iterdir()] or [path.stat().st_atime for path in paths]
335 used[unit] = max(stamps)
336 freed = 0
337 for unit in sorted(units, key=used.get):
338 if used[unit] >= started or total - freed <= budget * 0.8:
339 break
340 for path in units[unit]:
341 freed += size(path)
342 shutil.rmtree(path) if path.is_dir() and not path.is_symlink() else path.unlink()
343 return total - disk_usage()
344
345
346def disk_usage():
347 return int(subprocess.check_output(['du', '-sk', TARGET]).split()[0]) * 1024
348
349
350def table(results):
351 lines = []
352 for result in results:
353 seconds = result.get('seconds')
354 clock = f'{int(seconds // 60)}:{int(seconds % 60):02}' if seconds is not None else ''
355 note = result.get('note', '')
356 lines.append(f'{result["name"]:<16} {result["status"]:<8} {clock:>6} {note}'.rstrip())
357 for error in result.get('errors', [])[:6]:
358 where = f'{error["file"]} ' if error['file'] else ''
359 lines.append(f'{"":<33}{where}{error["error"].splitlines()[0][:160]}')
360 if len(result.get('errors', [])) > 6:
361 lines.append(f'{"":<33}… {len(result["errors"]) - 6} more in {result["log"]}')
362 elif result['status'] in ('failed', 'timeout'):
363 lines.append(f'{"":<33}{result["log"]}')
364 return '\n'.join(lines)
365
366
367def main():
368 every = lanes()
369 names = [lane['name'] for lane in every]
370 parser = argparse.ArgumentParser(description=__doc__)
371 source = parser.add_mutually_exclusive_group()
372 source.add_argument('--rev', default='main', help='The jj revision to gate; main by default')
373 source.add_argument('--working-copy', action='store_true',
374 help="This checkout's working copy, as it is when the run starts")
375 parser.add_argument('--lanes', nargs='+', metavar='LANE',
376 help=f'Only these lanes, or those starting LANE-: {", ".join(names)}')
377 parser.add_argument('--changed', action='store_true',
378 help="Only the lanes, and tests of the packages, that the revision's changes from main reach")
379 parser.add_argument('--jobs', type=int, default=4, help='Lanes at once (default 4)')
380 parser.add_argument('--test-jobs', type=int, default=4, help='Test executables at once (default 4)')
381 parser.add_argument('--timeout', type=float, metavar='MINUTES', help="Each lane's limit, overriding its own")
382 parser.add_argument('--budget', type=float, default=40, metavar='GB',
383 help='Prune workspaces/ci/target to this size after the run (default 40)')
384 args = parser.parse_args()
385 if ROOT == CI:
386 sys.exit(f'Run ci.py from another checkout; {CI} is its own.')
387 def named(lane, name):
388 return lane == name or lane.startswith(f'{name}-')
389 unknown = [name for name in args.lanes or () if not any(named(lane, name) for lane in names)]
390 if unknown:
391 parser.error(f'unknown lanes {unknown}; choose from {names}')
392 chosen = [lane for lane in every if not args.lanes or any(named(lane['name'], name) for name in args.lanes)]
393
394 RUNS.mkdir(parents=True, exist_ok=True)
395 lock = open(RUNS / 'lock', 'w')
396 try:
397 fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
398 except BlockingIOError:
399 print('Waiting for another ci.py run to finish…', file=sys.stderr, flush=True)
400 fcntl.flock(lock, fcntl.LOCK_EX)
401 started_at, started = datetime.now().astimezone(), time.monotonic()
402 wall = time.time()
403 rev = '@' if args.working_copy else args.rev
404 commit = checkout(rev)
405 described = jj('log', '--no-graph', '-r', commit, '-T',
406 'change_id.short() ++ " " ++ commit_id.short() ++ " " ++ description.first_line()').strip()
407 print(f'Gating {rev}: {described}', flush=True)
408
409 tested = None
410 changed = None
411 if args.changed:
412 changed = [line for line in jj('diff', '--from', 'main', '--to', commit, '--name-only').splitlines() if line]
413 metadata = json.loads(subprocess.check_output(['cargo', 'metadata', '--format-version=1', '--no-deps'], cwd=CI))
414 tested = affected(changed, metadata)
415 chosen = [lane for lane in chosen
416 if any(path.startswith(lane['paths']) for path in changed)
417 or tested & (dependencies(lane['packages'], metadata) if lane['packages'] else tested)]
418 print(f'Changed from main: {len(changed)} files, reaching {", ".join(sorted(tested)) or "no packages"}',
419 flush=True)
420
421 folder = RUNS / 'runs' / started_at.strftime('%Y%m%d-%H%M%S')
422 folder.mkdir(parents=True)
423 environment = {key: value for key, value in os.environ.items()
424 if not key.startswith(('ONESTORE_', 'SNOWBOUND_', 'CARGO_TARGET_DIR'))}
425 run = Run(folder, environment)
426
427 def gate(lane):
428 result = {'name': lane['name']}
429 if lane.get('missing'):
430 print(f'{lane["name"]}: skipped, {lane["missing"]}', flush=True)
431 return {**result, 'status': 'skipped', 'note': lane['missing']}
432 began = time.monotonic()
433 deadline = began + 60 * (args.timeout or lane['minutes'])
434 if lane['name'] == 'test':
435 outcome, logs, parts = run_tests(run, deadline, tested, args.test_jobs)
436 result['parts'] = parts
437 else:
438 outcome, logs = run_lane(run, lane, deadline)
439 errors = [error for log in logs for error in diagnose(log)] if outcome != 'passed' else []
440 if outcome == 'timeout':
441 errors.insert(0, {'error': f'timed out after {args.timeout or lane["minutes"]:g} minutes', 'file': None})
442 seconds = round(time.monotonic() - began, 1)
443 print(f'{lane["name"]}: {outcome} in {seconds:.0f}s', flush=True)
444 return {**result, 'status': outcome, 'seconds': seconds, 'log': str(logs[0]), 'errors': errors}
445
446 pool = ThreadPoolExecutor(args.jobs)
447 try:
448 results = list(pool.map(gate, chosen))
449 except KeyboardInterrupt:
450 pool.shutdown(wait=False, cancel_futures=True)
451 run.stop_all()
452 raise
453 passed = all(result['status'] in ('passed', 'skipped') for result in results)
454 freed = prune(args.budget * 1e9, wall)
455 summary = {
456 'status': 'passed' if passed else 'failed', 'revision': rev, 'commit': commit, 'described': described,
457 'workspace': str(CI), 'started': started_at.isoformat(timespec='seconds'),
458 'seconds': round(time.monotonic() - started, 1), 'changed': changed,
459 'tested_packages': sorted(tested) if tested is not None else None,
460 'pruned_bytes': freed, 'lanes': results,
461 }
462 (folder / 'summary.json').write_text(json.dumps(summary, indent=2) + '\n')
463 for old in sorted((RUNS / 'runs').iterdir())[:-20]:
464 shutil.rmtree(old)
465 print()
466 print(table(results))
467 print(f'\n{summary["status"]} in {summary["seconds"] / 60:.1f} min: {folder / "summary.json"}')
468 sys.exit(0 if passed else 1)
469
470
471if __name__ == '__main__':
472 main()