MongoDB: concurrent conditional UpdateItem creates on one key can each answer 200, and all but one write is lost #790

On the MongoDB backend, conditional UpdateItem calls that race to create the same item with attribute_not_exists(pk) can each answer 200, and the item ends up with only one writer's values. Some of the losers answer 500 InternalServerError instead. Amazon DynamoDB, measured on 2026-11-03 in us-east-1 with the reproduction below (32 writers x 20 rounds), gave exactly one 200 per round, stored that caller's w each time, and refused the remaining 620 calls with ConditionalCheckFailedException. Callers use attribute_not_exists creates to enforce uniqueness, so a second 200 tells two clients that each of them created the item.

Where

On a table without GSIs or streams, a conditional UpdateItem whose condition can be pushed down runs update_item_pushdown (crates/storage-mongodb/src/data_engine.rs, main 5ff323b). It reads the item with the key and condition filter, then by key alone. When neither read finds a document, it evaluates the condition against an empty item, builds the new item, and writes it with replace_one(key_filter, new_doc) and upsert: true (line 2961). That write runs outside a session and has no version filter.

Two writers that both read no item both reach that write. The later replace_one matches the document the earlier one created and replaces it, so both answer 200 with an empty ALL_OLD. When both upserts insert at the same moment, one of them fails on the _id_ unique index with E11000, which the call maps to StorageError::Internal and the server answers with 500.

The existing-item branch of the same function handles a concurrent writer. Its replace_one carries the _v version in the filter, and a miss falls back to the session path through the "raced by concurrent writer" error. The create branch has no equivalent. Creating the item with insert_one and returning that error on a duplicate key would send the loser to the session path, which reads the created item and answers ConditionalCheckFailedException.

Reproduction

ExtendDB on MongoDB 7.0.45 as a single-node replica set, a debug build of main 5ff323b. A table with a string hash key pk, on-demand, no GSIs and no stream. Each round, 32 threads with one client each and client retries off wait on a barrier, then send UpdateItem on one new key with SET w = :w, ConditionExpression attribute_not_exists(pk), :w set to the writer's name and ReturnValues ALL_OLD. A consistent GetItem reads the stored item after the round.

import threading
import uuid
from collections import Counter

import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

ENDPOINT = "https://127.0.0.1:18443"  # ExtendDB on MongoDB; credentials come from the environment
TABLE, WRITERS, ROUNDS = "create-race", 32, 50
cfg = Config(retries={"total_max_attempts": 1, "mode": "standard"})


def client():
    return boto3.client("dynamodb", endpoint_url=ENDPOINT, verify=False, region_name="us-east-1", config=cfg)


admin = client()
admin.create_table(
    TableName=TABLE,
    AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
    KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
    BillingMode="PAY_PER_REQUEST",
)
admin.get_waiter("table_exists").wait(TableName=TABLE)
clients = [client() for _ in range(WRITERS)]
totals = Counter()
for r in range(ROUNDS):
    pk, barrier, out = uuid.uuid4().hex, threading.Barrier(WRITERS), [None] * WRITERS

    def run(i):
        barrier.wait()
        try:
            resp = clients[i].update_item(
                TableName=TABLE,
                Key={"pk": {"S": pk}},
                UpdateExpression="SET w = :w",
                ConditionExpression="attribute_not_exists(pk)",
                ExpressionAttributeValues={":w": {"S": f"w{i}"}},
                ReturnValues="ALL_OLD",
            )
            out[i] = "200 with ALL_OLD" if resp.get("Attributes") else "200"
        except ClientError as e:
            out[i] = e.response["Error"]["Code"]

    threads = [threading.Thread(target=run, args=(i,)) for i in range(WRITERS)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    stored = admin.get_item(TableName=TABLE, Key={"pk": {"S": pk}}, ConsistentRead=True)["Item"]["w"]["S"]
    codes = Counter(out)
    totals.update(codes)
    if codes != Counter({"200": 1, "ConditionalCheckFailedException": WRITERS - 1}):
        print(r, dict(codes), "stored", stored)
print(dict(totals))

32 writers x 50 rounds, 1,600 calls:

21 of the 50 rounds did not end with one 200 and 31 ConditionalCheckFailedException. In 17 rounds more than one writer answered 200 (two in most of them, three in three rounds, eight in one), and the stored w was the value of one of them, so 26 writes answered 200 and were replaced. In the other 4 rounds one writer answered 200 and one or two answered InternalServerError. The server logged each of the 5 as E11000 duplicate key error collection: extenddb_data._ddb_<table id> index: _id_.

PR #431 doesn't change this path. Two runs of the script above against that branch had 18 and 30 rounds of 50 that did not end with one 200 and 31 ConditionalCheckFailedException, and one InternalServerError each.