src/DbIo.bend
mylsm-lsm-store@0.5.0.0 · 234 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.5.0.0/src/Db.bend as Db
3import mylsm-lsm-store@0.5.0.0/src/Wal.bend as Wal
4import mylsm-lsm-store@0.5.0.0/src/Fs.bend as Fs
5import mylsm-lsm-store@0.5.0.0/src/CrashPoint.bend as CrashPoint
6import mylsm-lsm-store@0.5.0.0/src/DurableCommit.bend as Commit
7import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
8
9# Host boundary for durable Db operations. Pure transitions and their laws live
10# in Db; filesystem ordering remains an explicitly unproved host interaction.
11
12# Handle wal tail in the database filesystem effects.
13def wal_tail(fr: (File & Result<&1, &1, U32 & String, Unit>), +dir: String) -> IO(Result<&1, &1, U32 & String, Unit>):
14 match fr:
15 case (f2, r):
16 do IO<Result<&1, &1, U32 & String, Unit>>:
17 res : Unit <- IO.try(Unit, IO.pure(Result<&1, &1, U32 & String, Unit>, r))
18 _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("wal.appended"))
19 _cls : Unit <- File.close(f2)
20 _syn : Unit <- IO.try(Unit, Fs.fsync(Db.wal_path(dir)))
21 _cp2 : Unit <- IO.try(Unit, CrashPoint.hit("wal.synced"))
22 _prm : Unit <- IO.try(Unit, Fs.chmod(Db.wal_path(dir), U32.from_nat(384n)))
23 return Done{res}
24
25def wal.initialize.fsync(
26 result: Result<&1, &1, U32 & String, Unit>,
27 +path: String
28) -> IO(Result<&1, &1, U32 & String, Unit>):
29 match result:
30 case Fail{error}:
31 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
32 case Done{Unit{}}:
33 Fs.chmod(path, U32.from_nat(384n))
34
35def wal.initialize.synced(
36 result: Result<&1, &1, U32 & String, Unit>,
37 +path: String
38) -> IO(Result<&1, &1, U32 & String, Unit>):
39 match result:
40 case Fail{error}:
41 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
42 case Done{Unit{}}:
43 do IO<Result<&1, &1, U32 & String, Unit>>:
44 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
45 wal.initialize.fsync(synced, path)
46
47def wal.initialize.written(
48 pair: File & Result<&1, &1, U32 & String, Unit>,
49 +path: String
50) -> IO(Result<&1, &1, U32 & String, Unit>):
51 match pair:
52 case (file, result):
53 do IO<Result<&1, &1, U32 & String, Unit>>:
54 _closed : Unit <- File.close(file)
55 wal.initialize.synced(result, path)
56
57def wal.initialize.opened(
58 opened: Result<&1, &1, U32 & String, File>,
59 +path: String
60) -> IO(Result<&1, &1, U32 & String, Unit>):
61 match opened:
62 case Fail{error}:
63 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
64 case Done{file}:
65 do IO<Result<&1, &1, U32 & String, Unit>>:
66 written : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, Wal.log_header())
67 wal.initialize.written(written, path)
68
69def wal.initialize.header(+path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
70 do IO<Result<&1, &1, U32 & String, Unit>>:
71 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "a")
72 wal.initialize.opened(opened, path)
73
74def wal.initialize.header_if_empty(
75 empty: Bool,
76 +path: String
77) -> IO(Result<&1, &1, U32 & String, Unit>):
78 match empty:
79 case True{}:
80 wal.initialize.header(path)
81 case False{}:
82 IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
83
84def wal.initialize.existing(
85 result: Result<&1, &1, U32 & String, Nat>,
86 +path: String
87) -> IO(Result<&1, &1, U32 & String, Unit>):
88 match result:
89 case Fail{error}:
90 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
91 case Done{size}:
92 wal.initialize.header_if_empty(Nat.is_eq(size, 0n), path)
93
94def wal.initialize.present(
95 present: Bool,
96 +path: String
97) -> IO(Result<&1, &1, U32 & String, Unit>):
98 match present:
99 case False{}:
100 wal.initialize.header(path)
101 case True{}:
102 do IO<Result<&1, &1, U32 & String, Unit>>:
103 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path)
104 wal.initialize.existing(size, path)
105
106def wal.initialize.exists(
107 result: Result<&1, &1, U32 & String, Bool>,
108 +path: String
109) -> IO(Result<&1, &1, U32 & String, Unit>):
110 match result:
111 case Fail{error}:
112 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
113 case Done{present}:
114 wal.initialize.present(present, path)
115
116# Creates the WAL only when it is absent.
117def wal_initialize(+dir: String) -> IO(Result<&1, &1, U32 & String, Unit>):
118 do IO<Result<&1, &1, U32 & String, Unit>>:
119 present : Result<&1, &1, U32 & String, Bool> <- Fs.exists(Db.wal_path(dir))
120 wal.initialize.exists(present, Db.wal_path(dir))
121
122# Handle wal encoded in the database filesystem effects.
123def wal_encoded(
124 +dir: String,
125 result: Result<&1, &1, Wal.Error, Bytes.Bytes>
126) -> IO(Result<&1, &1, U32 & String, Unit>):
127 match result:
128 case Fail{_}:
129 IO.pure(Result<&1, &1, U32 & String, Unit>,
130 Fail{(U32.from_nat(3n), "WAL batch cannot be encoded")})
131 case Done{frame}:
132 do IO<Result<&1, &1, U32 & String, Unit>>:
133 f : File <- IO.try(File, File.open(Db.wal_path(dir), "a"))
134 fr : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(f, frame)
135 wal_tail(fr, dir)
136
137# Handle wal append in the database filesystem effects.
138def wal_append(+dir: String, batch: Wal.Batch) -> IO(Result<&1, &1, U32 & String, Unit>):
139 wal_encoded(dir, Wal.encode_frame(batch))
140
141# Turns the fsync result into a durable or uncertain commit stage.
142def append_synced(result: Result<&1, &1, U32 & String, Unit>, file: File) -> IO(Commit.Stage):
143 match result:
144 case Fail{(code, message)}:
145 do IO<Commit.Stage>:
146 _closed : Unit <- File.close(file)
147 return Commit.AppendUnknown{code, message}
148 case Done{Unit{}}:
149 do IO<Commit.Stage>:
150 _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("wal.synced"))
151 _closed : Unit <- File.close(file)
152 return Commit.DurableAppend{}
153
154# Syncs the WAL after a successful append and closes the file.
155def append_written(pair: File & Result<&1, &1, U32 & String, Unit>, +path: String) -> IO(Commit.Stage):
156 match pair:
157 case (file, Fail{(code, message)}):
158 do IO<Commit.Stage>:
159 _closed : Unit <- File.close(file)
160 return Commit.AppendUnknown{code, message}
161 case (file, Done{Unit{}}):
162 do IO<Commit.Stage>:
163 _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("wal.appended"))
164 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
165 append_synced(synced, file)
166
167# Rejects an open failure before append or writes the encoded frame.
168def append_opened(opened: Result<&1, &1, U32 & String, File>, frame: Bytes.Bytes, +path: String) -> IO(Commit.Stage):
169 match opened:
170 case Fail{(code, message)}:
171 IO.pure(Commit.Stage, Commit.AppendRejected{code, message})
172 case Done{file}:
173 do IO<Commit.Stage>:
174 _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("wal.before_append"))
175 written : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, frame)
176 append_written(written, path)
177
178# Opens the WAL and appends one encoded frame.
179def append_frame(+path: String, frame: Bytes.Bytes) -> IO(Commit.Stage):
180 do IO<Commit.Stage>:
181 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "a")
182 append_opened(opened, frame, path)
183
184# Rejects encoding errors before opening the WAL.
185def encoded_append(
186 result: Result<&1, &1, Wal.Error, Bytes.Bytes>,
187 +dir: String,
188 muts: List<&2, Wal.Mut>,
189) -> IO(Commit.Stage & List<&2, Wal.Mut>):
190 match result:
191 case Fail{error}:
192 IO.pure(Commit.Stage & List<&2, Wal.Mut>, (Commit.EncodingRejected{error}, muts))
193 case Done{frame}:
194 do IO<Commit.Stage & List<&2, Wal.Mut>>:
195 stage : Commit.Stage <- append_frame(Db.wal_path(dir), frame)
196 return (stage, muts)
197
198# Encodes a batch and appends it to the WAL.
199def wal_commit(+dir: String, +batch: Wal.Batch) -> IO(Commit.Stage & List<&2, Wal.Mut>):
200 match batch:
201 case Wal.Batch{muts}:
202 encoded_append(Wal.encode_frame(Wal.Batch{muts}), dir, muts)
203
204# Persists a batch before applying its pure commit transition.
205def db_write_durable(+db: Db.Db, +batch: Wal.Batch) -> IO(Db.Db & Commit.Stage):
206 match db:
207 case Db.Db{dir, _, _, _, _, _, _, _, _, _}:
208 do IO<Db.Db & Commit.Stage>:
209 pair : Commit.Stage & List<&2, Wal.Mut> <- wal_commit(dir, batch)
210 return Commit.apply(pair, db)
211
212# Persists a batch and rotates the MemTable when needed.
213def db_write(+db: Db.Db, +batch: Wal.Batch) -> IO(Result<&1, &1, U32 & String, Db.Db>):
214 match db:
215 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
216 match batch:
217 case Wal.Batch{muts}:
218 do IO<Result<&1, &1, U32 & String, Db.Db>>:
219 _res : Unit <- IO.try(Unit, wal_append(dir, Wal.Batch{muts}))
220 return Db.rot_done(Db.apply_and_rotate(muts, mem, frozen, mem_count, frozen_count), dir, batch_cap, bcache, levels, flushed, manifest_token)
221
222# Writes one key/value pair through the legacy core path.
223def db_put(+db: Db.Db, +key: String, +val: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
224 db_write(db, Wal.Batch{Con{Wal.Put{key, val}, Nil{}}})
225
226# Restores staged mutation order before writing the batch.
227def db_write_staged(+db: Db.Db, +staged: List<&2, Wal.Batch>) -> IO(Result<&1, &1, U32 & String, Db.Db>):
228 match db:
229 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
230 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))})
231
232# Writes one tombstone through the legacy core path.
233def db_del(+db: Db.Db, +key: String) -> IO(Result<&1, &1, U32 & String, Db.Db>):
234 db_write(db, Wal.Batch{Con{Wal.Del{key}, Nil{}}})