From d5c0b50e63eff1f2fd9b9363aa504dd1a190e59e Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Mon, 17 Aug 2026 17:52:30 +0800 Subject: [PATCH] =?UTF-8?q?fix(storage):=20=E5=BD=92=E4=B8=80=E5=8C=96?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E5=89=8D=E7=BC=80=E7=B4=AF=E7=A7=AF=E6=B1=A1?= =?UTF-8?q?=E6=9F=93=E7=9A=84=E4=BC=9A=E8=AF=9D=20ID=EF=BC=88=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E6=80=A7=E8=BF=81=E7=A7=BB=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 早期版本重复拼接通道前缀产生 websocket:websocket: 一类的污染会话记录，会话列表因此出现多条前缀逐层累积的重复会话 - 新增 user_version=2 迁移：子表数据改指归一化 ID，污染会话并入既有会话或重命名，seq 按时间重编号，整体事务保证原子性 - 新增回归测试覆盖合并、重命名、seq 连续性与幂等重跑 --- src/storage/migrations.rs | 170 ++++++++++++++++++++++++++++++++++++++ src/storage/mod.rs | 3 +- src/storage/tests.rs | 77 +++++++++++++++++ 3 files changed, 249 insertions(+), 1 deletion(-) diff --git a/src/storage/migrations.rs b/src/storage/migrations.rs index 13e33b2..5665d22 100644 --- a/src/storage/migrations.rs +++ b/src/storage/migrations.rs @@ -5,6 +5,7 @@ //! every [`super::SessionStore`] construction and are idempotent. use rusqlite::Connection; +use rusqlite::params; use super::StorageError; @@ -213,6 +214,175 @@ pub(super) fn ensure_memory_scope_key_migration(conn: &Connection) -> Result<(), Ok(()) } +/// 修复历史 bug 污染的会话 ID（user_version 2，一次性迁移）。 +/// +/// 早期版本在 chat_id 已带通道前缀时会再次拼接 `{channel}:{chat_id}`， +/// 产生形如 `websocket:websocket:` 的污染会话记录，且每次重连都会 +/// 多累积一层前缀、多出一条孤儿会话。`persistent_session_id` 现已幂等， +/// 不会再产生新污染，但存量污染记录会在会话列表中显示为多个重复会话。 +/// +/// 本迁移将污染 ID 归一化：子表数据（消息/话题/技能事件/待办/子代理/记忆溯源） +/// 改指归一化 ID，污染会话记录本身被删除或重命名。通过 PRAGMA user_version +/// 追踪，仅执行一次；整体包在事务中，失败自动回滚，重启后可安全重跑。 +pub(super) fn repair_session_id_prefix_pollution( + conn: &mut Connection, +) -> Result<(), StorageError> { + const SESSION_ID_REPAIR_VERSION: i64 = 2; + + let current_version: i64 = conn.query_row("PRAGMA user_version", [], |row| row.get(0))?; + if current_version >= SESSION_ID_REPAIR_VERSION { + return Ok(()); + } + + let tx = conn.transaction()?; + repair_session_id_prefix_pollution_inner(&tx)?; + tx.execute( + &format!("PRAGMA user_version = {SESSION_ID_REPAIR_VERSION}"), + [], + )?; + tx.commit()?; + + Ok(()) +} + +fn repair_session_id_prefix_pollution_inner(conn: &Connection) -> Result<(), StorageError> { + let candidates: Vec<(String, String, String)> = { + let mut stmt = conn.prepare("SELECT id, channel_name, chat_id FROM sessions")?; + let rows = stmt.query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + )) + })?; + rows.collect::, _>>()? + }; + + let mut repaired = 0usize; + for (polluted_id, channel_name, chat_id) in candidates { + let normalized = super::persistent_session_id(&channel_name, &polluted_id); + if normalized == polluted_id { + continue; + } + + let target_exists: bool = conn + .query_row( + "SELECT COUNT(*) FROM sessions WHERE id = ?1", + params![normalized], + |row| row.get::<_, i64>(0), + ) + .map(|count| count > 0)?; + + if target_exists { + // 并入既有归一化会话：先把污染消息的 seq 偏移大数， + // 避免改指 session_id 时违反 UNIQUE(session_id, seq)，合并后统一重编号。 + conn.execute( + "UPDATE messages SET seq = seq + 1000000000 WHERE session_id = ?1", + params![polluted_id], + )?; + } else { + // 无归一化会话：先建归一化 ID 的记录（子表改指时父记录必须已存在， + // 否则会违反外键约束），最后删除污染记录。 + // websocket/cli 的 chat_id 与 session id 同源，顺带剥离累积前缀。 + let clean_chat_id = if channel_name == "cli" || channel_name == "websocket" { + super::persistent_session_id(&channel_name, &chat_id) + } else { + chat_id + }; + conn.execute( + "INSERT INTO sessions ( + id, title, channel_name, chat_id, summary, + created_at, updated_at, last_active_at, archived_at, deleted_at, + message_count, user_turn_count, agent_prompt_reinjection_count + ) + SELECT ?2, title, channel_name, ?3, summary, + created_at, updated_at, last_active_at, archived_at, deleted_at, + message_count, user_turn_count, agent_prompt_reinjection_count + FROM sessions WHERE id = ?1", + params![polluted_id, normalized, clean_chat_id], + )?; + } + + conn.execute( + "UPDATE messages SET session_id = ?2 WHERE session_id = ?1", + params![polluted_id, normalized], + )?; + conn.execute( + "UPDATE topics SET session_id = ?2 WHERE session_id = ?1", + params![polluted_id, normalized], + )?; + conn.execute( + "UPDATE skill_events SET session_id = ?2 WHERE session_id = ?1", + params![polluted_id, normalized], + )?; + conn.execute( + "UPDATE todos SET session_id = ?2 WHERE session_id = ?1", + params![polluted_id, normalized], + )?; + conn.execute( + "UPDATE pending_subagents SET parent_session_id = ?2 WHERE parent_session_id = ?1", + params![polluted_id, normalized], + )?; + conn.execute( + "UPDATE memories SET source_session_id = ?2 WHERE source_session_id = ?1", + params![polluted_id, normalized], + )?; + + if target_exists { + conn.execute( + "UPDATE sessions SET + created_at = MIN(created_at, (SELECT created_at FROM sessions WHERE id = ?2)), + updated_at = MAX(updated_at, (SELECT updated_at FROM sessions WHERE id = ?2)), + last_active_at = MAX(last_active_at, (SELECT last_active_at FROM sessions WHERE id = ?2)), + message_count = message_count + (SELECT message_count FROM sessions WHERE id = ?2), + user_turn_count = user_turn_count + (SELECT user_turn_count FROM sessions WHERE id = ?2) + WHERE id = ?1", + params![normalized, polluted_id], + )?; + } + + conn.execute("DELETE FROM sessions WHERE id = ?1", params![polluted_id])?; + + if target_exists { + // 重编号：按 created_at 恢复连续 seq。原地 UPDATE 会在行级更新过程中 + // 与尚未更新的行冲突 UNIQUE(session_id, seq)，因此分两步： + // 先写入"大偏移 + 行号"（互不冲突），再统一减去偏移还原为 1..N。 + const SEQ_RENUMBER_OFFSET: i64 = 2_000_000_000; + conn.execute( + "UPDATE messages SET seq = ?2 + ( + SELECT t.rn FROM ( + SELECT id AS mid, ROW_NUMBER() OVER (ORDER BY created_at ASC, seq ASC) AS rn + FROM messages WHERE session_id = ?1 + ) t WHERE t.mid = messages.id + ) + WHERE session_id = ?1", + params![normalized, SEQ_RENUMBER_OFFSET], + )?; + conn.execute( + "UPDATE messages SET seq = seq - ?2 WHERE session_id = ?1", + params![normalized, SEQ_RENUMBER_OFFSET], + )?; + } + + repaired += 1; + tracing::info!( + polluted_id = %polluted_id, + normalized_id = %normalized, + merged_into_existing = target_exists, + "Repaired polluted session id" + ); + } + + if repaired > 0 { + tracing::info!( + repaired_count = repaired, + "Session id prefix pollution repair complete" + ); + } + + Ok(()) +} + pub(super) fn ensure_todos_schema(conn: &Connection) -> Result<(), StorageError> { let table_exists: bool = conn .query_row( diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 408fe2e..07247f5 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -65,7 +65,7 @@ impl SessionStore { /// Initialize a SessionStore from a connection and its file path. /// The connection is used for schema initialization only; the pool /// manages subsequent connections using the same file path. - fn from_connection(conn: Connection, db_uri: &str) -> Result { + fn from_connection(mut conn: Connection, db_uri: &str) -> Result { conn.busy_timeout(std::time::Duration::from_secs(30))?; conn.execute_batch( " @@ -237,6 +237,7 @@ impl SessionStore { ensure_memory_scope_key_migration(&conn)?; ensure_todos_schema(&conn)?; ensure_pending_subagents_schema(&conn)?; + repair_session_id_prefix_pollution(&mut conn)?; drop(conn); diff --git a/src/storage/tests.rs b/src/storage/tests.rs index 586bcd0..bb12429 100644 --- a/src/storage/tests.rs +++ b/src/storage/tests.rs @@ -717,3 +717,80 @@ fn test_get_topic_message_count_uses_count_query() { 0 ); } + +#[test] +fn test_repair_session_id_prefix_pollution() { + let store = SessionStore::in_memory().unwrap(); + + let clean = store.ensure_channel_session("websocket", "abc").unwrap(); + assert_eq!(clean.id, "abc"); + + store + .append_message(&clean.id, &ChatMessage::user("clean msg")) + .unwrap(); + + let mut conn = store.pool.get().unwrap(); + conn.execute("PRAGMA user_version = 1", []).unwrap(); + + conn.execute( + "INSERT INTO sessions (id, title, channel_name, chat_id, summary, created_at, updated_at, last_active_at, archived_at, deleted_at, message_count, user_turn_count, agent_prompt_reinjection_count) VALUES ('websocket:websocket:abc', 'Polluted', 'websocket', 'websocket:websocket:abc', NULL, 100, 100, 100, NULL, NULL, 1, 1, 0)", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO messages (id, session_id, seq, role, content, media_refs_json, created_at) VALUES ('m-polluted', 'websocket:websocket:abc', 1, 'user', 'old polluted msg', '[]', 100)", + [], + ) + .unwrap(); + + conn.execute( + "INSERT INTO sessions (id, title, channel_name, chat_id, summary, created_at, updated_at, last_active_at, archived_at, deleted_at, message_count, user_turn_count, agent_prompt_reinjection_count) VALUES ('websocket:xyz', 'PollutedXyz', 'websocket', 'websocket:xyz', NULL, 200, 200, 200, NULL, NULL, 1, 1, 0)", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO messages (id, session_id, seq, role, content, media_refs_json, created_at) VALUES ('m-xyz', 'websocket:xyz', 1, 'user', 'xyz msg', '[]', 200)", + [], + ) + .unwrap(); + + super::migrations::repair_session_id_prefix_pollution(&mut conn).unwrap(); + + assert!( + store + .get_session("websocket:websocket:abc") + .unwrap() + .is_none() + ); + assert!(store.get_session("websocket:xyz").unwrap().is_none()); + + let merged = store.get_session("abc").unwrap().unwrap(); + assert_eq!(merged.message_count, 2); + assert_eq!(merged.user_turn_count, 2); + + let msgs = store.load_messages("abc").unwrap(); + assert_eq!(msgs.len(), 2); + assert_eq!(msgs[0].content, "old polluted msg"); + + { + let mut stmt = conn + .prepare("SELECT seq FROM messages WHERE session_id = 'abc' ORDER BY seq") + .unwrap(); + let seqs: Vec = stmt + .query_map([], |row| row.get(0)) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(seqs, vec![1, 2]); + } + + let xyz = store.get_session("xyz").unwrap().unwrap(); + assert_eq!(xyz.chat_id, "xyz"); + assert_eq!(store.load_messages("xyz").unwrap().len(), 1); + + assert_eq!(store.list_sessions("websocket", false).unwrap().len(), 2); + + super::migrations::repair_session_id_prefix_pollution(&mut conn).unwrap(); + assert_eq!(store.get_session("abc").unwrap().unwrap().message_count, 2); + assert_eq!(store.list_sessions("websocket", false).unwrap().len(), 2); +}