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