diff --git a/Cargo.lock b/Cargo.lock index 2f4d2d60..0470dd00 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2650,6 +2650,7 @@ dependencies = [ "serde_json", "serde_yaml", "tempfile", + "thiserror", "tokio", "tokio-tungstenite", "tower", diff --git a/crates/connect-agent/Cargo.toml b/crates/connect-agent/Cargo.toml index 11ecc4f6..0424bad9 100644 --- a/crates/connect-agent/Cargo.toml +++ b/crates/connect-agent/Cargo.toml @@ -27,6 +27,7 @@ serde.workspace = true serde_json.workspace = true serde_yaml.workspace = true tempfile.workspace = true +thiserror.workspace = true tokio.workspace = true tokio-tungstenite.workspace = true tracing.workspace = true diff --git a/crates/connect-agent/src/relay.rs b/crates/connect-agent/src/relay.rs index bbb66688..7d02ef69 100644 --- a/crates/connect-agent/src/relay.rs +++ b/crates/connect-agent/src/relay.rs @@ -3,6 +3,7 @@ use crate::admission::{ }; use crate::server::AgentState; use futures_util::{SinkExt, StreamExt}; +use mdbase_connect_core::ConnectError; use mdbase_connect_protocol::{ AgentConnectionState, ConnectContractSupport, ConnectOperationOutcome, ConnectProblem, RelayFileFrame, RelayFileKind, RelayMessage, CONTROL_PROTOCOL_VERSION, @@ -40,7 +41,12 @@ async fn connect_once( connector_token: &str, state: Arc, ) -> Result<(), Box> { - sync_collections(client, server_url, connector_token, &state).await?; + sync_collections(client, server_url, connector_token, &state) + .await + .map_err(|error| { + warn_collection_sync_error(&error); + Box::new(error) as Box + })?; let websocket_url = websocket_url(server_url)?; let mut request = websocket_url.as_str().into_client_request()?; request.headers_mut().insert( @@ -309,7 +315,7 @@ async fn connect_once( &connector_token, &state, ).await { - tracing::warn!(%error, "collection sync failed"); + warn_collection_sync_error(&error); } }); } @@ -451,9 +457,20 @@ async fn sync_collections( server_url: &str, connector_token: &str, state: &AgentState, -) -> Result<(), Box> { - let collections = state.collections()?; - let inventory_revision = state.next_inventory_revision()?; +) -> Result<(), CollectionSyncError> { + let collections = state + .collections() + .map_err(|source| CollectionSyncError::Registry { + operation: "list_collections", + source, + })?; + let inventory_revision = + state + .next_inventory_revision() + .map_err(|source| CollectionSyncError::Registry { + operation: "next_inventory_revision", + source, + })?; let payload = serde_json::json!({ "relay_public_key": state.relay_public_key(), "inventory_revision": inventory_revision, @@ -475,11 +492,49 @@ async fn sync_collections( .send() .await?; if !response.status().is_success() { - return Err(format!("collection sync failed with HTTP {}", response.status()).into()); + return Err(CollectionSyncError::Http(response.status())); } Ok(()) } +#[derive(Debug, thiserror::Error)] +enum CollectionSyncError { + #[error("{source}")] + Registry { + operation: &'static str, + #[source] + source: ConnectError, + }, + #[error(transparent)] + Request(#[from] reqwest::Error), + #[error("collection sync failed with HTTP {0}")] + Http(reqwest::StatusCode), +} + +fn warn_collection_sync_error(error: &CollectionSyncError) { + if let CollectionSyncError::Registry { operation, source } = error { + let sqlite = source.registry_sqlite_diagnostic(); + tracing::warn!( + error_code = source.code(), + registry_database = "connector", + registry_operation = *operation, + sqlite_diagnostic_available = sqlite.is_some(), + sqlite_primary_code = sqlite + .as_ref() + .map(|diagnostic| diagnostic.primary_code.as_str()) + .unwrap_or("unavailable"), + sqlite_extended_code = sqlite + .as_ref() + .map(|diagnostic| diagnostic.extended_code) + .unwrap_or(0), + %error, + "collection sync failed" + ); + } else { + tracing::warn!(%error, "collection sync failed"); + } +} + fn websocket_url(server_url: &str) -> Result> { let mut url = Url::parse(server_url)?; match url.scheme() { @@ -511,6 +566,30 @@ mod tests { ); } + #[test] + fn collection_sync_errors_preserve_registry_phase_and_sqlite_subcode() { + let source = ConnectError::Registry(rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_IOERR_FSYNC), + Some("fsync failed".to_string()), + )); + let error = CollectionSyncError::Registry { + operation: "next_inventory_revision", + source, + }; + + let CollectionSyncError::Registry { operation, source } = error else { + panic!("expected registry sync error"); + }; + assert_eq!(operation, "next_inventory_revision"); + assert_eq!( + source.registry_sqlite_diagnostic(), + Some(mdbase_connect_core::RegistrySqliteDiagnostic { + primary_code: "SystemIoFailure".to_string(), + extended_code: rusqlite::ffi::SQLITE_IOERR_FSYNC, + }) + ); + } + #[tokio::test] async fn policy_barrier_orders_generations_and_fails_closed() { let (sender, receiver) = tokio::sync::watch::channel((0_u64, true)); diff --git a/crates/connect-agent/src/runtime_notifications.rs b/crates/connect-agent/src/runtime_notifications.rs index 0e5e06c9..25588d49 100644 --- a/crates/connect-agent/src/runtime_notifications.rs +++ b/crates/connect-agent/src/runtime_notifications.rs @@ -166,7 +166,23 @@ impl RuntimeNotificationService { let collections = match self.local_registry.list() { Ok(collections) => collections, Err(error) => { - tracing::warn!(%error, "notification runtime could not list collections for recovery"); + let sqlite = error.registry_sqlite_diagnostic(); + tracing::warn!( + error_code = error.code(), + registry_database = "connector", + registry_operation = "list_collections", + sqlite_diagnostic_available = sqlite.is_some(), + sqlite_primary_code = sqlite + .as_ref() + .map(|diagnostic| diagnostic.primary_code.as_str()) + .unwrap_or("unavailable"), + sqlite_extended_code = sqlite + .as_ref() + .map(|diagnostic| diagnostic.extended_code) + .unwrap_or(0), + %error, + "notification runtime could not list collections for recovery" + ); return; } }; diff --git a/crates/connect-core/src/lib.rs b/crates/connect-core/src/lib.rs index 1302bf55..07930845 100644 --- a/crates/connect-core/src/lib.rs +++ b/crates/connect-core/src/lib.rs @@ -20,6 +20,7 @@ pub use registry::{ EncryptedRequestClaim, GrantReplayContext, MutationClaim, MutationClaimRequest, MutationJournalDiagnostics, MutationJournalState, MutationLease, MutationRecoveryData, RegistryBackupDiagnostic, RegistryBackupMetadata, RegistryDiagnostics, + RegistrySqliteDiagnostic, }; pub use secrets::SystemSecretStore; pub mod profiling; diff --git a/crates/connect-core/src/registry.rs b/crates/connect-core/src/registry.rs index fa101fa5..998f131f 100644 --- a/crates/connect-core/src/registry.rs +++ b/crates/connect-core/src/registry.rs @@ -32,6 +32,18 @@ pub struct ApplicationSetupResult { pub receipt: Value, } +/// Privacy-safe SQLite identifiers for correlating local registry failures. +/// +/// The primary name distinguishes broad classes such as `SystemIoFailure`, +/// while the extended integer preserves SQLite's exact failure subtype (for +/// example `SQLITE_IOERR_READ` versus `SQLITE_IOERR_FSYNC`). Neither field +/// contains SQL, database contents, or local filesystem paths. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RegistrySqliteDiagnostic { + pub primary_code: String, + pub extended_code: i32, +} + const CONNECT_EXTENSION: &str = "x-mdbase-connect"; const CONNECT_COLLECTION_ID: &str = "collection_id"; const MIRROR_MARKER_DIRECTORY: &str = ".mdbase"; @@ -262,6 +274,17 @@ impl ConnectError { } } + pub fn registry_sqlite_diagnostic(&self) -> Option { + let Self::Registry(error) = self else { + return None; + }; + let sqlite = error.sqlite_error()?; + Some(RegistrySqliteDiagnostic { + primary_code: format!("{:?}", sqlite.code), + extended_code: sqlite.extended_code, + }) + } + pub(crate) fn invalid_collection(diagnostics: Vec) -> Self { let message = diagnostics .iter() diff --git a/crates/connect-core/src/registry/tests.rs b/crates/connect-core/src/registry/tests.rs index e406ccb5..74cdba0f 100644 --- a/crates/connect-core/src/registry/tests.rs +++ b/crates/connect-core/src/registry/tests.rs @@ -242,3 +242,27 @@ implements: provision, ) } + +#[test] +fn registry_sqlite_diagnostic_preserves_extended_result_code() { + let error = ConnectError::Registry(rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_IOERR_READ), + Some("read failed".to_string()), + )); + + assert_eq!( + error.registry_sqlite_diagnostic(), + Some(RegistrySqliteDiagnostic { + primary_code: "SystemIoFailure".to_string(), + extended_code: rusqlite::ffi::SQLITE_IOERR_READ, + }) + ); +} + +#[test] +fn non_sqlite_registry_errors_have_no_sqlite_diagnostic() { + assert_eq!( + ConnectError::CollectionNotFound(Uuid::nil()).registry_sqlite_diagnostic(), + None + ); +}