runner.bend

bend-kit-runner@0.13.0.1 · 1,810 lines · 251 definitions

1# Durable job runner: a journaled scheduler queue, bounded by a resources pool, with retry and a breaker per target. Source: https://github.com/paymog/bend-kit/tree/main/runner
2import Base
3import bend-kit-bytes@0.3.3.0/bytes.bend as Bytes
4import bend-kit-journal@0.3.0.1/journal.bend as J
5import bend-kit-scheduler@0.4.1.0/queue.bend as Q
6import bend-kit-resources@0.1.0.0/resources.bend as Res
7import bend-kit-retry@0.3.0.0/retry.bend as Retry
8import bend-kit-time@0.3.0.0/time.bend as Time
9import bend-kit-collections@0.1.4.1/heap.bend as H
10import bend-kit-collections@0.1.4.1/omap.bend as O
11import bend-kit-collections@0.1.4.1/hmap.bend as M
12import bend-kit-collections@0.1.4.1/LAWS.bend as CL
13import bend-kit-collections@0.1.4.1/PROOF.bend as CP
14import bend-kit-files@0.3.0.0/files.bend as Files
15
16# Why retry gave up on a ticket: its budget was spent, its deadline passed, its failure must not repeat,
17# or another try would repeat its failure.
18type Why is Data:
19  WhySpent{}
20  WhyLate{}
21  WhyUnsafe{}
22  WhyFailed{}
23
24# A ticket that retry gave up on, with its payload, and when it gave up (wall-clock ms).
25# It stays until Revive, Cancel, or Purge takes it.
26type Dead<-A: Data> is Data:
27  Dead{ticket: Q.Ticket<A>, why: Why, at: Nat}
28
29# One durable transition. Each one is one journal record, appended before the state changes.
30# Select: the queue's select at now. Replay runs it at the same now, so it takes the same ticket.
31# Requeue: the attempt failed; it counts one failure. Defer: the breaker held the ticket back; no attempt ran.
32# Abandon: retry gave up at at; the running ticket becomes dead. Revive: a dead ticket is pending again at deadline, with a new budget.
33# Purge: the dead tickets abandoned before before are dropped. PurgeAll: every dead ticket is dropped.
34# Break: the breaker of target changed to state.
35type Op<-A: Data> is Data:
36  Enqueue{deadline: Nat, payload: A}
37  Select{now: Nat}
38  Complete{id: Nat}
39  Cancel{id: Nat}
40  Requeue{id: Nat, deadline: Nat}
41  Defer{id: Nat, deadline: Nat}
42  Abandon{id: Nat, why: Why, at: Nat}
43  Revive{id: Nat, deadline: Nat}
44  Purge{before: Nat}
45  PurgeAll{}
46  Break{target: String, state: Retry.State}
47
48# The failed attempts of a ticket, and its first deadline, from which the policy deadline counts.
49type Try is Data:
50  Try{fails: Nat, start: Nat}
51
52# A Try for each ticket that is not yet complete, cancelled, or dead, by ticket id.
53def Tries() -> Data:
54  O.OMap<Nat, Try>
55
56# The breaker state of each target that has one, by target. An absent target is closed with no failures.
57def Breakers() -> Data:
58  Map<&2, Retry.State>
59
60def dead.hash(i: Nat) -> U32:
61  U32.from_nat(i)
62
63# The dead tickets, by ticket id.
64def Graves(-A: Data) -> Data:
65  M.HMap<Nat, Dead<A>>
66
67# The durable state: the queue, the tries, the dead tickets, and the breakers.
68type State<-A: Data> is Data:
69  State{queue: Q.State<A>, tries: Tries(), dead: Graves(A), breakers: Breakers()}
70
71def new(-A: Data) -> State<A>:
72  State{Q.new(A), O.OMap.new(Nat, Try), M.HMap.new(Nat, Dead<A>), Map.new(&2, Retry.State)}
73
74def queue(-A: Data, s: State<A>) -> Q.State<A>:
75  State{q, ts, ds, bs} = s
76  q
77
78def tries(-A: Data, s: State<A>) -> Tries():
79  State{q, ts, ds, bs} = s
80  ts
81
82# The dead tickets, by id. A Revive of one makes it pending again; a Cancel or a Purge drops it.
83def dead(-A: Data, s: State<A>) -> Graves(A):
84  State{q, ts, ds, bs} = s
85  ds
86
87def breakers(-A: Data, s: State<A>) -> Breakers():
88  State{q, ts, ds, bs} = s
89  bs
90
91# ---- dead tickets
92
93def dead.get(~A: Data, ds: Graves(A), +i: Nat) -> Maybe<&2, Dead<A>>:
94  M.HMap.get(~Nat, ~Dead<A>, ~dead.hash, ~Nat.is_eq, ds, i)
95
96def dead.drop(~A: Data, ds: Graves(A), +i: Nat) -> Graves(A):
97  M.HMap.del(~Nat, ~Dead<A>, ~dead.hash, ~Nat.is_eq, ds, i)
98
99def dead.bury(~A: Data, m: Maybe<&2, Q.Ticket<A>>, ds: Graves(A), w: Why, +at: Nat) -> Graves(A):
100  match m:
101    case None{}:
102      ds
103    case Some{Q.Ticket{+i, d, x}}:
104      M.HMap.put(~Nat, ~Dead<A>, ~dead.hash, ~Nat.is_eq, ds, i, Dead{Q.Ticket{i, d, x}, w, at})
105
106def dead.list.entries(-A: Data, es: List<&2, M.Entry<Nat, Dead<A>>>, acc: List<&2, Dead<A>>) -> List<&2, Dead<A>>:
107  match es:
108    case Nil{}:
109      acc
110    case Con{M.Entry{k, d}, rest}:
111      dead.list.entries(A, rest, Con{d, acc})
112
113def dead.list.go(-A: Data, ds: Graves(A), acc: List<&2, Dead<A>>) -> List<&2, Dead<A>>:
114  match ds:
115    case M.HNil{}:
116      acc
117    case M.HLeaf{h, es}:
118      dead.list.entries(A, es, acc)
119    case M.HBranch{lo, hi}:
120      dead.list.go(A, lo, dead.list.go(A, hi, acc))
121
122# The dead tickets, in no set order.
123def dead.list(-A: Data, ds: Graves(A)) -> List<&2, Dead<A>>:
124  dead.list.go(A, ds, Nil{})
125
126def dead.old.keep(-A: Data, old: Bool, +i: Nat, rest: List<&2, Nat>) -> List<&2, Nat>:
127  match old:
128    case True{}:
129      Con{i, rest}
130    case False{}:
131      rest
132
133# The ids of the dead tickets abandoned before t.
134def dead.old(-A: Data, ds: List<&2, Dead<A>>, +t: Nat) -> List<&2, Nat>:
135  match ds:
136    case Nil{}:
137      Nil{}
138    case Con{Dead{Q.Ticket{i, d, x}, w, +at}, rest}:
139      dead.old.keep(A, Nat.is_lt(at, t), i, dead.old(A, rest, t))
140
141def dead.drops(~A: Data, ks: List<&2, Nat>, ds: Graves(A)) -> Graves(A):
142  match ks:
143    case Nil{}:
144      ds
145    case Con{k, rest}:
146      dead.drops(~A, rest, dead.drop(~A, ds, k))
147
148# The dead tickets without those abandoned before t. It walks every dead ticket once.
149def dead.purge(~A: Data, +ds: Graves(A), +t: Nat) -> Graves(A):
150  dead.drops(~A, dead.old(A, dead.list(A, ds), t), ds)
151
152# ---- tries
153
154def tries.drop(ts: Tries(), +i: Nat) -> Tries():
155  O.OMap.del(~Nat, ~Try, ~Nat.cmp, ts, i)
156
157def tries.get(ts: Tries(), +i: Nat) -> Maybe<&2, Try>:
158  O.OMap.get(~Nat, ~Try, ~Nat.cmp, ts, i)
159
160def tries.bump.got(m: Maybe<&2, Try>, ts: Tries(), +i: Nat) -> Tries():
161  match m:
162    case None{}:
163      ts
164    case Some{Try{f, st}}:
165      O.OMap.put(~Nat, ~Try, ~Nat.cmp, ts, i, Try{1n+f, st})
166
167def tries.bump(+ts: Tries(), +i: Nat) -> Tries():
168  tries.bump.got(tries.get(ts, i), ts, i)
169
170# ---- breakers
171
172def brk.got(r: Breakers() & Retry.State) -> Retry.State:
173  (m, b) = r
174  b
175
176# The breaker state of target key.
177def brk.get(-A: Data, s: State<A>, key: String) -> Retry.State:
178  brk.got(Map.get(Retry.State, Retry.CircuitClosed{Nil{}}, breakers(A, s), key))
179
180def brk.state(br: Retry.Breaker, old: Retry.State) -> Retry.State:
181  match br:
182    case Retry.BreakerOff{}:
183      old
184    case Retry.BreakerOn{c, st}:
185      st
186
187def brk.nats(a: List<&2, Nat>, b: List<&2, Nat>) -> Bool:
188  match a b:
189    case Nil{} Nil{}:
190      True{}
191    case Con{+x, xs} Con{+y, ys}:
192      Bool.and(Nat.is_eq(x, y), brk.nats(xs, ys))
193    case p q:
194      False{}
195
196def brk.same(a: Retry.State, b: Retry.State) -> Bool:
197  match a b:
198    case Retry.CircuitClosed{xs} Retry.CircuitClosed{ys}:
199      brk.nats(xs, ys)
200    case Retry.CircuitOpen{+x} Retry.CircuitOpen{+y}:
201      Nat.is_eq(x, y)
202    case Retry.CircuitHalf{+x} Retry.CircuitHalf{+y}:
203      Nat.is_eq(x, y)
204    case p q:
205      False{}
206
207def calm.state(b: Retry.State) -> Retry.State:
208  match b:
209    case Retry.CircuitClosed{fs}:
210      Retry.CircuitClosed{fs}
211    case Retry.CircuitOpen{u}:
212      Retry.CircuitOpen{u}
213    case Retry.CircuitHalf{k}:
214      Retry.CircuitHalf{0n}
215
216def calm.map(m: Breakers()) -> Breakers():
217  match m:
218    case MTip{}:
219      MTip{}
220    case MLeaf{k, v}:
221      MLeaf{k, calm.state(v)}
222    case MNode{p, lo, hi}:
223      MNode{p, calm.map(lo), calm.map(hi)}
224
225# The probes of a half-open breaker died with the process that ran them, so open counts none in flight.
226def calm(-A: Data, s: State<A>) -> State<A>:
227  State{q, ts, ds, bs} = s
228  State{q, ts, ds, calm.map(bs)}
229
230# ---- the transition
231
232def step.enqueued(-A: Data, r: Q.State<A> & Nat, ts: Tries(), ds: Graves(A), bs: Breakers(), d: Nat) -> State<A>:
233  (q, i) = r
234  State{q, O.OMap.put(~Nat, ~Try, ~Nat.cmp, ts, i, Try{0n, d}), ds, bs}
235
236def step.selected(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, ts: Tries(), ds: Graves(A), bs: Breakers()) -> State<A>:
237  (q, t) = r
238  State{q, ts, ds, bs}
239
240def step.requeued(-A: Data, r: Q.State<A> & Bool, ts: Tries(), ds: Graves(A), bs: Breakers()) -> State<A>:
241  (q, b) = r
242  State{q, ts, ds, bs}
243
244# Complete takes the ticket from the running list only: O(running), not O(pending).
245def step.completed(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, ts: Tries(), ds: Graves(A), bs: Breakers()) -> State<A>:
246  (q, t) = r
247  State{q, ts, ds, bs}
248
249# Abandon takes the ticket from the running list, as Complete does, and keeps it with its payload.
250def step.abandoned(~A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, ts: Tries(), ds: Graves(A), bs: Breakers(), w: Why, +at: Nat) -> State<A>:
251  (q, t) = r
252  State{q, ts, dead.bury(~A, t, ds, w, at), bs}
253
254# The revived ticket keeps its payload and takes id i, the key it was dead under.
255def step.revived(~A: Data, m: Maybe<&2, Dead<A>>, +i: Nat, n: Nat, h: H.Heap<Q.Ticket<A>>, r: List<&2, Q.Ticket<A>>, ts: Tries(), ds: Graves(A), bs: Breakers(), +d: Nat) -> State<A>:
256  match m:
257    case None{}:
258      State{Q.State{n, h, r}, ts, ds, bs}
259    case Some{Dead{Q.Ticket{k, old, x}, w, at}}:
260      State{Q.State{n, H.Heap.push(~Q.Ticket<A>, ~(a => b => Q.le(A, a, b)), Q.Ticket{i, d, x}, h), r}, O.OMap.put(~Nat, ~Try, ~Nat.cmp, ts, i, Try{0n, d}), ds, bs}
261
262# A closed breaker with no failures reads the same as an absent one, so it leaves the map.
263def step.broke(bs: Breakers(), k: String, b: Retry.State) -> Breakers():
264  match b:
265    case Retry.CircuitClosed{Nil{}}:
266      Map.del(&2, Retry.State, bs, k)
267    case b:
268      Map.set(&2, Retry.State, bs, k, b)
269
270# The state after op. Live and replay both run it, so they agree.
271def step(~A: Data, s: State<A>, op: Op<A>) -> State<A>:
272  match s op:
273    case State{q, ts, ds, bs} Enqueue{+d, x}:
274      step.enqueued(A, Q.enqueue(~A, d, x, q), ts, ds, bs, d)
275    case State{q, ts, ds, bs} Select{now}:
276      step.selected(A, Q.select(~A, now, q), ts, ds, bs)
277    case State{q, ts, ds, bs} Complete{+i}:
278      step.completed(A, Q.complete(A, i, q), tries.drop(ts, i), ds, bs)
279    case State{q, ts, ds, bs} Cancel{+i}:
280      State{Q.cancel(~A, i, q), tries.drop(ts, i), dead.drop(~A, ds, i), bs}
281    case State{q, ts, ds, bs} Requeue{+i, d}:
282      step.requeued(A, Q.requeue(~A, i, d, q), tries.bump(ts, i), ds, bs)
283    case State{q, ts, ds, bs} Defer{i, d}:
284      step.requeued(A, Q.requeue(~A, i, d, q), ts, ds, bs)
285    case State{q, ts, ds, bs} Abandon{+i, w, at}:
286      step.abandoned(~A, Q.complete(A, i, q), tries.drop(ts, i), ds, bs, w, at)
287    case State{Q.State{n, h, r}, ts, +ds, bs} Revive{+i, d}:
288      step.revived(~A, dead.get(~A, ds, i), i, n, h, r, ts, dead.drop(~A, ds, i), bs, d)
289    case State{q, ts, ds, bs} Purge{t}:
290      State{q, ts, dead.purge(~A, ds, t), bs}
291    case State{q, ts, ds, bs} PurgeAll{}:
292      State{q, ts, M.HMap.new(Nat, Dead<A>), bs}
293    case State{q, ts, ds, bs} Break{k, b}:
294      State{q, ts, ds, step.broke(bs, k, b)}
295
296def apply.op(~A: Data, m: Maybe<&2, Op<A>>, s: State<A>) -> Maybe<&2, State<A>>:
297  match m:
298    case None{}:
299      None{}
300    case Some{op}:
301      Some{step(~A, s, op)}
302
303# Folds one record into the replayed state. A record that does not decode ends the replay with None.
304def apply(~A: Data, ~R: Type, ~dec: R -> Maybe<&2, Op<A>>, acc: Maybe<&2, State<A>>, r: R) -> Maybe<&2, State<A>>:
305  match acc:
306    case None{}:
307      None{}
308    case Some{s}:
309      apply.op(~A, dec(r), s)
310
311# A journal record. Compaction writes one Snap of the whole state; every other record is one op.
312type Rec<-A: Data> is Data:
313  Snap{state: State<A>}
314  Do{op: Op<A>}
315
316def load.rec(~A: Data, m: Maybe<&2, Rec<A>>, s: State<A>) -> Maybe<&2, State<A>>:
317  match m:
318    case None{}:
319      None{}
320    case Some{Snap{t}}:
321      Some{t}
322    case Some{Do{op}}:
323      Some{step(~A, s, op)}
324
325# Folds one record into the replayed state: a Snap replaces it, and an op steps it, as apply does.
326# A record that does not decode ends the replay with None.
327def load(~A: Data, ~R: Type, ~dec: R -> Maybe<&2, Rec<A>>, acc: Maybe<&2, State<A>>, r: R) -> Maybe<&2, State<A>>:
328  match acc:
329    case None{}:
330      None{}
331    case Some{s}:
332      load.rec(~A, dec(r), s)
333
334# ---- pool
335
336# Picked: a due ticket and one unit of the pool for it. Idle: no due ticket, or no free unit; the pool is unchanged.
337type Pick<-A: Data> is Type:
338  Picked{pool: Res.Pool, ticket: Q.Ticket<A>, grant: Res.Grant}
339  Idle{pool: Res.Pool}
340
341def pick.took(-A: Data, r: Res.Reserve, t: Q.Ticket<A>) -> Pick<A>:
342  match r:
343    case Res.Took{p, g}:
344      Picked{p, t, g}
345    case Res.Refused{p}:
346      Idle{p}
347
348def pick.due(-A: Data, m: Maybe<&2, Q.Ticket<A>>, p: Res.Pool) -> Pick<A>:
349  match m:
350    case None{}:
351      Idle{p}
352    case Some{t}:
353      pick.took(A, Res.reserve(p, 1n), t)
354
355def pick.sel(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, p: Res.Pool) -> Pick<A>:
356  (q, m) = r
357  pick.due(A, m, p)
358
359# The ticket that Select{now} takes, with one unit of p, when there is a ticket due and a unit free.
360def pick(~A: Data, p: Res.Pool, s: State<A>, now: Nat) -> Pick<A>:
361  State{q, ts, ds, bs} = s
362  pick.sel(A, Q.select(~A, now, q), p)
363
364# ---- retry
365
366# The budget left for ticket i: the policy's attempts less its failures, until its start plus the policy deadline.
367def budget.of(+attempts: Nat, t: Maybe<&2, Try>, d: Maybe<&2, Nat>) -> Retry.Budget:
368  match t d:
369    case None{} m:
370      Retry.Budget{attempts, None{}}
371    case Some{Try{f, st}} None{}:
372      Retry.Budget{Nat.sub(attempts, f), None{}}
373    case Some{Try{f, +st}} Some{ms}:
374      Retry.Budget{Nat.sub(attempts, f), Some{Nat.add(st, ms)}}
375
376def budget(p: Retry.Policy, t: Maybe<&2, Try>) -> Retry.Budget:
377  Retry.Policy{a, base, cap, d} = p
378  budget.of(a, t, d)
379
380# The failures so far: the index of the next retry.
381def fails(t: Maybe<&2, Try>) -> Nat:
382  match t:
383    case None{}:
384      0n
385    case Some{Try{f, st}}:
386      f
387
388# Why retry stopped a transient failure: no attempt is left, or the next wait would reach the deadline.
389def gave(b: Retry.Budget) -> Why:
390  Retry.Budget{left, d} = b
391  match left:
392    case 0n:
393      WhySpent{}
394    case 1n+k:
395      WhyLate{}
396
397# A settled attempt completes its ticket; retry gave up at now on any other.
398def stopped(-A: Data, o: Retry.Outcome, b: Retry.Budget, +id: Nat, +now: Nat) -> Op<A>:
399  match o:
400    case Retry.Settled{}:
401      Complete{id}
402    case Retry.Failed{}:
403      Abandon{id, WhyFailed{}, now}
404    case Retry.Rejected{}:
405      Abandon{id, WhyFailed{}, now}
406    case Retry.Unsafe{}:
407      Abandon{id, WhyUnsafe{}, now}
408    case Retry.Transient{w}:
409      Abandon{id, gave(b), now}
410
411def after.plan(-A: Data, pl: Retry.Plan, o: Retry.Outcome, b: Retry.Budget, +id: Nat, +now: Nat) -> Op<A>:
412  match pl:
413    case Retry.PlanStop{}:
414      stopped(A, o, b, id, now)
415    case Retry.PlanWait{w}:
416      Requeue{id, Nat.add(now, w)}
417
418# The record after an attempt of ticket id ends at now with o: requeue at now plus retry's wait,
419# complete when o is settled, or abandon when retry gave up.
420def after(-A: Data, +o: Retry.Outcome, p: Retry.Policy, +b: Retry.Budget, +id: Nat, +n: Nat, +now: Nat, +r: Nat) -> Op<A>:
421  after.plan(A, Retry.plan(o, p, b, n, now, r), o, b, id, now)
422
423# A target's breaker after an attempt ends at now with o.
424def brk.after(+circ: Retry.Circuit, old: Retry.State, +now: Nat, o: Retry.Outcome) -> Retry.Breaker:
425  Retry.Breaker.done(Retry.BreakerOn{circ, old}, now, Retry.failed(o))
426
427# When an open breaker lets attempts start again.
428def until(br: Retry.Breaker, +now: Nat) -> Nat:
429  match br:
430    case Retry.BreakerOn{c, Retry.CircuitOpen{u}}:
431      u
432    case Retry.BreakerOn{c, Retry.CircuitClosed{fs}}:
433      Nat.add(now, Retry.Circuit.cooldown(c))
434    case Retry.BreakerOn{c, Retry.CircuitHalf{k}}:
435      Nat.add(now, Retry.Circuit.cooldown(c))
436    case Retry.BreakerOff{}:
437      now
438
439# The record for a ticket that admit refused: Defer to when the breaker may admit, or Abandon when its budget is spent or late.
440def held(-A: Data, why: Retry.Deny, br: Retry.Breaker, +id: Nat, +now: Nat) -> Op<A>:
441  match why:
442    case Retry.DenyOpen{}:
443      Defer{id, until(br, now)}
444    case Retry.DenyLate{}:
445      Abandon{id, WhyLate{}, now}
446    case Retry.DenySpent{}:
447      Abandon{id, WhySpent{}, now}
448
449# ---- the record codec: a u32le tag, then u64le fields; an Enqueue then holds the payload.
450# An Abandon is tag 8, then id, why.code, and at. A Purge is tag 11 and before; a PurgeAll is tag 12.
451# A Break is tag 10, then its target and its breaker state.
452
453def why.code(w: Why) -> Nat:
454  match w:
455    case WhySpent{}:
456      0n
457    case WhyLate{}:
458      1n
459    case WhyUnsafe{}:
460      2n
461    case WhyFailed{}:
462      3n
463
464def W() -> Nat:
465  Nat.mul(65536n, 65536n)
466
467# Nat fields hold values under 2^64.
468def put.u64(b: Bytes.Bytes, +at: U32, +n: Nat) -> Bytes.Bytes:
469  Bytes.set.u32le(Bytes.set.u32le(b, at, U32.from_nat(Nat.mod(n, W()))), (at + 4 : U32), U32.from_nat(Nat.div(n, W())))
470
471def head(+tag: U32, +a: Nat, +c: Nat) -> Bytes.Bytes:
472  put.u64(put.u64(Bytes.set.u32le(Bytes.new(20), 0, tag), 4, a), 12, c)
473
474def snap.u32(out: Bytes.Bytes, +v: U32) -> Bytes.Bytes:
475  Bytes.append(out, Bytes.set.u32le(Bytes.new(4), 0, v))
476
477def snap.u64(out: Bytes.Bytes, +n: Nat) -> Bytes.Bytes:
478  Bytes.append(out, put.u64(Bytes.new(8), 0, n))
479
480def snap.chars(s: String, out: Bytes.Bytes) -> Bytes.Bytes:
481  match s:
482    case SNil{}:
483      out
484    case SCon{Chr{c}, t}:
485      snap.chars(t, snap.u32(out, c))
486
487# A target is a u32 count of chars, then each char as a u32.
488def snap.key(out: Bytes.Bytes, +k: String) -> Bytes.Bytes:
489  snap.chars(k, snap.u32(out, Bytes.count(k, 0)))
490
491def snap.nats(xs: List<&2, Nat>, out: Bytes.Bytes) -> Bytes.Bytes:
492  match xs:
493    case Nil{}:
494      out
495    case Con{x, rest}:
496      snap.nats(rest, snap.u64(out, x))
497
498# A breaker state is a u32 kind. 0 is closed: a u32 count, then each failure as a u64. 1 is open: until. 2 is half-open: busy.
499def snap.brk(out: Bytes.Bytes, b: Retry.State) -> Bytes.Bytes:
500  match b:
501    case Retry.CircuitClosed{+fs}:
502      snap.nats(fs, snap.u32(snap.u32(out, 0), U32.from_nat(List.length(&2, Nat, fs))))
503    case Retry.CircuitOpen{u}:
504      snap.u64(snap.u32(out, 1), u)
505    case Retry.CircuitHalf{k}:
506      snap.u64(snap.u32(out, 2), k)
507
508def encode(~A: Data, ~penc: A -> Bytes.Bytes, op: Op<A>) -> Bytes.Bytes:
509  match op:
510    case Enqueue{+d, x}:
511      Bytes.append(put.u64(Bytes.set.u32le(Bytes.new(12), 0, 1), 4, d), penc(x))
512    case Select{+now}:
513      head(2, now, 0n)
514    case Complete{+i}:
515      head(3, i, 0n)
516    case Cancel{+i}:
517      head(4, i, 0n)
518    case Requeue{+i, +d}:
519      head(5, i, d)
520    case Defer{+i, +d}:
521      head(6, i, d)
522    case Abandon{+i, w, +at}:
523      Bytes.append(head(8, i, why.code(w)), put.u64(Bytes.new(8), 0, at))
524    case Revive{+i, +d}:
525      head(9, i, d)
526    case Purge{+t}:
527      head(11, t, 0n)
528    case PurgeAll{}:
529      head(12, 0n, 0n)
530    case Break{k, b}:
531      snap.brk(snap.key(snap.u32(Bytes.new(0), 10), k), b)
532
533def get.field(m: Maybe<&2, U32>) -> U32:
534  match m:
535    case None{}:
536      0
537    case Some{x}:
538      x
539
540def get.u32(r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & U32:
541  (b, m) = r
542  (b, get.field(m))
543
544def get.hi(+lo: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Nat:
545  (b, m) = r
546  (b, Nat.add(U32.to_nat(lo), Nat.mul(U32.to_nat(get.field(m)), W())))
547
548def get.lo(+at: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Nat:
549  (b, m) = r
550  +lo = get.field(m)
551  get.hi(lo, Bytes.get.u32le(b, (at + 4 : U32)))
552
553def get.u64(b: Bytes.Bytes, +at: U32) -> Bytes.Bytes & Nat:
554  get.lo(at, Bytes.get.u32le(b, at))
555
556# ---- the snapshot codec: tag 7 and the next id, then the heap, the running list, the tries, the dead tickets, and the breakers.
557# A u32 flag starts each tree node and list cell (1) and marks each end (0), so decode rebuilds the same shape.
558# A ticket is its u64 id and deadline, then a u32 length and the payload.
559# The dead tickets are a trie: each node is a u32 kind, 0 empty, 1 a bucket, or 2 a branch, then its lo and hi nodes.
560# A bucket is its u32 hash, then a list of entries: the u64 key, the ticket, a u32 why.code, and the u64 abandon time.
561# A breaker is its target, then its state, as in a Break record.
562
563def snap.payload(out: Bytes.Bytes, p: Bytes.Bytes) -> Bytes.Bytes:
564  Bytes.Bytes{+len, buf} = p
565  Bytes.append(snap.u32(out, len), Bytes.Bytes{len, buf})
566
567def snap.ticket(~A: Data, ~penc: A -> Bytes.Bytes, out: Bytes.Bytes, t: Q.Ticket<A>) -> Bytes.Bytes:
568  Q.Ticket{+i, +d, x} = t
569  snap.payload(snap.u64(snap.u64(out, i), d), penc(x))
570
571def snap.heap(~A: Data, ~penc: A -> Bytes.Bytes, h: H.Heap<Q.Ticket<A>>, out: Bytes.Bytes) -> Bytes.Bytes:
572  match h:
573    case H.HLeaf{}:
574      snap.u32(out, 0)
575    case H.HNode{+n, t, l, r}:
576      snap.heap(~A, ~penc, r, snap.heap(~A, ~penc, l, snap.ticket(~A, ~penc, snap.u64(snap.u32(out, 1), n), t)))
577
578def snap.run(~A: Data, ~penc: A -> Bytes.Bytes, xs: List<&2, Q.Ticket<A>>, out: Bytes.Bytes) -> Bytes.Bytes:
579  match xs:
580    case Nil{}:
581      snap.u32(out, 0)
582    case Con{t, rest}:
583      snap.run(~A, ~penc, rest, snap.ticket(~A, ~penc, snap.u32(out, 1), t))
584
585def snap.tries(ts: Tries(), out: Bytes.Bytes) -> Bytes.Bytes:
586  match ts:
587    case O.OTip{}:
588      snap.u32(out, 0)
589    case O.OBin{+n, +k, Try{+f, +st}, lo, hi}:
590      snap.tries(hi, snap.tries(lo, snap.u64(snap.u64(snap.u64(snap.u32(snap.u32(out, 1), n), k), f), st)))
591
592def snap.entries(~A: Data, ~penc: A -> Bytes.Bytes, es: List<&2, M.Entry<Nat, Dead<A>>>, out: Bytes.Bytes) -> Bytes.Bytes:
593  match es:
594    case Nil{}:
595      snap.u32(out, 0)
596    case Con{M.Entry{k, Dead{t, w, at}}, rest}:
597      snap.entries(~A, ~penc, rest, snap.u64(snap.u32(snap.ticket(~A, ~penc, snap.u64(snap.u32(out, 1), k), t), U32.from_nat(why.code(w))), at))
598
599def snap.graves(~A: Data, ~penc: A -> Bytes.Bytes, ds: Graves(A), out: Bytes.Bytes) -> Bytes.Bytes:
600  match ds:
601    case M.HNil{}:
602      snap.u32(out, 0)
603    case M.HLeaf{h, es}:
604      snap.entries(~A, ~penc, es, snap.u32(snap.u32(out, 1), h))
605    case M.HBranch{lo, hi}:
606      snap.graves(~A, ~penc, hi, snap.graves(~A, ~penc, lo, snap.u32(out, 2)))
607
608def snap.brks(xs: List<&2, Sigma<&2, &2, String, _ => Retry.State>>, out: Bytes.Bytes) -> Bytes.Bytes:
609  match xs:
610    case Nil{}:
611      snap.u32(out, 0)
612    case Con{(k, b), rest}:
613      snap.brks(rest, snap.brk(snap.key(snap.u32(out, 1), k), b))
614
615# The Snap record of s.
616def encode.snap(~A: Data, ~penc: A -> Bytes.Bytes, s: State<A>) -> Bytes.Bytes:
617  State{Q.State{+n, h, run}, ts, ds, bs} = s
618  snap.brks(Map.to_list(&2, Retry.State, bs),
619    snap.graves(~A, ~penc, ds, snap.tries(ts, snap.run(~A, ~penc, run, snap.heap(~A, ~penc, h, snap.u64(snap.u32(Bytes.new(0), 7), n))))))
620
621def encode.rec(~A: Data, ~penc: A -> Bytes.Bytes, r: Rec<A>) -> Bytes.Bytes:
622  match r:
623    case Snap{s}:
624      encode.snap(~A, ~penc, s)
625    case Do{op}:
626      encode(~A, ~penc, op)
627
628# A snapshot reader: the record, and the offset of the next field.
629type Rd is Type:
630  Rd{b: Bytes.Bytes, at: U32}
631
632# One step of the snapshot decoder. None: a field runs past the record or does not decode.
633def Dec(-T: Type) -> Type:
634  Rd -> Maybe<&1, Rd & T>
635
636def Dec.next(-A: Type, -B: Type, m: Maybe<&1, Rd & A>, f: A -> Dec(B)) -> Maybe<&1, Rd & B>:
637  match m:
638    case None{}:
639      None{}
640    case Some{(r, x)}:
641      f(x)(r)
642
643def Dec.bind(-A: Type, -B: Type, m: Dec(A), f: A -> Dec(B)) -> Dec(B):
644  r => Dec.next(A, B, m(r), f)
645
646def Dec.pure(-A: Type, x: A) -> Dec(A):
647  r => Some{(r, x)}
648
649def Dec.fail(-A: Type) -> Dec(A):
650  r => None{}
651
652def rd.u32.of(m: Maybe<&2, U32>, b: Bytes.Bytes, +at: U32) -> Maybe<&1, Rd & U32>:
653  match m:
654    case None{}:
655      None{}
656    case Some{x}:
657      Some{(Rd{b, (at + 4 : U32)}, x)}
658
659def rd.u32.got(+at: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Maybe<&1, Rd & U32>:
660  (b, m) = r
661  rd.u32.of(m, b, at)
662
663def rd.u32.at(r: Rd) -> Maybe<&1, Rd & U32>:
664  Rd{b, +at} = r
665  rd.u32.got(at, Bytes.get.u32le(b, at))
666
667def rd.u32() -> Dec(U32):
668  r => rd.u32.at(r)
669
670def rd.u64.of(+lo: U32, +hi: U32) -> Nat:
671  Nat.add(U32.to_nat(lo), Nat.mul(U32.to_nat(hi), W()))
672
673def rd.u64() -> Dec(Nat):
674  do Dec<Nat>:
675    lo : U32 <- rd.u32()
676    hi : U32 <- rd.u32()
677    return rd.u64.of(lo, hi)
678
679def rd.cut.got(+at: U32, +n: U32, r: Bytes.Bytes & Bytes.Bytes) -> Maybe<&1, Rd & Bytes.Bytes>:
680  (b, p) = r
681  Some{(Rd{b, (at + n : U32)}, p)}
682
683def rd.cut.if(ok: Bool, b: Bytes.Bytes, +at: U32, +n: U32) -> Maybe<&1, Rd & Bytes.Bytes>:
684  match ok:
685    case False{}:
686      None{}
687    case True{}:
688      rd.cut.got(at, n, Bytes.slice(b, at, n))
689
690def rd.cut.at(+n: U32, r: Rd) -> Maybe<&1, Rd & Bytes.Bytes>:
691  Rd{Bytes.Bytes{+len, buf}, +at} = r
692  rd.cut.if(Bytes.fits(len, at, n), Bytes.Bytes{len, buf}, at, n)
693
694# The next n octets.
695def rd.cut(+n: U32) -> Dec(Bytes.Bytes):
696  r => rd.cut.at(n, r)
697
698def rd.flag.of(+m: U32) -> Dec(Bool):
699  Bool.pick(Dec(Bool), U32.is_eq(m, 0), Dec.pure(Bool, False{}), Bool.pick(Dec(Bool), U32.is_eq(m, 1), Dec.pure(Bool, True{}), Dec.fail(Bool)))
700
701# A u32 flag: 1 is True, 0 is False, and any other value fails.
702def rd.flag() -> Dec(Bool):
703  Dec.bind(U32, Bool, rd.u32(), m => rd.flag.of(m))
704
705def rd.payload(-A: Data, +i: Nat, +d: Nat, m: Maybe<&2, A>) -> Dec(Q.Ticket<A>):
706  match m:
707    case None{}:
708      Dec.fail(Q.Ticket<A>)
709    case Some{x}:
710      Dec.pure(Q.Ticket<A>, Q.Ticket{i, d, x})
711
712def rd.ticket(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>) -> Dec(Q.Ticket<A>):
713  do Dec<Q.Ticket<A>>:
714    i : Nat <- rd.u64()
715    d : Nat <- rd.u64()
716    n : U32 <- rd.u32()
717    p : Bytes.Bytes <- rd.cut(n)
718    rd.payload(A, i, d, pdec(p))
719
720# Each rd.* tree reader takes the flag already read. fuel bounds the depth: every node holds at least one 4-octet flag.
721def rd.heap(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +fuel: Nat, node: Bool) -> Dec(H.Heap<Q.Ticket<A>>):
722  match fuel node:
723    case 0n b:
724      Dec.fail(H.Heap<Q.Ticket<A>>)
725    case 1n+f False{}:
726      Dec.pure(H.Heap<Q.Ticket<A>>, H.HLeaf{})
727    case 1n+f True{}:
728      do Dec<H.Heap<Q.Ticket<A>>>:
729        n : Nat <- rd.u64()
730        t : Q.Ticket<A> <- rd.ticket(~A, ~pdec)
731        lf : Bool <- rd.flag()
732        l : H.Heap<Q.Ticket<A>> <- rd.heap(~A, ~pdec, f, lf)
733        rf : Bool <- rd.flag()
734        r : H.Heap<Q.Ticket<A>> <- rd.heap(~A, ~pdec, f, rf)
735        return H.HNode{n, t, l, r}
736
737def rd.run(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +fuel: Nat, more: Bool) -> Dec(List<&2, Q.Ticket<A>>):
738  match fuel more:
739    case 0n b:
740      Dec.fail(List<&2, Q.Ticket<A>>)
741    case 1n+f False{}:
742      Dec.pure(List<&2, Q.Ticket<A>>, Nil{})
743    case 1n+f True{}:
744      do Dec<List<&2, Q.Ticket<A>>>:
745        t : Q.Ticket<A> <- rd.ticket(~A, ~pdec)
746        m : Bool <- rd.flag()
747        rest : List<&2, Q.Ticket<A>> <- rd.run(~A, ~pdec, f, m)
748        return Con{t, rest}
749
750def rd.tries(+fuel: Nat, node: Bool) -> Dec(O.OMap<Nat, Try>):
751  match fuel node:
752    case 0n b:
753      Dec.fail(O.OMap<Nat, Try>)
754    case 1n+f False{}:
755      Dec.pure(O.OMap<Nat, Try>, O.OTip{})
756    case 1n+f True{}:
757      do Dec<O.OMap<Nat, Try>>:
758        n : U32 <- rd.u32()
759        k : Nat <- rd.u64()
760        fails : Nat <- rd.u64()
761        st : Nat <- rd.u64()
762        lf : Bool <- rd.flag()
763        lo : O.OMap<Nat, Try> <- rd.tries(f, lf)
764        hf : Bool <- rd.flag()
765        hi : O.OMap<Nat, Try> <- rd.tries(f, hf)
766        return O.OBin{n, k, Try{fails, st}, lo, hi}
767
768def rd.why(+c: U32) -> Dec(Why):
769  Bool.pick(Dec(Why), U32.is_eq(c, 0), Dec.pure(Why, WhySpent{}),
770  Bool.pick(Dec(Why), U32.is_eq(c, 1), Dec.pure(Why, WhyLate{}),
771  Bool.pick(Dec(Why), U32.is_eq(c, 2), Dec.pure(Why, WhyUnsafe{}),
772  Bool.pick(Dec(Why), U32.is_eq(c, 3), Dec.pure(Why, WhyFailed{}), Dec.fail(Why)))))
773
774def rd.entries(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +fuel: Nat, more: Bool) -> Dec(List<&2, M.Entry<Nat, Dead<A>>>):
775  match fuel more:
776    case 0n b:
777      Dec.fail(List<&2, M.Entry<Nat, Dead<A>>>)
778    case 1n+f False{}:
779      Dec.pure(List<&2, M.Entry<Nat, Dead<A>>>, Nil{})
780    case 1n+f True{}:
781      do Dec<List<&2, M.Entry<Nat, Dead<A>>>>:
782        k : Nat <- rd.u64()
783        t : Q.Ticket<A> <- rd.ticket(~A, ~pdec)
784        c : U32 <- rd.u32()
785        w : Why <- rd.why(c)
786        at : Nat <- rd.u64()
787        m : Bool <- rd.flag()
788        rest : List<&2, M.Entry<Nat, Dead<A>>> <- rd.entries(~A, ~pdec, f, m)
789        return Con{M.Entry{k, Dead{t, w, at}}, rest}
790
791# A trie node whose kind is already read: nil is kind 0, leaf kind 1, and branch kind 2. Any other kind fails.
792def rd.graves(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +fuel: Nat, nil: Bool, leaf: Bool, branch: Bool) -> Dec(Graves(A)):
793  match fuel nil leaf branch:
794    case 0n a b c:
795      Dec.fail(Graves(A))
796    case 1n+f True{} b c:
797      Dec.pure(Graves(A), M.HNil{})
798    case 1n+f False{} True{} c:
799      do Dec<Graves(A)>:
800        h : U32 <- rd.u32()
801        m : Bool <- rd.flag()
802        es : List<&2, M.Entry<Nat, Dead<A>>> <- rd.entries(~A, ~pdec, f, m)
803        return M.HLeaf{h, es}
804    case 1n+f False{} False{} True{}:
805      do Dec<Graves(A)>:
806        +lk : U32 <- rd.u32()
807        lo : Graves(A) <- rd.graves(~A, ~pdec, f, U32.is_eq(lk, 0), U32.is_eq(lk, 1), U32.is_eq(lk, 2))
808        +hk : U32 <- rd.u32()
809        hi : Graves(A) <- rd.graves(~A, ~pdec, f, U32.is_eq(hk, 0), U32.is_eq(hk, 1), U32.is_eq(hk, 2))
810        return M.HBranch{lo, hi}
811    case 1n+f False{} False{} False{}:
812      Dec.fail(Graves(A))
813
814# The trie, when it is well formed as the never-selected laws assume.
815def rd.wf(-A: Data, ok: Bool, ds: Graves(A)) -> Dec(Graves(A)):
816  match ok:
817    case True{}:
818      Dec.pure(Graves(A), ds)
819    case False{}:
820      Dec.fail(Graves(A))
821
822# Each rd.* sequence reader takes the count or flag already read. fuel bounds its length: every item holds at least 4 octets.
823def rd.chars(+fuel: Nat, end: Bool, +n: U32) -> Dec(String):
824  match fuel end:
825    case 0n b:
826      Dec.fail(String)
827    case 1n+f True{}:
828      Dec.pure(String, SNil{})
829    case 1n+f False{}:
830      +k = (n - 1 : U32)
831      do Dec<String>:
832        c : U32 <- rd.u32()
833        rest : String <- rd.chars(f, U32.is_eq(k, 0), k)
834        return SCon{Chr{c}, rest}
835
836def rd.key(+fuel: Nat) -> Dec(String):
837  do Dec<String>:
838    +n : U32 <- rd.u32()
839    rd.chars(fuel, U32.is_eq(n, 0), n)
840
841def rd.nats(+fuel: Nat, end: Bool, +n: U32) -> Dec(List<&2, Nat>):
842  match fuel end:
843    case 0n b:
844      Dec.fail(List<&2, Nat>)
845    case 1n+f True{}:
846      Dec.pure(List<&2, Nat>, Nil{})
847    case 1n+f False{}:
848      +k = (n - 1 : U32)
849      do Dec<List<&2, Nat>>:
850        x : Nat <- rd.u64()
851        rest : List<&2, Nat> <- rd.nats(f, U32.is_eq(k, 0), k)
852        return Con{x, rest}
853
854def rd.closed(+fuel: Nat) -> Dec(Retry.State):
855  do Dec<Retry.State>:
856    +n : U32 <- rd.u32()
857    fs : List<&2, Nat> <- rd.nats(fuel, U32.is_eq(n, 0), n)
858    return Retry.CircuitClosed{fs}
859
860def rd.brk.of(+fuel: Nat, +kind: U32) -> Dec(Retry.State):
861  Bool.pick(Dec(Retry.State), U32.is_eq(kind, 0), rd.closed(fuel),
862  Bool.pick(Dec(Retry.State), U32.is_eq(kind, 1), Dec.bind(Nat, Retry.State, rd.u64(), u => Dec.pure(Retry.State, Retry.CircuitOpen{u})),
863  Bool.pick(Dec(Retry.State), U32.is_eq(kind, 2), Dec.bind(Nat, Retry.State, rd.u64(), k => Dec.pure(Retry.State, Retry.CircuitHalf{k})), Dec.fail(Retry.State))))
864
865def rd.brk(+fuel: Nat) -> Dec(Retry.State):
866  Dec.bind(U32, Retry.State, rd.u32(), k => rd.brk.of(fuel, k))
867
868def rd.break(-A: Data, +fuel: Nat) -> Dec(Op<A>):
869  do Dec<Op<A>>:
870    k : String <- rd.key(fuel)
871    b : Retry.State <- rd.brk(fuel)
872    return Break{k, b}
873
874def rd.brks(+fuel: Nat, more: Bool, acc: Map<&2, Retry.State>) -> Dec(Map<&2, Retry.State>):
875  match fuel more:
876    case 0n b:
877      Dec.fail(Map<&2, Retry.State>)
878    case 1n+f False{}:
879      Dec.pure(Map<&2, Retry.State>, acc)
880    case 1n+f True{}:
881      do Dec<Map<&2, Retry.State>>:
882        k : String <- rd.key(f)
883        b : Retry.State <- rd.brk(f)
884        m : Bool <- rd.flag()
885        rd.brks(f, m, Map.set(&2, Retry.State, acc, k, b))
886
887def rd.state(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +fuel: Nat) -> Dec(State<A>):
888  do Dec<State<A>>:
889    n : Nat <- rd.u64()
890    hf : Bool <- rd.flag()
891    h : H.Heap<Q.Ticket<A>> <- rd.heap(~A, ~pdec, fuel, hf)
892    rf : Bool <- rd.flag()
893    run : List<&2, Q.Ticket<A>> <- rd.run(~A, ~pdec, fuel, rf)
894    tf : Bool <- rd.flag()
895    ts : O.OMap<Nat, Try> <- rd.tries(fuel, tf)
896    +dk : U32 <- rd.u32()
897    +dt : Graves(A) <- rd.graves(~A, ~pdec, fuel, U32.is_eq(dk, 0), U32.is_eq(dk, 1), U32.is_eq(dk, 2))
898    ds : Graves(A) <- rd.wf(A, CL.hmap_wf(~Nat, ~Dead<A>, ~dead.hash, ~Nat.is_eq, dt), dt)
899    bf : Bool <- rd.flag()
900    bs : Map<&2, Retry.State> <- rd.brks(fuel, bf, Map.new(&2, Retry.State))
901    return State{Q.State{n, h, run}, ts, ds, bs}
902
903# The value read, when the reader ended at len, the record's end.
904def rd.done(-T: Data, +len: U32, m: Maybe<&1, Rd & T>) -> Maybe<&2, T>:
905  match m:
906    case None{}:
907      None{}
908    case Some{(Rd{b, +at}, +x)}:
909      Bool.pick(Maybe<&2, T>, U32.is_eq(at, len), Some{x}, None{})
910
911# The Abandon of ticket a at e whose why.code is c, or None for any other code.
912def decode.abandon(-A: Data, +a: Nat, +c: Nat, +e: Nat) -> Maybe<&2, Op<A>>:
913  Bool.pick(Maybe<&2, Op<A>>, Nat.is_eq(c, 0n), Some{Abandon{a, WhySpent{}, e}},
914  Bool.pick(Maybe<&2, Op<A>>, Nat.is_eq(c, 1n), Some{Abandon{a, WhyLate{}, e}},
915  Bool.pick(Maybe<&2, Op<A>>, Nat.is_eq(c, 2n), Some{Abandon{a, WhyUnsafe{}, e}},
916  Bool.pick(Maybe<&2, Op<A>>, Nat.is_eq(c, 3n), Some{Abandon{a, WhyFailed{}, e}}, None{}))))
917
918def decode.small(-A: Data, ok: Bool, +tag: U32, +a: Nat, +c: Nat) -> Maybe<&2, Op<A>>:
919  match ok:
920    case False{}:
921      None{}
922    case True{}:
923      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 2), Some{Select{a}},
924      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 3), Some{Complete{a}},
925      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 4), Some{Cancel{a}},
926      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 5), Some{Requeue{a, c}},
927      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 6), Some{Defer{a, c}},
928      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 9), Some{Revive{a, c}},
929      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 11), Some{Purge{a}},
930      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 12), Some{PurgeAll{}}, None{}))))))))
931
932def decode.long(-A: Data, ok: Bool, +a: Nat, +c: Nat, r: Bytes.Bytes & Nat) -> Maybe<&2, Op<A>>:
933  match ok:
934    case False{}:
935      None{}
936    case True{}:
937      (b, +e) = r
938      decode.abandon(A, a, c, e)
939
940def decode.payload(-A: Data, m: Maybe<&2, A>, +d: Nat) -> Maybe<&2, Op<A>>:
941  match m:
942    case None{}:
943      None{}
944    case Some{x}:
945      Some{Enqueue{d, x}}
946
947def decode.c.if(-A: Data, abandon: Bool, +tag: U32, +len: U32, +a: Nat, +c: Nat, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
948  match abandon:
949    case True{}:
950      decode.long(A, U32.is_eq(len, 28), a, c, get.u64(b, 20))
951    case False{}:
952      decode.small(A, U32.is_eq(len, 20), tag, a, c)
953
954# An Abandon is 28 octets; every other op without a payload is 20.
955def decode.c(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +tag: U32, +len: U32, +a: Nat, r: Bytes.Bytes & Nat) -> Maybe<&2, Op<A>>:
956  (b, +c) = r
957  decode.c.if(A, U32.is_eq(tag, 8), tag, len, a, c, b)
958
959def decode.kind(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, enq: Bool, +tag: U32, +len: U32, +a: Nat, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
960  match enq:
961    case True{}:
962      decode.payload(A, pdec(Bytes.snd(Bytes.slice(b, 12, (len - 12 : U32)))), a)
963    case False{}:
964      decode.c(~A, ~pdec, tag, len, a, get.u64(b, 12))
965
966def decode.a(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +tag: U32, +len: U32, r: Bytes.Bytes & Nat) -> Maybe<&2, Op<A>>:
967  (b, +a) = r
968  decode.kind(~A, ~pdec, U32.is_eq(tag, 1), tag, len, a, b)
969
970def decode.tag.if(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, brk: Bool, +tag: U32, +len: U32, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
971  match brk:
972    case True{}:
973      rd.done(Op<A>, len, rd.break(A, U32.to_nat((len / 4 + 1 : U32)))(Rd{b, 4}))
974    case False{}:
975      decode.a(~A, ~pdec, tag, len, get.u64(b, 4))
976
977def decode.tag(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +len: U32, r: Bytes.Bytes & U32) -> Maybe<&2, Op<A>>:
978  (b, +tag) = r
979  decode.tag.if(~A, ~pdec, U32.is_eq(tag, 10), tag, len, b)
980
981def decode.len(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, ok: Bool, +len: U32, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
982  match ok:
983    case False{}:
984      None{}
985    case True{}:
986      decode.tag(~A, ~pdec, len, get.u32(Bytes.get.u32le(b, 0)))
987
988# The op of a record, or None when it is not one that encode writes.
989def decode(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
990  Bytes.Bytes{+len, buf} = b
991  decode.len(~A, ~pdec, U32.is_ge(len, 12), len, Bytes.Bytes{len, buf})
992
993# The state of a Snap record whose tag is already checked. The decoder must end at the record's end.
994def decode.snap(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, b: Bytes.Bytes) -> Maybe<&2, State<A>>:
995  Bytes.Bytes{+len, buf} = b
996  rd.done(State<A>, len, rd.state(~A, ~pdec, U32.to_nat((len / 4 + 1 : U32)))(Rd{Bytes.Bytes{len, buf}, 4}))
997
998def decode.snapped(-A: Data, m: Maybe<&2, State<A>>) -> Maybe<&2, Rec<A>>:
999  match m:
1000    case None{}:
1001      None{}
1002    case Some{s}:
1003      Some{Snap{s}}
1004
1005def decode.did(-A: Data, m: Maybe<&2, Op<A>>) -> Maybe<&2, Rec<A>>:
1006  match m:
1007    case None{}:
1008      None{}
1009    case Some{op}:
1010      Some{Do{op}}
1011
1012def decode.rec.if(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, snap: Bool, b: Bytes.Bytes) -> Maybe<&2, Rec<A>>:
1013  match snap:
1014    case True{}:
1015      decode.snapped(A, decode.snap(~A, ~pdec, b))
1016    case False{}:
1017      decode.did(A, decode(~A, ~pdec, b))
1018
1019def decode.rec.tag(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, r: Bytes.Bytes & U32) -> Maybe<&2, Rec<A>>:
1020  (b, +tag) = r
1021  decode.rec.if(~A, ~pdec, U32.is_eq(tag, 7), b)
1022
1023# The Rec of a journal record, or None when it is neither a Snap nor an op that encode.rec writes.
1024def decode.rec(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, b: Bytes.Bytes) -> Maybe<&2, Rec<A>>:
1025  decode.rec.tag(~A, ~pdec, get.u32(Bytes.get.u32le(b, 0)))
1026
1027# ---- the runner
1028
1029# policy: retry's policy for each ticket. circuit: the breaker of each target. idle: ms that drain waits when no attempt started.
1030# compact: tick compacts the journal once records past this many octets were appended since open or the last compaction; 0 never.
1031type Config is Data:
1032  Config{policy: Retry.Policy, circuit: Retry.Circuit, idle: U32, compact: U32}
1033
1034# The journal, its path, its file cap, and its length at open or at the last compaction.
1035type Log is Type:
1036  Log{journal: J.Journal, path: String, limit: U32, base: U32}
1037
1038# An attempt that runs: its unit of the pool, the budget that admit left, and its target.
1039type Flight<-A: Data> is Type:
1040  Flight{ticket: Q.Ticket<A>, grant: Res.Grant, budget: Retry.Budget, target: String}
1041
1042def Out(-X: Type) -> Type:
1043  Result<&1, &1, U32 & String, X>
1044
1045# A message to the fiber that owns the runner. Ended: an attempt finished. Add and Drop: another fiber asks for an enqueue or a cancel.
1046type Msg<-A: Data> is Type:
1047  Ended{id: Nat, outcome: Retry.Outcome}
1048  Add{deadline: Nat, payload: A, reply: Chan(Out(Nat))}
1049  Drop{id: Nat, reply: Chan(Out(Unit))}
1050
1051# An open runner. Replaying its journal gives its state; its pool bounds the attempts that run at once.
1052# Each attempt sends its outcome, and each other fiber its ask, on inbox. flights are the attempts that run.
1053type Runner<-A: Data> is Type:
1054  Runner{log: Log, state: State<A>, pool: Res.Pool, config: Config, inbox: Chan(Msg<A>), flights: List<&1, Flight<A>>}
1055
1056# The runner, the claim fuel left, and the first error.
1057def Claimed(-A: Data) -> Type:
1058  Runner<A> & Nat & Out(Unit)
1059
1060def Ran(-A: Data) -> Type:
1061  Runner<A> & Out(Unit)
1062
1063# Wall-clock ms since the Unix epoch. Deadlines must keep their meaning across a restart.
1064def clock() -> IO(Nat):
1065  do IO<Nat>:
1066    t : Time.Instant <- Time.now()
1067    return Time.Duration.to_ms_nat(Time.Instant.since_epoch(t))
1068
1069def first(a: Out(Unit), b: Out(Unit)) -> Out(Unit):
1070  match a:
1071    case Fail{e}:
1072      Fail{e}
1073    case Done{u}:
1074      b
1075
1076def log.appended(+path: String, +limit: U32, +base: U32, x: J.Journal & Out(Unit)) -> Log & Out(Unit):
1077  (j, o) = x
1078  (Log{j, path, limit, base}, o)
1079
1080def log.append(l: Log, p: Bytes.Bytes) -> IO(Log & Out(Unit)):
1081  Log{j, +path, +limit, +base} = l
1082  do IO<Log & Out(Unit)>:
1083    x : J.Journal & Out(Unit) <- J.append(j, p)
1084    return log.appended(path, limit, base, x)
1085
1086def log.close(l: Log) -> IO(Unit):
1087  Log{j, path, limit, base} = l
1088  J.close(j)
1089
1090# The journal reads with limit as its record cap, so that a snapshot fits; appends keep max.
1091def log.of(j: J.Journal, +max: U32, path: String, +limit: U32) -> Log:
1092  J.Journal{file, cap, +end, ok} = j
1093  Log{J.Journal{file, max, end, ok}, path, limit, end}
1094
1095def commit.done(~A: Data, x: Log & Out(Unit), s: State<A>, +op: Op<A>, p: Res.Pool, c: Config, ib: Chan(Msg<A>), fs: List<&1, Flight<A>>) -> Ran(A):
1096  match x:
1097    case (j, Done{u}):
1098      (Runner{j, step(~A, s, op), p, c, ib, fs}, Done{u})
1099    case (j, Fail{e}):
1100      (Runner{j, s, p, c, ib, fs}, Fail{e})
1101
1102# Appends op, and applies it once the append is durable. After a failed append the state is as it was,
1103# and the journal refuses every later append until the runner is reopened.
1104def commit(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +op: Op<A>) -> IO(Ran(A)):
1105  Runner{j, s, p, c, ib, fs} = r
1106  do IO<Ran(A)>:
1107    x : Log & Out(Unit) <- log.append(j, encode(~A, ~penc, op))
1108    return commit.done(~A, x, s, op, p, c, ib, fs)
1109
1110def enqueue.done(-A: Data, +id: Nat, y: Ran(A)) -> Runner<A> & Out(Nat):
1111  match y:
1112    case (r, Done{u}):
1113      (r, Done{id})
1114    case (r, Fail{e}):
1115      (r, Fail{e})
1116
1117# Adds a job that is due at deadline (wall-clock ms) and answers its ticket id once the record is durable.
1118def enqueue(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +deadline: Nat, x: A) -> IO(Runner<A> & Out(Nat)):
1119  Runner{j, +s, p, c, ib, fs} = r
1120  do IO<Runner<A> & Out(Nat)>:
1121    y : Ran(A) <- commit(~A, ~penc, Runner{j, s, p, c, ib, fs}, Enqueue{deadline, x})
1122    return enqueue.done(A, Q.next_id(A, queue(A, s)), y)
1123
1124# Cancels ticket id. A running attempt finishes, but its ticket is never selected again. A dead ticket is dropped.
1125def cancel(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +id: Nat) -> IO(Ran(A)):
1126  commit(~A, ~penc, r, Cancel{id})
1127
1128# Makes dead ticket id pending again at deadline (wall-clock ms), with its payload and a new retry budget.
1129# A ticket that is not dead does not change.
1130def revive(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +id: Nat, +deadline: Nat) -> IO(Ran(A)):
1131  commit(~A, ~penc, r, Revive{id, deadline})
1132
1133# Drops, as one record, the dead tickets that retry gave up on before `before` (wall-clock ms).
1134def purge(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +before: Nat) -> IO(Ran(A)):
1135  commit(~A, ~penc, r, Purge{before})
1136
1137# Drops every dead ticket, as one record.
1138def purge.all(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>) -> IO(Ran(A)):
1139  commit(~A, ~penc, r, PurgeAll{})
1140
1141# ---- asks from other fibers
1142
1143# The runner back, and its inbox. Another fiber passes the inbox to ask.enqueue and ask.cancel.
1144def inbox(-A: Data, r: Runner<A>) -> Runner<A> & Chan(Msg<A>):
1145  Runner{j, s, p, c, +ib, fs} = r
1146  (Runner{j, s, p, c, ib, fs}, ib)
1147
1148def ask.answer(-X: Type, m: Maybe<&1, Out(X)>) -> Out(X):
1149  match m:
1150    case None{}:
1151      Fail{(32, "runner: the runner closed before it answered")}
1152    case Some{o}:
1153      o
1154
1155def ask.sent(-A: Data, -X: Type, x: Result<&1, &1, Msg<A>, Unit>, +rep: Chan(Out(X))) -> IO(Out(X)):
1156  match x:
1157    case Fail{m}:
1158      IO.pure(Out(X), Fail{(32, "runner: the inbox is closed")})
1159    case Done{u}:
1160      do IO<Out(X)>:
1161        a : Maybe<&1, Out(X)> <- Chan.recv(Out(X), rep)
1162        return ask.answer(X, a)
1163
1164# Sends m, built with a new reply channel, on the inbox and waits for the reply.
1165def ask(-A: Data, -X: Type, +ib: Chan(Msg<A>), m: Chan(Out(X)) -> Msg<A>) -> IO(Out(X)):
1166  do IO<Out(X)>:
1167    +rep : Chan(Out(X)) <- Chan.new(Out(X), 1)
1168    x : Result<&1, &1, Msg<A>, Unit> <- Chan.send(Msg<A>, ib, m(rep))
1169    ask.sent(A, X, x, rep)
1170
1171# From another fiber: adds a job due at deadline, and answers its ticket id once the record is durable.
1172# The answer comes when the fiber that owns the runner reads the inbox, in tick or drain.
1173def ask.enqueue(-A: Data, +ib: Chan(Msg<A>), +deadline: Nat, x: A) -> IO(Out(Nat)):
1174  ask(A, Nat, ib, rep => Add{deadline, x, rep})
1175
1176# From another fiber: cancels ticket id, as cancel does. The answer comes when the owner reads the inbox.
1177def ask.cancel(-A: Data, +ib: Chan(Msg<A>), +id: Nat) -> IO(Out(Unit)):
1178  ask(A, Unit, ib, rep => Drop{id, rep})
1179
1180# ---- one round
1181
1182# A grant of this runner's pool always fits it, so Kept does not happen.
1183def release.got(-A: Data, x: Res.Release, j: Log, s: State<A>, c: Config, ib: Chan(Msg<A>), fs: List<&1, Flight<A>>) -> Runner<A>:
1184  match x:
1185    case Res.Released{p}:
1186      Runner{j, s, p, c, ib, fs}
1187    case Res.Kept{p, g}:
1188      Runner{j, s, p, c, ib, fs}
1189
1190def release(-A: Data, r: Runner<A>, g: Res.Grant) -> Runner<A>:
1191  Runner{j, s, p, c, ib, fs} = r
1192  release.got(A, Res.release(p, g), j, s, c, ib, fs)
1193
1194def claim.held(-A: Data, y: Ran(A), g: Res.Grant, next: Runner<A> -> IO(Claimed(A))) -> IO(Claimed(A)):
1195  match y:
1196    case (r, Fail{e}):
1197      IO.pure(Claimed(A), (release(A, r, g), 0n, Fail{e}))
1198    case (r, Done{u}):
1199      next(release(A, r, g))
1200
1201def claim.tag(-A: Data, +id: Nat, o: Retry.Outcome) -> IO(Msg<A>):
1202  IO.pure(Msg<A>, Ended{id, o})
1203
1204def trip.if(~A: Data, ~penc: A -> Bytes.Bytes, same: Bool, r: Runner<A>, +key: String, st: Retry.State) -> IO(Ran(A)):
1205  match same:
1206    case True{}:
1207      IO.pure(Ran(A), (r, Done{Unit{}}))
1208    case False{}:
1209      commit(~A, ~penc, r, Break{key, st})
1210
1211# Appends a Break when the target's breaker left old, so that a reopen restores it.
1212def trip(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +key: String, +old: Retry.State, br: Retry.Breaker) -> IO(Ran(A)):
1213  +st = brk.state(br, old)
1214  trip.if(~A, ~penc, brk.same(old, st), r, key, st)
1215
1216def claim.go(~A: Data, ~penc: A -> Bytes.Bytes, ~work: A -> IO(Retry.Outcome), +now: Nat, gate: Retry.Gate, +br: Retry.Breaker, b: Retry.Budget, +key: String,
1217    r: Runner<A>, +id: Nat, d: Nat, +x: A, g: Res.Grant, next: Runner<A> -> IO(Claimed(A))) -> IO(Claimed(A)):
1218  match gate:
1219    case Retry.GateGo{}:
1220      Runner{j, s, p, c, +ib, fs} = r
1221      do IO<Claimed(A)>:
1222        u : Chan(Msg<A>) <- IO.fork.go(Msg<A>, IO.bind(Retry.Outcome, Msg<A>, work(x), o => claim.tag(A, id, o)), ib)
1223        next(Runner{j, s, p, c, ib, Con{Flight{Q.Ticket{id, d, x}, g, b, key}, fs}})
1224    case Retry.GateNo{why}:
1225      do IO<Claimed(A)>:
1226        y : Ran(A) <- commit(~A, ~penc, r, held(A, why, br, id, now))
1227        claim.held(A, y, g, next)
1228
1229def claim.tripped(~A: Data, ~penc: A -> Bytes.Bytes, ~work: A -> IO(Retry.Outcome), +now: Nat, y: Ran(A), gate: Retry.Gate, +br: Retry.Breaker, b: Retry.Budget, +key: String,
1230    +id: Nat, d: Nat, +x: A, g: Res.Grant, next: Runner<A> -> IO(Claimed(A))) -> IO(Claimed(A)):
1231  match y:
1232    case (r, Fail{e}):
1233      IO.pure(Claimed(A), (release(A, r, g), 0n, Fail{e}))
1234    case (r, Done{u}):
1235      claim.go(~A, ~penc, ~work, now, gate, br, b, key, r, id, d, x, g, next)
1236
1237# A change to the breaker is durable before the attempt starts.
1238def claim.gate(~A: Data, ~penc: A -> Bytes.Bytes, ~work: A -> IO(Retry.Outcome), +now: Nat, a: Retry.Breaker & Retry.Budget & Retry.Gate, +key: String, +old: Retry.State,
1239    r: Runner<A>, +id: Nat, d: Nat, +x: A, g: Res.Grant, next: Runner<A> -> IO(Claimed(A))) -> IO(Claimed(A)):
1240  (+br, b, gate) = a
1241  do IO<Claimed(A)>:
1242    y : Ran(A) <- trip(~A, ~penc, r, key, old, br)
1243    claim.tripped(~A, ~penc, ~work, now, y, gate, br, b, key, id, d, x, g, next)
1244
1245# The selected ticket asks its target's breaker and its own budget for an attempt.
1246def claim.admit(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +now: Nat, r: Runner<A>, t: Q.Ticket<A>, g: Res.Grant,
1247    next: Runner<A> -> IO(Claimed(A))) -> IO(Claimed(A)):
1248  Runner{j, +s, p, +c, ib, fs} = r
1249  Config{pol, circ, idle, every} = c
1250  Q.Ticket{+id, d, +x} = t
1251  +key = target(x)
1252  +old = brk.get(A, s, key)
1253  claim.gate(~A, ~penc, ~work, now, Retry.admit(Retry.BreakerOn{circ, old}, budget(pol, tries.get(tries(A, s), id)), now), key, old,
1254    Runner{j, s, p, c, ib, fs}, id, d, x, g, next)
1255
1256def claim.selected(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +now: Nat, y: Ran(A),
1257    t: Q.Ticket<A>, g: Res.Grant, next: Runner<A> -> IO(Claimed(A))) -> IO(Claimed(A)):
1258  match y:
1259    case (r, Fail{e}):
1260      IO.pure(Claimed(A), (release(A, r, g), 0n, Fail{e}))
1261    case (r, Done{u}):
1262      claim.admit(~A, ~penc, ~target, ~work, now, r, t, g, next)
1263
1264def claim.pick(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +left: Nat, +now: Nat, k: Pick<A>, j: Log, s: State<A>,
1265    c: Config, ib: Chan(Msg<A>), fs: List<&1, Flight<A>>, next: Runner<A> -> IO(Claimed(A))) -> IO(Claimed(A)):
1266  match k:
1267    case Idle{p}:
1268      IO.pure(Claimed(A), (Runner{j, s, p, c, ib, fs}, Nat.add(1n, left), Done{Unit{}}))
1269    case Picked{p, t, g}:
1270      do IO<Claimed(A)>:
1271        y : Ran(A) <- commit(~A, ~penc, Runner{j, s, p, c, ib, fs}, Select{now})
1272        claim.selected(~A, ~penc, ~target, ~work, now, y, t, g, next)
1273
1274# Starts an attempt for each due ticket while the pool has a free unit; each attempt sends its outcome on the inbox.
1275# fuel bounds the tickets taken, and the fuel left comes back.
1276def claim(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), fuel: Nat, +now: Nat, r: Runner<A>) -> IO(Claimed(A)):
1277  match fuel:
1278    case 0n:
1279      IO.pure(Claimed(A), (r, 0n, Done{Unit{}}))
1280    case 1n+n:
1281      Runner{j, +s, p, c, ib, fs} = r
1282      +m = n
1283      claim.pick(~A, ~penc, ~target, ~work, m, now, pick(~A, p, s, now), j, s, c, ib, fs, (r2 => claim(~A, ~penc, ~target, ~work, m, now, r2)))
1284
1285def settle.done(-A: Data, y: Ran(A), g: Res.Grant, err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1286  (r, res) = y
1287  next(release(A, r, g), first(err, res))
1288
1289def settle.br(~A: Data, ~penc: A -> Bytes.Bytes, +op: Op<A>, y: Ran(A), g: Res.Grant, err: Out(Unit),
1290    next: Runner<A> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1291  match y:
1292    case (r, Fail{e}):
1293      next(release(A, r, g), first(err, Fail{e}))
1294    case (r, Done{u}):
1295      do IO<Runner<A> & Out(Bool)>:
1296        z : Ran(A) <- commit(~A, ~penc, r, op)
1297        settle.done(A, z, g, err, next)
1298
1299# Tells the target's breaker, appends the outcome's op, and returns the attempt's unit to the pool.
1300# The op of a cancelled ticket changes no state: the queue no longer holds it.
1301def settle.op(~A: Data, ~penc: A -> Bytes.Bytes, +o: Retry.Outcome, b: Retry.Budget, +id: Nat, +key: String, +now: Nat, +rnd: Nat, r: Runner<A>, g: Res.Grant,
1302    err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1303  Runner{j, +s, p, +c, ib, fs} = r
1304  Config{pol, circ, idle, every} = c
1305  +old = brk.get(A, s, key)
1306  do IO<Runner<A> & Out(Bool)>:
1307    y : Ran(A) <- trip(~A, ~penc, Runner{j, s, p, c, ib, fs}, key, old, brk.after(circ, old, now, o))
1308    settle.br(~A, ~penc, after(A, o, pol, b, id, fails(tries.get(tries(A, s), id)), now, rnd), y, g, err, next)
1309
1310def random(r: Out(U32)) -> Nat:
1311  match r:
1312    case Done{x}:
1313      U32.to_nat(x)
1314    case Fail{e}:
1315      0n
1316
1317def take.keep(-A: Data, m: Maybe<&1, Flight<A>>, xs: List<&1, Flight<A>>) -> List<&1, Flight<A>>:
1318  match m:
1319    case None{}:
1320      xs
1321    case Some{f}:
1322      Con{f, xs}
1323
1324def take.at(-A: Data, eq: Bool, f: Flight<A>, r: Maybe<&1, Flight<A>> & List<&1, Flight<A>>) -> Maybe<&1, Flight<A>> & List<&1, Flight<A>>:
1325  match eq:
1326    case True{}:
1327      (m, xs) = r
1328      (Some{f}, take.keep(A, m, xs))
1329    case False{}:
1330      (m, xs) = r
1331      (m, Con{f, xs})
1332
1333# The flight of ticket id, and the other flights.
1334def take(-A: Data, fs: List<&1, Flight<A>>, +id: Nat) -> Maybe<&1, Flight<A>> & List<&1, Flight<A>>:
1335  match fs:
1336    case Nil{}:
1337      (None{}, Nil{})
1338    case Con{Flight{Q.Ticket{+i, d, x}, g, b, k}, rest}:
1339      take.at(A, Nat.is_eq(i, id), Flight{Q.Ticket{i, d, x}, g, b, k}, take(A, rest, id))
1340
1341def settle.flight(~A: Data, ~penc: A -> Bytes.Bytes, mf: Maybe<&1, Flight<A>>, +o: Retry.Outcome, +now: Nat, +rnd: Nat, r: Runner<A>, err: Out(Unit),
1342    next: Runner<A> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1343  match mf:
1344    case None{}:
1345      next(r, err)
1346    case Some{Flight{Q.Ticket{id, d, x}, g, b, key}}:
1347      settle.op(~A, ~penc, o, b, id, key, now, rnd, r, g, err, next)
1348
1349def settle.take(~A: Data, ~penc: A -> Bytes.Bytes, t: Maybe<&1, Flight<A>> & List<&1, Flight<A>>, +o: Retry.Outcome, +now: Nat, +rnd: Nat,
1350    j: Log, s: State<A>, p: Res.Pool, c: Config, ib: Chan(Msg<A>), err: Out(Unit),
1351    next: Runner<A> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1352  (mf, fs) = t
1353  settle.flight(~A, ~penc, mf, o, now, rnd, Runner{j, s, p, c, ib, fs}, err, next)
1354
1355# Settles the flight of ticket id with outcome o.
1356def settle(~A: Data, ~penc: A -> Bytes.Bytes, +id: Nat, +o: Retry.Outcome, +now: Nat, +rnd: Nat, r: Runner<A>, err: Out(Unit),
1357    next: Runner<A> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1358  Runner{j, s, p, c, ib, fs} = r
1359  settle.take(~A, ~penc, take(A, fs, id), o, now, rnd, j, s, p, c, ib, err, next)
1360
1361# Sends the failure to the asker, and keeps it as the round's error.
1362def asked.failed(-A: Data, -X: Type, +rep: Chan(Out(X)), r: Runner<A>, +c: U32, +m: String, err: Out(Unit),
1363    next: Runner<A> -> Nat -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1364  do IO<Runner<A> & Out(Bool)>:
1365    x : Result<&1, &1, Out(X), Unit> <- Chan.send(Out(X), rep, Fail{(c, m)})
1366    next(r, 0n, first(err, Fail{(c, m)}))
1367
1368def asked.fail(-A: Data, -X: Type, +rep: Chan(Out(X)), r: Runner<A>, e: U32 & String, err: Out(Unit),
1369    next: Runner<A> -> Nat -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1370  (c, m) = e
1371  asked.failed(A, X, rep, r, c, m, err, next)
1372
1373def added(-A: Data, +rep: Chan(Out(Nat)), y: Runner<A> & Out(Nat), err: Out(Unit),
1374    next: Runner<A> -> Nat -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1375  match y:
1376    case (r, Done{+i}):
1377      do IO<Runner<A> & Out(Bool)>:
1378        x : Result<&1, &1, Out(Nat), Unit> <- Chan.send(Out(Nat), rep, Done{i})
1379        next(r, 1n, err)
1380    case (r, Fail{e}):
1381      asked.fail(A, Nat, rep, r, e, err, next)
1382
1383def dropped(-A: Data, +rep: Chan(Out(Unit)), y: Ran(A), err: Out(Unit),
1384    next: Runner<A> -> Nat -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1385  match y:
1386    case (r, Done{u}):
1387      do IO<Runner<A> & Out(Bool)>:
1388        x : Result<&1, &1, Out(Unit), Unit> <- Chan.send(Out(Unit), rep, Done{Unit{}})
1389        next(r, 0n, err)
1390    case (r, Fail{e}):
1391      asked.fail(A, Unit, rep, r, e, err, next)
1392
1393# Settles an ended attempt, or appends an ask's op and answers it. next gets the claim fuel the message adds: 1 for an enqueue.
1394def handle(~A: Data, ~penc: A -> Bytes.Bytes, m: Msg<A>, +now: Nat, +rnd: Nat, r: Runner<A>, err: Out(Unit),
1395    next: Runner<A> -> Nat -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1396  match m:
1397    case Ended{+id, +o}:
1398      settle(~A, ~penc, id, o, now, rnd, r, err, (r2 => e2 => next(r2, 0n, e2)))
1399    case Add{+d, x, +rep}:
1400      IO.bind(Runner<A> & Out(Nat), Runner<A> & Out(Bool), enqueue(~A, ~penc, r, d, x), y => added(A, rep, y, err, next))
1401    case Drop{+i, +rep}:
1402      IO.bind(Ran(A), Runner<A> & Out(Bool), cancel(~A, ~penc, r, i), y => dropped(A, rep, y, err, next))
1403
1404def merged(-A: Data, e: Out(Unit), k: Claimed(A)) -> Claimed(A):
1405  (r, n, err) = k
1406  (r, n, first(e, err))
1407
1408def empty.heap(-A: Data, h: H.Heap<Q.Ticket<A>>) -> Bool:
1409  match h:
1410    case H.HLeaf{}:
1411      True{}
1412    case H.HNode{n, t, l, r}:
1413      False{}
1414
1415def empty.list(-A: Data, xs: List<&2, Q.Ticket<A>>) -> Bool:
1416  match xs:
1417    case Nil{}:
1418      True{}
1419    case Con{t, rest}:
1420      False{}
1421
1422# No ticket is pending or running. Dead tickets wait for a Revive, so they do not count.
1423def idle(-A: Data, s: State<A>) -> Bool:
1424  State{Q.State{n, h, run}, ts, ds, bs} = s
1425  Bool.and(empty.heap(A, h), empty.list(A, run))
1426
1427def pending(-A: Data, s: State<A>) -> Nat:
1428  State{Q.State{n, h, run}, ts, ds, bs} = s
1429  H.Heap.size(Q.Ticket<A>, h)
1430
1431# The flights back, and whether none runs.
1432def flights.none(-A: Data, fs: List<&1, Flight<A>>) -> List<&1, Flight<A>> & Bool:
1433  match fs:
1434    case Nil{}:
1435      (Nil{}, True{})
1436    case Con{f, rest}:
1437      (Con{f, rest}, False{})
1438
1439# ---- compaction
1440
1441# Kept: compaction failed before the rename, and the old journal stands.
1442# Lost: the rename happened, but the new journal did not open. Moved: the journal at path is the snapshot.
1443type Swap is Type:
1444  Kept{err: U32 & String}
1445  Lost{err: U32 & String}
1446  Moved{journal: J.Journal}
1447
1448def swap.reopened(x: Out(J.Journal & Unit)) -> Swap:
1449  match x:
1450    case Fail{e}:
1451      Lost{e}
1452    case Done{(j, u)}:
1453      Moved{j}
1454
1455# The reopen fsyncs the directory, so the rename is durable before the next append.
1456def swap.renamed(+path: String, +limit: U32, x: Out(Unit)) -> IO(Swap):
1457  match x:
1458    case Fail{e}:
1459      IO.pure(Swap, Kept{e})
1460    case Done{u}:
1461      do IO<Swap>:
1462        y : Out(J.Journal & Unit) <- J.open(~Unit, ~(a => b => a), path, limit, limit, Unit{})
1463        return swap.reopened(y)
1464
1465def swap.written(+path: String, +tmp: String, +limit: U32, x: J.Journal & Out(Unit)) -> IO(Swap):
1466  match x:
1467    case (t, Fail{e}):
1468      do IO<Swap>:
1469        J.close(t)
1470        return Kept{e}
1471    case (t, Done{u}):
1472      do IO<Swap>:
1473        J.close(t)
1474        y : Out(Unit) <- Files.rename(tmp, path)
1475        swap.renamed(path, limit, y)
1476
1477def swap.opened(+path: String, +tmp: String, +limit: U32, x: Out(J.Journal & Unit), snap: Bytes.Bytes) -> IO(Swap):
1478  match x:
1479    case Fail{e}:
1480      IO.pure(Swap, Kept{e})
1481    case Done{(t, u)}:
1482      do IO<Swap>:
1483        y : J.Journal & Out(Unit) <- J.append(t, snap)
1484        swap.written(path, tmp, limit, y)
1485
1486# Writes snap as the only record of path.compact, fsyncs it, and renames it over path.
1487def swap(+path: String, +limit: U32, snap: Bytes.Bytes) -> IO(Swap):
1488  +tmp = path ++ ".compact"
1489  do IO<Swap>:
1490    gone : Out(Unit) <- Files.remove(tmp)
1491    x : Out(J.Journal & Unit) <- J.open(~Unit, ~(a => b => a), tmp, limit, limit, Unit{})
1492    swap.opened(path, tmp, limit, x, snap)
1493
1494def log.stop(j: J.Journal) -> J.Journal:
1495  J.Journal{file, max, end, ok} = j
1496  J.Journal{file, max, end, False{}}
1497
1498def compact.moved(x: Swap, j: J.Journal, +max: U32, path: String, +limit: U32, +base: U32) -> IO(Log & Out(Unit)):
1499  match x:
1500    case Kept{e}:
1501      IO.pure(Log & Out(Unit), (Log{j, path, limit, base}, Fail{e}))
1502    case Lost{e}:
1503      IO.pure(Log & Out(Unit), (Log{log.stop(j), path, limit, base}, Fail{e}))
1504    case Moved{k}:
1505      do IO<Log & Out(Unit)>:
1506        J.close(j)
1507        return (log.of(k, max, path, limit), Done{Unit{}})
1508
1509def compact.if(~A: Data, ~penc: A -> Bytes.Bytes, ok: Bool, j: J.Journal, +max: U32, +path: String, +limit: U32, +base: U32, s: State<A>) -> IO(Log & Out(Unit)):
1510  match ok:
1511    case False{}:
1512      IO.pure(Log & Out(Unit), (Log{j, path, limit, base}, Fail{(5, "runner: an append failed; reopen the runner")}))
1513    case True{}:
1514      do IO<Log & Out(Unit)>:
1515        x : Swap <- swap(path, limit, encode.snap(~A, ~penc, s))
1516        compact.moved(x, j, max, path, limit, base)
1517
1518def compact.done(-A: Data, x: Log & Out(Unit), s: State<A>, p: Res.Pool, c: Config, ib: Chan(Msg<A>), fs: List<&1, Flight<A>>) -> Ran(A):
1519  (l, o) = x
1520  (Runner{l, s, p, c, ib, fs}, o)
1521
1522# Writes the state as one Snap record into a new journal, fsyncs it, and renames it over the old one.
1523# A crash leaves one of the two journals, and both replay to this state. After a failure the old
1524# journal stands; when the failure came after the rename, appends stop until the runner is reopened.
1525def compact(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>) -> IO(Ran(A)):
1526  Runner{Log{j, path, limit, base}, +s, p, c, ib, fs} = r
1527  J.Journal{file, +max, end, +ok} = j
1528  do IO<Ran(A)>:
1529    x : Log & Out(Unit) <- compact.if(~A, ~penc, ok, J.Journal{file, max, end, ok}, max, path, limit, base, s)
1530    return compact.done(A, x, s, p, c, ib, fs)
1531
1532def tick.ran(e: Out(Unit), b: Bool) -> Out(Bool):
1533  match e:
1534    case Fail{x}:
1535      Fail{x}
1536    case Done{u}:
1537      Done{b}
1538
1539def tick.compacted(-A: Data, +b: Bool, y: Ran(A)) -> Runner<A> & Out(Bool):
1540  (r, e) = y
1541  (r, tick.ran(e, b))
1542
1543def tick.due(~A: Data, ~penc: A -> Bytes.Bytes, due: Bool, +b: Bool, r: Runner<A>) -> IO(Runner<A> & Out(Bool)):
1544  match due:
1545    case False{}:
1546      IO.pure(Runner<A> & Out(Bool), (r, Done{b}))
1547    case True{}:
1548      do IO<Runner<A> & Out(Bool)>:
1549        y : Ran(A) <- compact(~A, ~penc, r)
1550        return tick.compacted(A, b, y)
1551
1552def tick.size(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +b: Bool) -> IO(Runner<A> & Out(Bool)):
1553  Runner{Log{J.Journal{file, max, +end, ok}, path, limit, +base}, s, p, +c, ib, fs} = r
1554  Config{pol, circ, idle, +every} = c
1555  tick.due(~A, ~penc, Bool.and(Bool.not(U32.is_eq(every, 0)), U32.is_gt((end - base : U32), every)), b,
1556    Runner{Log{J.Journal{file, max, end, ok}, path, limit, base}, s, p, c, ib, fs})
1557
1558# Compacts once the journal grew past Config.compact octets since open or the last compaction.
1559def tick.grown(~A: Data, ~penc: A -> Bytes.Bytes, t: Runner<A> & Out(Bool)) -> IO(Runner<A> & Out(Bool)):
1560  match t:
1561    case (r, Fail{e}):
1562      IO.pure(Runner<A> & Out(Bool), (r, Fail{e}))
1563    case (r, Done{b}):
1564      tick.size(~A, ~penc, r, b)
1565
1566def tick.polled(-A: Data, x: Poll<&1, &1, Maybe<&1, Msg<A>>, Unit>) -> Maybe<&1, Msg<A>>:
1567  match x:
1568    case Ready{m}:
1569      m
1570    case Wait{u}:
1571      None{}
1572
1573def tick.wait(-A: Data, t: List<&1, Flight<A>> & Bool, j: Log, s: State<A>, p: Res.Pool, c: Config, +ib: Chan(Msg<A>)) ->
1574    IO(Runner<A> & Maybe<&1, Msg<A>>):
1575  (fs, none) = t
1576  match none:
1577    case False{}:
1578      do IO<Runner<A> & Maybe<&1, Msg<A>>>:
1579        m : Maybe<&1, Msg<A>> <- Chan.recv(Msg<A>, ib)
1580        return (Runner{j, s, p, c, ib, fs}, m)
1581    case True{}:
1582      do IO<Runner<A> & Maybe<&1, Msg<A>>>:
1583        x : Poll<&1, &1, Maybe<&1, Msg<A>>, Unit> <- Chan.try_recv(Msg<A>, ib, 0)
1584        return (Runner{j, s, p, c, ib, fs}, tick.polled(A, x))
1585
1586# The next message. While an attempt runs, it waits for one; when none runs, it takes one only when the inbox holds one.
1587def tick.next(-A: Data, r: Runner<A>) -> IO(Runner<A> & Maybe<&1, Msg<A>>):
1588  Runner{j, s, p, c, ib, fs} = r
1589  tick.wait(A, flights.none(A, fs), j, s, p, c, ib)
1590
1591# Handles the message, then claims due tickets into free units with the claim fuel left. None ends the round.
1592def tick.got(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), x: Runner<A> & Maybe<&1, Msg<A>>, +ran: Bool, +left: Nat,
1593    err: Out(Unit), next: Claimed(A) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
1594  (r, m) = x
1595  match m:
1596    case None{}:
1597      IO.pure(Runner<A> & Out(Bool), (r, tick.ran(err, ran)))
1598    case Some{msg}:
1599      do IO<Runner<A> & Out(Bool)>:
1600        +now : Nat <- clock()
1601        rnd : Out(U32) <- IO.random_u32()
1602        handle(~A, ~penc, msg, now, random(rnd), r, err, (r2 => more => e2 =>
1603          IO.bind(Claimed(A), Runner<A> & Out(Bool), claim(~A, ~penc, ~target, ~work, Nat.add(left, more), now, r2), k2 => next(merged(A, e2, k2)))))
1604
1605# Handles messages until no attempt runs and the inbox is empty. fuel bounds the messages; attempts that still run
1606# when it ends stay in the runner, and the next tick settles them.
1607def tick.run(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), fuel: Nat, +ran: Bool, k: Claimed(A)) ->
1608    IO(Runner<A> & Out(Bool)):
1609  match fuel:
1610    case 0n:
1611      (r, left, err) = k
1612      IO.pure(Runner<A> & Out(Bool), (r, tick.ran(err, ran)))
1613    case 1n+n:
1614      (r, +left, err) = k
1615      do IO<Runner<A> & Out(Bool)>:
1616        x : Runner<A> & Maybe<&1, Msg<A>> <- tick.next(A, r)
1617        tick.got(~A, ~penc, ~target, ~work, x, ran, left, err, (k2 => tick.run(~A, ~penc, ~target, ~work, n, True{}, k2)))
1618
1619# Fuel for each pending ticket, each attempt that may already run, and the last look at the inbox.
1620def tick.at(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +now: Nat, r: Runner<A>) -> IO(Runner<A> & Out(Bool)):
1621  Runner{j, +s, +p, c, ib, fs} = r
1622  +fuel = Nat.add(2n, Nat.add(pending(A, s), Res.capacity(p)))
1623  do IO<Runner<A> & Out(Bool)>:
1624    k : Claimed(A) <- claim(~A, ~penc, ~target, ~work, fuel, now, Runner{j, s, p, c, ib, fs})
1625    tick.run(~A, ~penc, ~target, ~work, fuel, False{}, k)
1626
1627# One round: start an attempt for each due ticket the pool admits. Then read the inbox: as each attempt finishes, record its outcome,
1628# return its unit, and start the next due ticket in that unit; append and answer each ask from another fiber.
1629# The round ends when no attempt runs and the inbox is empty. Then compact the journal when it grew past Config.compact.
1630# Done{True} when an attempt ran or an ask was answered. work runs on its own; target names the breaker that guards it.
1631def tick(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), r: Runner<A>) -> IO(Runner<A> & Out(Bool)):
1632  do IO<Runner<A> & Out(Bool)>:
1633    now : Nat <- clock()
1634    t : Runner<A> & Out(Bool) <- tick.at(~A, ~penc, ~target, ~work, now, r)
1635    tick.grown(~A, ~penc, t)
1636
1637def finished.of(-A: Data, +quiet: Bool, t: List<&1, Flight<A>> & Bool, j: Log, s: State<A>, p: Res.Pool, c: Config, ib: Chan(Msg<A>)) ->
1638    Runner<A> & Bool:
1639  (fs, none) = t
1640  (Runner{j, s, p, c, ib, fs}, Bool.and(quiet, none))
1641
1642# The runner back, and whether no ticket is pending or running and no attempt runs.
1643def finished(-A: Data, r: Runner<A>) -> Runner<A> & Bool:
1644  Runner{j, +s, p, c, ib, fs} = r
1645  finished.of(A, idle(A, s), flights.none(A, fs), j, s, p, c, ib)
1646
1647def drain.handled(-A: Data, t: Runner<A> & Out(Bool), next: Runner<A> -> IO(Ran(A))) -> IO(Ran(A)):
1648  match t:
1649    case (r, Fail{e}):
1650      IO.pure(Ran(A), (r, Fail{e}))
1651    case (r, Done{b}):
1652      next(r)
1653
1654def drain.woke(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +idle: U32,
1655    x: Poll<&1, &1, Maybe<&1, Msg<A>>, Unit>, r: Runner<A>, next: Runner<A> -> IO(Ran(A))) -> IO(Ran(A)):
1656  match x:
1657    case Ready{Some{m}}:
1658      do IO<Ran(A)>:
1659        t : Runner<A> & Out(Bool) <- tick.got(~A, ~penc, ~target, ~work, (r, Some{m}), True{}, 0n, Done{Unit{}},
1660          (k => tick.run(~A, ~penc, ~target, ~work, 0n, True{}, k)))
1661        drain.handled(A, t, next)
1662    case Ready{None{}}:
1663      do IO<Ran(A)>:
1664        IO.sleep(idle)
1665        next(r)
1666    case Wait{u}:
1667      next(r)
1668
1669def drain.wait(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +idle: U32, r: Runner<A>,
1670    next: Runner<A> -> IO(Ran(A))) -> IO(Ran(A)):
1671  Runner{j, s, p, c, +ib, fs} = r
1672  do IO<Ran(A)>:
1673    x : Poll<&1, &1, Maybe<&1, Msg<A>>, Unit> <- Chan.try_recv(Msg<A>, ib, idle)
1674    drain.woke(~A, ~penc, ~target, ~work, idle, x, Runner{j, s, p, c, ib, fs}, next)
1675
1676def drain.ticked(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +idle: U32,
1677    t: Runner<A> & Out(Bool), next: Runner<A> -> IO(Ran(A))) -> IO(Ran(A)):
1678  match t:
1679    case (r, Fail{e}):
1680      IO.pure(Ran(A), (r, Fail{e}))
1681    case (r, Done{True{}}):
1682      next(r)
1683    case (r, Done{False{}}):
1684      drain.wait(~A, ~penc, ~target, ~work, idle, r, next)
1685
1686def drain.if(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), done: Bool, r: Runner<A>, next: Runner<A> -> IO(Ran(A))) -> IO(Ran(A)):
1687  match done:
1688    case True{}:
1689      IO.pure(Ran(A), (r, Done{Unit{}}))
1690    case False{}:
1691      Runner{j, s, p, +c, ib, fs} = r
1692      Config{pol, circ, +idle, every} = c
1693      do IO<Ran(A)>:
1694        t : Runner<A> & Out(Bool) <- tick(~A, ~penc, ~target, ~work, Runner{j, s, p, c, ib, fs})
1695        drain.ticked(~A, ~penc, ~target, ~work, idle, t, next)
1696
1697def drain.at(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), x: Runner<A> & Bool, next: Runner<A> -> IO(Ran(A))) -> IO(Ran(A)):
1698  (r, done) = x
1699  drain.if(~A, ~penc, ~target, ~work, done, r, next)
1700
1701def drain.end(-A: Data, x: Runner<A> & Bool) -> Ran(A):
1702  (r, done) = x
1703  (r, Bool.pick(Out(Unit), done, Done{Unit{}}, Fail{(62, "runner: drain ran out of rounds")}))
1704
1705# Ticks until no ticket is pending or running and no attempt runs. After a round that did nothing, it waits up to idle ms
1706# for a message, and handles one that arrives at once.
1707# Fails when fuel rounds leave work.
1708def drain(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), fuel: Nat, r: Runner<A>) -> IO(Ran(A)):
1709  match fuel:
1710    case 0n:
1711      IO.pure(Ran(A), drain.end(A, finished(A, r)))
1712    case 1n+n:
1713      drain.at(~A, ~penc, ~target, ~work, finished(A, r), (r2 => drain(~A, ~penc, ~target, ~work, n, r2)))
1714
1715def orphans.next(-A: Data, y: Ran(A), err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Ran(A))) -> IO(Ran(A)):
1716  (r, e) = y
1717  next(r, first(err, e))
1718
1719# A running ticket at open lost its attempt in a crash: it is requeued at now, as a failed attempt.
1720def orphans(~A: Data, ~penc: A -> Bytes.Bytes, ts: List<&2, Q.Ticket<A>>, +now: Nat, r: Runner<A>, err: Out(Unit)) -> IO(Ran(A)):
1721  match ts:
1722    case Nil{}:
1723      IO.pure(Ran(A), (r, err))
1724    case Con{Q.Ticket{id, d, x}, rest}:
1725      do IO<Ran(A)>:
1726        y : Ran(A) <- commit(~A, ~penc, r, Requeue{id, now})
1727        orphans.next(A, y, err, (r2 => e2 => orphans(~A, ~penc, rest, now, r2, e2)))
1728
1729def close.answer(-A: Data, m: Msg<A>) -> IO(Unit):
1730  match m:
1731    case Ended{id, o}:
1732      IO.pure(Unit, Unit{})
1733    case Add{d, x, rep}:
1734      do IO<Unit>:
1735        y : Result<&1, &1, Out(Nat), Unit> <- Chan.send(Out(Nat), rep, Fail{(32, "runner: the runner closed")})
1736        return Unit{}
1737    case Drop{i, rep}:
1738      do IO<Unit>:
1739        y : Result<&1, &1, Out(Unit), Unit> <- Chan.send(Out(Unit), rep, Fail{(32, "runner: the runner closed")})
1740        return Unit{}
1741
1742def close.polled(-A: Data, x: Poll<&1, &1, Maybe<&1, Msg<A>>, Unit>, next: Unit -> IO(Unit)) -> IO(Unit):
1743  match x:
1744    case Ready{Some{m}}:
1745      do IO<Unit>:
1746        close.answer(A, m)
1747        next(Unit{})
1748    case Ready{None{}}:
1749      IO.pure(Unit, Unit{})
1750    case Wait{u}:
1751      IO.pure(Unit, Unit{})
1752
1753# Answers each ask left in the closed inbox with a failure. fuel bounds the messages it reads.
1754def close.drain(-A: Data, fuel: Nat, +ib: Chan(Msg<A>)) -> IO(Unit):
1755  match fuel:
1756    case 0n:
1757      IO.pure(Unit, Unit{})
1758    case 1n+n:
1759      do IO<Unit>:
1760        x : Poll<&1, &1, Maybe<&1, Msg<A>>, Unit> <- Chan.try_recv(Msg<A>, ib, 0)
1761        close.polled(A, x, (u => close.drain(A, n, ib)))
1762
1763# Closes the inbox and the journal. Each ask still in the inbox gets a failure, and no op is appended for it.
1764def close(-A: Data, r: Runner<A>) -> IO(Unit):
1765  Runner{j, s, +p, c, +ib, fs} = r
1766  do IO<Unit>:
1767    Chan.close(Msg<A>, ib)
1768    close.drain(A, Nat.add(2n, Res.capacity(p)), ib)
1769    log.close(j)
1770
1771def open.done(-A: Data, y: Ran(A)) -> IO(Out(Runner<A>)):
1772  match y:
1773    case (r, Fail{e}):
1774      do IO<Out(Runner<A>)>:
1775        close(A, r)
1776        return Fail{e}
1777    case (r, Done{u}):
1778      IO.pure(Out(Runner<A>), Done{r})
1779
1780def open.got(~A: Data, ~penc: A -> Bytes.Bytes, x: Out(J.Journal & Maybe<&2, State<A>>), +max: U32, path: String, +limit: U32, c: Config, +capacity: Nat) ->
1781    IO(Out(Runner<A>)):
1782  match x:
1783    case Fail{e}:
1784      IO.pure(Out(Runner<A>), Fail{e})
1785    case Done{(j, None{})}:
1786      do IO<Out(Runner<A>)>:
1787        J.close(j)
1788        return Fail{(22, "runner: a journal record is not a runner record")}
1789    case Done{(j, Some{+s})}:
1790      do IO<Out(Runner<A>)>:
1791        now : Nat <- clock()
1792        ib : Chan(Msg<A>) <- Chan.new(Msg<A>, (U32.from_nat(capacity) + 1 : U32))
1793        y : Ran(A) <- orphans(~A, ~penc, Q.running(A, queue(A, s)), now, Runner{log.of(j, max, path, limit), calm(A, s), Res.new(0n, capacity), c, ib, Nil{}}, Done{Unit{}})
1794        open.done(A, y)
1795
1796def replay.step(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, acc: Maybe<&2, State<A>>, b: Bytes.Bytes) -> Maybe<&2, State<A>>:
1797  load(~A, ~Bytes.Bytes, ~(y => decode.rec(~A, ~pdec, y)), acc, b)
1798
1799# Opens the runner at path: replays its journal into the state, then requeues each ticket that was running.
1800# capacity bounds the attempts that run at once. max caps an op record's octets; limit caps the journal file and a snapshot.
1801# Fails when the journal fails to open, or a committed record is not a snapshot or an op.
1802def open(~A: Data, ~penc: A -> Bytes.Bytes, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +path: String, +max: U32, +limit: U32, c: Config, +capacity: Nat) -> IO(Out(Runner<A>)):
1803  do IO<Out(Runner<A>)>:
1804    x : Out(J.Journal & Maybe<&2, State<A>>) <- J.open(~Maybe<&2, State<A>>, ~(acc => b => replay.step(~A, ~pdec, acc, b)), path, limit, limit, Some{new(A)})
1805    open.got(~A, ~penc, x, max, path, limit, c, capacity)
1806
1807# The runner back, and a copy of its state.
1808def state(-A: Data, r: Runner<A>) -> Runner<A> & State<A>:
1809  Runner{j, +s, p, c, ib, fs} = r
1810  (Runner{j, s, p, c, ib, fs}, s)