Node.js中如何正确处理离线与在线数据库同步的异常问题？

当前同步逻辑的核心问题是云端数据操作与本地状态更新并非原子性操作：当执行完云端的insertOne/updateOne后，若在执行本地Product.update前发生断电、程序崩溃等情况，会导致该产品已同步到云端，但本地仍处于未同步状态（1或3）；反之若先更新本地状态再执行云端操作，可能出现本地标记为已同步但云端操作失败，导致数据丢失。这种非原子操作直接引发了数据一致性问题，出现你描述的“云端已同步但本地状态未更新”的遗漏情况。

要解决这个问题，我们通过状态机优化+幂等操作+重试机制保证数据最终一致性，具体步骤如下：

1. 扩展同步状态机 #

新增「同步中」状态（例如isSynced=4），用于标记正在处理的记录：

• 1: 待新增
• 3: 待更新
• 4: 同步中
• 2: 已同步

下次同步时可识别中途中断的「同步中」记录，进行重试处理，避免遗漏。

2. 保证云端操作的幂等性 #

将新增操作从insertOne改为updateOne + upsert，确保即使重复执行该操作，也不会创建重复数据，只会更新现有记录，避免因重试导致的重复数据问题。

3. 原子化单条记录的处理流程 #

调整每个产品的处理逻辑，保证状态变更与云端操作的连贯性：

• 先将本地记录标记为「同步中」（isSynced=4）
• 执行云端同步操作
• 成功后将本地标记为「已同步」（isSynced=2）
• 失败则将本地状态回滚为原状态（1或3）

若本地数据库支持事务（如MySQL），可将本地状态变更包裹在事务中，进一步保证状态变更的原子性。

4. 增加「同步中」记录的重试逻辑 #

每次同步开始时，同时拉取isSynced=1、3和4的记录，对中途中断的记录进行重试。

const index = async (req, res) => {
  const client = new MongoClient(mongoURI);
  try {
    await client.connect();
    const db = client.db(dbName);
    const productsCollection = db.collection(collectionName);

    // 拉取待同步（1、3）和同步中（4）的记录，处理中断重试
    const updatedProductsArray = await Product.findAll({
      where: { isSynced: [1, 3, 4] },
    });

    if (updatedProductsArray.length > 0) {
      for (const product of updatedProductsArray) {
        const originalStatus = product.isSynced;
        const { isSynced, ...productData } = product.dataValues;
        let syncSuccess = false;

        try {
          // 标记为同步中，避免重复处理
          await Product.update(
            { isSynced: 4 },
            { where: { product_id: productData.product_id } }
          );

          // 统一用upsert保证幂等性，适配新增/更新场景
          await productsCollection.updateOne(
            { product_id: productData.product_id },
            { $set: { ...productData, isSynced: 2 } },
            { upsert: true }
          );

          // 标记为已同步
          await Product.update(
            { isSynced: 2 },
            { where: { product_id: productData.product_id } }
          );

          syncSuccess = true;
        } catch (err) {
          console.error(`Error syncing product ID ${productData.product_id}:`, err);
          // 失败时回滚状态为原状态
          if (!syncSuccess) {
            await Product.update(
              { isSynced: originalStatus },
              { where: { product_id: productData.product_id } }
            );
          }
          continue;
        }
      }
    }

    res.json({ status: 1, msg: "Data Sync Successfully!!!" });
  } catch (error) {
    console.error("Error syncing data:", error);
    res.status(500).json({ error: "Error syncing data" });
  } finally {
    await client.close(); // 等待连接关闭完成，避免提前终止操作
  }
};

• 分批处理：若待同步记录数量极大，可分批次处理，避免单次同步时间过长。
• 日志增强：为每条同步记录添加详细日志（同步时间、状态变更、操作结果），方便问题排查。
• 定期校验：定期对比本地与云端数据，对不一致的记录进行补偿处理，强化最终一致性。

内容的提问来源于stack exchange，提问作者Swapnil Mane