fix(storage): repair call log columns in versioned user databases

memoria/crates/memoria-api/src/auth.rs

@@ -1047,6 +1047,10 @@ impl CallLogBatcher {
                         continue;
                     }
                 };
+                if let Err(e) = user_store.ensure_call_log_tool_outcome_schema().await {
+                    warn!("call_log schema repair failed for user {user_id}: {e}");
+                    continue;
+                }
                 let table = user_store.t("mem_api_call_log");
                 if let Err(e) = flush_call_log_chunked(user_store.pool(), &table, &entries).await {
                     warn!("call_log batch flush failed for user {user_id}: {e}");
@@ -1059,6 +1063,10 @@ impl CallLogBatcher {
             }
             return;
         }
+        if let Err(e) = sql.ensure_call_log_tool_outcome_schema().await {
+            warn!("call_log schema repair failed: {e}");
+            return;
+        }
         if let Err(e) = flush_call_log_chunked(sql.pool(), "mem_api_call_log", &entries).await {
             warn!(
                 "call_log batch flush failed ({} entries): {e}",

memoria/crates/memoria-api/tests/api_e2e.rs

@@ -2076,6 +2076,105 @@ async fn test_mcp_tool_outcomes_persist_and_remain_queryable_by_tool() {
     assert_eq!(backend["tool_backend_error_count"], 1);
 }
 
+#[tokio::test]
+async fn test_call_log_repair_failure_keeps_memory_api_available_and_background_recovers() {
+    let master = "log-repair-test-master";
+    let (base, client, server) = spawn_server_with_master_key(master).await;
+    let user = uid();
+    server.user_store(&user).await;
+    let pool = server.user_db_pool(&user).await;
+    // Simulate a version-2 database whose optional logging table is unavailable.
+    for column in ["tool_success", "tool_error_kind"] {
+        sqlx::query(&format!(
+            "ALTER TABLE mem_api_call_log DROP COLUMN {column}"
+        ))
+        .execute(&pool)
+        .await
+        .unwrap();
+    }
+    sqlx::query("ALTER TABLE mem_api_call_log RENAME TO saved_call_log")
+        .execute(&pool)
+        .await
+        .unwrap();
+    server.router().invalidate_user(&user).await;
+    let response = client
+        .post(format!("{base}/v1/memories"))
+        .bearer_auth(master)
+        .header("X-User-Id", &user)
+        .json(&json!({"content":"A durable preference during a logging outage"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), 201);
+    let memory: Value = response.json().await.unwrap();
+    let memory_id = memory["memory_id"].as_str().expect("stored memory id");
+    let response = client
+        .get(format!("{base}/v1/memories/{memory_id}"))
+        .bearer_auth(master)
+        .header("X-User-Id", &user)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), 200);
+    let cached_store = server.service().user_sql_store(&user).await.unwrap();
+    server
+        .state()
+        .call_log_batcher
+        .flush(&server.service())
+        .await;
+
+    // Restore the legacy table without invalidating the cached user store.
+    sqlx::query("ALTER TABLE saved_call_log RENAME TO mem_api_call_log")
+        .execute(&pool)
+        .await
+        .unwrap();
+    server.state().call_log_batcher.record_rpc(
+        user.clone(),
+        "POST".into(),
+        "/mcp/memory_retrieve".into(),
+        200,
+        1,
+        memoria_api::auth::RpcMeta {
+            success: true,
+            error_code: None,
+            tool_success: Some(false),
+            tool_error_kind: Some("backend"),
+        },
+    );
+    server
+        .state()
+        .call_log_batcher
+        .flush(&server.service())
+        .await;
+    let store = server.service().user_sql_store(&user).await.unwrap();
+    assert!(Arc::ptr_eq(&cached_store, &store));
+    let outcome: (Option<i8>, Option<String>) = sqlx::query_as(
+        "SELECT tool_success, tool_error_kind FROM mem_api_call_log
+         WHERE user_id = ? AND path = '/mcp/memory_retrieve'",
+    )
+    .bind(&user)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(outcome, (Some(0), Some("backend".into())));
+    let response = client
+        .get(format!("{base}/admin/users/{user}/call-stats"))
+        .bearer_auth(master)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), 200);
+    let stats: Value = response.json().await.unwrap();
+    let entry = stats["by_path"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .find(|entry| entry["path"] == "/mcp/memory_retrieve")
+        .expect("recovered tool statistics");
+    assert_eq!(entry["tool_backend_error_count"], 1);
+    pool.close().await;
+}
+
 async fn assert_scoped_user_has_no_database(server: &support::multi_db::ApiTestServer, user: &str) {
     let registry: i64 =
         sqlx::query_scalar("SELECT COUNT(*) FROM mem_user_registry WHERE user_id = ?")

memoria/crates/memoria-storage/src/store.rs

@@ -8,7 +8,7 @@ use memoria_core::{
 use sqlx::{mysql::MySqlPool, MySql, QueryBuilder, Row};
 use std::borrow::Cow;
 use std::str::FromStr;
-use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -900,6 +900,8 @@ fn mo_to_vec(s: &str) -> Result<Vec<f32>, MemoriaError> {
 pub struct SqlMemoryStore {
     pool: MySqlPool,
     branch_alter_capability: Arc<crate::branch_capability::BranchAlterCapability>,
+    /// Successful optional log-schema checks are shared by clones of this store.
+    call_log_tool_outcomes_ready: Arc<AtomicBool>,
     embedding_dim: usize,
     instance_id: String,
     database_url: Option<String>,
@@ -998,6 +1000,7 @@ impl SqlMemoryStore {
             branch_alter_capability: Arc::new(
                 crate::branch_capability::BranchAlterCapability::default(),
             ),
+            call_log_tool_outcomes_ready: Arc::new(AtomicBool::new(false)),
             embedding_dim,
             instance_id,
             database_url: None,
@@ -1090,12 +1093,14 @@ impl SqlMemoryStore {
     pub fn set_db_name(&mut self, name: String) {
         self.branch_alter_capability =
             Arc::new(crate::branch_capability::BranchAlterCapability::default());
+        self.call_log_tool_outcomes_ready = Arc::new(AtomicBool::new(false));
         self.db_name = Some(name);
     }
 
     pub fn set_database_url(&mut self, url: String) {
         self.branch_alter_capability =
             Arc::new(crate::branch_capability::BranchAlterCapability::default());
+        self.call_log_tool_outcomes_ready = Arc::new(AtomicBool::new(false));
         self.database_url = Some(url);
     }
 
@@ -1720,24 +1725,8 @@ impl SqlMemoryStore {
             }
         }
 
-        // Nullable fields distinguish historical/non-tool calls from actual tool
-        // success. Keep additive migrations compatible with older writers.
-        for column in [
-            "tool_success TINYINT(1) NULL",
-            "tool_error_kind VARCHAR(32) NULL",
-        ] {
-            if let Err(error) = sqlx::query(&format!(
-                "ALTER TABLE {api_call_log_table} ADD COLUMN {column}"
-            ))
-            .execute(pool)
-            .await
-            {
-                if !is_duplicate_column(&error) {
-                    tracing::error!(%error, column, "Cannot migrate tool outcome call-log columns");
-                    return Err(db_err(error));
-                }
-            }
-        }
+        self.ensure_call_log_tool_outcome_columns(pool, schema_name)
+            .await?;
 
         let has_feedback_memory_user_idx = info_schema_index_exists(
             pool,
@@ -2264,6 +2253,59 @@ impl SqlMemoryStore {
         }
     }
 
+    /// Retry optional call-log repairs from the background writer, even when the
+    /// router has already cached an otherwise usable user schema. Failed checks
+    /// are not cached; successful checks avoid metadata queries on later batches.
+    pub async fn ensure_call_log_tool_outcome_schema(&self) -> Result<(), MemoriaError> {
+        if self.call_log_tool_outcomes_ready.load(Ordering::Relaxed) {
+            return Ok(());
+        }
+        let schema_name = self.current_schema_name().await?;
+        self.ensure_call_log_tool_outcome_columns(&self.pool, schema_name.as_ref())
+            .await
+    }
+
+    /// Tool outcomes were added after version 2 was introduced, without a version
+    /// bump. Repair just these additive columns, without replaying historical
+    /// memory/branch migrations or changing the version marker during a rollout.
+    async fn ensure_call_log_tool_outcome_columns(
+        &self,
+        pool: &MySqlPool,
+        schema_name: &str,
+    ) -> Result<(), MemoriaError> {
+        if self.call_log_tool_outcomes_ready.load(Ordering::Relaxed) {
+            return Ok(());
+        }
+        let table = self.t("mem_api_call_log");
+        // NULL preserves historical/non-tool outcomes and permits older API
+        // instances to continue inserting their original column list.
+        for (name, definition) in [
+            ("tool_success", "tool_success TINYINT(1) NULL"),
+            ("tool_error_kind", "tool_error_kind VARCHAR(32) NULL"),
+        ] {
+            if info_schema_column_exists(pool, schema_name, "mem_api_call_log", name).await {
+                continue;
+            }
+            let sql = format!("ALTER TABLE {table} ADD COLUMN {definition}");
+            if let Err(error) = exec_ddl_with_retry(pool, &sql).await {
+                // Another instance can win between the metadata read and ALTER.
+                // MatrixOne may also report a stale secondary-index table; accept
+                // that race only when the requested column actually exists.
+                let repaired_by_other_instance = is_duplicate_column(&error)
+                    || (is_mo_concurrent_ddl_race(&error)
+                        && info_schema_column_exists(pool, schema_name, "mem_api_call_log", name)
+                            .await);
+                if !repaired_by_other_instance {
+                    tracing::error!(%error, column = name, "Cannot migrate tool outcome call-log columns");
+                    return Err(db_err(error));
+                }
+            }
+        }
+        self.call_log_tool_outcomes_ready
+            .store(true, Ordering::Relaxed);
+        Ok(())
+    }
+
     pub async fn migrate_user(&self) -> Result<(), MemoriaError> {
         let pool = &self.pool;
         let meta_table = self.t("mem_schema_meta");
@@ -2293,6 +2335,14 @@ impl SqlMemoryStore {
         if load_user_schema_version(pool, &meta_table).await? == Some(CURRENT_USER_SCHEMA_VERSION)
             && info_schema_table_exists(pool, schema_name, "mem_memories").await
         {
+            // Monitoring failures must not make an existing user's memories
+            // unavailable. The background log writer retries this optional repair.
+            if let Err(error) = self
+                .ensure_call_log_tool_outcome_columns(pool, schema_name)
+                .await
+            {
+                tracing::warn!(%error, schema_name, "Optional call-log schema repair deferred to background writer");
+            }
             return Ok(());
         }