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