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