commands.bend

bend-kit-commands@0.4.0.0 · 445 lines · 59 definitions

1# Retry-safe commands: deduplicate by command id, replay recorded results, and commit keyed rows, 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 row of the store's key-value table. Read: Some is the value, None means no row. Written: Some puts the value, None deletes the row.
159type Row is Type:
160  Row{key: Bytes.Bytes, value: Maybe<Bytes.Bytes>}
161
162# A database and its prepared statements. One Store serves one task at a time.
163type Store is Data:
164  Store{db: U32, begin: U32, commit: U32, rollback: U32, find: U32, read: U32, save: U32, del: U32, keep: U32, emit: U32}
165
166# Opens or creates the tables at path.
167# wait is how many ms a write waits for another connection's transaction before it fails with SQLITE_BUSY (5).
168def open(path: String, wait: U32) -> Run(Store):
169  do Run<Store>:
170    +db : U32 <- Sqlite.db.open(path)
171    sql(db, "PRAGMA busy_timeout = " ++ U32.show(wait))
172    sql(db, "CREATE TABLE IF NOT EXISTS commands_rows (k BLOB PRIMARY KEY, v BLOB NOT NULL) WITHOUT ROWID")
173    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)")
174    sql(db, "CREATE INDEX IF NOT EXISTS commands_seen_at ON commands_seen (at)")
175    sql(db, "CREATE TABLE IF NOT EXISTS commands_outbox (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT NOT NULL, effect BLOB NOT NULL)")
176    begin : U32 <- Sqlite.stmt.prepare(db, "BEGIN IMMEDIATE")
177    commit : U32 <- Sqlite.stmt.prepare(db, "COMMIT")
178    rollback : U32 <- Sqlite.stmt.prepare(db, "ROLLBACK")
179    find : U32 <- Sqlite.stmt.prepare(db, "SELECT payload, kind, result, at FROM commands_seen WHERE id = ?")
180    read : U32 <- Sqlite.stmt.prepare(db, "SELECT v FROM commands_rows WHERE k = ?")
181    save : U32 <- Sqlite.stmt.prepare(db, "INSERT OR REPLACE INTO commands_rows (k, v) VALUES (?, ?)")
182    del : U32 <- Sqlite.stmt.prepare(db, "DELETE FROM commands_rows WHERE k = ?")
183    keep : U32 <- Sqlite.stmt.prepare(db, "INSERT INTO commands_seen (id, payload, kind, result, at) VALUES (?, ?, ?, ?, ?)")
184    emit : U32 <- Sqlite.stmt.prepare(db, "INSERT INTO commands_outbox (id, effect) VALUES (?, ?)")
185    return Store{db, begin, commit, rollback, find, read, save, del, keep, emit}
186
187def close(st: Store) -> Run(Unit):
188  Store{db, begin, commit, rollback, find, read, save, del, keep, emit} = st
189  do Run<Unit>:
190    Sqlite.stmt.finalize(begin)
191    Sqlite.stmt.finalize(commit)
192    Sqlite.stmt.finalize(rollback)
193    Sqlite.stmt.finalize(find)
194    Sqlite.stmt.finalize(read)
195    Sqlite.stmt.finalize(save)
196    Sqlite.stmt.finalize(del)
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 row.of(row: Bool, +read: U32) -> Run(Maybe<Bytes.Bytes>):
230  match row:
231    case False{}:
232      Run.pure(Maybe<Bytes.Bytes>, None{})
233    case True{}:
234      do Run<Maybe<Bytes.Bytes>>:
235        v : Bytes.Bytes <- column(read, 0)
236        return Some{v}
237
238# The value at key k, if a row holds it.
239def row.get(+read: U32, k: Bytes.Bytes) -> Run(Maybe<Bytes.Bytes>):
240  do Run<Maybe<Bytes.Bytes>>:
241    bind(read, 1, k)
242    row : Bool <- Sqlite.stmt.step(read)
243    m : Maybe<Bytes.Bytes> <- row.of(row, read)
244    Sqlite.stmt.reset(read)
245    return m
246
247def row.at(+read: U32, kk: Bytes.Bytes & Bytes.Bytes) -> Run(Row):
248  (k, k2) = kk
249  do Run<Row>:
250    v : Maybe<Bytes.Bytes> <- row.get(read, k)
251    return Row{k2, v}
252
253# One row for each key, in order.
254def rows.get(ks: List<Bytes.Bytes>, +read: U32) -> Run(List<Row>):
255  match ks:
256    case Nil{}:
257      Run.pure(List<Row>, Nil{})
258    case Con{k, rest}:
259      do Run<List<Row>>:
260        r : Row <- row.at(read, Bytes.clone(k))
261        rs : List<Row> <- rows.get(rest, read)
262        return Con{r, rs}
263
264def seen.put.row(+keep: U32, id: String, p: Bytes.Bytes, kind: U32, r: Bytes.Bytes, at: U32) -> Run(Unit):
265  do Run<Unit>:
266    Sqlite.stmt.bind.text(keep, 1, id)
267    bind(keep, 2, p)
268    Sqlite.stmt.bind.int(keep, 3, kind)
269    bind(keep, 4, r)
270    Sqlite.stmt.bind.int(keep, 5, at)
271    exec(keep)
272
273def seen.put.of(o: Outcome<Bytes.Bytes>, +keep: U32, id: String, p: Bytes.Bytes, at: U32) -> Run(Unit):
274  match o:
275    case Accepted{r}:
276      seen.put.row(keep, id, p, 1, r, at)
277    case Refused{r}:
278      seen.put.row(keep, id, p, 2, r, at)
279
280def seen.put(rec: Record<Bytes.Bytes, Bytes.Bytes>, +keep: U32, id: String) -> Run(Unit):
281  Record{p, o, at} = rec
282  seen.put.of(o, keep, id, p, at)
283
284def outbox.put(es: List<Bytes.Bytes>, +emit: U32, +id: String) -> Run(Unit):
285  match es:
286    case Nil{}:
287      Run.pure(Unit, Unit{})
288    case Con{e, rest}:
289      do Run<Unit>:
290        Sqlite.stmt.bind.text(emit, 1, id)
291        bind(emit, 2, e)
292        exec(emit)
293        outbox.put(rest, emit, id)
294
295def row.put(v: Maybe<Bytes.Bytes>, +save: U32, +del: U32, k: Bytes.Bytes) -> Run(Unit):
296  match v:
297    case Some{x}:
298      do Run<Unit>:
299        bind(save, 1, k)
300        bind(save, 2, x)
301        exec(save)
302    case None{}:
303      do Run<Unit>:
304        bind(del, 1, k)
305        exec(del)
306
307# Writes the rows in order, so a later row for a key wins.
308def rows.put(rs: List<Row>, +save: U32, +del: U32) -> Run(Unit):
309  match rs:
310    case Nil{}:
311      Run.pure(Unit, Unit{})
312    case Con{Row{k, v}, rest}:
313      do Run<Unit>:
314        row.put(v, save, del, k)
315        rows.put(rest, save, del)
316
317# Applies a step inside the open transaction, then commits it.
318def written(
319  w: Step<List<Row>, Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>, +commit: U32, +save: U32, +del: U32, +keep: U32, +emit: U32, +id: String
320) -> Run(Reply<Bytes.Bytes>):
321  match w:
322    case Write{rs, rec, reply, es}:
323      do Run<Reply<Bytes.Bytes>>:
324        rows.put(rs, save, del)
325        seen.put(rec, keep, id)
326        outbox.put(es, emit, id)
327        exec(commit)
328        return reply
329    case Note{rec, reply}:
330      do Run<Reply<Bytes.Bytes>>:
331        seen.put(rec, keep, id)
332        exec(commit)
333        return reply
334    case Keep{reply}:
335      do Run<Reply<Bytes.Bytes>>:
336        exec(commit)
337        return reply
338
339def handle.new(
340  ~keys: Bytes.Bytes -> List<Bytes.Bytes>,
341  ~decide: List<Row> -> Bytes.Bytes -> Decision<List<Row>, Bytes.Bytes, Bytes.Bytes>,
342  pp: Bytes.Bytes & Bytes.Bytes, +read: U32, at: U32, +commit: U32, +save: U32, +del: U32, +keep: U32, +emit: U32, +id: String
343) -> Run(Reply<Bytes.Bytes>):
344  (p, p2) = pp
345  do Run<Reply<Bytes.Bytes>>:
346    rs : List<Row> <- rows.get(keys(p), read)
347    written(step(~List<Row>, ~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~same, ~Bytes.clone, ~Bytes.clone, ~decide, None{}, rs, p2, at), commit, save, del, keep, emit, id)
348
349# A duplicate reads no rows: step ignores them when found is Some.
350def handle.found(
351  ~keys: Bytes.Bytes -> List<Bytes.Bytes>,
352  ~decide: List<Row> -> Bytes.Bytes -> Decision<List<Row>, Bytes.Bytes, Bytes.Bytes>,
353  found: Maybe<Record<Bytes.Bytes, Bytes.Bytes>>, +read: U32, payload: Bytes.Bytes, at: U32,
354  +commit: U32, +save: U32, +del: U32, +keep: U32, +emit: U32, +id: String
355) -> Run(Reply<Bytes.Bytes>):
356  match found:
357    case None{}:
358      handle.new(~keys, ~decide, Bytes.clone(payload), read, at, commit, save, del, keep, emit, id)
359    case Some{rec}:
360      written(step(~List<Row>, ~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~same, ~Bytes.clone, ~Bytes.clone, ~decide, Some{rec}, Nil{}, payload, at), commit, save, del, keep, emit, id)
361
362def handle.txn(
363  ~keys: Bytes.Bytes -> List<Bytes.Bytes>,
364  ~decide: List<Row> -> Bytes.Bytes -> Decision<List<Row>, Bytes.Bytes, Bytes.Bytes>,
365  st: Store, +id: String, payload: Bytes.Bytes, at: U32
366) -> Run(Reply<Bytes.Bytes>):
367  Store{db, begin, commit, rollback, find, read, save, del, keep, emit} = st
368  do Run<Reply<Bytes.Bytes>>:
369    exec(begin)
370    found : Maybe<Record<Bytes.Bytes, Bytes.Bytes>> <- seen.get(find, id)
371    handle.found(~keys, ~decide, found, read, payload, at, commit, save, del, keep, emit, id)
372
373# On failure, resets the read statements and rolls back. Rollback errors are ignored: the transaction may not have started.
374def handle.end(r: Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>, st: Store) -> Run(Reply<Bytes.Bytes>):
375  match r:
376    case Done{x}:
377      Run.pure(Reply<Bytes.Bytes>, x)
378    case Fail{e}:
379      Store{db, begin, commit, rollback, find, read, save, del, keep, emit} = st
380      do IO<Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>>:
381        a : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(find)
382        b : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(read)
383        c : Result<&1, &1, U32 & String, Unit> <- exec(rollback)
384        return Fail{e}
385
386# Runs one command in one transaction: find its record, read the rows that keys names, decide, and commit rows, record, and effects.
387# at is the record's time for prune, in seconds. decide sees the committed rows, one for each key in order, and the payload.
388def handle(
389  ~keys: Bytes.Bytes -> List<Bytes.Bytes>,
390  ~decide: List<Row> -> Bytes.Bytes -> Decision<List<Row>, Bytes.Bytes, Bytes.Bytes>,
391  +st: Store, id: String, payload: Bytes.Bytes, at: U32
392) -> Run(Reply<Bytes.Bytes>):
393  do IO<Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>>:
394    r : Result<&1, &1, U32 & String, Reply<Bytes.Bytes>> <- handle.txn(~keys, ~decide, st, id, payload, at)
395    handle.end(r, st)
396
397# The committed value at key k.
398def get(st: Store, k: Bytes.Bytes) -> Run(Maybe<Bytes.Bytes>):
399  Store{db, begin, commit, rollback, find, read, save, del, keep, emit} = st
400  row.get(read, k)
401
402def prune.run(+s: U32, before: U32) -> Run(Unit):
403  do Run<Unit>:
404    Sqlite.stmt.bind.int(s, 1, before)
405    exec(s)
406
407# Forgets every record made before the time before. A later retry of a forgotten id runs as a new command.
408def prune(st: Store, before: U32) -> Run(Unit):
409  Store{db, begin, commit, rollback, find, read, save, del, keep, emit} = st
410  once(Unit, db, "DELETE FROM commands_seen WHERE at < ?", +s => prune.run(s, before))
411
412# An undelivered effect: its sequence number, its command id, and the effect.
413type Pending is Type:
414  Pending{seq: U32, id: String, effect: Bytes.Bytes}
415
416def outbox.row(row: Bool, +s: U32) -> Run(Maybe<Pending>):
417  match row:
418    case False{}:
419      Run.pure(Maybe<Pending>, None{})
420    case True{}:
421      do Run<Maybe<Pending>>:
422        seq : U32 <- Sqlite.stmt.column.int(s, 0)
423        id : String <- Sqlite.stmt.column.text(s, 1)
424        e : Bytes.Bytes <- column(s, 2)
425        return Some{Pending{seq, id, e}}
426
427def outbox.first(+s: U32) -> Run(Maybe<Pending>):
428  do Run<Maybe<Pending>>:
429    row : Bool <- Sqlite.stmt.step(s)
430    outbox.row(row, s)
431
432# The oldest undelivered effect.
433def outbox.next(st: Store) -> Run(Maybe<Pending>):
434  Store{db, begin, commit, rollback, find, read, save, del, keep, emit} = st
435  once(Maybe<Pending>, db, "SELECT seq, id, effect FROM commands_outbox ORDER BY seq LIMIT 1", +s => outbox.first(s))
436
437def outbox.ack.run(+s: U32, seq: U32) -> Run(Unit):
438  do Run<Unit>:
439    Sqlite.stmt.bind.int(s, 1, seq)
440    exec(s)
441
442# Removes a delivered effect. Deliver before ack: a crash between them delivers the effect again.
443def outbox.ack(st: Store, seq: U32) -> Run(Unit):
444  Store{db, begin, commit, rollback, find, read, save, del, keep, emit} = st
445  once(Unit, db, "DELETE FROM commands_outbox WHERE seq = ?", +s => outbox.ack.run(s, seq))