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