v0.3.2-pre.006
This commit is contained in:
312
crates/ksp-store-postgres-lib/src/migration.rs
Normal file
312
crates/ksp-store-postgres-lib/src/migration.rs
Normal file
@@ -0,0 +1,312 @@
|
||||
// file: crates/ksp-store-postgres-lib/src/migration.rs
|
||||
// version: 1
|
||||
|
||||
use sha2::Digest; // rust-rules: trait-import
|
||||
|
||||
const ADVISORY_LOCK_KEY: i64 = 0x4b53_5053_544f_5245;
|
||||
const BOOTSTRAP_MIGRATION_NAME: &str = "bootstrap";
|
||||
const BOOTSTRAP_MIGRATION_SQL: &str = include_str!("../migrations/V000__bootstrap.sql");
|
||||
const BOOTSTRAP_MIGRATION_VERSION: i64 = 0;
|
||||
const HEX_LOWER: &[u8; 16] = b"0123456789abcdef";
|
||||
const HISTORY_INSERT_SQL: &str = "INSERT INTO ksp_store_schema_migrations (version, name, checksum, applied_at) VALUES ($1, $2, $3, CURRENT_TIMESTAMP)";
|
||||
const HISTORY_LOAD_SQL: &str = "SELECT version, name, checksum FROM ksp_store_schema_migrations ORDER BY version";
|
||||
const LOCK_POLL_INTERVAL_MS: u64 = 25;
|
||||
const METADATA_EXISTS_SQL: &str = r#"SELECT EXISTS (
|
||||
SELECT 1 FROM information_schema.tables
|
||||
WHERE table_schema = current_schema()
|
||||
AND table_name = 'ksp_store_schema_migrations'
|
||||
AND table_type = 'BASE TABLE'
|
||||
)"#;
|
||||
const METADATA_PRIMARY_KEY_SQL: &str = r#"SELECT COUNT(*)::BIGINT,
|
||||
COUNT(*) FILTER (WHERE kcu.column_name = 'version')::BIGINT
|
||||
FROM information_schema.table_constraints tc
|
||||
JOIN information_schema.key_column_usage kcu
|
||||
ON tc.constraint_catalog = kcu.constraint_catalog
|
||||
AND tc.constraint_schema = kcu.constraint_schema
|
||||
AND tc.constraint_name = kcu.constraint_name
|
||||
WHERE tc.table_schema = current_schema()
|
||||
AND tc.table_name = 'ksp_store_schema_migrations'
|
||||
AND tc.constraint_type = 'PRIMARY KEY'"#;
|
||||
const METADATA_SHAPE_SQL: &str = r#"SELECT column_name, data_type, is_nullable
|
||||
FROM information_schema.columns
|
||||
WHERE table_schema = current_schema()
|
||||
AND table_name = 'ksp_store_schema_migrations'
|
||||
ORDER BY ordinal_position"#;
|
||||
const SET_STATEMENT_TIMEOUT_SQL: &str = "SELECT set_config('statement_timeout', $1, true)";
|
||||
|
||||
struct AppliedMigration {
|
||||
checksum: std::string::String,
|
||||
name: std::string::String,
|
||||
version: i64,
|
||||
}
|
||||
|
||||
/// Runs the private bounded PostgreSQL schema bootstrap on one dedicated pooled client.
|
||||
pub(crate) async fn bootstrap(
|
||||
client: &mut deadpool_postgres::Client,
|
||||
auto_migrate: bool,
|
||||
migration_timeout: std::time::Duration,
|
||||
migration_lock_timeout: std::time::Duration,
|
||||
) -> std::result::Result<(), crate::PostgresBackendError> {
|
||||
let bounded = tokio::time::timeout(migration_timeout, bootstrap_inner(client, auto_migrate, migration_timeout, migration_lock_timeout)).await;
|
||||
return match bounded {
|
||||
std::result::Result::Ok(result) => result,
|
||||
std::result::Result::Err(_) => {
|
||||
std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_timeout"))
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async fn bootstrap_inner(
|
||||
client: &mut deadpool_postgres::Client,
|
||||
auto_migrate: bool,
|
||||
migration_timeout: std::time::Duration,
|
||||
migration_lock_timeout: std::time::Duration,
|
||||
) -> std::result::Result<(), crate::PostgresBackendError> {
|
||||
let transaction_result = client.transaction().await;
|
||||
let transaction = match transaction_result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_begin"));
|
||||
},
|
||||
};
|
||||
let lock_result = acquire_advisory_lock(&transaction, migration_lock_timeout).await;
|
||||
if let std::result::Result::Err(error) = lock_result {
|
||||
return std::result::Result::Err(error);
|
||||
}
|
||||
let timeout_result = set_statement_timeout(&transaction, migration_timeout).await;
|
||||
if let std::result::Result::Err(error) = timeout_result {
|
||||
return std::result::Result::Err(error);
|
||||
}
|
||||
let exists_result = metadata_exists(&transaction).await;
|
||||
let metadata_exists = match exists_result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let checksum = bootstrap_checksum();
|
||||
if !metadata_exists {
|
||||
if !auto_migrate {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_pending"));
|
||||
}
|
||||
let create_result = transaction.batch_execute(BOOTSTRAP_MIGRATION_SQL).await;
|
||||
if create_result.is_err() {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_create"));
|
||||
}
|
||||
let shape_result = verify_metadata_shape(&transaction).await;
|
||||
if let std::result::Result::Err(error) = shape_result {
|
||||
return std::result::Result::Err(error);
|
||||
}
|
||||
let insert_result = transaction.execute(HISTORY_INSERT_SQL, &[&BOOTSTRAP_MIGRATION_VERSION, &BOOTSTRAP_MIGRATION_NAME, &checksum]).await;
|
||||
match insert_result {
|
||||
std::result::Result::Ok(1) => {},
|
||||
std::result::Result::Ok(_) | std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_insert"));
|
||||
},
|
||||
}
|
||||
} else {
|
||||
let shape_result = verify_metadata_shape(&transaction).await;
|
||||
if let std::result::Result::Err(error) = shape_result {
|
||||
return std::result::Result::Err(error);
|
||||
}
|
||||
let history_result = load_history(&transaction).await;
|
||||
let history = match history_result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let validation_result = validate_history(history.as_slice(), checksum.as_str());
|
||||
if let std::result::Result::Err(error) = validation_result {
|
||||
return std::result::Result::Err(error);
|
||||
}
|
||||
}
|
||||
let commit_result = transaction.commit().await;
|
||||
return match commit_result {
|
||||
std::result::Result::Ok(()) => std::result::Result::Ok(()),
|
||||
std::result::Result::Err(_) => {
|
||||
std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_commit"))
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async fn acquire_advisory_lock(
|
||||
transaction: &deadpool_postgres::Transaction<'_>,
|
||||
timeout: std::time::Duration,
|
||||
) -> std::result::Result<(), crate::PostgresBackendError> {
|
||||
let started = tokio::time::Instant::now();
|
||||
let deadline = match started.checked_add(timeout) {
|
||||
std::option::Option::Some(value) => value,
|
||||
std::option::Option::None => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock_timeout"));
|
||||
},
|
||||
};
|
||||
loop {
|
||||
let row_result = transaction.query_one("SELECT pg_try_advisory_xact_lock($1)", &[&ADVISORY_LOCK_KEY]).await;
|
||||
let row = match row_result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock"));
|
||||
},
|
||||
};
|
||||
let acquired_result = row.try_get::<usize, bool>(0);
|
||||
let acquired = match acquired_result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock_decode"));
|
||||
},
|
||||
};
|
||||
if acquired {
|
||||
return std::result::Result::Ok(());
|
||||
}
|
||||
let now = tokio::time::Instant::now();
|
||||
if now >= deadline {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock_timeout"));
|
||||
}
|
||||
let candidate = now + std::time::Duration::from_millis(LOCK_POLL_INTERVAL_MS);
|
||||
let wake = if candidate < deadline { candidate } else { deadline };
|
||||
tokio::time::sleep_until(wake).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn set_statement_timeout(
|
||||
transaction: &deadpool_postgres::Transaction<'_>,
|
||||
timeout: std::time::Duration,
|
||||
) -> std::result::Result<(), crate::PostgresBackendError> {
|
||||
let timeout_value = format!("{}ms", timeout.as_millis());
|
||||
let result = transaction.query_one(SET_STATEMENT_TIMEOUT_SQL, &[&timeout_value]).await;
|
||||
return match result {
|
||||
std::result::Result::Ok(_) => std::result::Result::Ok(()),
|
||||
std::result::Result::Err(_) => {
|
||||
std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "statement_timeout"))
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async fn metadata_exists(transaction: &deadpool_postgres::Transaction<'_>) -> std::result::Result<bool, crate::PostgresBackendError> {
|
||||
let result = transaction.query_one(METADATA_EXISTS_SQL, &[]).await;
|
||||
let row = match result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_probe"));
|
||||
},
|
||||
};
|
||||
return match row.try_get::<usize, bool>(0) {
|
||||
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
||||
std::result::Result::Err(_) => {
|
||||
std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_probe_decode"))
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async fn verify_metadata_shape(transaction: &deadpool_postgres::Transaction<'_>) -> std::result::Result<(), crate::PostgresBackendError> {
|
||||
let result = transaction.query(METADATA_SHAPE_SQL, &[]).await;
|
||||
let rows = match result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_shape"));
|
||||
},
|
||||
};
|
||||
let expected = [("version", "bigint", "NO"), ("name", "text", "NO"), ("checksum", "text", "NO"), ("applied_at", "timestamp with time zone", "NO")];
|
||||
let mut found = [false; 4];
|
||||
for row in rows {
|
||||
let column_result = row.try_get::<usize, std::string::String>(0);
|
||||
let data_type_result = row.try_get::<usize, std::string::String>(1);
|
||||
let nullable_result = row.try_get::<usize, std::string::String>(2);
|
||||
let (column, data_type, nullable) = match (column_result, data_type_result, nullable_result) {
|
||||
(std::result::Result::Ok(column), std::result::Result::Ok(data_type), std::result::Result::Ok(nullable)) => (column, data_type, nullable),
|
||||
_ => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_shape_decode"));
|
||||
},
|
||||
};
|
||||
for (index, expected_row) in expected.iter().enumerate() {
|
||||
if column == expected_row.0 {
|
||||
if found[index] || data_type != expected_row.1 || nullable != expected_row.2 {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "metadata_shape"));
|
||||
}
|
||||
found[index] = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
for required in found {
|
||||
if !required {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "metadata_shape"));
|
||||
}
|
||||
}
|
||||
let key_result = transaction.query_one(METADATA_PRIMARY_KEY_SQL, &[]).await;
|
||||
let key_row = match key_result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_primary_key"));
|
||||
},
|
||||
};
|
||||
let key_count = key_row.try_get::<usize, i64>(0);
|
||||
let version_count = key_row.try_get::<usize, i64>(1);
|
||||
return match (key_count, version_count) {
|
||||
(std::result::Result::Ok(1), std::result::Result::Ok(1)) => std::result::Result::Ok(()),
|
||||
(std::result::Result::Ok(_), std::result::Result::Ok(_)) => {
|
||||
std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "metadata_primary_key"))
|
||||
},
|
||||
_ => std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_primary_key_decode")),
|
||||
};
|
||||
}
|
||||
|
||||
async fn load_history(transaction: &deadpool_postgres::Transaction<'_>) -> std::result::Result<std::vec::Vec<AppliedMigration>, crate::PostgresBackendError> {
|
||||
let result = transaction.query(HISTORY_LOAD_SQL, &[]).await;
|
||||
let rows = match result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_load"));
|
||||
},
|
||||
};
|
||||
let mut history = std::vec::Vec::with_capacity(rows.len());
|
||||
for row in rows {
|
||||
let version_result = row.try_get::<usize, i64>(0);
|
||||
let name_result = row.try_get::<usize, std::string::String>(1);
|
||||
let checksum_result = row.try_get::<usize, std::string::String>(2);
|
||||
match (version_result, name_result, checksum_result) {
|
||||
(std::result::Result::Ok(version), std::result::Result::Ok(name), std::result::Result::Ok(checksum)) => {
|
||||
history.push(AppliedMigration { checksum, name, version });
|
||||
},
|
||||
_ => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_decode"));
|
||||
},
|
||||
}
|
||||
}
|
||||
return std::result::Result::Ok(history);
|
||||
}
|
||||
|
||||
fn validate_history(history: &[AppliedMigration], expected_checksum: &str) -> std::result::Result<(), crate::PostgresBackendError> {
|
||||
let mut sentinel_found = false;
|
||||
for applied in history {
|
||||
if applied.version > BOOTSTRAP_MIGRATION_VERSION {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::SchemaNewer, "history_newer"));
|
||||
}
|
||||
if applied.version < BOOTSTRAP_MIGRATION_VERSION {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_unknown"));
|
||||
}
|
||||
if applied.name != BOOTSTRAP_MIGRATION_NAME || applied.checksum != expected_checksum {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_diverged"));
|
||||
}
|
||||
sentinel_found = true;
|
||||
}
|
||||
if !sentinel_found {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_missing"));
|
||||
}
|
||||
return std::result::Result::Ok(());
|
||||
}
|
||||
|
||||
fn bootstrap_checksum() -> std::string::String {
|
||||
let mut hasher = sha2::Sha256::new();
|
||||
hasher.update(BOOTSTRAP_MIGRATION_SQL.as_bytes());
|
||||
let digest = hasher.finalize();
|
||||
let bytes = digest.as_slice();
|
||||
let mut encoded = std::string::String::with_capacity(bytes.len() * 2);
|
||||
for byte in bytes {
|
||||
let value = *byte;
|
||||
encoded.push(char::from(HEX_LOWER[(value >> 4) as usize]));
|
||||
encoded.push(char::from(HEX_LOWER[(value & 0x0f) as usize]));
|
||||
}
|
||||
return encoded;
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "../unit_tests/migration.rs"]
|
||||
mod tests;
|
||||
Reference in New Issue
Block a user