systemprompt_database/lifecycle/installation/extension/
current.rs

1//! Read-only check that the database schema matches the binary.
2//!
3//! A replica that boots without migrating (`--skip-migrate`, or
4//! `database.migrate_on_boot: false`) must still refuse a database the
5//! migration step never reached. [`schema_currency`] reports every enabled
6//! extension whose owned tables are all absent (never installed), every
7//! defined migration with no ledger row (pending), and every applied
8//! migration whose checksum no longer matches its file (drift). It reads the
9//! same primitives the installer uses and executes no extension DDL.
10//!
11//! The check also sees the opposite direction: an extension whose ledger holds
12//! a migration numbered past every one this binary declares is *ahead* — a
13//! newer release migrated it during a rolling update. That is reported in
14//! [`SchemaCurrency::ahead`] and is not a failure: under the expand/contract
15//! contract a newer schema stays readable by the previous release. Checksum
16//! drift in an ahead extension is the newer release's to judge, so it moves to
17//! [`SchemaCurrency::superseded_drift`] instead of failing the boot.
18//! The exception is a contract migration: an ahead migration the newer
19//! release recorded as `@compat: breaking` is listed again in
20//! [`SchemaCurrency::breaking`], and [`SchemaCurrency::breaks_this_release`]
21//! tells the caller this binary must not serve.
22//!
23//! Copyright (c) systemprompt.io — Business Source License 1.1.
24//! See <https://systemprompt.io> for licensing details.
25
26use systemprompt_extension::{ExtensionRegistry, LoaderError};
27use systemprompt_identifiers::ExtensionId;
28
29use super::super::prepare::prepare_extension_schema;
30use crate::lifecycle::migrations::{
31    ChecksumDrift, MigrationService, OrphanedMigration, PendingMigration,
32};
33use crate::services::DatabaseProvider;
34
35/// How far the live schema is behind — or ahead of — the binary's extensions.
36///
37/// [`SchemaCurrency::is_current`] ignores `ahead` and `superseded_drift`: a
38/// schema a newer release migrated is serviceable by this one.
39#[derive(Debug, Default)]
40pub struct SchemaCurrency {
41    pub fresh_extensions: Vec<ExtensionId>,
42    pub pending: Vec<PendingMigration>,
43    pub drift: Vec<ChecksumDrift>,
44    pub ahead: Vec<OrphanedMigration>,
45    pub superseded_drift: Vec<ChecksumDrift>,
46    pub breaking: Vec<OrphanedMigration>,
47}
48
49impl SchemaCurrency {
50    #[must_use]
51    pub const fn is_current(&self) -> bool {
52        self.fresh_extensions.is_empty() && self.pending.is_empty() && self.drift.is_empty()
53    }
54
55    #[must_use]
56    pub const fn is_ahead(&self) -> bool {
57        !self.ahead.is_empty()
58    }
59
60    #[must_use]
61    pub const fn breaks_this_release(&self) -> bool {
62        !self.breaking.is_empty()
63    }
64}
65
66pub async fn schema_currency(
67    db: &dyn DatabaseProvider,
68    registry: &ExtensionRegistry,
69    disabled_extensions: &[ExtensionId],
70) -> Result<SchemaCurrency, LoaderError> {
71    let migrations = MigrationService::new(db);
72    let mut currency = SchemaCurrency::default();
73
74    for ext in registry.enabled_schema_extensions(disabled_extensions)? {
75        let prepared = prepare_extension_schema(ext.as_ref())?;
76        let freshness = migrations
77            .assess_freshness(&prepared.extension_id, &prepared.owned_tables)
78            .await?;
79        if freshness.tables_total > 0 && freshness.tables_present == 0 {
80            currency.fresh_extensions.push(prepared.extension_id);
81            continue;
82        }
83        if ext.has_migrations() {
84            let latest = ext.migrations().iter().map(|m| m.version).max();
85            let status = migrations.status(ext.as_ref()).await?;
86            let ahead: Vec<OrphanedMigration> = status
87                .orphaned
88                .into_iter()
89                .filter(|m| latest.is_some_and(|latest| m.version > latest))
90                .collect();
91            currency.pending.extend(status.pending);
92            if ahead.is_empty() {
93                currency.drift.extend(status.drift);
94            } else {
95                let breaking = migrations.breaking_versions(&prepared.extension_id).await?;
96                currency.breaking.extend(
97                    ahead
98                        .iter()
99                        .filter(|m| breaking.contains(&m.version))
100                        .cloned(),
101                );
102                currency.superseded_drift.extend(status.drift);
103                currency.ahead.extend(ahead);
104            }
105        }
106    }
107
108    Ok(currency)
109}
systemprompt_database/lifecycle/installation/extension/current.rs

systemprompt_database/lifecycle/installation/extension/
current.rs
