133 lines
5.0 KiB
Python
133 lines
5.0 KiB
Python
import json
|
|
import sqlite3
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Reservation:
|
|
state: str
|
|
response: Any = None
|
|
|
|
|
|
class IdempotencyLedgerBusy(RuntimeError):
|
|
"""Raised when a durable completion cannot acquire the SQLite write lock."""
|
|
|
|
|
|
class IdempotencyLedger:
|
|
"""Small durable ledger for replaying successful Gitea mutations."""
|
|
|
|
def __init__(
|
|
self,
|
|
path: str | Path,
|
|
*,
|
|
ttl_seconds: float,
|
|
max_entries: int,
|
|
lock_timeout_seconds: float = 0.1,
|
|
clock: Callable[[], float] = time.time,
|
|
) -> None:
|
|
self.path = Path(path)
|
|
self.ttl_seconds = ttl_seconds
|
|
self.max_entries = max_entries
|
|
self.lock_timeout_seconds = lock_timeout_seconds
|
|
self.clock = clock
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._initialize()
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
connection = sqlite3.connect(self.path, timeout=self.lock_timeout_seconds)
|
|
connection.execute(
|
|
f"PRAGMA busy_timeout = {max(1, int(self.lock_timeout_seconds * 1000))}"
|
|
)
|
|
return connection
|
|
|
|
def _initialize(self) -> None:
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS idempotency_operations (
|
|
key TEXT PRIMARY KEY,
|
|
fingerprint TEXT NOT NULL,
|
|
status TEXT NOT NULL CHECK(status IN ('pending', 'completed')),
|
|
response_json TEXT,
|
|
created_at REAL NOT NULL,
|
|
completed_at REAL
|
|
)
|
|
"""
|
|
)
|
|
connection.execute(
|
|
"CREATE INDEX IF NOT EXISTS idempotency_completed_at_idx "
|
|
"ON idempotency_operations(completed_at)"
|
|
)
|
|
|
|
@staticmethod
|
|
def _fingerprint(value: tuple[Any, ...]) -> str:
|
|
return json.dumps(value, separators=(",", ":"), sort_keys=True)
|
|
|
|
def reserve(self, key: str, fingerprint: tuple[Any, ...]) -> Reservation:
|
|
encoded = self._fingerprint(fingerprint)
|
|
now = self.clock()
|
|
try:
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
connection.execute(
|
|
"DELETE FROM idempotency_operations "
|
|
"WHERE status = 'completed' AND completed_at <= ?",
|
|
(now - self.ttl_seconds,),
|
|
)
|
|
row = connection.execute(
|
|
"SELECT fingerprint, status, response_json FROM idempotency_operations "
|
|
"WHERE key = ?",
|
|
(key,),
|
|
).fetchone()
|
|
if row is not None:
|
|
if row[0] != encoded:
|
|
return Reservation("conflict")
|
|
if row[1] == "completed":
|
|
return Reservation("completed", json.loads(row[2]))
|
|
return Reservation("pending")
|
|
count = connection.execute(
|
|
"SELECT COUNT(*) FROM idempotency_operations"
|
|
).fetchone()[0]
|
|
if count >= self.max_entries:
|
|
completed = connection.execute(
|
|
"SELECT key FROM idempotency_operations "
|
|
"WHERE status = 'completed' ORDER BY completed_at, rowid LIMIT 1"
|
|
).fetchone()
|
|
if completed is None:
|
|
return Reservation("busy")
|
|
connection.execute(
|
|
"DELETE FROM idempotency_operations WHERE key = ?",
|
|
(completed[0],),
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO idempotency_operations "
|
|
"(key, fingerprint, status, created_at) VALUES (?, ?, 'pending', ?)",
|
|
(key, encoded, now),
|
|
)
|
|
except sqlite3.OperationalError as exc:
|
|
if "locked" in str(exc).lower() or "busy" in str(exc).lower():
|
|
return Reservation("busy")
|
|
raise
|
|
return Reservation("reserved")
|
|
|
|
def clear(self) -> None:
|
|
with self._connect() as connection:
|
|
connection.execute("DELETE FROM idempotency_operations")
|
|
|
|
def complete(self, key: str, response: Any) -> None:
|
|
encoded = json.dumps(response, separators=(",", ":"), sort_keys=True)
|
|
try:
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"UPDATE idempotency_operations SET status = 'completed', "
|
|
"response_json = ?, completed_at = ? WHERE key = ?",
|
|
(encoded, self.clock(), key),
|
|
)
|
|
except sqlite3.OperationalError as exc:
|
|
if "locked" in str(exc).lower() or "busy" in str(exc).lower():
|
|
raise IdempotencyLedgerBusy from exc
|
|
raise
|