class ScheduledTaskStore:
"""SQLite-backed confirmation, ownership, schedule and run history store.
Each method creates its own connection; SQLite's IMMEDIATE transactions make
confirmation and claim atomic across processes sharing the same database.
Claimed runs are not automatically replayed if their worker disappears.
"""
def __init__(self, path: str | Path) -> None:
self.path = str(path)
if self.path == ":memory:":
raise ValueError("use a persistent SQLite path")
with self._connect() as db:
db.executescript("""
CREATE TABLE IF NOT EXISTS scheduled_drafts (
draft_id TEXT PRIMARY KEY, payload TEXT NOT NULL, payload_hash TEXT NOT NULL,
expires_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS scheduled_tasks (
task_id TEXT PRIMARY KEY, owner_keys TEXT NOT NULL, chat_id TEXT NOT NULL,
instruction TEXT NOT NULL, schedule TEXT NOT NULL, policy TEXT NOT NULL,
timezone TEXT NOT NULL, state TEXT NOT NULL, next_run_at REAL,
revision INTEGER NOT NULL, created_at REAL NOT NULL, updated_at REAL NOT NULL,
tenant_key TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS scheduled_tasks_due ON scheduled_tasks(state, next_run_at);
CREATE TABLE IF NOT EXISTS scheduled_runs (
run_id TEXT PRIMARY KEY, task_id TEXT NOT NULL, scheduled_for REAL NOT NULL,
status TEXT NOT NULL, session_id TEXT NOT NULL, started_at REAL,
finished_at REAL, summary TEXT, delivery_status TEXT NOT NULL DEFAULT 'pending',
UNIQUE(task_id, scheduled_for)
);
CREATE INDEX IF NOT EXISTS scheduled_runs_task ON scheduled_runs(task_id, scheduled_for);
CREATE TABLE IF NOT EXISTS scheduled_reconciliations (
run_id TEXT PRIMARY KEY, tenant_key TEXT NOT NULL, actor TEXT NOT NULL,
decision TEXT NOT NULL, reason TEXT NOT NULL, reconciled_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS scheduled_delivery_attempts (
token TEXT PRIMARY KEY, run_id TEXT NOT NULL, status TEXT NOT NULL,
started_at REAL NOT NULL, finished_at REAL
);
""")
db.execute("BEGIN IMMEDIATE")
columns = {row["name"] for row in db.execute("PRAGMA table_info(scheduled_runs)")}
if "delivery_status" not in columns:
db.execute("ALTER TABLE scheduled_runs ADD COLUMN delivery_status TEXT NOT NULL DEFAULT 'pending'")
for name, field_type in (
("wait_kind", "TEXT"),
("wait_id", "TEXT"),
("wait_deadline", "REAL"),
("delivery_deadline", "REAL"),
("delivery_token", "TEXT"),
("lease_token", "TEXT"),
("lease_until", "REAL"),
):
if name not in columns:
db.execute(f"ALTER TABLE scheduled_runs ADD COLUMN {name} {field_type}")
task_columns = {row["name"] for row in db.execute("PRAGMA table_info(scheduled_tasks)")}
if "tenant_key" not in task_columns:
db.execute("ALTER TABLE scheduled_tasks ADD COLUMN tenant_key TEXT NOT NULL DEFAULT ''")
@contextmanager
def _connect(self) -> Iterator[sqlite3.Connection]:
db = sqlite_connect(self.path)
db.row_factory = sqlite3.Row
try:
with db:
yield db
finally:
db.close()
def create_draft(
self,
*,
tenant_key: str,
owner_keys: tuple[str, ...],
chat_id: str,
instruction: str,
schedule: Schedule,
now: float,
policy: str = "read_only",
timezone: str = "UTC",
ttl: float = 1800,
) -> ScheduledDraft:
tenant_key, owner = _tenant(tenant_key), _keys(owner_keys)
now, ttl = _number(now, "now"), _number(ttl, "ttl")
if (
ttl <= 0
or not isinstance(chat_id, str)
or not chat_id.strip()
or not isinstance(instruction, str)
or not instruction.strip()
):
raise ValueError("positive ttl, chat_id and instruction required")
if policy not in ("read_only", "confirm_writes"):
raise ValueError("unsupported scheduled policy")
if schedule.next_after(now, timezone) is None:
raise ValueError("schedule has no future occurrence")
payload = json.dumps(
{
"owner_keys": owner,
"tenant_key": tenant_key,
"chat_id": chat_id,
"instruction": instruction,
"schedule": asdict(schedule),
"policy": policy,
"timezone": timezone,
},
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
)
digest = hashlib.sha256(payload.encode()).hexdigest()
result = ScheduledDraft(uuid.uuid4().hex, digest, now + ttl)
with self._connect() as db:
db.execute(
"INSERT INTO scheduled_drafts VALUES (?, ?, ?, ?)",
(result.draft_id, payload, digest, result.expires_at),
)
return result
def confirm_draft(
self, draft_id: str, owner_keys: tuple[str, ...], payload_hash: str, *, tenant_key: str, now: float
) -> ScheduledTask:
now = _number(now, "now")
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute("SELECT * FROM scheduled_drafts WHERE draft_id=?", (draft_id,)).fetchone()
if row is None or row["expires_at"] <= now:
raise ValueError("draft missing or expired")
if (
row["payload_hash"] != payload_hash
or hashlib.sha256(row["payload"].encode()).hexdigest() != payload_hash
):
raise ValueError("draft payload hash mismatch")
data = json.loads(row["payload"])
if data.get("tenant_key") != _tenant(tenant_key) or not _is_owner(tuple(data["owner_keys"]), owner_keys):
raise PermissionError("only the creator may confirm")
schedule = Schedule.from_dict(data["schedule"])
next_run = schedule.next_after(now, data["timezone"])
if next_run is None:
raise ValueError("schedule has no future occurrence")
task_id = uuid.uuid4().hex
db.execute(
"INSERT INTO scheduled_tasks "
"(task_id, owner_keys, chat_id, instruction, schedule, policy, timezone, state, "
"next_run_at, revision, created_at, updated_at, tenant_key) "
"VALUES (?, ?, ?, ?, ?, ?, ?, 'active', ?, 1, ?, ?, ?)",
(
task_id,
json.dumps(data["owner_keys"]),
data["chat_id"],
data["instruction"],
json.dumps(data["schedule"]),
data["policy"],
data["timezone"],
next_run,
now,
now,
tenant_key,
),
)
db.execute("DELETE FROM scheduled_drafts WHERE draft_id=?", (draft_id,))
row = db.execute("SELECT * FROM scheduled_tasks WHERE task_id=?", (task_id,)).fetchone()
return _task(row)
def get(self, task_id: str) -> ScheduledTask | None:
with self._connect() as db:
row = db.execute("SELECT * FROM scheduled_tasks WHERE task_id=?", (task_id,)).fetchone()
return _task(row) if row else None
def list_owned(self, owner_keys: tuple[str, ...], *, tenant_key: str) -> list[ScheduledTask]:
owner = _keys(owner_keys)
tenant_key = _tenant(tenant_key)
with self._connect() as db:
return [
task
for row in db.execute(
"SELECT * FROM scheduled_tasks WHERE state != 'deleted' AND tenant_key=? ORDER BY created_at",
(tenant_key,),
)
if _is_owner((task := _task(row)).owner_keys, owner)
]
def set_state(
self,
task_id: str,
owner_keys: tuple[str, ...],
state: str,
*,
tenant_key: str,
expected_revision: int,
now: float,
) -> ScheduledTask:
if state not in ("active", "paused", "deleted"):
raise ValueError("invalid task state")
now = _number(now, "now")
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute("SELECT * FROM scheduled_tasks WHERE task_id=?", (task_id,)).fetchone()
if row is None:
raise ValueError("task not found")
task = _task(row)
if task.tenant_key != _tenant(tenant_key) or not _is_owner(task.owner_keys, owner_keys):
raise PermissionError("only the creator may manage this task")
if (
task.revision != expected_revision
or task.state == "deleted"
or (task.state == "completed" and state != "deleted")
):
raise ValueError("stale revision or terminal task")
next_run = task.next_run_at
if state == "active" and task.state != "active":
next_run = task.schedule.next_after(now, task.timezone)
if next_run is None:
if task.schedule.kind != "once" or task.next_run_at is None:
raise ValueError("schedule has no future occurrence")
run_id = uuid.uuid4().hex
db.execute(
"INSERT INTO scheduled_runs "
"(run_id, task_id, scheduled_for, status, session_id, finished_at) VALUES (?, ?, ?, ?, ?, ?)",
(run_id, task.task_id, task.next_run_at, "missed", f"scheduled:{task.task_id}:{run_id}", now),
)
state = "completed"
db.execute(
"UPDATE scheduled_tasks SET state=?, next_run_at=?, revision=revision+1, updated_at=? WHERE task_id=?",
(state, next_run, now, task_id),
)
return _task(db.execute("SELECT * FROM scheduled_tasks WHERE task_id=?", (task_id,)).fetchone())
def refresh_timezones(
self, timezone_for: Callable[[ScheduledTask], str], *, now: float, task_ids: tuple[str, ...] | None = None
) -> None:
"""Recompute future calendar triggers when their trusted timezone changes.
A trigger already due is left for claim_due to settle as run or missed.
The callback runs without holding a database transaction.
"""
now = _number(now, "now")
with self._connect() as db:
if task_ids is None:
rows = db.execute("SELECT * FROM scheduled_tasks WHERE state='active'").fetchall()
elif not task_ids:
return
elif len(task_ids) <= 1024:
marks = ",".join("?" for _ in task_ids)
rows = db.execute(
f"SELECT * FROM scheduled_tasks WHERE state='active' AND task_id IN ({marks})", task_ids
).fetchall()
else:
raise ValueError("too many timezone refresh tasks")
tasks = []
for row in rows:
try:
tasks.append(_task(row))
except (ValueError, TypeError, OverflowError, OSError, KeyError):
logger.warning("quarantining invalid scheduled task %s", row["task_id"])
with self._connect() as db:
db.execute(
"UPDATE scheduled_tasks SET state='error', next_run_at=NULL, revision=revision+1, updated_at=? "
"WHERE task_id=? AND state='active' AND revision=?",
(now, row["task_id"], row["revision"]),
)
for task in tasks:
if task.schedule.kind != "calendar":
continue
try:
zone = timezone_for(task)
if zone == task.timezone or task.next_run_at is None or task.next_run_at <= now:
continue
next_run = task.schedule.next_after(now, zone)
except Exception: # noqa: BLE001 - one broken resolver must not block every task
logger.warning("quarantining scheduled task with invalid timezone %s", task.task_id)
with self._connect() as db:
db.execute(
"UPDATE scheduled_tasks SET state='error', next_run_at=NULL, revision=revision+1, updated_at=? "
"WHERE task_id=? AND state='active' AND revision=?",
(now, task.task_id, task.revision),
)
continue
with self._connect() as db:
db.execute(
"UPDATE scheduled_tasks SET timezone=?, next_run_at=?, revision=revision+1, updated_at=? "
"WHERE task_id=? AND state='active' AND revision=?",
(zone, next_run, now, task.task_id, task.revision),
)
def active_calendar_batch(self, *, cursor: str = "", limit: int = 128) -> tuple[list[ScheduledTask], str]:
if (
not isinstance(cursor, str)
or not isinstance(limit, int)
or isinstance(limit, bool)
or not 1 <= limit <= 1024
):
raise ValueError("invalid calendar scan bounds")
with self._connect() as db:
rows = db.execute(
"SELECT * FROM scheduled_tasks WHERE state='active' AND task_id>? ORDER BY task_id LIMIT ?",
(cursor, limit),
).fetchall()
tasks = []
for row in rows:
try:
task = _task(row)
except (ValueError, TypeError, OverflowError, OSError, KeyError):
continue
if task.schedule.kind == "calendar":
tasks.append(task)
next_cursor = rows[-1]["task_id"] if len(rows) == limit else ""
return tasks, next_cursor
def claim_due(
self,
*,
now: float,
timezone_for: Callable[[ScheduledTask], str] | None = None,
grace_seconds: float = 60,
lease_seconds: float = 60,
) -> ScheduledRun | None:
now, grace_seconds = _number(now, "now"), _number(grace_seconds, "grace_seconds")
lease_seconds = _number(lease_seconds, "lease_seconds")
if grace_seconds < 0 or lease_seconds <= 0:
raise ValueError("grace_seconds must be nonnegative and lease_seconds positive")
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
while True:
row = db.execute(
"SELECT * FROM scheduled_tasks WHERE state='active' AND next_run_at<=? "
"ORDER BY next_run_at, task_id LIMIT 1",
(now,),
).fetchone()
if row is None:
return None
try:
task = _task(row)
_tenant(task.tenant_key)
_keys(task.owner_keys)
zone = timezone_for(task) if timezone_for else task.timezone
next_run = task.schedule.next_after(now, zone)
break
except Exception: # noqa: BLE001 - isolate invalid stored schedules or timezone resolvers
logger.warning("quarantining invalid scheduled task %s", row["task_id"])
db.execute(
"UPDATE scheduled_tasks SET state='error', next_run_at=NULL, revision=revision+1, updated_at=? "
"WHERE task_id=?",
(now, row["task_id"]),
)
state = "completed" if next_run is None else "active"
occurrence = float(task.next_run_at) # type: ignore[arg-type]
overlap = db.execute(
"SELECT 1 FROM scheduled_runs WHERE task_id=? AND status IN (?, ?, ?, ?) LIMIT 1",
(task.task_id, *_ACTIVE_RUNS),
).fetchone()
status = "missed" if now - occurrence > grace_seconds else "skipped_overlap" if overlap else "running"
run_id = uuid.uuid4().hex
session = f"scheduled:{task.task_id}:{run_id}"
db.execute(
"INSERT INTO scheduled_runs "
"(run_id, task_id, scheduled_for, status, session_id, started_at, "
"finished_at, lease_token, lease_until) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
run_id,
task.task_id,
occurrence,
status,
session,
now if status == "running" else None,
None if status == "running" else now,
uuid.uuid4().hex if status == "running" else None,
now + lease_seconds if status == "running" else None,
),
)
db.execute(
"UPDATE scheduled_tasks SET state=?, next_run_at=?, revision=revision+1, updated_at=? WHERE task_id=?",
(state, next_run, now, task.task_id),
)
return _run(db.execute("SELECT * FROM scheduled_runs WHERE run_id=?", (run_id,)).fetchone())
def finish_run(
self,
run_id: str,
status: str,
*,
now: float,
summary: str | None = None,
expected_status: str | None = None,
expected_token: str | None = None,
reset_delivery: bool = False,
) -> ScheduledRun:
if status not in _FINAL_RUNS:
raise ValueError("invalid run status")
now = _number(now, "now")
if summary is not None and (not isinstance(summary, str) or len(summary) > 2000):
raise ValueError("summary must be at most 2000 characters")
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute("SELECT * FROM scheduled_runs WHERE run_id=?", (run_id,)).fetchone()
if row is None or row["status"] not in _ACTIVE_RUNS or row["status"] == "unknown":
raise ScheduledRunConflict("run not active")
if expected_status is not None and row["status"] != expected_status:
raise ScheduledRunConflict("run status changed")
if row["status"] == "running" and not expected_token:
raise ScheduledRunConflict("running task requires its lease token")
if expected_token is not None and (
row["lease_token"] != expected_token or row["lease_until"] is None or row["lease_until"] <= now
):
raise ScheduledRunConflict("run lease expired or changed")
finished = None if status in _ACTIVE_RUNS else now
if reset_delivery and row["delivery_token"]:
db.execute(
"UPDATE scheduled_delivery_attempts SET status='unknown', finished_at=? "
"WHERE token=? AND status='sending'",
(now, row["delivery_token"]),
)
db.execute(
"UPDATE scheduled_runs SET status=?, finished_at=?, summary=?, "
"delivery_status=CASE WHEN ? THEN 'pending' ELSE delivery_status END, "
"delivery_token=CASE WHEN ? THEN NULL ELSE delivery_token END, "
"delivery_deadline=CASE WHEN ? THEN NULL ELSE delivery_deadline END, lease_until=NULL WHERE run_id=?",
(status, finished, summary, reset_delivery, reset_delivery, reset_delivery, run_id),
)
return _run(db.execute("SELECT * FROM scheduled_runs WHERE run_id=?", (run_id,)).fetchone())
def get_run(self, run_id: str) -> ScheduledRun | None:
with self._connect() as db:
row = db.execute("SELECT * FROM scheduled_runs WHERE run_id=?", (run_id,)).fetchone()
return _run(row) if row else None
def validate_lease(self, run_id: str, token: str | None, *, now: float) -> None:
if not token:
raise PermissionError("scheduled lease is missing")
with self._connect() as db:
row = db.execute(
"SELECT 1 FROM scheduled_runs WHERE run_id=? AND status='running' AND lease_token=? AND lease_until>?",
(run_id, token, _number(now, "now")),
).fetchone()
if row is None:
raise PermissionError("scheduled lease expired or changed")
def renew_lease(self, run_id: str, token: str | None, *, now: float, lease_seconds: float = 60) -> bool:
now, lease_seconds = _number(now, "now"), _number(lease_seconds, "lease_seconds")
if not token or lease_seconds <= 0:
raise ValueError("valid lease token and duration required")
with self._connect() as db:
cursor = db.execute(
"UPDATE scheduled_runs SET lease_until=? WHERE run_id=? AND status='running' "
"AND lease_token=? AND lease_until>?",
(now + lease_seconds, run_id, token, now),
)
return cursor.rowcount == 1
def expire_runs(self, *, now: float) -> int:
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
now = _number(now, "now")
db.execute(
"UPDATE scheduled_delivery_attempts SET status='unknown', finished_at=? "
"WHERE status='sending' AND token IN (SELECT delivery_token FROM scheduled_runs "
"WHERE status='running' AND (lease_until IS NULL OR lease_until<=?))",
(now, now),
)
cursor = db.execute(
"UPDATE scheduled_runs SET status='unknown', summary='execution lease expired', lease_until=NULL, "
"delivery_status='pending', delivery_token=NULL, delivery_deadline=NULL "
"WHERE status='running' AND (lease_until IS NULL OR lease_until<=?)",
(now,),
)
return cursor.rowcount
def unknown_runs(self, *, tenant_key: str) -> list[ScheduledRun]:
"""List a tenant's frozen runs; the caller must authorize the operator."""
tenant_key = _tenant(tenant_key)
with self._connect() as db:
return [
_run(row)
for row in db.execute(
"SELECT r.* FROM scheduled_runs r JOIN scheduled_tasks t ON r.task_id=t.task_id "
"WHERE r.status='unknown' AND t.tenant_key=? ORDER BY r.scheduled_for",
(tenant_key,),
)
]
def reconcile_unknown(
self, run_id: str, *, tenant_key: str, actor: str, decision: str, reason: str, now: float
) -> ScheduledRun:
"""Record an authorized operator's external reconciliation; never replay work.
Authorization of the actor and the correctness of the investigation are
the host's responsibility. Only a frozen run in this tenant is accepted.
"""
tenant_key, now = _tenant(tenant_key), _number(now, "now")
if (
not isinstance(actor, str)
or not actor.strip()
or not isinstance(reason, str)
or not reason.strip()
or len(reason) > 2000
or decision not in ("confirmed_succeeded", "confirmed_failed", "abandoned")
):
raise ValueError("actor, reason and explicit reconciliation decision are required")
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute(
"SELECT r.*, t.tenant_key FROM scheduled_runs r JOIN scheduled_tasks t ON r.task_id=t.task_id "
"WHERE r.run_id=?",
(run_id,),
).fetchone()
if row is None or row["tenant_key"] != tenant_key:
raise PermissionError("unknown run is unavailable in this tenant")
if row["status"] != "unknown":
raise ScheduledRunConflict("run is not frozen")
db.execute(
"INSERT INTO scheduled_reconciliations VALUES (?, ?, ?, ?, ?, ?)",
(run_id, tenant_key, actor, decision, reason, now),
)
status = "succeeded" if decision == "confirmed_succeeded" else "failed"
db.execute(
"UPDATE scheduled_runs SET status=?, finished_at=?, lease_token=NULL, lease_until=NULL, "
"wait_kind=NULL, wait_id=NULL, wait_deadline=NULL WHERE run_id=? AND status='unknown'",
(status, now, run_id),
)
return _run(db.execute("SELECT * FROM scheduled_runs WHERE run_id=?", (run_id,)).fetchone())
def claim_resume(
self, run_id: str, wait_id: str, *, expected_status: str, now: float, lease_seconds: float = 60
) -> ScheduledRun:
now, lease_seconds = _number(now, "now"), _number(lease_seconds, "lease_seconds")
if (
expected_status not in ("awaiting_confirmation", "awaiting_authorization")
or not wait_id
or lease_seconds <= 0
):
raise ValueError("invalid continuation lease")
token = uuid.uuid4().hex
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
cursor = db.execute(
"UPDATE scheduled_runs SET status='running', lease_token=?, lease_until=? "
"WHERE run_id=? AND status=? AND wait_id=? AND wait_kind=? AND EXISTS "
"(SELECT 1 FROM scheduled_tasks WHERE task_id=scheduled_runs.task_id "
"AND state IN ('active', 'completed') AND tenant_key != '')",
(
token,
now + lease_seconds,
run_id,
expected_status,
wait_id,
"approval" if expected_status == "awaiting_confirmation" else "authorization",
),
)
if cursor.rowcount != 1:
raise ScheduledRunConflict("scheduled continuation was already claimed or changed")
return _run(db.execute("SELECT * FROM scheduled_runs WHERE run_id=?", (run_id,)).fetchone())
def record_wait(
self, run_id: str, kind: str, wait_id: str, deadline: float, *, expected_token: str, now: float
) -> ScheduledRun:
"""Record a persisted continuation before its card becomes actionable."""
if kind not in ("approval", "authorization") or not wait_id:
raise ValueError("invalid continuation")
deadline, now = _number(deadline, "deadline"), _number(now, "now")
if not expected_token:
raise ValueError("lease token is required")
with self._connect() as db:
cursor = db.execute(
"UPDATE scheduled_runs SET wait_kind=?, wait_id=?, wait_deadline=? "
"WHERE run_id=? AND status='running' AND lease_token=? AND lease_until>?",
(kind, wait_id, deadline, run_id, expected_token, now),
)
if cursor.rowcount != 1:
raise ScheduledRunConflict("run cannot accept a continuation")
return _run(db.execute("SELECT * FROM scheduled_runs WHERE run_id=?", (run_id,)).fetchone())
def clear_wait(self, run_id: str, wait_id: str, *, expected_token: str, now: float) -> None:
with self._connect() as db:
cursor = db.execute(
"UPDATE scheduled_runs SET wait_kind=NULL, wait_id=NULL, wait_deadline=NULL "
"WHERE run_id=? AND wait_id=? AND status='running' AND lease_token=? AND lease_until>?",
(run_id, wait_id, expected_token, _number(now, "now")),
)
if cursor.rowcount != 1:
raise ScheduledRunConflict("run lease expired or continuation changed")
def expired_waits(self, *, now: float) -> list[ScheduledRun]:
with self._connect() as db:
return [
_run(row)
for row in db.execute(
"SELECT * FROM scheduled_runs WHERE "
"((status='awaiting_confirmation' AND wait_kind='approval') OR "
"(status='awaiting_authorization' AND wait_kind='authorization')) "
"AND wait_deadline<=? ORDER BY wait_deadline",
(_number(now, "now"),),
)
]
def runs(self, task_id: str) -> list[ScheduledRun]:
with self._connect() as db:
return [
_run(row)
for row in db.execute("SELECT * FROM scheduled_runs WHERE task_id=? ORDER BY scheduled_for", (task_id,))
]
def claim_delivery(self, run_id: str, *, now: float, lease_seconds: float = 60) -> str | None:
now, lease_seconds = _number(now, "now"), _number(lease_seconds, "lease_seconds")
if lease_seconds <= 0:
raise ValueError("delivery lease must be positive")
token = uuid.uuid4().hex
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
cursor = db.execute(
"UPDATE scheduled_runs SET delivery_status='sending', delivery_deadline=?, delivery_token=? "
"WHERE run_id=? AND status!='running' AND delivery_status='pending'",
(now + lease_seconds, token, run_id),
)
if cursor.rowcount != 1:
return None
db.execute(
"INSERT INTO scheduled_delivery_attempts VALUES (?, ?, 'sending', ?, NULL)", (token, run_id, now)
)
return token
def expire_deliveries(self, *, now: float) -> None:
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
now = _number(now, "now")
db.execute(
"UPDATE scheduled_delivery_attempts SET status='unknown', finished_at=? "
"WHERE status='sending' AND token IN (SELECT delivery_token FROM scheduled_runs "
"WHERE delivery_status='sending' AND delivery_deadline<=?)",
(now, now),
)
db.execute(
"UPDATE scheduled_runs SET delivery_status='unknown' "
"WHERE delivery_status='sending' AND delivery_deadline<=?",
(now,),
)
def record_delivery(self, run_id: str, status: str, *, token: str, now: float) -> ScheduledRun:
"""Record delivery independently from execution, without retrying an uncertain send."""
if status not in ("delivered", "unknown"):
raise ValueError("invalid delivery status")
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
cursor = db.execute(
"UPDATE scheduled_runs SET delivery_status=?, delivery_deadline=NULL, delivery_token=NULL "
"WHERE run_id=? AND delivery_status='sending' AND delivery_token=?",
(status, run_id, token),
)
if cursor.rowcount != 1:
raise ValueError("run not found or delivery already settled")
db.execute(
"UPDATE scheduled_delivery_attempts SET status=?, finished_at=? WHERE token=?",
(status, _number(now, "now"), token),
)
return _run(db.execute("SELECT * FROM scheduled_runs WHERE run_id=?", (run_id,)).fetchone())
def pending_deliveries(self, *, limit: int = 128) -> list[ScheduledRun]:
"""Return notices never yet submitted; uncertain attempts are never replayed."""
if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 1024:
raise ValueError("invalid delivery scan bound")
with self._connect() as db:
return [
_run(row)
for row in db.execute(
"SELECT * FROM scheduled_runs WHERE status != 'running' "
"AND delivery_status='pending' ORDER BY scheduled_for LIMIT ?",
(limit,),
)
]