Merge pull request #25 from Sreejay-Reddy/feat/event-log-0.5.0

feat: add sentinel_events append-only schema and TIMESTAMPTZ migration

Author: Sreejay-Reddy

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 <key>`** — 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

README.md

@@ -63,6 +63,7 @@ Sentinel ships with `sen`, a command-line tool for inspecting lease state direct
 
 ```bash
 sen inspect <key>
+sen history <key> --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 <key>
+sen history <key> --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

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

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"]

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)
 

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,15 +16,17 @@ 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()
                 RETURNING 1;
                 """, (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)

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()
 

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)