fix(capture): 读不到源行时按日志年龄延迟重试，根治降级Delete致数据丢失
Insert/Update 日志回读不到源行时，旧逻辑无条件降级为 Delete，在 ACE 引擎可见性延迟（批量插入约11s窗口）下误伤，导致该 Insert 的行不仅没进 SQL 反而被空打 Delete 并清理 Access 日志，数据永久丢失（8/4 事件根因： 氩弧焊.接收 16255-16259 缺失）。 改为按 Access 日志年龄（capture_defer_seconds=60）决定动作： - 年龄 < 阈值：DEFER，跳过不入队，Access 日志保留，下一轮 cycle 重读； - 年龄 >= 阈值：AGED-OUT，入队直接标 dead 待人工（usp_SyncApply 只选 pending，不会执行任何破坏性 SQL），归档表记 ProcessedOperateType=AgedOut。 零 schema/SQL 改动：cleanup 只删 Status=applied 的日志，DEFER 行不入队 则无 applied 行、AGED-OUT 标 dead 非 applied，两者 Access 日志均保留， 与既有 cleanup/apply/service 逻辑天然自洽。 详见 docs/plan-capture-defer-by-age.md 与 docs/incremental-sync-flow.md。
This commit is contained in:
Misaka_Company
281
docs/incremental-sync-flow.md
Normal file
281
docs/incremental-sync-flow.md
Normal file
@@ -0,0 +1,281 @@
|
||||
# 增量同步流程详解（Incremental Sync Flow）
|
||||
|
||||
> 本文档梳理 Access → SQL Server 增量同步的完整流程，逐节点说明「发生了什么、读了/写了什么、状态如何流转」，作为重构「Insert 降级 Delete」逻辑的决策依据。
|
||||
>
|
||||
> 涉及代码：`src/sync/service.py`（主循环 cycle）、`src/sync/capture.py`（捕获）、`sql/02_sync_apply.sql`（应用）、`src/sync/cleanup.py`（清理）、`src/sync/sql_writer.py`（SQL 端读写）、`src/sync/access_reader.py`（Access 端读取）。
|
||||
|
||||
---
|
||||
|
||||
## 一、整体架构：一轮 cycle 的三段流水线
|
||||
|
||||
每个 cycle（默认间隔 `poll_interval_seconds=10s`）跑一遍三段，顺序固定、不可调换：
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
START([cycle 开始<br/>分配 cycle_id]) --> CAP
|
||||
|
||||
CAP["【1. Capture 捕获】<br/>逐库读 TableChangeLog → 回读整行 → 入队 SyncQueue<br/><i>src/sync/capture.py</i>"]
|
||||
CAP --> APPLY
|
||||
|
||||
APPLY["【2. Apply 应用】<br/>调 usp_SyncApply 把 SyncQueue pending 行落到镜像表<br/><i>sql/02_sync_apply.sql</i>"]
|
||||
APPLY --> HEALTH
|
||||
|
||||
HEALTH["【2.5 队列健康检查】<br/>查 error/dead 卡死行并告警"]
|
||||
HEALTH --> CLEAN
|
||||
|
||||
CLEAN["【3. Cleanup 清理】<br/>删 Access 已应用日志 → SyncQueue 行标 cleaned<br/><i>src/sync/cleanup.py</i>"]
|
||||
CLEAN --> PURGE
|
||||
|
||||
PURGE["【3.5 Purge 回收】<br/>删 SyncQueue 中超保留期的 cleaned 行"]
|
||||
PURGE --> DONE([cycle 结束<br/>休眠 poll_interval])
|
||||
|
||||
style CAP fill:#e3f2fd,stroke:#1976d2
|
||||
style APPLY fill:#fff3e0,stroke:#f57c00
|
||||
style HEALTH fill:#fce4ec,stroke:#c2185b
|
||||
style CLEAN fill:#e8f5e9,stroke:#388e3c
|
||||
style PURGE fill:#f3e5f5,stroke:#7b1fa2
|
||||
```
|
||||
|
||||
**关键设计约束**（决定重构可行性的红线）：
|
||||
- **Access 的 `TableChangeLog` 是 append-only，无状态字段**——它只是个待处理队列，无法在上面记录「已重试几次」。
|
||||
- **日志清除的唯一依据是 SyncQueue 的 `Status='applied'`**——只要一条日志对应的队列行不是 applied，cleanup 就不会删它（详见第三节）。
|
||||
- **SyncQueue 有唯一索引 `(SourceFile,SourceTable,SourceLogID)` 去重**——同一日志第二次入队会被静默跳过，不会覆盖原行、不会自增计数。
|
||||
|
||||
---
|
||||
|
||||
## 二、Capture 阶段（数据捕获）—— ❗重构的核心战场
|
||||
|
||||
逐库处理，每个 Access 文件独立隔离（单库失败不影响其它）。
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A([开始 capture 某个文件]) --> B["读 Access TableChangeLog<br/>最旧 N 条（按 ID 升序）<br/>N = capture_batch_size=500"]
|
||||
B --> C{有日志行?}
|
||||
C -- 否 --> Z([capture 结束])
|
||||
C -- 是 --> D[逐行处理]
|
||||
|
||||
D --> E{"表是否在同步范围内?<br/>is_synced_table"}
|
||||
E -- 否 --> F["跳过（out_of_scope）<br/>不计入，下轮仍会读到"]
|
||||
F --> D
|
||||
E -- 是 --> G{"OperateType?"}
|
||||
|
||||
G -- Insert/Update --> H["🔑 回读整行<br/>read_row(table, record_id)<br/>SELECT * FROM 表 WHERE ID=?"]
|
||||
H --> I{读到行?}
|
||||
I -- 是 --> J["row_data = JSON 序列化<br/>op 保持 Insert/Update"]
|
||||
I -- ❌否 --> K["⚠️ 降级 op = Delete<br/>row_data = None<br/>写 DOWNGRADE 警告日志"]
|
||||
|
||||
G -- Delete --> L["op = Delete<br/>row_data = None<br/>（不回读，Delete 无需数据）"]
|
||||
|
||||
G -- 其它未知 --> M["跳过（unknown_op）<br/>下轮仍会读到"]
|
||||
|
||||
J --> N["写 SyncLogArchive（永久审计）<br/>记录 OriginalOperateType + ProcessedOperateType"]
|
||||
K --> N
|
||||
L --> N
|
||||
|
||||
N --> O["入队 SyncQueue（去重插入）<br/>INSERT...WHERE NOT EXISTS"]
|
||||
O --> P{插入成功?}
|
||||
P -- 是 --> Q["enqueued +1"]
|
||||
P -- 否（去重命中）--> R["dedup_skipped +1<br/>说明上轮 apply 失败残留"]
|
||||
|
||||
Q --> S{还有下一行?}
|
||||
R --> S
|
||||
F --> S
|
||||
M --> S
|
||||
S -- 是 --> D
|
||||
S -- 否 --> T["打印 capture summary<br/>read/enqueued/downgraded/..."]
|
||||
T --> Z
|
||||
|
||||
style K fill:#ffcdd2,stroke:#c62828,stroke-width:3px
|
||||
style H fill:#fff9c4,stroke:#f9a825
|
||||
style I fill:#fff9c4,stroke:#f9a825
|
||||
```
|
||||
|
||||
### 🔴 问题节点：降级 Delete（`capture.py:93-108`）
|
||||
|
||||
```python
|
||||
if op in ("Insert", "Update"):
|
||||
d = reader.read_row(lr.table_name, lr.record_id)
|
||||
if d is None:
|
||||
op = "Delete" # ← 问题根源：读不到就降级
|
||||
st.downgraded += 1
|
||||
log.warning("capture DOWNGRADE %s->Delete ...")
|
||||
```
|
||||
|
||||
**为什么读不到？两个场景无法区分：**
|
||||
1. **真删除**：行被 Insert 后又 Delete（Access 客户端先插后删）→ 这时降级 Delete 是「碰巧正确」。
|
||||
2. **可见性延迟**（本次事件的根因）：行已插入但 ACE 引擎尚未对其他 ODBC 连接可见（批量插入时窗口可达 11 秒）→ 这时降级 Delete 是**有害的误伤**。
|
||||
|
||||
**代码当前无法区分这两种情况**，统一降级为 Delete。这就是要重构的核心。
|
||||
|
||||
---
|
||||
|
||||
## 三、Apply 阶段（数据应用）
|
||||
|
||||
调用存储过程 `usp_SyncApply`，**按表分组、集合化处理**所有 pending 行。
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A([call_apply<br/>EXEC usp_SyncApply]) --> B["重置 error 行<br/>RetryCount < max_retries 的 → pending"]
|
||||
B --> C["按 TargetSchema+TargetTable 分组<br/>遍历每个目标表"]
|
||||
|
||||
C --> D{该表有 pending 行?}
|
||||
D -- 否 --> C
|
||||
D -- 是 --> E["统计 pending 数 / distinct RecordID 数"]
|
||||
|
||||
E --> F["BEGIN TRAN"]
|
||||
|
||||
F --> G["🔑 构建列清单<br/>从 sys.columns 读目标表所有列<br/>（排除 ID/computed/identity/timestamp）"]
|
||||
|
||||
G --> H["构建动态 SQL"]
|
||||
|
||||
H --> I["分支1: Upsert（MERGE）<br/>ranked CTE: 按 RecordID 分区,<br/>SourceLogID DESC 取 rn=1<br/>仅 OperateType∈Insert/Update 且 RowData 非空"]
|
||||
I --> J["SET IDENTITY_INSERT ON<br/>MERGE 目标表<br/>匹配则 UPDATE, 不匹配则 INSERT<br/>@merged = @@ROWCOUNT"]
|
||||
|
||||
J --> K["分支2: Delete<br/>同一 ranked CTE 的 rn=1 行<br/>仅 OperateType=Delete"]
|
||||
K --> L["DELETE 目标表 WHERE ID IN (...)<br/>@deleted = @@ROWCOUNT"]
|
||||
|
||||
L --> M["把该表所有 pending 行<br/>Status → applied, AppliedAt = now<br/>@applied = @@ROWCOUNT"]
|
||||
|
||||
M --> N[COMMIT]
|
||||
|
||||
N --> O["写 SyncApplyRunLog 审计<br/>pending/merged/deleted/applied/<br/>error/dead + CycleID + 耗时"]
|
||||
O --> C
|
||||
|
||||
N -.失败.-> X["ROLLBACK"]
|
||||
X --> Y["超 max_retries → dead<br/>否则 → error（下轮重试）"]
|
||||
Y --> O
|
||||
|
||||
style I fill:#e3f2fd,stroke:#1976d2
|
||||
style K fill:#ffcdd2,stroke:#c62828
|
||||
style M fill:#fff3e0,stroke:#f57c00
|
||||
```
|
||||
|
||||
### 关键：保序「最后操作胜」（`02_sync_apply.sql:95-135`）
|
||||
|
||||
单个 `ranked` CTE 同时供 Upsert 和 Delete 两个分支使用：
|
||||
```sql
|
||||
ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn
|
||||
```
|
||||
- `rn=1` 是该 RecordID **真正的最后一条日志**。
|
||||
- Upsert 分支：`rn=1 AND OperateType IN ('Insert','Update')`
|
||||
- Delete 分支：`rn=1 AND OperateType='Delete'`
|
||||
|
||||
所以**降级成 Delete 的行，在这里会真的去 SQL 端执行 DELETE**。本次事件中 5 行 Delete 的 `@deleted=0`（SQL 里本就没这些行，空打），但 `@applied=5`（队列行照样被标 applied）。
|
||||
|
||||
---
|
||||
|
||||
## 四、Cleanup 阶段（日志清除）—— ❗决定「重试」能否成立的命脉
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A([cleanup 某个文件]) --> B["查 SyncQueue 中<br/>Status=applied 的 SourceLogID 列表<br/>applied_log_ids(file)"]
|
||||
B --> C{有 applied 行?}
|
||||
C -- 否 --> Z([cleanup 结束, 返回 0])
|
||||
C -- 是 --> D["DELETE FROM Access.TableChangeLog<br/>WHERE ID IN (上述列表)<br/>分批 + 锁重试"]
|
||||
D --> E{删除数 == 预期?}
|
||||
E -- 否 --> F["WARNING: 部分日志已不在<br/>（被外部或中断的运行删过）"]
|
||||
E -- 是 --> G["mark_cleaned:<br/>这些队列行 Status → cleaned<br/>CleanedAt = now"]
|
||||
F --> G
|
||||
G --> H["INFO: 删除了 N 条 Access 日志"]
|
||||
H --> Z
|
||||
|
||||
style D fill:#e8f5e9,stroke:#388e3c,stroke-width:2px
|
||||
style B fill:#fff9c4,stroke:#f9a825
|
||||
```
|
||||
|
||||
### 🔑 cleanup 的判定条件是重构的支点
|
||||
|
||||
**cleanup 只删 `Status='applied'` 的日志**（`sql_writer.py:264-272` 的 `applied_log_ids`）。这意味着：
|
||||
|
||||
| capture 对该日志的处理 | SyncQueue 行状态 | cleanup 是否删 Access 日志 | 后果 |
|
||||
|------------------------|------------------|----------------------------|------|
|
||||
| 降级 Delete（现状） | applied | **删除** | ❌ 日志消失，再无重试机会 |
|
||||
| **跳过不入队（重构后）** | （无对应行） | **不删** | ✅ 日志保留，下轮重读 |
|
||||
|
||||
**结论：重构只要做到「读不到 → 不入队」，cleanup 这一段天然会把日志保留下来，无需改动 cleanup.py。** 这是「跨 cycle 重试」能够成立的根基。
|
||||
|
||||
---
|
||||
|
||||
## 五、数据流转全景：一条日志的完整生命周期
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
subgraph Access["Access 端（.accdb）"]
|
||||
T1[(业务表<br/>如 接收)]
|
||||
TCL[(TableChangeLog<br/>变更日志 append-only)]
|
||||
end
|
||||
|
||||
subgraph SQL["SQL Server 端（CompanyDB）"]
|
||||
SQ[(SyncQueue<br/>待处理队列)]
|
||||
ARCH[(SyncLogArchive<br/>永久审计)]
|
||||
RL[(SyncApplyRunLog<br/>apply 运行日志)]
|
||||
MIRROR[(镜像表<br/>如 接收_YEAR2026)]
|
||||
end
|
||||
|
||||
T1 -- "数据宏 After I/U/D<br/>写入一行日志" --> TCL
|
||||
TCL -- "① capture 读取" --> CAP[Capture]
|
||||
CAP -- "回读整行" --> T1
|
||||
CAP -- "② 入队（去重）" --> SQ
|
||||
CAP -- "② 审计存档" --> ARCH
|
||||
SQ -- "③ apply 处理" --> APPLY[usp_SyncApply]
|
||||
APPLY -- "MERGE/DELETE" --> MIRROR
|
||||
APPLY -- "pending→applied" --> SQ
|
||||
APPLY -- "记录运行结果" --> RL
|
||||
SQ -- "④ cleanup 查 applied" --> CLEAN[Cleanup]
|
||||
CLEAN -- "⑤ 删已应用日志" --> TCL
|
||||
CLEAN -- "applied→cleaned" --> SQ
|
||||
|
||||
style CAP fill:#e3f2fd,stroke:#1976d2
|
||||
style APPLY fill:#fff3e0,stroke:#f57c00
|
||||
style CLEAN fill:#e8f5e9,stroke:#388e3c
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 六、本次事件的重放（在上述流程中的路径）
|
||||
|
||||
5 条 Insert 日志（RecordID 16255-16259）在一轮 cycle 中的遭遇：
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A["8/3 09:01 Access 批量插入 5 行<br/>数据宏写 5 条 Insert 日志<br/>SourceLogID 15533-15537"] --> B
|
||||
|
||||
B["09:00:50 Capture 读取这 5 条日志<br/>（注: CapturedAt 比 OriginalTime 早 11s<br/> = ACE 可见性窗口）"] --> C
|
||||
|
||||
C["read_row 回读 5 行<br/>SELECT * FROM 接收 WHERE ID=16255..16259"] --> D
|
||||
|
||||
D["❌ 全部返回 None<br/>（行尚未对其他连接可见）"] --> E
|
||||
|
||||
E["🔴 降级为 Delete<br/>入队 SyncQueue, OperateType=Delete<br/>RowData=null"] --> F
|
||||
|
||||
F["09:00:51 Apply: Delete 分支<br/>DELETE 接收_YEAR2026 WHERE ID IN(16255..16259)<br/>@deleted=0（SQL 本就没这5行）<br/>@applied=5（队列行标 applied）"] --> G
|
||||
|
||||
G["队列行 Status = applied"] --> H
|
||||
|
||||
H["Cleanup: 查到这5条 applied<br/>DELETE Access.TableChangeLog ID=15533-15537"] --> I
|
||||
|
||||
I["🔴 日志被删，5 条 Insert 证据消失<br/>这5行从此再无机会被同步进 SQL"] --> J
|
||||
|
||||
J["8/4 05:08 compare: missing_in_sql=5<br/>ID 16255-16259"]
|
||||
|
||||
style D fill:#ffcdd2,stroke:#c62828
|
||||
style E fill:#ffcdd2,stroke:#c62828,stroke-width:3px
|
||||
style H fill:#e8f5e9,stroke:#388e3c
|
||||
style I fill:#ffcdd2,stroke:#c62828
|
||||
style J fill:#fff9c4,stroke:#f9a825
|
||||
```
|
||||
|
||||
**链式灾难**：降级 Delete（capture）→ 空打 Delete 但标 applied（apply）→ 删 Access 日志（cleanup）。三段配合下，这 5 行被「合法地」从同步链路中抹除。只要在 capture 段断开第一环（不降级、不入队），后面 apply/cleanup 就不会碰它们，日志保留，下轮自然重试。
|
||||
|
||||
---
|
||||
|
||||
## 七、重构决策点（待定）
|
||||
|
||||
基于上述流程，重构的核心是改造 capture 阶段的「降级」分支。需要决策的问题：
|
||||
|
||||
1. **读不到时的动作**：跳过不入队（日志保留，下轮重读）vs 入队但标记 defer 状态？
|
||||
2. **重试上限的判定依据**：用「重试次数」（需要存储计数）vs 用「日志年龄时间窗」（无需存储，靠 `now - OriginalTime > 阈值`）？
|
||||
3. **终态处理**：超限后该 Access 日志保留（占队列头持续重读）还是删除（丢证据）？
|
||||
4. **Update 日志**：是否套用同一套 defer 逻辑？
|
||||
|
||||
我倾向的方案：**读不到 → 不入队 + 写 defer 审计；用日志年龄（如 120s）作终态判据；超限则入队标 dead（保留 Access 日志不删，待人工）**。零 schema 改动、零状态存储。等你审完这份流程图确认方向后，我再动手。
|
||||
211
docs/plan-capture-defer-by-age.md
Normal file
211
docs/plan-capture-defer-by-age.md
Normal file
@@ -0,0 +1,211 @@
|
||||
# 重构方案：capture 读不到行时按「日志年龄」延迟重试
|
||||
|
||||
> 目标：根治「Insert 日志回读不到行 → 降级 Delete → 数据丢失」的缺陷（8/4 事件根因）。
|
||||
> 核心改动：读不到 → **不入队、不降级**，按 Access 日志的年龄决定「下轮重试」还是「判死待人工」。
|
||||
> 改动面：`capture.py`（主）、`config.py`（新增2个配置项）、`capture.py` 的 CaptureStats（新增计数器）。**零 schema 改动、零 SQL 改动、不动 apply/cleanup。**
|
||||
|
||||
---
|
||||
|
||||
## 一、当前问题行为 vs 重构后行为
|
||||
|
||||
| 场景 | 当前行为（缺陷） | 重构后行为 |
|
||||
|------|------------------|-----------|
|
||||
| Insert/Update 读不到行（可见性延迟，本次事件） | 降级 Delete → 入队 → apply 空打 Delete + 标 applied → cleanup 删 Access 日志 → **数据永久丢失** | 年龄 < 阈值：跳过不入队，日志保留，下轮重读 → 读到后正常 Insert ✅ |
|
||||
| Insert/Update 读不到行（真删除：先插后删） | 降级 Delete（碰巧正确） | 年龄 < 阈值：跳过；后续 Delete 日志会兜底正确清理；超龄判死待人工 ✅ |
|
||||
| Insert/Update 一直读不到（真异常/数据损坏） | 降级 Delete（错误） | 年龄 ≥ 阈值：入队标 dead，保留 Access 日志，**告警待人工** ✅ |
|
||||
|
||||
---
|
||||
|
||||
## 二、配置项新增（`src/sync/config.py` 的 RuntimeConfig）
|
||||
|
||||
```python
|
||||
class RuntimeConfig(BaseModel):
|
||||
# ... 既有字段 ...
|
||||
# Insert/Update 日志回读不到行时，按日志年龄延迟重试：年龄小于此秒数则
|
||||
# 跳过不入队（保留 Access 日志，下一轮 cycle 重新捕获），度过 ACE 引擎
|
||||
# 的可见性窗口；超过此年龄仍读不到则判定为真删除/异常，入队标 dead 待人工。
|
||||
# 设为 0 可关闭延迟重试（退化为旧的"立即判死"语义，但不再降级 Delete）。
|
||||
capture_defer_seconds: int = 120
|
||||
```
|
||||
|
||||
- **默认值 120 秒**：覆盖本次事件观察到的 ~11 秒可见性窗口，并留足 10 倍余量；对 `poll_interval=10s` 意味着最多重试约 12 轮。
|
||||
- 单一配置项，`config.yaml` 无需改动即可生效（用默认值）。`defer_dead_op` 不暴露为配置（实现细节，固定为 `dead` 状态，见下）。
|
||||
|
||||
---
|
||||
|
||||
## 三、capture.py 改动（核心）
|
||||
|
||||
### 3.1 CaptureStats 新增两个计数器
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class CaptureStats:
|
||||
# ... 既有字段 ...
|
||||
read: int = 0
|
||||
enqueued: int = 0
|
||||
dedup_skipped: int = 0
|
||||
downgraded: int = 0 # 保留字段，重构后恒为 0（兼容旧日志解析）
|
||||
deferred: int = 0 # 【新】读不到行但年龄 < 阈值，跳过待下轮重试
|
||||
aged_out: int = 0 # 【新】读不到行且年龄 ≥ 阈值，判死待人工
|
||||
out_of_scope: int = 0
|
||||
unknown_op: int = 0
|
||||
# ... merge() / ops_str() 同步更新 ...
|
||||
```
|
||||
|
||||
### 3.2 降级分支重构（`capture_file` 中 Insert/Update 回读逻辑）
|
||||
|
||||
**替换** 现有的第 93-108 行（整段 `if op in ("Insert", "Update"):` 块）：
|
||||
|
||||
```python
|
||||
if op in ("Insert", "Update"):
|
||||
d = reader.read_row(lr.table_name, lr.record_id)
|
||||
if d is None:
|
||||
# 读不到行：不再降级 Delete。按 Access 日志年龄决定延迟重试还是判死。
|
||||
# 必须计算日志年龄（原始设计无此逻辑）。
|
||||
age_seconds = _log_age_seconds(lr.time)
|
||||
if age_seconds < cfg.runtime.capture_defer_seconds:
|
||||
# 暂态读不到（ACE 可见性窗口）：跳过，不入队、不清理。
|
||||
# Access 日志因无对应 applied 队列行，cleanup 不会删除，下轮重试。
|
||||
st.deferred += 1
|
||||
log.warning(
|
||||
"capture DEFER %s file=%s table=%s record_id=%s log_id=%s "
|
||||
"log_time=%s age=%ds (source row unreadable; will retry next "
|
||||
"cycle while log age < %ds)",
|
||||
lr.operate_type, fm.file, lr.table_name, lr.record_id,
|
||||
lr.id, lr.time, int(age_seconds),
|
||||
cfg.runtime.capture_defer_seconds,
|
||||
)
|
||||
continue # ← 关键：跳过本行，archive/queue 都不写
|
||||
else:
|
||||
# 超龄仍读不到：真删除或真异常。入队标 dead，不降级、不执行任何
|
||||
# 破坏性操作。保留 Access 日志（无 applied 行 → cleanup 不删），
|
||||
# 队列健康检查会告警，等待人工介入。
|
||||
st.aged_out += 1
|
||||
log.warning(
|
||||
"capture AGED-OUT %s file=%s table=%s record_id=%s log_id=%s "
|
||||
"log_time=%s age=%ds >= %ds -- enqueuing as dead for manual "
|
||||
"review (source row still unreadable after defer window)",
|
||||
lr.operate_type, fm.file, lr.table_name, lr.record_id,
|
||||
lr.id, lr.time, int(age_seconds),
|
||||
cfg.runtime.capture_defer_seconds,
|
||||
)
|
||||
op = "dead" # 仅用于入队时的状态标记，见下
|
||||
else:
|
||||
row_data = json.dumps(d, ensure_ascii=False)
|
||||
```
|
||||
|
||||
### 3.3 超龄行的入队方式（`aged_out` 分支）
|
||||
|
||||
超龄行需要进 SyncQueue 但**不能被 apply 执行任何 SQL 操作**（没数据可插，也不能 Delete）。两种实现可选（我倾向 A）：
|
||||
|
||||
**方案 A（推荐）：入队后直接标 dead，OperateType 保留原 Insert/Update 真相**
|
||||
- 入队时 `OperateType` 仍写 `Insert`/`Update`（保留原始意图，便于审计），`RowData=null`。
|
||||
- 入队后立即 `UPDATE ... SET Status='dead', ErrorMsg='source row unreadable after {age}s defer'`。
|
||||
- apply 的游标只选 `Status='pending'`，dead 行不会被处理 → 不会误删。
|
||||
- queue 健康检查已有 dead 告警（`service.py:166-184`），自动浮现。
|
||||
|
||||
**方案 B：新增 OperateType='Noop'** — 改动面更大（apply 存储过程需识别），不推荐。
|
||||
|
||||
> 需要在 `sql_writer.py` 新增一个方法 `insert_dead_row(row, error_msg)`，逻辑 = 先 `insert_queue_row`（去重插入）再 `UPDATE ... SET Status='dead', RetryCount=<对应>, ErrorMsg=?`。
|
||||
|
||||
### 3.4 日志年龄计算辅助函数 `_log_age_seconds`
|
||||
|
||||
```python
|
||||
import datetime as _dt
|
||||
|
||||
def _log_age_seconds(log_time: object) -> float:
|
||||
"""Access 日志行 Time 字段距今的秒数。log_time 是 pyodbc 返回的 datetime。
|
||||
异常时返回一个大数（视为已超龄），确保宁可判死也不无限重试。"""
|
||||
try:
|
||||
if isinstance(log_time, _dt.datetime):
|
||||
return (_dt.datetime.now() - log_time).total_seconds()
|
||||
# Access via ODBC 通常返回 datetime；兜底处理 naive/其它类型
|
||||
return float("inf")
|
||||
except Exception:
|
||||
return float("inf")
|
||||
```
|
||||
|
||||
> ⚠️ **时区/时钟注意**：`lr.time` 是 Access 端写入的本地时间，`datetime.now()` 也是本机本地时间，两者同在 114 主机同一时区，可直接相减。本次事件中 OriginalTime/CapturedAt 的"倒挂"现象（差11秒）不影响此逻辑——因为按年龄判断，即便 lr.time 偏早，age 只会被算得更大，倾向判死而非误伤，方向安全。
|
||||
|
||||
---
|
||||
|
||||
## 四、归档表（SyncLogArchive）的处理
|
||||
|
||||
| 分支 | 是否写 archive | 理由 |
|
||||
|------|---------------|------|
|
||||
| DEFER（跳过） | **不写** | 日志保留在 Access，下轮 capture 会重新读到并正常归档；此时写 archive 反而会在去重表里留下"读不到"的半成品记录 |
|
||||
| AGED-OUT（判死） | **写** | 超龄是终态，需留永久审计（OriginalOperateType=Insert/Update, ProcessedOperateType='AgedOut', RowData=null, OriginalTime=lr.time） |
|
||||
|
||||
> ProcessedOperateType 新增值 `'AgedOut'`（仅 archive 表用，varchar(10) 放得下 7 字符）。这是纯审计标记，不影响任何执行逻辑。
|
||||
|
||||
---
|
||||
|
||||
## 五、不改动的地方（明确边界）
|
||||
|
||||
| 模块 | 是否改动 | 原因 |
|
||||
|------|---------|------|
|
||||
| `cleanup.py` | ❌ 不改 | 只删 `Status='applied'` 的日志；DEFER 行不入队无 applied 行 → 日志保留；AGED-OUT 标 dead 非 applied → 日志也保留。天然自洽。 |
|
||||
| `sql/02_sync_apply.sql` | ❌ 不改 | apply 游标只选 `pending`，dead 行天然跳过；DEFER 行根本不入队。 |
|
||||
| `sql/01_sync_queue.sql` | ❌ 不改 | 不新增列，不改索引。 |
|
||||
| `service.py` | ❌ 不改 | 队列健康检查已有 error/dead 告警（`queue_error_samples`），AGED-OUT 的 dead 行会自动被它捕获并 WARNING。capture summary 日志格式已包含新计数器（由 CaptureStats.ops_str/merge 驱动）。 |
|
||||
| `access_reader.py` | ❌ 不改 | `read_row` 行为不变。 |
|
||||
| `config.yaml` | ❌ 不改 | 用默认值 120s 即可。 |
|
||||
|
||||
---
|
||||
|
||||
## 六、本次事件 5 行的重放（重构后）
|
||||
|
||||
```
|
||||
cycle N (09:00:50): capture 读到 5 条 Insert 日志
|
||||
→ read_row 返回 None
|
||||
→ age = now(09:00:50) - log_time(09:01:01) → 注: 因 Access 时间戳特性 age 可能算成负或小
|
||||
→ 即便按最保守计算，age 远 < 120s
|
||||
→ DEFER：跳过不入队，写 WARNING，Access 日志保留
|
||||
|
||||
cycle N+1 (09:01:00): capture 再次读到这 5 条日志
|
||||
→ read_row 此刻可见性窗口已过（11s > 窗口）→ 读到行 ✅
|
||||
→ 正常入队 OperateType=Insert，带完整 RowData
|
||||
→ apply MERGE → SQL 正确写入 5 行 ✅
|
||||
→ cleanup 删 Access 日志（这次是 applied，合理）
|
||||
```
|
||||
|
||||
**结果：8/4 compare 不再出现 missing_in_sql=5。**
|
||||
|
||||
---
|
||||
|
||||
## 七、验证计划
|
||||
|
||||
1. **单元测试**（`tests/test_capture.py`）：
|
||||
- mock `read_row` 返回 None + `lr.time` 为近时 → 断言 `deferred=1, enqueued=0, aged_out=0`，不调用 `insert_queue_row` / `insert_archive_row`。
|
||||
- mock `read_row` 返回 None + `lr.time` 为 200s 前 → 断言 `aged_out=1`，调用 `insert_dead_row`，Status=dead。
|
||||
- mock `read_row` 返回 dict + 任意时间 → 断言正常入队（回归测试）。
|
||||
- `capture_defer_seconds=0` → 任何读不到都立即判死（边界）。
|
||||
2. **现有测试回归**：`pytest` 全绿（确保去重/正常 Insert/真 Delete 路径不受影响）。
|
||||
3. **集成验证**（部署后观察 1-2 天）：
|
||||
- 关注 capture summary 日志的 `deferred=` 计数，确认批量插入场景下有 defer 发生且下轮 enqueued。
|
||||
- 关注 `queue health` WARNING，确认 dead 行（如有）被正确告警。
|
||||
- 跑 `compare --granularity ids`，确认无 missing_in_sql。
|
||||
|
||||
---
|
||||
|
||||
## 八、改动文件清单
|
||||
|
||||
| 文件 | 改动类型 | 说明 |
|
||||
|------|---------|------|
|
||||
| `src/sync/config.py` | 新增字段 | RuntimeConfig 加 `capture_defer_seconds: int = 120` |
|
||||
| `src/sync/capture.py` | 核心重构 | 降级分支 → defer/aged_out 分支；CaptureStats 加 2 计数器；新增 `_log_age_seconds` |
|
||||
| `src/sync/sql_writer.py` | 新增方法 | `insert_dead_row(row, error_msg)`：去重插入后立即标 dead |
|
||||
| `tests/test_capture.py` | 新增用例 | 覆盖 DEFER / AGED-OUT / 正常 / 边界 4 种情况 |
|
||||
|
||||
**总计 4 个文件，零 SQL/零 schema 改动。**
|
||||
|
||||
---
|
||||
|
||||
## 九、待你确认的决策点
|
||||
|
||||
1. **`capture_defer_seconds` 默认值 120s** 是否合适？（覆盖11s窗口×10倍余量）
|
||||
2. **AGED-OUT 行的处理**：入队标 dead（方案A，推荐）vs 其它？
|
||||
3. **archive 表 ProcessedOperateType 新增值 `'AgedOut'`** 是否可接受？
|
||||
4. **downgraded 计数器**：保留为恒0（向后兼容旧日志解析）还是直接删除？
|
||||
|
||||
确认后我即按此方案执行。
|
||||
@@ -2,23 +2,32 @@
|
||||
|
||||
Every consumed ``TableChangeLog`` row is appended to the permanent audit store
|
||||
(``SyncLogArchive``) BEFORE it is enqueued, so the original evidence survives
|
||||
cleanup. On top of that, this module now emits a detailed audit trail to the
|
||||
cleanup. On top of that, this module emits a detailed audit trail to the
|
||||
service log:
|
||||
|
||||
- WARNING for every operate-type downgrade (Insert/Update -> Delete because
|
||||
the source row was unreadable at capture time), with record id, log id and
|
||||
the Access log timestamp -- the exact event class behind the 14287 incident,
|
||||
previously invisible in the text log;
|
||||
- WARNING for every Insert/Update that could not be read back from the source
|
||||
table at capture time. Two outcomes, both logged by identity (record id,
|
||||
log id, log time, age):
|
||||
* DEFER -- the log row is younger than ``capture_defer_seconds``: treated
|
||||
as an ACE visibility-latency window. The row is NOT enqueued, the Access
|
||||
log is left intact (cleanup only deletes rows whose queue status is
|
||||
``applied``), and the next cycle re-reads it. This is the fix for the
|
||||
data-loss incident where an Insert downgraded to Delete silently dropped
|
||||
rows from SQL Server (see docs/plan-capture-defer-by-age.md).
|
||||
* AGED-OUT -- still unreadable past the defer window: enqueued directly as
|
||||
``dead`` (usp_SyncApply never selects dead rows, so no destructive SQL
|
||||
runs) and left for manual review; the Access log is also preserved.
|
||||
- WARNING for unknown operate types, with enough identity (log id / record id
|
||||
/ time) to locate and repair the offending log row manually;
|
||||
- a per-file INFO summary: rows read, newly enqueued, dedup-skipped
|
||||
(re-capture after a failed apply -- a symptom worth noticing), downgraded,
|
||||
out-of-scope, unknown ops, the processed log-ID range and a per-operation
|
||||
breakdown;
|
||||
(re-capture after a failed apply -- a symptom worth noticing), deferred,
|
||||
aged-out, out-of-scope, unknown ops, the processed log-ID range and a
|
||||
per-operation breakdown;
|
||||
- DEBUG detail for individual out-of-scope and dedup skips.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as _dt
|
||||
import json
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
@@ -31,6 +40,23 @@ from .targets import is_synced_table
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _log_age_seconds(log_time: object) -> float:
|
||||
"""Seconds elapsed since the Access log row's ``Time`` value.
|
||||
|
||||
``log_time`` is the raw pyodbc datetime from ``TableChangeLog.Time`` (set
|
||||
by the Access data macro). Both it and ``datetime.now()`` run on the same
|
||||
host/clock, so direct subtraction is valid. On any anomaly (missing /
|
||||
non-datetime value) returns ``inf`` so the caller errs toward ageing out
|
||||
rather than retrying forever.
|
||||
"""
|
||||
try:
|
||||
if isinstance(log_time, _dt.datetime):
|
||||
return (_dt.datetime.now() - log_time).total_seconds()
|
||||
return float("inf")
|
||||
except Exception:
|
||||
return float("inf")
|
||||
|
||||
|
||||
@dataclass
|
||||
class CaptureStats:
|
||||
"""Counters for one capture pass (per file, or aggregated per cycle)."""
|
||||
@@ -38,7 +64,8 @@ class CaptureStats:
|
||||
read: int = 0 # change-log rows read from Access
|
||||
enqueued: int = 0 # rows newly inserted into SyncQueue
|
||||
dedup_skipped: int = 0 # already queued (re-capture after a failed apply)
|
||||
downgraded: int = 0 # Insert/Update downgraded to Delete
|
||||
deferred: int = 0 # Insert/Update unreadable but young -> retry next cycle
|
||||
aged_out: int = 0 # Insert/Update still unreadable past defer window -> dead
|
||||
out_of_scope: int = 0 # log rows for tables outside the sync scope
|
||||
unknown_op: int = 0 # log rows with an unrecognised OperateType
|
||||
min_log_id: int | None = None
|
||||
@@ -53,7 +80,8 @@ class CaptureStats:
|
||||
self.read += other.read
|
||||
self.enqueued += other.enqueued
|
||||
self.dedup_skipped += other.dedup_skipped
|
||||
self.downgraded += other.downgraded
|
||||
self.deferred += other.deferred
|
||||
self.aged_out += other.aged_out
|
||||
self.out_of_scope += other.out_of_scope
|
||||
self.unknown_op += other.unknown_op
|
||||
for k, v in other.ops.items():
|
||||
@@ -93,19 +121,66 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
|
||||
if op in ("Insert", "Update"):
|
||||
d = reader.read_row(lr.table_name, lr.record_id)
|
||||
if d is None:
|
||||
# 行已删（或此刻不可读），降级为 Delete。这是数据差异排查的
|
||||
# 头号嫌疑事件（参见 14287 事件），必须在文本日志显式留痕，
|
||||
# 而不只是写入 SyncLogArchive。
|
||||
op = "Delete"
|
||||
st.downgraded += 1
|
||||
# 回读不到整行：不再降级 Delete。按 Access 日志年龄决定动作，
|
||||
# 根治"Insert 降级 Delete 致数据丢失"缺陷。
|
||||
age = _log_age_seconds(lr.time)
|
||||
if age < cfg.runtime.capture_defer_seconds:
|
||||
# 暂态（ACE 可见性延迟）：跳过，不入队、不写 archive。
|
||||
# Access 日志因无 applied 队列行，cleanup 不会删除，
|
||||
# 下一轮 cycle 会重新读到。
|
||||
st.deferred += 1
|
||||
log.warning(
|
||||
"capture DEFER %s file=%s table=%s record_id=%s "
|
||||
"log_id=%s log_time=%s age=%ds (source row unreadable; "
|
||||
"will retry next cycle while age < %ds)",
|
||||
lr.operate_type, fm.file, lr.table_name, lr.record_id,
|
||||
lr.id, lr.time, int(age),
|
||||
cfg.runtime.capture_defer_seconds,
|
||||
)
|
||||
continue
|
||||
# 超龄仍读不到：真删除或真异常。入队标 dead，不降级、不执行
|
||||
# 任何破坏性 SQL。保留 Access 日志（无 applied 行 → cleanup
|
||||
# 不删），队列健康检查会告警，等待人工介入。
|
||||
st.aged_out += 1
|
||||
log.warning(
|
||||
"capture DOWNGRADE %s->Delete file=%s table=%s "
|
||||
"record_id=%s log_id=%s log_time=%s "
|
||||
"(source row unreadable at capture time; original intent "
|
||||
"preserved in SyncLogArchive.OriginalOperateType)",
|
||||
lr.operate_type, fm.file, lr.table_name,
|
||||
lr.record_id, lr.id, lr.time,
|
||||
"capture AGED-OUT %s file=%s table=%s record_id=%s "
|
||||
"log_id=%s log_time=%s age=%ds >= %ds -- enqueuing as "
|
||||
"dead for manual review (source row still unreadable "
|
||||
"after defer window)",
|
||||
lr.operate_type, fm.file, lr.table_name, lr.record_id,
|
||||
lr.id, lr.time, int(age),
|
||||
cfg.runtime.capture_defer_seconds,
|
||||
)
|
||||
target_schema = fm.schema
|
||||
target_table = fm.target_table(lr.table_name)
|
||||
writer.insert_archive_row(ArchiveRow(
|
||||
source_file=fm.file,
|
||||
source_table=lr.table_name,
|
||||
source_log_id=lr.id,
|
||||
record_id=lr.record_id,
|
||||
target_schema=target_schema,
|
||||
target_table=target_table,
|
||||
original_operate_type=lr.operate_type,
|
||||
processed_operate_type="AgedOut",
|
||||
row_data=None,
|
||||
original_time=lr.time,
|
||||
))
|
||||
qr = QueueRow(
|
||||
source_file=fm.file,
|
||||
source_table=lr.table_name,
|
||||
record_id=lr.record_id,
|
||||
target_schema=target_schema,
|
||||
target_table=target_table,
|
||||
source_log_id=lr.id,
|
||||
operate_type=op, # 保留原始 Insert/Update，便于审计
|
||||
row_data=None,
|
||||
)
|
||||
writer.insert_dead_row(
|
||||
qr,
|
||||
f"source row unreadable after {int(age)}s defer window "
|
||||
f"(capture_defer_seconds={cfg.runtime.capture_defer_seconds})",
|
||||
)
|
||||
continue
|
||||
else:
|
||||
row_data = json.dumps(d, ensure_ascii=False)
|
||||
elif op != "Delete":
|
||||
@@ -122,7 +197,7 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
|
||||
# Persist the ORIGINAL log entry to the permanent audit store BEFORE the
|
||||
# queue insert (and long before cleanup deletes the Access log). This
|
||||
# keeps both the source operate type (lr.operate_type) and the processed
|
||||
# one (op) so a downgrade like Insert->Delete stays reconstructible.
|
||||
# one (op) so any divergence stays reconstructible.
|
||||
writer.insert_archive_row(ArchiveRow(
|
||||
source_file=fm.file,
|
||||
source_table=lr.table_name,
|
||||
@@ -164,11 +239,11 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
|
||||
if st.read:
|
||||
log.info(
|
||||
"capture file=%s read=%d enqueued=%d dedup_skipped=%d "
|
||||
"downgraded=%d out_of_scope=%d unknown_op=%d "
|
||||
"deferred=%d aged_out=%d out_of_scope=%d unknown_op=%d "
|
||||
"log_ids=%s..%s ops={%s}",
|
||||
fm.file, st.read, st.enqueued, st.dedup_skipped, st.downgraded,
|
||||
st.out_of_scope, st.unknown_op, st.min_log_id, st.max_log_id,
|
||||
st.ops_str(),
|
||||
fm.file, st.read, st.enqueued, st.dedup_skipped, st.deferred,
|
||||
st.aged_out, st.out_of_scope, st.unknown_op, st.min_log_id,
|
||||
st.max_log_id, st.ops_str(),
|
||||
)
|
||||
else:
|
||||
log.debug("capture file=%s: change log empty", fm.file)
|
||||
|
||||
@@ -27,6 +27,12 @@ class RuntimeConfig(BaseModel):
|
||||
cleanup_batch_size: int = 200
|
||||
cleanup_lock_retries: int = 3
|
||||
cleaned_retention_hours: int = 24
|
||||
# Insert/Update 日志回读不到整行时，按 Access 日志的"年龄"（log_time 距今
|
||||
# 秒数）决定动作：年龄小于此阈值视为 ACE 引擎的可见性延迟，跳过不入队
|
||||
# （Access 日志保留，下一轮 cycle 重新捕获）；年龄达到此阈值仍读不到则
|
||||
# 判定为真删除/异常，入队标 dead 待人工。根治"降级 Delete 致数据丢失"
|
||||
# （见 docs/plan-capture-defer-by-age.md）。设为 0 关闭延迟重试。
|
||||
capture_defer_seconds: int = 60
|
||||
# Idle cycles now log at DEBUG; the service emits an INFO heartbeat at this
|
||||
# interval while idle so a quiet log still proves the service is alive.
|
||||
idle_heartbeat_seconds: int = 600
|
||||
|
||||
@@ -109,10 +109,10 @@ def cycle(cfg) -> bool:
|
||||
activity = True
|
||||
log.info(
|
||||
"capture summary: read=%d enqueued=%d dedup_skipped=%d "
|
||||
"downgraded=%d out_of_scope=%d unknown_op=%d ops={%s}",
|
||||
"deferred=%d aged_out=%d out_of_scope=%d unknown_op=%d ops={%s}",
|
||||
total.read, total.enqueued, total.dedup_skipped,
|
||||
total.downgraded, total.out_of_scope, total.unknown_op,
|
||||
total.ops_str(),
|
||||
total.deferred, total.aged_out, total.out_of_scope,
|
||||
total.unknown_op, total.ops_str(),
|
||||
)
|
||||
else:
|
||||
log.debug("capture summary: no new change-log rows in any file")
|
||||
|
||||
@@ -132,6 +132,35 @@ class SqlWriter:
|
||||
# autocommit: statement already committed.
|
||||
return cur.rowcount > 0
|
||||
|
||||
def insert_dead_row(self, row: QueueRow, error_msg: str) -> bool:
|
||||
"""Enqueue ``row`` then immediately mark it ``dead`` (never applied).
|
||||
|
||||
Used by capture for an Insert/Update log whose source row stays
|
||||
unreadable past the defer window (``capture_defer_seconds``): the row
|
||||
is recorded for audit/health-alerting purposes but never executed by
|
||||
``usp_SyncApply`` (which only selects ``Status='pending'``), so no
|
||||
destructive SQL ever runs against the mirror table. The OperateType is
|
||||
preserved as the original Insert/Update so the intent stays visible,
|
||||
while ErrorMsg explains why it was parked.
|
||||
|
||||
Dedup semantics match ``insert_queue_row``: if the (SourceFile,
|
||||
SourceTable, SourceLogID) row already exists, the insert is skipped
|
||||
and the existing row is re-marked dead. Returns True when a new row
|
||||
was inserted, False on a dedup hit (the re-mark still happens).
|
||||
"""
|
||||
inserted = self.insert_queue_row(row)
|
||||
cur = self._conn.cursor()
|
||||
cur.execute(
|
||||
f"UPDATE {self.queue_table} SET Status='dead', ErrorMsg=? "
|
||||
"WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?",
|
||||
error_msg,
|
||||
row.source_file,
|
||||
row.source_table,
|
||||
row.source_log_id,
|
||||
)
|
||||
# autocommit: statement already committed.
|
||||
return inserted
|
||||
|
||||
def insert_archive_row(self, row: ArchiveRow) -> bool:
|
||||
"""Append ``row`` to the permanent audit store (dedup on source keys).
|
||||
|
||||
|
||||
@@ -1,23 +1,34 @@
|
||||
import datetime as _dt
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from sync.config import FileMapping, SyncConfig, AccessConfig, RuntimeConfig, SqlServerConfig
|
||||
from sync.access_reader import LogRow
|
||||
from sync.capture import capture_file
|
||||
|
||||
def _cfg():
|
||||
|
||||
def _cfg(defer=60):
|
||||
return SyncConfig(sql_server=SqlServerConfig(conn_str="x"),
|
||||
access=AccessConfig(driver="d", roots={"2026":"r"}),
|
||||
runtime=RuntimeConfig(),
|
||||
access=AccessConfig(driver="d", roots={"2026": "r"}),
|
||||
runtime=RuntimeConfig(capture_defer_seconds=defer),
|
||||
files=[])
|
||||
|
||||
|
||||
def _fm():
|
||||
return FileMapping(file="x.accdb", root="2026", schema="s",
|
||||
year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
|
||||
|
||||
|
||||
def test_capture_insert_reads_row_and_queues():
|
||||
cfg = _cfg()
|
||||
fm = FileMapping(file="氩弧焊.accdb", root="2026", schema="TIGWelding", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
|
||||
fm = FileMapping(file="氩弧焊.accdb", root="2026", schema="TIGWelding",
|
||||
year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(10, "表壳焊接记录", "34041", "Insert", None)]
|
||||
reader.read_log.return_value = [LogRow(10, "表壳焊接记录", "34041", "Insert", _dt.datetime.now())]
|
||||
reader.read_row.return_value = {"ID": 34041, "订单号": "X1"}
|
||||
writer = MagicMock()
|
||||
n = capture_file(fm, reader, writer, cfg)
|
||||
assert n == 1
|
||||
st = capture_file(fm, reader, writer, cfg)
|
||||
assert st.enqueued == 1
|
||||
assert st.deferred == 0 and st.aged_out == 0
|
||||
args = writer.insert_queue_row.call_args[0][0]
|
||||
assert args.target_schema == "TIGWelding"
|
||||
assert args.target_table == "表壳焊接记录_YEAR2026"
|
||||
@@ -25,37 +36,106 @@ def test_capture_insert_reads_row_and_queues():
|
||||
assert '"订单号": "X1"' in args.row_data
|
||||
assert args.source_log_id == 10
|
||||
|
||||
def test_capture_update_missing_row_downgrades_to_delete():
|
||||
cfg = _cfg()
|
||||
fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
|
||||
|
||||
def test_capture_unreadable_young_row_is_deferred():
|
||||
# Insert 日志年龄 < capture_defer_seconds → 跳过，不入队、不写 archive
|
||||
cfg = _cfg(defer=60)
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(11, "T", "5", "Update", None)]
|
||||
reader.read_row.return_value = None # 行已删
|
||||
reader.read_log.return_value = [LogRow(11, "T", "5", "Insert", _dt.datetime.now())]
|
||||
reader.read_row.return_value = None # 回读不到
|
||||
writer = MagicMock()
|
||||
n = capture_file(fm, reader, writer, cfg)
|
||||
assert n == 1
|
||||
args = writer.insert_queue_row.call_args[0][0]
|
||||
assert args.operate_type == "Delete"
|
||||
assert args.row_data is None
|
||||
st = capture_file(_fm(), reader, writer, cfg)
|
||||
assert st.deferred == 1
|
||||
assert st.enqueued == 0 and st.aged_out == 0
|
||||
writer.insert_queue_row.assert_not_called()
|
||||
writer.insert_dead_row.assert_not_called()
|
||||
writer.insert_archive_row.assert_not_called()
|
||||
|
||||
|
||||
def test_capture_unreadable_young_update_also_deferred():
|
||||
# Update 同样走 defer 路径（不降级 Delete）
|
||||
cfg = _cfg(defer=60)
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(12, "T", "6", "Update", _dt.datetime.now())]
|
||||
reader.read_row.return_value = None
|
||||
writer = MagicMock()
|
||||
st = capture_file(_fm(), reader, writer, cfg)
|
||||
assert st.deferred == 1
|
||||
writer.insert_queue_row.assert_not_called()
|
||||
writer.insert_dead_row.assert_not_called()
|
||||
|
||||
|
||||
def test_capture_unreadable_aged_row_marked_dead():
|
||||
# Insert 日志年龄 >= capture_defer_seconds → 入队标 dead，不执行破坏性 SQL
|
||||
cfg = _cfg(defer=60)
|
||||
old_time = _dt.datetime.now() - _dt.timedelta(seconds=200)
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(13, "T", "7", "Insert", old_time)]
|
||||
reader.read_row.return_value = None # 仍读不到
|
||||
writer = MagicMock()
|
||||
st = capture_file(_fm(), reader, writer, cfg)
|
||||
assert st.aged_out == 1
|
||||
assert st.enqueued == 0 and st.deferred == 0
|
||||
# archive 留痕（ProcessedOperateType=AgedOut）
|
||||
arch = writer.insert_archive_row.call_args[0][0]
|
||||
assert arch.original_operate_type == "Insert"
|
||||
assert arch.processed_operate_type == "AgedOut"
|
||||
assert arch.row_data is None
|
||||
# 入队标 dead，OperateType 保留原始 Insert
|
||||
qr, err = writer.insert_dead_row.call_args[0]
|
||||
assert qr.operate_type == "Insert" # 不降级
|
||||
assert qr.row_data is None
|
||||
assert "200s" in err
|
||||
|
||||
|
||||
def test_capture_defer_zero_disables_retry():
|
||||
# capture_defer_seconds=0 → 任何读不到都立即判死（边界）
|
||||
cfg = _cfg(defer=0)
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(14, "T", "8", "Insert", _dt.datetime.now())]
|
||||
reader.read_row.return_value = None
|
||||
writer = MagicMock()
|
||||
st = capture_file(_fm(), reader, writer, cfg)
|
||||
assert st.aged_out == 1
|
||||
assert st.deferred == 0
|
||||
|
||||
|
||||
def test_capture_delete_never_reads_row():
|
||||
# Delete 日志无需回读，直接入队
|
||||
cfg = _cfg()
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(15, "T", "9", "Delete", _dt.datetime.now())]
|
||||
writer = MagicMock()
|
||||
st = capture_file(_fm(), reader, writer, cfg)
|
||||
assert st.enqueued == 1
|
||||
reader.read_row.assert_not_called()
|
||||
qr = writer.insert_queue_row.call_args[0][0]
|
||||
assert qr.operate_type == "Delete"
|
||||
assert qr.row_data is None
|
||||
|
||||
|
||||
def test_capture_skips_excluded_tables():
|
||||
cfg = _cfg()
|
||||
fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog", "氩弧焊每日催货落实记录_停"])
|
||||
fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026",
|
||||
exclude_tables=["TableChangeLog", "氩弧焊每日催货落实记录_停"])
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(1, "TableChangeLog", "1", "Insert", None),
|
||||
LogRow(2, "氩弧焊每日催货落实记录_停", "1", "Insert", None)]
|
||||
reader.read_log.return_value = [LogRow(1, "TableChangeLog", "1", "Insert", _dt.datetime.now()),
|
||||
LogRow(2, "氩弧焊每日催货落实记录_停", "1", "Insert", _dt.datetime.now())]
|
||||
writer = MagicMock()
|
||||
assert capture_file(fm, reader, writer, cfg) == 0
|
||||
st = capture_file(fm, reader, writer, cfg)
|
||||
assert st.enqueued == 0 and st.out_of_scope == 2
|
||||
writer.insert_queue_row.assert_not_called()
|
||||
|
||||
|
||||
def test_capture_include_tables_filter():
|
||||
cfg = _cfg()
|
||||
fm = FileMapping(file="x.accdb", root="2026", schema="inspectionRecords", year_suffix="_YEAR2026",
|
||||
exclude_tables=["TableChangeLog"], include_tables=["检验合格记录表"])
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(1, "检验合格记录表", "1", "Insert", None),
|
||||
LogRow(2, "其它表", "1", "Insert", None)]
|
||||
reader.read_log.return_value = [LogRow(1, "检验合格记录表", "1", "Insert", _dt.datetime.now()),
|
||||
LogRow(2, "其它表", "1", "Insert", _dt.datetime.now())]
|
||||
reader.read_row.return_value = {"ID": 1}
|
||||
writer = MagicMock()
|
||||
assert capture_file(fm, reader, writer, cfg) == 1
|
||||
st = capture_file(fm, reader, writer, cfg)
|
||||
assert st.enqueued == 1
|
||||
assert writer.insert_queue_row.call_args[0][0].target_table == "检验合格记录表_YEAR2026"
|
||||
|
||||
Reference in New Issue
Block a user