From faf7a46fd46024a6f00cb3c125bf9d157d5f976e Mon Sep 17 00:00:00 2001 From: lr90 Date: Fri, 9 Oct 2026 12:49:58 +0800 Subject: [PATCH 1/2] fix(storage): repair call log columns in versioned user databases --- memoria/crates/memoria-storage/src/store.rs | 60 +++-- .../tests/call_log_schema_upgrade.rs | 251 ++++++++++++++++++ 2 files changed, 293 insertions(+), 18 deletions(-) create mode 100644 memoria/crates/memoria-storage/tests/call_log_schema_upgrade.rs diff --git a/memoria/crates/memoria-storage/src/store.rs b/memoria/crates/memoria-storage/src/store.rs index 9a3fe5d2..1964b1ec 100644 --- a/memoria/crates/memoria-storage/src/store.rs +++ b/memoria/crates/memoria-storage/src/store.rs @@ -1720,24 +1720,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 +2248,44 @@ impl SqlMemoryStore { } } + /// Tool outcomes were added after schema version 2. Repair just these + /// additive columns on current-version databases, 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> { + 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; + } + if let Err(error) = sqlx::query(&format!("ALTER TABLE {table} ADD COLUMN {definition}")) + .execute(pool) + .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)); + } + } + } + Ok(()) + } + pub async fn migrate_user(&self) -> Result<(), MemoriaError> { let pool = &self.pool; let meta_table = self.t("mem_schema_meta"); @@ -2293,6 +2315,8 @@ 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 { + self.ensure_call_log_tool_outcome_columns(pool, schema_name) + .await?; return Ok(()); } diff --git a/memoria/crates/memoria-storage/tests/call_log_schema_upgrade.rs b/memoria/crates/memoria-storage/tests/call_log_schema_upgrade.rs new file mode 100644 index 00000000..6dab7405 --- /dev/null +++ b/memoria/crates/memoria-storage/tests/call_log_schema_upgrade.rs @@ -0,0 +1,251 @@ +use memoria_storage::store::CURRENT_USER_SCHEMA_VERSION; +use memoria_storage::{DbRouter, SqlMemoryStore}; +use sqlx::{mysql::MySqlPoolOptions, MySqlPool, Row}; +use uuid::Uuid; + +struct LegacyFixture { + router: DbRouter, + pool: MySqlPool, + user: String, + user_url: String, +} + +async fn legacy_fixture(version: i64, partial: bool) -> LegacyFixture { + let url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "mysql://root:111@localhost:6001/memoria_test".into()); + let (base, _) = url.rsplit_once('/').expect("database URL"); + let shared_url = format!("{base}/call_log_upgrade_{}", Uuid::new_v4().simple()); + let router = DbRouter::connect(&shared_url, 1024, Uuid::new_v4().to_string()) + .await + .expect("connect router"); + let user = format!("log_upgrade_{}", Uuid::new_v4().simple()); + router.user_store(&user).await.expect("create user schema"); + let db = router.user_db_name(&user).await.expect("user database"); + let user_url = format!("{base}/{db}"); + let pool = MySqlPoolOptions::new() + .max_connections(4) + .connect(&user_url) + .await + .expect("connect user pool"); + + // Recreate the actual pre-#258 table, including an existing history row. + sqlx::query("DROP TABLE mem_api_call_log") + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "CREATE TABLE mem_api_call_log ( + id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, + user_id VARCHAR(64) NOT NULL, + method VARCHAR(10) NOT NULL DEFAULT '', + path VARCHAR(256) NOT NULL, + status_code SMALLINT NOT NULL DEFAULT 0, + latency_ms INT NOT NULL DEFAULT 0, + called_at DATETIME(6) NOT NULL DEFAULT NOW(6), + rpc_success TINYINT(1) NOT NULL DEFAULT 1, + rpc_error_code INT NULL, + INDEX idx_user_called (user_id, called_at) + )", + ) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO mem_api_call_log (user_id, method, path, status_code, latency_ms) + VALUES (?, 'GET', '/historical', 200, 17)", + ) + .bind(&user) + .execute(&pool) + .await + .unwrap(); + if partial { + sqlx::query("ALTER TABLE mem_api_call_log ADD COLUMN tool_success TINYINT(1) NULL") + .execute(&pool) + .await + .unwrap(); + sqlx::query("UPDATE mem_api_call_log SET tool_success = 0 WHERE path = '/historical'") + .execute(&pool) + .await + .unwrap(); + } + sqlx::query( + "INSERT INTO mem_memories + (memory_id, user_id, memory_type, content, initial_confidence, trust_tier, + source_event_ids, extra_metadata, observed_at, created_at) + VALUES ('preserved-memory', ?, 'semantic', 'Keep existing memory content', 0.8, 'T3', + '[]', '{}', NOW(6), NOW(6))", + ) + .bind(&user) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "UPDATE mem_schema_meta SET schema_version = ?, updated_at = '2000-01-01 00:00:00' + WHERE schema_key = 'user_schema'", + ) + .bind(version) + .execute(&pool) + .await + .unwrap(); + router.invalidate_user(&user).await; + LegacyFixture { + router, + pool, + user, + user_url, + } +} + +async fn verify_usable_history(f: &LegacyFixture, partial: bool) { + let columns = sqlx::query( + "SELECT column_name, is_nullable FROM information_schema.columns + WHERE table_schema = DATABASE() AND table_name = 'mem_api_call_log' + AND column_name IN ('tool_success', 'tool_error_kind')", + ) + .fetch_all(&f.pool) + .await + .unwrap(); + assert_eq!(columns.len(), 2); + assert!(columns + .iter() + .all(|r| r.get::("is_nullable") == "YES")); + let historical = sqlx::query( + "SELECT latency_ms, tool_success, tool_error_kind FROM mem_api_call_log + WHERE path = '/historical'", + ) + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!(historical.get::("latency_ms"), 17); + assert_eq!( + historical.get::, _>("tool_success"), + partial.then_some(0) + ); + assert_eq!(historical.get::, _>("tool_error_kind"), None); + + // Old and new API instances must both be able to write during a rolling update. + sqlx::query( + "INSERT INTO mem_api_call_log + (user_id, method, path, status_code, latency_ms, rpc_success, rpc_error_code) + VALUES (?, 'GET', '/old-writer', 200, 11, 1, NULL)", + ) + .bind(&f.user) + .execute(&f.pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO mem_api_call_log + (user_id, method, path, status_code, latency_ms, rpc_success, rpc_error_code, + tool_success, tool_error_kind) + VALUES (?, 'POST', '/new-writer', 200, 12, 1, NULL, 0, 'backend')", + ) + .bind(&f.user) + .execute(&f.pool) + .await + .unwrap(); + let stats = sqlx::query( + "SELECT COUNT(*) AS total, + CAST(SUM(CASE WHEN status_code >= 400 OR rpc_success = 0 OR tool_success = 0 + THEN 1 ELSE 0 END) AS SIGNED) AS errors + FROM mem_api_call_log", + ) + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!(stats.get::("total"), 3); + assert_eq!(stats.get::("errors"), if partial { 2 } else { 1 }); + let content: String = + sqlx::query_scalar("SELECT content FROM mem_memories WHERE memory_id = 'preserved-memory'") + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!(content, "Keep existing memory content"); +} + +async fn routed_upgrade(version: i64, partial: bool) { + let f = legacy_fixture(version, partial).await; + f.router + .user_store(&f.user) + .await + .expect("upgrade legacy user"); + verify_usable_history(&f, partial).await; + let marker = sqlx::query("SELECT schema_version, updated_at FROM mem_schema_meta") + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!( + marker.get::("schema_version"), + CURRENT_USER_SCHEMA_VERSION + ); + let updated_at = marker.get::("updated_at"); + if version == CURRENT_USER_SCHEMA_VERSION { + assert_eq!(updated_at.to_string(), "2000-01-01 00:00:00"); + } + f.router.invalidate_user(&f.user).await; + f.router + .user_store(&f.user) + .await + .expect("idempotent reentry"); + let after: chrono::NaiveDateTime = sqlx::query_scalar( + "SELECT updated_at FROM mem_schema_meta WHERE schema_key = 'user_schema'", + ) + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!(after, updated_at); + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM mem_api_call_log") + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!(count, 3); +} + +#[tokio::test] +async fn version_1_user_can_upgrade_and_preserve_call_history() { + routed_upgrade(1, false).await; +} + +#[tokio::test] +async fn version_2_user_repairs_missing_tool_outcomes_without_rewriting_marker() { + routed_upgrade(2, false).await; +} + +#[tokio::test] +async fn version_2_user_finishes_partial_tool_outcome_migration() { + routed_upgrade(2, true).await; +} + +#[tokio::test] +async fn two_instances_can_repair_version_2_tool_outcomes() { + let f = legacy_fixture(2, false).await; + let a = SqlMemoryStore::connect(&f.user_url, 1024, "instance-a".into()) + .await + .unwrap(); + let b = SqlMemoryStore::connect(&f.user_url, 1024, "instance-b".into()) + .await + .unwrap(); + let (a_result, b_result) = tokio::join!(a.migrate_user(), b.migrate_user()); + a_result.expect("first instance migration"); + b_result.expect("second instance migration"); + verify_usable_history(&f, false).await; +} + +#[tokio::test] +async fn failed_log_repair_does_not_mark_user_schema_ready() { + let f = legacy_fixture(2, false).await; + sqlx::query("ALTER TABLE mem_api_call_log RENAME TO saved_call_log") + .execute(&f.pool) + .await + .unwrap(); + assert!(f.router.user_store(&f.user).await.is_err()); + sqlx::query("ALTER TABLE saved_call_log RENAME TO mem_api_call_log") + .execute(&f.pool) + .await + .unwrap(); + // Retry on the same router: failed initialization must not cache readiness. + f.router + .user_store(&f.user) + .await + .expect("retry repairs the restored legacy log table"); + verify_usable_history(&f, false).await; +} From 155c6302172ce249918a2fb3d558b35a285ae7b6 Mon Sep 17 00:00:00 2001 From: lr90 Date: Fri, 9 Oct 2026 13:04:33 +0800 Subject: [PATCH 2/2] fix(storage): keep optional call log repair failures non-fatal --- memoria/crates/memoria-api/src/auth.rs | 8 ++ memoria/crates/memoria-api/tests/api_e2e.rs | 99 +++++++++++++++++++ memoria/crates/memoria-storage/src/store.rs | 44 +++++++-- .../tests/call_log_schema_upgrade.rs | 56 +++++++++-- 4 files changed, 192 insertions(+), 15 deletions(-) diff --git a/memoria/crates/memoria-api/src/auth.rs b/memoria/crates/memoria-api/src/auth.rs index fb9970b3..0b23a4c9 100644 --- a/memoria/crates/memoria-api/src/auth.rs +++ b/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}", diff --git a/memoria/crates/memoria-api/tests/api_e2e.rs b/memoria/crates/memoria-api/tests/api_e2e.rs index 8c274871..ef572e8b 100644 --- a/memoria/crates/memoria-api/tests/api_e2e.rs +++ b/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, Option) = 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 = ?") diff --git a/memoria/crates/memoria-storage/src/store.rs b/memoria/crates/memoria-storage/src/store.rs index 1964b1ec..147183be 100644 --- a/memoria/crates/memoria-storage/src/store.rs +++ b/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, MemoriaError> { pub struct SqlMemoryStore { pool: MySqlPool, branch_alter_capability: Arc, + /// Successful optional log-schema checks are shared by clones of this store. + call_log_tool_outcomes_ready: Arc, embedding_dim: usize, instance_id: String, database_url: Option, @@ -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); } @@ -2248,14 +2253,29 @@ impl SqlMemoryStore { } } - /// Tool outcomes were added after schema version 2. Repair just these - /// additive columns on current-version databases, without replaying historical + /// 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. @@ -2266,10 +2286,8 @@ impl SqlMemoryStore { if info_schema_column_exists(pool, schema_name, "mem_api_call_log", name).await { continue; } - if let Err(error) = sqlx::query(&format!("ALTER TABLE {table} ADD COLUMN {definition}")) - .execute(pool) - .await - { + 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. @@ -2283,6 +2301,8 @@ impl SqlMemoryStore { } } } + self.call_log_tool_outcomes_ready + .store(true, Ordering::Relaxed); Ok(()) } @@ -2315,8 +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 { - self.ensure_call_log_tool_outcome_columns(pool, schema_name) - .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(()); } diff --git a/memoria/crates/memoria-storage/tests/call_log_schema_upgrade.rs b/memoria/crates/memoria-storage/tests/call_log_schema_upgrade.rs index 6dab7405..847932b6 100644 --- a/memoria/crates/memoria-storage/tests/call_log_schema_upgrade.rs +++ b/memoria/crates/memoria-storage/tests/call_log_schema_upgrade.rs @@ -1,3 +1,4 @@ +use memoria_core::interfaces::MemoryStore; use memoria_storage::store::CURRENT_USER_SCHEMA_VERSION; use memoria_storage::{DbRouter, SqlMemoryStore}; use sqlx::{mysql::MySqlPoolOptions, MySqlPool, Row}; @@ -8,13 +9,32 @@ struct LegacyFixture { pool: MySqlPool, user: String, user_url: String, + shared_db: String, + user_db: String, +} + +impl LegacyFixture { + async fn cleanup(self) { + self.router.invalidate_user(&self.user).await; + self.pool.close().await; + // Only databases owned by this fixture are removed. + for db in [&self.user_db, &self.shared_db] { + sqlx::query(&format!("DROP DATABASE IF EXISTS `{db}`")) + .execute(self.router.global_user_pool()) + .await + .expect("clean up fixture database"); + } + self.router.global_user_pool().close().await; + self.router.shared_pool().close().await; + } } async fn legacy_fixture(version: i64, partial: bool) -> LegacyFixture { let url = std::env::var("DATABASE_URL") .unwrap_or_else(|_| "mysql://root:111@localhost:6001/memoria_test".into()); let (base, _) = url.rsplit_once('/').expect("database URL"); - let shared_url = format!("{base}/call_log_upgrade_{}", Uuid::new_v4().simple()); + let shared_db = format!("call_log_upgrade_{}", Uuid::new_v4().simple()); + let shared_url = format!("{base}/{shared_db}"); let router = DbRouter::connect(&shared_url, 1024, Uuid::new_v4().to_string()) .await .expect("connect router"); @@ -93,6 +113,8 @@ async fn legacy_fixture(version: i64, partial: bool) -> LegacyFixture { pool, user, user_url, + shared_db, + user_db: db, } } @@ -198,6 +220,7 @@ async fn routed_upgrade(version: i64, partial: bool) { .await .unwrap(); assert_eq!(count, 3); + f.cleanup().await; } #[tokio::test] @@ -228,24 +251,45 @@ async fn two_instances_can_repair_version_2_tool_outcomes() { a_result.expect("first instance migration"); b_result.expect("second instance migration"); verify_usable_history(&f, false).await; + a.pool().close().await; + b.pool().close().await; + f.cleanup().await; } #[tokio::test] -async fn failed_log_repair_does_not_mark_user_schema_ready() { +async fn failed_log_repair_keeps_memories_available_and_retries_on_cached_store() { let f = legacy_fixture(2, false).await; sqlx::query("ALTER TABLE mem_api_call_log RENAME TO saved_call_log") .execute(&f.pool) .await .unwrap(); - assert!(f.router.user_store(&f.user).await.is_err()); + let store = f + .router + .user_store(&f.user) + .await + .expect("memory store stays available"); + let memory = store + .get("preserved-memory") + .await + .unwrap() + .expect("existing memory"); + assert_eq!(memory.content, "Keep existing memory content"); + assert!(store.ensure_call_log_tool_outcome_schema().await.is_err()); sqlx::query("ALTER TABLE saved_call_log RENAME TO mem_api_call_log") .execute(&f.pool) .await .unwrap(); - // Retry on the same router: failed initialization must not cache readiness. - f.router + // Retry through the same cached store without invalidating the router. + let cached = f + .router .user_store(&f.user) .await - .expect("retry repairs the restored legacy log table"); + .expect("cached user remains available"); + assert!(std::sync::Arc::ptr_eq(&store, &cached)); + cached + .ensure_call_log_tool_outcome_schema() + .await + .expect("retry repairs the restored log table"); verify_usable_history(&f, false).await; + f.cleanup().await; }