test(daemon): cover lifecycle races and recovery
Exercise real startup, cancellation, restart approval, and active-work recovery. Wait for the tool before testing a crash.
517 unmodified linesExpand all5185195205215225235245255265275284 unmodified lines53353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558646 unmodified lines633634635636637638639640641642643644645646647648358 unmodified linesExpand all517 unmodified linesExpand all self.connection = None self.base = None self._pid = None self._local = threading.local() self.boot_action = None self._launchers = [] self._commands = [] with _daemons_lock: _daemons.append(self)
@property4 unmodified lines """Start the daemon; the first fixture boots it after its prepare.""" extensions = self.home / "extensions.json" if not extensions.exists(): extensions.write_text(json.dumps(OFFLINE)) if self.boot_action is not None: self.boot_action(self) self.cli("sessions") self._refresh()
def start_launcher(self): """Run the real launcher directly, retaining its log and process owner.""" with (self.home / "daemon.log").open("ab") as log: process = subprocess.Popen( [self.env["ALBEDO_DAEMON"]], cwd=ROOT, env=self.env, stdout=log, stderr=log, ) self._launchers.append(process) # Direct launches are our children. Reap them as soon as they exit so # another CLI can observe that the old daemon is gone during upgrade. threading.Thread(target=process.wait, daemon=True).start() return process
def await_ready(self, timeout=20): """Adopt the published endpoint once its authenticated API answers.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: self._refresh() with self.api("/health") as response: if json.load(response).get("ok"): return except (OSError, ValueError, http.client.HTTPException): pass time.sleep(0.02) self._fail("launcher did not publish an authenticated endpoint")
def start_cli(self, *args, **streams): """Start a CLI command whose input or cancellation the fixture controls.""" process = subprocess.Popen( [str(ROOT / "cli/bin/albedo"), *args], cwd=ROOT, env=self.env, **streams, ) self._commands.append(process) return process
def _refresh(self): global daemon_boots self.connection = json.loads((self.home / "daemon.json").read_text()) self.base = f"http://127.0.0.1:{self.connection['port']}"46 unmodified lines if not self._stop(timeout=10): assert self._pid is not None os.kill(self._pid, signal.SIGKILL) self._pid = None for launcher in self._launchers: if launcher.poll() is None: launcher.kill() launcher.wait() for command in self._commands: if command.poll() is None: command.kill() command.wait() shutil.rmtree(self.root, ignore_errors=True) with _daemons_lock: if self in _daemons: _daemons.remove(self)358 unmodified linesExpand all47 unmodified lines484950515253545556525758596061832 unmodified linesExpand all89489589689789889990089390190290390490590697 unmodified lines47 unmodified lines "Path('example.txt').write_text('hello\\n')\n" "saved = Path('example.txt').read_text()\n" "assert 'ALBEDO_API_KEY' not in os.environ\n" "assert 'ALBEDO_TOKEN' not in os.environ\n" + ( "Path('recovery-started').write_text('ready')\n" if "hang then recover" in user else "" ) f"await asyncio.sleep({30 if 'hang' in user else 2 if 'first task' in user else 0.4})\nlen(saved)" + f"await asyncio.sleep({30 if 'hang' in user else 2 if 'first task' in user else 0.4})\nlen(saved)" ) return ( text("finished") if tools832 unmodified linesExpand all any(e.get("type") == "user" for e in app.events(second)) ) third = app.session() app.prompt(third, "hang then recover").close() ready = app.workspace / "recovery-started" deadline = time.monotonic() + 15 while not ready.exists() and time.monotonic() < deadline: time.sleep(0.5) time.sleep(0.05) self.assertTrue(ready.exists(), "recovery tool did not start") stamp = (app.workspace / "example.txt").stat().st_mtime_ns self.settings( app, protocol, active="alpha", beta="changed-beta-default" )97 unmodified lines123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382"""CLI attachment and startup must preserve an independently owned real daemon."""
import jsonimport osimport ptyimport selectimport signalimport socketimport statimport subprocessimport sysimport termiosimport timeimport unittest
from harness import Albedo, Provider, exclusive, operation_id, python, text
def direct_start(daemon): daemon.start_launcher() daemon.await_ready()
def runtime_limits(app): cookie = (app.home / "inspect.cookie").read_text().strip() node = f"albedo_{app.daemon._pid}@{socket.gethostname().split('.')[0]}" expression = ( f"Node = '{node}', " "Limits = [rpc:call(Node, erlang, system_info, [Key]) " "|| Key <- [process_limit, port_limit, schedulers_online]], " "io:put_chars(json:encode(Limits)), halt()." ) result = subprocess.run( [ "erl", "+S", "2:2", "-sname", f"lifecycle_probe_{app.daemon._pid}", "-setcookie", cookie, "-noshell", "-eval", expression, ], capture_output=True, text=True, timeout=20, check=True, ) return json.loads(result.stdout)
def gated_launcher(app): """A pipe barrier signals the real launch attempt before letting it exec.""" original = app.env["ALBEDO_DAEMON"] started, release = app.root / "launcher-started", app.root / "launcher-release" os.mkfifo(started) os.mkfifo(release) attempts = app.root / "launcher-attempts" wrapper = app.root / "gated-launcher" wrapper.write_text( f"#!{sys.executable}\n" "import os\n" f"with open({str(attempts)!r}, 'a') as attempts:\n" " attempts.write(str(os.getpid()) + '\\n')\n" " attempts.flush()\n" f"with open({str(started)!r}, 'w') as ready:\n" " ready.write(str(os.getpid()) + '\\n')\n" f"with open({str(release)!r}, 'rb', buffering=0) as gate:\n" " gate.read(1)\n" f"os.execv({original!r}, [{original!r}])\n" ) wrapper.chmod(0o700) app.env["ALBEDO_DAEMON"] = str(wrapper) return started, release, attempts
class TerminalCommand: def __init__(self, app, *arguments): self.master, slave = pty.openpty() termios.tcsetwinsize(slave, (24, 100)) self.process = app.daemon.start_cli( *arguments, stdin=slave, stdout=slave, stderr=slave ) os.close(slave) self.output = b""
def prompt(self, timeout=10): deadline = time.monotonic() + timeout while time.monotonic() < deadline: readable, _, _ = select.select( [self.master], [], [], max(0, deadline - time.monotonic()) ) if readable: try: chunk = os.read(self.master, 65536) except OSError as error: raise AssertionError( "CLI exited before restart prompt: " + self.output.decode(errors="replace") ) from error self.output += chunk if b"\x1b[6n" in chunk: os.write(self.master, b"\x1b[1;1R") if b"\x1b[c" in chunk: os.write(self.master, b"\x1b[?1;2c") if b"[y/N]" in self.output or b"[Y/n]" in self.output: return raise AssertionError( "CLI did not offer a restart: " + self.output.decode(errors="replace") )
def answer(self, value): os.write(self.master, (value + "\r").encode()) result = self.process.wait(timeout=20) os.close(self.master) return result
@exclusiveclass LifecycleTest(unittest.TestCase): def test_standalone_default_home_authentication_and_vm_defaults_match_cli(self): def prepare(app): app.home = app.daemon.home = app.root / "user-home" / ".albedo" app.home.mkdir(mode=0o755) app.env.pop("ALBEDO_HOME") app.env.pop("ALBEDO_TOKEN", None) app.env["ALBEDO_INSPECT"] = "1" app.daemon.boot_action = direct_start
with Albedo(prepare=prepare) as app: direct = dict(app.connection) self.assertGreaterEqual(len(direct["token"]), 32) self.assertEqual(stat.S_IMODE(app.home.stat().st_mode), 0o700) self.assertEqual( stat.S_IMODE((app.home / "daemon.json").stat().st_mode), 0o600 ) connection = socket.create_connection( ("127.0.0.1", direct["port"]), timeout=5 ) with connection: connection.sendall( b"GET /health HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n" ) self.assertIn(b"403", connection.recv(1024).split(b"\r\n", 1)[0]) direct_limits = runtime_limits(app) session = app.session() app.prompt(session, "independent daemon is usable").close() app.idle(session) self.assertIn("ok", [event.get("text") for event in app.events(session)]) self.assertEqual(app.connection, direct) app.restart() self.assertNotEqual(app.connection["pid"], direct["pid"]) self.assertNotEqual(app.connection["token"], direct["token"]) self.assertEqual(runtime_limits(app), direct_limits) app.restart( prepare=lambda fixture: fixture.env.update( ERL_FLAGS=fixture.env["ERL_FLAGS"] + " +P 131072 +Q 32768", ALBEDO_TOKEN="operator-chosen-authentication-token", ) ) self.assertEqual(runtime_limits(app)[:2], [131072, 32768]) self.assertEqual( app.connection["token"], "operator-chosen-authentication-token" ) self.assertEqual(app.api("/health").status, 200)
def test_cancelling_cli_startup_wait_leaves_the_started_daemon_owned(self): captured = {}
def prepare(app): started, release, attempts = gated_launcher(app)
def start(daemon): command = daemon.start_cli( "daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE ) with started.open() as ready: captured["child"] = int(ready.readline()) adopted = False try: command.send_signal(signal.SIGINT) stdout, stderr = command.communicate(timeout=5) captured["cancelled"] = command.returncode captured["output"] = stdout + stderr with release.open("wb", buffering=0) as gate: gate.write(b"x") daemon.await_ready() adopted = True finally: if not adopted: try: os.kill(captured["child"], signal.SIGKILL) except ProcessLookupError: pass captured["attempts"] = attempts.read_text().splitlines()
app.daemon.boot_action = start
with Albedo(prepare=prepare) as app: self.assertNotEqual(captured["cancelled"], 0, captured["output"]) self.assertEqual(app.connection["pid"], captured["child"]) self.assertEqual(len(captured["attempts"]), 1) session = app.session() app.prompt(session, "daemon survived waiter cancellation").close() app.idle(session) self.assertIn("ok", [event.get("text") for event in app.events(session)])
def test_simultaneous_cli_starters_converge_on_one_real_execution(self): captured = {}
def prepare(app): started, release, attempts = gated_launcher(app)
def start(daemon): first = daemon.start_cli( "daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE ) second = daemon.start_cli( "daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE ) with started.open() as ready: captured["owner"] = int(ready.readline()) adopted = False try: with release.open("wb", buffering=0) as gate: gate.write(b"x") captured["outputs"] = [ command.communicate(timeout=20) for command in (first, second) ] captured["codes"] = [first.returncode, second.returncode] daemon.await_ready() adopted = True finally: if not adopted: try: os.kill(captured["owner"], signal.SIGKILL) except ProcessLookupError: pass captured["attempts"] = attempts.read_text().splitlines()
app.daemon.boot_action = start
with Albedo(prepare=prepare) as app: self.assertEqual(captured["codes"], [0, 0], captured["outputs"]) self.assertEqual(captured["attempts"], [str(app.connection["pid"])]) self.assertEqual(app.connection["pid"], captured["owner"]) self.assertEqual(app.api("/sessions").status, 200)
def test_compatible_other_build_is_kept_without_a_terminal_and_explicit_yes_restarts( self, ): def prepare(app): app.env["ALBEDO_BUILD"] = "independent-compatible-build" app.daemon.boot_action = direct_start
with Albedo(prepare=prepare) as app: previous = dict(app.connection) self.assertEqual(previous["build"], "independent-compatible-build") app.env.pop("ALBEDO_BUILD") app.cli("daemon") self.assertEqual( json.loads((app.home / "daemon.json").read_text()), previous ) terminal = TerminalCommand(app, "daemon") terminal.prompt() self.assertEqual(terminal.answer(""), 0) self.assertEqual( json.loads((app.home / "daemon.json").read_text()), previous ) for cancel in ("\x03", "\x04"): terminal = TerminalCommand(app, "daemon") terminal.prompt() self.assertNotEqual(terminal.answer(cancel), 0) self.assertEqual( json.loads((app.home / "daemon.json").read_text()), previous ) terminal = TerminalCommand(app, "daemon") terminal.prompt() self.assertEqual(terminal.answer("y"), 0) app.daemon._refresh() self.assertNotEqual(app.connection["pid"], previous["pid"]) self.assertNotEqual(app.connection["token"], previous["token"])
def test_restart_confirmation_cannot_replace_a_newer_daemon(self): with Albedo() as app: terminal = TerminalCommand(app, "daemon") terminal.prompt() app.restart() replacement = dict(app.connection) self.assertNotEqual(terminal.answer("y"), 0) self.assertEqual( json.loads((app.home / "daemon.json").read_text()), replacement ) self.assertEqual(app.api("/health").status, 200)
def test_explicit_upgrade_recovers_history_receipt_and_uncertain_tool_without_repeating_effect( self, ): fixture = {}
def script(request): messages = request["messages"] if any( "daemon restart" in str(message.get("content", "")) for message in messages ): return text("recovered after upgrade") if messages[-1]["role"] == "tool": return text("tool finished") return python( "from pathlib import Path\nimport time\n" f"with Path({str(fixture['effect'])!r}).open('a') as effect:\n" " effect.write('effect applied\\n')\n" " effect.flush()\n" "time.sleep(60)" )
provider = Provider(script) self.addCleanup(provider.close) with Albedo(provider) as app: fixture["effect"] = app.workspace / "effect" session = app.session() operation = operation_id() app.api( f"/sessions/{session}/events", {"operationId": operation, "content": "apply an effect then wait"}, ).close() deadline = time.monotonic() + 20 while ( not fixture["effect"].exists() or fixture["effect"].read_text() != "effect applied\n" ): if time.monotonic() > deadline: app._fail("tool never published its external effect") time.sleep(0.02) self.assertTrue( json.load(app.api(f"/sessions/{session}/status"))["running"] ) previous = dict(app.connection) terminal = TerminalCommand(app, "daemon") terminal.prompt() self.assertEqual(terminal.answer("y"), 0) app.daemon._refresh() self.assertNotEqual(app.connection["pid"], previous["pid"]) app.idle(session) receipt = json.load(app.api(f"/operations/{operation}")) self.assertEqual(receipt["deliveryStatus"], "committed") events = app.events(session) self.assertEqual( len( [event for event in events if event.get("operationId") == operation] ), 1, ) self.assertTrue( any(event.get("source") == "daemon restart" for event in events) ) self.assertTrue( any(event.get("text") == "recovered after upgrade" for event in events) ) self.assertEqual(fixture["effect"].read_text(), "effect applied\n")
def test_failed_discovery_publication_exits_and_releases_home_for_retry(self): captured = {}
def prepare(app): def start(daemon): blocked = app.home / "daemon.json" blocked.mkdir() process = daemon.start_launcher() captured["exit"] = process.wait(timeout=20) captured["pid"] = process.pid blocked.rmdir()
app.daemon.boot_action = start
with Albedo(prepare=prepare) as app: self.assertNotEqual(captured["exit"], 0) self.assertNotEqual(app.connection["pid"], captured["pid"]) self.assertEqual(app.api("/health").status, 200)