diff --git a/solstone/observe/sense.py b/solstone/observe/sense.py index 583fd12c2..294e56335 100644 --- a/solstone/observe/sense.py +++ b/solstone/observe/sense.py @@ -1409,11 +1409,11 @@ def main(): sensor.register(f"*{ext}", "depict", ["journal", "depict", "{file}"]) from solstone.think.speakers_analyze_installation import ( - begin_speakers_analyze_generation, + enter_speakers_analyze_generation, ) try: - sensor._speakers_analyze_generation = begin_speakers_analyze_generation( + sensor._speakers_analyze_generation = enter_speakers_analyze_generation( journal_path=journal ) except Exception as exc: diff --git a/solstone/think/speakers_analyze_installation.py b/solstone/think/speakers_analyze_installation.py index 51ef77465..0336b3a75 100644 --- a/solstone/think/speakers_analyze_installation.py +++ b/solstone/think/speakers_analyze_installation.py @@ -8,6 +8,7 @@ from __future__ import annotations import hashlib import json import os +import secrets import sys import uuid from collections.abc import Callable @@ -27,9 +28,12 @@ from solstone.apps.speakers.encoder_config import ( from solstone.think import probe from solstone.think.journal_io import MalformedPolicy, atomic_replace, read_json from solstone.think.journal_io.lease import ( + BorrowedFileLease, FileLease, acquire_file_lease, - probe_file_lease_held, + adopt_inherited_file_lease_fd, + read_file_lease_fd, + set_file_lease_offset_token, ) from solstone.think.model_assets import ( ModelsDistributionUnavailable, @@ -45,7 +49,12 @@ ROOT_DIST_NAME = "solstone" INSTALL_GENERATION_SCHEMA = "solstone.speakers_analyze.install_generation.v1" PROOF_KEY_SCHEMA = "solstone.speakers_analyze.install_proof_key.v1" GENERATION_ENV_KEY = "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_ID" +GENERATION_FD_ENV_KEY = "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_FD" +GENERATION_TOKEN_ENV_KEY = "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_TOKEN" GENERATION_MODE = 0o600 +GENERATION_FD_MIN = 3 +GENERATION_FD_MAX = 1_048_576 +GENERATION_TOKEN_MAX = (1 << 62) - 1 SPEAKERS_ANALYZE_REPAIR_TEXT = ( "Repair: reinstall the journal host stack with solstone-journal, or " @@ -87,9 +96,15 @@ class SpeakersAnalyzeInstallationResult: @dataclass(frozen=True) class SpeakersAnalyzeGeneration: generation_id: str - lease: FileLease + lease: FileLease | BorrowedFileLease def release(self) -> None: + if isinstance(self.lease, FileLease) and ( + os.environ.get(GENERATION_ENV_KEY) == self.generation_id + ): + os.environ.pop(GENERATION_ENV_KEY, None) + os.environ.pop(GENERATION_FD_ENV_KEY, None) + os.environ.pop(GENERATION_TOKEN_ENV_KEY, None) self.lease.release() def __enter__(self) -> SpeakersAnalyzeGeneration: @@ -164,7 +179,7 @@ def check_speakers_analyze_installation( return _digest_assets(proof_key) -def begin_speakers_analyze_generation( +def enter_speakers_analyze_generation( *, journal_path: str | Path | None = None, executable: str | Path | None = None, @@ -174,45 +189,63 @@ def begin_speakers_analyze_generation( ] = probe.current_solstone_core_platform, ) -> SpeakersAnalyzeGeneration: root = _journal_root(journal_path) - lease = acquire_file_lease( - _generation_lease_path(root), attempts=1, retry_max_seconds=0 - ) - if lease is None: - raise RuntimeError("speakers-analyze generation lease is already held") - cheap = _cheap_installation_result( + borrowed = _borrow_speakers_analyze_generation( + root=root, executable=executable, version_reader=version_reader, platform_reader=platform_reader, - platform_tag_reader=_packaging_platform_tags, - executable_predicate=lambda path: os.access(path, os.X_OK), ) - if not cheap.ok: - lease.release() - raise RuntimeError(cheap.message) - proof_key = _installation_proof_key( - executable=executable, - version_reader=version_reader, - platform_reader=platform_reader, + if borrowed is not None: + return borrowed + + lease = acquire_file_lease( + _generation_lease_path(root), attempts=1, retry_max_seconds=0 ) - result, observed_assets = _validated_asset_digests(proof_key) - if not result.ok: + if lease is None: + raise RuntimeError("speakers-analyze generation lease is already held") + try: + cheap = _cheap_installation_result( + executable=executable, + version_reader=version_reader, + platform_reader=platform_reader, + platform_tag_reader=_packaging_platform_tags, + executable_predicate=lambda path: os.access(path, os.X_OK), + ) + if not cheap.ok: + raise RuntimeError(cheap.message) + proof_key = _installation_proof_key( + executable=executable, + version_reader=version_reader, + platform_reader=platform_reader, + ) + result, observed_assets = _validated_asset_digests(proof_key) + if not result.ok: + raise RuntimeError(result.message) + generation_id = uuid.uuid4().hex + token = secrets.randbelow(GENERATION_TOKEN_MAX) + 1 + set_file_lease_offset_token(lease, token, _generation_lease_path(root)) + record = { + "schema": INSTALL_GENERATION_SCHEMA, + "generation_id": generation_id, + "created_at": _now_iso(), + "verified_at": _now_iso(), + "proof_key": proof_key, + "assets": observed_assets, + "helper": proof_key["helper"], + "packages": proof_key["packages"], + "platform": proof_key["platform"], + } + _write_generation_record(root, record) + os.environ[GENERATION_ENV_KEY] = generation_id + os.environ[GENERATION_FD_ENV_KEY] = str( + read_file_lease_fd(lease, _generation_lease_path(root)) + ) + os.environ[GENERATION_TOKEN_ENV_KEY] = str(token) + return SpeakersAnalyzeGeneration(generation_id=generation_id, lease=lease) + except BaseException: + _clear_generation_env() lease.release() - raise RuntimeError(result.message) - generation_id = uuid.uuid4().hex - record = { - "schema": INSTALL_GENERATION_SCHEMA, - "generation_id": generation_id, - "created_at": _now_iso(), - "verified_at": _now_iso(), - "proof_key": proof_key, - "assets": observed_assets, - "helper": proof_key["helper"], - "packages": proof_key["packages"], - "platform": proof_key["platform"], - } - _write_generation_record(root, record) - os.environ[GENERATION_ENV_KEY] = generation_id - return SpeakersAnalyzeGeneration(generation_id=generation_id, lease=lease) + raise def _cheap_installation_result( @@ -379,8 +412,81 @@ def _generation_proves_digest( if not generation_id: return False root = _journal_root(journal_path) - if not probe_file_lease_held(_generation_lease_path(root)): + candidate = _inherited_generation_candidate() + if candidate is None: return False + fd, token = candidate + borrowed = adopt_inherited_file_lease_fd( + _generation_lease_path(root), fd=fd, token=token + ) + if borrowed is None: + return False + try: + return _generation_record_proves_digest( + root=root, + generation_id=generation_id, + proof_key=proof_key, + ) + finally: + borrowed.release() + + +def _borrow_speakers_analyze_generation( + *, + root: Path, + executable: str | Path | None, + version_reader: Callable[[str], str], + platform_reader: Callable[[], probe.CorePlatform], +) -> SpeakersAnalyzeGeneration | None: + generation_id = os.environ.get(GENERATION_ENV_KEY) + if not any( + os.environ.get(key) + for key in ( + GENERATION_ENV_KEY, + GENERATION_FD_ENV_KEY, + GENERATION_TOKEN_ENV_KEY, + ) + ): + return None + candidate = _inherited_generation_candidate() + if generation_id is None or candidate is None: + _reject_generation_borrow() + return None + fd, token = candidate + borrowed = adopt_inherited_file_lease_fd( + _generation_lease_path(root), fd=fd, token=token + ) + if borrowed is None: + _reject_generation_borrow(candidate_fd=fd) + return None + try: + proof_key = _installation_proof_key( + executable=executable, + version_reader=version_reader, + platform_reader=platform_reader, + ) + if _generation_record_proves_digest( + root=root, + generation_id=generation_id, + proof_key=proof_key, + ): + return SpeakersAnalyzeGeneration( + generation_id=generation_id, lease=borrowed + ) + except BaseException: + borrowed.release() + raise + borrowed.release() + _reject_generation_borrow(candidate_fd=fd) + return None + + +def _generation_record_proves_digest( + *, + root: Path, + generation_id: str, + proof_key: dict[str, object], +) -> bool: raw = read_json( _generation_record_path(root), on_error=MalformedPolicy.WARN_AND_SKIP, @@ -405,6 +511,57 @@ def _generation_proves_digest( return True +def _inherited_generation_candidate() -> tuple[int, int] | None: + fd = _parse_generation_fd(os.environ.get(GENERATION_FD_ENV_KEY)) + token = _parse_generation_token(os.environ.get(GENERATION_TOKEN_ENV_KEY)) + if fd is None or token is None: + return None + try: + os.fstat(fd) + except OSError: + return None + return fd, token + + +def _parse_generation_fd(value: str | None) -> int | None: + if value is None: + return None + try: + fd = int(value) + except (TypeError, ValueError): + return None + if fd < GENERATION_FD_MIN or fd > GENERATION_FD_MAX: + return None + return fd + + +def _parse_generation_token(value: str | None) -> int | None: + if value is None: + return None + try: + token = int(value) + except (TypeError, ValueError): + return None + if token <= 0 or token > GENERATION_TOKEN_MAX: + return None + return token + + +def _reject_generation_borrow(candidate_fd: int | None = None) -> None: + if candidate_fd is not None: + try: + os.close(candidate_fd) + except OSError: + pass + _clear_generation_env() + + +def _clear_generation_env() -> None: + os.environ.pop(GENERATION_ENV_KEY, None) + os.environ.pop(GENERATION_FD_ENV_KEY, None) + os.environ.pop(GENERATION_TOKEN_ENV_KEY, None) + + def _required_assets() -> tuple[tuple[str, Path, str], ...]: return ( ("wespeaker", resolve_wespeaker_model(), WESPEAKER_MODEL_SHA256), @@ -445,7 +602,9 @@ def _now_iso() -> str: __all__ = [ + "GENERATION_FD_ENV_KEY", "GENERATION_ENV_KEY", + "GENERATION_TOKEN_ENV_KEY", "HELPER_BINARY_NAME", "HELPER_DIST_NAME", "MODELS_DIST_NAME", @@ -453,8 +612,8 @@ __all__ = [ "SPEAKERS_ANALYZE_REPAIR_TEXT", "SpeakersAnalyzeGeneration", "SpeakersAnalyzeInstallationResult", - "begin_speakers_analyze_generation", "check_speakers_analyze_installation", + "enter_speakers_analyze_generation", "runtime_has_speakers_analyze_wheel_coverage", "speakers_analyze_path_for_executable", ] diff --git a/tests/conftest.py b/tests/conftest.py index 31501bd13..b1f515a0e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -191,7 +191,9 @@ def set_test_journal_path(monkeypatch, _isolate_os_environ): @pytest.fixture(autouse=True) def _speakers_analyze_startup_invariant_ready( - monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest + monkeypatch: pytest.MonkeyPatch, + request: pytest.FixtureRequest, + tmp_path: Path, ) -> None: """Keep unit tests independent of the installed native helper wheel. @@ -202,31 +204,14 @@ def _speakers_analyze_startup_invariant_ready( return from solstone.think import speakers_analyze_installation as installation - - class _NoopLease: - def release(self) -> None: - return None - - def begin_ready_generation( - **_kwargs: object, - ) -> installation.SpeakersAnalyzeGeneration: - generation_id = "test-speakers-analyze-generation" - os.environ[installation.GENERATION_ENV_KEY] = generation_id - return installation.SpeakersAnalyzeGeneration( - generation_id=generation_id, - lease=_NoopLease(), - ) + from tests.helpers.speakers_analyze import install_enter_generation_stub monkeypatch.setattr( installation, "check_speakers_analyze_installation", lambda **_kwargs: installation.SpeakersAnalyzeInstallationResult("ok"), ) - monkeypatch.setattr( - installation, - "begin_speakers_analyze_generation", - begin_ready_generation, - ) + install_enter_generation_stub(monkeypatch, tmp_path) @pytest.fixture(autouse=True) diff --git a/tests/helpers/speakers_analyze.py b/tests/helpers/speakers_analyze.py new file mode 100644 index 000000000..cfaaf7aa6 --- /dev/null +++ b/tests/helpers/speakers_analyze.py @@ -0,0 +1,56 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Test helpers for speakers-analyze generation startup seams.""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + + +def install_enter_generation_stub( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + generation_id: str = "test-speakers-analyze-generation", +) -> None: + from solstone.think import speakers_analyze_installation as installation + + class _NoopLease: + def __init__(self, fd: int) -> None: + self.fd = fd + + def release(self) -> None: + if os.environ.get(installation.GENERATION_ENV_KEY) == generation_id: + monkeypatch.delenv(installation.GENERATION_ENV_KEY, raising=False) + monkeypatch.delenv(installation.GENERATION_FD_ENV_KEY, raising=False) + monkeypatch.delenv(installation.GENERATION_TOKEN_ENV_KEY, raising=False) + try: + os.close(self.fd) + except OSError: + pass + + def enter_ready_generation(**_kwargs: object): + token = 1 + fd = os.open( + tmp_path / "speakers-analyze-generation.fake", + os.O_RDWR | os.O_CREAT, + 0o600, + ) + os.lseek(fd, token, os.SEEK_SET) + monkeypatch.setenv(installation.GENERATION_ENV_KEY, generation_id) + monkeypatch.setenv(installation.GENERATION_FD_ENV_KEY, str(fd)) + monkeypatch.setenv(installation.GENERATION_TOKEN_ENV_KEY, str(token)) + return installation.SpeakersAnalyzeGeneration( + generation_id=generation_id, + lease=_NoopLease(fd), + ) + + monkeypatch.setattr( + installation, + "enter_speakers_analyze_generation", + enter_ready_generation, + ) diff --git a/tests/test_sense.py b/tests/test_sense.py index ab9415cf3..73baaf26d 100644 --- a/tests/test_sense.py +++ b/tests/test_sense.py @@ -5,6 +5,7 @@ import json import logging +import os import signal import subprocess import sys @@ -67,7 +68,7 @@ def _speakers_analyze_generation_ready(monkeypatch): monkeypatch.setattr( "solstone.think.speakers_analyze_installation." - "begin_speakers_analyze_generation", + "enter_speakers_analyze_generation", lambda **_kwargs: Generation(), ) @@ -2890,7 +2891,7 @@ def test_main_speakers_analyze_failure_prints_canonical_message_once( monkeypatch.setattr( "solstone.think.speakers_analyze_installation." - "begin_speakers_analyze_generation", + "enter_speakers_analyze_generation", fail_generation, ) @@ -2903,6 +2904,43 @@ def test_main_speakers_analyze_failure_prints_canonical_message_once( assert message in captured.err +def test_main_live_generation_id_without_fd_still_prints_canonical_message_once( + tmp_path, monkeypatch, capsys +): + from solstone.observe import sense + + message = "speakers-analyze generation lease is already held" + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + monkeypatch.setenv( + "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_ID", + "live-generation", + ) + monkeypatch.setenv("SOL_SUPERVISOR_SPAWNED", "1") + monkeypatch.delenv("SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_FD", raising=False) + monkeypatch.delenv( + "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_TOKEN", + raising=False, + ) + monkeypatch.setattr(sense, "require_solstone", lambda: None) + monkeypatch.setattr(sys, "argv", ["sense", "--day", "20250101"]) + + def fail_generation(**_kwargs): + raise RuntimeError(message) + + monkeypatch.setattr( + "solstone.think.speakers_analyze_installation." + "enter_speakers_analyze_generation", + fail_generation, + ) + + with pytest.raises(SystemExit) as exc_info: + sense.main() + + captured = capsys.readouterr() + assert exc_info.value.code == 78 + assert captured.err.count(message) == 1 + + def _registered_describe_commands(sensor: FileSensor) -> list[list[str]]: return [ command @@ -3098,6 +3136,61 @@ def test_queue_wait_ms_reaches_child_env(tmp_path, monkeypatch): assert "SOL_QUEUE_WAIT_MS" not in captured["env"] +def test_generation_env_reaches_event_and_batch_transcribe_children( + tmp_path, monkeypatch +): + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + fd = os.open(tmp_path / "generation.lock", os.O_RDWR | os.O_CREAT, 0o600) + monkeypatch.setenv( + "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_ID", + "generation", + ) + monkeypatch.setenv("SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_FD", str(fd)) + monkeypatch.setenv("SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_TOKEN", "123") + sensor = FileSensor(tmp_path) + test_file = make_segment_file(tmp_path, "audio.flac") + captured: list[dict[str, str]] = [] + + def fake_runner_spawn(cmd, ref, callosum, env, day): + captured.append(env) + return FakeManaged(FakeProcess(0), ref=ref) + + monkeypatch.setattr( + "solstone.observe.sense.RunnerManagedProcess.spawn", fake_runner_spawn + ) + + try: + sensor._spawn_managed_process( + ["journal", "transcribe", str(test_file)], + test_file, + "event-ref", + "143022_300", + None, + None, + None, + ) + sensor._spawn_managed_process( + ["journal", "transcribe", str(test_file)], + test_file, + "batch-ref", + "143022_300", + None, + None, + "20250101", + ) + finally: + os.close(fd) + + assert [ + ( + env["SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_ID"], + env["SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_FD"], + env["SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_TOKEN"], + ) + for env in captured + ] == [("generation", str(fd), "123"), ("generation", str(fd), "123")] + + def test_run_handler_passes_queue_wait_from_queued_at(tmp_path, monkeypatch): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) sensor = FileSensor(tmp_path) diff --git a/tests/test_speakers_analyze_installation.py b/tests/test_speakers_analyze_installation.py index c936ce6c1..db4c41755 100644 --- a/tests/test_speakers_analyze_installation.py +++ b/tests/test_speakers_analyze_installation.py @@ -5,6 +5,7 @@ from __future__ import annotations +import json import os from importlib.metadata import PackageNotFoundError from pathlib import Path @@ -13,6 +14,13 @@ import pytest from solstone.think import probe from solstone.think import speakers_analyze_installation as installation +from solstone.think.journal_io.lease import ( + BorrowedFileLease, + FileLease, + acquire_file_lease, + read_file_lease_fd, + read_file_lease_offset_token, +) def _version_reader(dist_name: str) -> str: @@ -61,6 +69,34 @@ def _asset_fixtures(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: ) +def _entry_kwargs(tmp_path: Path, executable: Path) -> dict: + return { + "journal_path": tmp_path, + "executable": executable, + "version_reader": _version_reader, + "platform_reader": _platform_reader, + } + + +def _clear_generation_env() -> None: + os.environ.pop(installation.GENERATION_ENV_KEY, None) + os.environ.pop(installation.GENERATION_FD_ENV_KEY, None) + os.environ.pop(installation.GENERATION_TOKEN_ENV_KEY, None) + + +def _restore_generation_env( + monkeypatch: pytest.MonkeyPatch, generation_id: str, fd: int, token: int +) -> None: + monkeypatch.setenv(installation.GENERATION_ENV_KEY, generation_id) + monkeypatch.setenv(installation.GENERATION_FD_ENV_KEY, str(fd)) + monkeypatch.setenv(installation.GENERATION_TOKEN_ENV_KEY, str(token)) + + +def _assert_fd_closed(fd: int) -> None: + with pytest.raises(OSError): + os.fstat(fd) + + def test_coverage_gate_reads_helper_constants_not_core_constants(monkeypatch): core_platform = ("coreos", "core64") helper_platform = ("helperos", "helper64") @@ -225,11 +261,8 @@ def test_live_generation_record_reuses_digest_proof( ): executable = _helper(tmp_path) _asset_fixtures(tmp_path, monkeypatch) - generation = installation.begin_speakers_analyze_generation( - journal_path=tmp_path, - executable=executable, - version_reader=_version_reader, - platform_reader=_platform_reader, + generation = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) ) calls = 0 @@ -251,7 +284,7 @@ def test_live_generation_record_reuses_digest_proof( ) finally: generation.release() - os.environ.pop(installation.GENERATION_ENV_KEY, None) + _clear_generation_env() assert result.status == "ok" assert calls == 0 @@ -262,15 +295,12 @@ def test_stale_generation_record_degrades_to_full_digest( ): executable = _helper(tmp_path) _asset_fixtures(tmp_path, monkeypatch) - generation = installation.begin_speakers_analyze_generation( - journal_path=tmp_path, - executable=executable, - version_reader=_version_reader, - platform_reader=_platform_reader, + generation = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) ) generation_id = generation.generation_id generation.release() - os.environ.pop(installation.GENERATION_ENV_KEY, None) + _clear_generation_env() calls = 0 original_digest = installation._sha256_file @@ -292,3 +322,230 @@ def test_stale_generation_record_degrades_to_full_digest( assert result.status == "ok" assert calls == 2 + + +def test_owned_entry_publishes_fd_token_and_token_free_record( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + executable = _helper(tmp_path) + _asset_fixtures(tmp_path, monkeypatch) + + generation = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + try: + assert isinstance(generation.lease, FileLease) + fd = int(os.environ[installation.GENERATION_FD_ENV_KEY]) + token = int(os.environ[installation.GENERATION_TOKEN_ENV_KEY]) + assert os.environ[installation.GENERATION_ENV_KEY] == generation.generation_id + assert token > 0 + assert read_file_lease_fd(generation.lease) == fd + assert read_file_lease_offset_token(fd) == token + + record = json.loads( + ( + tmp_path / "health" / "speakers-analyze" / "install-generation.json" + ).read_text(encoding="utf-8") + ) + assert record["schema"] == installation.INSTALL_GENERATION_SCHEMA + assert record["generation_id"] == generation.generation_id + assert "token" not in record + assert installation.GENERATION_TOKEN_ENV_KEY not in record + finally: + generation.release() + _clear_generation_env() + + +def test_borrowed_entry_reuses_proof_without_hash_and_keeps_owner_live( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + executable = _helper(tmp_path) + _asset_fixtures(tmp_path, monkeypatch) + owner = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + owner_fd = int(os.environ[installation.GENERATION_FD_ENV_KEY]) + token = int(os.environ[installation.GENERATION_TOKEN_ENV_KEY]) + inherited_fd = os.dup(owner_fd) + _restore_generation_env(monkeypatch, owner.generation_id, inherited_fd, token) + calls = 0 + + def fail_digest(_path: Path) -> str: + nonlocal calls + calls += 1 + raise AssertionError("borrow must reuse the live proof") + + monkeypatch.setattr(installation, "_sha256_file", fail_digest) + borrower = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + try: + assert isinstance(borrower.lease, BorrowedFileLease) + assert calls == 0 + borrower.release() + assert ( + acquire_file_lease( + tmp_path / "health" / "speakers-analyze" / "install-generation.lock", + attempts=1, + ) + is None + ) + _restore_generation_env(monkeypatch, owner.generation_id, inherited_fd, token) + assert ( + installation.check_speakers_analyze_installation( + journal_path=tmp_path, + executable=executable, + version_reader=_version_reader, + platform_reader=_platform_reader, + platform_tag_reader=_platform_tags, + ).status + == "ok" + ) + finally: + try: + os.close(inherited_fd) + except OSError: + pass + owner.release() + _clear_generation_env() + + +def test_copied_or_separately_opened_generation_env_cannot_borrow( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + executable = _helper(tmp_path) + _asset_fixtures(tmp_path, monkeypatch) + owner = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + owner_fd = int(os.environ[installation.GENERATION_FD_ENV_KEY]) + token = int(os.environ[installation.GENERATION_TOKEN_ENV_KEY]) + lock_path = tmp_path / "health" / "speakers-analyze" / "install-generation.lock" + + cases: list[tuple[str, int | None]] = [] + cases.append(("missing-fd", None)) + unrelated_fd = os.open(tmp_path / "unrelated.lock", os.O_RDWR | os.O_CREAT, 0o600) + cases.append(("unrelated-fd", unrelated_fd)) + separate_fd = os.open(lock_path, os.O_RDWR) + os.lseek(separate_fd, token, os.SEEK_SET) + cases.append(("separate-same-path-fd", separate_fd)) + mismatched_id_fd = os.dup(owner_fd) + cases.append(("mismatched-id", mismatched_id_fd)) + + try: + for name, fd in cases: + _restore_generation_env( + monkeypatch, + "stale-generation" if name == "mismatched-id" else owner.generation_id, + fd if fd is not None else owner_fd, + token, + ) + if name == "missing-fd": + os.environ.pop(installation.GENERATION_FD_ENV_KEY, None) + monkeypatch.setenv("SOL_SUPERVISOR_SPAWNED", "1") + with pytest.raises(RuntimeError, match="generation lease is already held"): + installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + assert installation.GENERATION_ENV_KEY not in os.environ + assert installation.GENERATION_FD_ENV_KEY not in os.environ + assert installation.GENERATION_TOKEN_ENV_KEY not in os.environ + if fd is not None: + _assert_fd_closed(fd) + finally: + owner.release() + _clear_generation_env() + + +def test_final_duplicate_rejection_closes_candidate_before_fresh_entry( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + executable = _helper(tmp_path) + _asset_fixtures(tmp_path, monkeypatch) + owner = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + old_fd = int(os.environ[installation.GENERATION_FD_ENV_KEY]) + token = int(os.environ[installation.GENERATION_TOKEN_ENV_KEY]) + final_duplicate_fd = os.dup(old_fd) + os.close(old_fd) + owner.lease._fd = None + _restore_generation_env(monkeypatch, "stale-generation", final_duplicate_fd, token) + + calls = 0 + original_digest = installation._sha256_file + + def counted_digest(path: Path) -> str: + nonlocal calls + calls += 1 + return original_digest(path) + + monkeypatch.setattr(installation, "_sha256_file", counted_digest) + fresh = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + try: + assert fresh.generation_id != owner.generation_id + assert calls == 2 + _assert_fd_closed(final_duplicate_fd) + finally: + fresh.release() + _clear_generation_env() + + +@pytest.mark.parametrize( + "failure", ["proof-key", "validation", "token-init", "record-write"] +) +def test_owned_entry_failures_release_and_allow_reacquire( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: str +): + executable = _helper(tmp_path) + _asset_fixtures(tmp_path, monkeypatch) + if failure == "proof-key": + monkeypatch.setattr( + installation, + "_installation_proof_key", + lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("proof failed")), + ) + elif failure == "validation": + monkeypatch.setattr( + installation, + "_validated_asset_digests", + lambda _proof_key: ( + installation.SpeakersAnalyzeInstallationResult( + "asset-missing", "validation failed" + ), + [], + ), + ) + elif failure == "token-init": + monkeypatch.setattr( + installation, + "set_file_lease_offset_token", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + RuntimeError("token failed") + ), + ) + else: + monkeypatch.setattr( + installation, + "_write_generation_record", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + RuntimeError("write failed") + ), + ) + + with pytest.raises(RuntimeError): + installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + assert installation.GENERATION_ENV_KEY not in os.environ + assert installation.GENERATION_FD_ENV_KEY not in os.environ + assert installation.GENERATION_TOKEN_ENV_KEY not in os.environ + + lease = acquire_file_lease( + tmp_path / "health" / "speakers-analyze" / "install-generation.lock", + attempts=1, + ) + assert lease is not None + lease.release()