From a9d285aba765d52ae397cde629db5af56d294dd7 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Sun, 12 Jul 2026 05:20:19 -0600 Subject: [PATCH] fix(local): guarantee exclusive-admission slot locks release Ensure exclusive admission cleanup keeps releasing every held slot file even when an unlock or close operation fails, and release partial exclusive acquisitions after flock errors before propagating the original failure. Expand bundled capacity telemetry coverage to assert both first-attempt and retry records carry local_capacity_exhausted with retry_index 0 and 1 and no prompt/raw server content. Update the local capacity retry design note so pre-fix failure wording no longer describes the fixed commit as failing on HEAD. --- docs/design/local-provider-capacity-retry.md | 28 +++++++++++--------- solstone/think/providers/local_admission.py | 18 ++++++++++--- tests/test_local.py | 27 ++++++++++++++++--- tests/test_local_admission.py | 26 ++++++++++++++++++ 4 files changed, 78 insertions(+), 21 deletions(-) diff --git a/docs/design/local-provider-capacity-retry.md b/docs/design/local-provider-capacity-retry.md index 1572e547d..2bea8f355 100644 --- a/docs/design/local-provider-capacity-retry.md +++ b/docs/design/local-provider-capacity-retry.md @@ -198,20 +198,21 @@ emit fallback events. Cloud fallback behavior stays in the non-local branch. - Update `test_run_generate_bundled_context_rejection_backstop` to use the authoritative 400 body with `error.type == "exceed_context_size_error"`, - `n_prompt_tokens`, and `n_ctx`; expect `ContextBudgetExceeded`. This passes - on HEAD and pins preserved behavior. + `n_prompt_tokens`, and `n_ctx`; expect `ContextBudgetExceeded`. This passed + on the pre-fix tree too and pins preserved behavior. - Repoint the alt-phrasing test to the 500 `server_error` body with `Context size has been exceeded.`; expect `LocalCapacityExhausted` and - `local_capacity_exhausted`. Fails on HEAD because HEAD raises + `local_capacity_exhausted`. Failed on the pre-fix tree because it raised `ContextBudgetExceeded`. - Add a fallback case for a context-pattern body that is missing authoritative - structure; expect `LocalCapacityExhausted`. Fails on HEAD for the same reason. + structure; expect `LocalCapacityExhausted`. Failed on the pre-fix tree for + the same reason. - Add async parity for the 500 capacity body through fake `httpx.AsyncClient`; - expect `LocalCapacityExhausted`. Fails on HEAD because the shared helper - misclassifies it. + expect `LocalCapacityExhausted`. Failed on the pre-fix tree because the + shared helper misclassified it. - Add telemetry assertion for bundled generate failure: captured `record_local_inference` row has `reason_code == "local_capacity_exhausted"`. - Fails on HEAD with `context_budget_exceeded`. + Failed on the pre-fix tree with `context_budget_exceeded`. - Keep the no-POST fitter rejection test as-is: `local_budget.fit_contents()` still raises `ContextBudgetExceeded` before any HTTP request when preserved content cannot fit. @@ -219,24 +220,25 @@ emit fallback events. Cloud fallback behavior stays in the non-local branch. ### `tests/test_local_admission.py` - Exclusive acquisition blocks while a normal single-slot holder is active, then - succeeds after release. Fails on HEAD because no exclusive mode exists. + succeeds after release. Failed on the pre-fix tree because no exclusive mode + existed. - Exclusive acquisition with `capacity == 1` behaves like normal one-slot - acquisition. Fails on HEAD because no exclusive mode exists. + acquisition. Failed on the pre-fix tree because no exclusive mode existed. - Exclusive timeout releases any partially acquired locks before sleeping or timing out. Drive this by holding slot 1, leaving slot 0 free, then attempting exclusive acquire; after timeout, a normal waiter must be able to acquire - slot 0. Fails on HEAD because no all-slot primitive exists. + slot 0. Failed on the pre-fix tree because no all-slot primitive existed. - Exclusive permit releases all locks on exception. Acquire exclusive, raise inside the context manager, then acquire two normal permits at capacity 2. - Fails on HEAD because `LocalPermit` only represents one file. + Failed on the pre-fix tree because `LocalPermit` only represented one file. ### `tests/test_talent_fallback.py` - Add local capacity retry test where both first and retry calls raise `LocalCapacityExhausted`; assert exactly two calls, second call has `inference_retry_index=1` and `local_exclusive_admission=True`, no fallback is - consulted, and the emitted error event has `retries == 1`. Fails on HEAD - because local non-length errors re-raise without retry. + consulted, and the emitted error event has `retries == 1`. Failed on the + pre-fix tree because local non-length errors re-raised without retry. - Update existing incomplete-JSON retry assertions to prove the retry has `inference_retry_index=1` but no `local_exclusive_admission`. This guards against inferring exclusivity from retry index. diff --git a/solstone/think/providers/local_admission.py b/solstone/think/providers/local_admission.py index ebe8b3d84..5072b81fb 100644 --- a/solstone/think/providers/local_admission.py +++ b/solstone/think/providers/local_admission.py @@ -121,8 +121,16 @@ def _release_files(files: list[IO[str]]) -> None: continue try: fcntl.flock(lock_file, fcntl.LOCK_UN) - finally: - lock_file.close() + except Exception: + LOG.warning("failed to unlock local inference slot lock", exc_info=True) + _close_file(lock_file) + + +def _close_file(lock_file: IO[str]) -> None: + try: + lock_file.close() + except Exception: + LOG.warning("failed to close local inference slot lock", exc_info=True) def _try_acquire(capacity: int, started: float, root: Path) -> LocalPermit | None: @@ -157,8 +165,10 @@ def _try_acquire_exclusive( try: fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError as exc: - lock_file.close() - _release_files(lock_files) + try: + _close_file(lock_file) + finally: + _release_files(lock_files) if exc.errno in (errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK): return None raise diff --git a/tests/test_local.py b/tests/test_local.py index a9449dbaa..0572d428c 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -777,7 +777,7 @@ def test_run_agenerate_bundled_capacity_rejection_matches_sync(monkeypatch): assert exc.value.reason_code == "local_capacity_exhausted" -def test_run_generate_bundled_capacity_rejection_records_telemetry(monkeypatch): +def test_run_generate_bundled_capacity_rejection_records_retry_telemetry(monkeypatch): provider = _provider() monkeypatch.setattr(provider, "resolve_local_endpoint", _bundled_endpoint) _patch_bundled_server(monkeypatch) @@ -806,10 +806,29 @@ def test_run_generate_bundled_capacity_rejection_records_telemetry(monkeypatch): monkeypatch.setattr(httpx, "post", fake_post) with pytest.raises(provider.LocalProviderError): - provider.run_generate("hello", model=LOCAL_MODEL, max_output_tokens=16) + provider.run_generate( + "private prompt text", + model=LOCAL_MODEL, + max_output_tokens=16, + ) + with pytest.raises(provider.LocalProviderError): + provider.run_generate( + "private prompt text", + model=LOCAL_MODEL, + max_output_tokens=16, + inference_retry_index=1, + local_exclusive_admission=True, + ) - assert records - assert records[-1]["reason_code"] == "local_capacity_exhausted" + assert len(records) == 2 + assert [record["retry_index"] for record in records] == [0, 1] + for record in records: + assert record["reason_code"] == "local_capacity_exhausted" + assert record["outcome"] == "error" + serialized = json.dumps(record, sort_keys=True) + assert "private prompt text" not in serialized + assert "Context size has been exceeded." not in serialized + assert "server_error" not in serialized def test_openhands_local_llm_kwargs(monkeypatch): diff --git a/tests/test_local_admission.py b/tests/test_local_admission.py index 91669d4e3..311f75967 100644 --- a/tests/test_local_admission.py +++ b/tests/test_local_admission.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio +import errno import json import threading import time @@ -115,6 +116,31 @@ def test_exclusive_timeout_does_not_strand_partial_locks(monkeypatch, tmp_path): slot_one.release() +def test_exclusive_flock_error_releases_partial_slot_set(monkeypatch, tmp_path): + _isolated_journal(monkeypatch, tmp_path) + + from solstone.think.providers import local_admission + + real_flock = local_admission.fcntl.flock + + def fail_second_slot(lock_file, operation): + if ( + str(lock_file.name).endswith("slot-1.lock") + and operation & local_admission.fcntl.LOCK_EX + and operation & local_admission.fcntl.LOCK_NB + ): + raise OSError(errno.EIO, "simulated flock failure") + return real_flock(lock_file, operation) + + monkeypatch.setattr(local_admission.fcntl, "flock", fail_second_slot) + + with pytest.raises(OSError): + acquire_local_slot(2, 0.1, exclusive=True) + + with acquire_local_slot(1, 0.1) as permit: + assert permit.slot_index == 0 + + def test_exclusive_failure_releases_all_slots(monkeypatch, tmp_path): _isolated_journal(monkeypatch, tmp_path) -- 2.51.2