release: 0.10.2 — footgun 429 accionable + upsert masivo (dev→main)

docs/API.md

@@ -851,7 +851,7 @@ def load_equation_spec(path: str | Path) -> EquationSpec:
 
 | Implementación | Estado | Notas |
 |----------------|--------|-------|
-| `OpenAlexSource` | **v1** | **Referencia/backbone**, sobre `httpx`. Entrega mínimo + enriquecimiento (refs inline + afiliaciones per-autor + instituciones; `cited_by_id` lo puebla el chaining/Enricher, no el seed). Traducción **passthrough**: envuelve la ecuación en `title_and_abstract.search:(...)` y **reporta** los límites WoS (NEAR/comodín/tags) sin traducirlos. Flag `native=True` (query cruda). **Negaciones (`exclude`):** cada `AND NOT "<término>"` se inyecta **dentro** de la única expresión `search:((query) AND NOT "<término>")` (campo no repetido; el filtro de año queda como predicado separado por coma, fuera del `search`) y se reporta en el `translation_report`; ignorado con `native`. Credenciales inyectadas (arg → `OPENALEX_API_KEY` → `~/.openalex/credentials` → polite pool; ADR 0012). Cursor paging con tope `max_results` (default 200). Puebla `Manifest.openalex_version` (ADR 0017). `transport` inyectable (tests sin red). |
+| `OpenAlexSource` | **v1** | **Referencia/backbone**, sobre `httpx`. Entrega mínimo + enriquecimiento (refs inline + afiliaciones per-autor + instituciones; `cited_by_id` lo puebla el chaining/Enricher, no el seed). Traducción **passthrough**: envuelve la ecuación en `title_and_abstract.search:(...)` y **reporta** los límites WoS (NEAR/comodín/tags) sin traducirlos. Flag `native=True` (query cruda). **Negaciones (`exclude`):** cada `AND NOT "<término>"` se inyecta **dentro** de la única expresión `search:((query) AND NOT "<término>")` (campo no repetido; el filtro de año queda como predicado separado por coma, fuera del `search`) y se reporta en el `translation_report`; ignorado con `native`. Credenciales inyectadas (arg → `OPENALEX_API_KEY` → `~/.openalex/credentials` → polite pool; ADR 0012). Cursor paging con tope `max_results` (default 200). Puebla `Manifest.openalex_version` (ADR 0017). `transport` inyectable (tests sin red). Un **429** (rate limit del pool anónimo) en `seed()` → `NetworkError` (exit 4) con mensaje **accionable**: declarar `--email` mueve la petición al polite pool (remedio primario); api_key opcional (ADR 0012, #210). |
 | `BibtexSource` | **v1, secundaria** | Sembrar desde *pearls* vía `load()`. Extra **`[bibtex]`** (import perezoso de `bibtexparser`); acceso defensivo (campos faltantes sin `KeyError`). Mínimo universal. `seed()` lanza `NotImplementedError`. `.bib` con error grave → `ValueError`; sin entradas válidas → `UserWarning` (no no-op silencioso). Carga bulk con `from_arrow`. |
 | `ScieloSource` / `RedalycSource` / `LaReferenciaSource` | futuro | Fuentes regionales, mínimo universal. Declaradas, no implementadas (ADR 0018). |
 | `RisSource` / `CsvSource` | futuro | No implementados. |
@@ -860,7 +860,9 @@ def load_equation_spec(path: str | Path) -> EquationSpec:
 el `Forager` y el `Enricher`):
 
 - **`fetch_citing(openalex_id) -> list[dict]`** (singular, forward chaining): `GET works?filter=cites:`,
-  con retry/backoff ante 429/5xx.
+  con retry/backoff ante 429/5xx. Al **agotar** los reintentos con **429** → `NetworkError` (exit 4)
+  **accionable** (polite pool/`--email`; ADR 0012, #210); los 5xx agotados conservan
+  `httpx.HTTPStatusError`. Asimetría deliberada: solo el 429 tiene remedio del lado del usuario.
 - **`fetch_citing_batch(ids, *, max_per_paper, since=None) -> dict[seed_id, list[citer_id]]`**: trae los
   citantes **batcheando por OR** (`cites:W1|W2|...`, lotes ≤50), pagina por cursor y **atribuye página a
   página** con **presupuesto por semilla** (corta cuando todas alcanzan `max_per_paper`; sin starvation).

docs/decisiones/0012-openalex-credenciales.md

@@ -1,6 +1,7 @@
 # 0012 — Credenciales de OpenAlex: email del pool cortés + API key opcional, inyectados
 
-- **Estado:** Aceptada
+- **Estado:** Aceptada · **enmendada 2026-06-29** (#210: un 429 se traduce a `NetworkError`
+  accionable que apunta al polite pool, en `seed` y en el chaining — ver "Seguimiento" al final)
 - **Fecha:** 2026-06-15
 - **Decidido por:** IA (Claude Opus 4.8), validado por el Product Owner humano
   (ver [`registro-ia.md`](registro-ia.md))
@@ -44,3 +45,24 @@ secreto en código**, ningún `os.environ.get("...", "literal")` para la key.
   en el README cuando llegue el Hito 4, y testear ambos caminos (con y sin key) contra API
   mockeada — sin red en CI.
 - No cambia el núcleo puro: las credenciales viven en la costura `Source`, inyectadas.
+
+## Seguimiento — 2026-06-29 (#210: el 429 aflora como error accionable hacia el polite pool)
+
+> Realización de la consecuencia "se avisa" (lección 7): un **429 (Too Many Requests)** del pool
+> anónimo ya **no** aflora como `httpx.HTTPStatusError` pelado, sino como **`NetworkError`** (exit 4,
+> `service/errors.py`) con un mensaje que nombra el **remedio primario — declarar el email mueve la
+> petición al polite pool** (límite más generoso), la **api_key como opcional** y referencia este ADR
+> (`_MSG_RATE_LIMIT_429`). Aplica en **ambos caminos**: `seed()` (al recibir 429) y
+> `fetch_citing`/chaining (`_fetch_all_with_retry`, al **agotar** los reintentos con 429).
+>
+> No cambia la **política** de credenciales (la decisión de este ADR): la implementa. Statuses no-429
+> y los retryables 5xx (500/502/503/504) conservan su conducta anterior — asimetría deliberada: solo
+> el 429 tiene remedio del lado del usuario (el polite pool).
+>
+> Extender el fix más allá del `seed` del DoD original —que el chaining levante `NetworkError` en vez
+> de `HTTPStatusError` al agotar reintentos— fue **decisión consciente del PO**: dejar el footgun del
+> 429 cubierto en todos los caminos de cara a la promoción de la librería.
+>
+> **Decidido por:** Product Owner humano (2026-06-29, #210); implementado y verificado (gate verde).
+> Ver `src/bib2graph/sources/openalex.py` (`_MSG_RATE_LIMIT_429`, `seed`, `_fetch_all_with_retry`) y
+> `docs/API.md` §2.

pyproject.toml

@@ -204,12 +204,13 @@ addopts = [
     "--cov=bib2graph",
     "--cov-report=term-missing",
     "-m",
-    "not network",
+    "not network and not slow",
 ]
 markers = [
     "unit: tests puros, sin red ni I/O (default)",
     "integration: tests con I/O local (duckdb/store); corren en el gate",
     "network: tests que requieren red real (p.ej. OpenAlex); excluidos del gate por defecto (correr con -m network)",
+    "slow: benchmarks de escala (50 K+ filas); excluidos del gate por defecto (correr con -m slow)",
 ]
 
 [tool.coverage.run]

src/bib2graph/backends/duckdb.py

@@ -40,6 +40,7 @@
     _apply_curation_to_rows,
     _merge_curation_status,
     _merge_provenance,
+    _merge_rows,
     compute_corpus_hash,
 )
 from bib2graph.constants import LIST_COLUMNS, CurationStatus
@@ -254,6 +255,118 @@ def _build_upsert_sql() -> str:
 
 _UPSERT_SQL = _build_upsert_sql()
 
+
+def _build_bulk_update_existing_sql() -> str:
+    """SQL para actualizar filas que ya existen en ``corpus`` (conflictos reales).
+
+    Lee el lote desde la vista registrada ``_incoming_upsert`` y actualiza solo
+    las filas cuyo ``id`` ya existe en ``corpus``.  Las UDFs Python
+    ``_merge_curation_status_udf`` y ``_merge_provenance_udf`` se invocan
+    únicamente para las filas que efectivamente están en conflicto (no para
+    inserciones nuevas), eliminando el overhead de UDF para el caso de primer
+    persist sin conflictos.
+
+    La columna ``_seq`` NO se actualiza: las filas existentes conservan su
+    orden de primera aparición (ADR 0024, D3).
+    """
+    scalar_cols = [
+        "source_id",
+        "doi",
+        "title",
+        "year",
+        "abstract",
+        "source",
+        "language",
+        "publisher",
+        "is_seed",
+    ]
+    list_cols = list(LIST_COLUMNS)
+
+    scalar_sets = [f"    {c} = COALESCE(src.{c}, corpus.{c})" for c in scalar_cols]
+
+    # D3 para listas: unión ordenada deduplicada; NULL si ambos son NULL
+    list_sets = [
+        f"    {c} = CASE\n"
+        f"        WHEN corpus.{c} IS NULL AND src.{c} IS NULL THEN NULL\n"
+        f"        ELSE list_sort(list_distinct(list_concat(\n"
+        f"            COALESCE(corpus.{c}, []),\n"
+        f"            COALESCE(src.{c}, [])\n"
+        f"        )))\n"
+        f"    END"
+        for c in list_cols
+    ]
+
+    # UDFs para los campos especiales de merge
+    special_sets = [
+        "    curation_status = _merge_curation_status_udf(\n"
+        "        corpus.curation_status, corpus.provenance,\n"
+        "        src.curation_status, src.provenance\n"
+        "    )",
+        "    provenance = _merge_provenance_udf(corpus.provenance, src.provenance)",
+    ]
+
+    all_sets = scalar_sets + list_sets + special_sets
+    sets_sql = ",\n".join(all_sets)
+
+    return (
+        f"UPDATE corpus SET\n"
+        f"{sets_sql}\n"
+        f"FROM _incoming_upsert AS src\n"
+        f"WHERE corpus.id = src.id"
+    )
+
+
+def _build_bulk_insert_new_sql() -> str:
+    """SQL para insertar filas nuevas (sin conflicto con ``corpus``).
+
+    Lee el lote desde la vista registrada ``_incoming_upsert`` y hace INSERT
+    solo de las filas cuyo ``id`` NO existe todavía en ``corpus``.  No invoca
+    ninguna UDF Python (las filas son nuevas, no hay merge que hacer).
+
+    ADR 0024: ``_seq = CAST($1 AS BIGINT) + ROW_NUMBER() OVER (ORDER BY _row_idx)``
+    asigna valores únicos y crecientes en el orden de aparición del lote Arrow.
+    ``_row_idx`` es una columna auxiliar (índice 0-based) agregada al registrar
+    la vista, que garantiza que el ROW_NUMBER() respete el orden de la tabla
+    Arrow entrante sin depender del scan de DuckDB (no garantizado por SQL estándar
+    sin ORDER BY).  ``$1`` es el ``MAX(_seq)`` calculado antes de la llamada.
+    """
+    schema_cols = ", ".join(f.name for f in CORPUS_SCHEMA)
+    insert_cols = f"{schema_cols}, _seq"
+    return (
+        f"INSERT INTO corpus ({insert_cols})\n"
+        f"SELECT {schema_cols},\n"
+        f"    CAST($1 AS BIGINT) + ROW_NUMBER() OVER (ORDER BY _row_idx) AS _seq\n"
+        f"FROM _incoming_upsert\n"
+        f"WHERE id NOT IN (SELECT id FROM corpus)"
+    )
+
+
+def _build_simple_insert_sql() -> str:
+    """SQL de INSERT directo sin filtro de conflicto (usado tras DELETE en overwrite).
+
+    Después de un ``DELETE FROM corpus`` la tabla está vacía y cualquier
+    fila de ``_incoming_upsert`` es nueva: el filtro ``NOT IN`` sería un
+    no-op costoso.  Este SQL hace el INSERT sin ninguna condición.
+
+    ADR 0024: ``_seq`` empieza en 1.  ``ROW_NUMBER() OVER (ORDER BY _row_idx)``
+    garantiza que el orden de filas en la tabla Arrow se preserve fielmente en
+    ``_seq`` (la columna ``_row_idx`` es el índice 0-based agregado al registrar
+    la vista; sin ORDER BY el scan de DuckDB no tiene orden garantizado por SQL).
+    """
+    schema_cols = ", ".join(f.name for f in CORPUS_SCHEMA)
+    insert_cols = f"{schema_cols}, _seq"
+    return (
+        f"INSERT INTO corpus ({insert_cols})\n"
+        f"SELECT {schema_cols},\n"
+        f"    ROW_NUMBER() OVER (ORDER BY _row_idx) AS _seq\n"
+        f"FROM _incoming_upsert"
+    )
+
+
+_BULK_UPDATE_EXISTING_SQL = _build_bulk_update_existing_sql()
+_BULK_INSERT_NEW_SQL = _build_bulk_insert_new_sql()
+_SIMPLE_INSERT_SQL = _build_simple_insert_sql()
+
 # ---------------------------------------------------------------------------
 # Helpers de conversión Python ↔ DuckDB
 # ---------------------------------------------------------------------------
@@ -287,6 +400,45 @@ def _arrow_table_from_con(con: duckdb.DuckDBPyConnection) -> pa.Table:
     return result.cast(CORPUS_SCHEMA)
 
 
+def _dedup_merge_table(table: pa.Table) -> pa.Table:
+    """Deduplica las filas de ``table`` por ``id``, mergeando duplicados con semántica D3.
+
+    Si el lote entrante contiene múltiples filas con el mismo ``id``, las fusiona
+    progresivamente usando la misma lógica que ``_merge_rows`` de ``backends.memory``
+    (la misma semántica que las UDFs ``_merge_curation_status_udf`` y
+    ``_merge_provenance_udf``).  El orden de primera aparición se preserva.
+
+    Garantiza que el lote no viole el PRIMARY KEY de ``corpus`` (``id UNIQUE``),
+    que antes causaba ``ConstraintException`` en el upsert bulk cuando el batch
+    entrante ya contenía ids repetidos.
+
+    Args:
+        table: Tabla Arrow de entrada (puede tener ``id`` duplicados).
+
+    Returns:
+        Tabla Arrow sin ids duplicados, con el mismo schema canónico.
+        Si la tabla no tiene duplicados, la devuelve intacta.
+    """
+    if len(table) == 0:
+        return table
+    rows = table.to_pylist()
+    seen: dict[str, dict[str, object]] = {}
+    order: list[str] = []
+    for row in rows:
+        id_ = str(row["id"])
+        if id_ not in seen:
+            seen[id_] = row
+            order.append(id_)
+        else:
+            # Merge progresivo: preserva provenance acumulada y curación más reciente
+            seen[id_] = _merge_rows(seen[id_], row)
+    if len(seen) == len(rows):
+        # Sin duplicados: devolver la tabla original sin coste de reconstrucción
+        return table
+    deduped = [seen[id_] for id_ in order]
+    return pa.Table.from_pylist(deduped, schema=CORPUS_SCHEMA)
+
+
 # ---------------------------------------------------------------------------
 # DuckDBBackend
 # ---------------------------------------------------------------------------
@@ -421,24 +573,53 @@ def merge_provenance_udf(
     def _upsert_table(self, table: pa.Table) -> None:
         """Hace upsert de todas las filas de ``table`` en la base de datos.
 
-        ADR 0024: calcula ``start = COALESCE(MAX(_seq), 0)`` una sola vez y
-        asigna ``_seq = start + 1 + i`` (0-based) a cada fila. Filas ya
-        existentes (ON CONFLICT) ignoran el ``_seq`` provisto porque el
-        DO UPDATE no lo actualiza, preservando así el orden de primera
-        aparición (D3).
+        Ingesta vectorizada de dos pasos — registra la tabla Arrow en DuckDB
+        (zero-copy) y ejecuta exactamente 2 + 1 sentencias SQL por lote:
+
+        1. **UPDATE** filas ya existentes (las UDFs Python se invocan solo para
+           los conflictos reales; para un primer persist sin conflictos = 0 UDFs).
+        2. **INSERT** filas nuevas con ``WHERE id NOT IN (SELECT id FROM corpus)``
+           (vectorizado, cero UDFs).
+
+        Elimina el round-trip por fila previo (O(n) → O(1) viajes al motor SQL)
+        y evita además llamar a las UDFs Python para filas sin conflicto,
+        logrando la mayor ganancia de velocidad en el caso de primer persist
+        (corpus vacío → todas las filas son nuevas → 0 UDFs).
+
+        Pre-procesamiento del lote:
+        - **Dedup-MERGE** (``_dedup_merge_table``): si el lote contiene ids
+          duplicados, los fusiona con la misma semántica D3 que las UDFs
+          (provenance append-only, curación más reciente gana) antes de registrar
+          la vista, evitando violaciones de PRIMARY KEY.
+        - **Columna ``_row_idx``**: se agrega al registrar la vista para que
+          ``ROW_NUMBER() OVER (ORDER BY _row_idx)`` en ``_BULK_INSERT_NEW_SQL``
+          produzca un ``_seq`` determinista y estable que respeta el orden de
+          aparición de la tabla Arrow entrante (ADR 0024, D3).
+
+        Las filas existentes conservan su ``_seq`` original (D3).
         """
-        rows = table.to_pylist()
-        if not rows:
+        if len(table) == 0:
             return
-        # ADR 0024: base para el _seq monótono de esta inserción por lote
+        # Dedup-MERGE: fusiona ids duplicados en el lote ANTES del upsert SQL
+        table = _dedup_merge_table(table)
+        # ADR 0024: base para el _seq monótono de las filas nuevas en este lote
         result = self._con.execute(
             "SELECT COALESCE(MAX(_seq), 0) FROM corpus"
         ).fetchone()
         start: int = int(result[0]) if result else 0
-        for i, row in enumerate(rows):
-            # _seq = start + 1 + i garantiza valores únicos y crecientes en este lote
-            params = [*_row_to_params(row), start + 1 + i]
-            self._con.execute(_UPSERT_SQL, params)
+        # _row_idx: columna auxiliar de orden para ROW_NUMBER() OVER (ORDER BY _row_idx)
+        table_with_idx = table.append_column(
+            "_row_idx",
+            pa.array(range(len(table)), type=pa.int64()),
+        )
+        self._con.register("_incoming_upsert", table_with_idx)
+        try:
+            # Paso 1: actualizar filas existentes (UDFs solo para conflictos reales)
+            self._con.execute(_BULK_UPDATE_EXISTING_SQL)
+            # Paso 2: insertar filas nuevas (vectorizado, sin UDF)
+            self._con.execute(_BULK_INSERT_NEW_SQL, [start])
+        finally:
+            self._con.unregister("_incoming_upsert")
 
     def _clone(self) -> DuckDBBackend:
         """Crea una nueva instancia apuntando al mismo archivo (path compartido).
@@ -566,9 +747,12 @@ def merge(self, other_table: pa.Table) -> DuckDBBackend:
         luego las nuevas filas de ``other_table`` en su orden de aparición) se
         garantiza por la columna interna ``_seq``:
         - Filas de ``self`` ya tienen ``_seq`` asignado (se preservan en el
-          ON CONFLICT DO UPDATE, que no actualiza ``_seq``).
-        - Filas nuevas de ``other_table`` reciben ``_seq`` mayor (calculado
-          como ``MAX(_seq) + 1 + i`` en ``_upsert_table``).
+          UPDATE, que no actualiza ``_seq``).
+        - Filas nuevas de ``other_table`` reciben ``_seq`` mayor, calculado
+          con ``ROW_NUMBER() OVER (ORDER BY _row_idx)`` desde ``MAX(_seq)``
+          en ``_upsert_table``; la columna auxiliar ``_row_idx`` (índice
+          0-based del lote Arrow) garantiza que el orden de primera aparición
+          se preserve de forma determinista, sin depender del scan de DuckDB.
         - ``_arrow_table_from_con`` lee con ``ORDER BY _seq``.
         No se requiere DELETE+reinsert.
 
@@ -1005,9 +1189,10 @@ def close(self) -> None:
     def overwrite_corpus(self, table: pa.Table) -> None:
         """Reemplaza TODA la tabla ``corpus`` con el contenido de ``table``.
 
-        Hace TRUNCATE + INSERT (no upsert) para que el estado en disco sea
-        exactamente ``table``, sin residuos de filas previas.  Preserva las
-        tablas hermanas (``loop_state_log``, ``referenced_but_not_fetched``).
+        Hace TRUNCATE + INSERT masivo (no upsert fila-a-fila) para que el estado
+        en disco sea exactamente ``table``, sin residuos de filas previas.
+        Preserva las tablas hermanas (``loop_state_log``,
+        ``referenced_but_not_fetched``).
 
         Úsalo solo en la ruta de ingesta (``seed``, ``restore``, ``chain``,
         ``thesaurus``) donde ya tenés el corpus completo y correcto en
@@ -1024,10 +1209,22 @@ def overwrite_corpus(self, table: pa.Table) -> None:
                 Debe cumplir ``CORPUS_SCHEMA``.
         """
         self._con.execute("DELETE FROM corpus")
-        rows = table.to_pylist()
-        for i, row in enumerate(rows):
-            params = [*_row_to_params(row), i + 1]
-            self._con.execute(_UPSERT_SQL, params)
+        if len(table) == 0:
+            return
+        # Dedup-MERGE: fusiona ids duplicados en el lote ANTES del INSERT masivo
+        table = _dedup_merge_table(table)
+        # Tras DELETE la tabla está vacía: INSERT directo sin filtro NOT IN
+        # (más eficiente que _BULK_INSERT_NEW_SQL para el caso de tabla limpia)
+        # _row_idx: columna auxiliar de orden para ROW_NUMBER() OVER (ORDER BY _row_idx)
+        table_with_idx = table.append_column(
+            "_row_idx",
+            pa.array(range(len(table)), type=pa.int64()),
+        )
+        self._con.register("_incoming_upsert", table_with_idx)
+        try:
+            self._con.execute(_SIMPLE_INSERT_SQL)
+        finally:
+            self._con.unregister("_incoming_upsert")
 
     def query(self, sql: str) -> pa.Table:
         """Ejecuta una consulta SQL sobre el backend y devuelve tabla Arrow.