diff --git a/docs/kb/modules/05-faza-mailowa-plan.md b/docs/kb/modules/05-faza-mailowa-plan.md index 00640ee..d602cbb 100644 --- a/docs/kb/modules/05-faza-mailowa-plan.md +++ b/docs/kb/modules/05-faza-mailowa-plan.md @@ -589,6 +589,33 @@ mała zmiana w serwisie, nie w `packages/kb-retrieval`). **Szacunek: 1 sesja (run w tle).** +### Decyzje operatora do Etapu B (2026-08-04) — przed runem + +Recon przed Etapem B (mirror archiwum na SOLARII żyje: 225 057 plików / 27 GB; RTT +SOLARIA→PIHA 0,83 ms; PIHA 140 GB wolne, baza 397 MB; M1 — `NODE_TYPE=lte_node` +na node-agencie SOLARII — zdeployowane, więc kontener Ollamy nie zniknie po +zatrzymaniu) wykazał dwie rzeczy do rozstrzygnięcia. Decyzje: + +1. **Run w plastrach po 50k** (`--limit 50000 --offset 0/50k/100k/150k/200k`), + log per plaster, `nice`/`ionice`. Powód: brak checkpointu (restart = ponowny + parse od początku listy, ~1 h) + nocne wyłączanie SOLARII. Plaster ≈ 25–40 min. + Tempo kolejnych plastrów po obserwacji PIHA po pierwszym. +2. **Circuit breaker w jobie: TAK** — `--max-embed-failures` (domyślnie 5), + abort z kodem wyjścia 2 po N kolejnych nieudanych batchach embed. Powód: + parse jest jednowątkowy i wyprzedza GPU, więc martwa Ollama (4 incydenty) + zamieniłaby 2-godzinny przebieg w 200k+ `chunks_errors` bez ani jednego + zapisu. Licznik zeruje się po udanym batchu. +3. **Dry-run całości pomijamy** — idempotencja i odwracalność flag newsletterowych + wystarczają; ewentualna kalibracja heurystyki na dekadzie 2010–2015 po fakcie, + na już zapisanych flagach. +4. Przełączenie domyślnego `mode` kb-query na `hybrid` (DoD (d)) — **poza zakresem + Etapu B**, osobny task po PASS regresji. + +Uwaga do czytania wyników: na pełnym korpusie `exit 1` jest spodziewany +(pojedyncze `parse_errors` — §1.5 dokumentuje ~9 maili na fallbacku compat32). +Werdyktem jest bilans i liczniki w linii `summary`, nie kod wyjścia. `exit 2` +oznacza co innego: backend embed padł, trzeba wznowić plaster po naprawie Ollamy. + ## 10. Krok 7 — IMAP/JMAP przyrostówka (zarys; szczegóły = osobny recon) Zakotwiczone w kb-00 jako etapy 3–4 (`jobs/fastmail-poller`, diff --git a/jobs/mail-body-ingest/README.md b/jobs/mail-body-ingest/README.md index 99fc1ce..cbd43ad 100644 --- a/jobs/mail-body-ingest/README.md +++ b/jobs/mail-body-ingest/README.md @@ -38,6 +38,11 @@ mail-body-ingest --dsn postgresql://kb:@piha:5433/kb --archive-root /home/os # Etap A pilot — last 12 months only (plan Decyzja 9): mail-body-ingest --dsn ... --since 2025-07-01 --apply > mail-ingest-etapA.log 2>&1 +# Etap B — full archive in 50k slices (plan §9; ORDER BY id is stable, so slices are +# reproducible, and idempotency covers their boundaries): +nice -n 10 ionice -c2 -n7 mail-body-ingest --dsn ... --apply \ + --limit 50000 --offset 0 > mail-ingest-etapB-0.log 2>&1 + # Smoke-test slice: mail-body-ingest --dsn ... --apply --limit 10 ``` @@ -89,14 +94,39 @@ Any non-zero `read_errors`/`parse_errors`/`missing_file`/`chunks_errors`/ `chunks_conflict_skipped`, or an unbalanced sum, makes the CLI exit 1 — same convention as `gmail-header-backfill`/`documents-ingest`'s `chunk_embed`. -## Ollama-offline tolerance +**Reading exit 1 on a full-corpus run**: it is a "look at this", not "the run failed". Across +225k mails a handful of `parse_errors` is expected (plan §1.5 documents ~9 mails that need the +compat32 fallback), and any one of them alone trips exit 1. The verdict is the balance and the +counters in the `summary` line, not the exit code. Exit 2 is different — see below. + +## Exit codes + +| Code | Meaning | +|---|---| +| 0 | Balanced, zero errors | +| 1 | Balanced-but-imperfect (any `parse_errors`/`missing_file`/`read_errors`/`chunks_errors`/`chunks_conflict_skipped`), an unbalanced sum, or an embedding-dimension abort | +| 2 | `--max-embed-failures` consecutive embed batches failed — the embed backend is down; re-run once it is back | + +## Ollama-offline tolerance and the circuit breaker A failed `embed_batch()` call is caught per-batch (`aiohttp.ClientError` -> the whole batch, up to `--batch-size` chunks, counts as `chunks_errors`; the run logs a warning and continues). Those chunks never enter the idempotency set, so a later re-run retries them automatically — -no separate checkpointing needed. Only a wrong embedding dimension -(`EmbeddingDimensionError`) aborts the entire run, since that would otherwise silently index -a vector that doesn't match `document_chunk.embedding VECTOR(1024)`. +no separate checkpointing needed. + +Tolerating a *flaky* backend is right; surviving a *dead* one is not. The archive is parsed +single-threaded ahead of the GPU, so on a full-corpus run (Etap B) a dead Ollama would let the +job chew through 200k+ mails at parse speed, mark every chunk `chunks_errors`, and throw away a +multi-hour pass. `--max-embed-failures` (default 5, `0` disables) therefore stops the run after +that many *consecutive* failed batches, with exit code 2; a single successful batch resets the +counter. Ollama@SOLARIA's known failure mode is total (container vanishes, network-detached — +4 incidents, plan §1.4/§7), so the breaker trips within seconds of it. On abort, pending +`entities[type=threading]` appends are flushed first: they don't depend on Ollama, they're +idempotent, and re-deriving them would mean re-reading the same 27 GB. + +Only a wrong embedding dimension is more severe (`EmbeddingDimensionError`, exit 1) — it aborts +immediately, since that would otherwise silently index a vector that doesn't match +`document_chunk.embedding VECTOR(1024)`. ## Idempotency @@ -114,13 +144,15 @@ pip install -e "jobs/mail-body-ingest[dev]" cd jobs/mail-body-ingest && pytest ``` -Pure unit tests (48), no DB/Ollama — `run()` is tested by monkeypatching `asyncpg.connect` +Pure unit tests (55), no DB/Ollama — `run()` is tested by monkeypatching `asyncpg.connect` and `aiohttp.ClientSession` with in-memory fakes, `.eml` bytes written to `tmp_path`. Covers: quote-strip (EN/PL/Outlook markers, bare `>` lines), HTML->text (style/script/blockquote/ gmail_quote skipping), newsletter classification, threading extraction, prefix building, body extraction (plain-preferred, HTML fallback, attachment-only), the typed/compat32 parse fallback, stats balance, idempotency (second run inserts nothing new), newsletter chunks -never reaching Ollama, Ollama-offline batch isolation, and dimension-mismatch abort. +never reaching Ollama, Ollama-offline batch isolation, dimension-mismatch abort, and the +circuit breaker (trips on N consecutive failures, resets on a success, disabled by `0`, +flushes pending threading on abort). ## Definition of Done diff --git a/jobs/mail-body-ingest/src/mail_body_ingest/ingest.py b/jobs/mail-body-ingest/src/mail_body_ingest/ingest.py index 594a566..bf35e5e 100644 --- a/jobs/mail-body-ingest/src/mail_body_ingest/ingest.py +++ b/jobs/mail-body-ingest/src/mail_body_ingest/ingest.py @@ -43,9 +43,16 @@ stays in the DB (reversible: `UPDATE ... SET excluded_reason=NULL WHERE excluded 'newsletter'` + a re-embed run un-flags them later, per plan Decyzja 4). Ollama-offline tolerance: a failed `embed_batch()` call is caught per-batch (`chunks_errors += -len(batch)`), never aborting the run — those chunks never enter the idempotency set, so a later -re-run naturally retries them. Only a wrong embedding dimension aborts the whole run -(`EmbeddingDimensionError`) — never silently indexes a mismatched vector. +len(batch)`) — those chunks never enter the idempotency set, so a later re-run naturally retries +them. Transient single-batch failures therefore cost nothing. What they must NOT do is let a run +survive a *dead* backend: with the archive parsed single-threaded ahead of the GPU, a full-corpus +run (Etap B, plan §9) would otherwise keep chewing through 200k+ mails at parse speed, marking +every chunk `chunks_errors`, and the whole multi-hour pass would have to be repeated. Hence the +circuit breaker: `--max-embed-failures` consecutive failed batches (default 5, `0` disables) +raise `EmbedBackendUnavailableError` and stop the run early with exit code 2, after flushing the +threading updates already earned. A single successful batch resets the counter. Only a wrong +embedding dimension is more severe (`EmbeddingDimensionError`, exit 1) — never silently indexes +a mismatched vector. """ from __future__ import annotations @@ -81,6 +88,17 @@ _log = structlog.get_logger(__name__) DEFAULT_ARCHIVE_ROOT = Path("/home/oskar/kb/mail/archive") DEFAULT_BATCH_SIZE = 64 THREADING_UPDATE_BATCH_SIZE = 500 +# Circuit breaker: consecutive failed embed batches that mean "the backend is down, not flaky". +# 5 x --batch-size chunks written off before stopping; Ollama@SOLARIA's known failure mode is +# total (container vanishes / network-detached), not partial, so this trips within seconds of it. +DEFAULT_MAX_EMBED_FAILURES = 5 +EXIT_EMBED_BACKEND_UNAVAILABLE = 2 + + +class EmbedBackendUnavailableError(RuntimeError): + """Raised when `--max-embed-failures` consecutive embed batches failed — the run stops instead + of parsing the rest of the archive into `chunks_errors`. Nothing is corrupted: failed chunks + were never inserted, so a re-run picks them up through the ordinary idempotency path.""" _CHUNK_INSERT_SQL = """ INSERT INTO document_chunk (envelope_id, chunk_index, text, embedding, model, excluded_reason) @@ -374,6 +392,7 @@ def _new_stats() -> dict: "chunks_errors": 0, "quoted_chars_stripped_total": 0, "embed_calls": 0, + "embed_batch_failures": 0, # diagnostic only — not part of the balance equations "embed_seconds_total": 0.0, "threading_updated": 0, "threading_already_present": 0, @@ -390,6 +409,7 @@ async def run( offset: Optional[int] = None, apply: bool = False, batch_size: int = DEFAULT_BATCH_SIZE, + max_embed_failures: int = DEFAULT_MAX_EMBED_FAILURES, ) -> dict: """Process one --limit/--offset (optionally --since-filtered) slice of `source='gmail'` envelopes. dry-run (apply=False): parse + quote-strip + classify + chunk + count, zero @@ -413,8 +433,10 @@ async def run( embed_buffer: list[tuple[str, int, str]] = [] threading_pending: list[tuple[str, str]] = [] + consecutive_embed_failures = 0 async def flush_embed_buffer() -> None: + nonlocal consecutive_embed_failures if not embed_buffer: return texts = [t for (_eid, _idx, t) in embed_buffer] @@ -425,9 +447,18 @@ async def run( except aiohttp.ClientError: _log.warning("skip.embed_batch_error", count=len(embed_buffer), exc_info=True) stats["chunks_errors"] += len(embed_buffer) + stats["embed_batch_failures"] += 1 + consecutive_embed_failures += 1 embed_buffer.clear() + if max_embed_failures and consecutive_embed_failures >= max_embed_failures: + raise EmbedBackendUnavailableError( + f"{consecutive_embed_failures} consecutive embed batches failed " + f"({ollama_url}) — stopping instead of parsing the rest of the archive " + f"into chunks_errors; re-run to retry the missed chunks" + ) return + consecutive_embed_failures = 0 stats["embed_calls"] += 1 stats["embed_seconds_total"] += elapsed for (eid, idx, chunk), embedding in zip(embed_buffer, embeddings): @@ -540,6 +571,12 @@ async def run( if apply: await flush_embed_buffer() await flush_threading() + except EmbedBackendUnavailableError: + # The threading appends already earned by the mails parsed so far are independent of + # Ollama and idempotent — flush them rather than making the next run re-derive them. + await flush_threading() + _log.error("embed_backend_unavailable_abort", ollama_url=ollama_url, **stats) + raise finally: if session is not None: await session.close() @@ -586,6 +623,12 @@ def main() -> None: help="Slice offset, ordered by envelope id (default: 0)") parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE, help=f"Ollama /api/embed batch size (default: {DEFAULT_BATCH_SIZE})") + parser.add_argument("--max-embed-failures", type=int, default=DEFAULT_MAX_EMBED_FAILURES, + metavar="N", + help=f"Abort (exit {EXIT_EMBED_BACKEND_UNAVAILABLE}) after N consecutive " + f"failed embed batches — a dead Ollama must not turn a multi-hour run into " + f"200k chunks_errors. 0 disables the breaker. " + f"Default: {DEFAULT_MAX_EMBED_FAILURES}.") parser.add_argument("--apply", action="store_true", help="Actually call Ollama, insert chunks, and update threading entities. " "Default is dry-run (parse + classify + chunk + count only).") @@ -610,11 +653,17 @@ def main() -> None: offset=args.offset, apply=args.apply, batch_size=args.batch_size, + max_embed_failures=args.max_embed_failures, ) ) except EmbeddingDimensionError as exc: _log.error("dim_mismatch_abort", error=str(exc)) sys.exit(1) + except EmbedBackendUnavailableError as exc: + # Distinct exit code: unlike exit 1 (which a full-corpus run can legitimately reach on a + # handful of parse_errors), this one means "nothing more will succeed until Ollama is back". + _log.error("embed_backend_unavailable", error=str(exc)) + sys.exit(EXIT_EMBED_BACKEND_UNAVAILABLE) mode = "APPLY" if args.apply else "DRY-RUN" avg_embed_ms = ( diff --git a/jobs/mail-body-ingest/tests/test_ingest.py b/jobs/mail-body-ingest/tests/test_ingest.py index 60ce719..4e74e2d 100644 --- a/jobs/mail-body-ingest/tests/test_ingest.py +++ b/jobs/mail-body-ingest/tests/test_ingest.py @@ -17,6 +17,7 @@ import pytest from kb_retrieval.embed import EmbeddingDimensionError from mail_body_ingest.ingest import ( _CHUNK_INSERT_SQL, + EmbedBackendUnavailableError, _decode_jsonb, _has_threading, build_prefix, @@ -388,17 +389,26 @@ class _FakeTagsResponse: class _FakeOllamaSession: - def __init__(self, dim=1024, fail_batches=False, health_up=True): + def __init__(self, dim=1024, fail_batches=False, health_up=True, fail_pattern=None): self._dim = dim self._fail_batches = fail_batches self._health_up = health_up + # Per-request failure sequence (True = this batch fails); exhausting it falls back to + # `fail_batches`. Lets a test interleave failures and successes to exercise the breaker's + # "consecutive" semantics rather than a plain total. + self._fail_pattern = list(fail_pattern) if fail_pattern is not None else None self.requests: list[dict] = [] self.closed = False + def _should_fail(self) -> bool: + if self._fail_pattern: + return self._fail_pattern.pop(0) + return self._fail_batches + def post(self, url, json): assert url.endswith("/api/embed") self.requests.append({"url": url, "json": json}) - if self._fail_batches: + if self._should_fail(): return _FakeEmbedResponse({}, status=500) n = len(json["input"]) return _FakeEmbedResponse({"embeddings": [[0.01] * self._dim for _ in range(n)]}) @@ -558,6 +568,76 @@ class TestRun: + stats["chunks_conflict_skipped"] + stats["chunks_errors"] ) + def _write_n_mails(self, tmp_path, n: int) -> list: + rows = [] + for i in range(n): + eid = f"m{i}@x" + _write_eml(tmp_path, f"gmail/2025/08/{eid}.eml", _plain_eml( + {"From": "a@b.com", "Subject": f"hi {i}", "Date": "Fri, 01 Aug 2025 10:00:00 +0000"}, + f"hello there number {i}", + )) + rows.append(_env_row(eid, f"hi {i}")) + return rows + + async def test_breaker_aborts_after_consecutive_embed_failures(self, tmp_path, monkeypatch): + """A dead backend must stop the run, not let it parse the rest of the archive into + chunks_errors (plan §9 Etap B: 200k+ mails behind a single-threaded parse).""" + conn = _FakeConn(envelopes=self._write_n_mails(tmp_path, 8)) + ollama = _FakeOllamaSession(dim=1024, fail_batches=True) + self._patch(monkeypatch, conn, ollama) + + with pytest.raises(EmbedBackendUnavailableError): + await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True, + batch_size=1, max_embed_failures=5) + + # Stopped at the 5th failed batch — mails 6-8 were never parsed, let alone embedded. + assert len(ollama.requests) == 5 + assert conn.execute_calls == [] # nothing inserted: failed chunks stay retryable + + async def test_breaker_counter_resets_on_successful_batch(self, tmp_path, monkeypatch): + """"Consecutive", not "total" — flaky batches interleaved with successes must not trip it.""" + conn = _FakeConn(envelopes=self._write_n_mails(tmp_path, 5)) + ollama = _FakeOllamaSession(dim=1024, fail_pattern=[True, True, False, True, True]) + self._patch(monkeypatch, conn, ollama) + + stats = await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True, + batch_size=1, max_embed_failures=3) + + assert len(ollama.requests) == 5 # ran to completion + assert stats["chunks_errors"] == 4 + assert stats["chunks_inserted"] == 1 + assert stats["embed_batch_failures"] == 4 + assert stats["chunks_total"] == ( + stats["chunks_inserted"] + stats["chunks_newsletter_flagged"] + stats["chunks_already_embedded"] + + stats["chunks_conflict_skipped"] + stats["chunks_errors"] + ) + + async def test_breaker_disabled_with_zero(self, tmp_path, monkeypatch): + conn = _FakeConn(envelopes=self._write_n_mails(tmp_path, 6)) + ollama = _FakeOllamaSession(dim=1024, fail_batches=True) + self._patch(monkeypatch, conn, ollama) + + stats = await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True, + batch_size=1, max_embed_failures=0) + + assert len(ollama.requests) == 6 + assert stats["chunks_errors"] == 6 + + async def test_breaker_abort_flushes_pending_threading(self, tmp_path, monkeypatch): + """Threading appends are Ollama-independent and idempotent — the abort keeps them rather + than making the next run re-derive them from the same 27 GB read.""" + conn = _FakeConn(envelopes=self._write_n_mails(tmp_path, 8)) + ollama = _FakeOllamaSession(dim=1024, fail_batches=True) + self._patch(monkeypatch, conn, ollama) + + with pytest.raises(EmbedBackendUnavailableError): + await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True, + batch_size=1, max_embed_failures=5) + + assert len(conn.executemany_calls) == 1 + # The 5 mails processed before the breaker tripped, none of the 3 after it. + assert len(conn.executemany_calls[0][1]) == 5 + async def test_dimension_mismatch_aborts(self, tmp_path, monkeypatch): _write_eml(tmp_path, "gmail/2025/08/m1@x.eml", _plain_eml( {"From": "a@b.com", "Subject": "hi", "Date": "Fri, 01 Aug 2025 10:00:00 +0000"}, "hello there"