src/Fs.bend
mylsm-lsm-store@0.6.0.0 · 523 lines · 0 definitions
1import Base
2import bend-kit-bytes@0.3.2.0/bytes.bend as StorageBytes
3import mylsm-lsm-store@0.6.0.0/src/FilesCompat.bend as PackedFiles
4import mylsm-lsm-store@0.6.0.0/src/FsPolicy.bend as Policy
5import mylsm-lsm-store@0.6.0.0/src/coordinator/QueuePolicy.bend as QueuePolicy
6import mylsm-lsm-store@0.6.0.0/src/ReadAtPolicy.bend as ReadAtPolicy
7
8# Audited host effects (spec §3.3). Thin syscall wrappers with .c + .js
9# twins and matching intended semantics. No laws — validated empirically.
10
11# Read at most 1 MiB as packed words. Short reads are valid progress; zero is EOF.
12def read_words(
13 file: File,
14 max: U32
15) ->
16 IO(File & Result<&1, &1, U32 & String, U32 & Array<U32>>):
17 PackedFiles.read.words(file, max)
18
19# Tracks each bounded write syscall and its terminal outcome.
20type WriteWordsState is Type:
21 WriteWordsNeed{file: File, offsets: U32 & U32,
22 remaining: U32 & U32, words: Array<U32>, empty: Bool}
23 WriteWordsFailed{file: File, error: U32 & String}
24 WriteWordsNoProgress{file: File}
25
26def u32.clone.pair.result(copies: Nat & Nat) -> U32 & U32:
27 match copies:
28 case (first, second): (U32.from_nat(first), U32.from_nat(second))
29
30def u32.clone.pair(value: U32) -> U32 & U32:
31 u32.clone.pair.result(QueuePolicy.nat.clone.pair(U32.to_nat(value)))
32
33def u32.clone.triple.finish(
34 first: U32,
35 copies: U32 & U32
36) -> U32 & (U32 & U32):
37 match copies:
38 case (second, third): (first, (second, third))
39
40def u32.clone.triple.first(pair: U32 & U32) -> U32 & (U32 & U32):
41 match pair:
42 case (first, rest):
43 u32.clone.triple.finish(first, u32.clone.pair(rest))
44
45def u32.clone.triple(value: U32) -> U32 & (U32 & U32):
46 u32.clone.triple.first(u32.clone.pair(value))
47
48def write_words.state.remaining.copied(
49 file: File,
50 for_write_offset: U32,
51 for_read_offset: U32,
52 remaining: U32 & (U32 & U32),
53 words: Array<U32>
54) -> WriteWordsState:
55 match remaining:
56 case (for_write, (for_read, for_empty)):
57 WriteWordsNeed{file, (for_write_offset, for_read_offset),
58 (for_write, for_read), words, U32.is_zero(for_empty)}
59
60
61def write_words.state.remaining(
62 file: File,
63 offsets: U32 & U32,
64 remaining: U32 & (U32 & U32),
65 words: Array<U32>
66) -> WriteWordsState:
67 match offsets:
68 case (for_write_offset, for_read_offset):
69 write_words.state.remaining.copied(file, for_write_offset,
70 for_read_offset, remaining, words)
71
72def write_words.state(
73 file: File,
74 offset: U32,
75 remaining: U32,
76 words: Array<U32>
77) -> WriteWordsState:
78 write_words.state.remaining(file, u32.clone.pair(offset),
79 u32.clone.triple(remaining), words)
80
81def write_words.after.choose(
82 file: File,
83 offset: U32,
84 remaining: U32,
85 words: Array<U32>,
86 no_progress: Bool
87) -> WriteWordsState:
88 match no_progress:
89 case True{}: WriteWordsNoProgress{file}
90 case False{}: write_words.state(file, offset, remaining, words)
91
92def write_words.after.checked(
93 file: File,
94 offset: U32,
95 remaining: U32,
96 copies: U32 & (U32 & U32),
97 words: Array<U32>
98) -> WriteWordsState:
99 match copies:
100 case (for_offset, (for_subtract, for_zero)):
101 write_words.after.choose(
102 file, U32.add(offset, for_offset), U32.sub(remaining, for_subtract),
103 words, U32.is_zero(for_zero))
104
105def write_words.after(
106 pair: File & Result<&1, &1, U32 & String, U32 & Array<U32>>,
107 offset: U32,
108 remaining: U32
109) -> WriteWordsState:
110 match pair:
111 case (file, Fail{error}): WriteWordsFailed{file, error}
112 case (file, Done{(written, words)}):
113 write_words.after.checked(file, offset, remaining,
114 u32.clone.triple(written), words)
115
116def write_words.loop(
117 fuel: Nat,
118 state: WriteWordsState
119) ->
120 IO(File & Result<&1, &1, U32 & String, Unit>):
121 match fuel:
122 case 0n:
123 match state:
124 case WriteWordsNeed{file, _, _, _, empty}:
125 match empty:
126 case True{}:
127 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
128 (file, Done{Unit{}}))
129 case False{}:
130 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
131 (file, Fail{(7, "write ran out of bounded progress steps")}))
132 case WriteWordsFailed{file, error}:
133 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
134 (file, Fail{error}))
135 case WriteWordsNoProgress{file}:
136 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
137 (file, Fail{(5, "write made no progress")}))
138 case 1n+rest:
139 match state:
140 case WriteWordsNeed{file, (for_write_offset, for_read_offset),
141 (for_write, for_read), words, empty}:
142 match empty:
143 case True{}:
144 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
145 (file, Done{Unit{}}))
146 case False{}:
147 do IO<File & Result<&1, &1, U32 & String, Unit>>:
148 pair : File & Result<&1, &1, U32 & String, U32 & Array<U32>> <-
149 PackedFiles.write.words.once(file, for_write_offset,
150 U32.min(65536, for_write), words)
151 write_words.loop(rest,
152 write_words.after(pair, for_read_offset, for_read))
153 case WriteWordsFailed{file, error}:
154 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
155 (file, Fail{error}))
156 case WriteWordsNoProgress{file}:
157 IO.pure(File & Result<&1, &1, U32 & String, Unit>,
158 (file, Fail{(5, "write made no progress")}))
159
160# Writes one packed buffer through bounded single-syscall steps.
161def write_words(
162 file: File,
163 len: U32,
164 words: Array<U32>
165) ->
166 IO(File & Result<&1, &1, U32 & String, Unit>):
167 write_words.loop(4294967295n,
168 write_words.state(file, 0, len, words))
169
170def read_bytes.wrap(
171 result: Result<&1, &1, U32 & String, U32 & Array<U32>>
172) ->
173 Result<&1, &1, U32 & String, StorageBytes.Bytes>:
174 match result:
175 case Fail{err}:
176 Fail{err}
177 case Done{(len, words)}:
178 Done{StorageBytes.Bytes{len, words}}
179
180def read_bytes.pair(
181 pair: File & Result<&1, &1, U32 & String, U32 & Array<U32>>
182) ->
183 File & Result<&1, &1, U32 & String, StorageBytes.Bytes>:
184 match pair:
185 case (handle, result):
186 (handle, read_bytes.wrap(result))
187
188# Read one bounded packed chunk; nonempty short reads are not EOF.
189def read_bytes(
190 file: File,
191 max: U32
192) ->
193 IO(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>):
194 do IO<File & Result<&1, &1, U32 & String, StorageBytes.Bytes>>:
195 pair : File & Result<&1, &1, U32 & String, U32 & Array<U32>> <- read_words(file, max)
196 return read_bytes.pair(pair)
197
198def read_at.pack.finish(offset: U32, result: StorageBytes.Bytes) -> StorageBytes.Bytes:
199 match result:
200 case StorageBytes.Bytes{_, buf}:
201 StorageBytes.Bytes{offset, buf}
202
203def read_at.nat.clone(+value: Nat) -> Nat & Nat:
204 (Nat.mul(1n, value), Nat.mul(1n, value))
205
206def read_at.nat.clone.triple(+value: Nat) -> Nat & Nat & Nat:
207 (Nat.mul(1n, value), Nat.mul(1n, value), Nat.mul(1n, value))
208
209def read_at.pack.go(
210 fuel: Nat,
211 values: List<&2, U32>,
212 result: StorageBytes.Bytes,
213 +offset: U32
214) -> StorageBytes.Bytes:
215 match fuel:
216 case 0n: read_at.pack.finish(offset, result)
217 case 1n+rest:
218 match values:
219 case Nil{}: read_at.pack.finish(offset, result)
220 case Con{value, tail}:
221 read_at.pack.go(rest, tail,
222 StorageBytes.set(result, offset, value), (offset + 1 : U32))
223
224def read_at.pack.buffer.length(
225 values: List<&2, U32>,
226 lengths: Nat & Nat,
227 buf: Array<U32>
228) -> StorageBytes.Bytes:
229 match lengths:
230 case (fuel, capacity):
231 read_at.pack.go(fuel, values,
232 StorageBytes.Bytes{U32.from_nat(capacity), buf}, 0)
233
234def read_at.pack.buffer(
235 values: List<&2, U32>,
236 buffer: StorageBytes.Bytes
237) -> StorageBytes.Bytes:
238 match buffer:
239 case StorageBytes.Bytes{len, buf}:
240 read_at.pack.buffer.length(values, read_at.nat.clone(U32.to_nat(len)), buf)
241
242def read_at.pack(max: U32, values: List<&2, U32>) -> StorageBytes.Bytes:
243 read_at.pack.buffer(values, StorageBytes.new(max))
244
245# Retains the file and bounded positional-read progress between host calls.
246type ReadAtState is Type:
247 ReadAtNeed{file: File, offset: Nat, remaining: Nat,
248 chunks: List<&1, StorageBytes.Bytes>}
249 ReadAtRead{pair: File & Result<&1, &1, U32 & String, List<&2, U32>>,
250 offset: Nat, remaining: Nat,
251 chunks: List<&1, StorageBytes.Bytes>}
252 ReadAtPacked{file: File, offset: Nat, remaining: Nat,
253 chunks: List<&1, StorageBytes.Bytes>, buffer: StorageBytes.Bytes}
254
255def read_at.finish(
256 file: File,
257 chunks: List<&1, StorageBytes.Bytes>
258) -> IO(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>):
259 IO.pure(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>,
260 (file, Done{StorageBytes.concat(
261 List.reverse(&1, StorageBytes.Bytes, chunks))}))
262
263# Returns an explicit error if bounded progress unexpectedly runs out.
264def read_at.exhausted(
265 state: ReadAtState
266) -> IO(
267 File & Result<&1, &1, U32 & String, StorageBytes.Bytes>
268):
269 match state:
270 case ReadAtNeed{file, _, _, _}:
271 IO.pure(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>,
272 (file, Fail{(7, "positional read exhausted its progress budget")}))
273 case ReadAtRead{pair, _, _, _}:
274 match pair:
275 case (file, _):
276 IO.pure(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>,
277 (file, Fail{(7, "positional read exhausted its progress budget")}))
278 case ReadAtPacked{file, _, _, _, _}:
279 IO.pure(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>,
280 (file, Fail{(7, "positional read exhausted its progress budget")}))
281
282# Assembles a positional read from bounded chunks without moving the file cursor.
283def read_at.loop(
284 fuel: Nat,
285 state: ReadAtState
286) -> IO(
287 File & Result<&1, &1, U32 & String, StorageBytes.Bytes>
288):
289 match fuel:
290 case 0n: read_at.exhausted(state)
291 case 1n+rest:
292 match state:
293 case ReadAtNeed{file, +offset, +remaining, chunks}:
294 match remaining:
295 case 0n: read_at.finish(file, chunks)
296 case 1n+_:
297 do IO<File & Result<&1, &1, U32 & String, StorageBytes.Bytes>>:
298 pair : File & Result<&1, &1, U32 & String, List<&2, U32>> <-
299 File.read_at(file, U32.from_nat(offset),
300 U32.from_nat(Nat.min(8192n, remaining)))
301 read_at.loop(rest,
302 ReadAtRead{pair, offset, remaining, chunks})
303 case ReadAtRead{pair, offset, +remaining, chunks}:
304 match pair:
305 case (file, Fail{error}):
306 IO.pure(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>,
307 (file, Fail{error}))
308 case (file, Done{Nil{}}): read_at.finish(file, chunks)
309 case (file, Done{values}):
310 read_at.loop(rest, ReadAtPacked{file, offset, remaining,
311 chunks, read_at.pack(8192, values)})
312 case ReadAtPacked{file, offset, +remaining, chunks, buffer}:
313 match buffer:
314 case StorageBytes.Bytes{+len, buf}:
315 read_at.loop(rest, ReadAtNeed{file,
316 Nat.add(offset, U32.to_nat(len)),
317 Nat.sub(remaining, U32.to_nat(len)),
318 Con{StorageBytes.Bytes{len, buf}, chunks}})
319
320def read_at.range.valid(offset: Nat, max: Nat) -> Bool:
321 ReadAtPolicy.range_valid(Nat.add(4294967295n, 1n), offset, max)
322
323def read_at.start.validated(
324 file: File,
325 offset: Nat,
326 maxima: Nat & Nat,
327 valid: Bool
328) -> IO(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>):
329 match maxima:
330 case (for_fuel, for_remaining):
331 match valid:
332 case False{}:
333 IO.pure(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>,
334 (file, Fail{(22, "positional read range exceeds offset limits")}))
335 case True{}:
336 read_at.loop(Nat.add(Nat.mul(3n, for_fuel), 3n),
337 ReadAtNeed{file, offset, for_remaining, Nil{}})
338
339def read_at.start(
340 file: File,
341 offsets: Nat & Nat,
342 maxima: Nat & Nat & Nat
343) -> IO(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>):
344 match offsets:
345 case (offset, checked_offset):
346 match maxima:
347 case (for_validation, for_fuel, for_remaining):
348 read_at.start.validated(file, offset, (for_fuel, for_remaining),
349 read_at.range.valid(checked_offset, for_validation))
350
351# Reads bounded bytes at a file offset without changing its current position.
352def read_at(
353 file: File,
354 +offset: U32,
355 +max: U32
356) -> IO(File & Result<&1, &1, U32 & String, StorageBytes.Bytes>):
357 read_at.start(file, read_at.nat.clone(U32.to_nat(offset)),
358 read_at.nat.clone.triple(U32.to_nat(max)))
359
360# Write the significant bytes of a packed buffer through the pinned host effect.
361def write_bytes(
362 file: File,
363 bytes: StorageBytes.Bytes
364) ->
365 IO(File & Result<&1, &1, U32 & String, Unit>):
366 StorageBytes.Bytes{len, buf} = bytes
367 write_words(file, len, buf)
368
369# Sync file contents to stable storage.
370def fsync(path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
371 import "./effs/fsync.c"
372 import "./effs/fsync.js"
373
374# Reads one monotonic clock sample as nanoseconds.
375def clock_monotonic() -> IO(Result<&1, &1, U32 & String, Nat>):
376 import "./effs/clock_monotonic.c"
377 import "./effs/clock_monotonic.js"
378
379# Sleeps once for the requested bounded millisecond duration.
380def sleep_ms(milliseconds: U32) -> IO(Result<&1, &1, U32 & String, Unit>):
381 import "./effs/sleep.c"
382 import "./effs/sleep.js"
383
384# Handle truncate in the filesystem effects.
385def truncate(path: String, offset: Nat) -> IO(Result<&1, &1, U32 & String, Unit>):
386 import "./effs/truncate.c"
387 import "./effs/truncate.js"
388
389# Atomically rename a filesystem path.
390def rename(old: String, new: String) -> IO(Result<&1, &1, U32 & String, Unit>):
391 PackedFiles.rename(old, new)
392
393# Atomically rename a filesystem path only when the destination is absent.
394def rename_noreplace(old: String, new: String) -> IO(Result<&1, &1, U32 & String, Unit>):
395 import "./effs/rename_noreplace.c"
396 import "./effs/rename_noreplace.js"
397
398# Remove a file or empty directory.
399def remove(path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
400 PackedFiles.remove(path)
401
402# Return the number of entries available through the indexed directory API.
403def read_dir_count(path: String) -> IO(Result<&1, &1, U32 & String, Nat>):
404 import "./effs/read_dir.c"
405 import "./effs/read_dir.js"
406
407# Fetch one directory entry by index.
408def read_dir_at(path: String, idx: Nat) -> IO(Result<&1, &1, U32 & String, String>):
409 import "./effs/read_dir.c"
410 import "./effs/read_dir.js"
411
412# Create one directory.
413def make_dir(path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
414 import "./effs/make_dir.c"
415 import "./effs/make_dir.js"
416
417# Resolves an existing database path to its filesystem identity.
418def canonical_path(path: String) -> IO(Result<&1, &1, U32 & String, String>):
419 import "./effs/canonical_path.c"
420 import "./effs/canonical_path.js"
421
422# Set POSIX permission bits.
423def chmod(path: String, bits: U32) -> IO(Result<&1, &1, U32 & String, Unit>):
424 import "./effs/chmod.c"
425 import "./effs/chmod.js"
426
427# Check whether a filesystem path exists.
428def exists.result(result: Result<&1, &1, U32 & String, PackedFiles.Info>) -> Result<&1, &1, U32 & String, Bool>:
429 match result:
430 case Fail{(+code, message)}:
431 Policy.missing_path_result(U32.is_eq(code, 2), code, message)
432 case Done{_}:
433 Done{True{}}
434
435# Returns whether a path exists and preserves other stat errors.
436def exists(path: String) -> IO(Result<&1, &1, U32 & String, Bool>):
437 do IO<Result<&1, &1, U32 & String, Bool>>:
438 result : Result<&1, &1, U32 & String, PackedFiles.Info> <- PackedFiles.stat(path)
439 return exists.result(result)
440
441# Read-only byte size for Bend-side accounting (no shell wc needed).
442def file_size(path: String) -> IO(Result<&1, &1, U32 & String, Nat>):
443 import "./effs/file_size.c"
444 import "./effs/file_size.js"
445
446# Pure accumulator step (leaf: matches structurally, never recurses).
447# Past-end fetches answer "" (SNil{}), which is skipped.
448def grab_name(nm: String, acc: List<String>) -> List<String>:
449 match nm:
450 case SNil{}:
451 acc
452 case SCon{_, _}:
453 Con{nm, acc}
454
455# Fuel-bounded sweep 0..count-1. Startup-only use (spec §10 below): the
456# directory is stable during recovery, so count-then-fetch is exact.
457def read_go(fuel: Nat, +path: String, +idx: Nat, acc: List<String>) -> IO(List<String>):
458 match fuel:
459 case 0n:
460 IO.pure(List<String>, List.reverse(&1, String, acc))
461 case 1n+f:
462 do IO<List<String>>:
463 nm : String <- IO.try(String, read_dir_at(path, idx))
464 read_go(f, path, Nat.add(idx, 1n), grab_name(nm, acc))
465
466# List directory names using the bounded count-then-fetch host API.
467def read_dir(+path: String) -> IO(Result<&1, &1, U32 & String, List<String>>):
468 do IO<Result<&1, &1, U32 & String, List<String>>>:
469 n : Nat <- IO.try(Nat, read_dir_count(path))
470 names : List<String> <- read_go(Nat.add(n, 1n), path, 0n, Nil{})
471 return Done{names}
472
473def write_text.closed(
474 pair: File & Result<&1, &1, U32 & String, Unit>
475) -> IO(Result<&1, &1, U32 & String, Unit>):
476 match pair:
477 case (file, result):
478 do IO<Result<&1, &1, U32 & String, Unit>>:
479 _closed : Unit <- File.close(file)
480 return result
481
482def write_text.opened(
483 opened: Result<&1, &1, U32 & String, File>,
484 data: String
485) -> IO(Result<&1, &1, U32 & String, Unit>):
486 match opened:
487 case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
488 case Done{file}:
489 do IO<Result<&1, &1, U32 & String, Unit>>:
490 pair : File & Result<&1, &1, U32 & String, Unit> <- File.write(file, data)
491 write_text.closed(pair)
492
493# Writes a whole text file, replacing what was there.
494def write_text(path: String, data: String) -> IO(Result<&1, &1, U32 & String, Unit>):
495 do IO<Result<&1, &1, U32 & String, Unit>>:
496 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "w")
497 write_text.opened(opened, data)
498
499def read_text.closed(
500 pair: File & Result<&1, &1, U32 & String, String>
501) -> IO(Result<&1, &1, U32 & String, String>):
502 match pair:
503 case (file, result):
504 do IO<Result<&1, &1, U32 & String, String>>:
505 _closed : Unit <- File.close(file)
506 return result
507
508def read_text.opened(
509 opened: Result<&1, &1, U32 & String, File>
510) -> IO(Result<&1, &1, U32 & String, String>):
511 match opened:
512 case Fail{error}: IO.pure(Result<&1, &1, U32 & String, String>, Fail{error})
513 case Done{file}:
514 do IO<Result<&1, &1, U32 & String, String>>:
515 pair : File & Result<&1, &1, U32 & String, String> <-
516 File.read(file, U32.from_nat(4294967295n))
517 read_text.closed(pair)
518
519# Reads a whole text file in one bounded call.
520def read_text(path: String) -> IO(Result<&1, &1, U32 & String, String>):
521 do IO<Result<&1, &1, U32 & String, String>>:
522 opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
523 read_text.opened(opened)