src/coordinator/OptionsCodec.bend
mylsm-lsm-store@0.6.0.1 · 263 lines · 41 definitions
1import Base
2import mylsm-lsm-store@0.6.0.1/src/Keys.bend as Keys
3import mylsm-lsm-store@0.6.0.1/src/ResourceOptions.bend as ResourceOptions
4import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
5import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
6
7
8type Decision is Data:
9 Parsed{options: ResourceOptions.Options}
10 Malformed{}
11 UnsupportedVersion{version: U32}
12 InvalidOptions{}
13
14
15def word(value: U32) -> Bytes.Bytes:
16 Bytes.set.u32be(Bytes.new(4), 0, value)
17
18def magic.value(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
19 match value:
20 case Some{bytes}: bytes
21 case None{}: Bytes.new(4)
22
23
24def magic() -> Bytes.Bytes:
25 magic.value(Bytes.from_hex("4d4c4f36"))
26
27def encode.transport(limits: ResourceOptions.TransportLimits) -> Bytes.Bytes:
28 match limits:
29 case ResourceOptions.TransportLimits{connections, frame_bytes,
30 receive_bytes, response_bytes, frame_timeout_ms, response_timeout_ms}:
31 Bytes.concat([word(connections), word(frame_bytes), word(receive_bytes),
32 word(response_bytes), word(frame_timeout_ms), word(response_timeout_ms)])
33
34def encode.queue(limits: ResourceOptions.QueueLimits) -> Bytes.Bytes:
35 match limits:
36 case ResourceOptions.QueueLimits{requests, bytes, timeout_ms}:
37 Bytes.concat([word(requests), word(bytes), word(timeout_ms)])
38
39def encode.readers(limits: ResourceOptions.ReaderLimits) -> Bytes.Bytes:
40 match limits:
41 case ResourceOptions.ReaderLimits{snapshots, cursors,
42 cursors_per_snapshot, retained_root_bytes, retired_files,
43 index_page_bytes, index_buffer_bytes, restore_bytes}:
44 Bytes.concat([word(snapshots), word(cursors), word(cursors_per_snapshot),
45 word(retained_root_bytes), word(retired_files), word(index_page_bytes),
46 word(index_buffer_bytes), word(restore_bytes)])
47
48
49def encode(options: ResourceOptions.Options) -> Bytes.Bytes:
50 match options:
51 case ResourceOptions.Options{transport, queue, readers}:
52 Bytes.concat([magic(), word(1), encode.transport(transport),
53 encode.queue(queue), encode.readers(readers)])
54
55def read.word.result(pair: Bytes.Bytes & Maybe<&2, U32>) -> U32:
56 match pair:
57 case (_, None{}): 0
58 case (_, Some{word}): word
59
60def read.word(input: Bytes.Bytes, offset: U32) -> U32:
61 read.word.result(Bytes.get.u32be(input, offset))
62
63def read.fields(
64 remaining: Nat,
65 input: Bytes.Bytes & Bytes.Bytes,
66 +offset: U32,
67 fields: List<&2, U32>
68) -> List<&2, U32>:
69 match remaining:
70 case 0n:
71 match input:
72 case (_, _): fields
73 case 1n+rest:
74 match input:
75 case (to_read, retained):
76 read.fields(rest, ByteMemTable.bytes.clone(retained),
77 U32.add(offset, 4), Con{read.word(to_read, offset), fields})
78
79def fields.of.options(options: ResourceOptions.Options) -> List<&2, U32>:
80 match options:
81 case ResourceOptions.Options{transport, queue, readers}:
82 match transport:
83 case ResourceOptions.TransportLimits{connections, frame_bytes,
84 receive_bytes, response_bytes, frame_timeout_ms, response_timeout_ms}:
85 match queue:
86 case ResourceOptions.QueueLimits{requests, queue_bytes,
87 queue_timeout_ms}:
88 match readers:
89 case ResourceOptions.ReaderLimits{snapshots, cursors,
90 cursors_per_snapshot, retained_root_bytes, retired_files,
91 index_page_bytes, index_buffer_bytes, restore_bytes}:
92 Con{restore_bytes, Con{index_buffer_bytes,
93 Con{index_page_bytes, Con{retired_files,
94 Con{retained_root_bytes, Con{cursors_per_snapshot,
95 Con{cursors, Con{snapshots, Con{queue_timeout_ms,
96 Con{queue_bytes, Con{requests, Con{response_timeout_ms,
97 Con{frame_timeout_ms, Con{response_bytes,
98 Con{receive_bytes, Con{frame_bytes,
99 Con{connections, Nil{}}}}}}}}}}}}}}}}}}
100
101def options.from.fields(
102 fields: List<&2, U32>
103) -> Maybe<&2, ResourceOptions.Options>:
104 match fields:
105 case Con{restore_bytes, Con{index_buffer_bytes,
106 Con{index_page_bytes, Con{retired_files,
107 Con{retained_root_bytes, Con{cursors_per_snapshot,
108 Con{cursors, Con{snapshots, Con{queue_timeout_ms,
109 Con{queue_bytes, Con{requests, Con{response_timeout_ms,
110 Con{frame_timeout_ms, Con{response_bytes, Con{receive_bytes,
111 Con{frame_bytes, Con{connections, Nil{}}}}}}}}}}}}}}}}}}:
112 Some{ResourceOptions.Options{
113 ResourceOptions.TransportLimits{connections, frame_bytes,
114 receive_bytes, response_bytes, frame_timeout_ms,
115 response_timeout_ms},
116 ResourceOptions.QueueLimits{requests, queue_bytes, queue_timeout_ms},
117 ResourceOptions.ReaderLimits{snapshots, cursors,
118 cursors_per_snapshot, retained_root_bytes, retired_files,
119 index_page_bytes, index_buffer_bytes, restore_bytes}}}
120 case _: None{}
121
122def validated.result(result: ResourceOptions.Decision) -> Decision:
123 match result:
124 case ResourceOptions.Accepted{options}: Parsed{options}
125 case _: InvalidOptions{}
126
127
128def validated(options: ResourceOptions.Options) -> Decision:
129 validated.result(ResourceOptions.validate(options))
130
131def fields.options(
132 result: Maybe<&2, ResourceOptions.Options>
133) -> Decision:
134 match result:
135 case None{}: Malformed{}
136 case Some{options}: validated(options)
137
138def fields.result(fields: List<&2, U32>) -> Decision:
139 fields.options(options.from.fields(fields))
140
141
142def fields(input: Bytes.Bytes) -> Decision:
143 fields.result(read.fields(17n, ByteMemTable.bytes.clone(input), 8, Nil{}))
144
145def u32.duplicate.pair(+value: U32) -> U32 & U32:
146 (value, value)
147
148def version.classified(input: Bytes.Bytes, version: U32,
149 supported: Bool) -> Decision:
150 match supported:
151 case True{}: fields(input)
152 case False{}: UnsupportedVersion{version}
153
154def version.checked(input: Bytes.Bytes, pair: U32 & U32) -> Decision:
155 match pair:
156 case (version, check):
157 version.classified(input, version, U32.is_eq(check, 1))
158
159def version.result(pair: Bytes.Bytes & Maybe<&2, U32>) -> Decision:
160 match pair:
161 case (_, None{}): Malformed{}
162 case (retained, Some{version_code}):
163 version.checked(retained, u32.duplicate.pair(version_code))
164
165def version.read(pair: Bytes.Bytes & Bytes.Bytes) -> Decision:
166 match pair:
167 case (to_read, retained):
168 version.result(Bytes.get.u32be(to_read, 4))
169
170def magic.checked(input: Bytes.Bytes, matches: Bool) -> Decision:
171 match matches:
172 case True{}: version.read(ByteMemTable.bytes.clone(input))
173 case False{}: Malformed{}
174
175def magic.candidate(pair: Bytes.Bytes & Bytes.Bytes) -> Bool:
176 match pair:
177 case (_, candidate): Keys.Bytes.eq(candidate, magic())
178
179def magic.matches(pair: Bytes.Bytes & Bytes.Bytes) -> Decision:
180 match pair:
181 case (magic_source, retained):
182 magic.checked(retained, magic.candidate(Bytes.slice(magic_source, 0, 4)))
183
184def sized.result(input: Bytes.Bytes, valid: Bool) -> Decision:
185 match valid:
186 case False{}: Malformed{}
187 case True{}:
188 magic.matches(ByteMemTable.bytes.clone(input))
189
190
191def sized(input: Bytes.Bytes, size: U32) -> Decision:
192 sized.result(input, U32.is_eq(size, 76))
193
194
195def measured(measured: Bytes.Bytes & U32) -> Decision:
196 match measured:
197 case (input, size): sized(input, size)
198
199
200def decode(input: Bytes.Bytes) -> Decision:
201 measured(Bytes.length(input))
202
203def hex.decoded.result(bytes: Bytes.Bytes, canonical: Bool) -> Decision:
204 match canonical:
205 case True{}: decode(bytes)
206 case False{}: Malformed{}
207
208def hex.decoded.cloned(input: String,
209 copies: Bytes.Bytes & Bytes.Bytes) -> Decision:
210 match copies:
211 case (canonical_bytes, parse_bytes):
212 hex.decoded.result(parse_bytes,
213 String.eq(Bytes.to_hex(canonical_bytes), input))
214
215def hex.decoded(input: String,
216 result: Maybe<&1, Bytes.Bytes>) -> Decision:
217 match result:
218 case None{}: Malformed{}
219 case Some{bytes}:
220 hex.decoded.cloned(input, ByteMemTable.bytes.clone(bytes))
221
222def string.clone.pair.go(value: String, left: String,
223 right: String) -> String & String:
224 match value:
225 case SNil{}: (String.reverse(left), String.reverse(right))
226 case SCon{+character, rest}:
227 string.clone.pair.go(rest, SCon{character, left},
228 SCon{character, right})
229
230def string.clone.pair(value: String) -> String & String:
231 string.clone.pair.go(value, SNil{}, SNil{})
232
233def string.clone.triple.second(length_input: String,
234 parse_inputs: String & String) -> String & (String & String):
235 match parse_inputs:
236 case (canonical_input, decode_input):
237 (length_input, (canonical_input, decode_input))
238
239def string.clone.triple.first(copies: String & String) -> String & (String & String):
240 match copies:
241 case (length_input, parse_input):
242 string.clone.triple.second(length_input,
243 string.clone.pair(parse_input))
244
245def string.clone.triple(value: String) -> String & (String & String):
246 string.clone.triple.first(string.clone.pair(value))
247
248def decode.hex.sized(input: String & String, valid: Bool) -> Decision:
249 match input:
250 case (canonical_input, decode_input):
251 match valid:
252 case True{}:
253 hex.decoded(canonical_input, Bytes.from_hex(decode_input))
254 case False{}: Malformed{}
255
256def decode.hex.copies(copies: String & (String & String)) -> Decision:
257 match copies:
258 case (length_input, (canonical_input, decode_input)):
259 decode.hex.sized((canonical_input, decode_input),
260 Nat.is_eq(String.length(length_input), 152n))
261
262def decode.hex(input: String) -> Decision:
263 decode.hex.copies(string.clone.triple(input))