src/coordinator/Protocol.bend
mylsm-lsm-store@0.6.0.0 · 781 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/ByteMemTable.bend as ByteMemTable
3import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
4
5# Returns the maximum encoded request or response payload in bytes.
6def max_payload_bytes() -> Nat:
7 1073741824n
8
9# Admits a wire payload only when it fits the shared protocol ceiling.
10def payload.admitted(length: U32) -> Bool:
11 U32.is_le(length, 1073741824)
12
13# Carries one decoded request with its protocol and owner generation.
14type 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.
19type 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.
27type FrameDecision is Type:
28 NeedMore{}
29 Malformed{}
30 UnknownOperation{}
31 FrameTooLarge{}
32 Decoded{request: Request}
33
34# Carries the state for HeaderDecision.
35type 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.
43type 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.
59type 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.
64type 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.
73type ResponseStatusDecision is Data:
74 ValidResponseStatus{status: ResponseStatus}
75 InvalidResponseStatus{}
76
77def u32.bytes(value: U32) -> Bytes.Bytes:
78 Bytes.set.u32be(Bytes.new(4), 0, value)
79
80def snapshot.id.bytes(snapshot_id: U32) -> Bytes.Bytes:
81 u32.bytes(snapshot_id)
82
83def snapshot.id.read.parsed(
84 pair: Bytes.Bytes & Maybe<&2, U32>
85) -> Maybe<&2, U32>:
86 match pair:
87 case (_, parsed): parsed
88
89def 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
97def snapshot.id.read(payload: Bytes.Bytes) -> Maybe<&2, U32>:
98 snapshot.id.read.measured(Bytes.length(payload))
99
100def 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.
106def magic() -> Bytes.Bytes:
107 magic.from_hex(Bytes.from_hex("4d4c5336"))
108
109def 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.
122def 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
128def 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
144def 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
161def 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
183def 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
198def 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.
206def 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
212def 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
223def 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
238def 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
254def 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.
272def nat.duplicate.pair(+value: Nat) -> Nat & Nat:
273 (value, value)
274
275# Copies a fixed-width protocol word without converting it to unary recursion.
276def u32.duplicate.triple(+value: U32) -> U32 & U32 & U32:
277 (value, value, value)
278
279def 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
298def 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
315def 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.
331def 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
349def 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
365def 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
381def 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
401def 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
420def 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
436def 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
450def 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
464def 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
477def 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
489def 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
500def 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
512def 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
522def 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
532def 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
543def 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
551def 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
559def 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.
563def decode(input: Bytes.Bytes, max_payload_bytes: Nat) -> FrameDecision:
564 decode.header(input, max_payload_bytes)
565
566def 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
583def 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
588def 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
597def 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
611def 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
622def 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
631def header.inspect(input: Bytes.Bytes, max_payload_bytes: Nat) -> HeaderDecision:
632 header.inspect.size(Bytes.length(input), max_payload_bytes)
633
634def 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
645def 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
657def 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
670def 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
682def 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
695def 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
706def 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
718def 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.
728def decode_response(input: Bytes.Bytes, max_payload_bytes: Nat) -> ResponseDecision:
729 decode.response.frame(decode(input, max_payload_bytes))
730
731def 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
753def 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.
769def 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)