补充SQL切换联合故障验收工具并校准系统调用边界

test/deployment/test_runtime_sql_switch_fault.py

@@ -0,0 +1,85 @@
+"""联合故障工具的范围、阶段与失败判定回归，不连接真实服务。"""
+
+import http.client
+import json
+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_switch_fault as fault
+from runtime_sql import SQLFixture
+
+
+class RuntimeSQLSwitchFaultTest(unittest.TestCase):
+    def test_every_thread_must_be_stopped(self):
+        task = Mock()
+        for state, expected in (('S', False), ('R', False), ('T', True), ('t', True)):
+            task.__truediv__ = Mock(return_value=Mock(read_text=Mock(return_value='State:\t' + state + ' (state)\n')))
+            with patch.object(Path, 'iterdir', return_value=iter([task])):
+                self.assertEqual(fault.stopped_tasks('123'), expected)
+
+    def test_formal_scope_rejected_before_process_or_database_access(self):
+        with patch.object(fault.subprocess, 'Popen') as process, patch.object(fault, 'service_state') as state:
+            with self.assertRaises(AssertionError):
+                fault.verify_switch_fault(Path('/opt/smail'), 'smail.service', 'unused', {},
+                                          SQLFixture('mysql'), SQLFixture('mysql'), 'before-root')
+            process.assert_not_called()
+            state.assert_not_called()
+
+    def test_boundary_requires_active_directory_phase_and_expected_database(self):
+        source, target = SQLFixture('mysql'), SQLFixture('mysql')
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            archive = root / 'data/transaction/one'
+            archive.mkdir(parents=True)
+            (root / 'config').mkdir()
+            trace = root / 'trace.log'
+            manifest = archive / 'manifest.json'
+            for phase, name in (('before-root', source.settings['name']), ('after-root', target.settings['name'])):
+                (root / 'config/config.json').write_text(json.dumps({'database': {'name': name}}))
+                manifest.write_text(json.dumps({'phase': 'sections_committed'}))
+                for path, expected in (('/opt/smail/data/transaction/one/candidate', False),
+                                       ('/opt/smail/config', True), ('/opt/smail/config-other', False)):
+                    trace.write_text('renameat(4<' + path + '>, ".config.json-a", 4<' + path + '>, "config.json"')
+                    self.assertEqual(bool(fault.boundary(root, trace, phase, source, target)), expected)
+                trace.write_text('renameat(4</opt/smail/config>, ".config.json-a", 4</opt/smail/config>, "config.json"')
+                manifest.write_text(json.dumps({'phase': 'root_committed'}))
+                self.assertIsNone(fault.boundary(root, trace, phase, source, target))
+
+    def test_committed_requires_matching_nonempty_mail_and_preserved_history(self):
+        source, target = Mock(), Mock()
+        for mail, history in (([[], []], [['a'], ['a']]),
+                              ([['a'], ['b']], [['a'], ['a']]),
+                              ([['a'], ['a']], [[], []]),
+                              ([['a'], ['a']], [['a'], ['b']])):
+            with patch.object(fault, 'table_rows', side_effect=mail), \
+                    patch.object(fault, 'history_rows', side_effect=history):
+                with self.assertRaises(AssertionError):
+                    fault.assert_committed(source, target)
+        with patch.object(fault, 'table_rows', side_effect=[['mail'], ['mail']]), \
+                patch.object(fault, 'history_rows', side_effect=[['old'], ['old', 'new']]):
+            self.assertEqual(fault.assert_committed(source, target)['committed_history_rows'], 2)
+
+    def test_timeout_and_regular_response_cannot_pass_as_killed_process(self):
+        for outcome in (None, TimeoutError(), http.client.RemoteDisconnected(), ConnectionResetError()):
+            with tempfile.TemporaryDirectory() as directory:
+                root = Path(directory) / 'instance'
+                client = Mock()
+                client.success.return_value = {'migration': {'id': 'test-id'}, 'token': 'not-for-logs'}
+                client.request.side_effect = outcome
+                with patch.object(fault, 'login', return_value=(client, None)), \
+                        patch.object(fault, 'clear_pending_restart', return_value=1), \
+                        patch.object(fault, 'wait_until'):
+                    if isinstance(outcome, (http.client.RemoteDisconnected, ConnectionResetError)):
+                        self.assertTrue(fault.prepare(root, 'unused', SQLFixture('mysql'))['request_disconnected'])
+                    else:
+                        with self.assertRaises((AssertionError, TimeoutError)):
+                            fault.prepare(root, 'unused', SQLFixture('mysql'))
+                self.assertEqual((root.parent / 'switch-ready').read_bytes(), b'')
+
+
+if __name__ == '__main__':
+    unittest.main()

test/tools/runtime_fault_calibration.py

@@ -0,0 +1,109 @@
+"""在独立微型进程中校准系统调用延迟，不附加正式或测试应用。"""
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import time
+
+
+CHILD = r'''
+import json, os, sys, time
+from pathlib import Path
+root = Path(sys.argv[1])
+while not (root / 'gate').exists():
+    time.sleep(0.01)
+fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY)
+start = time.monotonic()
+os.rename('unrelated.tmp', 'unrelated.json', src_dir_fd=fd, dst_dir_fd=fd)
+(root / 'unrelated-result.json').write_text(json.dumps({'seconds': time.monotonic() - start}))
+(root / 'ready').touch()
+start = time.monotonic()
+os.rename('candidate.tmp', 'config.json', src_dir_fd=fd, dst_dir_fd=fd)
+(root / 'done.json').write_text(json.dumps({'seconds': time.monotonic() - start}))
+os.close(fd)
+'''
+
+
+def wait_until(predicate, timeout=10):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return
+        time.sleep(0.02)
+    raise AssertionError('校准等待超时')
+
+
+def tracer_pid(pid):
+    lines = Path('/proc/' + str(pid) + '/status').read_text().splitlines()
+    return int(next(line.split(':')[1] for line in lines if line.startswith('TracerPid:')))
+
+
+def calibrate(root, phase):
+    assert phase in ('enter', 'exit')
+    root.mkdir(mode=0o700)
+    for name, value in [('config.json', 'old'), ('candidate.tmp', 'new'), ('unrelated.tmp', 'other')]:
+        (root / name).write_text(value)
+    child = subprocess.Popen([sys.executable, '-c', CHILD, str(root)],
+                             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    trace = None
+    details = {'phase': phase, 'passed': False}
+    try:
+        with (root / 'trace.stderr').open('w') as errors:
+            trace = subprocess.Popen(['strace', '-f', '-ttt', '-T', '-yy',
+                '-e', 'trace=rename,renameat,renameat2', '-P', 'config.json',
+                '-e', 'inject=rename,renameat,renameat2:delay_' + phase + '=3s',
+                '-o', str(root / 'trace.log'), '-p', str(child.pid)],
+                stdout=subprocess.DEVNULL, stderr=errors)
+            wait_until(lambda: tracer_pid(child.pid) == trace.pid)
+            wait_until(lambda: 'attached' in (root / 'trace.stderr').read_text())
+            (root / 'gate').touch()
+            wait_until(lambda: (root / 'ready').exists())
+            unrelated = json.loads((root / 'unrelated-result.json').read_text())
+            assert unrelated['seconds'] < 1, '路径过滤没有排除无关替换'
+            time.sleep(0.4)
+            observed = (root / 'config.json').read_text()
+            assert observed == ('old' if phase == 'enter' else 'new'), '延迟边界不符合预期'
+            assert 'renameat(' in (root / 'trace.log').read_text(), '延迟期间没有可观测的调用记录'
+            assert not (root / 'done.json').exists(), '替换调用未处于延迟窗口'
+            assert child.wait(timeout=10) == 0
+            assert trace.wait(timeout=5) == 0
+            duration = json.loads((root / 'done.json').read_text())['seconds']
+            assert 2.5 <= duration <= 8
+            assert (root / 'config.json').read_text() == 'new'
+            details.update(passed=True, unrelated_seconds=unrelated['seconds'],
+                           target_seconds=duration, observed_during_delay=observed,
+                           child_exited=True, tracer_exited=True,
+                           trace_sha256=hashlib.sha256((root / 'trace.log').read_bytes()).hexdigest())
+    finally:
+        # 只回收本函数亲自创建的子进程，不接收外部 PID，也不删除任何证据。
+        if child.poll() is None:
+            child.kill()
+            child.wait(timeout=10)
+        if trace is not None and trace.poll() is None:
+            trace.send_signal(signal.SIGTERM)
+            trace.wait(timeout=10)
+        (root / 'result.json').write_text(json.dumps(details, indent=2), encoding='utf-8')
+    return details
+
+
+def main():
+    assert os.geteuid() == 0
+    root = Path('/var/tmp') / ('smail-review-fault-calibration-' + time.strftime('%Y%m%d-%H%M%S'))
+    root.mkdir(mode=0o700)
+    result = {'root': str(root), 'passed': False}
+    try:
+        result['strace_version'] = subprocess.check_output(['strace', '-V'], text=True).splitlines()[0]
+        result['ptrace_scope'] = Path('/proc/sys/kernel/yama/ptrace_scope').read_text().strip()
+        result['cases'] = [calibrate(root / phase, phase) for phase in ('enter', 'exit')]
+        result['passed'] = True
+    finally:
+        (root / 'result.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
+        print(json.dumps(result, indent=2), flush=True)
+
+
+if __name__ == '__main__':
+    main()

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 in ('runtime_mixed_migration.py', 'runtime_sql_interruption.py')
+                or (probe_script in ('runtime_mixed_migration.py', 'runtime_sql_interruption.py',
+                                     'runtime_sql_switch_fault.py')
                     and sql_settings is not None))
         script = probe_script
     command = ['systemd-run', '--wait', '--pipe', '--unit=' + probe,

test/tools/runtime_review.py

@@ -122,6 +122,7 @@ def main():
     parser.add_argument('--restore', action='store_true')
     parser.add_argument('--migration', action='store_true')
     parser.add_argument('--interruption', action='store_true')
+    parser.add_argument('--switch-fault', choices=('before-root', 'after-root'))
     parser.add_argument('--database-type', choices=('sqlite', 'mysql', 'postgres'), default='sqlite')
     parser.add_argument('--migration-target-type', choices=('sqlite', 'mysql', 'postgres'))
     args = parser.parse_args()
@@ -132,6 +133,9 @@ def main():
         parser.error(str(error))
     if args.interruption and not args.migration:
         parser.error('中断验收必须选择同引擎迁移独立场景')
+    if args.switch_fault and (not args.migration or args.interruption or
+                              args.database_type == 'sqlite' or target_kind != args.database_type):
+        parser.error('联合切换故障必须使用独立的同引擎 SQL 迁移场景')
     if args.migration and (args.nginx_template or args.restore):
         parser.error('迁移进程验收仅使用直接 TLS 的独立场景')
     if args.restore and args.database_type != 'sqlite':
@@ -186,7 +190,7 @@ def main():
         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:
+    if args.interruption or args.switch_fault:
         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')
@@ -194,6 +198,10 @@ def main():
             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.switch_fault:
+        shutil.copy2(Path(__file__).with_name('runtime_sql_switch_fault.py'), target / 'runtime_sql_switch_fault.py')
+        (target / 'runtime_sql_switch_fault.py').chmod(0o640)
+        shutil.chown(target / 'runtime_sql_switch_fault.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)
@@ -352,7 +360,10 @@ def main():
                 else:
                     from runtime_interruption import verify_interruption
                     verify_interruption(root, unit, password, result)
-            if mixed:
+            if args.switch_fault:
+                from runtime_sql_switch_fault import verify_switch_fault
+                verify_switch_fault(root, unit, password, result, sql_fixture, migration_target, args.switch_fault)
+            elif mixed:
                 from runtime_mixed_migration import verify_migration
                 verify_migration(root, unit, password, result, mixed_source, mixed_target)
             elif migration_target:
@@ -374,8 +385,9 @@ def main():
                     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)
+                    active_database = sql_fixture if args.switch_fault == 'before-root' else migration_target
+                    assert_active(root, active_database)
+                    result['integrity'] = active_database.integrity(root)
                 elif args.migration:
                     from runtime_migration import migration_integrity
                     result['integrity'] = migration_integrity(root, interrupted=args.interruption)

test/tools/runtime_sql_switch_fault.py

@@ -0,0 +1,220 @@
+"""在隔离应用的真实 SQL 提交与根配置替换边界终止进程。"""
+
+import argparse
+import hashlib
+import http.client
+import json
+import os
+from pathlib import Path
+import re
+import signal
+import subprocess
+import sys
+import threading
+import time
+
+from runtime_interruption import assert_scope
+from runtime_migration import clear_pending_restart, login, run_probe, service_state
+from runtime_sql import SQLDatabase
+from runtime_sql_migration import (
+    assert_active, assert_connections, connection_count, fingerprint,
+    history_rows, table_rows, target_payload,
+)
+
+
+def wait_until(predicate, timeout=60):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return
+        time.sleep(0.02)
+    raise AssertionError('联合故障验收等待超时')
+
+
+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),
+    })
+    # 令牌与会话只在探针内存中；就绪标记不包含秘密。
+    (root.parent / 'switch-ready').touch(exist_ok=False)
+    wait_until(lambda: (root.parent / 'switch-gate').is_file())
+    try:
+        client.request('/api/v1/database/migrations', {
+            'id': preview['migration']['id'], 'token': preview['token'],
+        })
+    except (http.client.RemoteDisconnected, ConnectionResetError):
+        return {'request_disconnected': True}
+    raise AssertionError('未观察到连接被终止，普通响应或超时不能算进程终止验收')
+
+
+def traced_tasks(pid, tracer):
+    tasks = list(Path('/proc/' + str(pid) + '/task').iterdir())
+    assert tasks
+    return all('TracerPid:\t' + str(tracer) + '\n' in (task / 'status').read_text() for task in tasks)
+
+
+def stopped_tasks(pid):
+    tasks = list(Path('/proc/' + str(pid) + '/task').iterdir())
+    assert tasks
+    for task in tasks:
+        status = (task / 'status').read_text()
+        state = next(line.split()[1] for line in status.splitlines() if line.startswith('State:'))
+        if state not in ('T', 't'):
+            return False
+    return True
+
+
+def boundary(root, trace, phase, source, target):
+    manifests = list((root / 'data/transaction').glob('*/manifest.json'))
+    if len(manifests) != 1:
+        return None
+    manifest = json.loads(manifests[0].read_text())
+    if manifest['phase'] != 'sections_committed':
+        return None
+    text = trace.read_text() if trace.exists() else ''
+    # 同名归档也会延迟；只有目录描述符明确指向活动配置的调用才可命中。
+    pattern = r'renameat(?:2)?\(\d+</opt/smail/config>, "[^"\n]+", \d+</opt/smail/config>, "config.json"'
+    if not re.search(pattern, text):
+        return None
+    expected = source if phase == 'before-root' else target
+    if json.loads((root / 'config/config.json').read_text())['database']['name'] != expected.settings['name']:
+        return None
+    return manifest
+
+
+def assert_committed(source, target):
+    source_mail, target_mail = table_rows(source, 'email'), table_rows(target, 'email')
+    assert source_mail and source_mail == target_mail, '独立连接没有看到完整已提交邮件'
+    source_history, target_history = history_rows(source), history_rows(target)
+    assert source_history and set(source_history).issubset(target_history), '目标缺少已提交配置历史'
+    return {'committed_mail_rows': len(target_mail), 'committed_history_rows': len(target_history),
+            'mail_equal': True, 'source_history_preserved': True}
+
+
+def verify_switch_fault(root, unit, password, result, source, target, phase):
+    assert_scope(root, unit)
+    assert phase in ('before-root', 'after-root') and source.kind == target.kind
+    assert source.settings['name'] != target.settings['name']
+    details = result['migration'] = {'passed': False, 'fault_phase': phase, 'database_type': source.kind}
+    settings = {'source': source.settings, 'target': target.settings}
+    outcome = {}
+    thread = trace = None
+    frozen = False
+    trace_path = root.parent / 'switch-trace.log'
+
+    def request():
+        try:
+            outcome['probe'] = run_probe(root, unit, password, 'switch-request', settings,
+                                         probe_script='runtime_sql_switch_fault.py')
+        except Exception as error:
+            outcome['error'] = type(error).__name__
+
+    try:
+        initial = service_state(unit)
+        pid = initial['MainPID']
+        assert pid.isdigit() and int(pid) > 0
+        assert Path('/proc/' + pid + '/root/opt/smail/smail').samefile(root / 'smail')
+        assert Path('/proc/' + pid + '/ns/net').readlink() != Path('/proc/self/ns/net').readlink()
+        thread = threading.Thread(target=request, daemon=True)
+        thread.start()
+        wait_until(lambda: (root.parent / 'switch-ready').exists())
+        assert thread.is_alive() and service_state(unit)['MainPID'] == pid
+        assert_active(root, source)
+        assert list((root / 'data/transaction').iterdir()) == []
+        with (root.parent / 'switch-trace.stderr').open('w') as errors:
+            delay = 'enter' if phase == 'before-root' else 'exit'
+            trace = subprocess.Popen(['strace', '-f', '-ttt', '-T', '-yy',
+                '-e', 'trace=rename,renameat,renameat2', '-P', 'config.json',
+                '-e', 'inject=rename,renameat,renameat2:delay_' + delay + '=6s',
+                '-o', str(trace_path), '-p', pid], stdout=subprocess.DEVNULL, stderr=errors)
+            wait_until(lambda: traced_tasks(pid, trace.pid), timeout=10)
+            (root.parent / 'switch-gate').touch(exist_ok=False)
+            wait_until(lambda: boundary(root, trace_path, phase, source, target), timeout=55)
+            assert thread.is_alive() and trace.poll() is None
+            subprocess.run(['systemctl', 'kill', '--kill-whom=all', '--signal=SIGSTOP', unit], check=True, timeout=15)
+            frozen = True
+            wait_until(lambda: stopped_tasks(pid), timeout=5)
+            manifest = boundary(root, trace_path, phase, source, target)
+            assert manifest is not None, '冻结后故障边界已经变化'
+            details['committed'] = assert_committed(source, target)
+            active, offline = (source, target) if phase == 'before-root' else (target, source)
+            assert_active(root, active)
+            offline_digest = fingerprint(offline)
+            details['observed'] = {'manifest_phase': manifest['phase'],
+                'from_generation': manifest['fromGeneration'], 'to_generation': manifest['toGeneration'],
+                'root_database': active.settings['name'], 'active_root_syscall_observed': True,
+                'all_application_threads_stopped': True}
+            subprocess.run(['systemctl', 'kill', '--kill-whom=all', '--signal=SIGKILL', unit], check=True, timeout=15)
+            frozen = False
+            wait_until(lambda: service_state(unit)['MainPID'] == '0')
+            killed = service_state(unit)
+            assert killed['ExecMainStatus'] == '9'
+            trace.wait(timeout=15)
+            thread.join(timeout=35)
+            assert not thread.is_alive() and 'error' not in outcome and outcome['probe']['request_disconnected']
+            assert connection_count(source) == connection_count(target) == 0
+            assert fingerprint(offline) == offline_digest
+            target.revoke_creation()
+            recovery_settings = {'source': offline.settings, 'target': active.settings}
+            subprocess.run(['systemctl', 'start', unit], check=True, timeout=45)
+            details['recovered'] = run_probe(root, unit, password, 'recovered', recovery_settings)
+            recovered = service_state(unit)
+            assert recovered['MainPID'] not in ('0', pid)
+            assert recovered['InvocationID'] != initial['InvocationID']
+            details['connections'] = assert_connections(offline, active)
+            assert fingerprint(offline) == offline_digest
+            assert list((root / 'data/transaction').iterdir()) == []
+            assert list((root / 'data/transaction-staging').iterdir()) == []
+            # 再次从磁盘启动，证明恢复结果可重复使用，不只依赖首次恢复的内存状态。
+            subprocess.run(['systemctl', 'restart', unit], check=True, timeout=45)
+            details['second_restart'] = run_probe(root, unit, password, 'process', recovery_settings)
+            final = service_state(unit)
+            assert final['MainPID'] not in ('0', pid, recovered['MainPID'])
+            assert fingerprint(offline) == offline_digest
+            assert_connections(offline, active)
+            details.update(passed=True, offline_database_unchanged=True,
+                           temporary_creation_privilege_revoked=True,
+                           killed_state=killed, recovered_state=recovered, final_state=final,
+                           trace_sha256=hashlib.sha256(trace_path.read_bytes()).hexdigest())
+    finally:
+        # 异常路径不让带有延迟器的迁移继续执行；仅停止已经核验的隔离应用。
+        if trace is not None and trace.poll() is None:
+            subprocess.run(['systemctl', 'kill', '--kill-whom=all', '--signal=SIGKILL', unit],
+                           capture_output=True, timeout=15)
+            try:
+                trace.wait(timeout=15)
+            except subprocess.TimeoutExpired:
+                trace.send_signal(signal.SIGTERM)
+                trace.wait(timeout=10)
+        elif frozen:
+            subprocess.run(['systemctl', 'kill', '--kill-whom=all', '--signal=SIGKILL', unit],
+                           capture_output=True, timeout=15)
+        if thread is not None:
+            thread.join(timeout=90)
+        (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=('switch-request',))
+    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']
+    result = prepare(args.root, password, 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()