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