gateway.bend
bend-kit-gateway@0.1.0.0 · 127 lines · 0 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.1.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))