diff --git a/sdk/python/README.md b/sdk/python/README.md index a828280..ee9dbe9 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -105,6 +105,7 @@ Sandbox( detached: bool = False, node_id: str | None = None, *, + failover: bool = False, xpu: str | None = None, storage_mb: int | None = None, network_policy: NetworkPolicy | None = None, @@ -435,11 +436,44 @@ public gateway. `Sandbox(failover=True)` opts into same-node recovery of the same logical sandbox after its physical runtime fails. `sandbox.reload()` requests the same -rollback explicitly. It returns `False` whenever the rollback is not completed, -including when no usable local anonymous checkpoint exists, the sandbox is -already closed, or the backend reports an operational failure. A successful -reload preserves `sandbox.id` and the existing commands, filesystem, and PTY -facades. +recovery explicitly. Recovery first queries the latest valid local anonymous +checkpoint. If one exists, the replacement runtime restores it. If the node +authoritatively reports that no anonymous checkpoint exists, the same +FunctionProxy and FunctionAgent cold-start a replacement from the sandbox's +original creation parameters. Query, authorization, metadata, download, +validation, and restore failures are errors; they never silently downgrade to +a cold start. + +Both successful paths preserve the logical `sandbox.id` and the existing +commands, filesystem, and PTY facade objects. They create a new physical +runtime, so callers must not cache a runtime ID or route address across reload. +Snapshot recovery restores the checkpointed process and writable-filesystem +state. Cold start does not preserve memory, running processes, or writable +files from the old runtime; only declarative creation inputs such as runtime, +rootfs, mounts, resources, environment, network policy, ports, and placement +are reused. + +`sandbox.reload()` intentionally keeps its public `bool` result. Internally, +the frontend and native Sandbox SDK propagate an optional `snapshot` or +`cold-start` mode to the AKernel backend. On `cold-start`, all tracked command +handles created before the recovery boundary fail locally before native +`wait`, `kill`, or `send_stdin` can target a reused PID. Completed +`CommandResult` objects remain ordinary immutable values. New commands receive +the new generation and work normally. Raw integer PID operations also fail +closed after an authoritative cold start because an integer carries no +generation identity. Snapshot recovery keeps old handles available. With an +older server or native SDK that omits the recovery mode, AKernel retains +conservative compatibility: an old handle remains usable only until a native +operation proves that its process was not restored. When the server explicitly +reports `snapshot`, a transient native operation error remains that operation's +error and does not poison the handle generation as though a cold start had +occurred. + +Reload returns `False` when recovery cannot start or complete, including a +closed sandbox, an in-flight command operation, an unsupported backend, or an +operational failure. A source-stop failure is not reported as cold-start +success. The actor-based `openyuanrong-sdk` backend does not support failover +or reload. Recovery points are local and follow the source sandbox lifecycle. They are created by sandbox workloads through RRT's internal `POST /checkpoint` @@ -459,8 +493,7 @@ AKERNEL_TEST_RUNTIME=runsc python examples/failover_reload.py ``` See [`examples/failover_reload.py`](./examples/failover_reload.py) for the -internal trigger used during integration. The actor-based -`openyuanrong-sdk` backend does not support failover or reload. +internal checkpoint trigger used during integration. ## Reverse tunnels diff --git a/sdk/python/akernel_sdk/_backends/openyuanrong_sandbox.py b/sdk/python/akernel_sdk/_backends/openyuanrong_sandbox.py index 8f577d8..5cc6fb1 100644 --- a/sdk/python/akernel_sdk/_backends/openyuanrong_sandbox.py +++ b/sdk/python/akernel_sdk/_backends/openyuanrong_sandbox.py @@ -18,7 +18,9 @@ import inspect import os -from collections.abc import Mapping +import threading +from collections.abc import Callable, Iterator, Mapping +from contextlib import contextmanager from typing import Any import yr_sandbox @@ -46,6 +48,12 @@ _NAMESPACE = "default" _DEFAULT_LISTEN_PORT = 8766 +_COLD_START_HANDLE_ERROR = ( + "pre-reload command handle was not restored after sandbox cold start" +) +_COMMAND_OPERATION_CONFLICT = ( + "command operation unavailable while another sandbox operation is in flight" +) def _native_port_range(value: PortRange | int | None) -> Any: @@ -149,10 +157,119 @@ def _entry_info(value: Any) -> EntryInfo: ) +class _TrackedCommandPid(int): + """PID-compatible token that keeps one native handle's SDK generation.""" + + generation: int + native_handle: Any + + def __new__( + cls, + pid: int, + generation: int, + native_handle: Any, + ) -> _TrackedCommandPid: + value = int.__new__(cls, pid) + value.generation = generation + value.native_handle = native_handle + return value + + class _CommandsDriver: def __init__(self, commands: Any) -> None: self._commands = commands self._handles: dict[int, Any] = {} + self._generation = 0 + self._invalid_generations: set[int] = set() + self._uncertain_generations: set[int] = set() + self._raw_pid_operations_invalid = False + self._generation_lock = threading.Lock() + self._operation_state_lock = threading.Lock() + self._active_command_operations = 0 + self._reload_in_progress = False + + @contextmanager + def _command_operation(self) -> Iterator[None]: + with self._operation_state_lock: + if self._reload_in_progress: + raise BackendOperationError(_COMMAND_OPERATION_CONFLICT) + self._active_command_operations += 1 + try: + yield + finally: + with self._operation_state_lock: + self._active_command_operations -= 1 + + def reload( + self, + native_reload: Callable[[], Any], + recovery_mode: Callable[[], Any], + ) -> bool: + """Run native reload and its generation commit as one boundary.""" + + with self._operation_state_lock: + if self._reload_in_progress or self._active_command_operations: + return False + self._reload_in_progress = True + try: + try: + reloaded = bool(native_reload()) + except Exception: + return False + if reloaded: + try: + mode = recovery_mode() + except Exception: + mode = None + with self._generation_lock: + self._generation += 1 + if mode == "cold-start": + self._invalid_generations.update(range(self._generation)) + self._raw_pid_operations_invalid = True + elif mode != "snapshot": + self._uncertain_generations.update(range(self._generation)) + return reloaded + finally: + with self._operation_state_lock: + self._reload_in_progress = False + + def _current_generation(self) -> int: + with self._generation_lock: + return self._generation + + def _tracked_handle(self, pid: int) -> _TrackedCommandPid | None: + if isinstance(pid, _TrackedCommandPid): + return pid + return None + + def _ensure_generation_valid(self, pid: _TrackedCommandPid) -> None: + with self._generation_lock: + invalid = pid.generation in self._invalid_generations + if invalid: + raise self._cold_start_handle_error(pid) + + def _ensure_raw_pid_operations_valid(self, pid: int) -> None: + with self._generation_lock: + invalid = self._raw_pid_operations_invalid + if invalid: + raise self._cold_start_handle_error(pid) + + def _invalidate_pre_reload_generation( + self, + pid: _TrackedCommandPid, + ) -> bool: + with self._generation_lock: + if ( + pid.generation >= self._generation + or pid.generation not in self._uncertain_generations + ): + return False + self._invalid_generations.add(pid.generation) + return True + + @staticmethod + def _cold_start_handle_error(pid: int) -> BackendOperationError: + return BackendOperationError(f"process {int(pid)}: {_COLD_START_HANDLE_ERROR}") def run( self, @@ -162,16 +279,17 @@ def run( cwd: str | None, timeout: int, ) -> CommandResult: - try: - value = self._commands.run( - cmd, - envs=dict(envs) if envs is not None else None, - cwd=cwd, - timeout=timeout, - ) - return _command_result(value) - except Exception as error: - raise _convert_error("command execution", error) from error + with self._command_operation(): + try: + value = self._commands.run( + cmd, + envs=dict(envs) if envs is not None else None, + cwd=cwd, + timeout=timeout, + ) + return _command_result(value) + except Exception as error: + raise _convert_error("command execution", error) from error def start( self, @@ -181,46 +299,80 @@ def start( cwd: str | None, stdin: bool, ) -> int: - try: - handle = self._commands.run( - cmd, - background=True, - envs=dict(envs) if envs is not None else None, - cwd=cwd, - stdin=stdin, - ) - except Exception as error: - raise _convert_error("background command start", error) from error - pid = int(handle.pid) - self._handles[pid] = handle - return pid + with self._command_operation(): + generation = self._current_generation() + try: + handle = self._commands.run( + cmd, + background=True, + envs=dict(envs) if envs is not None else None, + cwd=cwd, + stdin=stdin, + ) + except Exception as error: + raise _convert_error("background command start", error) from error + pid = int(handle.pid) + self._handles[pid] = handle + return _TrackedCommandPid(pid, generation, handle) def wait(self, pid: int, timeout: int | None) -> CommandResult: - handle = self._handles.get(pid) - if handle is None: - raise BackendOperationError(f"no command handle for pid {pid}") - try: - return _command_result(handle.wait(timeout)) - except Exception as error: - raise _convert_error(f"wait for process {pid}", error) from error + with self._command_operation(): + tracked = self._tracked_handle(pid) + if tracked is not None: + self._ensure_generation_valid(tracked) + handle = tracked.native_handle + else: + self._ensure_raw_pid_operations_valid(pid) + handle = self._handles.get(pid) + if handle is None: + raise BackendOperationError(f"no command handle for pid {pid}") + try: + return _command_result(handle.wait(timeout)) + except Exception as error: + if tracked is not None and self._invalidate_pre_reload_generation( + tracked + ): + raise self._cold_start_handle_error(pid) from error + raise _convert_error(f"wait for process {pid}", error) from error def kill(self, pid: int) -> bool: - try: - return bool(self._commands.kill(pid)) - except Exception as error: - raise _convert_error(f"kill process {pid}", error) from error + with self._command_operation(): + tracked = self._tracked_handle(pid) + if tracked is not None: + self._ensure_generation_valid(tracked) + else: + self._ensure_raw_pid_operations_valid(pid) + try: + return bool(self._commands.kill(int(pid))) + except Exception as error: + if tracked is not None and self._invalidate_pre_reload_generation( + tracked + ): + raise self._cold_start_handle_error(pid) from error + raise _convert_error(f"kill process {pid}", error) from error def send_stdin(self, pid: int, data: str, eof: bool) -> None: - try: - self._commands.send_stdin(pid, data, eof) - except Exception as error: - raise _convert_error(f"send stdin to process {pid}", error) from error + with self._command_operation(): + tracked = self._tracked_handle(pid) + if tracked is not None: + self._ensure_generation_valid(tracked) + else: + self._ensure_raw_pid_operations_valid(pid) + try: + self._commands.send_stdin(int(pid), data, eof) + except Exception as error: + if tracked is not None and self._invalidate_pre_reload_generation( + tracked + ): + raise self._cold_start_handle_error(pid) from error + raise _convert_error(f"send stdin to process {pid}", error) from error def list(self) -> list[CommandInfo]: - try: - return [_command_info(value) for value in self._commands.list()] - except Exception as error: - raise _convert_error("list processes", error) from error + with self._command_operation(): + try: + return [_command_info(value) for value in self._commands.list()] + except Exception as error: + raise _convert_error("list processes", error) from error class _FilesystemDriver: @@ -337,10 +489,10 @@ def reload(self) -> bool: "The installed openyuanrong-sandbox backend does not support " "sandbox reload. Upgrade it to a version with failover support." ) - try: - return bool(reload_sandbox()) - except Exception: - return False + return self.commands.reload( + reload_sandbox, + lambda: getattr(self._sandbox, "_last_reload_mode", None), + ) def update_network_policy(self, policy: NetworkPolicy | None) -> None: if self._terminated or self._closed: diff --git a/sdk/python/tests/integration/test_sandbox.py b/sdk/python/tests/integration/test_sandbox.py index 87ce95d..fc64dac 100644 --- a/sdk/python/tests/integration/test_sandbox.py +++ b/sdk/python/tests/integration/test_sandbox.py @@ -20,6 +20,7 @@ import unittest from akernel_sdk import HttpReverseTunnel, Sandbox +from akernel_sdk._backends.errors import BackendOperationError _ENABLED = ( os.environ.get("AKERNEL_RUN_INTEGRATION") == "1" @@ -27,6 +28,9 @@ and bool(os.environ.get("AKERNEL_TOKEN")) ) _RUNTIME = os.environ.get("AKERNEL_TEST_RUNTIME", "runsc") +_RECOVERY_ENABLED = ( + _ENABLED and os.environ.get("AKERNEL_RUN_RECOVERY_INTEGRATION") == "1" +) _INSTALL_CURL_COMMAND = ( "apt-get update && " @@ -228,5 +232,46 @@ def test_internal_checkpoint_reload_and_reverse_tunnel(self): server.server_close() +@unittest.skipUnless( + _RECOVERY_ENABLED, + "set AKERNEL_RUN_RECOVERY_INTEGRATION=1 with the SDK environment", +) +class SandboxColdRecoveryIntegrationTest(unittest.TestCase): + def test_reload_without_snapshot_cold_starts_same_logical_sandbox(self): + sandbox = Sandbox( + cpu=1000, + memory=2048, + runtime=_RUNTIME, + failover=True, + ) + try: + logical_id = sandbox.id + facades = (sandbox.commands, sandbox.files, sandbox.pty) + completed = sandbox.commands.run("printf completed-before-cold-start") + sandbox.files.write("/tmp/cold-start-only", "old-runtime") + pending = sandbox.commands.run("sleep 60", background=True) + + self.assertIs(sandbox.reload(), True) + + self.assertEqual(sandbox.id, logical_id) + self.assertEqual((sandbox.commands, sandbox.files, sandbox.pty), facades) + self.assertEqual(completed.stdout, "completed-before-cold-start") + self.assertEqual( + sandbox.commands.run("test ! -e /tmp/cold-start-only").exit_code, + 0, + ) + self.assertEqual( + sandbox.commands.run("printf command-after-cold-start").stdout, + "command-after-cold-start", + ) + with self.assertRaisesRegex( + BackendOperationError, + "pre-reload command handle was not restored after sandbox cold start", + ): + pending.wait(timeout=10) + finally: + sandbox.kill() + + if __name__ == "__main__": unittest.main() diff --git a/sdk/python/tests/unit/test_backends.py b/sdk/python/tests/unit/test_backends.py index 425a55d..8ff5fa5 100644 --- a/sdk/python/tests/unit/test_backends.py +++ b/sdk/python/tests/unit/test_backends.py @@ -13,7 +13,9 @@ # limitations under the License. import os +import threading import unittest +from concurrent.futures import ThreadPoolExecutor from types import MappingProxyType, SimpleNamespace from unittest.mock import MagicMock, patch @@ -30,6 +32,7 @@ InvalidBackendError, UnsupportedBackendFeatureError, ) +from akernel_sdk.commands import Commands as PublicCommands from akernel_sdk.types import ( CommandInfo, CommandResult, @@ -42,6 +45,26 @@ S3Config, ) +_COLD_START_HANDLE_ERROR = ( + "pre-reload command handle was not restored after sandbox cold start" +) +_OPERATION_CONFLICT = "another sandbox operation is in flight" + + +def _blocking_native(result=None, error=None): + entered = threading.Event() + release = threading.Event() + + def operation(*_args, **_kwargs): + entered.set() + if not release.wait(5): + raise TimeoutError("test did not release native operation") + if error is not None: + raise error + return result + + return operation, entered, release + def _spec(**overrides): values = { @@ -144,6 +167,18 @@ def setUp(self): self.addCleanup(self.environment.stop) self.backend = openyuanrong_sandbox.OpenYuanRongSandboxBackend(self.config) + def _create_session(self, native): + with patch.object( + openyuanrong_sandbox.yr_sandbox, + "Sandbox", + return_value=native, + ): + return self.backend.create(_spec()) + + @staticmethod + def _start(session, cmd="sleep 60", *, stdin=False): + return session.commands.start(cmd, envs=None, cwd=None, stdin=stdin) + def test_connection_config_maps_to_yr_environment(self): self.assertEqual(os.environ["YR_SERVER_ADDRESS"], "api.example:443") self.assertEqual(os.environ["YR_TLS"], "1") @@ -231,6 +266,657 @@ def test_old_native_sdk_omits_disabled_failover(self): self.assertNotIn("failover", sandbox_type.call_args.kwargs) + def test_reload_cold_start_success_keeps_native_session_and_facades(self): + native = MagicMock() + native.id = "default-reload" + native.reload.return_value = True + with patch.object( + openyuanrong_sandbox.yr_sandbox, + "Sandbox", + return_value=native, + ): + session = self.backend.create(_spec()) + + native_session = session._sandbox + commands = session.commands + files = session.files + + self.assertIs(session.reload(), True) + self.assertIs(session._sandbox, native_session) + self.assertIs(session.commands, commands) + self.assertIs(session.files, files) + native.reload.assert_called_once_with() + + def test_start_in_flight_makes_reload_fail_without_crossing_generation(self): + native = MagicMock() + native.id = "default-start-boundary" + old_handle = MagicMock(pid=321) + start_native, start_entered, start_release = _blocking_native(old_handle) + native.commands.run.side_effect = start_native + session = self._create_session(native) + + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit( + session.commands.start, + "sleep 60", + envs=None, + cwd=None, + stdin=False, + ) + self.assertTrue(start_entered.wait(1)) + try: + self.assertIs(session.reload(), False) + native.reload.assert_not_called() + finally: + start_release.set() + pid = future.result(timeout=1) + + self.assertEqual(pid.generation, 0) + native.reload.return_value = True + self.assertIs(session.reload(), True) + old_handle.wait.side_effect = RuntimeError("old process missing") + with self.assertRaisesRegex( + BackendOperationError, + _COLD_START_HANDLE_ERROR, + ): + session.commands.wait(pid, 30) + + def test_reload_in_flight_rejects_start_before_native_and_advances_once(self): + native = MagicMock() + native.id = "default-reload-boundary" + reload_native, reload_entered, reload_release = _blocking_native(True) + native.reload.side_effect = reload_native + session = self._create_session(native) + + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(session.reload) + self.assertTrue(reload_entered.wait(1)) + try: + with self.assertRaisesRegex( + BackendOperationError, + _OPERATION_CONFLICT, + ): + self._start(session, "printf new") + native.commands.run.assert_not_called() + finally: + reload_release.set() + self.assertIs(future.result(timeout=1), True) + + new_handle = MagicMock(pid=654) + native.commands.run.return_value = new_handle + new_pid = self._start(session, "printf new") + self.assertEqual(new_pid.generation, 1) + + def test_wait_in_flight_makes_reload_fail_without_invalidating_generation(self): + native = MagicMock() + native.id = "default-wait-boundary" + old_handle = MagicMock(pid=321) + native.commands.run.return_value = old_handle + wait_native, wait_entered, wait_release = _blocking_native( + error=RuntimeError("opaque wait failure") + ) + old_handle.wait.side_effect = wait_native + session = self._create_session(native) + + pid = self._start(session) + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(session.commands.wait, pid, 30) + self.assertTrue(wait_entered.wait(1)) + try: + self.assertIs(session.reload(), False) + native.reload.assert_not_called() + finally: + wait_release.set() + with self.assertRaisesRegex( + BackendOperationError, + "wait for process 321 failed: opaque wait failure", + ): + future.result(timeout=1) + + old_handle.wait.side_effect = None + old_handle.wait.return_value = SimpleNamespace( + stdout="still generation zero\n", + stderr="", + exit_code=0, + ) + self.assertEqual( + session.commands.wait(pid, 30), + CommandResult("still generation zero\n", "", 0), + ) + + def test_wait_allows_concurrent_stdin_and_kill_to_finish_process(self): + native = MagicMock() + native.id = "default-command-readers" + old_handle = MagicMock(pid=321) + native.commands.run.return_value = old_handle + wait_entered = threading.Event() + stdin_called = threading.Event() + kill_called = threading.Event() + process_finished = threading.Event() + + def wait_native(_timeout): + wait_entered.set() + if not process_finished.wait(5): + raise TimeoutError("test did not finish native process") + return SimpleNamespace(stdout="finished\n", stderr="", exit_code=0) + + def send_stdin_native(*_args): + stdin_called.set() + + def kill_native(_pid): + kill_called.set() + process_finished.set() + return True + + old_handle.wait.side_effect = wait_native + native.commands.send_stdin.side_effect = send_stdin_native + native.commands.kill.side_effect = kill_native + session = self._create_session(native) + pid = self._start(session, "cat", stdin=True) + + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(session.commands.wait, pid, None) + self.assertTrue(wait_entered.wait(1)) + try: + try: + session.commands.send_stdin(pid, "input", False) + self.assertIs(session.commands.kill(pid), True) + except BackendOperationError as error: + self.fail(f"concurrent command operation was rejected: {error}") + finally: + process_finished.set() + result = future.result(timeout=1) + + self.assertTrue(stdin_called.is_set()) + self.assertTrue(kill_called.is_set()) + self.assertEqual(result, CommandResult("finished\n", "", 0)) + + def test_reload_exception_clears_in_progress_flag(self): + native = MagicMock() + native.id = "default-reload-exception" + reload_native, reload_entered, reload_release = _blocking_native( + error=RuntimeError("reload failed") + ) + native.reload.side_effect = reload_native + session = self._create_session(native) + + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(session.reload) + self.assertTrue(reload_entered.wait(1)) + try: + with self.assertRaisesRegex( + BackendOperationError, + _OPERATION_CONFLICT, + ): + self._start(session, "printf blocked") + native.commands.run.assert_not_called() + finally: + reload_release.set() + self.assertIs(future.result(timeout=1), False) + + native.commands.run.return_value = MagicMock(pid=654) + pid = self._start(session, "printf after-failure") + self.assertEqual(pid.generation, 0) + native.reload.side_effect = None + native.reload.return_value = True + self.assertIs(session.reload(), True) + + def test_old_handle_native_failure_marks_generation_with_explicit_error(self): + native = MagicMock() + native.id = "default-cold-start-handle" + native.reload.return_value = True + old_handle = MagicMock() + old_handle.pid = 321 + old_handle.wait.side_effect = RuntimeError("opaque native failure") + native.commands.run.return_value = old_handle + with patch.object( + openyuanrong_sandbox.yr_sandbox, + "Sandbox", + return_value=native, + ): + session = self.backend.create(_spec()) + + pid = session.commands.start( + "sleep 60", envs=None, cwd=None, stdin=False + ) + self.assertIs(session.reload(), True) + + with self.assertRaisesRegex( + BackendOperationError, + "pre-reload command handle was not restored after sandbox cold start", + ) as raised: + session.commands.wait(pid, 30) + + self.assertNotIn("opaque native failure", str(raised.exception)) + + def test_invalid_old_generation_fails_closed_without_pid_operations(self): + native = MagicMock() + native.id = "default-cold-start-generation" + native.reload.return_value = True + failed_handle = MagicMock(pid=321) + failed_handle.wait.side_effect = RuntimeError("native process missing") + second_handle = MagicMock(pid=654) + native.commands.run.side_effect = [failed_handle, second_handle] + with patch.object( + openyuanrong_sandbox.yr_sandbox, + "Sandbox", + return_value=native, + ): + session = self.backend.create(_spec()) + + failed_pid = session.commands.start( + "sleep 60", envs=None, cwd=None, stdin=False + ) + second_pid = session.commands.start( + "cat", envs=None, cwd=None, stdin=True + ) + self.assertIs(session.reload(), True) + with self.assertRaises(BackendOperationError): + session.commands.wait(failed_pid, 30) + + with self.assertRaisesRegex( + BackendOperationError, + "pre-reload command handle was not restored after sandbox cold start", + ): + session.commands.kill(second_pid) + with self.assertRaisesRegex( + BackendOperationError, + "pre-reload command handle was not restored after sandbox cold start", + ): + session.commands.send_stdin(second_pid, "input", False) + + native.commands.kill.assert_not_called() + native.commands.send_stdin.assert_not_called() + + def test_new_generation_handle_works_after_old_generation_is_invalid(self): + native = MagicMock() + native.id = "default-new-generation" + native.reload.return_value = True + old_handle = MagicMock(pid=321) + old_handle.wait.side_effect = RuntimeError("native process missing") + new_handle = MagicMock(pid=321) + new_handle.wait.return_value = SimpleNamespace( + stdout="new runtime\n", + stderr="", + exit_code=0, + ) + native.commands.run.side_effect = [old_handle, new_handle] + with patch.object( + openyuanrong_sandbox.yr_sandbox, + "Sandbox", + return_value=native, + ): + session = self.backend.create(_spec()) + + old_pid = session.commands.start( + "sleep 60", envs=None, cwd=None, stdin=False + ) + self.assertIs(session.reload(), True) + with self.assertRaises(BackendOperationError): + session.commands.wait(old_pid, 30) + new_pid = session.commands.start( + "printf new", envs=None, cwd=None, stdin=False + ) + + with self.assertRaisesRegex( + BackendOperationError, + "pre-reload command handle was not restored after sandbox cold start", + ): + session.commands.wait(old_pid, 30) + self.assertEqual( + session.commands.wait(new_pid, 30), + CommandResult("new runtime\n", "", 0), + ) + old_handle.wait.assert_called_once_with(30) + new_handle.wait.assert_called_once_with(30) + + def test_snapshot_like_old_handle_native_success_remains_valid(self): + native = MagicMock() + native.id = "default-snapshot-handle" + native.reload.return_value = True + old_handle = MagicMock(pid=321) + old_handle.wait.return_value = SimpleNamespace( + stdout="restored\n", + stderr="", + exit_code=0, + ) + native.commands.run.return_value = old_handle + with patch.object( + openyuanrong_sandbox.yr_sandbox, + "Sandbox", + return_value=native, + ): + session = self.backend.create(_spec()) + + pid = session.commands.start( + "sleep 1", envs=None, cwd=None, stdin=False + ) + self.assertIs(session.reload(), True) + + self.assertEqual( + session.commands.wait(pid, 30), + CommandResult("restored\n", "", 0), + ) + + def test_cold_start_mode_invalidates_old_pid_before_native_operations(self): + native = MagicMock() + native.id = "default-authoritative-cold-start" + + def reload_native(): + native._last_reload_mode = "cold-start" + return True + + native.reload.side_effect = reload_native + native.commands.run.return_value = MagicMock(pid=321) + native.commands.kill.return_value = True + session = self._create_session(native) + pid = self._start(session, "cat", stdin=True) + + self.assertIs(session.reload(), True) + + with self.assertRaisesRegex(BackendOperationError, _COLD_START_HANDLE_ERROR): + session.commands.kill(pid) + with self.assertRaisesRegex(BackendOperationError, _COLD_START_HANDLE_ERROR): + session.commands.send_stdin(pid, "input", False) + native.commands.kill.assert_not_called() + native.commands.send_stdin.assert_not_called() + + def test_cold_start_mode_rejects_plain_pid_before_native_operations(self): + native = MagicMock() + native.id = "default-authoritative-cold-start-plain-pid" + + def reload_native(): + native._last_reload_mode = "cold-start" + return True + + native.reload.side_effect = reload_native + old_handle = MagicMock(pid=321) + old_handle.wait.return_value = SimpleNamespace( + stdout="wrong runtime\n", stderr="", exit_code=0 + ) + native.commands.run.return_value = old_handle + native.commands.kill.return_value = True + session = self._create_session(native) + pid = self._start(session, "cat", stdin=True) + + self.assertIs(session.reload(), True) + + plain_pid = int(pid) + with self.assertRaisesRegex(BackendOperationError, _COLD_START_HANDLE_ERROR): + session.commands.wait(plain_pid, 30) + with self.assertRaisesRegex(BackendOperationError, _COLD_START_HANDLE_ERROR): + session.commands.kill(plain_pid) + with self.assertRaisesRegex(BackendOperationError, _COLD_START_HANDLE_ERROR): + session.commands.send_stdin(plain_pid, "input", False) + old_handle.wait.assert_not_called() + native.commands.kill.assert_not_called() + native.commands.send_stdin.assert_not_called() + + def test_plain_pid_remains_fail_closed_for_new_generation_after_cold_start(self): + native = MagicMock() + native.id = "default-new-generation-plain-pid" + + def reload_native(): + native._last_reload_mode = "cold-start" + return True + + native.reload.side_effect = reload_native + old_handle = MagicMock(pid=321) + new_handle = MagicMock(pid=321) + new_handle.wait.return_value = SimpleNamespace( + stdout="new runtime\n", stderr="", exit_code=0 + ) + native.commands.run.side_effect = [old_handle, new_handle] + session = self._create_session(native) + self._start(session) + self.assertIs(session.reload(), True) + new_pid = self._start(session, "printf new") + + self.assertEqual( + session.commands.wait(new_pid, 30), + CommandResult("new runtime\n", "", 0), + ) + with self.assertRaisesRegex(BackendOperationError, _COLD_START_HANDLE_ERROR): + session.commands.wait(int(new_pid), 30) + new_handle.wait.assert_called_once_with(30) + + def test_snapshot_mode_keeps_old_pid_native_operations_available(self): + native = MagicMock() + native.id = "default-authoritative-snapshot" + + def reload_native(): + native._last_reload_mode = "snapshot" + return True + + native.reload.side_effect = reload_native + native.commands.run.return_value = MagicMock(pid=321) + native.commands.kill.return_value = True + session = self._create_session(native) + pid = self._start(session, "cat", stdin=True) + + self.assertIs(session.reload(), True) + + self.assertIs(session.commands.kill(pid), True) + session.commands.send_stdin(pid, "input", False) + native.commands.kill.assert_called_once_with(321) + native.commands.send_stdin.assert_called_once_with(321, "input", False) + + def test_snapshot_mode_keeps_plain_pid_operations_available(self): + native = MagicMock() + native.id = "default-authoritative-snapshot-plain-pid" + + def reload_native(): + native._last_reload_mode = "snapshot" + return True + + native.reload.side_effect = reload_native + native.commands.run.return_value = MagicMock(pid=321) + native.commands.kill.return_value = True + session = self._create_session(native) + pid = self._start(session) + + self.assertIs(session.reload(), True) + + self.assertIs(session.commands.kill(int(pid)), True) + native.commands.kill.assert_called_once_with(321) + + def test_snapshot_mode_native_error_does_not_invalidate_generation(self): + native = MagicMock() + native.id = "default-authoritative-snapshot-native-error" + + def reload_native(): + native._last_reload_mode = "snapshot" + return True + + native.reload.side_effect = reload_native + handle = MagicMock(pid=321) + handle.wait.side_effect = [ + RuntimeError("transient native failure"), + SimpleNamespace(stdout="restored\n", stderr="", exit_code=0), + ] + native.commands.run.return_value = handle + session = self._create_session(native) + pid = self._start(session) + + self.assertIs(session.reload(), True) + + with self.assertRaisesRegex( + BackendOperationError, + "wait for process 321 failed: transient native failure", + ) as raised: + session.commands.wait(pid, 30) + self.assertNotIn(_COLD_START_HANDLE_ERROR, str(raised.exception)) + self.assertEqual( + session.commands.wait(pid, 30), + CommandResult("restored\n", "", 0), + ) + + def test_old_native_sdk_without_reload_mode_keeps_conservative_behavior(self): + commands = MagicMock() + commands.run.return_value = MagicMock(pid=321) + commands.kill.return_value = True + native = SimpleNamespace( + id="default-old-sdk", + commands=commands, + files=MagicMock(), + reload=MagicMock(return_value=True), + ) + session = self._create_session(native) + pid = self._start(session) + + self.assertIs(session.reload(), True) + + self.assertIs(session.commands.kill(pid), True) + commands.kill.assert_called_once_with(321) + + def test_reload_false_does_not_advance_or_invalidate_handle_generation(self): + native = MagicMock() + native.id = "default-failed-reload-handle" + native.reload.return_value = False + old_handle = MagicMock(pid=321) + old_handle.wait.return_value = SimpleNamespace( + stdout="still running\n", + stderr="", + exit_code=0, + ) + native.commands.run.return_value = old_handle + with patch.object( + openyuanrong_sandbox.yr_sandbox, + "Sandbox", + return_value=native, + ): + session = self.backend.create(_spec()) + + pid = session.commands.start( + "sleep 1", envs=None, cwd=None, stdin=False + ) + self.assertIs(session.reload(), False) + + self.assertEqual( + session.commands.wait(pid, 30), + CommandResult("still running\n", "", 0), + ) + + def test_public_command_handle_preserves_tracked_pid_token(self): + native = MagicMock() + native.id = "default-public-token" + native.reload.return_value = True + old_handle = MagicMock(pid=321) + old_handle.wait.side_effect = RuntimeError("native process missing") + native.commands.run.return_value = old_handle + session = self._create_session(native) + + handle = PublicCommands(session.commands).run( + "sleep 60", background=True + ) + self.assertIsInstance(handle.pid, int) + self.assertIsNot(type(handle.pid), int) + self.assertIs(session.reload(), True) + + with self.assertRaisesRegex( + BackendOperationError, + _COLD_START_HANDLE_ERROR, + ): + handle.wait(30) + + def test_explicit_int_pid_uses_legacy_path_without_generation_claim(self): + native = MagicMock() + native.id = "default-legacy-pid" + native.reload.return_value = True + old_handle = MagicMock(pid=321) + old_handle.wait.side_effect = RuntimeError("opaque legacy failure") + native.commands.run.return_value = old_handle + session = self._create_session(native) + + tracked_pid = self._start(session) + self.assertIs(session.reload(), True) + + with self.assertRaisesRegex( + BackendOperationError, + "wait for process 321 failed: opaque legacy failure", + ) as raised: + session.commands.wait(int(tracked_pid), 30) + self.assertNotIn("not restored after sandbox cold start", str(raised.exception)) + + def test_old_generation_kill_false_does_not_invalidate(self): + native = MagicMock() + native.id = "default-old-kill-false" + native.reload.return_value = True + first_handle = MagicMock(pid=321) + second_handle = MagicMock(pid=654) + second_handle.wait.return_value = SimpleNamespace( + stdout="snapshot preserved\n", + stderr="", + exit_code=0, + ) + native.commands.run.side_effect = [first_handle, second_handle] + native.commands.kill.return_value = False + session = self._create_session(native) + + first_pid = self._start(session) + second_pid = self._start(session, "printf restored") + self.assertIs(session.reload(), True) + + self.assertIs(session.commands.kill(first_pid), False) + self.assertEqual( + session.commands.wait(second_pid, 30), + CommandResult("snapshot preserved\n", "", 0), + ) + + def test_old_generation_kill_exception_invalidates_other_handles(self): + native = MagicMock() + native.id = "default-old-kill-exception" + native.reload.return_value = True + first_handle = MagicMock(pid=321) + second_handle = MagicMock(pid=654) + native.commands.run.side_effect = [first_handle, second_handle] + native.commands.kill.side_effect = RuntimeError("opaque kill failure") + session = self._create_session(native) + + first_pid = self._start(session) + second_pid = self._start(session) + self.assertIs(session.reload(), True) + + with self.assertRaisesRegex( + BackendOperationError, + _COLD_START_HANDLE_ERROR, + ): + session.commands.kill(first_pid) + with self.assertRaisesRegex( + BackendOperationError, + _COLD_START_HANDLE_ERROR, + ): + session.commands.wait(second_pid, 30) + second_handle.wait.assert_not_called() + + def test_old_generation_stdin_exception_invalidates_other_handles(self): + native = MagicMock() + native.id = "default-old-stdin-exception" + native.reload.return_value = True + first_handle = MagicMock(pid=321) + second_handle = MagicMock(pid=654) + native.commands.run.side_effect = [first_handle, second_handle] + native.commands.send_stdin.side_effect = RuntimeError( + "opaque stdin failure" + ) + session = self._create_session(native) + + first_pid = self._start(session, "cat", stdin=True) + second_pid = self._start(session) + self.assertIs(session.reload(), True) + + with self.assertRaisesRegex( + BackendOperationError, + _COLD_START_HANDLE_ERROR, + ): + session.commands.send_stdin(first_pid, "input", False) + with self.assertRaisesRegex( + BackendOperationError, + _COLD_START_HANDLE_ERROR, + ): + session.commands.wait(second_pid, 30) + second_handle.wait.assert_not_called() + def test_old_native_sdk_rejects_enabled_failover(self): with ( patch.object( diff --git a/sdk/python/tests/unit/test_sandbox.py b/sdk/python/tests/unit/test_sandbox.py index 5ba7508..675dda7 100644 --- a/sdk/python/tests/unit/test_sandbox.py +++ b/sdk/python/tests/unit/test_sandbox.py @@ -33,7 +33,7 @@ from akernel_sdk._dockercontext import LocalDockerContext from akernel_sdk._dockerfile import DockerfileBuildError, DockerfileParseError from akernel_sdk._dockerfile_runner import DockerfileApplyResult -from akernel_sdk.types import SandboxInfo +from akernel_sdk.types import CommandResult, SandboxInfo class SandboxTest(unittest.TestCase): @@ -84,6 +84,53 @@ def test_default_constructor_and_info(self): self.session.terminate.assert_called_once_with() self.session.close.assert_called_once_with() + def test_failover_is_typed_and_forwarded(self): + sandbox = Sandbox(failover=True) + spec = self.backend.create.call_args.args[0] + + self.assertTrue(spec.failover) + sandbox.kill() + + def test_failover_rejects_non_boolean_values(self): + with self.assertRaisesRegex(TypeError, "failover"): + Sandbox(failover=1) + self.backend.create.assert_not_called() + + def test_reload_cold_start_success_returns_true_without_replacing_facades(self): + self.session.reload.return_value = True + sandbox = Sandbox() + before = (sandbox.commands, sandbox.files, sandbox.pty, sandbox._session) + + self.assertIs(sandbox.reload(), True) + for current, original in zip( + (sandbox.commands, sandbox.files, sandbox.pty, sandbox._session), + before, + strict=True, + ): + self.assertIs(current, original) + self.session.reload.assert_called_once_with() + + def test_completed_command_result_stays_readable_after_reload(self): + completed = CommandResult("completed\n", "", 0) + self.session.commands.run.return_value = completed + self.session.reload.return_value = True + sandbox = Sandbox() + + result = sandbox.commands.run("printf completed") + self.assertIs(sandbox.reload(), True) + + self.assertIs(result, completed) + self.assertEqual(result.stdout, "completed\n") + self.assertEqual(result.stderr, "") + self.assertEqual(result.exit_code, 0) + + def test_reload_returns_false_after_close(self): + sandbox = Sandbox() + sandbox.kill() + + self.assertIs(sandbox.reload(), False) + self.session.reload.assert_not_called() + def test_extra_config_is_validated_and_defensively_copied(self): labels = ["worker"] requested = {"featureFlag": True, "nested": {"labels": labels}}