//// A kernel or its owner going away must still end every process the kernel //// started, races beyond normal E2E teardown. Split from supervision_test so //// the two run side by side. import albedo/daemon/store import albedo/harness/extensions/python/kernel as python import gleam/erlang/process import gleam/int import gleeunit/should import harness/kernel_fixture /// Cells may not spawn processes themselves, but a library they call may: /// this defines one, `spawn`, compiled outside any cell. const library_spawn = "library = {}\nexec(compile('import subprocess\\ndef spawn(*args, **kwargs):\\n return subprocess.Popen(*args, **kwargs)', 'fixture_library.py', 'exec'), library)\nspawn = library['spawn']\n" pub fn owner_death_still_ends_reported_job_groups_test() -> Nil { let assert Ok(store) = store.start(":memory:", "") let assert Ok(kernel) = kernel_fixture.local(store, "/tmp") let assert Ok(started) = python.execute( kernel, "a", "late = run('sleep', '30', timeout=300)\nimport asyncio\nawait asyncio.sleep(0.3)\nlate.group.pgid", 10_000, ) let assert Ok(pgid) = int.parse(started.value) // Kill the kernel from inside a cell: no cleanup callback can run, so the // supervisor's own record of the job group is the only remaining owner. let assert Error(python.Lost) = python.execute( kernel, "b", "import os, signal\nos.kill(os.getpid(), signal.SIGKILL)", 5000, ) let assert Ok(probe) = kernel_fixture.local(store, "/tmp") let assert Ok(checked) = python.execute( probe, "c", "import os\nstate = 'alive'\ntry:\n os.killpg(" <> int.to_string(pgid) <> ", 0)\nexcept ProcessLookupError:\n state = 'gone'\nstate", 5000, ) checked.value |> should.equal("'gone'") let assert Ok(_) = python.stop(probe) store.close(store) } pub fn kernel_death_ends_unregistered_children_in_its_own_group_test() -> Nil { let assert Ok(store) = store.start(":memory:", "") let assert Ok(kernel) = kernel_fixture.local(store, "/tmp") let assert Ok(started) = python.execute(kernel, "start", library_spawn <> "import subprocess child = spawn(['sleep', '30'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) child.pid", 5000) let assert Ok(pid) = int.parse(started.value) let assert Error(python.Lost) = python.execute( kernel, "die", "import os, signal os.kill(os.getpid(), signal.SIGKILL)", 5000, ) let assert Ok(probe) = kernel_fixture.local(store, "/tmp") let assert Ok(checked) = python.execute(probe, "check", "import os state = 'alive' try: os.kill(" <> int.to_string(pid) <> ", 0) except ProcessLookupError: state = 'gone' state", 5000) checked.value |> should.equal("'gone'") let assert Ok(_) = python.stop(probe) store.close(store) } pub fn shutdown_keeps_ownership_of_late_job_registrations_test() -> Nil { let assert Ok(store) = store.start(":memory:", "") let assert Ok(kernel) = kernel_fixture.local(store, "/tmp") let assert Ok(path) = python.execute( kernel, "start", library_spawn <> "import __main__ as kernel, os, subprocess, tempfile from pathlib import Path fd, report = tempfile.mkstemp(prefix='albedo-owned-shutdown-') os.close(fd) def late(): child = spawn(['sleep', '30'], start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) Path(report).write_text(str(child.pid)) kernel.send({'type': 'job_start', 'id': 'late-fixture', 'pgid': child.pid}) os._exit(0) kernel.CLEANUP.append(late) report", 5000, ) let assert Ok(_) = python.stop(kernel) let assert Ok(probe) = kernel_fixture.local(store, "/tmp") let assert Ok(checked) = python.execute(probe, "check", "import os from pathlib import Path report = Path(" <> path.value <> ") pid = int(report.read_text()) report.unlink() state = 'alive' try: os.kill(pid, 0) except ProcessLookupError: state = 'gone' state", 5000) checked.value |> should.equal("'gone'") let assert Ok(_) = python.stop(probe) store.close(store) } pub fn owner_loss_terminates_a_kernel_held_in_native_code_test() -> Nil { let assert Ok(store) = store.start(":memory:", "") let assert Ok(kernel) = kernel_fixture.local(store, "/tmp") let assert Ok(started) = python.execute( kernel, "pid", "import os os.getpid()", 5000, ) let assert Ok(pid) = int.parse(started.value) let reply = process.new_subject() let _ = process.spawn_unlinked(fn() { process.send( reply, python.execute( kernel, "native", "import ctypes ctypes.PyDLL(None).sleep(30)", 60_000, ), ) }) process.sleep(100) store.close(store) process.receive(reply, 8000) |> should.equal(Ok(Error(python.Lost))) let assert Ok(other) = store.start(":memory:", "") let assert Ok(probe) = kernel_fixture.local(other, "/tmp") let assert Ok(checked) = python.execute(probe, "check", "import os state = 'alive' try: os.kill(" <> int.to_string(pid) <> ", 0) except ProcessLookupError: state = 'gone' state", 5000) checked.value |> should.equal("'gone'") let assert Ok(_) = python.stop(probe) store.close(other) }