src/ByteManifest.bend
mylsm-lsm-store@0.6.0.0 · 460 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/hub_sha/sha256.bend as SHA
3import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
4
5# Represents a legacy or generation-addressed byte-store table layout.
6type ManifestValue is Data:
7 Empty{}
8 HasSst{}
9 GenerationEmpty{}
10 GenerationSst{file_id: U32}
11
12# Distinguishes malformed bytes, unknown formats, and comparator mismatch.
13type Error is Data:
14 Malformed{}
15 UnsupportedVersion{generation: U32}
16 ComparatorMismatch{generation: U32}
17 ChecksumMismatch{}
18
19# Reports the next generation identifier or exhaustion without wraparound.
20type FileIdDecision is Data:
21 FileIdAvailable{id: U32}
22 FileIdExhausted{}
23
24# Unwraps a fixed manifest-format literal.
25def literal(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
26 match value:
27 case Some{bytes}: bytes
28 case None{}: Bytes.new(0)
29
30# Returns the byte-manifest family identifier shared by generations four and five.
31def magic() -> Bytes.Bytes:
32 literal(Bytes.from_hex("4d594c534d344d00"))
33
34# Encodes one unsigned big-endian field.
35def u32.bytes(value: U32) -> Bytes.Bytes:
36 Bytes.set.u32be(Bytes.new(4), 0, value)
37
38# Returns the exact fixed file name referenced by the single-run manifest.
39def sst.name() -> Bytes.Bytes:
40 literal(Bytes.from_hex("535354"))
41
42# Names the SST file of one generation.
43def sst.generation.name(file_id: U32) -> String:
44 "SST." ++ Bytes.to_hex(u32.bytes(file_id))
45
46def sst.generation.path(directory: String, file_id: U32) -> String:
47 directory ++ "/" ++ sst.generation.name(file_id)
48
49# Creates two owned copies for hashing and output.
50def clone.pair(+len: U32, copies: Array<U32> & Array<U32>) -> Bytes.Bytes & Bytes.Bytes:
51 match copies:
52 case (left, right): (Bytes.Bytes{len, left}, Bytes.Bytes{len, right})
53
54# Clones a packed buffer for the manifest checksum.
55def clone.split(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:
56 match bytes:
57 case Bytes.Bytes{+len, buf}:
58 clone.pair(len, Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
59
60# Packs the three independently owned manifest copies.
61def clone.triple.parts(
62 +len: U32,
63 first: Array<U32>,
64 second: Array<U32>,
65 third: Array<U32>
66) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
67 (Bytes.Bytes{len, first}, Bytes.Bytes{len, second}, Bytes.Bytes{len, third})
68
69# Selects the retained copy from the final byte duplication.
70def clone.triple.finish(
71 +len: U32,
72 second: Array<U32>,
73 copies: Array<U32> & Array<U32>
74) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
75 match copies:
76 case (first, third): clone.triple.parts(len, first, second, third)
77
78# Adds an independently copied third buffer.
79def clone.triple.first(
80 +len: U32,
81 copies: Array<U32> & Array<U32>
82) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
83 match copies:
84 case (first, second):
85 clone.triple.finish(len, second,
86 Bytes.copy.bytes(len, first, Bytes.alloc(len), 0, 0))
87
88# Creates three owned copies for independent header-field checks.
89def clone.triple(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
90 match bytes:
91 case Bytes.Bytes{+len, buf}:
92 clone.triple.first(len,
93 Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
94
95# Builds the checksummed body for one manifest value.
96def body(value: ManifestValue) -> Bytes.Bytes:
97 match value:
98 case Empty{}:
99 Bytes.concat([magic(), u32.bytes(4), u32.bytes(1), u32.bytes(1), u32.bytes(0)])
100 case HasSst{}:
101 Bytes.concat([magic(), u32.bytes(4), u32.bytes(1), u32.bytes(1),
102 u32.bytes(1), u32.bytes(3), sst.name()])
103 case GenerationEmpty{}:
104 Bytes.concat([magic(), u32.bytes(5), u32.bytes(1), u32.bytes(1),
105 u32.bytes(0)])
106 case GenerationSst{file_id}:
107 Bytes.concat([magic(), u32.bytes(5), u32.bytes(1), u32.bytes(1),
108 u32.bytes(1), u32.bytes(file_id)])
109
110# Appends the digest of all preceding manifest bytes.
111def encode.body.copies(copies: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
112 match copies:
113 case (encoded, for_hash):
114 Bytes.concat([encoded, SHA.sha256_packed_bytes(for_hash)])
115
116# Appends the digest of all preceding manifest bytes.
117def encode.body(packed: Bytes.Bytes) -> Bytes.Bytes:
118 encode.body.copies(clone.split(packed))
119
120# Encodes the currently supported comparator and table layout.
121def encode(value: ManifestValue) -> Bytes.Bytes:
122 encode.body(body(value))
123
124def u32.read.value(pair: Bytes.Bytes & Maybe<&2, U32>) -> U32:
125 match pair:
126 case (_, Some{value}): value
127 case _: 0
128
129def u32.read(input: Bytes.Bytes) -> U32:
130 u32.read.value(Bytes.get.u32be(input, 0))
131
132def u32.clone.pair.bytes(copies: Bytes.Bytes & Bytes.Bytes) -> U32 & U32:
133 match copies:
134 case (left, right):
135 (u32.read(left), u32.read(right))
136
137def clone.file.id(file_id: U32) -> U32 & U32:
138 u32.clone.pair.bytes(clone.split(u32.bytes(file_id)))
139
140def next.file.id.current.result(next_id: U32, is_max: Bool) -> FileIdDecision:
141 match is_max:
142 case True{}: FileIdExhausted{}
143 case False{}: FileIdAvailable{U32.add(next_id, 1)}
144
145def next.file.id.current.copies(copies: U32 & U32) -> FileIdDecision:
146 match copies:
147 case (next_id, max_check):
148 next.file.id.current.result(next_id,
149 U32.is_eq(max_check, u32.read(literal(Bytes.from_hex("ffffffff")))))
150
151def next.file.id.current(file_id: U32) -> FileIdDecision:
152 next.file.id.current.copies(
153 u32.clone.pair.bytes(clone.split(u32.bytes(file_id))))
154
155def next.file.id(value: ManifestValue) -> FileIdDecision:
156 match value:
157 case Empty{}: FileIdAvailable{1}
158 case HasSst{}: FileIdAvailable{1}
159 case GenerationEmpty{}: FileIdAvailable{1}
160 case GenerationSst{file_id}: next.file.id.current(file_id)
161
162# Returns the second packed buffer from a slice operation.
163def slice.result(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
164 match pair:
165 case (_, sliced): sliced
166
167# Compares two packed values while preserving the equality result.
168def equal(pair: Bytes.Bytes & Bytes.Bytes & Bool) -> Bool:
169 match pair:
170 case (_, _, same): same
171
172# Returns the decoded value only when its body checksum is valid.
173def decode.checksum.compare(
174 value: ManifestValue,
175 valid: Bool
176) -> Result<&1, &1, Error, ManifestValue>:
177 match valid:
178 case True{}: Done{value}
179 case False{}: Fail{ChecksumMismatch{}}
180
181# Hashes one owned copy and compares the other with the stored digest.
182def decode.checksum.copies(
183 value: ManifestValue,
184 +offset: U32,
185 copies: Bytes.Bytes & Bytes.Bytes
186) -> Result<&1, &1, Error, ManifestValue>:
187 match copies:
188 case (for_hash, for_digest):
189 decode.checksum.compare(value,
190 equal(Bytes.eq(SHA.sha256_packed_bytes(slice.result(
191 Bytes.slice(for_hash, 0, offset))),
192 slice.result(Bytes.slice(for_digest, offset, 32)))))
193
194# Rejects a truncated or extended manifest before hashing it.
195def decode.checksum.size(
196 input: Bytes.Bytes,
197 value: ManifestValue,
198 +offset: U32,
199 exact: Bool
200) -> Result<&1, &1, Error, ManifestValue>:
201 match exact:
202 case False{}: Fail{Malformed{}}
203 case True{}:
204 decode.checksum.copies(value, offset, clone.split(input))
205
206# Compares the measured file size with its selected canonical layout.
207def decode.checksum.measured(
208 value: ManifestValue,
209 +offset: U32,
210 +expected_size: U32,
211 measured: Bytes.Bytes & U32
212) -> Result<&1, &1, Error, ManifestValue>:
213 match measured:
214 case (input, size):
215 decode.checksum.size(input, value, offset, U32.is_eq(size, expected_size))
216
217# Checks exact size and the SHA-256 checksum over the canonical body.
218def decode.checksum(
219 input: Bytes.Bytes,
220 value: ManifestValue,
221 +offset: U32,
222 +expected_size: U32
223) -> Result<&1, &1, Error, ManifestValue>:
224 decode.checksum.measured(value, offset, expected_size, Bytes.length(input))
225
226# Checks the fixed table name and its exact encoded length.
227def decode.sst.name.valid(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, ManifestValue>:
228 match valid:
229 case False{}: Fail{Malformed{}}
230 case True{}: decode.checksum(input, HasSst{}, 31, 63)
231
232# Checks the fixed SST name using independent packed-byte copies.
233def decode.sst_name.copies(
234 copies: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes
235) -> Result<&1, &1, Error, ManifestValue>:
236 match copies:
237 case (input, length_bytes, name_bytes):
238 decode.sst.name.valid(input,
239 equal(Bytes.eq(slice.result(Bytes.slice(length_bytes, 24, 4)), u32.bytes(3))) &&
240 equal(Bytes.eq(slice.result(Bytes.slice(name_bytes, 28, 3)), sst.name())))
241
242# Validates the only permitted SST file name.
243def decode.sst_name(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
244 decode.sst_name.copies(clone.triple(input))
245
246# Selects the empty or one-table layout from its declared file count.
247def decode.sst_count.value(
248 input: Bytes.Bytes,
249 field: Maybe<&2, U32>
250) -> Result<&1, &1, Error, ManifestValue>:
251 match field:
252 case None{}: Fail{Malformed{}}
253 case Some{0}: decode.checksum(input, Empty{}, 24, 56)
254 case Some{1}: decode.sst_name(input)
255 case _: Fail{Malformed{}}
256
257def decode.sst_count.field(
258 pair: Bytes.Bytes & Maybe<&2, U32>
259) -> Result<&1, &1, Error, ManifestValue>:
260 match pair:
261 case (input, field): decode.sst_count.value(input, field)
262
263def decode.sst_count(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
264 decode.sst_count.field(Bytes.get.u32be(input, 20))
265
266# Requires the canonical single level used by the byte durable engine.
267def decode.level_count.value(
268 input: Bytes.Bytes,
269 field: Maybe<&2, U32>
270) -> Result<&1, &1, Error, ManifestValue>:
271 match field:
272 case None{}: Fail{Malformed{}}
273 case Some{1}: decode.sst_count(input)
274 case _: Fail{Malformed{}}
275
276def decode.level_count.field(
277 pair: Bytes.Bytes & Maybe<&2, U32>
278) -> Result<&1, &1, Error, ManifestValue>:
279 match pair:
280 case (input, field): decode.level_count.value(input, field)
281
282def decode.level_count(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
283 decode.level_count.field(Bytes.get.u32be(input, 16))
284
285# Rejects manifests written for a different key comparator.
286def decode.comparator.value(
287 input: Bytes.Bytes,
288 field: Maybe<&2, U32>
289) -> Result<&1, &1, Error, ManifestValue>:
290 match field:
291 case None{}: Fail{Malformed{}}
292 case Some{1}: decode.level_count(input)
293 case Some{generation}: Fail{ComparatorMismatch{generation}}
294
295def decode.comparator.field(
296 pair: Bytes.Bytes & Maybe<&2, U32>
297) -> Result<&1, &1, Error, ManifestValue>:
298 match pair:
299 case (input, field): decode.comparator.value(input, field)
300
301def decode.comparator(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
302 decode.comparator.field(Bytes.get.u32be(input, 12))
303
304def decode.v5.sst.value(
305 input: Bytes.Bytes,
306 field: Maybe<&2, U32>
307) -> Result<&1, &1, Error, ManifestValue>:
308 match field:
309 case None{}: Fail{Malformed{}}
310 case Some{file_id}:
311 decode.checksum(input, GenerationSst{file_id}, 28, 60)
312
313def decode.v5.sst.field(
314 pair: Bytes.Bytes & Maybe<&2, U32>
315) -> Result<&1, &1, Error, ManifestValue>:
316 match pair:
317 case (input, field): decode.v5.sst.value(input, field)
318
319def decode.v5.sst.count.value(
320 input: Bytes.Bytes,
321 count_field: Maybe<&2, U32>
322) -> Result<&1, &1, Error, ManifestValue>:
323 match count_field:
324 case None{}: Fail{Malformed{}}
325 case Some{0}:
326 decode.checksum(input, GenerationEmpty{}, 24, 56)
327 case Some{1}:
328 decode.v5.sst.field(Bytes.get.u32be(input, 24))
329 case _: Fail{Malformed{}}
330
331def decode.v5.sst.count.field(
332 pair: Bytes.Bytes & Maybe<&2, U32>
333) -> Result<&1, &1, Error, ManifestValue>:
334 match pair:
335 case (input, field): decode.v5.sst.count.value(input, field)
336
337def decode.v5.level_count.value(
338 input: Bytes.Bytes,
339 field: Maybe<&2, U32>
340) -> Result<&1, &1, Error, ManifestValue>:
341 match field:
342 case None{}: Fail{Malformed{}}
343 case Some{1}:
344 decode.v5.sst.count.field(Bytes.get.u32be(input, 20))
345 case _: Fail{Malformed{}}
346
347def decode.v5.level_count.field(
348 pair: Bytes.Bytes & Maybe<&2, U32>
349) -> Result<&1, &1, Error, ManifestValue>:
350 match pair:
351 case (input, field): decode.v5.level_count.value(input, field)
352
353def decode.v5.level_count(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
354 decode.v5.level_count.field(Bytes.get.u32be(input, 16))
355
356def decode.v5.comparator.value(
357 input: Bytes.Bytes,
358 field: Maybe<&2, U32>
359) -> Result<&1, &1, Error, ManifestValue>:
360 match field:
361 case None{}: Fail{Malformed{}}
362 case Some{1}:
363 decode.v5.level_count(input)
364 case Some{generation}: Fail{ComparatorMismatch{generation}}
365
366def decode.v5.comparator.field(
367 pair: Bytes.Bytes & Maybe<&2, U32>
368) -> Result<&1, &1, Error, ManifestValue>:
369 match pair:
370 case (input, field): decode.v5.comparator.value(input, field)
371
372def decode.v5.comparator(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
373 decode.v5.comparator.field(Bytes.get.u32be(input, 12))
374
375# Rejects unknown storage generations before comparator validation.
376def decode.version.value(
377 input: Bytes.Bytes,
378 field: Maybe<&2, U32>
379) -> Result<&1, &1, Error, ManifestValue>:
380 match field:
381 case None{}: Fail{Malformed{}}
382 case Some{4}: decode.comparator(input)
383 case Some{5}: decode.v5.comparator(input)
384 case Some{generation}: Fail{UnsupportedVersion{generation}}
385
386def decode.version.field(
387 pair: Bytes.Bytes & Maybe<&2, U32>
388) -> Result<&1, &1, Error, ManifestValue>:
389 match pair:
390 case (input, field): decode.version.value(input, field)
391
392def decode.version(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
393 decode.version.field(Bytes.get.u32be(input, 8))
394
395# Checks the magic before interpreting version fields.
396def decode.magic(input: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, ManifestValue>:
397 match valid:
398 case False{}: Fail{Malformed{}}
399 case True{}: decode.version(input)
400
401def decode.magic.pair(pair: Bytes.Bytes & Bool) -> Result<&1, &1, Error, ManifestValue>:
402 match pair:
403 case (input, valid): decode.magic(input, valid)
404
405def decode.magic.copies(
406 copies: Bytes.Bytes & Bytes.Bytes
407) -> Result<&1, &1, Error, ManifestValue>:
408 match copies:
409 case (input, for_magic):
410 decode.magic.pair((input,
411 equal(Bytes.eq(slice.result(Bytes.slice(for_magic, 0, 8)), magic()))))
412
413# Validates the size before reading fixed header fields.
414def decode.sized.check_magic(bytes: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
415 decode.magic.copies(clone.split(bytes))
416
417def decode.sized.valid(bytes: Bytes.Bytes, valid: Bool) -> Result<&1, &1, Error, ManifestValue>:
418 match valid:
419 case False{}: Fail{Malformed{}}
420 case True{}: decode.sized.check_magic(bytes)
421
422# Validates the size before reading fixed header fields.
423def decode.sized(input: Bytes.Bytes & U32) -> Result<&1, &1, Error, ManifestValue>:
424 match input:
425 case (bytes, size):
426 match size:
427 case 56: decode.sized.valid(bytes, True{})
428 case 63: decode.sized.valid(bytes, True{})
429 case 60: decode.sized.valid(bytes, True{})
430 case _: decode.sized.valid(bytes, False{})
431
432# Decodes bounded generation-four and generation-five byte manifests.
433def decode(input: Bytes.Bytes) -> Result<&1, &1, Error, ManifestValue>:
434 decode.sized(Bytes.length(input))
435
436def retired.current(value: ManifestValue) -> Maybe<&2, String>:
437 match value:
438 case GenerationSst{file_id}: Some{sst.generation.name(file_id)}
439 case _: None{}
440
441def retired.published(current: Maybe<&2, String>, +name: String) -> Bool:
442 match current:
443 case None{}: False{}
444 case Some{published}: String.eq(published, name)
445
446def retired.push(is_retired: Bool, name: String, rest: List<&2, String>) -> List<&2, String>:
447 match is_retired:
448 case True{}: Con{name, rest}
449 case False{}: rest
450
451def retired.go(names: List<&2, String>, +current: Maybe<&2, String>) -> List<&2, String>:
452 match names:
453 case Nil{}: Nil{}
454 case Con{+name, rest}:
455 retired.push(String.starts_with(name, "SST.") &&
456 Bool.not(retired.published(current, name)), name, retired.go(rest, current))
457
458# Lists the SST files in a directory listing that the manifest no longer references.
459def retired(value: ManifestValue, names: List<&2, String>) -> List<&2, String>:
460 retired.go(names, retired.current(value))