[FEATURE] Write a completion marker when every chunk for a document stored

[noel-improv]

Description

Nothing recorded whether a document's chunks all reached S3, so a truncated prefix and a short document read back identically. S3ChunkUploader._drain caught each chunk's exception, logged it and returned nothing, which #507 documented rather than fixed.

_drain now reports whether every upload landed. A document whose chunks all stored gets a marker holding its chunk ids, their count and a hash over them, written last. A document with a failed chunk carries no marker and is still yielded, exactly as today.

Related issue: #325.

Changes

• _drain returns whether all of a document's uploads succeeded.
• _write_completion_marker writes the marker after that check passes. A failed marker write is logged rather than raised. The chunks themselves uploaded, so stopping the run costs more than the re-stage a missing marker causes.
• Both downloaders exclude marker objects. Neither filtered by extension, and TextNode.from_json returns a node with a generated uuid and empty text for a marker rather than rejecting it, so an unfiltered listing would add one extra node to every read of that document.
• A document with nothing to write no longer creates a prefix. Previously an index-only document produced a prefix holding just a marker, which reads back as a document with no nodes, whose source_id() is None, and a re-stage cannot build a path from None.

Three duplications collapsed rather than extended:

	before	after
index-key node filter	two forms, one convoluted	written_nodes(doc)
node-id hash	inline in _doc_suffix	node_ids_hash()
KMS vs AES256	four copies	EncryptedPut._put

Why the marker name contains a digest

An auto-tuned run emits one source as several SourceDocuments, which share a prefix. With a fixed marker name, the last one written replaces the others. With two documents where the second loses a chunk:

objects : ['_COMPLETE', 'c1.json', 'c2.json', 'c3.json']
marker  : {'chunk_ids': ['c1','c2'], 'count': 2, ...}

Three of four objects are present, with a marker that reports the document as complete. #518 addressed the same collision for the document key by including a hash of the node ids; the marker name now does the same.

Five hex characters is 20 bits, the same budget _doc_suffix uses, and the same population: the SourceDocuments sharing one prefix. Brute-forced against realistic chunk ids, the first name collision appears at 1,672 documents in one prefix, with birthday odds of 0.47% at 100 and 38% at 1,000. So the bound is tens of documents per prefix, which is what the auto-tune path produces when it slices a source into max_batch_size buckets. A name collision loses a completeness record rather than data. The remaining marker still lists one document's chunks, so a reader comparing coverage re-stages rather than reporting the document complete.

Testing

• Unit tests added/updated
• Integration tests added (as appropriate)
• Existing tests pass (pytest)
• Tested manually

19 new tests. 2154 pass across tests/unit.

Each change was checked by mutation testing, not only by the tests passing. Reverting to a fixed marker name fails 2 tests; removing the ServerSideEncryption headers fails 3; yielding an empty document immediately rather than queueing it fails 2. The encryption check matters because that branch had no assertions at all before this, so collapsing four copies of it was otherwise unverifiable.

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)

Verified against real S3

Uploaded a two-chunk document to noel-graphrag-test and read it back:

coll/aws::live:test/_COMPLETE-23840          <- marker
coll/aws::live:test/aws::live:test:a.json
coll/aws::live:test/aws::live:test:b.json

marker encryption : AES256
read back         : ['aws::live:test:a', 'aws::live:test:b']
phantom chunk?    : False

Two things worth your call

Every ingest now writes one extra object per document, and nothing reads it yet. The restart work is expected to be enabled by a setting, but that setting does not exist yet, and adding one with no code reading it would be configuration no caller can use. The write is therefore unconditional. Say if you would rather it waited until the restart setting exists.

The encryption consolidation is separable from the feature. Reverting the marker would revert it too. I folded it in because writing the marker otherwise meant a fifth copy of the same KMS branch, but it touches S3DocUploader, which this feature does not otherwise change. Happy to split it into its own PR.

Known limitations

Nothing reads the marker yet, so chunk_ids, count and content_hash are stored and never validated. Read behaviour is deliberately out of scope here.

The marker name is the first five characters of the content_hash in its own body, so the two carry the same value at different precision. The name has to be short; the body field is what the acceptance criteria ask for.

A stale marker from an earlier run is not deleted. With per-document names a re-chunked document writes a different name and leaves the old one, which a reader catches by comparing the hash against a live listing. That comparison belongs with the read path.

S3DocUploader writes no marker, but S3DocDownloader skips one. Both uploaders write into the same prefix shape, so the skip is defensive rather than dead.

The marker PUT runs on the consumer's thread rather than the pool, adding one serialized round trip per document. The drain already blocks that thread, so this is additive rather than a change in shape.

[mykola-pereyma]

LGTM