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	
12	# One queue transition. Each one is one journal record, appended before the state changes.
13	# Select: the queue's select at now. Replay runs it at the same now, so it takes the same ticket.
14	# Requeue: the attempt failed; it counts one failure. Defer: the breaker held the ticket back; no attempt ran.
15	type Op<-A: Data> is Data:
16	  Enqueue{deadline: Nat, payload: A}
17	  Select{now: Nat}
18	  Complete{id: Nat}
19	  Cancel{id: Nat}
20	  Requeue{id: Nat, deadline: Nat}
21	  Defer{id: Nat, deadline: Nat}
22	
23	# The failed attempts of a ticket, and its first deadline, from which the policy deadline counts.
24	type Try is Data:
25	  Try{fails: Nat, start: Nat}
26	
27	# A Try for each ticket that is not yet complete or cancelled, by ticket id.
28	def Tries() -> Data:
29	  O.OMap<Nat, Try>
30	
31	# The durable state: the queue and the tries.
32	type State<-A: Data> is Data:
33	  State{queue: Q.State<A>, tries: Tries()}
34	
35	def new(-A: Data) -> State<A>:
36	  State{Q.new(A), O.OMap.new(Nat, Try)}
37	
38	def queue(-A: Data, s: State<A>) -> Q.State<A>:
39	  State{q, ts} = s
40	  q
41	
42	def tries(-A: Data, s: State<A>) -> Tries():
43	  State{q, ts} = s
44	  ts
45	
46	# ---- tries
47	
48	def tries.drop(ts: Tries(), +i: Nat) -> Tries():
49	  O.OMap.del(~Nat, ~Try, ~Nat.cmp, ts, i)
50	
51	def tries.get(ts: Tries(), +i: Nat) -> Maybe<&2, Try>:
52	  O.OMap.get(~Nat, ~Try, ~Nat.cmp, ts, i)
53	
54	def tries.bump.got(m: Maybe<&2, Try>, ts: Tries(), +i: Nat) -> Tries():
55	  match m:
56	    case None{}:
57	      ts
58	    case Some{Try{f, st}}:
59	      O.OMap.put(~Nat, ~Try, ~Nat.cmp, ts, i, Try{1n+f, st})
60	
61	def tries.bump(+ts: Tries(), +i: Nat) -> Tries():
62	  tries.bump.got(tries.get(ts, i), ts, i)
63	
64	# ---- the transition
65	
66	def step.enqueued(-A: Data, r: Q.State<A> & Nat, ts: Tries(), d: Nat) -> State<A>:
67	  (q, i) = r
68	  State{q, O.OMap.put(~Nat, ~Try, ~Nat.cmp, ts, i, Try{0n, d})}
69	
70	def step.selected(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, ts: Tries()) -> State<A>:
71	  (q, t) = r
72	  State{q, ts}
73	
74	def step.requeued(-A: Data, r: Q.State<A> & Bool, ts: Tries()) -> State<A>:
75	  (q, b) = r
76	  State{q, ts}
77	
78	# Complete takes the ticket from the running list only: O(running), not O(pending).
79	def step.completed(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, ts: Tries()) -> State<A>:
80	  (q, t) = r
81	  State{q, ts}
82	
83	# The state after op. Live and replay both run it, so they agree.
84	def step(~A: Data, s: State<A>, op: Op<A>) -> State<A>:
85	  match s op:
86	    case State{q, ts} Enqueue{+d, x}:
87	      step.enqueued(A, Q.enqueue(~A, d, x, q), ts, d)
88	    case State{q, ts} Select{now}:
89	      step.selected(A, Q.select(~A, now, q), ts)
90	    case State{q, ts} Complete{+i}:
91	      step.completed(A, Q.complete(A, i, q), tries.drop(ts, i))
92	    case State{q, ts} Cancel{+i}:
93	      State{Q.cancel(~A, i, q), tries.drop(ts, i)}
94	    case State{q, ts} Requeue{+i, d}:
95	      step.requeued(A, Q.requeue(~A, i, d, q), tries.bump(ts, i))
96	    case State{q, ts} Defer{i, d}:
97	      step.requeued(A, Q.requeue(~A, i, d, q), ts)
98	
99	def apply.op(~A: Data, m: Maybe<&2, Op<A>>, s: State<A>) -> Maybe<&2, State<A>>:
100	  match m:
101	    case None{}:
102	      None{}
103	    case Some{op}:
104	      Some{step(~A, s, op)}
105	
106	# Folds one record into the replayed state. A record that does not decode ends the replay with None.
107	def apply(~A: Data, ~R: Type, ~dec: R -> Maybe<&2, Op<A>>, acc: Maybe<&2, State<A>>, r: R) -> Maybe<&2, State<A>>:
108	  match acc:
109	    case None{}:
110	      None{}
111	    case Some{s}:
112	      apply.op(~A, dec(r), s)
113	
114	# ---- pool
115	
116	# Picked: a due ticket and one unit of the pool for it. Idle: no due ticket, or no free unit; the pool is unchanged.
117	type Pick<-A: Data> is Type:
118	  Picked{pool: Res.Pool, ticket: Q.Ticket<A>, grant: Res.Grant}
119	  Idle{pool: Res.Pool}
120	
121	def pick.took(-A: Data, r: Res.Reserve, t: Q.Ticket<A>) -> Pick<A>:
122	  match r:
123	    case Res.Took{p, g}:
124	      Picked{p, t, g}
125	    case Res.Refused{p}:
126	      Idle{p}
127	
128	def pick.due(-A: Data, m: Maybe<&2, Q.Ticket<A>>, p: Res.Pool) -> Pick<A>:
129	  match m:
130	    case None{}:
131	      Idle{p}
132	    case Some{t}:
133	      pick.took(A, Res.reserve(p, 1n), t)
134	
135	def pick.sel(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, p: Res.Pool) -> Pick<A>:
136	  (q, m) = r
137	  pick.due(A, m, p)
138	
139	# The ticket that Select{now} takes, with one unit of p, when there is a ticket due and a unit free.
140	def pick(~A: Data, p: Res.Pool, s: State<A>, now: Nat) -> Pick<A>:
141	  State{q, ts} = s
142	  pick.sel(A, Q.select(~A, now, q), p)
143	
144	# ---- retry
145	
146	# The budget left for ticket i: the policy's attempts less its failures, until its start plus the policy deadline.
147	def budget.of(+attempts: Nat, t: Maybe<&2, Try>, d: Maybe<&2, Nat>) -> Retry.Budget:
148	  match t d:
149	    case None{} m:
150	      Retry.Budget{attempts, None{}}
151	    case Some{Try{f, st}} None{}:
152	      Retry.Budget{Nat.sub(attempts, f), None{}}
153	    case Some{Try{f, +st}} Some{ms}:
154	      Retry.Budget{Nat.sub(attempts, f), Some{Nat.add(st, ms)}}
155	
156	def budget(p: Retry.Policy, t: Maybe<&2, Try>) -> Retry.Budget:
157	  Retry.Policy{a, base, cap, d} = p
158	  budget.of(a, t, d)
159	
160	# The failures so far: the index of the next retry.
161	def fails(t: Maybe<&2, Try>) -> Nat:
162	  match t:
163	    case None{}:
164	      0n
165	    case Some{Try{f, st}}:
166	      f
167	
168	def after.plan(-A: Data, pl: Retry.Plan, +id: Nat, +now: Nat) -> Op<A>:
169	  match pl:
170	    case Retry.PlanStop{}:
171	      Complete{id}
172	    case Retry.PlanWait{w}:
173	      Requeue{id, Nat.add(now, w)}
174	
175	# The record after an attempt of ticket id ends at now with o: requeue at now plus retry's wait, or complete.
176	def after(-A: Data, o: Retry.Outcome, p: Retry.Policy, b: Retry.Budget, +id: Nat, +n: Nat, +now: Nat, +r: Nat) -> Op<A>:
177	  after.plan(A, Retry.plan(o, p, b, n, now, r), id, now)
178	
179	# When an open breaker lets attempts start again.
180	def until(br: Retry.Breaker, +now: Nat) -> Nat:
181	  match br:
182	    case Retry.BreakerOn{c, Retry.CircuitOpen{u}}:
183	      u
184	    case Retry.BreakerOn{c, Retry.CircuitClosed{fs}}:
185	      Nat.add(now, Retry.Circuit.cooldown(c))
186	    case Retry.BreakerOn{c, Retry.CircuitHalf{k}}:
187	      Nat.add(now, Retry.Circuit.cooldown(c))
188	    case Retry.BreakerOff{}:
189	      now
190	
191	# The record for a ticket that admit refused: Defer to when the breaker may admit, or Complete when its budget is spent or late.
192	def held(-A: Data, why: Retry.Deny, br: Retry.Breaker, +id: Nat, +now: Nat) -> Op<A>:
193	  match why:
194	    case Retry.DenyOpen{}:
195	      Defer{id, until(br, now)}
196	    case Retry.DenyLate{}:
197	      Complete{id}
198	    case Retry.DenySpent{}:
199	      Complete{id}
200	
201	# ---- the record codec: a u32le tag, then u64le fields; an Enqueue then holds the payload.
202	
203	def W() -> Nat:
204	  Nat.mul(65536n, 65536n)
205	
206	# Nat fields hold values under 2^64.
207	def put.u64(b: Bytes.Bytes, +at: U32, +n: Nat) -> Bytes.Bytes:
208	  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())))
209	
210	def head(+tag: U32, +a: Nat, +c: Nat) -> Bytes.Bytes:
211	  put.u64(put.u64(Bytes.set.u32le(Bytes.new(20), 0, tag), 4, a), 12, c)
212	
213	def encode(~A: Data, ~penc: A -> Bytes.Bytes, op: Op<A>) -> Bytes.Bytes:
214	  match op:
215	    case Enqueue{+d, x}:
216	      Bytes.append(put.u64(Bytes.set.u32le(Bytes.new(12), 0, 1), 4, d), penc(x))
217	    case Select{+now}:
218	      head(2, now, 0n)
219	    case Complete{+i}:
220	      head(3, i, 0n)
221	    case Cancel{+i}:
222	      head(4, i, 0n)
223	    case Requeue{+i, +d}:
224	      head(5, i, d)
225	    case Defer{+i, +d}:
226	      head(6, i, d)
227	
228	def get.field(m: Maybe<&2, U32>) -> U32:
229	  match m:
230	    case None{}:
231	      0
232	    case Some{x}:
233	      x
234	
235	def get.u32(r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & U32:
236	  (b, m) = r
237	  (b, get.field(m))
238	
239	def get.hi(+lo: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Nat:
240	  (b, m) = r
241	  (b, Nat.add(U32.to_nat(lo), Nat.mul(U32.to_nat(get.field(m)), W())))
242	
243	def get.lo(+at: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Nat:
244	  (b, m) = r
245	  +lo = get.field(m)
246	  get.hi(lo, Bytes.get.u32le(b, (at + 4 : U32)))
247	
248	def get.u64(b: Bytes.Bytes, +at: U32) -> Bytes.Bytes & Nat:
249	  get.lo(at, Bytes.get.u32le(b, at))
250	
251	def decode.small(-A: Data, ok: Bool, +tag: U32, +a: Nat, +c: Nat) -> Maybe<&2, Op<A>>:
252	  match ok:
253	    case False{}:
254	      None{}
255	    case True{}:
256	      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 2), Some{Select{a}},
257	      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 3), Some{Complete{a}},
258	      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 4), Some{Cancel{a}},
259	      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 5), Some{Requeue{a, c}},
260	      Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 6), Some{Defer{a, c}}, None{})))))
261	
262	def decode.payload(-A: Data, m: Maybe<&2, A>, +d: Nat) -> Maybe<&2, Op<A>>:
263	  match m:
264	    case None{}:
265	      None{}
266	    case Some{x}:
267	      Some{Enqueue{d, x}}
268	
269	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>>:
270	  (b, +c) = r
271	  decode.small(A, U32.is_eq(len, 20), tag, a, c)
272	
273	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>>:
274	  match enq:
275	    case True{}:
276	      decode.payload(A, pdec(Bytes.snd(Bytes.slice(b, 12, (len - 12 : U32)))), a)
277	    case False{}:
278	      decode.c(~A, ~pdec, tag, len, a, get.u64(b, 12))
279	
280	def decode.a(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +tag: U32, +len: U32, r: Bytes.Bytes & Nat) -> Maybe<&2, Op<A>>:
281	  (b, +a) = r
282	  decode.kind(~A, ~pdec, U32.is_eq(tag, 1), tag, len, a, b)
283	
284	def decode.tag(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +len: U32, r: Bytes.Bytes & U32) -> Maybe<&2, Op<A>>:
285	  (b, +tag) = r
286	  decode.a(~A, ~pdec, tag, len, get.u64(b, 4))
287	
288	def decode.len(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, ok: Bool, +len: U32, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
289	  match ok:
290	    case False{}:
291	      None{}
292	    case True{}:
293	      decode.tag(~A, ~pdec, len, get.u32(Bytes.get.u32le(b, 0)))
294	
295	# The op of a record, or None when it is not one that encode writes.
296	def decode(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
297	  Bytes.Bytes{+len, buf} = b
298	  decode.len(~A, ~pdec, U32.is_ge(len, 12), len, Bytes.Bytes{len, buf})
299	
300	# ---- the runner
301	
302	# policy: retry's policy for each ticket. circuit: the breaker of each target. idle: ms that drain waits when no attempt started.
303	type Config is Data:
304	  Config{policy: Retry.Policy, circuit: Retry.Circuit, idle: U32}
305	
306	# An open runner. Its journal holds every op that made state; its pool bounds the attempts that run at once.
307	type Runner<-A: Data> is Type:
308	  Runner{journal: J.Journal, state: State<A>, pool: Res.Pool, breakers: Map<&2, Retry.Breaker>, config: Config}
309	
310	# An attempt that runs: its unit of the pool, the budget that admit left, its target, and the channel of its outcome.
311	type Flight<-A: Data> is Type:
312	  Flight{ticket: Q.Ticket<A>, grant: Res.Grant, budget: Retry.Budget, target: String, chan: Chan(Retry.Outcome)}
313	
314	def Out(-X: Type) -> Type:
315	  Result<&1, &1, U32 & String, X>
316	
317	def Claimed(-A: Data) -> Type:
318	  Runner<A> & List<&1, Flight<A>> & Out(Unit)
319	
320	def Ran(-A: Data) -> Type:
321	  Runner<A> & Out(Unit)
322	
323	# Wall-clock ms since the Unix epoch. Deadlines must keep their meaning across a restart.
324	def clock() -> IO(Nat):
325	  do IO<Nat>:
326	    t : Time.Instant <- Time.now()
327	    return Time.Duration.to_ms_nat(Time.Instant.since_epoch(t))
328	
329	def first(a: Out(Unit), b: Out(Unit)) -> Out(Unit):
330	  match a:
331	    case Fail{e}:
332	      Fail{e}
333	    case Done{u}:
334	      b
335	
336	def commit.done(~A: Data, x: J.Journal & Out(Unit), s: State<A>, +op: Op<A>, p: Res.Pool, bs: Map<&2, Retry.Breaker>, c: Config) -> Ran(A):
337	  match x:
338	    case (j, Done{u}):
339	      (Runner{j, step(~A, s, op), p, bs, c}, Done{u})
340	    case (j, Fail{e}):
341	      (Runner{j, s, p, bs, c}, Fail{e})
342	
343	# Appends op, and applies it once the append is durable. After a failed append the state is as it was,
344	# and the journal refuses every later append until the runner is reopened.
345	def commit(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +op: Op<A>) -> IO(Ran(A)):
346	  Runner{j, s, p, bs, c} = r
347	  do IO<Ran(A)>:
348	    x : J.Journal & Out(Unit) <- J.append(j, encode(~A, ~penc, op))
349	    return commit.done(~A, x, s, op, p, bs, c)
350	
351	# A grant of this runner's pool always fits it, so Kept does not happen.
352	def release.got(-A: Data, x: Res.Release, j: J.Journal, s: State<A>, bs: Map<&2, Retry.Breaker>, c: Config) -> Runner<A>:
353	  match x:
354	    case Res.Released{p}:
355	      Runner{j, s, p, bs, c}
356	    case Res.Kept{p, g}:
357	      Runner{j, s, p, bs, c}
358	
359	def release(-A: Data, r: Runner<A>, g: Res.Grant) -> Runner<A>:
360	  Runner{j, s, p, bs, c} = r
361	  release.got(A, Res.release(p, g), j, s, bs, c)
362	
363	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)):
364	  match y:
365	    case (r, Fail{e}):
366	      IO.pure(Claimed(A), (release(A, r, g), fs, Fail{e}))
367	    case (r, Done{u}):
368	      next(release(A, r, g), fs)
369	
370	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,
371	    j: J.Journal, s: State<A>, p: Res.Pool, bs: Map<&2, Retry.Breaker>, c: Config, +id: Nat, d: Nat, +x: A, g: Res.Grant,
372	    fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
373	  match a:
374	    case (+br, b, Retry.GateGo{}):
375	      do IO<Claimed(A)>:
376	        ch : Chan(Retry.Outcome) <- IO.fork(Retry.Outcome, work(x))
377	        next(Runner{j, s, p, Map.set(&2, Retry.Breaker, bs, key, br), c}, Con{Flight{Q.Ticket{id, d, x}, g, b, key, ch}, fs})
378	    case (+br, b, Retry.GateNo{why}):
379	      do IO<Claimed(A)>:
380	        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))
381	        claim.held(A, y, g, fs, next)
382	
383	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,
384	    j: J.Journal, +s: State<A>, p: Res.Pool, +c: Config, +id: Nat, d: Nat, +x: A, g: Res.Grant,
385	    fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
386	  (bs, br) = got
387	  Config{pol, circ, idle} = c
388	  claim.go(~A, ~penc, ~work, 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)
389	
390	# The selected ticket asks its target's breaker and its own budget for an attempt.
391	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,
392	    fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
393	  Runner{j, s, p, bs, +c} = r
394	  Config{pol, circ, idle} = c
395	  Q.Ticket{id, d, +x} = t
396	  +key = target(x)
397	  claim.gate(~A, ~penc, ~work, now, Map.get(Retry.Breaker, Retry.Breaker.new(circ), bs, key), key, j, s, p, c, id, d, x, g, fs, next)
398	
399	def claim.selected(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +now: Nat, y: Ran(A),
400	    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)):
401	  match y:
402	    case (r, Fail{e}):
403	      IO.pure(Claimed(A), (release(A, r, g), fs, Fail{e}))
404	    case (r, Done{u}):
405	      claim.admit(~A, ~penc, ~target, ~work, now, r, t, g, fs, next)
406	
407	def claim.pick(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +now: Nat, k: Pick<A>, j: J.Journal, s: State<A>,
408	    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)):
409	  match k:
410	    case Idle{p}:
411	      IO.pure(Claimed(A), (Runner{j, s, p, bs, c}, fs, Done{Unit{}}))
412	    case Picked{p, t, g}:
413	      do IO<Claimed(A)>:
414	        y : Ran(A) <- commit(~A, ~penc, Runner{j, s, p, bs, c}, Select{now})
415	        claim.selected(~A, ~penc, ~target, ~work, now, y, t, g, fs, next)
416	
417	# Starts an attempt for each due ticket while the pool has a free unit. fuel bounds the tickets taken.
418	def claim(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), fuel: Nat, +now: Nat, r: Runner<A>, fs: List<&1, Flight<A>>) ->
419	    IO(Claimed(A)):
420	  match fuel:
421	    case 0n:
422	      IO.pure(Claimed(A), (r, fs, Done{Unit{}}))
423	    case 1n+n:
424	      Runner{j, +s, p, bs, c} = r
425	      claim.pick(~A, ~penc, ~target, ~work, now, pick(~A, p, s, now), j, s, bs, c, fs, (r2 => fs2 => claim(~A, ~penc, ~target, ~work, n, now, r2, fs2)))
426	
427	def settle.done(-A: Data, y: Ran(A), g: Res.Grant, err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Ran(A))) -> IO(Ran(A)):
428	  (r, res) = y
429	  next(release(A, r, g), first(err, res))
430	
431	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>,
432	    j: J.Journal, s: State<A>, p: Res.Pool, c: Config, g: Res.Grant, err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Ran(A))) -> IO(Ran(A)):
433	  (bs, br) = got
434	  do IO<Ran(A)>:
435	    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)
436	    settle.done(A, y, g, err, next)
437	
438	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,
439	    err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Ran(A))) -> IO(Ran(A)):
440	  Runner{j, +s, p, bs, +c} = r
441	  Config{pol, circ, idle} = c
442	  settle.br(~A, ~penc, Retry.failed(o), key, now, Map.get(Retry.Breaker, Retry.Breaker.new(circ), bs, key),
443	    after(A, o, pol, b, id, fails(tries.get(tries(A, s), id)), now, rnd), j, s, p, c, g, err, next)
444	
445	def random(r: Out(U32)) -> Nat:
446	  match r:
447	    case Done{x}:
448	      U32.to_nat(x)
449	    case Fail{e}:
450	      0n
451	
452	# Waits for the attempt, tells its target's breaker, appends Complete or Requeue, and returns its unit to the pool.
453	def settle.one(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, f: Flight<A>, err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Ran(A))) -> IO(Ran(A)):
454	  Flight{Q.Ticket{id, d, x}, g, b, key, ch} = f
455	  do IO<Ran(A)>:
456	    o : Retry.Outcome <- IO.join(Retry.Outcome, ch)
457	    now : Nat <- clock()
458	    rnd : Out(U32) <- IO.random_u32()
459	    settle.op(~A, ~penc, o, b, id, key, now, random(rnd), r, g, err, next)
460	
461	def settle(~A: Data, ~penc: A -> Bytes.Bytes, fs: List<&1, Flight<A>>, r: Runner<A>, err: Out(Unit)) -> IO(Ran(A)):
462	  match fs:
463	    case Nil{}:
464	      IO.pure(Ran(A), (r, err))
465	    case Con{f, rest}:
466	      settle.one(~A, ~penc, r, f, err, (r2 => e2 => settle(~A, ~penc, rest, r2, e2)))
467	
468	def empty.heap(-A: Data, h: H.Heap<Q.Ticket<A>>) -> Bool:
469	  match h:
470	    case H.HLeaf{}:
471	      True{}
472	    case H.HNode{n, t, l, r}:
473	      False{}
474	
475	def empty.list(-A: Data, xs: List<&2, Q.Ticket<A>>) -> Bool:
476	  match xs:
477	    case Nil{}:
478	      True{}
479	    case Con{t, rest}:
480	      False{}
481	
482	# No ticket is pending or running.
483	def idle(-A: Data, s: State<A>) -> Bool:
484	  State{Q.State{n, h, run}, ts} = s
485	  Bool.and(empty.heap(A, h), empty.list(A, run))
486	
487	def pending(-A: Data, s: State<A>) -> Nat:
488	  State{Q.State{n, h, run}, ts} = s
489	  H.Heap.size(Q.Ticket<A>, h)
490	
491	def tick.ran(e: Out(Unit), b: Bool) -> Out(Bool):
492	  match e:
493	    case Fail{x}:
494	      Fail{x}
495	    case Done{u}:
496	      Done{b}
497	
498	def tick.done(-A: Data, y: Ran(A)) -> Runner<A> & Out(Bool):
499	  (r, e) = y
500	  (r, tick.ran(e, True{}))
501	
502	def tick.settle(~A: Data, ~penc: A -> Bytes.Bytes, c: Claimed(A)) -> IO(Runner<A> & Out(Bool)):
503	  match c:
504	    case (r, Nil{}, err):
505	      IO.pure(Runner<A> & Out(Bool), (r, tick.ran(err, False{})))
506	    case (r, Con{f, fs}, err):
507	      do IO<Runner<A> & Out(Bool)>:
508	        y : Ran(A) <- settle(~A, ~penc, Con{f, fs}, r, err)
509	        return tick.done(A, y)
510	
511	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)):
512	  Runner{j, +s, p, bs, c} = r
513	  do IO<Runner<A> & Out(Bool)>:
514	    k : Claimed(A) <- claim(~A, ~penc, ~target, ~work, 1n+pending(A, s), now, Runner{j, s, p, bs, c}, Nil{})
515	    tick.settle(~A, ~penc, k)
516	
517	# One round: start an attempt for each due ticket the pool admits, wait for them all, and record each outcome.
518	# Done{True} when an attempt ran. work runs on its own; target names the breaker that guards it.
519	def tick(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), r: Runner<A>) -> IO(Runner<A> & Out(Bool)):
520	  do IO<Runner<A> & Out(Bool)>:
521	    now : Nat <- clock()
522	    tick.at(~A, ~penc, ~target, ~work, now, r)
523	
524	def drain.ticked(-A: Data, +idle: U32, t: Runner<A> & Out(Bool), next: Runner<A> -> IO(Ran(A))) -> IO(Ran(A)):
525	  match t:
526	    case (r, Fail{e}):
527	      IO.pure(Ran(A), (r, Fail{e}))
528	    case (r, Done{True{}}):
529	      next(r)
530	    case (r, Done{False{}}):
531	      do IO<Ran(A)>:
532	        IO.sleep(idle)
533	        next(r)
534	
535	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)):
536	  match done:
537	    case True{}:
538	      IO.pure(Ran(A), (r, Done{Unit{}}))
539	    case False{}:
540	      Runner{j, s, p, bs, +c} = r
541	      Config{pol, circ, +idle} = c
542	      do IO<Ran(A)>:
543	        t : Runner<A> & Out(Bool) <- tick(~A, ~penc, ~target, ~work, Runner{j, s, p, bs, c})
544	        drain.ticked(A, idle, t, next)
545	
546	# Ticks until no ticket is pending or running. It waits idle ms after a round that started no attempt.
547	# Fails when fuel rounds leave work.
548	def drain(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), fuel: Nat, r: Runner<A>) -> IO(Ran(A)):
549	  match fuel:
550	    case 0n:
551	      Runner{j, +s, p, bs, c} = r
552	      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")})))
553	    case 1n+n:
554	      Runner{j, +s, p, bs, c} = r
555	      drain.if(~A, ~penc, ~target, ~work, idle(A, s), Runner{j, s, p, bs, c}, (r2 => drain(~A, ~penc, ~target, ~work, n, r2)))
556	
557	def orphans.next(-A: Data, y: Ran(A), err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Ran(A))) -> IO(Ran(A)):
558	  (r, e) = y
559	  next(r, first(err, e))
560	
561	# A running ticket at open lost its attempt in a crash: it is requeued at now, as a failed attempt.
562	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)):
563	  match ts:
564	    case Nil{}:
565	      IO.pure(Ran(A), (r, err))
566	    case Con{Q.Ticket{id, d, x}, rest}:
567	      do IO<Ran(A)>:
568	        y : Ran(A) <- commit(~A, ~penc, r, Requeue{id, now})
569	        orphans.next(A, y, err, (r2 => e2 => orphans(~A, ~penc, rest, now, r2, e2)))
570	
571	def open.done(-A: Data, y: Ran(A)) -> IO(Out(Runner<A>)):
572	  match y:
573	    case (Runner{j, s, p, bs, c}, Fail{e}):
574	      do IO<Out(Runner<A>)>:
575	        J.close(j)
576	        return Fail{e}
577	    case (r, Done{u}):
578	      IO.pure(Out(Runner<A>), Done{r})
579	
580	def open.got(~A: Data, ~penc: A -> Bytes.Bytes, x: Out(J.Journal & Maybe<&2, State<A>>), c: Config, +capacity: Nat) -> IO(Out(Runner<A>)):
581	  match x:
582	    case Fail{e}:
583	      IO.pure(Out(Runner<A>), Fail{e})
584	    case Done{(j, None{})}:
585	      do IO<Out(Runner<A>)>:
586	        J.close(j)
587	        return Fail{(22, "runner: a journal record is not a runner op")}
588	    case Done{(j, Some{+s})}:
589	      do IO<Out(Runner<A>)>:
590	        now : Nat <- clock()
591	        y : Ran(A) <- orphans(~A, ~penc, Q.running(A, queue(A, s)), now, Runner{j, s, Res.new(0n, capacity), Map.new(&2, Retry.Breaker), c}, Done{Unit{}})
592	        open.done(A, y)
593	
594	def replay.step(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, acc: Maybe<&2, State<A>>, b: Bytes.Bytes) -> Maybe<&2, State<A>>:
595	  apply(~A, ~Bytes.Bytes, ~(y => decode(~A, ~pdec, y)), acc, b)
596	
597	# Opens the runner at path: replays its journal into the state, then requeues each ticket that was running.
598	# capacity bounds the attempts that run at once. max caps a record's octets; limit caps the journal file.
599	# Fails when the journal fails to open, or a committed record is not an op.
600	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>)):
601	  do IO<Out(Runner<A>)>:
602	    x : Out(J.Journal & Maybe<&2, State<A>>) <- J.open(~Maybe<&2, State<A>>, ~(acc => b => replay.step(~A, ~pdec, acc, b)), path, max, limit, Some{new(A)})
603	    open.got(~A, ~penc, x, c, capacity)
604	
605	def enqueue.done(-A: Data, +id: Nat, y: Ran(A)) -> Runner<A> & Out(Nat):
606	  match y:
607	    case (r, Done{u}):
608	      (r, Done{id})
609	    case (r, Fail{e}):
610	      (r, Fail{e})
611	
612	# Adds a job that is due at deadline (wall-clock ms) and answers its ticket id once the record is durable.
613	def enqueue(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +deadline: Nat, x: A) -> IO(Runner<A> & Out(Nat)):
614	  Runner{j, +s, p, bs, c} = r
615	  do IO<Runner<A> & Out(Nat)>:
616	    y : Ran(A) <- commit(~A, ~penc, Runner{j, s, p, bs, c}, Enqueue{deadline, x})
617	    return enqueue.done(A, Q.next_id(A, queue(A, s)), y)
618	
619	# Cancels ticket id. A running attempt finishes, but its ticket is never selected again.
620	def cancel(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +id: Nat) -> IO(Ran(A)):
621	  commit(~A, ~penc, r, Cancel{id})
622	
623	# The runner back, and a copy of its state.
624	def state(-A: Data, r: Runner<A>) -> Runner<A> & State<A>:
625	  Runner{j, +s, p, bs, c} = r
626	  (Runner{j, s, p, bs, c}, s)
627	
628	def close(-A: Data, r: Runner<A>) -> IO(Unit):
629	  Runner{j, s, p, bs, c} = r
630	  J.close(j)