src/ByteWalIo.bend
mylsm-lsm-store@0.6.0.0 · 652 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/Fs.bend as Fs
5import mylsm-lsm-store@0.6.0.0/src/SstStreamIo.bend as SstStreamIo
6import mylsm-lsm-store@0.6.0.0/src/Wal.bend as Wal
7import mylsm-lsm-store@0.6.0.0/src/CrashPoint.bend as CrashPoint
8import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
9
10# Tracks the owned file and bounded phase state during byte WAL replay.
11type RecoveryState is Type:
12 At{file: File, size: Nat, offset: Nat, table: ByteMemTable.Table,
13 at_end: Bool, can_read_length: Bool}
14 Length{pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>,
15 size: Nat, offset: Nat, table: ByteMemTable.Table}
16 Prefix{file: File, size: Nat, offset: Nat, table: ByteMemTable.Table,
17 prefix: Bytes.Bytes, frame_length: U32, within_file: Bool}
18 Count{pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>,
19 size: Nat, offset: Nat, table: ByteMemTable.Table,
20 prefix: Bytes.Bytes, frame_length: U32}
21 Counted{file: File, size: Nat, offset: Nat, table: ByteMemTable.Table,
22 prefix: Bytes.Bytes, count_bytes: Bytes.Bytes, frame_length: U32,
23 within_limit: Bool}
24 Body{pair: File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult>,
25 size: Nat, offset: Nat, table: ByteMemTable.Table,
26 prefix: Bytes.Bytes, count_bytes: Bytes.Bytes, frame_length: U32}
27 Decoded{file: File, size: Nat, next_offset: Nat,
28 table: ByteMemTable.Table,
29 result: Result<&1, &1, Wal.Error, ByteWal.Batch>}
30 Failed{file: File, error: U32 & String}
31 Tail{file: File, path: String, offset: Nat, table: ByteMemTable.Table}
32
33# Assigns the stable file-corruption category used by this reader.
34def corruption_error_code() -> U32:
35 4294967292
36
37# Pairs a corruption category with its diagnostic text.
38def error(message: String) -> U32 & String:
39 (corruption_error_code(), message)
40
41def u32.read(pair: Bytes.Cursor & Maybe<&2, U32>) -> Maybe<&2, U32>:
42 match pair:
43 case (_, value):
44 value
45
46def frame.length(bytes: Bytes.Bytes) -> Maybe<&2, U32>:
47 u32.read(Bytes.Cursor.u32be(Bytes.Cursor.new(bytes)))
48
49def frame.count(bytes: Bytes.Bytes) -> Maybe<&2, U32>:
50 frame.length(bytes)
51
52def recover.at(
53 file: File,
54 +size: Nat,
55 +offset: Nat,
56 table: ByteMemTable.Table
57) -> RecoveryState:
58 At{file, size, offset, table, Nat.is_eq(offset, size),
59 Nat.is_le(Nat.add(offset, 4n), size)}
60
61def recover.prefix.next(
62 file: File,
63 +size: Nat,
64 +offset: Nat,
65 table: ByteMemTable.Table,
66 prefix: Bytes.Bytes,
67 parsed: Maybe<&2, U32>
68) -> RecoveryState:
69 match parsed:
70 case None{}:
71 Failed{file, error("malformed byte WAL frame length")}
72 case Some{+frame_length}:
73 Prefix{file, size, offset, table, prefix, frame_length,
74 Nat.is_le(Nat.add(offset, Nat.add(4n, U32.to_nat(frame_length))), size) &&
75 U32.is_le(36, frame_length)}
76
77def recover.prefix.cloned(
78 file: File,
79 pair: Bytes.Bytes & Bytes.Bytes,
80 +size: Nat,
81 +offset: Nat,
82 table: ByteMemTable.Table
83) -> RecoveryState:
84 match pair:
85 case (prefix, inspected):
86 recover.prefix.next(file, size, offset, table,
87 prefix, frame.length(inspected))
88
89def recover.counted(
90 file: File,
91 +size: Nat,
92 +offset: Nat,
93 table: ByteMemTable.Table,
94 prefix: Bytes.Bytes,
95 count_bytes: Bytes.Bytes,
96 +frame_length: U32,
97 parsed: Maybe<&2, U32>
98) -> RecoveryState:
99 match parsed:
100 case None{}:
101 Failed{file, error("malformed byte WAL frame count")}
102 case Some{+count}:
103 Counted{file, size, offset, table, prefix, count_bytes,
104 frame_length, Wal.frame_length_allowed(count, frame_length)}
105
106def recover.count.cloned(
107 file: File,
108 pair: Bytes.Bytes & Bytes.Bytes,
109 +size: Nat,
110 +offset: Nat,
111 table: ByteMemTable.Table,
112 prefix: Bytes.Bytes,
113 +frame_length: U32
114) -> RecoveryState:
115 match pair:
116 case (count_bytes, inspected):
117 recover.counted(file, size, offset, table, prefix,
118 count_bytes, frame_length, frame.count(inspected))
119
120def apply.batch.loop(
121 fuel: Nat,
122 table: ByteMemTable.Table,
123 mutations: List<&1, ByteWal.Mutation>
124) -> ByteMemTable.Table:
125 match fuel:
126 case 0n:
127 table
128 case 1n+rest:
129 match mutations:
130 case Nil{}:
131 table
132 case Con{ByteWal.Put{key, value}, tail}:
133 apply.batch.loop(rest, ByteMemTable.put(table, key, value), tail)
134 case Con{ByteWal.Delete{key}, tail}:
135 apply.batch.loop(rest, ByteMemTable.del(table, key), tail)
136
137def mutations.count.go(
138 mutations: List<&1, ByteWal.Mutation>,
139 reversed: List<&1, ByteWal.Mutation>,
140 count: Nat
141) -> List<&1, ByteWal.Mutation> & Nat:
142 match mutations:
143 case Nil{}:
144 (List.reverse(&1, ByteWal.Mutation, reversed), count)
145 case Con{mutation, tail}:
146 mutations.count.go(tail, Con{mutation, reversed}, Nat.add(count, 1n))
147
148def mutations.count(mutations: List<&1, ByteWal.Mutation>) -> List<&1, ByteWal.Mutation> & Nat:
149 mutations.count.go(mutations, Nil{}, 0n)
150
151def apply.batch.counted(
152 table: ByteMemTable.Table,
153 pair: List<&1, ByteWal.Mutation> & Nat
154) -> ByteMemTable.Table:
155 match pair:
156 case (mutations, count):
157 apply.batch.loop(Nat.add(count, 1n), table, mutations)
158
159def apply.batch(table: ByteMemTable.Table, batch: ByteWal.Batch) -> ByteMemTable.Table:
160 match batch:
161 case ByteWal.Batch{mutations}:
162 apply.batch.counted(table, mutations.count(mutations))
163
164def close.failed(file: File, failure: U32 & String) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
165 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
166 _closed : Unit <- File.close(file)
167 return Fail{failure}
168
169def close.done(file: File, table: ByteMemTable.Table) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
170 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
171 _closed : Unit <- File.close(file)
172 return Done{table}
173
174def recover.exhausted(state: RecoveryState) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
175 match state:
176 case At{file, _, _, _, _, _}:
177 close.failed(file, error("byte WAL recovery limit exceeded"))
178 case Length{pair, _, _, _}:
179 match pair:
180 case (file, _):
181 close.failed(file, error("byte WAL recovery limit exceeded"))
182 case Prefix{file, _, _, _, _, _, _}:
183 close.failed(file, error("byte WAL recovery limit exceeded"))
184 case Count{pair, _, _, _, _, _}:
185 match pair:
186 case (file, _):
187 close.failed(file, error("byte WAL recovery limit exceeded"))
188 case Counted{file, _, _, _, _, _, _, _}:
189 close.failed(file, error("byte WAL recovery limit exceeded"))
190 case Body{pair, _, _, _, _, _, _}:
191 match pair:
192 case (file, _):
193 close.failed(file, error("byte WAL recovery limit exceeded"))
194 case Decoded{file, _, _, _, _}:
195 close.failed(file, error("byte WAL recovery limit exceeded"))
196 case Failed{file, failure}:
197 close.failed(file, failure)
198 case Tail{file, _, _, _}:
199 close.failed(file, error("byte WAL recovery limit exceeded"))
200
201def recover.length.completed(
202 +path: String,
203 file: File,
204 +size: Nat,
205 +offset: Nat,
206 table: ByteMemTable.Table,
207 bytes: Bytes.Bytes,
208 complete: Bool
209) -> RecoveryState:
210 match complete:
211 case False{}:
212 Tail{file, path, offset, table}
213 case True{}:
214 recover.prefix.cloned(file, Wal.bytes.clone.split(bytes), size, offset, table)
215
216def recover.count.completed(
217 +path: String,
218 file: File,
219 +size: Nat,
220 +offset: Nat,
221 table: ByteMemTable.Table,
222 prefix: Bytes.Bytes,
223 bytes: Bytes.Bytes,
224 complete: Bool,
225 +frame_length: U32
226) -> RecoveryState:
227 match complete:
228 case False{}:
229 Tail{file, path, offset, table}
230 case True{}:
231 recover.count.cloned(file, Wal.bytes.clone.split(bytes), size, offset, table, prefix, frame_length)
232
233def tail.synced(
234 result: Result<&1, &1, U32 & String, Unit>,
235 table: ByteMemTable.Table
236) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
237 match result:
238 case Fail{failure}:
239 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{failure})
240 case Done{Unit{}}:
241 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Done{table})
242
243def tail.truncated(
244 result: Result<&1, &1, U32 & String, Unit>,
245 +path: String,
246 table: ByteMemTable.Table
247) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
248 match result:
249 case Fail{failure}:
250 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{failure})
251 case Done{Unit{}}:
252 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
253 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
254 tail.synced(synced, table)
255
256def tail.closed(
257 +path: String,
258 +offset: Nat,
259 table: ByteMemTable.Table
260) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
261 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
262 truncated : Result<&1, &1, U32 & String, Unit> <- Fs.truncate(path, offset)
263 tail.truncated(truncated, path, table)
264
265def recover.result(
266 +path: String,
267 fuel: Nat,
268 state: RecoveryState
269) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
270 match fuel:
271 case 0n:
272 recover.exhausted(state)
273 case 1n+rest:
274 match state:
275 case Tail{file, path, +offset, table}:
276 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
277 _closed : Unit <- File.close(file)
278 tail.closed(path, offset, table)
279 case At{file, +size, +offset, table, at_end, can_read_length}:
280 match at_end:
281 case True{}:
282 close.done(file, table)
283 case False{}:
284 match can_read_length:
285 case False{}:
286 recover.result(path, rest, Tail{file, path, offset, table})
287 case True{}:
288 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
289 pair : File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult> <-
290 SstStreamIo.exact.start(file, 4n)
291 recover.result(path, rest, Length{pair, size, offset, table})
292 case Length{pair, +size, +offset, table}:
293 match pair:
294 case (file, Fail{failure}):
295 close.failed(file, failure)
296 case (file, Done{SstStreamIo.ExactResult{bytes, complete}}):
297 recover.result(path, rest, recover.length.completed(path, file, size, offset, table, bytes, complete))
298 case Prefix{file, +size, +offset, table, prefix, +frame_length, within_file}:
299 match within_file:
300 case False{}:
301 recover.result(path, rest, Tail{file, path, offset, table})
302 case True{}:
303 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
304 pair : File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult> <-
305 SstStreamIo.exact.start(file, 4n)
306 recover.result(path, rest, Count{pair, size, offset, table,
307 prefix, frame_length})
308 case Count{pair, +size, +offset, table, prefix, +frame_length}:
309 match pair:
310 case (file, Fail{failure}):
311 close.failed(file, failure)
312 case (file, Done{SstStreamIo.ExactResult{bytes, complete}}):
313 recover.result(path, rest, recover.count.completed(path, file, size, offset, table, prefix, bytes, complete, frame_length))
314 case Counted{file, +size, +offset, table, prefix, count_bytes,
315 +frame_length, within_limit}:
316 match within_limit:
317 case False{}:
318 close.failed(file, error("byte WAL frame exceeds configured limit"))
319 case True{}:
320 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
321 pair : File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult> <-
322 SstStreamIo.exact.start(file, U32.to_nat((frame_length - 4 : U32)))
323 recover.result(path, rest, Body{pair, size, offset, table,
324 prefix, count_bytes, frame_length})
325 case Body{pair, +size, +offset, table, prefix, count_bytes, +frame_length}:
326 match pair:
327 case (file, Fail{failure}):
328 close.failed(file, failure)
329 case (file, Done{SstStreamIo.ExactResult{bytes, complete}}):
330 match complete:
331 case False{}:
332 recover.result(path, rest, Tail{file, path, offset, table})
333 case True{}:
334 recover.result(path, rest, Decoded{file, size,
335 Nat.add(offset, Nat.add(4n, U32.to_nat(frame_length))), table,
336 ByteWal.decode(Bytes.concat([prefix, count_bytes, bytes]))})
337 case Decoded{file, +size, +next_offset, table, result}:
338 match result:
339 case Fail{_}:
340 close.failed(file, error("corrupt byte WAL frame"))
341 case Done{batch}:
342 recover.result(path, rest, recover.at(file, size, next_offset,
343 apply.batch(table, batch)))
344 case Failed{file, failure}:
345 close.failed(file, failure)
346
347def header.checked(
348 +path: String,
349 file: File,
350 +size: Nat,
351 result: Result<&1, &1, Wal.Error, Unit>,
352 initial: ByteMemTable.Table
353) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
354 match result:
355 case Fail{_}:
356 close.failed(file, error("unsupported byte WAL format"))
357 case Done{Unit{}}:
358 recover.result(path, Nat.add(Nat.mul(size, 8n), 8n),
359 recover.at(file, size, 8n, initial))
360
361def header.valid(
362 +path: String,
363 file: File,
364 +size: Nat,
365 result: Result<&1, &1, U32 & String, Bytes.Bytes>,
366 initial: ByteMemTable.Table
367) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
368 match result:
369 case Fail{failure}:
370 close.failed(file, failure)
371 case Done{bytes}:
372 header.checked(path, file, size, ByteWal.decode_header(bytes), initial)
373
374def header.pair(
375 +path: String,
376 +size: Nat,
377 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
378 initial: ByteMemTable.Table
379) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
380 match pair:
381 case (file, result):
382 header.valid(path, file, size, result, initial)
383
384def recover.opened(
385 +path: String,
386 +size: Nat,
387 opened: Result<&1, &1, U32 & String, File>,
388 initial: ByteMemTable.Table
389) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
390 match opened:
391 case Fail{failure}:
392 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{failure})
393 case Done{file}:
394 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
395 pair : File & Result<&1, &1, U32 & String, SstStreamIo.ExactResult> <-
396 SstStreamIo.exact.start(file, 8n)
397 header.pair(path, size, SstStreamIo.read.exact.state(pair), initial)
398
399def recover.size.checked(
400 +path: String,
401 +size: Nat,
402 valid: Bool,
403 initial: ByteMemTable.Table
404) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
405 match valid:
406 case False{}:
407 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>,
408 Fail{error("malformed or oversized byte WAL")})
409 case True{}:
410 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
411 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
412 recover.opened(path, size, opened, initial)
413
414def recover.sized(
415 +path: String,
416 result: Result<&1, &1, U32 & String, Nat>,
417 initial: ByteMemTable.Table
418) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
419 match result:
420 case Fail{failure}:
421 IO.pure(Result<&1, &1, U32 & String, ByteMemTable.Table>, Fail{failure})
422 case Done{+size}:
423 recover.size.checked(path, size,
424 Nat.is_le(8n, size) && Nat.is_le(size, 4294967295n), initial)
425
426
427def recover.from(+path: String, initial: ByteMemTable.Table) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
428 do IO<Result<&1, &1, U32 & String, ByteMemTable.Table>>:
429 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path)
430 recover.sized(path, size, initial)
431
432# Replays the byte WAL over an empty initial mutation table.
433def recover(+path: String) -> IO(Result<&1, &1, U32 & String, ByteMemTable.Table>):
434 recover.from(path, ByteMemTable.empty())
435
436# Publishes a fresh WAL header without exposing a partially truncated active file.
437def create.closed(
438 result: Result<&1, &1, U32 & String, Unit>,
439 +path: String
440) -> IO(Result<&1, &1, U32 & String, Unit>):
441 match result:
442 case Fail{failure}:
443 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{failure})
444 case Done{Unit{}}:
445 Fs.fsync(path)
446
447def create.written(
448 pair: File & Result<&1, &1, U32 & String, Unit>,
449 +path: String
450) -> IO(Result<&1, &1, U32 & String, Unit>):
451 match pair:
452 case (file, result):
453 do IO<Result<&1, &1, U32 & String, Unit>>:
454 _closed : Unit <- File.close(file)
455 create.closed(result, path)
456
457def create.opened(
458 opened: Result<&1, &1, U32 & String, File>,
459 +path: String
460) -> IO(Result<&1, &1, U32 & String, Unit>):
461 match opened:
462 case Fail{failure}:
463 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{failure})
464 case Done{file}:
465 do IO<Result<&1, &1, U32 & String, Unit>>:
466 pair : File & Result<&1, &1, U32 & String, Unit> <-
467 Fs.write_bytes(file, ByteWal.header())
468 create.written(pair, path)
469
470# Creates and durably publishes an empty generation-four byte WAL.
471def create(+path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
472 do IO<Result<&1, &1, U32 & String, Unit>>:
473 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "w")
474 create.opened(opened, path)
475
476def reset.directory.synced(
477 +directory: String,
478 result: Result<&1, &1, U32 & String, Unit>
479) -> IO(Result<&1, &1, U32 & String, Unit>):
480 match result:
481 case Fail{failure}:
482 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{failure})
483 case Done{Unit{}}:
484 Fs.fsync(directory)
485
486# Creates and syncs a replacement WAL before atomic publication.
487
488def reset.published(
489 +directory: String,
490 result: Result<&1, &1, U32 & String, Unit>
491) -> IO(Result<&1, &1, U32 & String, Unit>):
492 match result:
493 case Fail{failure}:
494 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{failure})
495 case Done{Unit{}}:
496 do IO<Result<&1, &1, U32 & String, Unit>>:
497 renamed : Result<&1, &1, U32 & String, Unit> <-
498 Fs.rename(directory ++ "/WAL.tmp", directory ++ "/WAL")
499 reset.directory.synced(directory, renamed)
500
501# Syncs the directory entry after atomically replacing the active WAL.
502
503# Replaces the active WAL only after its new header is synced.
504def reset_atomic(+directory: String) -> IO(Result<&1, &1, U32 & String, Unit>):
505 do IO<Result<&1, &1, U32 & String, Unit>>:
506 created : Result<&1, &1, U32 & String, Unit> <- create(directory ++ "/WAL.tmp")
507 reset.published(directory, created)
508
509
510def append.synced(
511 result: Result<&1, &1, U32 & String, Unit>,
512 file: File,
513 +path: String
514) -> IO(Result<&1, &1, U32 & String, Unit>):
515 match result:
516 case Fail{failure}:
517 do IO<Result<&1, &1, U32 & String, Unit>>:
518 _closed : Unit <- File.close(file)
519 return Fail{failure}
520 case Done{Unit{}}:
521 do IO<Result<&1, &1, U32 & String, Unit>>:
522 _closed : Unit <- File.close(file)
523 Fs.fsync(path)
524
525def append.written(
526 pair: File & Result<&1, &1, U32 & String, Unit>,
527 +path: String
528) -> IO(Result<&1, &1, U32 & String, Unit>):
529 match pair:
530 case (file, result):
531 append.synced(result, file, path)
532
533# Retains the requested portion of a duplicated WAL frame.
534def append.slice(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
535 match pair:
536 case (_, bytes): bytes
537
538# Stops between WAL chunks or writes the remaining frame before sync.
539def append.first.written(
540 pair: File & Result<&1, &1, U32 & String, Unit>,
541 remaining: Bytes.Bytes,
542 +path: String
543) -> IO(Result<&1, &1, U32 & String, Unit>):
544 match pair:
545 case (file, Fail{failure}):
546 do IO<Result<&1, &1, U32 & String, Unit>>:
547 _closed : Unit <- File.close(file)
548 return Fail{failure}
549 case (file, Done{Unit{}}):
550 do IO<Result<&1, &1, U32 & String, Unit>>:
551 _checkpoint : Result<&1, &1, U32 & String, Unit> <-
552 CrashPoint.hit("wal.during_append")
553 written : File & Result<&1, &1, U32 & String, Unit> <-
554 Fs.write_bytes(file, remaining)
555 append.written(written, path)
556
557# Writes the first bounded WAL chunk and continues through the split point.
558def append.chunked.rest(
559 file: File,
560 first: Bytes.Bytes,
561 remaining: Bytes.Bytes,
562 +path: String
563) -> IO(Result<&1, &1, U32 & String, Unit>):
564 do IO<Result<&1, &1, U32 & String, Unit>>:
565 written : File & Result<&1, &1, U32 & String, Unit> <-
566 Fs.write_bytes(file, first)
567 append.first.written(written, remaining, path)
568
569# Splits duplicated frame buffers around one bounded WAL write chunk.
570def append.chunked.cloned(
571 file: File,
572 sources: Bytes.Bytes & Bytes.Bytes,
573 total: U32,
574 +path: String
575) -> IO(Result<&1, &1, U32 & String, Unit>):
576 match sources:
577 case (first_source, rest_source):
578 append.chunked.rest(file,
579 append.slice(Bytes.slice(first_source, 0, 65536)),
580 append.slice(Bytes.slice(rest_source, 65536,
581 (total - 65536 : U32))), path)
582
583# Clones a large WAL frame before forming its bounded append chunks.
584def append.chunked.slices(
585 file: File,
586 frame: Bytes.Bytes,
587 total: U32,
588 +path: String
589) -> IO(Result<&1, &1, U32 & String, Unit>):
590 do IO<Result<&1, &1, U32 & String, Unit>>:
591 sources : Bytes.Bytes & Bytes.Bytes <-
592 IO.pure(Bytes.Bytes & Bytes.Bytes, ByteMemTable.bytes.clone(frame))
593 append.chunked.cloned(file, sources, total, path)
594
595# Selects full-frame or chunked writing after opening the WAL.
596def append.opened.checked(
597 file: File,
598 frame: Bytes.Bytes,
599 total: U32,
600 +path: String,
601 large: Bool
602) -> IO(Result<&1, &1, U32 & String, Unit>):
603 match large:
604 case True{}:
605 append.chunked.slices(file, frame, total, path)
606 case False{}:
607 do IO<Result<&1, &1, U32 & String, Unit>>:
608 written : File & Result<&1, &1, U32 & String, Unit> <-
609 Fs.write_bytes(file, frame)
610 append.written(written, path)
611
612# Routes an opened WAL frame according to its encoded byte length.
613def append.opened.frame(
614 file: File,
615 frame: Bytes.Bytes,
616 +path: String
617) -> IO(Result<&1, &1, U32 & String, Unit>):
618 match frame:
619 case Bytes.Bytes{+total, _}:
620 append.opened.checked(file, frame, total, path,
621 U32.is_le(65537, total))
622
623def append.opened(
624 opened: Result<&1, &1, U32 & String, File>,
625 frame: Bytes.Bytes,
626 +path: String
627) -> IO(Result<&1, &1, U32 & String, Unit>):
628 match opened:
629 case Fail{failure}:
630 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{failure})
631 case Done{file}:
632 do IO<Result<&1, &1, U32 & String, Unit>>:
633 _checkpoint : Result<&1, &1, U32 & String, Unit> <-
634 CrashPoint.hit("wal.before_append")
635 append.opened.frame(file, frame, path)
636
637def append.encoded(
638 result: Result<&1, &1, Wal.Error, Bytes.Bytes>,
639 +path: String
640) -> IO(Result<&1, &1, U32 & String, Unit>):
641 match result:
642 case Fail{_}:
643 IO.pure(Result<&1, &1, U32 & String, Unit>,
644 Fail{(3, "byte WAL batch cannot be encoded")})
645 case Done{frame}:
646 do IO<Result<&1, &1, U32 & String, Unit>>:
647 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "a")
648 append.opened(opened, frame, path)
649
650# Appends one encoded batch and syncs the resulting WAL file.
651def append(+path: String, batch: ByteWal.Batch) -> IO(Result<&1, &1, U32 & String, Unit>):
652 append.encoded(ByteWal.encode(batch), path)