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