补充SQLite跨引擎实机验收并保留初始化超时反例

docs/SQLite跨引擎迁移补充验收.md

@@ -0,0 +1,70 @@
+# SQLite 跨引擎迁移补充验收
+
+## 范围与任务状态
+
+本记录补充 SQLite 与 MySQL、PostgreSQL 之间的四个应用级迁移方向。它不替代真实浏览器、外部邮件最终收件、凭据轮换、证书自动续期及全项目生产评审。
+
+- [✓] 已核对修改前基线 `d4f1b0c` 与远端一致，用户原报告保持不变。
+- [✓] 混合引擎验收工具与反例测试已补齐。
+- [✓] SQLite → MySQL、SQLite → PostgreSQL、MySQL → SQLite 三个方向实机通过。
+- [!] PostgreSQL → SQLite 两次均在初始化恢复码生成请求上超时，尚未进入迁移，不能放行。
+- [✓] 独立复核文件摘要、账号权限、测试单元及正式服务保护状态。
+- [✓] 最终 Linux 部署回归 149 项通过、零跳过；本地运行时回归 79 项通过。
+
+**结论：本轮只关闭三个迁移方向，不能宣称六个跨引擎方向全部通过，更不能宣称全项目达到生产部署级别。**
+
+## 候选与隔离边界
+
+本轮仅扩展验收工具，应用后端、前端和插件源码均未修改。使用上轮已编译候选 `smail-character`，版本 `review-character-20260921`，SHA256 为 `97ad202b67e57e6284d7707928cf4591a5d2d676344d7b67f003630242e28924`。启动前逐项校验 611 个后端 Go 源码摘要；不把复用候选描述为本轮全资产重新构建。
+
+每轮使用新建空实例、随机 SQL 库和专用账号。测试域名、随机管理员口令及自签测试证书只服务于隔离协议探针，不构成域名、正式证书或管理员交互验收。实例通过独立网络、受限 systemd 单元及 192 MiB 应用内存上限运行，不停止正式 Smail、Nginx、MySQL、PostgreSQL、Scert、FDM，不清空正式库。
+
+SQLite → SQL 仅保留已停用的 `smail.db` 作为离线源证据；SQL → SQLite 只允许活动文件 `migrated.db`。业务表与配置历史都使用同一目标数据库，不能生成第二个状态库。离线证据不是第二个活动数据库，也不代表正式部署必须保留旧库。
+
+## 工具与断言
+
+- `test/tools/runtime_mixed_migration.py`：只接受一端 SQLite、一端 SQL；SQLite 文件名固定、路径受限、只读查询且缺库不能新建。
+- 使用正式迁移预览、执行和系统重启 API，不直接改写配置模拟成功。待重启拒绝写入，内部重启继续使用原会话。
+- 内部重启、真实系统重启、故障恢复后三次分别检查 SQLite 文件描述符和 SQL 连接；要求源停止使用、目标承担活动读写。
+- 目标 SQL 账号锁定或 SQLite 文件权限收紧后，应用应明确启动失败；配置保持指向目标，不静默回退源库。异常路径也恢复账号或权限。
+- 迁移前邮件与各阶段新邮件通过 SMTP、IMAP、POP3 检查中文正文和附件；新邮件只进入目标库。
+- 配置历史比较四列原始字节并校验载荷摘要；源库逻辑指纹在后续写入、重启和故障注入期间保持一致。
+- 扫描实例中 SQLite 扩展名、文件头、侧文件和符号链接，拒绝隐藏库、改名库、嵌套额外库。
+
+## 当前实测记录
+
+下表实例目录均位于 `/var/tmp/smail-review-runtime-` 前缀下。
+
+| 方向 | 实例后缀 | 内部重启 PID | 系统重启 PID | 故障恢复 PID | 结果 |
+| --- | --- | --- | --- | --- | --- |
+| SQLite → MySQL | `20260921-074846` | `790031`，不变 | `790465` | `790531` | 通过 |
+| SQLite → PostgreSQL | `20260921-075415` | `790691`，不变 | `790937` | `791027` | 通过 |
+| MySQL → SQLite | `20260921-075919` | `791137`，不变 | `791541` | `791650` | 通过 |
+| PostgreSQL → SQLite | `20260921-080502` | 未执行 | 未执行 | 未执行 | 初始化请求超时 |
+| PostgreSQL → SQLite，原条件复测 | `20260921-080854` | 未执行 | 未执行 | 未执行 | 相同初始化请求超时 |
+
+两个 SQLite → SQL 方向在内部重启、系统重启、故障恢复后的 SQL 连接数均为 `3 / 2 / 2`，旧 SQLite 均未被进程打开。MySQL → SQLite 三阶段源 SQL 连接数均为零，活动 SQLite 均已打开，运行目录没有 `smail.db` 或第二个状态库。
+
+三个成功方向均保留同一条源配置历史的原始字节与摘要；待重启写入拒绝、旧会话恢复、旧邮件和各阶段新邮件的正文及附件均通过。目标不可用时 PID 为零、服务状态为 failed、退出码为 1，目标配置保留且源库逻辑指纹不变。恢复后读写正常。五轮均无 OOM、资源采集错误或清理错误。
+
+独立复核覆盖三个成功实例及两个失败实例，共 47 份云端／本地证据摘要、七个运行工具和 611 个后端源码摘要。五个 SQL 专用账号均锁定；三个 PostgreSQL 账号均不能登录、不能建库、不是超级用户。测试应用、探针、桥接和网络锚点全部停止；正式六服务的 PID、InvocationID、重启次数及正式程序摘要均未改变。
+
+复核索引为 `mixed-recheck-20260921-081310.json`，同时保留原始首轮与复测汇总，不以仅包含成功样本的新文件替代失败记录。
+
+## 回归与保留边界
+
+新增十三项反例测试，覆盖路径、缺库不创建、只读连接、历史损坏、隐藏数据库、源连接残留、单活动文件以及故障恢复。初次本地测试发现测试夹具未关闭 SQLite 连接，Windows 临时目录清理失败；修复夹具显式关闭后通过，不将其描述为产品缺陷。
+
+本地运行时最终回归 79 项通过，日志 `mixed-local-runtime-final.log`。第一次 Linux 全部署回归 148 项通过、零跳过，证据 `python-linux-result-20260921-074736.json`；随后增强隐藏库反例的相互隔离并增加单活动文件反例，最终 Linux 回归实际执行 149 项、零跳过、退出零，证据 `python-linux-result-20260921-081152.json`。最终套件运行约 28.64 秒、内存峰值约 46.3 MiB、交换零；这不是应用容量验收。
+
+### 保留的 PostgreSQL 首次安装超时
+
+实例 `/var/tmp/smail-review-runtime-20260921-080502/` 在初始恢复码生成请求上触发 30 秒 HTTP 读取超时，尚未开始迁移，不能算作 PostgreSQL → SQLite 的迁移成功或迁移失败反例。该轮应用 CPU 配额为 25%，应用峰值约 105.3 MiB、交换零、OOM 零，约 44.55 秒内消耗 10.296 秒 CPU。CPU 限额可能影响耗时，但这些指标不足以独立确认超时根因。
+
+该轮正式服务未改变，测试进程、桥接和网络锚点已停止，PostgreSQL 专用账号已锁定。失败证据保留在 `mixed-runtime-evidence/smail-review-runtime-20260921-080502/`，原始汇总为 `mixed-results-20260921-074833.json`。
+
+第二轮 `080854` 保持同一候选、资源限制和超时断言，新建空库与账号，仍在同一请求发生读取超时。结果为 `mixed-results-20260921-080840.json`，对应原始证据目录同样保留。收尾完整性检查因目标尚未创建而报告断言失败，不能误述成目标 SQLite 已损坏。没有继续盲目重试，也没有放宽限制或将失败从汇总中移除。
+
+源码 `backend/services/credentials/service.go` 的该路径逐个生成十枚恢复码，并串行调用 `password.Hash` 后写入事务。后续需要分别采集密码计算、数据库事务和请求截止时间的证据，才能确定超时原因与整改方案；不能只凭当前日志将其归为 PostgreSQL 迁移缺陷，也不能降低密码哈希强度来获取通过结果。
+
+本地证据根目录为 `test-log/production-remediation-20260921/`。成功与失败证据均保留，不删除源库、失败目标或用户文件。

docs/数据库迁移重启补充验收.md

@@ -89,3 +89,9 @@
 ## 后续 SQLite 复制中断补充
 
 9 月 21 日 06:17 的新空隔离实例完成未提交复制事务中的 SIGKILL 中断及恢复，随后正常迁移和重启恢复也通过。详见 `SQLite复制中断恢复补充验收.md`，包含两个夹具失败记录、原始目标及 journal 快照摘要、20 份证据复核与 131 项部署回归。本项仅关闭 SQLite 未提交复制阶段，不关闭提交临界点、配置切换临界点、SQL 数据库中断或跨引擎矩阵。
+
+## 后续混合引擎矩阵补充
+
+详见 `SQLite跨引擎迁移补充验收.md`。SQLite → MySQL、SQLite → PostgreSQL、MySQL → SQLite 三个方向通过正式迁移、两类重启、目标不可用及恢复后的读写，三阶段证明只有目标库活动，配置历史同库。
+
+PostgreSQL → SQLite 两次均在空安装恢复码生成请求的 30 秒读取超时处终止，尚未进入迁移；失败原样保留，不能宣称矩阵完整。47 份成功／失败证据及全部测试账号、测试单元和正式服务保护状态已独立复核。最终部署回归 149 项、零跳过，但不关闭初始化超时、迁移中断临界点或其他生产门禁。

docs/生产补充验收记录.md

@@ -331,3 +331,11 @@ SQLite 已完成迁移失败不覆盖目标、成功迁移、内部重启、系
 新候选完成 MySQL ↔ PostgreSQL 双向迁移、内部重启、系统级重启、目标账号故障及恢复后的真实三协议读写，源库零活动连接、源内容不变、配置历史同库、没有 SQLite 文件。26 份实机证据、六个工具及 611 个后端源码摘要再次核对一致，测试账号锁定且临时建库权限撤销，正式六服务和程序未变。
 
 云端完整部署回归实际为 136 项、零跳过；新增 Go 专项两个顶层测试零跳过。本地全量后端通过，但明确保留 35 个顶层和 12 个子测试跳过，不能称全平台零跳过。涉及 SQLite 的其余四个跨引擎应用级方向、中断临界点和此前其他生产门禁仍未关闭，整体生产验收尚未通过。
+
+### 9 月 21 日 SQLite 跨引擎补充
+
+详见 `SQLite跨引擎迁移补充验收.md`。沿用源码摘要一致的最新编译候选，只扩展验收工具；SQLite → MySQL、SQLite → PostgreSQL、MySQL → SQLite 三个方向通过。两个 SQLite 源均停止占用，MySQL 源活动连接归零；目标独立承载业务和配置历史。两类重启、目标故障拒绝启动与恢复、原会话、历史邮件及新增邮件正文和附件均通过。
+
+PostgreSQL → SQLite 两次在初始化恢复码生成时发生 30 秒 HTTP 读取超时，未进入迁移。没有降低哈希强度、减少码数量或放宽超时条件，不将失败标为通过。该初始化耗时问题和最后一个迁移方向继续待查。
+
+最终本地运行时回归 79 项、Linux 完整部署回归 149 项通过，Linux 零跳过；47 份成功与失败证据、七个工具及 611 个后端源码摘要已独立复核。五个 SQL 测试账号锁定，测试单元全部停止，正式六服务和程序未变；用户原测试报告未编辑。全项目生产验收仍未通过，历史凭据、其余安全告警、真实浏览器、外部投递、自动续期、容量及异机灾备等范围不能由本轮结果代替。

test/deployment/test_runtime_mixed_migration.py

@@ -0,0 +1,165 @@
+"""混合引擎验收器反例测试；本地临时库不连接云端。"""
+
+import hashlib
+from contextlib import closing
+import json
+from pathlib import Path
+import shutil
+import sqlite3
+import stat
+import sys
+import tempfile
+import unittest
+from unittest.mock import Mock, PropertyMock, patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools'))
+import runtime_mixed_migration as mixed
+from runtime_sql import SQLFixture
+
+
+class RuntimeMixedMigrationTest(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.root = Path(self.directory.name).resolve()
+        (self.root / 'data').mkdir()
+        (self.root / 'config').mkdir()
+        self.source = mixed.SQLiteDatabase(self.root, {'kind': 'sqlite', 'name': 'smail.db'})
+        self.target = SQLFixture('mysql')
+        with closing(sqlite3.connect(self.source.path)) as database:
+            database.execute('CREATE TABLE config_records(kind TEXT,id TEXT,payload BLOB,digest TEXT)')
+            payload = '中文\x00载荷'.encode()
+            database.execute('INSERT INTO config_records VALUES (?,?,?,?)',
+                             ('backup', 'item', payload, hashlib.sha256(payload).hexdigest()))
+            database.commit()
+        self.configure(self.target)
+
+    def configure(self, target):
+        (self.root / 'config/config.json').write_text(json.dumps({'database': mixed.target_payload(target)}))
+
+    def test_only_fixed_local_names_are_allowed(self):
+        for name in ('../smail.db', '/smail.db', 'nested/smail.db', 'state.db', ''):
+            with self.assertRaises(AssertionError):
+                mixed.SQLiteDatabase(self.root, {'kind': 'sqlite', 'name': name})
+        with self.assertRaises(AssertionError):
+            mixed.SQLiteDatabase(self.root, {'kind': 'sqlite', 'name': 'smail.db', 'password': 'extra'})
+
+    def test_missing_database_is_not_created(self):
+        missing = mixed.SQLiteDatabase(self.root, {'kind': 'sqlite', 'name': 'migrated.db'})
+        with self.assertRaises(sqlite3.OperationalError):
+            missing.query('SELECT 1')
+        self.assertFalse(missing.path.exists())
+
+    def test_read_only_query_cannot_change_database(self):
+        with self.assertRaises(sqlite3.OperationalError):
+            self.source.query('DELETE FROM config_records')
+        self.assertEqual(self.source.query('SELECT COUNT(*) FROM config_records'), '1')
+
+    def test_history_keeps_bytes_and_rejects_null_or_bad_digest(self):
+        values = mixed.history_rows(self.source)
+        self.assertEqual(bytes.fromhex(json.loads(values[0])[2]), '中文\x00载荷'.encode())
+        for digest in (None, 'invalid'):
+            with closing(sqlite3.connect(self.source.path)) as database:
+                database.execute('UPDATE config_records SET digest=?', (digest,))
+                database.commit()
+            with self.assertRaises(AssertionError):
+                mixed.history_rows(self.source)
+
+    def test_expected_offline_source_and_wrong_configuration(self):
+        mixed.assert_active(self.root, self.source, self.target)
+        self.configure(self.source)
+        with self.assertRaises(AssertionError):
+            mixed.assert_active(self.root, self.source, self.target)
+
+    def test_sql_to_sqlite_has_only_one_active_file(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory).resolve()
+            (root / 'data').mkdir()
+            (root / 'config').mkdir()
+            target = mixed.SQLiteDatabase(root, {'kind': 'sqlite', 'name': 'migrated.db'})
+            shutil.copyfile(self.source.path, target.path)
+            (root / 'config/config.json').write_text(json.dumps({'database': mixed.target_payload(target)}))
+            result = mixed.migration_integrity(root, self.target, target)
+            self.assertEqual(result['active_database'], 'migrated.db')
+            self.assertEqual(result['sqlite_files'], 1)
+            self.assertEqual(result['retained_offline_files'], [])
+            shutil.copyfile(target.path, root / 'data/smail.db')
+            with self.assertRaises(AssertionError):
+                mixed.migration_integrity(root, self.target, target)
+
+    def test_hidden_nested_and_renamed_databases_rejected(self):
+        for name in ('state.db', '.hidden.sqlite3', 'nested/smail.db', 'renamed.bin'):
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory).resolve()
+                (root / 'data').mkdir()
+                shutil.copyfile(self.source.path, root / 'data/smail.db')
+                source = mixed.SQLiteDatabase(root, self.source.settings)
+                mixed.assert_files(root, source, self.target)
+                path = root / name
+                path.parent.mkdir(parents=True, exist_ok=True)
+                with closing(sqlite3.connect(path)) as database:
+                    database.execute('CREATE TABLE marker(id INTEGER)')
+                with self.assertRaises(AssertionError):
+                    mixed.assert_files(root, source, self.target)
+
+    def test_symlink_is_rejected_before_open(self):
+        with patch.object(Path, 'is_symlink', return_value=True):
+            with self.assertRaises(AssertionError):
+                self.source.query('SELECT 1')
+            with self.assertRaises(AssertionError):
+                mixed.assert_files(self.root, self.source, self.target)
+
+    def test_sql_target_requires_source_closed_and_target_connected(self):
+        for names, count, accepted in ((set(), 3, True), ({'/opt/smail/data/smail.db'}, 3, False),
+                                      ({'/opt/smail/data/smail.db-wal'}, 3, False),
+                                      ({'/opt/smail/data/smail.db (deleted)'}, 3, False), (set(), 0, False)):
+            with patch.object(mixed, 'descriptor_names', return_value=names), \
+                    patch.object(mixed, 'connection_count', return_value=count):
+                if accepted:
+                    self.assertTrue(mixed.assert_connections(self.root, {'MainPID': '123'}, self.source, self.target)['source_inactive'])
+                else:
+                    with self.assertRaises(AssertionError):
+                        mixed.assert_connections(self.root, {'MainPID': '123'}, self.source, self.target)
+
+    def test_sql_source_requires_zero_connections_and_sqlite_open(self):
+        with patch.object(mixed, 'assert_active'):
+            for names, count, accepted in (({'/opt/smail/data/smail.db'}, 0, True),
+                                          ({'/opt/smail/data/smail.db'}, 1, False), (set(), 0, False)):
+                with patch.object(mixed, 'descriptor_names', return_value=names), \
+                        patch.object(mixed, 'connection_count', return_value=count):
+                    if accepted:
+                        self.assertTrue(mixed.assert_connections(self.root, {'MainPID': '123'}, self.target, self.source)['target_active'])
+                    else:
+                        with self.assertRaises(AssertionError):
+                            mixed.assert_connections(self.root, {'MainPID': '123'}, self.target, self.source)
+
+    def test_sql_account_restored_even_when_lock_or_body_fails(self):
+        for lock_failure in (False, True):
+            target = Mock(kind='mysql')
+            if lock_failure:
+                target.lock.side_effect = RuntimeError('lock failure')
+            with self.assertRaises(RuntimeError):
+                with mixed.unavailable_target(target):
+                    raise RuntimeError('probe failure')
+            target.unlock.assert_called_once_with()
+
+    def test_sqlite_permissions_restored_on_failure(self):
+        with patch.object(Path, 'stat', return_value=Mock(st_mode=stat.S_IFREG | 0o600)), \
+                patch.object(Path, 'is_file', return_value=True), \
+                patch.object(mixed.SQLiteDatabase, 'path', new_callable=PropertyMock,
+                             return_value=self.root / 'data/smail.db'), \
+                patch.object(Path, 'chmod') as chmod:
+            with self.assertRaises(RuntimeError):
+                with mixed.unavailable_target(self.source):
+                    raise RuntimeError('probe failure')
+            self.assertEqual([call.args[0] for call in chmod.call_args_list], [0, 0o600])
+
+    def test_formal_root_rejected_before_service_calls(self):
+        with patch.object(mixed, 'service_state') as state:
+            with self.assertRaises(AssertionError):
+                mixed.verify_migration(Path('/opt/smail'), 'smail.service', '', {}, self.source, self.target)
+            state.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()

test/deployment/test_runtime_sql_migration.py

@@ -18,8 +18,10 @@ class RuntimeSQLMigrationTest(unittest.TestCase):
         for source, target in (('mysql', 'postgres'), ('postgres', 'mysql')):
             self.assertEqual(review.migration_kind(source, target, True, False), target)
         self.assertEqual(review.migration_kind('sqlite', None, True, False), 'sqlite')
-        for arguments in (('sqlite', 'mysql', True, False), ('mysql', 'postgres', False, False),
-                          ('mysql', 'postgres', True, True), ('mysql', 'sqlite', True, False)):
+        for source, target in (('sqlite', 'mysql'), ('sqlite', 'postgres'), ('mysql', 'sqlite'), ('postgres', 'sqlite')):
+            self.assertEqual(review.migration_kind(source, target, True, False), target)
+        for arguments in (('mysql', 'postgres', False, False), ('mysql', 'postgres', True, True),
+                          ('sqlite', 'mysql', True, True), ('mysql', 'invalid', True, False)):
             with self.assertRaises(ValueError):
                 review.migration_kind(*arguments)
 

test/tools/runtime_migration.py

@@ -205,7 +205,8 @@ def run_probe(root, unit, password, phase, sql_settings=None, probe_script=None)
     probe = 'smail-migration-probe-' + unit.removeprefix('smail-runtime-review-') + '-' + phase
     script = 'runtime_sql_migration.py' if sql_settings else 'runtime_migration.py'
     if probe_script is not None:
-        assert probe_script == 'runtime_interruption.py' and sql_settings is None
+        assert ((probe_script == 'runtime_interruption.py' and sql_settings is None)
+                or (probe_script == 'runtime_mixed_migration.py' and sql_settings is not None))
         script = probe_script
     command = ['systemd-run', '--wait', '--pipe', '--unit=' + probe,
         '--property=User=smail', '--property=Group=smail',

test/tools/runtime_mixed_migration.py

@@ -0,0 +1,232 @@
+"""SQLite 与服务器数据库之间的隔离应用级验收，不操作正式库。"""
+
+import argparse
+import base64
+from contextlib import contextmanager
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import subprocess
+import sys
+import time
+
+from runtime_migration import database_fingerprint, read_database, run_probe, service_state
+from runtime_sql import SQLDatabase
+from runtime_sql_migration import connection_count, fingerprint, history_rows, prepare, target_payload, verify_mail
+
+
+class SQLiteDatabase:
+    def __init__(self, root, settings):
+        assert set(settings) == {'kind', 'name'} and settings['kind'] == 'sqlite'
+        assert settings['name'] in ('smail.db', 'migrated.db')
+        self.root = root
+        self.settings = settings.copy()
+        self.kind = 'sqlite'
+
+    @property
+    def path(self):
+        path = self.root / 'data' / self.settings['name']
+        assert self.root.is_absolute() and self.root.resolve(strict=True) == self.root
+        assert not self.root.is_symlink() and not path.parent.is_symlink() and not path.is_symlink()
+        assert path.resolve() == path
+        return path
+
+    def query(self, sql):
+        with read_database(self.path) as database:
+            rows = database.execute(sql).fetchall()
+        assert all(len(row) == 1 and row[0] is not None for row in rows)
+        return '\n'.join(str(row[0]) for row in rows)
+
+    def verify_message(self, message_id, account, body, attachment):
+        with read_database(self.path) as database:
+            rows = database.execute('SELECT id,text,attachments FROM email WHERE msg_id=?', (message_id,)).fetchall()
+            assert len(rows) == 1
+            key, text, attachments = rows[0]
+            assert body in text
+            values = json.loads(attachments)
+            assert len(values) == 1 and values[0]['Filename'] == '验收附件.txt'
+            assert base64.b64decode(values[0]['Content'], validate=True) == attachment
+            assert database.execute('SELECT u.account FROM user_email e JOIN "user" u ON u.id=e.user_id '
+                                    'WHERE e.email_id=?', (key,)).fetchall() == [(account,)]
+
+
+def assert_files(root, source, target):
+    expected = {root / 'data' / database.settings['name'] for database in (source, target)
+                if database.kind == 'sqlite'}
+    seen = set()
+    # 不只按扩展名找库：同时识别 SQLite 文件头，拒绝嵌套、改名和符号链接。
+    for path in root.rglob('*'):
+        assert not path.is_symlink(), '隔离实例不能通过符号链接隐藏第二个库'
+        if not path.is_file():
+            continue
+        with path.open('rb') as stream:
+            header = stream.read(16)
+        if header == b'SQLite format 3\x00' or path.suffix == '.db' or '.sqlite' in path.name:
+            assert path in expected, '发现非本轮来源或目标的 SQLite 文件'
+            seen.add(path)
+        if path.name.endswith(('-wal', '-shm', '-journal')):
+            assert path.with_name(path.name.rsplit('-', 1)[0]) in expected
+    assert seen == expected, '本轮 SQLite 证据文件缺失'
+    return sorted(path.name for path in seen)
+
+
+def assert_configuration(root, target):
+    database = json.loads((root / 'config/config.json').read_text())['database']
+    expected = target_payload(target)
+    for field in ('type', 'localRef') if target.kind == 'sqlite' else ('type', 'host', 'port', 'username', 'name'):
+        assert database.get(field) == expected[field], '活动配置没有指向本轮唯一目标'
+
+
+def assert_active(root, source, target):
+    assert_configuration(root, target)
+    assert_files(root, source, target)
+
+
+def descriptor_names(pid):
+    assert pid.isdigit() and int(pid) > 0
+    names = set()
+    for descriptor in Path('/proc', pid, 'fd').iterdir():
+        try:
+            names.add(str(descriptor.readlink()))
+        except FileNotFoundError:
+            pass
+    return names
+
+
+def assert_connections(root, state, source, target):
+    names = descriptor_names(state['MainPID'])
+    database = target if target.kind != 'sqlite' else source
+    count = connection_count(database)
+    sqlite = source if source.kind == 'sqlite' else target
+    prefix = '/opt/smail/data/' + sqlite.settings['name']
+    opened = any(name == prefix or name.startswith(prefix + '-') or name.startswith(prefix + ' (deleted)')
+                 for name in names)
+    if source.kind == 'sqlite':
+        assert not opened and count > 0, 'SQLite 源仍被打开或 SQL 目标无连接'
+    else:
+        assert prefix in names and count == 0, 'SQLite 目标未打开或 SQL 源仍有连接'
+    assert_active(root, source, target)
+    return {'sql_connections': count, 'sqlite_open': opened, 'source_inactive': True, 'target_active': True}
+
+
+def source_fingerprint(database):
+    return database_fingerprint(database.path) if database.kind == 'sqlite' else fingerprint(database)
+
+
+@contextmanager
+def unavailable_target(target):
+    if target.kind == 'sqlite':
+        path = target.path
+        assert path.is_file() and stat.S_IMODE(path.stat().st_mode) == 0o600
+        try:
+            path.chmod(0)
+            yield
+        finally:
+            path.chmod(0o600)
+    else:
+        try:
+            target.lock()
+            yield
+        finally:
+            target.unlock()
+
+
+def migration_integrity(root, source, target):
+    assert_active(root, source, target)
+    sqlite = source if source.kind == 'sqlite' else target
+    with read_database(sqlite.path) as database:
+        assert database.execute('PRAGMA integrity_check').fetchone() == ('ok',)
+    count = int(target.query('SELECT COUNT(*) FROM config_records'))
+    assert count > 0
+    return {'type': target.kind, 'active_database': target.settings['name'], 'config_records': count,
+            'sqlite_files': 1, 'retained_offline_files': [source.settings['name']] if source.kind == 'sqlite' else [],
+            'integrity': 'ok'}
+
+
+def verify_migration(root, unit, password, result, source, target):
+    assert root.parent.parent == Path('/var/tmp') and re.fullmatch(r'smail-review-runtime-\d{8}-\d{6}', root.parent.name)
+    assert root.name == 'instance' and not root.is_symlink() and root.resolve(strict=True) == root
+    assert source.kind != target.kind and 'sqlite' in (source.kind, target.kind)
+    assert unit == 'smail-runtime-review-' + root.parent.name.removeprefix('smail-review-runtime-')
+    details = result['migration'] = {'passed': False, 'database_type': source.kind, 'target_type': target.kind,
+                                     'cross_engine': True, 'source_database': source.settings['name'],
+                                     'target_database': target.settings['name']}
+    settings = {'source': source.settings, 'target': target.settings}
+
+    def probe(phase):
+        return run_probe(root, unit, password, phase, settings, 'runtime_mixed_migration.py')
+
+    before = service_state(unit)
+    details['prepare'] = probe('prepare')
+    if target.kind != 'sqlite':
+        target.revoke_creation()
+        details['temporary_creation_privilege_revoked'] = True
+    internal = service_state(unit)
+    assert before['MainPID'] == internal['MainPID'] and before['InvocationID'] == internal['InvocationID']
+    details['internal_restart_same_process'] = True
+    details['connections_after_internal'] = assert_connections(root, internal, source, target)
+    digest = source_fingerprint(source)
+    history = history_rows(source)
+    assert history
+    details['internal'] = probe('internal')
+    subprocess.run(['systemctl', 'restart', unit], check=True, timeout=55)
+    restarted = service_state(unit)
+    assert restarted['MainPID'] != internal['MainPID'] and restarted['InvocationID'] != internal['InvocationID']
+    details['process'] = probe('process')
+    details['connections_after_process'] = assert_connections(root, restarted, source, target)
+    details['process_states'] = {'before': before, 'internal': internal, 'restarted': restarted}
+    with unavailable_target(target):
+        subprocess.run(['systemctl', 'restart', unit], check=True, timeout=55)
+        for _ in range(70):
+            failed = service_state(unit)
+            if failed['MainPID'] == '0' and failed['ActiveState'] == 'failed':
+                break
+            time.sleep(1)
+        else:
+            raise AssertionError('目标不可用时应用没有明确停止，不能接受静默回退')
+        assert_configuration(root, target)
+        assert source_fingerprint(source) == digest
+        details['unavailable_target'] = {'failed_closed': True, 'source_unchanged': True,
+                                         'configuration_preserved': True, 'state': failed}
+    subprocess.run(['systemctl', 'start', unit], check=True, timeout=55)
+    details['recovered'] = probe('recovered')
+    final = service_state(unit)
+    assert final['MainPID'] not in (before['MainPID'], restarted['MainPID'])
+    details['connections_after_recovery'] = assert_connections(root, final, source, target)
+    assert source_fingerprint(source) == digest
+    current = history_rows(target)
+    assert set(history).issubset(current)
+    details['history'] = {'source_records': len(history), 'target_records': len(current), 'all_source_records_equal': True}
+    details.update(source_unchanged=True, target_access_restored=True, final_state=final, passed=True)
+    (root.parent / 'migration-result.json').write_text(json.dumps(details, ensure_ascii=False, indent=2), encoding='utf-8')
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('root', type=Path)
+    parser.add_argument('phase', choices=('prepare', 'internal', 'process', 'recovered'))
+    args = parser.parse_args()
+    assert os.geteuid() != 0
+    password = sys.stdin.readline().strip()
+    assert len(password) == 12
+    settings = json.loads(sys.stdin.readline())
+    assert set(settings) == {'source', 'target'}
+    source, target = (SQLiteDatabase(args.root, settings[key]) if settings[key]['kind'] == 'sqlite'
+                      else SQLDatabase(settings[key]) for key in ('source', 'target'))
+    assert source.kind != target.kind and 'sqlite' in (source.kind, target.kind)
+
+    def active_check(root, database):
+        assert_active(root, source, database)
+
+    result = (prepare(args.root, password, source, target, active_check) if args.phase == 'prepare' else
+              verify_mail(args.root, password, args.phase, source, target, active_check))
+    result['passed'] = True
+    (args.root.parent / ('migration-' + args.phase + '.json')).write_text(
+        json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+
+
+if __name__ == '__main__':
+    main()

test/tools/runtime_review.py

@@ -75,7 +75,7 @@ def collect_and_stop(result, unit, probe_unit, proxy_unit=None):
 def migration_kind(source, target, enabled, interrupted):
     if target is None:
         return source
-    if not enabled or interrupted or source == 'sqlite' or target not in ('mysql', 'postgres'):
+    if not enabled or interrupted or target not in ('sqlite', 'mysql', 'postgres'):
         raise ValueError('跨引擎目标必须用于独立 SQL 迁移验收')
     return target
 
@@ -103,7 +103,7 @@ def main():
     parser.add_argument('--migration', action='store_true')
     parser.add_argument('--interruption', action='store_true')
     parser.add_argument('--database-type', choices=('sqlite', 'mysql', 'postgres'), default='sqlite')
-    parser.add_argument('--migration-target-type', choices=('mysql', 'postgres'))
+    parser.add_argument('--migration-target-type', choices=('sqlite', 'mysql', 'postgres'))
     args = parser.parse_args()
     try:
         target_kind = migration_kind(args.database_type, args.migration_target_type, args.migration, args.interruption)
@@ -152,7 +152,8 @@ def main():
         shutil.copy2(Path(__file__).with_name('runtime_migration.py'), target / 'runtime_migration.py')
         (target / 'runtime_migration.py').chmod(0o640)
         shutil.chown(target / 'runtime_migration.py', user='root', group='smail')
-    if args.database_type != 'sqlite':
+    mixed = args.migration and args.database_type != target_kind and 'sqlite' in (args.database_type, target_kind)
+    if args.database_type != 'sqlite' or mixed:
         shutil.copy2(Path(__file__).with_name('runtime_sql.py'), target / 'runtime_sql.py')
         (target / 'runtime_sql.py').chmod(0o640)
         shutil.chown(target / 'runtime_sql.py', user='root', group='smail')
@@ -160,6 +161,10 @@ def main():
             shutil.copy2(Path(__file__).with_name('runtime_sql_migration.py'), target / 'runtime_sql_migration.py')
             (target / 'runtime_sql_migration.py').chmod(0o640)
             shutil.chown(target / 'runtime_sql_migration.py', user='root', group='smail')
+    if mixed:
+        shutil.copy2(Path(__file__).with_name('runtime_mixed_migration.py'), target / 'runtime_mixed_migration.py')
+        (target / 'runtime_mixed_migration.py').chmod(0o640)
+        shutil.chown(target / 'runtime_mixed_migration.py', user='root', group='smail')
     if args.interruption:
         shutil.copy2(Path(__file__).with_name('runtime_interruption.py'), target / 'runtime_interruption.py')
         (target / 'runtime_interruption.py').chmod(0o640)
@@ -197,11 +202,17 @@ def main():
         from runtime_sql import SQLFixture
         sql_fixture = SQLFixture(args.database_type)
         result['sql_database'] = sql_fixture.settings['name']
-        if args.migration:
-            from runtime_sql_migration import MigrationTarget
-            migration_target = MigrationTarget(target_kind)
-            result['sql_target_database'] = migration_target.settings['name']
-            result['migration_target_type'] = target_kind
+    if args.migration and target_kind != 'sqlite':
+        from runtime_sql_migration import MigrationTarget
+        migration_target = MigrationTarget(target_kind)
+        result['sql_target_database'] = migration_target.settings['name']
+    if args.migration:
+        result['migration_target_type'] = target_kind
+    mixed_source = mixed_target = None
+    if mixed:
+        from runtime_mixed_migration import SQLiteDatabase
+        mixed_source = sql_fixture or SQLiteDatabase(root, {'kind': 'sqlite', 'name': 'smail.db'})
+        mixed_target = migration_target or SQLiteDatabase(root, {'kind': 'sqlite', 'name': 'migrated.db'})
     if args.nginx_template:
         from runtime_proxy import prepare_nginx
         prepare_nginx(target, root, args.nginx_template)
@@ -218,6 +229,7 @@ def main():
         network_args = ['--property=PrivateNetwork=yes']
         if migration_target:
             migration_target.create()
+        if args.migration and (sql_fixture or migration_target):
             # 使用独立限时单元固定隔离网络，应用换进程后仍能访问本轮数据库桥接。
             network_unit = 'smail-runtime-netns-' + stamp
             subprocess.run([
@@ -255,11 +267,9 @@ def main():
         assert str(Path('/proc/' + pid + '/ns/net').readlink()) != live_net
         assert Path('/proc/' + pid + '/root/opt/smail/smail').stat().st_ino == (root / 'smail').stat().st_ino
         assert (root / 'smail').stat().st_ino != (live / 'smail').stat().st_ino
-        if sql_fixture:
+        if sql_fixture or migration_target:
             from runtime_sql_bridge import DatabaseBridge
-            ports = {sql_fixture.port}
-            if migration_target:
-                ports.add(migration_target.port)
+            ports = {fixture.port for fixture in (sql_fixture, migration_target) if fixture}
             for port in sorted(ports):
                 bridges[port] = DatabaseBridge(int(pid), port)
             result['database_bridge_ports'] = sorted(bridges)
@@ -309,7 +319,10 @@ def main():
             if args.interruption:
                 from runtime_interruption import verify_interruption
                 verify_interruption(root, unit, password, result)
-            if migration_target:
+            if mixed:
+                from runtime_mixed_migration import verify_migration
+                verify_migration(root, unit, password, result, mixed_source, mixed_target)
+            elif migration_target:
                 from runtime_sql_migration import verify_migration
                 verify_migration(root, unit, password, result, sql_fixture, migration_target)
             else:
@@ -323,7 +336,10 @@ def main():
                 result['live_unchanged'] = (result['after'] == before
                                             and hashlib.file_digest(stream, 'sha256').hexdigest() == live_sha)
             try:
-                if migration_target:
+                if mixed:
+                    from runtime_mixed_migration import migration_integrity
+                    result['integrity'] = migration_integrity(root, mixed_source, mixed_target)
+                elif migration_target:
                     from runtime_sql_migration import assert_active
                     assert_active(root, migration_target)
                     result['integrity'] = migration_target.integrity(root)
@@ -362,7 +378,10 @@ def main():
     assert result['live_unchanged'] and result['oom_events'] == 0
     assert not result['cleanup_errors'], '测试单元未全部停止，需要核查保留的结果'
     assert not result['resource_errors'], '资源证据不完整，需要核查保留的结果'
-    if sql_fixture:
+    if mixed:
+        assert result['migration']['passed'] and result['network_anchor_stopped']
+        assert result['integrity']['config_records'] > 0
+    elif sql_fixture:
         assert result['integrity'].get('sqlite_files') == 0 and result['integrity']['config_records'] > 0
         if args.migration:
             assert result['migration']['passed'] and result['network_anchor_stopped']

test/tools/runtime_sql_migration.py

@@ -62,6 +62,8 @@ class MigrationTarget(SQLFixture):
 
 def target_payload(database):
     settings = database.settings
+    if database.kind == 'sqlite':
+        return {'type': 'sqlite', 'localRef': settings['name']}
     return {'type': 'mysql' if database.kind == 'mysql' else 'postgresql',
             'host': '127.0.0.1', 'port': database.port, 'username': settings['user'],
             'name': settings['name'], 'password': {'plaintext': settings['password']}}
@@ -104,6 +106,10 @@ def history_rows(database):
     if database.kind == 'mysql':
         values = ','.join('LOWER(HEX(`' + column + '`))' for column in ('kind', 'id', 'payload', 'digest'))
         query = 'SELECT JSON_ARRAY(' + values + ') FROM config_records'
+    elif database.kind == 'sqlite':
+        values = ','.join('CASE WHEN "' + column + '" IS NULL THEN NULL ELSE lower(hex("' + column + '")) END'
+                          for column in ('kind', 'id', 'payload', 'digest'))
+        query = 'SELECT json_array(' + values + ') FROM config_records'
     else:
         values = ["encode(convert_to(" + column + ",'UTF8'),'hex')" for column in ('kind', 'id')]
         values += ["encode(payload,'hex')", "encode(convert_to(digest,'UTF8'),'hex')"]
@@ -120,20 +126,22 @@ def history_rows(database):
     return sorted(rows)
 
 
+def connection_count(database):
+    name, user = database.settings['name'], database.settings['user']
+    if database.kind == 'mysql':
+        query = "SELECT COUNT(*) FROM information_schema.processlist WHERE USER='" + user + "' AND DB='" + name + "';"
+    else:
+        query = "SELECT COUNT(*) FROM pg_stat_activity WHERE usename='" + user + "' AND datname='" + name + "';"
+    return int(database.admin(query))
+
+
 def assert_connections(source, target):
-    counts = []
-    for database in (source, target):
-        name, user = database.settings['name'], database.settings['user']
-        if database.kind == 'mysql':
-            query = "SELECT COUNT(*) FROM information_schema.processlist WHERE USER='" + user + "' AND DB='" + name + "';"
-        else:
-            query = "SELECT COUNT(*) FROM pg_stat_activity WHERE usename='" + user + "' AND datname='" + name + "';"
-        counts.append(int(database.admin(query)))
+    counts = [connection_count(database) for database in (source, target)]
     assert counts[0] == 0 and counts[1] > 0, '源库仍有活动连接或目标库没有应用连接'
     return {'source_connections': counts[0], 'target_connections': counts[1]}
 
 
-def prepare(root, password, source, target):
+def prepare(root, password, source, target, active_check=assert_active):
     client, _ = login(root, password)
     generation = clear_pending_restart(client)
     endpoint = '/api/v1/database/migrations'
@@ -142,7 +150,7 @@ def prepare(root, password, source, target):
     })
     record = client.success(endpoint, {'id': preview['migration']['id'], 'token': preview['token']})
     assert record['status'] == 'succeeded'
-    assert_active(root, target)
+    active_check(root, target)
     pending = client.success('/api/v1/config/status', method='GET')
     assert pending['activeGeneration'] > generation and pending['pendingRestart']
     status, denied, _ = client.request('/api/v1/security/credentials/protocols/generate', {'protocol': 'imap'})
@@ -154,9 +162,9 @@ def prepare(root, password, source, target):
             'existing_session_preserved': True, 'generation': pending['activeGeneration']}
 
 
-def verify_mail(root, password, phase, source, target):
+def verify_mail(root, password, phase, source, target, active_check=assert_active):
     client, context = login(root, password)
-    assert_active(root, target)
+    active_check(root, target)
     assert not client.success('/api/v1/config/status', method='GET')['pendingRestart']
     codes = {}
     for protocol in ('smtp', 'imap', 'pop3'):