src/DbIo.bend
mylsm-lsm-store@0.4.0.0 · 132 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.4.0.0/src/Db.bend as Db
3import mylsm-lsm-store@0.4.0.0/src/Wal.bend as Wal
4import mylsm-lsm-store@0.4.0.0/src/Fs.bend as Fs
5import mylsm-lsm-store@0.4.0.0/src/CrashPoint.bend as CrashPoint
6import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7
8# Host boundary for durable Db operations. Pure transitions and their laws live
9# in Db; filesystem ordering remains an explicitly unproved host interaction.
10
11# Handle wal tail in the database filesystem effects.
12def wal_tail(fr: (File & Result<&1, &1, U32 & String, Unit>), +dir: String) -> IO(Result<&1, &1, U32 & String, Unit>):
13 match fr:
14 case (f2, r):
15 do IO<Result<&1, &1, U32 & String, Unit>>:
16 res : Unit <- IO.try(Unit, IO.pure(Result<&1, &1, U32 & String, Unit>, r))
17 _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("wal.appended"))
18 _cls : Unit <- File.close(f2)
19 _syn : Unit <- IO.try(Unit, Fs.fsync(Db.wal_path(dir)))
20 _cp2 : Unit <- IO.try(Unit, CrashPoint.hit("wal.synced"))
21 _prm : Unit <- IO.try(Unit, Fs.chmod(Db.wal_path(dir), U32.from_nat(384n)))
22 return Done{res}
23
24def wal.initialize.synced(
25 result: Result<&1, &1, U32 & String, Unit>,
26 +path: String
27) -> IO(Result<&1, &1, U32 & String, Unit>):
28 match result:
29 case Fail{error}:
30 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
31 case Done{Unit{}}:
32 do IO<Result<&1, &1, U32 & String, Unit>>:
33 _synced : Unit <- IO.try(Unit, Fs.fsync(path))
34 changed : Result<&1, &1, U32 & String, Unit> <- Fs.chmod(path, U32.from_nat(384n))
35 return changed
36
37def wal.initialize.written(
38 pair: File & Result<&1, &1, U32 & String, Unit>,
39 +path: String
40) -> IO(Result<&1, &1, U32 & String, Unit>):
41 match pair:
42 case (file, result):
43 do IO<Result<&1, &1, U32 & String, Unit>>:
44 _closed : Unit <- File.close(file)
45 wal.initialize.synced(result, path)
46
47def wal.initialize.header(+path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
48 do IO<Result<&1, &1, U32 & String, Unit>>:
49 file : File <- IO.try(File, File.open(path, "a"))
50 written : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, Wal.log_header())
51 wal.initialize.written(written, path)
52
53def wal.initialize.header_if_empty(
54 empty: Bool,
55 +path: String
56) -> IO(Result<&1, &1, U32 & String, Unit>):
57 match empty:
58 case True{}:
59 wal.initialize.header(path)
60 case False{}:
61 IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
62
63def wal.initialize.existing(
64 result: Result<&1, &1, U32 & String, Nat>,
65 +path: String
66) -> IO(Result<&1, &1, U32 & String, Unit>):
67 match result:
68 case Fail{error}:
69 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
70 case Done{size}:
71 wal.initialize.header_if_empty(Nat.is_eq(size, 0n), path)
72
73def wal.initialize.present(
74 present: Bool,
75 +path: String
76) -> IO(Result<&1, &1, U32 & String, Unit>):
77 match present:
78 case False{}:
79 wal.initialize.header(path)
80 case True{}:
81 do IO<Result<&1, &1, U32 & String, Unit>>:
82 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path)
83 wal.initialize.existing(size, path)
84
85# Handle wal initialize in the database filesystem effects.
86def wal_initialize(+dir: String) -> IO(Result<&1, &1, U32 & String, Unit>):
87 do IO<Result<&1, &1, U32 & String, Unit>>:
88 present : Bool <- IO.try(Bool, Fs.exists(Db.wal_path(dir)))
89 wal.initialize.present(present, Db.wal_path(dir))
90
91# Handle wal encoded in the database filesystem effects.
92def wal_encoded(
93 +dir: String,
94 result: Result<&1, &1, Wal.Error, Bytes.Bytes>
95) -> IO(Result<&1, &1, U32 & String, Unit>):
96 match result:
97 case Fail{_}:
98 IO.pure(Result<&1, &1, U32 & String, Unit>,
99 Fail{(U32.from_nat(3n), "WAL batch cannot be encoded")})
100 case Done{frame}:
101 do IO<Result<&1, &1, U32 & String, Unit>>:
102 f : File <- IO.try(File, File.open(Db.wal_path(dir), "a"))
103 fr : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(f, frame)
104 wal_tail(fr, dir)
105
106# Handle wal append in the database filesystem effects.
107def wal_append(+dir: String, batch: Wal.Batch) -> IO(Result<&1, &1, U32 & String, Unit>):
108 wal_encoded(dir, Wal.encode_frame(batch))
109
110# Handle db write in the database filesystem effects.
111def db_write(+db: Db.Db, +batch: Wal.Batch) -> IO(Result<&1, &1, U32 & String, Db.Db>):
112 match db:
113 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
114 match batch:
115 case Wal.Batch{muts}:
116 do IO<Result<&1, &1, U32 & String, Db.Db>>:
117 _res : Unit <- IO.try(Unit, wal_append(dir, Wal.Batch{muts}))
118 return Db.rot_done(Db.apply_and_rotate(muts, mem, frozen, mem_count, frozen_count), dir, batch_cap, bcache, levels, flushed, manifest_token)
119
120# Handle db put in the database filesystem effects.
121def db_put(+db: Db.Db, +key: String, +val: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
122 db_write(db, Wal.Batch{Con{Wal.Put{key, val}, Nil{}}})
123
124# Handle db write staged in the database filesystem effects.
125def db_write_staged(+db: Db.Db, +staged: List<&2, Wal.Batch>) -> IO(Result<&1, &1, U32 & String, Db.Db>):
126 match db:
127 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
128 db_write(Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}, Wal.Batch{Db.staged_muts(List.reverse(&2, Wal.Batch, staged))})
129
130# Handle db del in the database filesystem effects.
131def db_del(+db: Db.Db, +key: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
132 db_write(db, Wal.Batch{Con{Wal.Del{key}, Nil{}}})