src/coordinator/CursorOpenCodec.bend
mylsm-lsm-store@0.6.0.0 · 428 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/coordinator/Protocol.bend as Protocol
3import mylsm-lsm-store@0.6.0.0/src/scan/RangeCursor.bend as RangeCursor
4import mylsm-lsm-store@0.6.0.0/src/ByteMemTable.bend as ByteMemTable
5import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
6
7# Carries one exact range or prefix request for an owner snapshot cursor.
8type Request is Type:
9 Request{snapshot_id: U32, bounds: RangeCursor.Bounds}
10
11# Reports a decoded cursor-open request or malformed argument bytes.
12type Decision is Type:
13 Decoded{request: Request}
14 Malformed{}
15
16# Carries an input buffer, one parsed optional bound, and its next offset.
17type BoundResult is Type:
18 BoundParsed{source: Bytes.Bytes, value: Maybe<&1, Bytes.Bytes>, next: U32}
19 BoundMalformed{}
20
21# Carries both range bounds after parsing the lower bound.
22type RangeResult is Type:
23 RangeParsed{lower: Maybe<&1, Bytes.Bytes>,
24 upper: Maybe<&1, Bytes.Bytes>, next: U32}
25 RangeMalformed{}
26
27# Encodes an absent bound with a reserved length outside the admitted frame cap.
28def bound.absent() -> Bytes.Bytes:
29 Protocol.u32.bytes(4294967295)
30
31# Encodes one present bound as a length followed by its raw bytes.
32def bound.present(measured: Bytes.Bytes & U32) -> Bytes.Bytes:
33 match measured:
34 case (value, length):
35 Bytes.concat([Protocol.u32.bytes(length), value])
36
37# Encodes an optional bound without interpreting or transforming its bytes.
38def bound.encode(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
39 match value:
40 case None{}: bound.absent()
41 case Some{bytes}: bound.present(Bytes.length(bytes))
42
43# Encodes one full range request with an owner snapshot token.
44def encode.range(
45 snapshot_id: U32,
46 lower: Maybe<&1, Bytes.Bytes>,
47 upper: Maybe<&1, Bytes.Bytes>
48) -> Bytes.Bytes:
49 Bytes.concat([Protocol.snapshot.id.bytes(snapshot_id),
50 Protocol.u32.bytes(0), bound.encode(lower), bound.encode(upper)])
51
52# Encodes one byte-prefix request with an owner snapshot token.
53def encode.prefix(snapshot_id: U32, prefix: Bytes.Bytes) -> Bytes.Bytes:
54 Bytes.concat([Protocol.snapshot.id.bytes(snapshot_id),
55 Protocol.u32.bytes(1), bound.present(Bytes.length(prefix))])
56
57# Copies a fixed-width word without recursion on its numeric value.
58def u32.duplicate.triple(+value: U32) -> U32 & U32 & U32:
59 (value, value, value)
60
61# Duplicates one fixed-width word for two bounded offset calculations.
62def u32.duplicate.pair(+value: U32) -> U32 & U32:
63 (value, value)
64
65# Retains the source buffer while storing one exact bound slice.
66def field.sized.sliced(
67 pair: Bytes.Bytes & Bytes.Bytes,
68 next: U32
69) -> BoundResult:
70 match pair:
71 case (source, value): BoundParsed{source, Some{value}, next}
72
73# Slices a validated span using separate copies for its start and end offsets.
74def field.sized.checked(
75 payload: Bytes.Bytes,
76 offsets: U32 & U32 & U32,
77 lengths: U32 & U32,
78 fits: Bool
79) -> BoundResult:
80 match offsets:
81 case (slice_offset, (next_offset, _)):
82 match lengths:
83 case (slice_length, next_length):
84 match fits:
85 case True{}:
86 field.sized.sliced(Bytes.slice(payload,
87 U32.from_nat(Nat.add(U32.to_nat(slice_offset), 4n)),
88 slice_length),
89 U32.from_nat(Nat.add(Nat.add(U32.to_nat(next_offset), 4n),
90 U32.to_nat(next_length))))
91 case False{}: BoundMalformed{}
92
93# Selects a bounded raw-byte slice or rejects a span beyond the frame.
94def field.sized.decision(
95 payload: Bytes.Bytes,
96 offset: U32,
97 length: U32,
98 fits: Bool
99) -> BoundResult:
100 field.sized.checked(payload,
101 u32.duplicate.triple(offset), u32.duplicate.pair(length), fits)
102
103# Separates validation inputs from the values passed to the slice decision.
104def field.sized.inputs(
105 payload: Bytes.Bytes,
106 totals: U32 & U32,
107 offsets: U32 & U32,
108 lengths: U32 & U32
109) -> BoundResult:
110 match totals:
111 case (_, check_total):
112 match offsets:
113 case (slice_offset, check_offset):
114 match lengths:
115 case (slice_length, check_length):
116 field.sized.decision(payload, slice_offset, slice_length,
117 Nat.is_le(Nat.add(Nat.add(U32.to_nat(check_offset), 4n),
118 U32.to_nat(check_length)), U32.to_nat(check_total)))
119
120# Validates the payload span before slicing one present bound.
121def field.sized(
122 payload: Bytes.Bytes,
123 total: U32,
124 offset: U32,
125 length: U32
126) -> BoundResult:
127 field.sized.inputs(payload, u32.duplicate.pair(total),
128 u32.duplicate.pair(offset),
129 u32.duplicate.pair(length))
130
131# Selects the absent sentinel or a validated present bound.
132def field.value.decision(
133 payload: Bytes.Bytes,
134 total: U32,
135 offset: U32,
136 length: U32,
137 absent: Bool
138) -> BoundResult:
139 match absent:
140 case True{}:
141 BoundParsed{payload, None{}, U32.from_nat(
142 Nat.add(U32.to_nat(offset), 4n))}
143 case False{}:
144 field.sized(payload, total, offset, length)
145
146# Separates the sentinel check from the length passed to the bound parser.
147def field.value.inputs(
148 payload: Bytes.Bytes,
149 total: U32,
150 offset: U32,
151 lengths: U32 & U32
152) -> BoundResult:
153 match lengths:
154 case (bound_length, check_length):
155 field.value.decision(payload, total, offset, bound_length,
156 U32.is_eq(check_length, 4294967295))
157
158# Converts one field length into a bounded slice or an absent bound.
159def field.value(
160 payload: Bytes.Bytes,
161 total: U32,
162 offset: U32,
163 length: U32
164) -> BoundResult:
165 field.value.inputs(payload, total, offset, u32.duplicate.pair(length))
166
167# Checks that one field's four-byte length is available at its offset.
168def field.length(
169 payload: Bytes.Bytes,
170 total: U32,
171 offset: U32,
172 parsed: Maybe<&2, U32>
173) -> BoundResult:
174 match parsed:
175 case None{}: BoundMalformed{}
176 case Some{length}:
177 field.value(payload, total, offset, length)
178
179# Parses one optional bound length from the supplied offset.
180def field.read.parsed(
181 payload: Bytes.Bytes,
182 total: U32,
183 offset: U32,
184 pair: Bytes.Bytes & Maybe<&2, U32>
185) -> BoundResult:
186 match pair:
187 case (_, parsed): field.length(payload, total, offset, parsed)
188
189# Keeps one source for slicing while reading its fixed-width length field.
190def field.read.copies(
191 sources: Bytes.Bytes & Bytes.Bytes,
192 total: U32,
193 offsets: U32 & U32
194) -> BoundResult:
195 match sources:
196 case (payload, measured):
197 match offsets:
198 case (slice_offset, read_offset):
199 field.read.parsed(payload, total, slice_offset,
200 Bytes.get.u32be(measured, read_offset))
201
202# Parses one optional bound length from the supplied offset.
203def field.read(
204 payload: Bytes.Bytes,
205 total: U32,
206 offset: U32
207) -> BoundResult:
208 field.read.copies(ByteMemTable.bytes.clone(payload), total,
209 u32.duplicate.pair(offset))
210
211# Combines both successfully parsed range bounds.
212def range.upper.result(
213 result: BoundResult,
214 lower: Maybe<&1, Bytes.Bytes>
215) -> RangeResult:
216 match result:
217 case BoundMalformed{}: RangeMalformed{}
218 case BoundParsed{_, upper, next}:
219 RangeParsed{lower, upper, next}
220
221# Reads the upper field from the source retained by lower-bound parsing.
222def range.upper.read(
223 source: Bytes.Bytes,
224 total: U32,
225 lower: Maybe<&1, Bytes.Bytes>,
226 offset: U32
227) -> RangeResult:
228 range.upper.result(field.read(source, total, offset), lower)
229
230# Parses the upper bound after preserving the lower-bound result.
231def range.upper(
232 lower_result: BoundResult,
233 total: U32
234) -> RangeResult:
235 match lower_result:
236 case BoundMalformed{}: RangeMalformed{}
237 case BoundParsed{source, lower, next}:
238 range.upper.read(source, total, lower, next)
239
240# Rejects trailing bytes or returns the decoded range cursor request.
241def range.finish.exact(
242 exact: Bool,
243 snapshot_id: U32,
244 lower: Maybe<&1, Bytes.Bytes>,
245 upper: Maybe<&1, Bytes.Bytes>
246) -> Decision:
247 match exact:
248 case True{}:
249 Decoded{Request{snapshot_id,
250 RangeCursor.Bounds{lower, upper, None{}}}}
251 case False{}: Malformed{}
252
253# Builds a range request only when its fields consume the entire frame.
254def range.finish(
255 parsed: RangeResult,
256 snapshot_id: U32,
257 total: U32
258) -> Decision:
259 match parsed:
260 case RangeMalformed{}: Malformed{}
261 case RangeParsed{lower, upper, next}:
262 range.finish.exact(U32.is_eq(next, total), snapshot_id, lower, upper)
263
264# Parses range bounds with separate total lengths for IO-free validation.
265def range.decode.totals(
266 payload: Bytes.Bytes,
267 snapshot_id: U32,
268 totals: U32 & U32 & U32
269) -> Decision:
270 match totals:
271 case (read_total, (upper_total, finish_total)):
272 range.finish(range.upper(field.read(payload, read_total, 8), upper_total),
273 snapshot_id, finish_total)
274
275# Parses both optional bounds and requires an exact request boundary.
276def range.decode(
277 payload: Bytes.Bytes,
278 snapshot_id: U32,
279 total: U32
280) -> Decision:
281 range.decode.totals(payload, snapshot_id,
282 u32.duplicate.triple(total))
283
284# Rejects trailing bytes or returns the decoded prefix cursor request.
285def prefix.finish.exact(
286 exact: Bool,
287 snapshot_id: U32,
288 prefix: Bytes.Bytes
289) -> Decision:
290 match exact:
291 case True{}:
292 Decoded{Request{snapshot_id,
293 RangeCursor.Bounds{None{}, None{}, Some{prefix}}}}
294 case False{}: Malformed{}
295
296# Builds a prefix request only when its field consumes the remaining frame.
297def prefix.finish(
298 parsed: BoundResult,
299 snapshot_id: U32,
300 total: U32
301) -> Decision:
302 match parsed:
303 case BoundMalformed{}: Malformed{}
304 case BoundParsed{_, Some{prefix}, next}:
305 prefix.finish.exact(U32.is_eq(next, total), snapshot_id, prefix)
306 case BoundParsed{_, None{}, _}: Malformed{}
307
308# Parses prefix bounds with separate lengths for reading and validation.
309def prefix.decode.totals(
310 payload: Bytes.Bytes,
311 snapshot_id: U32,
312 totals: U32 & U32
313) -> Decision:
314 match totals:
315 case (read_total, finish_total):
316 prefix.finish(field.read(payload, read_total, 8), snapshot_id, finish_total)
317
318# Parses one present prefix bound and requires an exact request boundary.
319def prefix.decode(
320 payload: Bytes.Bytes,
321 snapshot_id: U32,
322 total: U32
323) -> Decision:
324 prefix.decode.totals(payload, snapshot_id, u32.duplicate.pair(total))
325
326# Selects range or prefix decoding from the exact mode word.
327def mode(
328 payload: Bytes.Bytes,
329 snapshot_id: U32,
330 total: U32,
331 parsed: Maybe<&2, U32>
332) -> Decision:
333 match parsed:
334 case Some{0}: range.decode(payload, snapshot_id, total)
335 case Some{1}: prefix.decode(payload, snapshot_id, total)
336 case _: Malformed{}
337
338# Requires the mode field to exist before selecting a cursor-open shape.
339def mode.read.parsed(
340 payload: Bytes.Bytes,
341 snapshot_id: U32,
342 total: U32,
343 parsed: Maybe<&2, U32>
344) -> Decision:
345 mode(payload, snapshot_id, total, parsed)
346
347# Projects one fixed-width mode parse while retaining its request source.
348def mode.read.field(
349 payload: Bytes.Bytes,
350 snapshot_id: U32,
351 total: U32,
352 field: Bytes.Bytes & Maybe<&2, U32>
353) -> Decision:
354 match field:
355 case (_, parsed): mode.read.parsed(payload, snapshot_id, total, parsed)
356
357# Reads the mode from a copy while retaining the request for bound slicing.
358def mode.read.copies(
359 sources: Bytes.Bytes & Bytes.Bytes,
360 snapshot_id: U32,
361 total: U32
362) -> Decision:
363 match sources:
364 case (payload, reader):
365 mode.read.field(payload, snapshot_id, total,
366 Bytes.get.u32be(reader, 4))
367
368# Requires the mode field to exist before selecting a cursor-open shape.
369def mode.read(
370 payload: Bytes.Bytes,
371 snapshot_id: U32,
372 total: U32,
373 enough: Bool
374) -> Decision:
375 match enough:
376 case True{}:
377 mode.read.copies(ByteMemTable.bytes.clone(payload), snapshot_id, total)
378 case False{}: Malformed{}
379
380# Decodes an owner token followed by one supported cursor-open mode.
381def decode.payload.inputs(
382 payload: Bytes.Bytes,
383 totals: U32 & U32,
384 parsed: Maybe<&2, U32>
385) -> Decision:
386 match totals:
387 case (mode_total, check_total):
388 match parsed:
389 case Some{snapshot_id}:
390 mode.read(payload, snapshot_id, mode_total,
391 U32.is_le(8, check_total))
392 case None{}: Malformed{}
393
394# Decodes an owner token followed by one supported cursor-open mode.
395def decode.payload(
396 payload: Bytes.Bytes,
397 total: U32,
398 parsed: Maybe<&2, U32>
399) -> Decision:
400 decode.payload.inputs(payload, u32.duplicate.pair(total), parsed)
401
402# Projects one snapshot-token parse while retaining the full request.
403def decode.payload.field(
404 payload: Bytes.Bytes,
405 total: U32,
406 field: Bytes.Bytes & Maybe<&2, U32>
407) -> Decision:
408 match field:
409 case (_, parsed): decode.payload(payload, total, parsed)
410
411# Decodes an owner token from one input copy and retains another for mode parsing.
412def decode.payload.copies(
413 sources: Bytes.Bytes & Bytes.Bytes,
414 total: U32
415) -> Decision:
416 match sources:
417 case (payload, reader):
418 decode.payload.field(payload, total, Bytes.get.u32be(reader, 0))
419
420# Decodes one bounded cursor-open request without interpreting its bytes.
421def decode.measured(measured: Bytes.Bytes & U32) -> Decision:
422 match measured:
423 case (payload, total):
424 decode.payload.copies(ByteMemTable.bytes.clone(payload), total)
425
426# Decodes a full-range or prefix request from its exact wire payload.
427def decode(payload: Bytes.Bytes) -> Decision:
428 decode.measured(Bytes.length(payload))