feat: optimize database operations with batch upserts and mark inactive functionality

common/db.py

@@ -23,40 +23,78 @@ CREATE VIRTUAL TABLE IF NOT EXISTS papers_fts USING fts5(
 """
 
 
+# WAL + synchronous=NORMAL lets writers commit without an fsync per commit
+# (fsync only happens at checkpoint). On slow-fsync filesystems like WSL's
+# /mnt/c, this is the difference between minutes and hours for a full sync.
+_PRAGMAS = [
+    "PRAGMA journal_mode=WAL",
+    "PRAGMA synchronous=NORMAL",
+    "PRAGMA temp_store=MEMORY",
+]
+
+
 def init_db(path: str) -> sqlite3.Connection:
     conn = sqlite3.connect(path, check_same_thread=False)
     conn.row_factory = sqlite3.Row
     conn.executescript(SCHEMA)
+    for pragma in _PRAGMAS:
+        conn.execute(pragma)
     conn.commit()
     return conn
 
 
+def upsert_papers(conn: sqlite3.Connection, papers: list[dict]) -> None:
+    """Upsert many papers in a single transaction.
+
+    Batched so we pay the commit cost once per batch instead of once per
+    paper; the per-paper commit (and its fsync) was the dominant cost of a
+    full sync. `upsert_paper` delegates here with a one-element list.
+    """
+    if not papers:
+        return
+    rows = []
+    fts_rows = []
+    for paper in papers:
+        params = {**paper, "active": int(paper["active"])}
+        params.setdefault("faiss_id", None)
+        rows.append(params)
+        fts_rows.append((paper["id"], paper["title"], paper["abstract"]))
+    with conn:  # one transaction: commit on success, rollback on error
+        conn.executemany(
+            """
+            INSERT INTO papers (id, title, abstract, authors, venue, year, url, active, faiss_id)
+            VALUES (:id, :title, :abstract, :authors, :venue, :year, :url, :active, :faiss_id)
+            ON CONFLICT(id) DO UPDATE SET
+                title=excluded.title, abstract=excluded.abstract, authors=excluded.authors,
+                venue=excluded.venue, year=excluded.year, url=excluded.url, active=excluded.active,
+                faiss_id=COALESCE(excluded.faiss_id, papers.faiss_id)
+            """,
+            rows,
+        )
+        conn.executemany("DELETE FROM papers_fts WHERE id = ?", [(r["id"],) for r in rows])
+        conn.executemany(
+            "INSERT INTO papers_fts (id, title, abstract) VALUES (?, ?, ?)",
+            fts_rows,
+        )
+
+
 def upsert_paper(conn: sqlite3.Connection, paper: dict) -> None:
-    params = {**paper, "active": int(paper["active"])}
-    params.setdefault("faiss_id", None)
-    conn.execute(
-        """
-        INSERT INTO papers (id, title, abstract, authors, venue, year, url, active, faiss_id)
-        VALUES (:id, :title, :abstract, :authors, :venue, :year, :url, :active, :faiss_id)
-        ON CONFLICT(id) DO UPDATE SET
-            title=excluded.title, abstract=excluded.abstract, authors=excluded.authors,
-            venue=excluded.venue, year=excluded.year, url=excluded.url, active=excluded.active,
-            faiss_id=COALESCE(excluded.faiss_id, papers.faiss_id)
-        """,
-        params,
-    )
-    conn.execute("DELETE FROM papers_fts WHERE id = ?", (paper["id"],))
-    conn.execute(
-        "INSERT INTO papers_fts (id, title, abstract) VALUES (?, ?, ?)",
-        (paper["id"], paper["title"], paper["abstract"]),
-    )
-    conn.commit()
+    upsert_papers(conn, [paper])
+
+
+def mark_inactive_ids(conn: sqlite3.Connection, paper_ids: list[str]) -> None:
+    """Mark many papers inactive in a single transaction (batched for the
+    same reason as `upsert_papers`)."""
+    if not paper_ids:
+        return
+    ids = [(pid,) for pid in paper_ids]
+    with conn:
+        conn.executemany("UPDATE papers SET active = 0 WHERE id = ?", ids)
+        conn.executemany("DELETE FROM papers_fts WHERE id = ?", ids)
 
 
 def mark_inactive(conn: sqlite3.Connection, paper_id: str) -> None:
-    conn.execute("UPDATE papers SET active = 0 WHERE id = ?", (paper_id,))
-    conn.execute("DELETE FROM papers_fts WHERE id = ?", (paper_id,))
-    conn.commit()
+    mark_inactive_ids(conn, [paper_id])
 
 
 def get_paper_id_by_faiss_id(conn: sqlite3.Connection, faiss_id: int) -> str | None:

common/vector_index.py

@@ -4,7 +4,7 @@ import faiss
 import numpy as np
 
 
-def create_index(dim: int = 1024) -> faiss.IndexFlatIP:
+def create_index(dim: int) -> faiss.IndexFlatIP:
     return faiss.IndexFlatIP(dim)
 
 

pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     "pydantic>=2.6",
     "pydantic-settings>=2.2",
     "numpy>=1.26",
+    "tqdm>=4.66",
     "acl-anthology>=0.5",
 ]
 

sync/embeddings_client.py

@@ -1,20 +1,73 @@
-"""Client for the university's OpenAI-compatible Qwen3-Embedding-4B endpoint."""
+"""Client for the university's OpenAI-compatible Qwen3-Embedding-4B endpoint.
+
+The endpoint runs a 4B model, so a single request carrying an entire weekly
+delta (or a first-run full anthology) easily exceeds the read timeout. We
+therefore chunk the input and retry the transient timeouts the openai SDK
+does not retry on its own (it retries 429/5xx but treats read timeouts as
+non-retryable connection errors).
+"""
+import logging
+import time
+
 import numpy as np
-from openai import OpenAI, OpenAIError
+from openai import APIConnectionError, OpenAI, OpenAIError
+from tqdm import tqdm
+
+logger = logging.getLogger(__name__)
+
+# Max texts per embedding request. Small enough that each call to the 4B
+# model finishes well inside the read timeout, even on a slow week.
+_BATCH_SIZE = 32
+# Number of attempts per chunk before giving up (and propagating before any
+# writes in run_sync, per its "fail before writes" contract).
+_MAX_RETRIES = 3
+_RETRY_BACKOFF_SECONDS = 2.0
 
 
 class EmbeddingsClient:
-    def __init__(self, base_url: str, api_key: str, model: str = "qwen3-embedding-4b", dimensions: int = 1024):
+    def __init__(self, base_url: str, api_key: str, model: str = "qwen3-embedding-4b",
+                 timeout: float = 60.0):
+        # The hosted qwen3-embedding-4b does NOT support matryoshka representation,
+        # so we must not send `dimensions` — sending it 400s. The model returns its
+        # fixed native dimensionality; the index dim is derived from the returned
+        # vectors in run_sync rather than assumed here.
         self.model = model
-        self.dimensions = dimensions
-        self._client = OpenAI(base_url=base_url, api_key=api_key, timeout=60.0)
+        self._client = OpenAI(base_url=base_url, api_key=api_key, timeout=timeout)
 
     def embed_batch(self, texts: list[str]) -> list[np.ndarray]:
-        try:
-            response = self._client.embeddings.create(
-                model=self.model, input=texts, dimensions=self.dimensions
-            )
-        except OpenAIError as e:
-            raise RuntimeError(f"Embedding API request failed: {e}") from e
-        data = sorted(response.data, key=lambda d: d.index)
-        return [np.array(d.embedding, dtype=np.float32) for d in data]
+        if not texts:
+            return []
+        total_chunks = (len(texts) + _BATCH_SIZE - 1) // _BATCH_SIZE
+        vectors: list[np.ndarray] = []
+        starts = range(0, len(texts), _BATCH_SIZE)
+        for start in tqdm(starts, total=total_chunks, desc="embedding", unit="batch"):
+            chunk = texts[start:start + _BATCH_SIZE]
+            vectors.extend(self._embed_chunk(chunk))
+        return vectors
+
+    def _embed_chunk(self, chunk: list[str]) -> list[np.ndarray]:
+        last_error: Exception | None = None
+        for attempt in range(1, _MAX_RETRIES + 1):
+            try:
+                response = self._client.embeddings.create(
+                    model=self.model, input=chunk
+                )
+            except APIConnectionError as e:
+                # Read timeouts (APITimeoutError is a subclass) and other
+                # transient connection errors: the SDK won't retry, so we do.
+                last_error = e
+                if attempt == _MAX_RETRIES:
+                    break
+                logger.warning(
+                    "embedding request failed (attempt %d/%d), retrying in %.1fs: %s",
+                    attempt, _MAX_RETRIES, _RETRY_BACKOFF_SECONDS, e,
+                )
+                time.sleep(_RETRY_BACKOFF_SECONDS)
+                continue
+            except OpenAIError as e:
+                raise RuntimeError(f"Embedding API request failed: {e}") from e
+            data = sorted(response.data, key=lambda d: d.index)
+            return [np.array(d.embedding, dtype=np.float32) for d in data]
+        raise RuntimeError(
+            f"Embedding API request failed after {_MAX_RETRIES} attempts: {last_error}"
+        ) from last_error

sync/run_sync.py

@@ -3,7 +3,7 @@ import json
 import os
 from datetime import datetime, timezone
 
-from common.db import init_db, upsert_paper, mark_inactive
+from common.db import init_db, upsert_papers, mark_inactive_ids
 from common.vector_index import create_index, load_index, save_index, add_vector
 from sync.anthology_source import iter_papers
 from sync.delta import compute_delta, build_state
@@ -29,7 +29,7 @@ def run_sync(anthology_path: str, hf_repo_id: str, hf_token: str,
     try:
         index = load_index(index_path)
     except FileNotFoundError:
-        index = create_index(dim=1024)
+        index = None  # created lazily once the embedding dimension is known
 
     current_papers = list(iter_papers(anthology_path))
     delta = compute_delta(current_papers, last_state)
@@ -38,14 +38,27 @@ def run_sync(anthology_path: str, hf_repo_id: str, hf_token: str,
         client = EmbeddingsClient(base_url=embedding_base_url, api_key=embedding_api_key)
         texts = [_embedding_text(p) for p in delta.new_or_changed]
         vectors = client.embed_batch(texts)  # raises RuntimeError on failure, propagates before any writes below
+        # qwen3-embedding-4b returns a fixed native dim (no matryoshka), so the
+        # index dimension is whatever the model emits, derived from the first vector.
+        embedding_dim = len(vectors[0])
+        if index is None:
+            index = create_index(dim=embedding_dim)
+        elif index.d != embedding_dim:
+            raise RuntimeError(
+                f"Embedding dimension {embedding_dim} does not match existing index "
+                f"dimension {index.d}; delete the stale index.faiss (locally and on "
+                f"the HF Hub) and run a fresh full sync."
+            )
+        upsert_rows = []
         for paper, vector in zip(delta.new_or_changed, vectors):
             faiss_id = add_vector(index, vector)
-            upsert_paper(conn, {**paper, "active": True, "faiss_id": faiss_id})
+            upsert_rows.append({**paper, "active": True, "faiss_id": faiss_id})
+        upsert_papers(conn, upsert_rows)
 
-    for paper_id in delta.removed_ids:
-        mark_inactive(conn, paper_id)
+    mark_inactive_ids(conn, list(delta.removed_ids))
 
-    save_index(index, index_path)
+    if index is not None:
+        save_index(index, index_path)
     state = {
         "papers": build_state(current_papers),
         "last_synced_at": datetime.now(timezone.utc).isoformat(),

tests/sync/test_embeddings_client.py

@@ -18,10 +18,9 @@ class _FakeEmbeddingsResponse:
 def test_embed_batch_parses_response(monkeypatch):
     captured = {}
 
-    def fake_create(self, model, input, dimensions):
+    def fake_create(self, model, input):
         captured["model"] = model
         captured["input"] = input
-        captured["dimensions"] = dimensions
         return _FakeEmbeddingsResponse([
             _FakeEmbeddingDatum([0.2] * 1024, 1),
             _FakeEmbeddingDatum([0.1] * 1024, 0),
@@ -29,11 +28,10 @@ def test_embed_batch_parses_response(monkeypatch):
 
     monkeypatch.setattr("openai.resources.embeddings.Embeddings.create", fake_create)
 
-    client = EmbeddingsClient(base_url="http://fake", api_key="k", dimensions=1024)
+    client = EmbeddingsClient(base_url="http://fake", api_key="k")
     result = client.embed_batch(["a", "b"])
 
     assert captured["input"] == ["a", "b"]
-    assert captured["dimensions"] == 1024
     assert len(result) == 2
     assert result[0].shape == (1024,)
     assert result[0].dtype == np.float32
@@ -41,10 +39,91 @@ def test_embed_batch_parses_response(monkeypatch):
     assert np.isclose(result[1][0], 0.2)
 
 
+def test_embed_batch_empty_input_makes_no_requests(monkeypatch):
+    called = {"n": 0}
+
+    def fake_create(self, model, input):
+        called["n"] += 1
+        return _FakeEmbeddingsResponse([])
+
+    monkeypatch.setattr("openai.resources.embeddings.Embeddings.create", fake_create)
+
+    client = EmbeddingsClient(base_url="http://fake", api_key="k")
+    assert client.embed_batch([]) == []
+    assert called["n"] == 0
+
+
+def test_embed_batch_chunks_large_input(monkeypatch):
+    captured = []
+
+    def fake_create(self, model, input):
+        captured.append(list(input))
+        # index resets to 0 within each request, as on the real API
+        return _FakeEmbeddingsResponse([
+            _FakeEmbeddingDatum([float(i)] * 1024, i) for i in range(len(input))
+        ])
+
+    monkeypatch.setattr("openai.resources.embeddings.Embeddings.create", fake_create)
+    monkeypatch.setattr("sync.embeddings_client._BATCH_SIZE", 32)
+
+    client = EmbeddingsClient(base_url="http://fake", api_key="k")
+    texts = [f"t{i}" for i in range(80)]
+    result = client.embed_batch(texts)
+
+    assert len(captured) == 3  # 32 + 32 + 16
+    assert captured[0] == texts[:32]
+    assert captured[1] == texts[32:64]
+    assert captured[2] == texts[64:]
+    assert len(result) == 80
+    # order preserved within and across chunks (each chunk's vectors come back 0..n-1)
+    assert [round(v[0]) for v in result] == list(range(32)) + list(range(32)) + list(range(16))
+
+
+def test_embed_batch_retries_on_timeout_then_succeeds(monkeypatch):
+    import httpx
+    from openai import APITimeoutError
+
+    calls = {"n": 0}
+
+    def fake_create(self, model, input):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise APITimeoutError(request=httpx.Request("POST", "http://fake/embeddings"))
+        return _FakeEmbeddingsResponse([_FakeEmbeddingDatum([0.5] * 1024, 0)])
+
+    monkeypatch.setattr("openai.resources.embeddings.Embeddings.create", fake_create)
+    monkeypatch.setattr("sync.embeddings_client.time.sleep", lambda _: None)
+
+    client = EmbeddingsClient(base_url="http://fake", api_key="k")
+    result = client.embed_batch(["a"])
+
+    assert calls["n"] == 2
+    assert np.isclose(result[0][0], 0.5)
+
+
+def test_embed_batch_raises_after_exhausted_retries(monkeypatch):
+    import httpx
+    from openai import APITimeoutError
+
+    calls = {"n": 0}
+
+    def fake_create(self, model, input):
+        calls["n"] += 1
+        raise APITimeoutError(request=httpx.Request("POST", "http://fake/embeddings"))
+
+    monkeypatch.setattr("openai.resources.embeddings.Embeddings.create", fake_create)
+    monkeypatch.setattr("sync.embeddings_client.time.sleep", lambda _: None)
+
+    client = EmbeddingsClient(base_url="http://fake", api_key="k")
+    with pytest.raises(RuntimeError):
+        client.embed_batch(["a"])
+    assert calls["n"] == 3  # _MAX_RETRIES attempts
+
+
 def test_embed_batch_raises_runtime_error_on_api_error(monkeypatch):
     import httpx
 
-    def fake_create(self, model, input, dimensions):
+    def fake_create(self, model, input):
         request = httpx.Request("POST", "http://fake/embeddings")
         response = httpx.Response(500, request=request, json={"error": "boom"})
         raise APIStatusError("boom", response=response, body={"error": "boom"})