Concurrent bulk upserts can return HTTP 200 searches with missing hits #3028

Summary

A normal bulk import of new documents makes sequence IDs visible in the in-memory search index before their JSON documents have been written to RocksDB. A concurrent search can select those IDs during that window.

The search still returns HTTP 200 and keeps the pre-materialization found count, but silently omits every hit whose JSON is not yet in the store. This requires no collection deletion/recreation, JOIN, async reference, proxy, or multi-node cluster.

This supersedes #2789 with a current-version, production-faithful reproduction.

Environment

• Typesense 31.0.rc14
• Docker image typesense/typesense:31.0.rc14
• Image digest sha256:6f63c3de844ce3c399dee23e04530f38e80a7462de621f1c8444245325320baa
• Reproduced on one node and on an isolated three-node cluster with ordinary action=upsert&batch_size=40

The same signature is also currently present on a production three-node v30.2 cluster.

Minimal reproduction, including recovery timing

Start a clean server:

docker run --rm -d --name ts-index-store-repro -p 18108:8108   typesense/typesense:31.0.rc14   --data-dir=/data --api-key=xyz --thread-pool-size=16

Once /health is ready, run:

import json
import threading
import time
import urllib.parse
import urllib.request

BASE = "http://127.0.0.1:18108"
HEADERS = {"X-TYPESENSE-API-KEY": "xyz"}
PER_PAGE = 250
bad_seen = threading.Event()
bad_lock = threading.Lock()
first_bad = {}

def request(method, path, body=None, content_type="application/json"):
    headers = dict(HEADERS)
    data = None
    if body is not None:
        headers["Content-Type"] = content_type
        data = body.encode()
    req = urllib.request.Request(BASE + path, data=data, headers=headers, method=method)
    with urllib.request.urlopen(req, timeout=120) as response:
        return response.status, response.read().decode()

schema = {
    "name": "products",
    "fields": [
        {"name": "name", "type": "string"},
        {"name": "created_at", "type": "int64", "sort": True},
        {"name": "payload", "type": "string", "index": False},
    ],
    "default_sorting_field": "created_at",
}
request("POST", "/collections", json.dumps(schema))

params = urllib.parse.urlencode({
    "q": "*",
    "query_by": "name",
    "sort_by": "created_at:desc",
    "per_page": PER_PAGE,
    "include_fields": "id,name,created_at",
    "use_cache": "false",
})
search_path = "/collections/products/documents/search?" + params

def search():
    status, body = request("GET", search_path)
    result = json.loads(body)
    returned = len(result["hits"])
    expected = min(result["found"], PER_PAGE)
    return status, result, returned, expected

def search_worker():
    while not bad_seen.is_set():
        status, result, returned, expected = search()
        if returned < expected:
            with bad_lock:
                if not first_bad:
                    first_bad.update({
                        "status": status,
                        "found": result["found"],
                        "returned": returned,
                        "expected": expected,
                        "observed_at": time.perf_counter(),
                    })
            bad_seen.set()

threads = [threading.Thread(target=search_worker) for _ in range(4)]
for thread in threads:
    thread.start()

payload = "x" * 20_000  # Stored, non-indexed field; representative of larger product documents.
next_id = 1

# Stop submitting new batches as soon as a bad response is observed.
# The current import request is allowed to finish.
for batch in range(500):
    documents = [
        json.dumps({
            "id": f"product-{i}",
            "name": f"Product {i}",
            "created_at": i,
            "payload": payload,
        }, separators=(",", ":"))
        for i in range(next_id, next_id + 40)
    ]
    next_id += 40

    _, body = request(
        "POST",
        "/collections/products/documents/import?action=upsert&batch_size=40",
        "\n".join(documents),
        "text/plain",
    )
    rows = [json.loads(line) for line in body.splitlines()]
    assert all(row.get("success") for row in rows)

    if bad_seen.is_set():
        break

bad_seen.set()
for thread in threads:
    thread.join()

assert first_bad, "No incomplete response observed"

# No more writes are submitted. Poll the identical uncached query until it is consistent.
polls = 0
while True:
    polls += 1
    _, recovered, returned, expected = search()
    if returned == expected:
        recovered_at = time.perf_counter()
        break

print("bad_response", {
    "status": first_bad["status"],
    "found": first_bad["found"],
    "returned": first_bad["returned"],
    "expected": first_bad["expected"],
})
print("first_consistent_response", {
    "found": recovered["found"],
    "returned": returned,
    "expected": expected,
    "polls": polls,
    "recovered_after_ms": round(
        (recovered_at - first_bad["observed_at"]) * 1000,
        2,
    ),
})

The 20 KB field is stored but not indexed. It only makes the visibility window easy to observe with a server batch of 40; larger production product documents naturally have the same characteristic.

Results

Stress run

• 8,000 documents imported
• 0 import failures
• 4,000 searches returned HTTP 200
• 33 HTTP 200 searches returned fewer hits than min(found, per_page)
• 173 Document fetch error. Could not locate the JSON document... log entries
• 173 matching present in index but not in store entries

Examples:

{"status":200,"found":51,"returned":43}
{"status":200,"found":2440,"returned":249}
{"status":200,"found":3353,"returned":242}

Stop-importing-and-poll recovery test

Five independent trials stopped after the first bad response, allowed only the already in-flight 40-document import to finish, and then polled the identical uncached query:

• Recovery: min 5.89 ms, median 16.14 ms, max 25.78 ms.
• Every trial was consistent on the first poll.
• The in-flight import completed 0.85–7.54 ms after the incomplete response was observed.
• 42 Document fetch error entries were produced across these five trials.

This confirms that the uncached inconsistency is transient and closes when the same import finishes its store writes; no later repair/reconciliation process is required. These local timings are not a production latency bound. With use_cache=true, an incomplete HTTP 200 response can remain cached for the configured TTL after the underlying store has become consistent.

Three-node RC14 confirmation

The same test was also run against three 31.0.rc14 Docker nodes on one bridge network, using default lag thresholds and no cache or load balancer. Imports were sent to the current leader. Two search workers queried each node directly with the same 40-document batches and 20 KB stored/non-indexed field.

This does not need a second full reproduction script: the only variation from the script above is to define three API base URLs, start the same search worker twice per URL, and keep per-node mismatch/recovery state. In one mode, imports stop after the first mismatch anywhere. In a second mode, imports continue until every node has independently returned found > hits.length, with each node then polling the exact affected query until it returns all 40 hits.

Results:

At the first leader mismatch, each follower was commonly one Raft log entry behind. That initially gives the followers a consistently older view; once each follower applies the batch locally, it can enter the same index-before-store visibility gap. The cluster therefore creates staggered opportunities to observe the same local race, not a different race.

The Raft path supports this distinction: the leader submits the request for replication, then each node's on_apply enqueues the committed request into its local BatchedIndexer. In a three-node cluster, Raft needs a majority (2/3) to commit; it does not wait for all three nodes to finish their local collection/store work. Consensus latency occurs before the local index/store visibility gap, while follower scheduling, disk speed, and queued writes can stagger when that gap occurs on each node.

These are local lab timings, not production upper bounds. The three-node result strengthens the same fix direction below; it does not indicate a separate cluster-specific code path or justify a second issue.

Production prevalence

On a three-node v30.2 cluster, the exact search-materialization error currently occurs on all nodes:

• Last 24 hours: 1,808
• Last 7 days: 18,283
• Last 30 days: 78,746

The real workload sends up to 2,000 product documents per HTTP import request with Typesense's server-side batch_size=40, while searches run concurrently.

Code path / likely cause

In the RC14 source:

1. Collection::batch_index() first calls batch_index_in_memory().
2. That function publishes the batch through Index::batch_memory_index(); its collection lock is released when the function returns.
3. Only afterwards does batch_index() write each new document to RocksDB with store->batch_write().
4. Search locks the collection only while running the in-memory index search. During later result materialization it fetches each selected document from the store and continues on a miss.

This creates an index/store visibility gap for new documents and directly explains found > hits.length.

A possible fix direction is to publish stored JSON before making a new sequence ID searchable, or otherwise prevent readers from observing the newly indexed IDs until the corresponding store writes complete. The separate unlock between index selection and document materialization may also deserve a regression test for concurrent deletes.

Expected behavior

A successful search should represent a consistent view:

• It must not select a sequence ID whose document is unavailable.
• An HTTP 200 response should return min(found, per_page) hits when no grouping/filtering rule intentionally reduces the page.
• Successful concurrent imports must not produce present in index but not in store / Document fetch error logs.

Impact

Affected requests silently receive incomplete result pages. Counts, pagination and facets can claim documents that are absent from hits. If use_cache=true, the incomplete successful response can also be cached for the configured TTL, extending a millisecond-scale write race into a longer user-visible inconsistency.