1	import Base
2	import ../ByteMemTable.bend as ByteMemTable
3	import ../ByteWal.bend as ByteWal
4	import ../StorageBytes.bend as StorageBytes
5	import ../Wal.bend as Wal
6	import ../hub_sha/sha256.bend as SHA
7	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
8	
9	# Reports invalid ordering, configured block limits, and encoding failures.
10	type 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.
18	type Phase is Data:
19	  Header{}
20	  Blocks{}
21	  Footer{}
22	  Finished{}
23	
24	# Owns an immutable logical view and the remaining export position.
25	type 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.
30	type Step is Type:
31	  Chunk{stream: Stream, bytes: Bytes.Bytes}
32	  End{}
33	
34	# Holds one bounded entry page and the remaining ordered suffix.
35	type BatchSplit is Type:
36	  BatchSplit{batch: List<&1, ByteMemTable.Entry>,
37	    rest: List<&1, ByteMemTable.Entry>, count: U32}
38	
39	def 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	
44	def 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.
50	def 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.
56	def magic() -> Bytes.Bytes:
57	  marker(Bytes.from_hex("4d594c534d424b31"))
58	
59	def block.magic() -> Bytes.Bytes:
60	  marker(Bytes.from_hex("424c4b31"))
61	
62	def footer.magic() -> Bytes.Bytes:
63	  marker(Bytes.from_hex("454e4431"))
64	
65	def u32.bytes(value: U32) -> Bytes.Bytes:
66	  Bytes.set.u32be(Bytes.new(4), 0, value)
67	
68	def 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	
72	def 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	
77	def 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	
81	def u32.duplicate.pair(value: U32) -> U32 & U32:
82	  u32.duplicate.pair.result(nat.duplicate.pair(U32.to_nat(value)))
83	
84	def 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	
91	def 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	
97	def blocks.count(entries: Nat) -> U32:
98	  blocks.count.parts(nat.duplicate.pair(entries))
99	
100	# Reserves three transitions per chunk and two finalization transitions.
101	def write_fuel(block_count: Nat) -> Nat:
102	  Nat.add(Nat.mul(3n, Nat.add(block_count, 2n)), 2n)
103	
104	def 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	
119	def 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	
124	def split.batch(entries: List<&1, ByteMemTable.Entry>) -> BatchSplit:
125	  split.batch.ready(ByteMemTable.count.keep(entries))
126	
127	def 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	
136	def 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	
143	def 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	
151	def 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.
160	def start(entries: List<&1, ByteMemTable.Entry>) -> Result<&1, &1, Error, Stream>:
161	  start.cloned(ByteMemTable.entries.clone(entries))
162	
163	def 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	
177	def 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.
185	def 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.
191	def 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	
196	def 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	
206	def 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	
213	def 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	
220	def 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	
230	def 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	
240	def 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	
250	def 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	
257	def 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	
268	def 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	
276	def 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	
286	def 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	
300	def 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	
307	def 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	
321	def 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	
334	def next.block.footer(pair: Stream & Bytes.Bytes) -> Result<&1, &1, Error, Step>:
335	  match pair:
336	    case (stream, bytes): Done{Chunk{stream, bytes}}
337	
338	def 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	
352	def 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	
358	def 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	
365	def next.header.result(pair: Stream & Bytes.Bytes) -> Result<&1, &1, Error, Step>:
366	  match pair:
367	    case (updated, bytes): Done{Chunk{updated, bytes}}
368	
369	def 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.
374	def 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{}}