"""Nexus — Schedule Runner Background Task Every 60 seconds: check PushSchedule cron expressions, trigger due schedules. """ import asyncio import json import logging from datetime import datetime, timezone from server.infrastructure.database.session import AsyncSessionLocal from server.domain.models import PushSchedule logger = logging.getLogger("nexus.schedule_runner") SCHEDULE_CHECK_INTERVAL = 60 # seconds def _cron_match(cron_expr: str, now: datetime) -> bool: """Check if current time matches a 5-field cron expression (min hour dom month dow). Supports: * (any), */N (step), specific values, comma-separated lists. """ parts = cron_expr.strip().split() if len(parts) != 5: return False fields = [ (now.minute, range(0, 60)), (now.hour, range(0, 24)), (now.day, range(1, 32)), (now.month, range(1, 13)), ((now.weekday() + 1) % 7, range(0, 7)), # cron: 0=Sunday, Python weekday(): 0=Monday ] for (current_val, _), field_expr in zip(fields, parts): if not _field_match(field_expr, current_val): return False return True def rollback_schedule_after_failed_run(schedule, prev_last_run_at, run_mode: str) -> None: """Restore schedule so cron/once can fire again after execution failure.""" schedule.last_run_at = prev_last_run_at if run_mode == "once": schedule.enabled = True def _field_match(expr: str, value: int) -> bool: """Match a single cron field expression against a value. Supports: * (any), */N (step), specific values, comma-separated lists, and range (N-M). Each comma-separated part is tried independently; the function returns True as soon as any part matches. """ for part in expr.split(","): if part == "*": return True if part.startswith("*/"): step = int(part[2:]) if step == 0: continue # */0 is invalid; skip rather than ZeroDivisionError if value % step == 0: return True continue if "-" in part: low, high = part.split("-", 1) if int(low) <= value <= int(high): return True else: # cron: 7 also means Sunday (=0) if value == int(part) or (int(part) == 7 and value == 0): return True return False def _schedule_is_due(schedule: PushSchedule, now: datetime) -> bool: """Return True if *schedule* should fire at *now* (cron or once).""" run_mode = getattr(schedule, "run_mode", None) or "cron" if run_mode == "once": fire_at = getattr(schedule, "fire_at", None) if not fire_at: return False now_naive = now.replace(tzinfo=None) fire_naive = fire_at.replace(tzinfo=None) if fire_at.tzinfo else fire_at if fire_naive > now_naive: return False if schedule.last_run_at: return False return True cron_expr = getattr(schedule, "cron_expr", None) if not cron_expr or not _cron_match(cron_expr, now): return False if schedule.last_run_at: last_run = schedule.last_run_at now_naive = now.replace(tzinfo=None) if last_run.tzinfo is not None: last_run = last_run.replace(tzinfo=None) seconds_since = (now_naive - last_run).total_seconds() if seconds_since < SCHEDULE_CHECK_INTERVAL: return False return True async def schedule_runner_loop(): """Background task: check schedules every 60s, trigger due cron jobs.""" logger.info("Schedule runner loop started (interval: 60s)") while True: await asyncio.sleep(SCHEDULE_CHECK_INTERVAL) try: async with AsyncSessionLocal() as session: from sqlalchemy import select id_result = await session.execute( select(PushSchedule.id).where(PushSchedule.enabled == True) ) schedule_ids = list(id_result.scalars().all()) now = datetime.now(timezone.utc) triggered = 0 for sched_id in schedule_ids: try: lock_result = await session.execute( select(PushSchedule) .where( PushSchedule.id == sched_id, PushSchedule.enabled == True, ) .with_for_update(skip_locked=True) ) schedule = lock_result.scalar_one_or_none() if not schedule: continue if not _schedule_is_due(schedule, now): await session.rollback() continue run_mode = getattr(schedule, "run_mode", None) or "cron" # Parse server_ids from JSON try: server_ids = json.loads(schedule.server_ids) except (json.JSONDecodeError, TypeError): logger.error(f"Schedule {schedule.id}: invalid server_ids JSON") await session.rollback() continue if not server_ids: await session.rollback() continue # Mark as running BEFORE execution to prevent double-trigger prev_last_run_at = schedule.last_run_at schedule.last_run_at = now if run_mode == "once": schedule.enabled = False await session.commit() schedule_type = getattr(schedule, "schedule_type", "push") or "push" execution_ok = True try: if schedule_type == "script": # ── Script execution schedule ── from server.application.services.script_service import ScriptService from server.infrastructure.database.script_repo import ( ScriptRepositoryImpl, ScriptExecutionRepositoryImpl, ) from server.infrastructure.database.db_credential_repo import DbCredentialRepositoryImpl from server.infrastructure.database.server_repo import ServerRepositoryImpl from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl cmd = getattr(schedule, 'script_content', None) or '' script_id = getattr(schedule, 'script_id', None) # If no inline command, load from script library if not cmd and script_id: script = await ScriptRepositoryImpl(session).get_by_id(script_id) if script: cmd = script.content or '' if not cmd: logger.error(f"Schedule '{schedule.name}': no command or script content, skipping") execution_ok = False else: svc = ScriptService( script_repo=ScriptRepositoryImpl(session), execution_repo=ScriptExecutionRepositoryImpl(session), credential_repo=DbCredentialRepositoryImpl(session), server_repo=ServerRepositoryImpl(session), audit_repo=AuditLogRepositoryImpl(session), ) timeout = getattr(schedule, 'exec_timeout', None) or 60 long_task = bool(getattr(schedule, 'long_task', False)) execution = await svc.execute_command( command=cmd, server_ids=server_ids, script_id=script_id, timeout=timeout, operator=f"schedule:{schedule.name}", long_task=long_task, ) logger.info( f"Schedule '{schedule.name}' (script) triggered: " f"execution_id={execution.id} on {len(server_ids)} servers" ) else: # ── File push schedule (original behaviour) ── from server.application.services.sync_engine_v2 import SyncEngineV2 from server.infrastructure.database.server_repo import ServerRepositoryImpl from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl from server.infrastructure.database.push_schedule_repo import PushRetryJobRepositoryImpl engine = SyncEngineV2( server_repo=ServerRepositoryImpl(session), sync_log_repo=SyncLogRepositoryImpl(session), audit_repo=AuditLogRepositoryImpl(session), retry_repo=PushRetryJobRepositoryImpl(session), ) result = await engine.sync_files( server_ids=server_ids, source_path=schedule.source_path or '/tmp', sync_mode=getattr(schedule, 'sync_mode', None) or "incremental", trigger_type="schedule", operator=f"schedule:{schedule.name}", ) logger.info( f"Schedule '{schedule.name}' (push) triggered: " f"{result['completed']}/{result['total']} succeeded" ) if result.get("error"): execution_ok = False elif result.get("total", 0) > 0 and result.get("completed", 0) == 0: execution_ok = False except Exception as exec_err: logger.error(f"Schedule {schedule.id} execution failed: {exec_err}") execution_ok = False if not execution_ok: fresh = await session.get(PushSchedule, sched_id) if fresh: rollback_schedule_after_failed_run( fresh, prev_last_run_at, run_mode ) await session.commit() triggered += 1 except Exception as e: logger.error(f"Schedule {sched_id} setup/parse failed: {e}") await session.rollback() if triggered: logger.info(f"Schedule runner: {triggered} schedules triggered") except Exception as e: logger.error(f"Schedule runner check failed: {e}")