eric/geth
Watch
1
0
Fork
You've already forked geth
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add ordered store migrations

Browse source
This commit is contained in:
Eric Wendland 2026-07-05 17:54:02 +02:00
commit bf970fdc76
3 changed files with 178 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

184
crates/geth-store/src/lib.rs
View file

@ -1,12 +1,16 @@
use rusqlite::{Connection, params};
use rusqlite::{Connection, OptionalExtension, Transaction, params};
use std::path::Path;
pub const CURRENT_SCHEMA_VERSION: i64 = 2;
#[derive(Debug, thiserror::Error)]
pub enum StoreError {
#[error("sqlite error: {0}")]
Sqlite(#[from] rusqlite::Error),
#[error("json error: {0}")]
Json(#[from] serde_json::Error),
#[error("invalid schema version value: {0}")]
InvalidSchemaVersion(String),
}
pub struct Store {
@ -36,6 +40,45 @@ impl Store {
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
"#,
)?;
let mut version = self.schema_version()?;
if version == 0 {
self.apply_schema_v1()?;
version = 1;
}
while version < CURRENT_SCHEMA_VERSION {
let next = version + 1;
match next {
2 => self.apply_migration_v2()?,
_ => unreachable!("missing migration for schema version {next}"),
}
version = next;
}
Ok(())
}
fn schema_version(&self) -> Result<i64, StoreError> {
let version = self
.conn
.query_row(
"SELECT value FROM meta WHERE key = 'schema_version'",
[],
|row| row.get::<_, String>(0),
)
.optional()?;
let Some(version) = version else {
return Ok(0);
};
version
.parse::<i64>()
.map_err(|_| StoreError::InvalidSchemaVersion(version))
}
fn apply_schema_v1(&self) -> Result<(), StoreError> {
let tx = self.conn.unchecked_transaction()?;
tx.execute_batch(
r#"
CREATE TABLE IF NOT EXISTS agents (
agent_id TEXT PRIMARY KEY,
public_key TEXT NOT NULL,
@ -73,7 +116,6 @@ impl Store {
CREATE TABLE IF NOT EXISTS keychain_signatures (
op_id TEXT NOT NULL,
signer TEXT NOT NULL,
signer_public_key TEXT NOT NULL DEFAULT '',
namespace TEXT NOT NULL,
signature BLOB NOT NULL,
created_at_ms INTEGER NOT NULL,
@ -202,16 +244,14 @@ impl Store {
renewal_of TEXT,
reason TEXT,
status TEXT NOT NULL,
created_at_ms INTEGER NOT NULL,
provenance_json TEXT
created_at_ms INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS ssh_certificates (
cert_id TEXT PRIMARY KEY,
request_id TEXT NOT NULL,
certificate TEXT NOT NULL,
certificate_fingerprint TEXT NOT NULL,
imported_at_ms INTEGER NOT NULL,
provenance_json TEXT
imported_at_ms INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS ssh_revocations (
revocation_id TEXT PRIMARY KEY,
@ -219,35 +259,52 @@ impl Store {
target TEXT NOT NULL,
reason TEXT,
created_at_ms INTEGER NOT NULL,
published INTEGER NOT NULL DEFAULT 0,
provenance_json TEXT
published INTEGER NOT NULL DEFAULT 0
);
INSERT OR IGNORE INTO meta(key, value) VALUES ('schema_version', '1');
"#,
)?;
self.add_column_if_missing(
self.set_schema_version_tx(&tx, 1)?;
tx.commit()?;
Ok(())
}
fn apply_migration_v2(&self) -> Result<(), StoreError> {
let tx = self.conn.unchecked_transaction()?;
Self::add_column_if_missing_tx(
&tx,
"keychain_signatures",
"signer_public_key",
"TEXT NOT NULL DEFAULT ''",
)?;
self.add_column_if_missing("ssh_cert_requests", "provenance_json", "TEXT")?;
self.add_column_if_missing("ssh_certificates", "provenance_json", "TEXT")?;
self.add_column_if_missing("ssh_revocations", "provenance_json", "TEXT")?;
Self::add_column_if_missing_tx(&tx, "ssh_cert_requests", "provenance_json", "TEXT")?;
Self::add_column_if_missing_tx(&tx, "ssh_certificates", "provenance_json", "TEXT")?;
Self::add_column_if_missing_tx(&tx, "ssh_revocations", "provenance_json", "TEXT")?;
self.set_schema_version_tx(&tx, 2)?;
tx.commit()?;
Ok(())
}
fn add_column_if_missing(
&self,
fn set_schema_version_tx(&self, tx: &Transaction<'_>, version: i64) -> Result<(), StoreError> {
tx.execute(
"INSERT INTO meta(key, value) VALUES ('schema_version', ?1)
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
[version.to_string()],
)?;
Ok(())
}
fn add_column_if_missing_tx(
tx: &Transaction<'_>,
table: &str,
column: &str,
definition: &str,
) -> Result<(), StoreError> {
let mut stmt = self.conn.prepare(&format!("PRAGMA table_info({table})"))?;
let mut stmt = tx.prepare(&format!("PRAGMA table_info({table})"))?;
let columns = stmt
.query_map([], |row| row.get::<_, String>(1))?
.collect::<Result<Vec<_>, _>>()?;
if !columns.iter().any(|name| name == column) {
self.conn.execute(
tx.execute(
&format!("ALTER TABLE {table} ADD COLUMN {column} {definition}"),
[],
)?;
@ -1616,10 +1673,103 @@ mod tests {
let store = Store::open_memory().expect("open");
store.migrate().expect("migrate again");
store.migrate().expect("migrate third time");
assert_eq!(
store.schema_version().expect("schema version"),
CURRENT_SCHEMA_VERSION
);
let resources = store.list_resources().expect("resources");
assert!(resources.is_empty());
}
#[test]
fn schema_v1_fixture_migrates_to_current_schema() {
let path =
std::env::temp_dir().join(format!("geth-store-v1-migration-{}.sqlite", now_ms()));
{
let conn = Connection::open(&path).expect("open fixture");
conn.execute_batch(
r#"
CREATE TABLE meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
INSERT INTO meta(key, value) VALUES ('schema_version', '1');
CREATE TABLE keychain_signatures (
op_id TEXT NOT NULL,
signer TEXT NOT NULL,
namespace TEXT NOT NULL,
signature BLOB NOT NULL,
created_at_ms INTEGER NOT NULL,
PRIMARY KEY (op_id, signer, namespace)
);
CREATE TABLE ssh_cert_requests (
request_id TEXT PRIMARY KEY,
requester_node TEXT NOT NULL,
public_key TEXT NOT NULL,
public_key_fingerprint TEXT NOT NULL,
cert_kind TEXT NOT NULL,
principals_json TEXT NOT NULL,
requested_validity TEXT,
renewal_of TEXT,
reason TEXT,
status TEXT NOT NULL,
created_at_ms INTEGER NOT NULL
);
CREATE TABLE ssh_certificates (
cert_id TEXT PRIMARY KEY,
request_id TEXT NOT NULL,
certificate TEXT NOT NULL,
certificate_fingerprint TEXT NOT NULL,
imported_at_ms INTEGER NOT NULL
);
CREATE TABLE ssh_revocations (
revocation_id TEXT PRIMARY KEY,
kind TEXT NOT NULL,
target TEXT NOT NULL,
reason TEXT,
created_at_ms INTEGER NOT NULL,
published INTEGER NOT NULL DEFAULT 0
);
"#,
)
.expect("create v1 fixture");
}
let store = Store::open(&path).expect("migrate fixture");
assert_eq!(
store.schema_version().expect("schema version"),
CURRENT_SCHEMA_VERSION
);
assert!(column_exists(
&store,
"keychain_signatures",
"signer_public_key"
));
assert!(column_exists(
&store,
"ssh_cert_requests",
"provenance_json"
));
assert!(column_exists(&store, "ssh_certificates", "provenance_json"));
assert!(column_exists(&store, "ssh_revocations", "provenance_json"));
drop(store);
let _ = std::fs::remove_file(path);
}
fn column_exists(store: &Store, table: &str, column: &str) -> bool {
let mut stmt = store
.conn
.prepare(&format!("PRAGMA table_info({table})"))
.expect("table info");
let columns = stmt
.query_map([], |row| row.get::<_, String>(1))
.expect("columns")
.collect::<Result<Vec<_>, _>>()
.expect("column names");
columns.iter().any(|name| name == column)
}
#[test]
fn ssh_cert_request_and_revocation_roundtrip() {
let store = Store::open_memory().expect("open");