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