fix(storage): 严格解析本地策略并原子执行数据库迁移

This commit is contained in:
2026-09-04 19:57:26 +08:00
parent 78dd774bce
commit cec89494f9
6 changed files with 210 additions and 10 deletions
+4
View File
@@ -32,8 +32,12 @@ def connect() -> sqlite3.Connection:
# 关闭 Python sqlite3 的隐式事务,提交时机由 transaction() 或显式 commit 控制。 # 关闭 Python sqlite3 的隐式事务,提交时机由 transaction() 或显式 commit 控制。
conn.isolation_level = None conn.isolation_level = None
conn.execute("PRAGMA foreign_keys = ON") conn.execute("PRAGMA foreign_keys = ON")
try:
_load_extension(conn) _load_extension(conn)
migrate(conn) migrate(conn)
except BaseException:
conn.close()
raise
return conn return conn
+34 -2
View File
@@ -6,6 +6,7 @@
""" """
from datetime import datetime, timezone from datetime import datetime, timezone
import sqlite3
from app.constants import EMBEDDING_DIM from app.constants import EMBEDDING_DIM
@@ -134,6 +135,18 @@ MIGRATIONS: list[str] = [
] ]
def _statements(script: str):
"""Split complete SQLite statements without executescript's implicit COMMIT."""
pending = ""
for char in script:
pending += char
if char == ";" and sqlite3.complete_statement(pending):
yield pending
pending = ""
if pending.strip():
yield pending
def migrate(conn) -> None: def migrate(conn) -> None:
"""把尚未应用的迁移脚本按序应用到给定连接。""" """把尚未应用的迁移脚本按序应用到给定连接。"""
conn.execute( conn.execute(
@@ -145,9 +158,28 @@ def migrate(conn) -> None:
for idx, script in enumerate(MIGRATIONS, start=1): for idx, script in enumerate(MIGRATIONS, start=1):
if idx in applied: if idx in applied:
continue continue
conn.executescript(script) conn.execute("BEGIN IMMEDIATE")
try:
# Another connection may have migrated while this one waited.
if not conn.execute("SELECT 1 FROM schema_migrations WHERE version=?", (idx,)).fetchone():
recovered_v6 = False
if idx == 6:
column = next((row for row in conn.execute("PRAGMA table_info(blocks)")
if row["name"] == "embedding_local_only"), None)
if column is not None:
# Recover the precise partial state left by the old v6 runner.
if column["type"].upper() != "INTEGER" or column["notnull"] != 1 or column["dflt_value"] != "0":
raise sqlite3.DatabaseError("Unexpected embedding_local_only column schema")
recovered_v6 = True
if not recovered_v6:
for statement in _statements(script):
conn.execute(statement)
conn.execute( conn.execute(
"INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)", "INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)",
(idx, datetime.now(timezone.utc).isoformat()), (idx, datetime.now(timezone.utc).isoformat()),
) )
conn.commit() conn.execute("COMMIT")
except BaseException:
if conn.in_transaction:
conn.execute("ROLLBACK")
raise
+33 -1
View File
@@ -13,7 +13,10 @@ from dataclasses import dataclass, field
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
import yaml
from app.contracts import NoteBlock from app.contracts import NoteBlock
from app.errors import ApiError
from app.textutils import count_tokens from app.textutils import count_tokens
_HEADING_RE = re.compile(r"^(#{1,6})[ \t]+(.*?)\s*$") _HEADING_RE = re.compile(r"^(#{1,6})[ \t]+(.*?)\s*$")
@@ -70,7 +73,7 @@ def parse_note(
created_at=created_at, created_at=created_at,
updated_at=updated_at, updated_at=updated_at,
blocks=blocks, blocks=blocks,
embedding_local_only=str(frontmatter.get("embedding_local_only", "")).lower() == "true", embedding_local_only=_embedding_policy(markdown),
) )
@@ -184,6 +187,35 @@ def _utf16_len(text: str) -> int:
return len(text.encode("utf-16-le")) // 2 return len(text.encode("utf-16-le")) // 2
def _embedding_policy(markdown: str) -> bool:
end = markdown.find("\n---", 3) if markdown.startswith("---") else -1
if end == -1:
return False
try:
# Compose nodes without constructing objects. This accepts YAML comments,
# quoted keys and indentation while retaining duplicate-key information.
node = yaml.compose(markdown[3:end], Loader=yaml.SafeLoader)
except yaml.YAMLError as exc:
raise ApiError(422, "INVALID_EMBEDDING_POLICY", "Frontmatter YAML 无效,无法确认本地索引策略。") from exc
if node is None:
return False
if not isinstance(node, yaml.MappingNode):
raise ApiError(422, "INVALID_EMBEDDING_POLICY", "Frontmatter 必须是 YAML 键值映射。")
if any(key.tag == "tag:yaml.org,2002:merge" for key, _ in node.value):
raise ApiError(422, "INVALID_EMBEDDING_POLICY", "Frontmatter 不支持 YAML 合并键,请显式声明索引策略。")
values = [value for key, value in node.value
if isinstance(key, yaml.ScalarNode) and key.value.lower() == "embedding_local_only"]
if not values:
return False
if len(values) > 1:
raise ApiError(422, "INVALID_EMBEDDING_POLICY", "embedding_local_only 不能重复声明。")
value = values[0]
if (not isinstance(value, yaml.ScalarNode) or value.tag != "tag:yaml.org,2002:bool"
or value.value.lower() not in {"true", "false", "yes", "no", "on", "off"}):
raise ApiError(422, "INVALID_EMBEDDING_POLICY", "embedding_local_only 必须是 YAML 布尔值 true 或 false。")
return value.value.lower() in {"true", "yes", "on"}
def _extract_frontmatter(markdown: str) -> dict[str, str]: def _extract_frontmatter(markdown: str) -> dict[str, str]:
"""极简 frontmatter 解析,只提取 key: value 行。""" """极简 frontmatter 解析,只提取 key: value 行。"""
if not markdown.startswith("---"): if not markdown.startswith("---"):
+3 -1
View File
@@ -125,6 +125,8 @@ def test_local_only_export_and_rebuild_keep_local_embedding_policy(monkeypatch):
job = await jobs.create_transcription('lecture.txt', local_only=True) job = await jobs.create_transcription('lecture.txt', local_only=True)
note = await create_transcript_note(job.job_id, TranscriptNoteRequest(title='Private')) note = await create_transcript_note(job.job_id, TranscriptNoteRequest(title='Private'))
assert note.markdown.startswith('---\nembedding_local_only: true\n---') assert note.markdown.startswith('---\nembedding_local_only: true\n---')
await note_service.update_note(note.note_id, markdown=note.markdown.replace(
'embedding_local_only: true', 'embedding_local_only: true # keep local'))
await index_service.rebuild(IndexRebuildRequest()) await index_service.rebuild(IndexRebuildRequest())
assert len(calls) >= 2 and all(calls) assert len(calls) >= 3 and all(calls)
asyncio.run(scenario()) asyncio.run(scenario())
+110
View File
@@ -0,0 +1,110 @@
import sqlite3
from datetime import datetime, timezone
from concurrent.futures import ThreadPoolExecutor
import pytest
from app.database import migrations
from app.database.db import _load_extension
from app.errors import ApiError
from app.knowledge.parser import parse_note
def parsed(value):
return parse_note(markdown='---\nembedding_local_only: '+value+'\n---\nbody', file_path='note.md', folder='',
created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc))
@pytest.mark.parametrize('value,expected', [('true', True), ('true # keep local', True), ('TRUE # comment', True), ('false # explicit', False)])
def test_policy_parses_yaml_boolean_with_comments(value, expected):
assert parsed(value).embedding_local_only is expected
@pytest.mark.parametrize('value', ['truth', '1', '', 'null', '"true"', '[true]', '{broken', 'true\nembedding_local_only: false'])
def test_invalid_policy_never_silently_enables_remote(value):
with pytest.raises(ApiError) as error:
parsed(value)
assert error.value.code == 'INVALID_EMBEDDING_POLICY'
def connection(path, factory=sqlite3.Connection):
conn = sqlite3.connect(path, isolation_level=None, factory=factory)
conn.row_factory = sqlite3.Row
_load_extension(conn)
return conn
def seed_v5(path, monkeypatch):
conn = connection(path)
with monkeypatch.context() as patch:
patch.setattr(migrations, 'MIGRATIONS', migrations.MIGRATIONS[:5])
migrations.migrate(conn)
conn.execute("INSERT INTO search_history(query) VALUES ('retained')")
conn.close()
@pytest.mark.parametrize('failure', [sqlite3.OperationalError, KeyboardInterrupt])
def test_migration_and_version_write_rollback_together(tmp_path, monkeypatch, failure):
path = tmp_path / 'migration.db'
seed_v5(path, monkeypatch)
class Interrupted(sqlite3.Connection):
def execute(self, sql, parameters=()):
if sql.startswith('INSERT INTO schema_migrations') and parameters[0] == 6:
raise failure('interrupted')
return super().execute(sql, parameters)
conn = connection(path, Interrupted)
try:
with pytest.raises(failure):
migrations.migrate(conn)
assert not conn.in_transaction
assert not any(r['name'] == 'embedding_local_only' for r in conn.execute('pragma table_info(blocks)'))
finally:
conn.close()
conn = connection(path)
try:
migrations.migrate(conn)
assert conn.execute('select count(*) from schema_migrations where version=6').fetchone()[0] == 1
assert conn.execute('select query from search_history').fetchone()[0] == 'retained'
finally:
conn.close()
def test_old_partial_v6_recovers_without_duplicate_column(tmp_path, monkeypatch):
path = tmp_path / 'partial.db'
seed_v5(path, monkeypatch)
conn = connection(path)
try:
conn.executescript(migrations.MIGRATIONS[5])
migrations.migrate(conn)
migrations.migrate(conn)
assert conn.execute('select count(*) from schema_migrations where version=6').fetchone()[0] == 1
assert conn.execute('select query from search_history').fetchone()[0] == 'retained'
finally:
conn.close()
def test_concurrent_connections_can_upgrade(tmp_path, monkeypatch):
path = tmp_path / 'concurrent.db'
seed_v5(path, monkeypatch)
def upgrade(_):
conn = connection(path)
try:
migrations.migrate(conn)
return conn.execute('select count(*) from schema_migrations where version=6').fetchone()[0]
finally:
conn.close()
with ThreadPoolExecutor(max_workers=2) as pool:
assert list(pool.map(upgrade, range(2))) == [1, 1]
@pytest.mark.parametrize('header', ['"embedding_local_only": true # comment', ' embedding_local_only: true', 'embedding_local_only:\n true', 'local: &local true\nembedding_local_only: *local'])
def test_policy_supports_yaml_key_and_scalar_forms(header):
note = parse_note(markdown='---\n'+header+'\n---\nbody',file_path='note.md',folder='',created_at=datetime.now(timezone.utc),updated_at=datetime.now(timezone.utc))
assert note.embedding_local_only
def test_merge_policy_is_rejected_instead_of_ignored():
with pytest.raises(ApiError):
parsed('true\n<<: {embedding_local_only: false}')
with pytest.raises(ApiError):
parsed('!!bool invalid')
@@ -24,6 +24,8 @@
| F-06 | 本地转写导出未传递限制 | 正文可能发送给远程 Embedding | 持久化本地索引标记 | | F-06 | 本地转写导出未传递限制 | 正文可能发送给远程 Embedding | 持久化本地索引标记 |
| F-07 | 推理结束才读取当前模型标识 | 向量与空间错配 | 冻结模型、revision 和设备配置 | | F-07 | 推理结束才读取当前模型标识 | 向量与空间错配 | 冻结模型、revision 和设备配置 |
| F-08 | 将不同处理策略误判为配置漂移 | 普通与仅本地笔记共存时不能重建 | 按策略校验覆盖,独立检索并融合排名 | | F-08 | 将不同处理策略误判为配置漂移 | 普通与仅本地笔记共存时不能重建 | 按策略校验覆盖,独立检索并融合排名 |
| F-09 | 简单字符串比较忽略 YAML 语法 | 注释等合法写法可能关闭本地限制 | 解析 YAML 节点,非法策略明确拒绝 |
| F-10 | 迁移 DDL 与版本号分开提交 | 中断后重启报重复列 | 原子迁移、并发重检及旧半迁移恢复 |
## 3. F-01 / F-03:模型可用不等于索引可用 ## 3. F-01 / F-03:模型可用不等于索引可用
@@ -134,6 +136,24 @@ embedding_local_only: true
## 9. 工程经验 ## 9. 工程经验
### F-09:本地限制标记的 YAML 解析
再次审阅复现:`embedding_local_only: true # keep local` 被旧字符串比较解析为 `False`。加注释没有改变用户意图,却可能使保存或重建发送正文到远程 Embedding。
实际方案:使用 PyYAML SafeLoader 解析 frontmatter 节点,不构造任意对象;读取布尔节点,支持注释、带引号的键、缩进、多行布尔值和布尔锚点。普通的 `true/false` 与 YAML 布尔别名 `yes/no/on/off` 均按布尔值处理。字符串 `"true"`、数字、空值、非法值及重复声明返回 `INVALID_EMBEDDING_POLICY`,不静默转为普通索引。
无效 YAML、非映射 frontmatter 和 YAML 合并键也明确拒绝;合并键应展开为显式声明,以避免遗漏继承的限制。标题与标签的既有提取方式保持不变。回归包含添加行尾注释后保存笔记、重建仍只走本地索引。
### F-10:数据库迁移中断恢复
`executescript` 先提交 DDL,随后才写入 `schema_migrations`。若 v6 新增列后中断,重启会再次执行 `ALTER TABLE`,产生 `duplicate column name: embedding_local_only`
实际方案:用 SQLite 的完整语句检测拆分静态迁移脚本,逐句执行,避免 `executescript` 隐式提交。每个版本在 `BEGIN IMMEDIATE` 事务内执行 DDL 和版本写入;异常包括中断均回滚。获取写锁后重新检查版本,防止多个连接重复迁移。连接初始化失败时主动关闭连接。
兼容恢复仅针对旧版已发生的 v6 半迁移:确认现有列为预期的 `INTEGER NOT NULL DEFAULT 0` 后补记版本,不再重复新增列;形状不符合预期则报错,不擅自更改数据。测试注入写版本失败和中断,验证列与版本同时回滚、重新连接升级成功,并覆盖旧半迁移与并发连接升级。
F-09/F-10 修复后完整后端回归:516 项通过,新增 21 个参数化用例;仍仅有既有 Starlette/httpx 弃用提示。测试均使用隔离数据,不调用外部模型 API。
F-08 修复后完整后端回归:495 项通过;新增 5 个用例覆盖混合策略重建与查询、全部本地回退、仅本地查询不访问 API、跨分区回滚和不完整分区。现有同策略空间漂移拒绝用例仍通过。 F-08 修复后完整后端回归:495 项通过;新增 5 个用例覆盖混合策略重建与查询、全部本地回退、仅本地查询不访问 API、跨分区回滚和不完整分区。现有同策略空间漂移拒绝用例仍通过。
区分配置、权重安装、推理运行、索引覆盖四种状态;按用户实际启动方式验证;跨异步边界冻结身份;持久化处理限制;增加限制时也验证普通 API 回退没有被破坏。 区分配置、权重安装、推理运行、索引覆盖四种状态;按用户实际启动方式验证;跨异步边界冻结身份;持久化处理限制;增加限制时也验证普通 API 回退没有被破坏。