diff --git a/docs/SQL复制中断恢复补充验收.md b/docs/SQL复制中断恢复补充验收.md new file mode 100644 index 0000000..523e3e5 --- /dev/null +++ b/docs/SQL复制中断恢复补充验收.md @@ -0,0 +1,78 @@ +# SQL 复制中断恢复补充验收 + +## 范围与放行边界 + +本轮补充 MySQL、PostgreSQL 同引擎迁移的未提交复制中断验收。只操作随机隔离数据库和本轮 Smail 测试进程，不停止共享数据库，不替换 `/opt/smail` 正式程序，不修改 DNS 或正式证书。 + +本记录不覆盖事务提交瞬间、配置落盘切换临界点、跨引擎复制中断、真实浏览器、外部投递、自动续期、长时容量或异机灾备。未关闭项目仍是生产放行门禁，不能将本项通过扩大为全项目可部署。 + +## 构建与隔离条件 + +- 源码基线：`8dc040a9ab14c32efea33c8ceedfd441d739a457`，修改前已提交并推送。 +- 最新后端重新编译：Linux amd64、CGO 关闭、内置插件模式。 +- 应用摘要：`468c0ce4dd6555d58c8cb79a920c9a0e9008e41e8d7894d4704d6f212036d761`。 +- 709 份输入文件记录摘要，包括后端源码、模块文件及实际嵌入资源。前端和 Linux 插件资源复用并核对摘要，不宣称本轮完整发行包重新构建。 +- 应用硬内存上限 192 MiB、Go 内存目标 128 MiB、单线程运行、CPU 配额 25%，测试单元禁止交换。 +- 每轮使用全新源库、未创建的正常迁移目标和独立中断目标；账号随机且权限仅限验收所需。PostgreSQL 建库权限临时授予，结束后撤销。 +- 保留离线源库与失败目标作为证据，不删除任何文件或数据库；正常运行只连接一个业务库，配置历史与业务同库。 + +## 中断证据要求 + +1. 通过非特权身份在隔离源库补充 1,024 条邮件，每条正文 16 KiB，保留原始邮件及表结构。 +2. 通过正式迁移 API 发起复制，不给应用增加 HTTP 故障注入入口或改变迁移行为。 +3. MySQL 按唯一测试账号和库过滤活动事务，要求修改行数大于 128；PostgreSQL 同时检查事务号、未结束事务、邮件表写锁和实际分配的数据页。 +4. 仅冻结与当前隔离目录程序 inode 匹配的测试单元，再次验证事务仍存在，独立连接读取目标邮件和配置历史均为零行，活动配置仍指向源。 +5. SIGKILL 后确认进程退出信号为 9、目标连接归零、事务回滚、源库全表数据摘要及配置文件摘要保持一致。 +6. 重启恢复后验证登录、历史邮件正文及附件、新邮件 SMTP 收取和 IMAP／POP3 读取；失败目标预检必须返回 HTTP 409 且结构化错误码为 `DATABASE_MIGRATION_TARGET_EXISTS`，其逻辑内容不得被覆盖。 +7. 随后使用另一个未占用目标完成正常迁移、内部重启、进程重启、目标账号故障拒绝启动及恢复；确认源库无活动连接、新邮件只在活动目标、配置历史完整保留。 + +连接断开、空目标存在、普通重启、任意 HTTP 409 都不能单独视为通过。数据库服务器不在 SIGSTOP／SIGKILL 范围内。 + +## 本地测试与保留的失败 + +新增九项工具回归，覆盖正式路径拒绝、同引擎限制、非破坏性种子、标识校验、事务观察过滤、已提交及缺失事务反例、正常响应不能冒充中断、创建失败仍锁定账号并保留结果。 + +Windows 运行时工具回归 90 项通过。Windows 全部署套件发现 121 个测试条目，结果为两项断言失败和两个加载错误：Linux 证书工具依赖 `fcntl`，另外两个用例依赖 Linux 路径格式。失败原始日志保存在 `sql-interruption-deployment-windows.log`，不计为通过，也不通过忽略异常来消除失败。 + +## 实机结果与复审 + +两轮使用同一最新后端候选，实际退出码均为零： + +| 取证项 | MySQL | PostgreSQL | +| --- | --- | --- | +| 隔离实例编号 | `20260921-222939` | `20260921-223902` | +| 未提交事务数 | 1 | 1 | +| 事务修改行数 | 190 | 不使用此指标 | +| 邮件表物理页字节数 | 不使用此指标 | 57,344 | +| 独立查询的已提交邮件／配置历史行数 | 0／0 | 0／0 | +| 中断前 PID | 840882 | 842979 | +| 中断后恢复 PID | 841717 | 843862 | +| 中断退出信号 | 9 | 9 | +| 后续迁移源／目标配置历史记录数 | 1／1，完整行一致 | 1／1，完整行一致 | +| 正常迁移后三阶段源连接数 | 均为 0 | 均为 0 | +| 运行目录中的 SQLite 文件数 | 0 | 0 | +| 采集到的应用内存峰值字节数 | 142,495,744 | 142,200,832 | +| OOM／交换峰值／收尾错误 | 均为 0 | 均为 0 | + +内存峰值为最终采集时的 systemd 单元资源值，不据此宣称覆盖全部进程重启前的历史峰值。全表数据摘要不替代物理页快照、序列状态或断电耐久性验证。中断恢复邮件探针复用了普通迁移验证器，其中 `mail.new_mail_only_in_target` 的目标参数是恢复后的活动源库，不是中断残留库。 + +云端完整 Linux 部署工具回归在 `/var/tmp/smail-review-python-20260921-224608/` 执行：160 项、零跳过、零失败，测试耗时 24.709 秒。日志 `python-linux-tests-20260921-224608.log` 的 SHA-256 为 `73303ae1ef525cf297f667a256fcc4c70b051b9bfb7725feb5be12f43e37ae0f`。其中证书相关用例是隔离工具回归，不是真实 DNS／ACME 续期验收。 + +- `[✓]` MySQL、PostgreSQL 实机复制中断、恢复和后续正常迁移完成并复审通过。 +- `[✓]` 完整 Linux 部署工具回归 160 项通过，零跳过。 +- `[✓]` 本地与远端证据、服务状态、数据库账号和资源清理独立复核通过。 +- `[ ]` 事务提交、配置切换临界点及最终统一发行候选验收。 + +证据根目录为 `test-log/production-remediation-20260921/`： + +- 构建：`sql-interruption-build-manifest.json`、`sql-interruption-source-manifest.json`。 +- MySQL 汇总：`sql-interruption-results-20260921-222924.json`。 +- PostgreSQL 汇总：`sql-interruption-results-20260921-223851.json`。 +- 实机原始证据：`sql-interruption-runtime-evidence/` 下两个对应编号目录，各 20 份文件。 +- 本地回归：`sql-interruption-tool-tests-verified.log`。 +- Linux 回归：`python-linux-result-20260921-224608.json` 及上述原始日志。 +- 独立复审：`sql-interruption-final-recheck.json`。 + +北京时间 2026 年 9 月 21 日 22:56，独立复核 40 份实机证据、两份部署程序副本及两组八个工具的本地／远端摘要一致；709 份构建输入（含 619 份 Go 源文件）没有漂移。六个测试 SQL 账号全部锁定，PostgreSQL 三个账号均无建库权限，两轮测试库连接均归零，各十个应用／探针／网络单元停止，Linux 部署工具单元也已停止。正式六服务的 PID、InvocationID 和重启计数未变，正式程序摘要仍为 `c14ef1c5ba1dc2096c312c8357039a7330c985cbeac90765aed01d8151e05e1f`。用户原有 `云端全项目测试报告.md` 没有编辑或暂存。 + +测试账号密码、协议凭据和会话值不进入本记录。 diff --git a/docs/生产补充验收记录.md b/docs/生产补充验收记录.md index 3fa66c0..a3fa9ec 100644 --- a/docs/生产补充验收记录.md +++ b/docs/生产补充验收记录.md @@ -364,3 +364,15 @@ PostgreSQL → SQLite 两次在初始化恢复码生成时发生 30 秒 HTTP 读 本地全量后端 1888 个测试及子测试通过、41 个跳过、零失败；Linux 恢复码专项 19 个测试及子测试通过、零跳过；本地运行时工具 81 项及云端部署工具 151 项通过，云端零跳过。不同范围和不同候选的结果分别保留，不能叠加为全部生产门禁通过。 收尾独立复核 127 份文件本地与远端摘要、616 个后端源码摘要均一致，八个测试数据库账号锁定，三次超时失败证据保留。正式六服务和主程序未变，测试单元全部停止；用户原测试报告没有编辑或暂存。其余安全、续期、浏览器、外部投递、容量和灾备门禁继续未关闭。 + +### 9 月 21 日双 SQL 未提交复制中断补充 + +详见 `SQL复制中断恢复补充验收.md`。最新后端重新编译的同一候选 `468c0ce4dd6555d58c8cb79a920c9a0e9008e41e8d7894d4704d6f212036d761` 完成 MySQL、PostgreSQL 未提交复制事务中的真实进程终止及恢复。两轮均观察到唯一未结束事务，独立查询的目标邮件和配置历史均为零行；SIGKILL 后事务回滚，源库全表数据及活动配置摘要保持一致。恢复后历史与新邮件正文、附件、三协议读取及失败目标防覆盖均通过。 + +随后正常迁移、内部重启、进程重启、目标账号故障拒绝启动及恢复通过；活动库唯一、配置历史与业务同库，没有额外 SQLite。离线源库与失败目标保留为证据，不当作第二个活动业务库。使用 128 MiB Go 内存目标及 192 MiB 硬上限，不掩盖旧配置的超时失败。 + +本地运行时工具 90 项通过，云端完整 Linux 部署工具 160 项通过、零跳过。Windows 全部署套件的两项路径断言失败和两个 `fcntl` 加载错误单独保留，不能称全平台通过。前端及插件资源仅复用并核对摘要，不宣称本轮完整发行包已重新构建。 + +22:56 独立复核 40 份实机证据、两份程序及两组八个工具的远端摘要一致，709 份构建输入（含 619 份 Go 源文件）未漂移；六个测试账号锁定，PostgreSQL 建库权限撤销，测试库连接及本轮测试进程归零。正式六服务和程序未变，用户原全项目报告没有编辑或暂存。本项复审通过，但未替换正式部署。 + +事务提交和配置切换临界点、跨引擎复制中断及最终完整发行候选统一验收仍未关闭；其他生产门禁维持原状态，不将本轮双 SQL 结果扩展为整体生产放行。 diff --git a/test/deployment/test_runtime_sql_interruption.py b/test/deployment/test_runtime_sql_interruption.py new file mode 100644 index 0000000..e0dd30e --- /dev/null +++ b/test/deployment/test_runtime_sql_interruption.py @@ -0,0 +1,115 @@ +"""SQL 中断验收的范围、真实事务判定与清理回归，不连接真实数据库。""" + +import http.client +from pathlib import Path +import sys +import tempfile +import unittest +from unittest.mock import Mock, patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools')) +import runtime_sql_interruption as interruption +from runtime_sql import SQLFixture +import runtime_review as review + + +class RuntimeSQLInterruptionTest(unittest.TestCase): + def test_same_engine_interruption_only(self): + for kind in ('sqlite', 'mysql', 'postgres'): + self.assertEqual(review.migration_kind(kind, None, True, True), kind) + with self.assertRaises(ValueError): + review.migration_kind('mysql', 'postgres', True, True) + + def test_formal_instance_rejected_before_target_creation(self): + with patch.object(interruption, 'MigrationTarget') as target: + with self.assertRaises(AssertionError): + interruption.verify_interruption(Path('/opt/smail'), 'smail.service', 'unused', {}, SQLFixture('mysql')) + target.assert_not_called() + + def test_seed_does_not_copy_primary_key_or_change_existing_rows(self): + for kind in ('mysql', 'postgres'): + source = SQLFixture(kind) + with patch.object(source, 'query', side_effect=['id\nmsg_id\ntext\nsubject', '1', '', '1025']) as query: + self.assertEqual(interruption.seed_source(source)['added_rows'], 1024) + sql = query.call_args_list[2].args[0] + self.assertTrue(sql.startswith('INSERT INTO email')) + self.assertNotIn('UPDATE ', sql) + self.assertNotIn('DELETE ', sql) + self.assertNotIn('DROP ', sql) + self.assertIn('REPEAT(\'x\',16384)', sql) + self.assertIn('SELECT 1023 AS i', sql) + self.assertNotIn('id', sql.split(' SELECT ')[0].replace('msg_id', '')) + + def test_seed_rejects_unsafe_column_before_write(self): + source = SQLFixture('mysql') + with patch.object(source, 'query', return_value='id\nmsg_id\ntext\nbad;DROP') as query: + with self.assertRaises(AssertionError): + interruption.seed_source(source) + self.assertEqual(query.call_count, 1) + + def test_mysql_observation_filters_exact_account_and_database(self): + target = SQLFixture('mysql') + with patch.object(target, 'admin', return_value='150') as admin: + self.assertEqual(interruption.copy_observation(target), + {'open_copy_transactions': 1, 'modified_rows': 150}) + sql = admin.call_args.args[0] + for key in ('name', 'user'): + self.assertIn(target.settings[key], sql) + self.assertNotIn(target.settings['password'], sql) + self.assertIn('trx_rows_modified>128', sql) + with patch.object(target, 'admin', return_value=''): + self.assertEqual(interruption.copy_observation(target)['open_copy_transactions'], 0) + + def test_postgres_requires_database_pages_xid_and_write_lock(self): + target = SQLFixture('postgres') + with patch.object(target, 'admin', return_value='0') as admin: + self.assertEqual(interruption.copy_observation(target)['open_copy_transactions'], 0) + self.assertEqual(admin.call_count, 1) + for size, expected in (('0', 0), ('32768', 0), ('65536', 1)): + with patch.object(target, 'admin', side_effect=['1', '1', size]) as admin: + self.assertEqual(interruption.copy_observation(target)['open_copy_transactions'], expected) + sql = admin.call_args_list[1].args[0] + for value in (target.settings['name'], target.settings['user'], 'backend_xid', 'RowExclusiveLock'): + self.assertIn(value, sql) + self.assertEqual(admin.call_args_list[1].args[1], target.settings['name']) + + def test_committed_or_missing_or_ambiguous_copy_cannot_pass(self): + target = SQLFixture('mysql') + for count, rows in ((0, ['0', '0']), (2, ['0', '0']), + (1, ['1', '0']), (1, ['0', '1'])): + with patch.object(target, 'query', side_effect=rows): + with self.assertRaises(AssertionError): + interruption.assert_uncommitted(target, {'open_copy_transactions': count}) + with patch.object(target, 'query', side_effect=['0', '0']): + self.assertEqual(interruption.assert_uncommitted(target, {'open_copy_transactions': 1})['committed_mail_rows'], 0) + + def test_normal_response_is_not_interruption(self): + client = Mock() + client.success.return_value = {'migration': {'id': 'one'}, 'token': 'test-token'} + with patch.object(interruption, 'login', return_value=(client, None)), \ + patch.object(interruption, 'clear_pending_restart', return_value=1): + for response in ((200, {}, {}), (500, {}, {})): + client.request.return_value = response + with self.assertRaises(AssertionError): + interruption.prepare(Path('/unused'), 'unused', SQLFixture('postgres')) + client.request.side_effect = http.client.RemoteDisconnected() + self.assertEqual(interruption.prepare(Path('/unused'), 'unused', SQLFixture('postgres')), + {'request_interrupted': True}) + + def test_partial_create_failure_still_locks_account_and_preserves_result(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) / 'instance' + source, target = SQLFixture('postgres'), Mock() + target.settings = SQLFixture('postgres').settings + target.create.side_effect = RuntimeError('controlled failure') + result = {} + with patch.object(interruption, 'assert_scope'), patch.object(interruption, 'MigrationTarget', return_value=target): + with self.assertRaises(RuntimeError): + interruption.verify_interruption(root, 'unused', 'unused', result, source) + target.lock.assert_called_once_with() + self.assertFalse(result['interruption']['passed']) + self.assertTrue((root.parent / 'interruption-result.json').exists()) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/tools/runtime_migration.py b/test/tools/runtime_migration.py index 4da4f5e..1491268 100644 --- a/test/tools/runtime_migration.py +++ b/test/tools/runtime_migration.py @@ -206,7 +206,8 @@ def run_probe(root, unit, password, phase, sql_settings=None, probe_script=None) 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) - or (probe_script == 'runtime_mixed_migration.py' and sql_settings is not None)) + or (probe_script in ('runtime_mixed_migration.py', 'runtime_sql_interruption.py') + and sql_settings is not None)) script = probe_script command = ['systemd-run', '--wait', '--pipe', '--unit=' + probe, '--property=User=smail', '--property=Group=smail', diff --git a/test/tools/runtime_review.py b/test/tools/runtime_review.py index 92df35b..71ff3a3 100644 --- a/test/tools/runtime_review.py +++ b/test/tools/runtime_review.py @@ -130,8 +130,8 @@ def main(): target_kind = migration_kind(args.database_type, args.migration_target_type, args.migration, args.interruption) except ValueError as error: parser.error(str(error)) - if args.interruption and (not args.migration or args.database_type != 'sqlite'): - parser.error('中断验收必须选择 SQLite 迁移独立场景') + if args.interruption and not args.migration: + parser.error('中断验收必须选择同引擎迁移独立场景') if args.migration and (args.nginx_template or args.restore): parser.error('迁移进程验收仅使用直接 TLS 的独立场景') if args.restore and args.database_type != 'sqlite': @@ -190,6 +190,10 @@ def main(): shutil.copy2(Path(__file__).with_name('runtime_interruption.py'), target / 'runtime_interruption.py') (target / 'runtime_interruption.py').chmod(0o640) shutil.chown(target / 'runtime_interruption.py', user='root', group='smail') + if args.database_type != 'sqlite': + shutil.copy2(Path(__file__).with_name('runtime_sql_interruption.py'), target / 'runtime_sql_interruption.py') + (target / 'runtime_sql_interruption.py').chmod(0o640) + shutil.chown(target / 'runtime_sql_interruption.py', user='root', group='smail') if args.nginx_template: shutil.copy2(Path(__file__).with_name('runtime_proxy.py'), target / 'runtime_proxy.py') (target / 'runtime_proxy.py').chmod(0o640) @@ -342,8 +346,12 @@ def main(): result['live_message_absent'] = True if args.migration: if args.interruption: - from runtime_interruption import verify_interruption - verify_interruption(root, unit, password, result) + if sql_fixture: + from runtime_sql_interruption import verify_interruption + verify_interruption(root, unit, password, result, sql_fixture) + else: + from runtime_interruption import verify_interruption + verify_interruption(root, unit, password, result) if mixed: from runtime_mixed_migration import verify_migration verify_migration(root, unit, password, result, mixed_source, mixed_target) diff --git a/test/tools/runtime_sql_interruption.py b/test/tools/runtime_sql_interruption.py new file mode 100644 index 0000000..936f46f --- /dev/null +++ b/test/tools/runtime_sql_interruption.py @@ -0,0 +1,222 @@ +"""在随机 SQL 测试库中捕捉未提交复制，只终止本轮隔离应用。""" + +import argparse +import hashlib +import http.client +import json +import os +from pathlib import Path +import subprocess +import sys +import threading +import time + +from runtime_interruption import assert_occupied_target, assert_scope +from runtime_migration import clear_pending_restart, login, run_probe, service_state +from runtime_sql import SQLDatabase +from runtime_sql_migration import ( + MigrationTarget, assert_active, connection_count, fingerprint, identifier, + target_payload, verify_mail, +) + + +def seed_source(source): + schema = 'DATABASE()' if source.kind == 'mysql' else "'public'" + columns = source.query('SELECT column_name FROM information_schema.columns WHERE table_schema=' + + schema + " AND table_name='email' ORDER BY ordinal_position").splitlines() + assert {'id', 'msg_id', 'text'}.issubset(columns) + quote = '`' if source.kind == 'mysql' else '"' + columns = [identifier(column) for column in columns if column != 'id'] + names = ','.join(quote + column + quote for column in columns) + values = [] + for column in columns: + if column == 'msg_id': + values.append("CONCAT('interruption-fill-',n.i,'@runtime.invalid.internal')") + elif column == 'text': + values.append("REPEAT('x',16384)") + else: + values.append('e.' + quote + column + quote) + before = int(source.query('SELECT COUNT(*) FROM email')) + assert before > 0 + numbers = ' UNION ALL '.join('SELECT ' + str(index) + ' AS i' for index in range(1024)) + source.query('INSERT INTO email (' + names + ') SELECT ' + ','.join(values) + + ' FROM (SELECT * FROM email ORDER BY id LIMIT 1) e CROSS JOIN (' + numbers + ') n;') + assert int(source.query('SELECT COUNT(*) FROM email')) == before + 1024 + return {'added_rows': 1024, 'text_bytes_per_row': 16384} + + +def copy_observation(target): + name, user = target.settings['name'], target.settings['user'] + if target.kind == 'mysql': + raw = target.admin("SELECT t.trx_rows_modified FROM information_schema.innodb_trx t " + "JOIN information_schema.processlist p ON p.ID=t.trx_mysql_thread_id " + "WHERE p.USER='" + user + "' AND p.DB='" + name + + "' AND t.trx_rows_modified>128;") + rows = [int(value) for value in raw.splitlines()] + return {'open_copy_transactions': len(rows), 'modified_rows': max(rows, default=0)} + exists = target.admin("SELECT COUNT(*) FROM pg_database WHERE datname='" + name + "';") + if exists != '1': + return {'open_copy_transactions': 0, 'relation_bytes': 0} + raw = target.admin("SELECT COUNT(*) FROM pg_stat_activity a WHERE a.usename='" + user + + "' AND a.datname='" + name + "' AND a.xact_start IS NOT NULL " + "AND a.backend_xid IS NOT NULL AND EXISTS (SELECT 1 FROM pg_locks l " + "WHERE l.pid=a.pid AND l.locktype='relation' AND l.mode='RowExclusiveLock' " + "AND l.granted AND l.relation=to_regclass('public.email'));", name) + size = target.admin("SELECT COALESCE(pg_relation_size(to_regclass('public.email')),0);", name) + # 只接受真实分配了邮件页、持有写锁且分配了事务号的复制，不把空结构创建算成复制。 + count, size = int(raw), int(size) + return {'open_copy_transactions': count if size > 32768 else 0, + 'relation_bytes': size} + + +def assert_uncommitted(target, observation): + assert observation['open_copy_transactions'] == 1, '未捕获唯一未提交复制事务' + assert target.query('SELECT COUNT(*) FROM email') == '0', '目标邮件已经提交，不能算未提交中断' + assert target.query('SELECT COUNT(*) FROM config_records') == '0', '配置历史已经提交' + return dict(observation, committed_mail_rows=0, committed_config_rows=0) + + +def prepare(root, password, target): + client, _ = login(root, password) + generation = clear_pending_restart(client) + preview = client.success('/api/v1/database/migrations/preview', { + 'expectedGeneration': generation, 'target': target_payload(target), + }) + try: + client.request('/api/v1/database/migrations', { + 'id': preview['migration']['id'], 'token': preview['token'], + }) + except (OSError, http.client.HTTPException): + return {'request_interrupted': True} + raise AssertionError('请求正常结束，不能当作进程中断验收') + + +def verify_recovery(root, password, source, target): + client, _ = login(root, password) + assert_active(root, source) + status = client.success('/api/v1/config/status', method='GET') + assert not status['pendingRestart'] + digest = fingerprint(target) + denied, body, _ = client.request('/api/v1/database/migrations/preview', { + 'expectedGeneration': status['activeGeneration'], 'target': target_payload(target), + }) + rejected = assert_occupied_target(denied, body) + assert fingerprint(target) == digest + # 中断后源库仍是唯一活动库；复用三协议验收，验证新邮件不写入失败目标。 + mail = verify_mail(root, password, 'interrupt-recovery', target, source) + return {'occupied_target_rejected': rejected, 'retained_target_unchanged': True, + 'source_accepts_writes': True, 'mail': mail} + + +def verify_interruption(root, unit, password, result, source): + assert_scope(root, unit) + assert isinstance(source, SQLDatabase) and source.kind in ('mysql', 'postgres') + target = MigrationTarget(source.kind) + assert source.settings['name'] != target.settings['name'] + details = result['interruption'] = {'passed': False, 'database_type': source.kind, + 'retained_target_database': target.settings['name']} + settings = {'source': source.settings, 'target': target.settings} + frozen, thread, outcome = False, None, {} + + def probe(phase): + return run_probe(root, unit, password, phase, settings, + probe_script='runtime_sql_interruption.py') + + def request(): + try: + outcome['probe'] = probe('interrupt-copy') + except Exception as error: + outcome['error'] = type(error).__name__ + + try: + target.create() + before = service_state(unit) + details['seed'] = probe('interrupt-seed') + active = service_state(unit) + assert active['MainPID'] != '0' and active['MainPID'] == before['MainPID'] + assert Path('/proc/' + active['MainPID'] + '/root/opt/smail/smail').stat().st_ino == (root / 'smail').stat().st_ino + thread = threading.Thread(target=request, daemon=True) + thread.start() + deadline = time.monotonic() + 90 + while time.monotonic() < deadline: + observation = copy_observation(target) + if observation['open_copy_transactions'] == 1: + subprocess.run(['systemctl', 'kill', '--kill-whom=all', '--signal=SIGSTOP', unit], check=True, timeout=15) + frozen = True + break + assert thread.is_alive(), '复制请求提前退出，未注入中断' + time.sleep(0.02) + assert frozen, '未捕获未提交事务，不能把普通重启算成中断' + details['copy'] = assert_uncommitted(target, copy_observation(target)) + assert_active(root, source) + source_digest = fingerprint(source) + config_digest = hashlib.sha256((root / 'config/config.json').read_bytes()).hexdigest() + subprocess.run(['systemctl', 'kill', '--kill-whom=all', '--signal=SIGKILL', unit], check=True, timeout=15) + frozen = False + for _ in range(100): + failed = service_state(unit) + if failed['MainPID'] == '0' and connection_count(target) == 0: + break + time.sleep(0.1) + assert failed['MainPID'] == '0' and failed['ExecMainStatus'] == '9' + assert connection_count(target) == 0 + thread.join(timeout=30) + assert not thread.is_alive() and 'error' not in outcome and outcome['probe']['request_interrupted'] + assert fingerprint(source) == source_digest + assert hashlib.sha256((root / 'config/config.json').read_bytes()).hexdigest() == config_digest + assert target.query('SELECT COUNT(*) FROM email') == '0' + assert target.query('SELECT COUNT(*) FROM config_records') == '0' + target.revoke_creation() + details.update(source_unchanged=True, configuration_unchanged=True, + target_rolled_back=True, killed_state=failed) + subprocess.run(['systemctl', 'start', unit], check=True, timeout=45) + details['recovery'] = probe('interrupt-recovery') + restored = service_state(unit) + assert restored['MainPID'] not in ('0', active['MainPID']) + assert restored['InvocationID'] != active['InvocationID'] + assert connection_count(target) == 0 + details.update(before=active, recovered_state=restored, passed=True) + finally: + try: + if frozen: + subprocess.run(['systemctl', 'kill', '--kill-whom=all', '--signal=SIGCONT', unit], check=True, timeout=15) + if thread: + thread.join(timeout=90) + finally: + try: + target.lock() + details['retained_target_account_locked'] = True + except Exception: + details['passed'] = False + raise + finally: + (root.parent / 'interruption-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=('interrupt-seed', 'interrupt-copy', 'interrupt-recovery')) + 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 = (SQLDatabase(settings[key]) for key in ('source', 'target')) + assert source.kind == target.kind and source.settings['name'] != target.settings['name'] + if args.phase == 'interrupt-seed': + result = seed_source(source) + elif args.phase == 'interrupt-copy': + result = prepare(args.root, password, target) + else: + result = verify_recovery(args.root, password, source, target) + 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)) + + +if __name__ == '__main__': + main()