migrating-to-checkpoints

迁移到检查点

更新现有的 createSync() 代码以使用检查点。首先保留服务提供商的行为；只更改进度／恢复机制，以及确保检查点功能正确所需的删除处理。

范围

• 本技能适用于 Zero YAML TypeScript 同步。如果 nango.yaml 仍定义了该同步，请建议先迁移到 Zero YAML，或在迁移检查点的同时进行迁移。提及 migrating-to-zero-yaml 技能；如果尚未安装，请建议运行 npx skills add https://github.com/NangoHQ/skills --skill migrating-to-zero-yaml，并附上文档链接：https://nango.dev/docs/implementation-guides/platform/migrations/migrate-to-zero-yaml。
• 如果仓库有多个服务提供商，优先一次迁移一个同步。

注意事项

• 每次成功调用 batchSave() 后都必须立即调用 saveCheckpoint()（调用 batchUpdate()／batchDelete() 推进进度后也一样）。这适用于分页循环（for await 遍历 nango.paginate、手动游标／偏移量循环、嵌套抓取），不能只在 exec 末尾调用。只在辅助函数返回后或 exec 末尾保存，如果运行在分页过程中失败，就无法持久化进度。
• 不要用“还有更多页面”作为调用 saveCheckpoint() 的条件。 类似 if (nextCursor) { await nango.saveCheckpoint(...) } 的模式会跳过最后一页的调用。如果整个数据集刚好在第一页返回，就不会写入检查点行，随后 clearCheckpoint() 会失败并报 checkpoint_conflict（“自上次读取以来检查点已更新”）——这与真实的并发写入产生相同错误，因为 clearCheckpoint() 使用乐观锁，删除零行也会以相同方式被拒绝。每页都必须保存检查点，包括最后一页，并在清除检查点之前完成保存。
• 不要只保存检查点而不使用它。下一次运行必须在服务提供商请求参数或分页／恢复状态中使用该检查点。
• 检查点对象只能包含扁平值：字符串、数字或布尔值。日期应保存为 ISO 字符串；不要保存 Date 对象、数组或嵌套对象。
• 部署后的首次运行还没有检查点，因此会像初始同步一样运行，耗时可能比之后的增量运行更长。在相关情况下，请告知客户。
• 仅同步增量变更的任务不得使用 trackDeletesStart()／trackDeletesEnd()。这些端点会省略未变更的记录，因此 Nango 会将未变更记录标记为已删除。

清查

搜索旧式和部分实现的检查点模式：

rg -n "lastSyncDate|syncType:\s*['\"]incremental|deleteRecordsFromPreviousExecutions|trackDeletes:|track_deletes|getCheckpoint|saveCheckpoint|clearCheckpoint" .

对于每个同步，确定：

• 现有的下限或恢复标记：updated_at、modified_since、变更记录端点游标、页面令牌、偏移量／页码、since_id 等。
• 服务提供商请求是否只返回变更记录，还是会遍历完整数据集。
• 分页形式，以及是否可以在每页后保存进度。
• 传递给 batchSave、batchUpdate 和 batchDelete 的模型名称。
• 删除策略：显式删除记录端点／删除标记／Webhook、全量刷新差异比对，或没有可靠的删除信号。

选择检查点

• 时间戳下限：z.object({ updated_after: z.string() })。
• 游标或变更数据流：z.object({ cursor: z.string() })。
• 页码／偏移量恢复：z.object({ page: z.number() }) 或 z.object({ offset: z.number() })。
• 复合恢复标记：组合扁平原始值，例如 z.object({ updated_after: z.string(), page_token: z.string().optional() })。
• 全量刷新容错：在运行过程中保存游标／页码检查点，并且仅在成功获取并保存完整数据集后调用 clearCheckpoint()。

如果服务提供商无法筛选变更，也无法恢复分页，不要强行添加检查点。说明服务提供商的限制，并保留全量刷新，或将其转换为有明确理由的全量刷新。

迁移步骤

1. 在同步附近添加一个命名检查点模式，并将其传递给 createSync()：

const CheckpointSchema = z.object({
  updated_after: z.string(),
});

export default createSync({
  description: "Sync contacts",
  frequency: "every hour",
  checkpoint: CheckpointSchema,
  models: { Contact },
  exec: async (nango) => {
    // ...
  },
});

2. 从增量迁移中移除 syncType: 'incremental'。checkpoint 字段会取代这个旧信号。

3. 将 nango.lastSyncDate 的读取替换为 await nango.getCheckpoint()，并在提供商请求中使用检查点：

const checkpoint = await nango.getCheckpoint();

const response = await nango.get({
  endpoint: "/contacts",
  params: {
    ...(checkpoint?.updated_after && {
      updated_after: checkpoint.updated_after,
    }),
  },
  retries: 3,
});

4. 每次调用 batchSave() 后，都要在同一轮循环中调用 saveCheckpoint()——包括在 nango.paginate 内部以及任何提取出来的分页辅助函数中。将两者配对；不要把检查点写入推迟到 exec 末尾。

const contacts = response.data.items.map(mapContact);
await nango.batchSave(contacts, "Contact");

if (contacts.length > 0) {
  const lastContact = contacts[contacts.length - 1]!;
  await nango.saveCheckpoint({ updated_after: lastContact.updated_at });
}

分页同步：

for await (const page of nango.paginate<Item>(config)) {
  const records = page.map(mapRecord);
  await nango.batchSave(records, "Contact");
  await nango.saveCheckpoint({ updated_after: latestTimestamp(records) });
}

无效示例：调用 await getAndSaveUsers(nango)，该函数只在循环中调用 batchSave，然后在辅助函数返回后才于 exec 中调用一次 saveCheckpoint。

5. 对于时间戳检查点，优先使用提供商记录中经过排序的最后修改时间。如果多条记录可能具有相同的时间戳，或者 API 无法稳定排序，请使用复合检查点，包含提供商游标／页令牌或 last_id 之类的打破平局字段。

6. 不要仅为了保存一个最终检查点而将所有记录都累积在内存中。逐页处理、调用 batchSave 和 saveCheckpoint。

使用检查点进行全量刷新

仅当 API 无法返回有变更的行、但可以恢复分页时才使用此方式。检查点用于故障恢复，而非增量过滤——并且它是必需的，不是可选项。Nango 同步在有时间限制的执行窗口内运行。如果没有检查点，在该窗口内未完成的全量刷新会在下一次运行时从第 1 页重新开始，每次都重新获取相同的前几页，永远无法到达数据集的其余部分。

• 在获取数据前读取已保存的游标／页码。
• 在每次刷新执行时调用 trackDeletesStart()。重复调用是安全的——它不会覆盖同一逻辑刷新之前一次执行已开启的删除跟踪窗口的起始位置。
• 如果存在检查点，则从已保存的游标／页码开始分页。
• 每次 batchSave() 成功后，使用下一游标／页码调用 saveCheckpoint()（在同一个循环体中——绝不能只在 exec 末尾调用），包括最后一页，而不仅仅是在还有下一页时调用。
• 仅在保存最后一页后调用 clearCheckpoint()。保存每个已处理的页面（包括最后一页），可确保正常的页面处理路径有检查点记录可清除。如果某条独立路径根本没有创建检查点（例如，没有处理任何页面），不要在该路径调用 clearCheckpoint()；这会在运行时抛出 checkpoint_conflict。这不能替代保存最后一页。
• 仅在调用 clearCheckpoint() 后调用 trackDeletesEnd()，以确保它只会在实际完成整个数据集的那次执行中触发一次。
• 在下一次计划运行时，已清除的检查点会使同步再次从头开始。

删除处理

• 如果提供商会暴露已删除记录、墓碑记录或删除 Webhook，请使用同一检查点窗口，对这些 ID 调用 batchDelete()。
• 如果检测删除需要比较整个数据集，请使用全量刷新：在获取／保存数据前调用 trackDeletesStart('Model')，并且仅在保存整个数据集且清除检查点后调用 trackDeletesEnd('Model')。
• 在跨多次执行的检查点全量刷新中，可以在每次执行开始时调用 trackDeletesStart()——这是安全且幂等的，不会覆盖已开启窗口的起始位置。只有在完成全量刷新并清除检查点的那次执行中，才应调用 trackDeletesEnd()。
• 移除 deleteRecordsFromPreviousExecutions()。它与使用检查点的同步不兼容。

测试

从 Nango 项目根目录运行本地验证流程：

nango compile
nango dryrun <sync-name> <connection-id> --validate -e dev --no-interactive --auto-confirm
nango dryrun <sync-name> <connection-id> --validate -e dev --no-interactive --auto-confirm --checkpoint '{"updated_after":"2024-06-01T00:00:00Z"}'

如果同步需要元数据，请使用 --metadata；根据模式调整 --checkpoint 负载；如果仓库中已有测试套件，也请运行它。

最终检查清单

• 迁移后的同步中不再引用 nango.lastSyncDate
• 检查点增量同步中不再使用 syncType: 'incremental'
• 存在 checkpoint schema，且仅使用扁平的原始值
• 在发出提供方请求前调用 getCheckpoint()
• 保存的检查点会影响下一次提供方请求或恢复状态
• 每个会推进进度的 batchSave()/batchUpdate()/batchDelete() 都会在同一循环中立即调用 saveCheckpoint()（包括 nango.paginate 和分页辅助函数内部的调用，而不只是 exec 返回后调用一次）
• 全量刷新同步有检查点 schema，会从检查点恢复分页，在每一页之后（包括最后一页）保存检查点，并且仅在最后一页成功完成后调用 clearCheckpoint()
• 全量刷新中的 saveCheckpoint() 调用绝不受是否还有下一页的条件限制
• 每条调用 clearCheckpoint() 的路径都已创建或恢复检查点；未创建检查点的路径不会调用它
• 只有在 clearCheckpoint() 之后，才会在完成最后一页的那次执行中运行全量刷新的 trackDeletesEnd()
• 删除处理方式与同步类型匹配：显式删除由提供方发起时使用 batchDelete()，仅全量刷新使用 trackDeletesStart/End
• 使用符合实际情况的 --checkpoint 测试了 dryrun

实用文档

• 检查点：https://nango.dev/docs/implementation-guides/use-cases/syncs/checkpoints
• 迁移指南：https://nango.dev/docs/implementation-guides/platform/migrations/migrate-to-checkpoints
• 删除检测：https://nango.dev/docs/implementation-guides/use-cases/syncs/deletion-detection
• Functions SDK 参考：https://nango.dev/docs/reference/functions#checkpoints