NodeJS单次请求批量更新MongoDB多个文档不稳定如何保障全部更新成功

核心问题根因 #

你当前代码出现部分更新的核心原因如下：

• 未使用MongoDB事务，多集合/多文档更新不具备原子性，中间某一步出错时已经执行的更新不会自动回滚
• 未校验更新操作的返回结果，即使updateOne执行成功但未匹配到文档、没有实际修改内容，代码也会继续执行，无法感知到异常
• 未处理并发请求问题，同一请求短时间被多次调用时，后到的请求可能打断或覆盖前序的更新流程

具体实现方案 #

2.1 优先执行前置校验查询 #

完全满足你的要求，先执行现有findAccepted查询，确认请求未被处理过，再执行后续更新逻辑。

2.2 引入MongoDB事务保证原子性 #

所有更新操作放在同一个事务中执行，只要有任意一步失败，所有已执行的更新都会自动回滚，不会出现部分更新的情况。

注意：MongoDB 4.0+版本需要运行在副本集模式，4.2+版本支持分片集群模式下的多文档事务，使用前确认你的Mongo部署符合要求。

2.3 校验所有更新操作的执行结果 #

每个更新操作执行后校验返回的modifiedCount，如果和预期修改数量不一致，主动抛出异常触发事务回滚。

2.4 优化批量更新减少数据库请求 #

将循环调用updateOne改为bulkWrite批量操作，降低数据库交互次数，提升性能同时减少出错概率。

改造后代码示例 #

const { MongoServerError } = require('mongodb');
// 注意需要提前在Mongoose连接配置中开启会话支持
router.post('/requestAccept', async (req, res) => {
  // 原有参数校验逻辑保持不变
  if (req.body.sumDebit != req.body.sumCredit) return res.json({ success: false, message: 'Accepted debit not equal to accepted credit.' });
  if (req.body.sumDebit < 0) return res.json({ success: false, message: 'Invalid amount.' });
  if (req.body.minAccept != req.body.sumDebit) return res.json({ success: false, message: 'Allowed amount to accept is: ' + req.body.minAccept });
  if (!req.body.sumCredit || req.body.sumCredit < 0) return res.json({ success: false, message: 'Invalid amount.' });
  if (!req.body.ticketNumber ) return res.json({ success: false, message: 'Unexpected TIC error.' });
  if (!req.body.mainId ) return res.json({ success: false, message: 'Unexpected MAI error.' });
  if (!req.body.requestId ) return res.json({ success: false, message: 'Unexpected REQ error.' });
  if (!req.body.eeid ) return res.json({ success: false, message: 'Unexpected EEI error.' });
  const requestMainId = new ObjectID(req.body.mainId);
  const requestSubId = new ObjectID(req.body.requestId);
  // 第一步：优先执行前置校验查询，确认请求未被处理
  try {
    const findAccepted = await FinancialRequest.aggregate([
      { $unwind: { path: "$requests", preserveNullAndEmptyArrays: true } },
      {
        $project: {
          _id: 1,
          'subId': "$requests._id",
          'status': "$requests.status",
        }
      },
      { $match: { $and: [{ _id: requestMainId, subId: requestSubId, status: 'Accepted' }] } },
    ]);
    if (findAccepted.length > 0 ) return res.json({ success: false, message: 'Unexpected error. Please reload your system' });
  } catch (err) {
    return res.json({ success: false, message: '校验请求状态失败' });
  }
  // 第二步：开启会话和事务执行所有更新
  const session = await mongoose.startSession();
  session.startTransaction();
  try {
    // 批量更新Credit文档
    const creditUpdates = req.body.credit.map(item => ({
      updateOne: {
        filter: { accountNumber: item.creditAccountNumber },
        update: {
          $push: {
            "transactions": {
              ticketNumber: req.body.ticketNumber,
              mainId: req.body.mainId,
              requestId: req.body.requestId,
              requestType: req.body.requestType,
              type: 'Credit',
              eeid: req.body.eeid,
              firstName: req.body.firstName,
              lastName: req.body.lastName,
              amount: item.amountAccepted,
              date: new Date(),
            }
          }
        }
      }
    }));
    const creditRes = await Financial.bulkWrite(creditUpdates, { session });
    if (creditRes.modifiedCount !== req.body.credit.length) {
      throw new Error('Credit批量更新数量不符合预期');
    }
    // 批量更新Debit文档
    const debitUpdates = req.body.debit.map(item => ({
      updateOne: {
        filter: { accountNumber: item.debitAccountNumber },
        update: {
          $push: {
            "transactions": {
              mainId: req.body.mainId,
              ticketNumber: req.body.ticketNumber,
              requestId: req.body.requestId,
              requestType: req.body.requestType,
              type: 'Debit',
              eeid: req.body.eeid,
              firstName: req.body.firstName,
              lastName: req.body.lastName,
              amount: item.amountAccepted,
              date: new Date(),
            }
          }
        }
      }
    }));
    const debitRes = await Financial.bulkWrite(debitUpdates, { session });
    if (debitRes.modifiedCount !== req.body.debit.length) {
      throw new Error('Debit批量更新数量不符合预期');
    }
    // 批量更新Credit请求文档
    const creditReqUpdates = req.body.credit.map(item => ({
      updateOne: {
        filter: { _id: req.body.mainId },
        update: {
          $set: {
            "requests.$[l].credits.$[m].amountAccepted": item.amountAccepted,
            "requests.$[l].credits.$[m].dateAccepted": new Date(),
            "requests.$[l].credits.$[m].status": 'Accepted',
          }
        },
        arrayFilters: [{ "l._id": req.body.requestId }, { "m._id": item._id }]
      }
    }));
    const creditReqRes = await FinancialRequest.bulkWrite(creditReqUpdates, { session });
    if (creditReqRes.modifiedCount !== req.body.credit.length) {
      throw new Error('Credit请求批量更新数量不符合预期');
    }
    // 批量更新Debit请求文档
    const debitReqUpdates = req.body.debit.map(item => ({
      updateOne: {
        filter: { _id: req.body.mainId },
        update: {
          $set: {
            "requests.$[o].debits.$[p].amountAccepted": item.amountAccepted,
            "requests.$[o].debits.$[p].dateAccepted": new Date(),
            "requests.$[o].debits.$[p].status": 'Accepted',
          }
        },
        arrayFilters: [{ "o._id": req.body.requestId }, { "p._id": item._id }]
      }
    }));
    const debitReqRes = await FinancialRequest.bulkWrite(debitReqUpdates, { session });
    if (debitReqRes.modifiedCount !== req.body.debit.length) {
      throw new Error('Debit请求批量更新数量不符合预期');
    }
    // 更新主请求状态
    const updateStatusRes = await FinancialRequest.updateOne(
      { _id: req.body.mainId },
      {
        $set: {
          "requests.$[o].dateAccepted": new Date(),
          "requests.$[o].status": 'Accepted',
        },
      },
      { arrayFilters: [{ "o._id": req.body.requestId }], session }
    );
    if (updateStatusRes.modifiedCount !== 1) {
      throw new Error('主请求状态更新失败');
    }
    // 所有操作成功，提交事务
    await session.commitTransaction();
    res.json({ success: true, message: 'Request Accepted' });
  } catch (err) {
    // 任意步骤出错，回滚事务
    await session.abortTransaction();
    res.json({ success: false, message: `更新失败：${err.message || 'An error occured'}` });
  } finally {
    // 关闭会话
    session.endSession();
  }
});

额外优化建议 #

• 可以给requestId加唯一索引，避免重复请求被多次处理，提升接口幂等性
• 事务执行失败如果是因为写冲突，可以加3次以内的自动重试逻辑，降低偶发失败的概率
• 生产环境建议将错误日志落盘，方便排查异常情况

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