1	import Base
2	import ../ByteMemTable.bend as ByteMemTable
3	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
4	
5	# Returns the maximum encoded request or response payload in bytes.
6	def max_payload_bytes() -> Nat:
7	  1073741824n
8	
9	# Admits a wire payload only when it fits the shared protocol ceiling.
10	def payload.admitted(length: U32) -> Bool:
11	  U32.is_le(length, 1073741824)
12	
13	# Carries one decoded request with its protocol and owner generation.
14	type Request is Type:
15	  Request{version: U32, generation: U32, request_id: U32, operation: U32,
16	    payload: Bytes.Bytes}
17	
18	# Reports whether a request passes the pure protocol admission checks.
19	type Decision is Type:
20	  Accepted{request_id: U32, operation: U32, payload: Bytes.Bytes}
21	  VersionMismatch{}
22	  StaleGeneration{}
23	  UnsupportedRequestOperation{}
24	  ResourceLimit{}
25	
26	# Separates incomplete, malformed, oversized, unsupported, and decoded frames.
27	type FrameDecision is Type:
28	  NeedMore{}
29	  Malformed{}
30	  UnknownOperation{}
31	  FrameTooLarge{}
32	  Decoded{request: Request}
33	
34	# Carries the state for HeaderDecision.
35	type HeaderDecision is Data:
36	  HeaderIncomplete{}
37	  HeaderReady{frame_bytes: U32}
38	  HeaderMalformed{}
39	  HeaderUnknownOperation{}
40	  HeaderTooLarge{}
41	
42	# Defines response outcomes carried independently from operation payloads.
43	type ResponseStatus is Data:
44	  Success{}
45	  InvalidArgument{}
46	  NotFound{}
47	  Busy{}
48	  Timeout{}
49	  ResponseResourceLimit{}
50	  Corruption{}
51	  IoFailure{}
52	  CommitUnknown{}
53	  AlreadyExists{}
54	  UnsupportedPlatform{}
55	  InvalidUtf8{}
56	  UnsupportedFormat{}
57	
58	# Carries one correlated outcome and an optional raw detail payload.
59	type Response is Type:
60	  Response{version: U32, generation: U32, request_id: U32,
61	    status: ResponseStatus, payload: Bytes.Bytes}
62	
63	# Reports incremental framing and typed response-decoding outcomes.
64	type ResponseDecision is Type:
65	  ResponseNeedMore{}
66	  MalformedResponse{}
67	  ResponseTooLarge{}
68	  UnexpectedResponse{}
69	  UnknownResponseStatus{}
70	  DecodedResponse{response: Response}
71	
72	# Maps a numeric response code to a supported public status.
73	type ResponseStatusDecision is Data:
74	  ValidResponseStatus{status: ResponseStatus}
75	  InvalidResponseStatus{}
76	
77	def u32.bytes(value: U32) -> Bytes.Bytes:
78	  Bytes.set.u32be(Bytes.new(4), 0, value)
79	
80	def snapshot.id.bytes(snapshot_id: U32) -> Bytes.Bytes:
81	  u32.bytes(snapshot_id)
82	
83	def snapshot.id.read.parsed(
84	  pair: Bytes.Bytes & Maybe<&2, U32>
85	) -> Maybe<&2, U32>:
86	  match pair:
87	    case (_, parsed): parsed
88	
89	def snapshot.id.read.measured(
90	  measured: Bytes.Bytes & U32
91	) -> Maybe<&2, U32>:
92	  match measured:
93	    case (payload, 4):
94	      snapshot.id.read.parsed(Bytes.get.u32be(payload, 0))
95	    case _: None{}
96	
97	def snapshot.id.read(payload: Bytes.Bytes) -> Maybe<&2, U32>:
98	  snapshot.id.read.measured(Bytes.length(payload))
99	
100	def magic.from_hex(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
101	  match value:
102	    case Some{bytes}: bytes
103	    case None{}: Bytes.new(0)
104	
105	# Returns the fixed identifier for versioned coordinator request frames.
106	def magic() -> Bytes.Bytes:
107	  magic.from_hex(Bytes.from_hex("4d4c5336"))
108	
109	def encode.fields(
110	  +version: U32,
111	  +generation: U32,
112	  +request_id: U32,
113	  +operation: U32,
114	  measured: Bytes.Bytes & U32
115	) -> Bytes.Bytes:
116	  match measured:
117	    case (payload, +length):
118	      Bytes.concat([magic(), u32.bytes(version), u32.bytes(generation),
119	        u32.bytes(request_id), u32.bytes(operation), u32.bytes(length), payload])
120	
121	# Encodes one request with an exact payload length and operation tag.
122	def encode(request: Request) -> Bytes.Bytes:
123	  match request:
124	    case Request{+version, +generation, +request_id, +operation, payload}:
125	      encode.fields(version, generation, request_id, operation,
126	        Bytes.length(payload))
127	
128	def response.status.code(status: ResponseStatus) -> U32:
129	  match status:
130	    case Success{}: 0
131	    case InvalidArgument{}: 1
132	    case NotFound{}: 2
133	    case Busy{}: 3
134	    case Timeout{}: 4
135	    case ResponseResourceLimit{}: 5
136	    case Corruption{}: 6
137	    case IoFailure{}: 7
138	    case CommitUnknown{}: 8
139	    case AlreadyExists{}: 9
140	    case UnsupportedPlatform{}: 10
141	    case InvalidUtf8{}: 11
142	    case UnsupportedFormat{}: 12
143	
144	def response.status.decode(code: U32) -> ResponseStatusDecision:
145	  match code:
146	    case 0: ValidResponseStatus{Success{}}
147	    case 1: ValidResponseStatus{InvalidArgument{}}
148	    case 2: ValidResponseStatus{NotFound{}}
149	    case 3: ValidResponseStatus{Busy{}}
150	    case 4: ValidResponseStatus{Timeout{}}
151	    case 5: ValidResponseStatus{ResponseResourceLimit{}}
152	    case 6: ValidResponseStatus{Corruption{}}
153	    case 7: ValidResponseStatus{IoFailure{}}
154	    case 8: ValidResponseStatus{CommitUnknown{}}
155	    case 9: ValidResponseStatus{AlreadyExists{}}
156	    case 10: ValidResponseStatus{UnsupportedPlatform{}}
157	    case 11: ValidResponseStatus{InvalidUtf8{}}
158	    case 12: ValidResponseStatus{UnsupportedFormat{}}
159	    case _: InvalidResponseStatus{}
160	
161	def response.encode.header(
162	  buffer: Array<U32>,
163	  version: U32,
164	  generation: U32,
165	  request_id: U32,
166	  status_code: U32,
167	  payload_length: U32
168	) -> Array<U32>:
169	  Bytes.put.go(4n,
170	    Bytes.put.go(4n,
171	      Bytes.put.go(4n,
172	        Bytes.put.go(4n,
173	          Bytes.put.go(4n,
174	            Bytes.put.go(4n,
175	              Bytes.put.go(4n, buffer, 3, False{}, 1296847670),
176	              7, False{}, version),
177	            11, False{}, generation),
178	          15, False{}, request_id),
179	        19, False{}, 7),
180	      23, False{}, U32.add(payload_length, 4)),
181	    27, False{}, status_code)
182	
183	def response.encode.payload(
184	  version: U32,
185	  generation: U32,
186	  request_id: U32,
187	  status_code: U32,
188	  payload: Bytes.Bytes
189	) -> Bytes.Bytes:
190	  match payload:
191	    case Bytes.Bytes{+length, buffer}:
192	      Bytes.Bytes{U32.add(length, 28),
193	        response.encode.header(
194	          Bytes.dst(Bytes.copy(length, buffer,
195	            Bytes.alloc(U32.add(length, 28)), 0, 28)),
196	          version, generation, request_id, status_code, length)}
197	
198	def header.control(header: Bytes.Bytes, status_code: U32) -> Bytes.Bytes:
199	  match header:
200	    case Bytes.Bytes{24, buffer}:
201	      Bytes.set.u32be(Bytes.set.u32be(Bytes.set.u32be(
202	        Bytes.Bytes{28, buffer}, 16, 7), 20, 4), 24, status_code)
203	    case _: Bytes.new(0)
204	
205	# Encodes a typed response status before its raw detail bytes.
206	def encode_response(response: Response) -> Bytes.Bytes:
207	  match response:
208	    case Response{version, generation, request_id, status, payload}:
209	      response.encode.payload(version, generation, request_id,
210	        response.status.code(status), payload)
211	
212	def decode.complete.payload(
213	  +version: U32,
214	  +generation: U32,
215	  +request_id: U32,
216	  +operation: U32,
217	  sliced: Bytes.Bytes & Bytes.Bytes
218	) -> FrameDecision:
219	  match sliced:
220	    case (_, payload):
221	      Decoded{Request{version, generation, request_id, operation, payload}}
222	
223	def decode.complete(
224	  input: Bytes.Bytes,
225	  +version: U32,
226	  +generation: U32,
227	  +request_id: U32,
228	  +operation: U32,
229	  +payload_length: U32,
230	  exact: Bool
231	) -> FrameDecision:
232	  match exact:
233	    case False{}: Malformed{}
234	    case True{}:
235	      decode.complete.payload(version, generation, request_id,
236	        operation, Bytes.slice(input, 24, payload_length))
237	
238	def decode.complete.classified(
239	  input: Bytes.Bytes,
240	  +version: U32,
241	  +generation: U32,
242	  +request_id: U32,
243	  +operation: U32,
244	  +payload_length: U32,
245	  exact: Bool,
246	  short: Bool
247	) -> FrameDecision:
248	  match short:
249	    case True{}: NeedMore{}
250	    case False{}:
251	      decode.complete(input, version, generation, request_id, operation,
252	        payload_length, exact)
253	
254	def decode.sized(
255	  input: Bytes.Bytes,
256	  +version: U32,
257	  +generation: U32,
258	  +request_id: U32,
259	  +operation: U32,
260	  +payload_length: U32,
261	  exact: Bool,
262	  short: Bool,
263	  within_limit: Bool
264	) -> FrameDecision:
265	  match within_limit:
266	    case False{}: FrameTooLarge{}
267	    case True{}:
268	      decode.complete.classified(input, version, generation, request_id,
269	        operation, payload_length, exact, short)
270	
271	# Copies a natural directly without recursion on its numeric value.
272	def nat.duplicate.pair(+value: Nat) -> Nat & Nat:
273	  (value, value)
274	
275	# Copies a fixed-width protocol word without converting it to unary recursion.
276	def u32.duplicate.triple(+value: U32) -> U32 & U32 & U32:
277	  (value, value, value)
278	
279	def decode.words.payload_length.size.copies(
280	  input: Bytes.Bytes,
281	  +version: U32,
282	  +generation: U32,
283	  +request_id: U32,
284	  +operation: U32,
285	  +payload_length: U32,
286	  exact_expected: Nat,
287	  short_expected: Nat,
288	  within_limit: Bool,
289	  sizes: Nat & Nat
290	) -> FrameDecision:
291	  match sizes:
292	    case (exact_size, short_size):
293	      decode.sized(input, version, generation, request_id, operation,
294	        payload_length,
295	        Nat.is_eq(exact_size, exact_expected),
296	        Nat.is_lt(short_size, short_expected), within_limit)
297	
298	def decode.words.payload_length.expected.copies(
299	  input: Bytes.Bytes,
300	  +size: U32,
301	  +version: U32,
302	  +generation: U32,
303	  +request_id: U32,
304	  +operation: U32,
305	  +payload_length: U32,
306	  expected: Nat & Nat,
307	  within_limit: Bool
308	) -> FrameDecision:
309	  match expected:
310	    case (exact_expected, short_expected):
311	      decode.words.payload_length.size.copies(input, version,
312	        generation, request_id, operation, payload_length, exact_expected, short_expected,
313	        within_limit, nat.duplicate.pair(U32.to_nat(size)))
314	
315	def decode.words.payload_length.expected(
316	  input: Bytes.Bytes,
317	  +size: U32,
318	  +version: U32,
319	  +generation: U32,
320	  +request_id: U32,
321	  +operation: U32,
322	  +payload_length: U32,
323	  expected: Nat,
324	  within_limit: Bool
325	) -> FrameDecision:
326	  decode.words.payload_length.expected.copies(input, size, version,
327	    generation, request_id, operation, payload_length, nat.duplicate.pair(expected),
328	    within_limit)
329	
330	# Rejects oversized declarations before computing or comparing frame offsets.
331	def decode.words.payload_length.admitted(
332	  input: Bytes.Bytes,
333	  +size: U32,
334	  +version: U32,
335	  +generation: U32,
336	  +request_id: U32,
337	  +operation: U32,
338	  +payload_length: U32,
339	  exact_length: U32,
340	  within_limit: Bool
341	) -> FrameDecision:
342	  match within_limit:
343	    case False{}: FrameTooLarge{}
344	    case True{}:
345	      decode.words.payload_length.expected(input, size, version, generation,
346	        request_id, operation, payload_length,
347	        Nat.add(24n, U32.to_nat(exact_length)), True{})
348	
349	def decode.words.payload_length.copies(
350	  input: Bytes.Bytes,
351	  +size: U32,
352	  +version: U32,
353	  +generation: U32,
354	  +request_id: U32,
355	  +operation: U32,
356	  copies: U32 & U32 & U32,
357	  max_payload_bytes: Nat
358	) -> FrameDecision:
359	  match copies:
360	    case (frame_length, exact_length, bounded_length):
361	      decode.words.payload_length.admitted(input, size, version, generation,
362	        request_id, operation, frame_length, exact_length,
363	        U32.is_le(bounded_length, U32.from_nat(max_payload_bytes)))
364	
365	def decode.words.payload_length(
366	  pair: Bytes.Bytes & Maybe<&2, U32>,
367	  +size: U32,
368	  max_payload_bytes: Nat,
369	  +version: U32,
370	  +generation: U32,
371	  +request_id: U32,
372	  +operation: U32
373	) -> FrameDecision:
374	  match pair:
375	    case (input, Some{payload_length}):
376	      decode.words.payload_length.copies(input, size, version, generation,
377	        request_id, operation, u32.duplicate.triple(payload_length),
378	        max_payload_bytes)
379	    case _: Malformed{}
380	
381	def operation.valid(code: U32) -> Bool:
382	  match code:
383	    case 1: True{}
384	    case 2: True{}
385	    case 3: True{}
386	    case 4: True{}
387	    case 5: True{}
388	    case 6: True{}
389	    case 7: True{}
390	    case 8: True{}
391	    case 9: True{}
392	    case 10: True{}
393	    case 11: True{}
394	    case 12: True{}
395	    case 13: True{}
396	    case 14: True{}
397	    case 15: True{}
398	    case 16: True{}
399	    case _: False{}
400	
401	def operation.request.valid(code: U32) -> Bool:
402	  match code:
403	    case 1: True{}
404	    case 2: True{}
405	    case 3: True{}
406	    case 4: True{}
407	    case 5: True{}
408	    case 6: True{}
409	    case 8: True{}
410	    case 9: True{}
411	    case 10: True{}
412	    case 11: True{}
413	    case 12: True{}
414	    case 13: True{}
415	    case 14: True{}
416	    case 15: True{}
417	    case 16: True{}
418	    case _: False{}
419	
420	def decode.words.operation.validated(
421	  input: Bytes.Bytes,
422	  +size: U32,
423	  +version: U32,
424	  +generation: U32,
425	  +request_id: U32,
426	  +operation: U32,
427	  valid: Bool,
428	  max_payload_bytes: Nat
429	) -> FrameDecision:
430	  match valid:
431	    case False{}: UnknownOperation{}
432	    case True{}:
433	      decode.words.payload_length(Bytes.get.u32be(input, 20), size,
434	        max_payload_bytes, version, generation, request_id, operation)
435	
436	def decode.words.operation.checked(
437	  input: Bytes.Bytes,
438	  +size: U32,
439	  +version: U32,
440	  +generation: U32,
441	  +request_id: U32,
442	  codes: U32 & U32 & U32,
443	  max_payload_bytes: Nat
444	) -> FrameDecision:
445	  match codes:
446	    case (operation, code, _):
447	      decode.words.operation.validated(input, size, version, generation,
448	        request_id, operation, operation.valid(code), max_payload_bytes)
449	
450	def decode.words.operation(
451	  pair: Bytes.Bytes & Maybe<&2, U32>,
452	  +size: U32,
453	  max_payload_bytes: Nat,
454	  +version: U32,
455	  +generation: U32,
456	  +request_id: U32
457	) -> FrameDecision:
458	  match pair:
459	    case (input, Some{operation}):
460	      decode.words.operation.checked(input, size, version, generation,
461	        request_id, u32.duplicate.triple(operation), max_payload_bytes)
462	    case _: Malformed{}
463	
464	def decode.words.request_id(
465	  pair: Bytes.Bytes & Maybe<&2, U32>,
466	  +size: U32,
467	  max_payload_bytes: Nat,
468	  +version: U32,
469	  +generation: U32
470	) -> FrameDecision:
471	  match pair:
472	    case (input, Some{request_id}):
473	      decode.words.operation(Bytes.get.u32be(input, 16), size,
474	        max_payload_bytes, version, generation, request_id)
475	    case _: Malformed{}
476	
477	def decode.words.generation(
478	  pair: Bytes.Bytes & Maybe<&2, U32>,
479	  +size: U32,
480	  max_payload_bytes: Nat,
481	  +version: U32
482	) -> FrameDecision:
483	  match pair:
484	    case (input, Some{generation}):
485	      decode.words.request_id(Bytes.get.u32be(input, 12), size,
486	        max_payload_bytes, version, generation)
487	    case _: Malformed{}
488	
489	def decode.words.version(
490	  pair: Bytes.Bytes & Maybe<&2, U32>,
491	  +size: U32,
492	  max_payload_bytes: Nat
493	) -> FrameDecision:
494	  match pair:
495	    case (input, Some{version}):
496	      decode.words.generation(Bytes.get.u32be(input, 8), size,
497	        max_payload_bytes, version)
498	    case _: Malformed{}
499	
500	def decode.magic.checked(
501	  input: Bytes.Bytes,
502	  +size: U32,
503	  max_payload_bytes: Nat,
504	  valid_magic: Bool
505	) -> FrameDecision:
506	  match valid_magic:
507	    case False{}: Malformed{}
508	    case True{}:
509	      decode.words.version(Bytes.get.u32be(input, 4), size,
510	        max_payload_bytes)
511	
512	def decode.magic.comparison(
513	  comparison: Bytes.Bytes & Bytes.Bytes & Bool,
514	  input: Bytes.Bytes,
515	  +size: U32,
516	  max_payload_bytes: Nat
517	) -> FrameDecision:
518	  match comparison:
519	    case (_, _, valid):
520	      decode.magic.checked(input, size, max_payload_bytes, valid)
521	
522	def decode.magic.slice(
523	  sliced: Bytes.Bytes & Bytes.Bytes,
524	  +size: U32,
525	  max_payload_bytes: Nat
526	) -> FrameDecision:
527	  match sliced:
528	    case (input, candidate):
529	      decode.magic.comparison(Bytes.eq(magic(), candidate), input, size,
530	        max_payload_bytes)
531	
532	def decode.available(
533	  input: Bytes.Bytes,
534	  +size: U32,
535	  max_payload_bytes: Nat,
536	  enough_header: Bool
537	) -> FrameDecision:
538	  match enough_header:
539	    case False{}: NeedMore{}
540	    case True{}:
541	      decode.magic.slice(Bytes.slice(input, 0, 4), size, max_payload_bytes)
542	
543	def decode.header.size(
544	  input: Bytes.Bytes,
545	  max_payload_bytes: Nat,
546	  +size: U32
547	) -> FrameDecision:
548	  decode.available(input, size, max_payload_bytes,
549	    Nat.is_le(24n, U32.to_nat(size)))
550	
551	def decode.header.pair(
552	  pair: Bytes.Bytes & U32,
553	  max_payload_bytes: Nat
554	) -> FrameDecision:
555	  match pair:
556	    case (input, size):
557	      decode.header.size(input, max_payload_bytes, size)
558	
559	def decode.header(input: Bytes.Bytes, max_payload_bytes: Nat) -> FrameDecision:
560	  decode.header.pair(Bytes.length(input), max_payload_bytes)
561	
562	# Decodes one complete frame or reports whether more input is required.
563	def decode(input: Bytes.Bytes, max_payload_bytes: Nat) -> FrameDecision:
564	  decode.header(input, max_payload_bytes)
565	
566	def header.classify(
567	  decision: FrameDecision,
568	  frame_bytes: Maybe<&2, U32>
569	) -> HeaderDecision:
570	  match decision:
571	    case NeedMore{}:
572	      match frame_bytes:
573	        case Some{length}: HeaderReady{length}
574	        case _: HeaderMalformed{}
575	    case Decoded{_}:
576	      match frame_bytes:
577	        case Some{length}: HeaderReady{length}
578	        case _: HeaderMalformed{}
579	    case Malformed{}: HeaderMalformed{}
580	    case UnknownOperation{}: HeaderUnknownOperation{}
581	    case FrameTooLarge{}: HeaderTooLarge{}
582	
583	def header.frame.bytes(pair: Bytes.Bytes & Maybe<&2, U32>) -> Maybe<&2, U32>:
584	  match pair:
585	    case (_, Some{payload_bytes}): Some{U32.add(24, payload_bytes)}
586	    case (_, None{}): None{}
587	
588	def header.inspect.cloned(
589	  pair: Bytes.Bytes & Bytes.Bytes,
590	  max_payload_bytes: Nat
591	) -> HeaderDecision:
592	  match pair:
593	    case (header, validation):
594	      header.classify(decode(validation, max_payload_bytes),
595	        header.frame.bytes(Bytes.get.u32be(header, 20)))
596	
597	def header.inspect.sized(
598	  input: Bytes.Bytes,
599	  max_payload_bytes: Nat,
600	  enough_header: Bool,
601	  exact_header: Bool
602	) -> HeaderDecision:
603	  match enough_header:
604	    case False{}: HeaderIncomplete{}
605	    case True{}:
606	      match exact_header:
607	        case False{}: HeaderMalformed{}
608	        case True{}:
609	          header.inspect.cloned(ByteMemTable.bytes.clone(input), max_payload_bytes)
610	
611	def header.inspect.sized.copies(
612	  input: Bytes.Bytes,
613	  sizes: U32 & U32 & U32,
614	  max_payload_bytes: Nat
615	) -> HeaderDecision:
616	  match sizes:
617	    case (size, (checked_size, _)):
618	      header.inspect.sized(input, max_payload_bytes,
619	        Nat.is_le(24n, U32.to_nat(size)),
620	        Nat.is_eq(24n, U32.to_nat(checked_size)))
621	
622	def header.inspect.size(
623	  pair: Bytes.Bytes & U32,
624	  max_payload_bytes: Nat
625	) -> HeaderDecision:
626	  match pair:
627	    case (input, size):
628	      header.inspect.sized.copies(input, u32.duplicate.triple(size),
629	        max_payload_bytes)
630	
631	def header.inspect(input: Bytes.Bytes, max_payload_bytes: Nat) -> HeaderDecision:
632	  header.inspect.size(Bytes.length(input), max_payload_bytes)
633	
634	def decode.response.payload.slice(
635	  sliced: Bytes.Bytes & Bytes.Bytes,
636	  +version: U32,
637	  +generation: U32,
638	  +request_id: U32,
639	  status: ResponseStatus
640	) -> ResponseDecision:
641	  match sliced:
642	    case (_, payload):
643	      DecodedResponse{Response{version, generation, request_id, status, payload}}
644	
645	def decode.response.payload(
646	  pair: Bytes.Bytes & U32,
647	  +version: U32,
648	  +generation: U32,
649	  +request_id: U32,
650	  status: ResponseStatus
651	) -> ResponseDecision:
652	  match pair:
653	    case (payload, +length):
654	      decode.response.payload.slice(Bytes.slice(payload, 4, (length - 4 : U32)),
655	        version, generation, request_id, status)
656	
657	def decode.response.status.ready(
658	  input: Bytes.Bytes,
659	  +version: U32,
660	  +generation: U32,
661	  +request_id: U32,
662	  result: ResponseStatusDecision
663	) -> ResponseDecision:
664	  match result:
665	    case InvalidResponseStatus{}: UnknownResponseStatus{}
666	    case ValidResponseStatus{status}:
667	      decode.response.payload(Bytes.length(input), version, generation,
668	        request_id, status)
669	
670	def decode.response.status(
671	  pair: Bytes.Bytes & Maybe<&2, U32>,
672	  +version: U32,
673	  +generation: U32,
674	  +request_id: U32
675	) -> ResponseDecision:
676	  match pair:
677	    case (input, Some{code}):
678	      decode.response.status.ready(input, version, generation, request_id,
679	        response.status.decode(code))
680	    case _: MalformedResponse{}
681	
682	def decode.response.minimum(
683	  input: Bytes.Bytes,
684	  +version: U32,
685	  +generation: U32,
686	  +request_id: U32,
687	  enough_status: Bool
688	) -> ResponseDecision:
689	  match enough_status:
690	    case False{}: MalformedResponse{}
691	    case True{}:
692	      decode.response.status(Bytes.get.u32be(input, 0), version, generation,
693	        request_id)
694	
695	def decode.response.size(
696	  pair: Bytes.Bytes & U32,
697	  +version: U32,
698	  +generation: U32,
699	  +request_id: U32
700	) -> ResponseDecision:
701	  match pair:
702	    case (input, +size):
703	      decode.response.minimum(input, version, generation, request_id,
704	        Nat.is_le(4n, U32.to_nat(size)))
705	
706	def decode.response.operation(
707	  +version: U32,
708	  +generation: U32,
709	  +request_id: U32,
710	  +operation: U32,
711	  payload: Bytes.Bytes
712	) -> ResponseDecision:
713	  match operation:
714	    case 7:
715	      decode.response.size(Bytes.length(payload), version, generation, request_id)
716	    case _: UnexpectedResponse{}
717	
718	def decode.response.frame(decision: FrameDecision) -> ResponseDecision:
719	  match decision:
720	    case NeedMore{}: ResponseNeedMore{}
721	    case Malformed{}: MalformedResponse{}
722	    case UnknownOperation{}: UnexpectedResponse{}
723	    case FrameTooLarge{}: ResponseTooLarge{}
724	    case Decoded{Request{version, generation, request_id, operation, payload}}:
725	      decode.response.operation(version, generation, request_id, operation, payload)
726	
727	# Decodes a response only after validating its correlated outer frame.
728	def decode_response(input: Bytes.Bytes, max_payload_bytes: Nat) -> ResponseDecision:
729	  decode.response.frame(decode(input, max_payload_bytes))
730	
731	def validate.policy(
732	  request_id: U32,
733	  operation: U32,
734	  payload: Bytes.Bytes,
735	  version_valid: Bool,
736	  generation_valid: Bool,
737	  operation_valid: Bool,
738	  size_valid: Bool
739	) -> Decision:
740	  match version_valid:
741	    case False{}: VersionMismatch{}
742	    case True{}:
743	      match generation_valid:
744	        case False{}: StaleGeneration{}
745	        case True{}:
746	          match operation_valid:
747	            case False{}: UnsupportedRequestOperation{}
748	            case True{}:
749	              match size_valid:
750	                case False{}: ResourceLimit{}
751	                case True{}: Accepted{request_id, operation, payload}
752	
753	def validate.sized(
754	  request_id: U32,
755	  operation: U32,
756	  version_valid: Bool,
757	  generation_valid: Bool,
758	  operation_valid: Bool,
759	  sizes: Bytes.Bytes & U32,
760	  max_payload_bytes: Nat
761	) -> Decision:
762	  match sizes:
763	    case (request_payload, +payload_length):
764	      validate.policy(request_id, operation, request_payload, version_valid,
765	        generation_valid, operation_valid,
766	        U32.is_le(payload_length, U32.from_nat(max_payload_bytes)))
767	
768	# Validates protocol identity, active generation, and payload size.
769	def validate(
770	  request: Request,
771	  expected_version: U32,
772	  active_generation: U32,
773	  max_payload_bytes: Nat
774	) -> Decision:
775	  match request:
776	    case Request{+version, +generation, +request_id, +operation, payload}:
777	      validate.sized(request_id, operation,
778	        U32.is_eq(version, expected_version),
779	        U32.is_eq(generation, active_generation),
780	        operation.request.valid(operation),
781	        Bytes.length(payload), max_payload_bytes)