MongoDB upsert 乐观并发中如何避免 DuplicateKeyException

结论

不要把 Version < incomingVersion 写进 upsert 的 filter。可以改为：

1. filter 只用唯一键 Id；
2. 通过 aggregation pipeline update 在更新表达式中比较 Version；
3. document 不存在，或者传入版本更新时，写入传入的数据；
4. 已存版本相同或更高时保留原 document。

这样只需一次单 document 操作，就能以原子方式完成版本判断和写入。旧版本事件仍会匹配已有 document，不会因为匹配失败而触发插入。

在特定场景下，可以捕获并忽略 DuplicateKeyException 作为兜底，但不要用异常代替正常的版本判断。

原代码为什么会触发重复键异常

upsert 会根据 filter 的匹配结果决定更新还是插入：

• 匹配到 document 时执行 update；
• 没有匹配到 document 时尝试 insert。

原 filter 同时包含 Id 和版本条件：

Id == user.Id AND Version < user.Version

假设数据库中的版本为 10，传入版本为 8。此时，相同 Id 的 document 虽然已经存在，但完整条件不成立，MongoDB 会认为没有匹配结果。

由于 IsUpsert = true，MongoDB 随后会尝试插入具有相同 Id 的 document，于是与 _id 唯一索引冲突。

SetOnInsert 只能控制插入时设置哪些字段，无法改变 filter 未匹配时尝试插入的行为。

使用 aggregation pipeline update

MongoDB 支持使用 aggregation pipeline 执行 update。filter 先按 Id 匹配现有 document，再由 pipeline 判断是采用传入数据，还是保留 $$ROOT。

下面用 $replaceWith 实现完整替换：

using MongoDB.Bson;
using MongoDB.Driver;

public Task<UpdateResult> UpsertAsync(
    User user,
    CancellationToken cancellationToken)
{
    var incoming = user.ToBsonDocument();

    var shouldWrite = new BsonDocument(
        "$or",
        new BsonArray
        {
            // upsert 创建的基础 document 中还没有 Version
            new BsonDocument(
                "$eq",
                new BsonArray
                {
                    new BsonDocument("$type", "$Version"),
                    "missing"
                }),

            // 已有 document 的版本较旧
            new BsonDocument(
                "$lt",
                new BsonArray
                {
                    "$Version",
                    user.Version
                })
        });

    var pipeline = PipelineDefinition<User, User>.Create(
        new[]
        {
            new BsonDocument(
                "$replaceWith",
                new BsonDocument(
                    "$cond",
                    new BsonArray
                    {
                        shouldWrite,

                        // 防止传入 document 中的字符串被当作字段路径解析
                        new BsonDocument("$literal", incoming),

                        // 相同或更高版本时保持原 document
                        "$$ROOT"
                    }))
        });

    return context.Users.UpdateOneAsync(
        Builders<User>.Filter.Eq(u => u.Id, user.Id),
        pipeline,
        new UpdateOptions { IsUpsert = true },
        cancellationToken);
}

filter 中只保留：

Builders<User>.Filter.Eq(u => u.Id, user.Id)

只要相同 Id 的 document 已经存在，filter 就能匹配。是否允许更新，则由 pipeline 根据版本决定。

三种情况的处理结果

假设 Id 对应 _id，或者对应另一个具有唯一索引的字段：

最后一种情况仍会成功完成操作，但 ModifiedCount 通常为 0。调用方可以通过这个值判断此次事件是否更新了 projection。

保留 document 中的额外字段

$replaceWith 会用传入对象替换整个 document。如果数据库中还保存了不属于 User 模型的字段，这些字段会被删除。

需要保留额外字段时，可以改用 $set，并让每个字段使用相同的 $cond：

[
  {
    $set: {
      Version: {
        $cond: [
          {
            $or: [
              { $eq: [{ $type: "$Version" }, "missing"] },
              { $lt: ["$Version", incomingVersion] }
            ]
          },
          incomingVersion,
          "$Version"
        ]
      },
      Name: {
        $cond: [
          {
            $or: [
              { $eq: [{ $type: "$Version" }, "missing"] },
              { $lt: ["$Version", incomingVersion] }
            ]
          },
          incomingName,
          "$Name"
        ]
      }
    }
  }
]

实际使用时，IsDeleted、UserName、Media 等字段也要采用同一个条件。所有字段必须共享完全一致的版本判断，否则一个 document 中可能混入新旧版本的数据。

是否应该忽略 DuplicateKeyException

继续使用原 filter 时，旧版本事件引发重复键异常是可以预见的。业务确实允许忽略这类异常时，应严格限定捕获范围：

try
{
    await UpsertWithVersionInFilterAsync(user, cancellationToken);
}
catch (MongoWriteException ex)
    when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
{
    // 只有确认冲突来自 Id 唯一索引，
    // 且业务含义确实是“已有相同或更高版本”时才能忽略。
}

DuplicateKeyException 也可能来自 UserName 等其他唯一索引。如果直接捕获所有重复键异常，真正的数据冲突也会被当成旧事件忽略。仅检查错误码，也未必能准确表示当前 document 已有相同或更高版本这一业务条件。

异常处理适合用于并发插入冲突的重试或兜底，不应替代版本比较。

并发写入的注意事项

这条 pipeline 能以原子方式完成单个 document 的判断和修改。它不需要预先读取 document，因此没有 read-modify-write 之间的竞态窗口。

多个请求同时对尚不存在的 Id 执行 upsert 时，仍可能在首次创建 document 时发生竞争。具体行为取决于 MongoDB 服务端版本、唯一索引和 filter 的形式。处理时应注意以下几点：

• 为 Id 建立唯一索引，通常可以直接使用 _id；
• filter 使用唯一键的等值条件；
• 对少量并发首次插入引起的 duplicate key 进行有限重试；
• 重试时再次执行同一条 pipeline update，此时已有 document 会进入版本比较分支；
• 不要直接忽略其他唯一索引产生的冲突。

在 CQRS query side，这种写法会按照事件版本维护 projection。高版本事件先到时可以创建 document，后来到达的低版本事件仍会匹配相同 Id，但不会覆盖较新的数据。它处理的是事件乱序和重复消费，与 command side 通过精确期望版本检测并发冲突的 optimistic concurrency 不同。

aggregation pipeline update 需要 MongoDB 服务端支持 pipeline 形式的更新。示例中的 BSON 字段名也要与数据库中的实际名称一致。如果项目配置了 camelCase convention 或自定义字段映射，需要把 Version 等名称改成对应的 BSON 字段名。

备注：内容仅供参考。