From bf970fdc766c667efeed5ea8ce6ccb7c8c3e87ac Mon Sep 17 00:00:00 2001 From: Eric Wendland Date: Sun, 5 Jul 2026 17:54:02 +0200 Subject: [PATCH] feat: add ordered store migrations --- crates/geth-store/src/lib.rs | 184 ++++++++++++++++++++++++--- docs/architecture.md | 6 + docs/production-readiness-roadmap.md | 10 +- 3 files changed, 178 insertions(+), 22 deletions(-) diff --git a/crates/geth-store/src/lib.rs b/crates/geth-store/src/lib.rs index 0792150..b88d8f4 100644 --- a/crates/geth-store/src/lib.rs +++ b/crates/geth-store/src/lib.rs @@ -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 { + 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::() + .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::, _>>()?; 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::, _>>() + .expect("column names"); + columns.iter().any(|name| name == column) + } + #[test] fn ssh_cert_request_and_revocation_roundtrip() { let store = Store::open_memory().expect("open"); diff --git a/docs/architecture.md b/docs/architecture.md index ff63b3f..3cc298c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -6,6 +6,12 @@ the shared Iroh endpoint, resource registry, module routing, and local control socket. Control commands connect to the Unix socket and send typed JSONL requests. +The local metadata store is SQLite product state. `geth-store` tracks a numeric +`schema_version` in the `meta` table and applies ordered migrations up to the +crate's current schema version when the store opens. Fresh database creation and +repeated opens are idempotent; migrations that change existing schemas run in a +SQLite transaction where SQLite supports it. + Service management is also exposed through the single binary. `geth daemon service ...` installs and controls a user-level service definition for the local daemon. The initial backends are systemd user units on Linux, launchd user agents diff --git a/docs/production-readiness-roadmap.md b/docs/production-readiness-roadmap.md index 452f27f..66f40e1 100644 --- a/docs/production-readiness-roadmap.md +++ b/docs/production-readiness-roadmap.md @@ -120,12 +120,12 @@ and downstream projects. Goal: treat local SQLite state as durable product data before users depend on it. -- `[ ]` Replace opportunistic schema setup with ordered migrations. +- `[x]` Replace opportunistic schema setup with ordered migrations. Acceptance criteria: - - `[ ]` The store tracks numeric schema versions. - - `[ ]` Each migration is transactional where SQLite supports it. - - `[ ]` Fresh database creation and repeated opens are idempotent. - - `[ ]` Old schema fixtures migrate to the current schema in tests. + - `[x]` The store tracks numeric schema versions. + - `[x]` Each migration is transactional where SQLite supports it. + - `[x]` Fresh database creation and repeated opens are idempotent. + - `[x]` Old schema fixtures migrate to the current schema in tests. - `[ ]` Make multi-table writes transactional. Acceptance criteria: