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