1	import Base
2	import ../ByteMemTable.bend as ByteMemTable
3	import ./Protocol.bend as Protocol
4	import ./IoPolicy.bend as IoPolicy
5	import ./UnixSocketPolicy.bend as UnixSocketPolicy
6	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7	
8	# Classifies framing, resource, and host failures from socket IO.
9	type 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.
22	type ExactResult is Type:
23	  ExactResult{bytes: Bytes.Bytes, complete: Bool}
24	
25	# Carries the state for ReadAttempt.
26	type ReadAttempt is Type:
27	  ReadAvailable{bytes: Bytes.Bytes}
28	  ReadInterrupted{}
29	  ReadWouldBlock{}
30	  ReadFailed{error: Error}
31	
32	# Carries the state for WriteAttempt.
33	type 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.
40	type 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.
46	type 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.
61	def max_payload_bytes() -> Nat:
62	  Protocol.max_payload_bytes()
63	
64	def 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	
74	def 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	
83	def 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.
87	def 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	
96	def 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	
104	def 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	
111	def 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	
117	def 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	
128	def 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	
140	def 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	
149	def 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	
156	def 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.
166	def 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	
176	def 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	
191	def 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	
201	def 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	
213	def 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	
225	def 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	
235	def 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	
243	def write.chunk(remaining: U32) -> U32:
244	  IoPolicy.write_chunk(remaining)
245	
246	# Writes one raw byte chunk to a local socket connection.
247	def 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	
256	def 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	
269	def 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	
286	def 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	
305	def 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	
373	def 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	
386	def 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	
396	def 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.
404	def 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	
414	def 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	
423	def 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	
469	def 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	
473	def 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	
478	def 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	
488	def 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	
496	def frame.error(decision: Protocol.FrameDecision) -> Error:
497	  match decision:
498	    case Protocol.UnknownOperation{}: UnknownOperation{}
499	    case Protocol.FrameTooLarge{}: FrameTooLarge{}
500	    case _: Malformed{}
501	
502	def 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	
510	def 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	
529	def 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	
540	def 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	
552	def 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	
560	def 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	
573	def 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	
583	def 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	
599	def 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.
611	def 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	
618	def 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	
631	def 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	
650	def 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	
660	def 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	
672	def 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	
682	def 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	
695	def 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	
702	def 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	
714	def 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.
725	def 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)