1	import Base
2	import ./MemTable.bend as MemTable
3	import ./SstFile.bend as SstFile
4	import ./StorageBytes.bend as StorageBytes
5	import ./hub_sha/sha256.bend as SHA
6	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7	
8	# Represent Mut data used by the write-ahead log data model.
9	type 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.
14	type Batch is Data:
15	  Batch{muts: List<&2, Mut>}
16	
17	# Handle stage in the write-ahead log data model.
18	def 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.
22	type Error is Data:
23	  TooLarge{}
24	  InvalidUtf8{}
25	  Malformed{}
26	  ChecksumMismatch{}
27	
28	def 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	
39	def 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	
50	def 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.
56	def 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.
60	def u8_bytes(value: U32) -> Bytes.Bytes:
61	  Bytes.set(Bytes.new(1), 0, value)
62	
63	def 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.
71	def log_header() -> Bytes.Bytes:
72	  header.finish(Bytes.from_hex("4d594c534d335700"))
73	
74	def 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.
84	def decode_header(bytes: Bytes.Bytes) -> Result<&1, &1, Error, Unit>:
85	  header.valid(Bytes.eq(log_header(), bytes))
86	
87	def 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	
100	def 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	
109	def 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	
124	def 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	
139	def 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	
149	def 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	
158	def 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	
171	def 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	
181	def 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	
190	def 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.
206	def 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	
215	def 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	
223	def 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	
228	def 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	
241	def 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	
251	def 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.
267	def 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	
272	def 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	
286	def 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	
296	def 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	
306	def 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.
317	def 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.
323	type DigestRead is Type:
324	  DigestInput{cursor: Bytes.Cursor, value: Maybe<&2, U32>}
325	
326	def digest.read.wrap(pair: Bytes.Cursor & Maybe<&2, U32>) -> DigestRead:
327	  match pair:
328	    case (cursor, value):
329	      DigestInput{cursor, value}
330	
331	def 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	
348	def 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.
355	type RecordRead is Type:
356	  RecordInput{cursor: Bytes.Cursor,
357	    result: Result<&1, &1, Error, Mut>}
358	
359	def 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	
366	def 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	
382	def 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	
393	def frame.digest.prefix(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
394	  match pair:
395	    case (_, prefix):
396	      prefix
397	
398	def 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	
410	def 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	
427	def 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	
437	def 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	
449	def 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	
459	def 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	
470	def 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	
482	def 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	
495	def 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	
508	def 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	
522	def 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	
532	def 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.
543	def decode_frame(
544	  bytes: Bytes.Bytes
545	) -> Result<&1, &1, Error, Batch>:
546	  frame.bound.length(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))