fs-lawrisk/analysis/checkpoint_analysis.md

检查点功能安全分析报告

🚨 严重Bug汇总

1. 数据丢失风险 - CRITICAL

位置: _restore_table() (第380-409行)

问题:

truncate_sql = f"TRUNCATE TABLE {table_name} CASCADE"
cur.execute(truncate_sql)

风险:

• 直接TRUNCATE表会永久删除现有数据
• CASCADE会级联删除依赖表的数据
• 如果恢复过程中出错，原始数据已丢失且无法恢复

2. 表依赖顺序错误 - HIGH

问题: 恢复表时没有考虑外键依赖关系

• 如果先恢复子表，再恢复父表时会因为外键约束失败
• 当前代码假设所有表都可以直接TRUNCATE，实际情况并非如此

3. 并发安全问题 - HIGH

问题: 恢复过程没有数据库锁

• 其他会话可能在恢复期间写入数据
• 导致数据不一致或恢复失败

4. 事务管理问题 - MEDIUM

问题: create_checkpoint中每个表独立事务

• 部分表备份失败不会影响已完成的部分
• 导致checkpoint数据不一致

详细分析

Bug #1: TRUNCATE CASCADE 危险操作

def _restore_table(conn, table_name, data):
    # 危险：直接清空表！
    truncate_sql = f"TRUNCATE TABLE {table_name} CASCADE"
    cur.execute(truncate_sql)

影响:

• 假设表A有外键指向表B
• 如果先TRUNCATE表B，CASCADE会删除表A中相关的行
• 即使后续恢复表B，表A的数据已经永久丢失

Bug #2: 没有表依赖拓扑排序

PostgreSQL表的外键依赖关系:

regions (父表)
  ├── region_themes
  ├── region_scopes
  ├── region_theme_permits
  ├── region_permit_risks
  └── region_permit_details

正确的恢复顺序:

1. 先恢复没有外键依赖的表 (regions, themes, business_scopes, permits, risks)
2. 再恢复引用其他表的表 (region_themes, region_scopes, region_theme_permits, etc.)

Bug #3: 缺少表锁定

恢复期间应该使用:

BEGIN;
LOCK TABLE table_name IN EXCLUSIVE MODE;
-- 恢复数据
COMMIT;

修复方案

方案1: 安全恢复流程

1. 自动备份当前状态 - 恢复前创建临时checkpoint
2. 表拓扑排序 - 按外键依赖逆序恢复
3. 表级锁 - 防止并发写入
4. 单事务 - 全部成功或全部失败
5. 回滚机制 - 恢复失败时自动回滚到备份

方案2: 安全restore实现

def restore_checkpoint_safe(checkpoint_id: str) -> Dict[str, Any]:
    """安全的checkpoint恢复，带自动回退"""
    # 1. 创建自动备份 (自动命名为auto_backup_<timestamp>)
    auto_backup = create_checkpoint(f"auto_backup_before_restore_{checkpoint_id}")
    
    # 2. 获取拓扑排序后的表列表
    ordered_tables = _get_tables_topological_order()
    
    # 3. 开始事务
    with _lic_pg_conn(autocommit=False) as conn:
        try:
            # 4. 锁定所有表
            for table in ordered_tables:
                conn.execute(f"LOCK TABLE {table} IN EXCLUSIVE MODE")
            
            # 5. 按依赖顺序恢复
            for table in ordered_tables:
                data = checkpoint_data["tables"].get(table, [])
                _restore_table_safe(conn, table, data)
            
            # 6. 提交
            conn.commit()
            return {"status": "success", "restored_from": checkpoint_id}
            
        except Exception as e:
            # 7. 回滚
            conn.rollback()
            # 可选：自动从auto_backup恢复
            return {"status": "error", "message": str(e)}

方案3: 表依赖图构建

def _get_tables_topological_order() -> List[str]:
    """获取按外键依赖排序的表列表"""
    sql = """
        SELECT
            tc.table_name,
            array_agg(ccu.table_name ORDER BY ccu.table_name) AS referenced_by
        FROM information_schema.table_constraints tc
        JOIN information_schema.key_column_usage kcu
          ON tc.constraint_name = kcu.constraint_name
          AND tc.table_schema = kcu.table_schema
        JOIN information_schema.constraint_column_usage ccu
          ON tc.constraint_name = ccu.constraint_name
          AND tc.table_schema = ccu.table_schema
        WHERE tc.constraint_type = 'FOREIGN KEY'
          AND tc.table_schema = 'public'
        GROUP BY tc.table_name
        ORDER BY tc.table_name
    """
    # 实现拓扑排序算法

立即修复建议

立即可做的修复:

1. 添加安全警告 - 在API文档中强调restore是危险操作
2. 表排序 - 按依赖关系排序恢复
3. 添加表锁 - 防止并发写入
4. 单事务 - 全部成功或全部失败

建议的新流程:

用户调用 restore
    ↓
1. 自动创建auto_backup (可选)
2. 获取依赖顺序
3. 锁定所有表
4. 开始事务
5. 逐表恢复 (TRUNCATE + INSERT)
6. 提交/回滚
7. 返回结果

测试建议

需要测试的场景:

1. 正常恢复流程
2. 恢复过程中服务器断电
3. 并发写入时恢复
4. 部分表恢复失败
5. 恢复后的数据完整性验证