commands.bend
bend-kit-commands@0.3.0.0 · 396 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.3.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
156# ---- the store
157
158# A database and its prepared statements. One Store serves one task at a time.
159type Store is Data:
160 Store{db: U32, begin: U32, commit: U32, rollback: U32, find: U32, read: U32, save: U32, keep: U32, emit: U32}
161
162def init.run(+s: U32, init: Bytes.Bytes) -> Run(Unit):
163 do Run<Unit>:
164 bind(s, 1, init)
165 exec(s)
166
167# Opens or creates the tables at path. A new database starts with domain state init.
168# wait is how many ms a write waits for another connection's transaction before it fails with SQLITE_BUSY (5).
169def open(path: String, init: Bytes.Bytes, wait: U32) -> Run(Store):
170 do Run<Store>:
171 +db : U32 <- Sqlite.db.open(path)
172 sql(db, "PRAGMA busy_timeout = " ++ U32.show(wait))
173 sql(db, "CREATE TABLE IF NOT EXISTS commands_state (k INTEGER PRIMARY KEY CHECK (k = 0), state BLOB NOT NULL)")
174 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)")
175 sql(db, "CREATE INDEX IF NOT EXISTS commands_seen_at ON commands_seen (at)")
176 sql(db, "CREATE TABLE IF NOT EXISTS commands_outbox (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT NOT NULL, effect BLOB NOT NULL)")
177 once(Unit, db, "INSERT OR IGNORE INTO commands_state (k, state) VALUES (0, ?)", +s => init.run(s, init))
178 begin : U32 <- Sqlite.stmt.prepare(db, "BEGIN IMMEDIATE")
179 commit : U32 <- Sqlite.stmt.prepare(db, "COMMIT")
180 rollback : U32 <- Sqlite.stmt.prepare(db, "ROLLBACK")
181 find : U32 <- Sqlite.stmt.prepare(db, "SELECT payload, kind, result, at FROM commands_seen WHERE id = ?")
182 read : U32 <- Sqlite.stmt.prepare(db, "SELECT state FROM commands_state WHERE k = 0")
183 save : U32 <- Sqlite.stmt.prepare(db, "UPDATE commands_state SET state = ? WHERE k = 0")
184 keep : U32 <- Sqlite.stmt.prepare(db, "INSERT INTO commands_seen (id, payload, kind, result, at) VALUES (?, ?, ?, ?, ?)")
185 emit : U32 <- Sqlite.stmt.prepare(db, "INSERT INTO commands_outbox (id, effect) VALUES (?, ?)")
186 return Store{db, begin, commit, rollback, find, read, save, keep, emit}
187
188def close(st: Store) -> Run(Unit):
189 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
190 do Run<Unit>:
191 Sqlite.stmt.finalize(begin)
192 Sqlite.stmt.finalize(commit)
193 Sqlite.stmt.finalize(rollback)
194 Sqlite.stmt.finalize(find)
195 Sqlite.stmt.finalize(read)
196 Sqlite.stmt.finalize(save)
197 Sqlite.stmt.finalize(keep)
198 Sqlite.stmt.finalize(emit)
199 Sqlite.db.close(db)
200
201def outcome(accepted: Bool, r: Bytes.Bytes) -> Outcome<Bytes.Bytes>:
202 match accepted:
203 case True{}:
204 Accepted{r}
205 case False{}:
206 Refused{r}
207
208def seen.row(row: Bool, +find: U32) -> Run(Maybe<Record<Bytes.Bytes, Bytes.Bytes>>):
209 match row:
210 case False{}:
211 Run.pure(Maybe<Record<Bytes.Bytes, Bytes.Bytes>>, None{})
212 case True{}:
213 do Run<Maybe<Record<Bytes.Bytes, Bytes.Bytes>>>:
214 p : Bytes.Bytes <- column(find, 0)
215 kind : U32 <- Sqlite.stmt.column.int(find, 1)
216 r : Bytes.Bytes <- column(find, 2)
217 at : U32 <- Sqlite.stmt.column.int(find, 3)
218 return Some{Record{p, outcome(U32.is_eq(kind, 1), r), at}}
219
220# The record for id, if the store holds one.
221def seen.get(+find: U32, id: String) -> Run(Maybe<Record<Bytes.Bytes, Bytes.Bytes>>):
222 do Run<Maybe<Record<Bytes.Bytes, Bytes.Bytes>>>:
223 Sqlite.stmt.bind.text(find, 1, id)
224 row : Bool <- Sqlite.stmt.step(find)
225 m : Maybe<Record<Bytes.Bytes, Bytes.Bytes>> <- seen.row(row, find)
226 Sqlite.stmt.reset(find)
227 return m
228
229def state.row(row: Bool, +read: U32) -> Run(Bytes.Bytes):
230 match row:
231 case True{}:
232 column(read, 0)
233 case False{}:
234 Run.fail(Bytes.Bytes, (1, "commands: the state row is missing"))
235
236def state.get(+read: U32) -> Run(Bytes.Bytes):
237 do Run<Bytes.Bytes>:
238 row : Bool <- Sqlite.stmt.step(read)
239 s : Bytes.Bytes <- state.row(row, read)
240 Sqlite.stmt.reset(read)
241 return s
242
243def seen.put.row(+keep: U32, id: String, p: Bytes.Bytes, kind: U32, r: Bytes.Bytes, at: U32) -> Run(Unit):
244 do Run<Unit>:
245 Sqlite.stmt.bind.text(keep, 1, id)
246 bind(keep, 2, p)
247 Sqlite.stmt.bind.int(keep, 3, kind)
248 bind(keep, 4, r)
249 Sqlite.stmt.bind.int(keep, 5, at)
250 exec(keep)
251
252def seen.put.of(o: Outcome<Bytes.Bytes>, +keep: U32, id: String, p: Bytes.Bytes, at: U32) -> Run(Unit):
253 match o:
254 case Accepted{r}:
255 seen.put.row(keep, id, p, 1, r, at)
256 case Refused{r}:
257 seen.put.row(keep, id, p, 2, r, at)
258
259def seen.put(rec: Record<Bytes.Bytes, Bytes.Bytes>, +keep: U32, id: String) -> Run(Unit):
260 Record{p, o, at} = rec
261 seen.put.of(o, keep, id, p, at)
262
263def outbox.put(es: List<Bytes.Bytes>, +emit: U32, +id: String) -> Run(Unit):
264 match es:
265 case Nil{}:
266 Run.pure(Unit, Unit{})
267 case Con{e, rest}:
268 do Run<Unit>:
269 Sqlite.stmt.bind.text(emit, 1, id)
270 bind(emit, 2, e)
271 exec(emit)
272 outbox.put(rest, emit, id)
273
274def state.put(+save: U32, s: Bytes.Bytes) -> Run(Unit):
275 do Run<Unit>:
276 bind(save, 1, s)
277 exec(save)
278
279# Applies a step inside the open transaction, then commits it.
280def written(
281 w: Step<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>, +commit: U32, +save: U32, +keep: U32, +emit: U32, +id: String
282) -> Run(Reply<Bytes.Bytes>):
283 match w:
284 case Write{s, rec, reply, es}:
285 do Run<Reply<Bytes.Bytes>>:
286 state.put(save, s)
287 seen.put(rec, keep, id)
288 outbox.put(es, emit, id)
289 exec(commit)
290 return reply
291 case Note{rec, reply}:
292 do Run<Reply<Bytes.Bytes>>:
293 seen.put(rec, keep, id)
294 exec(commit)
295 return reply
296 case Keep{reply}:
297 do Run<Reply<Bytes.Bytes>>:
298 exec(commit)
299 return reply
300
301# A duplicate needs no state: step ignores s when found is Some, so it gets an empty one.
302def handle.found(
303 ~decide: Bytes.Bytes -> Bytes.Bytes -> Decision<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>,
304 found: Maybe<Record<Bytes.Bytes, Bytes.Bytes>>, +read: U32, payload: Bytes.Bytes, at: U32,
305 +commit: U32, +save: U32, +keep: U32, +emit: U32, +id: String
306) -> Run(Reply<Bytes.Bytes>):
307 match found:
308 case None{}:
309 do Run<Reply<Bytes.Bytes>>:
310 s : Bytes.Bytes <- state.get(read)
311 written(step(~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~same, ~Bytes.clone, ~Bytes.clone, ~decide, None{}, s, payload, at), commit, save, keep, emit, id)
312 case Some{rec}:
313 written(step(~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~same, ~Bytes.clone, ~Bytes.clone, ~decide, Some{rec}, Bytes.new(0), payload, at), commit, save, keep, emit, id)
314
315def handle.txn(
316 ~decide: Bytes.Bytes -> Bytes.Bytes -> Decision<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>,
317 st: Store, +id: String, payload: Bytes.Bytes, at: U32
318) -> Run(Reply<Bytes.Bytes>):
319 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
320 do Run<Reply<Bytes.Bytes>>:
321 exec(begin)
322 found : Maybe<Record<Bytes.Bytes, Bytes.Bytes>> <- seen.get(find, id)
323 handle.found(~decide, found, read, payload, at, commit, save, keep, emit, id)
324
325# On failure, resets the read statements and rolls back. Rollback errors are ignored: the transaction may not have started.
326def handle.end(r: Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>, st: Store) -> Run(Reply<Bytes.Bytes>):
327 match r:
328 case Done{x}:
329 Run.pure(Reply<Bytes.Bytes>, x)
330 case Fail{e}:
331 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
332 do IO<Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>>:
333 a : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(find)
334 b : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(read)
335 c : Result<&1, &1, U32 & String, Unit> <- exec(rollback)
336 return Fail{e}
337
338# Runs one command in one transaction: find its record, decide, and commit state, record, and effects.
339# at is the record's time for prune, in seconds. decide sees the committed state and the payload.
340def handle(
341 ~decide: Bytes.Bytes -> Bytes.Bytes -> Decision<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>,
342 +st: Store, id: String, payload: Bytes.Bytes, at: U32
343) -> Run(Reply<Bytes.Bytes>):
344 do IO<Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>>:
345 r : Result<&1, &1, U32 & String, Reply<Bytes.Bytes>> <- handle.txn(~decide, st, id, payload, at)
346 handle.end(r, st)
347
348# The committed domain state.
349def state(st: Store) -> Run(Bytes.Bytes):
350 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
351 state.get(read)
352
353def prune.run(+s: U32, before: U32) -> Run(Unit):
354 do Run<Unit>:
355 Sqlite.stmt.bind.int(s, 1, before)
356 exec(s)
357
358# Forgets every record made before the time before. A later retry of a forgotten id runs as a new command.
359def prune(st: Store, before: U32) -> Run(Unit):
360 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
361 once(Unit, db, "DELETE FROM commands_seen WHERE at < ?", +s => prune.run(s, before))
362
363# An undelivered effect: its sequence number, its command id, and the effect.
364type Pending is Type:
365 Pending{seq: U32, id: String, effect: Bytes.Bytes}
366
367def outbox.row(row: Bool, +s: U32) -> Run(Maybe<Pending>):
368 match row:
369 case False{}:
370 Run.pure(Maybe<Pending>, None{})
371 case True{}:
372 do Run<Maybe<Pending>>:
373 seq : U32 <- Sqlite.stmt.column.int(s, 0)
374 id : String <- Sqlite.stmt.column.text(s, 1)
375 e : Bytes.Bytes <- column(s, 2)
376 return Some{Pending{seq, id, e}}
377
378def outbox.first(+s: U32) -> Run(Maybe<Pending>):
379 do Run<Maybe<Pending>>:
380 row : Bool <- Sqlite.stmt.step(s)
381 outbox.row(row, s)
382
383# The oldest undelivered effect.
384def outbox.next(st: Store) -> Run(Maybe<Pending>):
385 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
386 once(Maybe<Pending>, db, "SELECT seq, id, effect FROM commands_outbox ORDER BY seq LIMIT 1", +s => outbox.first(s))
387
388def outbox.ack.run(+s: U32, seq: U32) -> Run(Unit):
389 do Run<Unit>:
390 Sqlite.stmt.bind.int(s, 1, seq)
391 exec(s)
392
393# Removes a delivered effect. Deliver before ack: a crash between them delivers the effect again.
394def outbox.ack(st: Store, seq: U32) -> Run(Unit):
395 Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
396 once(Unit, db, "DELETE FROM commands_outbox WHERE seq = ?", +s => outbox.ack.run(s, seq))