1	import Base
2	import ./Db.bend as Db
3	import ./Wal.bend as Wal
4	import ./Fs.bend as Fs
5	import ./CrashPoint.bend as CrashPoint
6	import ./DurableCommit.bend as Commit
7	import 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.
13	def 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	
25	def 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	
35	def 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	
47	def 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	
57	def 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	
69	def 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	
74	def 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	
84	def 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	
94	def 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	
106	def 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.
117	def 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.
123	def 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.
138	def 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.
142	def 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.
155	def 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.
168	def 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.
179	def 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.
185	def 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.
199	def 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.
205	def 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.
213	def 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.
223	def 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.
227	def 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.
233	def 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{}}})