systemprompt_database/projection/
mod.rs

1//! Store, don't compute: the data paradigm and the queue that carries it.
2//!
3//! Every analytics-shaped read in the system follows one rule. The answer a
4//! page, report or policy needs was known once — when the row it summarises
5//! was written — and is stored then, so no reader ever re-derives it from
6//! the raw history. The cost of a read is then the size of the answer, never
7//! the size of the ledger behind it.
8//!
9//! The vocabulary, from raw to read:
10//!
11//! 1. **Ledger** — the append-only raw tables the gateway, the hook plane, the
12//!    tool plane and the governance spine write (`ai_requests`,
13//!    `ai_request_payloads`, `ai_request_messages`, `mcp_tool_executions`,
14//!    `plugin_usage_events`, `governance_decisions`, `logs`, …). Read by
15//!    exactly two things: projectors, and single-entity detail queries (`WHERE
16//!    pk = $1`, no aggregate).
17//! 2. **Stamped fact** — a pure function of the row and what is in hand when it
18//!    is written, stored as a column in the same transaction by the table's
19//!    owner (`has_offered_tools`, `is_builtin`, `skill_ref`).
20//! 3. **Counter** — an additive aggregate the writer keeps current through a
21//!    `FOR EACH STATEMENT … REFERENCING NEW TABLE` trigger (never a row
22//!    trigger: a row trigger's cost is the fan-out that turned a 3,644-row
23//!    update into 27 minutes).
24//! 4. **Projection** — one denormalised row per entity (`<entity>_facts`),
25//!    maintained by `refresh_<projection>(keys TEXT[])` and drained from the
26//!    generic queue in this module.
27//! 5. **Rollup** — `<subject>_rollup_hourly(bucket, dims…)` of additive
28//!    measures, upserted at write time. Hourly, because analytics windows are
29//!    arbitrary timestamps; percentiles, distinct counts and top-N substrings
30//!    are not additive and keep reading raw rows for their window.
31//!
32//! **Every fact has four parts, all in the base schema**: the structure, the
33//! projector or stamp, the statement trigger (stamp directly when the
34//! transition table alone determines the value, else queue the keys) and the
35//! drainer job. An upgrade migration is the same DDL text plus one
36//! `INSERT INTO projection_backlog SELECT '<projection>', <key> FROM …`.
37//! Nothing else: a fresh install stamps migrations without running them, so
38//! anything a migration computes exists only on upgraded databases.
39//!
40//! **NULL means pending.** A reader never falls back to the ledger expression
41//! (`COALESCE(fact, <ledger expr>)`); it shows the fact as pending and
42//! [`depth`] says how many are. Two static gates keep this true: the
43//! ledger-reads gate over repository SQL, and `no_backfill_after_floor` over
44//! migrations, built on [`crate::backfill_statements`].
45//!
46//! ## The queue
47//!
48//! `projection_backlog(projection, key, queued_at, attempts)` is one table
49//! for every projection. Writers queue keys ([`queue`] or the SQL
50//! `queue_projection_backlog`); a drainer job calls [`drain`] with the
51//! projection's refresh, which takes a batch with `FOR UPDATE SKIP LOCKED`,
52//! refreshes it and deletes the rows in one transaction. A failed batch is
53//! rolled back and re-queued one attempt dearer; a key past its retry budget
54//! stays queued for an operator and never blocks the rest. A refresh that
55//! can tell which key of its batch is the bad one may [`park`] that key
56//! on its own transaction instead of failing the batch: the key goes
57//! straight back to the queue past the budget, counted as pending, and
58//! the rest of the batch commits. A statement that keeps a
59//! `refresh_*(keys TEXT[])` SQL function can use [`drain_with_sql`] and
60//! name it.
61//!
62//! Copyright (c) systemprompt.io — Business Source License 1.1.
63//! See <https://systemprompt.io> for licensing details.
64
65use std::time::{Duration, Instant};
66
67use sqlx::{PgConnection, PgExecutor, PgPool};
68use systemprompt_traits::RepositoryError;
69use tracing::warn;
70
71use crate::services::BoxFuture;
72
73/// How a drainer tick went. `remaining` is the queue depth when it stopped,
74/// so a job can report "partly drained" without another query.
75#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
76pub struct DrainReport {
77    pub keys: u64,
78    pub batches: u32,
79    pub remaining: i64,
80    pub elapsed: Duration,
81    pub requeued: u64,
82}
83
84/// A projection's refresh: given a connection inside the drain transaction
85/// and the keys taken, rebuild them and return how many rows were written.
86pub type RefreshFuture<'c> = BoxFuture<'c, Result<u64, RepositoryError>>;
87
88pub async fn queue<'e, E>(
89    executor: E,
90    projection: &str,
91    keys: &[String],
92) -> Result<i64, RepositoryError>
93where
94    E: PgExecutor<'e>,
95{
96    let queued: i64 = sqlx::query_scalar("SELECT queue_projection_backlog($1, $2)")
97        .bind(projection)
98        .bind(keys)
99        .fetch_one(executor)
100        .await?;
101    Ok(queued)
102}
103
104// Why: Postgres — take_projection_backlog's max_attempts defaults to 5; a
105// parked key sits one past it so the next take leaves it for an operator.
106pub const PARKED_ATTEMPTS: i32 = 6;
107
108pub async fn park<'e, E>(executor: E, projection: &str, key: &str) -> Result<(), RepositoryError>
109where
110    E: PgExecutor<'e>,
111{
112    sqlx::query(
113        "INSERT INTO projection_backlog (projection, key, attempts) VALUES ($1, $2, $3)
114         ON CONFLICT (projection, key) DO UPDATE
115         SET attempts = GREATEST(projection_backlog.attempts, EXCLUDED.attempts)",
116    )
117    .bind(projection)
118    .bind(key)
119    .bind(PARKED_ATTEMPTS)
120    .execute(executor)
121    .await?;
122    Ok(())
123}
124
125pub async fn depth<'e, E>(executor: E, projection: &str) -> Result<i64, RepositoryError>
126where
127    E: PgExecutor<'e>,
128{
129    let depth: i64 = sqlx::query_scalar("SELECT projection_backlog_depth($1)")
130        .bind(projection)
131        .fetch_one(executor)
132        .await?;
133    Ok(depth)
134}
135
136pub async fn depths<'e, E>(executor: E) -> Result<Vec<(String, i64)>, RepositoryError>
137where
138    E: PgExecutor<'e>,
139{
140    let rows: Vec<(String, i64)> = sqlx::query_as(
141        "SELECT projection, COUNT(*)::BIGINT FROM projection_backlog GROUP BY projection",
142    )
143    .fetch_all(executor)
144    .await?;
145    Ok(rows)
146}
147
148pub async fn drain<F>(
149    pool: &PgPool,
150    projection: &str,
151    batch: i32,
152    budget: Duration,
153    refresh: F,
154) -> Result<DrainReport, RepositoryError>
155where
156    F: for<'c> Fn(&'c mut PgConnection, Vec<String>) -> RefreshFuture<'c> + Send + Sync,
157{
158    let started = Instant::now();
159    let mut report = DrainReport::default();
160    while started.elapsed() < budget {
161        let mut tx = pool.begin().await?;
162        let keys: Vec<String> = sqlx::query_scalar("SELECT take_projection_backlog($1, $2)")
163            .bind(projection)
164            .bind(batch)
165            .fetch_one(&mut *tx)
166            .await?;
167        if keys.is_empty() {
168            tx.rollback().await?;
169            break;
170        }
171        let taken = u64::try_from(keys.len()).unwrap_or(u64::MAX);
172        match refresh(&mut tx, keys.clone()).await {
173            Ok(_) => {
174                tx.commit().await?;
175                report.keys = report.keys.saturating_add(taken);
176                report.batches = report.batches.saturating_add(1);
177            },
178            Err(error) => {
179                tx.rollback().await?;
180                let requeued: i64 = sqlx::query_scalar("SELECT requeue_projection_backlog($1, $2)")
181                    .bind(projection)
182                    .bind(&keys)
183                    .fetch_one(pool)
184                    .await?;
185                report.requeued = u64::try_from(requeued).unwrap_or(0);
186                warn!(
187                    projection,
188                    keys = keys.len(),
189                    error = %error,
190                    "projection refresh failed; batch re-queued one attempt dearer"
191                );
192                break;
193            },
194        }
195    }
196    report.remaining = depth(pool, projection).await?;
197    report.elapsed = started.elapsed();
198    Ok(report)
199}
200
201pub async fn drain_with_sql(
202    pool: &PgPool,
203    projection: &str,
204    batch: i32,
205    budget: Duration,
206    projector: &str,
207) -> Result<DrainReport, RepositoryError> {
208    let statement = projector_statement(projector)?;
209    let statement = statement.as_str();
210    drain(pool, projection, batch, budget, move |conn, keys| {
211        let statement = statement.to_owned();
212        Box::pin(async move {
213            // Why: Postgres — the only dynamic part is the projector's name,
214            // checked as a bare identifier by `projector_statement`; the keys
215            // are a bind parameter.
216            let written: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(statement.as_str()))
217                .bind(&keys)
218                .fetch_one(conn)
219                .await?;
220            Ok(u64::try_from(written).unwrap_or(0))
221        })
222    })
223    .await
224}
225
226fn projector_statement(projector: &str) -> Result<String, RepositoryError> {
227    let mut chars = projector.chars();
228    let leads = chars
229        .next()
230        .is_some_and(|c| c.is_ascii_lowercase() || c == '_');
231    let rest = chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_');
232    if !(leads && rest) {
233        return Err(RepositoryError::InvalidArgument {
234            field: "projector",
235            reason: format!("{projector:?} is not a bare lower-case identifier"),
236        });
237    }
238    Ok(format!("SELECT {projector}($1)"))
239}
systemprompt_database/projection/mod.rs

systemprompt_database/projection/
mod.rs
