From ec3933d5de741edbba9ccf91227598290aaa24df Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 11 Jul 2026 14:25:14 -0400 Subject: [PATCH] Move MCP servers to SQLite storage --- Cargo.lock | 6 + .../2026-07-11-mcp-servers-sqlite-plan.md | 70 ++ docs/public/agents/mcp.mdx | 15 + .../migrations/2026071102_mcp_servers.sql | 73 ++ lib/crates/fabro-db/tests/sqlite.rs | 157 +++ lib/crates/fabro-mcp-store/Cargo.toml | 6 + lib/crates/fabro-mcp-store/src/error.rs | 48 +- lib/crates/fabro-mcp-store/src/lib.rs | 11 +- lib/crates/fabro-mcp-store/src/model.rs | 34 +- lib/crates/fabro-mcp-store/src/store.rs | 933 +++++++++++------- lib/crates/fabro-mcp-store/tests/store.rs | 375 +++++++ lib/crates/fabro-server/src/server.rs | 25 +- .../src/server/handler/mcp_servers.rs | 11 +- lib/crates/fabro-server/src/server/tests.rs | 36 +- .../fabro-server/tests/it/api/mcp_servers.rs | 154 ++- lib/crates/fabro-types/src/mcp_store.rs | 52 +- lib/crates/fabro-types/src/settings/run.rs | 22 +- 17 files changed, 1610 insertions(+), 418 deletions(-) create mode 100644 docs/plans/2026-07-11-mcp-servers-sqlite-plan.md create mode 100644 lib/crates/fabro-db/migrations/2026071102_mcp_servers.sql create mode 100644 lib/crates/fabro-mcp-store/tests/store.rs diff --git a/Cargo.lock b/Cargo.lock index c66de1be7..1b8b7c8ba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2846,12 +2846,18 @@ dependencies = [ name = "fabro-mcp-store" version = "0.302.0-nightly.1" dependencies = [ + "chrono", + "fabro-db", "fabro-types", "serde", + "serde_json", + "sqlx", + "strum 0.28.0", "tempfile", "thiserror 2.0.18", "tokio", "toml 0.8.23", + "tracing", ] [[package]] diff --git a/docs/plans/2026-07-11-mcp-servers-sqlite-plan.md b/docs/plans/2026-07-11-mcp-servers-sqlite-plan.md new file mode 100644 index 000000000..66179f999 --- /dev/null +++ b/docs/plans/2026-07-11-mcp-servers-sqlite-plan.md @@ -0,0 +1,70 @@ +# MCP Servers to SQLite + +## Goal + +Move server-managed MCP definitions from sibling `mcps/*.toml` files into the +shared SQLite database. Preserve the REST contract, value-omitting read views, +content-hash ETags, synchronous manifest catalog reads, and all three transport +types. + +## Schema + +Use one `mcp_servers` row per definition: + +- Scalar columns: id, revision, display name, description, transport type, + protocol, URL, port, and timeouts. +- Typed JSON columns: command array, env map, and header map. +- A transport-shape constraint requires exactly the columns belonging to + `stdio`, `http`, or `sandbox`. +- ID, revision, protocol, port, timeout, and top-level JSON shape constraints + provide database-level defense in depth. +- No secondary indexes: supported queries are primary-key lookup and full + sorted listing. + +Keep `McpServerRevision` as lowercase SHA-256 hex. Create and replace derive it +from the existing canonical representation. Legacy import preserves the hash +of the original TOML bytes so an ETag remains valid across upgrade. + +## Store + +- `fabro-mcp-store` owns SQL, row mapping, typed JSON encoding, validation, + revisions, caching, and legacy import. +- Retain the synchronous in-memory catalog required by manifest resolution. +- Serialize in-process mutations, then enforce replace/delete revisions in SQL + with `WHERE id = ? AND revision = ?`. +- Update the cache only after a successful transaction. +- Revalidate every decoded row through the existing domain model. +- Encode env/header maps through sorted maps for deterministic JSON. +- Never log or debug-print transport env/header values. + +## Legacy import + +At startup, inspect `mcps/` next to the active `settings.toml`: + +1. Missing directory: no-op. +2. Parse and validate every TOML definition before mutating SQLite. +3. Insert all definitions in one transaction with + `ON CONFLICT(id) DO NOTHING`; SQLite wins. +4. Commit, then rename the directory to + `mcps.imported-.bak`. +5. If backup rename fails, leave the source directory for a retry; the next + import skips existing SQLite rows and retries the rename. + +Logs contain only paths, counts, and MCP ids. They never contain commands, +URLs, env values, or headers. + +## Tests + +- Schema accepts every transport and rejects invalid variant shapes. +- CRUD, sorted listing, reload persistence, and all transport round trips. +- Independent store instances enforce SQL revision conflicts. +- Deterministic map JSON and typed corrupted-row errors. +- Import success, SQLite precedence, stable imported revision, retry no-op, + malformed input unchanged, and directory backup. +- API persistence/restart behavior, value-omitting reads, legacy startup import, + malformed legacy startup failure, and existing auth requirements. +- Workspace build, formatter, Clippy, and relevant Nextest suites. + +## Unresolved questions + +None. diff --git a/docs/public/agents/mcp.mdx b/docs/public/agents/mcp.mdx index b3fda877c..76fb68054 100644 --- a/docs/public/agents/mcp.mdx +++ b/docs/public/agents/mcp.mdx @@ -119,6 +119,21 @@ MCP servers available to Fabro agents can be configured in two places: Each server entry specifies a transport type and optional timeouts. The server name is the TOML table key and is used in qualified tool names. +### Server-managed catalog + +Fabro servers can also manage a shared MCP catalog through the MCP servers REST API. Workflows reference a catalog definition by id instead of repeating its transport configuration: + +```toml +[run.agent.mcps.sentry] +id = "sentry" +``` + +Server-managed definitions are stored in the server's shared SQLite database. Read APIs return configured env/header names but never their values. + +When upgrading an installation that stored definitions as `mcps/*.toml` next to the active `settings.toml`, Fabro validates and imports the directory during startup. Existing SQLite rows win on id conflicts. After a successful transaction, Fabro renames the source directory to a timestamped backup such as `mcps.imported-20260711T120000000000Z.bak`. + +Transport env/header values preserve their existing plaintext-at-rest behavior in SQLite. Prefer `{{ secrets.NAME }}` interpolation over literal credentials where possible. + ## Transports ### Stdio diff --git a/lib/crates/fabro-db/migrations/2026071102_mcp_servers.sql b/lib/crates/fabro-db/migrations/2026071102_mcp_servers.sql new file mode 100644 index 000000000..e5e2d009e --- /dev/null +++ b/lib/crates/fabro-db/migrations/2026071102_mcp_servers.sql @@ -0,0 +1,73 @@ +CREATE TABLE mcp_servers ( + id TEXT PRIMARY KEY NOT NULL, + revision TEXT NOT NULL, + display_name TEXT NOT NULL, + description TEXT, + transport_type TEXT NOT NULL, + protocol TEXT, + command_json TEXT, + url TEXT, + port INTEGER, + env_json TEXT, + headers_json TEXT, + startup_timeout_secs INTEGER NOT NULL, + tool_timeout_secs INTEGER NOT NULL, + CHECK (length(id) BETWEEN 1 AND 63), + CHECK (substr(id, 1, 1) GLOB '[a-z0-9]'), + CHECK (id NOT GLOB '*[^a-z0-9-]*'), + CHECK (length(revision) = 64), + CHECK (revision NOT GLOB '*[^0-9a-f]*'), + CHECK (length(trim(display_name)) > 0), + CHECK (transport_type IN ('stdio', 'http', 'sandbox')), + CHECK (protocol IS NULL OR protocol IN ('streamable_http', 'sse')), + CHECK (startup_timeout_secs >= 0), + CHECK (tool_timeout_secs >= 0), + CHECK ( + ( + transport_type = 'stdio' + AND protocol IS NULL + AND command_json IS NOT NULL + AND json_valid(command_json) + AND json_type(command_json) = 'array' + AND json_array_length(command_json) > 0 + AND json_type(command_json, '$[0]') = 'text' + AND length(trim(json_extract(command_json, '$[0]'))) > 0 + AND url IS NULL + AND port IS NULL + AND env_json IS NOT NULL + AND json_valid(env_json) + AND json_type(env_json) = 'object' + AND headers_json IS NULL + ) + OR + ( + transport_type = 'http' + AND protocol IS NOT NULL + AND command_json IS NULL + AND url IS NOT NULL + AND length(trim(url)) > 0 + AND port IS NULL + AND env_json IS NULL + AND headers_json IS NOT NULL + AND json_valid(headers_json) + AND json_type(headers_json) = 'object' + ) + OR + ( + transport_type = 'sandbox' + AND protocol IS NOT NULL + AND command_json IS NOT NULL + AND json_valid(command_json) + AND json_type(command_json) = 'array' + AND json_array_length(command_json) > 0 + AND json_type(command_json, '$[0]') = 'text' + AND length(trim(json_extract(command_json, '$[0]'))) > 0 + AND url IS NULL + AND port BETWEEN 1 AND 65535 + AND env_json IS NOT NULL + AND json_valid(env_json) + AND json_type(env_json) = 'object' + AND headers_json IS NULL + ) + ) +); diff --git a/lib/crates/fabro-db/tests/sqlite.rs b/lib/crates/fabro-db/tests/sqlite.rs index 53fc0d803..60c371ffa 100644 --- a/lib/crates/fabro-db/tests/sqlite.rs +++ b/lib/crates/fabro-db/tests/sqlite.rs @@ -49,6 +49,13 @@ async fn connect_creates_parent_directory_and_migrate_is_idempotent() -> anyhow: .await?; assert_eq!(secrets_table_count, 1); + let mcp_servers_table_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'mcp_servers'", + ) + .fetch_one(database.pool()) + .await?; + assert_eq!(mcp_servers_table_count, 1); + let legacy_import_table_count: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'legacy_imports'", ) @@ -65,6 +72,156 @@ async fn connect_creates_parent_directory_and_migrate_is_idempotent() -> anyhow: Ok(()) } +#[tokio::test] +async fn mcp_servers_schema_rejects_invalid_transport_rows() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; + database.migrate().await?; + + insert_mcp_server( + database.pool(), + "stdio", + "stdio", + None, + Some(r#"["server"]"#), + None, + None, + Some("{}"), + None, + ) + .await?; + insert_mcp_server( + database.pool(), + "http", + "http", + Some("streamable_http"), + None, + Some("https://example.com/mcp"), + None, + None, + Some("{}"), + ) + .await?; + insert_mcp_server( + database.pool(), + "sandbox", + "sandbox", + Some("sse"), + Some(r#"["server"]"#), + None, + Some(3000), + Some("{}"), + None, + ) + .await?; + + for result in [ + insert_mcp_server( + database.pool(), + "bad-id_", + "stdio", + None, + Some(r#"["server"]"#), + None, + None, + Some("{}"), + None, + ) + .await, + insert_mcp_server( + database.pool(), + "empty-command", + "stdio", + None, + Some("[]"), + None, + None, + Some("{}"), + None, + ) + .await, + insert_mcp_server( + database.pool(), + "http-with-env", + "http", + Some("streamable_http"), + None, + Some("https://example.com/mcp"), + None, + Some("{}"), + Some("{}"), + ) + .await, + insert_mcp_server( + database.pool(), + "sandbox-port", + "sandbox", + Some("streamable_http"), + Some(r#"["server"]"#), + None, + Some(65_536), + Some("{}"), + None, + ) + .await, + ] { + assert!(result.is_err(), "invalid MCP server row should be rejected"); + } + + Ok(()) +} + +#[expect( + clippy::too_many_arguments, + reason = "schema test helper mirrors the mutually exclusive transport columns" +)] +async fn insert_mcp_server( + pool: &fabro_db::DbPool, + id: &str, + transport_type: &str, + protocol: Option<&str>, + command_json: Option<&str>, + url: Option<&str>, + port: Option, + env_json: Option<&str>, + headers_json: Option<&str>, +) -> Result<(), sqlx::Error> { + sqlx::query( + r" + INSERT INTO mcp_servers ( + id, + revision, + display_name, + transport_type, + protocol, + command_json, + url, + port, + env_json, + headers_json, + startup_timeout_secs, + tool_timeout_secs + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ", + ) + .bind(id) + .bind("a".repeat(64)) + .bind("MCP Server") + .bind(transport_type) + .bind(protocol) + .bind(command_json) + .bind(url) + .bind(port) + .bind(env_json) + .bind(headers_json) + .bind(10_i64) + .bind(60_i64) + .execute(pool) + .await?; + Ok(()) +} + #[tokio::test] async fn environments_schema_rejects_invalid_rows() -> anyhow::Result<()> { let dir = tempfile::tempdir()?; diff --git a/lib/crates/fabro-mcp-store/Cargo.toml b/lib/crates/fabro-mcp-store/Cargo.toml index 2fe21ece3..0172d9bae 100644 --- a/lib/crates/fabro-mcp-store/Cargo.toml +++ b/lib/crates/fabro-mcp-store/Cargo.toml @@ -13,11 +13,17 @@ doctest = false workspace = true [dependencies] +chrono.workspace = true +fabro-db = { path = "../fabro-db" } fabro-types = { path = "../fabro-types" } serde.workspace = true +serde_json.workspace = true +sqlx.workspace = true +strum.workspace = true thiserror.workspace = true tokio.workspace = true toml.workspace = true +tracing.workspace = true [dev-dependencies] tempfile = "3" diff --git a/lib/crates/fabro-mcp-store/src/error.rs b/lib/crates/fabro-mcp-store/src/error.rs index f4956b18d..9949fe9b5 100644 --- a/lib/crates/fabro-mcp-store/src/error.rs +++ b/lib/crates/fabro-mcp-store/src/error.rs @@ -21,6 +21,38 @@ pub enum McpServerStoreError { #[from] source: McpServerValidationError, }, + #[error("mcp server database error")] + Db { + #[from] + source: sqlx::Error, + }, + #[error("stored mcp server {id} has invalid revision")] + StoredRevision { + id: McpServerId, + #[source] + source: fabro_types::McpServerRevisionParseError, + }, + #[error("stored mcp server {id} has invalid transport: {reason}")] + StoredTransport { id: McpServerId, reason: String }, + #[error("stored mcp server {id} has invalid {column} value {value}")] + StoredInteger { + id: McpServerId, + column: &'static str, + value: i64, + }, + #[error("encoding mcp server {field} as JSON")] + JsonEncode { + field: &'static str, + #[source] + source: serde_json::Error, + }, + #[error("decoding stored mcp server {id}.{field} JSON")] + JsonDecode { + id: McpServerId, + field: &'static str, + #[source] + source: serde_json::Error, + }, #[error("invalid mcp server filename at {path:?}")] InvalidFilename { path: PathBuf, reason: String }, #[error("failed to parse mcp server TOML at {path:?}")] @@ -46,6 +78,13 @@ pub enum McpServerStoreError { #[source] source: std::io::Error, }, + #[error("renaming legacy mcp server directory {source_path:?} to backup {backup_path:?}")] + LegacyBackup { + source_path: PathBuf, + backup_path: PathBuf, + #[source] + source: std::io::Error, + }, } impl McpServerStoreError { @@ -77,10 +116,15 @@ impl McpServerStoreError { Self::AlreadyExists { .. } => "already_exists", Self::StaleRevision { .. } => "stale_revision", Self::Validation { .. } => "validation", + Self::Db { .. } => "database", + Self::StoredRevision { .. } + | Self::StoredTransport { .. } + | Self::StoredInteger { .. } + | Self::JsonDecode { .. } => "stored_data", + Self::JsonEncode { .. } | Self::Serialize { .. } => "serialize", Self::InvalidFilename { .. } => "invalid_filename", Self::Parse { .. } | Self::InvalidUtf8 { .. } => "parse", - Self::Serialize { .. } => "serialize", - Self::Io { .. } => "io", + Self::Io { .. } | Self::LegacyBackup { .. } => "io", } } } diff --git a/lib/crates/fabro-mcp-store/src/lib.rs b/lib/crates/fabro-mcp-store/src/lib.rs index 7000a5fc5..793c84ae4 100644 --- a/lib/crates/fabro-mcp-store/src/lib.rs +++ b/lib/crates/fabro-mcp-store/src/lib.rs @@ -1,14 +1,13 @@ //! Durable storage for server-managed MCP server definitions. //! -//! Concrete [`McpServerStore`] modeled on `fabro-automation`'s -//! `AutomationStore`: per-file TOML under `{config}/mcps/{id}.toml`, in-memory -//! cache, SHA-256 revision for optimistic concurrency, and async -//! storage-agnostic methods. The domain model lives in `fabro-types`; this -//! crate owns persistence. +//! Concrete SQLite-backed [`McpServerStore`] with a synchronous catalog cache, +//! SHA-256 revisions for optimistic concurrency, and one-time import from the +//! legacy per-file TOML directory. The domain model lives in `fabro-types`; +//! this crate owns persistence. mod error; mod model; mod store; pub use error::McpServerStoreError; -pub use store::McpServerStore; +pub use store::{ImportReport, McpServerStore, import_legacy_directory_once}; diff --git a/lib/crates/fabro-mcp-store/src/model.rs b/lib/crates/fabro-mcp-store/src/model.rs index 5ae656b2b..c467abb60 100644 --- a/lib/crates/fabro-mcp-store/src/model.rs +++ b/lib/crates/fabro-mcp-store/src/model.rs @@ -3,8 +3,8 @@ //! The domain types (`McpServerDefinition`, `McpServerDraft`, //! `McpServerReplace`, `McpServerId`, `McpServerRevision`) live in //! `fabro-types` so they stay persistence-independent. This module owns the -//! store-side glue: validating, serializing to canonical TOML bytes, deriving -//! the revision, and reconstructing definitions from persisted bytes. +//! store-side glue for validating definitions, deriving revisions from +//! canonical TOML bytes, and reconstructing definitions during legacy import. use std::path::PathBuf; @@ -12,12 +12,14 @@ use fabro_types::settings::McpTransport; use fabro_types::{ McpServerDefinition, McpServerId, McpServerReplace, McpServerRevision, mcp_store, }; +use serde::de::Error as _; use serde::{Deserialize, Serialize}; +use toml::de::Error as TomlDeError; use crate::error::McpServerStoreError; -/// The on-disk body of a definition. Excludes `id`/`revision`, which are -/// derived from the filename and content hash rather than persisted. +/// The legacy TOML body of a definition. It excludes `id`/`revision`, which old +/// installations derived from the filename and content hash. #[derive(Debug, Clone, PartialEq, Deserialize)] #[serde(deny_unknown_fields)] struct PersistedMcpServer { @@ -63,11 +65,10 @@ impl From for McpServerReplace { } } -/// Build a definition + its canonical persisted bytes from a replace payload. +/// Build a definition + its canonical revision bytes from a replace payload. /// /// The revision is the SHA-256 of the freshly serialized canonical bytes, so a -/// caller can compare it to the on-disk content hash for optimistic -/// concurrency. +/// caller can use it for optimistic concurrency. pub(crate) fn definition_from_replace( id: McpServerId, replace: McpServerReplace, @@ -94,6 +95,17 @@ pub(crate) fn definition_from_persisted_path( Ok(assemble(id, revision, replace)) } +/// Reconstruct a definition from normalized durable fields, revalidating the +/// same domain invariants enforced for API writes and legacy TOML imports. +pub(crate) fn definition_from_stored_parts( + id: McpServerId, + revision: McpServerRevision, + replace: McpServerReplace, +) -> Result { + mcp_store::validate_mcp_server_fields(&replace)?; + Ok(assemble(id, revision, replace)) +} + fn assemble( id: McpServerId, revision: McpServerRevision, @@ -119,5 +131,11 @@ pub(crate) fn canonical_bytes(replace: &McpServerReplace) -> Result, Mcp fn parse_persisted(bytes: &[u8], path: PathBuf) -> Result { let content = std::str::from_utf8(bytes) .map_err(|err| McpServerStoreError::invalid_utf8(path.clone(), err))?; - toml::from_str(content).map_err(|err| McpServerStoreError::parse(path, err)) + toml::from_str(content).map_err(|err| { + // TOML parse errors retain and display the source line by default. + // MCP transport lines may contain literal credentials, so retain only + // the parser's safe reason and discard its source-text context. + let safe = TomlDeError::custom(err.message()); + McpServerStoreError::parse(path, safe) + }) } diff --git a/lib/crates/fabro-mcp-store/src/store.rs b/lib/crates/fabro-mcp-store/src/store.rs index 7ba18c2e0..0b3c025f9 100644 --- a/lib/crates/fabro-mcp-store/src/store.rs +++ b/lib/crates/fabro-mcp-store/src/store.rs @@ -1,48 +1,90 @@ -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; +use std::ffi::OsString; use std::io::ErrorKind; use std::path::{Path, PathBuf}; +use std::str::FromStr as _; use std::sync::RwLock; -use std::time::{SystemTime, UNIX_EPOCH}; -use fabro_types::settings::run::McpServerSettings; +use chrono::{DateTime, Utc}; +use fabro_db::DbPool; +use fabro_types::settings::run::{McpHttpProtocol, McpServerSettings, McpTransport}; use fabro_types::{ McpServerDefinition, McpServerDraft, McpServerId, McpServerReplace, McpServerRevision, - McpServerView, + McpServerValidationError, McpServerView, }; +use serde::de::DeserializeOwned; +use sqlx::query::Query; +use sqlx::sqlite::{SqliteArguments, SqliteRow}; +use sqlx::{Row as _, Sqlite, Transaction}; +use strum::{Display, EnumString, IntoStaticStr}; use tokio::fs; -use tokio::io::AsyncWriteExt as _; use tokio::sync::Mutex; +use tracing::info; use crate::error::McpServerStoreError; use crate::model; -/// Durable per-file TOML store for server-managed MCP server definitions. +/// SQLite-backed durable store for server-managed MCP server definitions. /// -/// Concrete by design (no trait): a future FS→SQL move is a one-time migration, -/// not a runtime backend choice. The method surface keeps callers storage- -/// agnostic; only [`McpServerStore::load`] knows about the filesystem. -#[derive(Debug)] +/// Reads use a synchronous in-memory catalog because manifest resolution is +/// synchronous. Mutations are serialized within this process and use +/// revision-guarded SQL so SQLite remains authoritative for concurrency. pub struct McpServerStore { - dir: PathBuf, + pool: DbPool, mutations: Mutex<()>, defs: RwLock>, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ImportReport { + pub source_path: PathBuf, + pub backup_path: PathBuf, + pub imported_rows: usize, + pub skipped_rows: usize, + pub mcp_server_ids: Vec, +} + +#[derive(Debug, Clone, Copy, Display, EnumString, IntoStaticStr)] +#[strum(serialize_all = "snake_case")] +enum TransportType { + Stdio, + Http, + Sandbox, +} + +impl TransportType { + fn as_str(self) -> &'static str { + self.into() + } +} + +impl std::fmt::Debug for McpServerStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("McpServerStore").finish_non_exhaustive() + } +} + impl McpServerStore { - /// Synchronously load every persisted definition in `dir`. Returns an error - /// if any file fails to parse or validate; the caller decides startup - /// failure policy. Synchronous because it runs once at construction time - /// (typically during server startup) and is invoked from non-async code. - pub fn load(dir: impl Into) -> Result { - let dir = dir.into(); - let defs = load_definitions(&dir)?; + /// Load every persisted definition and build the synchronous catalog. + pub async fn load(pool: DbPool) -> Result { + let defs = load_definitions(&pool).await?; Ok(Self { - dir, + pool, mutations: Mutex::new(()), defs: RwLock::new(defs), }) } + /// Import a legacy `mcps/*.toml` directory when present, then load the + /// SQLite-backed catalog. + pub async fn open( + pool: DbPool, + legacy_dir: impl AsRef, + ) -> Result { + import_legacy_directory_once(&pool, legacy_dir).await?; + Self::load(pool).await + } + fn read_defs( &self, ) -> std::sync::RwLockReadGuard<'_, HashMap> { @@ -76,9 +118,8 @@ impl McpServerStore { .collect() } - /// Sorted ids only, without cloning the (potentially sensitive) env/header - /// maps carried by full definitions. Used by missing-reference errors to - /// list available ids cheaply. + /// Sorted ids only, without cloning the potentially sensitive env/header + /// maps carried by full definitions. pub fn ids(&self) -> Vec { let defs = self.read_defs(); let mut ids = defs.keys().cloned().collect::>(); @@ -95,19 +136,14 @@ impl McpServerStore { draft: McpServerDraft, ) -> Result { let (id, replace) = draft.into(); + let (definition, _) = model::definition_from_replace(id.clone(), replace)?; let _mutation = self.mutations.lock().await; - if self.read_defs().contains_key(&id) { + let mut transaction = self.pool.begin().await?; + if !insert_definition_ignoring_conflict(&mut transaction, &definition).await? { return Err(McpServerStoreError::AlreadyExists { id }); } - let (definition, bytes) = model::definition_from_replace(id.clone(), replace)?; - - let path = definition_path(&self.dir, &id); - write_new(&self.dir, &path, &bytes) - .await - .map_err(|err| create_error_for(id.clone(), err))?; - - let mut defs = self.write_defs(); - defs.insert(id, definition.clone()); + transaction.commit().await?; + self.write_defs().insert(id, definition.clone()); Ok(definition) } @@ -117,16 +153,12 @@ impl McpServerStore { expected: &McpServerRevision, replace: McpServerReplace, ) -> Result { + let (definition, _) = model::definition_from_replace(id.clone(), replace)?; let _mutation = self.mutations.lock().await; - { - let defs = self.read_defs(); - check_revision(&defs, id, expected)?; - } - let (definition, bytes) = model::definition_from_replace(id.clone(), replace)?; - - write_atomic(&self.dir, &definition_path(&self.dir, id), &bytes).await?; - let mut defs = self.write_defs(); - defs.insert(id.clone(), definition.clone()); + let mut transaction = self.pool.begin().await?; + update_definition(&mut transaction, &definition, expected).await?; + transaction.commit().await?; + self.write_defs().insert(id.clone(), definition.clone()); Ok(definition) } @@ -136,39 +168,218 @@ impl McpServerStore { expected: &McpServerRevision, ) -> Result<(), McpServerStoreError> { let _mutation = self.mutations.lock().await; - { - let defs = self.read_defs(); - check_revision(&defs, id, expected)?; + let mut transaction = self.pool.begin().await?; + let result = sqlx::query("DELETE FROM mcp_servers WHERE id = ? AND revision = ?") + .bind(id.as_str()) + .bind(expected.as_str()) + .execute(&mut *transaction) + .await?; + if result.rows_affected() == 0 { + return Err(revision_mismatch_error(&mut transaction, id, expected).await?); } - - let path = definition_path(&self.dir, id); - fs::remove_file(&path) - .await - .map_err(|err| McpServerStoreError::io(path, err))?; - let mut defs = self.write_defs(); - defs.remove(id); + transaction.commit().await?; + self.write_defs().remove(id); Ok(()) } } -fn check_revision( - defs: &HashMap, +async fn load_definitions( + pool: &DbPool, +) -> Result, McpServerStoreError> { + let rows = sqlx::query( + r" + SELECT + id, + revision, + display_name, + description, + transport_type, + protocol, + command_json, + url, + port, + env_json, + headers_json, + startup_timeout_secs, + tool_timeout_secs + FROM mcp_servers + ORDER BY id + ", + ) + .fetch_all(pool) + .await?; + + rows.iter() + .map(definition_from_row) + .map(|result| result.map(|definition| (definition.id.clone(), definition))) + .collect() +} + +fn definition_from_row(row: &SqliteRow) -> Result { + let id = McpServerId::new(row.try_get::("id")?)?; + let revision_text = row.try_get::("revision")?; + let revision = McpServerRevision::from_str(&revision_text).map_err(|source| { + McpServerStoreError::StoredRevision { + id: id.clone(), + source, + } + })?; + let transport_type_text = row.try_get::("transport_type")?; + let transport_type = TransportType::from_str(&transport_type_text).map_err(|_| { + McpServerStoreError::StoredTransport { + id: id.clone(), + reason: format!("unknown transport type {transport_type_text:?}"), + } + })?; + let protocol = row.try_get::, _>("protocol")?; + let command_json = row.try_get::, _>("command_json")?; + let url = row.try_get::, _>("url")?; + let port = row.try_get::, _>("port")?; + let env_json = row.try_get::, _>("env_json")?; + let headers_json = row.try_get::, _>("headers_json")?; + let transport = match transport_type { + TransportType::Stdio => { + require_absent(&id, "protocol", protocol.as_ref())?; + require_absent(&id, "url", url.as_ref())?; + require_absent(&id, "port", port.as_ref())?; + require_absent(&id, "headers_json", headers_json.as_ref())?; + McpTransport::Stdio { + command: decode_json( + &id, + "command_json", + &require_field(&id, "command_json", command_json)?, + )?, + env: decode_string_map( + &id, + "env_json", + &require_field(&id, "env_json", env_json)?, + )?, + } + } + TransportType::Http => { + require_absent(&id, "command_json", command_json.as_ref())?; + require_absent(&id, "port", port.as_ref())?; + require_absent(&id, "env_json", env_json.as_ref())?; + McpTransport::Http { + protocol: parse_protocol(&id, require_field(&id, "protocol", protocol)?.as_str())?, + url: require_field(&id, "url", url)?, + headers: decode_string_map( + &id, + "headers_json", + &require_field(&id, "headers_json", headers_json)?, + )?, + } + } + TransportType::Sandbox => { + require_absent(&id, "url", url.as_ref())?; + require_absent(&id, "headers_json", headers_json.as_ref())?; + McpTransport::Sandbox { + protocol: parse_protocol(&id, require_field(&id, "protocol", protocol)?.as_str())?, + command: decode_json( + &id, + "command_json", + &require_field(&id, "command_json", command_json)?, + )?, + port: decode_port(&id, require_field(&id, "port", port)?)?, + env: decode_string_map( + &id, + "env_json", + &require_field(&id, "env_json", env_json)?, + )?, + } + } + }; + let replace = McpServerReplace { + display_name: row.try_get("display_name")?, + description: row.try_get("description")?, + transport, + startup_timeout_secs: decode_timeout( + &id, + "startup_timeout_secs", + row.try_get("startup_timeout_secs")?, + )?, + tool_timeout_secs: decode_timeout( + &id, + "tool_timeout_secs", + row.try_get("tool_timeout_secs")?, + )?, + }; + model::definition_from_stored_parts(id, revision, replace) +} + +fn require_field( id: &McpServerId, - expected: &McpServerRevision, + field: &'static str, + value: Option, +) -> Result { + value.ok_or_else(|| McpServerStoreError::StoredTransport { + id: id.clone(), + reason: format!("{field} is missing"), + }) +} + +fn require_absent( + id: &McpServerId, + field: &'static str, + value: Option<&T>, ) -> Result<(), McpServerStoreError> { - let current = defs - .get(id) - .ok_or_else(|| McpServerStoreError::NotFound { id: id.clone() })?; - if ¤t.revision != expected { - return Err(McpServerStoreError::StaleRevision { - id: id.clone(), - expected: expected.clone(), - actual: current.revision.clone(), + if value.is_some() { + return Err(McpServerStoreError::StoredTransport { + id: id.clone(), + reason: format!("{field} must be absent"), }); } Ok(()) } +fn parse_protocol(id: &McpServerId, value: &str) -> Result { + McpHttpProtocol::from_str(value).map_err(|_| McpServerStoreError::StoredTransport { + id: id.clone(), + reason: format!("unknown protocol {value:?}"), + }) +} + +fn decode_port(id: &McpServerId, value: i64) -> Result { + u16::try_from(value).map_err(|_| McpServerStoreError::StoredInteger { + id: id.clone(), + column: "port", + value, + }) +} + +fn decode_timeout( + id: &McpServerId, + column: &'static str, + value: i64, +) -> Result { + u64::try_from(value).map_err(|_| McpServerStoreError::StoredInteger { + id: id.clone(), + column, + value, + }) +} + +fn decode_json( + id: &McpServerId, + field: &'static str, + value: &str, +) -> Result { + serde_json::from_str(value).map_err(|source| McpServerStoreError::JsonDecode { + id: id.clone(), + field, + source, + }) +} + +fn decode_string_map( + id: &McpServerId, + field: &'static str, + value: &str, +) -> Result, McpServerStoreError> { + let ordered = decode_json::>(id, field, value)?; + Ok(ordered.into_iter().collect()) +} + fn server_settings_from_definition(definition: &McpServerDefinition) -> McpServerSettings { McpServerSettings { name: definition.id.to_string(), @@ -180,43 +391,321 @@ fn server_settings_from_definition(definition: &McpServerDefinition) -> McpServe } } -#[expect( - clippy::disallowed_methods, - reason = "MCP server directory scan runs once at startup, before the runtime needs to make progress; std::fs avoids needing a Tokio runtime for the caller." -)] -fn load_definitions( - dir: &Path, -) -> Result, McpServerStoreError> { - let entries = match std::fs::read_dir(dir) { +async fn current_revision( + transaction: &mut Transaction<'_, Sqlite>, + id: &McpServerId, +) -> Result, McpServerStoreError> { + let revision: Option = + sqlx::query_scalar("SELECT revision FROM mcp_servers WHERE id = ?") + .bind(id.as_str()) + .fetch_optional(&mut **transaction) + .await?; + revision + .map(|revision| { + McpServerRevision::from_str(&revision).map_err(|source| { + McpServerStoreError::StoredRevision { + id: id.clone(), + source, + } + }) + }) + .transpose() +} + +async fn revision_mismatch_error( + transaction: &mut Transaction<'_, Sqlite>, + id: &McpServerId, + expected: &McpServerRevision, +) -> Result { + let Some(actual) = current_revision(transaction, id).await? else { + return Ok(McpServerStoreError::NotFound { id: id.clone() }); + }; + Ok(McpServerStoreError::StaleRevision { + id: id.clone(), + expected: expected.clone(), + actual, + }) +} + +async fn insert_definition_ignoring_conflict( + transaction: &mut Transaction<'_, Sqlite>, + definition: &McpServerDefinition, +) -> Result { + let row = McpServerSqlRow::from_definition(definition)?; + let result = bind_definition(sqlx::query(INSERT_DEFINITION_SQL), row) + .execute(&mut **transaction) + .await?; + Ok(result.rows_affected() == 1) +} + +async fn update_definition( + transaction: &mut Transaction<'_, Sqlite>, + definition: &McpServerDefinition, + expected: &McpServerRevision, +) -> Result<(), McpServerStoreError> { + let row = McpServerSqlRow::from_definition(definition)?; + let result = sqlx::query(UPDATE_DEFINITION_SQL) + .bind(row.revision) + .bind(row.display_name) + .bind(row.description) + .bind(row.transport_type) + .bind(row.protocol) + .bind(row.command_json) + .bind(row.url) + .bind(row.port) + .bind(row.env_json) + .bind(row.headers_json) + .bind(row.startup_timeout_secs) + .bind(row.tool_timeout_secs) + .bind(row.id) + .bind(expected.as_str()) + .execute(&mut **transaction) + .await?; + if result.rows_affected() == 0 { + return Err(revision_mismatch_error(transaction, &definition.id, expected).await?); + } + Ok(()) +} + +fn bind_definition( + query: Query<'_, Sqlite, SqliteArguments>, + row: McpServerSqlRow, +) -> Query<'_, Sqlite, SqliteArguments> { + query + .bind(row.id) + .bind(row.revision) + .bind(row.display_name) + .bind(row.description) + .bind(row.transport_type) + .bind(row.protocol) + .bind(row.command_json) + .bind(row.url) + .bind(row.port) + .bind(row.env_json) + .bind(row.headers_json) + .bind(row.startup_timeout_secs) + .bind(row.tool_timeout_secs) +} + +struct McpServerSqlRow { + id: String, + revision: String, + display_name: String, + description: Option, + transport_type: &'static str, + protocol: Option<&'static str>, + command_json: Option, + url: Option, + port: Option, + env_json: Option, + headers_json: Option, + startup_timeout_secs: i64, + tool_timeout_secs: i64, +} + +impl McpServerSqlRow { + fn from_definition(definition: &McpServerDefinition) -> Result { + let (transport_type, protocol, command_json, url, port, env_json, headers_json) = + match &definition.transport { + McpTransport::Stdio { command, env } => ( + TransportType::Stdio, + None, + Some(encode_json("command_json", command)?), + None, + None, + Some(encode_string_map("env_json", env)?), + None, + ), + McpTransport::Http { + protocol, + url, + headers, + } => ( + TransportType::Http, + Some(protocol.as_str()), + None, + Some(url.clone()), + None, + None, + Some(encode_string_map("headers_json", headers)?), + ), + McpTransport::Sandbox { + protocol, + command, + port, + env, + } => ( + TransportType::Sandbox, + Some(protocol.as_str()), + Some(encode_json("command_json", command)?), + None, + Some(i64::from(*port)), + Some(encode_string_map("env_json", env)?), + None, + ), + }; + Ok(Self { + id: definition.id.to_string(), + revision: definition.revision.to_string(), + display_name: definition.display_name.clone(), + description: definition.description.clone(), + transport_type: transport_type.as_str(), + protocol, + command_json, + url, + port, + env_json, + headers_json, + startup_timeout_secs: encode_timeout( + "startup_timeout_secs", + definition.startup_timeout_secs, + )?, + tool_timeout_secs: encode_timeout("tool_timeout_secs", definition.tool_timeout_secs)?, + }) + } +} + +fn encode_timeout(field: &'static str, value: u64) -> Result { + i64::try_from(value) + .map_err(|_| McpServerValidationError::TimeoutOutOfRange { field, value }.into()) +} + +fn encode_json( + field: &'static str, + value: &T, +) -> Result { + serde_json::to_string(value).map_err(|source| McpServerStoreError::JsonEncode { field, source }) +} + +fn encode_string_map( + field: &'static str, + map: &HashMap, +) -> Result { + let ordered = map + .iter() + .map(|(key, value)| (key.as_str(), value.as_str())) + .collect::>(); + encode_json(field, &ordered) +} + +const INSERT_DEFINITION_SQL: &str = r" +INSERT INTO mcp_servers ( + id, + revision, + display_name, + description, + transport_type, + protocol, + command_json, + url, + port, + env_json, + headers_json, + startup_timeout_secs, + tool_timeout_secs +) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(id) DO NOTHING +"; + +const UPDATE_DEFINITION_SQL: &str = r" +UPDATE mcp_servers SET + revision = ?, + display_name = ?, + description = ?, + transport_type = ?, + protocol = ?, + command_json = ?, + url = ?, + port = ?, + env_json = ?, + headers_json = ?, + startup_timeout_secs = ?, + tool_timeout_secs = ? +WHERE id = ? AND revision = ? +"; + +pub async fn import_legacy_directory_once( + pool: &DbPool, + source_dir: impl AsRef, +) -> Result, McpServerStoreError> { + let source_dir = source_dir.as_ref(); + let Some(paths) = legacy_definition_paths(source_dir).await? else { + return Ok(None); + }; + let definitions = read_legacy_definitions(paths).await?; + + let mut transaction = pool.begin().await?; + let mut imported_ids = Vec::new(); + let mut skipped_rows = 0; + for definition in &definitions { + if insert_definition_ignoring_conflict(&mut transaction, definition).await? { + imported_ids.push(definition.id.to_string()); + } else { + skipped_rows += 1; + } + } + transaction.commit().await?; + + let backup_path = rename_imported_legacy_directory(source_dir).await?; + let report = ImportReport { + source_path: source_dir.to_path_buf(), + backup_path, + imported_rows: imported_ids.len(), + skipped_rows, + mcp_server_ids: imported_ids, + }; + info!( + source_path = %report.source_path.display(), + backup_path = %report.backup_path.display(), + imported_rows = report.imported_rows, + skipped_rows = report.skipped_rows, + mcp_server_ids = ?report.mcp_server_ids, + "Imported legacy MCP server directory into SQLite" + ); + Ok(Some(report)) +} + +async fn legacy_definition_paths( + source_dir: &Path, +) -> Result>, McpServerStoreError> { + let mut entries = match fs::read_dir(source_dir).await { Ok(entries) => entries, - Err(err) if err.kind() == ErrorKind::NotFound => return Ok(HashMap::new()), - Err(err) => return Err(McpServerStoreError::io(dir, err)), + Err(source) if source.kind() == ErrorKind::NotFound => return Ok(None), + Err(source) => return Err(McpServerStoreError::io(source_dir, source)), }; - let mut defs = HashMap::new(); - for entry in entries { - let entry = entry.map_err(|err| McpServerStoreError::io(dir, err))?; + let mut paths = Vec::new(); + while let Some(entry) = entries + .next_entry() + .await + .map_err(|source| McpServerStoreError::io(source_dir, source))? + { let path = entry.path(); let file_type = entry .file_type() - .map_err(|err| McpServerStoreError::io(&path, err))?; - if !file_type.is_file() || !is_toml_file(&path) { - continue; + .await + .map_err(|source| McpServerStoreError::io(&path, source))?; + if file_type.is_file() && is_toml_file(&path) { + paths.push((id_from_path(&path)?, path)); } - let definition = load_definition_file(&path)?; - defs.insert(definition.id.clone(), definition); } - Ok(defs) + paths.sort_by(|left, right| left.1.cmp(&right.1)); + Ok(Some(paths)) } -#[expect( - clippy::disallowed_methods, - reason = "Sync sibling of `load_definitions`; only invoked from the synchronous startup load path." -)] -fn load_definition_file(path: &Path) -> Result { - let id = id_from_path(path)?; - let bytes = std::fs::read(path).map_err(|err| McpServerStoreError::io(path, err))?; - model::definition_from_persisted_path(id, &bytes, path) +async fn read_legacy_definitions( + paths: Vec<(McpServerId, PathBuf)>, +) -> Result, McpServerStoreError> { + let mut definitions = Vec::with_capacity(paths.len()); + for (id, path) in paths { + let bytes = fs::read(&path) + .await + .map_err(|source| McpServerStoreError::io(&path, source))?; + definitions.push(model::definition_from_persisted_path(id, &bytes, path)?); + } + definitions.sort_by(|left, right| left.id.cmp(&right.id)); + Ok(definitions) } fn id_from_path(path: &Path) -> Result { @@ -239,259 +728,25 @@ fn is_toml_file(path: &Path) -> bool { .is_some_and(|extension| extension == "toml") } -async fn write_atomic(dir: &Path, path: &Path, bytes: &[u8]) -> Result<(), McpServerStoreError> { - let temp_path = write_temp_file(dir, path, bytes).await?; - if let Err(err) = fs::rename(&temp_path, path).await { - cleanup_temp(&temp_path).await; - return Err(McpServerStoreError::io(path, err)); - } - - Ok(()) -} - -async fn write_new(dir: &Path, path: &Path, bytes: &[u8]) -> Result<(), McpServerStoreError> { - let temp_path = write_temp_file(dir, path, bytes).await?; - if let Err(err) = fs::hard_link(&temp_path, path).await { - cleanup_temp(&temp_path).await; - return Err(McpServerStoreError::io(path, err)); - } - cleanup_temp(&temp_path).await; - Ok(()) -} - -async fn write_temp_file( - dir: &Path, - path: &Path, - bytes: &[u8], +async fn rename_imported_legacy_directory( + source_dir: &Path, ) -> Result { - fs::create_dir_all(dir) + let backup_path = legacy_backup_path(source_dir, Utc::now()); + fs::rename(source_dir, &backup_path) .await - .map_err(|err| McpServerStoreError::io(dir, err))?; - let temp_path = temp_path_for(path); - let mut file = fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(&temp_path) - .await - .map_err(|err| McpServerStoreError::io(&temp_path, err))?; - - if let Err(err) = file.write_all(bytes).await { - cleanup_temp(&temp_path).await; - return Err(McpServerStoreError::io(&temp_path, err)); - } - if let Err(err) = file.sync_all().await { - cleanup_temp(&temp_path).await; - return Err(McpServerStoreError::io(&temp_path, err)); - } - drop(file); - - Ok(temp_path) + .map_err(|source| McpServerStoreError::LegacyBackup { + source_path: source_dir.to_path_buf(), + backup_path: backup_path.clone(), + source, + })?; + Ok(backup_path) } -async fn cleanup_temp(path: &Path) { - let _ = fs::remove_file(path).await; -} - -fn create_error_for(id: McpServerId, err: McpServerStoreError) -> McpServerStoreError { - match err { - McpServerStoreError::Io { source, .. } if source.kind() == ErrorKind::AlreadyExists => { - McpServerStoreError::AlreadyExists { id } - } - err => err, - } -} - -fn temp_path_for(path: &Path) -> PathBuf { - let parent = path.parent().unwrap_or_else(|| Path::new(".")); - let file_name = path +fn legacy_backup_path(source_dir: &Path, imported_at: DateTime) -> PathBuf { + let timestamp = imported_at.format("%Y%m%dT%H%M%S%fZ"); + let mut file_name = source_dir .file_name() - .and_then(|name| name.to_str()) - .unwrap_or("mcp-server.toml"); - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map_or(0, |duration| duration.as_nanos()); - parent.join(format!(".{file_name}.{}.{}.tmp", std::process::id(), now)) -} - -fn definition_path(dir: &Path, id: &McpServerId) -> PathBuf { - dir.join(format!("{id}.toml")) -} - -#[cfg(test)] -mod tests { - use std::collections::HashMap; - - use fabro_types::settings::McpTransport; - use fabro_types::settings::run::McpHttpProtocol; - use fabro_types::{McpServerDraft, McpServerId, McpServerReplace, McpServerRevision}; - use tokio::fs; - - use crate::error::McpServerStoreError; - use crate::store::McpServerStore; - - fn http_transport(url: &str) -> McpTransport { - McpTransport::Http { - protocol: McpHttpProtocol::default(), - url: url.to_string(), - headers: HashMap::new(), - } - } - - fn draft(id: &str, display_name: &str) -> McpServerDraft { - McpServerDraft { - id: McpServerId::new(id).unwrap(), - display_name: display_name.to_string(), - description: None, - transport: http_transport("https://example.com/mcp"), - startup_timeout_secs: 10, - tool_timeout_secs: 60, - } - } - - fn replacement(display_name: &str) -> McpServerReplace { - McpServerReplace { - display_name: display_name.to_string(), - description: Some("updated".to_string()), - transport: http_transport("https://example.com/mcp/v2"), - startup_timeout_secs: 15, - tool_timeout_secs: 90, - } - } - - #[tokio::test] - async fn missing_directory_loads_empty_store() { - let dir = tempfile::tempdir().unwrap(); - let store = McpServerStore::load(dir.path().join("mcps")).unwrap(); - - assert!(store.list().is_empty()); - assert!(store.ids().is_empty()); - } - - #[tokio::test] - async fn load_ignores_non_toml_files_and_keeps_valid_definitions() { - let dir = tempfile::tempdir().unwrap(); - let mcp_dir = dir.path().join("mcps"); - fs::create_dir_all(&mcp_dir).await.unwrap(); - fs::write(mcp_dir.join("notes.txt"), "ignore") - .await - .unwrap(); - fs::write( - mcp_dir.join("sentry.toml"), - r#" -display_name = "Sentry" -startup_timeout_secs = 10 -tool_timeout_secs = 60 - -[transport] -type = "http" -url = "https://sentry.example.com/mcp" - -[transport.headers] -"#, - ) - .await - .unwrap(); - - let store = McpServerStore::load(&mcp_dir).unwrap(); - let defs = store.list(); - - assert_eq!(defs.len(), 1); - assert_eq!(defs[0].id.as_str(), "sentry"); - assert_eq!(defs[0].display_name, "Sentry"); - } - - #[tokio::test] - async fn load_fails_on_malformed_toml() { - let dir = tempfile::tempdir().unwrap(); - let mcp_dir = dir.path().join("mcps"); - fs::create_dir_all(&mcp_dir).await.unwrap(); - fs::write(mcp_dir.join("broken.toml"), "not valid toml =") - .await - .unwrap(); - - let err = McpServerStore::load(&mcp_dir).unwrap_err(); - assert!(matches!(err, McpServerStoreError::Parse { .. })); - } - - #[tokio::test] - async fn load_fails_on_invalid_filename_id() { - let dir = tempfile::tempdir().unwrap(); - let mcp_dir = dir.path().join("mcps"); - fs::create_dir_all(&mcp_dir).await.unwrap(); - fs::write(mcp_dir.join("Bad Name.toml"), "display_name = \"Bad\"") - .await - .unwrap(); - - let err = McpServerStore::load(&mcp_dir).unwrap_err(); - assert!(matches!(err, McpServerStoreError::InvalidFilename { .. })); - } - - #[tokio::test] - async fn create_get_list_replace_and_delete_round_trip_files_and_revisions() { - let dir = tempfile::tempdir().unwrap(); - let mcp_dir = dir.path().join("mcps"); - let store = McpServerStore::load(&mcp_dir).unwrap(); - - let created = store.create(draft("sentry", "Sentry")).await.unwrap(); - let path = mcp_dir.join("sentry.toml"); - let persisted = fs::read_to_string(&path).await.unwrap(); - assert!(persisted.contains("display_name = \"Sentry\"")); - assert!(!top_level_lines(&persisted).any(|line| line.starts_with("id = "))); - assert!(!top_level_lines(&persisted).any(|line| line.starts_with("revision = "))); - assert_eq!( - created.revision, - McpServerRevision::from_bytes(persisted.as_bytes()) - ); - - assert_eq!(store.get(&created.id).unwrap(), created); - let listed = store.list(); - assert_eq!(listed.len(), 1); - assert_eq!(store.ids(), vec![created.id.clone()]); - - let replaced = store - .replace(&created.id, &created.revision, replacement("Sentry v2")) - .await - .unwrap(); - assert_ne!(replaced.revision, created.revision); - assert_eq!(replaced.display_name, "Sentry v2"); - assert_eq!(store.get(&created.id).unwrap().revision, replaced.revision); - - store.delete(&created.id, &replaced.revision).await.unwrap(); - assert!(store.get(&created.id).is_none()); - assert!(!path.exists()); - } - - #[tokio::test] - async fn replace_with_stale_revision_is_rejected() { - let dir = tempfile::tempdir().unwrap(); - let store = McpServerStore::load(dir.path().join("mcps")).unwrap(); - let created = store.create(draft("sentry", "Sentry")).await.unwrap(); - - let stale = McpServerRevision::from_bytes(b"stale"); - let err = store - .replace(&created.id, &stale, replacement("Updated")) - .await - .unwrap_err(); - assert!(matches!(err, McpServerStoreError::StaleRevision { .. })); - - // The on-disk and in-memory definition is unchanged after a rejected replace. - assert_eq!(store.get(&created.id).unwrap(), created); - } - - #[tokio::test] - async fn duplicate_create_is_rejected() { - let dir = tempfile::tempdir().unwrap(); - let store = McpServerStore::load(dir.path().join("mcps")).unwrap(); - store.create(draft("sentry", "Sentry")).await.unwrap(); - - let err = store - .create(draft("sentry", "Duplicate")) - .await - .unwrap_err(); - assert!(matches!(err, McpServerStoreError::AlreadyExists { .. })); - } - - fn top_level_lines(toml: &str) -> impl Iterator { - toml.lines().take_while(|line| !line.starts_with('[')) - } + .map_or_else(|| OsString::from("mcps"), OsString::from); + file_name.push(format!(".imported-{timestamp}.bak")); + source_dir.with_file_name(file_name) } diff --git a/lib/crates/fabro-mcp-store/tests/store.rs b/lib/crates/fabro-mcp-store/tests/store.rs new file mode 100644 index 000000000..f39cd1ea9 --- /dev/null +++ b/lib/crates/fabro-mcp-store/tests/store.rs @@ -0,0 +1,375 @@ +#![expect( + clippy::unwrap_used, + reason = "SQLite MCP store integration tests use panic-on-failure fixture setup" +)] + +use std::collections::HashMap; +use std::error::Error as _; + +use fabro_db::Database; +use fabro_mcp_store::{McpServerStore, McpServerStoreError, import_legacy_directory_once}; +use fabro_types::settings::run::{McpHttpProtocol, McpTransport}; +use fabro_types::{McpServerDraft, McpServerId, McpServerReplace, McpServerRevision}; +use tokio::fs; + +async fn test_database() -> (tempfile::TempDir, Database) { + let dir = tempfile::tempdir().unwrap(); + let database = Database::connect(dir.path().join("fabro.sqlite3")) + .await + .unwrap(); + database.migrate().await.unwrap(); + (dir, database) +} + +fn http_transport(url: &str) -> McpTransport { + McpTransport::Http { + protocol: McpHttpProtocol::default(), + url: url.to_string(), + headers: HashMap::new(), + } +} + +fn draft(id: &str, display_name: &str) -> McpServerDraft { + McpServerDraft { + id: McpServerId::new(id).unwrap(), + display_name: display_name.to_string(), + description: None, + transport: http_transport("https://example.com/mcp"), + startup_timeout_secs: 10, + tool_timeout_secs: 60, + } +} + +fn replacement(display_name: &str) -> McpServerReplace { + McpServerReplace { + display_name: display_name.to_string(), + description: Some("updated".to_string()), + transport: http_transport("https://example.com/mcp/v2"), + startup_timeout_secs: 15, + tool_timeout_secs: 90, + } +} + +fn legacy_toml(display_name: &str, url: &str) -> String { + format!( + r#"display_name = "{display_name}" +startup_timeout_secs = 10 +tool_timeout_secs = 60 + +[transport] +type = "http" +url = "{url}" + +[transport.headers] +"# + ) +} + +#[tokio::test] +async fn empty_database_loads_empty_store() { + let (_dir, database) = test_database().await; + let store = McpServerStore::load(database.clone_pool()).await.unwrap(); + + assert!(store.list().is_empty()); + assert!(store.ids().is_empty()); +} + +#[tokio::test] +async fn create_get_list_replace_delete_and_reload_round_trip() { + let (_dir, database) = test_database().await; + let store = McpServerStore::load(database.clone_pool()).await.unwrap(); + + let created = store.create(draft("sentry", "Sentry")).await.unwrap(); + assert_eq!(store.get(&created.id).unwrap(), created); + assert_eq!(store.list(), vec![created.clone()]); + assert_eq!(store.ids(), vec![created.id.clone()]); + + let reloaded = McpServerStore::load(database.clone_pool()).await.unwrap(); + assert_eq!(reloaded.get(&created.id).unwrap(), created); + + let replaced = store + .replace(&created.id, &created.revision, replacement("Sentry v2")) + .await + .unwrap(); + assert_ne!(replaced.revision, created.revision); + assert_eq!(replaced.display_name, "Sentry v2"); + + store + .delete(&replaced.id, &replaced.revision) + .await + .unwrap(); + assert!(store.get(&replaced.id).is_none()); + assert!( + McpServerStore::load(database.clone_pool()) + .await + .unwrap() + .list() + .is_empty() + ); +} + +#[tokio::test] +async fn all_transport_variants_round_trip_with_sorted_map_json() { + let (_dir, database) = test_database().await; + let store = McpServerStore::load(database.clone_pool()).await.unwrap(); + let definitions = [ + McpServerDraft { + id: McpServerId::new("stdio").unwrap(), + display_name: "Stdio".to_string(), + description: None, + transport: McpTransport::Stdio { + command: vec!["server".to_string(), "--flag".to_string()], + env: HashMap::from([ + ("Z_KEY".to_string(), "z".to_string()), + ("A_KEY".to_string(), "a".to_string()), + ]), + }, + startup_timeout_secs: 1, + tool_timeout_secs: 2, + }, + McpServerDraft { + id: McpServerId::new("http").unwrap(), + display_name: "HTTP".to_string(), + description: None, + transport: McpTransport::Http { + protocol: McpHttpProtocol::Sse, + url: "https://example.com/mcp".to_string(), + headers: HashMap::from([("Authorization".to_string(), "secret".to_string())]), + }, + startup_timeout_secs: 3, + tool_timeout_secs: 4, + }, + McpServerDraft { + id: McpServerId::new("sandbox").unwrap(), + display_name: "Sandbox".to_string(), + description: None, + transport: McpTransport::Sandbox { + protocol: McpHttpProtocol::StreamableHttp, + command: vec!["server".to_string()], + port: 3000, + env: HashMap::from([("TOKEN".to_string(), "secret".to_string())]), + }, + startup_timeout_secs: 5, + tool_timeout_secs: 6, + }, + ]; + + for definition in definitions.clone() { + store.create(definition).await.unwrap(); + } + let reloaded = McpServerStore::load(database.clone_pool()).await.unwrap(); + for expected in definitions { + let actual = reloaded.get(&expected.id).unwrap(); + assert_eq!(actual.display_name, expected.display_name); + assert_eq!(actual.transport, expected.transport); + } + + let env_json: String = + sqlx::query_scalar("SELECT env_json FROM mcp_servers WHERE id = 'stdio'") + .fetch_one(database.pool()) + .await + .unwrap(); + assert_eq!(env_json, r#"{"A_KEY":"a","Z_KEY":"z"}"#); +} + +#[tokio::test] +async fn duplicate_create_is_rejected() { + let (_dir, database) = test_database().await; + let first = McpServerStore::load(database.clone_pool()).await.unwrap(); + let second = McpServerStore::load(database.clone_pool()).await.unwrap(); + first.create(draft("sentry", "Sentry")).await.unwrap(); + + let err = second + .create(draft("sentry", "Duplicate")) + .await + .unwrap_err(); + assert!(matches!(err, McpServerStoreError::AlreadyExists { .. })); +} + +#[tokio::test] +async fn revision_guard_rejects_stale_independent_store() { + let (_dir, database) = test_database().await; + let creator = McpServerStore::load(database.clone_pool()).await.unwrap(); + let created = creator.create(draft("sentry", "Sentry")).await.unwrap(); + let first = McpServerStore::load(database.clone_pool()).await.unwrap(); + let second = McpServerStore::load(database.clone_pool()).await.unwrap(); + + first + .replace(&created.id, &created.revision, replacement("Winner")) + .await + .unwrap(); + let err = second + .replace(&created.id, &created.revision, replacement("Loser")) + .await + .unwrap_err(); + + assert!(matches!(err, McpServerStoreError::StaleRevision { .. })); + assert_eq!(second.get(&created.id).unwrap(), created); +} + +#[tokio::test] +async fn imports_legacy_directory_once_without_overwriting_sql() { + let (dir, database) = test_database().await; + let store = McpServerStore::load(database.clone_pool()).await.unwrap(); + store.create(draft("existing", "SQLite")).await.unwrap(); + let source = dir.path().join("mcps"); + fs::create_dir_all(&source).await.unwrap(); + let existing_bytes = legacy_toml("Legacy", "https://legacy.example.com/mcp"); + let imported_bytes = legacy_toml("Imported", "https://new.example.com/mcp"); + fs::write(source.join("existing.toml"), &existing_bytes) + .await + .unwrap(); + fs::write(source.join("new.toml"), &imported_bytes) + .await + .unwrap(); + fs::write(source.join("notes.txt"), "preserve in backup") + .await + .unwrap(); + + let report = import_legacy_directory_once(database.pool(), &source) + .await + .unwrap() + .unwrap(); + + assert_eq!(report.imported_rows, 1); + assert_eq!(report.skipped_rows, 1); + assert_eq!(report.mcp_server_ids, vec!["new"]); + assert!(!source.exists()); + assert!(report.backup_path.join("notes.txt").exists()); + let reloaded = McpServerStore::load(database.clone_pool()).await.unwrap(); + assert_eq!( + reloaded + .get(&McpServerId::new("existing").unwrap()) + .unwrap() + .display_name, + "SQLite" + ); + let imported = reloaded.get(&McpServerId::new("new").unwrap()).unwrap(); + assert_eq!(imported.display_name, "Imported"); + assert_eq!( + imported.revision, + McpServerRevision::from_bytes(imported_bytes.as_bytes()) + ); + assert!( + import_legacy_directory_once(database.pool(), &source) + .await + .unwrap() + .is_none() + ); +} + +#[tokio::test] +async fn malformed_legacy_toml_does_not_import_or_rename() { + let (dir, database) = test_database().await; + let source = dir.path().join("mcps"); + fs::create_dir_all(&source).await.unwrap(); + fs::write( + source.join("valid.toml"), + legacy_toml("Valid", "https://example.com/mcp"), + ) + .await + .unwrap(); + fs::write(source.join("broken.toml"), "not valid toml =") + .await + .unwrap(); + + let err = import_legacy_directory_once(database.pool(), &source) + .await + .unwrap_err(); + + assert!(matches!(err, McpServerStoreError::Parse { .. })); + assert!(source.exists()); + assert!( + McpServerStore::load(database.clone_pool()) + .await + .unwrap() + .list() + .is_empty() + ); +} + +#[tokio::test] +async fn legacy_parse_error_chain_does_not_expose_transport_values() { + let (dir, database) = test_database().await; + let source = dir.path().join("mcps"); + fs::create_dir_all(&source).await.unwrap(); + fs::write( + source.join("broken.toml"), + r#" +display_name = "Broken" +startup_timeout_secs = 10 +tool_timeout_secs = 60 + +[transport] +type = "http" +url = "https://example.com/mcp" + +[transport.headers] +Authorization = "do-not-print" trailing-invalid-content +"#, + ) + .await + .unwrap(); + + let err = import_legacy_directory_once(database.pool(), &source) + .await + .unwrap_err(); + let mut rendered = err.to_string(); + let mut source = err.source(); + while let Some(err) = source { + rendered.push_str(&err.to_string()); + source = err.source(); + } + + assert!(!rendered.contains("do-not-print")); +} + +#[tokio::test] +async fn corrupted_stored_row_returns_typed_error() { + let (_dir, database) = test_database().await; + let mut connection = database.pool().acquire().await.unwrap(); + sqlx::query("PRAGMA ignore_check_constraints = ON") + .execute(&mut *connection) + .await + .unwrap(); + sqlx::query( + r" + INSERT INTO mcp_servers ( + id, revision, display_name, transport_type, command_json, env_json, + startup_timeout_secs, tool_timeout_secs + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ", + ) + .bind("broken") + .bind("not-a-revision") + .bind("Broken") + .bind("stdio") + .bind(r#"["server"]"#) + .bind("{}") + .bind(10_i64) + .bind(60_i64) + .execute(&mut *connection) + .await + .unwrap(); + + let err = McpServerStore::load(database.clone_pool()) + .await + .unwrap_err(); + assert!(matches!(err, McpServerStoreError::StoredRevision { .. })); +} + +#[tokio::test] +async fn store_debug_does_not_expose_transport_values() { + let (_dir, database) = test_database().await; + let store = McpServerStore::load(database.clone_pool()).await.unwrap(); + let mut secret_draft = draft("secret", "Secret"); + secret_draft.transport = McpTransport::Http { + protocol: McpHttpProtocol::default(), + url: "https://example.com/mcp".to_string(), + headers: HashMap::from([("Authorization".to_string(), "do-not-print".to_string())]), + }; + store.create(secret_draft).await.unwrap(); + + let debug = format!("{store:?}"); + assert!(!debug.contains("do-not-print")); +} diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index cfba70bd7..6d46d3d6d 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -2345,6 +2345,28 @@ fn load_environment_store_blocking( .expect("environment store load thread should not panic") } +#[expect( + clippy::disallowed_methods, + reason = "synchronous app-state assembly may run inside an async runtime; a short-lived OS \ + thread avoids nested Tokio runtimes" +)] +fn load_mcp_server_store_blocking( + pool: DbPool, + legacy_dir: PathBuf, +) -> anyhow::Result { + std::thread::spawn(move || { + let runtime = TokioRuntimeBuilder::new_current_thread() + .enable_all() + .build() + .context("build MCP server store runtime")?; + runtime + .block_on(McpServerStore::open(pool, legacy_dir)) + .map_err(anyhow::Error::new) + }) + .join() + .expect("MCP server store load thread should not panic") +} + pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result> { let AppStateConfig { resolved_settings, @@ -2388,8 +2410,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result for ApiError { } // Remaining variants are persistence/parse faults that indicate an // internal problem rather than a client one. - McpServerStoreError::InvalidFilename { .. } + McpServerStoreError::Db { .. } + | McpServerStoreError::StoredRevision { .. } + | McpServerStoreError::StoredTransport { .. } + | McpServerStoreError::StoredInteger { .. } + | McpServerStoreError::JsonEncode { .. } + | McpServerStoreError::JsonDecode { .. } + | McpServerStoreError::InvalidFilename { .. } | McpServerStoreError::Parse { .. } | McpServerStoreError::InvalidUtf8 { .. } | McpServerStoreError::Serialize { .. } - | McpServerStoreError::Io { .. } => Self::new( + | McpServerStoreError::Io { .. } + | McpServerStoreError::LegacyBackup { .. } => Self::new( StatusCode::INTERNAL_SERVER_ERROR, "mcp server store operation failed", ), diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index 9a863ed50..cf2679165 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -117,6 +117,32 @@ fn test_environment_store( (temp, store) } +fn test_mcp_server_store() -> (tempfile::TempDir, McpServerStore) { + let temp = tempfile::tempdir().expect("MCP server store tempdir should be created"); + let db_path = temp.path().join("fabro.sqlite3"); + let pool = std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("MCP server store setup runtime should build"); + runtime.block_on(async move { + let database = fabro_db::Database::connect(db_path) + .await + .expect("test MCP server database should connect"); + database + .migrate() + .await + .expect("test MCP server database should migrate"); + database.clone_pool() + }) + }) + .join() + .expect("MCP server store setup thread should not panic"); + let store = load_mcp_server_store_blocking(pool, temp.path().join("mcps")) + .expect("test MCP server store should load"); + (temp, store) +} + fn server_settings_from_toml(source: &str) -> ServerSettings { ServerSettingsBuilder::from_toml(source).expect("server settings should resolve") } @@ -1275,10 +1301,9 @@ id = "missing" #[test] fn system_sandbox_provider_uses_manifest_defaults() { - let (temp, environment_store) = + let (_environment_temp, environment_store) = test_environment_store(Some(EnvironmentProvider::Daytona), true); - let mcp_server_store = - McpServerStore::load(temp.path().join("mcps")).expect("mcp server store should load"); + let (_mcp_temp, mcp_server_store) = test_mcp_server_store(); let source = r#" _version = 1 @@ -1296,9 +1321,8 @@ id = "default" #[test] fn system_sandbox_provider_defaults_when_manifest_run_settings_do_not_resolve() { - let (temp, environment_store) = test_environment_store(None, true); - let mcp_server_store = - McpServerStore::load(temp.path().join("mcps")).expect("mcp server store should load"); + let (_environment_temp, environment_store) = test_environment_store(None, true); + let (_mcp_temp, mcp_server_store) = test_mcp_server_store(); let source = r#" _version = 1 diff --git a/lib/crates/fabro-server/tests/it/api/mcp_servers.rs b/lib/crates/fabro-server/tests/it/api/mcp_servers.rs index 50e78b2a3..5c71d671e 100644 --- a/lib/crates/fabro-server/tests/it/api/mcp_servers.rs +++ b/lib/crates/fabro-server/tests/it/api/mcp_servers.rs @@ -2,8 +2,12 @@ use std::path::{Path, PathBuf}; use axum::body::Body; use axum::http::{Method, Request, StatusCode, header}; +use fabro_db::Database; +use fabro_mcp_store::McpServerStore; use fabro_server::server::build_router; use fabro_server::test_support::{TestAppStateBuilder, build_test_router, test_auth_mode}; +use fabro_types::settings::McpTransport; +use fabro_types::{McpServerDefinition, McpServerId}; use serde_json::{Value, json}; use tower::ServiceExt; @@ -47,11 +51,13 @@ fn replacement_body(display_name: &str) -> Value { fn mcp_server_app() -> (axum::Router, tempfile::TempDir, PathBuf) { let temp_dir = tempfile::tempdir().expect("mcp server test tempdir should be created"); let active_config_path = temp_dir.path().join("settings.toml"); - let mcp_dir = temp_dir.path().join("mcps"); + let vault_path = temp_dir.path().join("secrets.json"); + let db_path = temp_dir.path().join("db").join("fabro.sqlite3"); let state = TestAppStateBuilder::new() .active_config_path(active_config_path) + .vault_path(vault_path) .build(); - (build_test_router(state), temp_dir, mcp_dir) + (build_test_router(state), temp_dir, db_path) } fn json_request(method: Method, path: &str, body: &Value) -> Request { @@ -114,11 +120,14 @@ fn revision_from(body: &Value) -> &str { .expect("mcp server response should include a revision") } -async fn persisted_mcp_server_toml(mcp_dir: &Path, id: &str) -> toml::Value { - let persisted = tokio::fs::read_to_string(mcp_dir.join(format!("{id}.toml"))) +async fn persisted_mcp_server(db_path: &Path, id: &str) -> Option { + let database = Database::connect(db_path) .await - .expect("persisted mcp server TOML should be readable"); - toml::from_str(&persisted).expect("persisted mcp server TOML should parse") + .expect("persisted MCP server database should connect"); + let store = McpServerStore::load(database.clone_pool()) + .await + .expect("persisted MCP server store should load"); + store.get(&McpServerId::new(id).expect("fixture MCP server id should be valid")) } #[tokio::test] @@ -141,8 +150,8 @@ async fn empty_mcp_server_list_returns_total_zero() { } #[tokio::test] -async fn create_mcp_server_returns_etag_and_persists_sibling_toml_file() { - let (app, _temp_dir, mcp_dir) = mcp_server_app(); +async fn create_mcp_server_returns_etag_and_persists_sqlite_row() { + let (app, _temp_dir, db_path) = mcp_server_app(); let response = app .clone() @@ -167,7 +176,7 @@ async fn create_mcp_server_returns_etag_and_persists_sibling_toml_file() { assert_eq!(body["id"], "sentry"); assert_eq!(body["display_name"], "Sentry"); assert_eq!(etag, format!("\"{}\"", revision_from(&body))); - assert!(mcp_dir.join("sentry.toml").exists()); + assert!(db_path.exists()); // The response is the value-omitting view: header *names* are returned, but // the stored header value is not. @@ -177,27 +186,20 @@ async fn create_mcp_server_returns_etag_and_persists_sibling_toml_file() { "response must not echo transport header values" ); - let persisted = persisted_mcp_server_toml(&mcp_dir, "sentry").await; - assert_eq!( - persisted.get("display_name").and_then(toml::Value::as_str), - Some("Sentry") - ); - assert!(persisted.get("id").is_none()); - assert!(persisted.get("revision").is_none()); - // The value the view omits is still persisted on disk for the runtime to use. - assert_eq!( - persisted - .get("transport") - .and_then(|transport| transport.get("headers")) - .and_then(|headers| headers.get("X-Org")) - .and_then(toml::Value::as_str), - Some("fabro") - ); + let persisted = persisted_mcp_server(&db_path, "sentry") + .await + .expect("MCP server should be persisted"); + assert_eq!(persisted.display_name, "Sentry"); + assert_eq!(persisted.revision.as_str(), revision_from(&body)); + let McpTransport::Http { headers, .. } = persisted.transport else { + panic!("persisted MCP server should use HTTP transport") + }; + assert_eq!(headers.get("X-Org").map(String::as_str), Some("fabro")); } #[tokio::test] -async fn mcp_server_round_trips_through_create_get_and_toml() { - let (app, _temp_dir, mcp_dir) = mcp_server_app(); +async fn mcp_server_round_trips_through_create_get_and_sqlite() { + let (app, _temp_dir, db_path) = mcp_server_app(); let created = create_mcp_server(&app, "sentry", "Sentry").await; assert_eq!(created["transport"]["type"], "http"); @@ -211,14 +213,10 @@ async fn mcp_server_round_trips_through_create_get_and_toml() { let retrieved = response_json(response, StatusCode::OK, "GET /api/v1/mcp-servers/sentry").await; assert_eq!(retrieved, created); - let persisted = persisted_mcp_server_toml(&mcp_dir, "sentry").await; - assert_eq!( - persisted - .get("transport") - .and_then(|transport| transport.get("type")) - .and_then(toml::Value::as_str), - Some("http") - ); + let persisted = persisted_mcp_server(&db_path, "sentry") + .await + .expect("MCP server should be persisted"); + assert!(matches!(persisted.transport, McpTransport::Http { .. })); } #[tokio::test] @@ -432,8 +430,8 @@ async fn replace_and_delete_mcp_server_require_if_match() { } #[tokio::test] -async fn delete_mcp_server_removes_file_and_resource() { - let (app, _temp_dir, mcp_dir) = mcp_server_app(); +async fn delete_mcp_server_removes_sqlite_row_and_resource() { + let (app, _temp_dir, db_path) = mcp_server_app(); let created = create_mcp_server(&app, "sentry", "Sentry").await; let revision = revision_from(&created); @@ -454,7 +452,7 @@ async fn delete_mcp_server_removes_file_and_resource() { ) .await; - assert!(!mcp_dir.join("sentry.toml").exists()); + assert!(persisted_mcp_server(&db_path, "sentry").await.is_none()); let response = app .oneshot(empty_request(Method::GET, "/mcp-servers/sentry")) .await @@ -593,7 +591,7 @@ id = "sentry" } #[tokio::test] -async fn mcp_server_store_malformed_persisted_toml_fails_startup() { +async fn malformed_legacy_mcp_server_toml_fails_startup() { let temp_dir = tempfile::tempdir().expect("mcp server test tempdir should be created"); let mcp_dir = temp_dir.path().join("mcps"); tokio::fs::create_dir_all(&mcp_dir) @@ -610,6 +608,84 @@ async fn mcp_server_store_malformed_persisted_toml_fails_startup() { assert!(result.is_err()); } +#[tokio::test] +async fn legacy_mcp_server_toml_imports_at_startup_and_renames_directory() { + let temp_dir = tempfile::tempdir().expect("mcp server test tempdir should be created"); + let active_config_path = temp_dir.path().join("settings.toml"); + let vault_path = temp_dir.path().join("secrets.json"); + let db_path = temp_dir.path().join("db").join("fabro.sqlite3"); + let mcp_dir = temp_dir.path().join("mcps"); + tokio::fs::create_dir_all(&mcp_dir) + .await + .expect("legacy MCP server directory should be created"); + tokio::fs::write( + mcp_dir.join("sentry.toml"), + r#" +display_name = "Sentry" +startup_timeout_secs = 10 +tool_timeout_secs = 60 + +[transport] +type = "http" +url = "https://example.com/mcp" + +[transport.headers] +Authorization = "secret-value" +"#, + ) + .await + .expect("legacy MCP server fixture should be written"); + + let state = TestAppStateBuilder::new() + .active_config_path(active_config_path) + .vault_path(vault_path) + .build(); + let app = build_test_router(state); + + let response = app + .oneshot(empty_request(Method::GET, "/mcp-servers/sentry")) + .await + .expect("imported MCP server should respond"); + let body = response_json( + response, + StatusCode::OK, + "GET imported /api/v1/mcp-servers/sentry", + ) + .await; + assert_eq!(body["display_name"], "Sentry"); + assert_eq!(body["transport"]["header_keys"], json!(["Authorization"])); + assert!(!mcp_dir.exists()); + let mut entries = tokio::fs::read_dir(temp_dir.path()) + .await + .expect("test directory should be readable"); + let mut backup_found = false; + while let Some(entry) = entries + .next_entry() + .await + .expect("test directory entry should be readable") + { + if entry + .file_name() + .to_string_lossy() + .starts_with("mcps.imported-") + { + backup_found = true; + break; + } + } + assert!(backup_found); + let persisted = persisted_mcp_server(&db_path, "sentry") + .await + .expect("imported MCP server should be in SQLite"); + let McpTransport::Http { headers, .. } = persisted.transport else { + panic!("imported MCP server should use HTTP transport") + }; + assert_eq!( + headers.get("Authorization").map(String::as_str), + Some("secret-value") + ); +} + #[tokio::test] async fn mcp_servers_routes_require_authenticated_user() { let temp_dir = tempfile::tempdir().expect("mcp server test tempdir should be created"); diff --git a/lib/crates/fabro-types/src/mcp_store.rs b/lib/crates/fabro-types/src/mcp_store.rs index 18cedbe42..5a7d64535 100644 --- a/lib/crates/fabro-types/src/mcp_store.rs +++ b/lib/crates/fabro-types/src/mcp_store.rs @@ -3,8 +3,8 @@ //! These types describe MCP server definitions that are stored once on a Fabro //! server and later referenced by name from workflow configs. They are //! persistence-independent: the durable storage lives in the `fabro-mcp-store` -//! crate, which derives `id` (filename stem) and `revision` (content hash) and -//! never persists them inside the TOML body. +//! crate, which persists `id` and a content-hash `revision` alongside the +//! normalized definition fields. //! //! Transport is the existing [`McpTransport`](crate::settings::McpTransport) //! reused verbatim, so a stored definition uses the same `stdio`/`http`/ @@ -28,11 +28,9 @@ use crate::settings::run::McpHttpProtocol; /// A server-managed MCP server definition. /// -/// `id` and `revision` are derived (filename stem + content hash of the -/// persisted TOML bytes) and are not stored in the persisted TOML body. This is -/// the internal/persistence model and carries full transport values; it is not -/// serialized to clients. Read APIs return [`McpServerView`] instead, which -/// omits env/header values. +/// This is the internal/persistence model and carries full transport values; it +/// is not serialized to clients. Read APIs return [`McpServerView`] instead, +/// which omits env/header values. #[derive(Debug, Clone, PartialEq)] pub struct McpServerDefinition { pub id: McpServerId, @@ -44,8 +42,8 @@ pub struct McpServerDefinition { pub tool_timeout_secs: u64, } -/// Fields supplied when creating a new definition. Carries an `id` (the create -/// call assigns the filename) but no `revision` (the store derives it). +/// Fields supplied when creating a new definition. Carries an `id` but no +/// `revision` (the store derives it). #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct McpServerDraft { @@ -236,6 +234,7 @@ fn sorted_keys_ref(map: &HashMap) -> Vec { pub enum McpServerValidationError { InvalidMcpServerId { value: String }, EmptyName, + TimeoutOutOfRange { field: &'static str, value: u64 }, InvalidTransport { reason: String }, } @@ -249,6 +248,12 @@ impl fmt::Display for McpServerValidationError { ) } Self::EmptyName => f.write_str("mcp server display name must not be empty"), + Self::TimeoutOutOfRange { field, value } => { + write!( + f, + "mcp server {field} value {value} exceeds the signed 64-bit range" + ) + } Self::InvalidTransport { reason } => { write!(f, "mcp server transport is invalid: {reason}") } @@ -258,8 +263,7 @@ impl fmt::Display for McpServerValidationError { impl std::error::Error for McpServerValidationError {} -/// An MCP server id: lowercase, matches `^[a-z0-9][a-z0-9-]{0,62}$`, and equals -/// the persisted file's stem. +/// An MCP server id: lowercase and matches `^[a-z0-9][a-z0-9-]{0,62}$`. #[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)] pub struct McpServerId(String); @@ -312,8 +316,9 @@ impl<'de> Deserialize<'de> for McpServerId { } } -/// A revision: the lowercase SHA-256 hex of a definition's canonical persisted -/// TOML bytes. Used as an ETag for optimistic concurrency. +/// A revision: lowercase SHA-256 hex of a definition's canonical bytes. Used as +/// an ETag for optimistic concurrency. Legacy imports preserve the hash of the +/// original TOML bytes. #[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)] pub struct McpServerRevision(String); @@ -406,9 +411,17 @@ pub fn validate_mcp_server_fields( if replace.display_name.trim().is_empty() { return Err(McpServerValidationError::EmptyName); } + validate_timeout("startup_timeout_secs", replace.startup_timeout_secs)?; + validate_timeout("tool_timeout_secs", replace.tool_timeout_secs)?; validate_transport(&replace.transport) } +fn validate_timeout(field: &'static str, value: u64) -> Result<(), McpServerValidationError> { + i64::try_from(value) + .map(|_| ()) + .map_err(|_| McpServerValidationError::TimeoutOutOfRange { field, value }) +} + fn validate_transport(transport: &McpTransport) -> Result<(), McpServerValidationError> { match transport { McpTransport::Stdio { command, .. } | McpTransport::Sandbox { command, .. } => { @@ -515,6 +528,19 @@ mod tests { assert!(validate_mcp_server_fields(&replace).is_err()); } + #[test] + fn validation_rejects_timeout_outside_sqlite_integer_range() { + let replace = McpServerReplace { + display_name: "Sentry".to_string(), + description: None, + transport: http_transport(), + startup_timeout_secs: u64::MAX, + tool_timeout_secs: 60, + }; + + assert!(validate_mcp_server_fields(&replace).is_err()); + } + #[test] fn validation_accepts_well_formed_definition() { let replace = McpServerReplace { diff --git a/lib/crates/fabro-types/src/settings/run.rs b/lib/crates/fabro-types/src/settings/run.rs index 51410c44d..a582ddee9 100644 --- a/lib/crates/fabro-types/src/settings/run.rs +++ b/lib/crates/fabro-types/src/settings/run.rs @@ -2180,14 +2180,34 @@ pub enum McpTransport { }, } -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[derive( + Debug, + Clone, + Copy, + Default, + PartialEq, + Eq, + strum::Display, + strum::EnumString, + strum::IntoStaticStr, + Serialize, + Deserialize, +)] #[serde(rename_all = "snake_case")] +#[strum(serialize_all = "snake_case")] pub enum McpHttpProtocol { #[default] StreamableHttp, Sse, } +impl McpHttpProtocol { + #[must_use] + pub fn as_str(self) -> &'static str { + self.into() + } +} + #[expect( clippy::trivially_copy_pass_by_ref, reason = "serde skip_serializing_if helpers receive borrowed field values"