src/transport/server.bend

bendonnect-rpc@0.1.0.0 · 364 lines · 63 definitions

1# Connect server transport on bend-kit-http's serve.write.* (HTTP/1.1 only).
2# Request bodies are buffered before the handler runs (no client/bidi streaming); responses are
3# written incrementally (server streaming). Unary and error replies use a known length so the body
4# is not chunked. Run it with `bend file.bend` (runner) or a native build (`bend f.bend -o f`).
5import Base
6import bend-kit-http@0.36.0.0/http.bend as Http
7import bend-kit-http@0.36.0.0/completion.bend as Completion
8import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
9import bendonnect-rpc@0.1.0.0/src/code.bend as Code
10import bendonnect-rpc@0.1.0.0/src/error.bend as E
11import bendonnect-rpc@0.1.0.0/src/metadata.bend as Md
12import bendonnect-rpc@0.1.0.0/src/envelope.bend as Env
13import bendonnect-rpc@0.1.0.0/src/protocol.bend as P
14
15# ---------------------------------------------------------------- public API
16
17type Meta is Data:
18  Meta{headers: Map<&2, List<&2, String>>, trailers: Map<&2, List<&2, String>>}
19
20type Call is Type:
21  Call{procedure: String, kind: P.CallKind, codec: P.Codec, headers: Map<&2, List<&2, String>>,
22    timeout_ms: Maybe<&2, U32>, messages: List<&1, Bytes.Bytes>}
23
24# Wraps the HTTP writer of a server stream.
25type Sink is Type:
26  Sink{w: Http.Writer}
27
28type Reply is Type:
29  Single{result: Result<&1, &1, E.Error, Bytes.Bytes>, meta: Meta}
30  Many{run: Sink -> IO(Sink & Result<&1, &1, E.Error, Unit>), meta: Meta}
31  NoRoute{}
32
33def meta.empty() -> Meta:
34  Meta{Map.new(&2, List<&2, String>), Map.new(&2, List<&2, String>)}
35
36def ok(msg: Bytes.Bytes) -> Reply:
37  Single{Done{msg}, meta.empty()}
38
39def fail(code: Code.Code, message: String) -> Reply:
40  Single{Fail{E.new(code, message)}, meta.empty()}
41
42# One enveloped message, written (and flushed) immediately.
43def send(sink: Sink, msg: Bytes.Bytes) -> IO(Sink):
44  Sink{w} = sink
45  do IO<Sink>:
46    w2 : Http.Writer <- Http.writer.write(w, Env.encode(Env.Envelope{0, msg}))
47    return Sink{w2}
48
49# ---------------------------------------------------------------- internals
50
51# Per-request state carried from `start` to `write`.
52type Job is Type:
53  JBytes{body: Bytes.Bytes}
54  JMany{run: Sink -> IO(Sink & Result<&1, &1, E.Error, Unit>), trailers: Map<&2, List<&2, String>>}
55
56# Result of the pure half of `start`.
57type Pre is Type:
58  PEarly{status: U32, headers: Map<&2, List<&2, String>>, body: Bytes.Bytes}
59  PReady{call: Call, timeout: Maybe<&2, U32>, stream: Bool, codec: P.Codec}
60
61def msg_max() -> U32:
62  4194304
63
64def merge.go(xs: List<&2, Sigma<&2, &2, String, _ => List<&2, String>>>, acc: Map<&2, List<&2, String>>) -> Map<&2, List<&2, String>>:
65  match xs:
66    case Nil{}:
67      acc
68    case Con{Tuple{k, v}, rest}:
69      merge.go(rest, Map.set(&2, List<&2, String>, acc, k, v))
70
71def merge(a: Map<&2, List<&2, String>>, b: Map<&2, List<&2, String>>) -> Map<&2, List<&2, String>>:
72  merge.go(Map.to_list(&2, List<&2, String>, b), a)
73
74def end_env(e: Maybe<&1, E.Error>, trailers: Map<&2, List<&2, String>>) -> Bytes.Bytes:
75  Env.encode(Env.Envelope{Env.flag_end(), E.end_encode(E.EndStream{e, trailers})})
76
77def end_bytes(res: Result<&1, &1, E.Error, Unit>, trailers: Map<&2, List<&2, String>>) -> Bytes.Bytes:
78  match res:
79    case Fail{e}:
80      end_env(Some{e}, trailers)
81    case Done{u}:
82      end_env(None{}, trailers)
83
84def json_hdr() -> Map<&2, List<&2, String>>:
85  Http.set(Http.empty(), "content-type", "application/json")
86
87def head(status: U32, headers: Map<&2, List<&2, String>>, body: Bytes.Bytes) -> IO(Http.WriteHead<Job>):
88  Bytes.Bytes{+len, buf} = body
89  IO.pure(Http.WriteHead<Job>, Http.WriteHead{status, headers, Some{len}, JBytes{Bytes.Bytes{len, buf}}, Completion.Unobserved{}})
90
91def err_status(e: E.Error) -> U32:
92  E.Error{+code, message, details} = e
93  Code.to_http(code)
94
95# HTTP-level error (unary style): status from the code, JSON body.
96def early_json(e: E.Error) -> Pre:
97  E.Error{+code, +message, details} = e
98  PEarly{Code.to_http(code), json_hdr(), E.encode(E.Error{code, message, details})}
99
100# Streaming-style error: 200 + a single EndStream envelope.
101def early_stream(c: P.Codec, e: E.Error) -> Pre:
102  PEarly{200, Http.set(Http.empty(), "content-type", P.stream_content_type(c)), end_env(Some{e}, Map.new(&2, List<&2, String>))}
103
104def early.pick(stream: Bool, c: P.Codec, e: E.Error) -> Pre:
105  match stream:
106    case True{}:
107      early_stream(c, e)
108    case False{}:
109      early_json(e)
110
111def reject_pre(r: P.Reject, stream: Bool, c: P.Codec) -> Pre:
112  match r:
113    case P.Status{+s, +m}:
114      PEarly{s, json_hdr(), E.encode(E.new(Code.from_http(s), m))}
115    case P.Connect{code, m}:
116      early.pick(stream, c, E.new(code, m))
117
118# --- request envelopes -> messages
119
120def env_flags_ok(+flags: U32) -> Bool:
121  U32.is_eq(flags, 0)
122
123def msgs.cons(m: Bytes.Bytes, r: Result<&1, &1, E.Error, List<&1, Bytes.Bytes>>) -> Result<&1, &1, E.Error, List<&1, Bytes.Bytes>>:
124  match r:
125    case Fail{e}:
126      Fail{e}
127    case Done{xs}:
128      Done{m <> xs}
129
130def msgs.one(ok: Bool, data: Bytes.Bytes, rest: Result<&1, &1, E.Error, List<&1, Bytes.Bytes>>) -> Result<&1, &1, E.Error, List<&1, Bytes.Bytes>>:
131  match ok:
132    case True{}:
133      msgs.cons(data, rest)
134    case False{}:
135      Fail{E.new(Code.InvalidArgument{}, "unexpected envelope flags in request (compression or end-stream with identity encoding)")}
136
137def msgs(envs: List<&1, Env.Envelope>) -> Result<&1, &1, E.Error, List<&1, Bytes.Bytes>>:
138  match envs:
139    case Nil{}:
140      Done{Nil{}}
141    case Con{Env.Envelope{+flags, data}, rest}:
142      msgs.one(env_flags_ok(flags), data, msgs(rest))
143
144def too_big(+reason: String) -> Bool:
145  String.starts_with(reason, "message size")
146
147def decode_err(+reason: String) -> E.Error:
148  E.new(Bool.pick(Code.Code, too_big(reason), Code.ResourceExhausted{}, Code.InvalidArgument{}), reason)
149
150# --- building the Call
151
152def ready(procedure: String, kind: P.CallKind, +codec: P.Codec, headers: Map<&2, List<&2, String>>, +t: Maybe<&2, U32>, stream: Bool, ms: List<&1, Bytes.Bytes>) -> Pre:
153  PReady{Call{procedure, kind, codec, headers, t, ms}, t, stream, codec}
154
155def stream_done(procedure: String, codec: P.Codec, headers: Map<&2, List<&2, String>>, t: Maybe<&2, U32>, r: Result<&1, &1, E.Error, List<&1, Bytes.Bytes>>) -> Pre:
156  match r:
157    case Fail{e}:
158      early_stream(codec, e)
159    case Done{ms}:
160      ready(procedure, P.Stream{}, codec, headers, t, True{}, ms)
161
162def stream_decoded(procedure: String, codec: P.Codec, headers: Map<&2, List<&2, String>>, t: Maybe<&2, U32>, r: Result<&2, &1, String, List<&1, Env.Envelope>>) -> Pre:
163  match r:
164    case Fail{reason}:
165      early_stream(codec, decode_err(reason))
166    case Done{envs}:
167      stream_done(procedure, codec, headers, t, msgs(envs))
168
169def build.go(kind: P.CallKind, procedure: String, codec: P.Codec, t: Maybe<&2, U32>, message: Bytes.Bytes, headers: Map<&2, List<&2, String>>, body: Bytes.Bytes) -> Pre:
170  match kind:
171    case P.Unary{}:
172      ready(procedure, P.Unary{}, codec, headers, t, False{}, body <> Nil{})
173    case P.UnaryGet{}:
174      ready(procedure, P.UnaryGet{}, codec, headers, t, False{}, message <> Nil{})
175    case P.Stream{}:
176      stream_decoded(procedure, codec, headers, t, Env.decode_all(body, msg_max()))
177
178# Only JSON and Protobuf are served: the spec answers any other codec with 415.
179def build.codec(+codec: P.Codec, kind: P.CallKind, procedure: String, t: Maybe<&2, U32>, message: Bytes.Bytes, headers: Map<&2, List<&2, String>>, body: Bytes.Bytes) -> Pre:
180  match codec:
181    case P.Custom{name}:
182      reject_pre(P.Status{415, "unsupported codec \"" ++ name ++ "\": supported codecs are json and proto"}, False{}, P.Json{})
183    case _:
184      build.go(kind, procedure, codec, t, message, headers, body)
185
186def build(spec: P.Spec, headers: Map<&2, List<&2, String>>, body: Bytes.Bytes) -> Pre:
187  P.Spec{procedure, kind, codec, timeout_ms, encoding, accept, message} = spec
188  build.codec(codec, kind, procedure, timeout_ms, message, headers, body)
189
190def pre.go(r: Result<&2, &1, P.Reject, P.Spec>, headers: Map<&2, List<&2, String>>, body: Bytes.Bytes, ct: P.Ct) -> Pre:
191  match r:
192    case Fail{rej}:
193      P.Ct{+stream, +codec} = ct
194      reject_pre(rej, stream, codec)
195    case Done{spec}:
196      build(spec, headers, body)
197
198# Content type -> (stream?, codec), used only to shape rejections. Falls back to unary JSON.
199def ct_of.fin(m: Maybe<&2, P.Ct>) -> P.Ct:
200  match m:
201    case None{}:
202      P.Ct{False{}, P.Json{}}
203    case Some{c}:
204      c
205
206def ct_of(+h: Map<&2, List<&2, String>>) -> P.Ct:
207  ct_of.fin(P.parse_content_type(Http.header(h, "content-type")))
208
209def split_q.cons(c: U32, r: String & String) -> String & String:
210  (a, b) = r
211  (SCon{Chr{c}, a}, b)
212
213def split_q.pick(hit: Bool, +c: U32, +t: String, rest: Unit -> String & String) -> String & String:
214  match hit:
215    case True{}:
216      ("", t)
217    case False{}:
218      split_q.cons(c, rest(Unit{}))
219
220# "/p/q?a=b" -> ("/p/q", "a=b")
221def split_q(+s: String) -> String & String:
222  match s:
223    case SNil{}:
224      ("", "")
225    case SCon{Chr{+c}, +t}:
226      split_q.pick(U32.is_eq(c, 63), c, t, u => split_q(t))
227
228def pre.split(+m: String, pq: String & String, +headers: Map<&2, List<&2, String>>, body: Bytes.Bytes, +prefix: String) -> Pre:
229  (path, query) = pq
230  pre.go(P.classify(m, path, query, headers, prefix), headers, body, ct_of(headers))
231
232def pre(req: Http.Req, +prefix: String) -> Pre:
233  Http.Req{+method, +rawpath, +headers, body} = req
234  pre.split(method, split_q(rawpath), headers, body, prefix)
235
236# --- running the handler
237
238def timed.fin(m: Maybe<&1, Reply>) -> Reply:
239  match m:
240    case None{}:
241      fail(Code.DeadlineExceeded{}, "deadline exceeded")
242    case Some{r}:
243      r
244
245# NOTE: IO.within does not cancel the loser; a handler that overruns keeps running unobserved.
246def timed(~handle: Call -> IO(Reply), call: Call, t: Maybe<&2, U32>) -> IO(Reply):
247  match t:
248    case None{}:
249      handle(call)
250    case Some{+ms}:
251      do IO<Reply>:
252        m : Maybe<&1, Reply> <- IO.within(Reply, ms, handle(call))
253        return timed.fin(m)
254
255# --- rendering a Reply into a response head (+ body or streaming state)
256
257def ct_hdr(m: Map<&2, List<&2, String>>, ct: String) -> Map<&2, List<&2, String>>:
258  Http.set(m, "content-type", ct)
259
260def all_hdrs(meta: Meta, ct: String) -> Map<&2, List<&2, String>>:
261  Meta{hs, ts} = meta
262  ct_hdr(merge(hs, Md.trailers_to_headers(ts)), ct)
263
264def unary_ok(codec: P.Codec, b: Bytes.Bytes, meta: Meta) -> IO(Http.WriteHead<Job>):
265  head(200, all_hdrs(meta, P.unary_content_type(codec)), b)
266
267def unary_err(e: E.Error, meta: Meta) -> IO(Http.WriteHead<Job>):
268  E.Error{+code, +message, details} = e
269  head(Code.to_http(code), all_hdrs(meta, "application/json"), E.encode(E.Error{code, message, details}))
270
271def single_unary(codec: P.Codec, r: Result<&1, &1, E.Error, Bytes.Bytes>, meta: Meta) -> IO(Http.WriteHead<Job>):
272  match r:
273    case Done{b}:
274      unary_ok(codec, b, meta)
275    case Fail{e}:
276      unary_err(e, meta)
277
278def stream_single.go(r: Result<&1, &1, E.Error, Bytes.Bytes>, ct: String, hs: Map<&2, List<&2, String>>, ts: Map<&2, List<&2, String>>) -> IO(Http.WriteHead<Job>):
279  match r:
280    case Done{b}:
281      head(200, ct_hdr(hs, ct), Bytes.concat(Env.encode(Env.Envelope{0, b}) <> end_env(None{}, ts) <> Nil{}))
282    case Fail{e}:
283      head(200, ct_hdr(hs, ct), end_env(Some{e}, ts))
284
285def stream_single(codec: P.Codec, r: Result<&1, &1, E.Error, Bytes.Bytes>, meta: Meta) -> IO(Http.WriteHead<Job>):
286  Meta{+hs, +ts} = meta
287  stream_single.go(r, P.stream_content_type(codec), hs, ts)
288
289def many_head(codec: P.Codec, run: Sink -> IO(Sink & Result<&1, &1, E.Error, Unit>), meta: Meta) -> IO(Http.WriteHead<Job>):
290  Meta{hs, ts} = meta
291  IO.pure(Http.WriteHead<Job>, Http.WriteHead{200, ct_hdr(hs, P.stream_content_type(codec)), None{}, JMany{run, ts}, Completion.Unobserved{}})
292
293def not_found() -> IO(Http.WriteHead<Job>):
294  head(404, json_hdr(), E.encode(E.new(Code.NotFound{}, "no such procedure")))
295
296def internal(msg: String) -> IO(Http.WriteHead<Job>):
297  head(500, json_hdr(), E.encode(E.new(Code.Internal{}, msg)))
298
299def render.unary(codec: P.Codec, reply: Reply) -> IO(Http.WriteHead<Job>):
300  match reply:
301    case Single{r, meta}:
302      single_unary(codec, r, meta)
303    case Many{run, meta}:
304      internal("server-streaming reply returned for a unary call")
305    case NoRoute{}:
306      not_found()
307
308def render.stream(codec: P.Codec, reply: Reply) -> IO(Http.WriteHead<Job>):
309  match reply:
310    case Single{r, meta}:
311      stream_single(codec, r, meta)
312    case Many{run, meta}:
313      many_head(codec, run, meta)
314    case NoRoute{}:
315      not_found()
316
317def render(stream: Bool, codec: P.Codec, reply: Reply) -> IO(Http.WriteHead<Job>):
318  match stream:
319    case True{}:
320      render.stream(codec, reply)
321    case False{}:
322      render.unary(codec, reply)
323
324def start2(~handle: Call -> IO(Reply), p: Pre) -> IO(Http.WriteHead<Job>):
325  match p:
326    case PEarly{status, headers, body}:
327      head(status, headers, body)
328    case PReady{call, timeout, stream, codec}:
329      do IO<Http.WriteHead<Job>>:
330        reply : Reply <- timed(~handle, call, timeout)
331        render(stream, codec, reply)
332
333def start(~handle: Call -> IO(Reply), ~prefix: String, req: Http.Req) -> IO(Http.WriteHead<Job>):
334  start2(~handle, pre(req, prefix))
335
336def many.end(r: Sink & Result<&1, &1, E.Error, Unit>, trailers: Map<&2, List<&2, String>>, receipt: Completion.Receipt) -> IO(Http.Writer & Completion.Receipt):
337  (s, res) = r
338  Sink{w} = s
339  do IO<Http.Writer & Completion.Receipt>:
340    w2 : Http.Writer <- Http.writer.write(w, end_bytes(res, trailers))
341    return (w2, receipt)
342
343def write(st: Job, w: Http.Writer, receipt: Completion.Receipt) -> IO(Http.Writer & Completion.Receipt):
344  match st:
345    case JBytes{b}:
346      do IO<Http.Writer & Completion.Receipt>:
347        w2 : Http.Writer <- Http.writer.write(w, b)
348        return (w2, receipt)
349    case JMany{run, trailers}:
350      do IO<Http.Writer & Completion.Receipt>:
351        r : Sink & Result<&1, &1, E.Error, Unit> <- run(Sink{w})
352        many.end(r, trailers, receipt)
353
354def dispose(st: Job) -> IO(Unit):
355  match st:
356    case JBytes{b}:
357      IO.pure(Unit, Unit{})
358    case JMany{run, trailers}:
359      IO.pure(Unit, Unit{})
360
361# Serves forever on host:port. `prefix` is "" or e.g. "/api". It is a template argument
362# (pass a string literal) because serve.write.on.with takes closed templates.
363def serve(~handle: Call -> IO(Reply), ~prefix: String, host: String, +port: U32) -> IO(Unit):
364  Http.serve.write.on.with(~Job, ~(req => start(~handle, ~prefix, req)), ~write, ~dispose, ~Completion.ignore, host, port, 33554432n)