gateway.bend

bend-kit-gateway@0.3.0.0 · 127 lines · 21 definitions

1# A guarded read-through gateway: authorize, admit, serve from cache, and fetch on a miss. Source: https://github.com/paymog/bend-kit/tree/main/gateway
2import Base
3import bend-kit-authz@0.1.0.0/authz.bend as Az
4import bend-kit-resources@0.1.0.0/resources.bend as Res
5import bend-kit-cache@0.3.0.0/cache.bend as C
6
7# The shared state of one gateway. `held` joins every unit that the pool lent
8# and that no finish has given back yet.
9type Gate is Type:
10  Gate{pool: Res.Pool, held: Res.Grant, cache: C.Cache<String>}
11
12# A fetch in flight, as copyable data, so it can cross to the request task and back.
13type Pending is Data:
14  Pending{key: String, gen: Nat, at: Nat}
15
16# What the upstream fetch did. Failed covers an upstream error, a spent budget, and an open breaker.
17type Upstream is Data:
18  Fetched{found: C.Found<String>}
19  Failed{}
20
21type Begin is Type:
22  Denied{gate: Gate}
23  Busy{gate: Gate}
24  Fresh{gate: Gate, entry: C.Entry<String>}
25  Fetch{gate: Gate, pending: Pending}
26
27type Finish is Type:
28  Settled{gate: Gate}
29  Unheld{gate: Gate}
30
31# An empty gateway: `capacity` requests at once, and an empty cache.
32def new(+owner: Nat, capacity: Nat, policy: C.Policy) -> Gate:
33  Gate{Res.new(owner, capacity), Res.Grant{owner, 0n}, C.Cache.new(String, policy)}
34
35def give.released(r: Res.Release, rest: Res.Grant, cache: C.Cache<String>) -> Finish:
36  match r:
37    case Res.Released{pool}:
38      Settled{Gate{pool, rest, cache}}
39    case Res.Kept{pool, one}:
40      Unheld{Gate{pool, rest, cache}}
41
42def give.split(s: Res.Split, pool: Res.Pool, cache: C.Cache<String>) -> Finish:
43  match s:
44    case Res.Parts{one, rest}:
45      give.released(Res.release(pool, one), rest, cache)
46    case Res.Whole{held}:
47      Unheld{Gate{pool, held, cache}}
48
49# Gives one lent unit back to the pool. Unheld means that no unit was lent.
50def give(pool: Res.Pool, held: Res.Grant, cache: C.Cache<String>) -> Finish:
51  give.split(Res.split(held, 1n), pool, cache)
52
53# Starts a fetch of key at clock `at`, as data.
54def start.ticket(r: C.Ticket & C.Cache<String>) -> Pending & C.Cache<String>:
55  (t, cache) = r
56  C.Ticket{key, gen, at} = t
57  (Pending{key, gen, at}, cache)
58
59def start(cache: C.Cache<String>, +key: String, at: Nat) -> Pending & C.Cache<String>:
60  start.ticket(C.Cache.start(~String, cache, key, at))
61
62# Stores what a fetch found. A failed fetch stores nothing.
63def settle(cache: C.Cache<String>, p: Pending, u: Upstream) -> C.Cache<String>:
64  Pending{key, gen, at} = p
65  match u:
66    case Fetched{found}:
67      C.Cache.complete(~String, cache, C.Ticket{key, gen, at}, found)
68    case Failed{}:
69      cache
70
71def begin.fetch(pool: Res.Pool, held: Res.Grant, r: Pending & C.Cache<String>) -> Begin:
72  (p, cache) = r
73  Fetch{Gate{pool, held, cache}, p}
74
75# The joined grant names another pool: give the new unit back and refuse.
76def begin.refund(r: Res.Release, held: Res.Grant, cache: C.Cache<String>) -> Begin:
77  match r:
78    case Res.Released{pool}:
79      Busy{Gate{pool, held, cache}}
80    case Res.Kept{pool, one}:
81      Busy{Gate{pool, held, cache}}
82
83def begin.fresh(f: Finish, e: C.Entry<String>) -> Begin:
84  match f:
85    case Settled{gate}:
86      Fresh{gate, e}
87    case Unheld{gate}:
88      Busy{gate}
89
90def begin.read(m: Maybe<&2, C.Entry<String>>, pool: Res.Pool, held: Res.Grant, cache: C.Cache<String>, +key: String, +now: Nat) -> Begin:
91  match m:
92    case Some{e}:
93      begin.fresh(give(pool, held, cache), e)
94    case None{}:
95      begin.fetch(pool, held, start(cache, key, now))
96
97def begin.joined(j: Res.Combine, pool: Res.Pool, +cache: C.Cache<String>, +key: String, +now: Nat) -> Begin:
98  match j:
99    case Res.Joined{held}:
100      begin.read(C.Cache.read(~String, cache, key, now), pool, held, cache, key, now)
101    case Res.Apart{held, one}:
102      begin.refund(Res.release(pool, one), held, cache)
103
104def begin.took(r: Res.Reserve, held: Res.Grant, cache: C.Cache<String>, +key: String, +now: Nat) -> Begin:
105  match r:
106    case Res.Took{pool, one}:
107      begin.joined(Res.combine(held, one), pool, cache, key, now)
108    case Res.Refused{pool}:
109      Busy{Gate{pool, held, cache}}
110
111def begin.granted(ok: Bool, gate: Gate, +key: String, +now: Nat) -> Begin:
112  match ok:
113    case True{}:
114      Gate{pool, held, cache} = gate
115      begin.took(Res.reserve(pool, 1n), held, cache, key, now)
116    case False{}:
117      Denied{gate}
118
119# Authorizes req, takes one unit, and reads key at `now`. Fresh has given its
120# unit back; Fetch keeps it until finish.
121def begin(policy: Az.Policy, req: Az.Request, gate: Gate, +key: String, +now: Nat) -> Begin:
122  begin.granted(Az.granted(policy, req), gate, key, now)
123
124# Stores what the fetch did and gives its unit back.
125def finish(gate: Gate, p: Pending, u: Upstream) -> Finish:
126  Gate{pool, held, cache} = gate
127  give(pool, held, settle(cache, p, u))