okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffaufrefactor(runtime): decouple core storage from work9dd3a56

main
branchesmainall refs →
4.0 KiB99 linesraw
1//// Process-group termination must clean up descendants and races beyond normal E2E teardown.
2//// Kernel-level supervision: owned process groups, quota accounting, verified stops.
3
4import albedo/daemon/store
5import albedo/harness/extensions/python/kernel as python
6import gleam/int
7import gleeunit/should
8import harness/kernel_fixture
9
10/// A group and the descendant that outlived the command must become
11/// absent. Permission errors do not prove termination.
12const reaped_probe =
13  "import os, time\nstate = 'alive'\nfor _ in range(50):\n"
14  <> "    reachable = 0\n"
15  <> "    for probe in (lambda: os.killpg(job.group.pgid, 0), lambda: os.kill(grandchild, 0)):\n"
16  <> "        try:\n            probe()\n            reachable += 1\n"
17  <> "        except ProcessLookupError:\n            pass\n"
18  <> "    if reachable == 0:\n        state = 'gone'\n        break\n"
19  <> "    time.sleep(0.02)\n"
20
21/// Parent programs for a job, as `program`: each prints the pid of a sleeping
22/// child that stays in the job's group, then waits for it or leaves it behind.
23const waits_on_a_sleeper =
24  "program = 'import subprocess; child = subprocess.Popen([\"sleep\", \"30\"]); print(child.pid, flush=True); child.wait()'\n"
25
26const leaves_a_sleeper =
27  "program = 'import subprocess; child = subprocess.Popen([\"sleep\", \"30\"]); print(child.pid, flush=True)'\n"
28
29pub fn deadline_ends_the_group_and_reports_it_test() -> Nil {
30  let assert Ok(store) = store.start(":memory:", "")
31  let assert Ok(kernel) = kernel_fixture.local(store, "/tmp")
32  let assert Ok(outcome) =
33    python.execute(
34      kernel,
35      "a",
36      waits_on_a_sleeper
37        // The deadline must outlast the parent's startup, or the job ends
38        // before the grandchild it is meant to take down exists.
39        <> "import os, sys\njob = run(sys.executable, '-c', program, timeout=1)\nawait job\n"
40        <> "grandchild = int(job.tail().split()[0])\n"
41        <> reaped_probe
42        <> "(job.timed_out, job.termination.gone, state, job.poll() < 0, "
43        <> "'deadline exceeded after 1s' in job.tail(), 'terminated' in job.tail())",
44      10_000,
45    )
46  outcome.status |> should.equal(python.Succeeded)
47  outcome.value |> should.equal("(True, True, 'gone', True, True, True)")
48  let assert Ok(_) = python.stop(kernel)
49  store.close(store)
50}
51
52pub fn command_exit_ends_a_descendant_that_holds_output_test() -> Nil {
53  let assert Ok(store) = store.start(":memory:", "")
54  let assert Ok(kernel) = kernel_fixture.local(store, "/tmp")
55  let assert Ok(outcome) =
56    python.execute(
57      kernel,
58      "a",
59      leaves_a_sleeper
60        <> "import os, sys, time\nstarted = time.monotonic()\n"
61        <> "job = run(sys.executable, '-c', program, timeout=30)\nawait job\n"
62        <> "grandchild = int(job.tail().split()[0])\n"
63        <> reaped_probe
64        <> "(job.timed_out, job.termination.gone, state, time.monotonic() - started < 5)",
65      10_000,
66    )
67  outcome.status |> should.equal(python.Succeeded)
68  outcome.value |> should.equal("(False, True, 'gone', True)")
69  let assert Ok(_) = python.stop(kernel)
70  store.close(store)
71}
72
73pub fn stop_ends_a_running_job_group_test() -> Nil {
74  let assert Ok(store) = store.start(":memory:", "")
75  let assert Ok(kernel) = kernel_fixture.local(store, "/tmp")
76  let assert Ok(started) =
77    python.execute(
78      kernel,
79      "a",
80      "import asyncio\nlate = run('sleep', '30', timeout=300)\nawait asyncio.sleep(0.3)\nlate.group.pgid",
81      10_000,
82    )
83  let assert Ok(pgid) = int.parse(started.value)
84  // The typed stop is the supervisor's verdict: every owned group is gone.
85  let assert Ok(_) = python.stop(kernel)
86  let assert Ok(probe) = kernel_fixture.local(store, "/tmp")
87  let assert Ok(checked) =
88    python.execute(
89      probe,
90      "b",
91      "import os\nstate = 'alive'\ntry:\n    os.killpg("
92        <> int.to_string(pgid)
93        <> ", 0)\nexcept ProcessLookupError:\n    state = 'gone'\nstate",
94      5000,
95    )
96  checked.value |> should.equal("'gone'")
97  let assert Ok(_) = python.stop(probe)
98  store.close(store)
99}