Add SQLite IOERR correlation diagnostics

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

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<AgentState>,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    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<dyn std::error::Error + Send + Sync>
+        })?;
     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<dyn std::error::Error + Send + Sync>> {
-    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<Url, Box<dyn std::error::Error + Send + Sync>> {
     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));

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;
             }
         };

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;

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<RegistrySqliteDiagnostic> {
+        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<mdbase::v03::Diagnostic>) -> Self {
         let message = diagnostics
             .iter()

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
+    );
+}