"""Isolated Linux flock/timeout demonstration. No cron changes or network calls. Creates synthetic start/end records in a new directory. Requires Python 3.9+, util-linux flock and GNU timeout. This does not measure scheduler reliability. """ import argparse import datetime as dt import json import os from pathlib import Path import platform import shutil import signal import subprocess import sys import tempfile import time def emit(path, label, phase): event = {'worker': label, 'phase': phase, 'monotonic_ns': time.monotonic_ns(), 'utc': dt.datetime.now(dt.timezone.utc).isoformat()} data = (json.dumps(event) + '\n').encode() fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600) try: assert os.write(fd, data) == len(data) finally: os.close(fd) def events(path): return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] def finish(proc, started): try: out, err = proc.communicate(timeout=4) except subprocess.TimeoutExpired: os.killpg(proc.pid, signal.SIGKILL) proc.communicate() raise RuntimeError('Watchdog stopped a synthetic subprocess group') return {'exit_code': proc.returncode, 'elapsed_s': time.monotonic() - started, 'stdout': out, 'stderr': err} def launch(command): start = time.monotonic() proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True) return proc, start def run(out): if sys.platform != 'linux': raise SystemExit('Run on Linux; Windows file locks are not equivalent.') flock, timeout = shutil.which('flock'), shutil.which('timeout') if not flock or not timeout: raise SystemExit('Requires existing util-linux flock and GNU timeout; installs nothing.') versions = {name: subprocess.check_output([tool, '--version'], text=True).splitlines()[0] for name, tool in [('flock', flock), ('timeout', timeout)]} assert 'GNU coreutils' in versions['timeout'], 'This drill requires GNU timeout semantics' out.mkdir(parents=True, exist_ok=True) root = Path(tempfile.mkdtemp(prefix='cron-overlap-', dir=out)) script = str(Path(__file__).resolve()) def worker(log, label, delay, code=0): return [sys.executable, script, '--worker', label, '--events', str(log), '--delay', str(delay), '--exit-code', str(code)] specifications = [ # All times are demonstration parameters, not production recommendations. ('default_wait', 0.8, [], 0.05, 0, 'after_holder', False), ('skip_busy', 0.8, ['-n', '-E', '75'], 0.05, 75, 'never_started', False), ('wait_expires', 0.8, ['-w', '0.15', '-E', '75'], 0.05, 75, 'never_started', False), ('wait_is_not_runtime', 0.15, ['-w', '0.5', '-E', '75'], 0.8, 0, 'long_worker', False), ('execution_timeout', None, [], 0.8, 124, 'started_not_finished', False), ('timeout_includes_wait', 0.8, [], 0.05, 124, 'never_started', True), ('worker_failure', None, [], 0.01, 23, 'finished_failure', False), ] results = [] for name, hold_s, options, work_s, expected_code, expected_events, wrap_wait in specifications: case = root / name case.mkdir() log, lock = case / 'events.jsonl', case / 'job.lock' holder = None try: if hold_s is not None: holder = launch([flock, str(lock)] + worker(log, 'holder', hold_s)) deadline = time.monotonic() + 2 # A worker start record proves the holder acquired the actual flock. while not any(v['worker']=='holder' and v['phase']=='start' for v in events(log)): if time.monotonic() > deadline or holder[0].poll() is not None: raise RuntimeError('Holder failed to acquire its isolated lock') time.sleep(0.005) inner = worker(log, 'contender', work_s, 23 if name=='worker_failure' else 0) if name == 'execution_timeout': inner = [timeout, '--kill-after=0.2s', '0.15s'] + inner command = [flock] + options + [str(lock)] + inner if wrap_wait: command = [timeout, '--kill-after=0.2s', '0.15s'] + command contender_result = finish(*launch(command)) holder_result = finish(*holder) if holder else None holder = None seen = events(log) contender_events = [v for v in seen if v['worker']=='contender'] phases = [v['phase'] for v in contender_events] passed = contender_result['exit_code'] == expected_code if holder_result: passed = passed and holder_result['exit_code']==0 if expected_events == 'never_started': passed = passed and not contender_events elif expected_events in ('after_holder', 'long_worker'): holder_end = next(v['monotonic_ns'] for v in seen if v['worker']=='holder' and v['phase']=='end') passed = passed and phases==['start','end'] and contender_events[0]['monotonic_ns'] >= holder_end if expected_events == 'long_worker': duration = (contender_events[1]['monotonic_ns']-contender_events[0]['monotonic_ns'])/1e9 passed = passed and duration > 0.5 elif expected_events == 'started_not_finished': passed = passed and phases == ['start'] elif expected_events == 'finished_failure': passed = passed and phases == ['start','end'] # Test reacquisition with a real subprocess, not file deletion/existence. probe = subprocess.run([flock,'-n','-E','75',str(lock),sys.executable,'-c','pass'], capture_output=True, text=True, timeout=2) passed = passed and probe.returncode==0 result = {'case':name,'holder_s':hold_s,'flock_options':options, 'worker_delay_s':work_s,'expected_exit':expected_code, 'expected_events':expected_events,'contender':contender_result, 'holder':holder_result,'events':seen, 'lock_reacquired_after_case':probe.returncode==0,'passed':bool(passed)} results.append(result) finally: if holder is not None and holder[0].poll() is None: os.killpg(holder[0].pid, signal.SIGKILL) holder[0].communicate() result={'completed_at_utc':dt.datetime.now(dt.timezone.utc).isoformat(), 'environment':{'os':platform.system(),'kernel':platform.release(), 'python':platform.python_version(),**versions}, 'workload':'Synthetic start/end JSON records and bounded sleeps only', 'scheduler_executed':False,'network_calls':False,'cron_modified':False, 'cases':results,'passed':all(r['passed'] for r in results), 'limits':'One local Linux filesystem; no real cron daemon, fairness, multi-host locks, child-daemon inheritance, retries, production side effects or performance conclusions.'} (root/'result.json').write_text(json.dumps(result,indent=2)+'\n') print(json.dumps({'result_file':str(root/'result.json'),'passed':result['passed'], 'cases':len(results)},indent=2)) return 0 if result['passed'] else 1 def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--out',type=Path) parser.add_argument('--worker',help=argparse.SUPPRESS) parser.add_argument('--events',type=Path,help=argparse.SUPPRESS) parser.add_argument('--delay',type=float,default=0,help=argparse.SUPPRESS) parser.add_argument('--exit-code',type=int,default=0,help=argparse.SUPPRESS) args=parser.parse_args() if args.worker: if args.events is None or not 0 <= args.delay <= 2: parser.error('Invalid synthetic worker parameters') emit(args.events,args.worker,'start') time.sleep(args.delay) emit(args.events,args.worker,'end') return args.exit_code if args.out is None: parser.error('--out is required') return run(args.out) if __name__=='__main__': raise SystemExit(main())