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