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