A batch waits for its slowest worker, so one slow source stalls indexing and blocks the next fetch

[bakiburakogun]

Describe the Bug

A batch makes no progress until its slowest worker returns, so one slow source stalls the whole batch and no new queue items are fetched meanwhile.

files_indexing_thread submits the chunks and then walks the futures in submission order:

file_futures = [executor.submit(_load_sources, chunk) for chunk in file_chunks]
...
for i, future in enumerate(file_futures):
    LOGGER.debug('Waiting for file chunk %d/%d future to complete', i + 1, len(file_futures))
    files_result.update(future.result())

Results are reported and the queue rows deleted only after that loop, so nothing a fast worker produced is committed until the slowest one is done.

That is fine when workers finish in comparable times. It is expensive when they do not, and they often do not: in our runs the workers of a single batch finished anywhere between 1.4 s and 300 s. The 300 s ones were an embedding request that never got a response, retried up to three times against request_timeout.

How visible this is

Doubling doc_indexing_batch_size from 32 to 64 halved our throughput, from ~37 to ~15–20 documents/min. Each worker then carries twice as many sources, so the chance that a batch contains a slow one goes up and the tail it imposes doubles. We reverted it.

The default request_timeout of 1800 s compounds it: before we lowered it, a single unresponsive embedding request held a batch for ~35 minutes (3 attempts plus back-offs), during which all four of our backend instances sat idle. Lowering it to 30 s — healthy requests answer in about 6 s — was the single largest improvement we made, from ~5 to ~60–100 documents/min.

Suggested direction

Two things would help independently:

• Consume the futures as they complete (concurrent.futures.as_completed) and report results incrementally, so a slow worker delays only its own sources rather than the batch and the next fetch.
• Reconsider the default request_timeout. 1800 s is a long time to hold a worker for a live indexing pipeline; something in the tens of seconds, with the retry budget in mind, matches how the embedding services actually behave.

I have not sent a patch for the first one because changing the batch bookkeeping touches the result and deletion paths, and I would rather not guess at the invariants you want there. Happy to write it if you tell me which shape you prefer.

Setup Details

Nextcloud version: 34.0.3
context_chat / context_chat_backend: 5.4.0 / 5.4.1
Deployment: manual-install deploy daemon, four backend instances, external embedding service, PostgreSQL 18 with pgvector, ~1.5 M eligible files

[kyteinsky]

hello,
thanks, I'll consider this a feature request.

yeah a system to continuously consume and process batches, with results reported for each batch independently would be quite nice to have. We haven't thought of what the change could look like yet so suggestions are most welcome.

Reconsider the default request_timeout

the timeout of 30 mins was set considering a batch could have 100s of MBs of content which would run on the internal embedding service which is shipped with the app by default.
it is changable easily in /nc_app_context_chat_backend_data/config.yaml.

[bakiburakogun]

Thanks for taking it. Here is the shape I would suggest, kept as close to the current code as I could manage.

Three things move, the rest stays.

The coupling comes from the executor living inside the loop. The pool is built per batch, all chunks are submitted, the futures are walked in submission order, and the with block joins on exit. So from the moment the first chunk finishes until the last one does, the pool is idle and no fetch happens.

1. Hoist the executor out of the while loop.
2. Wait for whichever chunk finishes first instead of walking the futures in order.
3. Scope the DELETE to the chunk that just finished rather than to the whole batch.

MAX_INFLIGHT = 2 * file_parsing_cpu_count

with ThreadPoolExecutor(
    max_workers=file_parsing_cpu_count,
    thread_name_prefix='IndexingPool',
) as executor:
    inflight: dict[Future, tuple[Kind, dict[int, SourceItem | ReceivedFileItem]]] = {}

    while not THREAD_STOP_EVENT.is_set():
        ...  # enabled-state and connection checks unchanged

        # keep the pool fed
        while len(inflight) < MAX_INFLIGHT:
            batch = fetch_batch()   # today's GET, validation and chunking, unchanged
            if batch is None:
                break
            for kind, chunk in batch:
                inflight[executor.submit(_load_sources, chunk)] = (kind, chunk)

        if not inflight:
            sleep(POLLING_COOLDOWN)
            continue

        done, _ = wait(inflight, return_when=FIRST_COMPLETED)

        for fut in done:
            kind, chunk = inflight.pop(fut)
            report_and_delete(kind, fut.result())   # today's DELETE body, one chunk's ids

_load_sources, the chunking, the error classification, the retryable and non-retryable split, and the DELETE payload all stay as they are. No API change, no new config key, no schema change.

It does not raise the resource ceiling. _load_sources goes through exec_in_proc, so a chunk is a subprocess and max_workers is what bounds how many exist at once. That bound does not move. The pool simply stops being idle, so peak memory and CPU stay where they are today.

On how much to keep in flight. getFromQueue hands out a lease rather than a reservation that gets released: locked_at is set on fetch and the row only becomes eligible again after a day. So anything fetched and not deleted waits a day if the backend goes away, which argues for a bound rather than fetching freely. Two chunks per worker keeps roughly one batch of work queued behind the running one, which is enough to cover a refill without growing the number of rows a crash would strand. It is derived from the existing worker count, so it does not need to become configurable.

Invariants. The work in a batch already completes out of order across chunks; what changes is that the rows are removed as each chunk lands instead of all at the end. Retryable errors are still left in the queue untouched. The one place I would look at twice is existsQueueItemsUpToDbId, since the deleted set becomes a moving subset, though it already has holes today because retryable failures stay behind.

One side effect worth having: countLocked, the running figure, becomes a steady number instead of a sawtooth, which makes it usable as a progress signal.

On request_timeout — your reasoning makes sense and I would leave the default where it is. What made 1800 s expensive for us was not the number but that one request holding a worker held the batch. Once a slow chunk only delays its own sources, a long timeout costs one worker for its duration instead of stalling the pipeline. So I would treat these as one fix rather than two, and drop that half of my suggestion.

Happy to write this if the shape looks right. The one call I would want from you before starting is whether report_and_delete should fire per completed chunk or coalesce a few completions: the first is simpler to follow, the second is kinder to the PHP side when the worker count is high.