okami.mom/albedo

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

dawnkernel: a restarted daemon re-counts held heavy slots and ends a dead kernel's jobsfbcd4e8

main
branchesmainall refs →
2.4 KiB61 linesraw
1//// A replayed host call must never run twice, and only a call an earlier
2//// daemon started without answering may come back as "outcome unknown". The
3//// replay that reaches this ledger needs a daemon restart between a call and
4//// its acknowledgement, which E2E cannot place; the ledger is driven here.
5
6import albedo/daemon/store
7import albedo/harness/extensions/python/link
8import gleam/option
9import gleeunit/should
10
11fn recorded() -> #(store.Store, link.Record) {
12  let assert Ok(ledger) = store.start(":memory:", "")
13  let record =
14    link.Record(
15      session: "s",
16      kernel: "k1",
17      token: "t",
18      run_dir: "/nowhere",
19      cwd: "/tmp",
20      modules: "[]",
21      out_seq: 0,
22      owned: "{}",
23    )
24  let assert Ok(_) = link.create(ledger, record)
25  #(ledger, record)
26}
27
28pub fn a_replayed_call_is_answered_not_run_again_test() -> Nil {
29  let #(ledger, record) = recorded()
30  link.call(ledger, record, "c1") |> should.equal(link.Fresh)
31  let assert Ok(_) = link.reply(ledger, record, "c1", 1, "{\"type\":\"reply\"}")
32  // The kernel never saw our ack and sends the call again: its reply is
33  // already on the way through the outbox.
34  link.call(ledger, record, "c1") |> should.equal(link.Answered)
35  link.pending(ledger, record) |> should.equal([#(1, "{\"type\":\"reply\"}")])
36  // Once the kernel has the reply, the call and its frame are gone.
37  let assert Ok(_) = link.ack(ledger, record, 1)
38  link.pending(ledger, record) |> should.equal([])
39  store.close(ledger)
40}
41
42pub fn a_call_an_earlier_daemon_never_answered_is_unknown_test() -> Nil {
43  let #(ledger, record) = recorded()
44  link.call(ledger, record, "c1") |> should.equal(link.Fresh)
45  // The daemon restarted before answering: a replay must not run it again.
46  link.call(ledger, record, "c1") |> should.equal(link.Unknown)
47  store.close(ledger)
48}
49
50pub fn a_restarted_daemon_resends_and_keeps_counting_test() -> Nil {
51  let #(ledger, record) = recorded()
52  let assert Ok(_) = link.persist(ledger, record, 1, "{\"a\":1}")
53  let assert Ok(_) = link.persist(ledger, record, 2, "{\"a\":2}")
54  let assert Ok(_) = link.ack(ledger, record, 1)
55  let assert Ok(found) = link.find(ledger, "s") |> option.to_result(Nil)
56  found.out_seq |> should.equal(2)
57  link.pending(ledger, found) |> should.equal([#(2, "{\"a\":2}")])
58  let assert Ok(_) = link.forget(ledger, found)
59  link.find(ledger, "s") |> should.equal(option.None)
60  store.close(ledger)
61}