commands.bend
bend-kit-commands@0.1.0.0 · 327 lines · 0 definitions
1# Retry-safe commands: deduplicate by command id, replay recorded results, and commit state, record, and outbox in one SQLite transaction. Source: https://github.com/paymog/bend-kit/tree/main/commands
2import Base
3import bend-kit-sqlite@0.1.0.1/sqlite.bend as Sqlite
4
5# ---- the pure processor
6
7type Outcome<-R: Data> is Data:
8 Accepted{result: R}
9 Refused{result: R}
10
11# What the store keeps for one command id: its payload, its outcome, and when it was recorded.
12type Record<-P: Data, -R: Data> is Data:
13 Record{payload: P, outcome: Outcome<R>, at: U32}
14
15type Reply<-R: Data> is Data:
16 Fresh{outcome: Outcome<R>}
17 Replayed{outcome: Outcome<R>}
18 Conflict{}
19
20# The domain's answer to a new command.
21type Decision<-S: Data, -R: Data, -E: Data> is Data:
22 Accept{state: S, result: R, effects: List<&2, E>}
23 Refuse{result: R}
24
25# Write: commit the state, the record, and the effects together. Keep: write nothing.
26type Step<-S: Data, -P: Data, -R: Data, -E: Data> is Data:
27 Write{state: S, record: Record<P, R>, reply: Reply<R>, effects: List<&2, E>}
28 Keep{reply: Reply<R>}
29
30def decided(-S: Data, -P: Data, -R: Data, -E: Data, d: Decision<S, R, E>, s: S, p: P, at: U32) -> Step<S, P, R, E>:
31 match d:
32 case Accept{s2, +r, es}:
33 Write{s2, Record{p, Accepted{r}, at}, Fresh{Accepted{r}}, es}
34 case Refuse{+r}:
35 Write{s, Record{p, Refused{r}, at}, Fresh{Refused{r}}, Nil{}}
36
37def replayed(-R: Data, same: Bool, o: Outcome<R>) -> Reply<R>:
38 match same:
39 case True{}:
40 Replayed{o}
41 case False{}:
42 Conflict{}
43
44# One command, given the record that the store holds for its id and the domain state s.
45def step(
46 ~S: Data, ~P: Data, ~R: Data, ~E: Data, ~eq: P -> P -> Bool, ~decide: S -> P -> Decision<S, R, E>,
47 found: Maybe<&2, Record<P, R>>, +s: S, +p: P, at: U32
48) -> Step<S, P, R, E>:
49 match found:
50 case None{}:
51 decided(S, P, R, E, decide(s, p), s, p, at)
52 case Some{Record{q, o, t}}:
53 Keep{replayed(R, eq(q, p), o)}
54
55# ---- IO that can fail with a SQLite error
56
57def Run(-A: Type) -> Type:
58 IO(Result<&1, &1, U32 & String, A>)
59
60def Run.pure(-A: Type, x: A) -> Run(A):
61 IO.pure(Result<&1, &1, U32 & String, A>, Done{x})
62
63def Run.fail(-A: Type, e: U32 & String) -> Run(A):
64 IO.pure(Result<&1, &1, U32 & String, A>, Fail{e})
65
66def Run.next(-A: Type, -B: Type, r: Result<&1, &1, U32 & String, A>, f: A -> Run(B)) -> Run(B):
67 match r:
68 case Done{x}:
69 f(x)
70 case Fail{e}:
71 Run.fail(B, e)
72
73def Run.bind(-A: Type, -B: Type, m: Run(A), f: A -> Run(B)) -> Run(B):
74 IO.bind(Result<&1, &1, U32 & String, A>, Result<&1, &1, U32 & String, B>, m, r => Run.next(A, B, r, f))
75
76# r's value, unless r or then z failed.
77def first(-A: Type, r: Result<&1, &1, U32 & String, A>, z: Result<&1, &1, U32 & String, Unit>) -> Result<&1, &1, U32 & String, A>:
78 match r z:
79 case Fail{e} _:
80 Fail{e}
81 case Done{x} Fail{e}:
82 Fail{e}
83 case Done{x} Done{u}:
84 Done{x}
85
86def unit(-A: Type, r: Result<&1, &1, U32 & String, A>) -> Result<&1, &1, U32 & String, Unit>:
87 match r:
88 case Fail{e}:
89 Fail{e}
90 case Done{x}:
91 Done{Unit{}}
92
93# ---- SQLite helpers
94
95# Steps a statement and ignores any row, then resets it, also after a failed step.
96def exec(+stmt: U32) -> Run(Unit):
97 do IO<Result<&1, &1, U32 & String, Unit>>:
98 r : Result<&1, &1, U32 & String, Bool> <- Sqlite.stmt.step(stmt)
99 z : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(stmt)
100 return first(Unit, unit(Bool, r), z)
101
102def once.run(-A: Type, +s: U32, body: U32 -> Run(A)) -> Run(A):
103 do IO<Result<&1, &1, U32 & String, A>>:
104 r : Result<&1, &1, U32 & String, A> <- body(s)
105 z : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.finalize(s)
106 return first(A, r, z)
107
108# Prepares one statement, runs body on it, and finalizes it, also after body fails.
109def once(-A: Type, +db: U32, text: String, body: U32 -> Run(A)) -> Run(A):
110 do Run<A>:
111 s : U32 <- Sqlite.stmt.prepare(db, text)
112 once.run(A, s, body)
113
114def sql(+db: U32, text: String) -> Run(Unit):
115 once(Unit, db, text, +s => exec(s))
116
117# ---- the store
118
119# A database and its prepared statements. One Store serves one task at a time.
120type Store is Data:
121 Store{db: U32, begin: U32, commit: U32, rollback: U32, find: U32, read: U32, save: U32, keep: U32, emit: U32}
122
123def init.run(+s: U32, init: String) -> Run(Unit):
124 do Run<Unit>:
125 Sqlite.stmt.bind.text(s, 1, init)
126 exec(s)
127
128# Opens or creates the tables at path. A new database starts with domain state init.
129# wait is how many ms a write waits for another connection's transaction before it fails with SQLITE_BUSY (5).
130def open(path: String, init: String, wait: U32) -> Run(Store):
131 do Run<Store>:
132 +db : U32 <- Sqlite.db.open(path)
133 sql(db, "PRAGMA busy_timeout = " ++ U32.show(wait))
134 sql(db, "CREATE TABLE IF NOT EXISTS commands_state (k INTEGER PRIMARY KEY CHECK (k = 0), state TEXT NOT NULL)")
135 sql(db, "CREATE TABLE IF NOT EXISTS commands_seen (id TEXT PRIMARY KEY, payload TEXT NOT NULL, kind INTEGER NOT NULL, result TEXT NOT NULL, at INTEGER NOT NULL)")
136 sql(db, "CREATE INDEX IF NOT EXISTS commands_seen_at ON commands_seen (at)")
137 sql(db, "CREATE TABLE IF NOT EXISTS commands_outbox (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT NOT NULL, effect TEXT NOT NULL)")
138 once(Unit, db, "INSERT OR IGNORE INTO commands_state (k, state) VALUES (0, ?)", +s => init.run(s, init))
139 begin : U32 <- Sqlite.stmt.prepare(db, "BEGIN IMMEDIATE")
140 commit : U32 <- Sqlite.stmt.prepare(db, "COMMIT")
141 rollback : U32 <- Sqlite.stmt.prepare(db, "ROLLBACK")
142 find : U32 <- Sqlite.stmt.prepare(db, "SELECT payload, kind, result, at FROM commands_seen WHERE id = ?")
143 read : U32 <- Sqlite.stmt.prepare(db, "SELECT state FROM commands_state WHERE k = 0")
144 save : U32 <- Sqlite.stmt.prepare(db, "UPDATE commands_state SET state = ? WHERE k = 0")
145 keep : U32 <- Sqlite.stmt.prepare(db, "INSERT INTO commands_seen (id, payload, kind, result, at) VALUES (?, ?, ?, ?, ?)")
146 emit : U32 <- Sqlite.stmt.prepare(db, "INSERT INTO commands_outbox (id, effect) VALUES (?, ?)")
147 return Store{db, begin, commit, rollback, find, read, save, keep, emit}
148
149def close(st: Store) -> Run(Unit):
150 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
151 do Run<Unit>:
152 Sqlite.stmt.finalize(begin)
153 Sqlite.stmt.finalize(commit)
154 Sqlite.stmt.finalize(rollback)
155 Sqlite.stmt.finalize(find)
156 Sqlite.stmt.finalize(read)
157 Sqlite.stmt.finalize(save)
158 Sqlite.stmt.finalize(keep)
159 Sqlite.stmt.finalize(emit)
160 Sqlite.db.close(db)
161
162def outcome(+kind: U32, +r: String) -> Outcome<String>:
163 Bool.pick(Outcome<String>, U32.is_eq(kind, 1), Accepted{r}, Refused{r})
164
165def seen.row(row: Bool, +find: U32) -> Run(Maybe<&2, Record<String, String>>):
166 match row:
167 case False{}:
168 Run.pure(Maybe<&2, Record<String, String>>, None{})
169 case True{}:
170 do Run<Maybe<&2, Record<String, String>>>:
171 p : String <- Sqlite.stmt.column.text(find, 0)
172 kind : U32 <- Sqlite.stmt.column.int(find, 1)
173 r : String <- Sqlite.stmt.column.text(find, 2)
174 at : U32 <- Sqlite.stmt.column.int(find, 3)
175 return Some{Record{p, outcome(kind, r), at}}
176
177# The record for id, if the store holds one.
178def seen.get(+find: U32, id: String) -> Run(Maybe<&2, Record<String, String>>):
179 do Run<Maybe<&2, Record<String, String>>>:
180 Sqlite.stmt.bind.text(find, 1, id)
181 row : Bool <- Sqlite.stmt.step(find)
182 m : Maybe<&2, Record<String, String>> <- seen.row(row, find)
183 Sqlite.stmt.reset(find)
184 return m
185
186def state.row(row: Bool, +read: U32) -> Run(String):
187 match row:
188 case True{}:
189 Sqlite.stmt.column.text(read, 0)
190 case False{}:
191 Run.fail(String, (1, "commands: the state row is missing"))
192
193def state.get(+read: U32) -> Run(String):
194 do Run<String>:
195 row : Bool <- Sqlite.stmt.step(read)
196 s : String <- state.row(row, read)
197 Sqlite.stmt.reset(read)
198 return s
199
200def seen.put.row(+keep: U32, id: String, p: String, kind: U32, r: String, at: U32) -> Run(Unit):
201 do Run<Unit>:
202 Sqlite.stmt.bind.text(keep, 1, id)
203 Sqlite.stmt.bind.text(keep, 2, p)
204 Sqlite.stmt.bind.int(keep, 3, kind)
205 Sqlite.stmt.bind.text(keep, 4, r)
206 Sqlite.stmt.bind.int(keep, 5, at)
207 exec(keep)
208
209def seen.put.of(o: Outcome<String>, +keep: U32, id: String, p: String, at: U32) -> Run(Unit):
210 match o:
211 case Accepted{r}:
212 seen.put.row(keep, id, p, 1, r, at)
213 case Refused{r}:
214 seen.put.row(keep, id, p, 2, r, at)
215
216def seen.put(rec: Record<String, String>, +keep: U32, id: String) -> Run(Unit):
217 Record{p, o, at} = rec
218 seen.put.of(o, keep, id, p, at)
219
220def outbox.put(es: List<&2, String>, +emit: U32, +id: String) -> Run(Unit):
221 match es:
222 case Nil{}:
223 Run.pure(Unit, Unit{})
224 case Con{e, rest}:
225 do Run<Unit>:
226 Sqlite.stmt.bind.text(emit, 1, id)
227 Sqlite.stmt.bind.text(emit, 2, e)
228 exec(emit)
229 outbox.put(rest, emit, id)
230
231def state.put(+save: U32, s: String) -> Run(Unit):
232 do Run<Unit>:
233 Sqlite.stmt.bind.text(save, 1, s)
234 exec(save)
235
236# Applies a step inside the open transaction, then commits it.
237def written(w: Step<String, String, String, String>, +commit: U32, +save: U32, +keep: U32, +emit: U32, +id: String) -> Run(Reply<String>):
238 match w:
239 case Write{s, rec, reply, es}:
240 do Run<Reply<String>>:
241 state.put(save, s)
242 seen.put(rec, keep, id)
243 outbox.put(es, emit, id)
244 exec(commit)
245 return reply
246 case Keep{reply}:
247 do Run<Reply<String>>:
248 exec(commit)
249 return reply
250
251def handle.txn(~decide: String -> String -> Decision<String, String, String>, st: Store, +id: String, payload: String, at: U32) -> Run(Reply<String>):
252 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
253 do Run<Reply<String>>:
254 exec(begin)
255 found : Maybe<&2, Record<String, String>> <- seen.get(find, id)
256 s : String <- state.get(read)
257 written(step(~String, ~String, ~String, ~String, ~String.eq, ~decide, found, s, payload, at), commit, save, keep, emit, id)
258
259# On failure, resets the read statements and rolls back. Rollback errors are ignored: the transaction may not have started.
260def handle.end(r: Result<&1, &1, U32 & String, Reply<String>>, st: Store) -> Run(Reply<String>):
261 match r:
262 case Done{x}:
263 Run.pure(Reply<String>, x)
264 case Fail{e}:
265 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
266 do IO<Result<&1, &1, U32 & String, Reply<String>>>:
267 a : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(find)
268 b : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(read)
269 c : Result<&1, &1, U32 & String, Unit> <- exec(rollback)
270 return Fail{e}
271
272# Runs one command in one transaction: find its record, decide, and commit state, record, and effects.
273# at is the record's time for prune, in seconds. decide sees the committed state and the payload.
274def handle(~decide: String -> String -> Decision<String, String, String>, +st: Store, id: String, payload: String, at: U32) -> Run(Reply<String>):
275 do IO<Result<&1, &1, U32 & String, Reply<String>>>:
276 r : Result<&1, &1, U32 & String, Reply<String>> <- handle.txn(~decide, st, id, payload, at)
277 handle.end(r, st)
278
279# The committed domain state.
280def state(st: Store) -> Run(String):
281 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
282 state.get(read)
283
284def prune.run(+s: U32, before: U32) -> Run(Unit):
285 do Run<Unit>:
286 Sqlite.stmt.bind.int(s, 1, before)
287 exec(s)
288
289# Forgets every record made before the time before. A later retry of a forgotten id runs as a new command.
290def prune(st: Store, before: U32) -> Run(Unit):
291 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
292 once(Unit, db, "DELETE FROM commands_seen WHERE at < ?", +s => prune.run(s, before))
293
294# An undelivered effect: its sequence number, its command id, and the effect.
295type Pending is Data:
296 Pending{seq: U32, id: String, effect: String}
297
298def outbox.row(row: Bool, +s: U32) -> Run(Maybe<&2, Pending>):
299 match row:
300 case False{}:
301 Run.pure(Maybe<&2, Pending>, None{})
302 case True{}:
303 do Run<Maybe<&2, Pending>>:
304 seq : U32 <- Sqlite.stmt.column.int(s, 0)
305 id : String <- Sqlite.stmt.column.text(s, 1)
306 e : String <- Sqlite.stmt.column.text(s, 2)
307 return Some{Pending{seq, id, e}}
308
309def outbox.first(+s: U32) -> Run(Maybe<&2, Pending>):
310 do Run<Maybe<&2, Pending>>:
311 row : Bool <- Sqlite.stmt.step(s)
312 outbox.row(row, s)
313
314# The oldest undelivered effect.
315def outbox.next(st: Store) -> Run(Maybe<&2, Pending>):
316 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
317 once(Maybe<&2, Pending>, db, "SELECT seq, id, effect FROM commands_outbox ORDER BY seq LIMIT 1", +s => outbox.first(s))
318
319def outbox.ack.run(+s: U32, seq: U32) -> Run(Unit):
320 do Run<Unit>:
321 Sqlite.stmt.bind.int(s, 1, seq)
322 exec(s)
323
324# Removes a delivered effect. Deliver before ack: a crash between them delivers the effect again.
325def outbox.ack(st: Store, seq: U32) -> Run(Unit):
326 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
327 once(Unit, db, "DELETE FROM commands_outbox WHERE seq = ?", +s => outbox.ack.run(s, seq))