src/coordinator/CursorCodec.bend

mylsm-lsm-store@0.6.0.1 · 375 lines · 44 definitions

1import Base
2import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
3
4# Represents one incremental cursor response frame.
5type Frame is Type:
6  CursorEnd{}
7  CursorItem{key: Bytes.Bytes, value: Bytes.Bytes}
8
9# Reports a decoded cursor frame or malformed wire bytes.
10type Decision is Type:
11  Decoded{frame: Frame}
12  Malformed{}
13
14# Encodes one exact byte offset and length into a fixed-width word.
15def u32.bytes(value: U32) -> Bytes.Bytes:
16  Bytes.set.u32be(Bytes.new(4), 0, value)
17
18def nat.duplicate.step(pair: Nat & Nat) -> Nat & Nat:
19  match pair:
20    case (first, second): (Nat.add(1n, first), Nat.add(1n, second))
21
22def nat.duplicate(value: Nat) -> Nat & Nat:
23  match value:
24    case 0n: (0n, 0n)
25    case 1n+rest: nat.duplicate.step(nat.duplicate(rest))
26
27def u32.duplicate.result(pair: Nat & Nat) -> U32 & U32:
28  match pair:
29    case (first, second): (U32.from_nat(first), U32.from_nat(second))
30
31def u32.duplicate(value: U32) -> U32 & U32:
32  u32.duplicate.result(nat.duplicate(U32.to_nat(value)))
33
34# Extracts the owned byte sequence from the bytes package slice result.
35def slice.result(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
36  match pair:
37    case (_, sliced): sliced
38
39# Returns the one-byte tag for terminal cursor state.
40def end.tag() -> Bytes.Bytes:
41  Bytes.new(1)
42
43# Returns the one-byte tag for an item carrying raw key and value bytes.
44def item.tag() -> Bytes.Bytes:
45  slice.result(Bytes.slice(u32.bytes(1), 3, 1))
46
47# Encodes retained key/value buffers after measuring each exact byte length.
48def item.measured(
49  key_size: Bytes.Bytes & U32,
50  value_size: Bytes.Bytes & U32
51) -> Bytes.Bytes:
52  match key_size:
53    case (key, key_length):
54      match value_size:
55        case (value, value_length):
56          Bytes.concat([item.tag(), u32.bytes(key_length), key,
57            u32.bytes(value_length), value])
58
59# Encodes a cursor item using explicit lengths and no text conversion.
60def item(
61  key: Bytes.Bytes,
62  value: Bytes.Bytes
63) -> Bytes.Bytes:
64  item.measured(Bytes.length(key), Bytes.length(value))
65
66# Encodes End separately from an item whose key and value are empty.
67def encode(frame: Frame) -> Bytes.Bytes:
68  match frame:
69    case CursorEnd{}: end.tag()
70    case CursorItem{key, value}: item(key, value)
71
72# Extracts exact byte equality from the bytes package comparison result.
73def tag.matches(pair: Bytes.Bytes & Bytes.Bytes & Bool) -> Bool:
74  match pair:
75    case (_, _, equal): equal
76
77# Decodes the terminal marker only when no trailing bytes remain.
78def end.checked(
79  is_end: Bool,
80  exact: Bool
81) -> Decision:
82  match is_end:
83    case True{}:
84      match exact:
85        case True{}: Decoded{CursorEnd{}}
86        case False{}: Malformed{}
87    case False{}: Malformed{}
88
89# Combines the exact key and value byte slices into one cursor item.
90def item.value.paired(
91  pair: Bytes.Bytes & Bytes.Bytes,
92  key: Bytes.Bytes
93) -> Decision:
94  match pair:
95    case (_, value): Decoded{CursorItem{key, value}}
96
97# Creates the value slice from the source retained by the key slice.
98def item.value.key.sliced(
99  pair: Bytes.Bytes & Bytes.Bytes,
100  value_offset: U32,
101  value_length: U32
102) -> Decision:
103  match pair:
104    case (source, key):
105      item.value.paired(Bytes.slice(source, value_offset, value_length), key)
106
107# Slices the key first so the source remains available for the value slice.
108def item.value.slices(
109  payload: Bytes.Bytes,
110  key_lengths: U32 & U32,
111  value_length: U32
112) -> Decision:
113  match key_lengths:
114    case (key_length, value_key_length):
115      item.value.key.sliced(Bytes.slice(payload, 5, key_length),
116        U32.from_nat(Nat.add(9n, U32.to_nat(value_key_length))), value_length)
117
118# Accepts only an exact frame length after reading the value length.
119def item.value.bound.checked(
120  exact: Bool,
121  payload: Bytes.Bytes,
122  key_length: U32,
123  value_length: U32
124) -> Decision:
125  match exact:
126    case True{}:
127      item.value.slices(payload, u32.duplicate(key_length), value_length)
128    case False{}: Malformed{}
129
130# Accepts only a value length that ends exactly at the frame boundary.
131def item.value.bound.compare(
132  payload: Bytes.Bytes,
133  size: U32,
134  key_lengths: U32 & U32,
135  value_lengths: U32 & U32
136) -> Decision:
137  match key_lengths:
138    case (checked_key_length, retained_key_length):
139      match value_lengths:
140        case (checked_value_length, retained_value_length):
141          item.value.bound.checked(Nat.is_eq(U32.to_nat(size),
142            Nat.add(9n, Nat.add(U32.to_nat(checked_key_length),
143              U32.to_nat(checked_value_length)))), payload,
144            retained_key_length, retained_value_length)
145
146# Duplicates lengths for exact validation and the returned value slice.
147def item.value.bound(
148  payload: Bytes.Bytes,
149  size: U32,
150  key_length: U32,
151  value_length: U32
152) -> Decision:
153  item.value.bound.compare(payload, size, u32.duplicate(key_length),
154    u32.duplicate(value_length))
155
156# Distinguishes a complete value length from a truncated header.
157def item.value.parsed(
158  payload: Bytes.Bytes,
159  size: U32,
160  key_length: U32,
161  parsed: Maybe<&2, U32>
162) -> Decision:
163  match parsed:
164    case Some{value_length}:
165      item.value.bound(payload, size, key_length, value_length)
166    case None{}: Malformed{}
167
168# Reads the value length at the boundary immediately after the key.
169def item.value.length(
170  payload: Bytes.Bytes,
171  size: U32,
172  key_length: U32,
173  parsed: Maybe<&2, U32>
174) -> Decision:
175  item.value.parsed(payload, size, key_length, parsed)
176
177# Checks that the key length leaves room for the value length word.
178def item.key.bound.read(
179  read: Bytes.Bytes & Maybe<&2, U32>,
180  size: U32,
181  key_length: U32
182) -> Decision:
183  match read:
184    case (payload, parsed):
185      item.value.length(payload, size, key_length, parsed)
186
187# Checks that the key length leaves room for the value length word.
188def item.key.bound.read.offset(
189  payload: Bytes.Bytes,
190  size: U32,
191  key_lengths: U32 & U32
192) -> Decision:
193  match key_lengths:
194    case (offset_length, retained_length):
195      item.key.bound.read(Bytes.get.u32be(payload,
196        U32.from_nat(Nat.add(5n, U32.to_nat(offset_length)))), size,
197        retained_length)
198
199# Checks that the key length leaves room for the value length word.
200def item.key.bound.checked(
201  has_value_header: Bool,
202  payload: Bytes.Bytes,
203  size: U32,
204  key_length: U32
205) -> Decision:
206  match has_value_header:
207    case True{}:
208      item.key.bound.read.offset(payload, size,
209        u32.duplicate(key_length))
210    case False{}: Malformed{}
211
212# Accepts key lengths that leave a complete value length word.
213def item.key.bound.check(
214  payload: Bytes.Bytes,
215  sizes: U32 & U32,
216  key_lengths: U32 & U32
217) -> Decision:
218  match sizes:
219    case (checked_size, retained_size):
220      match key_lengths:
221        case (checked_key_length, retained_key_length):
222          item.key.bound.checked(Nat.is_le(
223            Nat.add(9n, U32.to_nat(checked_key_length)),
224            U32.to_nat(checked_size)), payload, retained_size,
225            retained_key_length)
226
227# Accepts key lengths that leave a complete value length word.
228def item.key.bound.compare(
229  payload: Bytes.Bytes,
230  size: U32,
231  key_lengths: U32 & U32
232) -> Decision:
233  item.key.bound.check(payload, u32.duplicate(size), key_lengths)
234
235# Accepts key lengths that leave a complete value length word.
236def item.key.bound(
237  payload: Bytes.Bytes,
238  size: U32,
239  key_length: U32
240) -> Decision:
241  item.key.bound.compare(payload, size, u32.duplicate(key_length))
242
243# Accepts the key-length field only when all required bytes are present.
244def item.key.parsed(
245  payload: Bytes.Bytes,
246  size: U32,
247  parsed: Maybe<&2, U32>
248) -> Decision:
249  match parsed:
250    case Some{key_length}: item.key.bound(payload, size, key_length)
251    case None{}: Malformed{}
252
253# Reads the key length before deriving the value-length offset.
254def item.key.length(
255  payload: Bytes.Bytes,
256  size: U32,
257  parsed: Maybe<&2, U32>
258) -> Decision:
259  item.key.parsed(payload, size, parsed)
260
261# Routes complete item headers to the key-length decoder.
262def item.header.read(
263  read: Bytes.Bytes & Maybe<&2, U32>,
264  size: U32
265) -> Decision:
266  match read:
267    case (payload, parsed): item.key.length(payload, size, parsed)
268
269# Routes complete item headers to the key-length decoder.
270def item.header.checked(
271  has_header: Bool,
272  payload: Bytes.Bytes,
273  size: U32
274) -> Decision:
275  match has_header:
276    case True{}:
277      item.header.read(Bytes.get.u32be(payload, 1), size)
278    case False{}: Malformed{}
279
280# Checks header capacity while retaining its total for later framing checks.
281def item.header.compare(
282  payload: Bytes.Bytes,
283  sizes: U32 & U32,
284  is_item: Bool
285) -> Decision:
286  match sizes:
287    case (checked_size, retained_size):
288      item.header.checked(U32.is_le(9, checked_size) && is_item, payload,
289        retained_size)
290
291# Rejects item headers that cannot contain both length words.
292def item.header.size(
293  payload: Bytes.Bytes,
294  size: U32,
295  is_item: Bool
296) -> Decision:
297  item.header.compare(payload, u32.duplicate(size), is_item)
298
299# Routes the item tag to exact length-prefixed decoding.
300def decode.item(
301  payload: Bytes.Bytes,
302  size: U32,
303  is_item: Bool
304) -> Decision:
305  item.header.size(payload, size, is_item)
306
307# Decodes the item marker from the retained response source.
308def decode.item.tag.sliced(
309  sliced: Bytes.Bytes & Bytes.Bytes,
310  size: U32
311) -> Decision:
312  match sliced:
313    case (source, tag_bytes):
314      decode.item(source, size,
315        tag.matches(Bytes.eq(tag_bytes, item.tag())))
316
317# Retains the response source while comparing the item marker.
318def decode.item.tag(
319  payload: Bytes.Bytes,
320  size: U32
321) -> Decision:
322  decode.item.tag.sliced(Bytes.slice(payload, 0, 1), size)
323
324# Selects End validation or item decoding after the terminal tag check.
325def decode.tag.end.checked(
326  is_end: Bool,
327  payload: Bytes.Bytes,
328  sizes: U32 & U32
329) -> Decision:
330  match is_end:
331    case True{}:
332      match sizes:
333        case (checked_size, _):
334          end.checked(True{}, U32.is_eq(checked_size, 1))
335    case False{}:
336      match sizes:
337        case (_, retained_size): decode.item.tag(payload, retained_size)
338
339# Checks the terminal marker while retaining bytes for item-tag decoding.
340def decode.tag.end(
341  sliced: Bytes.Bytes & Bytes.Bytes,
342  size: U32
343) -> Decision:
344  match sliced:
345    case (payload, tag_bytes):
346      decode.tag.end.checked(tag.matches(Bytes.eq(tag_bytes, end.tag())),
347        payload, u32.duplicate(size))
348
349# Validates that a nonempty encoded response has a supported tag.
350def decode.nonempty(
351  payload: Bytes.Bytes,
352  size: U32,
353  is_nonempty: Bool
354) -> Decision:
355  match is_nonempty:
356    case True{}: decode.tag.end(Bytes.slice(payload, 0, 1), size)
357    case False{}: Malformed{}
358
359# Decodes a complete cursor response, rejecting unknown tags and trailing data.
360def decode.size.checked(
361  payload: Bytes.Bytes,
362  sizes: U32 & U32
363) -> Decision:
364  match sizes:
365    case (checked_size, retained_size):
366      decode.nonempty(payload, retained_size, U32.is_le(1, checked_size))
367
368# Decodes a complete cursor response, rejecting unknown tags and trailing data.
369def decode.measured(measured: Bytes.Bytes & U32) -> Decision:
370  match measured:
371    case (payload, size): decode.size.checked(payload, u32.duplicate(size))
372
373# Decodes a complete cursor response from its exact byte payload.
374def decode(payload: Bytes.Bytes) -> Decision:
375  decode.measured(Bytes.length(payload))