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