mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
fix(cron): harden execution attempt ledger
This commit is contained in:
teknium1
Teknium
parent
d9dd05b69d
commit
abc22cdf1a
12 changed files with 387 additions and 203 deletions
|
|
@ -1,222 +1,228 @@
|
|||
"""Durable audit ledger for cron execution attempts.
|
||||
"""Profile-local durable audit ledger for cron execution attempts.
|
||||
|
||||
The ledger records what is known about each attempt; it is not a retry queue.
|
||||
An attempt left claimed or running when a scheduler process restarts is marked
|
||||
``unknown`` because the new process cannot prove whether its side effects ran.
|
||||
Interrupted attempts become ``unknown`` only after their exact owner process is
|
||||
proved gone. Terminal states are immutable.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import sqlite3
|
||||
import threading
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
try:
|
||||
import fcntl
|
||||
except ImportError: # pragma: no cover - non-Unix
|
||||
fcntl = None
|
||||
try:
|
||||
import msvcrt
|
||||
except ImportError: # pragma: no cover - non-Windows
|
||||
msvcrt = None
|
||||
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_time import now as _hermes_now
|
||||
from utils import atomic_replace
|
||||
|
||||
EXECUTIONS_FILE = get_hermes_home().resolve() / "cron" / "executions.json"
|
||||
EXECUTIONS_LOCK_FILE = get_hermes_home().resolve() / "cron" / ".executions.lock"
|
||||
EXECUTIONS_FILE = get_hermes_home().resolve() / "cron" / "executions.db"
|
||||
MAX_TERMINAL_EXECUTIONS = 1000
|
||||
_TERMINAL_STATES = ("completed", "failed", "unknown")
|
||||
_lock = threading.RLock()
|
||||
_lock_state = threading.local()
|
||||
_PROCESS_ID = uuid.uuid4().hex
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _ledger_lock():
|
||||
depth = getattr(_lock_state, "depth", 0)
|
||||
if depth:
|
||||
_lock_state.depth = depth + 1
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_lock_state.depth -= 1
|
||||
return
|
||||
|
||||
with _lock:
|
||||
EXECUTIONS_LOCK_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
lock_file = open(EXECUTIONS_LOCK_FILE, "a+", encoding="utf-8")
|
||||
try:
|
||||
if fcntl:
|
||||
fcntl.flock(lock_file, fcntl.LOCK_EX)
|
||||
elif msvcrt: # pragma: no cover - Windows
|
||||
lock_file.seek(0)
|
||||
if not lock_file.read(1):
|
||||
lock_file.write("0")
|
||||
lock_file.flush()
|
||||
lock_file.seek(0)
|
||||
msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1)
|
||||
_lock_state.depth = 1
|
||||
yield
|
||||
finally:
|
||||
_lock_state.depth = 0
|
||||
if fcntl:
|
||||
fcntl.flock(lock_file, fcntl.LOCK_UN)
|
||||
elif msvcrt: # pragma: no cover - Windows
|
||||
lock_file.seek(0)
|
||||
msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
|
||||
lock_file.close()
|
||||
|
||||
|
||||
def _load_unlocked() -> List[Dict[str, Any]]:
|
||||
if not EXECUTIONS_FILE.exists():
|
||||
return []
|
||||
try:
|
||||
data = json.loads(EXECUTIONS_FILE.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return []
|
||||
records = data.get("executions", []) if isinstance(data, dict) else []
|
||||
return records if isinstance(records, list) else []
|
||||
|
||||
|
||||
def _save_unlocked(records: List[Dict[str, Any]]) -> None:
|
||||
def _connect() -> sqlite3.Connection:
|
||||
EXECUTIONS_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
fd, tmp_name = tempfile.mkstemp(
|
||||
dir=str(EXECUTIONS_FILE.parent), prefix=".executions_", suffix=".tmp"
|
||||
conn = sqlite3.connect(EXECUTIONS_FILE, timeout=5)
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute("PRAGMA busy_timeout=5000")
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA synchronous=FULL")
|
||||
conn.execute(
|
||||
"""CREATE TABLE IF NOT EXISTS executions (
|
||||
id TEXT PRIMARY KEY,
|
||||
job_id TEXT NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
process_id TEXT NOT NULL,
|
||||
pid INTEGER NOT NULL,
|
||||
process_started_at INTEGER,
|
||||
status TEXT NOT NULL CHECK(status IN
|
||||
('claimed','running','completed','failed','unknown')),
|
||||
claimed_at TEXT NOT NULL,
|
||||
started_at TEXT,
|
||||
finished_at TEXT,
|
||||
error TEXT
|
||||
)"""
|
||||
)
|
||||
conn.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_executions_job_claimed "
|
||||
"ON executions(job_id, claimed_at DESC, id DESC)"
|
||||
)
|
||||
conn.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_executions_status_claimed "
|
||||
"ON executions(status, claimed_at DESC, id DESC)"
|
||||
)
|
||||
return conn
|
||||
|
||||
|
||||
def _record(row: Optional[sqlite3.Row]) -> Optional[Dict[str, Any]]:
|
||||
return dict(row) if row is not None else None
|
||||
|
||||
|
||||
def _process_start_time(pid: int) -> Optional[int]:
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
json.dump({"version": 1, "executions": records}, handle, indent=2)
|
||||
handle.write("\n")
|
||||
handle.flush()
|
||||
os.fsync(handle.fileno())
|
||||
atomic_replace(tmp_name, EXECUTIONS_FILE)
|
||||
try:
|
||||
os.chmod(EXECUTIONS_FILE, 0o600)
|
||||
except OSError:
|
||||
pass
|
||||
from gateway.status import get_process_start_time
|
||||
return get_process_start_time(pid)
|
||||
except Exception:
|
||||
try:
|
||||
os.unlink(tmp_name)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
return None
|
||||
|
||||
|
||||
def _owner_is_live(pid: int, started_at: Optional[int]) -> bool:
|
||||
try:
|
||||
from gateway.status import _pid_exists
|
||||
if not _pid_exists(pid):
|
||||
return False
|
||||
except Exception:
|
||||
return True # fail safe: inability to prove death must not rewrite state
|
||||
if started_at is None:
|
||||
return pid == os.getpid()
|
||||
current = _process_start_time(pid)
|
||||
return current is not None and current == started_at
|
||||
|
||||
|
||||
def _prune_unlocked(conn: sqlite3.Connection) -> None:
|
||||
limit = max(0, int(MAX_TERMINAL_EXECUTIONS))
|
||||
conn.execute(
|
||||
"""DELETE FROM executions WHERE id IN (
|
||||
SELECT id FROM executions
|
||||
WHERE status IN ('completed','failed','unknown')
|
||||
ORDER BY claimed_at DESC, id DESC LIMIT -1 OFFSET ?
|
||||
)""",
|
||||
(limit,),
|
||||
)
|
||||
|
||||
|
||||
def create_execution(job_id: str, *, source: str) -> Dict[str, Any]:
|
||||
"""Persist a claimed attempt before it is submitted for execution."""
|
||||
"""Persist a claimed attempt before executor/provider dispatch."""
|
||||
now = _hermes_now().isoformat()
|
||||
record: Dict[str, Any] = {
|
||||
"id": uuid.uuid4().hex,
|
||||
"job_id": str(job_id),
|
||||
"source": str(source),
|
||||
"process_id": _PROCESS_ID,
|
||||
"pid": os.getpid(),
|
||||
"status": "claimed",
|
||||
"claimed_at": now,
|
||||
"started_at": None,
|
||||
"finished_at": None,
|
||||
"error": None,
|
||||
}
|
||||
with _ledger_lock():
|
||||
records = _load_unlocked()
|
||||
records.append(record)
|
||||
_save_unlocked(records)
|
||||
return copy.deepcopy(record)
|
||||
|
||||
|
||||
def _transition(execution_id: str, status: str, **updates: Any) -> Optional[Dict[str, Any]]:
|
||||
with _ledger_lock():
|
||||
records = _load_unlocked()
|
||||
for record in records:
|
||||
if record.get("id") != execution_id:
|
||||
continue
|
||||
record["status"] = status
|
||||
record.update(updates)
|
||||
_save_unlocked(records)
|
||||
return copy.deepcopy(record)
|
||||
return None
|
||||
execution_id = uuid.uuid4().hex
|
||||
pid = os.getpid()
|
||||
with _lock, _connect() as conn:
|
||||
conn.execute(
|
||||
"""INSERT INTO executions
|
||||
(id, job_id, source, process_id, pid, process_started_at,
|
||||
status, claimed_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, 'claimed', ?)""",
|
||||
(execution_id, str(job_id), str(source), _PROCESS_ID, pid,
|
||||
_process_start_time(pid), now),
|
||||
)
|
||||
row = conn.execute(
|
||||
"SELECT * FROM executions WHERE id=?", (execution_id,)
|
||||
).fetchone()
|
||||
return _record(row) # type: ignore[return-value]
|
||||
|
||||
|
||||
def mark_execution_running(execution_id: str) -> Optional[Dict[str, Any]]:
|
||||
return _transition(
|
||||
execution_id,
|
||||
"running",
|
||||
started_at=_hermes_now().isoformat(),
|
||||
)
|
||||
"""Transition one claimed attempt to running exactly once."""
|
||||
now = _hermes_now().isoformat()
|
||||
with _lock, _connect() as conn:
|
||||
cur = conn.execute(
|
||||
"""UPDATE executions SET status='running', started_at=?
|
||||
WHERE id=? AND status='claimed'""",
|
||||
(now, execution_id),
|
||||
)
|
||||
if cur.rowcount != 1:
|
||||
return None
|
||||
return _record(conn.execute(
|
||||
"SELECT * FROM executions WHERE id=?", (execution_id,)
|
||||
).fetchone())
|
||||
|
||||
|
||||
def finish_execution(
|
||||
execution_id: str,
|
||||
*,
|
||||
success: bool,
|
||||
error: Optional[str] = None,
|
||||
execution_id: str, *, success: bool, error: Optional[str] = None,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
return _transition(
|
||||
execution_id,
|
||||
"completed" if success else "failed",
|
||||
finished_at=_hermes_now().isoformat(),
|
||||
error=None if success else (str(error) if error else "unknown failure"),
|
||||
)
|
||||
"""Write a terminal result once; terminal attempts cannot be rewritten."""
|
||||
now = _hermes_now().isoformat()
|
||||
status = "completed" if success else "failed"
|
||||
detail = None if success else (str(error) if error else "unknown failure")
|
||||
with _lock, _connect() as conn:
|
||||
cur = conn.execute(
|
||||
"""UPDATE executions SET status=?, finished_at=?, error=?
|
||||
WHERE id=? AND status IN ('claimed','running')""",
|
||||
(status, now, detail, execution_id),
|
||||
)
|
||||
if cur.rowcount != 1:
|
||||
return None
|
||||
_prune_unlocked(conn)
|
||||
return _record(conn.execute(
|
||||
"SELECT * FROM executions WHERE id=?", (execution_id,)
|
||||
).fetchone())
|
||||
|
||||
|
||||
def recover_interrupted_executions() -> int:
|
||||
"""Classify prior in-flight attempts as unknown; never enqueue a retry."""
|
||||
"""Mark provably abandoned attempts unknown without scheduling retries."""
|
||||
now = _hermes_now().isoformat()
|
||||
changed = 0
|
||||
|
||||
def _owner_is_live(record: Dict[str, Any]) -> bool:
|
||||
try:
|
||||
owner_pid = int(record.get("pid"))
|
||||
except (TypeError, ValueError):
|
||||
return False
|
||||
if owner_pid <= 0:
|
||||
return False
|
||||
if owner_pid == os.getpid():
|
||||
return True
|
||||
try:
|
||||
os.kill(owner_pid, 0) # windows-footgun: ok -- liveness probe
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
with _ledger_lock():
|
||||
records = _load_unlocked()
|
||||
for record in records:
|
||||
if record.get("status") not in {"claimed", "running"}:
|
||||
with _lock, _connect() as conn:
|
||||
rows = conn.execute(
|
||||
"""SELECT id, process_id, pid, process_started_at FROM executions
|
||||
WHERE status IN ('claimed','running')"""
|
||||
).fetchall()
|
||||
for row in rows:
|
||||
if row["process_id"] == _PROCESS_ID:
|
||||
continue
|
||||
# Multiple scheduler surfaces can start in one process. Their
|
||||
# startup recovery must not relabel work this same process is
|
||||
# currently executing. A real restart imports this module in a new
|
||||
# process and therefore has a distinct process id.
|
||||
if record.get("process_id") == _PROCESS_ID or _owner_is_live(record):
|
||||
if _owner_is_live(int(row["pid"]), row["process_started_at"]):
|
||||
continue
|
||||
record["status"] = "unknown"
|
||||
record["finished_at"] = now
|
||||
record["error"] = (
|
||||
"Scheduler restarted before this execution reached a durable "
|
||||
"terminal state; whether side effects ran is unknown."
|
||||
cur = conn.execute(
|
||||
"""UPDATE executions SET status='unknown', finished_at=?, error=?
|
||||
WHERE id=? AND status IN ('claimed','running')""",
|
||||
(now,
|
||||
"Scheduler restarted after this execution's owner exited before a durable "
|
||||
"terminal state; whether side effects ran is unknown.",
|
||||
row["id"]),
|
||||
)
|
||||
changed += 1
|
||||
changed += cur.rowcount
|
||||
if changed:
|
||||
_save_unlocked(records)
|
||||
_prune_unlocked(conn)
|
||||
return changed
|
||||
|
||||
|
||||
def list_executions(*, job_id: Optional[str] = None) -> List[Dict[str, Any]]:
|
||||
with _ledger_lock():
|
||||
records = copy.deepcopy(_load_unlocked())
|
||||
def list_executions(
|
||||
*, job_id: Optional[str] = None, limit: int = 50,
|
||||
before_claimed_at: Optional[str] = None,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Return indexed, newest-first execution history with cursor pagination."""
|
||||
clauses: List[str] = []
|
||||
params: List[Any] = []
|
||||
if job_id is not None:
|
||||
records = [record for record in records if record.get("job_id") == job_id]
|
||||
records.sort(key=lambda record: str(record.get("claimed_at", "")), reverse=True)
|
||||
return records
|
||||
clauses.append("job_id=?")
|
||||
params.append(str(job_id))
|
||||
if before_claimed_at is not None:
|
||||
clauses.append("claimed_at < ?")
|
||||
params.append(str(before_claimed_at))
|
||||
where = " WHERE " + " AND ".join(clauses) if clauses else ""
|
||||
params.append(max(1, min(int(limit), 500)))
|
||||
with _lock, _connect() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM executions" + where
|
||||
+ " ORDER BY claimed_at DESC, id DESC LIMIT ?",
|
||||
params,
|
||||
).fetchall()
|
||||
return [dict(row) for row in rows]
|
||||
|
||||
|
||||
def latest_execution(job_id: str) -> Optional[Dict[str, Any]]:
|
||||
records = list_executions(job_id=job_id)
|
||||
return records[0] if records else None
|
||||
rows = list_executions(job_id=job_id, limit=1)
|
||||
return rows[0] if rows else None
|
||||
|
||||
|
||||
def latest_executions(job_ids: List[str]) -> Dict[str, Dict[str, Any]]:
|
||||
"""Load latest execution for many jobs in one indexed query."""
|
||||
clean = [str(job_id) for job_id in dict.fromkeys(job_ids) if job_id]
|
||||
if not clean:
|
||||
return {}
|
||||
placeholders = ",".join("?" for _ in clean)
|
||||
with _lock, _connect() as conn:
|
||||
rows = conn.execute(
|
||||
f"""SELECT e.* FROM executions e
|
||||
WHERE e.job_id IN ({placeholders})
|
||||
AND e.id=(SELECT e2.id FROM executions e2
|
||||
WHERE e2.job_id=e.job_id
|
||||
ORDER BY e2.claimed_at DESC, e2.id DESC LIMIT 1)""",
|
||||
clean,
|
||||
).fetchall()
|
||||
return {row["job_id"]: dict(row) for row in rows}
|
||||
|
|
|
|||
16
cron/jobs.py
16
cron/jobs.py
|
|
@ -455,14 +455,6 @@ def _normalize_job_record(job: Dict[str, Any]) -> Dict[str, Any]:
|
|||
state = "scheduled" if normalized.get("enabled", True) else "paused"
|
||||
normalized["state"] = state
|
||||
|
||||
# Expose the independent latest attempt through existing job detail/list
|
||||
# readers without coupling execution history into schedule persistence.
|
||||
try:
|
||||
from cron.executions import latest_execution
|
||||
normalized["latest_execution"] = latest_execution(job_id)
|
||||
except Exception:
|
||||
normalized["latest_execution"] = None
|
||||
|
||||
return normalized
|
||||
|
||||
|
||||
|
|
@ -1324,6 +1316,14 @@ def list_jobs(include_disabled: bool = False) -> List[Dict[str, Any]]:
|
|||
jobs = [_normalize_job_record(j) for j in load_jobs()]
|
||||
if not include_disabled:
|
||||
jobs = [j for j in jobs if j.get("enabled", True)]
|
||||
try:
|
||||
from cron.executions import latest_executions
|
||||
|
||||
latest = latest_executions([job.get("id", "") for job in jobs])
|
||||
except Exception:
|
||||
latest = {}
|
||||
for job in jobs:
|
||||
job["latest_execution"] = latest.get(job.get("id", ""))
|
||||
return jobs
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -3946,23 +3946,28 @@ def tick(
|
|||
|
||||
try:
|
||||
return pool.submit(_run_and_release)
|
||||
except RuntimeError as submit_err:
|
||||
except Exception as submit_err:
|
||||
with _running_lock:
|
||||
_running_job_ids.discard(job_id)
|
||||
finish_execution(
|
||||
execution["id"],
|
||||
success=False,
|
||||
error=f"Executor dispatch failed: {submit_err}",
|
||||
)
|
||||
# Interpreter began finalizing between the guard above and the
|
||||
# submit — release the in-flight claim we just took and skip.
|
||||
if _interpreter_shutting_down(submit_err):
|
||||
with _running_lock:
|
||||
_running_job_ids.discard(job_id)
|
||||
finish_execution(
|
||||
execution["id"],
|
||||
success=False,
|
||||
error="Interpreter shutdown prevented executor dispatch.",
|
||||
)
|
||||
if isinstance(submit_err, RuntimeError) and _interpreter_shutting_down(submit_err):
|
||||
logger.warning(
|
||||
"Job '%s' not dispatched — interpreter is shutting down",
|
||||
job.get("name", job_id),
|
||||
)
|
||||
return None
|
||||
raise
|
||||
logger.error(
|
||||
"Job '%s' not dispatched: %s",
|
||||
job.get("name", job_id),
|
||||
submit_err,
|
||||
)
|
||||
return None
|
||||
|
||||
# Sequential pass for env-mutating (workdir) jobs.
|
||||
# Queued to a persistent single-thread pool so they run one at a time
|
||||
|
|
|
|||
|
|
@ -82,6 +82,12 @@ class CronScheduler(ABC):
|
|||
Built-in: no-op (it re-reads jobs.json on every tick)."""
|
||||
return None
|
||||
|
||||
def recover_interrupted(self) -> int:
|
||||
"""Run profile-local attempt recovery for every provider lifecycle."""
|
||||
from cron.executions import recover_interrupted_executions
|
||||
|
||||
return recover_interrupted_executions()
|
||||
|
||||
def fire_due(self, job_id: str, *, adapters: Any = None, loop: Any = None) -> bool:
|
||||
"""Run a single job NOW via the shared orchestrator. Called by the
|
||||
inbound fire webhook when an external scheduler signals a job is due.
|
||||
|
|
@ -171,11 +177,10 @@ class InProcessCronScheduler(CronScheduler):
|
|||
import logging
|
||||
from cron.scheduler import tick as cron_tick
|
||||
from cron.jobs import record_ticker_heartbeat
|
||||
from cron.executions import recover_interrupted_executions
|
||||
|
||||
logger = logging.getLogger("cron.scheduler_provider")
|
||||
logger.info("In-process cron scheduler started (interval=%ds)", interval)
|
||||
recovered = recover_interrupted_executions()
|
||||
recovered = self.recover_interrupted()
|
||||
if recovered:
|
||||
logger.warning(
|
||||
"Marked %d interrupted cron execution(s) unknown after restart",
|
||||
|
|
|
|||
|
|
@ -769,6 +769,7 @@ _QUICK_STATE_FILES = (
|
|||
".env",
|
||||
"auth.json",
|
||||
"cron/jobs.json",
|
||||
"cron/executions.db",
|
||||
"gateway_state.json",
|
||||
"channel_directory.json",
|
||||
"channel_aliases.json",
|
||||
|
|
|
|||
|
|
@ -196,6 +196,24 @@ def cron_tick():
|
|||
tick(verbose=True)
|
||||
|
||||
|
||||
def cron_runs(job_id: Optional[str] = None, limit: int = 20):
|
||||
"""Show indexed durable cron execution history."""
|
||||
from cron.executions import list_executions
|
||||
|
||||
records = list_executions(job_id=job_id, limit=limit)
|
||||
if not records:
|
||||
print("No cron execution attempts recorded.")
|
||||
return
|
||||
for record in records:
|
||||
print(
|
||||
f"{record.get('id', '?')} {record.get('status', '?'):<9} "
|
||||
f"job={record.get('job_id', '?')} source={record.get('source', '?')} "
|
||||
f"{record.get('claimed_at', '?')}"
|
||||
)
|
||||
if record.get("error"):
|
||||
print(f" {record['error']}")
|
||||
|
||||
|
||||
def cron_status():
|
||||
"""Show cron execution status."""
|
||||
from cron.jobs import list_jobs
|
||||
|
|
@ -440,6 +458,10 @@ def cron_command(args):
|
|||
cron_tick()
|
||||
return 0
|
||||
|
||||
if subcmd in {"runs", "history"}:
|
||||
cron_runs(getattr(args, "job_id", None), getattr(args, "limit", 20))
|
||||
return 0
|
||||
|
||||
if subcmd in {"create", "add"}:
|
||||
return cron_create(args)
|
||||
|
||||
|
|
@ -459,5 +481,5 @@ def cron_command(args):
|
|||
return _job_action("remove", args.job_id, "Removed")
|
||||
|
||||
print(f"Unknown cron command: {subcmd}")
|
||||
print("Usage: hermes cron [list|create|edit|pause|resume|run|remove|status|tick]")
|
||||
print("Usage: hermes cron [list|create|edit|pause|resume|run|remove|status|runs|tick]")
|
||||
sys.exit(1)
|
||||
|
|
|
|||
|
|
@ -156,6 +156,12 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None:
|
|||
# cron status
|
||||
cron_subparsers.add_parser("status", help="Check if cron scheduler is running")
|
||||
|
||||
cron_runs = cron_subparsers.add_parser(
|
||||
"runs", aliases=["history"], help="Show durable execution attempts"
|
||||
)
|
||||
cron_runs.add_argument("job_id", nargs="?", help="Optional job ID filter")
|
||||
cron_runs.add_argument("--limit", type=int, default=20, help="Rows to show (1-500)")
|
||||
|
||||
# cron tick (mostly for debugging)
|
||||
cron_tick = cron_subparsers.add_parser("tick", help="Run due jobs once and exit")
|
||||
add_accept_hooks_flag(cron_tick)
|
||||
|
|
|
|||
|
|
@ -109,8 +109,7 @@ class ChronosCronScheduler(CronScheduler):
|
|||
# A new provider lifecycle cannot prove what an interrupted prior
|
||||
# process did. Classify those attempts unknown for audit only; do not
|
||||
# requeue them here.
|
||||
from cron.executions import recover_interrupted_executions
|
||||
recover_interrupted_executions()
|
||||
self.recover_interrupted()
|
||||
try:
|
||||
self.reconcile()
|
||||
except Exception as e:
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ from __future__ import annotations
|
|||
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
|
@ -12,8 +13,7 @@ from pathlib import Path
|
|||
def _point_ledger(monkeypatch, tmp_path):
|
||||
import cron.executions as executions
|
||||
|
||||
monkeypatch.setattr(executions, "EXECUTIONS_FILE", tmp_path / "cron" / "executions.json")
|
||||
monkeypatch.setattr(executions, "EXECUTIONS_LOCK_FILE", tmp_path / "cron" / ".executions.lock")
|
||||
monkeypatch.setattr(executions, "EXECUTIONS_FILE", tmp_path / "cron" / "executions.db")
|
||||
return executions
|
||||
|
||||
|
||||
|
|
@ -39,6 +39,78 @@ def test_execution_transitions_are_durable(monkeypatch, tmp_path):
|
|||
assert persisted == [completed]
|
||||
|
||||
|
||||
def test_terminal_execution_cannot_be_rewritten(monkeypatch, tmp_path):
|
||||
executions = _point_ledger(monkeypatch, tmp_path)
|
||||
record = executions.create_execution("immutable", source="builtin")
|
||||
executions.mark_execution_running(record["id"])
|
||||
executions.finish_execution(record["id"], success=True)
|
||||
|
||||
assert executions.finish_execution(
|
||||
record["id"], success=False, error="late writer"
|
||||
) is None
|
||||
assert executions.latest_execution("immutable")["status"] == "completed"
|
||||
|
||||
|
||||
def test_retention_bounds_terminal_history_but_preserves_inflight(monkeypatch, tmp_path):
|
||||
executions = _point_ledger(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(executions, "MAX_TERMINAL_EXECUTIONS", 3)
|
||||
inflight = executions.create_execution("live", source="builtin")
|
||||
executions.mark_execution_running(inflight["id"])
|
||||
for index in range(8):
|
||||
row = executions.create_execution(f"done-{index}", source="builtin")
|
||||
executions.finish_execution(row["id"], success=True)
|
||||
|
||||
records = executions.list_executions(limit=100)
|
||||
assert len([row for row in records if row["status"] == "completed"]) == 3
|
||||
assert executions.latest_execution("live")["status"] == "running"
|
||||
|
||||
|
||||
def test_corrupt_store_fails_closed_without_overwrite(monkeypatch, tmp_path):
|
||||
executions = _point_ledger(monkeypatch, tmp_path)
|
||||
executions.EXECUTIONS_FILE.parent.mkdir(parents=True)
|
||||
executions.EXECUTIONS_FILE.write_bytes(b"not a sqlite database")
|
||||
|
||||
with __import__("pytest").raises(sqlite3.DatabaseError):
|
||||
executions.create_execution("new", source="builtin")
|
||||
assert executions.EXECUTIONS_FILE.read_bytes() == b"not a sqlite database"
|
||||
|
||||
|
||||
def test_execution_history_is_paginated(monkeypatch, tmp_path):
|
||||
executions = _point_ledger(monkeypatch, tmp_path)
|
||||
ids = []
|
||||
for _index in range(5):
|
||||
row = executions.create_execution("paged", source="builtin")
|
||||
executions.finish_execution(row["id"], success=True)
|
||||
ids.append(row["id"])
|
||||
|
||||
first = executions.list_executions(job_id="paged", limit=2)
|
||||
second = executions.list_executions(
|
||||
job_id="paged", limit=2, before_claimed_at=first[-1]["claimed_at"]
|
||||
)
|
||||
assert [row["id"] for row in first] == list(reversed(ids))[:2]
|
||||
assert set(row["id"] for row in first).isdisjoint(row["id"] for row in second)
|
||||
|
||||
|
||||
def test_cron_runs_cli_prints_execution_history(monkeypatch, tmp_path, capsys):
|
||||
executions = _point_ledger(monkeypatch, tmp_path)
|
||||
row = executions.create_execution("cli-job", source="builtin")
|
||||
executions.finish_execution(row["id"], success=False, error="boom")
|
||||
from hermes_cli.cron import cron_runs
|
||||
|
||||
cron_runs("cli-job", limit=10)
|
||||
|
||||
output = capsys.readouterr().out
|
||||
assert row["id"] in output
|
||||
assert "failed" in output
|
||||
assert "boom" in output
|
||||
|
||||
|
||||
def test_quick_backup_includes_execution_ledger():
|
||||
from hermes_cli.backup import _QUICK_STATE_FILES
|
||||
|
||||
assert "cron/executions.db" in _QUICK_STATE_FILES
|
||||
|
||||
|
||||
def test_failed_execution_keeps_error(monkeypatch, tmp_path):
|
||||
executions = _point_ledger(monkeypatch, tmp_path)
|
||||
|
||||
|
|
@ -61,15 +133,29 @@ def test_recovery_does_not_mark_live_process_execution_unknown(monkeypatch, tmp_
|
|||
def test_recovery_does_not_mark_other_live_owner_unknown(monkeypatch, tmp_path):
|
||||
executions = _point_ledger(monkeypatch, tmp_path)
|
||||
record = executions.create_execution("other-live", source="builtin")
|
||||
records = json.loads(executions.EXECUTIONS_FILE.read_text())["executions"]
|
||||
records[0]["process_id"] = "another-import"
|
||||
records[0]["pid"] = os.getpid()
|
||||
executions.EXECUTIONS_FILE.write_text(json.dumps({"version": 1, "executions": records}))
|
||||
with sqlite3.connect(executions.EXECUTIONS_FILE) as conn:
|
||||
conn.execute(
|
||||
"UPDATE executions SET process_id=?, pid=? WHERE id=?",
|
||||
("another-import", os.getpid(), record["id"]),
|
||||
)
|
||||
|
||||
assert executions.recover_interrupted_executions() == 0
|
||||
assert executions.latest_execution("other-live")["status"] == "claimed"
|
||||
|
||||
|
||||
def test_recovery_rejects_recycled_pid(monkeypatch, tmp_path):
|
||||
executions = _point_ledger(monkeypatch, tmp_path)
|
||||
record = executions.create_execution("recycled", source="builtin")
|
||||
with sqlite3.connect(executions.EXECUTIONS_FILE) as conn:
|
||||
conn.execute(
|
||||
"UPDATE executions SET process_id=?, process_started_at=? WHERE id=?",
|
||||
("old-import", -1, record["id"]),
|
||||
)
|
||||
|
||||
assert executions.recover_interrupted_executions() == 1
|
||||
assert executions.latest_execution("recycled")["status"] == "unknown"
|
||||
|
||||
|
||||
def test_restart_marks_interrupted_execution_unknown_without_requeue(tmp_path):
|
||||
"""Real temp-HERMES_HOME subprocess restart: in-flight is audit-only unknown."""
|
||||
home = tmp_path / "home"
|
||||
|
|
@ -121,6 +207,36 @@ def test_restart_marks_interrupted_execution_unknown_without_requeue(tmp_path):
|
|||
assert [r["status"] for r in records] == ["unknown"]
|
||||
|
||||
|
||||
def test_generic_submit_failure_finishes_attempt_and_releases_guard(monkeypatch):
|
||||
import cron.scheduler as scheduler
|
||||
|
||||
class BrokenPool:
|
||||
def submit(self, _callable):
|
||||
raise ValueError("executor rejected")
|
||||
|
||||
finished = []
|
||||
monkeypatch.setattr(
|
||||
scheduler, "create_execution",
|
||||
lambda *_args, **_kwargs: {"id": "exec-submit-fail"},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
scheduler, "finish_execution",
|
||||
lambda execution_id, **kwargs: finished.append((execution_id, kwargs)),
|
||||
)
|
||||
monkeypatch.setattr(scheduler, "get_due_jobs", lambda: [{"id": "submit-fail"}])
|
||||
monkeypatch.setattr(scheduler, "advance_next_run", lambda _job_id: None)
|
||||
monkeypatch.setattr(scheduler, "_get_parallel_pool", lambda _workers: BrokenPool())
|
||||
|
||||
assert scheduler.tick(verbose=False, sync=False) == 0
|
||||
assert finished == [
|
||||
("exec-submit-fail", {
|
||||
"success": False,
|
||||
"error": "Executor dispatch failed: executor rejected",
|
||||
})
|
||||
]
|
||||
assert "submit-fail" not in scheduler.get_running_job_ids()
|
||||
|
||||
|
||||
def test_run_one_job_records_running_then_terminal(monkeypatch):
|
||||
import cron.scheduler as scheduler
|
||||
|
||||
|
|
|
|||
|
|
@ -25,8 +25,8 @@ def _build():
|
|||
|
||||
def test_cron_subactions_present():
|
||||
parser = _build()
|
||||
for action in ("list", "create", "edit", "pause", "resume", "run", "remove", "status", "tick"):
|
||||
ns = parser.parse_args(["cron", action] if action in ("list", "status", "tick")
|
||||
for action in ("list", "create", "edit", "pause", "resume", "run", "remove", "status", "runs", "tick"):
|
||||
ns = parser.parse_args(["cron", action] if action in ("list", "status", "runs", "tick")
|
||||
else ["cron", action, "jobid"] if action in ("pause", "resume", "run", "remove", "edit")
|
||||
else ["cron", "create", "30m"])
|
||||
assert ns.command == "cron"
|
||||
|
|
@ -42,6 +42,10 @@ def test_cron_aliases():
|
|||
for alias in ("rm", "delete"):
|
||||
ns = parser.parse_args(["cron", alias, "jid"])
|
||||
assert ns.cron_command == alias
|
||||
ns = parser.parse_args(["cron", "history", "jid", "--limit", "7"])
|
||||
assert ns.cron_command == "history"
|
||||
assert ns.job_id == "jid"
|
||||
assert ns.limit == 7
|
||||
|
||||
|
||||
def test_cron_create_options():
|
||||
|
|
|
|||
|
|
@ -225,6 +225,20 @@ On each tick Hermes:
|
|||
|
||||
A file lock at `~/.hermes/cron/.tick.lock` prevents overlapping scheduler ticks from double-running the same job batch.
|
||||
|
||||
### Execution history
|
||||
|
||||
Hermes records each claimed cron attempt in the profile-local
|
||||
`~/.hermes/cron/executions.db` before executor or provider dispatch. Attempts
|
||||
move through `claimed`, `running`, and one immutable terminal state:
|
||||
`completed`, `failed`, or `unknown`. After restart, Hermes marks an abandoned
|
||||
attempt `unknown` only when the original PID and process-start fingerprint prove
|
||||
that its owner is gone. Unknown attempts are audit records and are never
|
||||
automatically rerun.
|
||||
|
||||
Inspect recent attempts with `hermes cron runs [job-id] --limit 20` (alias:
|
||||
`history`). Terminal history is bounded; active attempts are never pruned. The
|
||||
ledger is included in quick backups.
|
||||
|
||||
## Delivery options
|
||||
|
||||
When scheduling jobs, you specify where the output goes:
|
||||
|
|
|
|||
|
|
@ -244,7 +244,13 @@ hermes cron status
|
|||
6. 投递最终响应
|
||||
7. 更新运行元数据和下次调度时间
|
||||
|
||||
`~/.hermes/cron/.tick.lock` 处的文件锁防止重叠的调度器 tick 重复运行同一批任务。
|
||||
`~/.hermes/cron/.tick.lock` 文件锁可防止重叠的调度器 tick 重复运行同一批任务。
|
||||
|
||||
### 执行历史
|
||||
|
||||
Hermes 会在执行器或调度提供程序分派之前，将每次已领取的 cron 尝试记录到当前 profile 的 `~/.hermes/cron/executions.db`。尝试会依次进入 `claimed`、`running`，然后进入不可变的终态：`completed`、`failed` 或 `unknown`。重启后，只有原 PID 与进程启动时间指纹能够证明所有者已经消失时，Hermes 才会将遗留尝试标记为 `unknown`。未知尝试仅用于审计，绝不会自动重跑。
|
||||
|
||||
使用 `hermes cron runs [job-id] --limit 20`（别名：`history`）查看最近的尝试。终态历史有界，活动尝试不会被清理；快速备份也包含该账本。
|
||||
|
||||
## 投递选项
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue