From d72c05463552ffbd8ec8b31f199f28f3d41835f5 Mon Sep 17 00:00:00 2001 From: summer97souls Date: Sat, 20 Jun 2026 00:19:22 +0530 Subject: [PATCH] feat: add inspect API, sentinel CLI, and decouple reconciliation from once() - add sync and async inspect APIs - add inspect test suites for sync and async core implementations - add Sentinel CLI - remove reconciliation from once() and expose it as a standalone client API - simplify reconciliation API by removing owner_id and fencing_token requirements - refactor reconciliation flow and public API boundaries --- sentinel-py/pyproject.toml | 4 + sentinel-py/sentinel/async_core.py | 28 ++++- sentinel-py/sentinel/async_once.py | 11 +- sentinel-py/sentinel/async_reconcilliation.py | 40 ++----- sentinel-py/sentinel/async_sentinel.py | 11 +- sentinel-py/sentinel/cli.py | 50 +++++++++ sentinel-py/sentinel/core.py | 28 ++++- sentinel-py/sentinel/once.py | 11 +- sentinel-py/sentinel/reconciliation.py | 40 ++----- sentinel-py/sentinel/result.py | 18 +++- sentinel-py/sentinel/sentinel.py | 11 +- tests/test_async_core.py | 101 +++++++++++++++++- tests/test_async_once.py | 35 ------ tests/test_async_reconciliation.py | 24 ++--- tests/test_core.py | 82 ++++++++++++++ tests/test_once.py | 43 -------- tests/test_reconciliation.py | 24 ++--- 17 files changed, 342 insertions(+), 219 deletions(-) create mode 100644 sentinel-py/sentinel/cli.py diff --git a/sentinel-py/pyproject.toml b/sentinel-py/pyproject.toml index a5a01e7..6cb2db8 100644 --- a/sentinel-py/pyproject.toml +++ b/sentinel-py/pyproject.toml @@ -35,8 +35,12 @@ classifiers = [ license = { file = "LICENSE" } +[project.scripts] +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 51fd047..a867a1d 100644 --- a/sentinel-py/sentinel/async_core.py +++ b/sentinel-py/sentinel/async_core.py @@ -1,6 +1,6 @@ import json from .utils import get_owner_id, row_to_dict -from .result import AcquireResult, OperationResult +from .result import AcquireResult, OperationResult, InspectResult async def acquire(conn, key, *, owner_id=None, ttl_ms=10000, hard_ttl_ms = None): @@ -182,4 +182,30 @@ async def expire_lease(conn, key, *, owner_id, fencing_token): await conn.commit() return OperationResult(success) +async def inspect(conn, key): + async with conn.cursor() as cur: + await cur.execute(""" + SELECT owner_id, lease_expires_at, fencing_token, status, + lease_expires_at > NOW() AS lease_alive, lease_updated_at, + hard_expires_at, execution_result + FROM sentinel_leases + WHERE key = %s + """, (key,)) + + result = await cur.fetchone() + if result is None: + return None + + row = row_to_dict(cur, result) + + return InspectResult( + key = key, + owner_id = row["owner_id"], + fencing_token = row["fencing_token"], + status = row["status"], + lease_alive = row["lease_alive"], + lease_expires_at = row["lease_expires_at"], + lease_updated_at = row["lease_updated_at"], + hard_expires_at = row["hard_expires_at"], + execution_result = row["execution_result"]) diff --git a/sentinel-py/sentinel/async_once.py b/sentinel-py/sentinel/async_once.py index 74d838a..a80be2c 100644 --- a/sentinel-py/sentinel/async_once.py +++ b/sentinel-py/sentinel/async_once.py @@ -13,7 +13,6 @@ from .heartbeat_config import get_manager from .logging import logger -from .async_reconcilliation import AsyncReconcile from .result import OnceResult @@ -36,7 +35,6 @@ def __init__( self.kwargs = kwargs or {} self.owns_connection = owns_connection - self.reconcile = AsyncReconcile(get_conn) self._task = None async def run(self): @@ -76,8 +74,7 @@ async def run(self): success=False, status="executing", uncertain=True, - execution_alive=False, - reconcile=self.reconcile + execution_alive=False ) if acquired.status == "executing" and acquired.lease_alive: @@ -162,8 +159,7 @@ async def run(self): status="executing", execution_alive=False, uncertain=True, - exception=e, - reconcile=self.reconcile + exception=e ) # Finalize canonical completion @@ -184,8 +180,7 @@ async def run(self): return OnceResult( success=False, status=completed.status, - uncertain=True, - reconcile=self.reconcile + uncertain=True ) return OnceResult( diff --git a/sentinel-py/sentinel/async_reconcilliation.py b/sentinel-py/sentinel/async_reconcilliation.py index f15549b..4971c40 100644 --- a/sentinel-py/sentinel/async_reconcilliation.py +++ b/sentinel-py/sentinel/async_reconcilliation.py @@ -6,7 +6,7 @@ def __init__(self, get_conn, namespace=None): self.get_conn = get_conn self.namespace = namespace - async def reconcile(self, key, *, owner_id, fencing_token): + async def reconcile(self, key): conn = await self.get_conn() try: @@ -17,12 +17,10 @@ async def reconcile(self, key, *, owner_id, fencing_token): status = 'reconciling', lease_updated_at = NOW() WHERE key = %s - AND owner_id = %s - AND fencing_token = %s AND status = 'executing' AND lease_expires_at < NOW() RETURNING 1; - """, (key, owner_id, fencing_token)) + """, (key,)) success = await cur.fetchone() is not None @@ -33,14 +31,7 @@ async def reconcile(self, key, *, owner_id, fencing_token): finally: await conn.close() - async def force_complete( - self, - key, - *, - owner_id, - fencing_token, - execution_result - ): + async def force_complete(self, key, execution_result): conn = await self.get_conn() try: @@ -52,16 +43,9 @@ async def force_complete( execution_result = %s, lease_updated_at = NOW() WHERE key = %s - AND owner_id = %s - AND fencing_token = %s AND status = 'reconciling' RETURNING 1; - """, ( - execution_result, - key, - owner_id, - fencing_token - )) + """, (execution_result, key)) success = await cur.fetchone() is not None @@ -72,13 +56,7 @@ async def force_complete( finally: await conn.close() - async def reset( - self, - key, - *, - owner_id, - fencing_token - ): + async def reset(self, key): conn = await self.get_conn() try: @@ -89,15 +67,9 @@ async def reset( status = 'claimed', lease_updated_at = NOW() WHERE key = %s - AND owner_id = %s - AND fencing_token = %s AND status = 'reconciling' RETURNING 1; - """, ( - key, - owner_id, - fencing_token - )) + """, (key,)) success = await cur.fetchone() is not None diff --git a/sentinel-py/sentinel/async_sentinel.py b/sentinel-py/sentinel/async_sentinel.py index 98100cb..36910f6 100644 --- a/sentinel-py/sentinel/async_sentinel.py +++ b/sentinel-py/sentinel/async_sentinel.py @@ -1,4 +1,4 @@ -from .lease import Lease +from .async_reconcilliation import AsyncReconcile from .async_once import AsyncOnce from .heartbeat_config import get_manager @@ -16,6 +16,8 @@ def __init__(self, get_conn = None, default_ttl_ms=3000, namespace=None, owns_co raise ValueError( "No database connection provider found." ) + + self.reconcile = AsyncReconcile(self._conn, namespace=self.namespace) self.manager = get_manager( self._conn, @@ -35,13 +37,6 @@ def _hard_ttl(self, ttl_ms, hard_ttl_ms): def _key(self, key): return f"{self.namespace}:{key}" if self.namespace else key - - # def lease(self, key, ttl_ms=None, hard_ttl_ms=None): - # key = self._key(key) - # ttl = self._ttl(ttl_ms) - # hard_ttl = self._hard_ttl(ttl,hard_ttl_ms) - - # return Lease(None, key, ttl, hard_ttl, self._conn) async def once(self, key, fn, ttl_ms=None, hard_ttl_ms=None, kwargs=None): key = self._key(key) diff --git a/sentinel-py/sentinel/cli.py b/sentinel-py/sentinel/cli.py new file mode 100644 index 0000000..f494c88 --- /dev/null +++ b/sentinel-py/sentinel/cli.py @@ -0,0 +1,50 @@ +import os +import argparse +import psycopg + +try: + from dotenv import load_dotenv + load_dotenv() +except ImportError: + pass + +from .core import inspect + +def get_conn(): + db_url = os.environ.get("DATABASE_URL") + if not db_url: + raise EnvironmentError( + "No DATABASE_URL found. Set it in your environment or .env file.\n" + "Example: DATABASE_URL=postgresql://user:pass@localhost/mydb" + ) + return psycopg.connect(db_url) + +def main(): + parser = argparse.ArgumentParser(prog="sen", description="Sentinel CLI") + subparsers = parser.add_subparsers(dest="command") + + inspect_parser = subparsers.add_parser("inspect", help="Inspect a lease by key") + inspect_parser.add_argument("key", type=str) + + args = parser.parse_args() + + if args.command == "inspect": + conn = get_conn() + result = inspect(conn, args.key) + + if result is None: + print(f"No lease found for key: {args.key}") + else: + print(f"status: {result.status}") + print(f"lease_alive: {result.lease_alive}") + print(f"owner_id: {result.owner_id}") + print(f"fencing_token: {result.fencing_token}") + print(f"lease_expires_at: {result.lease_expires_at}") + 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}") + else: + parser.print_help() + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/sentinel-py/sentinel/core.py b/sentinel-py/sentinel/core.py index 814f213..818c903 100644 --- a/sentinel-py/sentinel/core.py +++ b/sentinel-py/sentinel/core.py @@ -1,6 +1,6 @@ import json from .utils import get_owner_id, row_to_dict -from .result import AcquireResult, OperationResult +from .result import AcquireResult, OperationResult, InspectResult def acquire(conn, key, *, owner_id=None, ttl_ms=10000, hard_ttl_ms = None): @@ -182,4 +182,30 @@ def expire_lease(conn, key, *, owner_id, fencing_token): conn.commit() return OperationResult(success) +def inspect(conn, key): + with conn.cursor() as cur: + cur.execute(""" + SELECT owner_id, lease_expires_at, fencing_token, status, + lease_expires_at > NOW() AS lease_alive, lease_updated_at, + hard_expires_at, execution_result + FROM sentinel_leases + WHERE key = %s + """, (key,)) + + result = cur.fetchone() + if result is None: + return None + + row = row_to_dict(cur, result) + + return InspectResult( + key = key, + owner_id = row["owner_id"], + fencing_token = row["fencing_token"], + status = row["status"], + lease_alive = row["lease_alive"], + lease_expires_at = row["lease_expires_at"], + lease_updated_at = row["lease_updated_at"], + hard_expires_at = row["hard_expires_at"], + execution_result = row["execution_result"]) diff --git a/sentinel-py/sentinel/once.py b/sentinel-py/sentinel/once.py index 24b73ac..d8efb60 100644 --- a/sentinel-py/sentinel/once.py +++ b/sentinel-py/sentinel/once.py @@ -13,7 +13,6 @@ from .heartbeat_config import get_manager from .logging import logger -from .reconciliation import Reconcile from .result import OnceResult @@ -36,7 +35,6 @@ def __init__( self.kwargs = kwargs or {} self.owns_connection = owns_connection - self.reconcile = Reconcile(get_conn) self._task = None def run(self): @@ -76,8 +74,7 @@ def run(self): success=False, status="executing", uncertain=True, - execution_alive=False, - reconcile=self.reconcile + execution_alive=False ) if acquired.status == "executing" and acquired.lease_alive: @@ -162,8 +159,7 @@ def run(self): status="executing", execution_alive=False, uncertain=True, - exception=e, - reconcile=self.reconcile + exception=e ) # Finalize canonical completion @@ -184,8 +180,7 @@ def run(self): return OnceResult( success=False, status=completed.status, - uncertain=True, - reconcile=self.reconcile + uncertain=True ) return OnceResult( diff --git a/sentinel-py/sentinel/reconciliation.py b/sentinel-py/sentinel/reconciliation.py index 7e4f51c..4c3a076 100644 --- a/sentinel-py/sentinel/reconciliation.py +++ b/sentinel-py/sentinel/reconciliation.py @@ -6,7 +6,7 @@ def __init__(self, get_conn, namespace=None): self.get_conn = get_conn self.namespace = namespace - def reconcile(self, key, *, owner_id, fencing_token): + def reconcile(self, key): conn = self.get_conn() try: @@ -17,12 +17,10 @@ def reconcile(self, key, *, owner_id, fencing_token): status = 'reconciling', lease_updated_at = NOW() WHERE key = %s - AND owner_id = %s - AND fencing_token = %s AND status = 'executing' AND lease_expires_at < NOW() RETURNING 1; - """, (key, owner_id, fencing_token)) + """, (key,)) success = cur.fetchone() is not None @@ -33,14 +31,7 @@ def reconcile(self, key, *, owner_id, fencing_token): finally: conn.close() - def force_complete( - self, - key, - *, - owner_id, - fencing_token, - execution_result - ): + def force_complete(self, key, execution_result): conn = self.get_conn() try: @@ -52,16 +43,9 @@ def force_complete( execution_result = %s, lease_updated_at = NOW() WHERE key = %s - AND owner_id = %s - AND fencing_token = %s AND status = 'reconciling' RETURNING 1; - """, ( - execution_result, - key, - owner_id, - fencing_token - )) + """, (execution_result, key)) success = cur.fetchone() is not None @@ -72,13 +56,7 @@ def force_complete( finally: conn.close() - def reset( - self, - key, - *, - owner_id, - fencing_token - ): + def reset(self, key): conn = self.get_conn() try: @@ -89,15 +67,9 @@ def reset( status = 'claimed', lease_updated_at = NOW() WHERE key = %s - AND owner_id = %s - AND fencing_token = %s AND status = 'reconciling' RETURNING 1; - """, ( - key, - owner_id, - fencing_token - )) + """, (key,)) success = cur.fetchone() is not None diff --git a/sentinel-py/sentinel/result.py b/sentinel-py/sentinel/result.py index 37504e1..50f53f7 100644 --- a/sentinel-py/sentinel/result.py +++ b/sentinel-py/sentinel/result.py @@ -13,6 +13,22 @@ def __init__(self, success: bool, status=None): self.success = success self.status = status +class InspectResult: + def __init__( + self, key, owner_id, fencing_token, status, lease_alive, + lease_expires_at, lease_updated_at, hard_expires_at, execution_result=None + ): + self.key = key + self.owner_id = owner_id + self.fencing_token = fencing_token + self.status = status + self.lease_alive = lease_alive + self.lease_expires_at = lease_expires_at + self.lease_updated_at = lease_updated_at + self.hard_expires_at = hard_expires_at + self.execution_result = execution_result + + class OnceResult: def __init__( self, @@ -21,7 +37,6 @@ def __init__( response=None, cached=False, exception=None, - reconcile=None, execution_alive=None, uncertain=False ): @@ -30,6 +45,5 @@ def __init__( self.response = response self.cached = cached self.exception = exception - self.reconcile = reconcile self.execution_alive = execution_alive self.uncertain = uncertain \ No newline at end of file diff --git a/sentinel-py/sentinel/sentinel.py b/sentinel-py/sentinel/sentinel.py index 3fb7bbb..2e1b2e6 100644 --- a/sentinel-py/sentinel/sentinel.py +++ b/sentinel-py/sentinel/sentinel.py @@ -1,4 +1,4 @@ -from .lease import Lease +from .reconciliation import Reconcile from .once import Once from .heartbeat_config import get_manager @@ -16,6 +16,8 @@ def __init__(self, get_conn = None, default_ttl_ms=3000, namespace=None, owns_co raise ValueError( "No database connection provider found." ) + + self.reconcile = Reconcile(self._conn, namespace=self.namespace) self.manager = get_manager( self._conn, @@ -35,13 +37,6 @@ def _hard_ttl(self, ttl_ms, hard_ttl_ms): def _key(self, key): return f"{self.namespace}:{key}" if self.namespace else key - - # def lease(self, key, ttl_ms=None, hard_ttl_ms=None): - # key = self._key(key) - # ttl = self._ttl(ttl_ms) - # hard_ttl = self._hard_ttl(ttl,hard_ttl_ms) - - # return Lease(None, key, ttl, hard_ttl, self._conn) def once(self, key, fn, ttl_ms=None, hard_ttl_ms=None, kwargs=None): key = self._key(key) diff --git a/tests/test_async_core.py b/tests/test_async_core.py index 4380d72..669c0ee 100644 --- a/tests/test_async_core.py +++ b/tests/test_async_core.py @@ -1,6 +1,7 @@ import time import pytest import pytest_asyncio +import asyncio import psycopg from sentinel.async_core import ( acquire, @@ -9,6 +10,7 @@ complete, expire_lease, release, + inspect, ) DSN = "postgresql://sentinel_test:sentinel_test@localhost/sentinel_test" @@ -146,4 +148,101 @@ async def test_async_release_allows_reacquire(aconn): r = await acquire(aconn, "async:core:release:basic", ttl_ms=5000, hard_ttl_ms=10000) await release(aconn, "async:core:release:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) r2 = await acquire(aconn, "async:core:release:basic", ttl_ms=5000, hard_ttl_ms=10000) - assert r2.acquired is True \ No newline at end of file + assert r2.acquired is True + + +# ─── INSPECT ──────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_async_inspect_missing_key(aconn): + result = await inspect( + aconn, + "async:inspect:missing", + ) + + assert result is None + + +@pytest.mark.asyncio +async def test_async_inspect_executing(aconn): + r = await acquire( + aconn, + "async:inspect:executing", + ttl_ms=5000, + hard_ttl_ms=10000, + ) + + await start_execution( + aconn, + "async:inspect:executing", + owner_id=r.owner_id, + fencing_token=r.fencing_token, + ) + + result = await inspect( + aconn, + "async:inspect:executing", + ) + + assert result.key == "async:inspect:executing" + assert result.status == "executing" + assert result.lease_alive is True + + +@pytest.mark.asyncio +async def test_async_inspect_completed(aconn): + r = await acquire( + aconn, + "async:inspect:completed", + ttl_ms=5000, + hard_ttl_ms=10000, + ) + + await start_execution( + aconn, + "async:inspect:completed", + owner_id=r.owner_id, + fencing_token=r.fencing_token, + ) + + await complete( + aconn, + "async:inspect:completed", + execution_result={"ok": True}, + owner_id=r.owner_id, + fencing_token=r.fencing_token, + ) + + result = await inspect( + aconn, + "async:inspect:completed", + ) + + assert result.status == "completed" + assert result.execution_result == {"ok": True} + + +@pytest.mark.asyncio +async def test_async_inspect_expired_execution(aconn): + r = await acquire( + aconn, + "async:inspect:expired", + ttl_ms=100, + hard_ttl_ms=10000, + ) + + await start_execution( + aconn, + "async:inspect:expired", + owner_id=r.owner_id, + fencing_token=r.fencing_token, + ) + + await asyncio.sleep(0.2) + + result = await inspect( + aconn, + "async:inspect:expired", + ) + + assert result.lease_alive is False \ No newline at end of file diff --git a/tests/test_async_once.py b/tests/test_async_once.py index 9d393fc..56e7600 100644 --- a/tests/test_async_once.py +++ b/tests/test_async_once.py @@ -110,41 +110,6 @@ async def boom(): assert r2.uncertain is True assert r2.execution_alive is False -# ─── RECONCILE VIA ONCE ───────────────────────────────────────────────────── - -@pytest.mark.asyncio -async def test_async_once_uncertain_exposes_reconcile(asentinel): - async def boom(): - raise ValueError("intentional") - result = await asentinel.once("async:once:reconcile_exposed", fn=boom, ttl_ms=5000, hard_ttl_ms=10000) - assert result.uncertain is True - assert result.reconcile is not None - assert isinstance(result.reconcile, AsyncReconcile) - -@pytest.mark.asyncio -async def test_async_once_execution_alive_no_reconcile(): - ready = asyncio.Event() - results = {} - - async def slow_fn(): - ready.set() - await asyncio.sleep(0.5) - return {"done": True} - - async def run_first(): - s = AsyncSentinel(get_conn=get_async_conn) - results["first"] = await s.once("async:once:alive", fn=slow_fn, ttl_ms=5000, hard_ttl_ms=10000) - - async def run_second(): - await ready.wait() - await asyncio.sleep(0.05) - s = AsyncSentinel(get_conn=get_async_conn) - results["second"] = await s.once("async:once:alive", fn=slow_fn, ttl_ms=5000, hard_ttl_ms=10000) - - await asyncio.gather(run_first(), run_second()) - assert results["second"].execution_alive is True - assert results["second"].reconcile is None - # ─── NAMESPACE ────────────────────────────────────────────────────────────── @pytest.mark.asyncio diff --git a/tests/test_async_reconciliation.py b/tests/test_async_reconciliation.py index ac53341..16f4dd9 100644 --- a/tests/test_async_reconciliation.py +++ b/tests/test_async_reconciliation.py @@ -27,22 +27,14 @@ async def test_async_reconcile_marks_reconciling(aconn, areconcile): r = await acquire(aconn, "async:rec:reconcile:basic", ttl_ms=100, hard_ttl_ms=10000) await start_execution(aconn, "async:rec:reconcile:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) time.sleep(0.2) - result = await areconcile.reconcile("async:rec:reconcile:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) + result = await areconcile.reconcile("async:rec:reconcile:basic") assert result.success is True @pytest.mark.asyncio async def test_async_reconcile_fails_if_lease_still_alive(aconn, areconcile): r = await acquire(aconn, "async:rec:reconcile:alive", ttl_ms=5000, hard_ttl_ms=10000) await start_execution(aconn, "async:rec:reconcile:alive", owner_id=r.owner_id, fencing_token=r.fencing_token) - result = await areconcile.reconcile("async:rec:reconcile:alive", owner_id=r.owner_id, fencing_token=r.fencing_token) - assert result.success is False - -@pytest.mark.asyncio -async def test_async_reconcile_fails_wrong_token(aconn, areconcile): - r = await acquire(aconn, "async:rec:reconcile:bad_token", ttl_ms=100, hard_ttl_ms=10000) - await start_execution(aconn, "async:rec:reconcile:bad_token", owner_id=r.owner_id, fencing_token=r.fencing_token) - time.sleep(0.2) - result = await areconcile.reconcile("async:rec:reconcile:bad_token", owner_id=r.owner_id, fencing_token=r.fencing_token + 999) + result = await areconcile.reconcile("async:rec:reconcile:alive") assert result.success is False # ─── FORCE COMPLETE ───────────────────────────────────────────────────────── @@ -52,11 +44,9 @@ async def test_async_force_complete_from_reconciling(aconn, areconcile): r = await acquire(aconn, "async:rec:force:basic", ttl_ms=100, hard_ttl_ms=10000) await start_execution(aconn, "async:rec:force:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) time.sleep(0.2) - await areconcile.reconcile("async:rec:force:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) + await areconcile.reconcile("async:rec:force:basic") result = await areconcile.force_complete( "async:rec:force:basic", - owner_id=r.owner_id, - fencing_token=r.fencing_token, execution_result='{"value": 1}' ) assert result.success is True @@ -67,8 +57,6 @@ async def test_async_force_complete_fails_if_not_reconciling(aconn, areconcile): await start_execution(aconn, "async:rec:force:bad_status", owner_id=r.owner_id, fencing_token=r.fencing_token) result = await areconcile.force_complete( "async:rec:force:bad_status", - owner_id=r.owner_id, - fencing_token=r.fencing_token, execution_result='{"value": 1}' ) assert result.success is False @@ -80,13 +68,13 @@ async def test_async_reset_allows_rerun(aconn, areconcile): r = await acquire(aconn, "async:rec:reset:basic", ttl_ms=100, hard_ttl_ms=10000) await start_execution(aconn, "async:rec:reset:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) time.sleep(0.2) - await areconcile.reconcile("async:rec:reset:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) - result = await areconcile.reset("async:rec:reset:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) + await areconcile.reconcile("async:rec:reset:basic") + result = await areconcile.reset("async:rec:reset:basic") assert result.success is True @pytest.mark.asyncio async def test_async_reset_fails_if_not_reconciling(aconn, areconcile): r = await acquire(aconn, "async:rec:reset:bad_status", ttl_ms=5000, hard_ttl_ms=10000) await start_execution(aconn, "async:rec:reset:bad_status", owner_id=r.owner_id, fencing_token=r.fencing_token) - result = await areconcile.reset("async:rec:reset:bad_status", owner_id=r.owner_id, fencing_token=r.fencing_token) + result = await areconcile.reset("async:rec:reset:bad_status") assert result.success is False \ No newline at end of file diff --git a/tests/test_core.py b/tests/test_core.py index 357442b..cddedee 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -7,6 +7,7 @@ complete, expire_lease, release, + inspect, ) @@ -195,3 +196,84 @@ def test_release_wrong_token_fails(conn): assert result.success is False r2 = acquire(conn, "core:release:bad_token", ttl_ms=5000, hard_ttl_ms=10000) assert r2.acquired is False + + +# ─── INSPECT ──────────────────────────────────────────────────────────────── + +def test_inspect_missing_key(conn): + result = inspect(conn, "inspect:missing") + + assert result is None + + +def test_inspect_executing(conn): + r = acquire( + conn, + "inspect:executing", + ttl_ms=5000, + hard_ttl_ms=10000, + ) + + start_execution( + conn, + "inspect:executing", + owner_id=r.owner_id, + fencing_token=r.fencing_token, + ) + + result = inspect(conn, "inspect:executing") + + assert result.key == "inspect:executing" + assert result.status == "executing" + assert result.lease_alive is True + + +def test_inspect_completed(conn): + r = acquire( + conn, + "inspect:completed", + ttl_ms=5000, + hard_ttl_ms=10000, + ) + + start_execution( + conn, + "inspect:completed", + owner_id=r.owner_id, + fencing_token=r.fencing_token, + ) + + complete( + conn, + "inspect:completed", + execution_result={"ok": True}, + owner_id=r.owner_id, + fencing_token=r.fencing_token, + ) + + result = inspect(conn, "inspect:completed") + + assert result.status == "completed" + assert result.execution_result == {"ok": True} + + +def test_inspect_expired_execution(conn): + r = acquire( + conn, + "inspect:expired", + ttl_ms=100, + hard_ttl_ms=10000, + ) + + start_execution( + conn, + "inspect:expired", + owner_id=r.owner_id, + fencing_token=r.fencing_token, + ) + + time.sleep(0.2) + + result = inspect(conn, "inspect:expired") + + assert result.lease_alive is False \ No newline at end of file diff --git a/tests/test_once.py b/tests/test_once.py index d88131f..c6c088a 100644 --- a/tests/test_once.py +++ b/tests/test_once.py @@ -104,49 +104,6 @@ def boom(): assert r2.execution_alive is False -# ─── RECONCILE EXPOSED ────────────────────────────────────────────────────── - -def test_once_uncertain_exposes_reconcile(sentinel): - def boom(): - raise ValueError("intentional") - - result = sentinel.once("once:reconcile_exposed", fn=boom, ttl_ms=5000, hard_ttl_ms=10000) - assert result.uncertain is True - assert result.reconcile is not None - assert isinstance(result.reconcile, Reconcile) - - -def test_once_execution_alive_no_reconcile(get_conn_fixture): - barrier = threading.Barrier(2) - results = [] - - def slow_fn(): - barrier.wait() - time.sleep(0.5) - return {"done": True} - - def run_first(): - s = Sentinel(get_conn=get_conn_fixture) - results.append(("first", s.once("once:alive", fn=slow_fn, ttl_ms=5000, hard_ttl_ms=10000))) - - def run_second(): - barrier.wait() - time.sleep(0.05) - s = Sentinel(get_conn=get_conn_fixture) - results.append(("second", s.once("once:alive", fn=slow_fn, ttl_ms=5000, hard_ttl_ms=10000))) - - t1 = threading.Thread(target=run_first) - t2 = threading.Thread(target=run_second) - t1.start() - t2.start() - t1.join() - t2.join() - - second_result = next(r for label, r in results if label == "second") - assert second_result.execution_alive is True - assert second_result.reconcile is None - - # ─── NAMESPACE ────────────────────────────────────────────────────────────── def test_once_namespace_isolates_keys(get_conn_fixture): diff --git a/tests/test_reconciliation.py b/tests/test_reconciliation.py index bce3f12..4467863 100644 --- a/tests/test_reconciliation.py +++ b/tests/test_reconciliation.py @@ -15,22 +15,14 @@ def test_reconcile_marks_reconciling(conn, reconcile): r = acquire(conn, "rec:reconcile:basic", ttl_ms=100, hard_ttl_ms=10000) start_execution(conn, "rec:reconcile:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) time.sleep(0.2) - result = reconcile.reconcile("rec:reconcile:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) + result = reconcile.reconcile("rec:reconcile:basic") assert result.success is True def test_reconcile_fails_if_lease_still_alive(conn, reconcile): r = acquire(conn, "rec:reconcile:alive", ttl_ms=5000, hard_ttl_ms=10000) start_execution(conn, "rec:reconcile:alive", owner_id=r.owner_id, fencing_token=r.fencing_token) - result = reconcile.reconcile("rec:reconcile:alive", owner_id=r.owner_id, fencing_token=r.fencing_token) - assert result.success is False - - -def test_reconcile_fails_wrong_token(conn, reconcile): - r = acquire(conn, "rec:reconcile:bad_token", ttl_ms=100, hard_ttl_ms=10000) - start_execution(conn, "rec:reconcile:bad_token", owner_id=r.owner_id, fencing_token=r.fencing_token) - time.sleep(0.2) - result = reconcile.reconcile("rec:reconcile:bad_token", owner_id=r.owner_id, fencing_token=r.fencing_token + 999) + result = reconcile.reconcile("rec:reconcile:alive") assert result.success is False @@ -40,11 +32,9 @@ def test_force_complete_from_reconciling(conn, reconcile): r = acquire(conn, "rec:force:basic", ttl_ms=100, hard_ttl_ms=10000) start_execution(conn, "rec:force:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) time.sleep(0.2) - reconcile.reconcile("rec:force:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) + reconcile.reconcile("rec:force:basic") result = reconcile.force_complete( "rec:force:basic", - owner_id=r.owner_id, - fencing_token=r.fencing_token, execution_result='{"value": 1}' ) assert result.success is True @@ -55,8 +45,6 @@ def test_force_complete_fails_if_not_reconciling(conn, reconcile): start_execution(conn, "rec:force:bad_status", owner_id=r.owner_id, fencing_token=r.fencing_token) result = reconcile.force_complete( "rec:force:bad_status", - owner_id=r.owner_id, - fencing_token=r.fencing_token, execution_result='{"value": 1}' ) assert result.success is False @@ -68,13 +56,13 @@ def test_reset_allows_rerun(conn, reconcile): r = acquire(conn, "rec:reset:basic", ttl_ms=100, hard_ttl_ms=10000) start_execution(conn, "rec:reset:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) time.sleep(0.2) - reconcile.reconcile("rec:reset:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) - result = reconcile.reset("rec:reset:basic", owner_id=r.owner_id, fencing_token=r.fencing_token) + reconcile.reconcile("rec:reset:basic") + result = reconcile.reset("rec:reset:basic") assert result.success is True def test_reset_fails_if_not_reconciling(conn, reconcile): r = acquire(conn, "rec:reset:bad_status", ttl_ms=5000, hard_ttl_ms=10000) start_execution(conn, "rec:reset:bad_status", owner_id=r.owner_id, fencing_token=r.fencing_token) - result = reconcile.reset("rec:reset:bad_status", owner_id=r.owner_id, fencing_token=r.fencing_token) + result = reconcile.reset("rec:reset:bad_status") assert result.success is False