gwk-kernel 0.0.2

The GridWork kernel: the PostgreSQL backend behind the gwk contract
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
//! Projection snapshots: the recovery shortcut, never the truth.
//!
//! A checkpoint is the projection tables, canonicalized, hashed, and stored as
//! one encrypted blob at one `global_sequence`. It is EVIDENCE, not a restore
//! point: this schema's `born_initial` and `no_truncate` guards mean projection
//! rows can only ever be produced by replaying the log, so what a checkpoint
//! buys is the ability to check that result, not to skip it (see
//! [`crate::recover`]). The worst a missing or corrupt one costs is a
//! comparison — which is why an append is allowed to take one and none of the
//! contract depends on it existing.
//!
//! Two things make the hash mean anything.
//!
//! **Records are re-serialized through the CONTRACT type, never passed
//! through.** `to_jsonb` returns keys in whatever order the row's physical
//! layout gives, and that layout is a property of one database's history — add
//! a column and it moves. Deserializing each row into its [`ProjectionRecord`]
//! and serializing that back means the bytes are determined by the type
//! declaration instead, so two kernels serving the same log agree, and so does
//! the same kernel after a `VACUUM FULL`.
//!
//! **The visit order is a written-down constant, not a catalog query.** The
//! hash depends on which table comes first, so that has to be something a
//! reader can see and a diff can show changing.
//!
//! The records blob's plaintext IS the canonical bytes, so `projection_hash`
//! and the blob's own content address are the same digest. That is stated as
//! an invariant and asserted, not left as a coincidence for someone to
//! discover while debugging a restore.

use gwk_domain::blob::BlobAddress;
use gwk_domain::checkpoint::{CHECKPOINT_SCHEMA_VERSION, Checkpoint};
use gwk_domain::envelope::PayloadRef;
use gwk_domain::ids::{ByteCount, Seq, Timestamp};
use gwk_domain::port::BlobStore;
use gwk_domain::protocol::{ProjectionKind, ProjectionRecord};
use sha2::{Digest, Sha256};
use sqlx::{PgConnection, Row};

use crate::blob::container;
use crate::blob::store::PgBlobStore;
use crate::numeric::{from_numeric_text, to_numeric_text};
use crate::project::Refusal;

/// What the records blob is: one canonical record per line.
///
/// Line-delimited rather than one JSON array, because recovery streams it back
/// a chunk at a time and a line is a frame it can complete without holding the
/// whole document.
pub const RECORDS_MEDIA_TYPE: &str = "application/x-ndjson";

/// Every projection table, in the order a snapshot visits them, each shaped
/// into the exact wire form of a [`ProjectionRecord`].
///
/// Alphabetical, and spelled out one query at a time. Two kinds of adjustment
/// appear, and both are deliberate rather than convenient:
///
/// * **A `::text` cast on four columns.** `to_jsonb` renders a `numeric` as a
///   JSON NUMBER, while the contract carries 64-bit counters as decimal
///   STRINGS. The difference is invisible until the value passes 2^53, at which
///   point the number silently comes back as a different one — so the cast goes
///   on every such column, not on the ones that have gotten large so far.
///
/// * **Two renamed columns and one subtracted one.** `gwk.receipt` stores
///   `from`/`to` as `from_state`/`to_state` because the bare words are SQL
///   reserved words, and `gwk.orchestrator_checkpoint.updated_at` is row
///   bookkeeping the contract type does not carry — a reader orders those by
///   the checkpoint's own `seq`. Each is handled BY NAME rather than by a
///   general tolerance, so a column added to any other table still fails the
///   round trip. That failure IS the parity check between the DDL and
///   `gwk-domain`, and nothing else in this kernel performs it.
///
/// The `derived` flag is the third adjustment, and the one with teeth. Two
/// tables hold rows the log cannot reproduce:
///
/// * `gwk.receipt` — `submit` writes the authority receipt itself, and on the
///   PAGED path it does so for a command that is REFUSED and therefore appends
///   no event at all ("the one refusal in the kernel that leaves rows behind").
/// * `gwk.attention_item` — the same paged path raises its item directly
///   through `page_attention`, again with no event behind it.
///
/// A hash over those tables could never be reproduced by a replay, so a
/// checkpoint carrying them would fail every scratch rebuild forever and the
/// failure would say nothing. They stay in the canonical dump — that is where
/// the DDL-to-contract parity check happens, and it must cover all fifteen —
/// and stay OUT of the digest. What guards them instead is what always did:
/// `receipt_append_only` and the delete guards, which no privilege can bypass.
struct Projection {
    tag: &'static str,
    /// The column `read` orders and pages by, which is also the field a client
    /// reads a continuation cursor out of. Written down rather than inferred:
    /// fourteen projections key on `id` and `orchestrator_checkpoint` does not,
    /// and a rule that guessed would hand back the wrong cursor for the one
    /// table that differs — a page that silently restarts, not one that fails.
    key: &'static str,
    query: &'static str,
    /// The same record, one page at a time: `$1` a cursor (exclusive), `$2` an
    /// exact key, `$3` the row ceiling. One query serves both reads because
    /// get-by-id IS a one-row page — a second near-identical string per table
    /// would be one more place for the record shape to drift from `query`.
    ///
    /// Ordering and the cursor comparison are both `COLLATE "C"`. Under a
    /// locale collation — which is what the stock PostgreSQL image gives you —
    /// ids differing only in punctuation can TIE, and two ties either side of a
    /// page boundary drop a row or repeat one. Byte order has no ties.
    read: &'static str,
    /// Whether replaying the log through `apply_event` rebuilds this table.
    derived: bool,
}

const fn derived(
    tag: &'static str,
    key: &'static str,
    query: &'static str,
    read: &'static str,
) -> Projection {
    Projection {
        tag,
        key,
        query,
        read,
        derived: true,
    }
}

const fn written_beside_the_log(
    tag: &'static str,
    key: &'static str,
    query: &'static str,
    read: &'static str,
) -> Projection {
    Projection {
        tag,
        key,
        query,
        read,
        derived: false,
    }
}

/// The paged read for one projection and the column it pages by, or `None` if
/// the kind has no table — which cannot happen while
/// [`every_projection_is_visited_exactly_once_in_a_written_down_order`] passes,
/// and is returned rather than panicked because this is reached from a client
/// request.
///
/// The two travel together on purpose: a caller that had the query but chose
/// the cursor field for itself would be free to choose a different one.
pub fn read_query(kind: ProjectionKind) -> Option<(&'static str, &'static str)> {
    PROJECTIONS
        .iter()
        .find(|p| p.tag == kind.as_str())
        .map(|p| (p.read, p.key))
}

const PROJECTIONS: &[Projection] = &[
    derived(
        "attempt",
        "id",
        "SELECT jsonb_build_object('projection_type', 'attempt', 'attempt', to_jsonb(t))::text \
         FROM gwk.attempt t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'attempt', 'attempt', to_jsonb(t))::text \
         FROM gwk.attempt t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    written_beside_the_log(
        "attention_item",
        "id",
        "SELECT jsonb_build_object('projection_type', 'attention_item', 'attention_item', \
           to_jsonb(t))::text FROM gwk.attention_item t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'attention_item', 'attention_item', \
           to_jsonb(t))::text FROM gwk.attention_item t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "authority_grant",
        "id",
        "SELECT jsonb_build_object('projection_type', 'authority_grant', 'authority_grant', \
           to_jsonb(t))::text FROM gwk.authority_grant t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'authority_grant', 'authority_grant', \
           to_jsonb(t))::text FROM gwk.authority_grant t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "command",
        "id",
        "SELECT jsonb_build_object('projection_type', 'command', 'command', to_jsonb(t))::text \
         FROM gwk.command t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'command', 'command', to_jsonb(t))::text \
         FROM gwk.command t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "dispatch_node",
        "id",
        "SELECT jsonb_build_object('projection_type', 'dispatch_node', 'dispatch_node', \
           to_jsonb(t))::text FROM gwk.dispatch_node t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'dispatch_node', 'dispatch_node', \
           to_jsonb(t))::text FROM gwk.dispatch_node t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "engine_session",
        "id",
        "SELECT jsonb_build_object('projection_type', 'engine_session', 'engine_session', \
           to_jsonb(t))::text FROM gwk.engine_session t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'engine_session', 'engine_session', \
           to_jsonb(t))::text FROM gwk.engine_session t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "evidence",
        "id",
        "SELECT jsonb_build_object('projection_type', 'evidence', 'evidence', \
           to_jsonb(t) || jsonb_build_object('byte_size', t.byte_size::text))::text \
         FROM gwk.evidence t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'evidence', 'evidence', \
           to_jsonb(t) || jsonb_build_object('byte_size', t.byte_size::text))::text \
         FROM gwk.evidence t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "gate",
        "id",
        "SELECT jsonb_build_object('projection_type', 'gate', 'gate', to_jsonb(t))::text \
         FROM gwk.gate t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'gate', 'gate', to_jsonb(t))::text \
         FROM gwk.gate t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "ingested_record",
        "id",
        "SELECT jsonb_build_object('projection_type', 'ingested_record', 'ingested_record', \
           to_jsonb(t) || jsonb_build_object('event_seq', t.event_seq::text))::text \
         FROM gwk.ingested_record t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'ingested_record', 'ingested_record', \
           to_jsonb(t) || jsonb_build_object('event_seq', t.event_seq::text))::text \
         FROM gwk.ingested_record t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "lease",
        "id",
        "SELECT jsonb_build_object('projection_type', 'lease', 'lease', \
           to_jsonb(t) || jsonb_build_object('fence_token', t.fence_token::text))::text \
         FROM gwk.lease t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'lease', 'lease', \
           to_jsonb(t) || jsonb_build_object('fence_token', t.fence_token::text))::text \
         FROM gwk.lease t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "message",
        "id",
        "SELECT jsonb_build_object('projection_type', 'message', 'message', to_jsonb(t))::text \
         FROM gwk.message t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'message', 'message', to_jsonb(t))::text \
         FROM gwk.message t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "orchestrator_checkpoint",
        "orchestrator_id",
        "SELECT jsonb_build_object('projection_type', 'orchestrator_checkpoint', \
           'orchestrator_checkpoint', \
           (to_jsonb(t) - 'updated_at') || jsonb_build_object('seq', t.seq::text))::text \
         FROM gwk.orchestrator_checkpoint t ORDER BY t.orchestrator_id",
        // Keyed on `orchestrator_id`: this is the one projection whose primary
        // key is not called `id`, and paging it by a column it does not have
        // would fail at the database rather than quietly.
        "SELECT jsonb_build_object('projection_type', 'orchestrator_checkpoint', \
           'orchestrator_checkpoint', \
           (to_jsonb(t) - 'updated_at') || jsonb_build_object('seq', t.seq::text))::text \
         FROM gwk.orchestrator_checkpoint t \
         WHERE ($1::text IS NULL OR t.orchestrator_id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.orchestrator_id = $2) \
         ORDER BY t.orchestrator_id COLLATE \"C\" LIMIT $3",
    ),
    written_beside_the_log(
        "receipt",
        "id",
        "SELECT jsonb_build_object('projection_type', 'receipt', 'receipt', \
           (to_jsonb(t) - 'from_state' - 'to_state') \
           || jsonb_strip_nulls(jsonb_build_object('from', t.from_state, 'to', t.to_state)))::text \
         FROM gwk.receipt t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'receipt', 'receipt', \
           (to_jsonb(t) - 'from_state' - 'to_state') \
           || jsonb_strip_nulls(jsonb_build_object('from', t.from_state, 'to', t.to_state)))::text \
         FROM gwk.receipt t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "task",
        "id",
        "SELECT jsonb_build_object('projection_type', 'task', 'task', to_jsonb(t))::text \
         FROM gwk.task t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'task', 'task', to_jsonb(t))::text \
         FROM gwk.task t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
    derived(
        "worktree",
        "id",
        "SELECT jsonb_build_object('projection_type', 'worktree', 'worktree', to_jsonb(t))::text \
         FROM gwk.worktree t ORDER BY t.id",
        "SELECT jsonb_build_object('projection_type', 'worktree', 'worktree', to_jsonb(t))::text \
         FROM gwk.worktree t \
         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
           AND ($2::text IS NULL OR t.id = $2) \
         ORDER BY t.id COLLATE \"C\" LIMIT $3",
    ),
];

/// The canonical bytes of every projection row: one record per line, each one
/// having made the round trip through its contract type.
///
/// Reading through the caller's connection is deliberate — inside an append
/// transaction it sees that transaction's own uncommitted projections, which is
/// exactly the state the checkpoint claims to describe.
pub async fn canonical_records(conn: &mut PgConnection) -> Result<Vec<u8>, Refusal> {
    records(conn, false).await
}

/// The subset a replay can rebuild — what a checkpoint actually hashes.
///
/// The two tables left out are written beside the log rather than from it (see
/// [`Projection`]), so including them would produce a digest no rebuild could
/// ever match. Everything downstream — the checkpoint, the readiness compare,
/// the scratch rebuild — uses THIS one, and the difference between the two
/// functions is the whole reason the invariant holds.
pub async fn derived_records(conn: &mut PgConnection) -> Result<Vec<u8>, Refusal> {
    records(conn, true).await
}

async fn records(conn: &mut PgConnection, derived_only: bool) -> Result<Vec<u8>, Refusal> {
    let mut out = Vec::new();
    for projection in PROJECTIONS {
        if derived_only && !projection.derived {
            continue;
        }
        // Read through the struct field, which is a `&'static str`: sqlx 0.9
        // accepts a literal-lifetime query and refuses anything else.
        let rows = sqlx::query(projection.query)
            .fetch_all(&mut *conn)
            .await
            .map_err(|e| Refusal::storage(format!("read {} projections: {e}", projection.tag)))?;
        for row in &rows {
            let raw: String = row
                .try_get(0)
                .map_err(|e| Refusal::storage(format!("projection row: {e}")))?;
            // `deny_unknown_fields` on every entity makes this the parity check
            // between the DDL and the contract types: a column with no field
            // fails here rather than silently dropping out of the hash.
            let record: ProjectionRecord = serde_json::from_str(&raw).map_err(|e| {
                Refusal::storage(format!(
                    "projection row does not match the contract type: {e}"
                ))
            })?;
            serde_json::to_writer(&mut out, &record)
                .map_err(|e| Refusal::storage(format!("serialize projection record: {e}")))?;
            out.push(b'\n');
        }
    }
    Ok(out)
}

/// The digest the checkpoint records, over exactly the bytes it stores.
pub fn projection_hash(records: &[u8]) -> String {
    let digest: [u8; 32] = Sha256::digest(records).into();
    container::hex_lower(&digest)
}

/// Snapshot the projections as `conn`'s transaction will leave them, store the
/// records, and record the checkpoint.
///
/// The blob is committed through the blob store's OWN connections, so it lands
/// before the caller's transaction does. If that transaction then rolls back,
/// what is left is a blob nothing references — which is precisely what sweep
/// reclaims, and why sweep has to consider checkpoints as well as events.
///
/// The reverse order is what has no recovery: a checkpoint row committed beside
/// a blob whose write was rolled back is a checkpoint that fails validation
/// forever, and the fallback ladder would walk past it on every single startup.
pub async fn snapshot(
    conn: &mut PgConnection,
    blobs: &PgBlobStore,
    through: Seq,
    created_at: &Timestamp,
) -> Result<Checkpoint, Refusal> {
    let records = derived_records(conn).await?;
    let hash = projection_hash(&records);
    let address =
        BlobAddress::from_digest(&hash).map_err(|e| Refusal::storage(format!("hash: {e}")))?;

    let byte_size = ByteCount::new(records.len() as u64);
    let store_blob = async {
        let upload = blobs
            .begin(RECORDS_MEDIA_TYPE.to_owned(), byte_size)
            .await?;
        for (sequence, chunk) in records
            .chunks(gwk_domain::blob::BLOB_CHUNK_BYTES)
            .enumerate()
        {
            let sequence = u32::try_from(sequence).map_err(|_| {
                gwk_domain::port::BlobError::Storage("snapshot has too many chunks".to_owned())
            })?;
            blobs.write_chunk(&upload, sequence, chunk).await?;
        }
        if records.is_empty() {
            // An empty projection set is a real snapshot: a kernel with no work
            // yet still has a state, and it is the empty one.
            blobs.write_chunk(&upload, 0, &[]).await?;
        }
        blobs.commit(upload, address.clone()).await
    };
    let (descriptor, _deduped) = store_blob
        .await
        .map_err(|e| Refusal::storage(format!("store checkpoint records: {e}")))?;
    // The invariant this whole design rests on: the blob's plaintext IS the
    // bytes that were hashed, so its content address and the projection hash
    // are one digest. If these ever diverge, one of them is describing
    // something other than what was stored.
    debug_assert_eq!(descriptor.address, address);

    let checkpoint = Checkpoint {
        schema_version: CHECKPOINT_SCHEMA_VERSION,
        through_sequence: through,
        projection_hash: hash,
        records_ref: PayloadRef {
            digest: address.as_str().to_owned(),
            media_type: RECORDS_MEDIA_TYPE.to_owned(),
            byte_size,
            retention_class: None,
            evidence_pin: None,
        },
        created_at: created_at.clone(),
    };

    // `DO NOTHING` rather than an upsert: two snapshots at one sequence are the
    // same state by definition, so the second is a no-op and never an
    // overwrite of a checkpoint someone may already be restoring from.
    sqlx::query(
        "INSERT INTO gwk_internal.checkpoint \
           (through_seq, schema_version, projection_hash, records_ref, created_at) \
         VALUES ($1::numeric, $2, $3, $4, $5::timestamptz) \
         ON CONFLICT (through_seq) DO NOTHING",
    )
    .bind(to_numeric_text(through.value()))
    .bind(i64::from(checkpoint.schema_version))
    .bind(&checkpoint.projection_hash)
    .bind(
        serde_json::to_value(&checkpoint.records_ref)
            .map_err(|e| Refusal::storage(format!("serialize records_ref: {e}")))?,
    )
    .bind(created_at.as_str())
    .execute(&mut *conn)
    .await
    .map_err(|e| Refusal::storage(format!("record checkpoint: {e}")))?;

    // The barrier's counters move in the same transaction as the row they
    // describe, so a rolled-back append leaves the barrier exactly where it
    // was and the next one is still due.
    sqlx::query(
        "UPDATE gwk_internal.writer SET checkpoint_seq = $1::numeric, checkpoint_at = $2::timestamptz \
         WHERE id = 1",
    )
    .bind(to_numeric_text(through.value()))
    .bind(created_at.as_str())
    .execute(&mut *conn)
    .await
    .map_err(|e| Refusal::storage(format!("advance the checkpoint barrier: {e}")))?;

    Ok(checkpoint)
}

/// Every checkpoint, newest first — the order the recovery ladder walks.
pub async fn checkpoints(conn: &mut PgConnection) -> Result<Vec<Checkpoint>, Refusal> {
    let rows = sqlx::query(
        "SELECT through_seq::text AS through_text, schema_version, projection_hash, records_ref, \
                to_json(created_at) #>> '{}' AS created_at \
         FROM gwk_internal.checkpoint ORDER BY through_seq DESC",
    )
    .fetch_all(conn)
    .await
    .map_err(|e| Refusal::storage(format!("read checkpoints: {e}")))?;

    rows.iter()
        .map(|row| {
            let get = |name: &str| -> Result<String, Refusal> {
                row.try_get(name)
                    .map_err(|e| Refusal::storage(format!("column {name}: {e}")))
            };
            let schema_version: i64 = row
                .try_get("schema_version")
                .map_err(|e| Refusal::storage(format!("column schema_version: {e}")))?;
            let records_ref: serde_json::Value = row
                .try_get("records_ref")
                .map_err(|e| Refusal::storage(format!("column records_ref: {e}")))?;
            Ok(Checkpoint {
                schema_version: u32::try_from(schema_version)
                    .map_err(|e| Refusal::storage(format!("schema_version: {e}")))?,
                through_sequence: Seq::new(
                    from_numeric_text(&get("through_text")?)
                        .map_err(|e| Refusal::storage(format!("column through_seq: {e}")))?,
                ),
                projection_hash: get("projection_hash")?,
                records_ref: serde_json::from_value(records_ref)
                    .map_err(|e| Refusal::storage(format!("column records_ref: {e}")))?,
                created_at: Timestamp::new(get("created_at")?),
            })
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn every_projection_is_visited_exactly_once_in_a_written_down_order() {
        // The hash depends on this order, so the list is asserted rather than
        // trusted: a table appearing twice would double its rows into the
        // digest, and one appearing under the wrong tag would deserialize into
        // the wrong contract type.
        let ordered: Vec<&str> = PROJECTIONS.iter().map(|p| p.tag).collect();
        let mut tags = ordered.clone();
        tags.sort_unstable();
        tags.dedup();
        assert_eq!(tags.len(), PROJECTIONS.len(), "a tag appears twice");
        assert_eq!(ordered, tags, "the visit order must be alphabetical");

        for projection in PROJECTIONS {
            let tag = projection.tag;
            // Every query names its tag TWICE — once as the `projection_type`
            // value, once as the single field, because that is the shape the
            // contract's records have. Reading the tag off the struct and
            // asserting the SQL agrees is what keeps the flag attached to the
            // table it actually describes.
            assert!(
                projection
                    .query
                    .contains(&format!("'projection_type', '{tag}', '{tag}',")),
                "{tag}: the record's one field must be named for its tag"
            );
            assert!(
                projection.query.contains(&format!(" FROM gwk.{tag} t ")),
                "{tag}: the query must read the table it is tagged for"
            );
            // Unordered rows would hash differently on every read, which is a
            // checkpoint that fails its own validation at random.
            assert!(
                projection.query.contains(" ORDER BY "),
                "{tag}: rows must be ordered"
            );
        }
    }

    #[test]
    fn only_the_tables_a_replay_can_rebuild_reach_the_hash() {
        // The exclusions are named here so adding a table cannot quietly join
        // them, and so the reason survives: `submit` writes both of these
        // itself, and on the paged path it does so for a command that appends
        // NO event — a row no replay will ever produce.
        let excluded: Vec<&str> = PROJECTIONS
            .iter()
            .filter(|p| !p.derived)
            .map(|p| p.tag)
            .collect();
        assert_eq!(excluded, ["attention_item", "receipt"]);
        assert_eq!(
            PROJECTIONS.iter().filter(|p| p.derived).count(),
            PROJECTIONS.len() - 2,
            "everything else must be rebuildable from the log"
        );
    }

    #[test]
    fn the_hash_is_over_the_stored_bytes_and_nothing_else() {
        // Whatever else changes, these two must stay the same function, or a
        // checkpoint's own address stops proving what it contains.
        let records = b"{\"projection_type\":\"task\"}\n".to_vec();
        let hash = projection_hash(&records);
        let address = BlobAddress::from_digest(&hash).expect("a legal address");
        assert_eq!(address.digest_hex(), hash);
        assert_eq!(
            hash,
            {
                let digest: [u8; 32] = Sha256::digest(&records).into();
                container::hex_lower(&digest)
            },
            "the hash must be a plain SHA-256 over the bytes"
        );
        // An empty projection set still has a hash — the digest of nothing.
        assert_eq!(
            projection_hash(&[]),
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        );
    }

    #[test]
    fn a_served_row_is_the_same_row_the_hash_canonicalizes() {
        // Each projection now spells its record-building expression twice, once
        // for the dump and once for the paged read. An edit landing on one and
        // not the other would make the row a client is served differ from the
        // row the checkpoint hashed — a disagreement nothing else in the system
        // is positioned to notice. Only the tail after `FROM` may differ.
        for projection in PROJECTIONS {
            let head = |q: &'static str| {
                q.split_once(" FROM ")
                    .expect("every projection query selects FROM a table")
                    .0
                    .to_owned()
            };
            assert_eq!(
                head(projection.query),
                head(projection.read),
                "{} builds a different record for the hash than for a read",
                projection.tag
            );
        }
    }

    #[test]
    fn the_cursor_key_is_the_column_the_page_was_ordered_by() {
        // `key` is what a client's next cursor is read out of and what the SQL
        // compares that cursor against. If they were ever different columns the
        // page would still return rows — the wrong ones, skipping or repeating
        // at every boundary, with nothing failing.
        for projection in PROJECTIONS {
            let key = projection.key;
            assert!(
                projection
                    .read
                    .contains(&format!("ORDER BY t.{key} COLLATE")),
                "{} pages by {key} but does not order by it",
                projection.tag
            );
            assert!(
                projection
                    .read
                    .contains(&format!("t.{key} COLLATE \"C\" > $1")),
                "{} orders by {key} but compares the cursor against another column",
                projection.tag
            );
        }
    }

    #[test]
    fn every_projection_a_client_can_name_has_a_read_behind_it() {
        // `ProjectionKind` is the request's vocabulary and `PROJECTIONS` is the
        // server's; a kind with no entry is a request that parses, is accepted,
        // and can never be answered.
        for kind in ProjectionKind::ALL {
            assert!(
                read_query(*kind).is_some(),
                "{} has no read query",
                kind.as_str()
            );
        }
    }
}