src/coordinator/Io.bend
mylsm-lsm-store@0.6.0.0 · 731 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.0/src/coordinator/Protocol.bend as Protocol
4import mylsm-lsm-store@0.6.0.0/src/coordinator/IoPolicy.bend as IoPolicy
5import mylsm-lsm-store@0.6.0.0/src/coordinator/UnixSocketPolicy.bend as UnixSocketPolicy
6import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7
8# Classifies framing, resource, and host failures from socket IO.
9type Error is Data:
10 UnsupportedPlatform{}
11 Transient{code: U32}
12 Host{code: U32, message: String}
13 Truncated{}
14 Malformed{}
15 UnknownOperation{}
16 FrameTooLarge{}
17 MalformedResponse{}
18 UnexpectedResponse{}
19 UnknownResponseStatus{}
20
21# Returns the terminal outcome of an exact-length socket transfer.
22type ExactResult is Type:
23 ExactResult{bytes: Bytes.Bytes, complete: Bool}
24
25# Carries the state for ReadAttempt.
26type ReadAttempt is Type:
27 ReadAvailable{bytes: Bytes.Bytes}
28 ReadInterrupted{}
29 ReadWouldBlock{}
30 ReadFailed{error: Error}
31
32# Carries the state for WriteAttempt.
33type WriteAttempt is Type:
34 NonblockingWriteSent{bytes: Bytes.Bytes, sent: U32}
35 NonblockingWriteInterrupted{bytes: Bytes.Bytes}
36 NonblockingWriteWouldBlock{bytes: Bytes.Bytes}
37 NonblockingWriteFailed{bytes: Bytes.Bytes, error: Error}
38
39# Tracks partial transfer progress until the requested length is complete.
40type ExactState is Type:
41 Need{file: File, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
42 Read{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
43 remaining: Nat, chunks: List<&1, Bytes.Bytes>}
44
45# Carries bounded chunks through one-syscall socket write steps.
46type WriteState is Type:
47 WriteNeed{file: File, remaining: U32 & U32 & U32, total: U32,
48 offsets: U32 & U32 & U32, bytes: Bytes.Bytes}
49 WriteDone{file: File}
50 WriteSend{file: File, total: U32, offset: U32 & U32 & U32,
51 remaining: U32 & U32 & U32,
52 words: Array<U32>}
53 WriteSent{pair: File & Result<&1, &1, U32 & String, U32 & Array<U32>>,
54 total: U32, offset: U32, remaining: U32}
55 WriteAdvance{file: File, total: U32, offset: U32, remaining: U32,
56 sent: U32 & U32 & U32, words: Array<U32>}
57 WriteFailed{file: File}
58
59
60# Sets the protocol payload ceiling enforced before socket allocation.
61def max_payload_bytes() -> Nat:
62 Protocol.max_payload_bytes()
63
64def host.error.classified(
65 code: U32,
66 message: String,
67 classification: UnixSocketPolicy.Classification
68) -> Error:
69 match classification:
70 case UnixSocketPolicy.UnsupportedPlatform{}: UnsupportedPlatform{}
71 case UnixSocketPolicy.Transient{code}: Transient{code}
72 case UnixSocketPolicy.Host{_}: Host{code, message}
73
74def host.error.copies(
75 copies: U32 & U32 & U32,
76 message: String
77) -> Error:
78 match copies:
79 case (classification_code, (retained_code, _)):
80 host.error.classified(retained_code, message,
81 UnixSocketPolicy.classify_code(classification_code))
82
83def host.error(code: U32, message: String) -> Error:
84 host.error.copies(Protocol.u32.duplicate.triple(code), message)
85
86# Reads one raw byte chunk from an accepted local socket.
87def unix_socket_read(
88 file: File,
89 max_bytes: U32
90) -> IO(
91 File & Result<&1, &1, U32 & String, U32 & Array<U32>>
92):
93 import "../effs/unix_socket_read.c"
94 import "../effs/unix_socket_read.js"
95
96def read.words(
97 file: File,
98 max_bytes: U32
99) -> IO(
100 File & Result<&1, &1, U32 & String, U32 & Array<U32>>
101):
102 unix_socket_read(file, max_bytes)
103
104def read.wrap(
105 result: Result<&1, &1, U32 & String, U32 & Array<U32>>
106) -> Result<&1, &1, U32 & String, Bytes.Bytes>:
107 match result:
108 case Fail{error}: Fail{error}
109 case Done{(length, words)}: Done{Bytes.Bytes{length, words}}
110
111def read.pair(
112 pair: File & Result<&1, &1, U32 & String, U32 & Array<U32>>
113) -> File & Result<&1, &1, U32 & String, Bytes.Bytes>:
114 match pair:
115 case (file, result): (file, read.wrap(result))
116
117def read.bytes(
118 file: File,
119 max_bytes: U32
120) -> IO(
121 File & Result<&1, &1, U32 & String, Bytes.Bytes>
122):
123 do IO<File & Result<&1, &1, U32 & String, Bytes.Bytes>>:
124 pair : File & Result<&1, &1, U32 & String, U32 & Array<U32>> <-
125 read.words(file, max_bytes)
126 return read.pair(pair)
127
128def read.nonblocking.classified(
129 file: File,
130 +code: U32,
131 message: String,
132 disposition: IoPolicy.ReadDisposition
133) -> File & ReadAttempt:
134 match disposition:
135 case IoPolicy.ReadInterrupted{}: (file, ReadInterrupted{})
136 case IoPolicy.ReadWouldBlock{}: (file, ReadWouldBlock{})
137 case IoPolicy.ReadFailure{}:
138 (file, ReadFailed{host.error(code, message)})
139
140def read.nonblocking.failed(
141 file: File,
142 pair: U32 & String
143) -> File & ReadAttempt:
144 match pair:
145 case (+code, message):
146 read.nonblocking.classified(file, code, message,
147 IoPolicy.read.disposition(code))
148
149def read.nonblocking.result(
150 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
151) -> File & ReadAttempt:
152 match pair:
153 case (file, Done{bytes}): (file, ReadAvailable{bytes})
154 case (file, Fail{error}): read.nonblocking.failed(file, error)
155
156def read.nonblocking(
157 file: File,
158 max_bytes: U32
159) -> IO(File & ReadAttempt):
160 do IO<File & ReadAttempt>:
161 pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
162 read.bytes(file, max_bytes)
163 return read.nonblocking.result(pair)
164
165# Handles unix socket write nonblocking.
166def unix_socket_write_nonblocking(
167 file: File,
168 offset: U32,
169 length: U32,
170 words: Array<U32>
171) -> IO(File & (Array<U32> &
172 Result<&1, &1, U32 & String, U32>)):
173 import "../effs/unix_socket_write_nonblocking.c"
174 import "../effs/unix_socket_write_nonblocking.js"
175
176def write.nonblocking.classified(
177 file: File,
178 bytes: Bytes.Bytes,
179 +code: U32,
180 message: String,
181 disposition: IoPolicy.WriteDisposition
182) -> File & WriteAttempt:
183 match disposition:
184 case IoPolicy.WriteInterrupted{}:
185 (file, NonblockingWriteInterrupted{bytes})
186 case IoPolicy.WriteWouldBlock{}:
187 (file, NonblockingWriteWouldBlock{bytes})
188 case IoPolicy.WriteFailure{}:
189 (file, NonblockingWriteFailed{bytes, host.error(code, message)})
190
191def write.nonblocking.failed(
192 file: File,
193 bytes: Bytes.Bytes,
194 error: U32 & String
195) -> File & WriteAttempt:
196 match error:
197 case (+code, message):
198 write.nonblocking.classified(file, bytes, code, message,
199 IoPolicy.write.disposition(code))
200
201def write.nonblocking.wrap(
202 pair: File & (Array<U32> & Result<&1, &1, U32 & String, U32>),
203 total: U32
204) -> File & WriteAttempt:
205 match pair:
206 case (file, (words, result)):
207 match result:
208 case Done{sent}:
209 (file, NonblockingWriteSent{Bytes.Bytes{total, words}, sent})
210 case Fail{error}:
211 write.nonblocking.failed(file, Bytes.Bytes{total, words}, error)
212
213def write.nonblocking.start(
214 file: File,
215 offset: U32,
216 length: U32,
217 total: U32,
218 words: Array<U32>
219) -> IO(File & WriteAttempt):
220 do IO<File & WriteAttempt>:
221 pair : File & (Array<U32> & Result<&1, &1, U32 & String, U32>) <-
222 unix_socket_write_nonblocking(file, offset, length, words)
223 return write.nonblocking.wrap(pair, total)
224
225def write.nonblocking(
226 file: File,
227 offset: U32,
228 length: U32,
229 bytes: Bytes.Bytes
230) -> IO(File & WriteAttempt):
231 match bytes:
232 case Bytes.Bytes{total, words}:
233 write.nonblocking.start(file, offset, length, total, words)
234
235def write.words(
236 file: File,
237 offset: U32,
238 length: U32,
239 words: Array<U32>
240) -> IO(File & Result<&1, &1, U32 & String, U32 & Array<U32>>):
241 unix_socket_write(file, offset, length, words)
242
243def write.chunk(remaining: U32) -> U32:
244 IoPolicy.write_chunk(remaining)
245
246# Writes one raw byte chunk to a local socket connection.
247def unix_socket_write(
248 file: File,
249 offset: U32,
250 length: U32,
251 words: Array<U32>
252) -> IO(File & Result<&1, &1, U32 & String, U32 & Array<U32>>):
253 import "../effs/unix_socket_write.c"
254 import "../effs/unix_socket_write.js"
255
256def write.advance.total(
257 copies: U32 & U32 & U32,
258 file: File,
259 offset: U32,
260 remaining: U32,
261 words: Array<U32>
262) -> WriteState:
263 match copies:
264 case (total_state, (total_bytes, _)):
265 WriteNeed{file, Protocol.u32.duplicate.triple(remaining), total_state,
266 Protocol.u32.duplicate.triple(offset),
267 Bytes.Bytes{total_bytes, words}}
268
269def write.advance(
270 zero: Bool,
271 file: File,
272 total: U32,
273 offset: U32,
274 remaining: U32,
275 sent_offset: U32,
276 sent_remaining: U32,
277 words: Array<U32>
278) -> WriteState:
279 match zero:
280 case True{}: WriteFailed{file}
281 case False{}:
282 write.advance.total(Protocol.u32.duplicate.triple(total), file,
283 (offset + sent_offset : U32), (remaining - sent_remaining : U32),
284 words)
285
286def write.need(
287 zero: Bool,
288 file: File,
289 remaining_send: U32,
290 total: U32,
291 offsets: U32 & U32 & U32,
292 bytes: Bytes.Bytes
293) -> WriteState:
294 match zero:
295 case True{}: WriteDone{file}
296 case False{}:
297 match offsets:
298 case (write_offset, _, _):
299 match bytes:
300 case Bytes.Bytes{_, words}:
301 WriteSend{file, total,
302 Protocol.u32.duplicate.triple(write_offset),
303 Protocol.u32.duplicate.triple(remaining_send), words}
304
305def write.loop(
306 fuel: Nat,
307 state: WriteState
308) -> IO(
309 File & Result<&1, &1, U32 & String, Unit>
310):
311 match fuel:
312 case 0n:
313 match state:
314 case WriteNeed{file, _, _, _, _}:
315 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
316 (file, Fail{(7, "socket write exhausted its progress bound")}))
317 case WriteSent{pair, _, _, _}:
318 match pair:
319 case (file, _):
320 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
321 (file, Fail{(7, "socket write exhausted its progress bound")}))
322 case WriteAdvance{file, _, _, _, _, _}:
323 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
324 (file, Fail{(7, "socket write exhausted its progress bound")}))
325 case WriteFailed{file}:
326 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
327 (file, Fail{(5, "socket write made no progress")}))
328 case WriteDone{file}:
329 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
330 (file, Done{Unit{}}))
331 case WriteSend{file, _, _, _, _}:
332 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
333 (file, Fail{(7, "socket write exhausted its progress bound")}))
334 case 1n+rest:
335 match state:
336 case WriteNeed{file, remaining, total, offsets, bytes}:
337 match remaining:
338 case (remaining_check, (remaining_send, _)):
339 write.loop(rest, write.need(U32.is_eq(remaining_check, 0), file,
340 remaining_send, total, offsets, bytes))
341 case WriteDone{file}:
342 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
343 (file, Done{Unit{}}))
344 case WriteFailed{file}:
345 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
346 (file, Fail{(5, "socket write made no progress")}))
347 case WriteSend{file, total, offset, remaining, words}:
348 match offset:
349 case (offset_send, (offset_keep, _)):
350 match remaining:
351 case (remaining_send, (remaining_keep, _)):
352 do IO<File & Result<&1, &1, U32 & String, Unit>>:
353 pair : File & Result<&1, &1, U32 & String, U32 & Array<U32>> <-
354 write.words(file, offset_send,
355 write.chunk(remaining_send), words)
356 write.loop(rest, WriteSent{pair, total, offset_keep,
357 remaining_keep})
358 case WriteSent{pair, total, offset, remaining}:
359 match pair:
360 case (file, Fail{error}):
361 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
362 (file, Fail{error}))
363 case (file, Done{(sent, words)}):
364 write.loop(rest, WriteAdvance{file, total, offset, remaining,
365 Protocol.u32.duplicate.triple(sent), words})
366 case WriteAdvance{file, total, offset, remaining, copies, words}:
367 match copies:
368 case (sent_check, (sent_offset, sent_remaining)):
369 write.loop(rest, write.advance(U32.is_eq(sent_check, 0),
370 file, total, offset, remaining, sent_offset, sent_remaining,
371 words))
372
373def write.start.total(
374 file: File,
375 words: Array<U32>,
376 fuel_length: U32,
377 remaining_seed: U32,
378 total_copies: U32 & U32 & U32
379) -> IO(File & Result<&1, &1, U32 & String, Unit>):
380 match total_copies:
381 case (total, (bytes_length, _)):
382 write.loop(Nat.add(Nat.mul(4n, U32.to_nat(fuel_length)), 2n),
383 WriteNeed{file, Protocol.u32.duplicate.triple(remaining_seed), total,
384 Protocol.u32.duplicate.triple(0), Bytes.Bytes{bytes_length, words}})
385
386def write.start.copies(
387 file: File,
388 words: Array<U32>,
389 copies: U32 & U32 & U32
390) -> IO(File & Result<&1, &1, U32 & String, Unit>):
391 match copies:
392 case (fuel_length, (total_seed, remaining_seed)):
393 write.start.total(file, words, fuel_length, remaining_seed,
394 Protocol.u32.duplicate.triple(total_seed))
395
396def write.start(
397 file: File,
398 length: U32,
399 words: Array<U32>
400) -> IO(File & Result<&1, &1, U32 & String, Unit>):
401 write.start.copies(file, words, Protocol.u32.duplicate.triple(length))
402
403# Completes a frame by advancing over bounded partial writes.
404def write_all(
405 file: File,
406 bytes: Bytes.Bytes
407) -> IO(
408 File & Result<&1, &1, U32 & String, Unit>
409):
410 match bytes:
411 case Bytes.Bytes{length, words}:
412 write.start(file, length, words)
413
414def exact.complete(
415 file: File,
416 chunks: List<&1, Bytes.Bytes>,
417 complete: Bool
418) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
419 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
420 (file, Done{ExactResult{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks)),
421 complete}}))
422
423def exact.loop(
424 fuel: Nat,
425 state: ExactState
426) -> IO(
427 File & Result<&1, &1, U32 & String, ExactResult>
428):
429 match fuel:
430 case 0n:
431 match state:
432 case Need{file, remaining, _}:
433 match remaining:
434 case 0n: exact.complete(file, Nil{}, True{})
435 case 1n+_: IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
436 (file, Fail{(7, "socket frame read exhausted its bound")}))
437 case Read{pair, _, _}:
438 match pair:
439 case (file, Fail{error}):
440 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
441 (file, Fail{error}))
442 case (file, Done{_}):
443 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
444 (file, Fail{(7, "socket frame read exhausted its bound")}))
445 case 1n+rest:
446 match state:
447 case Need{file, remaining, chunks}:
448 match remaining:
449 case 0n: exact.complete(file, chunks, True{})
450 case 1n+rest_remaining:
451 +total = Nat.add(1n, rest_remaining)
452 do IO<File & Result<&1, &1, U32 & String, ExactResult>>:
453 pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
454 read.bytes(file,
455 U32.from_nat(Nat.min(65536n, total)))
456 exact.loop(rest, Read{pair, total, chunks})
457 case Read{pair, remaining, chunks}:
458 match pair:
459 case (file, Fail{error}):
460 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
461 (file, Fail{error}))
462 case (file, Done{Bytes.Bytes{0, _}}):
463 exact.complete(file, chunks, False{})
464 case (file, Done{Bytes.Bytes{+length, words}}):
465 exact.loop(rest, Need{file,
466 Nat.sub(remaining, U32.to_nat(length)),
467 Con{Bytes.Bytes{length, words}, chunks}})
468
469def exact.nat.duplicate.step(pair: Nat & Nat) -> Nat & Nat:
470 match pair:
471 case (left, right): (Nat.add(1n, left), Nat.add(1n, right))
472
473def exact.nat.duplicate(value: Nat) -> Nat & Nat:
474 match value:
475 case 0n: (0n, 0n)
476 case 1n+rest: exact.nat.duplicate.step(exact.nat.duplicate(rest))
477
478def exact.start.copied(
479 file: File,
480 copies: Nat & Nat
481) -> IO(
482 File & Result<&1, &1, U32 & String, ExactResult>
483):
484 match copies:
485 case (fuel, needed):
486 exact.loop(Nat.add(Nat.mul(2n, fuel), 1n), Need{file, needed, Nil{}})
487
488def exact.start(
489 file: File,
490 needed: Nat
491) -> IO(
492 File & Result<&1, &1, U32 & String, ExactResult>
493):
494 exact.start.copied(file, exact.nat.duplicate(needed))
495
496def frame.error(decision: Protocol.FrameDecision) -> Error:
497 match decision:
498 case Protocol.UnknownOperation{}: UnknownOperation{}
499 case Protocol.FrameTooLarge{}: FrameTooLarge{}
500 case _: Malformed{}
501
502def frame.result(
503 file: File,
504 decision: Protocol.FrameDecision
505) -> File & Result<&1, &1, Error, Protocol.Request>:
506 match decision:
507 case Protocol.Decoded{request}: (file, Done{request})
508 case _: (file, Fail{frame.error(decision)})
509
510def payload.result(
511 header: Bytes.Bytes,
512 pair: File & Result<&1, &1, U32 & String, ExactResult>,
513 max_payload: Nat
514) -> IO(File & Result<&1, &1, Error, Protocol.Request>):
515 match pair:
516 case (file, Fail{(code, message)}):
517 IO.pure(File & Result<&1, &1, Error, Protocol.Request>,
518 (file, Fail{host.error(code, message)}))
519 case (file, Done{ExactResult{bytes, complete}}):
520 match complete:
521 case False{}:
522 IO.pure(File & Result<&1, &1, Error, Protocol.Request>,
523 (file, Fail{Truncated{}}))
524 case True{}:
525 IO.pure(File & Result<&1, &1, Error, Protocol.Request>,
526 frame.result(file, Protocol.decode(Bytes.concat([header, bytes]),
527 max_payload)))
528
529def payload.header(
530 file: File,
531 header: Bytes.Bytes,
532 length: U32,
533 max_payload: Nat
534) -> IO(File & Result<&1, &1, Error, Protocol.Request>):
535 do IO<File & Result<&1, &1, Error, Protocol.Request>>:
536 payload : File & Result<&1, &1, U32 & String, ExactResult> <-
537 exact.start(file, U32.to_nat(length))
538 payload.result(header, payload, max_payload)
539
540def header.length(
541 file: File,
542 bytes: Bytes.Bytes,
543 length: Maybe<&2, U32>,
544 max_payload: Nat
545) -> IO(File & Result<&1, &1, Error, Protocol.Request>):
546 match length:
547 case Some{length}: payload.header(file, bytes, length, max_payload)
548 case None{}:
549 IO.pure(File & Result<&1, &1, Error, Protocol.Request>,
550 (file, Fail{Malformed{}}))
551
552def header.length.result(
553 file: File,
554 pair: Bytes.Bytes & Maybe<&2, U32>,
555 max_payload: Nat
556) -> IO(File & Result<&1, &1, Error, Protocol.Request>):
557 match pair:
558 case (bytes, length): header.length(file, bytes, length, max_payload)
559
560def header.decision(
561 file: File,
562 bytes: Bytes.Bytes,
563 decision: Protocol.FrameDecision,
564 max_payload: Nat
565) -> IO(File & Result<&1, &1, Error, Protocol.Request>):
566 match decision:
567 case Protocol.NeedMore{}:
568 header.length.result(file, Bytes.get.u32be(bytes, 20), max_payload)
569 case _:
570 IO.pure(File & Result<&1, &1, Error, Protocol.Request>,
571 frame.result(file, decision))
572
573def header.cloned(
574 file: File,
575 pair: Bytes.Bytes & Bytes.Bytes,
576 +max_payload: Nat
577) -> IO(File & Result<&1, &1, Error, Protocol.Request>):
578 match pair:
579 case (for_decode, retained):
580 header.decision(file, retained,
581 Protocol.decode(for_decode, max_payload), max_payload)
582
583def header.result(
584 pair: File & Result<&1, &1, U32 & String, ExactResult>,
585 max_payload: Nat
586) -> IO(File & Result<&1, &1, Error, Protocol.Request>):
587 match pair:
588 case (file, Fail{(code, message)}):
589 IO.pure(File & Result<&1, &1, Error, Protocol.Request>,
590 (file, Fail{host.error(code, message)}))
591 case (file, Done{ExactResult{bytes, complete}}):
592 match complete:
593 case False{}:
594 IO.pure(File & Result<&1, &1, Error, Protocol.Request>,
595 (file, Fail{Truncated{}}))
596 case True{}:
597 header.cloned(file, ByteMemTable.bytes.clone(bytes), max_payload)
598
599def read_request.limited(
600 file: File,
601 max_payload: Nat
602) -> IO(
603 File & Result<&1, &1, Error, Protocol.Request>
604):
605 do IO<File & Result<&1, &1, Error, Protocol.Request>>:
606 header : File & Result<&1, &1, U32 & String, ExactResult> <-
607 exact.start(file, 24n)
608 header.result(header, max_payload)
609
610# Reads and validates one length-bounded request frame.
611def read_request(
612 file: File
613) -> IO(
614 File & Result<&1, &1, Error, Protocol.Request>
615):
616 read_request.limited(file, max_payload_bytes())
617
618def response.frame.result(
619 file: File,
620 decision: Protocol.ResponseDecision
621) -> File & Result<&1, &1, Error, Protocol.Response>:
622 match decision:
623 case Protocol.DecodedResponse{response}: (file, Done{response})
624 case Protocol.ResponseNeedMore{}: (file, Fail{Truncated{}})
625 case Protocol.ResponseTooLarge{}: (file, Fail{FrameTooLarge{}})
626 case Protocol.MalformedResponse{}: (file, Fail{MalformedResponse{}})
627 case Protocol.UnexpectedResponse{}: (file, Fail{UnexpectedResponse{}})
628 case Protocol.UnknownResponseStatus{}:
629 (file, Fail{UnknownResponseStatus{}})
630
631def response.payload.result(
632 header: Bytes.Bytes,
633 pair: File & Result<&1, &1, U32 & String, ExactResult>
634) -> IO(File & Result<&1, &1, Error, Protocol.Response>):
635 match pair:
636 case (file, Fail{(code, message)}):
637 IO.pure(File & Result<&1, &1, Error, Protocol.Response>,
638 (file, Fail{host.error(code, message)}))
639 case (file, Done{ExactResult{bytes, complete}}):
640 match complete:
641 case False{}:
642 IO.pure(File & Result<&1, &1, Error, Protocol.Response>,
643 (file, Fail{Truncated{}}))
644 case True{}:
645 IO.pure(File & Result<&1, &1, Error, Protocol.Response>,
646 response.frame.result(file,
647 Protocol.decode_response(Bytes.concat([header, bytes]),
648 max_payload_bytes())))
649
650def response.payload(
651 file: File,
652 header: Bytes.Bytes,
653 length: U32
654) -> IO(File & Result<&1, &1, Error, Protocol.Response>):
655 do IO<File & Result<&1, &1, Error, Protocol.Response>>:
656 payload : File & Result<&1, &1, U32 & String, ExactResult> <-
657 exact.start(file, U32.to_nat(length))
658 response.payload.result(header, payload)
659
660def response.length.admitted(
661 file: File,
662 header: Bytes.Bytes,
663 length: U32,
664 admitted: Bool
665) -> IO(File & Result<&1, &1, Error, Protocol.Response>):
666 match admitted:
667 case True{}: response.payload(file, header, length)
668 case False{}:
669 IO.pure(File & Result<&1, &1, Error, Protocol.Response>,
670 (file, Fail{FrameTooLarge{}}))
671
672def response.length.checked(
673 file: File,
674 header: Bytes.Bytes,
675 lengths: U32 & U32 & U32
676) -> IO(File & Result<&1, &1, Error, Protocol.Response>):
677 match lengths:
678 case (read_length, (checked_length, _)):
679 response.length.admitted(file, header, read_length,
680 U32.is_le(checked_length, U32.from_nat(max_payload_bytes())))
681
682def response.length(
683 file: File,
684 header: Bytes.Bytes,
685 length: Maybe<&2, U32>
686) -> IO(File & Result<&1, &1, Error, Protocol.Response>):
687 match length:
688 case Some{length}:
689 response.length.checked(file, header,
690 Protocol.u32.duplicate.triple(length))
691 case _:
692 IO.pure(File & Result<&1, &1, Error, Protocol.Response>,
693 (file, Fail{MalformedResponse{}}))
694
695def response.header.length(
696 file: File,
697 pair: Bytes.Bytes & Maybe<&2, U32>
698) -> IO(File & Result<&1, &1, Error, Protocol.Response>):
699 match pair:
700 case (header, length): response.length(file, header, length)
701
702def response.header.complete(
703 file: File,
704 header: Bytes.Bytes,
705 complete: Bool
706) -> IO(File & Result<&1, &1, Error, Protocol.Response>):
707 match complete:
708 case True{}:
709 response.header.length(file, Bytes.get.u32be(header, 20))
710 case False{}:
711 IO.pure(File & Result<&1, &1, Error, Protocol.Response>,
712 (file, Fail{Truncated{}}))
713
714def response.header.result(
715 pair: File & Result<&1, &1, U32 & String, ExactResult>
716) -> IO(File & Result<&1, &1, Error, Protocol.Response>):
717 match pair:
718 case (file, Fail{(code, message)}):
719 IO.pure(File & Result<&1, &1, Error, Protocol.Response>,
720 (file, Fail{host.error(code, message)}))
721 case (file, Done{ExactResult{bytes, complete}}):
722 response.header.complete(file, bytes, complete)
723
724# Reads and validates one length-bounded response frame.
725def read_response(
726 file: File
727) -> IO(File & Result<&1, &1, Error, Protocol.Response>):
728 do IO<File & Result<&1, &1, Error, Protocol.Response>>:
729 header : File & Result<&1, &1, U32 & String, ExactResult> <-
730 exact.start(file, 24n)
731 response.header.result(header)