systemprompt_database/lifecycle/migrations/
compat.rs

1//! The `@compat: breaking` ledger: contract migrations recorded where a
2//! previous release, which never saw their files, can still read them.
3//!
4//! The ledger is created on first use by the migrate step, under the
5//! bootstrap lock, and also declared by the database extension's schema.
6//! A row is written before the migration runs, so a crash between the two
7//! leaves a marker for an unapplied version; readers only consult markers for
8//! versions the ledger records as applied.
9//!
10//! Copyright (c) systemprompt.io — Business Source License 1.1.
11//! See <https://systemprompt.io> for licensing details.
12
13use systemprompt_extension::LoaderError;
14use systemprompt_identifiers::ExtensionId;
15
16use super::MigrationService;
17
18const RECORD_BREAKING_SQL: &str = "INSERT INTO extension_migration_compat (extension_id, \
19                                   version, breaks_older) VALUES ($1, $2, TRUE) ON CONFLICT \
20                                   (extension_id, version) DO UPDATE SET breaks_older = TRUE";
21
22const LEDGER_EXISTS_SQL: &str =
23    "SELECT to_regclass('extension_migration_compat') IS NOT NULL AS present";
24
25const BREAKING_VERSIONS_SQL: &str = "SELECT version FROM extension_migration_compat WHERE \
26                                     extension_id = $1 AND breaks_older ORDER BY version";
27
28impl MigrationService<'_> {
29    pub(super) async fn record_breaking(
30        &self,
31        ext_id: &ExtensionId,
32        version: u32,
33    ) -> Result<(), LoaderError> {
34        let ledger = include_str!("../../../schema/extension_migration_compat.sql");
35        crate::services::SqlExecutor::execute_statements_parsed(self.db, ledger)
36            .await
37            .map_err(|e| LoaderError::MigrationStepFailed {
38                extension: ext_id.clone(),
39                context: "Failed to ensure the migration compatibility ledger exists".to_owned(),
40                source: Box::new(e),
41            })?;
42        self.db
43            .execute(&RECORD_BREAKING_SQL, &[&ext_id, &version])
44            .await
45            .map_err(|e| LoaderError::MigrationStepFailed {
46                extension: ext_id.clone(),
47                context: format!("Failed to record migration {version} as breaking"),
48                source: Box::new(e),
49            })?;
50        Ok(())
51    }
52
53    // Why: this runs on verify-only boots, outside the bootstrap lock, so it
54    // must not race other replicas creating the ledger; a database no migrate
55    // step has touched since the ledger was introduced has no breaking rows.
56    pub async fn breaking_versions(&self, ext_id: &ExtensionId) -> Result<Vec<u32>, LoaderError> {
57        let ledger = self
58            .db
59            .query_raw_with(&LEDGER_EXISTS_SQL, &[])
60            .await
61            .map_err(|e| LoaderError::MigrationStepFailed {
62                extension: ext_id.clone(),
63                context: "Failed to look up the migration compatibility ledger".to_owned(),
64                source: Box::new(e),
65            })?;
66        let present = ledger
67            .rows
68            .first()
69            .and_then(|row| row.get("present"))
70            .and_then(serde_json::Value::as_bool)
71            .unwrap_or(false);
72        if !present {
73            return Ok(Vec::new());
74        }
75        let result = self
76            .db
77            .query_raw_with(&BREAKING_VERSIONS_SQL, &[&ext_id])
78            .await
79            .map_err(|e| LoaderError::MigrationStepFailed {
80                extension: ext_id.clone(),
81                context: "Failed to read the migration compatibility ledger".to_owned(),
82                source: Box::new(e),
83            })?;
84        Ok(result
85            .rows
86            .iter()
87            .filter_map(|row| row.get("version").and_then(serde_json::Value::as_i64))
88            .filter_map(|v| u32::try_from(v).ok())
89            .collect())
90    }
91}
systemprompt_database/lifecycle/migrations/compat.rs

systemprompt_database/lifecycle/migrations/
compat.rs
