[FIX] Surface a failed document upload instead of hanging S3 staging

[noel-improv]

Description

A failed document upload during S3 staging either hangs the run with no exception, or is reported as successfully staged. Both are silent, so a failed upload looks like a slow run or a clean one.

_upload_batch polls until it has seen one item per submitted document and treats queue.Empty as "keep waiting", never checking whether its producer is still alive. Three except Exception: log blocks could each leave that count unreachable.

What stops it firing today is _upload_doc swallowing its own S3 exception and returning None. The callback puts that None, the count advances, and the document is yielded and counted in the "Finished writing N source documents" total. The two defects are a pair: removing the swallow alone turns a silent success into a hang.

Changes

One invariant — every submitted document puts exactly one item on the queue.

• _get_callback_fn always puts, a failure marker when the upload raised, and releases the semaphore in a finally.
• _submit_proxy reports whether it submitted, so the publisher counts only what the consumer will see.
• _doc_publisher puts its count in a finally and catches BaseException, so a producer that dies still releases the consumer.
• _upload_batch breaks out when the queue is empty and the producer is gone, and raises the first failure once the batch drains.
• _upload_doc stops swallowing.

The queue poll drops from 60s to 1s. It paces the liveness check rather than the work, so a dead producer is noticed in about a second instead of a minute.

Problem

Related issue (if any): #

Introduced in 4b8743c0 ("Improved batch extract and S3BasedDocs for large ingests") rather than by a targeted change. #418 is the same symptom in the same path from a different cause and is already fixed.

Testing

• Unit tests added/updated
• Integration tests added (as appropriate)
• Existing tests pass (pytest) — 2,112 in tests/unit
• Tested manually (describe below)

Five tests, red before green, covering a failing document mid-batch, a dead producer, and the all-succeed path. No test previously failed a document mid-batch, which is the gap that hid this.

Checked against a real boto3 S3 client rather than a mock: put_object returns a genuine ClientError, the error propagates to the caller, no document is reported as staged, and the call returns instead of hanging.

One pre-existing failure is unrelated and reproduces on a clean tree: test_integ_dependency_compatibility.py errors with No module named pip in this venv.

Checklist

• Code follows existing style and conventions
• License headers present on new files
• Documentation updated (if applicable)
• No breaking changes (or clearly documented)

A caller that relied on a failed upload being silently skipped will now see the exception. That is the intended change.

[mykola-pereyma]

LGTM

[acarbonetto]

should this catch BaseException like we do in _doc_publisher?

[acarbonetto]

Only the first upload failure in a batch is surfaced, can we store and report all document failures?

[acarbonetto]

breaking out here leaves failure as None and count < target_count, so the generator just ends normally. The if failure is not None: raise failure below is skipped, and upload() goes on to report "Finished writing N source documents" for a batch that was actually truncated.

That's the exact silent partial-success mode this PR set out to eliminate — it's just narrowed to the abnormal path where the producer dies before publishing its count (e.g. hard-killed thread, or put(count) itself never completing).

Suggest raising instead of returning silently when we break on a dead producer:

except queue.Empty:
    # A producer that died before putting its count would otherwise
    # keep the consumer here for the rest of the run.
    if not thread.is_alive():
        if failure is None:
            failure = RuntimeError(
                f'Upload producer died before completing '
                f'[count: {count}, target_count: {target_count}]'
            )
        break

That way a truncated batch surfaces as an error rather than a success, keeping the guarantee consistent with the normal-path _UploadFailed handling.