feat(scheduler,conversations,api): cron and schedule in the gateway tz, job runs persisted
This commit is contained in:
@@ -210,6 +210,7 @@ async def _async_main() -> None:
|
||||
rotation=Rotation(
|
||||
conversations, gateway.rotation or RotationPolicy(tz=gateway.tz)
|
||||
),
|
||||
tz=gateway.tz,
|
||||
)
|
||||
conversations.scheduler = scheduler
|
||||
|
||||
|
||||
@@ -23,7 +23,7 @@ import logging
|
||||
import re
|
||||
import uuid
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, date, datetime, timedelta
|
||||
from datetime import UTC, date, datetime, timedelta, tzinfo
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
|
||||
from claude_agent_sdk import (
|
||||
@@ -298,6 +298,10 @@ class Conversations:
|
||||
self._tasks: set[asyncio.Task[None]] = set()
|
||||
self._idle_task: asyncio.Task[None] | None = None
|
||||
|
||||
@property
|
||||
def db(self) -> Database:
|
||||
return self._db
|
||||
|
||||
@property
|
||||
def queue(self) -> InjectQueue:
|
||||
return self._queue
|
||||
@@ -964,6 +968,7 @@ class Conversations:
|
||||
if result.text.strip():
|
||||
await self.inject(parent, result.text, urgency="normal", origin="слив")
|
||||
await self.set_status(conv, "merged")
|
||||
await self.mark_closed(conv)
|
||||
self._bus.publish(
|
||||
"conversation.merged",
|
||||
conversation_id=conv.external_id,
|
||||
@@ -1946,7 +1951,7 @@ class Conversations:
|
||||
)
|
||||
|
||||
|
||||
def parse_at(at: str) -> datetime:
|
||||
def parse_at(at: str, tz: tzinfo = UTC) -> datetime:
|
||||
raw = at.strip()
|
||||
match = _RELATIVE.match(raw.replace(" ", ""))
|
||||
if match:
|
||||
@@ -1954,7 +1959,7 @@ def parse_at(at: str) -> datetime:
|
||||
return datetime.now(UTC) + timedelta(seconds=int(amount) * _UNITS[unit])
|
||||
parsed = datetime.fromisoformat(raw)
|
||||
if parsed.tzinfo is None:
|
||||
parsed = parsed.astimezone()
|
||||
parsed = parsed.replace(tzinfo=tz)
|
||||
return parsed.astimezone(UTC)
|
||||
|
||||
|
||||
|
||||
@@ -14,19 +14,27 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import functools
|
||||
import json
|
||||
import logging
|
||||
import traceback
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from datetime import UTC, datetime, timedelta, tzinfo
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from croniter import croniter
|
||||
from pgqueuer import PgQueuer, Queries
|
||||
from pgqueuer.core.executors import ScheduleExecutor
|
||||
from pgqueuer.models import JobId
|
||||
from sqlalchemy import func
|
||||
from sqlmodel import col, select
|
||||
from starlette.applications import Starlette
|
||||
from starlette.responses import JSONResponse
|
||||
from starlette.routing import Route
|
||||
|
||||
from beaver_gateway.core.conversations import parse_at
|
||||
from beaver_gateway.storage.models import JobRunRecord
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Awaitable, Callable, Coroutine, Sequence
|
||||
@@ -42,12 +50,27 @@ if TYPE_CHECKING:
|
||||
from beaver_gateway.core.rotation import Rotation
|
||||
from beaver_gateway.storage.models import Conversation
|
||||
|
||||
__all__ = ["INJECT", "Budget", "Job", "JobRun", "Scheduler"]
|
||||
__all__ = ["INJECT", "Budget", "Job", "JobRun", "LocalCron", "Scheduler", "next_run"]
|
||||
|
||||
_log = logging.getLogger("beaver_gateway.core.scheduler")
|
||||
|
||||
INJECT = "inject"
|
||||
RETRY = timedelta(minutes=15)
|
||||
ERROR_MAX = 2000
|
||||
|
||||
|
||||
def next_run(expression: str, tz: tzinfo, now: datetime | None = None) -> datetime:
|
||||
"""The cron's next fire time, read in ``tz``, returned in UTC."""
|
||||
start = (now or datetime.now(UTC)).astimezone(tz)
|
||||
return croniter(expression, start_time=start).get_next(datetime).astimezone(UTC)
|
||||
|
||||
|
||||
@dataclass
|
||||
class LocalCron(ScheduleExecutor):
|
||||
tz: tzinfo = UTC
|
||||
|
||||
def get_next(self) -> datetime:
|
||||
return next_run(self.parameters.expression, self.tz)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
@@ -165,9 +188,11 @@ class Scheduler:
|
||||
budget: Budget | None = None,
|
||||
rotation: Rotation | None = None,
|
||||
heartbeat: timedelta = timedelta(seconds=30),
|
||||
tz: str = "UTC",
|
||||
) -> None:
|
||||
self.conversations = conversations
|
||||
self.rotation = rotation
|
||||
self.tz = ZoneInfo(tz)
|
||||
self.budget = budget or Budget()
|
||||
self._jobs = {job.name: job for job in jobs}
|
||||
self._driver = driver
|
||||
@@ -229,7 +254,12 @@ class Scheduler:
|
||||
|
||||
if job.cron:
|
||||
|
||||
@pgq.schedule(job.name, job.cron, clean_old=True)
|
||||
@pgq.schedule(
|
||||
job.name,
|
||||
job.cron,
|
||||
clean_old=True,
|
||||
executor_factory=functools.partial(LocalCron, tz=self.tz),
|
||||
)
|
||||
async def cron(_schedule: PgSchedule) -> None:
|
||||
await self._dispatch(job, trigger="cron", payload={})
|
||||
|
||||
@@ -257,19 +287,33 @@ class Scheduler:
|
||||
return
|
||||
started = datetime.now(UTC)
|
||||
bus.publish("job.start", job=job.name, trigger=trigger)
|
||||
status = "done"
|
||||
status, error = "done", None
|
||||
try:
|
||||
await job.run(JobRun(job, trigger, payload, self))
|
||||
except Exception: # noqa: BLE001
|
||||
status = "failed"
|
||||
except Exception as exc: # noqa: BLE001
|
||||
status, error = "failed", _error_text(exc)
|
||||
_log.exception("job %s (%s) failed", job.name, trigger)
|
||||
self._runs[job.name] = {
|
||||
"trigger": trigger,
|
||||
"started_at": started.isoformat(timespec="seconds"),
|
||||
"status": status,
|
||||
}
|
||||
record = JobRunRecord(
|
||||
job=job.name,
|
||||
trigger=trigger,
|
||||
started_at=started,
|
||||
finished_at=datetime.now(UTC),
|
||||
status=status,
|
||||
error=error,
|
||||
payload=payload,
|
||||
)
|
||||
await self._record(record)
|
||||
self._runs[job.name] = self._run_public(record)
|
||||
bus.publish("job.end", job=job.name, trigger=trigger, status=status)
|
||||
|
||||
async def _record(self, record: JobRunRecord) -> None:
|
||||
try:
|
||||
async with self.conversations.db.session() as session:
|
||||
session.add(record)
|
||||
await session.commit()
|
||||
except Exception: # noqa: BLE001
|
||||
_log.exception("job %s: run not recorded", record.job)
|
||||
|
||||
async def trigger(
|
||||
self,
|
||||
job: Job,
|
||||
@@ -325,14 +369,14 @@ class Scheduler:
|
||||
if self._queries is None:
|
||||
msg = "scheduler needs postgres; `schedule` is unavailable"
|
||||
raise RuntimeError(msg)
|
||||
when = parse_at(at)
|
||||
when = parse_at(at, self.tz).astimezone(self.tz)
|
||||
delay = max(when - datetime.now(UTC), timedelta(0))
|
||||
payload = _encode(
|
||||
{
|
||||
"conversation": conv.external_id,
|
||||
"text": text,
|
||||
"urgency": urgency,
|
||||
"at": when.isoformat(timespec="seconds"),
|
||||
"at": when.isoformat(timespec="minutes"),
|
||||
}
|
||||
)
|
||||
ids = await self._queries.enqueue(
|
||||
@@ -347,7 +391,7 @@ class Scheduler:
|
||||
"schedule.created",
|
||||
conversation_id=conv.external_id,
|
||||
job=job_id,
|
||||
execute_at=when.isoformat(timespec="seconds"),
|
||||
execute_at=when.isoformat(timespec="minutes"),
|
||||
)
|
||||
return job_id, when
|
||||
|
||||
@@ -411,16 +455,22 @@ class Scheduler:
|
||||
for row in await self._queries.peek_schedule():
|
||||
crons[str(row.entrypoint)] = row
|
||||
queue = [
|
||||
_job_public(row) for row in await self._queries.browse_queue(limit=200)
|
||||
_job_public(row, self.tz)
|
||||
for row in await self._queries.browse_queue(limit=200)
|
||||
]
|
||||
utilization = await self.utilization()
|
||||
last = await self._last_runs()
|
||||
return {
|
||||
"enabled": self.enabled,
|
||||
"tz": str(self.tz),
|
||||
"utilization": utilization,
|
||||
"throttled": utilization is not None
|
||||
and utilization > self.budget.threshold,
|
||||
"threshold": self.budget.threshold,
|
||||
"jobs": [self._job_public(job, crons.get(job.name)) for job in self.jobs],
|
||||
"jobs": [
|
||||
self._job_public(job, crons.get(job.name), last.get(job.name))
|
||||
for job in self.jobs
|
||||
],
|
||||
"queue": queue,
|
||||
}
|
||||
|
||||
@@ -428,24 +478,60 @@ class Scheduler:
|
||||
if self._queries is None:
|
||||
return []
|
||||
rows = await self._queries.browse_queue(limit=500, entrypoints=[INJECT])
|
||||
out = [_job_public(row) for row in rows]
|
||||
out = [_job_public(row, self.tz) for row in rows]
|
||||
if conv is not None:
|
||||
out = [
|
||||
j for j in out if j["payload"].get("conversation") == conv.external_id
|
||||
]
|
||||
return out
|
||||
|
||||
def _job_public(self, job: Job, cron: PgSchedule | None) -> dict[str, Any]:
|
||||
async def runs(self, name: str, limit: int = 50) -> list[dict[str, Any]]:
|
||||
stmt = (
|
||||
select(JobRunRecord)
|
||||
.where(JobRunRecord.job == name)
|
||||
.order_by(col(JobRunRecord.id).desc())
|
||||
.limit(limit)
|
||||
)
|
||||
async with self.conversations.db.session() as session:
|
||||
rows = (await session.exec(stmt)).all()
|
||||
return [self._run_public(row) for row in rows]
|
||||
|
||||
async def _last_runs(self) -> dict[str, dict[str, Any]]:
|
||||
newest = select(func.max(JobRunRecord.id)).group_by(JobRunRecord.job)
|
||||
stmt = select(JobRunRecord).where(col(JobRunRecord.id).in_(newest))
|
||||
try:
|
||||
async with self.conversations.db.session() as session:
|
||||
rows = (await session.exec(stmt)).all()
|
||||
except Exception: # noqa: BLE001
|
||||
_log.exception("job runs unreadable, using memory")
|
||||
return dict(self._runs)
|
||||
return {row.job: self._run_public(row) for row in rows}
|
||||
|
||||
def _job_public(
|
||||
self, job: Job, cron: PgSchedule | None, run: dict[str, Any] | None
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"name": job.name,
|
||||
"cron": job.cron,
|
||||
"webhook": job.webhook,
|
||||
"events": list(job.events),
|
||||
"critical": job.critical,
|
||||
"next_run": _iso(cron.next_run) if cron is not None else None,
|
||||
"last_run": _iso(cron.last_run) if cron is not None else None,
|
||||
"next_run": _iso(cron.next_run, self.tz) if cron is not None else None,
|
||||
"last_run": _iso(cron.last_run, self.tz) if cron is not None else None,
|
||||
"status": str(cron.status) if cron is not None else None,
|
||||
"run": self._runs.get(job.name),
|
||||
"run": run or self._runs.get(job.name),
|
||||
}
|
||||
|
||||
def _run_public(self, row: JobRunRecord) -> dict[str, Any]:
|
||||
return {
|
||||
"id": row.id,
|
||||
"job": row.job,
|
||||
"trigger": row.trigger,
|
||||
"started_at": _iso(row.started_at, self.tz),
|
||||
"finished_at": _iso(row.finished_at, self.tz),
|
||||
"status": row.status,
|
||||
"error": row.error,
|
||||
"payload": row.payload,
|
||||
}
|
||||
|
||||
# ---- http ----------------------------------------------------------
|
||||
@@ -506,21 +592,28 @@ def _decode(payload: bytes | None) -> dict[str, Any]:
|
||||
return data if isinstance(data, dict) else {"raw": data}
|
||||
|
||||
|
||||
def _job_public(row: PgJob) -> dict[str, Any]:
|
||||
def _job_public(row: PgJob, tz: tzinfo) -> dict[str, Any]:
|
||||
return {
|
||||
"id": int(row.id),
|
||||
"entrypoint": row.entrypoint,
|
||||
"status": str(row.status),
|
||||
"execute_after": _iso(row.execute_after),
|
||||
"created": _iso(row.created),
|
||||
"execute_after": _iso(row.execute_after, tz),
|
||||
"created": _iso(row.created, tz),
|
||||
"attempts": row.attempts,
|
||||
"payload": _decode(row.payload),
|
||||
}
|
||||
|
||||
|
||||
def _error_text(exc: BaseException) -> str:
|
||||
head = f"{type(exc).__name__}: {exc}"
|
||||
return f"{head}\n{''.join(traceback.format_exception(exc))}"[:ERROR_MAX]
|
||||
|
||||
|
||||
def _aware(value: datetime) -> datetime:
|
||||
return value if value.tzinfo is not None else value.replace(tzinfo=UTC)
|
||||
|
||||
|
||||
def _iso(value: datetime | None) -> str | None:
|
||||
return _aware(value).isoformat(timespec="seconds") if value is not None else None
|
||||
def _iso(value: datetime | None, tz: tzinfo = UTC) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
return _aware(value).astimezone(tz).isoformat(timespec="seconds")
|
||||
|
||||
@@ -22,7 +22,9 @@ from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
try:
|
||||
from claude_agent_sdk._cli_version import __cli_version__ as _cli_version
|
||||
from claude_agent_sdk._cli_version import __cli_version__
|
||||
|
||||
_cli_version = str(__cli_version__)
|
||||
except ImportError: # pragma: no cover
|
||||
_cli_version = "2.1.248"
|
||||
|
||||
|
||||
@@ -62,7 +62,7 @@ if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
from beaver_gateway.core.conversations import Conversations
|
||||
from beaver_gateway.core.scheduler import Scheduler
|
||||
from beaver_gateway.core.scheduler import Job, Scheduler
|
||||
from beaver_gateway.frontends.base import GatewayRuntime
|
||||
|
||||
_log = logging.getLogger("beaver_gateway.frontends.api")
|
||||
@@ -580,15 +580,27 @@ def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> Fa
|
||||
await require_token(request, runtime, scope=SCOPE)
|
||||
return await scheduler_of().snapshot()
|
||||
|
||||
def job_of(scheduler: Scheduler, name: str) -> Job:
|
||||
job = scheduler.job(name)
|
||||
if job is None:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, f"no job {name!r}")
|
||||
return job
|
||||
|
||||
@app.post("/jobs/{name}/run", status_code=status.HTTP_202_ACCEPTED)
|
||||
async def run_job(name: str, request: Request) -> dict[str, Any]:
|
||||
await require_token(request, runtime, scope=SCOPE)
|
||||
scheduler = scheduler_of()
|
||||
job = scheduler.job(name)
|
||||
if job is None:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, f"no job {name!r}")
|
||||
job = job_of(scheduler, name)
|
||||
return {"job": await scheduler.trigger(job, await body_of(request))}
|
||||
|
||||
@app.get("/jobs/{name}/runs")
|
||||
async def job_runs(name: str, request: Request) -> dict[str, Any]:
|
||||
await require_token(request, runtime, scope=SCOPE)
|
||||
scheduler = scheduler_of()
|
||||
job = job_of(scheduler, name)
|
||||
limit = min(int(request.query_params.get("limit", 50)), 500)
|
||||
return {"job": job.name, "runs": await scheduler.runs(job.name, limit)}
|
||||
|
||||
@app.delete("/jobs/queue/{job_id}")
|
||||
async def cancel_job(job_id: int, request: Request) -> dict[str, Any]:
|
||||
await require_token(request, runtime, scope=SCOPE)
|
||||
|
||||
@@ -360,6 +360,32 @@ class RateLimit(SQLModel, table=True):
|
||||
)
|
||||
|
||||
|
||||
class JobRunRecord(SQLModel, table=True):
|
||||
"""One finished run of a scheduler job (§3.6): what fired it, how it ended.
|
||||
|
||||
``error`` keeps the head of the traceback of a failed run; ``payload``
|
||||
is the trigger's data (``{}`` for cron) and should stay small.
|
||||
"""
|
||||
|
||||
__tablename__ = "job_runs"
|
||||
|
||||
id: int | None = Field(default=None, primary_key=True)
|
||||
job: str = Field(index=True)
|
||||
trigger: str
|
||||
started_at: datetime
|
||||
finished_at: datetime
|
||||
status: str
|
||||
error: str | None = Field(default=None)
|
||||
payload: dict[str, Any] = Field(
|
||||
default_factory=dict,
|
||||
sa_column=Column(
|
||||
JSON().with_variant(JSONB(), "postgresql"),
|
||||
nullable=False,
|
||||
server_default=text("'{}'"),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"AuditLog",
|
||||
"Conversation",
|
||||
@@ -367,6 +393,7 @@ __all__ = [
|
||||
"ConversationMessage",
|
||||
"Delivery",
|
||||
"InjectQueueItem",
|
||||
"JobRunRecord",
|
||||
"RateLimit",
|
||||
"TelegramUpdate",
|
||||
"Token",
|
||||
|
||||
+36
-1
@@ -21,6 +21,7 @@ from test_conversations import ScriptedClient, World
|
||||
from beaver_gateway.core.conversation_store import rewrite_messages
|
||||
from beaver_gateway.core.auth import TokenStore
|
||||
from beaver_gateway.core.registry import McpRegistry
|
||||
from beaver_gateway.core.scheduler import Job, JobRun, Scheduler
|
||||
from beaver_gateway.core.transcript import build_entries
|
||||
from beaver_gateway.frontends.admin import AdminFrontend
|
||||
from beaver_gateway.frontends.admin.frontend import build_app as build_admin
|
||||
@@ -96,7 +97,12 @@ class ToolClient(ScriptedClient):
|
||||
|
||||
|
||||
class Api:
|
||||
def __init__(self, world: World, memory_root: Path | None = None) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
world: World,
|
||||
memory_root: Path | None = None,
|
||||
scheduler: Scheduler | None = None,
|
||||
) -> None:
|
||||
self.world = world
|
||||
self.store = TokenStore(world.db, bootstrap={"t": TOKEN})
|
||||
self.runtime = GatewayRuntime(
|
||||
@@ -112,6 +118,7 @@ class Api:
|
||||
conversations=world.conversations,
|
||||
bus=world.bus,
|
||||
pool=world.pool,
|
||||
scheduler=scheduler,
|
||||
)
|
||||
self.app = build_api(self.runtime, memory_root=memory_root)
|
||||
self.http = AsyncClient(
|
||||
@@ -146,6 +153,34 @@ async def seed_usage(world: World, rows: list[dict[str, Any]]) -> None:
|
||||
await session.commit()
|
||||
|
||||
|
||||
async def test_job_runs_come_from_the_db(world: World) -> None:
|
||||
async def ping(_run: JobRun) -> None:
|
||||
pass
|
||||
|
||||
scheduler = Scheduler(
|
||||
conversations=world.conversations, jobs=[Job("ping", ping)], tz="Europe/Warsaw"
|
||||
)
|
||||
api = Api(world, scheduler=scheduler)
|
||||
await scheduler.start()
|
||||
res = await api.http.post("/jobs/ping/run", headers=HEADERS)
|
||||
assert res.status_code == 202, res.text
|
||||
for _ in range(100):
|
||||
runs = (await api.get("/jobs/ping/runs", {"limit": 10}))["runs"]
|
||||
if runs:
|
||||
break
|
||||
await asyncio.sleep(0.02)
|
||||
assert [(r["trigger"], r["status"], r["payload"]) for r in runs] == [
|
||||
("manual", "done", {})
|
||||
]
|
||||
jobs = await api.get("/jobs")
|
||||
assert jobs["tz"] == "Europe/Warsaw"
|
||||
assert jobs["jobs"][0]["run"]["id"] == runs[0]["id"]
|
||||
assert jobs["jobs"][0]["run"]["finished_at"].endswith("+02:00")
|
||||
missing = await api.http.get("/jobs/nope/runs", headers=HEADERS)
|
||||
assert missing.status_code == 404
|
||||
await scheduler.stop()
|
||||
|
||||
|
||||
async def test_usage_groups_by_agent_day_and_model(world: World) -> None:
|
||||
api = Api(world)
|
||||
now = datetime.now(UTC).replace(tzinfo=None)
|
||||
|
||||
@@ -4,6 +4,7 @@ import uuid
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import pytest
|
||||
from claude_agent_sdk import (
|
||||
@@ -629,6 +630,9 @@ async def test_schedule_without_scheduler_and_parse_at(world: World) -> None:
|
||||
assert parse_at("2026-09-01T10:00:00+02:00") == datetime(
|
||||
2026, 9, 1, 8, 0, tzinfo=UTC
|
||||
)
|
||||
assert parse_at("2026-09-01T10:00", ZoneInfo("Europe/Warsaw")) == datetime(
|
||||
2026, 9, 1, 8, 0, tzinfo=UTC
|
||||
)
|
||||
delta = (parse_at("+15m") - datetime.now(UTC)).total_seconds()
|
||||
assert 14 * 60 < delta <= 15 * 60
|
||||
with pytest.raises(ValueError, match="Invalid isoformat"):
|
||||
|
||||
+100
-7
@@ -1,9 +1,11 @@
|
||||
import asyncio
|
||||
import inspect
|
||||
import os
|
||||
import tempfile
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import psycopg
|
||||
import pytest
|
||||
@@ -11,14 +13,12 @@ from httpx import ASGITransport, AsyncClient
|
||||
from pgqueuer import PsycopgDriver, Queries
|
||||
from test_conversations import World
|
||||
|
||||
from beaver_gateway.core.scheduler import Budget, Job, JobRun, Scheduler
|
||||
from beaver_gateway.core.conversations import parse_at
|
||||
from beaver_gateway.core.scheduler import Budget, Job, JobRun, Scheduler, next_run
|
||||
from beaver_gateway.storage.models import RateLimit
|
||||
|
||||
DATABASE_URL = os.environ.get("TEST_DATABASE_URL")
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not DATABASE_URL, reason="TEST_DATABASE_URL (postgres) is not set"
|
||||
)
|
||||
WARSAW = ZoneInfo("Europe/Warsaw")
|
||||
|
||||
|
||||
class Pg:
|
||||
@@ -37,6 +37,8 @@ class Pg:
|
||||
|
||||
@pytest.fixture
|
||||
async def pg() -> Pg:
|
||||
if not DATABASE_URL:
|
||||
pytest.skip("TEST_DATABASE_URL (postgres) is not set")
|
||||
handle = Pg()
|
||||
queries = Queries(await handle.driver())
|
||||
if await queries.has_table("pgqueuer"):
|
||||
@@ -61,6 +63,8 @@ async def until(pred, timeout: float = 5.0) -> Any:
|
||||
deadline = asyncio.get_running_loop().time() + timeout
|
||||
while asyncio.get_running_loop().time() < deadline:
|
||||
value = pred()
|
||||
if inspect.isawaitable(value):
|
||||
value = await value
|
||||
if value:
|
||||
return value
|
||||
await asyncio.sleep(0.05)
|
||||
@@ -68,16 +72,91 @@ async def until(pred, timeout: float = 5.0) -> Any:
|
||||
raise AssertionError(msg)
|
||||
|
||||
|
||||
def test_cron_is_read_in_the_gateway_tz() -> None:
|
||||
summer = datetime(2026, 7, 1, 12, tzinfo=UTC)
|
||||
assert next_run("0 4 * * *", WARSAW, summer) == datetime(2026, 7, 2, 2, tzinfo=UTC)
|
||||
assert next_run("0 4 * * *", UTC, summer) == datetime(2026, 7, 2, 4, tzinfo=UTC)
|
||||
before_fallback = datetime(2026, 10, 24, 12, tzinfo=UTC)
|
||||
assert next_run("0 4 * * *", WARSAW, before_fallback) == datetime(
|
||||
2026, 10, 25, 3, tzinfo=UTC
|
||||
)
|
||||
during_fallback = datetime(2026, 10, 25, 1, 30, tzinfo=UTC)
|
||||
assert next_run("0 4 * * *", WARSAW, during_fallback) == datetime(
|
||||
2026, 10, 25, 3, tzinfo=UTC
|
||||
)
|
||||
assert next_run("*/15 * * * *", WARSAW, during_fallback) == datetime(
|
||||
2026, 10, 25, 1, 45, tzinfo=UTC
|
||||
)
|
||||
soon = next_run("* * * * *", WARSAW)
|
||||
assert soon.tzinfo is UTC
|
||||
assert timedelta(0) < soon - datetime.now(UTC) <= timedelta(minutes=1)
|
||||
|
||||
|
||||
def test_parse_at_reads_naive_iso_in_tz() -> None:
|
||||
assert parse_at("2026-09-01T10:00", WARSAW) == datetime(
|
||||
2026, 9, 1, 8, 0, tzinfo=UTC
|
||||
)
|
||||
assert parse_at("2026-12-01T10:00", WARSAW) == datetime(
|
||||
2026, 12, 1, 9, 0, tzinfo=UTC
|
||||
)
|
||||
assert parse_at("2026-09-01T10:00:00+02:00", WARSAW) == datetime(
|
||||
2026, 9, 1, 8, 0, tzinfo=UTC
|
||||
)
|
||||
assert parse_at("2026-09-01T10:00") == datetime(2026, 9, 1, 10, 0, tzinfo=UTC)
|
||||
|
||||
|
||||
async def test_runs_are_recorded_without_postgres(world: World) -> None:
|
||||
async def ok(run: JobRun) -> None:
|
||||
run.payload["seen"] = True
|
||||
|
||||
async def boom(_run: JobRun) -> None:
|
||||
msg = "no network"
|
||||
raise ConnectionError(msg)
|
||||
|
||||
scheduler = Scheduler(
|
||||
conversations=world.conversations,
|
||||
jobs=[Job("ok", ok, webhook=True), Job("boom", boom)],
|
||||
tz="Europe/Warsaw",
|
||||
)
|
||||
await scheduler.start()
|
||||
await scheduler.hook("ok", {"stack": "x"})
|
||||
await scheduler.trigger(scheduler.job("boom"))
|
||||
await until(lambda: len(scheduler._runs) == 2) # noqa: SLF001
|
||||
runs = await scheduler.runs("boom")
|
||||
assert [r["status"] for r in runs] == ["failed"]
|
||||
assert runs[0]["error"].startswith("ConnectionError: no network\n")
|
||||
assert runs[0]["trigger"] == "manual"
|
||||
assert runs[0]["started_at"] <= runs[0]["finished_at"]
|
||||
assert runs[0]["started_at"].endswith("+02:00")
|
||||
ok_runs = await scheduler.runs("ok")
|
||||
assert ok_runs[0]["payload"] == {"stack": "x", "seen": True}
|
||||
assert ok_runs[0]["trigger"] == "webhook"
|
||||
await scheduler.hook("ok", {"stack": "y"})
|
||||
await until(lambda: scheduler._runs["ok"]["payload"].get("stack") == "y") # noqa: SLF001
|
||||
assert len(await scheduler.runs("ok")) == 2
|
||||
assert len(await scheduler.runs("ok", limit=1)) == 1
|
||||
snapshot = await scheduler.snapshot()
|
||||
assert snapshot["tz"] == "Europe/Warsaw"
|
||||
runs_by_name = {j["name"]: j["run"] for j in snapshot["jobs"]}
|
||||
assert runs_by_name["ok"]["payload"]["stack"] == "y"
|
||||
assert runs_by_name["boom"]["status"] == "failed"
|
||||
await scheduler.stop()
|
||||
|
||||
|
||||
async def test_schedule_survives_a_restart(world: World, pg: Pg) -> None:
|
||||
conv = await world.conversations.create(kind="master", agent="a", origin="test")
|
||||
first = Scheduler(conversations=world.conversations, driver=await pg.driver())
|
||||
first = Scheduler(
|
||||
conversations=world.conversations, driver=await pg.driver(), tz="Europe/Warsaw"
|
||||
)
|
||||
world.conversations.scheduler = first
|
||||
await first.start()
|
||||
job_id, when = await world.conversations.schedule(conv, "+1s", "push X")
|
||||
assert job_id is not None
|
||||
assert 0 < (when - datetime.now(UTC)).total_seconds() <= 1
|
||||
assert when.utcoffset() == datetime.now(WARSAW).utcoffset()
|
||||
queued = await world.conversations.schedules(conv)
|
||||
assert [q["payload"]["text"] for q in queued] == ["push X"]
|
||||
assert queued[0]["payload"]["at"] == when.isoformat(timespec="minutes")
|
||||
await first.stop()
|
||||
|
||||
second = Scheduler(conversations=world.conversations, driver=await pg.driver())
|
||||
@@ -138,9 +217,14 @@ async def test_webhook_runs_the_job_with_its_payload(world: World, pg: Pg) -> No
|
||||
assert missing.status_code == 404
|
||||
await until(lambda: seen)
|
||||
assert seen == [{"trigger": "webhook", "stack": "x"}]
|
||||
await until(lambda: scheduler._runs.get("ping")) # noqa: SLF001
|
||||
snapshot = await scheduler.snapshot()
|
||||
assert [j["name"] for j in snapshot["jobs"]] == ["ping", "quiet"]
|
||||
assert snapshot["enabled"] and snapshot["jobs"][0]["webhook"]
|
||||
assert snapshot["jobs"][0]["run"]["status"] == "done"
|
||||
assert snapshot["jobs"][0]["run"]["trigger"] == "webhook"
|
||||
runs = await scheduler.runs("ping")
|
||||
assert [r["payload"] for r in runs] == [{"stack": "x"}]
|
||||
await scheduler.stop()
|
||||
|
||||
|
||||
@@ -170,6 +254,7 @@ async def test_non_critical_jobs_wait_while_the_window_is_hot(
|
||||
],
|
||||
driver=await pg.driver(),
|
||||
budget=Budget(threshold=0.7),
|
||||
tz="Europe/Warsaw",
|
||||
)
|
||||
await scheduler.start()
|
||||
events: list[dict[str, Any]] = []
|
||||
@@ -193,5 +278,13 @@ async def test_non_critical_jobs_wait_while_the_window_is_hot(
|
||||
assert snapshot["throttled"] and snapshot["utilization"] == 0.9
|
||||
crons = {j["name"]: j["cron"] for j in snapshot["jobs"]}
|
||||
assert crons == {"vibegram": "*/10 * * * *", "rotation": "0 * * * *"}
|
||||
await until(lambda: all(j["next_run"] for j in snapshot["jobs"]) or True)
|
||||
|
||||
async def crons_ready() -> dict[str, Any] | None:
|
||||
snap = await scheduler.snapshot()
|
||||
return snap if all(j["next_run"] for j in snap["jobs"]) else None
|
||||
|
||||
snapshot = await until(crons_ready)
|
||||
expected = datetime.now(WARSAW).utcoffset()
|
||||
for job_info in snapshot["jobs"]:
|
||||
assert datetime.fromisoformat(job_info["next_run"]).utcoffset() == expected
|
||||
await scheduler.stop()
|
||||
|
||||
+12
-1
@@ -270,6 +270,17 @@ export interface QueuedJob {
|
||||
status: string;
|
||||
}
|
||||
|
||||
export interface JobRunInfo {
|
||||
error: string | null;
|
||||
finished_at: string | null;
|
||||
id: number | null;
|
||||
job: string;
|
||||
payload: Record<string, unknown>;
|
||||
started_at: string;
|
||||
status: string;
|
||||
trigger: string;
|
||||
}
|
||||
|
||||
export interface JobInfo {
|
||||
critical: boolean;
|
||||
cron: string | null;
|
||||
@@ -277,7 +288,7 @@ export interface JobInfo {
|
||||
last_run: string | null;
|
||||
name: string;
|
||||
next_run: string | null;
|
||||
run: { trigger: string; started_at: string; status: string } | null;
|
||||
run: JobRunInfo | null;
|
||||
status: string | null;
|
||||
webhook: boolean;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user