src/Wal.bend
mylsm-lsm-store@0.4.0.0 · 543 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.4.0.0/src/MemTable.bend as MemTable
3import mylsm-lsm-store@0.4.0.0/src/SstFile.bend as SstFile
4import mylsm-lsm-store@0.4.0.0/src/StorageBytes.bend as StorageBytes
5import mylsm-lsm-store@0.4.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
266def frame.payload.fin(
267 +count: U32,
268 payload: Bytes.Bytes,
269 hash_payload: Bytes.Bytes,
270 frame_len: Maybe<&2, U32>
271) -> Result<&1, &1, Error, Bytes.Bytes>:
272 match frame_len:
273 case None{}:
274 Fail{TooLarge{}}
275 case Some{+length}:
276 frame.payload.valid(count, payload, length, hash_payload,
277 U32.is_le(length, StorageBytes.MAX_BLOCK_BYTES()) ||
278 U32.is_eq(count, 1) && U32.is_le(length,
279 (StorageBytes.MAX_RECORD_BYTES() + 36 : U32)))
280
281
282
283def frame.payload.output(
284 count: U32,
285 hash_payload: Bytes.Bytes,
286 measured: Bytes.Bytes & U32
287) -> Result<&1, &1, Error, Bytes.Bytes>:
288 match measured:
289 case (payload, +payload_len):
290 frame.payload.fin(count, payload, hash_payload,
291 StorageBytes.checked_add(payload_len, 36))
292
293def frame.payload.size(
294 count: U32,
295 pair: Bytes.Bytes & Bytes.Bytes
296) -> Result<&1, &1, Error, Bytes.Bytes>:
297 match pair:
298 case (output, hash_payload):
299 frame.payload.output(count, hash_payload, Bytes.length(output))
300
301
302
303def frame.payload(
304 +count: U32,
305 records: Result<&1, &1, Error, List<&1, Bytes.Bytes>>
306) -> Result<&1, &1, Error, Bytes.Bytes>:
307 match records:
308 case Fail{error}:
309 Fail{error}
310 case Done{parts}:
311 frame.payload.size(count, bytes.clone.split(Bytes.concat(parts)))
312
313# Encode frame for the packed WAL codec.
314def encode_frame(batch: Batch) -> Result<&1, &1, Error, Bytes.Bytes>:
315 match batch:
316 case Batch{+muts}:
317 frame.payload(U32.from_nat(List.length(&2, Mut, muts)), records(muts))
318
319# Represent DigestRead data used by the packed WAL codec.
320type DigestRead is Type:
321 DigestInput{cursor: Bytes.Cursor, value: Maybe<&2, U32>}
322
323def digest.read.wrap(pair: Bytes.Cursor & Maybe<&2, U32>) -> DigestRead:
324 match pair:
325 case (cursor, value):
326 DigestInput{cursor, value}
327
328def digest.read.go(
329 remaining: Nat,
330 input: DigestRead,
331 digest: Bytes.Bytes,
332 +offset: U32
333) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:
334 match remaining input:
335 case 0n DigestInput{cursor, _}:
336 (cursor, Done{digest})
337 case 1n+0n DigestInput{cursor, Some{word}}:
338 (cursor, Done{Bytes.set.u32be(digest, offset, word)})
339 case 1n+1n+rest DigestInput{cursor, Some{word}}:
340 digest.read.go(1n+rest, digest.read.wrap(Bytes.Cursor.u32be(cursor)),
341 Bytes.set.u32be(digest, offset, word), (offset + 4 : U32))
342 case 1n+_ DigestInput{cursor, None{}}:
343 (cursor, Fail{Malformed{}})
344
345def digest.read(
346 cursor: Bytes.Cursor
347) -> Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>:
348 digest.read.go(8n, digest.read.wrap(Bytes.Cursor.u32be(cursor)),
349 Bytes.new(32), 0)
350
351# Represent RecordRead data used by the packed WAL codec.
352type RecordRead is Type:
353 RecordInput{cursor: Bytes.Cursor,
354 result: Result<&1, &1, Error, Mut>}
355
356def records.read.wrap(
357 pair: Bytes.Cursor & Result<&1, &1, Error, Mut>
358) -> RecordRead:
359 match pair:
360 case (cursor, result):
361 RecordInput{cursor, result}
362
363def records.read.go(
364 remaining: Nat,
365 input: RecordRead,
366 reversed: List<&2, Mut>
367) -> Bytes.Cursor & Result<&1, &1, Error, List<&2, Mut>>:
368 match remaining input:
369 case 0n RecordInput{cursor, Fail{error}}:
370 (cursor, Fail{error})
371 case 0n RecordInput{cursor, Done{mut}}:
372 (cursor, Done{List.reverse(&2, Mut, mut <> reversed)})
373 case 1n+rest RecordInput{cursor, Fail{error}}:
374 (cursor, Fail{error})
375 case 1n+rest RecordInput{cursor, Done{mut}}:
376 records.read.go(rest, records.read.wrap(record.read(cursor)),
377 mut <> reversed)
378
379def records.read.start(
380 cursor: Bytes.Cursor,
381 +count: U32
382) -> Bytes.Cursor & Result<&1, &1, Error, List<&2, Mut>>:
383 match count:
384 case 0:
385 (cursor, Done{Nil{}})
386 case _:
387 records.read.go(Nat.sub(U32.to_nat(count), 1n),
388 records.read.wrap(record.read(cursor)), Nil{})
389
390def frame.digest.prefix(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
391 match pair:
392 case (_, prefix):
393 prefix
394
395def frame.verify.equal(
396 records: List<&2, Mut>,
397 comparison: Bytes.Bytes & Bytes.Bytes & Bool
398) -> Result<&1, &1, Error, Batch>:
399 match comparison:
400 case (_, _, False{}):
401 Fail{ChecksumMismatch{}}
402 case (_, _, True{}):
403 Done{Batch{records}}
404
405
406
407def frame.verify.end(
408 at_end: Bool,
409 bytes: Bytes.Bytes,
410 claimed: Bytes.Bytes,
411 records: List<&2, Mut>,
412 +end: U32
413) -> Result<&1, &1, Error, Batch>:
414 match at_end:
415 case False{}:
416 Fail{Malformed{}}
417 case True{}:
418 frame.verify.equal(records, Bytes.eq(claimed,
419 SHA.sha256_packed_bytes(frame.digest.prefix(
420 Bytes.slice(bytes, 0, (end - 32 : U32))))))
421
422
423
424def frame.verify.finish(
425 cursor: Bytes.Cursor,
426 claimed: Bytes.Bytes,
427 records: List<&2, Mut>
428) -> Result<&1, &1, Error, Batch>:
429 match cursor:
430 case Bytes.Cursor{bytes, +pos, _, +end}:
431 frame.verify.end(U32.is_eq(pos, end), bytes, claimed, records, end)
432
433
434def frame.verify.cursor(
435 pair: Bytes.Cursor & Result<&1, &1, Error, Bytes.Bytes>,
436 records: List<&2, Mut>
437) -> Result<&1, &1, Error, Batch>:
438 match pair:
439 case (_, Fail{error}):
440 Fail{error}
441 case (cursor, Done{claimed}):
442 frame.verify.finish(cursor, claimed, records)
443
444
445
446def frame.records(
447 pair: Bytes.Cursor & Result<&1, &1, Error, List<&2, Mut>>
448) -> Result<&1, &1, Error, Batch>:
449 match pair:
450 case (_, Fail{error}):
451 Fail{error}
452 case (cursor, Done{muts}):
453 frame.verify.cursor(digest.read(cursor), muts)
454
455
456def frame.bound.valid(
457 +frame_len: U32,
458 +count: U32,
459 valid: Bool
460) -> Bool:
461 valid && U32.is_le(count,
462 U32.from_nat(Nat.div(U32.to_nat(frame_len), 5n))) &&
463 (U32.is_le(frame_len, StorageBytes.MAX_BLOCK_BYTES()) ||
464 U32.is_eq(count, 1) && U32.is_le(frame_len,
465 (StorageBytes.MAX_RECORD_BYTES() + 36 : U32)))
466
467def frame.bound.checked(
468 cursor: Bytes.Cursor,
469 +count: U32,
470 valid: Bool
471) -> Result<&1, &1, Error, Batch>:
472 match valid:
473 case False{}:
474 Fail{Malformed{}}
475 case True{}:
476 frame.records(records.read.start(cursor, count))
477
478
479def frame.bound.count(
480 pair: Bytes.Cursor & Maybe<&2, U32>,
481 +frame_len: U32,
482 within: Bool
483) -> Result<&1, &1, Error, Batch>:
484 match pair:
485 case (_, None{}):
486 Fail{Malformed{}}
487 case (cursor, Some{+count}):
488 frame.bound.checked(cursor, count,
489 frame.bound.valid(frame_len, count, within))
490
491
492def frame.bound.total.valid(
493 cursor: Bytes.Cursor,
494 +frame_len: U32,
495 valid: Bool
496) -> Result<&1, &1, Error, Batch>:
497 match valid:
498 case False{}:
499 Fail{Malformed{}}
500 case True{}:
501 frame.bound.count(Bytes.Cursor.u32be(cursor), frame_len,
502 U32.is_le(36, frame_len))
503
504
505def frame.bound.total(
506 cursor: Bytes.Cursor,
507 +frame_len: U32,
508 +end: U32,
509 checked: Maybe<&2, U32>
510) -> Result<&1, &1, Error, Batch>:
511 match checked:
512 case None{}:
513 Fail{TooLarge{}}
514 case Some{+expected_end}:
515 frame.bound.total.valid(cursor, frame_len,
516 U32.is_eq(expected_end, end))
517
518
519def frame.bound.end(
520 cursor: Bytes.Cursor,
521 +frame_len: U32
522) -> Result<&1, &1, Error, Batch>:
523 match cursor:
524 case Bytes.Cursor{bytes, +pos, +start, +end}:
525 frame.bound.total(Bytes.Cursor{bytes, pos, start, end},
526 frame_len, end, StorageBytes.checked_add(frame_len, 4))
527
528
529def frame.bound.length(
530 pair: Bytes.Cursor & Maybe<&2, U32>
531) -> Result<&1, &1, Error, Batch>:
532 match pair:
533 case (_, None{}):
534 Fail{Malformed{}}
535 case (cursor, Some{+frame_len}):
536 frame.bound.end(cursor, frame_len)
537
538
539# Decode frame for the packed WAL codec.
540def decode_frame(
541 bytes: Bytes.Bytes
542) -> Result<&1, &1, Error, Batch>:
543 frame.bound.length(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))