runner.bend
bend-kit-runner@0.2.0.0 · 630 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.2.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.2.0/heap.bend as H
10import bend-kit-collections@0.1.2.0/omap.bend as O
11
12# One queue transition. Each one is one journal record, appended before the state changes.
13# Select: the queue's select at now. Replay runs it at the same now, so it takes the same ticket.
14# Requeue: the attempt failed; it counts one failure. Defer: the breaker held the ticket back; no attempt ran.
15type Op<-A: Data> is Data:
16 Enqueue{deadline: Nat, payload: A}
17 Select{now: Nat}
18 Complete{id: Nat}
19 Cancel{id: Nat}
20 Requeue{id: Nat, deadline: Nat}
21 Defer{id: Nat, deadline: Nat}
22
23# The failed attempts of a ticket, and its first deadline, from which the policy deadline counts.
24type Try is Data:
25 Try{fails: Nat, start: Nat}
26
27# A Try for each ticket that is not yet complete or cancelled, by ticket id.
28def Tries() -> Data:
29 O.OMap<Nat, Try>
30
31# The durable state: the queue and the tries.
32type State<-A: Data> is Data:
33 State{queue: Q.State<A>, tries: Tries()}
34
35def new(-A: Data) -> State<A>:
36 State{Q.new(A), O.OMap.new(Nat, Try)}
37
38def queue(-A: Data, s: State<A>) -> Q.State<A>:
39 State{q, ts} = s
40 q
41
42def tries(-A: Data, s: State<A>) -> Tries():
43 State{q, ts} = s
44 ts
45
46# ---- tries
47
48def tries.drop(ts: Tries(), +i: Nat) -> Tries():
49 O.OMap.del(~Nat, ~Try, ~Nat.cmp, ts, i)
50
51def tries.get(ts: Tries(), +i: Nat) -> Maybe<&2, Try>:
52 O.OMap.get(~Nat, ~Try, ~Nat.cmp, ts, i)
53
54def tries.bump.got(m: Maybe<&2, Try>, ts: Tries(), +i: Nat) -> Tries():
55 match m:
56 case None{}:
57 ts
58 case Some{Try{f, st}}:
59 O.OMap.put(~Nat, ~Try, ~Nat.cmp, ts, i, Try{1n+f, st})
60
61def tries.bump(+ts: Tries(), +i: Nat) -> Tries():
62 tries.bump.got(tries.get(ts, i), ts, i)
63
64# ---- the transition
65
66def step.enqueued(-A: Data, r: Q.State<A> & Nat, ts: Tries(), d: Nat) -> State<A>:
67 (q, i) = r
68 State{q, O.OMap.put(~Nat, ~Try, ~Nat.cmp, ts, i, Try{0n, d})}
69
70def step.selected(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, ts: Tries()) -> State<A>:
71 (q, t) = r
72 State{q, ts}
73
74def step.requeued(-A: Data, r: Q.State<A> & Bool, ts: Tries()) -> State<A>:
75 (q, b) = r
76 State{q, ts}
77
78# Complete takes the ticket from the running list only: O(running), not O(pending).
79def step.completed(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, ts: Tries()) -> State<A>:
80 (q, t) = r
81 State{q, ts}
82
83# The state after op. Live and replay both run it, so they agree.
84def step(~A: Data, s: State<A>, op: Op<A>) -> State<A>:
85 match s op:
86 case State{q, ts} Enqueue{+d, x}:
87 step.enqueued(A, Q.enqueue(~A, d, x, q), ts, d)
88 case State{q, ts} Select{now}:
89 step.selected(A, Q.select(~A, now, q), ts)
90 case State{q, ts} Complete{+i}:
91 step.completed(A, Q.complete(A, i, q), tries.drop(ts, i))
92 case State{q, ts} Cancel{+i}:
93 State{Q.cancel(~A, i, q), tries.drop(ts, i)}
94 case State{q, ts} Requeue{+i, d}:
95 step.requeued(A, Q.requeue(~A, i, d, q), tries.bump(ts, i))
96 case State{q, ts} Defer{i, d}:
97 step.requeued(A, Q.requeue(~A, i, d, q), ts)
98
99def apply.op(~A: Data, m: Maybe<&2, Op<A>>, s: State<A>) -> Maybe<&2, State<A>>:
100 match m:
101 case None{}:
102 None{}
103 case Some{op}:
104 Some{step(~A, s, op)}
105
106# Folds one record into the replayed state. A record that does not decode ends the replay with None.
107def apply(~A: Data, ~R: Type, ~dec: R -> Maybe<&2, Op<A>>, acc: Maybe<&2, State<A>>, r: R) -> Maybe<&2, State<A>>:
108 match acc:
109 case None{}:
110 None{}
111 case Some{s}:
112 apply.op(~A, dec(r), s)
113
114# ---- pool
115
116# Picked: a due ticket and one unit of the pool for it. Idle: no due ticket, or no free unit; the pool is unchanged.
117type Pick<-A: Data> is Type:
118 Picked{pool: Res.Pool, ticket: Q.Ticket<A>, grant: Res.Grant}
119 Idle{pool: Res.Pool}
120
121def pick.took(-A: Data, r: Res.Reserve, t: Q.Ticket<A>) -> Pick<A>:
122 match r:
123 case Res.Took{p, g}:
124 Picked{p, t, g}
125 case Res.Refused{p}:
126 Idle{p}
127
128def pick.due(-A: Data, m: Maybe<&2, Q.Ticket<A>>, p: Res.Pool) -> Pick<A>:
129 match m:
130 case None{}:
131 Idle{p}
132 case Some{t}:
133 pick.took(A, Res.reserve(p, 1n), t)
134
135def pick.sel(-A: Data, r: Q.State<A> & Maybe<&2, Q.Ticket<A>>, p: Res.Pool) -> Pick<A>:
136 (q, m) = r
137 pick.due(A, m, p)
138
139# The ticket that Select{now} takes, with one unit of p, when there is a ticket due and a unit free.
140def pick(~A: Data, p: Res.Pool, s: State<A>, now: Nat) -> Pick<A>:
141 State{q, ts} = s
142 pick.sel(A, Q.select(~A, now, q), p)
143
144# ---- retry
145
146# The budget left for ticket i: the policy's attempts less its failures, until its start plus the policy deadline.
147def budget.of(+attempts: Nat, t: Maybe<&2, Try>, d: Maybe<&2, Nat>) -> Retry.Budget:
148 match t d:
149 case None{} m:
150 Retry.Budget{attempts, None{}}
151 case Some{Try{f, st}} None{}:
152 Retry.Budget{Nat.sub(attempts, f), None{}}
153 case Some{Try{f, +st}} Some{ms}:
154 Retry.Budget{Nat.sub(attempts, f), Some{Nat.add(st, ms)}}
155
156def budget(p: Retry.Policy, t: Maybe<&2, Try>) -> Retry.Budget:
157 Retry.Policy{a, base, cap, d} = p
158 budget.of(a, t, d)
159
160# The failures so far: the index of the next retry.
161def fails(t: Maybe<&2, Try>) -> Nat:
162 match t:
163 case None{}:
164 0n
165 case Some{Try{f, st}}:
166 f
167
168def after.plan(-A: Data, pl: Retry.Plan, +id: Nat, +now: Nat) -> Op<A>:
169 match pl:
170 case Retry.PlanStop{}:
171 Complete{id}
172 case Retry.PlanWait{w}:
173 Requeue{id, Nat.add(now, w)}
174
175# The record after an attempt of ticket id ends at now with o: requeue at now plus retry's wait, or complete.
176def after(-A: Data, o: Retry.Outcome, p: Retry.Policy, b: Retry.Budget, +id: Nat, +n: Nat, +now: Nat, +r: Nat) -> Op<A>:
177 after.plan(A, Retry.plan(o, p, b, n, now, r), id, now)
178
179# When an open breaker lets attempts start again.
180def until(br: Retry.Breaker, +now: Nat) -> Nat:
181 match br:
182 case Retry.BreakerOn{c, Retry.CircuitOpen{u}}:
183 u
184 case Retry.BreakerOn{c, Retry.CircuitClosed{fs}}:
185 Nat.add(now, Retry.Circuit.cooldown(c))
186 case Retry.BreakerOn{c, Retry.CircuitHalf{k}}:
187 Nat.add(now, Retry.Circuit.cooldown(c))
188 case Retry.BreakerOff{}:
189 now
190
191# The record for a ticket that admit refused: Defer to when the breaker may admit, or Complete when its budget is spent or late.
192def held(-A: Data, why: Retry.Deny, br: Retry.Breaker, +id: Nat, +now: Nat) -> Op<A>:
193 match why:
194 case Retry.DenyOpen{}:
195 Defer{id, until(br, now)}
196 case Retry.DenyLate{}:
197 Complete{id}
198 case Retry.DenySpent{}:
199 Complete{id}
200
201# ---- the record codec: a u32le tag, then u64le fields; an Enqueue then holds the payload.
202
203def W() -> Nat:
204 Nat.mul(65536n, 65536n)
205
206# Nat fields hold values under 2^64.
207def put.u64(b: Bytes.Bytes, +at: U32, +n: Nat) -> Bytes.Bytes:
208 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())))
209
210def head(+tag: U32, +a: Nat, +c: Nat) -> Bytes.Bytes:
211 put.u64(put.u64(Bytes.set.u32le(Bytes.new(20), 0, tag), 4, a), 12, c)
212
213def encode(~A: Data, ~penc: A -> Bytes.Bytes, op: Op<A>) -> Bytes.Bytes:
214 match op:
215 case Enqueue{+d, x}:
216 Bytes.append(put.u64(Bytes.set.u32le(Bytes.new(12), 0, 1), 4, d), penc(x))
217 case Select{+now}:
218 head(2, now, 0n)
219 case Complete{+i}:
220 head(3, i, 0n)
221 case Cancel{+i}:
222 head(4, i, 0n)
223 case Requeue{+i, +d}:
224 head(5, i, d)
225 case Defer{+i, +d}:
226 head(6, i, d)
227
228def get.field(m: Maybe<&2, U32>) -> U32:
229 match m:
230 case None{}:
231 0
232 case Some{x}:
233 x
234
235def get.u32(r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & U32:
236 (b, m) = r
237 (b, get.field(m))
238
239def get.hi(+lo: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Nat:
240 (b, m) = r
241 (b, Nat.add(U32.to_nat(lo), Nat.mul(U32.to_nat(get.field(m)), W())))
242
243def get.lo(+at: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Nat:
244 (b, m) = r
245 +lo = get.field(m)
246 get.hi(lo, Bytes.get.u32le(b, (at + 4 : U32)))
247
248def get.u64(b: Bytes.Bytes, +at: U32) -> Bytes.Bytes & Nat:
249 get.lo(at, Bytes.get.u32le(b, at))
250
251def decode.small(-A: Data, ok: Bool, +tag: U32, +a: Nat, +c: Nat) -> Maybe<&2, Op<A>>:
252 match ok:
253 case False{}:
254 None{}
255 case True{}:
256 Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 2), Some{Select{a}},
257 Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 3), Some{Complete{a}},
258 Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 4), Some{Cancel{a}},
259 Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 5), Some{Requeue{a, c}},
260 Bool.pick(Maybe<&2, Op<A>>, U32.is_eq(tag, 6), Some{Defer{a, c}}, None{})))))
261
262def decode.payload(-A: Data, m: Maybe<&2, A>, +d: Nat) -> Maybe<&2, Op<A>>:
263 match m:
264 case None{}:
265 None{}
266 case Some{x}:
267 Some{Enqueue{d, x}}
268
269def decode.c(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +tag: U32, +len: U32, +a: Nat, r: Bytes.Bytes & Nat) -> Maybe<&2, Op<A>>:
270 (b, +c) = r
271 decode.small(A, U32.is_eq(len, 20), tag, a, c)
272
273def 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>>:
274 match enq:
275 case True{}:
276 decode.payload(A, pdec(Bytes.snd(Bytes.slice(b, 12, (len - 12 : U32)))), a)
277 case False{}:
278 decode.c(~A, ~pdec, tag, len, a, get.u64(b, 12))
279
280def decode.a(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +tag: U32, +len: U32, r: Bytes.Bytes & Nat) -> Maybe<&2, Op<A>>:
281 (b, +a) = r
282 decode.kind(~A, ~pdec, U32.is_eq(tag, 1), tag, len, a, b)
283
284def decode.tag(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, +len: U32, r: Bytes.Bytes & U32) -> Maybe<&2, Op<A>>:
285 (b, +tag) = r
286 decode.a(~A, ~pdec, tag, len, get.u64(b, 4))
287
288def decode.len(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, ok: Bool, +len: U32, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
289 match ok:
290 case False{}:
291 None{}
292 case True{}:
293 decode.tag(~A, ~pdec, len, get.u32(Bytes.get.u32le(b, 0)))
294
295# The op of a record, or None when it is not one that encode writes.
296def decode(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, b: Bytes.Bytes) -> Maybe<&2, Op<A>>:
297 Bytes.Bytes{+len, buf} = b
298 decode.len(~A, ~pdec, U32.is_ge(len, 12), len, Bytes.Bytes{len, buf})
299
300# ---- the runner
301
302# policy: retry's policy for each ticket. circuit: the breaker of each target. idle: ms that drain waits when no attempt started.
303type Config is Data:
304 Config{policy: Retry.Policy, circuit: Retry.Circuit, idle: U32}
305
306# An open runner. Its journal holds every op that made state; its pool bounds the attempts that run at once.
307type Runner<-A: Data> is Type:
308 Runner{journal: J.Journal, state: State<A>, pool: Res.Pool, breakers: Map<&2, Retry.Breaker>, config: Config}
309
310# An attempt that runs: its unit of the pool, the budget that admit left, its target, and the channel of its outcome.
311type Flight<-A: Data> is Type:
312 Flight{ticket: Q.Ticket<A>, grant: Res.Grant, budget: Retry.Budget, target: String, chan: Chan(Retry.Outcome)}
313
314def Out(-X: Type) -> Type:
315 Result<&1, &1, U32 & String, X>
316
317def Claimed(-A: Data) -> Type:
318 Runner<A> & List<&1, Flight<A>> & Out(Unit)
319
320def Ran(-A: Data) -> Type:
321 Runner<A> & Out(Unit)
322
323# Wall-clock ms since the Unix epoch. Deadlines must keep their meaning across a restart.
324def clock() -> IO(Nat):
325 do IO<Nat>:
326 t : Time.Instant <- Time.now()
327 return Time.Duration.to_ms_nat(Time.Instant.since_epoch(t))
328
329def first(a: Out(Unit), b: Out(Unit)) -> Out(Unit):
330 match a:
331 case Fail{e}:
332 Fail{e}
333 case Done{u}:
334 b
335
336def commit.done(~A: Data, x: J.Journal & Out(Unit), s: State<A>, +op: Op<A>, p: Res.Pool, bs: Map<&2, Retry.Breaker>, c: Config) -> Ran(A):
337 match x:
338 case (j, Done{u}):
339 (Runner{j, step(~A, s, op), p, bs, c}, Done{u})
340 case (j, Fail{e}):
341 (Runner{j, s, p, bs, c}, Fail{e})
342
343# Appends op, and applies it once the append is durable. After a failed append the state is as it was,
344# and the journal refuses every later append until the runner is reopened.
345def commit(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +op: Op<A>) -> IO(Ran(A)):
346 Runner{j, s, p, bs, c} = r
347 do IO<Ran(A)>:
348 x : J.Journal & Out(Unit) <- J.append(j, encode(~A, ~penc, op))
349 return commit.done(~A, x, s, op, p, bs, c)
350
351# A grant of this runner's pool always fits it, so Kept does not happen.
352def release.got(-A: Data, x: Res.Release, j: J.Journal, s: State<A>, bs: Map<&2, Retry.Breaker>, c: Config) -> Runner<A>:
353 match x:
354 case Res.Released{p}:
355 Runner{j, s, p, bs, c}
356 case Res.Kept{p, g}:
357 Runner{j, s, p, bs, c}
358
359def release(-A: Data, r: Runner<A>, g: Res.Grant) -> Runner<A>:
360 Runner{j, s, p, bs, c} = r
361 release.got(A, Res.release(p, g), j, s, bs, c)
362
363def 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)):
364 match y:
365 case (r, Fail{e}):
366 IO.pure(Claimed(A), (release(A, r, g), fs, Fail{e}))
367 case (r, Done{u}):
368 next(release(A, r, g), fs)
369
370def claim.go(~A: Data, ~penc: A -> Bytes.Bytes, ~work: A -> IO(Retry.Outcome), +now: Nat, a: Retry.Breaker & Retry.Budget & Retry.Gate, +key: String,
371 j: J.Journal, s: State<A>, p: Res.Pool, bs: Map<&2, Retry.Breaker>, c: Config, +id: Nat, d: Nat, +x: A, g: Res.Grant,
372 fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
373 match a:
374 case (+br, b, Retry.GateGo{}):
375 do IO<Claimed(A)>:
376 ch : Chan(Retry.Outcome) <- IO.fork(Retry.Outcome, work(x))
377 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})
378 case (+br, b, Retry.GateNo{why}):
379 do IO<Claimed(A)>:
380 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))
381 claim.held(A, y, g, fs, next)
382
383def claim.gate(~A: Data, ~penc: A -> Bytes.Bytes, ~work: A -> IO(Retry.Outcome), +now: Nat, got: Map<&2, Retry.Breaker> & Retry.Breaker, +key: String,
384 j: J.Journal, +s: State<A>, p: Res.Pool, +c: Config, +id: Nat, d: Nat, +x: A, g: Res.Grant,
385 fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
386 (bs, br) = got
387 Config{pol, circ, idle} = c
388 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)
389
390# The selected ticket asks its target's breaker and its own budget for an attempt.
391def 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,
392 fs: List<&1, Flight<A>>, next: Runner<A> -> List<&1, Flight<A>> -> IO(Claimed(A))) -> IO(Claimed(A)):
393 Runner{j, s, p, bs, +c} = r
394 Config{pol, circ, idle} = c
395 Q.Ticket{id, d, +x} = t
396 +key = target(x)
397 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)
398
399def claim.selected(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +now: Nat, y: Ran(A),
400 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)):
401 match y:
402 case (r, Fail{e}):
403 IO.pure(Claimed(A), (release(A, r, g), fs, Fail{e}))
404 case (r, Done{u}):
405 claim.admit(~A, ~penc, ~target, ~work, now, r, t, g, fs, next)
406
407def claim.pick(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), +now: Nat, k: Pick<A>, j: J.Journal, s: State<A>,
408 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)):
409 match k:
410 case Idle{p}:
411 IO.pure(Claimed(A), (Runner{j, s, p, bs, c}, fs, Done{Unit{}}))
412 case Picked{p, t, g}:
413 do IO<Claimed(A)>:
414 y : Ran(A) <- commit(~A, ~penc, Runner{j, s, p, bs, c}, Select{now})
415 claim.selected(~A, ~penc, ~target, ~work, now, y, t, g, fs, next)
416
417# Starts an attempt for each due ticket while the pool has a free unit. fuel bounds the tickets taken.
418def 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>>) ->
419 IO(Claimed(A)):
420 match fuel:
421 case 0n:
422 IO.pure(Claimed(A), (r, fs, Done{Unit{}}))
423 case 1n+n:
424 Runner{j, +s, p, bs, c} = r
425 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)))
426
427def settle.done(-A: Data, y: Ran(A), g: Res.Grant, err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Ran(A))) -> IO(Ran(A)):
428 (r, res) = y
429 next(release(A, r, g), first(err, res))
430
431def settle.br(~A: Data, ~penc: A -> Bytes.Bytes, +failed: Bool, +key: String, +now: Nat, got: Map<&2, Retry.Breaker> & Retry.Breaker, +op: Op<A>,
432 j: J.Journal, 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)):
433 (bs, br) = got
434 do IO<Ran(A)>:
435 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)
436 settle.done(A, y, g, err, next)
437
438def 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,
439 err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Ran(A))) -> IO(Ran(A)):
440 Runner{j, +s, p, bs, +c} = r
441 Config{pol, circ, idle} = c
442 settle.br(~A, ~penc, Retry.failed(o), key, now, Map.get(Retry.Breaker, Retry.Breaker.new(circ), bs, key),
443 after(A, o, pol, b, id, fails(tries.get(tries(A, s), id)), now, rnd), j, s, p, c, g, err, next)
444
445def random(r: Out(U32)) -> Nat:
446 match r:
447 case Done{x}:
448 U32.to_nat(x)
449 case Fail{e}:
450 0n
451
452# Waits for the attempt, tells its target's breaker, appends Complete or Requeue, and returns its unit to the pool.
453def 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)):
454 Flight{Q.Ticket{id, d, x}, g, b, key, ch} = f
455 do IO<Ran(A)>:
456 o : Retry.Outcome <- IO.join(Retry.Outcome, ch)
457 now : Nat <- clock()
458 rnd : Out(U32) <- IO.random_u32()
459 settle.op(~A, ~penc, o, b, id, key, now, random(rnd), r, g, err, next)
460
461def settle(~A: Data, ~penc: A -> Bytes.Bytes, fs: List<&1, Flight<A>>, r: Runner<A>, err: Out(Unit)) -> IO(Ran(A)):
462 match fs:
463 case Nil{}:
464 IO.pure(Ran(A), (r, err))
465 case Con{f, rest}:
466 settle.one(~A, ~penc, r, f, err, (r2 => e2 => settle(~A, ~penc, rest, r2, e2)))
467
468def empty.heap(-A: Data, h: H.Heap<Q.Ticket<A>>) -> Bool:
469 match h:
470 case H.HLeaf{}:
471 True{}
472 case H.HNode{n, t, l, r}:
473 False{}
474
475def empty.list(-A: Data, xs: List<&2, Q.Ticket<A>>) -> Bool:
476 match xs:
477 case Nil{}:
478 True{}
479 case Con{t, rest}:
480 False{}
481
482# No ticket is pending or running.
483def idle(-A: Data, s: State<A>) -> Bool:
484 State{Q.State{n, h, run}, ts} = s
485 Bool.and(empty.heap(A, h), empty.list(A, run))
486
487def pending(-A: Data, s: State<A>) -> Nat:
488 State{Q.State{n, h, run}, ts} = s
489 H.Heap.size(Q.Ticket<A>, h)
490
491def tick.ran(e: Out(Unit), b: Bool) -> Out(Bool):
492 match e:
493 case Fail{x}:
494 Fail{x}
495 case Done{u}:
496 Done{b}
497
498def tick.done(-A: Data, y: Ran(A)) -> Runner<A> & Out(Bool):
499 (r, e) = y
500 (r, tick.ran(e, True{}))
501
502def tick.settle(~A: Data, ~penc: A -> Bytes.Bytes, c: Claimed(A)) -> IO(Runner<A> & Out(Bool)):
503 match c:
504 case (r, Nil{}, err):
505 IO.pure(Runner<A> & Out(Bool), (r, tick.ran(err, False{})))
506 case (r, Con{f, fs}, err):
507 do IO<Runner<A> & Out(Bool)>:
508 y : Ran(A) <- settle(~A, ~penc, Con{f, fs}, r, err)
509 return tick.done(A, y)
510
511def 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)):
512 Runner{j, +s, p, bs, c} = r
513 do IO<Runner<A> & Out(Bool)>:
514 k : Claimed(A) <- claim(~A, ~penc, ~target, ~work, 1n+pending(A, s), now, Runner{j, s, p, bs, c}, Nil{})
515 tick.settle(~A, ~penc, k)
516
517# One round: start an attempt for each due ticket the pool admits, wait for them all, and record each outcome.
518# Done{True} when an attempt ran. work runs on its own; target names the breaker that guards it.
519def tick(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), r: Runner<A>) -> IO(Runner<A> & Out(Bool)):
520 do IO<Runner<A> & Out(Bool)>:
521 now : Nat <- clock()
522 tick.at(~A, ~penc, ~target, ~work, now, r)
523
524def drain.ticked(-A: Data, +idle: U32, t: Runner<A> & Out(Bool), next: Runner<A> -> IO(Ran(A))) -> IO(Ran(A)):
525 match t:
526 case (r, Fail{e}):
527 IO.pure(Ran(A), (r, Fail{e}))
528 case (r, Done{True{}}):
529 next(r)
530 case (r, Done{False{}}):
531 do IO<Ran(A)>:
532 IO.sleep(idle)
533 next(r)
534
535def 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)):
536 match done:
537 case True{}:
538 IO.pure(Ran(A), (r, Done{Unit{}}))
539 case False{}:
540 Runner{j, s, p, bs, +c} = r
541 Config{pol, circ, +idle} = c
542 do IO<Ran(A)>:
543 t : Runner<A> & Out(Bool) <- tick(~A, ~penc, ~target, ~work, Runner{j, s, p, bs, c})
544 drain.ticked(A, idle, t, next)
545
546# Ticks until no ticket is pending or running. It waits idle ms after a round that started no attempt.
547# Fails when fuel rounds leave work.
548def drain(~A: Data, ~penc: A -> Bytes.Bytes, ~target: A -> String, ~work: A -> IO(Retry.Outcome), fuel: Nat, r: Runner<A>) -> IO(Ran(A)):
549 match fuel:
550 case 0n:
551 Runner{j, +s, p, bs, c} = r
552 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")})))
553 case 1n+n:
554 Runner{j, +s, p, bs, c} = r
555 drain.if(~A, ~penc, ~target, ~work, idle(A, s), Runner{j, s, p, bs, c}, (r2 => drain(~A, ~penc, ~target, ~work, n, r2)))
556
557def orphans.next(-A: Data, y: Ran(A), err: Out(Unit), next: Runner<A> -> Out(Unit) -> IO(Ran(A))) -> IO(Ran(A)):
558 (r, e) = y
559 next(r, first(err, e))
560
561# A running ticket at open lost its attempt in a crash: it is requeued at now, as a failed attempt.
562def orphans(~A: Data, ~penc: A -> Bytes.Bytes, ts: List<&2, Q.Ticket<A>>, +now: Nat, r: Runner<A>, err: Out(Unit)) -> IO(Ran(A)):
563 match ts:
564 case Nil{}:
565 IO.pure(Ran(A), (r, err))
566 case Con{Q.Ticket{id, d, x}, rest}:
567 do IO<Ran(A)>:
568 y : Ran(A) <- commit(~A, ~penc, r, Requeue{id, now})
569 orphans.next(A, y, err, (r2 => e2 => orphans(~A, ~penc, rest, now, r2, e2)))
570
571def open.done(-A: Data, y: Ran(A)) -> IO(Out(Runner<A>)):
572 match y:
573 case (Runner{j, s, p, bs, c}, Fail{e}):
574 do IO<Out(Runner<A>)>:
575 J.close(j)
576 return Fail{e}
577 case (r, Done{u}):
578 IO.pure(Out(Runner<A>), Done{r})
579
580def open.got(~A: Data, ~penc: A -> Bytes.Bytes, x: Out(J.Journal & Maybe<&2, State<A>>), c: Config, +capacity: Nat) -> IO(Out(Runner<A>)):
581 match x:
582 case Fail{e}:
583 IO.pure(Out(Runner<A>), Fail{e})
584 case Done{(j, None{})}:
585 do IO<Out(Runner<A>)>:
586 J.close(j)
587 return Fail{(22, "runner: a journal record is not a runner op")}
588 case Done{(j, Some{+s})}:
589 do IO<Out(Runner<A>)>:
590 now : Nat <- clock()
591 y : Ran(A) <- orphans(~A, ~penc, Q.running(A, queue(A, s)), now, Runner{j, s, Res.new(0n, capacity), Map.new(&2, Retry.Breaker), c}, Done{Unit{}})
592 open.done(A, y)
593
594def replay.step(~A: Data, ~pdec: Bytes.Bytes -> Maybe<&2, A>, acc: Maybe<&2, State<A>>, b: Bytes.Bytes) -> Maybe<&2, State<A>>:
595 apply(~A, ~Bytes.Bytes, ~(y => decode(~A, ~pdec, y)), acc, b)
596
597# Opens the runner at path: replays its journal into the state, then requeues each ticket that was running.
598# capacity bounds the attempts that run at once. max caps a record's octets; limit caps the journal file.
599# Fails when the journal fails to open, or a committed record is not an op.
600def 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>)):
601 do IO<Out(Runner<A>)>:
602 x : Out(J.Journal & Maybe<&2, State<A>>) <- J.open(~Maybe<&2, State<A>>, ~(acc => b => replay.step(~A, ~pdec, acc, b)), path, max, limit, Some{new(A)})
603 open.got(~A, ~penc, x, c, capacity)
604
605def enqueue.done(-A: Data, +id: Nat, y: Ran(A)) -> Runner<A> & Out(Nat):
606 match y:
607 case (r, Done{u}):
608 (r, Done{id})
609 case (r, Fail{e}):
610 (r, Fail{e})
611
612# Adds a job that is due at deadline (wall-clock ms) and answers its ticket id once the record is durable.
613def enqueue(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +deadline: Nat, x: A) -> IO(Runner<A> & Out(Nat)):
614 Runner{j, +s, p, bs, c} = r
615 do IO<Runner<A> & Out(Nat)>:
616 y : Ran(A) <- commit(~A, ~penc, Runner{j, s, p, bs, c}, Enqueue{deadline, x})
617 return enqueue.done(A, Q.next_id(A, queue(A, s)), y)
618
619# Cancels ticket id. A running attempt finishes, but its ticket is never selected again.
620def cancel(~A: Data, ~penc: A -> Bytes.Bytes, r: Runner<A>, +id: Nat) -> IO(Ran(A)):
621 commit(~A, ~penc, r, Cancel{id})
622
623# The runner back, and a copy of its state.
624def state(-A: Data, r: Runner<A>) -> Runner<A> & State<A>:
625 Runner{j, +s, p, bs, c} = r
626 (Runner{j, s, p, bs, c}, s)
627
628def close(-A: Data, r: Runner<A>) -> IO(Unit):
629 Runner{j, s, p, bs, c} = r
630 J.close(j)