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 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.
12	def 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	
24	def 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	
37	def 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	
47	def 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	
53	def 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	
63	def 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	
73	def 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.
86	def 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.
92	def 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.
107	def 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.
111	def 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.
121	def 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.
125	def 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.
131	def 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{}}})