src/transfer/FormatInspectIo.bend
mylsm-lsm-store@0.6.0.0 · 177 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/DbLock.bend as DbLock
3import mylsm-lsm-store@0.6.0.0/src/DurableError.bend as DurableError
4import mylsm-lsm-store@0.6.0.0/src/Fs.bend as Fs
5import mylsm-lsm-store@0.6.0.0/src/StorageBytes.bend as StorageBytes
6import mylsm-lsm-store@0.6.0.0/src/transfer/FormatInspect.bend as FormatInspect
7import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
8
9# Separates lock ownership failures from read-only filesystem failures.
10type Error is Data:
11 Lock{error: DurableError.Error}
12 Host{code: U32, message: String}
13 ResourceLimit{}
14
15def identity.unknown() -> FormatInspect.Identity:
16 FormatInspect.Identity{"unknown", "unknown", 0}
17
18def inspection.resource_limit() -> FormatInspect.Inspection:
19 FormatInspect.Inspection{identity.unknown(), FormatInspect.ResourceLimit{}}
20
21def read.failed.released(
22 error: U32 & String,
23 result: Result<&1, &1, DurableError.Error, Unit>
24) -> Result<&1, &1, Error, FormatInspect.Inspection>:
25 match error:
26 case (code, message):
27 match result:
28 case Fail{lock_error}: Fail{Lock{lock_error}}
29 case Done{Unit{}}: Fail{Host{code, message}}
30
31def read.success.released(
32 bytes: Bytes.Bytes,
33 result: Result<&1, &1, DurableError.Error, Unit>
34) -> Result<&1, &1, Error, FormatInspect.Inspection>:
35 match result:
36 case Fail{error}: Fail{Lock{error}}
37 case Done{Unit{}}:
38 Done{FormatInspect.inspect_manifest_bytes(bytes)}
39
40def read.result(
41 +path: String,
42 lock: DbLock.Lock,
43 file_result: File & Result<&1, &1, U32 & String, Bytes.Bytes>
44) -> IO(Result<&1, &1, Error, FormatInspect.Inspection>):
45 match file_result:
46 case (file, Fail{error}):
47 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
48 _closed : Unit <- File.close(file)
49 released : Result<&1, &1, DurableError.Error, Unit> <-
50 DbLock.release(lock, "inspect", path)
51 return read.failed.released(error, released)
52 case (file, Done{bytes}):
53 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
54 _closed : Unit <- File.close(file)
55 released : Result<&1, &1, DurableError.Error, Unit> <-
56 DbLock.release(lock, "inspect", path)
57 return read.success.released(bytes, released)
58
59def read.opened(
60 +path: String,
61 lock: DbLock.Lock,
62 size: Nat,
63 opened: Result<&1, &1, U32 & String, File>
64) -> IO(Result<&1, &1, Error, FormatInspect.Inspection>):
65 match opened:
66 case Fail{error}:
67 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
68 released : Result<&1, &1, DurableError.Error, Unit> <-
69 DbLock.release(lock, "inspect", path)
70 return read.failed.released(error, released)
71 case Done{file}:
72 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
73 result : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
74 Fs.read_at(file, 0, U32.from_nat(size))
75 read.result(path, lock, result)
76
77def read.limit.released(
78 result: Result<&1, &1, DurableError.Error, Unit>
79) -> IO(Result<&1, &1, Error, FormatInspect.Inspection>):
80 match result:
81 case Fail{error}:
82 IO.pure(Result<&1, &1, Error, FormatInspect.Inspection>, Fail{Lock{error}})
83 case Done{Unit{}}:
84 IO.pure(Result<&1, &1, Error, FormatInspect.Inspection>,
85 Done{inspection.resource_limit()})
86
87def read.size.checked(
88 +path: String,
89 lock: DbLock.Lock,
90 +size: Nat,
91 within_limit: Bool
92) -> IO(Result<&1, &1, Error, FormatInspect.Inspection>):
93 match within_limit:
94 case True{}:
95 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
96 opened : Result<&1, &1, U32 & String, File> <-
97 File.open(path ++ "/MANIFEST", "r")
98 read.opened(path, lock, size, opened)
99 case False{}:
100 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
101 released : Result<&1, &1, DurableError.Error, Unit> <-
102 DbLock.release(lock, "inspect", path)
103 read.limit.released(released)
104
105def read.size(
106 +path: String,
107 lock: DbLock.Lock,
108 result: Result<&1, &1, U32 & String, Nat>
109) -> IO(Result<&1, &1, Error, FormatInspect.Inspection>):
110 match result:
111 case Fail{error}:
112 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
113 released : Result<&1, &1, DurableError.Error, Unit> <-
114 DbLock.release(lock, "inspect", path)
115 return read.failed.released(error, released)
116 case Done{+size}:
117 read.size.checked(path, lock, size,
118 Nat.is_le(size, U32.to_nat(StorageBytes.MAX_MANIFEST_BYTES())))
119
120def inspect.locked(
121 +path: String,
122 result: Result<&1, &1, DurableError.Error, DbLock.Lock>
123) -> IO(Result<&1, &1, Error, FormatInspect.Inspection>):
124 match result:
125 case Fail{error}:
126 IO.pure(Result<&1, &1, Error, FormatInspect.Inspection>, Fail{Lock{error}})
127 case Done{lock}:
128 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
129 size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path ++ "/MANIFEST")
130 read.size(path, lock, size)
131
132def inspect.lock.error(
133 error: U32 & String
134) -> IO(Result<&1, &1, Error, FormatInspect.Inspection>):
135 match error:
136 case (code, message):
137 IO.pure(Result<&1, &1, Error, FormatInspect.Inspection>,
138 Fail{Host{code, message}})
139
140def inspect.lock.present(
141 +path: String,
142 result: Result<&1, &1, U32 & String, Bool>
143) -> IO(Result<&1, &1, Error, FormatInspect.Inspection>):
144 match result:
145 case Fail{error}:
146 inspect.lock.error(error)
147 case Done{False{}}:
148 IO.pure(Result<&1, &1, Error, FormatInspect.Inspection>,
149 Fail{Host{2, "database LOCK file is missing"}})
150 case Done{True{}}:
151 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
152 acquired : Result<&1, &1, DurableError.Error, DbLock.Lock> <-
153 DbLock.acquire(path, "inspect")
154 inspect.locked(path, acquired)
155
156def inspect.canonical(
157 result: Result<&1, &1, U32 & String, String>
158) -> IO(Result<&1, &1, Error, FormatInspect.Inspection>):
159 match result:
160 case Fail{error}:
161 match error:
162 case (code, message):
163 IO.pure(Result<&1, &1, Error, FormatInspect.Inspection>,
164 Fail{Host{code, message}})
165 case Done{+path}:
166 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
167 exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path ++ "/LOCK")
168 inspect.lock.present(path, exists)
169
170# Reads and validates a database manifest under an existing exclusive lock.
171def inspect(
172 +path: String
173) ->
174 IO(Result<&1, &1, Error, FormatInspect.Inspection>):
175 do IO<Result<&1, &1, Error, FormatInspect.Inspection>>:
176 canonical : Result<&1, &1, U32 & String, String> <- Fs.canonical_path(path)
177 inspect.canonical(canonical)