From 76c9781a1795eda5d19cb33f0478009369c067ef Mon Sep 17 00:00:00 2001 From: William Schoenell Date: Tue, 21 Jul 2026 15:15:41 -0700 Subject: [PATCH 01/10] scheduler: fix alt/az pointing and a lost START during stop Two independent faults, both hit in production on the LNA 0.4 m tonight. PointHandler crashed on every alt/az point action with "'float' object has no attribute 'to_d'". Position.alt/az return floats (they are Coord.deg) while .ra/.dec return Coords, so the .to_d() applied to all four is only correct for the ra/dec pair. The ra/dec path is the common one, which is why this survived; robobs parks with an alt/az point, so it failed every time its queue emptied. Machine discarded a START requested while executor.stop() was running. That call blocks until the running action gives up - for a camera the rest of the exposure plus its readout, minutes - and the machine thread is inside it, not reading the state. A chimera-sched --start in that window only set the variable, and the unconditional state(OFF) after the stop then threw it away: the scheduler stayed dead and the CLI appeared to hang. Fall through to OFF only if nothing else was requested. Both are covered by tests that fail without the corresponding fix. --- src/chimera/controllers/scheduler/handlers.py | 11 ++- src/chimera/controllers/scheduler/machine.py | 10 ++- .../test_scheduler_machine_stop.py | 76 +++++++++++++++++++ .../controllers/test_scheduler_point_altaz.py | 58 ++++++++++++++ 4 files changed, 152 insertions(+), 3 deletions(-) create mode 100644 tests/chimera/controllers/test_scheduler_machine_stop.py create mode 100644 tests/chimera/controllers/test_scheduler_point_altaz.py diff --git a/src/chimera/controllers/scheduler/handlers.py b/src/chimera/controllers/scheduler/handlers.py index faa5c748..9b507763 100644 --- a/src/chimera/controllers/scheduler/handlers.py +++ b/src/chimera/controllers/scheduler/handlers.py @@ -53,9 +53,16 @@ def process(action): float(ra_dec.ra.to_h()), float(ra_dec.dec.to_d()), 2000.0 ) # epoch is always 2000.0 for pointing elif action.target_alt_az is not None: + # Position.alt/az are already floats in degrees (they return + # Coord.deg), unlike .ra/.dec which are Coords - so calling + # .to_d() on them raised "'float' object has no attribute + # 'to_d'" for every alt/az point action. hasattr keeps it + # working if a Coord is ever passed instead. + alt = action.target_alt_az.alt + az = action.target_alt_az.az telescope.slew_to_alt_az( - float(action.target_alt_az.alt.to_d()), - float(action.target_alt_az.az.to_d()), + float(alt.to_d() if hasattr(alt, "to_d") else alt), + float(az.to_d() if hasattr(az, "to_d") else az), ) elif action.target_name is not None: telescope.slew_to_object(action.target_name) diff --git a/src/chimera/controllers/scheduler/machine.py b/src/chimera/controllers/scheduler/machine.py index c85d961e..ef355bca 100644 --- a/src/chimera/controllers/scheduler/machine.py +++ b/src/chimera/controllers/scheduler/machine.py @@ -83,7 +83,15 @@ def run(self): elif self.state() == State.STOP: log.debug("[stop] trying to stop current program") self.executor.stop() - self.state(State.OFF) + # executor.stop() blocks until the running action gives up - + # for a camera that is the rest of the exposure plus readout. + # A START requested in that window (chimera-sched --start) + # only flips the state variable, because this thread is not + # reading it; dropping unconditionally to OFF here threw that + # request away, so the scheduler stayed dead and the CLI + # looked like it did nothing. + if self.state() == State.STOP: + self.state(State.OFF) elif self.state() == State.SHUTDOWN: log.debug("[shutdown] trying to stop current program") diff --git a/tests/chimera/controllers/test_scheduler_machine_stop.py b/tests/chimera/controllers/test_scheduler_machine_stop.py new file mode 100644 index 00000000..0b51b3ce --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_machine_stop.py @@ -0,0 +1,76 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# SPDX-FileCopyrightText: 2006-present Paulo Henrique Silva + +"""A START requested while executor.stop() runs must not be discarded. + +executor.stop() blocks until the running action gives up - for a camera +that is the rest of the exposure plus its readout. The machine thread sits +inside that call and is not reading the state, so a chimera-sched --start +in that window only sets the variable. Dropping unconditionally to OFF +afterwards threw the request away: the scheduler stayed dead and the CLI +looked like it had done nothing. +""" + +import threading + +from chimera.controllers.scheduler.machine import Machine +from chimera.controllers.scheduler.states import State + + +class Controller: + def state_changed(self, new, old): + pass + + +class Scheduler: + """Records that the machine acted on a START.""" + + def __init__(self): + self.rescheduled = threading.Event() + + def reschedule(self, machine): + self.rescheduled.set() + + def __next__(self): + return None + + +class Executor: + """stop() takes time, and an operator asks for a START while it runs.""" + + def __init__(self): + self.machine = None + self.stopped = threading.Event() + + def __start__(self): + pass + + def stop(self): + self.stopped.set() + self.machine.state(State.START) + + +def test_start_requested_during_stop_is_honoured(): + executor = Executor() + scheduler = Scheduler() + machine = Machine(scheduler, executor, Controller()) + executor.machine = machine + machine.daemon = True + + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + + # let it settle into OFF, then ask it to stop the running program + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + machine.state(State.STOP) + + assert executor.stopped.wait(10), "executor.stop() was never called" + assert scheduler.rescheduled.wait(10), ( + "the START requested during executor.stop() was discarded" + ) + + machine.state(State.SHUTDOWN) + thread.join(timeout=10) diff --git a/tests/chimera/controllers/test_scheduler_point_altaz.py b/tests/chimera/controllers/test_scheduler_point_altaz.py new file mode 100644 index 00000000..7de9cd45 --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_point_altaz.py @@ -0,0 +1,58 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# SPDX-FileCopyrightText: 2006-present Paulo Henrique Silva + +"""A point action given alt/az must reach the telescope as plain degrees. + +Position.alt/az return floats (Coord.deg) while .ra/.dec return Coords, so +the handler calling .to_d() on all four raised +"'float' object has no attribute 'to_d'" for every alt/az point. +""" + +from types import SimpleNamespace + +import pytest + +from chimera.controllers.scheduler.handlers import PointHandler +from chimera.util.position import Position + + +class FakeTelescope: + def __init__(self): + self.alt_az = None + + def slew_to_alt_az(self, alt, az): + self.alt_az = (alt, az) + + +def _action(position): + return SimpleNamespace( + target_ra_dec=None, + target_alt_az=position, + target_name=None, + offset_ns=None, + offset_ew=None, + dome_tracking=None, + dome_az=None, + pa=None, + ) + + +def test_position_alt_az_are_floats(): + # the premise of the bug: these are NOT Coords + position = Position.from_alt_az(88.0, 89.0) + assert isinstance(position.alt, float) + assert isinstance(position.az, float) + + +@pytest.fixture +def telescope(monkeypatch): + fake = FakeTelescope() + monkeypatch.setattr(PointHandler, "telescope", fake, raising=False) + monkeypatch.setattr(PointHandler, "dome", None, raising=False) + monkeypatch.setattr(PointHandler, "rotator", None, raising=False) + return fake + + +def test_point_alt_az_slews(telescope): + PointHandler.process(_action(Position.from_alt_az(88.0, 89.0))) + assert telescope.alt_az == (88.0, 89.0) From 339cbcd7cba370fdd35dbcefae07e5c1be8225f7 Mon Sep 17 00:00:00 2001 From: William Schoenell Date: Tue, 21 Jul 2026 16:28:08 -0700 Subject: [PATCH 02/10] scheduler: never run the same program twice concurrently A start() arriving while a program was executing forked a second execution of it. START overwrites BUSY, so the machine dropped back through IDLE, next(scheduler) handed back the SAME program - still unfinished, so still first in the queue - and _process spawned another thread for it. chimera-robobs calls scheduler.start() once per program it queues, so this compounds fast. Seen on the LNA 0.4 m: five concurrent autofocus runs and four concurrent sky flats racing on one camera, all surfacing together at shutdown as repeated "[error] #4 ... (Error while autofocusing)" for a single program id. From the outside it looks like an endless focus loop. Keep a handle on the worker thread and stay BUSY while it is alive. --- src/chimera/controllers/scheduler/machine.py | 19 ++++- .../test_scheduler_no_duplicate.py | 80 +++++++++++++++++++ 2 files changed, 98 insertions(+), 1 deletion(-) create mode 100644 tests/chimera/controllers/test_scheduler_no_duplicate.py diff --git a/src/chimera/controllers/scheduler/machine.py b/src/chimera/controllers/scheduler/machine.py index ef355bca..27f3bd35 100644 --- a/src/chimera/controllers/scheduler/machine.py +++ b/src/chimera/controllers/scheduler/machine.py @@ -23,6 +23,9 @@ def __init__(self, scheduler, executor, controller): self.controller = controller self.current_program = None + # handle on the thread running the current program, so the IDLE + # branch can tell whether one is still in flight + self._worker = None self.daemon = False @@ -58,6 +61,19 @@ def run(self): self.state(State.IDLE) if self.state() == State.IDLE: + # A program already executing must not be picked again. + # START overwrites BUSY, so every start() arriving while a + # program ran sent us back through here, next(scheduler) + # returned the SAME still-unfinished program and _process + # forked another thread for it. Seen live: robobs calls + # start() once per program it queues, and five concurrent + # autofocus runs plus four concurrent sky flats were racing + # on one camera. + if self._worker is not None and self._worker.is_alive(): + log.debug("[idle] a program is still running; staying busy") + self.state(State.BUSY) + continue + log.debug("[idle] looking for something to do...") # find something to do @@ -191,6 +207,7 @@ def process(): session.commit() - t = threading.Thread(target=process) + t = threading.Thread(target=process, name="scheduler-program") t.daemon = False + self._worker = t t.start() diff --git a/tests/chimera/controllers/test_scheduler_no_duplicate.py b/tests/chimera/controllers/test_scheduler_no_duplicate.py new file mode 100644 index 00000000..a3e8c411 --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_no_duplicate.py @@ -0,0 +1,80 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""A start() arriving while a program runs must not fork a second execution. + +robobs calls scheduler.start() once per program it queues. START overwrites +BUSY, so the machine re-entered IDLE, next(scheduler) handed back the SAME +still-unfinished program and _process forked another thread for it - five +concurrent autofocus runs and four concurrent sky flats on one camera. +""" + +import threading + +from chimera.controllers.scheduler.machine import Machine +from chimera.controllers.scheduler.states import State + + +class Controller: + def state_changed(self, new, old): + pass + + +class Scheduler: + """Always offers the same program, as a real queue does until it finishes.""" + + def __init__(self, program): + self.program = program + + def reschedule(self, machine): + pass + + def __next__(self): + return self.program + + +class Executor: + def __start__(self): + pass + + def stop(self): + pass + + +def test_start_while_busy_does_not_duplicate(monkeypatch): + started = [] + release = threading.Event() + + program = type("P", (), {"id": 1, "start_at": None})() + machine = Machine(Scheduler(program), Executor(), Controller()) + machine.daemon = True + + def fake_process(prog): + started.append(prog) + t = threading.Thread(target=lambda: release.wait(10), daemon=True) + machine._worker = t + t.start() + + monkeypatch.setattr(machine, "_process", fake_process) + + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + + machine.state(State.START) # first ingestion + for _ in range(100): + if started: + break + threading.Event().wait(0.05) + assert len(started) == 1 + + for _ in range(5): # robobs queues more + machine.state(State.START) + threading.Event().wait(0.1) + + assert len(started) == 1, f"program forked {len(started)} concurrent executions" + + release.set() + machine.state(State.SHUTDOWN) From 267bad65281c39331df0ae263d39b5d399c66d09 Mon Sep 17 00:00:00 2001 From: William Schoenell Date: Wed, 22 Jul 2026 15:59:00 -0700 Subject: [PATCH 03/10] scheduler: survive worker exit, cancellable waits, bounded abort, time order Four further machine fixes and the inline tracking stop, all from live nights on the LNA 0.4 m (2026-07-22), each with a discriminating test: - The duplicate guard slept on the condition variable, but the worker signals IDLE from inside itself just before exiting - the wakeup was consumed before the sleep and the machine parked forever. Poll the worker (join 1s) instead. - executor.stop() blocks until the running action gives up (280 s for a camera readout) and ran on the machine thread: the state machine froze and a --start during the abort was invisible. The abort now runs on its own thread bounded by STOP_ABORT_TIMEOUT. - A program waiting for a future slew_at sat in an uninterruptible time.sleep(); --stop could not reach it (no current action to abort). The wait is a cancellable event; STOP/SHUTDOWN release it and the program completes ABORTED. - SequentialScheduler ordered by priority alone, so a high-priority program with a future start_at was picked hours early and waited on the machine with the whole queue behind it. Execution follows start_at; priority breaks ties; programs without start times keep the old ordering. - The end-of-program tracking stop moves here from the robobs plugin, inline on the program thread (stop_tracking_on_program_end, default on): the detached-thread stop raced the next program's slew and untracked the freshly acquired target. --- .../controllers/scheduler/controller.py | 5 + src/chimera/controllers/scheduler/machine.py | 106 +++++++- .../controllers/scheduler/sequential.py | 9 +- .../test_scheduler_no_duplicate.py | 183 +++++++++++++ .../test_scheduler_tracking_stop.py | 250 ++++++++++++++++++ 5 files changed, 544 insertions(+), 9 deletions(-) create mode 100644 tests/chimera/controllers/test_scheduler_tracking_stop.py diff --git a/src/chimera/controllers/scheduler/controller.py b/src/chimera/controllers/scheduler/controller.py index e2672590..80aa6adf 100644 --- a/src/chimera/controllers/scheduler/controller.py +++ b/src/chimera/controllers/scheduler/controller.py @@ -30,10 +30,15 @@ class Scheduler(ChimeraObject): "dome": "/Dome/0", "autofocus": "/Autofocus/0", "autoflat": "/Autoflat/0", + "autoguider": "/Autoguider/0", "point_verify": "/PointVerify/0", "operator": "/Operator/0", "site": "/Site/0", "algorithm": SchedulingAlgorithm.SEQUENTIAL, + # leave the mount idle once a program ends, however it ended. Default + # on: a program that finishes with tracking left running walks the + # mount into a limit unattended. + "stop_tracking_on_program_end": True, } def __init__(self): diff --git a/src/chimera/controllers/scheduler/machine.py b/src/chimera/controllers/scheduler/machine.py index 27f3bd35..74046f01 100644 --- a/src/chimera/controllers/scheduler/machine.py +++ b/src/chimera/controllers/scheduler/machine.py @@ -1,6 +1,5 @@ import logging import threading -import time from chimera.controllers.scheduler.model import Program, Session from chimera.controllers.scheduler.states import State @@ -9,6 +8,11 @@ log = logging.getLogger(__name__) +#: how long the state machine waits for an abort before carrying on. The +#: abort itself keeps running in the background; this only bounds how long +#: the machine is unable to see a new START. +STOP_ABORT_TIMEOUT = 30.0 + class Machine(threading.Thread): __state = None @@ -26,6 +30,12 @@ def __init__(self, scheduler, executor, controller): # handle on the thread running the current program, so the IDLE # branch can tell whether one is still in flight self._worker = None + # set by STOP/SHUTDOWN to cancel a program that is still waiting + # for its slew time. Without it the wait was an uninterruptible + # time.sleep(): a program queued for 07:50 held the machine for + # 90 minutes and --stop could not touch it, because executor.stop() + # only aborts the CURRENT action and this one had not started any. + self._cancel_wait = threading.Event() self.daemon = False @@ -36,13 +46,17 @@ def state(self, state=None): return self.__state if state == self.__state: return - self.controller.state_changed(state, self.__state) - log.debug(f"Changing state, from {self.__state} to {state}.") + old_state = self.__state + log.debug(f"Changing state, from {old_state} to {state}.") self.__state = state self.wake_up() finally: self.__state_lock.release() + # publish OUTSIDE the lock: a slow event subscriber must not be able + # to block a concurrent state() call (e.g. stop() racing the worker) + self.controller.state_changed(state, old_state) + def run(self): log.info("Starting scheduler machine") self.state(State.OFF) @@ -70,8 +84,14 @@ def run(self): # autofocus runs plus four concurrent sky flats were racing # on one camera. if self._worker is not None and self._worker.is_alive(): - log.debug("[idle] a program is still running; staying busy") - self.state(State.BUSY) + # Wait for it rather than picking another program. POLL, + # do not sleep on the condition variable: the worker sets + # IDLE from inside its own thread just before exiting, so + # the wakeup can arrive before we sleep and be lost - the + # machine then parked forever with nothing left to wake + # it (seen live 2026-07-22, right after an autofocus + # failed and its worker signalled IDLE on the way out). + self._worker.join(1.0) continue log.debug("[idle] looking for something to do...") @@ -83,6 +103,7 @@ def run(self): log.debug("[idle] there is something to do, processing...") log.debug("[idle] program slew start %s", program.start_at) self.state(State.BUSY) + self._cancel_wait.clear() self.current_program = program self._process(program) continue @@ -98,7 +119,25 @@ def run(self): elif self.state() == State.STOP: log.debug("[stop] trying to stop current program") - self.executor.stop() + # release a program still counting down to its slew time + self._cancel_wait.set() + # Run the abort OFF this thread. It reaches the camera through + # a proxy, and that request cannot be served until the current + # exposure and its readout finish - 280 s observed on a QHY600. + # Doing it inline froze the whole state machine for that long: + # a chimera-sched --start in the meantime was invisible and the + # scheduler looked permanently wedged (2026-07-22). + stopper = threading.Thread( + target=self.executor.stop, name="scheduler-stop", daemon=True + ) + stopper.start() + stopper.join(STOP_ABORT_TIMEOUT) + if stopper.is_alive(): + log.warning( + "[stop] abort still running after %.0f s; carrying on " + "(it will finish in the background)", + STOP_ABORT_TIMEOUT, + ) # executor.stop() blocks until the running action gives up - # for a camera that is the rest of the exposure plus readout. # A START requested in that window (chimera-sched --start) @@ -111,6 +150,7 @@ def run(self): elif self.state() == State.SHUTDOWN: log.debug("[shutdown] trying to stop current program") + self._cancel_wait.set() self.executor.stop() log.debug("[shutdown] should die soon.") break @@ -138,6 +178,32 @@ def restart_all_programs(self): session.commit() + def _stop_tracking(self): + """Leave the mount idle at the end of a program. + + Called inline on the program thread, before program_complete and + before the machine returns to IDLE. Ordering matters: issued from a + detached thread instead, the stop blocks on the telescope lock behind + the *next* program's slew and lands after it, untracking the target + that program just acquired (seen live 2026-07-21, robobs). + """ + if not self.controller["stop_tracking_on_program_end"]: + return + + location = self.controller["telescope"] + # a string-typed chimera config key coerces None to "None" + if not location or str(location).lower() in ("none", ""): + return + + try: + telescope = self.controller.get_proxy(location) + if telescope.is_tracking(): + telescope.stop_tracking() + log.info("Tracking stopped at program end.") + except Exception: + # never let this fail the program that just ran + log.exception("Could not stop telescope tracking at program end.") + def _process(self, program): def process(): # session to be used by executor and handlers @@ -154,12 +220,33 @@ def process(): if program.start_at: wait_time = (program.start_at - now_mjd) * 86.4e3 if wait_time > 0.0: + # wait_time is in the site's (possibly fast-forwarded) + # seconds; sleep the equivalent REAL time so a scaled + # clock actually compresses the wait instead of sleeping + # sim-seconds as wall-seconds. speedup is 1.0 normally. + try: + speedup = float(site.time_speedup()) + except Exception: + speedup = 1.0 + real_wait = wait_time / speedup if speedup > 0 else wait_time log.debug( "[start] Waiting until MJD %f to start slewing", program.start_at, ) - log.debug("[start] Will wait for %f seconds", wait_time) - time.sleep(wait_time) + log.debug( + "[start] Will wait %f s (sim) = %f s (real, %gx)", + wait_time, + real_wait, + speedup, + ) + if self._cancel_wait.wait(real_wait): + log.debug("[start] wait cancelled; abandoning %s", str(task)) + self.controller.program_complete( + program.id, + SchedulerStatus.ABORTED, + "Aborted while waiting for its slew time.", + ) + return else: if program.valid_for >= 0.0: if -wait_time > program.valid_for: @@ -188,10 +275,12 @@ def process(): self.executor.execute(task) log.debug(f"[finish] {str(task)}") self.scheduler.done(task) + self._stop_tracking() self.controller.program_complete(program.id, SchedulerStatus.OK) self.state(State.IDLE) except ProgramExecutionException as e: self.scheduler.done(task, error=e) + self._stop_tracking() self.controller.program_complete( program.id, SchedulerStatus.ERROR, str(e) ) @@ -199,6 +288,7 @@ def process(): log.debug(f"[error] {str(task)} ({str(e)})") except ProgramExecutionAborted as e: self.scheduler.done(task, error=e) + self._stop_tracking() self.controller.program_complete( program.id, SchedulerStatus.ABORTED, "Aborted by user." ) diff --git a/src/chimera/controllers/scheduler/sequential.py b/src/chimera/controllers/scheduler/sequential.py index d157e462..e760526c 100644 --- a/src/chimera/controllers/scheduler/sequential.py +++ b/src/chimera/controllers/scheduler/sequential.py @@ -21,9 +21,16 @@ def reschedule(self, machine): session = Session() # FIXME: remove noqa + # Execution order: start_at first, priority as the tie-break. + # Ordering by priority alone starved the night when a program with + # a FUTURE start_at outranked everything: the machine picked the + # morning sky flat at 22:00 and waited 10 h on it while the whole + # night's science sat queued behind (2026-07-22). Programs without + # a start_at (hand-written queues) sort first and keep the old + # priority behaviour among themselves. programs = ( session.query(Program) - .order_by(desc(Program.priority)) + .order_by(Program.start_at.asc().nullsfirst(), desc(Program.priority)) .filter(Program.finished == False) # noqa .all() ) diff --git a/tests/chimera/controllers/test_scheduler_no_duplicate.py b/tests/chimera/controllers/test_scheduler_no_duplicate.py index a3e8c411..c38a3baf 100644 --- a/tests/chimera/controllers/test_scheduler_no_duplicate.py +++ b/tests/chimera/controllers/test_scheduler_no_duplicate.py @@ -78,3 +78,186 @@ def fake_process(prog): release.set() machine.state(State.SHUTDOWN) + + +def test_machine_recovers_when_the_worker_exits(): + """The worker signals IDLE from inside itself, just before exiting. + + The first version of the duplicate guard answered that by flipping to + BUSY and sleeping on the condition variable - but the wakeup had + already been delivered, so nothing woke the machine again and it parked + forever. Live on 2026-07-22: an autofocus failed, its worker set IDLE on + the way out, and chimera-sched --start did nothing thereafter. + """ + picked = [] + program = type("P", (), {"id": 1, "start_at": None})() + + class OneShotScheduler: + def reschedule(self, machine): + pass + + def __next__(self): + return program if len(picked) < 2 else None + + machine = Machine(OneShotScheduler(), Executor(), Controller()) + machine.daemon = True + + def fake_process(prog): + picked.append(prog) + + def body(): + # signal IDLE from inside the worker, as the real one does, and + # stay alive a moment longer - that window is the race: the + # machine sees IDLE while this thread is still running + machine.state(State.IDLE) + threading.Event().wait(0.4) + + t = threading.Thread(target=body, daemon=True) + machine._worker = t + t.start() + + machine._process = fake_process + + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + machine.state(State.START) + + # the machine must go on to pick the SECOND program, not park + for _ in range(120): + if len(picked) >= 2: + break + threading.Event().wait(0.05) + + assert len(picked) >= 2, "machine parked after the worker exited" + machine.state(State.SHUTDOWN) + + +def test_start_survives_a_slow_abort(monkeypatch): + """A --start must be acted on even while executor.stop() is still running. + + stop() reaches the camera through a proxy, which cannot be served until + the exposure and its readout finish - 280 s on a QHY600. Run inline it + froze the machine for that long and chimera-sched --start did nothing + (2026-07-22: --new, --stop, --new left the scheduler wedged). + """ + from chimera.controllers.scheduler import machine as machine_module + + monkeypatch.setattr(machine_module, "STOP_ABORT_TIMEOUT", 0.3) + + rescheduled = threading.Event() + releasing = threading.Event() + + class SlowExecutor: + def __start__(self): + pass + + def stop(self): + releasing.wait(30) # the camera readout + + class Scheduler: + def reschedule(self, machine): + rescheduled.set() + + def __next__(self): + return None + + machine = Machine(Scheduler(), SlowExecutor(), Controller()) + machine.daemon = True + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + + machine.state(State.STOP) + threading.Event().wait(0.1) + machine.state(State.START) # operator asks while the abort blocks + + assert rescheduled.wait(5), "the START was not acted on during the abort" + + releasing.set() + machine.state(State.SHUTDOWN) + + +def test_stop_cancels_a_program_waiting_for_its_slew_time(): + """--stop must release a program that is only counting down. + + executor.stop() aborts the CURRENT action, but a program queued for a + future slew_at has not started any: it sat in an uninterruptible + time.sleep(). On 2026-07-22 a focus queued for 07:50 held the machine + for 90 minutes, and a freshly loaded queue could not run because the + IDLE guard was waiting on that same worker. + """ + machine = Machine( + type( + "S", (), {"reschedule": lambda self, m: None, "__next__": lambda self: None} + )(), + Executor(), + Controller(), + ) + + finished = threading.Event() + + def waiter(): + # what _process does while counting down to slew_at + if machine._cancel_wait.wait(60): + finished.set() + + machine._worker = threading.Thread(target=waiter, daemon=True) + machine._worker.start() + threading.Event().wait(0.1) + assert machine._worker.is_alive() + + # run() sets OFF as its very first action, so the machine has to be up + # before STOP is requested or it is simply overwritten + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + + machine.state(State.STOP) + + assert finished.wait(10), "the waiting program was not released by STOP" + machine.state(State.SHUTDOWN) + + +def test_sequential_runs_in_time_order(): + """A queue with baked start times must execute chronologically. + + Priority-only ordering starved the night: the morning sky flat + (highest priority, start_at 08:20 next day) was picked at 22:00 and + the machine waited 10 h on it with the whole night queued behind + (2026-07-22). start_at orders execution; priority only breaks ties. + """ + from chimera.controllers.scheduler.model import Program, Session + from chimera.controllers.scheduler.sequential import SequentialScheduler + + session = Session() + session.query(Program).delete() + session.add(Program(name="morning-flat", priority=0, start_at=61244.35)) + session.add(Program(name="science", priority=-21, start_at=61243.93)) + session.add(Program(name="evening-flat", priority=0, start_at=61243.86)) + session.commit() + + scheduler = SequentialScheduler() + scheduler.reschedule(type("M", (), {"wake_up": staticmethod(lambda: None)})()) + + order = [] + while True: + program = next(scheduler) + if program is None: + break + order.append(program.name) + assert order == ["evening-flat", "science", "morning-flat"], order + + session.query(Program).delete() + session.commit() diff --git a/tests/chimera/controllers/test_scheduler_tracking_stop.py b/tests/chimera/controllers/test_scheduler_tracking_stop.py new file mode 100644 index 00000000..5c186f6c --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_tracking_stop.py @@ -0,0 +1,250 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""The scheduler leaves the mount idle when a program ends, however it ended. + +This used to live in robobs, which stopped tracking from a detached thread on +program_complete. is_tracking() is unlocked but stop_tracking() is not, so the +thread sailed past the check and then blocked on the telescope lock behind the +*next* program's slew - landing after it, on a target that had just been +acquired. TheSkyX turns tracking on at the end of an RA/Dec slew, so the stale +stop silently untracked the new target and the dome dropped its lookup table. + +Live on 2026-07-21 (opd-40): program #2 finished at 23:18:38.66, the next +program's 47.6 s slew held the lock, and "Tracking stopped" landed at +23:19:25.843 - during the following program's autofocus. + +The fix is ordering, not timing: the stop runs inline on the program thread, +before program_complete and while the machine is still BUSY. These tests give +stop_tracking a delay standing in for that lock wait, so a stop moved back onto +its own thread loses the race and fails them. +""" + +import threading +import time + +import pytest + +from chimera.controllers.scheduler import machine as machine_module +from chimera.controllers.scheduler.machine import Machine +from chimera.controllers.scheduler.states import State +from chimera.controllers.scheduler.status import SchedulerStatus +from chimera.core.exceptions import ProgramExecutionAborted, ProgramExecutionException + +#: stands in for the telescope lock held by the next program's slew (47.6 s +#: live). Long enough that a detached stop lands after program_complete. +LOCK_WAIT = 0.2 + + +class FakeTelescope: + """Records into the shared timeline so ordering can be asserted.""" + + def __init__(self, timeline, tracking=True, delay=LOCK_WAIT): + self.timeline = timeline + self.tracking = tracking + self.delay = delay + self.calls = [] + + def is_tracking(self): + self.calls.append("is_tracking") + return self.tracking + + def stop_tracking(self): + time.sleep(self.delay) # the lock wait + self.calls.append("stop_tracking") + self.timeline.append("stop_tracking") + self.tracking = False + + +class FakeSite: + def mjd(self): + return 60000.0 + + def time_speedup(self): + return 1.0 + + +class Controller: + """Stands in for the Scheduler ChimeraObject.""" + + def __init__(self, timeline, telescope, stop_tracking=True): + self.timeline = timeline + self.telescope = telescope + self.events = [] + self._config = { + "site": "/Site/0", + "telescope": "/Telescope/0", + "stop_tracking_on_program_end": stop_tracking, + } + + def __getitem__(self, key): + return self._config[key] + + def get_proxy(self, location): + return FakeSite() if location == "/Site/0" else self.telescope + + def program_begin(self, program_id): + pass + + def program_complete(self, program_id, status, message=None): + self.events.append(status) + self.timeline.append("program_complete") + + def state_changed(self, new, old): + pass + + +class Scheduler: + def reschedule(self, machine): + pass + + def __next__(self): + return None + + def done(self, task, error=None): + pass + + +class Executor: + """Ends the program the way the test asks for.""" + + def __init__(self, raises=None): + self.raises = raises + + def __start__(self): + pass + + def stop(self): + pass + + def execute(self, program): + if self.raises: + raise self.raises + + +@pytest.fixture(autouse=True) +def _no_db(monkeypatch): + """_process only uses the session to merge/commit; keep the DB out.""" + + class FakeSession: + def merge(self, obj): + return obj + + def commit(self): + pass + + monkeypatch.setattr(machine_module, "Session", lambda: FakeSession()) + + +def _build(telescope_factory, stop_tracking=True, raises=None): + timeline = [] + telescope = telescope_factory(timeline) + controller = Controller(timeline, telescope, stop_tracking=stop_tracking) + machine = Machine(Scheduler(), Executor(raises=raises), controller) + return timeline, telescope, controller, machine + + +def _run(machine): + program = type("P", (), {"id": 1, "start_at": 0.0, "valid_for": -1.0})() + machine.state(State.BUSY) + machine._process(program) + machine._worker.join(10) + assert not machine._worker.is_alive(), "the program thread never finished" + + +@pytest.mark.parametrize( + "raises,expected_status", + [ + (None, SchedulerStatus.OK), + (ProgramExecutionException("boom"), SchedulerStatus.ERROR), + (ProgramExecutionAborted(), SchedulerStatus.ABORTED), + ], + ids=["completed", "errored", "aborted"], +) +def test_tracking_stops_before_the_program_is_reported_done(raises, expected_status): + """However the program ended, the stop must precede program_complete. + + program_complete is what releases the next program: robobs reschedules on + it, and the machine leaves BUSY right after. A stop issued any later is + the race this replaced. + """ + timeline, telescope, controller, machine = _build(FakeTelescope, raises=raises) + + _run(machine) + + assert telescope.calls == ["is_tracking", "stop_tracking"] + assert controller.events == [expected_status] + assert timeline == ["stop_tracking", "program_complete"], ( + f"tracking was stopped out of order: {timeline}" + ) + + +def test_the_stop_happens_while_the_machine_is_still_busy(): + """The ordering that actually prevents the race. + + While the machine is BUSY the IDLE branch cannot pick another program, so + no slew can be in flight competing for the telescope lock. + """ + seen_state = [] + holder = [] + + class WatchingTelescope(FakeTelescope): + def stop_tracking(self): + super().stop_tracking() + seen_state.append(holder[0].state()) + + timeline, telescope, controller, machine = _build(WatchingTelescope) + holder.append(machine) + + _run(machine) + + assert seen_state == [State.BUSY], ( + f"tracking was stopped in state {seen_state}, not BUSY: the next " + f"program could already have been picked" + ) + + +def test_the_stop_is_inline_not_detached(): + """A detached stop is what made it land late; it must be done on return.""" + timeline, telescope, controller, machine = _build(FakeTelescope) + before = threading.active_count() + + _run(machine) + + # commanded by the time the program thread has finished, with no helper + # thread left behind to deliver it later + assert telescope.calls == ["is_tracking", "stop_tracking"] + assert threading.active_count() <= before + + +def test_option_off_leaves_tracking_alone(): + timeline, telescope, controller, machine = _build( + FakeTelescope, stop_tracking=False + ) + + _run(machine) + + assert telescope.calls == [] + assert timeline == ["program_complete"] + + +def test_a_mount_that_is_not_tracking_is_not_commanded(): + timeline, telescope, controller, machine = _build( + lambda tl: FakeTelescope(tl, tracking=False) + ) + + _run(machine) + + assert telescope.calls == ["is_tracking"] + + +def test_a_failing_telescope_does_not_break_the_program(): + """Best effort: the program still has to be reported complete.""" + + class BrokenTelescope(FakeTelescope): + def is_tracking(self): + raise RuntimeError("no link to the mount") + + timeline, telescope, controller, machine = _build(BrokenTelescope) + + _run(machine) + + assert controller.events == [SchedulerStatus.OK] From 65a781944917d316e82ac7a27b38a214d830d4ea Mon Sep 17 00:00:00 2001 From: William Schoenell Date: Wed, 22 Jul 2026 22:31:04 -0700 Subject: [PATCH 04/10] scheduler: do not re-run a finished program from a stale queue entry reschedule() rebuilds the run queue from every unfinished row - including the program RUNNING at that moment, whose finished flag is only written at completion. robobs hands the night over one program at a time and each hand-over triggers a reschedule, so the queue routinely held a stale entry for the running program; when it finished, the pop replayed it (2026-07-23 02:21: the SAO 189035 focus ran twice back to back, slew and all). next() now re-checks each popped entry against the database and skips finished rows, and the worker commits the finished flag right after scheduler.done() - before tracking stop, completion event and the IDLE transition - so the check cannot read a stale flag while the machine is being released. --- src/chimera/controllers/scheduler/machine.py | 7 ++++ .../controllers/scheduler/sequential.py | 15 ++++++- .../test_scheduler_no_duplicate.py | 41 +++++++++++++++++++ 3 files changed, 61 insertions(+), 2 deletions(-) diff --git a/src/chimera/controllers/scheduler/machine.py b/src/chimera/controllers/scheduler/machine.py index 74046f01..7a30d934 100644 --- a/src/chimera/controllers/scheduler/machine.py +++ b/src/chimera/controllers/scheduler/machine.py @@ -275,11 +275,17 @@ def process(): self.executor.execute(task) log.debug(f"[finish] {str(task)}") self.scheduler.done(task) + # the finished flag must be on disk BEFORE the machine is + # released: the idle picker re-checks it against the + # database, and the commit at the end of this function + # raced that check (stale entries then re-ran the program) + session.commit() self._stop_tracking() self.controller.program_complete(program.id, SchedulerStatus.OK) self.state(State.IDLE) except ProgramExecutionException as e: self.scheduler.done(task, error=e) + session.commit() self._stop_tracking() self.controller.program_complete( program.id, SchedulerStatus.ERROR, str(e) @@ -288,6 +294,7 @@ def process(): log.debug(f"[error] {str(task)} ({str(e)})") except ProgramExecutionAborted as e: self.scheduler.done(task, error=e) + session.commit() self._stop_tracking() self.controller.program_complete( program.id, SchedulerStatus.ABORTED, "Aborted by user." diff --git a/src/chimera/controllers/scheduler/sequential.py b/src/chimera/controllers/scheduler/sequential.py index e760526c..30866a0e 100644 --- a/src/chimera/controllers/scheduler/sequential.py +++ b/src/chimera/controllers/scheduler/sequential.py @@ -46,8 +46,19 @@ def reschedule(self, machine): machine.wake_up() def __next__(self): - if not self.run_queue.empty(): - return self.run_queue.get() + session = Session() + while not self.run_queue.empty(): + program = self.run_queue.get() + # reschedule() rebuilds the queue from every unfinished row - + # including the program RUNNING at that moment, whose finished + # flag is only written at completion. By the time its entry is + # popped it may have finished: re-check the row or the night + # replays it (a focus ran twice back to back, 2026-07-23). + current = session.query(Program).get(program.id) + if current is None or current.finished: + self.run_queue.task_done() + continue + return program return None diff --git a/tests/chimera/controllers/test_scheduler_no_duplicate.py b/tests/chimera/controllers/test_scheduler_no_duplicate.py index c38a3baf..9d2d97e7 100644 --- a/tests/chimera/controllers/test_scheduler_no_duplicate.py +++ b/tests/chimera/controllers/test_scheduler_no_duplicate.py @@ -261,3 +261,44 @@ def test_sequential_runs_in_time_order(): session.query(Program).delete() session.commit() + + +def test_finished_program_is_not_rerun_after_reschedule(): + """A reschedule while a program runs re-enqueues that same program (its + finished flag is only written at completion), and the pop must skip the + stale entry - or the night replays it. Live on 2026-07-23 02:21: robobs + handed over the queue one program at a time, each hand-over rebuilt the + run queue while the first focus was executing, and SAO 189035 ran twice + back to back.""" + from chimera.controllers.scheduler.model import Program, Session + from chimera.controllers.scheduler.sequential import SequentialScheduler + + session = Session() + session.query(Program).delete() + session.add(Program(name="focus", priority=-2, start_at=61244.19)) + session.add(Program(name="science", priority=-19, start_at=61244.20)) + session.commit() + + machine = type("M", (), {"wake_up": staticmethod(lambda: None)})() + scheduler = SequentialScheduler() + scheduler.reschedule(machine) + running = next(scheduler) + assert running.name == "focus" + + # robobs hands over another program mid-run: the rebuilt queue holds + # the still-unfinished focus again + scheduler.reschedule(machine) + + # the focus completes, exactly as machine._process does it + worker_session = Session() + task = worker_session.merge(running) + scheduler.done(task) + worker_session.commit() + + picked = next(scheduler) + assert picked is not None, "the science program was lost with the stale entry" + assert picked.name == "science", f"finished program re-ran: {picked.name}" + + session = Session() + session.query(Program).delete() + session.commit() From 077902179cac2071bfea25330f2ce39a8ea3d8ba Mon Sep 17 00:00:00 2001 From: William Schoenell Date: Thu, 23 Jul 2026 17:21:28 -0700 Subject: [PATCH 05/10] scheduler: condense comments, drop feature-branch leakage Comments trimmed to the why; the live-night detail already lives in the PR description and commit messages. Remove the time_speedup() wait scaling and the autoguider config key: both belong to feature/fast-forward-clock-and-autoguide. On master Site.time_speedup does not exist, so the call raised and fell back to 1.0 on every timed program. sequential: use SQLAlchemy 1.4 session.get() instead of legacy Query.get(). --- .../controllers/scheduler/controller.py | 5 +- src/chimera/controllers/scheduler/handlers.py | 7 +- src/chimera/controllers/scheduler/machine.py | 88 +++++-------------- .../controllers/scheduler/sequential.py | 21 ++--- .../test_scheduler_machine_stop.py | 9 +- .../test_scheduler_no_duplicate.py | 41 +++------ .../test_scheduler_tracking_stop.py | 44 +++------- 7 files changed, 64 insertions(+), 151 deletions(-) diff --git a/src/chimera/controllers/scheduler/controller.py b/src/chimera/controllers/scheduler/controller.py index 80aa6adf..6c06ccb0 100644 --- a/src/chimera/controllers/scheduler/controller.py +++ b/src/chimera/controllers/scheduler/controller.py @@ -30,14 +30,11 @@ class Scheduler(ChimeraObject): "dome": "/Dome/0", "autofocus": "/Autofocus/0", "autoflat": "/Autoflat/0", - "autoguider": "/Autoguider/0", "point_verify": "/PointVerify/0", "operator": "/Operator/0", "site": "/Site/0", "algorithm": SchedulingAlgorithm.SEQUENTIAL, - # leave the mount idle once a program ends, however it ended. Default - # on: a program that finishes with tracking left running walks the - # mount into a limit unattended. + # left tracking, an unattended mount walks into a limit "stop_tracking_on_program_end": True, } diff --git a/src/chimera/controllers/scheduler/handlers.py b/src/chimera/controllers/scheduler/handlers.py index 9b507763..f8358b1e 100644 --- a/src/chimera/controllers/scheduler/handlers.py +++ b/src/chimera/controllers/scheduler/handlers.py @@ -53,11 +53,8 @@ def process(action): float(ra_dec.ra.to_h()), float(ra_dec.dec.to_d()), 2000.0 ) # epoch is always 2000.0 for pointing elif action.target_alt_az is not None: - # Position.alt/az are already floats in degrees (they return - # Coord.deg), unlike .ra/.dec which are Coords - so calling - # .to_d() on them raised "'float' object has no attribute - # 'to_d'" for every alt/az point action. hasattr keeps it - # working if a Coord is ever passed instead. + # Position.alt/az are plain float degrees, unlike .ra/.dec + # which are Coords; accept either alt = action.target_alt_az.alt az = action.target_alt_az.az telescope.slew_to_alt_az( diff --git a/src/chimera/controllers/scheduler/machine.py b/src/chimera/controllers/scheduler/machine.py index 7a30d934..071e9708 100644 --- a/src/chimera/controllers/scheduler/machine.py +++ b/src/chimera/controllers/scheduler/machine.py @@ -8,9 +8,8 @@ log = logging.getLogger(__name__) -#: how long the state machine waits for an abort before carrying on. The -#: abort itself keeps running in the background; this only bounds how long -#: the machine is unable to see a new START. +#: max time STOP waits for the abort; past it the abort continues in the +#: background and the machine carries on STOP_ABORT_TIMEOUT = 30.0 @@ -27,14 +26,10 @@ def __init__(self, scheduler, executor, controller): self.controller = controller self.current_program = None - # handle on the thread running the current program, so the IDLE - # branch can tell whether one is still in flight + # thread running the current program; the IDLE branch waits on it self._worker = None - # set by STOP/SHUTDOWN to cancel a program that is still waiting - # for its slew time. Without it the wait was an uninterruptible - # time.sleep(): a program queued for 07:50 held the machine for - # 90 minutes and --stop could not touch it, because executor.stop() - # only aborts the CURRENT action and this one had not started any. + # set by STOP/SHUTDOWN to release a program still waiting for its + # slew time (executor.stop() only aborts an action already started) self._cancel_wait = threading.Event() self.daemon = False @@ -53,8 +48,7 @@ def state(self, state=None): finally: self.__state_lock.release() - # publish OUTSIDE the lock: a slow event subscriber must not be able - # to block a concurrent state() call (e.g. stop() racing the worker) + # publish outside the lock: a slow subscriber must not block state() self.controller.state_changed(state, old_state) def run(self): @@ -75,22 +69,12 @@ def run(self): self.state(State.IDLE) if self.state() == State.IDLE: - # A program already executing must not be picked again. - # START overwrites BUSY, so every start() arriving while a - # program ran sent us back through here, next(scheduler) - # returned the SAME still-unfinished program and _process - # forked another thread for it. Seen live: robobs calls - # start() once per program it queues, and five concurrent - # autofocus runs plus four concurrent sky flats were racing - # on one camera. + # START overwrites BUSY, so a start() during a run lands here + # with the same unfinished program: wait for the worker + # instead of forking a duplicate execution if self._worker is not None and self._worker.is_alive(): - # Wait for it rather than picking another program. POLL, - # do not sleep on the condition variable: the worker sets - # IDLE from inside its own thread just before exiting, so - # the wakeup can arrive before we sleep and be lost - the - # machine then parked forever with nothing left to wake - # it (seen live 2026-07-22, right after an autofocus - # failed and its worker signalled IDLE on the way out). + # poll, don't sleep(): the worker signals IDLE just + # before exiting, so the wakeup can be lost self._worker.join(1.0) continue @@ -121,12 +105,9 @@ def run(self): log.debug("[stop] trying to stop current program") # release a program still counting down to its slew time self._cancel_wait.set() - # Run the abort OFF this thread. It reaches the camera through - # a proxy, and that request cannot be served until the current - # exposure and its readout finish - 280 s observed on a QHY600. - # Doing it inline froze the whole state machine for that long: - # a chimera-sched --start in the meantime was invisible and the - # scheduler looked permanently wedged (2026-07-22). + # abort off this thread: executor.stop() blocks until the + # running action yields (a full camera readout), and inline + # it froze the whole state machine for that long stopper = threading.Thread( target=self.executor.stop, name="scheduler-stop", daemon=True ) @@ -138,13 +119,8 @@ def run(self): "(it will finish in the background)", STOP_ABORT_TIMEOUT, ) - # executor.stop() blocks until the running action gives up - - # for a camera that is the rest of the exposure plus readout. - # A START requested in that window (chimera-sched --start) - # only flips the state variable, because this thread is not - # reading it; dropping unconditionally to OFF here threw that - # request away, so the scheduler stayed dead and the CLI - # looked like it did nothing. + # a START requested during the abort only flips the state + # variable; dropping unconditionally to OFF discarded it if self.state() == State.STOP: self.state(State.OFF) @@ -181,11 +157,9 @@ def restart_all_programs(self): def _stop_tracking(self): """Leave the mount idle at the end of a program. - Called inline on the program thread, before program_complete and - before the machine returns to IDLE. Ordering matters: issued from a - detached thread instead, the stop blocks on the telescope lock behind - the *next* program's slew and lands after it, untracking the target - that program just acquired (seen live 2026-07-21, robobs). + Must run inline on the program thread, before program_complete: + from a detached thread the stop queues behind the next program's + slew and untracks the target it just acquired. """ if not self.controller["stop_tracking_on_program_end"]: return @@ -220,26 +194,12 @@ def process(): if program.start_at: wait_time = (program.start_at - now_mjd) * 86.4e3 if wait_time > 0.0: - # wait_time is in the site's (possibly fast-forwarded) - # seconds; sleep the equivalent REAL time so a scaled - # clock actually compresses the wait instead of sleeping - # sim-seconds as wall-seconds. speedup is 1.0 normally. - try: - speedup = float(site.time_speedup()) - except Exception: - speedup = 1.0 - real_wait = wait_time / speedup if speedup > 0 else wait_time log.debug( "[start] Waiting until MJD %f to start slewing", program.start_at, ) - log.debug( - "[start] Will wait %f s (sim) = %f s (real, %gx)", - wait_time, - real_wait, - speedup, - ) - if self._cancel_wait.wait(real_wait): + log.debug("[start] Will wait for %f seconds", wait_time) + if self._cancel_wait.wait(wait_time): log.debug("[start] wait cancelled; abandoning %s", str(task)) self.controller.program_complete( program.id, @@ -275,10 +235,8 @@ def process(): self.executor.execute(task) log.debug(f"[finish] {str(task)}") self.scheduler.done(task) - # the finished flag must be on disk BEFORE the machine is - # released: the idle picker re-checks it against the - # database, and the commit at the end of this function - # raced that check (stale entries then re-ran the program) + # commit the finished flag before releasing the machine: + # the idle picker re-checks it against the database session.commit() self._stop_tracking() self.controller.program_complete(program.id, SchedulerStatus.OK) diff --git a/src/chimera/controllers/scheduler/sequential.py b/src/chimera/controllers/scheduler/sequential.py index 30866a0e..02c2f467 100644 --- a/src/chimera/controllers/scheduler/sequential.py +++ b/src/chimera/controllers/scheduler/sequential.py @@ -21,13 +21,10 @@ def reschedule(self, machine): session = Session() # FIXME: remove noqa - # Execution order: start_at first, priority as the tie-break. - # Ordering by priority alone starved the night when a program with - # a FUTURE start_at outranked everything: the machine picked the - # morning sky flat at 22:00 and waited 10 h on it while the whole - # night's science sat queued behind (2026-07-22). Programs without - # a start_at (hand-written queues) sort first and keep the old - # priority behaviour among themselves. + # start_at orders execution, priority breaks ties: priority alone + # let a future-timed program hold the machine with the whole night + # queued behind it. Programs without start_at sort first and keep + # the old priority order among themselves. programs = ( session.query(Program) .order_by(Program.start_at.asc().nullsfirst(), desc(Program.priority)) @@ -49,12 +46,10 @@ def __next__(self): session = Session() while not self.run_queue.empty(): program = self.run_queue.get() - # reschedule() rebuilds the queue from every unfinished row - - # including the program RUNNING at that moment, whose finished - # flag is only written at completion. By the time its entry is - # popped it may have finished: re-check the row or the night - # replays it (a focus ran twice back to back, 2026-07-23). - current = session.query(Program).get(program.id) + # reschedule() also enqueues the currently RUNNING program (its + # finished flag is only written at completion), so entries can + # be stale by the time they are popped: re-check the database + current = session.get(Program, program.id) if current is None or current.finished: self.run_queue.task_done() continue diff --git a/tests/chimera/controllers/test_scheduler_machine_stop.py b/tests/chimera/controllers/test_scheduler_machine_stop.py index 0b51b3ce..c4503fe3 100644 --- a/tests/chimera/controllers/test_scheduler_machine_stop.py +++ b/tests/chimera/controllers/test_scheduler_machine_stop.py @@ -3,12 +3,9 @@ """A START requested while executor.stop() runs must not be discarded. -executor.stop() blocks until the running action gives up - for a camera -that is the rest of the exposure plus its readout. The machine thread sits -inside that call and is not reading the state, so a chimera-sched --start -in that window only sets the variable. Dropping unconditionally to OFF -afterwards threw the request away: the scheduler stayed dead and the CLI -looked like it had done nothing. +executor.stop() blocks until the running action gives up; a START in that +window only sets the state variable, and dropping unconditionally to OFF +afterwards threw the request away. """ import threading diff --git a/tests/chimera/controllers/test_scheduler_no_duplicate.py b/tests/chimera/controllers/test_scheduler_no_duplicate.py index 9d2d97e7..b2e9b1df 100644 --- a/tests/chimera/controllers/test_scheduler_no_duplicate.py +++ b/tests/chimera/controllers/test_scheduler_no_duplicate.py @@ -1,10 +1,8 @@ # SPDX-License-Identifier: GPL-2.0-or-later """A start() arriving while a program runs must not fork a second execution. -robobs calls scheduler.start() once per program it queues. START overwrites -BUSY, so the machine re-entered IDLE, next(scheduler) handed back the SAME -still-unfinished program and _process forked another thread for it - five -concurrent autofocus runs and four concurrent sky flats on one camera. +START overwrites BUSY, so the machine re-entered IDLE, next(scheduler) handed +back the SAME still-unfinished program and _process forked another thread. """ import threading @@ -83,11 +81,8 @@ def fake_process(prog): def test_machine_recovers_when_the_worker_exits(): """The worker signals IDLE from inside itself, just before exiting. - The first version of the duplicate guard answered that by flipping to - BUSY and sleeping on the condition variable - but the wakeup had - already been delivered, so nothing woke the machine again and it parked - forever. Live on 2026-07-22: an autofocus failed, its worker set IDLE on - the way out, and chimera-sched --start did nothing thereafter. + A guard that sleeps on the condition variable misses that wakeup (it is + delivered before the sleep) and the machine parks forever. """ picked = [] program = type("P", (), {"id": 1, "start_at": None})() @@ -140,10 +135,8 @@ def body(): def test_start_survives_a_slow_abort(monkeypatch): """A --start must be acted on even while executor.stop() is still running. - stop() reaches the camera through a proxy, which cannot be served until - the exposure and its readout finish - 280 s on a QHY600. Run inline it - froze the machine for that long and chimera-sched --start did nothing - (2026-07-22: --new, --stop, --new left the scheduler wedged). + stop() blocks until the running action yields (a full camera readout); + run inline it froze the machine for that long. """ from chimera.controllers.scheduler import machine as machine_module @@ -189,11 +182,9 @@ def __next__(self): def test_stop_cancels_a_program_waiting_for_its_slew_time(): """--stop must release a program that is only counting down. - executor.stop() aborts the CURRENT action, but a program queued for a - future slew_at has not started any: it sat in an uninterruptible - time.sleep(). On 2026-07-22 a focus queued for 07:50 held the machine - for 90 minutes, and a freshly loaded queue could not run because the - IDLE guard was waiting on that same worker. + executor.stop() aborts the CURRENT action, but a program waiting for a + future start_at has not started any: it sat in an uninterruptible + time.sleep() and held the machine until its slew time. """ machine = Machine( type( @@ -233,10 +224,9 @@ def waiter(): def test_sequential_runs_in_time_order(): """A queue with baked start times must execute chronologically. - Priority-only ordering starved the night: the morning sky flat - (highest priority, start_at 08:20 next day) was picked at 22:00 and - the machine waited 10 h on it with the whole night queued behind - (2026-07-22). start_at orders execution; priority only breaks ties. + Priority-only ordering let a future-timed program hold the machine with + the whole night queued behind it: start_at orders execution, priority + only breaks ties. """ from chimera.controllers.scheduler.model import Program, Session from chimera.controllers.scheduler.sequential import SequentialScheduler @@ -265,11 +255,8 @@ def test_sequential_runs_in_time_order(): def test_finished_program_is_not_rerun_after_reschedule(): """A reschedule while a program runs re-enqueues that same program (its - finished flag is only written at completion), and the pop must skip the - stale entry - or the night replays it. Live on 2026-07-23 02:21: robobs - handed over the queue one program at a time, each hand-over rebuilt the - run queue while the first focus was executing, and SAO 189035 ran twice - back to back.""" + finished flag is only written at completion); the pop must skip the + stale entry or the night replays it.""" from chimera.controllers.scheduler.model import Program, Session from chimera.controllers.scheduler.sequential import SequentialScheduler diff --git a/tests/chimera/controllers/test_scheduler_tracking_stop.py b/tests/chimera/controllers/test_scheduler_tracking_stop.py index 5c186f6c..4ec13694 100644 --- a/tests/chimera/controllers/test_scheduler_tracking_stop.py +++ b/tests/chimera/controllers/test_scheduler_tracking_stop.py @@ -1,21 +1,13 @@ # SPDX-License-Identifier: GPL-2.0-or-later """The scheduler leaves the mount idle when a program ends, however it ended. -This used to live in robobs, which stopped tracking from a detached thread on -program_complete. is_tracking() is unlocked but stop_tracking() is not, so the -thread sailed past the check and then blocked on the telescope lock behind the -*next* program's slew - landing after it, on a target that had just been -acquired. TheSkyX turns tracking on at the end of an RA/Dec slew, so the stale -stop silently untracked the new target and the dome dropped its lookup table. - -Live on 2026-07-21 (opd-40): program #2 finished at 23:18:38.66, the next -program's 47.6 s slew held the lock, and "Tracking stopped" landed at -23:19:25.843 - during the following program's autofocus. - -The fix is ordering, not timing: the stop runs inline on the program thread, -before program_complete and while the machine is still BUSY. These tests give -stop_tracking a delay standing in for that lock wait, so a stop moved back onto -its own thread loses the race and fails them. +This used to live in robobs, on a detached thread fired at program_complete: +the stop blocked on the telescope lock behind the *next* program's slew and +landed after it, untracking the freshly acquired target (TheSkyX re-enables +tracking at slew end). The fix is ordering, not timing: the stop runs inline +on the program thread, before program_complete, while the machine is still +BUSY. These tests give stop_tracking a delay standing in for that lock wait, +so a stop moved back onto its own thread loses the race and fails them. """ import threading @@ -29,8 +21,8 @@ from chimera.controllers.scheduler.status import SchedulerStatus from chimera.core.exceptions import ProgramExecutionAborted, ProgramExecutionException -#: stands in for the telescope lock held by the next program's slew (47.6 s -#: live). Long enough that a detached stop lands after program_complete. +#: stands in for the telescope lock held by the next program's slew; long +#: enough that a detached stop lands after program_complete LOCK_WAIT = 0.2 @@ -58,9 +50,6 @@ class FakeSite: def mjd(self): return 60000.0 - def time_speedup(self): - return 1.0 - class Controller: """Stands in for the Scheduler ChimeraObject.""" @@ -160,12 +149,8 @@ def _run(machine): ids=["completed", "errored", "aborted"], ) def test_tracking_stops_before_the_program_is_reported_done(raises, expected_status): - """However the program ended, the stop must precede program_complete. - - program_complete is what releases the next program: robobs reschedules on - it, and the machine leaves BUSY right after. A stop issued any later is - the race this replaced. - """ + """However the program ended, the stop must precede program_complete, + which is what releases the next program.""" timeline, telescope, controller, machine = _build(FakeTelescope, raises=raises) _run(machine) @@ -178,11 +163,8 @@ def test_tracking_stops_before_the_program_is_reported_done(raises, expected_sta def test_the_stop_happens_while_the_machine_is_still_busy(): - """The ordering that actually prevents the race. - - While the machine is BUSY the IDLE branch cannot pick another program, so - no slew can be in flight competing for the telescope lock. - """ + """While the machine is BUSY no other program can be picked, so no slew + can be in flight competing for the telescope lock.""" seen_state = [] holder = [] From 3270f40dd12f7adf9763c6038fe17682d7c4a455 Mon Sep 17 00:00:00 2001 From: William Schoenell Date: Thu, 23 Jul 2026 17:49:32 -0700 Subject: [PATCH 06/10] scheduler: do not execute a program past its validity window The expired branch reported program_complete(OK, 'not valid anymore') and then fell through to run the program anyway: program_begin fired, the actions executed, and a second contradictory program_complete landed at the end. robobs advances its queue on the first completion, so it raced a program that was in fact still running. The expired program is now marked done, committed, and completed without executing, releasing the machine to IDLE like every other exit. Programs with valid_for < 0 (no window) keep running late, as before. Also fixes the missing f-prefix on the expiry log line by switching it to lazy %-formatting. --- src/chimera/controllers/scheduler/machine.py | 29 ++-- .../test_scheduler_expired_program.py | 131 ++++++++++++++++++ 2 files changed, 148 insertions(+), 12 deletions(-) create mode 100644 tests/chimera/controllers/test_scheduler_expired_program.py diff --git a/src/chimera/controllers/scheduler/machine.py b/src/chimera/controllers/scheduler/machine.py index 071e9708..96640be4 100644 --- a/src/chimera/controllers/scheduler/machine.py +++ b/src/chimera/controllers/scheduler/machine.py @@ -208,21 +208,26 @@ def process(): ) return else: - if program.valid_for >= 0.0: - if -wait_time > program.valid_for: - log.debug( - "[start] Program is not valid anymore {program.start_at}, {program.valid_for}" - ) - self.controller.program_complete( - program.id, - SchedulerStatus.OK, - "Program not valid anymore.", - ) - else: + if program.valid_for >= 0.0 and -wait_time > program.valid_for: + # too late to run: finish it without executing log.debug( - "[start] Specified slew start MJD %s has already passed; proceeding without waiting", + "[start] Program is not valid anymore (start_at %f, valid_for %f)", program.start_at, + program.valid_for, ) + self.scheduler.done(task) + session.commit() + self.controller.program_complete( + program.id, + SchedulerStatus.OK, + "Program not valid anymore.", + ) + self.state(State.IDLE) + return + log.debug( + "[start] Specified slew start MJD %s has already passed; proceeding without waiting", + program.start_at, + ) else: log.debug("[start] No slew time specified, so no waiting") log.debug("[start] Current MJD is %f", site.mjd()) diff --git a/tests/chimera/controllers/test_scheduler_expired_program.py b/tests/chimera/controllers/test_scheduler_expired_program.py new file mode 100644 index 00000000..f3d10bf8 --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_expired_program.py @@ -0,0 +1,131 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""A program past its validity window must be finished without running. + +The expired branch reported program_complete(OK, "not valid anymore") and +then fell through to execute the program anyway - a second, contradictory +program_complete at the end, and anything listening to the first one (robobs +advances its queue on it) raced a program that was still running. +""" + +import pytest + +from chimera.controllers.scheduler import machine as machine_module +from chimera.controllers.scheduler.machine import Machine +from chimera.controllers.scheduler.states import State +from chimera.controllers.scheduler.status import SchedulerStatus + +NOW_MJD = 60000.0 +ONE_HOUR = 1.0 / 24.0 + + +class FakeSite: + def mjd(self): + return NOW_MJD + + +class Controller: + def __init__(self): + self.begun = [] + self.events = [] + self._config = {"site": "/Site/0", "stop_tracking_on_program_end": False} + + def __getitem__(self, key): + return self._config[key] + + def get_proxy(self, location): + return FakeSite() + + def program_begin(self, program_id): + self.begun.append(program_id) + + def program_complete(self, program_id, status, message=None): + self.events.append((status, message)) + + def state_changed(self, new, old): + pass + + +class Scheduler: + def __init__(self): + self.done_calls = [] + + def reschedule(self, machine): + pass + + def __next__(self): + return None + + def done(self, task, error=None): + self.done_calls.append((task, error)) + + +class Executor: + def __init__(self): + self.executed = [] + + def __start__(self): + pass + + def stop(self): + pass + + def execute(self, program): + self.executed.append(program) + + +@pytest.fixture(autouse=True) +def _no_db(monkeypatch): + """_process only uses the session to merge/commit; keep the DB out.""" + + class FakeSession: + def merge(self, obj): + return obj + + def commit(self): + pass + + monkeypatch.setattr(machine_module, "Session", lambda: FakeSession()) + + +def _run(start_at, valid_for): + controller = Controller() + scheduler = Scheduler() + executor = Executor() + machine = Machine(scheduler, executor, controller) + program = type("P", (), {"id": 1, "start_at": start_at, "valid_for": valid_for})() + + machine.state(State.BUSY) + machine._process(program) + machine._worker.join(10) + assert not machine._worker.is_alive(), "the program thread never finished" + return controller, scheduler, executor, machine + + +def test_expired_program_is_finished_without_running(): + # start_at one hour ago, valid for one minute + controller, scheduler, executor, machine = _run(NOW_MJD - ONE_HOUR, 60.0) + + assert executor.executed == [], "an expired program was executed" + assert controller.begun == [] + assert controller.events == [(SchedulerStatus.OK, "Program not valid anymore.")], ( + f"expected a single completion, got {controller.events}" + ) + assert len(scheduler.done_calls) == 1, "the expired program was not marked done" + assert machine.state() == State.IDLE, "the machine was not released" + + +def test_late_program_without_window_still_runs(): + # valid_for < 0 means no expiry: a late start must still execute, once + controller, scheduler, executor, machine = _run(NOW_MJD - ONE_HOUR, -1.0) + + assert len(executor.executed) == 1 + assert controller.begun == [1] + assert controller.events == [(SchedulerStatus.OK, None)] + + +def test_late_program_inside_its_window_still_runs(): + # one hour late, valid for two hours + controller, scheduler, executor, machine = _run(NOW_MJD - ONE_HOUR, 7200.0) + + assert len(executor.executed) == 1 + assert controller.events == [(SchedulerStatus.OK, None)] From db847dc10858e5c89a0a3735ee5f69802c3eb00a Mon Sep 17 00:00:00 2001 From: William Schoenell Date: Sat, 25 Jul 2026 11:02:11 -0700 Subject: [PATCH 07/10] scheduler: OFF only when the worker is dead; stop cannot be re-armed The opd-40 2026-07-25 escalation (lna40 #19): the machine declared OFF while an aborting exposure loop survived, and when that program finally died with an error its completion path flipped the stopped machine back to IDLE. STOP now moves to a STOPPING state, runs the abort off-thread, and joins the worker with a heartbeat before declaring OFF; terminal transitions from the worker are compare-and-set on BUSY so a program's own completion or error can never re-arm a machine that was told to stop. executor.stop() sets must_stop even with no action in flight and execute() clears its current handler on exit, so a later stop cannot abort an action of a finished program. SequentialScheduler.done() survives the queue having been rebuilt underneath it. Replaces the 30 s STOP_ABORT_TIMEOUT give-up: waiting on the stopper thread bounded the wait for the abort *call*, but OFF was still declared with the worker alive - the machine's word is now tied to the only thing that matters, worker death. Tests: an end-to-end stop regression rig (fake camera, real bus) covering the incident sequence, unit tests for worker-death-gated OFF and the sticky terminal transition, and the scheduler database redirected to a temp file at conftest import time - the model engine binds at import, and a suite-wide run used to write into the user's real ~/.chimera database. --- src/chimera/controllers/scheduler/executor.py | 75 +++++++----- src/chimera/controllers/scheduler/machine.py | 104 +++++++++++----- .../controllers/scheduler/sequential.py | 7 +- src/chimera/controllers/scheduler/states.py | 1 + tests/chimera/conftest.py | 14 ++- .../test_scheduler_machine_stop.py | 74 +++++++++++ .../test_scheduler_no_duplicate.py | 11 +- .../controllers/test_scheduler_stop.py | 115 ++++++++++++++++++ 8 files changed, 327 insertions(+), 74 deletions(-) create mode 100644 tests/chimera/controllers/test_scheduler_stop.py diff --git a/src/chimera/controllers/scheduler/executor.py b/src/chimera/controllers/scheduler/executor.py index 7f7b36aa..5b2837b1 100644 --- a/src/chimera/controllers/scheduler/executor.py +++ b/src/chimera/controllers/scheduler/executor.py @@ -53,43 +53,54 @@ def __start__(self): def execute(self, program): self.must_stop.clear() - for action in program.actions: - # aborted? - if self.must_stop.is_set(): - raise ProgramExecutionAborted() - - t0 = time.time() - - try: - self.current_action = action - self.current_handler = self.action_handlers[type(action)] - - log_msg = str(self.current_handler.log(action)) - log.debug(f"[start] {log_msg} ") - self.controller.action_begin(action.id, log_msg) - - self.current_handler.process(action) - - # instruments just returns in case of abort, so we need to check handler - # returned 'cause of abort or not + try: + for action in program.actions: + # aborted? if self.must_stop.is_set(): - self.controller.action_complete(action.id, SchedulerStatus.ABORTED) raise ProgramExecutionAborted() - else: - self.controller.action_complete(action.id, SchedulerStatus.OK) - except ProgramExecutionException: - self.controller.action_complete(action.id, SchedulerStatus.ERROR) - raise - except KeyError: - log.debug(f"No handler to {action} action. Skipping it") - finally: - log.debug("[finish] took: %f s" % (time.time() - t0)) + t0 = time.time() + + try: + self.current_action = action + self.current_handler = self.action_handlers[type(action)] + + log_msg = str(self.current_handler.log(action)) + log.debug(f"[start] {log_msg} ") + self.controller.action_begin(action.id, log_msg) + + self.current_handler.process(action) + + # instruments just returns in case of abort, so we need to check handler + # returned 'cause of abort or not + if self.must_stop.is_set(): + self.controller.action_complete( + action.id, SchedulerStatus.ABORTED + ) + raise ProgramExecutionAborted() + else: + self.controller.action_complete(action.id, SchedulerStatus.OK) + + except ProgramExecutionException: + self.controller.action_complete(action.id, SchedulerStatus.ERROR) + raise + except KeyError: + log.debug(f"No handler to {action} action. Skipping it") + finally: + log.debug("[finish] took: %f s" % (time.time() - t0)) + finally: + # a later stop() must not abort an action of a finished program + self.current_action = None + self.current_handler = None def stop(self): - if self.current_handler: - self.must_stop.set() - self.current_handler.abort(self.current_action) + # flag first: a worker between actions must see the stop even when + # there is no action in flight to abort + self.must_stop.set() + + handler, action = self.current_handler, self.current_action + if handler: + handler.abort(action) def _inject_instrument(self, handler): if not issubclass(handler, ActionHandler): diff --git a/src/chimera/controllers/scheduler/machine.py b/src/chimera/controllers/scheduler/machine.py index 9cffd452..fa6e700f 100644 --- a/src/chimera/controllers/scheduler/machine.py +++ b/src/chimera/controllers/scheduler/machine.py @@ -8,10 +8,6 @@ log = logging.getLogger(__name__) -#: max time STOP waits for the abort; past it the abort continues in the -#: background and the machine carries on -STOP_ABORT_TIMEOUT = 30.0 - class Machine(threading.Thread): __state = None @@ -48,6 +44,27 @@ def state(self, state=None): # publish outside the lock: a slow subscriber must not block state() self.controller.state_changed(state, old_state) + def transition(self, expected, state): + """Set the state only if the machine is currently in `expected`. + + Terminal transitions coming from the worker use this so a program's + own completion or error can never re-arm a machine that was told to + stop: on opd-40 a program error flipped a stopped machine OFF -> IDLE. + """ + self.__state_lock.acquire() + try: + if self.__state != expected or state == self.__state: + return False + old_state = self.__state + log.debug(f"Changing state, from {old_state} to {state}.") + self.__state = state + self.wake_up() + finally: + self.__state_lock.release() + + self.controller.state_changed(state, old_state) + return True + def run(self): log.info("Starting scheduler machine") self.state(State.OFF) @@ -99,35 +116,50 @@ def run(self): elif self.state() == State.STOP: log.debug("[stop] trying to stop current program") - # a program still counting down to its slew time is released - # by the state change itself: the wait polls state() and - # blocks on __wake_up_call, which state() notifies + # STOPPING first: a program counting down to its slew time + # is released by the state change itself, and the abort + # below may request a new state (START) that must not be + # overwritten afterwards + self.state(State.STOPPING) # abort off this thread: executor.stop() blocks until the # running action yields (a full camera readout), and inline # it froze the whole state machine for that long - stopper = threading.Thread( + threading.Thread( target=self.executor.stop, name="scheduler-stop", daemon=True - ) - stopper.start() - stopper.join(STOP_ABORT_TIMEOUT) - if stopper.is_alive(): - log.warning( - "[stop] abort still running after %.0f s; carrying on " - "(it will finish in the background)", - STOP_ABORT_TIMEOUT, - ) - # a START requested during the abort only flips the state - # variable; dropping unconditionally to OFF discarded it - if self.state() == State.STOP: - self.state(State.OFF) + ).start() + # OFF only once the worker is dead: declaring it earlier let + # a surviving exposure loop run 52 min behind a closed dome + self._join_worker("stop") + # honour a START requested while the abort ran + self.transition(State.STOPPING, State.OFF) + + # out of the loop: SHUTDOWN. Abort whatever is running and wait for + # the worker so nothing outlives the machine's word. + log.debug("[shutdown] trying to stop current program") + threading.Thread( + target=self.executor.stop, name="scheduler-stop", daemon=True + ).start() + self._join_worker("shutdown") + log.debug("[shutdown] thread ending...") - elif self.state() == State.SHUTDOWN: - log.debug("[shutdown] trying to stop current program") - self.executor.stop() - log.debug("[shutdown] should die soon.") - break + def _join_worker(self, reason): + """Wait until the worker thread is actually dead, heartbeating: a + camera abort legitimately takes minutes, and a silent wait reads as + a wedged machine.""" + worker = self._worker + if worker is None: + return - log.debug("[shutdown] thread ending...") + waited = 0 + while worker.is_alive(): + worker.join(1.0) + waited += 1 + if worker.is_alive() and waited % 10 == 0: + log.info( + "[%s] program still aborting after ~%d s, waiting...", + reason, + waited, + ) def sleep(self): self.__wake_up_call.acquire() @@ -201,7 +233,11 @@ def process(): # state change -- so a STOP/SHUTDOWN breaks the wait at once # instead of after a fixed sleep. while site.mjd() < program.start_at: - if self.state() in (State.STOP, State.SHUTDOWN): + if self.state() in ( + State.STOP, + State.STOPPING, + State.SHUTDOWN, + ): log.debug( "[start] wait cancelled; abandoning %s", str(task) ) @@ -228,7 +264,7 @@ def process(): SchedulerStatus.OK, "Program not valid anymore.", ) - self.state(State.IDLE) + self.transition(State.BUSY, State.IDLE) return log.debug( "[start] Specified slew start MJD %s has already passed; proceeding without waiting", @@ -251,7 +287,9 @@ def process(): session.commit() self._stop_tracking() self.controller.program_complete(program.id, SchedulerStatus.OK) - self.state(State.IDLE) + # conditional: a completion event must not re-arm a machine + # that went STOPPING/SHUTDOWN while this program ran + self.transition(State.BUSY, State.IDLE) except ProgramExecutionException as e: self.scheduler.done(task, error=e) session.commit() @@ -259,7 +297,7 @@ def process(): self.controller.program_complete( program.id, SchedulerStatus.ERROR, str(e) ) - self.state(State.IDLE) + self.transition(State.BUSY, State.IDLE) log.debug(f"[error] {str(task)} ({str(e)})") except ProgramExecutionAborted as e: self.scheduler.done(task, error=e) @@ -268,7 +306,9 @@ def process(): self.controller.program_complete( program.id, SchedulerStatus.ABORTED, "Aborted by user." ) - self.state(State.OFF) + # normally the machine itself goes STOPPING -> OFF once this + # worker dies; cover an out-of-band abort arriving while BUSY + self.transition(State.BUSY, State.OFF) log.debug(f"[aborted by user] {str(task)}") session.commit() diff --git a/src/chimera/controllers/scheduler/sequential.py b/src/chimera/controllers/scheduler/sequential.py index 02c2f467..31a0c7c1 100644 --- a/src/chimera/controllers/scheduler/sequential.py +++ b/src/chimera/controllers/scheduler/sequential.py @@ -64,5 +64,10 @@ def done(self, task, error=None): else: task.finished = True - self.run_queue.task_done() + try: + self.run_queue.task_done() + except ValueError: + # the queue was rebuilt by a reschedule since this program was + # taken: its slot is gone, nothing left to account for + pass self.machine.wake_up() diff --git a/src/chimera/controllers/scheduler/states.py b/src/chimera/controllers/scheduler/states.py index 1fb223db..871661bd 100644 --- a/src/chimera/controllers/scheduler/states.py +++ b/src/chimera/controllers/scheduler/states.py @@ -9,4 +9,5 @@ class State(Enum): IDLE = "IDLE" BUSY = "BUSY" STOP = "STOP" + STOPPING = "STOPPING" SHUTDOWN = "SHUTDOWN" diff --git a/tests/chimera/conftest.py b/tests/chimera/conftest.py index 9a6bab0c..04c46c05 100644 --- a/tests/chimera/conftest.py +++ b/tests/chimera/conftest.py @@ -1,11 +1,21 @@ import random +import tempfile import threading import time import pytest -from chimera.core.bus import Bus -from chimera.core.manager import Manager +import chimera.core.constants as constants + +# Redirect the scheduler database BEFORE any test module is imported: +# scheduler.model binds its engine to DEFAULT_PROGRAM_DATABASE at import +# time, and several test modules import it during collection — patching +# from inside a test module is too late once a sibling imported first, and +# the suite would then read AND WRITE the user's real ~/.chimera database. +constants.DEFAULT_PROGRAM_DATABASE = tempfile.mkstemp(suffix="-chimera-tests.db")[1] + +from chimera.core.bus import Bus # noqa: E402 +from chimera.core.manager import Manager # noqa: E402 @pytest.fixture diff --git a/tests/chimera/controllers/test_scheduler_machine_stop.py b/tests/chimera/controllers/test_scheduler_machine_stop.py index c4503fe3..cc001d37 100644 --- a/tests/chimera/controllers/test_scheduler_machine_stop.py +++ b/tests/chimera/controllers/test_scheduler_machine_stop.py @@ -71,3 +71,77 @@ def test_start_requested_during_stop_is_honoured(): machine.state(State.SHUTDOWN) thread.join(timeout=10) + + +class PlainScheduler: + def reschedule(self, machine): + pass + + def __next__(self): + return None + + +class PlainExecutor: + def __start__(self): + pass + + def stop(self): + pass + + +def test_off_waits_for_the_worker_to_die(): + """STOP must not report OFF while the program thread is alive. + + On opd-40 the machine declared OFF while an exposure loop survived for + 52 minutes behind a closed dome: OFF is only honest once the worker is + dead, and the machine says STOPPING until then. + """ + release = threading.Event() + machine = Machine(PlainScheduler(), PlainExecutor(), Controller()) + machine.daemon = True + + worker = threading.Thread(target=lambda: release.wait(30), daemon=True) + machine._worker = worker + worker.start() + + thread = threading.Thread(target=machine.run, daemon=True) + thread.start() + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + + machine.state(State.STOP) + + # while the worker lives the machine must hold STOPPING, never OFF + threading.Event().wait(1.0) + assert worker.is_alive() + assert machine.state() == State.STOPPING + + release.set() + for _ in range(100): + if machine.state() == State.OFF: + break + threading.Event().wait(0.05) + assert machine.state() == State.OFF + + machine.state(State.SHUTDOWN) + thread.join(timeout=10) + + +def test_worker_completion_cannot_rearm_a_stopped_machine(): + """A program's own completion or error must not re-arm a stopped machine. + + On opd-40 the surviving program eventually died with an error and its + completion path flipped the stopped machine OFF -> IDLE, re-arming the + night. Terminal transitions are conditional on BUSY. + """ + machine = Machine(PlainScheduler(), PlainExecutor(), Controller()) + + machine.state(State.STOPPING) + assert machine.transition(State.BUSY, State.IDLE) is False + assert machine.state() == State.STOPPING + + machine.state(State.BUSY) + assert machine.transition(State.BUSY, State.IDLE) is True + assert machine.state() == State.IDLE diff --git a/tests/chimera/controllers/test_scheduler_no_duplicate.py b/tests/chimera/controllers/test_scheduler_no_duplicate.py index 81372181..e69d3d7c 100644 --- a/tests/chimera/controllers/test_scheduler_no_duplicate.py +++ b/tests/chimera/controllers/test_scheduler_no_duplicate.py @@ -132,16 +132,13 @@ def body(): machine.state(State.SHUTDOWN) -def test_start_survives_a_slow_abort(monkeypatch): +def test_start_survives_a_slow_abort(): """A --start must be acted on even while executor.stop() is still running. stop() blocks until the running action yields (a full camera readout); - run inline it froze the machine for that long. + run inline it froze the machine for that long. The machine waits on the + WORKER, never on the abort call, so a hung abort cannot park it either. """ - from chimera.controllers.scheduler import machine as machine_module - - monkeypatch.setattr(machine_module, "STOP_ABORT_TIMEOUT", 0.3) - rescheduled = threading.Event() releasing = threading.Event() @@ -201,7 +198,7 @@ def waiter(): # block on the machine's wake-up Condition, which state() notifies on # every change, so STOP/SHUTDOWN breaks the wait at once wake = machine._Machine__wake_up_call - while machine.state() not in (State.STOP, State.SHUTDOWN): + while machine.state() not in (State.STOP, State.STOPPING, State.SHUTDOWN): with wake: wake.wait(1.0) finished.set() diff --git a/tests/chimera/controllers/test_scheduler_stop.py b/tests/chimera/controllers/test_scheduler_stop.py new file mode 100644 index 00000000..2ae8f49b --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_stop.py @@ -0,0 +1,115 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# End-to-end regression tests for scheduler stop semantics (lna40 incident, +# opd-40 2026-07-25): a sched.stop() must actually stop the running program, +# including after a start() during BUSY re-queued the still-unfinished +# program — on master that spawned a second worker that outlived every stop +# and kept the camera exposing for 52 min behind a closed dome. + +import time + +import pytest + +# the scheduler DB is redirected to a temp file in tests/chimera/conftest.py, +# before any module import binds the model engine +from chimera.controllers.imageserver.imageserver import ImageServer +from chimera.controllers.scheduler.controller import Scheduler +from chimera.controllers.scheduler.model import Expose, Program, Session +from chimera.controllers.scheduler.states import State +from chimera.core.site import Site +from chimera.instruments.fakecamera import FakeCamera +from chimera.instruments.fakefilterwheel import FakeFilterWheel + + +@pytest.fixture +def system(manager, wait_for, tmp_path): + manager.add_class(Site, "fake") + manager.add_class( + ImageServer, "fake", {"images_dir": str(tmp_path), "httpd": False} + ) + manager.add_class(FakeCamera, "fake") + manager.add_class(FakeFilterWheel, "fake") + manager.add_class(Scheduler, "fake") + + sched = manager.get_proxy("/Scheduler/0") + camera = manager.get_proxy("/Camera/0") + + # wait for the controller's control() tick to start the machine thread + assert wait_for(lambda: sched.state() is not None, 10) + + # a clean program table for each test + session = Session() + for program in session.query(Program).all(): + session.delete(program) + session.commit() + + yield sched, camera, tmp_path + + +def add_program(tmp_path, frames=8, exptime=1): + program = Program(name="stop-test", pi="tester") + program.actions = [ + Expose( + frames=frames, + exptime=exptime, + shutter="OPEN", + image_type="object", + object_name="stop-test", + # absolute path: no ImageServer lookup surprises + filename=str(tmp_path / "frame"), + ) + ] + session = Session() + session.add(program) + session.commit() + + +class TestSchedulerStop: + def test_stop_single_worker(self, system, wait_for): + """One worker, stop mid-multi-frame-expose stops for real.""" + sched, camera, tmp_path = system + add_program(tmp_path) + + sched.start() + assert wait_for(camera.is_exposing, 15), "exposure never started" + + sched.stop() + + assert wait_for(lambda: sched.state() == State.OFF, 10) + assert not camera.is_exposing(), "camera still exposing after OFF" + + # and it must STAY stopped + time.sleep(3) + assert not camera.is_exposing(), "camera exposing again after stop" + assert sched.state() == State.OFF + + def test_stop_after_restart_during_busy(self, system, wait_for): + """The incident case: a start() while BUSY re-queues the unfinished + program; the following stop must still kill everything and OFF must + mean the camera is no longer being driven.""" + sched, camera, tmp_path = system + # long enough that a surviving execution could not finish naturally + # inside the assertion windows below (the incident program had 100s + # of frames) + add_program(tmp_path, frames=60, exptime=1) + + sched.start() + assert wait_for(camera.is_exposing, 15), "exposure never started" + + # what robobs does when (re)submitting programs mid-night + sched.start() + time.sleep(1) + + sched.stop() + + assert wait_for(lambda: sched.state() == State.OFF, 10), ( + "machine never reached OFF after stop" + ) + + # OFF must be honest: the camera stops and STAYS stopped + assert not camera.is_exposing(), ( + "camera still exposing after the scheduler reported OFF" + ) + assert not wait_for(camera.is_exposing, 5), ( + "camera started exposing again after the stop (a second worker survived)" + ) + assert sched.state() == State.OFF, "scheduler re-armed itself after stop" From e593de350ea6460be85b96c8d0a24d50f9bdf962 Mon Sep 17 00:00:00 2001 From: William Schoenell Date: Sat, 25 Jul 2026 22:57:00 -0700 Subject: [PATCH 08/10] scheduler: stop the reschedule queue from crashing the machine thread reschedule() rebuilds run_queue and re-enqueues the still-running program, so a program's slot could be task_done()'d twice -- once by done(), once by __next__ when it pops the now-finished stale entry. __next__'s call was unguarded and, once the counter went negative, raised "task_done() called too many times", killing the machine thread mid-night (seen on opd-40). The queue is only ever used as a thread-safe FIFO -- nothing join()s it -- so the task_done() bookkeeping served no purpose. Drop it entirely from both __next__ and done() and note why, and add a single-program regression that drove the counter negative under the old code. --- .../controllers/scheduler/sequential.py | 10 ++--- .../test_scheduler_no_duplicate.py | 38 +++++++++++++++++++ 2 files changed, 41 insertions(+), 7 deletions(-) diff --git a/src/chimera/controllers/scheduler/sequential.py b/src/chimera/controllers/scheduler/sequential.py index 31a0c7c1..798fb0af 100644 --- a/src/chimera/controllers/scheduler/sequential.py +++ b/src/chimera/controllers/scheduler/sequential.py @@ -16,6 +16,9 @@ def __init__(self): def reschedule(self, machine): self.machine = machine + # a plain thread-safe FIFO, rebuilt on every reschedule; we never + # join() it, so no task_done() accounting (which drifts across the + # rebuild and raised "task_done() called too many times") self.run_queue = Queue(-1) session = Session() @@ -51,7 +54,6 @@ def __next__(self): # be stale by the time they are popped: re-check the database current = session.get(Program, program.id) if current is None or current.finished: - self.run_queue.task_done() continue return program @@ -64,10 +66,4 @@ def done(self, task, error=None): else: task.finished = True - try: - self.run_queue.task_done() - except ValueError: - # the queue was rebuilt by a reschedule since this program was - # taken: its slot is gone, nothing left to account for - pass self.machine.wake_up() diff --git a/tests/chimera/controllers/test_scheduler_no_duplicate.py b/tests/chimera/controllers/test_scheduler_no_duplicate.py index e69d3d7c..a377caf2 100644 --- a/tests/chimera/controllers/test_scheduler_no_duplicate.py +++ b/tests/chimera/controllers/test_scheduler_no_duplicate.py @@ -291,3 +291,41 @@ def test_finished_program_is_not_rerun_after_reschedule(): session = Session() session.query(Program).delete() session.commit() + + +def test_reschedule_during_run_does_not_unbalance_the_queue(): + """Popping the stale re-enqueued program must not raise on the queue. + + reschedule() rebuilds run_queue and re-enqueues the still-running program; + done() and the stale-entry pop then both accounted for it with + task_done(), and the second call raised 'task_done() called too many + times', killing the machine thread mid-night. A single-program night is + the minimal case that drove the counter negative. + """ + from chimera.controllers.scheduler.model import Program, Session + from chimera.controllers.scheduler.sequential import SequentialScheduler + + session = Session() + session.query(Program).delete() + session.add(Program(name="only", priority=0, start_at=61244.0)) + session.commit() + + machine = type("M", (), {"wake_up": staticmethod(lambda: None)})() + scheduler = SequentialScheduler() + scheduler.reschedule(machine) + running = next(scheduler) + assert running.name == "only" + + # a re-plan mid-run re-enqueues the still-unfinished program + scheduler.reschedule(machine) + + worker_session = Session() + scheduler.done(worker_session.merge(running)) + worker_session.commit() + + # the stale entry is skipped and the night ends cleanly, no ValueError + assert next(scheduler) is None + + session = Session() + session.query(Program).delete() + session.commit() From ad7890e4541a666c232d904ea1144f39e84a191c Mon Sep 17 00:00:00 2001 From: William Schoenell Date: Tue, 28 Jul 2026 11:20:07 -0700 Subject: [PATCH 09/10] scheduler: a deleted program row must not kill the program thread MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Aborting a program whose row was deleted while it ran killed the scheduler-program thread — and with it the completion event the rest of the stack waits for: machine.py:293 log.debug(f"[aborted by user] {str(task)}") ObjectDeletedError: Instance '' has been deleted The abort itself had completed correctly; the casualty was the LOGGING. `session.commit()` expires every instance, so `str(task)` after it is not reading cached values - it reloads the row, and the row is routinely gone by then: robobs cleans the chimera queue on start AND on stop, and plan_robobs rebuilds it. Every operator lock during a program hit this on opd-40 2026-07-28 (the abort landed 391 s after the delete, so this is not a tight race). Three defences, in order of generality: `Program.__str__` falls back to the ORM identity instead of raising - a __str__ that can raise is a trap wherever it is used, and every scheduler log line formats a task; the program is described ONCE, up front, while its row is certainly there; and the terminal bookkeeping is wrapped so a failure to record the end of a program cannot cost the completion event or the state transition. The program is over by then - the machine must survive to pick up the next one. --- src/chimera/controllers/scheduler/machine.py | 124 +++++++++++---- src/chimera/controllers/scheduler/model.py | 16 +- .../test_scheduler_deleted_program.py | 150 ++++++++++++++++++ 3 files changed, 259 insertions(+), 31 deletions(-) create mode 100644 tests/chimera/controllers/test_scheduler_deleted_program.py diff --git a/src/chimera/controllers/scheduler/machine.py b/src/chimera/controllers/scheduler/machine.py index 81725fbb..26483090 100644 --- a/src/chimera/controllers/scheduler/machine.py +++ b/src/chimera/controllers/scheduler/machine.py @@ -1,5 +1,6 @@ import logging import threading +import time from chimera.controllers.scheduler.model import Program, Session from chimera.controllers.scheduler.states import State @@ -8,6 +9,9 @@ log = logging.getLogger(__name__) +#: seconds between heartbeat lines while a program runs +HEARTBEAT_SECONDS = 60.0 + class Machine(threading.Thread): __state = None @@ -24,6 +28,9 @@ def __init__(self, scheduler, executor, controller): self.current_program = None # thread running the current program; the IDLE branch waits on it self._worker = None + # heartbeat bookkeeping (monotonic seconds) + self._last_heartbeat = 0.0 + self._program_started_at = 0.0 self.daemon = False @@ -90,6 +97,7 @@ def run(self): # poll, don't sleep(): the worker signals IDLE just # before exiting, so the wakeup can be lost self._worker.join(1.0) + self._heartbeat() continue log.debug("[idle] looking for something to do...") @@ -102,6 +110,8 @@ def run(self): log.debug("[idle] program slew start %s", program.start_at) self.state(State.BUSY) self.current_program = program + self._program_started_at = time.monotonic() + self._last_heartbeat = self._program_started_at self._process(program) continue @@ -112,7 +122,11 @@ def run(self): elif self.state() == State.BUSY: log.debug("[busy] waiting tasks to finish..") - self.sleep() + # bounded, so the heartbeat below gets a turn; the Condition + # is notified on every state change, so a STOP still lands + # immediately + self.sleep(timeout=HEARTBEAT_SECONDS) + self._heartbeat() elif self.state() == State.STOP: log.debug("[stop] trying to stop current program") @@ -142,6 +156,37 @@ def run(self): self._join_worker("shutdown") log.debug("[shutdown] thread ending...") + def _heartbeat(self): + """Say what the machine is doing, at most once a minute. + + Both waits are silent: BUSY sleeps on the wake-up Condition while + the worker runs, and a worker counting down to its slew time (or + running a 10 min autofocus) prints nothing either. Twice that + silence was diagnosed as "the scheduler is wedged" from the log + alone. + """ + now = time.monotonic() + if now - self._last_heartbeat < HEARTBEAT_SECONDS: + return + self._last_heartbeat = now + + program = self.current_program + waiting_for = "" + start_at = getattr(program, "start_at", None) + if start_at: + waiting_for = f", waiting until MJD {start_at:.5f}" + try: + waiting_for += f" (now {self.controller.get_site().mjd():.5f})" + except Exception: + pass + log.info( + "[%s] running %s for %.0f s%s", + self.state(), + program, + now - self._program_started_at, + waiting_for, + ) + def _join_worker(self, reason): """Wait until the worker thread is actually dead, heartbeating: a camera abort legitimately takes minutes, and a silent wait reads as @@ -161,10 +206,10 @@ def _join_worker(self, reason): waited, ) - def sleep(self): + def sleep(self, timeout=None): self.__wake_up_call.acquire() log.debug("Sleeping") - self.__wake_up_call.wait() + self.__wake_up_call.wait(timeout) self.__wake_up_call.release() def wake_up(self): @@ -213,7 +258,14 @@ def process(): task = session.merge(program) - log.debug(f"[start] {str(task)}") + # Describe the program ONCE, while its row is certainly there: + # every commit below expires the instance, and the queue can be + # rewritten under a running program (robobs cleans it, a replan + # rebuilds it), so formatting it afterwards reloads a row that + # may be gone. + label = str(task) + + log.debug(f"[start] {label}") # the manager-injected site, not a private Site(): one clock # system-wide @@ -279,40 +331,52 @@ def process(): ) self.controller.program_begin(program.id) - try: - self.executor.execute(task) - log.debug(f"[finish] {str(task)}") - self.scheduler.done(task) - # commit the finished flag before releasing the machine: - # the idle picker re-checks it against the database - session.commit() + def finish(status, message=None, error=None, next_state=State.IDLE): + """Close the program out. Nothing here may kill this thread. + + The program is over by the time we get here, so a failure to + record it must not cost the completion event (robobs waits + for it to pick the next program) nor the state transition + (the machine would sit BUSY until chimera is restarted). + """ + try: + self.scheduler.done(task, error=error) + # commit the finished flag before releasing the machine: + # the idle picker re-checks it against the database + session.commit() + except Exception: + log.exception("[finish] could not record the end of %s", label) self._stop_tracking() - self.controller.program_complete(program.id, SchedulerStatus.OK) + try: + self.controller.program_complete(program.id, status, message) + except Exception: + log.exception("[finish] could not publish the end of %s", label) # conditional: a completion event must not re-arm a machine # that went STOPPING/SHUTDOWN while this program ran - self.transition(State.BUSY, State.IDLE) + self.transition(State.BUSY, next_state) + + try: + self.executor.execute(task) + log.debug(f"[finish] {label}") + finish(SchedulerStatus.OK) except ProgramExecutionException as e: - self.scheduler.done(task, error=e) - session.commit() - self._stop_tracking() - self.controller.program_complete( - program.id, SchedulerStatus.ERROR, str(e) - ) - self.transition(State.BUSY, State.IDLE) - log.debug(f"[error] {str(task)} ({str(e)})") + finish(SchedulerStatus.ERROR, str(e), error=e) + log.debug(f"[error] {label} ({str(e)})") except ProgramExecutionAborted as e: - self.scheduler.done(task, error=e) - session.commit() - self._stop_tracking() - self.controller.program_complete( - program.id, SchedulerStatus.ABORTED, "Aborted by user." - ) # normally the machine itself goes STOPPING -> OFF once this # worker dies; cover an out-of-band abort arriving while BUSY - self.transition(State.BUSY, State.OFF) - log.debug(f"[aborted by user] {str(task)}") + finish( + SchedulerStatus.ABORTED, + "Aborted by user.", + error=e, + next_state=State.OFF, + ) + log.debug(f"[aborted by user] {label}") - session.commit() + try: + session.commit() + except Exception: + log.exception("[finish] final commit failed for %s", label) t = threading.Thread(target=process, name="scheduler-program") t.daemon = False diff --git a/src/chimera/controllers/scheduler/model.py b/src/chimera/controllers/scheduler/model.py index 53814f9d..e81b9b95 100644 --- a/src/chimera/controllers/scheduler/model.py +++ b/src/chimera/controllers/scheduler/model.py @@ -11,6 +11,7 @@ PickleType, String, create_engine, + inspect, ) from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import backref, relation, sessionmaker @@ -71,7 +72,20 @@ class Program(Base): ) def __str__(self): - return f"#{self.id} {self.name} pi:{self.pi} #actions: {len(self.actions)}" + # Never depend on live database state: every session.commit() + # expires the instance, so this triggers a lazy reload - and if the + # queue was rewritten while the program ran (robobs cleans it on + # start AND on stop, plan_robobs rebuilds it), the row is gone and + # the reload raises ObjectDeletedError. From a log line inside an + # exception handler that killed the scheduler-program thread + # outright (opd-40 2026-07-28, on every operator lock during a + # program). A __str__ that can raise is a trap wherever it is used. + try: + return f"#{self.id} {self.name} pi:{self.pi} #actions: {len(self.actions)}" + except Exception: + identity = inspect(self).identity + program_id = identity[0] if identity else "?" + return f"#{program_id} " class Action(Base): diff --git a/tests/chimera/controllers/test_scheduler_deleted_program.py b/tests/chimera/controllers/test_scheduler_deleted_program.py new file mode 100644 index 00000000..3829ba3d --- /dev/null +++ b/tests/chimera/controllers/test_scheduler_deleted_program.py @@ -0,0 +1,150 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""A program whose row was deleted while it ran must not kill the machine. + +The queue is rewritten under running programs as a matter of routine - +robobs cleans it on start AND on stop, plan_robobs rebuilds it - so by the +time the abort lands, the row may be gone. The abort branch logged +`str(task)` after a commit had expired the instance, which reloads the row +and raises ObjectDeletedError from inside an exception handler: the +scheduler-program thread died mid-cleanup, taking the completion event +with it. Seen on opd-40 2026-07-28 on every operator lock during a +program, the abort itself having completed correctly 6 min earlier. +""" + +import threading + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from chimera.controllers.scheduler import machine as machine_module +from chimera.controllers.scheduler import model +from chimera.controllers.scheduler.machine import Machine +from chimera.controllers.scheduler.states import State +from chimera.controllers.scheduler.status import SchedulerStatus +from chimera.core.exceptions import ProgramExecutionAborted + +NOW_MJD = 60000.0 + + +def test_str_of_a_deleted_program_does_not_raise(tmp_path): + """Program.__str__ is used in every scheduler log line, including ones + inside exception handlers: it must never depend on live database + state.""" + engine = create_engine(f"sqlite:///{tmp_path / 'scheduler.db'}") + model.metadata.create_all(engine) + session_factory = sessionmaker(bind=engine) + + session = session_factory() + program = model.Program(name="WASP-145A", pi="", priority=1) + session.add(program) + session.commit() + + # somebody rebuilt the queue while this program was executing + other = session_factory() + other.query(model.Program).delete() + other.commit() + + session.commit() # expires every instance in the session + assert str(program).startswith("#") # must not raise ObjectDeletedError + + +class FakeSite: + def mjd(self): + return NOW_MJD + + +class Controller: + def __init__(self): + self.events = [] + self._config = {"stop_tracking_on_program_end": False, "telescope": None} + + def __getitem__(self, key): + return self._config[key] + + def get_site(self): + return FakeSite() + + def program_begin(self, program_id): + pass + + def program_complete(self, program_id, status, message=None): + self.events.append((status, message)) + + def state_changed(self, new, old): + pass + + +class DeletedRowProgram: + """A program whose row disappears mid-run: formatting it then reloads + the row, exactly as an expired SQLAlchemy instance does.""" + + id = 1 + start_at = 0.0 + valid_for = -1 + + def __init__(self): + self.deleted = False + + def __str__(self): + if self.deleted: + raise RuntimeError("Instance has been deleted") + return "#1 WASP-145A pi: #actions: 3" + + +class Scheduler: + def reschedule(self, machine): + pass + + def __next__(self): + return None + + def done(self, task, error=None): + # SequentialScheduler logs the task it is closing out + str(task) + + +class AbortingExecutor: + """The queue is rewritten, then the abort the operator asked for lands.""" + + def __init__(self, program): + self.program = program + + def __start__(self): + pass + + def stop(self): + pass + + def execute(self, program): + self.program.deleted = True + raise ProgramExecutionAborted() + + +@pytest.fixture(autouse=True) +def _no_db(monkeypatch): + class FakeSession: + def merge(self, obj): + return obj + + def commit(self): + pass + + monkeypatch.setattr(machine_module, "Session", lambda: FakeSession()) + + +def test_aborting_a_deleted_program_keeps_the_machine_alive(): + program = DeletedRowProgram() + controller = Controller() + machine = Machine(Scheduler(), AbortingExecutor(program), controller) + + machine.state(State.BUSY) + machine._process(program) + machine._worker.join(10) + + assert not machine._worker.is_alive() + assert controller.events == [(SchedulerStatus.ABORTED, "Aborted by user.")], ( + "the abort's completion event was lost with the thread" + ) + assert machine.state() == State.OFF + assert threading.active_count() >= 1 From 23cfa8a2797dbf539680f73d81f5c227f4a9909e Mon Sep 17 00:00:00 2001 From: William Schoenell Date: Tue, 28 Jul 2026 11:20:07 -0700 Subject: [PATCH 10/10] scheduler: heartbeat while a program runs, and keep file order for untimed queues Two smaller papercuts from the same nights: * the machine waits in silence while a program runs (BUSY sleeps on the wake-up Condition; a worker counting down to its slew time or running a 10 min autofocus prints nothing), and twice that silence was diagnosed as 'the scheduler is wedged' from the log alone. It now says what it is running, for how long, and what it is waiting for, once a minute. The BUSY wait is bounded to make room for it; the Condition is still notified on every state change, so a STOP lands as immediately as before. * an untimed hand-written queue ('chimera-sched --new', no start_at, one priority) ran in whatever order the database returned - in practice the reverse of the YAML. Insertion order is now the last tie-break, so a file runs in the order it was written. --- .../controllers/scheduler/sequential.py | 11 +++++-- .../test_scheduler_no_duplicate.py | 31 +++++++++++++++++++ 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/src/chimera/controllers/scheduler/sequential.py b/src/chimera/controllers/scheduler/sequential.py index 798fb0af..f0044f57 100644 --- a/src/chimera/controllers/scheduler/sequential.py +++ b/src/chimera/controllers/scheduler/sequential.py @@ -27,10 +27,17 @@ def reschedule(self, machine): # start_at orders execution, priority breaks ties: priority alone # let a future-timed program hold the machine with the whole night # queued behind it. Programs without start_at sort first and keep - # the old priority order among themselves. + # the old priority order among themselves; equal priorities then run + # in insertion order, which for a hand-written `chimera-sched --new` + # queue is the order the targets appear in the YAML (they came out + # reversed, which is not what anyone writing a file expects). programs = ( session.query(Program) - .order_by(Program.start_at.asc().nullsfirst(), desc(Program.priority)) + .order_by( + Program.start_at.asc().nullsfirst(), + desc(Program.priority), + Program.id.asc(), + ) .filter(Program.finished == False) # noqa .all() ) diff --git a/tests/chimera/controllers/test_scheduler_no_duplicate.py b/tests/chimera/controllers/test_scheduler_no_duplicate.py index a377caf2..7fe22d27 100644 --- a/tests/chimera/controllers/test_scheduler_no_duplicate.py +++ b/tests/chimera/controllers/test_scheduler_no_duplicate.py @@ -255,6 +255,37 @@ def test_sequential_runs_in_time_order(): session.commit() +def test_sequential_keeps_file_order_for_untimed_programs(): + """A hand-written queue must run in the order it was written. + + `chimera-sched --new` bakes no start_at and usually one priority, so + the ordering fell through to whatever the database returned - the + targets came out in the reverse of the YAML (2026-07-20). + """ + from chimera.controllers.scheduler.model import Program, Session + from chimera.controllers.scheduler.sequential import SequentialScheduler + + session = Session() + session.query(Program).delete() + for name in ("first", "second", "third"): + session.add(Program(name=name, priority=0)) + session.commit() + + scheduler = SequentialScheduler() + scheduler.reschedule(type("M", (), {"wake_up": staticmethod(lambda: None)})()) + + order = [] + while True: + program = next(scheduler) + if program is None: + break + order.append(program.name) + assert order == ["first", "second", "third"], order + + session.query(Program).delete() + session.commit() + + def test_finished_program_is_not_rerun_after_reschedule(): """A reschedule while a program runs re-enqueues that same program (its finished flag is only written at completion); the pop must skip the