- Architecture: iOS export → Rust EventBuffer → API batch upload - Event types, sequence numbers, ACK mechanism - Session state machine, error handling, performance comparison - iOS UploadScheduler integration, app lifecycle handling Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
5.8 KiB
5.8 KiB
V2 事件 Pipeline 架构
1. 概述
V2 事件 Pipeline 是 Document Runtime(Rust FFI)+ iOS Swift 层协作的阅读行为采集链路。
与 V1 的区别:
- V1:每个事件独立 push,iOS 手动管理 buffer
- V2:Rust 层内置 EventBuffer + 序列号 + ACK 机制,iOS 仅负责定时 export
2. 架构
┌──────────────────────────────────────────────────┐
│ iOS Swift │
│ UploadScheduler (30s timer) │
│ → documentRuntime.exportPending() │
│ → POST /api/reading-events/batch │
│ → documentRuntime.ackExported(count) │
└──────────────────┬───────────────────────────────┘
│ FFI (uniffi)
┌──────────────────▼───────────────────────────────┐
│ Rust: zx_document_core │
│ │
│ ┌─ push_opened() │
│ ├─ push_heartbeat_v2(position?) │
│ ├─ push_position_changed() │
│ ├─ push_closed() │
│ ├─ push_marked_as_read() │
│ └─ mark_events_failed_v2(ids) │
│ │ │
│ ▼ │
│ ┌──────────────────────────┐ │
│ │ EventBuffer (Mutex<Vec>) │ │
│ │ - max 1000 events │ │
│ │ - auto-increment seq │ │
│ │ - session state tracking │ │
│ └──────────────────────────┘ │
│ │ │
│ ▼ │
│ export_pending() → Vec<BufferedEvent> │
│ clear_exported(count) │
│ mark_events_failed_v2(ids) │
└──────────────────────────────────────────────────┘
3. 事件类型
| 事件 | 触发时机 | 携带数据 |
|---|---|---|
opened |
openMaterial() |
materialId |
heartbeat |
定时器 (30s) | position? |
position_changed |
滚动/翻页 | position |
closed |
closeMaterial() |
totalActiveMs |
marked_as_read |
markAsRead() |
— |
failed |
异常/超时 | errorCode |
4. 序列号与 ACK
每个事件分配递增序列号(session 内自增)
iOS export 获得 Vec<(seq, event)>
上传 API 成功 → ACK exported count → Rust 移除已确认事件
上传失败 → mark_events_failed_v2(ids) → 事件保留,下次重试
4.1 export → ACK 流程
1. iOS 调用 exportPending() → Vec<BufferedEvent>
2. iOS 将 events 序列化为 JSON,POST /api/reading-events/batch
3. 成功 → clearExported(count) → Rust 移除前 count 条
4. 失败 → 事件保留在 buffer,下次 export 仍包含
4.2 会话恢复
App 重启 → SessionManager.reload_stale_events()
→ 将上次未 ack 的事件标记为 failed
→ 返回可重试的事件列表
5. iOS 集成
5.1 UploadScheduler
class UploadScheduler {
let interval: TimeInterval = 30
func start() {
Timer.scheduledTimer(withTimeInterval: interval, repeats: true) { _ in
Task { await self.flush() }
}
}
func flush() async {
let events = documentRuntime.exportPending()
guard !events.isEmpty else { return }
do {
try await api.uploadEvents(events)
documentRuntime.clearExported(events.count)
} catch {
// Events remain in buffer for retry
}
}
}
5.2 App 生命周期
// sceneDidEnterBackground
UploadScheduler.shared.flush() // 强制上传
// App 启动
RuntimeRecoveryService.shared.recover() // 恢复未完成会话
6. Session 状态机
openMaterial()
┌─ idle ─────────────────► opened
│ │
│ heartbeat/
│ position_change
│ │
│ ▼
│ reading
│ │
│ closeMaterial()
│ │
└──────── closed ◄───────────┘
7. 错误处理
| 场景 | 处理 |
|---|---|
| Buffer 满(1000条) | 拒绝新事件,iOS 应触发 flush |
| FFI 调用失败 | fallback 到 V1 NoopReadingRuntimeAdapter |
| 网络不可用 | 事件保留,下次 export 包含未 ACK 事件 |
| Session 超时(无活动 30min) | cleanup_stale_sessions_v2 自动清理 |
8. 性能特征
| 指标 | V1 | V2 |
|---|---|---|
| FFI 调用频率 | 每事件 1 次 | 每批次 1 次 |
| 内存占用 | ~50 events × iOS heap | ~1000 events × Rust heap |
| 序列号 | 无 | Session 内自增 |
| ACK 粒度 | 无 | Batch 级别 |