poimen-memory/tasks/M8.2-dual-write-indexer.md

M8.2 — Dual-write indexing pipeline

Field	Value
Phase	M8 — Hybrid Search
Size	M — 1–2 days
Status	⬜
Flags	—
Spec	inlined below
Blocks	M8.4, M8.5
Depends	M8.1 (OpenSearch running), M2.4 (pgvector repo)

Goal

When a document is ingested, write to both pgvector (embedding) and OpenSearch (raw text) atomically. Same chunk_id in both stores. If one write fails, log error but don't block the other — eventual consistency, not transactions.

Design

Unified ID mapping: Both stores use the same chunk_id (UUID). The ingest worker generates the ID once, writes to both.

Chunking policy: 512-token chunks with 10% (51-token) overlap. Deterministic — same input always produces same chunks with same IDs.

Dual write sequence:

1. Chunk document (heading-boundary or fixed-size).
2. Generate embedding via LLM.
3. Write to pgvector: INSERT INTO chunks (id, embedding, text, source, project, level, breadcrumb).
4. Write to OpenSearch: PUT vault-{project}/_doc/{chunk_id} with {content, source, level, breadcrumb, project_id, indexed_at}.
5. If OpenSearch write fails: log warning, mark chunk as opensearch_pending=true in pgvector. Background retry later.

OpenSearch index mapping:

{
  "content": {"type": "text", "analyzer": "standard", "boost": 2.0},
  "section_title": {"type": "text", "boost": 1.5},
  "breadcrumb": {"type": "keyword"},
  "source": {"type": "keyword"},
  "project_id": {"type": "keyword"},
  "level": {"type": "keyword"},
  "indexed_at": {"type": "date"}
}

Deduplication: Before writing, check chunk_hash (SHA256 of text). If hash exists and is_indexed=true in both stores, skip.

Steps

1. Add opensearch_pending boolean column to chunks table (migration).
2. Update IngestWorker::process_ingest() to call OpenSearch after pgvector write.
3. Make OpenSearchClient::index_document() public, fix method signature.
4. Add background task: retry opensearch_pending=true chunks every 5 minutes.
5. Add dedup check before dual write.

Acceptance

1. mem ingest --dry-run on a test doc shows chunks written to both stores.
2. Same chunk_id exists in both SELECT id FROM chunks and GET vault-*/_doc/{id}.
3. Kill OpenSearch mid-ingest: pgvector write succeeds, chunk marked opensearch_pending=true.
4. Restart OpenSearch: background retry picks up pending chunks within 5 minutes.
5. Re-ingest same document: dedup skips already-indexed chunks (0 new writes).

Verify

# Ingest a test document
cargo run -- ingest --project test --source fixtures/refcorpus/small.md

# Check pgvector
psql -c "SELECT id, source, opensearch_pending FROM chunks WHERE project='test'"

# Check OpenSearch
curl -k -H "Authorization: Bearer $TOKEN" \
  https://opensearch-internal:9200/vault-test/_search | jq '.hits.total'

# IDs must match

False pass: Both stores have data but with different IDs — the join on chunk_id finds zero matches. Assert SELECT count(*) FROM chunks WHERE id IN (opensearch_ids) equals total indexed.

Artifacts

• Modified crates/mem-cli/src/ingest_worker.rs
• Modified crates/mem-store/src/lib.rs (migration)
• Modified crates/mem-cli/src/opensearch_client.rs