Skip to main content

gwk_kernel/
checkpoint.rs

1//! Projection snapshots: the recovery shortcut, never the truth.
2//!
3//! A checkpoint is the projection tables, canonicalized, hashed, and stored as
4//! one encrypted blob at one `global_sequence`. It is EVIDENCE, not a restore
5//! point: this schema's `born_initial` and `no_truncate` guards mean projection
6//! rows can only ever be produced by replaying the log, so what a checkpoint
7//! buys is the ability to check that result, not to skip it (see
8//! [`crate::recover`]). The worst a missing or corrupt one costs is a
9//! comparison — which is why an append is allowed to take one and none of the
10//! contract depends on it existing.
11//!
12//! Two things make the hash mean anything.
13//!
14//! **Records are re-serialized through the CONTRACT type, never passed
15//! through.** `to_jsonb` returns keys in whatever order the row's physical
16//! layout gives, and that layout is a property of one database's history — add
17//! a column and it moves. Deserializing each row into its [`ProjectionRecord`]
18//! and serializing that back means the bytes are determined by the type
19//! declaration instead, so two kernels serving the same log agree, and so does
20//! the same kernel after a `VACUUM FULL`.
21//!
22//! **The visit order is a written-down constant, not a catalog query.** The
23//! hash depends on which table comes first, so that has to be something a
24//! reader can see and a diff can show changing.
25//!
26//! The records blob's plaintext IS the canonical bytes, so `projection_hash`
27//! and the blob's own content address are the same digest. That is stated as
28//! an invariant and asserted, not left as a coincidence for someone to
29//! discover while debugging a restore.
30
31use gwk_domain::blob::BlobAddress;
32use gwk_domain::checkpoint::{CHECKPOINT_SCHEMA_VERSION, Checkpoint};
33use gwk_domain::envelope::PayloadRef;
34use gwk_domain::ids::{ByteCount, Seq, Timestamp};
35use gwk_domain::port::BlobStore;
36use gwk_domain::protocol::{ProjectionKind, ProjectionRecord};
37use sha2::{Digest, Sha256};
38use sqlx::{PgConnection, Row};
39
40use crate::blob::container;
41use crate::blob::store::PgBlobStore;
42use crate::numeric::{from_numeric_text, to_numeric_text};
43use crate::project::Refusal;
44
45/// What the records blob is: one canonical record per line.
46///
47/// Line-delimited rather than one JSON array, because recovery streams it back
48/// a chunk at a time and a line is a frame it can complete without holding the
49/// whole document.
50pub const RECORDS_MEDIA_TYPE: &str = "application/x-ndjson";
51
52/// Every projection table, in the order a snapshot visits them, each shaped
53/// into the exact wire form of a [`ProjectionRecord`].
54///
55/// Alphabetical, and spelled out one query at a time. Two kinds of adjustment
56/// appear, and both are deliberate rather than convenient:
57///
58/// * **A `::text` cast on four columns.** `to_jsonb` renders a `numeric` as a
59///   JSON NUMBER, while the contract carries 64-bit counters as decimal
60///   STRINGS. The difference is invisible until the value passes 2^53, at which
61///   point the number silently comes back as a different one — so the cast goes
62///   on every such column, not on the ones that have gotten large so far.
63///
64/// * **Two renamed columns and one subtracted one.** `gwk.receipt` stores
65///   `from`/`to` as `from_state`/`to_state` because the bare words are SQL
66///   reserved words, and `gwk.orchestrator_checkpoint.updated_at` is row
67///   bookkeeping the contract type does not carry — a reader orders those by
68///   the checkpoint's own `seq`. Each is handled BY NAME rather than by a
69///   general tolerance, so a column added to any other table still fails the
70///   round trip. That failure IS the parity check between the DDL and
71///   `gwk-domain`, and nothing else in this kernel performs it.
72///
73/// The `derived` flag is the third adjustment, and the one with teeth. Two
74/// tables hold rows the log cannot reproduce:
75///
76/// * `gwk.receipt` — `submit` writes the authority receipt itself, and on the
77///   PAGED path it does so for a command that is REFUSED and therefore appends
78///   no event at all ("the one refusal in the kernel that leaves rows behind").
79/// * `gwk.attention_item` — the same paged path raises its item directly
80///   through `page_attention`, again with no event behind it.
81///
82/// A hash over those tables could never be reproduced by a replay, so a
83/// checkpoint carrying them would fail every scratch rebuild forever and the
84/// failure would say nothing. They stay in the canonical dump — that is where
85/// the DDL-to-contract parity check happens, and it must cover all fifteen —
86/// and stay OUT of the digest. What guards them instead is what always did:
87/// `receipt_append_only` and the delete guards, which no privilege can bypass.
88struct Projection {
89    tag: &'static str,
90    /// The column `read` orders and pages by, which is also the field a client
91    /// reads a continuation cursor out of. Written down rather than inferred:
92    /// fourteen projections key on `id` and `orchestrator_checkpoint` does not,
93    /// and a rule that guessed would hand back the wrong cursor for the one
94    /// table that differs — a page that silently restarts, not one that fails.
95    key: &'static str,
96    query: &'static str,
97    /// The same record, one page at a time: `$1` a cursor (exclusive), `$2` an
98    /// exact key, `$3` the row ceiling. One query serves both reads because
99    /// get-by-id IS a one-row page — a second near-identical string per table
100    /// would be one more place for the record shape to drift from `query`.
101    ///
102    /// Ordering and the cursor comparison are both `COLLATE "C"`. Under a
103    /// locale collation — which is what the stock PostgreSQL image gives you —
104    /// ids differing only in punctuation can TIE, and two ties either side of a
105    /// page boundary drop a row or repeat one. Byte order has no ties.
106    read: &'static str,
107    /// Whether replaying the log through `apply_event` rebuilds this table.
108    derived: bool,
109}
110
111const fn derived(
112    tag: &'static str,
113    key: &'static str,
114    query: &'static str,
115    read: &'static str,
116) -> Projection {
117    Projection {
118        tag,
119        key,
120        query,
121        read,
122        derived: true,
123    }
124}
125
126const fn written_beside_the_log(
127    tag: &'static str,
128    key: &'static str,
129    query: &'static str,
130    read: &'static str,
131) -> Projection {
132    Projection {
133        tag,
134        key,
135        query,
136        read,
137        derived: false,
138    }
139}
140
141/// The paged read for one projection and the column it pages by, or `None` if
142/// the kind has no table — which cannot happen while
143/// [`every_projection_is_visited_exactly_once_in_a_written_down_order`] passes,
144/// and is returned rather than panicked because this is reached from a client
145/// request.
146///
147/// The two travel together on purpose: a caller that had the query but chose
148/// the cursor field for itself would be free to choose a different one.
149pub fn read_query(kind: ProjectionKind) -> Option<(&'static str, &'static str)> {
150    PROJECTIONS
151        .iter()
152        .find(|p| p.tag == kind.as_str())
153        .map(|p| (p.read, p.key))
154}
155
156const PROJECTIONS: &[Projection] = &[
157    derived(
158        "attempt",
159        "id",
160        "SELECT jsonb_build_object('projection_type', 'attempt', 'attempt', to_jsonb(t))::text \
161         FROM gwk.attempt t ORDER BY t.id",
162        "SELECT jsonb_build_object('projection_type', 'attempt', 'attempt', to_jsonb(t))::text \
163         FROM gwk.attempt t \
164         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
165           AND ($2::text IS NULL OR t.id = $2) \
166         ORDER BY t.id COLLATE \"C\" LIMIT $3",
167    ),
168    written_beside_the_log(
169        "attention_item",
170        "id",
171        "SELECT jsonb_build_object('projection_type', 'attention_item', 'attention_item', \
172           to_jsonb(t))::text FROM gwk.attention_item t ORDER BY t.id",
173        "SELECT jsonb_build_object('projection_type', 'attention_item', 'attention_item', \
174           to_jsonb(t))::text FROM gwk.attention_item t \
175         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
176           AND ($2::text IS NULL OR t.id = $2) \
177         ORDER BY t.id COLLATE \"C\" LIMIT $3",
178    ),
179    derived(
180        "authority_grant",
181        "id",
182        "SELECT jsonb_build_object('projection_type', 'authority_grant', 'authority_grant', \
183           to_jsonb(t))::text FROM gwk.authority_grant t ORDER BY t.id",
184        "SELECT jsonb_build_object('projection_type', 'authority_grant', 'authority_grant', \
185           to_jsonb(t))::text FROM gwk.authority_grant t \
186         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
187           AND ($2::text IS NULL OR t.id = $2) \
188         ORDER BY t.id COLLATE \"C\" LIMIT $3",
189    ),
190    derived(
191        "command",
192        "id",
193        "SELECT jsonb_build_object('projection_type', 'command', 'command', to_jsonb(t))::text \
194         FROM gwk.command t ORDER BY t.id",
195        "SELECT jsonb_build_object('projection_type', 'command', 'command', to_jsonb(t))::text \
196         FROM gwk.command t \
197         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
198           AND ($2::text IS NULL OR t.id = $2) \
199         ORDER BY t.id COLLATE \"C\" LIMIT $3",
200    ),
201    derived(
202        "dispatch_node",
203        "id",
204        "SELECT jsonb_build_object('projection_type', 'dispatch_node', 'dispatch_node', \
205           to_jsonb(t))::text FROM gwk.dispatch_node t ORDER BY t.id",
206        "SELECT jsonb_build_object('projection_type', 'dispatch_node', 'dispatch_node', \
207           to_jsonb(t))::text FROM gwk.dispatch_node t \
208         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
209           AND ($2::text IS NULL OR t.id = $2) \
210         ORDER BY t.id COLLATE \"C\" LIMIT $3",
211    ),
212    derived(
213        "engine_session",
214        "id",
215        "SELECT jsonb_build_object('projection_type', 'engine_session', 'engine_session', \
216           to_jsonb(t))::text FROM gwk.engine_session t ORDER BY t.id",
217        "SELECT jsonb_build_object('projection_type', 'engine_session', 'engine_session', \
218           to_jsonb(t))::text FROM gwk.engine_session t \
219         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
220           AND ($2::text IS NULL OR t.id = $2) \
221         ORDER BY t.id COLLATE \"C\" LIMIT $3",
222    ),
223    derived(
224        "evidence",
225        "id",
226        "SELECT jsonb_build_object('projection_type', 'evidence', 'evidence', \
227           to_jsonb(t) || jsonb_build_object('byte_size', t.byte_size::text))::text \
228         FROM gwk.evidence t ORDER BY t.id",
229        "SELECT jsonb_build_object('projection_type', 'evidence', 'evidence', \
230           to_jsonb(t) || jsonb_build_object('byte_size', t.byte_size::text))::text \
231         FROM gwk.evidence t \
232         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
233           AND ($2::text IS NULL OR t.id = $2) \
234         ORDER BY t.id COLLATE \"C\" LIMIT $3",
235    ),
236    derived(
237        "gate",
238        "id",
239        "SELECT jsonb_build_object('projection_type', 'gate', 'gate', to_jsonb(t))::text \
240         FROM gwk.gate t ORDER BY t.id",
241        "SELECT jsonb_build_object('projection_type', 'gate', 'gate', to_jsonb(t))::text \
242         FROM gwk.gate t \
243         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
244           AND ($2::text IS NULL OR t.id = $2) \
245         ORDER BY t.id COLLATE \"C\" LIMIT $3",
246    ),
247    derived(
248        "ingested_record",
249        "id",
250        "SELECT jsonb_build_object('projection_type', 'ingested_record', 'ingested_record', \
251           to_jsonb(t) || jsonb_build_object('event_seq', t.event_seq::text))::text \
252         FROM gwk.ingested_record t ORDER BY t.id",
253        "SELECT jsonb_build_object('projection_type', 'ingested_record', 'ingested_record', \
254           to_jsonb(t) || jsonb_build_object('event_seq', t.event_seq::text))::text \
255         FROM gwk.ingested_record t \
256         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
257           AND ($2::text IS NULL OR t.id = $2) \
258         ORDER BY t.id COLLATE \"C\" LIMIT $3",
259    ),
260    derived(
261        "lease",
262        "id",
263        "SELECT jsonb_build_object('projection_type', 'lease', 'lease', \
264           to_jsonb(t) || jsonb_build_object('fence_token', t.fence_token::text))::text \
265         FROM gwk.lease t ORDER BY t.id",
266        "SELECT jsonb_build_object('projection_type', 'lease', 'lease', \
267           to_jsonb(t) || jsonb_build_object('fence_token', t.fence_token::text))::text \
268         FROM gwk.lease t \
269         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
270           AND ($2::text IS NULL OR t.id = $2) \
271         ORDER BY t.id COLLATE \"C\" LIMIT $3",
272    ),
273    derived(
274        "message",
275        "id",
276        "SELECT jsonb_build_object('projection_type', 'message', 'message', to_jsonb(t))::text \
277         FROM gwk.message t ORDER BY t.id",
278        "SELECT jsonb_build_object('projection_type', 'message', 'message', to_jsonb(t))::text \
279         FROM gwk.message t \
280         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
281           AND ($2::text IS NULL OR t.id = $2) \
282         ORDER BY t.id COLLATE \"C\" LIMIT $3",
283    ),
284    derived(
285        "orchestrator_checkpoint",
286        "orchestrator_id",
287        "SELECT jsonb_build_object('projection_type', 'orchestrator_checkpoint', \
288           'orchestrator_checkpoint', \
289           (to_jsonb(t) - 'updated_at') || jsonb_build_object('seq', t.seq::text))::text \
290         FROM gwk.orchestrator_checkpoint t ORDER BY t.orchestrator_id",
291        // Keyed on `orchestrator_id`: this is the one projection whose primary
292        // key is not called `id`, and paging it by a column it does not have
293        // would fail at the database rather than quietly.
294        "SELECT jsonb_build_object('projection_type', 'orchestrator_checkpoint', \
295           'orchestrator_checkpoint', \
296           (to_jsonb(t) - 'updated_at') || jsonb_build_object('seq', t.seq::text))::text \
297         FROM gwk.orchestrator_checkpoint t \
298         WHERE ($1::text IS NULL OR t.orchestrator_id COLLATE \"C\" > $1) \
299           AND ($2::text IS NULL OR t.orchestrator_id = $2) \
300         ORDER BY t.orchestrator_id COLLATE \"C\" LIMIT $3",
301    ),
302    written_beside_the_log(
303        "receipt",
304        "id",
305        "SELECT jsonb_build_object('projection_type', 'receipt', 'receipt', \
306           (to_jsonb(t) - 'from_state' - 'to_state') \
307           || jsonb_strip_nulls(jsonb_build_object('from', t.from_state, 'to', t.to_state)))::text \
308         FROM gwk.receipt t ORDER BY t.id",
309        "SELECT jsonb_build_object('projection_type', 'receipt', 'receipt', \
310           (to_jsonb(t) - 'from_state' - 'to_state') \
311           || jsonb_strip_nulls(jsonb_build_object('from', t.from_state, 'to', t.to_state)))::text \
312         FROM gwk.receipt t \
313         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
314           AND ($2::text IS NULL OR t.id = $2) \
315         ORDER BY t.id COLLATE \"C\" LIMIT $3",
316    ),
317    derived(
318        "task",
319        "id",
320        "SELECT jsonb_build_object('projection_type', 'task', 'task', to_jsonb(t))::text \
321         FROM gwk.task t ORDER BY t.id",
322        "SELECT jsonb_build_object('projection_type', 'task', 'task', to_jsonb(t))::text \
323         FROM gwk.task t \
324         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
325           AND ($2::text IS NULL OR t.id = $2) \
326         ORDER BY t.id COLLATE \"C\" LIMIT $3",
327    ),
328    derived(
329        "worktree",
330        "id",
331        "SELECT jsonb_build_object('projection_type', 'worktree', 'worktree', to_jsonb(t))::text \
332         FROM gwk.worktree t ORDER BY t.id",
333        "SELECT jsonb_build_object('projection_type', 'worktree', 'worktree', to_jsonb(t))::text \
334         FROM gwk.worktree t \
335         WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
336           AND ($2::text IS NULL OR t.id = $2) \
337         ORDER BY t.id COLLATE \"C\" LIMIT $3",
338    ),
339];
340
341/// The canonical bytes of every projection row: one record per line, each one
342/// having made the round trip through its contract type.
343///
344/// Reading through the caller's connection is deliberate — inside an append
345/// transaction it sees that transaction's own uncommitted projections, which is
346/// exactly the state the checkpoint claims to describe.
347pub async fn canonical_records(conn: &mut PgConnection) -> Result<Vec<u8>, Refusal> {
348    records(conn, false).await
349}
350
351/// The subset a replay can rebuild — what a checkpoint actually hashes.
352///
353/// The two tables left out are written beside the log rather than from it (see
354/// [`Projection`]), so including them would produce a digest no rebuild could
355/// ever match. Everything downstream — the checkpoint, the readiness compare,
356/// the scratch rebuild — uses THIS one, and the difference between the two
357/// functions is the whole reason the invariant holds.
358pub async fn derived_records(conn: &mut PgConnection) -> Result<Vec<u8>, Refusal> {
359    records(conn, true).await
360}
361
362async fn records(conn: &mut PgConnection, derived_only: bool) -> Result<Vec<u8>, Refusal> {
363    let mut out = Vec::new();
364    for projection in PROJECTIONS {
365        if derived_only && !projection.derived {
366            continue;
367        }
368        // Read through the struct field, which is a `&'static str`: sqlx 0.9
369        // accepts a literal-lifetime query and refuses anything else.
370        let rows = sqlx::query(projection.query)
371            .fetch_all(&mut *conn)
372            .await
373            .map_err(|e| Refusal::storage(format!("read {} projections: {e}", projection.tag)))?;
374        for row in &rows {
375            let raw: String = row
376                .try_get(0)
377                .map_err(|e| Refusal::storage(format!("projection row: {e}")))?;
378            // `deny_unknown_fields` on every entity makes this the parity check
379            // between the DDL and the contract types: a column with no field
380            // fails here rather than silently dropping out of the hash.
381            let record: ProjectionRecord = serde_json::from_str(&raw).map_err(|e| {
382                Refusal::storage(format!(
383                    "projection row does not match the contract type: {e}"
384                ))
385            })?;
386            serde_json::to_writer(&mut out, &record)
387                .map_err(|e| Refusal::storage(format!("serialize projection record: {e}")))?;
388            out.push(b'\n');
389        }
390    }
391    Ok(out)
392}
393
394/// The digest the checkpoint records, over exactly the bytes it stores.
395pub fn projection_hash(records: &[u8]) -> String {
396    let digest: [u8; 32] = Sha256::digest(records).into();
397    container::hex_lower(&digest)
398}
399
400/// Snapshot the projections as `conn`'s transaction will leave them, store the
401/// records, and record the checkpoint.
402///
403/// The blob is committed through the blob store's OWN connections, so it lands
404/// before the caller's transaction does. If that transaction then rolls back,
405/// what is left is a blob nothing references — which is precisely what sweep
406/// reclaims, and why sweep has to consider checkpoints as well as events.
407///
408/// The reverse order is what has no recovery: a checkpoint row committed beside
409/// a blob whose write was rolled back is a checkpoint that fails validation
410/// forever, and the fallback ladder would walk past it on every single startup.
411pub async fn snapshot(
412    conn: &mut PgConnection,
413    blobs: &PgBlobStore,
414    through: Seq,
415    created_at: &Timestamp,
416) -> Result<Checkpoint, Refusal> {
417    let records = derived_records(conn).await?;
418    let hash = projection_hash(&records);
419    let address =
420        BlobAddress::from_digest(&hash).map_err(|e| Refusal::storage(format!("hash: {e}")))?;
421
422    let byte_size = ByteCount::new(records.len() as u64);
423    let store_blob = async {
424        let upload = blobs
425            .begin(RECORDS_MEDIA_TYPE.to_owned(), byte_size)
426            .await?;
427        for (sequence, chunk) in records
428            .chunks(gwk_domain::blob::BLOB_CHUNK_BYTES)
429            .enumerate()
430        {
431            let sequence = u32::try_from(sequence).map_err(|_| {
432                gwk_domain::port::BlobError::Storage("snapshot has too many chunks".to_owned())
433            })?;
434            blobs.write_chunk(&upload, sequence, chunk).await?;
435        }
436        if records.is_empty() {
437            // An empty projection set is a real snapshot: a kernel with no work
438            // yet still has a state, and it is the empty one.
439            blobs.write_chunk(&upload, 0, &[]).await?;
440        }
441        blobs.commit(upload, address.clone()).await
442    };
443    let (descriptor, _deduped) = store_blob
444        .await
445        .map_err(|e| Refusal::storage(format!("store checkpoint records: {e}")))?;
446    // The invariant this whole design rests on: the blob's plaintext IS the
447    // bytes that were hashed, so its content address and the projection hash
448    // are one digest. If these ever diverge, one of them is describing
449    // something other than what was stored.
450    debug_assert_eq!(descriptor.address, address);
451
452    let checkpoint = Checkpoint {
453        schema_version: CHECKPOINT_SCHEMA_VERSION,
454        through_sequence: through,
455        projection_hash: hash,
456        records_ref: PayloadRef {
457            digest: address.as_str().to_owned(),
458            media_type: RECORDS_MEDIA_TYPE.to_owned(),
459            byte_size,
460            retention_class: None,
461            evidence_pin: None,
462        },
463        created_at: created_at.clone(),
464    };
465
466    // `DO NOTHING` rather than an upsert: two snapshots at one sequence are the
467    // same state by definition, so the second is a no-op and never an
468    // overwrite of a checkpoint someone may already be restoring from.
469    sqlx::query(
470        "INSERT INTO gwk_internal.checkpoint \
471           (through_seq, schema_version, projection_hash, records_ref, created_at) \
472         VALUES ($1::numeric, $2, $3, $4, $5::timestamptz) \
473         ON CONFLICT (through_seq) DO NOTHING",
474    )
475    .bind(to_numeric_text(through.value()))
476    .bind(i64::from(checkpoint.schema_version))
477    .bind(&checkpoint.projection_hash)
478    .bind(
479        serde_json::to_value(&checkpoint.records_ref)
480            .map_err(|e| Refusal::storage(format!("serialize records_ref: {e}")))?,
481    )
482    .bind(created_at.as_str())
483    .execute(&mut *conn)
484    .await
485    .map_err(|e| Refusal::storage(format!("record checkpoint: {e}")))?;
486
487    // The barrier's counters move in the same transaction as the row they
488    // describe, so a rolled-back append leaves the barrier exactly where it
489    // was and the next one is still due.
490    sqlx::query(
491        "UPDATE gwk_internal.writer SET checkpoint_seq = $1::numeric, checkpoint_at = $2::timestamptz \
492         WHERE id = 1",
493    )
494    .bind(to_numeric_text(through.value()))
495    .bind(created_at.as_str())
496    .execute(&mut *conn)
497    .await
498    .map_err(|e| Refusal::storage(format!("advance the checkpoint barrier: {e}")))?;
499
500    Ok(checkpoint)
501}
502
503/// Every checkpoint, newest first — the order the recovery ladder walks.
504pub async fn checkpoints(conn: &mut PgConnection) -> Result<Vec<Checkpoint>, Refusal> {
505    let rows = sqlx::query(
506        "SELECT through_seq::text AS through_text, schema_version, projection_hash, records_ref, \
507                to_json(created_at) #>> '{}' AS created_at \
508         FROM gwk_internal.checkpoint ORDER BY through_seq DESC",
509    )
510    .fetch_all(conn)
511    .await
512    .map_err(|e| Refusal::storage(format!("read checkpoints: {e}")))?;
513
514    rows.iter()
515        .map(|row| {
516            let get = |name: &str| -> Result<String, Refusal> {
517                row.try_get(name)
518                    .map_err(|e| Refusal::storage(format!("column {name}: {e}")))
519            };
520            let schema_version: i64 = row
521                .try_get("schema_version")
522                .map_err(|e| Refusal::storage(format!("column schema_version: {e}")))?;
523            let records_ref: serde_json::Value = row
524                .try_get("records_ref")
525                .map_err(|e| Refusal::storage(format!("column records_ref: {e}")))?;
526            Ok(Checkpoint {
527                schema_version: u32::try_from(schema_version)
528                    .map_err(|e| Refusal::storage(format!("schema_version: {e}")))?,
529                through_sequence: Seq::new(
530                    from_numeric_text(&get("through_text")?)
531                        .map_err(|e| Refusal::storage(format!("column through_seq: {e}")))?,
532                ),
533                projection_hash: get("projection_hash")?,
534                records_ref: serde_json::from_value(records_ref)
535                    .map_err(|e| Refusal::storage(format!("column records_ref: {e}")))?,
536                created_at: Timestamp::new(get("created_at")?),
537            })
538        })
539        .collect()
540}
541
542#[cfg(test)]
543mod tests {
544    use super::*;
545
546    #[test]
547    fn every_projection_is_visited_exactly_once_in_a_written_down_order() {
548        // The hash depends on this order, so the list is asserted rather than
549        // trusted: a table appearing twice would double its rows into the
550        // digest, and one appearing under the wrong tag would deserialize into
551        // the wrong contract type.
552        let ordered: Vec<&str> = PROJECTIONS.iter().map(|p| p.tag).collect();
553        let mut tags = ordered.clone();
554        tags.sort_unstable();
555        tags.dedup();
556        assert_eq!(tags.len(), PROJECTIONS.len(), "a tag appears twice");
557        assert_eq!(ordered, tags, "the visit order must be alphabetical");
558
559        for projection in PROJECTIONS {
560            let tag = projection.tag;
561            // Every query names its tag TWICE — once as the `projection_type`
562            // value, once as the single field, because that is the shape the
563            // contract's records have. Reading the tag off the struct and
564            // asserting the SQL agrees is what keeps the flag attached to the
565            // table it actually describes.
566            assert!(
567                projection
568                    .query
569                    .contains(&format!("'projection_type', '{tag}', '{tag}',")),
570                "{tag}: the record's one field must be named for its tag"
571            );
572            assert!(
573                projection.query.contains(&format!(" FROM gwk.{tag} t ")),
574                "{tag}: the query must read the table it is tagged for"
575            );
576            // Unordered rows would hash differently on every read, which is a
577            // checkpoint that fails its own validation at random.
578            assert!(
579                projection.query.contains(" ORDER BY "),
580                "{tag}: rows must be ordered"
581            );
582        }
583    }
584
585    #[test]
586    fn only_the_tables_a_replay_can_rebuild_reach_the_hash() {
587        // The exclusions are named here so adding a table cannot quietly join
588        // them, and so the reason survives: `submit` writes both of these
589        // itself, and on the paged path it does so for a command that appends
590        // NO event — a row no replay will ever produce.
591        let excluded: Vec<&str> = PROJECTIONS
592            .iter()
593            .filter(|p| !p.derived)
594            .map(|p| p.tag)
595            .collect();
596        assert_eq!(excluded, ["attention_item", "receipt"]);
597        assert_eq!(
598            PROJECTIONS.iter().filter(|p| p.derived).count(),
599            PROJECTIONS.len() - 2,
600            "everything else must be rebuildable from the log"
601        );
602    }
603
604    #[test]
605    fn the_hash_is_over_the_stored_bytes_and_nothing_else() {
606        // Whatever else changes, these two must stay the same function, or a
607        // checkpoint's own address stops proving what it contains.
608        let records = b"{\"projection_type\":\"task\"}\n".to_vec();
609        let hash = projection_hash(&records);
610        let address = BlobAddress::from_digest(&hash).expect("a legal address");
611        assert_eq!(address.digest_hex(), hash);
612        assert_eq!(
613            hash,
614            {
615                let digest: [u8; 32] = Sha256::digest(&records).into();
616                container::hex_lower(&digest)
617            },
618            "the hash must be a plain SHA-256 over the bytes"
619        );
620        // An empty projection set still has a hash — the digest of nothing.
621        assert_eq!(
622            projection_hash(&[]),
623            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
624        );
625    }
626
627    #[test]
628    fn a_served_row_is_the_same_row_the_hash_canonicalizes() {
629        // Each projection now spells its record-building expression twice, once
630        // for the dump and once for the paged read. An edit landing on one and
631        // not the other would make the row a client is served differ from the
632        // row the checkpoint hashed — a disagreement nothing else in the system
633        // is positioned to notice. Only the tail after `FROM` may differ.
634        for projection in PROJECTIONS {
635            let head = |q: &'static str| {
636                q.split_once(" FROM ")
637                    .expect("every projection query selects FROM a table")
638                    .0
639                    .to_owned()
640            };
641            assert_eq!(
642                head(projection.query),
643                head(projection.read),
644                "{} builds a different record for the hash than for a read",
645                projection.tag
646            );
647        }
648    }
649
650    #[test]
651    fn the_cursor_key_is_the_column_the_page_was_ordered_by() {
652        // `key` is what a client's next cursor is read out of and what the SQL
653        // compares that cursor against. If they were ever different columns the
654        // page would still return rows — the wrong ones, skipping or repeating
655        // at every boundary, with nothing failing.
656        for projection in PROJECTIONS {
657            let key = projection.key;
658            assert!(
659                projection
660                    .read
661                    .contains(&format!("ORDER BY t.{key} COLLATE")),
662                "{} pages by {key} but does not order by it",
663                projection.tag
664            );
665            assert!(
666                projection
667                    .read
668                    .contains(&format!("t.{key} COLLATE \"C\" > $1")),
669                "{} orders by {key} but compares the cursor against another column",
670                projection.tag
671            );
672        }
673    }
674
675    #[test]
676    fn every_projection_a_client_can_name_has_a_read_behind_it() {
677        // `ProjectionKind` is the request's vocabulary and `PROJECTIONS` is the
678        // server's; a kind with no entry is a request that parses, is accepted,
679        // and can never be answered.
680        for kind in ProjectionKind::ALL {
681            assert!(
682                read_query(*kind).is_some(),
683                "{} has no read query",
684                kind.as_str()
685            );
686        }
687    }
688}