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
2	import Base
3	import bend-kit-bytes@0.3.3.0/bytes.bend as Bytes
4	import bend-kit-sqlite@0.1.0.1/sqlite.bend as Sqlite
5	
6	# ---- the pure processor
7	
8	type 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.
13	type Record<-P: Type, -R: Type> is Type:
14	  Record{payload: P, outcome: Outcome<R>, at: U32}
15	
16	type 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.
22	type 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.
27	type 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	
32	def 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	
36	def 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	
40	def 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	
47	def 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	
54	def 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.
63	def 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	
75	def Run(-A: Type) -> Type:
76	  IO(Result<&1, &1, U32 & String, A>)
77	
78	def Run.pure(-A: Type, x: A) -> Run(A):
79	  IO.pure(Result<&1, &1, U32 & String, A>, Done{x})
80	
81	def Run.fail(-A: Type, e: U32 & String) -> Run(A):
82	  IO.pure(Result<&1, &1, U32 & String, A>, Fail{e})
83	
84	def 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	
91	def 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.
95	def 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	
104	def 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.
114	def 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	
120	def 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.
127	def 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	
132	def sql(+db: U32, text: String) -> Run(Unit):
133	  once(Unit, db, text, +s => exec(s))
134	
135	def blob.of(lw: U32 & Array<U32>) -> Bytes.Bytes:
136	  (n, w) = lw
137	  Bytes.Bytes{n, w}
138	
139	def 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	
144	def 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	
148	def 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.
153	def 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.
159	type 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.
163	type 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).
168	def 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	
187	def 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	
201	def 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	
208	def 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.
221	def 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	
229	def 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.
239	def 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	
247	def 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.
254	def 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	
264	def 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	
273	def 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	
280	def 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	
284	def 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	
295	def 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.
308	def 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.
318	def 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	
339	def 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.
350	def 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	
362	def 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.
374	def 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.
388	def 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.
398	def 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	
402	def 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.
408	def 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.
413	type Pending is Type:
414	  Pending{seq: U32, id: String, effect: Bytes.Bytes}
415	
416	def 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	
427	def 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.
433	def 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	
437	def 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.
443	def 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))