1	import Base
2	import ./DurableError.bend as DurableError
3	import ./DbLockPolicy.bend as Policy
4	import ./Fs.bend as Fs
5	
6	# Owns the open LOCK file until release.
7	type Lock is Type:
8	  Lock{file: File}
9	
10	# Determines whether a clock sample selects backoff or deadline validation.
11	type WaitPurpose is Data:
12	  ChooseDelay{}
13	  CheckDeadline{}
14	
15	# Carries the raw result of one nonblocking lock attempt.
16	type AttemptDecision is Type:
17	  Contended{}
18	  Failed{error: U32 & String}
19	
20	# Carries file ownership through the single recursive wait state machine.
21	type WaitState is Type:
22	  TryLock{file: File, started_us: Nat, timeout_ms: Nat, backoff_ms: U32,
23	    operation: String, path: String}
24	  LockAttempt{pair: File & Result<&1, &1, U32 & String, Unit>,
25	    started_us: Nat, timeout_ms: Nat, backoff_ms: U32, operation: String, path: String}
26	  ReadClock{file: File, started_us: Nat, timeout_ms: Nat, backoff_ms: U32,
27	    operation: String, path: String, purpose: WaitPurpose}
28	  ClockSample{file: File, started_us: Nat, timeout_ms: Nat, backoff_ms: U32,
29	    operation: String, path: String, purpose: WaitPurpose,
30	    result: Result<&1, &1, U32 & String, Nat>}
31	  Pause{file: File, started_us: Nat, timeout_ms: Nat, backoff_ms: U32,
32	    operation: String, path: String, duration_ms: U32}
33	  PauseResult{file: File, started_us: Nat, timeout_ms: Nat, backoff_ms: U32,
34	    operation: String, path: String,
35	    result: Result<&1, &1, U32 & String, Unit>}
36	  WaitFatal{file: File, error: U32 & String, operation: String, path: String}
37	  WaitExpired{file: File, operation: String, path: String}
38	
39	
40	# Builds the stable LOCK path for a database directory.
41	def lock_path(path: String) -> String:
42	  path ++ "/LOCK"
43	
44	def DbLock.try_lock(file: File) -> IO(File & Result<&1, &1, U32 & String, Unit>):
45	  import "./effs/db_lock.c"
46	  import "./effs/db_lock.js"
47	
48	def DbLock.try_unlock(file: File) -> IO(File & Result<&1, &1, U32 & String, Unit>):
49	  import "./effs/db_unlock.c"
50	  import "./effs/db_unlock.js"
51	
52	# Classifies a failure to open the LOCK file.
53	def open_failed(error: U32 & String, operation: String, path: String) -> IO(Result<&1, &1, DurableError.Error, Lock>):
54	  match error:
55	    case (code, message):
56	      IO.pure(Result<&1, &1, DurableError.Error, Lock>,
57	        Fail{Policy.classify_lock_error(code, message, operation, path)})
58	
59	# Closes the file and classifies a failed lock attempt.
60	def lock_failed(
61	  file: File,
62	  error: U32 & String,
63	  operation: String,
64	  path: String
65	) -> IO(Result<&1, &1, DurableError.Error, Lock>):
66	  match error:
67	    case (code, message):
68	      do IO<Result<&1, &1, DurableError.Error, Lock>>:
69	        _closed : Unit <- File.close(file)
70	        return Fail{Policy.classify_lock_error(code, message, operation, path)}
71	
72	# Returns an owned lock on success or closes the failed handle.
73	def lock_result(
74	  pair: File & Result<&1, &1, U32 & String, Unit>,
75	  operation: String,
76	  path: String
77	) -> IO(Result<&1, &1, DurableError.Error, Lock>):
78	  match pair:
79	    case (file, Fail{error}):
80	      lock_failed(file, error, operation, path)
81	    case (file, Done{Unit{}}):
82	      IO.pure(Result<&1, &1, DurableError.Error, Lock>, Done{Lock{file}})
83	
84	# Attempts the nonblocking OS lock on the open file.
85	def lock_file(file: File, operation: String, path: String) -> IO(Result<&1, &1, DurableError.Error, Lock>):
86	  do IO<Result<&1, &1, DurableError.Error, Lock>>:
87	    pair : File & Result<&1, &1, U32 & String, Unit> <- DbLock.try_lock(file)
88	    lock_result(pair, operation, path)
89	
90	# Locks the opened file or classifies its open error.
91	def opened(
92	  result: Result<&1, &1, U32 & String, File>,
93	  operation: String,
94	  path: String
95	) -> IO(Result<&1, &1, DurableError.Error, Lock>):
96	  match result:
97	    case Fail{error}:
98	      open_failed(error, operation, path)
99	    case Done{file}:
100	      lock_file(file, operation, path)
101	
102	# Opens the stable LOCK file and acquires its OS lock.
103	def acquire(+path: String, operation: String) -> IO(Result<&1, &1, DurableError.Error, Lock>):
104	  do IO<Result<&1, &1, DurableError.Error, Lock>>:
105	    result : Result<&1, &1, U32 & String, File> <- File.open(lock_path(path), "a")
106	    opened(result, operation, path)
107	
108	# Closes the file and preserves the unlock error.
109	def release_failed(
110	  file: File,
111	  error: U32 & String,
112	  operation: String,
113	  path: String
114	) -> IO(Result<&1, &1, DurableError.Error, Unit>):
115	  match error:
116	    case (code, message):
117	      do IO<Result<&1, &1, DurableError.Error, Unit>>:
118	        _closed : Unit <- File.close(file)
119	        return Fail{DurableError.classify_open(DurableError.OpenHostFailure{code, message}, operation, path)}
120	
121	# Closes the file after a successful unlock.
122	def release_succeeded(file: File) -> IO(Result<&1, &1, DurableError.Error, Unit>):
123	  do IO<Result<&1, &1, DurableError.Error, Unit>>:
124	    _closed : Unit <- File.close(file)
125	    return Done{Unit{}}
126	
127	# Closes the file after either unlock result.
128	def released(
129	  pair: File & Result<&1, &1, U32 & String, Unit>,
130	  operation: String,
131	  path: String
132	) -> IO(Result<&1, &1, DurableError.Error, Unit>):
133	  match pair:
134	    case (file, Fail{error}):
135	      release_failed(file, error, operation, path)
136	    case (file, Done{Unit{}}):
137	      release_succeeded(file)
138	
139	# Unlocks the file and closes it after the result.
140	def unlock_file(file: File, operation: String, path: String) -> IO(Result<&1, &1, DurableError.Error, Unit>):
141	  do IO<Result<&1, &1, DurableError.Error, Unit>>:
142	    pair : File & Result<&1, &1, U32 & String, Unit> <- DbLock.try_unlock(file)
143	    released(pair, operation, path)
144	
145	# Consumes the lock and releases its file handle.
146	def release(lock: Lock, operation: String, path: String) -> IO(Result<&1, &1, DurableError.Error, Unit>):
147	  match lock:
148	    case Lock{file}:
149	      unlock_file(file, operation, path)
150	
151	# Closes the lock file and returns a distinct timeout error.
152	def wait.timeout(file: File, operation: String, path: String) -> IO(Result<&1, &1, DurableError.Error, Lock>):
153	  do IO<Result<&1, &1, DurableError.Error, Lock>>:
154	    _closed : Unit <- File.close(file)
155	    return Fail{DurableError.Error{DurableError.Timeout{}, operation, path, 0,
156	      "lock acquisition timed out"}}
157	
158	# Closes the lock file and classifies a clock or sleep host failure.
159	def wait.host_error(
160	  file: File,
161	  error: U32 & String,
162	  operation: String,
163	  path: String
164	) -> IO(Result<&1, &1, DurableError.Error, Lock>):
165	  match error:
166	    case (code, message):
167	      do IO<Result<&1, &1, DurableError.Error, Lock>>:
168	        _closed : Unit <- File.close(file)
169	        return Fail{DurableError.classify_open(
170	          DurableError.OpenHostFailure{code, message}, operation, path)}
171	
172	# Separates contention from fatal lock errors while preserving host details.
173	def wait.attempt.decision(error: U32 & String) -> AttemptDecision:
174	  match error:
175	    case (+code, message):
176	      Bool.pick(AttemptDecision, Policy.busy_code(code),
177	        Contended{}, Failed{(code, message)})
178	
179	# Converts a raw lock failure into the next owned state.
180	def wait.attempt.state(
181	  decision: AttemptDecision,
182	  file: File,
183	  started_us: Nat,
184	  timeout_ms: Nat,
185	  backoff_ms: U32,
186	  operation: String,
187	  path: String
188	) -> WaitState:
189	  match decision:
190	    case Contended{}:
191	      ReadClock{file, started_us, timeout_ms, backoff_ms,
192	        operation, path, ChooseDelay{}}
193	    case Failed{error}:
194	      WaitFatal{file, error, operation, path}
195	
196	# Converts backoff policy into either one sleep or a terminal state.
197	def wait.delay.state(
198	  decision: Policy.WaitDecision,
199	  file: File,
200	  started_us: Nat,
201	  timeout_ms: Nat,
202	  backoff_ms: U32,
203	  operation: String,
204	  path: String
205	) -> WaitState:
206	  match decision:
207	    case Policy.LockRetry{sleep_ms}:
208	      Pause{file, started_us, timeout_ms, backoff_ms, operation, path,
209	        U32.from_nat(sleep_ms)}
210	    case _:
211	      WaitExpired{file, operation, path}
212	
213	# Converts a post-sleep deadline sample into a retry or timeout state.
214	def wait.deadline.state(
215	  remaining_ms: Nat,
216	  file: File,
217	  started_us: Nat,
218	  timeout_ms: Nat,
219	  backoff_ms: U32,
220	  operation: String,
221	  path: String
222	) -> WaitState:
223	  match remaining_ms:
224	    case 0n:
225	      WaitExpired{file, operation, path}
226	    case 1n+_:
227	      TryLock{file, started_us, timeout_ms, backoff_ms, operation, path}
228	
229	# Selects backoff or deadline logic from a monotonic elapsed-time sample.
230	def wait.clock.decision(
231	  purpose: WaitPurpose,
232	  file: File,
233	  started_us: Nat,
234	  +timeout_ms: Nat,
235	  +backoff_ms: U32,
236	  operation: String,
237	  path: String,
238	  elapsed_ms: Nat
239	) -> WaitState:
240	  match purpose:
241	    case ChooseDelay{}:
242	      wait.delay.state(Policy.busy_wait_decision(elapsed_ms, timeout_ms,
243	        U32.to_nat(backoff_ms)), file, started_us, timeout_ms, backoff_ms,
244	        operation, path)
245	    case CheckDeadline{}:
246	      wait.deadline.state(Nat.sub(timeout_ms, elapsed_ms), file, started_us,
247	        timeout_ms, backoff_ms, operation, path)
248	
249	# Closes a handle if the finite retry budget is unexpectedly exhausted.
250	def wait.exhausted(state: WaitState) -> IO(Result<&1, &1, DurableError.Error, Lock>):
251	  match state:
252	    case TryLock{file, _, _, _, operation, path}:
253	      wait.timeout(file, operation, path)
254	    case LockAttempt{pair, _, _, _, operation, path}:
255	      match pair:
256	        case (file, _):
257	          wait.timeout(file, operation, path)
258	    case ReadClock{file, _, _, _, operation, path, _}:
259	      wait.timeout(file, operation, path)
260	    case ClockSample{file, _, _, _, operation, path, _, _}:
261	      wait.timeout(file, operation, path)
262	    case Pause{file, _, _, _, operation, path, _}:
263	      wait.timeout(file, operation, path)
264	    case PauseResult{file, _, _, _, operation, path, _}:
265	      wait.timeout(file, operation, path)
266	    case WaitFatal{file, _, operation, path}:
267	      wait.timeout(file, operation, path)
268	    case WaitExpired{file, operation, path}:
269	      wait.timeout(file, operation, path)
270	
271	# Runs lock, clock, and sleep transitions with one decreasing structural fuel.
272	def wait.loop(fuel: Nat, state: WaitState) -> IO(Result<&1, &1, DurableError.Error, Lock>):
273	  match fuel:
274	    case 0n:
275	      wait.exhausted(state)
276	    case 1n+rest:
277	      match state:
278	        case TryLock{file, started_us, timeout_ms, backoff_ms, operation, path}:
279	          do IO<Result<&1, &1, DurableError.Error, Lock>>:
280	            pair : File & Result<&1, &1, U32 & String, Unit> <- DbLock.try_lock(file)
281	            wait.loop(rest, LockAttempt{pair, started_us, timeout_ms,
282	              backoff_ms, operation, path})
283	        case LockAttempt{pair, started_us, timeout_ms, backoff_ms, operation, path}:
284	          match pair:
285	            case (file, Done{Unit{}}):
286	              IO.pure(Result<&1, &1, DurableError.Error, Lock>, Done{Lock{file}})
287	            case (file, Fail{error}):
288	              wait.loop(rest, wait.attempt.state(wait.attempt.decision(error),
289	                file, started_us, timeout_ms, backoff_ms, operation, path))
290	        case ReadClock{file, started_us, timeout_ms, backoff_ms, operation, path, purpose}:
291	          do IO<Result<&1, &1, DurableError.Error, Lock>>:
292	            result : Result<&1, &1, U32 & String, Nat> <- Fs.clock_monotonic()
293	            wait.loop(rest, ClockSample{file, started_us, timeout_ms,
294	              backoff_ms, operation, path, purpose, result})
295	        case ClockSample{file, +started_us, +timeout_ms, +backoff_ms,
296	            operation, path, purpose, result}:
297	          match result:
298	            case Fail{error}:
299	              wait.host_error(file, error, operation, path)
300	            case Done{now_us}:
301	              +elapsed_us = Nat.sub(now_us, started_us)
302	              +elapsed_ms = Nat.div(elapsed_us, 1000n)
303	              wait.loop(rest, wait.clock.decision(purpose, file, started_us,
304	                timeout_ms, backoff_ms, operation, path, elapsed_ms))
305	        case Pause{file, started_us, timeout_ms, backoff_ms, operation, path, duration_ms}:
306	          do IO<Result<&1, &1, DurableError.Error, Lock>>:
307	            result : Result<&1, &1, U32 & String, Unit> <- Fs.sleep_ms(duration_ms)
308	            wait.loop(rest, PauseResult{file, started_us, timeout_ms,
309	              backoff_ms, operation, path, result})
310	        case PauseResult{file, started_us, timeout_ms, backoff_ms, operation, path, result}:
311	          match result:
312	            case Fail{error}:
313	              wait.host_error(file, error, operation, path)
314	            case Done{Unit{}}:
315	              wait.loop(rest, ReadClock{file, started_us, timeout_ms,
316	                backoff_ms, operation, path, CheckDeadline{}})
317	        case WaitFatal{file, error, operation, path}:
318	          lock_failed(file, error, operation, path)
319	        case WaitExpired{file, operation, path}:
320	          wait.timeout(file, operation, path)
321	
322	# Begins retries after validating the monotonic clock source.
323	def acquire_wait.started(
324	  file: File,
325	  operation: String,
326	  path: String,
327	  timeout_ms: U32,
328	  backoff_ms: U32,
329	  result: Result<&1, &1, U32 & String, Nat>
330	) -> IO(Result<&1, &1, DurableError.Error, Lock>):
331	  match result:
332	    case Fail{error}:
333	      wait.host_error(file, error, operation, path)
334	    case Done{started_us}:
335	      +timeout = U32.to_nat(timeout_ms)
336	      fuel = Nat.add(Nat.mul(timeout, 8n), 8n)
337	      wait.loop(fuel, TryLock{file, started_us, timeout, backoff_ms,
338	        operation, path})
339	
340	# Opens the stable lock file and samples the monotonic start time.
341	def acquire_wait.opened(
342	  result: Result<&1, &1, U32 & String, File>,
343	  operation: String,
344	  path: String,
345	  timeout_ms: U32,
346	  backoff_ms: U32
347	) -> IO(Result<&1, &1, DurableError.Error, Lock>):
348	  match result:
349	    case Fail{error}:
350	      open_failed(error, operation, path)
351	    case Done{file}:
352	      do IO<Result<&1, &1, DurableError.Error, Lock>>:
353	        started : Result<&1, &1, U32 & String, Nat> <- Fs.clock_monotonic()
354	        acquire_wait.started(file, operation, path, timeout_ms, backoff_ms, started)
355	
356	# Opens and waits only after the pure option validation succeeds.
357	def acquire_wait.valid(
358	  valid: Bool,
359	  +path: String,
360	  operation: String,
361	  +timeout_ms: U32,
362	  +backoff_ms: U32
363	) -> IO(Result<&1, &1, DurableError.Error, Lock>):
364	  match valid:
365	    case False{}:
366	      IO.pure(Result<&1, &1, DurableError.Error, Lock>,
367	        Fail{DurableError.classify_open(DurableError.InvalidInput{}, operation, path)})
368	    case True{}:
369	      do IO<Result<&1, &1, DurableError.Error, Lock>>:
370	        opened : Result<&1, &1, U32 & String, File> <- File.open(lock_path(path), "a")
371	        acquire_wait.opened(opened, operation, path, timeout_ms, backoff_ms)
372	
373	# Acquires the lock immediately or waits with a finite monotonic timeout.
374	def acquire_wait(
375	  +path: String,
376	  operation: String,
377	  +timeout_ms: U32,
378	  +backoff_ms: U32
379	) -> IO(Result<&1, &1, DurableError.Error, Lock>):
380	  acquire_wait.valid(Policy.wait_options_valid(timeout_ms, backoff_ms),
381	    path, operation, timeout_ms, backoff_ms)