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