PR #416 Review: perf: parallel file processing and batched upserts in mine()

URL: #416
Author: tbaldur | Status: Open | Branch: feat/parallel-mine → main
Scope: 1 file (mempalace/miner.py) | +106 / -17

Executive Summary

Goal: Speed up mine() by parallelizing file I/O + chunking and batching ChromaDB upserts.
Risk: HIGH — concurrency in the core data ingestion pipeline.
Recommendation: REQUEST CHANGES — Contains copy-paste duplicates, a data-loss path on exceptions, and logic duplication that creates maintenance risk.

Ratings

Issues

1. CRITICAL — Duplicate import line (copy-paste error)

Confidence: HIGH
Location: mempalace/miner.py, diff +17-18

The same import is added twice:

+from concurrent.futures import ThreadPoolExecutor, as_completed
+from concurrent.futures import ThreadPoolExecutor, as_completed

Python silently ignores duplicate imports, but this signals sloppy copy-paste and violates the project's code quality bar.

Fix: Remove one of the two lines.

2. CRITICAL — Duplicate constant definitions (copy-paste error)

Confidence: HIGH
Location: mempalace/miner.py, diff +60-63

Both constants are defined twice:

+BATCH_SIZE = 128  # chunks per upsert call
+MAX_WORKERS = min(32, (os.cpu_count() or 4) * 2)
+BATCH_SIZE = 128  # chunks per upsert call
+MAX_WORKERS = min(32, (os.cpu_count() or 4) * 2)

The second definitions silently overwrite the first. No functional harm today, but if someone later edits "the" definition and it's the wrong one, the change will be silently lost.

Fix: Remove the duplicate pair.

3. HIGH — Data loss on exception: unflushed batch is silently dropped

Confidence: HIGH
Location: mine() function — the for future in as_completed(futures) loop + final flush_batch()

future.result() is called without a try/except. If any file causes an unexpected exception in process_file_cpu, the exception propagates out of the loop, exits the with ThreadPoolExecutor block, and the final flush_batch() is never reached:

        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
            futures = { ... }
            for future in as_completed(futures):
                result = future.result()       # <-- unhandled exception
                # ...
                    if len(batch_ids) >= BATCH_SIZE:
                        flush_batch()
                # ...

        flush_batch()  # <-- NEVER REACHED if exception above

All chunks accumulated in batch_ids / batch_docs / batch_metas from successfully processed files that haven't been flushed yet are permanently lost. On a large codebase with BATCH_SIZE=128, this could be up to 127 chunks worth of work silently discarded.

Fix:

with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
    futures = { ... }
    for future in as_completed(futures):
        try:
            result = future.result()
        except Exception as exc:
            fp = futures[future]
            print(f"  ✗ {fp.name}: {exc}")
            continue
        # ... rest of loop ...

flush_batch()  # now always reached

Or at minimum, move flush_batch() into a finally block.

4. MEDIUM — Logic duplication between process_file and process_file_cpu

Confidence: HIGH

Both functions implement the same sequence: read file → strip → MIN_CHUNK_SIZE check → detect_room → chunk_text → build metadata with hashlib drawer IDs. The only difference is process_file calls add_drawer (single ChromaDB upserts) while process_file_cpu returns records for batching.

process_file is still used for the dry_run path. If someone later changes room detection, metadata fields, or chunking logic, they must update both functions or the dry_run output will diverge from real behavior.

Fix: Refactor process_file to delegate to process_file_cpu:

def process_file(filepath, project_path, collection, wing, rooms, agent, dry_run):
    if not dry_run and file_already_mined(collection, str(filepath), check_mtime=True):
        return 0, None

    result = process_file_cpu(filepath, project_path, wing, rooms, agent)
    if result is None:
        return 0, None

    source_file, room, records = result
    if dry_run:
        print(f"    [DRY RUN] {filepath.name} → room:{room} ({len(records)} drawers)")
        return len(records), room

    for drawer_id, content, meta in records:
        collection.upsert(documents=[content], ids=[drawer_id], metadatas=[meta])
    return len(records), room

5. MEDIUM — Behavioral change in files_skipped count

Confidence: HIGH

The output message says "Files skipped (already filed)" so the new behavior is arguably more correct, but it's an undocumented change. Files that are too small (< MIN_CHUNK_SIZE) or produce no chunks will no longer appear in the skip count.

Suggestion: Document this behavioral change in the PR description, or add a separate counter for truly empty/tiny files.

6. LOW — No progress feedback during pre-filtering

Confidence: MEDIUM

pending = [fp for fp in files if not file_already_mined(collection, str(fp), check_mtime=True)]

For large file lists (thousands of files), this fires a ChromaDB query per file with no output. Users may think the tool is hung.

Suggestion: Add a brief progress line before the list comprehension.

Thread Safety Analysis

The concurrency model (threads do CPU work, main thread does all ChromaDB I/O) is sound.

Flow Impact Analysis

• process_file: Still called in the dry_run path. No external callers found outside miner.py.
• mine(): Public API — signature unchanged. Callers unaffected.
• file_already_mined: Now called in a sequential pre-filter rather than inside the per-file loop. No functional impact.
• Blast radius: Low — changes are internal to the mine() hot path.

Guidelines Compliance

Summary

• Duplicate import/constants — fixed in this PR, won't appear in future changes
• Data loss on exception — wrapped the futures loop in try/finally so flush_batch() always runs
• Logic duplication — process_file now delegates to process_file_cpu; shared logic lives in one place
• files_skipped accuracy — now counts both already-mined and too-small/unreadable files
• Pre-filter progress — added "Checking N files for changes..." print before the mtime check loop

Great work on this PR, the parallelization pattern for miner.py is really clean and the batched upserts make a measurable difference.

One thing I noticed while testing this branch on a large Claude Code conversation archive (1.5 GB of JSONL in /.claude/projects/): convo_miner.py still uses the sequential per-chunk collection.add() path, so users mining conversation corpora via mempalace mine

I ported your pattern to convo_miner.py on my fork: FabioLissi/mempalace@fe74a7d (FabioLissi@fe74a7d). Same structure as your miner.py changes — process_convo_file_cpu worker function, ThreadPoolExecutor(max_workers=MAX_WORKERS) for parallel normalize/chunk, batched collection.upsert() with BATCH_SIZE=128, try/finally around the executor to guarantee final flush on exception.

Benchmark on a 10-file, 502-drawer sample from my Claude Code history (M-series Mac, MPS + BGE-base):
• Before any batching: 23.35s
• With your miner.py batching only: unchanged (convo path not covered)
• With the convo_miner.py port: 5.69s (4.1x speedup)

All 556 tests pass including tests/test_convo_miner.py::test_convo_mining which exercises the real ChromaDB write path end-to-end.

Totally happy to:
• Let you cherry-pick the commit onto this PR branch
• Open it as a separate follow-up PR after this one merges
• Or skip it entirely if you'd rather keep this PR scoped to miner.py

Whichever works best for you. Thanks again for the contribution!

Thank you for the analysis, changes made:

• flush_batch now takes explicit (ids, docs, metas) args

• Added idempotency assertion to the parallel test (re-mine same data, verify count unchanged)

• Pre-scan optimization: replaced the sequential file_already_mined() loop with build_mined_cache(), one bulk ChromaDB fetch + dict lookups. Greatly improved on my 32k+ files project.

• FabioLissi herry-picked your convo_miner.py port onto this branch — you're credited as author on the commit. Applied the same flush_batch hardening (explicit args) for consistency with the miner.py changes. Thank you very much for the contribution and the benchmark numbers!