src/Wal.bend
mylsm-lsm-store@0.5.0.0 · 546 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.5.0.0/src/MemTable.bend as MemTable
3import mylsm-lsm-store@0.5.0.0/src/SstFile.bend as SstFile
4import mylsm-lsm-store@0.5.0.0/src/StorageBytes.bend as StorageBytes
5import mylsm-lsm-store@0.5.0.0/src/hub_sha/sha256.bend as SHA
6import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7
8# Represent Mut data used by the write-ahead log data model.
9type Mut is Data:
10 Put{key: String, val: String}
11 Del{key: String}
12
13# Represent Batch data used by the write-ahead log data model.
14type Batch is Data:
15 Batch{muts: List<&2, Mut>}
16
17# Handle stage in the write-ahead log data model.
18def stage(+staged: List<&2, Batch>, +batch: Batch) -> List<&2, Batch>:
19 Con{batch, staged}
20
21# Represent Error data used by the packed WAL codec.
22type Error is Data:
23 TooLarge{}
24 InvalidUtf8{}
25 Malformed{}
26 ChecksumMismatch{}
27
28def record.error(error: SstFile.Error) -> Error:
29 match error:
30 case SstFile.TooLarge{}:
31 TooLarge{}
32 case SstFile.InvalidUtf8{}:
33 InvalidUtf8{}
34 case SstFile.ChecksumMismatch{}:
35 ChecksumMismatch{}
36 case _:
37 Malformed{}
38
39def record.convert(
40 pair: Bytes.Cursor & Result<&1, &1, SstFile.Error, SstFile.ParsedRecord>
41) -> Bytes.Cursor & Result<&1, &1, Error, Mut>:
42 match pair:
43 case (cursor, Fail{error}):
44 (cursor, Fail{record.error(error)})
45 case (cursor, Done{SstFile.Rec{MemTable.Entry{key, None{}}, _}}):
46 (cursor, Done{Del{key}})
47 case (cursor, Done{SstFile.Rec{MemTable.Entry{key, Some{value}}, _}}):
48 (cursor, Done{Put{key, value}})
49
50def record.read(
51 cursor: Bytes.Cursor
52) -> Bytes.Cursor & Result<&1, &1, Error, Mut>:
53 record.convert(SstFile.parse.record(cursor))
54
55# Encode or decode a 32-bit word for bytes for the packed WAL codec.
56def u32_bytes(value: U32) -> Bytes.Bytes:
57 Bytes.set.u32be(Bytes.new(4), 0, value)
58
59# Encode an 8-bit value for bytes for the packed WAL codec.
60def u8_bytes(value: U32) -> Bytes.Bytes:
61 Bytes.set(Bytes.new(1), 0, value)
62
63def header.finish(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
64 match value:
65 case Some{bytes}:
66 bytes
67 case None{}:
68 Bytes.new(0)
69
70# Handle log header in the packed WAL codec.
71def log_header() -> Bytes.Bytes:
72 header.finish(Bytes.from_hex("4d594c534d335700"))
73
74def header.valid(
75 comparison: Bytes.Bytes & Bytes.Bytes & Bool
76) -> Result<&1, &1, Error, Unit>:
77 match comparison:
78 case (_, _, True{}):
79 Done{Unit{}}
80 case (_, _, False{}):
81 Fail{Malformed{}}
82
83# Decode header for the packed WAL codec.
84def decode_header(bytes: Bytes.Bytes) -> Result<&1, &1, Error, Unit>:
85 header.valid(Bytes.eq(log_header(), bytes))
86
87def record.delete.valid(
88 tag: U32,
89 key: Bytes.Bytes,
90 +key_len: U32,
91 valid: Bool
92) -> Result<&1, &1, Error, Bytes.Bytes>:
93 match valid:
94 case False{}:
95 Fail{TooLarge{}}
96 case True{}:
97 Done{Bytes.concat([u8_bytes(tag), u32_bytes(key_len), key])}
98
99
100def record.delete(
101 pair: Bytes.Bytes & U32
102) -> Result<&1, &1, Error, Bytes.Bytes>:
103 match pair:
104 case (key, +key_len):
105 record.delete.valid(1, key, key_len,
106 U32.is_le(key_len, (StorageBytes.MAX_RECORD_BYTES() - 5 : U32)))
107
108
109def record.put.valid(
110 key: Bytes.Bytes,
111 +key_len: U32,
112 value: Bytes.Bytes,
113 +value_len: U32,
114 valid: Bool
115) -> Result<&1, &1, Error, Bytes.Bytes>:
116 match valid:
117 case False{}:
118 Fail{TooLarge{}}
119 case True{}:
120 Done{Bytes.concat([Bytes.new(1), u32_bytes(key_len), key,
121 u32_bytes(value_len), value])}
122
123
124def record.put.checked(
125 key: Bytes.Bytes,
126 +key_len: U32,
127 value: Bytes.Bytes,
128 +value_len: U32,
129 sum: Maybe<&2, U32>
130) -> Result<&1, &1, Error, Bytes.Bytes>:
131 match sum:
132 case None{}:
133 Fail{TooLarge{}}
134 case Some{size}:
135 record.put.valid(key, key_len, value, value_len,
136 U32.is_le(size, (StorageBytes.MAX_RECORD_BYTES() - 9 : U32)))
137
138
139def record.put.value(
140 key: Bytes.Bytes,
141 +key_len: U32,
142 value_pair: Bytes.Bytes & U32
143) -> Result<&1, &1, Error, Bytes.Bytes>:
144 match value_pair:
145 case (value, +value_len):
146 record.put.checked(key, key_len, value, value_len,
147 StorageBytes.checked_add(key_len, value_len))
148
149def record.put(
150 key_pair: Bytes.Bytes & U32,
151 value_pair: Bytes.Bytes & U32
152) -> Result<&1, &1, Error, Bytes.Bytes>:
153 match key_pair:
154 case (key, +key_len):
155 record.put.value(key, key_len, value_pair)
156
157
158def record.encode.put(
159 key_result: Result<&1, &1, U32 & String, Bytes.Bytes>,
160 value_result: Result<&1, &1, U32 & String, Bytes.Bytes>
161) -> Result<&1, &1, Error, Bytes.Bytes>:
162 match key_result value_result:
163 case Fail{_} _:
164 Fail{TooLarge{}}
165 case _ Fail{_}:
166 Fail{TooLarge{}}
167 case Done{key} Done{value}:
168 record.put(Bytes.length(key), Bytes.length(value))
169
170
171def record.encode.delete(
172 key_result: Result<&1, &1, U32 & String, Bytes.Bytes>
173) -> Result<&1, &1, Error, Bytes.Bytes>:
174 match key_result:
175 case Fail{_}:
176 Fail{TooLarge{}}
177 case Done{key}:
178 record.delete(Bytes.length(key))
179
180
181def record.encode(mut: Mut) -> Result<&1, &1, Error, Bytes.Bytes>:
182 match mut:
183 case Put{key, val}:
184 record.encode.put(StorageBytes.from_string(key), StorageBytes.from_string(val))
185 case Del{key}:
186 record.encode.delete(StorageBytes.from_string(key))
187
188
189
190def records.step(
191 rest: List<&2, Mut>,
192 current: Result<&1, &1, Error, Bytes.Bytes>,
193 acc: List<&1, Bytes.Bytes>
194) -> Result<&1, &1, Error, List<&1, Bytes.Bytes>>:
195 match rest current:
196 case Nil{} Fail{error}:
197 Fail{error}
198 case Nil{} Done{bytes}:
199 Done{List.reverse(&1, Bytes.Bytes, bytes <> acc)}
200 case Con{_mut, _tail} Fail{error}:
201 Fail{error}
202 case Con{mut, tail} Done{bytes}:
203 records.step(tail, record.encode(mut), bytes <> acc)
204
205# Handle the record s for the packed WAL codec.
206def records(
207 muts: List<&2, Mut>
208) -> Result<&1, &1, Error, List<&1, Bytes.Bytes>>:
209 match muts:
210 case Nil{}:
211 Done{Nil{}}
212 case Con{mut, tail}:
213 records.step(tail, record.encode(mut), Nil{})
214
215def bytes.clone.pair(
216 +len: U32,
217 copied: Array<U32> & Array<U32>
218) -> Bytes.Bytes & Bytes.Bytes:
219 match copied:
220 case (left, right):
221 (Bytes.Bytes{len, left}, Bytes.Bytes{len, right})
222
223def bytes.clone.split(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:
224 match bytes:
225 case Bytes.Bytes{+len, buf}:
226 bytes.clone.pair(len, Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
227
228def frame.digest.pair(
229 +count: U32,
230 +frame_len: U32,
231 output_payload: Bytes.Bytes,
232 hash_payload: Bytes.Bytes
233) -> Result<&1, &1, Error, Bytes.Bytes>:
234 digest = SHA.sha256_packed_bytes(Bytes.concat([u32_bytes(frame_len),
235 u32_bytes(count), hash_payload]))
236 Done{Bytes.concat([u32_bytes(frame_len), u32_bytes(count),
237 output_payload, digest])}
238
239
240
241def frame.digest(
242 +count: U32,
243 payload: Bytes.Bytes,
244 +frame_len: U32,
245 hash_payload: Bytes.Bytes
246) -> Result<&1, &1, Error, Bytes.Bytes>:
247 frame.digest.pair(count, frame_len, payload, hash_payload)
248
249
250
251def frame.payload.valid(
252 +count: U32,
253 payload: Bytes.Bytes,
254 +frame_len: U32,
255 hash_payload: Bytes.Bytes,
256 valid: Bool
257) -> Result<&1, &1, Error, Bytes.Bytes>:
258 match valid:
259 case False{}:
260 Fail{TooLarge{}}
261 case True{}:
262 frame.digest(count, payload, frame_len, hash_payload)
263
264
265
266# Checks an encoded WAL frame against its byte cap.
267def frame_length_allowed(+count: U32, +length: U32) -> Bool:
268 U32.is_le(length, StorageBytes.MAX_BLOCK_BYTES()) ||
269 U32.is_eq(count, 1) && U32.is_le(length,
270 (StorageBytes.MAX_RECORD_BYTES() + 36 : U32))
271
272def frame.payload.fin(
273 +count: U32,
274 payload: Bytes.Bytes,
275 hash_payload: Bytes.Bytes,
276 frame_len: Maybe<&2, U32>
277) -> Result<&1, &1, Error, Bytes.Bytes>:
278 match frame_len:
279 case None{}:
280 Fail{TooLarge{}}
281 case Some{+length}:
282 frame.payload.valid(count, payload, length, hash_payload, frame_length_allowed(count, length))
283
284
285
286def frame.payload.output(
287 count: U32,
288 hash_payload: Bytes.Bytes,
289 measured: Bytes.Bytes & U32
290) -> Result<&1, &1, Error, Bytes.Bytes>:
291 match measured:
292 case (payload, +payload_len):
293 frame.payload.fin(count, payload, hash_payload,
294 StorageBytes.checked_add(payload_len, 36))
295
296def frame.payload.size(
297 count: U32,
298 pair: Bytes.Bytes & Bytes.Bytes
299) -> Result<&1, &1, Error, Bytes.Bytes>:
300 match pair:
301 case (output, hash_payload):
302 frame.payload.output(count, hash_payload, Bytes.length(output))
303
304
305
306def frame.payload(
307 +count: U32,
308 records: Result<&1, &1, Error, List<&1, Bytes.Bytes>>
309) -> Result<&1, &1, Error, Bytes.Bytes>:
310 match records:
311 case Fail{error}:
312 Fail{error}
313 case Done{parts}:
314 frame.payload.size(count, bytes.clone.split(Bytes.concat(parts)))
315
316# Encode frame for the packed WAL codec.
317def encode_frame(batch: Batch) -> Result<&1, &1, Error, Bytes.Bytes>:
318 match batch:
319 case Batch{+muts}:
320 frame.payload(U32.from_nat(List.length(&2, Mut, muts)), records(muts))
321
322# Represent DigestRead data used by the packed WAL codec.
323type DigestRead is Type:
324 DigestInput{cursor: Bytes.Cursor, value: Maybe<&2, U32>}
325
326def digest.read.wrap(pair: Bytes.Cursor & Maybe<&2, U32>) -> DigestRead:
327 match pair:
328 case (cursor, value):
329 DigestInput{cursor, value}
330
331def digest.read.go(
332 remaining: Nat,
333 input: DigestRead,
334 digest: Bytes.Bytes,
335 +offset: U32
336) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:
337 match remaining input:
338 case 0n DigestInput{cursor, _}:
339 (cursor, Done{digest})
340 case 1n+0n DigestInput{cursor, Some{word}}:
341 (cursor, Done{Bytes.set.u32be(digest, offset, word)})
342 case 1n+1n+rest DigestInput{cursor, Some{word}}:
343 digest.read.go(1n+rest, digest.read.wrap(Bytes.Cursor.u32be(cursor)),
344 Bytes.set.u32be(digest, offset, word), (offset + 4 : U32))
345 case 1n+_ DigestInput{cursor, None{}}:
346 (cursor, Fail{Malformed{}})
347
348def digest.read(
349 cursor: Bytes.Cursor
350) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:
351 digest.read.go(8n, digest.read.wrap(Bytes.Cursor.u32be(cursor)),
352 Bytes.new(32), 0)
353
354# Represent RecordRead data used by the packed WAL codec.
355type RecordRead is Type:
356 RecordInput{cursor: Bytes.Cursor,
357 result: Result<&1, &1, Error, Mut>}
358
359def records.read.wrap(
360 pair: Bytes.Cursor & Result<&1, &1, Error, Mut>
361) -> RecordRead:
362 match pair:
363 case (cursor, result):
364 RecordInput{cursor, result}
365
366def records.read.go(
367 remaining: Nat,
368 input: RecordRead,
369 reversed: List<&2, Mut>
370) -> Bytes.Cursor & Result<&1, &1, Error, List<&2, Mut>>:
371 match remaining input:
372 case 0n RecordInput{cursor, Fail{error}}:
373 (cursor, Fail{error})
374 case 0n RecordInput{cursor, Done{mut}}:
375 (cursor, Done{List.reverse(&2, Mut, mut <> reversed)})
376 case 1n+rest RecordInput{cursor, Fail{error}}:
377 (cursor, Fail{error})
378 case 1n+rest RecordInput{cursor, Done{mut}}:
379 records.read.go(rest, records.read.wrap(record.read(cursor)),
380 mut <> reversed)
381
382def records.read.start(
383 cursor: Bytes.Cursor,
384 +count: U32
385) -> Bytes.Cursor & Result<&1, &1, Error, List<&2, Mut>>:
386 match count:
387 case 0:
388 (cursor, Done{Nil{}})
389 case _:
390 records.read.go(Nat.sub(U32.to_nat(count), 1n),
391 records.read.wrap(record.read(cursor)), Nil{})
392
393def frame.digest.prefix(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
394 match pair:
395 case (_, prefix):
396 prefix
397
398def frame.verify.equal(
399 records: List<&2, Mut>,
400 comparison: Bytes.Bytes & Bytes.Bytes & Bool
401) -> Result<&1, &1, Error, Batch>:
402 match comparison:
403 case (_, _, False{}):
404 Fail{ChecksumMismatch{}}
405 case (_, _, True{}):
406 Done{Batch{records}}
407
408
409
410def frame.verify.end(
411 at_end: Bool,
412 bytes: Bytes.Bytes,
413 claimed: Bytes.Bytes,
414 records: List<&2, Mut>,
415 +end: U32
416) -> Result<&1, &1, Error, Batch>:
417 match at_end:
418 case False{}:
419 Fail{Malformed{}}
420 case True{}:
421 frame.verify.equal(records, Bytes.eq(claimed,
422 SHA.sha256_packed_bytes(frame.digest.prefix(
423 Bytes.slice(bytes, 0, (end - 32 : U32))))))
424
425
426
427def frame.verify.finish(
428 cursor: Bytes.Cursor,
429 claimed: Bytes.Bytes,
430 records: List<&2, Mut>
431) -> Result<&1, &1, Error, Batch>:
432 match cursor:
433 case Bytes.Cursor{bytes, +pos, _, +end}:
434 frame.verify.end(U32.is_eq(pos, end), bytes, claimed, records, end)
435
436
437def frame.verify.cursor(
438 pair: Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>,
439 records: List<&2, Mut>
440) -> Result<&1, &1, Error, Batch>:
441 match pair:
442 case (_, Fail{error}):
443 Fail{error}
444 case (cursor, Done{claimed}):
445 frame.verify.finish(cursor, claimed, records)
446
447
448
449def frame.records(
450 pair: Bytes.Cursor & Result<&1, &1, Error, List<&2, Mut>>
451) -> Result<&1, &1, Error, Batch>:
452 match pair:
453 case (_, Fail{error}):
454 Fail{error}
455 case (cursor, Done{muts}):
456 frame.verify.cursor(digest.read(cursor), muts)
457
458
459def frame.bound.valid(
460 +frame_len: U32,
461 +count: U32,
462 valid: Bool
463) -> Bool:
464 valid && U32.is_le(count,
465 U32.from_nat(Nat.div(U32.to_nat(frame_len), 5n))) &&
466 (U32.is_le(frame_len, StorageBytes.MAX_BLOCK_BYTES()) ||
467 U32.is_eq(count, 1) && U32.is_le(frame_len,
468 (StorageBytes.MAX_RECORD_BYTES() + 36 : U32)))
469
470def frame.bound.checked(
471 cursor: Bytes.Cursor,
472 +count: U32,
473 valid: Bool
474) -> Result<&1, &1, Error, Batch>:
475 match valid:
476 case False{}:
477 Fail{Malformed{}}
478 case True{}:
479 frame.records(records.read.start(cursor, count))
480
481
482def frame.bound.count(
483 pair: Bytes.Cursor & Maybe<&2, U32>,
484 +frame_len: U32,
485 within: Bool
486) -> Result<&1, &1, Error, Batch>:
487 match pair:
488 case (_, None{}):
489 Fail{Malformed{}}
490 case (cursor, Some{+count}):
491 frame.bound.checked(cursor, count,
492 frame.bound.valid(frame_len, count, within))
493
494
495def frame.bound.total.valid(
496 cursor: Bytes.Cursor,
497 +frame_len: U32,
498 valid: Bool
499) -> Result<&1, &1, Error, Batch>:
500 match valid:
501 case False{}:
502 Fail{Malformed{}}
503 case True{}:
504 frame.bound.count(Bytes.Cursor.u32be(cursor), frame_len,
505 U32.is_le(36, frame_len))
506
507
508def frame.bound.total(
509 cursor: Bytes.Cursor,
510 +frame_len: U32,
511 +end: U32,
512 checked: Maybe<&2, U32>
513) -> Result<&1, &1, Error, Batch>:
514 match checked:
515 case None{}:
516 Fail{TooLarge{}}
517 case Some{+expected_end}:
518 frame.bound.total.valid(cursor, frame_len,
519 U32.is_eq(expected_end, end))
520
521
522def frame.bound.end(
523 cursor: Bytes.Cursor,
524 +frame_len: U32
525) -> Result<&1, &1, Error, Batch>:
526 match cursor:
527 case Bytes.Cursor{bytes, +pos, +start, +end}:
528 frame.bound.total(Bytes.Cursor{bytes, pos, start, end},
529 frame_len, end, StorageBytes.checked_add(frame_len, 4))
530
531
532def frame.bound.length(
533 pair: Bytes.Cursor & Maybe<&2, U32>
534) -> Result<&1, &1, Error, Batch>:
535 match pair:
536 case (_, None{}):
537 Fail{Malformed{}}
538 case (cursor, Some{+frame_len}):
539 frame.bound.end(cursor, frame_len)
540
541
542# Decode frame for the packed WAL codec.
543def decode_frame(
544 bytes: Bytes.Bytes
545) -> Result<&1, &1, Error, Batch>:
546 frame.bound.length(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))