From 1b94ae1bc8b758eb9bd52b9fc911539f4b71e569 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Tue, 21 Jul 2026 05:04:52 -0600 Subject: [PATCH] feat(providers): probe install lease state without creating it Add a non-creating provider install lease probe that reports held, free, or missing by trial-flocking an existing lease file only. Cover missing, free, held, and unexpected flock error behavior without changing the existing creating lease acquisition or boolean free probe. --- solstone/think/providers/install_lease.py | 33 ++++++++++++- tests/test_install_lease.py | 56 +++++++++++++++++++++++ 2 files changed, 88 insertions(+), 1 deletion(-) diff --git a/solstone/think/providers/install_lease.py b/solstone/think/providers/install_lease.py index 6c4eb6f47..7170d1115 100644 --- a/solstone/think/providers/install_lease.py +++ b/solstone/think/providers/install_lease.py @@ -12,7 +12,7 @@ import random import time from dataclasses import dataclass from pathlib import Path -from typing import Any +from typing import Any, Literal from solstone.think.providers.install_state import PROVIDERS, ProviderName from solstone.think.utils import get_journal @@ -112,6 +112,36 @@ def assert_install_lease_owned( return lease +def probe_install_lease_state( + provider: str, + *, + journal_path: str | Path | None = None, +) -> Literal["held", "free", "missing"]: + """Probe an existing provider lease with a trial flock. + + The only sanctioned side effect is a transient flock on an existing lease + file; this function never creates the lease path. + """ + validated = _validate_provider(provider) + path = lease_path(validated, journal_path=journal_path) + fd: int | None = None + try: + fd = os.open(path, os.O_RDONLY) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + if exc.errno in (errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK): + return "held" + raise + fcntl.flock(fd, fcntl.LOCK_UN) + return "free" + except (FileNotFoundError, NotADirectoryError): + return "missing" + finally: + if fd is not None: + os.close(fd) + + def probe_install_lease_free( provider: str, *, @@ -162,5 +192,6 @@ __all__ = [ "assert_install_lease_owned", "lease_path", "probe_install_lease_free", + "probe_install_lease_state", "prune_unowned_lease_file", ] diff --git a/tests/test_install_lease.py b/tests/test_install_lease.py index d6bb0dc3e..7e31fc3bb 100644 --- a/tests/test_install_lease.py +++ b/tests/test_install_lease.py @@ -9,10 +9,13 @@ import sys import time from pathlib import Path +import pytest + from solstone.think.providers.install_lease import ( acquire_install_lease, lease_path, probe_install_lease_free, + probe_install_lease_state, prune_unowned_lease_file, ) @@ -45,6 +48,59 @@ def test_context_release_paths(tmp_path, monkeypatch) -> None: assert probe_install_lease_free("local") is True +def test_probe_install_lease_state_missing_does_not_create_path( + tmp_path, monkeypatch +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + path = lease_path("local") + + assert probe_install_lease_state("local") == "missing" + assert not path.exists() + assert not path.parent.exists() + + +def test_probe_install_lease_state_free_existing_file(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + path = lease_path("local") + path.parent.mkdir(parents=True) + path.write_text("", encoding="utf-8") + + assert probe_install_lease_state("local") == "free" + assert path.exists() + + +def test_probe_install_lease_state_held_existing_file(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + + lease = acquire_install_lease("local") + assert lease is not None + try: + assert probe_install_lease_state("local") == "held" + finally: + lease.release() + + assert probe_install_lease_state("local") == "free" + + +def test_probe_install_lease_state_propagates_unexpected_oserror( + tmp_path, monkeypatch +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + path = lease_path("local") + path.parent.mkdir(parents=True) + path.write_text("", encoding="utf-8") + + from solstone.think.providers import install_lease + + def fail_flock(_fd: int, _operation: int) -> None: + raise OSError("unexpected flock failure") + + monkeypatch.setattr(install_lease.fcntl, "flock", fail_flock) + + with pytest.raises(OSError, match="unexpected flock failure"): + probe_install_lease_state("local") + + def test_prune_unowned_lease_file(tmp_path, monkeypatch) -> None: monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) path = lease_path("local") -- 2.51.2