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	# One queue transition. Each one is one journal record, appended before the state changes.
14	# Select: the queue's select at now. Replay runs it at the same now, so it takes the same ticket.
15	# Requeue: the attempt failed; it counts one failure. Defer: the breaker held the ticket back; no attempt ran.
16	type Op<-A: Data> is Data:
17	  Enqueue{deadline: Nat, payload: A}
18	  Select{now: Nat}
19	  Complete{id: Nat}
20	  Cancel{id: Nat}
21	  Requeue{id: Nat, deadline: Nat}
22	  Defer{id: Nat, deadline: Nat}
23	
24	# The failed attempts of a ticket, and its first deadline, from which the policy deadline counts.
25	type Try is Data:
26	  Try{fails: Nat, start: Nat}
27	
28	# A Try for each ticket that is not yet complete or cancelled, by ticket id.
29	def Tries() -> Data:
30	  O.OMap<Nat, Try>
31	
32	# The durable state: the queue and the tries.
33	type State<-A: Data> is Data:
34	  State{queue: Q.State<A>, tries: Tries()}
35	
36	def new(-A: Data) -> State<A>:
37	  State{Q.new(A), O.OMap.new(Nat, Try)}
38	
39	def queue(-A: Data, s: State<A>) -> Q.State<A>:
40	  State{q, ts} = s
41	  q
42	
43	def tries(-A: Data, s: State<A>) -> Tries():
44	  State{q, ts} = s
45	  ts
46	
47	# ---- tries
48	
49	def tries.drop(ts: Tries(), +i: Nat) -> Tries():
50	  O.OMap.del(~Nat, ~Try, ~Nat.cmp, ts, i)
51	
52	def tries.get(ts: Tries(), +i: Nat) -> Maybe<&2, Try>:
53	  O.OMap.get(~Nat, ~Try, ~Nat.cmp, ts, i)
54	
55	def tries.bump.got(m: Maybe<&2, Try>, ts: Tries(), +i: Nat) -> Tries():
56	  match m:
57	    case None{}:
58	      ts
59	    case Some{Try{f, st}}:
60	      O.OMap.put(~Nat, ~Try, ~Nat.cmp, ts, i, Try{1n+f, st})
61	
62	def tries.bump(+ts: Tries(), +i: Nat) -> Tries():
63	  tries.bump.got(tries.get(ts, i), ts, i)
64	
65	# ---- the transition
66	
67	def step.enqueued(-A: Data, r: Q.State<A> & Nat, ts: Tries(), d: Nat) -> State<A>:
68	  (q, i) = r
69	  State{q, O.OMap.put(~Nat, ~Try, ~Nat.cmp, ts, i, Try{0n, d})}
70	
71	def step.selected(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, ts: Tries()) -> State<A>:
72	  (q, t) = r
73	  State{q, ts}
74	
75	def step.requeued(-A: Data, r: Q.State<A> & Bool, ts: Tries()) -> State<A>:
76	  (q, b) = r
77	  State{q, ts}
78	
79	# Complete takes the ticket from the running list only: O(running), not O(pending).
80	def step.completed(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, ts: Tries()) -> State<A>:
81	  (q, t) = r
82	  State{q, ts}
83	
84	# The state after op. Live and replay both run it, so they agree.
85	def step(~A: Data, s: State<A>, op: Op<A>) -> State<A>:
86	  match s op:
87	    case State{q, ts} Enqueue{+d, x}:
88	      step.enqueued(A, Q.enqueue(~A, d, x, q), ts, d)
89	    case State{q, ts} Select{now}:
90	      step.selected(A, Q.select(~A, now, q), ts)
91	    case State{q, ts} Complete{+i}:
92	      step.completed(A, Q.complete(A, i, q), tries.drop(ts, i))
93	    case State{q, ts} Cancel{+i}:
94	      State{Q.cancel(~A, i, q), tries.drop(ts, i)}
95	    case State{q, ts} Requeue{+i, d}:
96	      step.requeued(A, Q.requeue(~A, i, d, q), tries.bump(ts, i))
97	    case State{q, ts} Defer{i, d}:
98	      step.requeued(A, Q.requeue(~A, i, d, q), ts)
99	
100	def apply.op(~A: Data, m: Maybe<&2, Op<A>>, s: State<A>) -> Maybe<&2, State<A>>:
101	  match m:
102	    case None{}:
103	      None{}
104	    case Some{op}:
105	      Some{step(~A, s, op)}
106	
107	# Folds one record into the replayed state. A record that does not decode ends the replay with None.
108	def apply(~A: Data, ~R: Type, ~dec: R -> Maybe<&2, Op<A>>, acc: Maybe<&2, State<A>>, r: R) -> Maybe<&2, State<A>>:
109	  match acc:
110	    case None{}:
111	      None{}
112	    case Some{s}:
113	      apply.op(~A, dec(r), s)
114	
115	# A journal record. Compaction writes one Snap of the whole state; every other record is one op.
116	type Rec<-A: Data> is Data:
117	  Snap{state: State<A>}
118	  Do{op: Op<A>}
119	
120	def load.rec(~A: Data, m: Maybe<&2, Rec<A>>, s: State<A>) -> Maybe<&2, State<A>>:
121	  match m:
122	    case None{}:
123	      None{}
124	    case Some{Snap{t}}:
125	      Some{t}
126	    case Some{Do{op}}:
127	      Some{step(~A, s, op)}
128	
129	# Folds one record into the replayed state: a Snap replaces it, and an op steps it, as apply does.
130	# A record that does not decode ends the replay with None.
131	def load(~A: Data, ~R: Type, ~dec: R -> Maybe<&2, Rec<A>>, acc: Maybe<&2, State<A>>, r: R) -> Maybe<&2, State<A>>:
132	  match acc:
133	    case None{}:
134	      None{}
135	    case Some{s}:
136	      load.rec(~A, dec(r), s)
137	
138	# ---- pool
139	
140	# Picked: a due ticket and one unit of the pool for it. Idle: no due ticket, or no free unit; the pool is unchanged.
141	type Pick<-A: Data> is Type:
142	  Picked{pool: Res.Pool, ticket: Q.Ticket<A>, grant: Res.Grant}
143	  Idle{pool: Res.Pool}
144	
145	def pick.took(-A: Data, r: Res.Reserve, t: Q.Ticket<A>) -> Pick<A>:
146	  match r:
147	    case Res.Took{p, g}:
148	      Picked{p, t, g}
149	    case Res.Refused{p}:
150	      Idle{p}
151	
152	def pick.due(-A: Data, m: Maybe<&2, Q.Ticket<A>>, p: Res.Pool) -> Pick<A>:
153	  match m:
154	    case None{}:
155	      Idle{p}
156	    case Some{t}:
157	      pick.took(A, Res.reserve(p, 1n), t)
158	
159	def pick.sel(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, p: Res.Pool) -> Pick<A>:
160	  (q, m) = r
161	  pick.due(A, m, p)
162	
163	# The ticket that Select{now} takes, with one unit of p, when there is a ticket due and a unit free.
164	def pick(~A: Data, p: Res.Pool, s: State<A>, now: Nat) -> Pick<A>:
165	  State{q, ts} = s
166	  pick.sel(A, Q.select(~A, now, q), p)
167	
168	# ---- retry
169	
170	# The budget left for ticket i: the policy's attempts less its failures, until its start plus the policy deadline.
171	def budget.of(+attempts: Nat, t: Maybe<&2, Try>, d: Maybe<&2, Nat>) -> Retry.Budget:
172	  match t d:
173	    case None{} m:
174	      Retry.Budget{attempts, None{}}
175	    case Some{Try{f, st}} None{}:
176	      Retry.Budget{Nat.sub(attempts, f), None{}}
177	    case Some{Try{f, +st}} Some{ms}:
178	      Retry.Budget{Nat.sub(attempts, f), Some{Nat.add(st, ms)}}
179	
180	def budget(p: Retry.Policy, t: Maybe<&2, Try>) -> Retry.Budget:
181	  Retry.Policy{a, base, cap, d} = p
182	  budget.of(a, t, d)
183	
184	# The failures so far: the index of the next retry.
185	def fails(t: Maybe<&2, Try>) -> Nat:
186	  match t:
187	    case None{}:
188	      0n
189	    case Some{Try{f, st}}:
190	      f
191	
192	def after.plan(-A: Data, pl: Retry.Plan, +id: Nat, +now: Nat) -> Op<A>:
193	  match pl:
194	    case Retry.PlanStop{}:
195	      Complete{id}
196	    case Retry.PlanWait{w}:
197	      Requeue{id, Nat.add(now, w)}
198	
199	# The record after an attempt of ticket id ends at now with o: requeue at now plus retry's wait, or complete.
200	def after(-A: Data, o: Retry.Outcome, p: Retry.Policy, b: Retry.Budget, +id: Nat, +n: Nat, +now: Nat, +r: Nat) -> Op<A>:
201	  after.plan(A, Retry.plan(o, p, b, n, now, r), id, now)
202	
203	# When an open breaker lets attempts start again.
204	def until(br: Retry.Breaker, +now: Nat) -> Nat:
205	  match br:
206	    case Retry.BreakerOn{c, Retry.CircuitOpen{u}}:
207	      u
208	    case Retry.BreakerOn{c, Retry.CircuitClosed{fs}}:
209	      Nat.add(now, Retry.Circuit.cooldown(c))
210	    case Retry.BreakerOn{c, Retry.CircuitHalf{k}}:
211	      Nat.add(now, Retry.Circuit.cooldown(c))
212	    case Retry.BreakerOff{}:
213	      now
214	
215	# The record for a ticket that admit refused: Defer to when the breaker may admit, or Complete when its budget is spent or late.
216	def held(-A: Data, why: Retry.Deny, br: Retry.Breaker, +id: Nat, +now: Nat) -> Op<A>:
217	  match why:
218	    case Retry.DenyOpen{}:
219	      Defer{id, until(br, now)}
220	    case Retry.DenyLate{}:
221	      Complete{id}
222	    case Retry.DenySpent{}:
223	      Complete{id}
224	
225	# ---- the record codec: a u32le tag, then u64le fields; an Enqueue then holds the payload.
226	
227	def W() -> Nat:
228	  Nat.mul(65536n, 65536n)
229	
230	# Nat fields hold values under 2^64.
231	def put.u64(b: Bytes.Bytes, +at: U32, +n: Nat) -> Bytes.Bytes:
232	  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())))
233	
234	def head(+tag: U32, +a: Nat, +c: Nat) -> Bytes.Bytes:
235	  put.u64(put.u64(Bytes.set.u32le(Bytes.new(20), 0, tag), 4, a), 12, c)
236	
237	def encode(~A: Data, ~penc: A -> Bytes.Bytes, op: Op<A>) -> Bytes.Bytes:
238	  match op:
239	    case Enqueue{+d, x}:
240	      Bytes.append(put.u64(Bytes.set.u32le(Bytes.new(12), 0, 1), 4, d), penc(x))
241	    case Select{+now}:
242	      head(2, now, 0n)
243	    case Complete{+i}:
244	      head(3, i, 0n)
245	    case Cancel{+i}:
246	      head(4, i, 0n)
247	    case Requeue{+i, +d}:
248	      head(5, i, d)
249	    case Defer{+i, +d}:
250	      head(6, i, d)
251	
252	def get.field(m: Maybe<&2, U32>) -> U32:
253	  match m:
254	    case None{}:
255	      0
256	    case Some{x}:
257	      x
258	
259	def get.u32(r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & U32:
260	  (b, m) = r
261	  (b, get.field(m))
262	
263	def get.hi(+lo: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Nat:
264	  (b, m) = r
265	  (b, Nat.add(U32.to_nat(lo), Nat.mul(U32.to_nat(get.field(m)), W())))
266	
267	def get.lo(+at: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Nat:
268	  (b, m) = r
269	  +lo = get.field(m)
270	  get.hi(lo, Bytes.get.u32le(b, (at + 4 : U32)))
271	
272	def get.u64(b: Bytes.Bytes, +at: U32) -> Bytes.Bytes & Nat:
273	  get.lo(at, Bytes.get.u32le(b, at))
274	
275	def decode.small(-A: Data, ok: Bool, +tag: U32, +a: Nat, +c: Nat) -> Maybe<&2, Op<A>>:
276	  match ok:
277	    case False{}:
278	      None{}
279	    case True{}:
280	      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 2), Some{Select{a}},
281	      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 3), Some{Complete{a}},
282	      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 4), Some{Cancel{a}},
283	      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 5), Some{Requeue{a, c}},
284	      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 6), Some{Defer{a, c}}, None{})))))
285	
286	def decode.payload(-A: Data, m: Maybe<&2, A>, +d: Nat) -> Maybe<&2, Op<A>>:
287	  match m:
288	    case None{}:
289	      None{}
290	    case Some{x}:
291	      Some{Enqueue{d, x}}
292	
293	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>>:
294	  (b, +c) = r
295	  decode.small(A, U32.is_eq(len, 20), tag, a, c)
296	
297	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>>:
298	  match enq:
299	    case True{}:
300	      decode.payload(A, pdec(Bytes.snd(Bytes.slice(b, 12, (len - 12 : U32)))), a)
301	    case False{}:
302	      decode.c(~A, ~pdec, tag, len, a, get.u64(b, 12))
303	
304	def decode.a(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +tag: U32, +len: U32, r: Bytes.Bytes & Nat) -> Maybe<&2, Op<A>>:
305	  (b, +a) = r
306	  decode.kind(~A, ~pdec, U32.is_eq(tag, 1), tag, len, a, b)
307	
308	def decode.tag(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +len: U32, r: Bytes.Bytes & U32) -> Maybe<&2, Op<A>>:
309	  (b, +tag) = r
310	  decode.a(~A, ~pdec, tag, len, get.u64(b, 4))
311	
312	def decode.len(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, ok: Bool, +len: U32, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
313	  match ok:
314	    case False{}:
315	      None{}
316	    case True{}:
317	      decode.tag(~A, ~pdec, len, get.u32(Bytes.get.u32le(b, 0)))
318	
319	# The op of a record, or None when it is not one that encode writes.
320	def decode(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
321	  Bytes.Bytes{+len, buf} = b
322	  decode.len(~A, ~pdec, U32.is_ge(len, 12), len, Bytes.Bytes{len, buf})
323	
324	# ---- the snapshot codec: tag 7 and the next id, then the heap, the running list, and the tries.
325	# A u32 flag starts each tree node and list cell (1) and marks each end (0), so decode rebuilds the same shape.
326	# A ticket is its u64 id and deadline, then a u32 length and the payload.
327	
328	def snap.u32(out: Bytes.Bytes, +v: U32) -> Bytes.Bytes:
329	  Bytes.append(out, Bytes.set.u32le(Bytes.new(4), 0, v))
330	
331	def snap.u64(out: Bytes.Bytes, +n: Nat) -> Bytes.Bytes:
332	  Bytes.append(out, put.u64(Bytes.new(8), 0, n))
333	
334	def snap.payload(out: Bytes.Bytes, p: Bytes.Bytes) -> Bytes.Bytes:
335	  Bytes.Bytes{+len, buf} = p
336	  Bytes.append(snap.u32(out, len), Bytes.Bytes{len, buf})
337	
338	def snap.ticket(~A: Data, ~penc: A -> Bytes.Bytes, out: Bytes.Bytes, t: Q.Ticket<A>) -> Bytes.Bytes:
339	  Q.Ticket{+i, +d, x} = t
340	  snap.payload(snap.u64(snap.u64(out, i), d), penc(x))
341	
342	def snap.heap(~A: Data, ~penc: A -> Bytes.Bytes, h: H.Heap<Q.Ticket<A>>, out: Bytes.Bytes) -> Bytes.Bytes:
343	  match h:
344	    case H.HLeaf{}:
345	      snap.u32(out, 0)
346	    case H.HNode{+n, t, l, r}:
347	      snap.heap(~A, ~penc, r, snap.heap(~A, ~penc, l, snap.ticket(~A, ~penc, snap.u64(snap.u32(out, 1), n), t)))
348	
349	def snap.run(~A: Data, ~penc: A -> Bytes.Bytes, xs: List<&2, Q.Ticket<A>>, out: Bytes.Bytes) -> Bytes.Bytes:
350	  match xs:
351	    case Nil{}:
352	      snap.u32(out, 0)
353	    case Con{t, rest}:
354	      snap.run(~A, ~penc, rest, snap.ticket(~A, ~penc, snap.u32(out, 1), t))
355	
356	def snap.tries(ts: Tries(), out: Bytes.Bytes) -> Bytes.Bytes:
357	  match ts:
358	    case O.OTip{}:
359	      snap.u32(out, 0)
360	    case O.OBin{+n, +k, Try{+f, +st}, lo, hi}:
361	      snap.tries(hi, snap.tries(lo, snap.u64(snap.u64(snap.u64(snap.u32(snap.u32(out, 1), n), k), f), st)))
362	
363	# The Snap record of s.
364	def encode.snap(~A: Data, ~penc: A -> Bytes.Bytes, s: State<A>) -> Bytes.Bytes:
365	  State{Q.State{+n, h, run}, ts} = s
366	  snap.tries(ts, snap.run(~A, ~penc, run, snap.heap(~A, ~penc, h, snap.u64(snap.u32(Bytes.new(0), 7), n))))
367	
368	def encode.rec(~A: Data, ~penc: A -> Bytes.Bytes, r: Rec<A>) -> Bytes.Bytes:
369	  match r:
370	    case Snap{s}:
371	      encode.snap(~A, ~penc, s)
372	    case Do{op}:
373	      encode(~A, ~penc, op)
374	
375	# A snapshot reader: the record, and the offset of the next field.
376	type Rd is Type:
377	  Rd{b: Bytes.Bytes, at: U32}
378	
379	# One step of the snapshot decoder. None: a field runs past the record or does not decode.
380	def Dec(-T: Type) -> Type:
381	  Rd -> Maybe<&1, Rd & T>
382	
383	def Dec.next(-A: Type, -B: Type, m: Maybe<&1, Rd & A>, f: A -> Dec(B)) -> Maybe<&1, Rd & B>:
384	  match m:
385	    case None{}:
386	      None{}
387	    case Some{(r, x)}:
388	      f(x)(r)
389	
390	def Dec.bind(-A: Type, -B: Type, m: Dec(A), f: A -> Dec(B)) -> Dec(B):
391	  r => Dec.next(A, B, m(r), f)
392	
393	def Dec.pure(-A: Type, x: A) -> Dec(A):
394	  r => Some{(r, x)}
395	
396	def Dec.fail(-A: Type) -> Dec(A):
397	  r => None{}
398	
399	def rd.u32.of(m: Maybe<&2, U32>, b: Bytes.Bytes, +at: U32) -> Maybe<&1, Rd & U32>:
400	  match m:
401	    case None{}:
402	      None{}
403	    case Some{x}:
404	      Some{(Rd{b, (at + 4 : U32)}, x)}
405	
406	def rd.u32.got(+at: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Maybe<&1, Rd & U32>:
407	  (b, m) = r
408	  rd.u32.of(m, b, at)
409	
410	def rd.u32.at(r: Rd) -> Maybe<&1, Rd & U32>:
411	  Rd{b, +at} = r
412	  rd.u32.got(at, Bytes.get.u32le(b, at))
413	
414	def rd.u32() -> Dec(U32):
415	  r => rd.u32.at(r)
416	
417	def rd.u64.of(+lo: U32, +hi: U32) -> Nat:
418	  Nat.add(U32.to_nat(lo), Nat.mul(U32.to_nat(hi), W()))
419	
420	def rd.u64() -> Dec(Nat):
421	  do Dec<Nat>:
422	    lo : U32 <- rd.u32()
423	    hi : U32 <- rd.u32()
424	    return rd.u64.of(lo, hi)
425	
426	def rd.cut.got(+at: U32, +n: U32, r: Bytes.Bytes & Bytes.Bytes) -> Maybe<&1, Rd & Bytes.Bytes>:
427	  (b, p) = r
428	  Some{(Rd{b, (at + n : U32)}, p)}
429	
430	def rd.cut.if(ok: Bool, b: Bytes.Bytes, +at: U32, +n: U32) -> Maybe<&1, Rd & Bytes.Bytes>:
431	  match ok:
432	    case False{}:
433	      None{}
434	    case True{}:
435	      rd.cut.got(at, n, Bytes.slice(b, at, n))
436	
437	def rd.cut.at(+n: U32, r: Rd) -> Maybe<&1, Rd & Bytes.Bytes>:
438	  Rd{Bytes.Bytes{+len, buf}, +at} = r
439	  rd.cut.if(Bytes.fits(len, at, n), Bytes.Bytes{len, buf}, at, n)
440	
441	# The next n octets.
442	def rd.cut(+n: U32) -> Dec(Bytes.Bytes):
443	  r => rd.cut.at(n, r)
444	
445	def rd.flag.of(+m: U32) -> Dec(Bool):
446	  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)))
447	
448	# A u32 flag: 1 is True, 0 is False, and any other value fails.
449	def rd.flag() -> Dec(Bool):
450	  Dec.bind(U32, Bool, rd.u32(), m => rd.flag.of(m))
451	
452	def rd.payload(-A: Data, +i: Nat, +d: Nat, m: Maybe<&2, A>) -> Dec(Q.Ticket<A>):
453	  match m:
454	    case None{}:
455	      Dec.fail(Q.Ticket<A>)
456	    case Some{x}:
457	      Dec.pure(Q.Ticket<A>, Q.Ticket{i, d, x})
458	
459	def rd.ticket(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>) -> Dec(Q.Ticket<A>):
460	  do Dec<Q.Ticket<A>>:
461	    i : Nat <- rd.u64()
462	    d : Nat <- rd.u64()
463	    n : U32 <- rd.u32()
464	    p : Bytes.Bytes <- rd.cut(n)
465	    rd.payload(A, i, d, pdec(p))
466	
467	# Each rd.* tree reader takes the flag already read. fuel bounds the depth: every node holds at least one 4-octet flag.
468	def rd.heap(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +fuel: Nat, node: Bool) -> Dec(H.Heap<Q.Ticket<A>>):
469	  match fuel node:
470	    case 0n b:
471	      Dec.fail(H.Heap<Q.Ticket<A>>)
472	    case 1n+f False{}:
473	      Dec.pure(H.Heap<Q.Ticket<A>>, H.HLeaf{})
474	    case 1n+f True{}:
475	      do Dec<H.Heap<Q.Ticket<A>>>:
476	        n : Nat <- rd.u64()
477	        t : Q.Ticket<A> <- rd.ticket(~A, ~pdec)
478	        lf : Bool <- rd.flag()
479	        l : H.Heap<Q.Ticket<A>> <- rd.heap(~A, ~pdec, f, lf)
480	        rf : Bool <- rd.flag()
481	        r : H.Heap<Q.Ticket<A>> <- rd.heap(~A, ~pdec, f, rf)
482	        return H.HNode{n, t, l, r}
483	
484	def rd.run(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +fuel: Nat, more: Bool) -> Dec(List<&2, Q.Ticket<A>>):
485	  match fuel more:
486	    case 0n b:
487	      Dec.fail(List<&2, Q.Ticket<A>>)
488	    case 1n+f False{}:
489	      Dec.pure(List<&2, Q.Ticket<A>>, Nil{})
490	    case 1n+f True{}:
491	      do Dec<List<&2, Q.Ticket<A>>>:
492	        t : Q.Ticket<A> <- rd.ticket(~A, ~pdec)
493	        m : Bool <- rd.flag()
494	        rest : List<&2, Q.Ticket<A>> <- rd.run(~A, ~pdec, f, m)
495	        return Con{t, rest}
496	
497	def rd.tries(+fuel: Nat, node: Bool) -> Dec(O.OMap<Nat, Try>):
498	  match fuel node:
499	    case 0n b:
500	      Dec.fail(O.OMap<Nat, Try>)
501	    case 1n+f False{}:
502	      Dec.pure(O.OMap<Nat, Try>, O.OTip{})
503	    case 1n+f True{}:
504	      do Dec<O.OMap<Nat, Try>>:
505	        n : U32 <- rd.u32()
506	        k : Nat <- rd.u64()
507	        fails : Nat <- rd.u64()
508	        st : Nat <- rd.u64()
509	        lf : Bool <- rd.flag()
510	        lo : O.OMap<Nat, Try> <- rd.tries(f, lf)
511	        hf : Bool <- rd.flag()
512	        hi : O.OMap<Nat, Try> <- rd.tries(f, hf)
513	        return O.OBin{n, k, Try{fails, st}, lo, hi}
514	
515	def rd.state(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +fuel: Nat) -> Dec(State<A>):
516	  do Dec<State<A>>:
517	    n : Nat <- rd.u64()
518	    hf : Bool <- rd.flag()
519	    h : H.Heap<Q.Ticket<A>> <- rd.heap(~A, ~pdec, fuel, hf)
520	    rf : Bool <- rd.flag()
521	    run : List<&2, Q.Ticket<A>> <- rd.run(~A, ~pdec, fuel, rf)
522	    tf : Bool <- rd.flag()
523	    ts : O.OMap<Nat, Try> <- rd.tries(fuel, tf)
524	    return State{Q.State{n, h, run}, ts}
525	
526	def decode.snap.end(-A: Data, +len: U32, m: Maybe<&1, Rd & State<A>>) -> Maybe<&2, State<A>>:
527	  match m:
528	    case None{}:
529	      None{}
530	    case Some{(Rd{b, +at}, +s)}:
531	      Bool.pick(Maybe<&2, State<A>>, U32.is_eq(at, len), Some{s}, None{})
532	
533	# The state of a Snap record whose tag is already checked. The decoder must end at the record's end.
534	def decode.snap(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, b: Bytes.Bytes) -> Maybe<&2, State<A>>:
535	  Bytes.Bytes{+len, buf} = b
536	  decode.snap.end(A, len, rd.state(~A, ~pdec, U32.to_nat((len / 4 + 1 : U32)))(Rd{Bytes.Bytes{len, buf}, 4}))
537	
538	def decode.snapped(-A: Data, m: Maybe<&2, State<A>>) -> Maybe<&2, Rec<A>>:
539	  match m:
540	    case None{}:
541	      None{}
542	    case Some{s}:
543	      Some{Snap{s}}
544	
545	def decode.did(-A: Data, m: Maybe<&2, Op<A>>) -> Maybe<&2, Rec<A>>:
546	  match m:
547	    case None{}:
548	      None{}
549	    case Some{op}:
550	      Some{Do{op}}
551	
552	def decode.rec.if(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, snap: Bool, b: Bytes.Bytes) -> Maybe<&2, Rec<A>>:
553	  match snap:
554	    case True{}:
555	      decode.snapped(A, decode.snap(~A, ~pdec, b))
556	    case False{}:
557	      decode.did(A, decode(~A, ~pdec, b))
558	
559	def decode.rec.tag(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, r: Bytes.Bytes & U32) -> Maybe<&2, Rec<A>>:
560	  (b, +tag) = r
561	  decode.rec.if(~A, ~pdec, U32.is_eq(tag, 7), b)
562	
563	# The Rec of a journal record, or None when it is neither a Snap nor an op that encode.rec writes.
564	def decode.rec(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, b: Bytes.Bytes) -> Maybe<&2, Rec<A>>:
565	  decode.rec.tag(~A, ~pdec, get.u32(Bytes.get.u32le(b, 0)))
566	
567	# ---- the runner
568	
569	# policy: retry's policy for each ticket. circuit: the breaker of each target. idle: ms that drain waits when no attempt started.
570	# compact: tick compacts the journal once records past this many octets were appended since open or the last compaction; 0 never.
571	type Config is Data:
572	  Config{policy: Retry.Policy, circuit: Retry.Circuit, idle: U32, compact: U32}
573	
574	# The journal, its path, its file cap, and its length at open or at the last compaction.
575	type Log is Type:
576	  Log{journal: J.Journal, path: String, limit: U32, base: U32}
577	
578	# An open runner. Replaying its journal gives its state; its pool bounds the attempts that run at once.
579	type Runner<-A: Data> is Type:
580	  Runner{log: Log, state: State<A>, pool: Res.Pool, breakers: Map<&2, Retry.Breaker>, config: Config}
581	
582	# An attempt that runs: its unit of the pool, the budget that admit left, and its target.
583	type Flight<-A: Data> is Type:
584	  Flight{ticket: Q.Ticket<A>, grant: Res.Grant, budget: Retry.Budget, target: String}
585	
586	def Out(-X: Type) -> Type:
587	  Result<&1, &1, U32 & String, X>
588	
589	# Each attempt sends its ticket id and outcome on the round's one channel, so tick settles them in the order they finish.
590	def Msg() -> Type:
591	  Nat & Retry.Outcome
592	
593	# The runner, the attempts that run, the claim fuel left, and the first error.
594	def Claimed(-A: Data) -> Type:
595	  Runner<A> & List<&1, Flight<A>> & Nat & Out(Unit)
596	
597	def Ran(-A: Data) -> Type:
598	  Runner<A> & Out(Unit)
599	
600	# Wall-clock ms since the Unix epoch. Deadlines must keep their meaning across a restart.
601	def clock() -> IO(Nat):
602	  do IO<Nat>:
603	    t : Time.Instant <- Time.now()
604	    return Time.Duration.to_ms_nat(Time.Instant.since_epoch(t))
605	
606	def first(a: Out(Unit), b: Out(Unit)) -> Out(Unit):
607	  match a:
608	    case Fail{e}:
609	      Fail{e}
610	    case Done{u}:
611	      b
612	
613	def log.appended(+path: String, +limit: U32, +base: U32, x: J.Journal & Out(Unit)) -> Log & Out(Unit):
614	  (j, o) = x
615	  (Log{j, path, limit, base}, o)
616	
617	def log.append(l: Log, p: Bytes.Bytes) -> IO(Log & Out(Unit)):
618	  Log{j, +path, +limit, +base} = l
619	  do IO<Log & Out(Unit)>:
620	    x : J.Journal & Out(Unit) <- J.append(j, p)
621	    return log.appended(path, limit, base, x)
622	
623	def log.close(l: Log) -> IO(Unit):
624	  Log{j, path, limit, base} = l
625	  J.close(j)
626	
627	# The journal reads with limit as its record cap, so that a snapshot fits; appends keep max.
628	def log.of(j: J.Journal, +max: U32, path: String, +limit: U32) -> Log:
629	  J.Journal{file, cap, +end, ok} = j
630	  Log{J.Journal{file, max, end, ok}, path, limit, end}
631	
632	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) -> Ran(A):
633	  match x:
634	    case (j, Done{u}):
635	      (Runner{j, step(~A, s, op), p, bs, c}, Done{u})
636	    case (j, Fail{e}):
637	      (Runner{j, s, p, bs, c}, Fail{e})
638	
639	# Appends op, and applies it once the append is durable. After a failed append the state is as it was,
640	# and the journal refuses every later append until the runner is reopened.
641	def commit(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +op: Op<A>) -> IO(Ran(A)):
642	  Runner{j, s, p, bs, c} = r
643	  do IO<Ran(A)>:
644	    x : Log & Out(Unit) <- log.append(j, encode(~A, ~penc, op))
645	    return commit.done(~A, x, s, op, p, bs, c)
646	
647	# A grant of this runner's pool always fits it, so Kept does not happen.
648	def release.got(-A: Data, x: Res.Release, j: Log, s: State<A>, bs: Map<&2, Retry.Breaker>, c: Config) -> Runner<A>:
649	  match x:
650	    case Res.Released{p}:
651	      Runner{j, s, p, bs, c}
652	    case Res.Kept{p, g}:
653	      Runner{j, s, p, bs, c}
654	
655	def release(-A: Data, r: Runner<A>, g: Res.Grant) -> Runner<A>:
656	  Runner{j, s, p, bs, c} = r
657	  release.got(A, Res.release(p, g), j, s, bs, c)
658	
659	def claim.held(-A: Data, y: Ran(A), g: Res.Grant, fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
660	  match y:
661	    case (r, Fail{e}):
662	      IO.pure(Claimed(A), (release(A, r, g), fs, 0n, Fail{e}))
663	    case (r, Done{u}):
664	      next(release(A, r, g), fs)
665	
666	def claim.tag(+id: Nat, o: Retry.Outcome) -> IO(Msg()):
667	  IO.pure(Msg(), (id, o))
668	
669	def claim.go(~A: Data, ~penc: A -> Bytes.Bytes, ~work: A -> IO(Retry.Outcome), +ch: Chan(Msg()), +now: Nat, a: Retry.Breaker & Retry.Budget & Retry.Gate, +key: String,
670	    j: Log, s: State<A>, p: Res.Pool, bs: Map<&2, Retry.Breaker>, c: Config, +id: Nat, d: Nat, +x: A, g: Res.Grant,
671	    fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
672	  match a:
673	    case (+br, b, Retry.GateGo{}):
674	      do IO<Claimed(A)>:
675	        u : Chan(Msg()) <- IO.fork.go(Msg(), IO.bind(Retry.Outcome, Msg(), work(x), o => claim.tag(id, o)), ch)
676	        next(Runner{j, s, p, Map.set(&2, Retry.Breaker, bs, key, br), c}, Con{Flight{Q.Ticket{id, d, x}, g, b, key}, fs})
677	    case (+br, b, Retry.GateNo{why}):
678	      do IO<Claimed(A)>:
679	        y : Ran(A) <- commit(~A, ~penc, Runner{j, s, p, Map.set(&2, Retry.Breaker, bs, key, br), c}, held(A, why, br, id, now))
680	        claim.held(A, y, g, fs, next)
681	
682	def claim.gate(~A: Data, ~penc: A -> Bytes.Bytes, ~work: A -> IO(Retry.Outcome), +ch: Chan(Msg()), +now: Nat, got: Map<&2, Retry.Breaker> & Retry.Breaker, +key: String,
683	    j: Log, +s: State<A>, p: Res.Pool, +c: Config, +id: Nat, d: Nat, +x: A, g: Res.Grant,
684	    fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
685	  (bs, br) = got
686	  Config{pol, circ, idle, every} = c
687	  claim.go(~A, ~penc, ~work, ch, now, Retry.admit(br, budget(pol, tries.get(tries(A, s), id)), now), key, j, s, p, bs, c, id, d, x, g, fs, next)
688	
689	# The selected ticket asks its target's breaker and its own budget for an attempt.
690	def claim.admit(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +ch: Chan(Msg()), +now: Nat, r: Runner<A>, t: Q.Ticket<A>, g: Res.Grant,
691	    fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
692	  Runner{j, s, p, bs, +c} = r
693	  Config{pol, circ, idle, every} = c
694	  Q.Ticket{id, d, +x} = t
695	  +key = target(x)
696	  claim.gate(~A, ~penc, ~work, ch, now, Map.get(Retry.Breaker, Retry.Breaker.new(circ), bs, key), key, j, s, p, c, id, d, x, g, fs, next)
697	
698	def claim.selected(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +ch: Chan(Msg()), +now: Nat, y: Ran(A),
699	    t: Q.Ticket<A>, g: Res.Grant, fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
700	  match y:
701	    case (r, Fail{e}):
702	      IO.pure(Claimed(A), (release(A, r, g), fs, 0n, Fail{e}))
703	    case (r, Done{u}):
704	      claim.admit(~A, ~penc, ~target, ~work, ch, now, r, t, g, fs, next)
705	
706	def claim.pick(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +ch: Chan(Msg()), +left: Nat, +now: Nat, k: Pick<A>, j: Log, s: State<A>,
707	    bs: Map<&2, Retry.Breaker>, c: Config, fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
708	  match k:
709	    case Idle{p}:
710	      IO.pure(Claimed(A), (Runner{j, s, p, bs, c}, fs, Nat.add(1n, left), Done{Unit{}}))
711	    case Picked{p, t, g}:
712	      do IO<Claimed(A)>:
713	        y : Ran(A) <- commit(~A, ~penc, Runner{j, s, p, bs, c}, Select{now})
714	        claim.selected(~A, ~penc, ~target, ~work, ch, now, y, t, g, fs, next)
715	
716	# Starts an attempt for each due ticket while the pool has a free unit; each attempt sends its outcome on ch.
717	# fuel bounds the tickets taken, and the fuel left comes back.
718	def claim(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), fuel: Nat, +ch: Chan(Msg()), +now: Nat, r: Runner<A>, fs: List<&1, Flight<A>>) ->
719	    IO(Claimed(A)):
720	  match fuel:
721	    case 0n:
722	      IO.pure(Claimed(A), (r, fs, 0n, Done{Unit{}}))
723	    case 1n+n:
724	      Runner{j, +s, p, bs, c} = r
725	      +m = n
726	      claim.pick(~A, ~penc, ~target, ~work, ch, m, now, pick(~A, p, s, now), j, s, bs, c, fs, (r2 => fs2 => claim(~A, ~penc, ~target, ~work, m, ch, now, r2, fs2)))
727	
728	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)):
729	  (r, res) = y
730	  next(release(A, r, g), first(err, res))
731	
732	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>,
733	    j: Log, s: State<A>, p: Res.Pool, c: Config, g: Res.Grant, err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
734	  (bs, br) = got
735	  do IO<Runner<A> & Out(Bool)>:
736	    y : Ran(A) <- commit(~A, ~penc, Runner{j, s, p, Map.set(&2, Retry.Breaker, bs, key, Retry.Breaker.done(br, now, failed)), c}, op)
737	    settle.done(A, y, g, err, next)
738	
739	# Tells the target's breaker, appends Complete or Requeue, and returns the attempt's unit to the pool.
740	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,
741	    err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
742	  Runner{j, +s, p, bs, +c} = r
743	  Config{pol, circ, idle, every} = c
744	  settle.br(~A, ~penc, Retry.failed(o), key, now, Map.get(Retry.Breaker, Retry.Breaker.new(circ), bs, key),
745	    after(A, o, pol, b, id, fails(tries.get(tries(A, s), id)), now, rnd), j, s, p, c, g, err, next)
746	
747	def random(r: Out(U32)) -> Nat:
748	  match r:
749	    case Done{x}:
750	      U32.to_nat(x)
751	    case Fail{e}:
752	      0n
753	
754	def take.keep(-A: Data, m: Maybe<&1, Flight<A>>, xs: List<&1, Flight<A>>) -> List<&1, Flight<A>>:
755	  match m:
756	    case None{}:
757	      xs
758	    case Some{f}:
759	      Con{f, xs}
760	
761	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>>:
762	  match eq:
763	    case True{}:
764	      (m, xs) = r
765	      (Some{f}, take.keep(A, m, xs))
766	    case False{}:
767	      (m, xs) = r
768	      (m, Con{f, xs})
769	
770	# The flight of ticket id, and the other flights.
771	def take(-A: Data, fs: List<&1, Flight<A>>, +id: Nat) -> Maybe<&1, Flight<A>> & List<&1, Flight<A>>:
772	  match fs:
773	    case Nil{}:
774	      (None{}, Nil{})
775	    case Con{Flight{Q.Ticket{+i, d, x}, g, b, k}, rest}:
776	      take.at(A, Nat.is_eq(i, id), Flight{Q.Ticket{i, d, x}, g, b, k}, take(A, rest, id))
777	
778	def settle.flight(~A: Data, ~penc: A -> Bytes.Bytes, mf: Maybe<&1, Flight<A>>, fs: List<&1, Flight<A>>, +o: Retry.Outcome, +now: Nat, +rnd: Nat, r: Runner<A>, err: Out(Unit),
779	    next: Runner<A> -> List<&1, Flight<A>> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
780	  match mf:
781	    case None{}:
782	      next(r, fs, err)
783	    case Some{Flight{Q.Ticket{id, d, x}, g, b, key}}:
784	      settle.op(~A, ~penc, o, b, id, key, now, rnd, r, g, err, (r2 => e2 => next(r2, fs, e2)))
785	
786	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, r: Runner<A>, err: Out(Unit),
787	    next: Runner<A> -> List<&1, Flight<A>> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
788	  (mf, fs) = t
789	  settle.flight(~A, ~penc, mf, fs, o, now, rnd, r, err, next)
790	
791	# Settles the flight that m names. A closed channel is an error, and no flight settles.
792	def settle(~A: Data, ~penc: A -> Bytes.Bytes, m: Maybe<&1, Msg()>, +now: Nat, +rnd: Nat, r: Runner<A>, fs: List<&1, Flight<A>>, err: Out(Unit),
793	    next: Runner<A> -> List<&1, Flight<A>> -> Out(Unit) -> IO(Runner<A> & Out(Bool))) -> IO(Runner<A> & Out(Bool)):
794	  match m:
795	    case None{}:
796	      next(r, fs, first(err, Fail{(32, "runner: the outcome channel closed")}))
797	    case Some{(+id, +o)}:
798	      settle.take(~A, ~penc, take(A, fs, id), o, now, rnd, r, err, next)
799	
800	def merged(-A: Data, e: Out(Unit), k: Claimed(A)) -> Claimed(A):
801	  (r, fs, n, err) = k
802	  (r, fs, n, first(e, err))
803	
804	def empty.heap(-A: Data, h: H.Heap<Q.Ticket<A>>) -> Bool:
805	  match h:
806	    case H.HLeaf{}:
807	      True{}
808	    case H.HNode{n, t, l, r}:
809	      False{}
810	
811	def empty.list(-A: Data, xs: List<&2, Q.Ticket<A>>) -> Bool:
812	  match xs:
813	    case Nil{}:
814	      True{}
815	    case Con{t, rest}:
816	      False{}
817	
818	# No ticket is pending or running.
819	def idle(-A: Data, s: State<A>) -> Bool:
820	  State{Q.State{n, h, run}, ts} = s
821	  Bool.and(empty.heap(A, h), empty.list(A, run))
822	
823	def pending(-A: Data, s: State<A>) -> Nat:
824	  State{Q.State{n, h, run}, ts} = s
825	  H.Heap.size(Q.Ticket<A>, h)
826	
827	# ---- compaction
828	
829	# Kept: compaction failed before the rename, and the old journal stands.
830	# Lost: the rename happened, but the new journal did not open. Moved: the journal at path is the snapshot.
831	type Swap is Type:
832	  Kept{err: U32 & String}
833	  Lost{err: U32 & String}
834	  Moved{journal: J.Journal}
835	
836	def swap.reopened(x: Out(J.Journal & Unit)) -> Swap:
837	  match x:
838	    case Fail{e}:
839	      Lost{e}
840	    case Done{(j, u)}:
841	      Moved{j}
842	
843	# The reopen fsyncs the directory, so the rename is durable before the next append.
844	def swap.renamed(+path: String, +limit: U32, x: Out(Unit)) -> IO(Swap):
845	  match x:
846	    case Fail{e}:
847	      IO.pure(Swap, Kept{e})
848	    case Done{u}:
849	      do IO<Swap>:
850	        y : Out(J.Journal & Unit) <- J.open(~Unit, ~(a => b => a), path, limit, limit, Unit{})
851	        return swap.reopened(y)
852	
853	def swap.written(+path: String, +tmp: String, +limit: U32, x: J.Journal & Out(Unit)) -> IO(Swap):
854	  match x:
855	    case (t, Fail{e}):
856	      do IO<Swap>:
857	        J.close(t)
858	        return Kept{e}
859	    case (t, Done{u}):
860	      do IO<Swap>:
861	        J.close(t)
862	        y : Out(Unit) <- Files.rename(tmp, path)
863	        swap.renamed(path, limit, y)
864	
865	def swap.opened(+path: String, +tmp: String, +limit: U32, x: Out(J.Journal & Unit), snap: Bytes.Bytes) -> IO(Swap):
866	  match x:
867	    case Fail{e}:
868	      IO.pure(Swap, Kept{e})
869	    case Done{(t, u)}:
870	      do IO<Swap>:
871	        y : J.Journal & Out(Unit) <- J.append(t, snap)
872	        swap.written(path, tmp, limit, y)
873	
874	# Writes snap as the only record of path.compact, fsyncs it, and renames it over path.
875	def swap(+path: String, +limit: U32, snap: Bytes.Bytes) -> IO(Swap):
876	  +tmp = path ++ ".compact"
877	  do IO<Swap>:
878	    gone : Out(Unit) <- Files.remove(tmp)
879	    x : Out(J.Journal & Unit) <- J.open(~Unit, ~(a => b => a), tmp, limit, limit, Unit{})
880	    swap.opened(path, tmp, limit, x, snap)
881	
882	def log.stop(j: J.Journal) -> J.Journal:
883	  J.Journal{file, max, end, ok} = j
884	  J.Journal{file, max, end, False{}}
885	
886	def compact.moved(-A: Data, x: Swap, j: J.Journal, +max: U32, path: String, +limit: U32, +base: U32, s: State<A>, p: Res.Pool, bs: Map<&2, Retry.Breaker>, c: Config) ->
887	    IO(Ran(A)):
888	  match x:
889	    case Kept{e}:
890	      IO.pure(Ran(A), (Runner{Log{j, path, limit, base}, s, p, bs, c}, Fail{e}))
891	    case Lost{e}:
892	      IO.pure(Ran(A), (Runner{Log{log.stop(j), path, limit, base}, s, p, bs, c}, Fail{e}))
893	    case Moved{k}:
894	      do IO<Ran(A)>:
895	        J.close(j)
896	        return (Runner{log.of(k, max, path, limit), s, p, bs, c}, Done{Unit{}})
897	
898	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>, p: Res.Pool,
899	    bs: Map<&2, Retry.Breaker>, c: Config) -> IO(Ran(A)):
900	  match ok:
901	    case False{}:
902	      IO.pure(Ran(A), (Runner{Log{j, path, limit, base}, s, p, bs, c}, Fail{(5, "runner: an append failed; reopen the runner")}))
903	    case True{}:
904	      do IO<Ran(A)>:
905	        x : Swap <- swap(path, limit, encode.snap(~A, ~penc, s))
906	        compact.moved(A, x, j, max, path, limit, base, s, p, bs, c)
907	
908	# Writes the state as one Snap record into a new journal, fsyncs it, and renames it over the old one.
909	# A crash leaves one of the two journals, and both replay to this state. After a failure the old
910	# journal stands; when the failure came after the rename, appends stop until the runner is reopened.
911	def compact(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>) -> IO(Ran(A)):
912	  Runner{Log{j, path, limit, base}, s, p, bs, c} = r
913	  J.Journal{file, +max, end, +ok} = j
914	  compact.if(~A, ~penc, ok, J.Journal{file, max, end, ok}, max, path, limit, base, s, p, bs, c)
915	
916	def tick.ran(e: Out(Unit), b: Bool) -> Out(Bool):
917	  match e:
918	    case Fail{x}:
919	      Fail{x}
920	    case Done{u}:
921	      Done{b}
922	
923	def tick.compacted(-A: Data, +b: Bool, y: Ran(A)) -> Runner<A> & Out(Bool):
924	  (r, e) = y
925	  (r, tick.ran(e, b))
926	
927	def tick.due(~A: Data, ~penc: A -> Bytes.Bytes, due: Bool, +b: Bool, r: Runner<A>) -> IO(Runner<A> & Out(Bool)):
928	  match due:
929	    case False{}:
930	      IO.pure(Runner<A> & Out(Bool), (r, Done{b}))
931	    case True{}:
932	      do IO<Runner<A> & Out(Bool)>:
933	        y : Ran(A) <- compact(~A, ~penc, r)
934	        return tick.compacted(A, b, y)
935	
936	def tick.size(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +b: Bool) -> IO(Runner<A> & Out(Bool)):
937	  Runner{Log{J.Journal{file, max, +end, ok}, path, limit, +base}, s, p, bs, +c} = r
938	  Config{pol, circ, idle, +every} = c
939	  tick.due(~A, ~penc, Bool.and(Bool.not(U32.is_eq(every, 0)), U32.is_gt((end - base : U32), every)), b,
940	    Runner{Log{J.Journal{file, max, end, ok}, path, limit, base}, s, p, bs, c})
941	
942	# Compacts once the journal grew past Config.compact octets since open or the last compaction.
943	def tick.grown(~A: Data, ~penc: A -> Bytes.Bytes, t: Runner<A> & Out(Bool)) -> IO(Runner<A> & Out(Bool)):
944	  match t:
945	    case (r, Fail{e}):
946	      IO.pure(Runner<A> & Out(Bool), (r, Fail{e}))
947	    case (r, Done{b}):
948	      tick.size(~A, ~penc, r, b)
949	
950	# Waits for the next attempt to finish, settles it, then claims due tickets into the freed unit with the claim fuel left.
951	# fuel bounds the settles. claim starts at most as many attempts as the same fuel, so every attempt settles before fuel ends.
952	def tick.run(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), fuel: Nat, +ch: Chan(Msg()), +ran: Bool, k: Claimed(A)) ->
953	    IO(Runner<A> & Out(Bool)):
954	  match fuel:
955	    case 0n:
956	      match k:
957	        case (r, Nil{}, left, err):
958	          IO.pure(Runner<A> & Out(Bool), (r, tick.ran(err, ran)))
959	        case (r, Con{f, fs}, left, err):
960	          IO.pure(Runner<A> & Out(Bool), (r, tick.ran(first(err, Fail{(62, "runner: an attempt outlived the round")}), ran)))
961	    case 1n+n:
962	      match k:
963	        case (r, Nil{}, left, err):
964	          IO.pure(Runner<A> & Out(Bool), (r, tick.ran(err, ran)))
965	        case (r, Con{f, fs}, +left, err):
966	          do IO<Runner<A> & Out(Bool)>:
967	            m : Maybe<&1, Msg()> <- Chan.recv(Msg(), ch)
968	            +now : Nat <- clock()
969	            rnd : Out(U32) <- IO.random_u32()
970	            settle(~A, ~penc, m, now, random(rnd), r, Con{f, fs}, err, (r2 => fs2 => e2 =>
971	              IO.bind(Claimed(A), Runner<A> & Out(Bool), claim(~A, ~penc, ~target, ~work, left, ch, now, r2, fs2),
972	                k2 => tick.run(~A, ~penc, ~target, ~work, n, ch, True{}, merged(A, e2, k2)))))
973	
974	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)):
975	  Runner{j, +s, +p, bs, c} = r
976	  +fuel = Nat.add(1n, pending(A, s))
977	  do IO<Runner<A> & Out(Bool)>:
978	    +ch : Chan(Msg()) <- Chan.new(Msg(), (U32.from_nat(Res.capacity(p)) + 1 : U32))
979	    k : Claimed(A) <- claim(~A, ~penc, ~target, ~work, fuel, ch, now, Runner{j, s, p, bs, c}, Nil{})
980	    t : Runner<A> & Out(Bool) <- tick.run(~A, ~penc, ~target, ~work, fuel, ch, False{}, k)
981	    Chan.close(Msg(), ch)
982	    return t
983	
984	# One round: start an attempt for each due ticket the pool admits. As each attempt finishes, record its outcome,
985	# return its unit, and start the next due ticket in that unit. The round ends when no attempt runs.
986	# Then compact the journal when it grew past Config.compact. Done{True} when an attempt ran.
987	# work runs on its own; target names the breaker that guards it.
988	def tick(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), r: Runner<A>) -> IO(Runner<A> & Out(Bool)):
989	  do IO<Runner<A> & Out(Bool)>:
990	    now : Nat <- clock()
991	    t : Runner<A> & Out(Bool) <- tick.at(~A, ~penc, ~target, ~work, now, r)
992	    tick.grown(~A, ~penc, t)
993	
994	def drain.ticked(-A: Data, +idle: U32, t: Runner<A> & Out(Bool), next: Runner<A> -> IO(Ran(A))) -> IO(Ran(A)):
995	  match t:
996	    case (r, Fail{e}):
997	      IO.pure(Ran(A), (r, Fail{e}))
998	    case (r, Done{True{}}):
999	      next(r)
1000	    case (r, Done{False{}}):
1001	      do IO<Ran(A)>:
1002	        IO.sleep(idle)
1003	        next(r)
1004	
1005	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)):
1006	  match done:
1007	    case True{}:
1008	      IO.pure(Ran(A), (r, Done{Unit{}}))
1009	    case False{}:
1010	      Runner{j, s, p, bs, +c} = r
1011	      Config{pol, circ, +idle, every} = c
1012	      do IO<Ran(A)>:
1013	        t : Runner<A> & Out(Bool) <- tick(~A, ~penc, ~target, ~work, Runner{j, s, p, bs, c})
1014	        drain.ticked(A, idle, t, next)
1015	
1016	# Ticks until no ticket is pending or running. It waits idle ms after a round that started no attempt.
1017	# Fails when fuel rounds leave work.
1018	def drain(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), fuel: Nat, r: Runner<A>) -> IO(Ran(A)):
1019	  match fuel:
1020	    case 0n:
1021	      Runner{j, +s, p, bs, c} = r
1022	      IO.pure(Ran(A), (Runner{j, s, p, bs, c}, Bool.pick(Out(Unit), idle(A, s), Done{Unit{}}, Fail{(62, "runner: drain ran out of rounds")})))
1023	    case 1n+n:
1024	      Runner{j, +s, p, bs, c} = r
1025	      drain.if(~A, ~penc, ~target, ~work, idle(A, s), Runner{j, s, p, bs, c}, (r2 => drain(~A, ~penc, ~target, ~work, n, r2)))
1026	
1027	def orphans.next(-A: Data, y: Ran(A), err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Ran(A))) -> IO(Ran(A)):
1028	  (r, e) = y
1029	  next(r, first(err, e))
1030	
1031	# A running ticket at open lost its attempt in a crash: it is requeued at now, as a failed attempt.
1032	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)):
1033	  match ts:
1034	    case Nil{}:
1035	      IO.pure(Ran(A), (r, err))
1036	    case Con{Q.Ticket{id, d, x}, rest}:
1037	      do IO<Ran(A)>:
1038	        y : Ran(A) <- commit(~A, ~penc, r, Requeue{id, now})
1039	        orphans.next(A, y, err, (r2 => e2 => orphans(~A, ~penc, rest, now, r2, e2)))
1040	
1041	def open.done(-A: Data, y: Ran(A)) -> IO(Out(Runner<A>)):
1042	  match y:
1043	    case (Runner{j, s, p, bs, c}, Fail{e}):
1044	      do IO<Out(Runner<A>)>:
1045	        log.close(j)
1046	        return Fail{e}
1047	    case (r, Done{u}):
1048	      IO.pure(Out(Runner<A>), Done{r})
1049	
1050	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) ->
1051	    IO(Out(Runner<A>)):
1052	  match x:
1053	    case Fail{e}:
1054	      IO.pure(Out(Runner<A>), Fail{e})
1055	    case Done{(j, None{})}:
1056	      do IO<Out(Runner<A>)>:
1057	        J.close(j)
1058	        return Fail{(22, "runner: a journal record is not a runner record")}
1059	    case Done{(j, Some{+s})}:
1060	      do IO<Out(Runner<A>)>:
1061	        now : Nat <- clock()
1062	        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}, Done{Unit{}})
1063	        open.done(A, y)
1064	
1065	def replay.step(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, acc: Maybe<&2, State<A>>, b: Bytes.Bytes) -> Maybe<&2, State<A>>:
1066	  load(~A, ~Bytes.Bytes, ~(y => decode.rec(~A, ~pdec, y)), acc, b)
1067	
1068	# Opens the runner at path: replays its journal into the state, then requeues each ticket that was running.
1069	# capacity bounds the attempts that run at once. max caps an op record's octets; limit caps the journal file and a snapshot.
1070	# Fails when the journal fails to open, or a committed record is not a snapshot or an op.
1071	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>)):
1072	  do IO<Out(Runner<A>)>:
1073	    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)})
1074	    open.got(~A, ~penc, x, max, path, limit, c, capacity)
1075	
1076	def enqueue.done(-A: Data, +id: Nat, y: Ran(A)) -> Runner<A> & Out(Nat):
1077	  match y:
1078	    case (r, Done{u}):
1079	      (r, Done{id})
1080	    case (r, Fail{e}):
1081	      (r, Fail{e})
1082	
1083	# Adds a job that is due at deadline (wall-clock ms) and answers its ticket id once the record is durable.
1084	def enqueue(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +deadline: Nat, x: A) -> IO(Runner<A> & Out(Nat)):
1085	  Runner{j, +s, p, bs, c} = r
1086	  do IO<Runner<A> & Out(Nat)>:
1087	    y : Ran(A) <- commit(~A, ~penc, Runner{j, s, p, bs, c}, Enqueue{deadline, x})
1088	    return enqueue.done(A, Q.next_id(A, queue(A, s)), y)
1089	
1090	# Cancels ticket id. A running attempt finishes, but its ticket is never selected again.
1091	def cancel(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +id: Nat) -> IO(Ran(A)):
1092	  commit(~A, ~penc, r, Cancel{id})
1093	
1094	# The runner back, and a copy of its state.
1095	def state(-A: Data, r: Runner<A>) -> Runner<A> & State<A>:
1096	  Runner{j, +s, p, bs, c} = r
1097	  (Runner{j, s, p, bs, c}, s)
1098	
1099	def close(-A: Data, r: Runner<A>) -> IO(Unit):
1100	  Runner{j, s, p, bs, c} = r
1101	  log.close(j)