Cron 按日历重复触发任务

from pathlib import Path
from tempfile import mkdtemp
import fcntl
import json
import os
import shutil
import subprocess
import sys
import time


def command(arguments, **options):
    return subprocess.run(arguments, capture_output=True, text=True, timeout=5,
                          env={**os.environ, 'LC_ALL': 'C'}, **options)


def table_absent(result):
    # Debian cron 3.0pl1-162: other listing failures must abort.
    return (result.returncode == 1 and result.stdout == ''
            and result.stderr == 'no crontab for root\n')


def remove_table():
    current = command(['crontab', '-l'])
    if table_absent(current):
        return
    current.check_returncode()
    command(['crontab', '-r']).check_returncode()
    if not table_absent(command(['crontab', '-l'])):
        raise RuntimeError('could not verify table removal')


def attempt(errors, name, operation):
    try:
        operation()
    except Exception as error:
        errors.append(f'{name}: {error!r}')


def stop_daemon(daemon, errors):
    if daemon is None:
        return
    attempt(errors, 'terminate cron', daemon.terminate)
    try:
        daemon.wait(timeout=5)
    except subprocess.TimeoutExpired:
        attempt(errors, 'kill owned cron', daemon.kill)
        attempt(errors, 'reap cron', lambda: daemon.wait(timeout=5))
    except Exception as error:
        errors.append(f'wait cron: {error!r}')


def read_events(root):
    return sorted((json.loads(path.read_text()) for path in root.glob('*.json')),
                  key=lambda event: event['monotonic'])


def wait_workers(root):
    # Workers hold a shared lifetime lock before checking the run permit or publishing.
    with (root / 'lifetime.lock').open() as gate:
        deadline = time.monotonic() + 5
        while True:
            try:
                fcntl.flock(gate, fcntl.LOCK_EX | fcntl.LOCK_NB)
                break
            except BlockingIOError:
                if time.monotonic() >= deadline:
                    raise TimeoutError('workers did not release lifetime lock')
                time.sleep(0.1)
        events = read_events(root)
        started = {e['pid'] for e in events if e['event'] == 'start'}
        ended = {e['pid'] for e in events if e['event'] == 'end'}
        if not started <= ended:
            raise RuntimeError('a started worker has no completion record')


worker_source = r'''from pathlib import Path
import fcntl,json,os,sys,time
root=Path(__file__).parent
kind=sys.argv[1]
def record(event):
    destination=root/f'{os.getpid()}-{event}.json'
    temporary=destination.with_suffix('.tmp')
    temporary.write_text(json.dumps({'event':event,'kind':kind,
                         'pid':os.getpid(),'monotonic':time.monotonic()}))
    temporary.replace(destination)
def main():
    with (root/'lifetime.lock').open() as gate:
        fcntl.flock(gate,fcntl.LOCK_SH)
        if not (root/'run-permit').exists(): return
        lock=None
        try:
            if kind=='locked':
                lock=open(root/'job.lock','a')
                try: fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
                except BlockingIOError:
                    record('skip')
                    return
            record('start')
            try:
                until=time.monotonic()+70
                while time.monotonic()<until and (root/'run-permit').exists():
                    time.sleep(0.1)
            finally: record('end')
        finally:
            if lock is not None: lock.close()
main()
'''


def main():
    # This flag is an operator prerequisite, NOT evidence of containment.
    if sys.argv[1:] != ['--isolated-disposable-lab']:
        raise SystemExit('requires a separately verified disposable Debian 12 lab')
    if os.geteuid() != 0:
        raise RuntimeError('root is required, but does not prove isolation')
    probe = command(['pgrep', '-x', 'cron'])
    if probe.returncode != 1 or probe.stdout or probe.stderr:
        raise RuntimeError(f'cron absence not established: {probe!r}')
    if not table_absent(command(['crontab', '-l'])):
        raise RuntimeError('expected absent root table; existing table or listing error')
    root = Path(mkdtemp(prefix='cron-overlap-', dir='/tmp'))
    daemon = None
    table_attempted = False
    try:
        (root / 'lifetime.lock').touch()
        (root / 'run-permit').touch()
        worker = root / 'worker.py'
        worker.write_text(worker_source)
        table = 'MAILTO=""\n' + ''.join(f'* * * * * /usr/bin/python3 {worker} {kind} >>{root}/worker.log 2>&1\n'
                        for kind in ('plain', 'locked'))
        table_attempted = True
        command(['crontab', '-'], input=table).check_returncode()
        daemon = subprocess.Popen(['cron', '-f'], stdout=subprocess.DEVNULL,
                                  stderr=subprocess.DEVNULL)
        deadline = time.monotonic() + 140
        while time.monotonic() < deadline:
            if daemon.poll() is not None:
                raise RuntimeError('cron exited')
            events = read_events(root)
            plain = [e for e in events if (e['kind'], e['event']) == ('plain', 'start')]
            locked = [e for e in events if (e['kind'], e['event']) == ('locked', 'start')]
            skipped = [e for e in events if (e['kind'], e['event']) == ('locked', 'skip')]
            if len(plain) >= 2 and skipped:
                break
            time.sleep(0.2)
        else:
            raise AssertionError('two real cron ticks were not observed')
        assert len(plain) == 2 and len(locked) == 1 and len(skipped) == 1
        assert plain[1]['monotonic'] - plain[0]['monotonic'] < 70
        assert not any(e['event'] == 'end' and e['pid'] == plain[0]['pid'] for e in events)
        print(json.dumps({'actual_cron_starts': len(plain), 'overlap_observed': True,
                          'seconds_between_starts': round(plain[1]['monotonic']-plain[0]['monotonic'], 3),
                          'locked_starts': len(locked), 'locked_skips': len(skipped)}), flush=True)
    finally:
        original = sys.exc_info()[1]
        errors = []
        attempt(errors, 'revoke run permit', lambda: (root / 'run-permit').unlink(missing_ok=True))
        if table_attempted:
            attempt(errors, 'remove root table', remove_table)
        stop_daemon(daemon, errors)
        attempt(errors, 'wait for workers', lambda: wait_workers(root))
        # Retain private files after failed cleanup; discard the whole lab.
        if not errors:
            attempt(errors, 'remove private files', lambda: shutil.rmtree(root))
        if errors:
            message = 'cleanup incomplete; discard lab: ' + '; '.join(errors)
            if original is not None:
                original.add_note(message)
            else:
                raise RuntimeError(message)


if __name__ == '__main__':
    main()