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