journal.bend
bend-kit-journal@0.1.0.0 · 400 lines · 0 definitions
1# Append-only journal: checksummed frames, fsync'd appends, and crash recovery of the committed prefix. Source: https://github.com/paymog/bend-kit/tree/main/journal
2import Base
3# bend-kit-bytes@0.3.0.0, the version bend-kit-hash@0.1.0.0 takes, so Hash.crc32 accepts these Bytes.
4import bend-kit-bytes@0.3.0.0/bytes.bend as Bytes
5import bend-kit-hash@0.1.0.0/hash.bend as Hash
6
7# A frame is one record and one commit: three u32 little-endian fields, then the payload.
8# len, then U32.not(len), then the CRC-32 of the payload, then len payload octets.
9# A frame is committed when all 12 + len of its octets are in the file.
10
11# Clean: the input ends on a frame boundary.
12# Torn: the input ends inside a frame. A crash cut the last append short.
13# Corrupt: a whole header or a whole frame fails its check. Recovery does not go past it.
14type Tail is Data:
15 Clean{}
16 Torn{}
17 Corrupt{why: String}
18
19# One decoding step. Short, Bad, and End give the input back untouched, at the frame start.
20type Step<-P: Type, -S: Type> is Type:
21 End{s: S}
22 Short{s: S}
23 Bad{why: String, s: S}
24 Got{p: P, s: S}
25
26# The committed records in order, how the input ends, and the input from the first frame not taken.
27type Recovery<a, -P: Kind(a), -S: Type> is Type:
28 Recovery{records: List<a, P>, tail: Tail, rest: S}
29
30# The state after every committed record, how the input ends, and the input from the first frame not taken.
31type Replay<-A: Type, -S: Type> is Type:
32 Replay{acc: A, tail: Tail, rest: S}
33
34# The decoder is generic over the input S and the payload P. Bytes runs it; a list of octets proves it.
35# fits(s, at, n): n octets from at fit in s. u32(s, at): the field at at. cut(s, hs, len): a copy of
36# [hs, hs + len). skip(s, hs, len): s from hs + len. sum(p): the checksum of p.
37
38def next.verdict(~P: Type, ~S: Type, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ok: Bool, +len: U32, s: S, p: P) -> Step<P, S>:
39 match ok:
40 case True{}:
41 Got{p, skip(s, 12, len)}
42 case False{}:
43 Bad{"checksum mismatch", s}
44
45def next.summed(~P: Type, ~S: Type, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, +want: U32, +len: U32, s: S, r: P & U32) -> Step<P, S>:
46 (p, +got) = r
47 next.verdict(~P, ~S, ~skip, U32.is_eq(got, want), len, s, p)
48
49def next.cut(~P: Type, ~S: Type, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, +want: U32, +len: U32, r: S & P) -> Step<P, S>:
50 (s, p) = r
51 next.summed(~P, ~S, ~skip, want, len, s, sum(p))
52
53def next.want(~P: Type, ~S: Type, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, +len: U32, r: S & U32) -> Step<P, S>:
54 (s, +want) = r
55 next.cut(~P, ~S, ~skip, ~sum, want, len, cut(s, 12, len))
56
57def next.body(~P: Type, ~S: Type, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, +len: U32, r: S & Bool) -> Step<P, S>:
58 (s, whole) = r
59 match whole:
60 case True{}:
61 next.want(~P, ~S, ~cut, ~skip, ~sum, len, u32(s, 8))
62 case False{}:
63 Short{s}
64
65def next.size(~P: Type, ~S: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, over: Bool, +len: U32, s: S) -> Step<P, S>:
66 match over:
67 case True{}:
68 Bad{"length over max", s}
69 case False{}:
70 next.body(~P, ~S, ~u32, ~cut, ~skip, ~sum, len, fits(s, 12, len))
71
72def next.head(~P: Type, ~S: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, ok: Bool, +max: U32, +len: U32, s: S) -> Step<P, S>:
73 match ok:
74 case True{}:
75 next.size(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, U32.is_gt(len, max), len, s)
76 case False{}:
77 Bad{"length check mismatch", s}
78
79def next.lenx(~P: Type, ~S: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, +max: U32, +len: U32, r: S & U32) -> Step<P, S>:
80 (s, +lenx) = r
81 next.head(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, U32.is_eq(lenx, U32.not(len)), max, len, s)
82
83def next.len(~P: Type, ~S: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, +max: U32, r: S & U32) -> Step<P, S>:
84 (s, +len) = r
85 next.lenx(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, max, len, u32(s, 4))
86
87# The two length fields are checked before the payload is waited for, so a damaged length is
88# Bad and never a Torn tail that hides the frames after it.
89def next.start(~P: Type, ~S: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, +max: U32, r: S & Bool) -> Step<P, S>:
90 (s, both) = r
91 match both:
92 case True{}:
93 next.len(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, max, u32(s, 0))
94 case False{}:
95 Short{s}
96
97def next.any(~P: Type, ~S: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, +max: U32, r: S & Bool) -> Step<P, S>:
98 (s, some) = r
99 match some:
100 case True{}:
101 next.start(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, max, fits(s, 0, 8))
102 case False{}:
103 End{s}
104
105# One frame from s. max caps a payload's length.
106def next(~P: Type, ~S: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, +max: U32, s: S) -> Step<P, S>:
107 next.any(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, max, fits(s, 0, 1))
108
109def recover.con(~a: Quant, ~P: Kind(a), ~S: Type, p: P, r: Recovery<a, P, S>) -> Recovery<a, P, S>:
110 Recovery{rs, t, s} = r
111 Recovery{Con{p, rs}, t, s}
112
113# Every frame from st on. Once n frames are taken, the next one is the last: the tail is Corrupt,
114# and rest starts after it. Callers give n at least the frame count, so this never happens.
115def recover.go(~a: Quant, ~P: Kind(a), ~S: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, n: Nat, +max: U32, st: Step<P, S>) -> Recovery<a, P, S>:
116 match n st:
117 case 0n End{s}:
118 Recovery{Nil{}, Clean{}, s}
119 case 0n Short{s}:
120 Recovery{Nil{}, Torn{}, s}
121 case 0n Bad{why, s}:
122 Recovery{Nil{}, Corrupt{why}, s}
123 case 0n Got{p, s}:
124 Recovery{Con{p, Nil{}}, Corrupt{"frame budget exhausted"}, s}
125 case 1n+m End{s}:
126 Recovery{Nil{}, Clean{}, s}
127 case 1n+m Short{s}:
128 Recovery{Nil{}, Torn{}, s}
129 case 1n+m Bad{why, s}:
130 Recovery{Nil{}, Corrupt{why}, s}
131 case 1n+m Got{p, s}:
132 recover.con(~a, ~P, ~S, p, recover.go(~a, ~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, m, max, next(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, max, s)))
133
134# Every committed record of s, in order.
135def recover(~a: Quant, ~P: Kind(a), ~S: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, n: Nat, +max: U32, s: S) -> Recovery<a, P, S>:
136 recover.go(~a, ~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, n, max, next(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, max, s))
137
138# step folded over every frame from st on, without a list of the records. n as in recover.go.
139def replay.go(~P: Type, ~S: Type, ~A: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, ~step: A -> P -> A, n: Nat, +max: U32, st: Step<P, S>, acc: A) -> Replay<A, S>:
140 match n st:
141 case 0n End{s}:
142 Replay{acc, Clean{}, s}
143 case 0n Short{s}:
144 Replay{acc, Torn{}, s}
145 case 0n Bad{why, s}:
146 Replay{acc, Corrupt{why}, s}
147 case 0n Got{p, s}:
148 Replay{step(acc, p), Corrupt{"frame budget exhausted"}, s}
149 case 1n+m End{s}:
150 Replay{acc, Clean{}, s}
151 case 1n+m Short{s}:
152 Replay{acc, Torn{}, s}
153 case 1n+m Bad{why, s}:
154 Replay{acc, Corrupt{why}, s}
155 case 1n+m Got{p, s}:
156 replay.go(~P, ~S, ~A, ~fits, ~u32, ~cut, ~skip, ~sum, ~step, m, max, next(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, max, s), step(acc, p))
157
158def replay(~P: Type, ~S: Type, ~A: Type, ~fits: S -> @+at:U32 -> @+n:U32 -> S & Bool, ~u32: S -> @+at:U32 -> S & U32, ~cut: S -> @+hs:U32 -> @+len:U32 -> S & P, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ~sum: P -> P & U32, ~step: A -> P -> A, n: Nat, +max: U32, s: S, acc: A) -> Replay<A, S>:
159 replay.go(~P, ~S, ~A, ~fits, ~u32, ~cut, ~skip, ~sum, ~step, n, max, next(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, max, s), acc)
160
161# The model: a list of octets. The laws are stated over it.
162# A model field holds its value in the first of its four cells and 0 in the others. Bytes holds it
163# as four little-endian octets. Both take four cells, so offsets agree; check.bend tests the octets.
164
165def list.fits(xs: List<&2, U32>, +at: U32, +n: U32) -> List<&2, U32> & Bool:
166 +ys = xs
167 (ys, Nat.is_le(Nat.add(U32.to_nat(at), U32.to_nat(n)), List.length(&2, U32, ys)))
168
169def list.field(xs: List<&2, U32>) -> U32:
170 match xs:
171 case Con{x, t}:
172 x
173 case Nil{}:
174 0
175
176def list.u32(xs: List<&2, U32>, +at: U32) -> List<&2, U32> & U32:
177 +ys = xs
178 (ys, list.field(List.drop(&2, U32, ys, U32.to_nat(at))))
179
180def list.cut(xs: List<&2, U32>, +hs: U32, +len: U32) -> List<&2, U32> & List<&2, U32>:
181 +ys = xs
182 (ys, List.take(&2, U32, List.drop(&2, U32, ys, U32.to_nat(hs)), U32.to_nat(len)))
183
184def list.skip(xs: List<&2, U32>, +hs: U32, +len: U32) -> List<&2, U32>:
185 List.drop(&2, U32, xs, Nat.add(U32.to_nat(hs), U32.to_nat(len)))
186
187def crc.bit(n: Nat, +c: U32) -> U32:
188 match n:
189 case 0n:
190 c
191 case 1n+m:
192 crc.bit(m, Bool.pick(U32, U32.is_eq((c .&. 1 : U32), 1), ((c >> 1n) .^. 3988292384 : U32), (c >> 1n : U32)))
193
194def crc.go(xs: List<&2, U32>, +c: U32) -> U32:
195 match xs:
196 case Nil{}:
197 c
198 case Con{+x, t}:
199 crc.go(t, crc.bit(8n, (c .^. (x .&. 255 : U32) : U32)))
200
201# CRC-32 (ISO-HDLC, as zlib and Hash.crc32) of a list of octets, bit by bit.
202def list.crc(xs: List<&2, U32>) -> U32:
203 U32.not(crc.go(xs, 4294967295))
204
205def list.sum(xs: List<&2, U32>) -> List<&2, U32> & U32:
206 +ys = xs
207 (ys, list.crc(ys))
208
209def list.next(+max: U32, xs: List<&2, U32>) -> Step<List<&2, U32>, List<&2, U32>>:
210 next(~List<&2, U32>, ~List<&2, U32>, ~list.fits, ~list.u32, ~list.cut, ~list.skip, ~list.sum, max, xs)
211
212def list.recover(n: Nat, +max: U32, xs: List<&2, U32>) -> Recovery<&2, List<&2, U32>, List<&2, U32>>:
213 recover(~&2, ~List<&2, U32>, ~List<&2, U32>, ~list.fits, ~list.u32, ~list.cut, ~list.skip, ~list.sum, n, max, xs)
214
215def list.replay(~A: Type, ~step: A -> List<&2, U32> -> A, n: Nat, +max: U32, xs: List<&2, U32>, acc: A) -> Replay<A, List<&2, U32>>:
216 replay(~List<&2, U32>, ~List<&2, U32>, ~A, ~list.fits, ~list.u32, ~list.cut, ~list.skip, ~list.sum, ~step, n, max, xs, acc)
217
218def list.head(+len: U32, +sum: U32) -> List<&2, U32>:
219 [len, 0, 0, 0, U32.not(len), 0, 0, 0, sum, 0, 0, 0]
220
221def list.frame(+p: List<&2, U32>) -> List<&2, U32>:
222 List.append(&2, U32, list.head(U32.from_nat(List.length(&2, U32, p)), list.crc(p)), p)
223
224# The frames of rs, one after another: what appending each record writes.
225def list.log(rs: List<&2, List<&2, U32>>) -> List<&2, U32>:
226 match rs:
227 case Nil{}:
228 Nil{}
229 case Con{p, t}:
230 List.append(&2, U32, list.frame(p), list.log(t))
231
232# The Bytes instance. A cursor reads b from pos; pos only moves forward, past whole frames.
233type Cur is Type:
234 Cur{b: Bytes.Bytes, pos: U32}
235
236def cur.fits(c: Cur, +at: U32, +n: U32) -> Cur & Bool:
237 Cur{Bytes.Bytes{+len, buf}, +pos} = c
238 (Cur{Bytes.Bytes{len, buf}, pos}, Bytes.fits((len - pos : U32), at, n))
239
240def cur.field(m: Maybe<&2, U32>) -> U32:
241 match m:
242 case Some{x}:
243 x
244 case None{}:
245 0
246
247def cur.u32.fin(+pos: U32, r: Bytes.Bytes & Maybe<&2, U32>) -> Cur & U32:
248 (b, m) = r
249 (Cur{b, pos}, cur.field(m))
250
251def cur.u32(c: Cur, +at: U32) -> Cur & U32:
252 Cur{b, +pos} = c
253 cur.u32.fin(pos, Bytes.get.u32le(b, (pos + at : U32)))
254
255def cur.cut.fin(+pos: U32, r: Bytes.Bytes & Bytes.Bytes) -> Cur & Bytes.Bytes:
256 (b, p) = r
257 (Cur{b, pos}, p)
258
259def cur.cut(c: Cur, +hs: U32, +len: U32) -> Cur & Bytes.Bytes:
260 Cur{b, +pos} = c
261 cur.cut.fin(pos, Bytes.slice(b, (pos + hs : U32), len))
262
263def cur.skip(c: Cur, +hs: U32, +len: U32) -> Cur:
264 Cur{b, +pos} = c
265 Cur{b, (pos + hs + len : U32)}
266
267# ponytail: Hash.crc32 builds its 256-entry table per record; share one table if recovery gets hot.
268def cur.sum(p: Bytes.Bytes) -> Bytes.Bytes & U32:
269 Hash.crc32(p)
270
271# Every frame of b holds at least 12 octets, so len / 12 + 1 steps reach the end.
272def fuel(+len: U32) -> Nat:
273 U32.to_nat((len / 12 + 1 : U32))
274
275# step folded over every committed record of b, in order. rest.pos is where the committed prefix ends.
276def replay.bytes(~A: Type, ~step: A -> Bytes.Bytes -> A, +max: U32, b: Bytes.Bytes, acc: A) -> Replay<A, Cur>:
277 Bytes.Bytes{+len, buf} = b
278 replay(~Bytes.Bytes, ~Cur, ~A, ~cur.fits, ~cur.u32, ~cur.cut, ~cur.skip, ~cur.sum, ~step, fuel(len), max, Cur{Bytes.Bytes{len, buf}, 0}, acc)
279
280def frame.of(+len: U32, r: Bytes.Bytes & U32) -> Bytes.Bytes:
281 (p, +sum) = r
282 Bytes.append(Bytes.set.u32le(Bytes.set.u32le(Bytes.set.u32le(Bytes.new(12), 0, len), 4, U32.not(len)), 8, sum), p)
283
284# The frame that appending p writes.
285def frame(p: Bytes.Bytes) -> Bytes.Bytes:
286 Bytes.Bytes{+len, buf} = p
287 frame.of(len, Hash.crc32(Bytes.Bytes{len, buf}))
288
289# Effects. Each one finishes its write with fsync (F_FULLFSYNC on macOS in C) before it answers.
290
291# Opens path for reading and appending, creating it, and fsyncs its directory. Answers the file and
292# its whole contents. Fails with EFBIG when the file holds more than limit octets.
293def open.words(path: String, limit: U32) -> IO(Result<&1, &1, U32 & String, File & U32 & Array<U32>>):
294 import "./effs/journal.c"
295 import "./effs/journal.js"
296
297# Writes len octets of words at the end of file, then fsyncs it.
298def append.words(file: File, len: U32, words: Array<U32>) -> IO(File & Result<&1, &1, U32 & String, Unit>):
299 import "./effs/journal.c"
300 import "./effs/journal.js"
301
302# Cuts file to len octets, then fsyncs it.
303def truncate(file: File, len: U32) -> IO(File & Result<&1, &1, U32 & String, Unit>):
304 import "./effs/journal.c"
305 import "./effs/journal.js"
306
307# An open journal. max caps a record's length. end is the file's length. ok turns False when an
308# append fails: what reached the disk is then unknown, so appends stop until the journal is reopened.
309type Journal is Type:
310 Journal{file: File, max: U32, end: U32, ok: Bool}
311
312def open.cut(r: File & Result<&1, &1, U32 & String, Unit>, +max: U32, +end: U32, -A: Type, acc: A) -> IO(Result<&1, &1, U32 & String, Journal & A>):
313 match r:
314 case (file, Fail{e}):
315 do IO<Result<&1, &1, U32 & String, Journal & A>>:
316 File.close(file)
317 return Fail{e}
318 case (file, Done{u}):
319 do IO<Result<&1, &1, U32 & String, Journal & A>>:
320 return Done{(Journal{file, max, end, True{}}, acc)}
321
322def open.tail(t: Tail, file: File, +max: U32, +pos: U32, -A: Type, acc: A) -> IO(Result<&1, &1, U32 & String, Journal & A>):
323 match t:
324 case Clean{}:
325 do IO<Result<&1, &1, U32 & String, Journal & A>>:
326 return Done{(Journal{file, max, pos, True{}}, acc)}
327 case Torn{}:
328 do IO<Result<&1, &1, U32 & String, Journal & A>>:
329 r : File & Result<&1, &1, U32 & String, Unit> <- truncate(file, pos)
330 open.cut(r, max, pos, A, acc)
331 case Corrupt{why}:
332 do IO<Result<&1, &1, U32 & String, Journal & A>>:
333 File.close(file)
334 return Fail{(22, "journal: corrupt frame at offset " ++ U32.show(pos) ++ ": " ++ why)}
335
336def open.replayed(file: File, +max: U32, -A: Type, r: Replay<A, Cur>) -> IO(Result<&1, &1, U32 & String, Journal & A>):
337 Replay{acc, t, rest} = r
338 Cur{b, +pos} = rest
339 open.tail(t, file, max, pos, A, acc)
340
341def open.read(~A: Type, ~step: A -> Bytes.Bytes -> A, r: Result<&1, &1, U32 & String, File & U32 & Array<U32>>, +max: U32, acc: A) -> IO(Result<&1, &1, U32 & String, Journal & A>):
342 match r:
343 case Fail{e}:
344 do IO<Result<&1, &1, U32 & String, Journal & A>>:
345 return Fail{e}
346 case Done{(file, (len, words))}:
347 open.replayed(file, max, A, replay.bytes(~A, ~step, max, Bytes.Bytes{len, words}, acc))
348
349# Opens the journal at path and folds step over its committed records, from acc.
350# A torn tail is cut off and fsynced before open answers. A corrupt frame fails with EINVAL and
351# its offset; the file is left as it is. max caps a record's length; limit caps the file's size.
352def open(~A: Type, ~step: A -> Bytes.Bytes -> A, path: String, +max: U32, limit: U32, acc: A) -> IO(Result<&1, &1, U32 & String, Journal & A>):
353 do IO<Result<&1, &1, U32 & String, Journal & A>>:
354 r : Result<&1, &1, U32 & String, File & U32 & Array<U32>> <- open.words(path, limit)
355 open.read(~A, ~step, r, max, acc)
356
357def append.back(+max: U32, +end: U32, +len: U32, r: File & Result<&1, &1, U32 & String, Unit>) -> Journal & Result<&1, &1, U32 & String, Unit>:
358 match r:
359 case (file, Done{u}):
360 (Journal{file, max, (end + len : U32), True{}}, Done{u})
361 case (file, Fail{e}):
362 (Journal{file, max, end, False{}}, Fail{e})
363
364def append.write(file: File, +max: U32, +end: U32, f: Bytes.Bytes) -> IO(Journal & Result<&1, &1, U32 & String, Unit>):
365 Bytes.Bytes{+len, words} = f
366 do IO<Journal & Result<&1, &1, U32 & String, Unit>>:
367 r : File & Result<&1, &1, U32 & String, Unit> <- append.words(file, len, words)
368 return append.back(max, end, len, r)
369
370def append.refuse(file: File, +max: U32, +end: U32, +ok: Bool, e: U32 & String) -> IO(Journal & Result<&1, &1, U32 & String, Unit>):
371 do IO<Journal & Result<&1, &1, U32 & String, Unit>>:
372 return (Journal{file, max, end, ok}, Fail{e})
373
374# The file stays under 4 GiB, so every offset and the next open's read fit a U32.
375def append.size(over: Bool, big: Bool, file: File, +max: U32, +end: U32, p: Bytes.Bytes) -> IO(Journal & Result<&1, &1, U32 & String, Unit>):
376 match over big:
377 case True{} _:
378 append.refuse(file, max, end, True{}, (22, "journal: record over max"))
379 case False{} True{}:
380 append.refuse(file, max, end, True{}, (27, "journal: file would reach 4 GiB"))
381 case False{} False{}:
382 append.write(file, max, end, frame(p))
383
384def append.ok(ok: Bool, file: File, +max: U32, +end: U32, p: Bytes.Bytes) -> IO(Journal & Result<&1, &1, U32 & String, Unit>):
385 match ok:
386 case False{}:
387 append.refuse(file, max, end, False{}, (5, "journal: an append failed; reopen the journal"))
388 case True{}:
389 Bytes.Bytes{+len, buf} = p
390 append.size(U32.is_gt(len, max), Bool.or(U32.is_gt(end, 4294967283), U32.is_lt((4294967283 - end : U32), len)), file, max, end, Bytes.Bytes{len, buf})
391
392# Appends p as one frame and answers once it is fsynced: Done means durable, under the trust
393# assumptions in README.md. After a Fail the record may or may not survive a reopen.
394def append(j: Journal, p: Bytes.Bytes) -> IO(Journal & Result<&1, &1, U32 & String, Unit>):
395 Journal{file, +max, +end, ok} = j
396 append.ok(ok, file, max, end, p)
397
398def close(j: Journal) -> IO(Unit):
399 Journal{file, max, end, ok} = j
400 File.close(file)