src/Store.bend
ber-core-store@0.1.2.0 · 213 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.3.2.0/mylsm.bend as MyLsmStore
3import mylsm-lsm-store@0.3.2.0/src/Db.bend as DbTypes
4import mylsm-lsm-store@0.3.2.0/src/Wal.bend as WalTypes
5
6# Session effects for ber-core. Op is a newtype over Db transitions and
7# Handle over MyLSM Db; do-blocks desugar through Op.bind/Op.pure exactly
8# like Sess. Pairs sit only in return and parameter positions (never as a
9# Sess answer), so Op stays kind-polymorphic; computed pairs thread through
10# helpers (match on parameters, never on computed values). Defs are ordered
11# leaf-first: no def calls one written below it.
12#
13# Every op returns its answer with a journal of physical mutations
14# (execution order). Pure run_op drops it; run_op_tracked exposes it and
15# run_op_durable appends it through one WAL frame plus fsync. oput/odel
16# journal exactly what they apply, so the WAL mirrors the memtable.
17
18# Journal rope: O(1) per-bind combination (no calls, just nesting); flattened
19# once at flush with an explicit work stack (constant native stack) into
20# chronological order. Reads build Empty spines that flatten to Nil.
21type Journal is Data:
22 JEmpty{}
23 JOne{mut: WalTypes.Mut}
24 JMore{prior: Journal, post: Journal}
25
26# Counts rope nodes structurally; the work loop below visits each once.
27def rope_nodes(+journal: Journal) -> Nat:
28 match journal:
29 case JEmpty{}:
30 1n
31 case JOne{unlogged_mut}:
32 1n
33 case JMore{prior, post}:
34 Nat.add(1n, Nat.add(rope_nodes(prior), rope_nodes(post)))
35
36# Flattens a rope oldest-first with an explicit work stack; the accumulator
37# holds the reversed prefix. Fuel covers nodes plus the final empty check.
38def flatten_go(fuel: Nat, +work: List<&2, Journal>, +acc: List<&2, WalTypes.Mut>) -> List<&2, WalTypes.Mut>:
39 match fuel:
40 case 0n:
41 acc
42 case 1n+fuel_left:
43 match work:
44 case Nil{}:
45 List.reverse(&2, WalTypes.Mut, acc)
46 case Con{head, tail}:
47 match head:
48 case JEmpty{}:
49 flatten_go(fuel_left, tail, acc)
50 case JOne{mut}:
51 flatten_go(fuel_left, tail, Con{mut, acc})
52 case JMore{prior, post}:
53 flatten_go(fuel_left, Con{prior, Con{post, tail}}, acc)
54
55# Materializes a journal rope into its chronological mutation list.
56def flatten_journal(+journal: Journal) -> List<&2, WalTypes.Mut>:
57 flatten_go(Nat.add(rope_nodes(journal), 1n), Con{journal, Nil{}}, Nil{})
58
59# One session step chain: a Db transition answering with its journal.
60type Op<a, -A: Kind(a)> is Kind(a <&> &1):
61 Op{run: DbTypes.Db -> (DbTypes.Db & A) & Journal}
62
63# An opened store handle; affine, threaded through run_op.
64type Handle is Type:
65 Handle{db: DbTypes.Db}
66
67# Unwraps the inner transition for translators like bind and run_op.
68def Op.inner(b, -B: Kind(b), op: Op<b, B>) -> DbTypes.Db -> (DbTypes.Db & B) & Journal:
69 match op:
70 case Op{run_fn}:
71 run_fn
72
73# Lifts a pure value with an empty journal.
74def Op.pure(a, -A: Kind(a), val: A) -> Op<a, A>:
75 Op{st => ((st, val), JEmpty{})}
76
77# Concatenates a prior journal before a later one, finishing the second step.
78def Op.seq2(a, -B: Kind(a), +prior: Journal, second: (DbTypes.Db & B) & Journal) -> (DbTypes.Db & B) & Journal:
79 match second:
80 case ((db_next, answer), journal):
81 ((db_next, answer), JMore{prior, journal})
82
83# Runs the next op on the first answer, threading Db and journal.
84def Op.seq(a, -A: Kind(a), -B: Kind(a), first: (DbTypes.Db & A) & Journal, fun: A -> Op<a, B>) -> (DbTypes.Db & B) & Journal:
85 match first:
86 case ((db_mid, answer), prior):
87 Op.seq2(a, B, prior, Op.inner(a, B, fun(answer))(db_mid))
88
89# Sequences two ops, concatenating journals in execution order.
90def Op.bind(a, -A: Kind(a), -B: Kind(a), op: Op<a, A>, fun: A -> Op<a, B>) -> Op<a, B>:
91 match op:
92 case Op{run1}:
93 Op{st => Op.seq(a, A, B, run1(st), fun)}
94
95# Pairs a finished point write with its journaled mutation.
96def pair_put(+key: String, +val: String, db_next: DbTypes.Db) -> (DbTypes.Db & Unit) & Journal:
97 ((db_next, Unit{}), JOne{WalTypes.Put{key, val}})
98
99# Writes one key and journals the mutation.
100def oput(+key: String, +val: String) -> Op<&2, Unit>:
101 Op{st => pair_put(key, val, MyLsmStore.put_go(st, key, val))}
102
103# Pairs a finished point read with an empty journal.
104def pair_get(finished: DbTypes.Db & Maybe<&2, String>) -> (DbTypes.Db & Maybe<&2, String>) & Journal:
105 match finished:
106 case (db_next, found_val):
107 ((db_next, found_val), JEmpty{})
108
109# Reads one key; missing reads as None. Reads journal nothing.
110def oget(+key: String) -> Op<&2, Maybe<&2, String>>:
111 Op{st => pair_get(MyLsmStore.sget_go(st, key))}
112
113# Pairs a finished point delete with its journaled mutation.
114def pair_del(+key: String, db_next: DbTypes.Db) -> (DbTypes.Db & Unit) & Journal:
115 ((db_next, Unit{}), JOne{WalTypes.Del{key}})
116
117# Deletes one key and journals the mutation.
118def odel(+key: String) -> Op<&2, Unit>:
119 Op{st => pair_del(key, MyLsmStore.del_go(st, key))}
120
121# Opens a store handle over a directory label (pure; no IO touched).
122def open_store(+dir: String) -> Handle:
123 Handle{MyLsmStore.open(dir)}
124
125# Default store label for CLI use (pure; no IO touched).
126def open_default_store() -> Handle:
127 open_store("./.ber")
128
129# Wraps a reopened database; IO errors propagate as Fail.
130def wrap_reopened(reopened: Result<&1, &1, U32 & String, DbTypes.Db>) -> Result<&1, &1, U32 & String, Handle>:
131 match reopened:
132 case Fail{error}:
133 Fail{error}
134 case Done{db}:
135 Done{Handle{db}}
136
137# Opens a durable handle, replaying wal.log when present; missing file
138# answers empty. The caller ensures the directory exists.
139def open_durable(+dir: String) -> IO(Result<&1, &1, U32 & String, Handle>):
140 do IO<Result<&1, &1, U32 & String, Handle>>:
141 reopened : Result<&1, &1, U32 & String, DbTypes.Db> <- MyLsmStore.reopen(dir)
142 return wrap_reopened(reopened)
143
144# Projects the materialized journal out of a tracked run (laws observe this).
145def tracked_journal(a, -A: Kind(a), tracked: (Handle & A) & Journal) -> List<&2, WalTypes.Mut>:
146 match tracked:
147 case ((unused_handle, unused_answer), journal):
148 flatten_journal(journal)
149
150# Rewraps a computed runner pair into Handle form, dropping the journal.
151def rewrap_pair(a, -A: Kind(a), computed: (DbTypes.Db & A) & Journal) -> Handle & A:
152 match computed:
153 case ((db_next, answer), unused_journal):
154 (Handle{db_next}, answer)
155
156# Executes a whole op chain against a handle (pure; journal dropped).
157def run_op(a, -A: Kind(a), store: Handle, op: Op<a, A>) -> Handle & A:
158 match store:
159 case Handle{db}:
160 rewrap_pair(a, A, Op.inner(a, A, op)(db))
161
162# Rewraps a tracked runner pair, keeping the journal.
163def rewrap_tracked(a, -A: Kind(a), computed: (DbTypes.Db & A) & Journal) -> (Handle & A) & Journal:
164 match computed:
165 case ((db_next, answer), journal):
166 ((Handle{db_next}, answer), journal)
167
168# Executes a whole op chain, exposing the journal for the WAL runner.
169def run_op_tracked(a, -A: Kind(a), store: Handle, op: Op<a, A>) -> (Handle & A) & Journal:
170 match store:
171 case Handle{db}:
172 rewrap_tracked(a, A, Op.inner(a, A, op)(db))
173
174# Wraps a written database with its answer; IO errors propagate as Fail.
175def wrap_written(a, -A: Kind(a), written: Result<&1, &1, U32 & String, DbTypes.Db>, answer: A) -> Result<&1, &1, U32 & String, Handle & A>:
176 match written:
177 case Fail{error}:
178 Fail{error}
179 case Done{db_next}:
180 Done{(Handle{db_next}, answer)}
181
182# Skips the WAL when the journal is empty (pure reads never fsync).
183def flush_clean(a, -A: Kind(a), store_next: Handle, answer: A) -> IO(Result<&1, &1, U32 & String, Handle & A>):
184 match store_next:
185 case Handle{db}:
186 IO.pure(Result<&1, &1, U32 & String, Handle & A>, Done{(Handle{db}, answer)})
187
188# Flushes a nonempty journal through one WAL frame plus fsync.
189def flush_dirty(a, -A: Kind(a), store_next: Handle, answer: A, +head_mut: WalTypes.Mut, +tail_muts: List<&2, WalTypes.Mut>) -> IO(Result<&1, &1, U32 & String, Handle & A>):
190 match store_next:
191 case Handle{db}:
192 do IO<Result<&1, &1, U32 & String, Handle & A>>:
193 written : Result<&1, &1, U32 & String, DbTypes.Db> <- DbTypes.db_write(db, WalTypes.Batch{Con{head_mut, tail_muts}})
194 return wrap_written(a, A, written, answer)
195
196# Dispatches on journal emptiness without matching computed values.
197def flush_list(a, -A: Kind(a), store_next: Handle, answer: A, +mutations: List<&2, WalTypes.Mut>) -> IO(Result<&1, &1, U32 & String, Handle & A>):
198 match mutations:
199 case Nil{}:
200 flush_clean(a, A, store_next, answer)
201 case Con{head_mut, tail_muts}:
202 flush_dirty(a, A, store_next, answer, head_mut, tail_muts)
203
204# Splits a tracked runner pair for the flush dispatch.
205def flush_tracked(a, -A: Kind(a), tracked: (Handle & A) & Journal) -> IO(Result<&1, &1, U32 & String, Handle & A>):
206 match tracked:
207 case ((store_next, answer), journal):
208 flush_list(a, A, store_next, answer, flatten_journal(journal))
209
210# Executes a whole op chain durably: pure run, then one WAL append plus
211# fsync of exactly the journaled mutations.
212def run_op_durable(a, -A: Kind(a), store: Handle, op: Op<a, A>) -> IO(Result<&1, &1, U32 & String, Handle & A>):
213 flush_tracked(a, A, run_op_tracked(a, A, store, op))