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	def 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	
283	def 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	
293	def 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	
303	def 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.
314	def 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.
320	type DigestRead is Type:
321	  DigestInput{cursor: Bytes.Cursor, value: Maybe<&2, U32>}
322	
323	def digest.read.wrap(pair: Bytes.Cursor & Maybe<&2, U32>) -> DigestRead:
324	  match pair:
325	    case (cursor, value):
326	      DigestInput{cursor, value}
327	
328	def 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	
345	def 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.
352	type RecordRead is Type:
353	  RecordInput{cursor: Bytes.Cursor,
354	    result: Result<&1, &1, Error, Mut>}
355	
356	def 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	
363	def 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	
379	def 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	
390	def frame.digest.prefix(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
391	  match pair:
392	    case (_, prefix):
393	      prefix
394	
395	def 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	
407	def 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	
424	def 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	
434	def 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	
446	def 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	
456	def 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	
467	def 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	
479	def 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	
492	def 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	
505	def 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	
519	def 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	
529	def 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.
540	def decode_frame(
541	  bytes: Bytes.Bytes
542	) -> Result<&1, &1, Error, Batch>:
543	  frame.bound.length(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))