kernel: a restarted daemon re-counts held heavy slots and ends a dead kernel's jobs
#59: the hello reports jobs holding a granted slot, and the new pool claims them; kernel_links records the job groups a kernel owns, so an attach that finds the kernel gone still ends them. Darwin now gets a leader token (start time via sysctl), so a reused pid is never signalled. kernel.md spells out why replayed kernel frames are harmless after a crash restart. Signed-off-by: dawn <90008@klbr.net>
36 unmodified lines3738394041414243444546474546474849505152538 unmodified lines626364656466676869708 unmodified lines79808182838485868788899091928693949596979899100101175 unmodified linesExpand all27727827928027027127227328128236 unmodified linesthe daemon writes `{"attach": {"kernel", "token", "ack", "grace"}}` first. thekernel checks the token (kept in the daemon's sqlite), bumps its epoch, cutsoff any older connection (newest attach wins), and answers`{"hello": {protocol, bundle, epoch, ack, pid, pgid, leader, ready, jobs,external, slots, dropped}}`, or `{"refused": reason}`. `protocol` (1) and theheld, external, slots, dropped}}`, or `{"refused": reason}`. `protocol` (1) and thehello/snapshot/shutdown frames never change shape. `bundle` is`albedo_bundle.digest()`, the same content hash the remote plugin stagesunder; a hello whose bundle or protocol differs from the bridge's makes thekernel stale (see version skew). `jobs` (live `job_start` frames), `external`and `slots` (job slots still waited for) let a fresh port owner take over jobownership and admission.kernel stale (see version skew). `jobs` (live `job_start` frames), `external`,`held` (jobs granted a heavy slot whose end isn't proven yet) and `slots` (jobslots still waited for) let a fresh port owner take over job ownership andadmission: held slots count in its pool at once (`albedo_job_slots:claim`,even past the limit), waited-for ones are asked for again.
## session layer
after the handshake every frame is `{"seq": n, "ack": m, "frame": {...}}`, or8 unmodified lines acknowledged at once.- **daemon side** (`python/link.gleam`, tables from the python extension's `kernel_links` schema migration): `kernel_links` (one row per session: kernel id, token, run dir, workspace, modules, pid/pgid/leader, epoch, last seq sent), `kernel_outbox` keyed (session, kernel, seq), bounded at 2048 frames / sent, and the job groups it owns, rewritten as they change), `kernel_outbox` keyed (session, kernel, seq), bounded at 2048 frames / 64 MiB per kernel, and `kernel_calls`. the port owner persists each frame before writing it, deletes what the kernel acknowledges, and acknowledges kernel frames within 100 ms. its own last-seen seq lives in memory: after a daemon restart it accepts everything the kernel resends, and the dedupe8 unmodified lines kernel acknowledges its reply.- execute, by cell id: a finished cell (the newest 16) resends its `done`; a queued or running one is ignored. callers must not reuse cell ids.- interrupt, reply, release are idempotent already.- what the kernel resends after a restart: job wakes are host calls (`jobs.completed`), so the call ledger covers them; `job_start`, `job`, `job_cancel` and `jobs` only update ownership and slot bookkeeping, which converges; `job_acquire` asks the new pool, which never saw it; a `done` nobody waits for rewrites the same outcome; traces save by cell id.
## drops
- the bridge dies: the port owner reattaches with backoff (50 ms doubling to 2 s, 40 tries); exit 3 or a refusal means the kernel is gone and the old abandon path reaps it. the cell keeps running meanwhile. abandon path reaps it. a port owner attaching after a restart starts from the record's pid/pgid/leader and groups, so a kernel that died hard while nobody was attached still has its process group and its jobs' groups ended. the leader token (start time, from `/proc` or Darwin's `sysctl`) keeps a reused pid from being signalled. the cell keeps running meanwhile.- the cell deadline passes, or an interrupt goes unanswered for 2 s, while detached: the interrupt waits in the outbox and the caller gets `kernel.Detached`, which leaves the journaled cell `started` (the model is told to check `cells.info`). a `done` that arrives for a cell nobody waits175 unmodified linesExpand allaway, or the session moves.
## not yet
- after a daemon restart, job groups of a kernel that died hard (SIGKILL) while nobody was attached are not reaped: the recorded pid could be reused.- job slots a reattached kernel's running jobs held are not re-counted in the new daemon's pool; only slots still waited for are asked for again.- profile output a remote login shell prints reaches the daemon's own stderr log on every remote command.221 unmodified linesExpand all2222232242252262272282292301416 unmodified linesExpand all221 unmodified linesExpand all entry = JOB_SLOTS.get(message["id"]) if entry is not None and not entry[0].done(): slot, on_queued = entry if message.get("ok") is True: LINK.jobs.hold(message["id"]) slot.set_result(None) elif message.get("queued") is True: on_queued() else:1416 unmodified linesExpand all158 unmodified linesExpand all1591601611621631641631641651661671681691701711721731741751761771781791802 unmodified lines183184185186187188189190191138 unmodified linesExpand all33033133233333433533633733830 unmodified lines158 unmodified linesExpand allclass JobBook: """Job groups this kernel still owns, read from the frames it sends.
The same bookkeeping the daemon keeps: a started job's group is owned until a job frame proves it gone. A reattaching daemon learns them from the hello, and the kernel stays alive past its grace while any are live. until a job frame proves it gone, and so is the heavy-job slot it was granted. A reattaching daemon learns both from the hello, and the kernel stays alive past its grace while any job is live. """
def __init__(self) -> None: self.started: dict[str, dict[str, object]] = {} self.held: set[str] = set() self.external = 0
def hold(self, id: str) -> None: """The daemon granted this job a slot.""" self.held.add(id)
def observe(self, frame: dict[str, object]) -> None: kind, id = frame.get("type"), frame.get("id") if kind == "job_start" and isinstance(id, str): pgid = frame.get("pgid")2 unmodified lines elif kind == "job" and isinstance(id, str): cleanup = frame.get("cleanup") if isinstance(cleanup, dict) and cleanup.get("gone") is True: self.started.pop(id, None) self.held.discard(id) elif kind == "jobs": live = frame.get("live") if isinstance(live, int) and not isinstance(live, bool) and live >= 0: self.external = live138 unmodified linesExpand all "bundle": self.bundle, "epoch": self.epoch, "ack": self.inbound.last, "jobs": list(self.jobs.started.values()), "held": sorted(self.jobs.held), "external": self.jobs.external, "dropped": self.outbox.dropped, } write_frame(connection.send, encode({"hello": hello}))30 unmodified lines10 unmodified lines1112131415161718192021222324252627282938 unmodified lines6869707169727374757641 unmodified lines118119120121119122123124125126127128129130131132133134135136137138139140141142143144117 unmodified linesExpand all10 unmodified linesfrom collections.abc import Callable, Sequencefrom dataclasses import dataclassfrom typing import Anyimport asyncioimport ctypesimport osimport signalimport structimport subprocessimport sys
TERM_GRACE = 0.25 # seconds a group gets to honour TERMKILL_GRACE = 2.0 # seconds a group gets to die after KILLPROBE_INTERVAL = 0.02LIBC = ctypes.CDLL(None)
def reap_stopped_safely(loop: asyncio.AbstractEventLoop) -> None: """Let `loop` run children that get paused with SIGSTOP.38 unmodified lines@dataclass(frozen=True)class Group: """A target to end: a whole process group, or one process that leads none.
`leader` identifies the leader where /proc exists, so a reused id is refused. `leader` identifies the leader by its start time, so a reused id is refused. """
pgid: int leader: str | None = None41 unmodified lines return None
def leader_token(pid: int) -> str | None: """Identity token for a live pid, where the platform exposes /proc.""" """Identity token for a live pid: its start time, from /proc or, on Darwin, its process table entry. None where neither can tell.""" try: with open(f"/proc/{pid}/stat", "rb") as stat: return start_token(stat.read()) except OSError: return darwin_start(pid) if sys.platform == "darwin" else None
def darwin_start(pid: int) -> str | None: """A Darwin process's start time: the timeval that opens its kinfo_proc (sysctl CTL_KERN, KERN_PROC, KERN_PROC_PID). None for a pid not in use.""" mib = (ctypes.c_int * 4)(1, 14, 1, pid) buffer = ctypes.create_string_buffer(1024) size = ctypes.c_size_t(len(buffer)) if LIBC.sysctl(mib, 4, buffer, ctypes.byref(size), None, 0) != 0 or size.value < 12: return None seconds, micros = struct.unpack_from("=qi", buffer.raw) return f"{seconds}.{micros:06d}"
def live_members(pgid: int) -> list[int] | None: """Pids in a group that are not zombies; None where /proc is unavailable."""117 unmodified linesExpand all1 unmodified line23456789101116 unmodified lines282930313233343536373839404142434 unmodified lines4849505152535455565758596042616263646543 unmodified lines1091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441 unmodified line
The kernel runs detached behind a bridge process; the daemon reattaches to itafter a restart (graceful or crash) and after the bridge dies mid-cell, so thenamespace, the kernel process, and an in-flight cell's result all survive.What the kernel owns survives with it: a reattached kernel's running job keepsits heavy-job slot, and one found dead after a restart has its jobs ended."""
import jsonimport os16 unmodified lines if "albedo_bridge.py" in line and str(home) in line ]
def alive(pid): try: os.kill(pid, 0) except ProcessLookupError: return False return True
@exclusiveclass KernelDetachTests(unittest.TestCase): def setUp(self): self.cells = []4 unmodified lines return text("ok")
self.provider = Provider(script) self.addCleanup(self.provider.close)
def prepare(app): # One heavy slot, granted to a job half a second in. app.daemon.env.update( ALBEDO_MAX_LOCAL_JOBS="1", ALBEDO_JOB_LOAD="0", ALBEDO_JOB_GRACE_SECONDS="0.5", )
self.app = Albedo(self.provider, protocol="responses") self.app = Albedo(self.provider, protocol="responses", prepare=prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.session = self.app.session()
43 unmodified lines self.assertEqual(slept["value"], first["value"]) (replacement,) = bridges(self.app.home) self.assertNotEqual(replacement, bridge)
def test_a_reattached_kernels_running_job_keeps_its_slot(self): held = self.cell( "import asyncio\na = run('sleep', '60')\nawait asyncio.sleep(1.5)\na.queued" ) self.assertEqual(held["value"], "False") self.app.restart() # The new daemon's pool counts the slot a's job already holds. waiting = self.cell( "b = run('sleep', '60')\nawait asyncio.sleep(1.5)\n(a.queued, b.queued)" ) self.assertEqual(waiting["value"], "(False, True)") freed = self.cell("await a.stop()\nawait asyncio.sleep(1)\nb.queued") self.assertEqual(freed["value"], "False") self.cell("await b.stop()")
def test_a_kernel_killed_while_the_daemon_was_down_has_its_jobs_ended(self): started = self.cell( "import asyncio, os\nj = run('sleep', '300')\nawait asyncio.sleep(0.5)\n(os.getpid(), j.process.pid)" ) kernel, job = (int(part) for part in started["value"].strip("()").split(","))
# Graceful stop leaves the kernel running; it dies hard before the # next daemon reaches it, and its job's group outlives it. self.app.restart(prepare=lambda _: os.kill(kernel, signal.SIGKILL)) deadline = time.monotonic() + 15 while alive(job) and time.monotonic() < deadline: time.sleep(0.1) self.assertFalse(alive(job), "the dead kernel's job is still running")
if __name__ == "__main__": unittest.main()105 unmodified linesExpand all10610710810911011011111211311411511611611711811912028 unmodified lines105 unmodified linesExpand all%% A fresh detached kernel in /tmp whose session layer persists nothing: these%% tests never reattach, so the link only has to answer.start(Python, Script, Host, Modules) -> Link = {link, fun(_, _) -> nil end, fun(_) -> nil end, fun() -> [] end, fun(_) -> fresh end, fun(_, _, _) -> nil end, fun(_) -> nil end, fun() -> nil end}, fun(_) -> fresh end, fun(_, _, _) -> nil end, fun(_) -> nil end, fun() -> nil end, fun(_) -> nil end}, Id = binary:part(albedo_native:new_id(), 0, 16), RunDir = filename:join(os:getenv("TMPDIR", "/tmp"), <<"albedo-run-", Id/binary>>), Bridge = filename:join(filename:dirname(Script), <<"albedo_bridge.py">>), albedo_python:start({boot, self(), Python, Bridge, <<"/tmp">>, Host, iolist_to_binary(json:encode(Modules)), Link, unicode:characters_to_binary(RunDir), Id, albedo_native:new_id(), 20, 0, true, none}). Id, albedo_native:new_id(), 20, 0, <<"{}">>, true, none}).
cell(Kernel, Code) -> %% A kernel answers a cell id it already ran with that run's result. Request = iolist_to_binary(json:encode(#{type => <<"execute">>, id => albedo_native:new_id(), code => Code})),28 unmodified lines17 unmodified lines18192021222324252635 unmodified lines17 unmodified lines run_dir: "/nowhere", cwd: "/tmp", modules: "[]", out_seq: 0, owned: "{}", ) let assert Ok(_) = link.create(ledger, record) #(ledger, record)}35 unmodified lines179 unmodified linesExpand all180181182183184185186185186187188189187188189190191192118 unmodified linesExpand all179 unmodified linesExpand all leader = run( asyncio.create_subprocess_shell("sleep 30", start_new_session=True) ) try: # Linux and Darwin both give a process an identity beyond its pid. self.assertIsNotNone(albedo_proc.leader_token(leader.pid)) group = albedo_proc.Group(leader.pid, "not-the-leader-we-created") if ( albedo_proc.leader_token(leader.pid) is not None ): # identity source present self.assertFalse(albedo_proc.alive(group)) self.assertTrue(present(leader.pid)) # and it was left alone self.assertFalse(albedo_proc.alive(group)) self.assertTrue(present(leader.pid)) # and it was left alone finally: run(albedo_proc.terminate([albedo_proc.Group(leader.pid)])) run(leader.wait())
118 unmodified linesExpand all16 unmodified lines1718192021212223242510 unmodified lines363738394041424344454655 unmodified lines10210310410510610710810911011111211311411511611772 unmodified lines16 unmodified lines%% local pool starts and watches the host pools; the admission protocol is the%% same.-module(albedo_job_slots).-behaviour(gen_server).-export([ensure/0, ensure/2, acquire/2, release/2, release_owner/1, limit/0]).-export([ensure/0, ensure/2, acquire/2, claim/2, release/2, release_owner/1, limit/0]).-export([init/1, handle_call/3, handle_cast/2, handle_info/2]).
-record(state, {base, running = 0, entries = #{}, waiting = [], load = false, ticking = false, hosts = #{}}).-define(MAX_WAITING, 1024).10 unmodified lines%% The pool for kernels on Host, which has Cpus cores.ensure(Host, Cpus) -> gen_server:call(ensure(), {host, Host, Cpus}).
acquire(Pool, Id) -> gen_server:cast(Pool, {acquire, self(), Id}).%% A slot a running job already holds, granted by an earlier daemon: it counts%% at once, even past the limit, and nothing is sent back.claim(Pool, Id) -> gen_server:cast(Pool, {claim, self(), Id}).release(Pool, Id) -> gen_server:cast(Pool, {release, {self(), Id}}).release_owner(Pool) -> gen_server:cast(Pool, {release_owner, self()}).
%% The slot count right now, for tests and status.55 unmodified lines _ -> ok end, {noreply, tick(S1)} end;handle_cast({claim, Owner, Id}, S = #state{entries = Entries, running = Running}) -> Key = {Owner, Id}, case maps:is_key(Key, Entries) of true -> {noreply, S}; false -> Mon = monitor(process, Owner), {noreply, S#state{entries = Entries#{Key => {active, Mon}}, running = Running + 1}} end;handle_cast({release, Key}, S) -> {noreply, drain(drop(Key, S))};handle_cast({base, Cpus}, S) -> {noreply, drain(S#state{base = remote_base(Cpus)})};handle_cast({release_owner, Owner}, S) -> {noreply, drop_keys([K || {P, _} = K := _ <- S#state.entries, P =:= Owner], S)}.72 unmodified lines38 unmodified lines394041424343444546471 unmodified line4950515253545556545758596061354 unmodified linesExpand all41641741841941741841942042142242342442542642742842960 unmodified lines49049149249349449549649749849950050120 unmodified lines52252352452551652652752851952953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956026 unmodified lines587588589590555556557591592593594559595596597598599564565600601602567603604605606607608253 unmodified linesExpand all38 unmodified lines-define(REMOTE_REATTACH_MAX_MS, 30000).-define(GRACE_MARGIN_S, 60). %% past a remote kernel's grace before it is given up
%% Boot is the Gleam kernel.Boot record; Link is kernel.Link.start({boot, Owner, Python, Bridge, Cwd, Host, Modules, Link, RunDir, Kernel, Token, Grace, OutSeq, Fresh, Remote}) ->start({boot, Owner, Python, Bridge, Cwd, Host, Modules, Link, RunDir, Kernel, Token, Grace, OutSeq, Owned, Fresh, Remote}) -> Parent = self(), Ref = make_ref(), {Pid, Mon} = spawn_monitor(fun() -> process_flag(trap_exit, true), monitor(process, Owner),1 unmodified line {some, {remote, _, Machine, Cpus}} -> albedo_job_slots:ensure(Machine, Cpus); none -> albedo_job_slots:ensure() end, monitor(process, Pool), %% What the record says the kernel owned, so a kernel found dead at %% this attach still has its groups ended. {Target, Groups} = owned(Owned), S = #{port => none, host => Host, link => Link, owner => Owner, active => none, events => [], groups => #{}, external => 0, pool => Pool, slots => #{}, target => none, groups => Groups, external => 0, pool => Pool, slots => #{}, target => Target, python => Python, bridge => Bridge, cwd => Cwd, remote => Remote, modules => Modules, run_dir => RunDir, kernel => Kernel, token => Token, grace => Grace, in => 0, acked => 0, kack => 0, out => OutSeq, calls => #{}, bundle => none, retries => 0, flush => none, give_up => none, stale => none, forced => false},354 unmodified linesExpand all S2 = case maps:get(<<"external">>, Hello, 0) of Live when is_integer(Live), Live >= 0 -> S1#{external => Live}; _ -> S1 end, Waiting = [Id || Id <- maps:get(<<"slots">>, Hello, []), is_binary(Id), not maps:is_key(Id, maps:get(slots, S2))], lists:foldl(fun acquire_slot/2, S2, Waiting). %% Slots its running jobs hold count again in this daemon's pool; the %% ones still waited for are asked for again. Untracked = fun(Field) -> [Id || Id <- maps:get(Field, Hello, []), is_binary(Id), not maps:is_key(Id, maps:get(slots, S2))] end, S3 = lists:foldl(fun claim_slot/2, S2, Untracked(<<"held">>)), lists:foldl(fun acquire_slot/2, S3, Untracked(<<"slots">>)).
%% A kernel speaking another protocol, or running another bundle than the%% bridge that reached it, is stale: the session swaps it at its next idle%% moment (kernel.upgrade). A protocol difference outranks a bundle one.60 unmodified linesacquire_slot(Id, S = #{pool := Pool, slots := Slots}) -> albedo_job_slots:acquire(Pool, Id), S#{slots => Slots#{Id => true}}.
claim_slot(Id, S = #{pool := Pool, slots := Slots}) -> albedo_job_slots:claim(Pool, Id), S#{slots => Slots#{Id => true}}.
slot_reply(Id, Result, S = #{slots := Slots}) -> case maps:is_key(Id, Slots) of false -> S; %% cancelled before the grant reached us true ->20 unmodified lines%% Job ownership bookkeeping shared by the main loop and the startup and%% shutdown drains: a started job's group is recorded; a job proven gone%% releases its slot and its group.start_job(Message, Id, S = #{groups := Groups}) -> S#{groups => Groups#{Id => group_of(Message)}}. owns(S, Groups#{Id => group_of(Message)}).
gone(Id, S = #{groups := Groups}) -> (release_slot(Id, S))#{groups => maps:remove(Id, Groups)}. owns(release_slot(Id, S), maps:remove(Id, Groups)).
%% The groups are recorded as they change, so a daemon that finds the kernel%% dead after a restart can still end them.owns(S = #{groups := Groups}, Groups) -> S;owns(S, Groups) -> link_own(S, json:encode(maps:map(fun(_, Spec) -> maps:map(fun nil_null/2, Spec) end, Groups))), S#{groups => Groups}.
nil_null(_, nil) -> null;nil_null(_, V) -> V.
%% The record's {pid, pgid, leader, groups}: the kernel's identity from its%% last hello and the job groups it owned.owned(Owned) -> M = try json:decode(Owned) catch _:_ -> #{} end, Spec = fun(Fields) -> maps:map(fun(_, null) -> nil; (_, V) -> V end, maps:with([<<"pid">>, <<"pgid">>, <<"leader">>], Fields)) end, Target = case maps:get(<<"pid">>, M, null) of Pid when is_integer(Pid), Pid > 1 -> target_of(Spec(M)); _ -> none end, Groups = case maps:get(<<"groups">>, M, #{}) of G when is_map(G) -> #{Id => group_of(Spec(Fields)) || Id := Fields <- G, is_map(Fields)}; _ -> #{} end, {Target, Groups}.
%% The newest 100 job and trace frames, what events/1 hands out.journal(Message, S = #{events := Events}) -> S#{events => lists:sublist([iolist_to_binary(json:encode(Message)) | Events], 100)}.
26 unmodified lines
%% The Gleam link: the durable outbox and host-call ledger. A storage failure%% is logged and never stops the kernel; the frame still goes out. Once the%% store is gone (its owner died) there is nothing left to record.link_persist(S = #{link := {link, Persist, _, _, _, _, _, _}}, Seq, Frame) -> guarded(S, fun() -> Persist(Seq, Frame) end).link_ack(S = #{link := {link, _, Ack, _, _, _, _, _}}, Upto) -> guarded(S, fun() -> Ack(Upto) end).link_pending(S = #{link := {link, _, _, Pending, _, _, _, _}}) ->link_persist(S = #{link := {link, Persist, _, _, _, _, _, _, _}}, Seq, Frame) -> guarded(S, fun() -> Persist(Seq, Frame) end).link_ack(S = #{link := {link, _, Ack, _, _, _, _, _, _}}, Upto) -> guarded(S, fun() -> Ack(Upto) end).link_pending(S = #{link := {link, _, _, Pending, _, _, _, _, _}}) -> case guarded(S, Pending) of Frames when is_list(Frames) -> Frames; _ -> [] end.link_call(S = #{link := {link, _, _, _, Call, _, _, _}}, Id) ->link_call(S = #{link := {link, _, _, _, Call, _, _, _, _}}, Id) -> case guarded(S, fun() -> Call(Id) end) of State when State =:= fresh; State =:= answered; State =:= unknown -> State; _ -> fresh end.link_reply(S = #{link := {link, _, _, _, _, Reply, _, _}}, Id, Seq, Frame) -> guarded(S, fun() -> Reply(Id, Seq, Frame) end).link_record(S = #{link := {link, _, _, _, _, _, Record, _}}, Fields) ->link_reply(S = #{link := {link, _, _, _, _, Reply, _, _, _}}, Id, Seq, Frame) -> guarded(S, fun() -> Reply(Id, Seq, Frame) end).link_record(S = #{link := {link, _, _, _, _, _, Record, _, _}}, Fields) -> guarded(S, fun() -> Record(iolist_to_binary(json:encode(maps:filter(fun(_, V) -> V =/= nil end, Fields)))) end).link_forget(S = #{link := {link, _, _, _, _, _, _, Forget}}) -> guarded(S, Forget).link_forget(S = #{link := {link, _, _, _, _, _, _, Forget, _}}) -> guarded(S, Forget).link_own(S = #{link := {link, _, _, _, _, _, _, _, Own}}, Groups) -> guarded(S, fun() -> Own(iolist_to_binary(Groups)) end).
guarded(#{owner := Owner}, Fun) -> case is_process_alive(Owner) of false -> error;253 unmodified linesExpand all75 unmodified lines767778798081828384320 unmodified linesExpand all40540640740840941041141241356 unmodified lines470471472473474475476477478207 unmodified linesExpand all75 unmodified lines kernel: String, token: String, grace: Int, out_seq: Int, owned: String, fresh: Bool, remote: Option(Remote), )}320 unmodified linesExpand all run_dir: runs <> "/" <> kernel, cwd: cwd, modules: modules, out_seq: 0, owned: "{}", ) use _ <- result.try( link.create(store, record) |> result.map_error(Unavailable), )56 unmodified lines kernel: record.kernel, token: record.token, grace: grace(), out_seq: record.out_seq, owned: record.owned, fresh: fresh, remote: remote, )) }),207 unmodified linesExpand all123456789456789101112131416 unmodified lines313233343536373839404115 unmodified lines5758596061626364656166676869707167727374757677787975808182838485868726 unmodified lines11411511611711011811912012112212312412512612712812913011313113213313413570 unmodified lines206207208209210211212213214215216217218219220221222223224225226227228229183 unmodified linesExpand all//// The daemon's half of a detached kernel's durable session layer.//////// One row per session names its kernel: where its run directory is, the//// token an attach presents, the workspace and modules it booted with, and//// the last sequence number the daemon sent it. Frames the daemon sent and//// the kernel has not acknowledged stay in `kernel_outbox`, so a reattach//// (or a restarted daemon) resends them. `kernel_calls` remembers every host//// call the kernel made until the kernel has its reply, so a replayed call//// never runs twice.//// token an attach presents, the workspace and modules it booted with, the//// last sequence number the daemon sent it, and the process groups it owns,//// so a kernel found dead after a restart still has its jobs ended. Frames//// the daemon sent and the kernel has not acknowledged stay in//// `kernel_outbox`, so a reattach (or a restarted daemon) resends them.//// `kernel_calls` remembers every host call the kernel made until the kernel//// has its reply, so a replayed call never runs twice.
import albedo/daemon/storeimport gleam/dynamic/decodeimport gleam/io16 unmodified lines run_dir: String, cwd: String, modules: String, out_seq: Int, /// `{pid, pgid, leader, groups}`: the kernel's identity from its last /// hello and its job groups by job id, as JSON. owned: String, )}
/// What the ledger knows about a host call id the kernel sent.15 unmodified lines call: fn(String) -> CallState, reply: fn(String, Int, String) -> Nil, record: fn(String) -> Nil, forget: fn() -> Nil, own: fn(String) -> Nil, )}
pub fn apply(db: sqlight.Connection) -> Result(Nil, String) { store.exec( use _ <- result.try(store.exec( db, "CREATE TABLE IF NOT EXISTS kernel_links ( session TEXT PRIMARY KEY, kernel TEXT NOT NULL UNIQUE, token TEXT NOT NULL, run_dir TEXT NOT NULL, cwd TEXT NOT NULL, modules TEXT NOT NULL, pid INTEGER, pgid INTEGER, leader TEXT, epoch INTEGER NOT NULL DEFAULT 0, out_seq INTEGER NOT NULL DEFAULT 0, out_seq INTEGER NOT NULL DEFAULT 0, groups TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))); CREATE TABLE IF NOT EXISTS kernel_outbox ( session TEXT NOT NULL, kernel TEXT NOT NULL, seq INTEGER NOT NULL, frame BLOB NOT NULL, PRIMARY KEY (session, kernel, seq)); CREATE TABLE IF NOT EXISTS kernel_calls ( kernel TEXT NOT NULL, call TEXT NOT NULL, reply_seq INTEGER, PRIMARY KEY (kernel, call));", ) )) store.add_columns(db, "kernel_links", [ #("groups", "TEXT NOT NULL DEFAULT '{}'"), ])}
/// Record a kernel about to boot, replacing whatever the session had.pub fn create(storage: store.Store, record: Record) -> Result(Nil, String) {26 unmodified lines use run_dir <- decode.field(3, decode.string) use cwd <- decode.field(4, decode.string) use modules <- decode.field(5, decode.string) use out_seq <- decode.field(6, decode.int) decode.success(Record(session, kernel, token, run_dir, cwd, modules, out_seq)) use owned <- decode.field(7, decode.string) decode.success(Record( session, kernel, token, run_dir, cwd, modules, out_seq, owned, ))}
const columns = "session,kernel,token,run_dir,cwd,modules,out_seq"const columns = "session,kernel,token,run_dir,cwd,modules,out_seq,json_object('pid',pid,'pgid',pgid,'leader',leader,'groups',json(groups))"
/// The kernel a session last booted, if one is recorded.pub fn find(storage: store.Store, session: String) -> Option(Record) { store.read(70 unmodified lines call: fn(id) { call(storage, record, id) }, reply: fn(id, seq, frame) { log(reply(storage, record, id, seq, frame)) }, record: fn(fields) { log(identify(storage, record, fields)) }, forget: fn() { log(forget(storage, record)) }, own: fn(groups) { log(own(storage, record, groups)) }, )}
/// The job groups the kernel owns now, by job id.fn own( storage: store.Store, record: Record, groups: String,) -> Result(Nil, String) { store.write(storage, "UPDATE kernel_links SET groups=? WHERE kernel=?", [ sqlight.text(groups), sqlight.text(record.kernel), ])}
fn log(outcome: Result(Nil, String)) -> Nil { case outcome { Ok(_) -> Nil Error(reason) -> io.println_error("kernel link: " <> reason)183 unmodified linesExpand all