Codex Release Bot
293 lines
13 KiB
Python
293 lines
13 KiB
Python
"""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, timedelta, timezone
|
|
|
|
from server.infrastructure.database.session import AsyncSessionLocal
|
|
from server.domain.models import PushSchedule
|
|
from server.utils.display_time import ensure_utc, operator_wall, to_beijing, to_naive_utc
|
|
|
|
logger = logging.getLogger("nexus.schedule_runner")
|
|
|
|
SCHEDULE_CHECK_INTERVAL = 60 # seconds
|
|
|
|
|
|
def _cron_match(cron_expr: str, at_utc: datetime) -> bool:
|
|
"""Match 5-field cron against operator (Beijing) wall clock at UTC instant *at_utc*."""
|
|
parts = cron_expr.strip().split()
|
|
if len(parts) != 5:
|
|
return False
|
|
|
|
minute, hour, day, month, dow = operator_wall(at_utc)
|
|
fields = [minute, hour, day, month, dow]
|
|
|
|
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 _compute_next_cron_run(cron_expr: str, after_utc: datetime) -> datetime | None:
|
|
"""Return next UTC naive instant strictly after *after_utc* matching *cron_expr* (Beijing wall)."""
|
|
after = ensure_utc(after_utc)
|
|
candidate_bj = to_beijing(after).replace(second=0, microsecond=0) + timedelta(minutes=1)
|
|
limit_bj = to_beijing(after) + timedelta(days=366)
|
|
while candidate_bj <= limit_bj:
|
|
at_utc = candidate_bj.astimezone(timezone.utc)
|
|
if _cron_match(cron_expr, at_utc):
|
|
return to_naive_utc(at_utc)
|
|
candidate_bj += timedelta(minutes=1)
|
|
return None
|
|
|
|
|
|
def compute_schedule_next_run(schedule: PushSchedule, now: datetime | None = None) -> datetime | None:
|
|
"""Compute next scheduled run time for API display."""
|
|
if not schedule.enabled:
|
|
return None
|
|
now = now or datetime.now(timezone.utc)
|
|
run_mode = getattr(schedule, "run_mode", None) or "cron"
|
|
|
|
if run_mode == "once":
|
|
fire_at = getattr(schedule, "fire_at", None)
|
|
if not fire_at or schedule.last_run_at:
|
|
return None
|
|
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 None
|
|
return fire_at
|
|
|
|
cron_expr = schedule.cron_expr
|
|
if not cron_expr:
|
|
return None
|
|
return _compute_next_cron_run(cron_expr.strip(), now)
|
|
|
|
|
|
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, ensure_utc(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 = to_naive_utc(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',
|
|
target_path=getattr(schedule, "target_path", None) or None,
|
|
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}")
|