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