#!/usr/bin/env python3 # SPDX-License-Identifier: MIT # Copyright (c) 2026 CyberNative AI LLC """Will my APScheduler 3.x cron schedule loop, go backwards or skip runs at the next DST fall-back? python check_fallback.py --zone America/New_York --cron "*/5 * * * *" python check_fallback.py --zone Europe/London --cron "*/30 * * * *" --zone-lib zoneinfo Uses the APScheduler that is INSTALLED in this interpreter. It finds the next instant the zone's UTC offset drops (the start of the repeated wall-clock period, to the second) and tests five situations around it, each in its own Python process: four synthetic due/now pairs on a real Job of a paused BackgroundScheduler (Job._get_run_times), and the trigger's get_next_fire_time walked through the change as a scheduler waking on time would. Each situation is stopped at the time limit (--timeout); a situation that is stopped, or runs out of memory, is LOOP. Where the system allows it, each check process first limits its own address space (--mem-mb, RLIMIT_AS) before it imports APScheduler; the output says whether that limit was applied, refused or unavailable. --jobs processes run at a time (default 2). OK finite, increasing run times (walk: fires in the repeated period where the schedule matches it) LOOP stopped at the time limit, out of memory, or the walk goes backwards / never passes the change BACKWARDS finite answer, but the run times go backwards in time SKIPS the walk fires in the first pass of the repeated period but never in the second N/A the situation does not apply to this schedule (the reason is printed); it changes no exit status Exit status: 0 = every situation listed was checked and is OK or N/A; 1 = at least one LOOP, BACKWARDS or SKIPS; 2 = the check could not finish: a zone library you asked for is missing, this APScheduler does not accept that kind of time zone, the crontab is not valid, or a situation ended in ERROR. 2 wins over 1 when both apply. By default it checks the zone libraries that are installed (pytz and/or zoneinfo) and says which ones it checked. It needs only the standard library plus the installed APScheduler (and pytz, if used). It does not edit your scheduler or application data and uses no network. Tested on Linux with CPython 3.12. """ import sys if __name__ == "__main__" and sys.argv[1:2] == ["--child"]: # First act of a check process, before APScheduler is imported: cap our own address space. def _cap_self(mb): if mb <= 0: return {"status": "off"} try: import resource lim = resource.RLIMIT_AS except (ImportError, AttributeError): return {"status": "unavailable"} try: soft, hard = resource.getrlimit(lim) want = mb * 1024 * 1024 for cur in (soft, hard): if cur != resource.RLIM_INFINITY and cur < want: want = cur # an existing limit is already lower: keep it resource.setrlimit(lim, (want, hard)) got = resource.getrlimit(lim)[0] return {"status": "applied", "mb": got // (1024 * 1024)} except (ValueError, OSError) as e: return {"status": "refused", "why": f"{e.__class__.__name__}: {e}"} _CAP = _cap_self(int(sys.argv[2])) import json print(json.dumps({"cap": _CAP}), flush=True) import argparse, datetime as dt, json, os, subprocess, threading from concurrent.futures import ThreadPoolExecutor from datetime import timedelta, timezone UTC = timezone.utc LIBS = ("pytz", "zoneinfo") SCENARIOS = [ # name, description (F = first instant of the repeated period, UTC) ("a", "last fire before the change, now = F-1s"), ("a_t", "previous fire F-15min, now = F-1s"), ("b1", "both inside the repeated hour: F+15min / F+20min"), ("b2", "restart inside the repeated hour: first fire after F+20min, then run it"), ("w", "scheduler walking through the change on time: prev = now = each answer, F-1s to F+90min"), ] OOM_LINE = b'{"verdict": "OOM"}\n' # ---------------------------------------------------------------- the transition (parent, stdlib only) def next_fallback(zone, after): """First whole-second UTC instant F after `after` where the zone's UTC offset drops; returns (F, drop in seconds) or None.""" import zoneinfo tz = zoneinfo.ZoneInfo(zone) off = lambda t: t.astimezone(tz).utcoffset() lo = after.replace(microsecond=0) for _ in range(2 * 366 * 24): # bracket hour by hour, up to two years ahead hi = lo + timedelta(hours=1) if off(hi) < off(lo): break lo = hi else: return None a, b = int(lo.timestamp()), int(hi.timestamp()) # off(a) is the old offset, off(b) the new lower one old = off(lo) while b - a > 1: # bisect to the second m = (a + b) // 2 if off(dt.datetime.fromtimestamp(m, UTC)) < old: b = m else: a = m F = dt.datetime.fromtimestamp(b, UTC) drop = int((off(F - timedelta(seconds=1)) - off(F)).total_seconds()) if drop <= 0: raise RuntimeError(f"no positive offset drop at {F:%Y-%m-%d %H:%M:%S}Z") return F, drop # ---------------------------------------------------------------- check processes def mk_tz(zone, lib): if lib == "pytz": import pytz return pytz.timezone(zone) import zoneinfo return zoneinfo.ZoneInfo(zone) def child_probe(spec): """Can this APScheduler take the zone library and the crontab?""" import importlib.metadata as md try: tz = mk_tz(spec["zone"], spec["lib"]) except ImportError as e: return {"probe": "missing", "msg": f"cannot import {spec['lib']}: {e}"} from apscheduler.triggers.cron import CronTrigger import apscheduler.util as au out = {"aps": md.version("apscheduler")} try: if hasattr(au, "astimezone"): au.astimezone(tz) from apscheduler.schedulers.background import BackgroundScheduler BackgroundScheduler(timezone=tz) except Exception as e: return dict(out, probe="unsupported", msg=f"{e.__class__.__name__}: {e}") try: CronTrigger.from_crontab(spec["cron"], timezone=tz) except Exception as e: return dict(out, probe="invalid_cron", msg=f"{e.__class__.__name__}: {e}") return dict(out, probe="ok") def child_situation(spec): from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger F = dt.datetime.fromisoformat(spec["fold"]).replace(tzinfo=UTC) span = timedelta(seconds=spec["span_s"]) # length of the repeated wall-clock period tz = mk_tz(spec["zone"], spec["lib"]) loc = lambda u: u.astimezone(tz) def trig(): return CronTrigger.from_crontab(spec["cron"], timezone=tz) def last_fire_before(limit): t, last = F - timedelta(hours=26), None for _ in range(5000): f = trig().get_next_fire_time(None, loc(t)) if f is None or f.astimezone(UTC) > limit: break last = f.astimezone(UTC); t = last + timedelta(seconds=1) return last sc = spec["scenario"] if sc == "w": t = trig(); prev = now_l = loc(F - span - timedelta(seconds=1)); n_first = 0 for _ in range(400): # first pass through the same wall-clock period, before the change nxt = t.get_next_fire_time(prev, now_l) if nxt is None or nxt.astimezone(UTC) >= F: break if nxt.astimezone(UTC) <= prev.astimezone(UTC): n_first = None; break n_first += 1; prev = now_l = nxt else: n_first = None t = trig(); prev = now_l = loc(F - timedelta(seconds=1)); seq = []; back = False for _ in range(400): if now_l.astimezone(UTC) >= F + timedelta(minutes=90): break nxt = t.get_next_fire_time(prev, now_l) if nxt is None: break if nxt.astimezone(UTC) <= prev.astimezone(UTC): back = True seq.append(nxt.astimezone(UTC)); prev = now_l = nxt else: back = True # never got past F+90min return {"verdict": "LOOP" if back else "WALKED", "n_run_times": len(seq), "fires_in_repeated_hour": sum(1 for x in seq if F <= x < F + span), "fires_in_first_pass": n_first, "first_utc": [x.strftime("%H:%M:%S") for x in seq[:8]]} if sc == "a": nrt, now = last_fire_before(F - timedelta(minutes=15)), F - timedelta(seconds=1) if nrt is None: return {"verdict": "N/A", "reason": "no run in the 26 h before the change to start from"} elif sc == "a_t": nrt, now = F - timedelta(minutes=15), F - timedelta(seconds=1) elif sc == "b1": nrt, now = F + timedelta(minutes=15), F + timedelta(minutes=20) else: first = trig().get_next_fire_time(None, loc(F + timedelta(minutes=20))) if first is None: return {"verdict": "N/A", "reason": "the schedule has no run after the change"} nrt = first.astimezone(UTC); now = nrt + timedelta(seconds=1) s = BackgroundScheduler(timezone=tz); s.start(paused=True) s.add_job(lambda: None, trig(), id="j", next_run_time=loc(nrt)) rt = s.get_job("j")._get_run_times(loc(now)); s.shutdown(wait=False) ut = [r.astimezone(UTC) for r in rt] return {"verdict": "RAN", "n_run_times": len(rt), "monotonic": all(a < b for a, b in zip(ut, ut[1:])), "run_times_utc": [u.strftime("%H:%M:%S") for u in ut[:12]]} def child_main(spec): try: r = child_probe(spec) if spec["scenario"] == "probe" else child_situation(spec) except MemoryError: os.write(1, OOM_LINE); os._exit(3) except Exception as e: r = {"verdict": "ERROR", "error": f"{e.__class__.__name__}: {e}"} print(json.dumps(r), flush=True) os._exit(0) # do not wait for scheduler threads # ---------------------------------------------------------------- parent class Runner: """Starts check processes, stops each at the time limit and reaps it; remembers live ones for Ctrl-C.""" def __init__(self, args): self.args, self.live, self.lock, self.stopping, self.caps = args, set(), threading.Lock(), False, [] def run(self, spec): cmd = [sys.executable, os.path.abspath(__file__), "--child", str(self.args.mem_mb), json.dumps(spec)] with self.lock: if self.stopping: return None, None, "", False p = subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) self.live.add(p) timed_out = False try: out, err = p.communicate(timeout=self.args.timeout) except subprocess.TimeoutExpired: timed_out = True p.kill() out, err = p.communicate() # waits for the killed process: nothing is left behind finally: with self.lock: self.live.discard(p) result = None for line in out.splitlines(): try: d = json.loads(line) except ValueError: continue if isinstance(d, dict) and "cap" in d: with self.lock: self.caps.append(d["cap"]) elif isinstance(d, dict): result = d return result, p.returncode, err, timed_out def stop_all(self): with self.lock: self.stopping = True for p in self.live: p.kill() def classify(spec, got, args): r, rc, err, timed_out = got if rc is None: return "ERROR", "not run (interrupted)" if timed_out: return "LOOP", f"no answer within {args.timeout:g}s (stopped)" if (r and r.get("verdict") == "OOM") or (r is None and "MemoryError" in err): return "LOOP", f"out of memory (address-space limit {args.mem_mb} MB)" v = r.get("verdict") if r else None if v == "LOOP": return "LOOP", "answers go backwards or never pass the change: " + ", ".join(r.get("first_utc", [])) if v == "WALKED": F = dt.datetime.fromisoformat(spec["fold"]); E = F + timedelta(seconds=spec["span_s"]) hour = ("hour " if spec["span_s"] == 3600 else f"{spec['span_s'] // 60} min ") + f"{F:%H:%M}-{E:%H:%M}Z" if r["fires_in_repeated_hour"] == 0 and r.get("fires_in_first_pass"): return "SKIPS", f"0 fires in the repeated {hour}, {r['fires_in_first_pass']} in the same wall-clock period before it" return "OK", f"{r['fires_in_repeated_hour']} fire(s) in the repeated {hour}" if v == "RAN": if not r["monotonic"]: return "BACKWARDS", "run times go backwards in time: " + ", ".join(r["run_times_utc"]) return "OK", f"{r['n_run_times']} run time(s)" if v == "N/A": return "N/A", r["reason"] if v == "ERROR": return "ERROR", r["error"][:160] tail = (err.strip().splitlines() or [f"exit status {rc}, no result"])[-1] return "ERROR", tail[:160] SEVERITY = ("ERROR", "LOOP", "BACKWARDS", "SKIPS") def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--zone", help="IANA zone the scheduler uses, e.g. America/New_York") ap.add_argument("--cron", help='5-field crontab, e.g. "*/5 * * * *"') ap.add_argument("--zone-lib", choices=["installed", "pytz", "zoneinfo", "both"], default="installed", help="zone library to check (default: every installed one; naming one makes a missing one an error)") ap.add_argument("--timeout", type=float, default=10, help="seconds per situation (default 10)") ap.add_argument("--mem-mb", type=int, default=512, help="address-space limit per check process in MB, 0 = none (default 512)") ap.add_argument("--jobs", type=int, default=2, help="check processes at a time (default 2)") ap.add_argument("--after", help="start the search at this time, ISO format, UTC unless an offset is given (default: now)") ap.add_argument("--json", action="store_true", help="print one JSON object instead of text") if sys.argv[1:2] == ["--child"]: return child_main(json.loads(sys.argv[3])) args = ap.parse_args() if not args.zone or not args.cron: ap.error("--zone and --cron are required") if sys.version_info < (3, 9): print("this check needs Python 3.9 or later (it uses zoneinfo to find the change)"); return 2 import importlib.metadata as md try: ver = md.version("apscheduler") except md.PackageNotFoundError: print("APScheduler is not installed in this interpreter (" + sys.executable + ")"); return 2 if not ver.startswith("3."): print(f"APScheduler {ver} is installed; this check is for APScheduler 3.x"); return 2 if args.after: after = dt.datetime.fromisoformat(args.after.replace("Z", "+00:00")) after = after.replace(tzinfo=UTC) if after.tzinfo is None else after.astimezone(UTC) else: after = dt.datetime.now(UTC) try: found = next_fallback(args.zone, after) except Exception as e: print(f"zone {args.zone!r} not found ({e.__class__.__name__}); on Windows, pip install tzdata"); return 2 out = {"apscheduler": ver, "zone": args.zone, "cron": args.cron} P = (lambda *a: None) if args.json else print P(f"APScheduler {ver} | zone {args.zone} | cron \"{args.cron}\"") if found is None: msg = f"no fall-back in {args.zone} in the two years after {after:%Y-%m-%d %H:%M}Z: nothing to check" P("RESULT: " + msg) if args.json: print(json.dumps(dict(out, fold_utc=None, result=msg, exit=0))) return 0 F, span_s = found import zoneinfo z = zoneinfo.ZoneInfo(args.zone) P(f"next fall-back: {F:%Y-%m-%d %H:%M:%S} UTC ({(F - timedelta(seconds=1)).astimezone(z):%H:%M:%S %Z} -> " f"{F.astimezone(z):%H:%M:%S %Z}), clocks go back {span_s // 60} min") asked = list(LIBS) if args.zone_lib in ("installed", "both") else [args.zone_lib] base = dict(zone=args.zone, cron=args.cron, fold=F.isoformat().replace("+00:00", ""), span_s=span_s) runner = Runner(args) ex = ThreadPoolExecutor(max(1, args.jobs)) try: probes = dict(zip(asked, ex.map(lambda lib: runner.run(dict(base, lib=lib, scenario="probe")), asked))) lib_state = {} # lib -> None (checked) or (status, message) for lib in asked: r, rc, err, timed_out = probes[lib] if r and r.get("probe") == "ok": lib_state[lib] = None elif r and r.get("probe") == "missing": lib_state[lib] = ("not installed", r["msg"]) elif r and r.get("probe") == "unsupported": lib_state[lib] = ("UNSUPPORTED", "this APScheduler does not accept " + lib + " time zones: " + r["msg"]) elif r and r.get("probe") == "invalid_cron": lib_state[lib] = ("INVALID CRON", r["msg"]) elif timed_out: lib_state[lib] = ("ERROR", f"loading APScheduler with {lib} gave no answer within {args.timeout:g}s (stopped)") else: lib_state[lib] = ("ERROR", (r or {}).get("error") or (err.strip().splitlines() or [f"exit status {rc}"])[-1][:160]) checked = [lib for lib in asked if lib_state[lib] is None] jobs = [(lib, sc) for lib in checked for sc in SCENARIOS] got = list(ex.map(lambda j: runner.run(dict(base, lib=j[0], scenario=j[1][0])), jobs)) except KeyboardInterrupt: runner.stop_all(); ex.shutdown(wait=True, cancel_futures=True) print("interrupted; check processes stopped", file=sys.stderr); return 130 ex.shutdown() caps = {json.dumps(c, sort_keys=True): c for c in runner.caps}.values() def cap_text(c): if c["status"] == "applied": return f"applied, {c['mb']} MB address space per check process" if c["status"] == "refused": return f"refused by this system ({c['why']}); only the time limit applies" if c["status"] == "unavailable": return "unavailable here (no RLIMIT_AS); only the time limit applies" return "off (--mem-mb 0); only the time limit applies" P(f"memory limit: {'; '.join(cap_text(c) for c in caps) or 'unknown (no check process reported it)'}" f" | time limit {args.timeout:g} s per situation | {max(1, args.jobs)} at a time") P("zone libraries checked: " + (", ".join(checked) or "none")) worst = 0; rows = []; per_lib = {} if any(lib_state[lib] and lib_state[lib][0] == "INVALID CRON" for lib in asked): msg = next(lib_state[lib][1] for lib in asked if lib_state[lib] and lib_state[lib][0] == "INVALID CRON") P(f"crontab not valid: {msg}") P("RESULT: not checked, the crontab is not valid") if args.json: print(json.dumps(dict(out, fold_utc=F.isoformat(), span_s=span_s, memory_limit=list(caps), checked=[], rows=[], per_lib={lib: "INVALID CRON" for lib in asked}, error=msg, exit=2))) return 2 for lib in asked: if lib_state[lib] is None: continue status, msg = lib_state[lib] P(f" {lib:8} {status}: {msg} -- not checked") per_lib[lib] = status + " (not checked)" if status != "not installed" or args.zone_lib != "installed": worst = 2 for (lib, sc), g in zip(jobs, got): v, note = classify(dict(base, lib=lib, scenario=sc[0]), g, args) rows.append(dict(zone_lib=lib, scenario=sc[0], verdict=v, note=note)) P(f" {lib:8} {sc[0]:4} {v:11} {note} [{sc[1]}]") if v in ("LOOP", "BACKWARDS", "SKIPS"): worst = max(worst, 1) elif v == "ERROR": worst = 2 for lib in checked: got_v = [r["verdict"] for r in rows if r["zone_lib"] == lib] bad = [v for v in SEVERITY if v in got_v] n_na = got_v.count("N/A") if bad: per_lib[lib] = "+".join(bad) elif got_v.count("OK"): per_lib[lib] = "OK" + (f" ({got_v.count('OK')} OK, {n_na} N/A)" if n_na else "") else: per_lib[lib] = "nothing checked (all N/A)"; worst = 2 if not checked: worst = 2 per_lib = {lib: per_lib[lib] for lib in asked} if args.json: print(json.dumps(dict(out, fold_utc=F.isoformat(), span_s=span_s, memory_limit=list(caps), checked=checked, rows=rows, per_lib=per_lib, exit=worst))) else: print("RESULT: " + ", ".join(f"{lib}={v}" for lib, v in per_lib.items()) + f" (exit {worst})") return worst if __name__ == "__main__": sys.exit(main())