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 2 import 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. 4 import 0x49814d83de8f70993a43e1002be29ecd/bytes.bend as Bytes 5 import 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. 14 type 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. 20 type 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. 27 type Recovery is Type: 28 Recovery{records: List, 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. 31 type 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 38 def next.verdict(~P: Type, ~S: Type, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, ok: Bool, +len: U32, s: S, p: P) -> Step: 39 match ok: 40 case True{}: 41 Got{p, skip(s, 12, len)} 42 case False{}: 43 Bad{"checksum mismatch", s} 44 45 def next.summed(~P: Type, ~S: Type, ~skip: S -> @+hs:U32 -> @+len:U32 -> S, +want: U32, +len: U32, s: S, r: P & U32) -> Step: 46 (p, +got) = r 47 next.verdict(~P, ~S, ~skip, U32.is_eq(got, want), len, s, p) 48 49 def 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: 50 (s, p) = r 51 next.summed(~P, ~S, ~skip, want, len, s, sum(p)) 52 53 def 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: 54 (s, +want) = r 55 next.cut(~P, ~S, ~skip, ~sum, want, len, cut(s, 12, len)) 56 57 def 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: 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 65 def 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: 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 72 def 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: 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 79 def 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: 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 83 def 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: 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. 89 def 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: 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 97 def 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: 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. 106 def 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: 107 next.any(~P, ~S, ~fits, ~u32, ~cut, ~skip, ~sum, max, fits(s, 0, 1)) 108 109 def recover.con(~a: Quant, ~P: Kind(a), ~S: Type, p: P, r: Recovery) -> Recovery: 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. 115 def 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) -> Recovery: 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. 135 def 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: 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. 139 def 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, acc: A) -> Replay: 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 158 def 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: 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 165 def 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 169 def list.field(xs: List<&2, U32>) -> U32: 170 match xs: 171 case Con{x, t}: 172 x 173 case Nil{}: 174 0 175 176 def 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 180 def 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 184 def 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 187 def 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 194 def 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. 202 def list.crc(xs: List<&2, U32>) -> U32: 203 U32.not(crc.go(xs, 4294967295)) 204 205 def list.sum(xs: List<&2, U32>) -> List<&2, U32> & U32: 206 +ys = xs 207 (ys, list.crc(ys)) 208 209 def list.next(+max: U32, xs: List<&2, U32>) -> Step, List<&2, U32>>: 210 next(~List<&2, U32>, ~List<&2, U32>, ~list.fits, ~list.u32, ~list.cut, ~list.skip, ~list.sum, max, xs) 211 212 def 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 215 def list.replay(~A: Type, ~step: A -> List<&2, U32> -> A, n: Nat, +max: U32, xs: List<&2, U32>, acc: A) -> Replay>: 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 218 def 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 221 def 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. 225 def 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. 233 type Cur is Type: 234 Cur{b: Bytes.Bytes, pos: U32} 235 236 def 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 240 def cur.field(m: Maybe<&2, U32>) -> U32: 241 match m: 242 case Some{x}: 243 x 244 case None{}: 245 0 246 247 def 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 251 def 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 255 def cur.cut.fin(+pos: U32, r: Bytes.Bytes & Bytes.Bytes) -> Cur & Bytes.Bytes: 256 (b, p) = r 257 (Cur{b, pos}, p) 258 259 def 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 263 def 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. 268 def 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. 272 def 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. 276 def replay.bytes(~A: Type, ~step: A -> Bytes.Bytes -> A, +max: U32, b: Bytes.Bytes, acc: A) -> Replay: 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 280 def 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. 285 def 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. 293 def open.words(path: String, limit: U32) -> IO(Result<&1, &1, U32 & String, File & U32 & Array>): 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. 298 def append.words(file: File, len: U32, words: Array) -> 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. 303 def 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. 309 type Journal is Type: 310 Journal{file: File, max: U32, end: U32, ok: Bool} 311 312 def 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>: 316 File.close(file) 317 return Fail{e} 318 case (file, Done{u}): 319 do IO>: 320 return Done{(Journal{file, max, end, True{}}, acc)} 321 322 def 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>: 326 return Done{(Journal{file, max, pos, True{}}, acc)} 327 case Torn{}: 328 do IO>: 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>: 333 File.close(file) 334 return Fail{(22, "journal: corrupt frame at offset " ++ U32.show(pos) ++ ": " ++ why)} 335 336 def open.replayed(file: File, +max: U32, -A: Type, r: Replay) -> 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 341 def open.read(~A: Type, ~step: A -> Bytes.Bytes -> A, r: Result<&1, &1, U32 & String, File & U32 & Array>, +max: U32, acc: A) -> IO(Result<&1, &1, U32 & String, Journal & A>): 342 match r: 343 case Fail{e}: 344 do IO>: 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. 352 def 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>: 354 r : Result<&1, &1, U32 & String, File & U32 & Array> <- open.words(path, limit) 355 open.read(~A, ~step, r, max, acc) 356 357 def 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 364 def 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>: 367 r : File & Result<&1, &1, U32 & String, Unit> <- append.words(file, len, words) 368 return append.back(max, end, len, r) 369 370 def append.refuse(file: File, +max: U32, +end: U32, +ok: Bool, e: U32 & String) -> IO(Journal & Result<&1, &1, U32 & String, Unit>): 371 do IO>: 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. 375 def 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 384 def 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. 394 def 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 398 def close(j: Journal) -> IO(Unit): 399 Journal{file, max, end, ok} = j 400 File.close(file)