Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
27 changes: 26 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -102,6 +103,8 @@ result = sentinel.once(
)
```

---

### Reading the result

```python
Expand All @@ -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`:
Expand Down Expand Up @@ -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
Expand Down
8 changes: 4 additions & 4 deletions ROADMAP.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

---

Expand All @@ -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
Expand Down
10 changes: 6 additions & 4 deletions sentinel-py/pyproject.toml
Original file line number Diff line number Diff line change
@@ -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",
Expand All @@ -16,7 +16,10 @@ keywords = [
"execution-coordination",
"fault-tolerance",
"concurrency",
"fencing-tokens"
"fencing-tokens",
"append-only log",
"execution history",
"audit log"
]

classifiers = [
Expand All @@ -40,7 +43,6 @@ sen = "sentinel.cli:main"

[project.optional-dependencies]
django = ["django>=4.2"]
cli = ["python-dotenv"]

[build-system]
requires = ["setuptools>=61.0"]
Expand Down
11 changes: 10 additions & 1 deletion sentinel-py/sentinel/async_core.py
Original file line number Diff line number Diff line change
@@ -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):

Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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:
Expand All @@ -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)
Expand All @@ -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)

Expand Down Expand Up @@ -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)

Expand Down
13 changes: 9 additions & 4 deletions sentinel-py/sentinel/async_reconcilliation.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from sentinel.result import OperationResult
from .events import SentinelEvent, async_write_event


class AsyncReconcile:
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand Down
21 changes: 19 additions & 2 deletions sentinel-py/sentinel/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
pass

from .core import inspect
from .events import history

def get_conn():
db_url = os.environ.get("DATABASE_URL")
Expand All @@ -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}")
Expand All @@ -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()

Expand Down
13 changes: 13 additions & 0 deletions sentinel-py/sentinel/core.py
Original file line number Diff line number Diff line change
@@ -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):

Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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:
Expand All @@ -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)
Expand All @@ -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)
Expand Down Expand Up @@ -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)

Expand Down
Loading
Loading