diff --git a/OpenOrchestrator/database/db_util.py b/OpenOrchestrator/database/db_util.py index 5d5bd3c..1473877 100644 --- a/OpenOrchestrator/database/db_util.py +++ b/OpenOrchestrator/database/db_util.py @@ -24,25 +24,68 @@ def connect(conn_string: str) -> bool: - """Connects to the database using the given connection string. + """Connect to the database using the given connection string. + + The engine is created with ``pool_pre_ping=True`` and ``pool_recycle=1800`` + so pooled connections that go stale across server restarts or firewall + idle-disconnects are transparently replaced on the next checkout. + + For pyodbc/MSSQL connection strings a 5-second ODBC login timeout is + also set; the default is 15 s, which would freeze the scheduler's Tk + event loop for that long on each failed attempt during an outage. + + If the engine is created successfully but the probe fails because the + server is unreachable, the engine is still stored: ``pool_pre_ping`` + will retry on the next session checkout and recover when the network + comes back. Args: - conn_string: The connection string. + conn_string: The SQLAlchemy connection string. Returns: - bool: True if successful. + True if both the engine creation and the probe succeeded. False if + the connection string is malformed or the server is currently + unreachable. A False return when the server is unreachable still + leaves a usable engine in place for ``pool_pre_ping`` to recover. """ global _connection_engine # pylint: disable=global-statement try: - engine = create_engine(conn_string) - engine.connect() + engine = create_engine( + conn_string, + pool_pre_ping=True, + pool_recycle=1800, + connect_args=_driver_connect_args(conn_string), + ) + except alc_exc.ArgumentError: + _connection_engine = None + return False + + try: + engine.connect().close() _connection_engine = engine return True - except (alc_exc.InterfaceError, alc_exc.ArgumentError, alc_exc.OperationalError): - _connection_engine = None + except (alc_exc.InterfaceError, alc_exc.OperationalError): + _connection_engine = engine + return False + + +def _driver_connect_args(conn_string: str) -> dict: + """Return driver-specific ``connect_args`` for :func:`create_engine`. + + Only pyodbc/MSSQL gets a short login timeout. Other drivers (sqlite, + psycopg2, ...) use their own defaults to avoid passing unknown keyword + arguments through to ``DBAPI.connect``. - return False + Args: + conn_string: The SQLAlchemy connection string. + + Returns: + A keyword-argument mapping to pass as ``connect_args``. + """ + if conn_string.startswith("mssql+pyodbc"): + return {"timeout": 5} + return {} def disconnect() -> None: diff --git a/OpenOrchestrator/scheduler/inflight.py b/OpenOrchestrator/scheduler/inflight.py new file mode 100644 index 0000000..7253021 --- /dev/null +++ b/OpenOrchestrator/scheduler/inflight.py @@ -0,0 +1,132 @@ +"""Per-machine registry of in-flight trigger IDs. + +Used by :mod:`OpenOrchestrator.scheduler.runner` to remember which triggers +this scheduler instance has flipped to RUNNING in the database but has not +yet finished cleanly. The registry is persisted as a small JSON file so it +survives a scheduler crash or reboot: + +- ``runner.run_trigger`` calls :func:`add` after the DB row is set to RUNNING. +- ``runner.end_job`` / ``runner.fail_job`` / ``runner.kill_job`` call + :func:`remove` once the job has finished. +- ``runner.reconcile_orphans`` reads the registry on scheduler startup and + marks any still-RUNNING orphans as FAILED. + +Without this, a scheduler that crashes between ``begin_*_trigger`` and the +job finishing leaves the trigger in RUNNING state with no real process, +and the pending-trigger queries (which only look at IDLE rows) never pick +it up again. See issues #152 and #106. + +The file lives at ``%APPDATA%\\OpenOrchestrator\\inflight.json`` on Windows +and ``~/.OpenOrchestrator/OpenOrchestrator/inflight.json`` elsewhere. +""" + +from __future__ import annotations + +import json +import os +import threading +from pathlib import Path +from typing import Iterable +from uuid import UUID + +_LOCK = threading.Lock() + + +def _path() -> Path: + """Return the in-flight JSON file path, creating parent dirs as needed. + + Returns: + Absolute path to ``inflight.json`` in the per-user data directory. + """ + appdata = os.environ.get("APPDATA") + base = Path(appdata) if appdata else Path.home() / ".OpenOrchestrator" + folder = base / "OpenOrchestrator" + folder.mkdir(parents=True, exist_ok=True) + return folder / "inflight.json" + + +def _load() -> list[str]: + """Read the persisted in-flight list. + + Returns: + List of trigger IDs as strings, or ``[]`` if the file is missing, + unreadable, or malformed. + """ + p = _path() + if not p.exists(): + return [] + try: + with p.open("r", encoding="utf-8") as f: + data = json.load(f) + if isinstance(data, list): + return [str(x) for x in data] + return [] + except (json.JSONDecodeError, OSError): + return [] + + +def _save(items: Iterable[str]) -> None: + """Atomically replace the persisted in-flight list. + + Writes to a temporary file in the same directory and then ``os.replace``\\ s + it over the destination, which is atomic on both Windows and POSIX. + + Args: + items: Trigger IDs (as strings) to persist. + """ + p = _path() + tmp = p.with_suffix(".tmp") + with tmp.open("w", encoding="utf-8") as f: + json.dump(list(items), f) + os.replace(tmp, p) + + +def add(trigger_id: UUID | str) -> None: + """Record that this scheduler is now running the given trigger. + + No-op if the ID is already present. + + Args: + trigger_id: Identifier of the trigger that was just flipped to RUNNING. + """ + s = str(trigger_id) + with _LOCK: + items = _load() + if s not in items: + items.append(s) + _save(items) + + +def remove(trigger_id: UUID | str) -> None: + """Record that this scheduler has finished with the given trigger. + + No-op if the ID is not present. + + Args: + trigger_id: Identifier of the trigger whose job just ended. + """ + s = str(trigger_id) + with _LOCK: + items = _load() + if s in items: + items.remove(s) + _save(items) + + +def get_all() -> list[str]: + """Return all trigger IDs currently marked as in-flight on this machine. + + Returns: + Snapshot of the trigger ID list at the time of the call. + """ + with _LOCK: + return _load() + + +def clear() -> None: + """Forget every in-flight trigger. + + Provided for tests and manual recovery; not used by the scheduler itself. + """ + with _LOCK: + _save([]) diff --git a/OpenOrchestrator/scheduler/run_tab.py b/OpenOrchestrator/scheduler/run_tab.py index 9d6d7e0..b3eb64f 100644 --- a/OpenOrchestrator/scheduler/run_tab.py +++ b/OpenOrchestrator/scheduler/run_tab.py @@ -7,6 +7,9 @@ import tkinter from tkinter import ttk import sys +import traceback +from datetime import datetime +from pathlib import Path from sqlalchemy import exc as alc_exc @@ -19,6 +22,53 @@ from OpenOrchestrator.scheduler.application import Application +# Consecutive DB-error ticks since the last successful one. Tk's after-loop +# is single-threaded so a module-level int is safe. +_consecutive_db_failures = 0 + +_BASE_TICK_MS = 6_000 +_MAX_BACKOFF_MS = 600_000 # 10 minutes +_DESKTOP_LOG_NAME = "OpenOrchestrator_scheduler_errors.log" + + +def _backoff_delay_ms() -> int: + """Compute the next-tick delay using exponential backoff. + + Returns: + Delay in milliseconds: ``_BASE_TICK_MS * 2**_consecutive_db_failures`` + clamped to ``_MAX_BACKOFF_MS``. + """ + return min(_BASE_TICK_MS * (2 ** _consecutive_db_failures), _MAX_BACKOFF_MS) + + +def _log_to_desktop(exc: BaseException, context: str = "") -> None: + """Append a timestamped traceback to a desktop log file. + + Used to surface scheduler-loop crashes outside the Tk text widget so they + survive across restarts. Never raises - if the log can't be written the + failure is swallowed so the scheduler loop itself can continue. + + Args: + exc: The exception to record. + context: Optional short string identifying the code path that raised. + """ + try: + desktop = Path.home() / "Desktop" + desktop.mkdir(parents=True, exist_ok=True) + log_path = desktop / _DESKTOP_LOG_NAME + ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + with log_path.open("a", encoding="utf-8") as f: + f.write("\n" + ("=" * 80) + "\n") + f.write(f"[{ts}] {exc.__class__.__name__}: {exc}\n") + if context: + f.write(f"Context: {context}\n") + f.write(("-" * 80) + "\n") + f.write(traceback.format_exc()) + f.write(("=" * 80) + "\n") + except Exception: # pylint: disable=broad-except + pass + + # pylint: disable-next=too-many-ancestors class RunTab(ttk.Frame): """A ttk.frame object containing the functionality of the run tab in Scheduler.""" @@ -77,6 +127,15 @@ def run(self): self.button.configure(text="Pause") print('Running...\n') + + try: + n = runner.reconcile_orphans() + if n > 0: + print(f"*** Failed {n} orphan trigger(s) from previous run ***\n") + except Exception as e: # pylint: disable=broad-except + print(f"Orphan reconciliation failed: {e.__class__.__name__}: {e}") + _log_to_desktop(e, context="reconcile_orphans during run()") + self.app.running = True # Only start a new loop if it's not already running @@ -105,60 +164,110 @@ def print_text(self, text: str) -> None: def loop(app: Application) -> None: - """The main loop function of the Scheduler. - Checks heartbeats, check triggers, and schedules the next loop. + """Run one Scheduler tick: pings, heartbeats, triggers, and reschedule. + + The body is fully guarded: any uncaught exception in a tick would + otherwise break Tk's ``after``-chain and leave the scheduler frozen + (see issues #152, #106). The reschedule lives in ``finally`` so the next + tick is always queued. + + DB-class errors (``SQLAlchemyError`` and the ``RuntimeError`` raised by + ``_get_session`` when there is no engine) route through an exponential + backoff capped at ``_MAX_BACKOFF_MS``. ``pool_pre_ping`` on the engine + is responsible for actually re-establishing the connection on subsequent + ticks; this function only needs to keep ticking. Args: app: The Scheduler Application object. """ + global _consecutive_db_failures # pylint: disable=global-statement + + delay_ms = _BASE_TICK_MS + try: - send_ping_to_orchestrator() + try: + send_ping_to_orchestrator() - check_heartbeats(app) + check_heartbeats(app) - if app.running: - check_triggers(app) + if app.running: + check_triggers(app) - except (alc_exc.OperationalError, alc_exc.ProgrammingError) as e: - print(f"Couldn't connect to database. {e}") + except (alc_exc.SQLAlchemyError, RuntimeError) as e: + _consecutive_db_failures += 1 + delay_ms = _backoff_delay_ms() + print("\n!!! LOST DATABASE CONNECTION " + f"(consecutive failures: {_consecutive_db_failures}) !!!") + print(f"Error: {e.__class__.__name__}: {e}") + print(f"Retrying in {delay_ms // 1000} seconds...\n") + _log_to_desktop(e, context="scheduler loop - DB error") - if len(app.running_jobs) == 0: - print("Doing cleanup...") - runner.clear_repo_folder() + except Exception as e: # pylint: disable=broad-except + print("\n!!! UNEXPECTED SCHEDULER ERROR !!!") + print(f"{e.__class__.__name__}: {e}") + print("Will continue on next tick.\n") + _log_to_desktop(e, context="scheduler loop - unexpected error") - # Schedule next loop - if app.running or len(app.running_jobs) > 0: - print('Waiting 6 seconds...\n') - app.after(6_000, loop, app) - else: - print("Scheduler is paused and no more processes are running.") + else: + if _consecutive_db_failures > 0: + print(f"\n*** Database connection restored after " + f"{_consecutive_db_failures} failed attempt(s) ***\n") + _consecutive_db_failures = 0 + + if len(app.running_jobs) == 0: + try: + print("Doing cleanup...") + runner.clear_repo_folder() + except Exception as e: # pylint: disable=broad-except + print(f"Cleanup failed: {e.__class__.__name__}: {e}") + _log_to_desktop(e, context="clear_repo_folder") + + finally: + if app.running or len(app.running_jobs) > 0: + print(f'Waiting {delay_ms // 1000} seconds...\n') + app.after(delay_ms, loop, app) + else: + print("Scheduler is paused and no more processes are running.") def check_heartbeats(app: Application) -> None: - """Check if any running jobs are still running, failed or done. + """Reconcile each running job's state against its subprocess and the DB. + + Each per-job check is wrapped so a DB failure on one job does not abort + the whole sweep; if any DB error occurred, the first one is re-raised + after the loop so :func:`loop` enters the backoff path on this tick. Args: app: The Scheduler Application object. """ print('Checking heartbeats...') - for job in app.running_jobs: - if job.process.poll() is not None: - if job.process.returncode == 0: - print(f"Process '{job.trigger.process_name}' is done") - runner.end_job(job) - else: - print(f"Process '{job.trigger.process_name}' failed. Check process log for more info.") - runner.fail_job(job) - - app.running_jobs.remove(job) + db_errors: list[BaseException] = [] + for job in list(app.running_jobs): + try: + if job.process.poll() is not None: + if job.process.returncode == 0: + print(f"Process '{job.trigger.process_name}' is done") + runner.end_job(job) + else: + print(f"Process '{job.trigger.process_name}' failed. Check process log for more info.") + runner.fail_job(job) + + app.running_jobs.remove(job) + + elif db_util.get_trigger(job.trigger.id).process_status == TriggerStatus.KILLING: + runner.kill_job(job) + print(f"Process '{job.trigger.process_name}' has been killed.") + app.running_jobs.remove(job) - elif db_util.get_trigger(job.trigger.id).process_status == TriggerStatus.KILLING: - runner.kill_job(job) - print(f"Process '{job.trigger.process_name}' has been killed.") - app.running_jobs.remove(job) - - else: - print(f"Process '{job.trigger.process_name}' is still running") + else: + print(f"Process '{job.trigger.process_name}' is still running") + except (alc_exc.SQLAlchemyError, RuntimeError) as e: + print(f"DB unavailable while checking '{job.trigger.process_name}': " + f"{e.__class__.__name__}. Will retry next tick.") + db_errors.append(e) + + if db_errors: + raise db_errors[0] def check_triggers(app: Application) -> None: diff --git a/OpenOrchestrator/scheduler/runner.py b/OpenOrchestrator/scheduler/runner.py index cbe3504..cb613f4 100644 --- a/OpenOrchestrator/scheduler/runner.py +++ b/OpenOrchestrator/scheduler/runner.py @@ -8,11 +8,13 @@ from dataclasses import dataclass import uuid +from sqlalchemy import exc as alc_exc + from OpenOrchestrator.common import crypto_util from OpenOrchestrator.database import db_util from OpenOrchestrator.database.triggers import Trigger, SingleTrigger, ScheduledTrigger, QueueTrigger, TriggerStatus from OpenOrchestrator.database.logs import LogLevel -from OpenOrchestrator.scheduler import util +from OpenOrchestrator.scheduler import util, inflight from OpenOrchestrator.database.jobs import Job, JobStatus if TYPE_CHECKING: @@ -67,27 +69,38 @@ def poll_triggers(app: Application) -> Trigger | None: def run_trigger(trigger: Trigger) -> SchedulerJob | None: - """Mark a trigger as running in the database - and start the process. + """Mark a trigger as running in the database and start its process. + + The trigger ID is added to the in-flight registry only after the DB row + has been flipped to RUNNING, so a crash inside ``begin_*_trigger`` leaves + nothing to reconcile (the DB is still IDLE in that case). Args: trigger: The trigger to run. Returns: - A Job object describing the process if successful. + A ``SchedulerJob`` describing the running process, or ``None`` if the + trigger could not be marked as running or the process failed to launch. """ print('Running trigger: ', trigger.trigger_name) - if isinstance(trigger, SingleTrigger) and db_util.begin_single_trigger(trigger.id): - return run_process(trigger) + began = False + if isinstance(trigger, SingleTrigger): + began = db_util.begin_single_trigger(trigger.id) + elif isinstance(trigger, ScheduledTrigger): + began = db_util.begin_scheduled_trigger(trigger.id) + elif isinstance(trigger, QueueTrigger): + began = db_util.begin_queue_trigger(trigger.id) - if isinstance(trigger, ScheduledTrigger) and db_util.begin_scheduled_trigger(trigger.id): - return run_process(trigger) + if not began: + return None - if isinstance(trigger, QueueTrigger) and db_util.begin_queue_trigger(trigger.id): - return run_process(trigger) + inflight.add(trigger.id) - return None + job = run_process(trigger) + if job is None: + inflight.remove(trigger.id) + return job def clone_git_repo(repo_url: str, branch: str | None) -> str: @@ -190,6 +203,8 @@ def end_job(job: SchedulerJob) -> None: if job.process_folder: clear_folder(job.process_folder) + inflight.remove(job.trigger.id) + def fail_job(job: SchedulerJob) -> None: """Mark a job as failed in the triggers table in the database. @@ -206,6 +221,8 @@ def fail_job(job: SchedulerJob) -> None: if job.process_folder: clear_folder(job.process_folder) + inflight.remove(job.trigger.id) + def kill_job(job: SchedulerJob) -> None: """Kill the job's process and mark is as killed in the database. @@ -225,6 +242,55 @@ def kill_job(job: SchedulerJob) -> None: if job.process_folder: clear_folder(job.process_folder) + inflight.remove(job.trigger.id) + + +def reconcile_orphans() -> int: + """Fail orphan triggers left behind by a previous crashed scheduler. + + Called once at scheduler startup. For every trigger ID still listed in + the local in-flight registry, fetch the row: if the DB still shows it + as RUNNING the job was never finished cleanly (the scheduler crashed, + the machine rebooted, or the DB was unreachable when the job ended), + so the trigger is set to FAILED and a log entry is written. Triggers + whose DB status is no longer RUNNING are silently dropped from the + local registry. + + A passed-in ``job_id`` of ``None`` is used in ``create_log`` because the + original :class:`Job` object died with the scheduler. + + Returns: + Number of orphan triggers that were transitioned to FAILED. + """ + failed = 0 + for tid in inflight.get_all(): + try: + trigger = db_util.get_trigger(tid) + except (alc_exc.SQLAlchemyError, ValueError, RuntimeError) as e: + print(f"Could not check orphan {tid}: {e.__class__.__name__}: {e}") + continue + + if trigger.process_status == TriggerStatus.RUNNING: + try: + db_util.set_trigger_status(tid, TriggerStatus.FAILED) + db_util.create_log( + trigger.process_name, + LogLevel.ERROR, + None, + "Trigger failed by scheduler startup orphan-reconciliation: " + "process state was lost (scheduler crashed, machine " + "rebooted, or DB was unreachable when the job ended)." + ) + failed += 1 + print(f"Failed orphan trigger '{trigger.trigger_name}' ({tid})") + except alc_exc.SQLAlchemyError as e: + print(f"Could not fail orphan {tid}: {e.__class__.__name__}: {e}") + continue + + inflight.remove(tid) + + return failed + def run_process(trigger: Trigger) -> SchedulerJob | None: """Runs the process of the given trigger with the necessary inputs: diff --git a/changelog.md b/changelog.md index 9cc71a4..2f180a5 100644 --- a/changelog.md +++ b/changelog.md @@ -10,6 +10,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Chips on triggers now update on "blur" events, when the user clicks away from the input. +- Orphan-trigger reconciliation on Scheduler startup. Trigger IDs the Scheduler has flipped to RUNNING are persisted to a small JSON file in the per-user data directory; on the next start any of those still showing RUNNING in the database (because the previous Scheduler crashed mid-job) are marked FAILED with a matching log entry. + +### Fixed + +- Scheduler no longer freezes when the SQL Server is restarted or briefly unreachable (closes #152 and #106). The main loop is wrapped so the Tk after-chain cannot die, and `pool_pre_ping` plus `pool_recycle` on the SQLAlchemy engine handle reconnection transparently. Failed ticks back off exponentially up to a 10-minute cap; the counter resets on the next successful tick. Tracebacks are appended to a desktop crash log so failures that happen overnight are visible afterwards. +- ODBC login timeout for MSSQL connections is now 5 s instead of the 15 s pyodbc default, so a failed connection attempt no longer freezes the Scheduler's Tk loop for 15 s at a time during an outage. ## [3.0.0] - 2026-02-23