src/transfer/Backup.bend
mylsm-lsm-store@0.6.0.0 · 383 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.0/src/ByteWal.bend as ByteWal
4import mylsm-lsm-store@0.6.0.0/src/StorageBytes.bend as StorageBytes
5import mylsm-lsm-store@0.6.0.0/src/Wal.bend as Wal
6import mylsm-lsm-store@0.6.0.0/src/hub_sha/sha256.bend as SHA
7import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
8
9# Reports invalid ordering, configured block limits, and encoding failures.
10type Error is Data:
11 UnsupportedPlatform{}
12 InvalidOrder{}
13 ResourceLimit{}
14 CodecFailure{}
15 SourceFailure{code: U32, message: String}
16
17# Tracks emission of the header, data blocks, footer, and terminal state.
18type Phase is Data:
19 Header{}
20 Blocks{}
21 Footer{}
22 Finished{}
23
24# Owns an immutable logical view and the remaining export position.
25type Stream is Type:
26 Stream{phase: Phase, entries: List<&1, ByteMemTable.Entry>,
27 entry_count: U32, block_count: U32, block_index: U32}
28
29# Returns one independently checksummed chunk or terminal completion.
30type Step is Type:
31 Chunk{stream: Stream, bytes: Bytes.Bytes}
32 End{}
33
34# Holds one bounded entry page and the remaining ordered suffix.
35type BatchSplit is Type:
36 BatchSplit{batch: List<&1, ByteMemTable.Entry>,
37 rest: List<&1, ByteMemTable.Entry>, count: U32}
38
39def clone.pair(+len: U32, copied: Array<U32> & Array<U32>) -> Bytes.Bytes & Bytes.Bytes:
40 match copied:
41 case (left, right):
42 (Bytes.Bytes{len, left}, Bytes.Bytes{len, right})
43
44def clone.split(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:
45 match bytes:
46 case Bytes.Bytes{+len, buf}:
47 clone.pair(len, Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
48
49# Unwraps a fixed format marker decoded from its hexadecimal literal.
50def marker(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
51 match value:
52 case Some{bytes}: bytes
53 case None{}: Bytes.new(0)
54
55# Returns the version-independent backup file identity.
56def magic() -> Bytes.Bytes:
57 marker(Bytes.from_hex("4d594c534d424b31"))
58
59def block.magic() -> Bytes.Bytes:
60 marker(Bytes.from_hex("424c4b31"))
61
62def footer.magic() -> Bytes.Bytes:
63 marker(Bytes.from_hex("454e4431"))
64
65def u32.bytes(value: U32) -> Bytes.Bytes:
66 Bytes.set.u32be(Bytes.new(4), 0, value)
67
68def nat.duplicate.pair.step(pair: Nat & Nat) -> Nat & Nat:
69 match pair:
70 case (left, right): (Nat.add(1n, left), Nat.add(1n, right))
71
72def nat.duplicate.pair(value: Nat) -> Nat & Nat:
73 match value:
74 case 0n: (0n, 0n)
75 case 1n+rest: nat.duplicate.pair.step(nat.duplicate.pair(rest))
76
77def u32.duplicate.pair.result(pair: Nat & Nat) -> U32 & U32:
78 match pair:
79 case (left, right): (U32.from_nat(left), U32.from_nat(right))
80
81def u32.duplicate.pair(value: U32) -> U32 & U32:
82 u32.duplicate.pair.result(nat.duplicate.pair(U32.to_nat(value)))
83
84def blocks.count.remainder(pair: Nat & Nat) -> Nat:
85 match pair:
86 case (quotient, remainder):
87 match remainder:
88 case 0n: quotient
89 case _: Nat.add(quotient, 1n)
90
91def blocks.count.parts(entries: Nat & Nat) -> U32:
92 match entries:
93 case (for_division, for_remainder):
94 U32.from_nat(blocks.count.remainder((Nat.div(for_division, 16n),
95 Nat.mod(for_remainder, 16n))))
96
97def blocks.count(entries: Nat) -> U32:
98 blocks.count.parts(nat.duplicate.pair(entries))
99
100# Reserves three transitions per chunk and two finalization transitions.
101def write_fuel(block_count: Nat) -> Nat:
102 Nat.add(Nat.mul(3n, Nat.add(block_count, 2n)), 2n)
103
104def split.batch.go(
105 fuel: Nat,
106 entries: List<&1, ByteMemTable.Entry>,
107 reversed: List<&1, ByteMemTable.Entry>,
108 count: U32
109) -> BatchSplit:
110 match fuel:
111 case 0n: BatchSplit{List.reverse(&1, ByteMemTable.Entry, reversed), entries, count}
112 case 1n+rest:
113 match entries:
114 case Nil{}:
115 BatchSplit{List.reverse(&1, ByteMemTable.Entry, reversed), Nil{}, count}
116 case Con{entry, tail}:
117 split.batch.go(rest, tail, Con{entry, reversed}, (count + 1 : U32))
118
119def split.batch.ready(pair: List<&1, ByteMemTable.Entry> & Nat) -> BatchSplit:
120 match pair:
121 case (entries, count):
122 split.batch.go(Nat.min(16n, count), entries, Nil{}, 0)
123
124def split.batch(entries: List<&1, ByteMemTable.Entry>) -> BatchSplit:
125 split.batch.ready(ByteMemTable.count.keep(entries))
126
127def start.counted.count(
128 entries: List<&1, ByteMemTable.Entry>,
129 counts: Nat & Nat
130) -> Result<&1, &1, Error, Stream>:
131 match counts:
132 case (entry_count, block_total):
133 Done{Stream{Header{}, entries, U32.from_nat(entry_count),
134 blocks.count(block_total), 0}}
135
136def start.counted(
137 measured: List<&1, ByteMemTable.Entry> & Nat
138) -> Result<&1, &1, Error, Stream>:
139 match measured:
140 case (entries, count):
141 start.counted.count(entries, nat.duplicate.pair(count))
142
143def start.ordered(
144 entries: List<&1, ByteMemTable.Entry>,
145 ordered: Bool
146) -> Result<&1, &1, Error, Stream>:
147 match ordered:
148 case False{}: Fail{InvalidOrder{}}
149 case True{}: start.counted(ByteMemTable.count.keep(entries))
150
151def start.cloned(
152 copies: List<&1, ByteMemTable.Entry> & List<&1, ByteMemTable.Entry>
153) -> Result<&1, &1, Error, Stream>:
154 match copies:
155 case (validation, source):
156 start.ordered(ByteMemTable.compact.live(source),
157 ByteMemTable.ordered.strict(validation))
158
159# Starts export after removing tombstones and validating key order.
160def start(entries: List<&1, ByteMemTable.Entry>) -> Result<&1, &1, Error, Stream>:
161 start.cloned(ByteMemTable.entries.clone(entries))
162
163def header.bytes.counts(
164 entries: List<&1, ByteMemTable.Entry>,
165 entry_counts: U32 & U32,
166 block_counts: U32 & U32,
167 +block_index: U32
168) -> Stream & Bytes.Bytes:
169 match entry_counts:
170 case (state_entries, encoded_entries):
171 match block_counts:
172 case (state_blocks, encoded_blocks):
173 (Stream{Blocks{}, entries, state_entries, state_blocks, block_index},
174 Bytes.concat([magic(), u32.bytes(1), u32.bytes(encoded_entries),
175 u32.bytes(encoded_blocks)]))
176
177def header.bytes(stream: Stream) -> Stream & Bytes.Bytes:
178 match stream:
179 case Stream{Header{}, entries, entry_count, block_count, block_index}:
180 header.bytes.counts(entries, u32.duplicate.pair(entry_count),
181 u32.duplicate.pair(block_count), block_index)
182 case _: (stream, Bytes.new(0))
183
184# Converts one surviving entry into its byte WAL put representation.
185def mutation(entry: ByteMemTable.Entry) -> ByteWal.Mutation:
186 match entry:
187 case ByteMemTable.Entry{key, Some{value}}: ByteWal.Put{key, value}
188 case ByteMemTable.Entry{key, None{}}: ByteWal.Delete{key}
189
190# Converts one bounded page of entries to byte WAL mutations.
191def mutations(entries: List<&1, ByteMemTable.Entry>) -> List<&1, ByteWal.Mutation>:
192 match entries:
193 case Nil{}: Nil{}
194 case Con{entry, tail}: Con{mutation(entry), mutations(tail)}
195
196def block.header(
197 +index: U32,
198 +count: U32,
199 pair: Bytes.Bytes & U32
200) -> Bytes.Bytes:
201 match pair:
202 case (frame, +length):
203 Bytes.concat([block.magic(), u32.bytes(index), u32.bytes(count),
204 u32.bytes(length), frame])
205
206def block.authenticated(
207 pair: Bytes.Bytes & Bytes.Bytes
208) -> Bytes.Bytes:
209 match pair:
210 case (output, hash_input):
211 Bytes.concat([output, SHA.sha256_packed_bytes(hash_input)])
212
213def block.bytes(
214 frame: Bytes.Bytes,
215 +index: U32,
216 +count: U32
217) -> Bytes.Bytes:
218 block.authenticated(clone.split(block.header(index, count, Bytes.length(frame))))
219
220def block.frame.bounded(
221 frame: Bytes.Bytes,
222 +index: U32,
223 +count: U32,
224 fits: Bool
225) -> Result<&1, &1, Error, Bytes.Bytes>:
226 match fits:
227 case False{}: Fail{ResourceLimit{}}
228 case True{}: Done{block.bytes(frame, index, count)}
229
230def block.frame.length(
231 pair: Bytes.Bytes & U32,
232 +index: U32,
233 +count: U32
234) -> Result<&1, &1, Error, Bytes.Bytes>:
235 match pair:
236 case (frame, +length):
237 block.frame.bounded(frame, index, count,
238 U32.is_le(length, StorageBytes.MAX_BLOCK_BYTES()))
239
240def block.frame(
241 result: Result<&1, &1, Wal.Error, Bytes.Bytes>,
242 +index: U32,
243 +count: U32
244) -> Result<&1, &1, Error, Bytes.Bytes>:
245 match result:
246 case Fail{_}: Fail{CodecFailure{}}
247 case Done{frame}:
248 block.frame.length(Bytes.length(frame), index, count)
249
250def block.encoded(
251 entries: List<&1, ByteMemTable.Entry>,
252 +index: U32,
253 +count: U32
254) -> Result<&1, &1, Error, Bytes.Bytes>:
255 block.frame(ByteWal.encode(ByteWal.Batch{mutations(entries)}), index, count)
256
257def footer.bytes.output(
258 stream: Stream,
259 output: Bytes.Bytes,
260 checksum: Bytes.Bytes
261) -> Stream & Bytes.Bytes:
262 match stream:
263 case Stream{Finished{}, entries, entry_count, block_count, block_index}:
264 (Stream{Finished{}, entries, entry_count, block_count, block_index},
265 Bytes.concat([output, checksum]))
266 case _: (stream, Bytes.new(0))
267
268def footer.bytes.hashed(
269 stream: Stream,
270 pair: Bytes.Bytes & Bytes.Bytes
271) -> Stream & Bytes.Bytes:
272 match pair:
273 case (output, hash_input):
274 footer.bytes.output(stream, output, SHA.sha256_packed_bytes(hash_input))
275
276def footer.bytes.authenticated(
277 entries: List<&1, ByteMemTable.Entry>,
278 +entry_count: U32,
279 +block_count: U32,
280 +block_index: U32,
281 pair: Bytes.Bytes & Bytes.Bytes
282) -> Stream & Bytes.Bytes:
283 footer.bytes.hashed(Stream{Finished{}, entries, entry_count, block_count,
284 block_index}, pair)
285
286def footer.bytes.counts(
287 entries: List<&1, ByteMemTable.Entry>,
288 entry_counts: U32 & U32,
289 block_counts: U32 & U32,
290 +block_index: U32
291) -> Stream & Bytes.Bytes:
292 match entry_counts:
293 case (state_entries, encoded_entries):
294 match block_counts:
295 case (state_blocks, encoded_blocks):
296 footer.bytes.authenticated(entries, state_entries, state_blocks,
297 block_index, clone.split(Bytes.concat([footer.magic(),
298 u32.bytes(encoded_entries), u32.bytes(encoded_blocks)])))
299
300def footer.bytes(stream: Stream) -> Stream & Bytes.Bytes:
301 match stream:
302 case Stream{Footer{}, entries, entry_count, block_count, block_index}:
303 footer.bytes.counts(entries, u32.duplicate.pair(entry_count),
304 u32.duplicate.pair(block_count), block_index)
305 case _: (stream, Bytes.new(0))
306
307def next.block.encoded(
308 stream: Stream,
309 tail: List<&1, ByteMemTable.Entry>,
310 result: Result<&1, &1, Error, Bytes.Bytes>
311) -> Result<&1, &1, Error, Step>:
312 match stream:
313 case Stream{Blocks{}, _, entry_count, block_count, +block_index}:
314 match result:
315 case Fail{error}: Fail{error}
316 case Done{bytes}:
317 Done{Chunk{Stream{Blocks{}, tail, entry_count, block_count,
318 (block_index + 1 : U32)}, bytes}}
319 case _: Fail{CodecFailure{}}
320
321def next.block.ready.index(
322 +entry_count: U32,
323 +block_count: U32,
324 indexes: U32 & U32,
325 batch: List<&1, ByteMemTable.Entry>,
326 tail: List<&1, ByteMemTable.Entry>,
327 +count: U32
328) -> Result<&1, &1, Error, Step>:
329 match indexes:
330 case (state_index, encoded_index):
331 next.block.encoded(Stream{Blocks{}, Nil{}, entry_count, block_count,
332 state_index}, tail, block.encoded(batch, encoded_index, count))
333
334def next.block.footer(pair: Stream & Bytes.Bytes) -> Result<&1, &1, Error, Step>:
335 match pair:
336 case (stream, bytes): Done{Chunk{stream, bytes}}
337
338def next.block.ready.split(
339 +entry_count: U32,
340 +block_count: U32,
341 +block_index: U32,
342 split: BatchSplit
343) -> Result<&1, &1, Error, Step>:
344 match split:
345 case BatchSplit{Nil{}, entries, _}:
346 next.block.footer(footer.bytes(Stream{Footer{}, entries, entry_count,
347 block_count, block_index}))
348 case BatchSplit{batch, tail, +count}:
349 next.block.ready.index(entry_count, block_count,
350 u32.duplicate.pair(block_index), batch, tail, count)
351
352def next.block.ready(stream: Stream, split: BatchSplit) -> Result<&1, &1, Error, Step>:
353 match stream:
354 case Stream{Blocks{}, _, entry_count, block_count, block_index}:
355 next.block.ready.split(entry_count, block_count, block_index, split)
356 case _: Fail{CodecFailure{}}
357
358def next.block(stream: Stream) -> Result<&1, &1, Error, Step>:
359 match stream:
360 case Stream{Blocks{}, entries, entry_count, block_count, block_index}:
361 next.block.ready(Stream{Blocks{}, Nil{}, entry_count, block_count,
362 block_index}, split.batch(entries))
363 case _: Fail{CodecFailure{}}
364
365def next.header.result(pair: Stream & Bytes.Bytes) -> Result<&1, &1, Error, Step>:
366 match pair:
367 case (updated, bytes): Done{Chunk{updated, bytes}}
368
369def next.footer.result(pair: Stream & Bytes.Bytes) -> Result<&1, &1, Error, Step>:
370 match pair:
371 case (updated, bytes): Done{Chunk{updated, bytes}}
372
373# Advances an export by one header, block, footer, or terminal state.
374def next(stream: Stream) -> Result<&1, &1, Error, Step>:
375 match stream:
376 case Stream{Header{}, _, _, _, _}:
377 next.header.result(header.bytes(stream))
378 case Stream{Blocks{}, _, _, _, _}:
379 next.block(stream)
380 case Stream{Footer{}, _, _, _, _}:
381 next.footer.result(footer.bytes(stream))
382 case Stream{Finished{}, _, _, _, _}:
383 Done{End{}}