Compare commits
2 commits
efd8d2fb01
...
17640526ef
| Author | SHA1 | Date | |
|---|---|---|---|
|
17640526ef | ||
|
|
7c40d1dc32 |
|
|
@ -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`,
|
||||
|
|
@ -621,16 +648,16 @@ Zakotwiczone w kb-00 jako etapy 3–4 (`jobs/fastmail-poller`,
|
|||
|
||||
## 12. Plan implementacji (kolejność = zależności)
|
||||
|
||||
| # | Krok | Zależy od | Szacunek |
|
||||
|---|---|---|---|
|
||||
| 0 | Chunker → `packages/kb-mail` | — | 0,5 sesji |
|
||||
| 1 | `embed_batch` w kb-retrieval | — | 0,5 sesji |
|
||||
| 2 | Job `mail-body-ingest` | 0, 1 | 2 sesje |
|
||||
| 3 | Tryb hybrid (kb-retrieval + kb-query) | — (równolegle z 2) | 1 sesja |
|
||||
| 4 | rsync + Etap A (12 mies.) + kalibracja | 2 | 1 sesja |
|
||||
| 5 | Bramka jakościowa (eval mailowy + regresja) | 3, 4 + zapytania od operatora | 1 sesja |
|
||||
| 6 | Etap B (pełne archiwum) + regresja + obserwacja PIHA | 5 = PASS | 1 sesja |
|
||||
| 7 | Recon przyrostówki IMAP/JMAP | — (po 6) | 1 sesja (poza DoD fazy) |
|
||||
| # | Krok | Zależy od | Szacunek | Stan | Dowód (2026-08-04) |
|
||||
|---|---|---|---|---|---|
|
||||
| 0 | Chunker → `packages/kb-mail` | — | 0,5 sesji | **WYKONANE** | `348ce10`; `packages/kb-mail/src/kb_mail/chunking.py` + `tests/test_chunking.py` |
|
||||
| 1 | `embed_batch` w kb-retrieval | — | 0,5 sesji | **WYKONANE** | `51998fd`; `kb_retrieval/embed.py:61` (`embed_batch`) + `tests/test_embed.py` |
|
||||
| 2 | Job `mail-body-ingest` | 0, 1 | 2 sesje | **WYKONANE** | `ad0ef40` (job), `a95524c` (README), `fc5c698` (fix html_to_text); `jobs/mail-body-ingest/` + `tests/test_ingest.py` |
|
||||
| 3 | Tryb hybrid (kb-retrieval + kb-query) | — (równolegle z 2) | 1 sesja | **WYKONANE** | `a640cf1`; `kb_retrieval/retrieval.py:112` (`hybrid_retrieve`), `:195` (`hybrid_query`), `kb-query/app/main.py:119` (`mode` pattern). Uwaga: domyślny `mode` to nadal `cascade` — przełączenie to follow-up z §8, nie część Kroku 3 |
|
||||
| 4 | rsync + Etap A (12 mies.) + kalibracja | 2 | 1 sesja | **WYKONANE** | §7 „Wynik Etapu A" (run na żywo 2026-07-23); potwierdzone na żywej bazie 2026-08-04: `document_chunk` gmail = 33 871 (6 398 z embeddingiem + 27 473 `newsletter`) — zgodne co do sztuki z tabelą §7 |
|
||||
| 5 | Bramka jakościowa (eval mailowy + regresja) | 3, 4 + zapytania od operatora | 1 sesja | **WYKONANE** (PASS) | §8 „Wynik bramki"; `56f64e9` (eval + queries.yaml dla hybrid), `bce635c` (`mail_hit@3`, próg N2, werdykt PASS), `71eb264` (`--transport http`) |
|
||||
| 6 | Etap B (pełne archiwum) + regresja + obserwacja PIHA | 5 = PASS | 1 sesja | **OTWARTE** | Brak commitu, brak sekcji z wynikiem w tym dokumencie; żywa baza pokazuje wyłącznie wolumen Etapu A (33 871 chunków gmail vs oczekiwane ~496k), więc run bez `--since` nie był wykonany |
|
||||
| 7 | Recon przyrostówki IMAP/JMAP | — (po 6) | 1 sesja (poza DoD fazy) | **OTWARTE** | Brak `jobs/fastmail-poller` / `jobs/gmail-imap-poller`, brak dokumentu reconu; IMAP/JMAP występuje wyłącznie jako zarys w §10 i w `kb-00-overview.md` |
|
||||
|
||||
**Kryterium ukończenia fazy mailowej:** (a) pełny korpus gmail zchunkowany
|
||||
(bilans domknięty, `parse_errors` na poziomie pojedynczych sztuk jak
|
||||
|
|
|
|||
|
|
@ -38,6 +38,11 @@ mail-body-ingest --dsn postgresql://kb:<pw>@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
|
||||
|
||||
|
|
|
|||
|
|
@ -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 = (
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
Loading…
Reference in a new issue