diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md index ad6c12e6..aa3f2f53 100644 --- a/docs/resource-budgets.md +++ b/docs/resource-budgets.md @@ -518,3 +518,18 @@ sensor reads when disabled. Finite-default calibration, workspace reservations, system headroom, bounded recovery and actual Mac/MPS memory and thermal gates remain pending. This checkpoint does not establish that the original 26.24 GB Mac symptom is fixed. + +### Failed embedding transfer retention checkpoint + +If embedding recovery's CPU transfer fails or is cancelled, the daemon drops +the cached embedding snapshot only when it still names that failed instance. +The original exception propagates unchanged, with no extra load, retry or +partial successful result. Sticky CPU selection remains set; the next admitted +request rebuilds the same requested model on CPU. An unrelated replacement +snapshot survives. Existing generation fencing retires idle obsolete fast +clones and lets active captured work finish before retirement. + +Synthetic tests cover ordinary errors, memory errors and cancellation, retained +identity, ownership release and weak references after exception tracebacks are +released. This bounds failed Python cache retention, not native allocator +reclamation, CPU-transfer peak memory or the original Mac memory incident. diff --git a/tests/test-model-process-admission-297.py b/tests/test-model-process-admission-297.py index 2c0dbb3c..afe380d8 100644 --- a/tests/test-model-process-admission-297.py +++ b/tests/test-model-process-admission-297.py @@ -5,6 +5,7 @@ import importlib.util import os import sys +import threading import types import unittest import weakref @@ -559,6 +560,185 @@ def test_invalid_sensor_cases_do_not_activate_unrelated_throttler(self) -> None: with patch.object(self.server, "_activate_throttler", side_effect=AssertionError("Unrelated throttler")): self.test_invalid_and_missing_measurements_refuse_before_construction() + def assert_failed_transfer(self, failure: BaseException, count: int) -> None: + responses = [] + try: + responses.append(self.request(count=count, tier="base")) + except BaseException as caught: + self.assertIs(caught, failure) + else: + self.fail("A failed transfer returned a response") + self.assertEqual(responses, []) + self.assertIn("embed", self.server._sticky_cpu) + self.assertEqual(self.moves, ["cpu"]) + self.assertFalse(self.server._lock.locked()) + self.assertFalse(self.server._inference_lock.locked()) + + def test_failed_cpu_transfer_drops_cache_and_next_request_rebuilds_same_identity(self) -> None: + for count, fail_at in ((1, 1), (5, 2)): + for error_type in (RuntimeError, MemoryError, ValueError, KeyboardInterrupt, SystemExit): + with self.subTest(count=count, error=error_type.__name__): + self.clear_models() + self.server._sticky_cpu.clear() + self.moves.clear() + failure = error_type("synthetic failed CPU transfer") + workspaces = [] + + def fail_inference(model: Model) -> None: + if len(self.calls) == fail_at: + self.fail_mps(model) + + def fail_transfer(model: Model, device: str) -> None: + self.assertEqual(model.identity, "qwen3_256") + self.assertEqual(model.device, "mps") + self.assertTrue(self.server._inference_lock.locked()) + self.assertTrue(self.server._lock.locked()) + self.moves.append(device) + workspace = SUPPORT.Workspace() + workspaces.append(weakref.ref(workspace)) + raise failure + + self.inference_hook = fail_inference + with patch.object(Model, "to", fail_transfer): + self.assert_failed_transfer(failure, count) + self.assertIsNone(self.server._embed_state) + self.assertEqual(len(self.calls), fail_at) + self.assertEqual(self.builds, [("qwen3_256", "mps")]) + self.assertEqual(self.server._inflight, 0) + self.assertFalse(self.server._fast_lock.locked()) + self.assertIsNotNone(workspaces[0]()) + self.assertIsNotNone(self.refs[-1]()) + failure.__traceback__ = None + self.assertIsNone(workspaces[0]()) + self.assertIsNone(self.refs[-1]()) + + self.inference_hook = self.fail_mps + response = self.request(count=count, tier="base") + self.assertTrue(response["ok"]) + self.assertEqual(self.builds, [("qwen3_256", "mps"), ("qwen3_256", "cpu")]) + self.assertEqual(self.moves, ["cpu"]) + self.assertEqual(response["vectors"].data, + [value for index in range(count) for value in (float(index), 0.5)]) + self.assertEqual(response["vectors"].dtype, "float32") + + def test_failed_transfer_preserves_an_unrelated_replacement_snapshot(self) -> None: + for count in (1, 3): + with self.subTest(count=count): + self.clear_models() + self.server._sticky_cpu.clear() + self.moves.clear() + failure = ValueError("synthetic transfer failure after replacement") + replacement = Model(self, "qwen3_256", "cpu") + replacement_state = self.ms._EmbedState(replacement, "base", "qwen3_256") + + def fail_transfer(model: Model, device: str) -> None: + self.moves.append(device) + self.server._publish_embed_state(replacement_state) + raise failure + + self.inference_hook = self.fail_mps + with patch.object(Model, "to", fail_transfer): + self.assert_failed_transfer(failure, count) + self.assertIs(self.server._embed_state, replacement_state) + failed = self.refs[-1] + failure.__traceback__ = None + self.assertIsNone(failed()) + self.assertTrue(self.request(count=count, tier="base")["ok"]) + self.assertEqual(self.builds, [("qwen3_256", "mps")]) + self.assertIs(self.server._embed_state.model, replacement) + + def test_failed_transfer_retires_idle_fast_clone(self) -> None: + self.assertTrue(self.request(tier="base")["ok"]) + main = self.refs[-1] + with self.server._inference_lock: + self.assertTrue(self.request(count=1, tier="base")["ok"]) + clone = self.refs[-1] + self.assertEqual(self.builds, [("qwen3_256", "mps"), ("qwen3_256", "cpu")]) + failure = MemoryError("synthetic transfer allocation failure") + + def fail_transfer(model: Model, device: str) -> None: + self.moves.append(device) + raise failure + + self.inference_hook = self.fail_mps + with patch.object(Model, "to", fail_transfer): + self.assert_failed_transfer(failure, 3) + self.assertIsNone(self.server._embed_state) + self.assertIsNone(self.server._fast_encoder) + self.assertIsNone(self.server._fast_model_id) + self.assertIsNone(self.server._fast_generation) + self.assertIsNone(clone()) + failure.__traceback__ = None + self.assertIsNone(main()) + + def test_failed_transfer_allows_active_fast_work_to_finish_without_stale_publication(self) -> None: + for phase in ("construction", "inference"): + with self.subTest(phase=phase): + self.clear_models() + self.server._sticky_cpu.clear() + self.moves.clear() + self.inference_hook = lambda model: None + self.build_hook = lambda model: None + self.assertTrue(self.request(tier="base")["ok"]) + main = self.refs[-1] + entered, release = threading.Event(), threading.Event() + outcomes, errors = [], [] + failure = ValueError("synthetic transfer failure beside active fast work") + + def wait_for_release() -> None: + entered.set() + self.assertTrue(release.wait(2), "Synthetic fast work was not released") + + def on_build(model: Model) -> None: + if model.device == "cpu" and phase == "construction": + wait_for_release() + + def on_inference(model: Model) -> None: + if model.device == "cpu" and phase == "inference": + wait_for_release() + elif model.device == "mps": + self.fail_mps(model) + + def run_fast() -> None: + try: + outcomes.append(self.request(count=1, tier="base")) + except BaseException as error: + errors.append(error) + + def fail_transfer(model: Model, device: str) -> None: + self.moves.append(device) + raise failure + + self.build_hook = on_build + self.inference_hook = on_inference + worker = threading.Thread(target=run_fast, daemon=True) + try: + with self.server._inference_lock: + worker.start() + self.assertTrue(entered.wait(1), "Fast owner never entered its native stage") + clone = self.refs[-1] + with patch.object(Model, "to", fail_transfer): + self.assert_failed_transfer(failure, 3) + self.assertIsNone(self.server._embed_state) + self.assertTrue(self.server._fast_lock.locked()) + self.assertTrue(worker.is_alive()) + failure.__traceback__ = None + self.assertIsNone(main()) + self.assertIsNotNone(clone()) + finally: + release.set() + if worker.ident is not None: + worker.join(2) + self.assertFalse(worker.is_alive(), "Fast owner remained blocked") + self.assertEqual(errors, []) + self.assertEqual(len(outcomes), 1) + self.assertTrue(outcomes[0]["ok"]) + self.assertEqual(outcomes[0]["vectors"].data, [0.0, 0.5]) + self.assertIsNone(self.server._fast_encoder) + self.assertIsNone(self.server._fast_model_id) + self.assertIsNone(self.server._fast_generation) + self.assertIsNone(clone()) + if __name__ == "__main__": unittest.main() diff --git a/truememory/model_server.py b/truememory/model_server.py index f8727c24..af92bee2 100644 --- a/truememory/model_server.py +++ b/truememory/model_server.py @@ -490,7 +490,15 @@ def _recover_embed_oom_locked( if deadline is not None: self._check_embed_recovery_deadline_locked(model, deadline) if hasattr(model, "to"): - model.to("cpu") + try: + model.to("cpu") + except BaseException: + # Failed or cancelled transfers cannot certify this instance + # as CPU-ready. Preserve the original error and sticky choice. + state = self._embed_state + if state is not None and state.model is model: + self._publish_embed_state(None) + raise def _check_embed_recovery_deadline_locked( self, model: object, deadline: _RequestDeadline,