From 97248c3703c15dde9c4953ed489144b659bae532 Mon Sep 17 00:00:00 2001 From: summer97souls Date: Sat, 27 Jun 2026 22:43:38 +0530 Subject: [PATCH] feat: add sentinel_events append-only schema and TIMESTAMPTZ migration feat: add SentinelEvent enum and write_event/async_write_event helpers feat: wire write_event into all sync state transitions in core.py feat: wire async_write_event into all async state transitions in async_core.py feat: add fencing token rotation on reconcile in sync and async reconciler feat: add RESET and RELEASED events to reconciler and release() feat: add sen history CLI command with --limit flag feat: add history tests for happy path and reconciliation in sync and async docs: update README with execution history section and CLI history command chore: move python-dotenv to core dependencies, remove cli optional release: bump version to 0.5.0, update CHANGELOG --- CHANGELOG.md | 16 ++++ README.md | 27 ++++++- ROADMAP.md | 8 +- sentinel-py/pyproject.toml | 10 ++- sentinel-py/sentinel/async_core.py | 11 ++- sentinel-py/sentinel/async_reconcilliation.py | 13 +++- sentinel-py/sentinel/cli.py | 21 ++++- sentinel-py/sentinel/core.py | 13 ++++ sentinel-py/sentinel/events.py | 73 ++++++++++++++++++ .../migrations/0003_sentinel_events.py | 76 +++++++++++++++++++ sentinel-py/sentinel/integrations/models.py | 47 +++++++++++- sentinel-py/sentinel/reconciliation.py | 11 ++- sentinel-py/sentinel/schema.py | 19 ++++- tests/conftest.py | 10 ++- tests/test_async_core.py | 7 -- tests/test_async_history.py | 56 ++++++++++++++ tests/test_async_reconciliation.py | 6 -- tests/test_history.py | 46 +++++++++++ 18 files changed, 434 insertions(+), 36 deletions(-) create mode 100644 sentinel-py/sentinel/events.py create mode 100644 sentinel-py/sentinel/integrations/migrations/0003_sentinel_events.py create mode 100644 tests/test_async_history.py create mode 100644 tests/test_history.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9826216..c76d4f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,22 @@ The format loosely follows Keep a Changelog. --- +## 0.5.0 — 2026-06-27 + +### Added +- **`sentinel_events` append-only event log** — new table recording every state transition with `key`, `event`, `owner_id`, `fencing_token`, `metadata`, and `occurred_at`. Indexed on `(key, occurred_at, fencing_token)` for efficient history queries. +- **`write_event` / `async_write_event`** — internal helpers that write events atomically within the same transaction as each state transition. Events and lease changes commit together or not at all. +- **`SentinelEvent` enum** — typed event constants (`ACQUIRED`, `REJECTED`, `EXECUTING`, `COMPLETED`, `EXPIRED`, `RECONCILING`, `RESET`, `RELEASED`) used across sync and async paths. +- **`history(conn, key, *, limit=50)`** — query the event log for any key, returns a list of `EventRecord` dataclasses ordered oldest first. +- **`sen history `** — CLI command to print the full execution history for a key directly from the terminal. Accepts `--limit` to cap results. +- **Fencing token rotation on reconciliation** — `reconcile()` now issues a new fencing token via `nextval('sentinel_token_seq')` at the moment it transitions a lease to `reconciling`. The original worker's token is immediately invalidated. +- **`RESET` and `RELEASED` events** — explicit events written when a lease is reset by the reconciler or explicitly released by the caller. + +### Changed +- `TIMESTAMPTZ` replaces `TIMESTAMP` across all lease columns for correct timezone-aware behaviour in distributed environments. + +--- + ## 0.4.2 — 2026-06-21 ### Added diff --git a/README.md b/README.md index a41c17b..9aba13e 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,7 @@ Sentinel ships with `sen`, a command-line tool for inspecting lease state direct ```bash sen inspect +sen history --limit 20 ``` `sen` reads `DATABASE_URL` from your environment or a `.env` file automatically. @@ -102,6 +103,8 @@ result = sentinel.once( ) ``` +--- + ### Reading the result ```python @@ -124,6 +127,29 @@ else: --- +## Execution History + +Sentinel records every state transition to an append-only event log. Every acquire, rejection, execution start, completion, expiry, and reconciliation is written atomically with the lease change that caused it. + +```bash +sen history +sen history --limit 20 +``` + +Example output: +History for key: payment-order-789 (3 events) +2026-06-27 14:02:01 acquired token=42 owner=worker-a +2026-06-27 14:02:01 executing token=42 owner=worker-a +2026-06-27 14:02:03 completed token=42 owner=worker-a + +### What the event log tells you + +The sequence of events is the ground truth for what happened to any execution key. A `reconciling` event followed by `acquired` means a new worker took over. A `reconciling` event followed by `completed` means the original worker finished inside the uncertainty window. An `expired` event means the worker raised an exception and the lease was collapsed immediately. + +The log does not resolve uncertainty — it records it honestly. + +--- + ## Async If you're working in an async context, use `AsyncSentinel`: @@ -254,7 +280,6 @@ The core execution semantics are stable as of 0.4.0. Reconciliation tooling and ## Roadmap - Redis cache for better throughput -- Append-only execution event log (`sentinel_events`) - FastAPI integration - Correlate — cross-service execution observability - Stronger reconciliation tooling diff --git a/ROADMAP.md b/ROADMAP.md index ee58e9a..e93bdc5 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -47,10 +47,9 @@ The focus is to keep the core small, explicit, and reliable rather than continuo ## Near Term (0.5.0) -- [ ] `sentinel_events` append-only event log written in the same transaction as state transitions -- [ ] FastAPI integration after async core stabilizes -- [ ] Expanded test coverage for edge cases and contention scenarios -- [ ] Improved examples and integration guides +- [x] `sentinel_events` append-only event log written in the same transaction as state transitions +- [x] Expanded test coverage for edge cases and contention scenarios +- [x] Improved examples and integration guides --- @@ -59,6 +58,7 @@ The focus is to keep the core small, explicit, and reliable rather than continuo These are ideas being explored and are not guaranteed. - [ ] Correlate — separate OSS library that reads `sentinel_events` for cross-service execution observability +- [ ] FastAPI integration after async core stabilizes - [ ] Batched adaptive heartbeats — bucket-level batch `UPDATE` with adaptive interval - [ ] Additional framework adapters (Flask, Starlette) - [ ] Embeddable reconciliation dashboard diff --git a/sentinel-py/pyproject.toml b/sentinel-py/pyproject.toml index 132ce70..9a97155 100644 --- a/sentinel-py/pyproject.toml +++ b/sentinel-py/pyproject.toml @@ -1,13 +1,13 @@ [project] name = "sentinel-coordination" -version = "0.4.1" +version = "0.5.0" description = "PostgreSQL-backed execution coordination primitive for correctness-sensitive distributed work." authors = [ { name = "Sreejay Reddy", email = "reddysreejay@gmail.com" } ] readme = "README.md" requires-python = ">=3.9" -dependencies = ["psycopg[binary]"] +dependencies = ["psycopg[binary]", "python-dotenv"] keywords = [ "postgresql", @@ -16,7 +16,10 @@ keywords = [ "execution-coordination", "fault-tolerance", "concurrency", - "fencing-tokens" + "fencing-tokens", + "append-only log", + "execution history", + "audit log" ] classifiers = [ @@ -40,7 +43,6 @@ sen = "sentinel.cli:main" [project.optional-dependencies] django = ["django>=4.2"] -cli = ["python-dotenv"] [build-system] requires = ["setuptools>=61.0"] diff --git a/sentinel-py/sentinel/async_core.py b/sentinel-py/sentinel/async_core.py index fc4bd3e..4c63be9 100644 --- a/sentinel-py/sentinel/async_core.py +++ b/sentinel-py/sentinel/async_core.py @@ -1,6 +1,7 @@ import json from .utils import get_owner_id, row_to_dict from .result import AcquireResult, OperationResult, InspectResult +from .events import SentinelEvent, async_write_event async def acquire(conn, key, *, owner_id=None, ttl_ms=10000, hard_ttl_ms = None): @@ -44,6 +45,7 @@ async def acquire(conn, key, *, owner_id=None, ttl_ms=10000, hard_ttl_ms = None) if result is not None: row = row_to_dict(cur, result) + await async_write_event(cur, key, SentinelEvent.ACQUIRED, owner_id=row["owner_id"], fencing_token=row["fencing_token"]) await conn.commit() @@ -71,6 +73,7 @@ async def acquire(conn, key, *, owner_id=None, ttl_ms=10000, hard_ttl_ms = None) if result is not None: row = row_to_dict(cur, result) + await async_write_event(cur, key, SentinelEvent.REJECTED, owner_id=row["owner_id"], fencing_token=row["fencing_token"]) await conn.commit() @@ -102,6 +105,7 @@ async def start_execution(conn, key, *, owner_id, fencing_token): success = result is not None if result is not None: row = row_to_dict(cur, result) + await async_write_event(cur, key, SentinelEvent.EXECUTING, owner_id=owner_id, fencing_token=fencing_token) await conn.commit() if row is None: @@ -118,6 +122,8 @@ async def release(conn, key, *, owner_id, fencing_token): """, (key, owner_id, fencing_token)) success = await cur.fetchone() is not None + if success: + await async_write_event(cur, key, SentinelEvent.RELEASED, owner_id=owner_id, fencing_token=fencing_token) await conn.commit() return OperationResult(success) @@ -144,7 +150,8 @@ async def complete(conn, key, *, owner_id, fencing_token, execution_result=None) """, (serialized_result, key, owner_id, fencing_token)) success = await cur.fetchone() is not None - + if success: + await async_write_event(cur, key, SentinelEvent.COMPLETED, owner_id=owner_id, fencing_token=fencing_token) await conn.commit() return OperationResult(success) @@ -180,6 +187,8 @@ async def expire_lease(conn, key, *, owner_id, fencing_token): RETURNING 1; """, (key, owner_id, fencing_token)) success = await cur.fetchone() is not None + if success: + await async_write_event(cur, key, SentinelEvent.EXPIRED, owner_id=owner_id, fencing_token=fencing_token) await conn.commit() return OperationResult(success) diff --git a/sentinel-py/sentinel/async_reconcilliation.py b/sentinel-py/sentinel/async_reconcilliation.py index 4971c40..8c2f902 100644 --- a/sentinel-py/sentinel/async_reconcilliation.py +++ b/sentinel-py/sentinel/async_reconcilliation.py @@ -1,4 +1,5 @@ from sentinel.result import OperationResult +from .events import SentinelEvent, async_write_event class AsyncReconcile: @@ -15,7 +16,8 @@ async def reconcile(self, key): UPDATE sentinel_leases SET status = 'reconciling', - lease_updated_at = NOW() + lease_updated_at = NOW(), + fencing_token = nextval('sentinel_token_seq') WHERE key = %s AND status = 'executing' AND lease_expires_at < NOW() @@ -23,7 +25,8 @@ async def reconcile(self, key): """, (key,)) success = await cur.fetchone() is not None - + if success: + await async_write_event(cur, key, SentinelEvent.RECONCILING) await conn.commit() return OperationResult(success) @@ -48,7 +51,8 @@ async def force_complete(self, key, execution_result): """, (execution_result, key)) success = await cur.fetchone() is not None - + if success: + await async_write_event(cur, key, SentinelEvent.COMPLETED) await conn.commit() return OperationResult(success) @@ -72,7 +76,8 @@ async def reset(self, key): """, (key,)) success = await cur.fetchone() is not None - + if success: + await async_write_event(cur, key, SentinelEvent.RESET) await conn.commit() return OperationResult(success) diff --git a/sentinel-py/sentinel/cli.py b/sentinel-py/sentinel/cli.py index 86f63da..b61fa4c 100644 --- a/sentinel-py/sentinel/cli.py +++ b/sentinel-py/sentinel/cli.py @@ -9,6 +9,7 @@ pass from .core import inspect +from .events import history def get_conn(): db_url = os.environ.get("DATABASE_URL") @@ -26,11 +27,15 @@ def main(): inspect_parser = subparsers.add_parser("inspect", help="Inspect a lease by key") inspect_parser.add_argument("key", type=str) + history_parser = subparsers.add_parser("history", help="Show execution history for a key") + history_parser.add_argument("key", type=str) + history_parser.add_argument("--limit", type=int, default=50, help="Max events to show (default: 50)") + args = parser.parse_args() if args.command == "inspect": - conn = get_conn() - result = inspect(conn, args.key) + with get_conn() as conn: + result = inspect(conn, args.key) if result is None: print(f"No lease found for key: {args.key}") @@ -43,6 +48,18 @@ def main(): print(f"lease_updated_at: {result.lease_updated_at}") print(f"hard_expires_at: {result.hard_expires_at}") print(f"execution_result: {result.execution_result}") + + elif args.command == "history": + with get_conn() as conn: + events = history(conn, args.key, limit=args.limit) + + if not events: + print(f"No history found for key: {args.key}") + else: + print(f"History for key: {args.key} ({len(events)} events)\n") + for e in events: + print(f" {e}") + else: parser.print_help() diff --git a/sentinel-py/sentinel/core.py b/sentinel-py/sentinel/core.py index deb625d..b66f4de 100644 --- a/sentinel-py/sentinel/core.py +++ b/sentinel-py/sentinel/core.py @@ -1,6 +1,7 @@ import json from .utils import get_owner_id, row_to_dict from .result import AcquireResult, OperationResult, InspectResult +from .events import SentinelEvent, write_event def acquire(conn, key, *, owner_id=None, ttl_ms=10000, hard_ttl_ms = None): @@ -47,6 +48,7 @@ def acquire(conn, key, *, owner_id=None, ttl_ms=10000, hard_ttl_ms = None): if result is not None: row = row_to_dict(cur, result) + write_event(cur, key, SentinelEvent.ACQUIRED, owner_id=row["owner_id"], fencing_token=row["fencing_token"]) conn.commit() @@ -74,6 +76,8 @@ def acquire(conn, key, *, owner_id=None, ttl_ms=10000, hard_ttl_ms = None): if result is not None: row = row_to_dict(cur, result) + write_event(cur, key, SentinelEvent.REJECTED, owner_id=row["owner_id"], fencing_token=row["fencing_token"]) + conn.commit() @@ -105,6 +109,8 @@ def start_execution(conn, key, *, owner_id, fencing_token): success = result is not None if result is not None: row = row_to_dict(cur, result) + write_event(cur, key, SentinelEvent.EXECUTING, owner_id=owner_id, fencing_token=fencing_token) + conn.commit() if row is None: @@ -121,6 +127,8 @@ def release(conn, key, *, owner_id, fencing_token): """, (key, owner_id, fencing_token)) success = cur.fetchone() is not None + if success: + write_event(cur, key, SentinelEvent.RELEASED, owner_id=owner_id, fencing_token=fencing_token) conn.commit() return OperationResult(success) @@ -147,6 +155,8 @@ def complete(conn, key, *, owner_id, fencing_token, execution_result=None): """, (serialized_result, key, owner_id, fencing_token)) success = cur.fetchone() is not None + if success: + write_event(cur, key, SentinelEvent.COMPLETED, owner_id=owner_id, fencing_token=fencing_token) conn.commit() return OperationResult(success) @@ -199,6 +209,9 @@ def expire_lease(conn, key, *, owner_id, fencing_token): RETURNING 1; """, (key, owner_id, fencing_token)) success = cur.fetchone() is not None + if success: + write_event(cur, key, SentinelEvent.EXPIRED, owner_id=owner_id, fencing_token=fencing_token) + conn.commit() return OperationResult(success) diff --git a/sentinel-py/sentinel/events.py b/sentinel-py/sentinel/events.py new file mode 100644 index 0000000..5ffc2cd --- /dev/null +++ b/sentinel-py/sentinel/events.py @@ -0,0 +1,73 @@ +from enum import Enum +from dataclasses import dataclass +from datetime import datetime +from typing import Optional, Any + +class SentinelEvent(Enum): + ACQUIRED = "acquired" + REJECTED = "rejected" + EXECUTING = "executing" + COMPLETED = "completed" + EXPIRED = "expired" + RECONCILING = "reconciling" + RELEASED = "released" + RESET = "reset" + +def write_event(cur, key, event, *, owner_id=None, fencing_token=None, metadata=None): + cur.execute(""" + INSERT INTO sentinel_events (key, event, owner_id, fencing_token, metadata) + VALUES (%s, %s, %s, %s, %s) + """, (key, event.value, owner_id, fencing_token, metadata)) + +async def async_write_event(cur, key, event, *, owner_id=None, fencing_token=None, metadata=None): + await cur.execute(""" + INSERT INTO sentinel_events (key, event, owner_id, fencing_token, metadata) + VALUES (%s, %s, %s, %s, %s) + """, (key, event.value, owner_id, fencing_token, metadata)) + +@dataclass +class EventRecord: + id: int + key: str + event: str + owner_id: Optional[str] + fencing_token: Optional[int] + metadata: Optional[Any] + occurred_at: datetime + + def __str__(self): + parts = [ + self.occurred_at.strftime('%Y-%m-%d %H:%M:%S'), + f"{self.event:<12}", + f"token={self.fencing_token}", + f"owner={self.owner_id}", + ] + if self.metadata: + parts.append(f"meta={self.metadata}") + return " ".join(parts) + + +def history(conn, key, *, limit=50): + with conn.cursor() as cur: + cur.execute(""" + SELECT id, key, event, owner_id, fencing_token, metadata, occurred_at + FROM sentinel_events + WHERE key = %s + ORDER BY occurred_at ASC, id ASC + LIMIT %s + """, (key, limit)) + + rows = cur.fetchall() + + return [ + EventRecord( + id=row[0], + key=row[1], + event=row[2], + owner_id=row[3], + fencing_token=row[4], + metadata=row[5], + occurred_at=row[6], + ) + for row in rows + ] \ No newline at end of file diff --git a/sentinel-py/sentinel/integrations/migrations/0003_sentinel_events.py b/sentinel-py/sentinel/integrations/migrations/0003_sentinel_events.py new file mode 100644 index 0000000..38ecb03 --- /dev/null +++ b/sentinel-py/sentinel/integrations/migrations/0003_sentinel_events.py @@ -0,0 +1,76 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("integrations", "0002_add_ttl_ms"), + ] + + operations = [ + migrations.CreateModel( + name="SentinelEvent", + fields=[ + ( + "id", + models.BigAutoField( + primary_key=True, + serialize=False, + ), + ), + ("key", models.TextField()), + ( + "event", + models.CharField( + max_length=32, + choices=[ + ("acquired", "Acquired"), + ("rejected", "Rejected"), + ("executing", "Executing"), + ("completed", "Completed"), + ("expired", "Expired"), + ("reconciling", "Reconciling"), + ], + ), + ), + ("occurred_at", models.DateTimeField()), + ( + "fencing_token", + models.BigIntegerField( + blank=True, + null=True, + ), + ), + ( + "owner_id", + models.TextField( + blank=True, + null=True, + ), + ), + ], + options={ + "db_table": "sentinel_events", + "ordering": ["occurred_at", "id"], + }, + ), + migrations.AddIndex( + model_name="sentinelevent", + index=models.Index( + fields=["key", "occurred_at", "fencing_token"], + name="idx_sentinel_events", + ), + ), + migrations.RunSQL( + """ + ALTER TABLE sentinel_events + ALTER COLUMN occurred_at + SET DEFAULT now(); + """, + reverse_sql=""" + ALTER TABLE sentinel_events + ALTER COLUMN occurred_at + DROP DEFAULT; + """, + ), + ] \ No newline at end of file diff --git a/sentinel-py/sentinel/integrations/models.py b/sentinel-py/sentinel/integrations/models.py index 1258f89..be83e73 100644 --- a/sentinel-py/sentinel/integrations/models.py +++ b/sentinel-py/sentinel/integrations/models.py @@ -52,4 +52,49 @@ class Meta: fields=["lease_expires_at"], name="idx_sentinel_expiry" ) - ] \ No newline at end of file + ] + +class SentinelEvent(models.Model): + class Event(models.TextChoices): + ACQUIRED = "acquired", "Acquired" + REJECTED = "rejected", "Rejected" + EXECUTING = "executing", "Executing" + COMPLETED = "completed", "Completed" + EXPIRED = "expired", "Expired" + RECONCILING = "reconciling", "Reconciling" + + id = models.BigAutoField(primary_key=True) + + key = models.TextField() + + event = models.CharField( + max_length=32, + choices=Event.choices, + ) + + occurred_at = models.DateTimeField() + + fencing_token = models.BigIntegerField( + null=True, + blank=True, + ) + + owner_id = models.TextField( + null=True, + blank=True, + ) + + class Meta: + db_table = "sentinel_events" + + indexes = [ + models.Index( + fields=["key", "occurred_at", "fencing_token"], + name="idx_sentinel_events", + ) + ] + + ordering = ["occurred_at", "id"] + + def __str__(self): + return f"{self.key} [{self.event}]" \ No newline at end of file diff --git a/sentinel-py/sentinel/reconciliation.py b/sentinel-py/sentinel/reconciliation.py index 4c3a076..8aa2524 100644 --- a/sentinel-py/sentinel/reconciliation.py +++ b/sentinel-py/sentinel/reconciliation.py @@ -1,4 +1,5 @@ from sentinel.result import OperationResult +from .events import SentinelEvent, write_event class Reconcile: @@ -15,7 +16,8 @@ def reconcile(self, key): UPDATE sentinel_leases SET status = 'reconciling', - lease_updated_at = NOW() + lease_updated_at = NOW(), + fencing_token = nextval('sentinel_token_seq') WHERE key = %s AND status = 'executing' AND lease_expires_at < NOW() @@ -23,6 +25,8 @@ def reconcile(self, key): """, (key,)) success = cur.fetchone() is not None + if success: + write_event(cur, key, SentinelEvent.RECONCILING) conn.commit() @@ -48,6 +52,8 @@ def force_complete(self, key, execution_result): """, (execution_result, key)) success = cur.fetchone() is not None + if success: + write_event(cur, key, SentinelEvent.COMPLETED) conn.commit() @@ -72,7 +78,8 @@ def reset(self, key): """, (key,)) success = cur.fetchone() is not None - + if success: + write_event(cur, key, SentinelEvent.RESET) conn.commit() return OperationResult(success) diff --git a/sentinel-py/sentinel/schema.py b/sentinel-py/sentinel/schema.py index 11de326..1f9f11c 100644 --- a/sentinel-py/sentinel/schema.py +++ b/sentinel-py/sentinel/schema.py @@ -4,9 +4,9 @@ CREATE TABLE IF NOT EXISTS sentinel_leases ( key TEXT PRIMARY KEY, owner_id TEXT NOT NULL, - lease_expires_at TIMESTAMP NOT NULL, - lease_updated_at TIMESTAMP, - hard_expires_at TIMESTAMP, + lease_expires_at TIMESTAMPTZ NOT NULL, + lease_updated_at TIMESTAMPTZ, + hard_expires_at TIMESTAMPTZ, execution_result JSONB, status TEXT NOT NULL DEFAULT 'claimed' CHECK (status IN ('claimed','executing','completed','reconciling')), fencing_token BIGINT NOT NULL DEFAULT 1, @@ -15,4 +15,17 @@ CREATE INDEX IF NOT EXISTS idx_sentinel_expiry ON sentinel_leases (lease_expires_at); + +CREATE TABLE IF NOT EXISTS sentinel_events ( + id BIGSERIAL PRIMARY KEY, + key TEXT NOT NULL, + event TEXT NOT NULL CHECK(event IN('acquired','rejected','executing','completed','expired','reconciling','released','reset')), + occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(), + fencing_token BIGINT, + owner_id TEXT, + metadata JSONB +); + +CREATE INDEX IF NOT EXISTS idx_sentinel_events + ON sentinel_events (key, occurred_at, fencing_token); """ \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py index 4f6b377..47c56ab 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ import pytest import asyncio +import pytest_asyncio import sys import psycopg from sentinel import init_db @@ -16,6 +17,7 @@ def clean_db(conn): with conn.cursor() as cur: cur.execute("DELETE FROM sentinel_leases;") cur.execute("ALTER SEQUENCE sentinel_token_seq RESTART WITH 1;") + cur.execute("DELETE FROM sentinel_events;") conn.commit() @@ -73,4 +75,10 @@ def clean_db_(): with conn.cursor() as cur: cur.execute("DELETE FROM sentinel_leases") conn.commit() - conn.close() \ No newline at end of file + conn.close() + +@pytest_asyncio.fixture +async def aconn(): + conn = await psycopg.AsyncConnection.connect(DSN) + yield conn + await conn.close() \ No newline at end of file diff --git a/tests/test_async_core.py b/tests/test_async_core.py index 669c0ee..fad891b 100644 --- a/tests/test_async_core.py +++ b/tests/test_async_core.py @@ -13,13 +13,6 @@ inspect, ) -DSN = "postgresql://sentinel_test:sentinel_test@localhost/sentinel_test" - -@pytest_asyncio.fixture -async def aconn(): - conn = await psycopg.AsyncConnection.connect(DSN) - yield conn - await conn.close() # ─── ACQUIRE ──────────────────────────────────────────────────────────────── diff --git a/tests/test_async_history.py b/tests/test_async_history.py new file mode 100644 index 0000000..35dcfc9 --- /dev/null +++ b/tests/test_async_history.py @@ -0,0 +1,56 @@ +import pytest +import pytest_asyncio +import asyncio +import psycopg +from sentinel.async_core import acquire, start_execution, complete +from sentinel.events import history, SentinelEvent + +DSN = "postgresql://sentinel_test:sentinel_test@localhost/sentinel_test" + +# ─── HISTORY ──────────────────────────────────────────────────────────────── + +from sentinel.events import history, SentinelEvent + +@pytest.mark.asyncio +async def test_async_history_happy_path(aconn, conn): + key = "async:history:happy" + r = await acquire(aconn, key, ttl_ms=5000, hard_ttl_ms=10000) + await start_execution(aconn, key, owner_id=r.owner_id, fencing_token=r.fencing_token) + await complete(aconn, key, owner_id=r.owner_id, fencing_token=r.fencing_token, execution_result={"ok": True}) + + events = history(conn, key) + event_types = [e.event for e in events] + + assert event_types == [ + SentinelEvent.ACQUIRED.value, + SentinelEvent.EXECUTING.value, + SentinelEvent.COMPLETED.value, + ] + assert all(e.fencing_token == r.fencing_token for e in events) + +@pytest.mark.asyncio +async def test_async_history_reconciliation_path(aconn, conn): + from sentinel.async_reconcilliation import AsyncReconcile + + key = "async:history:reconcile" + r = await acquire(aconn, key, ttl_ms=100, hard_ttl_ms=200) + await start_execution(aconn, key, owner_id=r.owner_id, fencing_token=r.fencing_token) + await asyncio.sleep(0.25) + + async def get_conn(): + return await psycopg.AsyncConnection.connect(DSN) + + reconciler = AsyncReconcile(get_conn) + await reconciler.reconcile(key) + await reconciler.reset(key) + + events = history(conn, key) + event_types = [e.event for e in events] + + assert SentinelEvent.ACQUIRED.value in event_types + assert SentinelEvent.EXECUTING.value in event_types + assert SentinelEvent.RECONCILING.value in event_types + assert SentinelEvent.RESET.value in event_types + + reconciling_token = next(e.fencing_token for e in events if e.event == SentinelEvent.RECONCILING.value) + assert reconciling_token != r.fencing_token \ No newline at end of file diff --git a/tests/test_async_reconciliation.py b/tests/test_async_reconciliation.py index 16f4dd9..913c364 100644 --- a/tests/test_async_reconciliation.py +++ b/tests/test_async_reconciliation.py @@ -10,12 +10,6 @@ async def get_async_conn(): return await psycopg.AsyncConnection.connect(DSN) -@pytest_asyncio.fixture -async def aconn(): - conn = await psycopg.AsyncConnection.connect(DSN) - yield conn - await conn.close() - @pytest_asyncio.fixture async def areconcile(): return AsyncReconcile(get_conn=get_async_conn) diff --git a/tests/test_history.py b/tests/test_history.py new file mode 100644 index 0000000..8576684 --- /dev/null +++ b/tests/test_history.py @@ -0,0 +1,46 @@ +import time +import pytest +import psycopg +from sentinel.core import acquire, start_execution, complete +from sentinel.events import history, SentinelEvent + +DSN = "postgresql://sentinel_test:sentinel_test@localhost/sentinel_test" + +def test_sync_history_happy_path(conn): + key = "sync:history:happy" + r = acquire(conn, key, ttl_ms=5000, hard_ttl_ms=10000) + start_execution(conn, key, owner_id=r.owner_id, fencing_token=r.fencing_token) + complete(conn, key, owner_id=r.owner_id, fencing_token=r.fencing_token, execution_result={"ok": True}) + + events = history(conn, key) + event_types = [e.event for e in events] + + assert event_types == [ + SentinelEvent.ACQUIRED.value, + SentinelEvent.EXECUTING.value, + SentinelEvent.COMPLETED.value, + ] + assert all(e.fencing_token == r.fencing_token for e in events) + +def test_sync_history_reconciliation_path(conn): + from sentinel.reconciliation import Reconcile + + key = "sync:history:reconcile" + r = acquire(conn, key, ttl_ms=100, hard_ttl_ms=200) + start_execution(conn, key, owner_id=r.owner_id, fencing_token=r.fencing_token) + time.sleep(0.25) + + reconciler = Reconcile(lambda: psycopg.connect(DSN)) + reconciler.reconcile(key) + reconciler.reset(key) + + events = history(conn, key) + event_types = [e.event for e in events] + + assert SentinelEvent.ACQUIRED.value in event_types + assert SentinelEvent.EXECUTING.value in event_types + assert SentinelEvent.RECONCILING.value in event_types + assert SentinelEvent.RESET.value in event_types + + reconciling_token = next(e.fencing_token for e in events if e.event == SentinelEvent.RECONCILING.value) + assert reconciling_token != r.fencing_token \ No newline at end of file