---

proceso_id: syntage-consistencia nombre: Syntage · Consistencia de invoices (capa 1 BQ↔ELK · capa 3 BQ↔API Syntage) criticidad: media patron: api-pull-programado # Cloud Run agendado (data-quality): capa 1 reconcilia BQ vs ELK a nivel uuid (diario desde 2026-07-24, antes horario), capa 3 reconcilia BQ vs la API de Syntage por RFC (diario) keywords: [consistencia, data quality, calidad de datos, elk, elasticsearch, bigquery, syntage, invoices, uuid, emisor, receptor, issuer, receiver, colision, invoices_ingestion_consistency_summary, invoices_ingestion_consistency_detail, invoices_source_consistency_summary, invoices_source_consistency_detail, invoices_consistency_report, invoices_consistency_report_90, invoices_consistency_issues, reconciliación, drift, frescura, watermark, load_dts, backfill, persistencia, event log, dedup, append-only, capa 1, capa 3, fuente de verdad, dlq, hydra, totalItems] fuentes: [Elasticsearch (ELK), BigQuery, API de Syntage] raw: - tabla: solvento-data-prod.report.invoices_ingestion_consistency_summary cargada_por: cloud run consistencia-elk-bq-syntage (WRITE_APPEND, partición diaria por run_at) grano: una fila por corrida (mode='elk-ingestion' desde 2026-07-24, antes '90days'; publish_reprocess.py y las cards filtran por ese valor — ver danger box sobre por qué se sacó el modo histórico) — run_at (frescura del job), window_lower/window_end (watermark de datos), faltantes_uuid (EXACTO, suma por RFC — la métrica de alerta), excedente_elk, faltantes_uuid_aprox/elk_distinct_uuid_aprox (cardinality ES, solo diagnóstico), pct_uuid_ok, pct_rfc_ok - tabla: solvento-data-prod.report.invoices_ingestion_consistency_detail cargada_por: cloud run consistencia-elk-bq-syntage (WRITE_APPEND, partición diaria por run_at) grano: una fila por (rfc, corrida) — bq_uuid, elk_count, colision_otro_rfc, faltantes, cuadra - tabla: solvento-data-prod.report.invoices_source_consistency_summary cargada_por: cloud run consistencia-elk-bq-syntage, endpoint /consistency/api-source (CAPA 3, WRITE_APPEND, partición diaria por run_at) grano: una fila por corrida diaria — faltantes_bq (métrica de alerta: facturas que Syntage tiene y BQ no), excedente_bq, rfcs_cuadran/descuadran/no_comparables, pct_rfc_ok - tabla: solvento-data-prod.report.invoices_source_consistency_detail cargada_por: cloud run consistencia-elk-bq-syntage, endpoint /consistency/api-source (CAPA 3) grano: una fila por (rfc, corrida) — bq_count, syntage_count, faltantes, cuadra, estado (ok|no_encontrado|error) depende_de: - solvento-data-prod.syntage.invoices # capa 1: dedup a última fila por (uuid, rfc) por load_dts, luego COUNT(DISTINCT uuid) filtrado por issuedat. capa 3: COUNT(DISTINCT id) GROUP BY rfc - Elasticsearch (ELK) índices invoices,invoices-* # capa 1: cardinality de uuid + doc_count por rfc - API de Syntage /taxpayers/{rfc}/invoices # capa 3: hydra:totalItems por RFC (1 request por RFC, rate limit ~10 req/s) consumo: - tipo: analytics quien: Dashboard Metabase 917 "Panel de inconsistencias V2" (colección "Consistencia", separado del 232 "Panel de Inconsistencias") lee_de: solvento-data-prod.report.invoices_ingestion_consistency_summary / _detail estado: dashboard creado (2026-07-08) con 4 cards del modo 90 días (faltantes en ELK, % cuadrado, watermark, tendencia). Faltan del spec (consistencia/METABASE_CARDS.md): frescura del job, detalle por RFC, y sobre todo la card de faltantes persistentes (única pensada para alertar). La sección "Historical" que se había creado en el dashboard queda descartada (ver danger box "Por qué este job ya no compara histórico") — no va a tener cards, el modo histórico BQ-vs-ELK se sacó del job. - tipo: analytics (pendiente) quien: capa 3 — cards a agregar al mismo dashboard 917 lee_de: solvento-data-prod.report.invoices_source_consistency_summary / _detail estado: spec listo en consistencia/METABASE_CARDS.md (sección "Capa 3"). La card de faltantes_bq debería alertar directo, no esperar persistencia — ver danger box sobre la ausencia de DLQ. repos: [https://gitlab.com/solventomx/data/consistencia] ultima_actualizacion: 2026-07-24

---

Syntage · Consistencia de invoices

Dos cuadraturas independientes sobre el mismo flujo de invoices de Syntage, que responden preguntas distintas:

• Capa 1 — ingesta (BQ ↔ ELK), diaria (0 4 * * * MX; era horaria hasta 2026-07-24). ¿La data que entró llegó a los dos sinks que la ingesta debería llenar? Compara a nivel uuid, no por conteo de RFC.
• Capa 3 — fuente de verdad (BQ ↔ API de Syntage), diaria. ¿Entró todo lo que Syntage tiene? Compara COUNT(DISTINCT id) por RFC.

Las dos hacen falta y no se solapan: capa 1 es ciega al caso en que una factura falta en BQ y en ELK por igual (los dos "coinciden" en no tenerla y reporta 0 faltantes). Ese caso solo lo ve capa 3. Validado en la práctica: de los 22 RFCs que descuadraron en capa 3 (2026-07-24), solo 3 estaban en la lista de descuadres de capa 1 — son modos de falla independientes.

(Capa 2 — RAW vs las tablas transformadas de dbt/AA — sigue pendiente, no implementada.)

Rediseño 2026-07: el grano por RFC generaba falsos positivos

La versión anterior de este job comparaba COUNT(DISTINCT id) por RFC en BigQuery contra doc_count por RFC en ELK. Desde que ELK dejó de tener un índice por RFC (pasó a uno único invoices,invoices-*), el _id de cada documento es el uuid de la factura — y un mismo uuid que en BigQuery aparece dos veces (una fila como emisor, otra como receptor, con RFCs distintos) colapsa a un solo documento en ELK. Comparar por RFC mezclaba unidades distintas y marcaba como "faltante" algo que en realidad estaba cargado. Al día de este rediseño, ~13% de los RFCs de los últimos 90 días (193/1452) mostraban ese mismatch en la versión vieja, con BQ≈Syntage casi siempre exacto — el patrón clásico de este bug, no de datos realmente perdidos. El job ahora reconcilia a nivel uuid y distingue "colisión con el otro RFC" (cargado, no es faltante) de "faltante real" (uuid ausente de ELK).

Las cards del dashboard 232 llevaban ~11 meses mostrando data falsa

El bloque "Inconsistencia Facturación" del dashboard 232 "Panel de Inconsistencias" (Metabase) lee de report.invoices_consistency_issues / invoices_consistency_issues_90. Verificado en BQ: esas tablas están en 0 filas desde agosto de 2025 — nada las escribe. El scorecard mostraba "0 inconsistencias" de forma indistinguible entre "todo bien" y "el job murió hace 11 meses". El job que sí corría todo este tiempo (consistencia-elk-bq-syntage, hasta hoy con TRUNCATE) escribía en invoices_consistency_report[_90] — vivas, pero nunca conectadas a ese dashboard. Este rediseño no reutiliza esas tablas: escribe en invoices_ingestion_consistency_summary/_detail (nuevas, con historial) y el dashboard de Metabase se construye aparte (ver consumo arriba). Las tablas viejas (*_issues* y *_report*) quedan para archivar una vez validado el reemplazo — no borrarlas todavía.

La pata de Syntage API salió de este job y se implementó aparte como capa 3 (2026-07-24)

La versión anterior comparaba contra la API de Syntage dentro del mismo job. Se separó: es la pata más frágil (rate limit 10 req/s, corridas largas, dependencia externa) y mezclada con capa 1 forzaba a las dos a compartir cadencia. Ahora vive en consistencia_bq_syntage.py con su propio endpoint y cadencia diaria — ver la sección "Capa 3" más abajo.

Nota sobre una idea que se descartó en el camino: se pensó acotar capa 3 a los RFCs que capa 1 ya marcara con faltante. Está mal: si el webhook nunca se disparó, la factura falta en BQ y en ELK por igual, capa 1 los ve cuadrados y ese RFC nunca aparecería en su lista — justo el caso que capa 3 existe para detectar. Capa 3 barre todos los RFCs. El filtrado por "ya sospechoso" solo aplica al drill-down de fase B, donde el barrido sin filtrar ya lo hizo fase A.

Validado en corridas reales: la cardinality de ES no es confiable como métrica global

La primera corrida de prueba dio faltantes_uuid negativo (-10,026) usando la agregación cardinality de Elasticsearch (HyperLogLog, precision_threshold topeado en 40.000 con ~3M de uuids reales — muy por debajo de lo necesario para ese volumen). Contradecía la reconciliación exacta por RFC (terms agregación, sin aproximar), que daba ~+1,534. Se corrigió: la columna faltantes_uuid que se escribe en invoices_ingestion_consistency_summary (y el pct_uuid_ok derivado) ahora sale de la suma exacta por RFC (faltantes_uuid_exacto en el código). La cardinality de ES queda solo en faltantes_uuid_aprox/elk_distinct_uuid_aprox, como diagnóstico — nunca usar esas dos para alertar.

Buffer de asentamiento (SETTLE_BUFFER_HOURS) y watermark de datos

También validado en corridas reales: RFCs de muy alto volumen pueden mostrar "faltantes" que cambian de signo entre una corrida y la siguiente (ej. +671 en una corrida, -203 catorce minutos después) — no es data perdida, es que BQ y ELK se escriben en paralelo desde el mismo pipeline sin sincronía perfecta para clientes de tráfico muy alto. Por eso la ventana de reconciliación excluye las últimas SETTLE_BUFFER_HOURS horas (default 3; mide el lag de propagación entre los dos sinks, que no depende de cada cuánto corra el job — por eso siguió siendo correcto al pasar de horario a diario). El límite superior de esa ventana se guarda como window_end — el watermark: hasta qué issuedat quedó certificada la cuadratura, distinto de run_at (cuándo corrió el job). Los dos hacen falta en el dashboard: run_at para saber si el job está vivo, window_end para saber hasta qué dato se puede confiar en el resultado.

Se probó filtrar también por load_dts para tapar backfills — se REVIRTIÓ, empeoró el resultado

Corriendo el script dos veces con minutos de diferencia, bq_uuid de la RFC rot7802026za creció miles de filas — un backfill activo: facturas con issuedat de abril-junio (fuera del buffer) pero load_dts (columna de carga en syntage.invoices) de "ahora mismo". Se agregó AND load_dts < window_end en fetch_bq() para tapar ese caso. Resultado, confirmado con una query directa: de 187,912 filas con load_dts en las últimas 3h, 60,679 tenían createdat viejo — load_dts se actualiza en cualquier escritura a la fila (cambio de status, pago, cancelación...), no solo en la primera carga. El filtro excluía del lado BQ facturas que ya estaban hace tiempo en ELK sin ninguna razón real: el match global cayó de 98% a 79.6% y excedente_elk se disparó a ~100,684. Se revirtió — fetch_bq() vuelve a filtrar solo por issuedat. No hay en syntage.invoices un campo confiable de "primera carga" distinto de load_dts (que en realidad es "última escritura").

La defensa contra backfills y otros eventos transitorios queda del lado de Metabase, no del job: la card de faltante persistente en varias corridas consecutivas (METABASE_CARDS.md) no necesita entender la causa de un pico puntual, solo exige que no se sostenga en el tiempo. Es más robusta que intentar filtrar la causa raíz en la query, porque no arriesga excluir data real como pasó acá.

syntage.invoices es un log de eventos, no una tabla upsert — se agregó dedup a última fila por (uuid, rfc)

Aclaración de arquitectura: por cada evento invoice.created/invoice.updated se agrega una fila nueva en syntage.invoices (no se pisa la existente). Validado con datos reales (90 días): 3,043,007 uuids distintos en 4,517,720 filas → 866,594 uuids (28.5%) tienen más de una fila, y 25,035 uuids (0.8%) tienen issuedat distinto entre sus filas duplicadas (wobble de ~1h). ELK, en cambio, siempre pisa el documento por _id=uuid — un único doc con el último evento. Comparar "cualquier fila de BQ en la ventana" contra "el único doc de ELK" no es una comparación justa cuando issuedat varía entre duplicados: un uuid puede entrar o salir del filtro de ventana de una corrida a la siguiente sin que la factura haya cambiado. Fix: fetch_bq() ahora dedupea PRIMERO a la fila más reciente por (uuid, rfc) (ROW_NUMBER() OVER (PARTITION BY uuid, rfc ORDER BY load_dts DESC, issuedat DESC)) y recién ahí filtra por la ventana de issuedat — la misma semántica "último estado" que ya usa ELK.

Bug encontrado y corregido en el primer --dry-run de esta versión: la primera implementación particionaba el ROW_NUMBER() solo por uuid (sin rfc), lo que colapsaba el caso emisor/receptor — la misma factura tiene 2 filas en BQ, una con el RFC emisor y otra con el RFC receptor, mismo uuid — a una sola fila, perdiendo la otra pata por completo. Resultado observado: RFCs que cuadran cayó de ~98% a 55.5% y el excedente ELK (exacto) saltó a 33,837, un falso "ELK tiene de más" sistemático en cientos de RFCs. Corregido particionando por (uuid, rfc). Re-validado con --dry-run: RFCs que cuadran 99.2%, excedente ELK exacto 0, faltantes reales 1,531 (en línea con los rangos sanos previos de 877-1,574).

¿load_dts viaja a ELK? Confirmado con inspect_elk_fields.py: no, pero hay un equivalente que no conviene usar

Corriendo el script de diagnóstico contra el mapping y un documento real de ELK: no existe ningún campo load_dts/equivalente nombrado así. createdat/updatedat SÍ están en el doc de ELK, pero son passthrough de los mismos campos de negocio que ya existen en syntage.invoices (no timestamps de ingesta de ELK) — de hecho es el mismo createdat que ya se usó para detectar que el filtro de load_dts en BQ era engañoso (ver danger box arriba). El campo que sí parece ser propio del pipeline de ingesta de ELK es _processing_timestamp (epoch ms, no estaba en la lista de candidatos original del script). No se usó: por la misma razón que load_dts falló en BQ, un timestamp de "última escritura" en ELK arriesga el mismo problema (excluye documentos viejos que se re-tocaron, no solo los recién llegados) — y ya no hace falta, porque el dedup por (uuid, rfc) de la nota anterior resuelve el problema sin necesitar ningún filtro de recencia de carga en ninguno de los dos lados.

Nota aparte, no relacionada con el fix: el mapping de invoices,invoices-* devolvió 4 índices reales — invoices (genérico) y tres índices dedicados por RFC (invoices-cas000522rc7, invoices-tsp0004051i4, invoices-rot7802026za — este último es el RFC del backfill). No es deuda de migración ni índice legacy: es una decisión de diseño a propósito — hay una lista puntual de RFCs de volumen muy alto a los que se les aisló su propio índice para evitar degradación de performance sobre el índice genérico, y hoy son solo esos 3. No afecta la lógica de este job (el patrón invoices,invoices-* los cubre a todos igual).

Se sacó el modo --historical (BQ completo vs ELK completo) — no tiene sentido conceptual

Corriendo el job en modo histórico se observó ~50% de RFCs cuadrando y una completitud de uuids muy baja — no era un bug, era la comparación en sí la que estaba mal planteada. Dos evidencias, no solo intuición:

1. BigQuery tiene un backfill que ELK nunca pudo recibir. Conteo de uuids distintos por año de issuedat en syntage.invoices: 2014-2019 se mueve entre 183K y 312K por año; 2020 salta a 10.3M y sigue creciendo hasta 25M+ en 2025. Es la firma de una carga histórica única (vía API de Syntage) directo a BigQuery. El volumen real de producción — y el webhook en tiempo real que también escribe a ELK — arranca recién en 2020. ELK nunca tuvo esas ~1.4M facturas de 2014-2019 porque el pipeline que lo alimenta no existía todavía.
2. Elasticsearch tiene una política de retención activa, a diferencia de BigQuery. Confirmado en Slack (#engineering, 2026-05-17, Gerónimo Afonso, hilo "Cleanup + Migración de Conductor", ticket DO-930): "Cleanup (retención 90 días) · Elasticsearch: 27 GB → 5 GB (-80%), 310 índices temporales viejos eliminados". Salvedad honesta: el mensaje dice "índices temporales viejos", no confirma al 100% que sea sobre el contenido del índice invoices en particular — pero sí confirma que ES en Solvento se trata como almacenamiento operativo/reciente, se poda periódicamente, y no es un sistema de verdad histórico como BigQuery.

Además, se confirmó (Martin) que el dashboard 232 al que esta capa reemplaza tampoco comparaba histórico contra ELK — su "histórico" era BigQuery vs Syntage (API), una comparación completamente distinta que vive en otra capa (capa 3, pendiente).

Decisión: el job (consistencia_elk_bq_syntage.py) ya no tiene modo --historical. Solo queda la ventana de 90 días (la única comparación BQ-vs-ELK que tiene sentido, dado que ambos lados deberían tener esa data reciente). El endpoint /consistency/historical en main.py se dejó respondiendo 410 (en vez de borrarlo) para que, si el Cloud Scheduler consistencia-historical todavía le pega, falle de forma visible en los logs en vez de 404 silencioso. Pendiente: borrar ese Cloud Scheduler. Si en el futuro hace falta una cuadratura de largo plazo, el camino correcto es BQ vs Syntage (capa 3), no BQ vs ELK.

Se implementó y se descartó la cuadratura por presencia de uuid (two-tier / uuid-set) — 2026-09-02

Rama feat/consistencia-uuid-set (commit e7326ec), nunca desplegada. La idea era reemplazar la comparación por conteos con una por presencia de uuid, con un nivel 1 barato (conteos por RFC) que filtrara y un nivel 2 exacto (uuid por uuid) solo sobre los RFCs que el nivel 1 marcara. Traía además una tabla nueva report.invoices_ingestion_consistency_missing con el uuid, issuedat y load_dts de cada faltante — o sea, el criterio "backfill activo vs gap real" resuelto en el dato, sin tener que ir a buscarlo a mano con investigate_rfc.py.

El nivel 1 no filtra nada. Medido con compare_methods() sobre datos reales:

	2026-09-02
RFCs con diferencia de conteo	896 de 1.387 (65%)
uuids que van al nivel 2	3.856.032 de 4.114.739 (93,7%)
ETA de la corrida	83 min

No es un bug: es consecuencia directa de escalar un RFC ante cualquier diferencia de conteo, en los dos sentidos. Esa decisión es correcta —un excedente puede enmascarar un faltante dentro del mismo RFC, ver el caso rot7802026za— y es exactamente lo que anula el ahorro. Con estos datos two-tier degenera en el barrido completo uuid-set, que era justamente el método que pretendía evitar.

Por qué no se puede desplegar así: el Cloud Run tiene timeout = "3600s" (60 min, el techo que acepta Cloud Run) y max_instances = 1. Una corrida de 83 min falla por timeout — y falla más los días de incidente, porque son los días en que el nivel 1 escala más RFCs. El método se vuelve más caro justo cuando más se lo necesita.

Dónde está el cuello, para no volver a optimizar la parte equivocada: no es ELK. La corrida se cortó paginando 3,86M de filas por la API REST de BigQuery (page_iterator). Si se retoma, el camino no es afinar el filtro del nivel 1 sino cambiar la forma: llevar los uuids de ELK a BigQuery y hacer el anti-join del lado de BQ, en vez de traer un lado entero a Python.

Proporcionalidad — la razón de fondo. El método simple que está en producción detectó el incidente del 2026-09-01 sin problemas (4.989 faltantes en 2 RFCs, sin falsos negativos). Lo único que no dio fue "¿backfill o gap real?", y eso lo responde investigate_rfc.py en segundos para los 2 RFCs que importan. El método nuevo resolvía esa misma pregunta para los 1.387 RFCs, todos los días. Ahí está el 83x.

Punto medio, si en algún momento hace falta la tabla _missing y no el método nuevo: acotar write_missing() a los RFCs que el método actual ya marca con déficit (no excedente). El 2026-09-02 eso eran 2 RFCs y 4.989 uuids — milisegundos. Da el diagnóstico (uuid + issuedat + load_dts) sin tocar la cuadratura. Es una feature nueva y chica, no la continuación de esta rama.

Capa 3 — BigQuery vs la API de Syntage (implementada 2026-07-24)

consistencia_bq_syntage.py, endpoint POST /consistency/api-source, diario (no horario: consulta la API una vez por RFC, ~1439 requests con rate limit de ~10 req/s).

Comparación: COUNT(DISTINCT id) GROUP BY rfc en BQ contra hydra:totalItems de /taxpayers/{rfc}/invoices, con el mismo rango de issuedAt en los dos lados.

Por qué acá el conteo simple alcanza (y en capa 1 no)

Capa 1 necesita dedup por (uuid, rfc) y lógica de colisión emisor/receptor porque ELK colapsa la misma factura a un documento (_id = uuid), así que las 2 filas de BQ (una como emisor, otra como receptor) chocan. La API de Syntage se consulta por RFC, igual que cada fila de BQ tiene su rfc: los dos lados están scopeados al mismo taxpayer, la comparación es simétrica y el conteo simple es correcto.

Validado con datos antes de escribir el job: de 848,348 pares (uuid, rfc) con más de una fila en 90 días, 848,348 tienen un solo id distinto y 0 tienen id variable — o sea COUNT(DISTINCT id) colapsa el log de eventos correctamente sin necesitar ROW_NUMBER(). Además COUNT(DISTINCT id) == COUNT(DISTINCT uuid) por RFC.

issuedAt[before]: el job viejo comparaba mal

La versión anterior solo usaba issuedAt[after], sin techo de ventana — contaba facturas más nuevas que el rango filtrado en BQ, inflando sistemáticamente el lado Syntage. Ahora se manda también issuedAt[before]. Confirmado que funciona con el modo --validate: 4 RFCs de referencia dieron diferencia exactamente 0 contra BQ en una ventana fija y cerrada (2026-04-01 → 2026-07-01), lo que a la vez prueba que [before] se respeta, que la API cuenta ambos roles (emisor y receptor) igual que BQ, y que el grano coincide.

Resultado del primer barrido completo (2026-07-24)

1439 RFCs · 98.5% cuadran (1417) · 22 descuadran · 44 facturas faltantes en BQ de ~3M (0.0015%) · 2 de excedente · 0 errores de API, 0 404.

Los 44 faltantes NO están dispersos ni en el borde reciente de la ventana: están en ventanas cerradas por RFC (ej. utm931109i9a 4-jun 15:10→5-jun 03:36; lsm100325a99 13-may 15:38→18:39). Diagnóstico con investigate_rfc_syntage.py + verificación en BQ: Syntage entrega en tandas (createdat agrupado en horas puntuales, issuedat amplio), y el lote sí llegó — para utm931109i9a el lote de prefijo a1f38486 (creado 5-jun 15:18) tiene 41 facturas en BQ y las 4 faltantes de ese mismo prefijo caen dentro de su rango. Es pérdida parcial de lote: se cayeron facturas sueltas de un lote que por lo demás se procesó bien.

Corolario para el buffer: no hay ni un faltante cerca del borde reciente, así que SETTLE_BUFFER_HOURS = 6 no es corto (sigue sin validarse si es más de lo necesario).

NO existe DLQ en el Dataflow — una factura que falla se pierde sin rastro recuperable

Verificado en templates-dataflow/invoices-api/dataflow_template.py: el parámetro --dlq_topic y Config.PUBSUB_DLQ_TOPIC existen pero no se referencian en ningún lado del pipeline. Los elementos fallidos van a ErrorLogger, que solo hace logging.error(...) y descarta — el comentario en el pipeline lo dice literal: # Log errores detallados (reemplaza DLQ). El topic de DLQ está vacío porque nunca nada lo escribió.

Consecuencias:

1. Capa 3 no es un "nice to have": es la única red de seguridad. Sin DLQ y con logs que expiran, la única forma de saber que una factura faltó es compararse contra Syntage.
2. La cadencia diaria es un requisito, no una preferencia. Cloud Logging retiene 30 días en el bucket _Default. Si el descuadre se detecta al día siguiente, quedan ~29 días para investigar; si se detecta a los 2 meses (como pasó con estos 44), los logs ya expiraron y no hay nada que debuggear. Esto argumenta por una alerta de Metabase, no solo un dashboard que se mire de vez en cuando.
3. Ese --dlq_topic muerto es una trampa: cualquiera que lea el código asume que los fallos se recuperan. Conviene borrarlo o implementarlo de verdad, pero no dejarlo sugiriendo una garantía inexistente.

Runbook: qué hacer cuando capa 3 marca un descuadre

1. python3 investigate_rfc_syntage.py --from-last-run (o pasando los RFCs) — pagina la lista real de ids de Syntage y la diferencia contra BQ. Devuelve los id/uuid/issuedAt exactos que faltan, más un histograma por día para distinguir buffer corto vs pérdida real.
2. Buscar ese id en Cloud Logging (dentro de los 30 días). Se puede buscar por el id directamente: el id de Syntage viaja en la URL del webhook (.../invoices/{id}), y ErrorLogger loguea Full message: {message_str} con el JSON completo. Strings útiles: === [invoices] FAILED MESSAGE === y [invoices] API FETCH FAILED for URL: (este último es el que se dispara cuando APIClient.fetch_data() agota sus 3 intentos — el escenario de una factura suelta que no se pudo traer).

3. Reprocesar. A diferencia de capa 1 (donde la factura estaba en BQ y solo faltaba empujarla a ELK), acá no está en ninguno de los dos sinks: hay que traerla de Syntage. El camino es el mismo que ya se validó en capa 1 — encender invoices-api-prod y publicarle mensajes con el shape {"rfc", "via_api": true, "url": ".../invoices/{id}", "webhook_uuid"} — pero la parte que arma la lista todavía no está escrita: publish_reprocess.py la saca de BigQuery, y acá justamente el dato en BQ no está. Ver "Cómo se adapta este patrón a capa 3". El insumo es la lista de ids que ya devuelve investigate_rfc_syntage.py.

   El fix de índices dedicados ya está aplicado (2026-07-27)

   ElkWriter.process() tenía un RFC con índice dedicado hardcodeado (tsp0004051i4) cuando ELK tiene tres (cas000522rc7, tsp0004051i4, rot7802026za). Ya se corrigió a un set en dataflow_template.py, y se validó en producción reprocesando los 30.675 documentos de rot7802026za. La misma lista está duplicada en DEDICATED_INDEX_RFCS de consistencia/reindex_missing.py: si ELK agrega o quita un índice dedicado, actualizar los dos.

   Recordar que arreglar el fuente no alcanza: hay que republicar el template (./deploy_templates.sh invoices-api prod) y apuntar template_gcs_path al nuevo, o el job levanta con el artefacto viejo.

Blind spot aceptado a propósito: el universo de RFCs sale de BigQuery

Fase A saca la lista de RFCs con SELECT DISTINCT rfc de syntage.invoices, no de Syntage. Consecuencia: un RFC que Syntage tenga y que nunca haya entrado a BQ (ni una factura, jamás) es invisible para este job. Se aceptó para no depender del endpoint de taxpayers de Syntage (call_credentials() existía en la versión vieja del job pero nunca se llamaba). La URL sí se conoce: está en reprocesos/invoices/invoices-prod-publisher.py, call_credentials_api() — GET /credentials?properties[]=id&properties[]=rfc&status=valid&itemsPerPage=1000, header X-API-Key, paginado con hydra:next. O sea que el blind spot es una decisión de alcance, no un desconocimiento, y cerrarlo cuesta una función que ya está escrita. Si algún día importa cubrir ese caso, hay que sacar la lista de RFCs de Syntage y unirla con la de BQ.

¿Llegaste hasta acá desde…?

Si ves…	Eso sale de…	La lógica está en…
Faltantes reales de invoices en ELK (capa 1, últimos 90 días)	solvento-data-prod.report.invoices_ingestion_consistency_summary (mode=elk-ingestion)	Cloud Run consistencia-elk-bq-syntage → reconcile() en consistencia_elk_bq_syntage.py
Detalle de qué RFC/uuid falta en ELK y por qué	solvento-data-prod.report.invoices_ingestion_consistency_detail	Distingue colision_otro_rfc (cargado, no es faltante) de faltantes (real)
Facturas que Syntage tiene y BigQuery no (capa 3)	solvento-data-prod.report.invoices_source_consistency_summary, columna faltantes_bq	consistencia_bq_syntage.py → endpoint /consistency/api-source, diario
Detalle por RFC de capa 3 (incluye por qué un RFC no se pudo comparar)	...invoices_source_consistency_detail	Columna estado: ok / no_encontrado (404) / error — los dos últimos NO cuentan como faltante
⚠️ Buscando un reporte histórico BQ-vs-ELK (sin filtro de fecha)	No existe — se sacó (ver danger box "Se sacó el modo --historical")	/consistency/historical responde 410
"¿Hasta qué fecha de factura está confirmada la cuadratura?" (no "¿cuándo corrió el job?")	Columna window_end en invoices_ingestion_consistency_summary (última fila por run_at)	El watermark — ver compute_window()
⚠️ invoices_consistency_issues[_90] en 0 filas	Tabla muerta desde ago-2025	No la alimenta nada; no usar como referencia viva
⚠️ invoices_consistency_report[_90] con datos pero por conteo de RFC	Versión previa de este mismo job (pre-rediseño)	Superada por el grano uuid; migrar consumo cuando el dashboard nuevo esté validado

Resumen

Campo	Valor
Origen	ELK (invoices,invoices-*), BigQuery (syntage.invoices), API de Syntage (/taxpayers/{rfc}/invoices)
Patrón	api-pull-programado (Cloud Run agendado por Cloud Scheduler)
Destino	capa 1: report.invoices_ingestion_consistency_summary + _detail · capa 3: report.invoices_source_consistency_summary + _detail (todas WRITE_APPEND, partición diaria por run_at)
Frecuencia	Las dos diarias (2026-07-24): capa 1 0 4 * * * MX (consistencia-elk-ingestion, antes horaria) · capa 3 30 6 * * * MX (consistencia-api-source)
Despliegue	./publish_container.sh prod en el repo consistencia solo buildea y publica la imagen (tags: latest, short SHA, timestamp). El Cloud Run y los dos schedulers los crea terraform-gcp-infrastructure (cloud_run_v2_services_prod + cloud_scheduler_jobs_prod, módulo cloud_scheduler nuevo)
Acceso	Servicio privado (allow_unauthenticated = false). Los schedulers lo invocan con OIDC usando sa-scheduler-data-prod@solvento-data-prod.iam.gserviceaccount.com. Para llamarlo a mano: curl -X POST -H "Authorization: Bearer $(gcloud auth print-identity-token)" <url>/consistency/elk-ingestion
Criticidad	🟡 Media — es la señal de salud de la ingesta del flujo más crítico del ecosistema (Syntage invoices), aunque el job en sí no alimenta producto
Dueño	Martin
Repos	consistencia

Diagrama de linaje

graph LR
    ELK[["🔎 Elasticsearch<br/>invoices,invoices-*"]]
    API[["🌐 API de Syntage<br/>/taxpayers/{rfc}/invoices"]]
    BQSRC[("solvento-data-prod<br/>syntage.invoices")]
    SCH(["⏰ Cloud Scheduler (Terraform)<br/>consistencia-elk-ingestion · 4:00 MX<br/>consistencia-api-source · 6:30 MX"])

    subgraph job["Cloud Run · consistencia-elk-bq-syntage"]
        CR1["CAPA 1 · reconcile()<br/><i>grano uuid, distingue colisión<br/>emisor/receptor de faltante real</i>"]
        CR3["CAPA 3 · reconcile()<br/><i>COUNT(DISTINCT id) por rfc<br/>vs hydra:totalItems</i>"]
        SUM[("report.invoices_ingestion_<br/>consistency_summary + _detail")]
        SRC[("report.invoices_source_<br/>consistency_summary + _detail")]
    end

    MB["📊 Metabase 917<br/>Panel de inconsistencias V2"]

    SCH --> CR1
    SCH --> CR3
    ELK --> CR1
    BQSRC --> CR1
    BQSRC --> CR3
    API --> CR3
    CR1 --> SUM
    CR3 --> SRC
    SUM --> MB
    SRC --> MB

    classDef src fill:#fde68a,stroke:#b45309,color:#000;
    classDef store fill:#bfdbfe,stroke:#1e40af,color:#000;
    classDef logic fill:#dcfce7,stroke:#15803d,color:#000;
    class ELK,API,SCH src;
    class BQSRC,SUM,SRC store;
    class CR1,CR3 logic;

Paso a paso

Cloud Run consistencia-elk-bq-syntage (repo, consistencia_elk_bq_syntage.py + main.py)

Flask app con endpoints GET /, POST /consistency, /consistency/elk-ingestion (ambos hacen lo mismo: ventana de 90 días), disparados por Cloud Scheduler. /consistency/historical sigue existiendo pero solo responde 410 — ver danger box arriba. Cada llamada ejecuta run():

1. compute_window() — recalcula en cada llamada (no a nivel de módulo — Cloud Run reutiliza contenedores "calientes" por horas/días) la ventana de fechas: límite superior window_end = ahora - SETTLE_BUFFER_HOURS (el watermark) y límite inferior window_end - 90d.
2. fetch_bq() — en BigQuery, primero dedupea a la última fila por (uuid, rfc) (por load_dts, ver info box "log de eventos" arriba — syntage.invoices agrega una fila por evento, no upsertea), y recién sobre ese resultado deduplicado cuenta uuids distintos por rfc (COUNT(DISTINCT uuid) GROUP BY rfc, normalizando rfc a minúsculas) filtrando por issuedat fuera del buffer (se probó filtrar además por load_dts para tapar backfills y se revirtió — ver danger box arriba). Detecta por separado los uuids que aparecen bajo más de un RFC (el caso emisor/receptor — set chico, es lo único que después se cruza contra ELK).
3. Consultas a ELK — elk_total_docs, elk_doc_count_per_rfc (agregación terms sobre rfc.keyword normalizado con script Painless) y elk_distinct_uuid (cardinality sobre uuid.keyword, aproximado — ver warning arriba), todas con el mismo filtro de ventana.
4. elk_rfcs_for_uuids() — para los uuids compartidos (emisor/receptor), busca en ELK bajo qué RFC(s) quedó guardado cada uno (por el routing de shards, a veces coexiste en ambos).
5. reconcile() — calcula la cuadratura por RFC (exacta, sin aproximar), separando colision_otro_rfc (compartido, cargado bajo el otro RFC — no es faltante) de faltantes (uuid genuinamente ausente en ELK). De ahí suma faltantes_uuid_exacto (la métrica real de data ausente) y excedente_elk_exacto (caso inverso: ELK con más que BQ para ese RFC). La cuadratura global por cardinality de ES (faltantes_uuid_aprox) se calcula también pero solo queda como diagnóstico.
6. write_summary() / write_detail() — escriben con insert_rows_json + WRITE_APPEND (no truncan). Las tablas se crean solas en la primera corrida, particionadas por día de run_at (expiración configurable, default 400 días) para que Metabase pueda mostrar tendencia y frescura, y para que un job caído se note (el run_at más reciente deja de avanzar) en vez de desaparecer sin dejar rastro.

Tablas que alimenta

Tabla	Contenido	Job que la carga
solvento-data-prod.report.invoices_ingestion_consistency_summary	Capa 1 — 1 fila por corrida: métricas globales (faltantes reales en ELK, % uuid ok, % rfc ok)	Cloud Run consistencia-elk-bq-syntage
solvento-data-prod.report.invoices_ingestion_consistency_detail	Capa 1 — 1 fila por (rfc, corrida): detalle para drill-down	Cloud Run consistencia-elk-bq-syntage
solvento-data-prod.report.invoices_source_consistency_summary	Capa 3 — 1 fila por corrida diaria: faltantes_bq (facturas que Syntage tiene y BQ no), excedente_bq, % rfc ok	Cloud Run, endpoint /consistency/api-source
solvento-data-prod.report.invoices_source_consistency_detail	Capa 3 — 1 fila por (rfc, corrida): bq_count, syntage_count, faltantes, estado	Cloud Run, endpoint /consistency/api-source
⚠️ invoices_consistency_report[_90]	Versión previa (grano RFC-count), sigue viva pero deprecada conceptualmente	Versión anterior del mismo Cloud Run — migrar y luego retirar
❌ invoices_consistency_issues[_90]	Muerta desde ago-2025 (0 filas)	Nada la escribe hoy — no usar

Consumo aguas abajo

• Dashboard Metabase "Data Quality · Syntage Invoices — Ingesta" (a crear, spec en consistencia/METABASE_CARDS.md): scorecards de faltantes reales y % ok por modo, frescura del último run_at, tendencia en el tiempo, y detalle por RFC. Separado del 232 para no arrastrar su deuda de tablas muertas.

Si falla 🛠️

Impacto: si el job deja de correr, MAX(run_at) en invoices_ingestion_consistency_summary deja de avanzar — a diferencia de la versión con TRUNCATE, esto sí se puede detectar con un scorecard de frescura en Metabase (ver spec de cards). Criticidad media: no alimenta producto, pero es la única señal de salud real de la ingesta de invoices.

Síntoma	Dónde mirar	Acción
run_at no avanza (frescura vieja)	Logs Cloud Run consistencia-elk-bq-syntage; ¿corrió el Scheduler?	Re-invocar POST /consistency/elk-ingestion
El job tarda o hace timeout	Volumen de uuids compartidos a verificar en ELK (elk_rfcs_for_uuids, batches de 1000)	Debería ser rápido (solo cruza el set de uuids compartidos, no todo el índice); si tarda, revisar tamaño de ese set
ELK devuelve 0 / error	Conectividad a ELK_HOST, credenciales ELK (env var ELK_PASSWORD, sin default hardcodeado desde este rediseño)	Verificar Elasticsearch; revisar logs de la query a ELK
faltantes_uuid sube de forma sostenida (no un pico puntual)	invoices_ingestion_consistency_detail, columna faltantes por RFC	Esto sí es señal real de data ausente en ELK: revisar la ingesta (webhook/Dataflow) de syntage.invoices, no este job

Escalación: Martin (Cloud Run / ingesta). Si la discrepancia apunta a la ingesta de invoices, seguir el runbook de Syntage invoices.

Remediación: qué hacer con un RFC que descuadra

Esta capa hasta acá solo detecta (compara agregados). Para pasar de "sé que a dte130830ix6 le faltan 47" a "esos 47 ya están en ELK" hay dos caminos, y la elección depende de la escala:

	Faltantes sueltos	Faltantes a escala
Cuándo	decenas de uuids, 1-3 RFCs	miles de uuids, decenas o cientos de RFCs
Herramienta	reindex_missing.py	publish_reprocess.py
Camino	BQ → ELK, escritura directa	BQ → Pub/Sub → invoices-api-prod → BQ + ELK
Formato del doc	lo arma el script (réplica manual)	lo arma el pipeline de producción
Reanudable	no	sí (checkpoint)
Requiere infra	no	prender el Dataflow y apagarlo al terminar

Los dos son manuales, no desplegados, y viven en la raíz de consistencia/ (se corren desde la raíz, importan de cloud_run.*).

Paso 0 de cualquiera de los dos: investigate_rfc.py <rfc> — a diferencia del job de producción (que solo compara conteos), lista uno por uno los uuids exactos que están en BQ y no en ELK, con su issuedat y load_dts. Esas dos fechas son la pista para distinguir ruido de problema real: load_dts recientísimo + issuedat viejo = probable backfill en curso (se resuelve solo, no hace falta actuar); issuedat y load_dts ambos viejos y estables = gap real.

Camino A — faltantes sueltos: reindex_missing.py

reindex_missing.py <rfc> [--verify] [--apply] toma la lista de faltantes, trae la fila completa desde BigQuery (fuente de verdad — ya sabemos que ahí el dato está bien) y la reindexa directo en ELK vía _update con doc_as_upsert, replicando el mapeo índice/routing de ElkWriter. Sin --apply es dry-run. Correr --verify antes de --apply la primera vez que se toca un RFC nuevo.

Sus dos límites, que son la razón de existir del camino B

1. Escribe doc por doc, secuencial y sin checkpoint. Para decenas de miles de documentos son horas, y si el proceso se corta no hay forma de saber dónde quedó.
2. El documento lo arma este repo, no el pipeline (CANONICAL_FIELDS, TIMESTAMP_FIELDS, DEDICATED_INDEX_RFCS). Cada divergencia con lo que SchemaManager escribe hoy es un bug silencioso. Y --verify no puede detectarlo cuando el RFC no tiene ningún doc en ELK contra el que comparar — que es justo el caso de un onboarding nuevo cuya carga inicial se perdió entera.

Camino B — faltantes a escala: publish_reprocess.py

Publica a Pub/Sub el mismo mensaje que publicaría el webhook y deja que el pipeline de producción haga el trabajo: Pub/Sub marca el ritmo, Dataflow reintenta y autoescala, y el documento sale con el formato correcto porque lo produce el mismo código que el resto de la ingesta.

Shape del mensaje — es el shape vía API que espera InvoiceProcessor.process(), el mismo que usaba el publicador viejo. Ojo: el webhook real publica invoices con el otro shape que soporta el pipeline (via_api: false + body con el JSON del evento), así que esto no es "lo mismo que el webhook" sino la otra rama del mismo consumidor. El campo id es redundante (el pipeline no lo lee, viaja solo para poder leer el dry-run):

{"webhook_uuid": "rp0727:<uuid4>", "rfc": "<RFC MAYÚSCULA>", "via_api": true,
 "url": "https://api.syntage.com/invoices/<id de Syntage>", "id": "<id de Syntage>"}

Topic: invoices-api-prod (dedicado al reproceso, distinto del invoices-webhook-prod del webhook). La subscription que consume el Dataflow es invoices-api-subscription.

Secuencia completa:

# 1. Prerequisitos de infra (una vez)
#    - secretos `syntage-api-key` y `elk-password` en Secret Manager
#    - roles/secretmanager.secretAccessor a la SA del worker de Dataflow
# 2. Republicar el template — NO alcanza con arreglar el fuente
cd templates-dataflow && ./deploy_templates.sh invoices-api prod
# 3. terraform.tfvars -> descomentar el bloque invoices-api-prod,
#    template_gcs_path = el path nuevo con timestamp -> plan -> apply
# 4. Verificar en los logs del worker que los secretos se resuelven:
#    "API Key de Syntage resuelto desde Secret Manager: projects/..."
# 5. Test chico: un RFC del índice genérico y OTRO de índice dedicado
cd consistencia
python3 publish_reprocess.py --rfc <rfc chico>            # dry-run
python3 publish_reprocess.py --rfc <rfc chico> --publish
python3 publish_reprocess.py --rfc rot7802026za --limit 1 --publish
python3 investigate_rfc.py <rfc chico>                    # confirmar en ELK
# 6. Corrida completa (toma los RFCs y la ventana de la última corrida de capa 1)
python3 publish_reprocess.py --publish [--load-dts-desde YYYY-MM-DD]
# 7. Verificar, y recién entonces APAGAR el Dataflow (comentar el bloque en Terraform)

El paso 2 es el que se olvida, y falla en silencio

terraform.tfvars apunta a un template_gcs_path con timestamp fijo. Arreglar dataflow_template.py en el repo no cambia el artefacto en GCS: si se levanta el job sin republicar, corre el código viejo. En el incidente del 25-jul eso habría significado escribir los 30.675 documentos de rot7802026za en el índice equivocado.

Probar SIEMPRE un RFC de índice dedicado, no solo uno genérico

ELK tiene tres índices dedicados (cas000522rc7, tsp0004051i4, rot7802026za) y el resto va al genérico invoices con routing. Son dos caminos de código distintos en ElkWriter.process(). Un test que solo cubre el genérico no prueba nada sobre los RFCs de mayor volumen — uno de ellos, rot7802026za, concentró el 20,5% de los faltantes de este incidente.

Decisiones de diseño del camino B, y la evidencia detrás

Por qué lista dirigida de uuids y NO ventana de fechas (que era el método histórico)

El método anterior (reprocesos/invoices/invoices-prod-publisher.py, corrido en dic-2025) paginaba la API de Syntage por RFC con issuedAt[after]=<hoy-91d> y publicaba todo lo que devolvía. Su log de diciembre lo cuantifica: 1.232.593 facturas publicadas, y una sola corrida (2025-12-16) publicó 1.037.105 en 2h46m.

No es solo ineficiente: no cubre el caso. El eje de una caída de ELK es load_dts (cuándo se escribió), no issuedat (cuándo se emitió), y los dos no se superponen. Medido en el incidente del 25-jul:

RFC	faltantes	issuedat en el finde	issuedat últ. 7d
rot7802026za	30.675	9.347	47.122
tec130927pd9	10.833	148	670

tec130927pd9 es un onboarding nuevo: load_dts de esa semana, issuedat repartido en 90 días. Una ventana de 7 días por emisión habría recuperado 670 de 10.833, el 6%. Cubrirlo con fechas obliga a barrer los 90 días completos de cada RFC: ~3,06M facturas reprocesadas para arreglar 149.609.

--load-dts-desde: cuándo es válido acotar por fecha de escritura, y cómo verificarlo

Acota los candidatos a lo escrito desde una fecha. Medido en el incidente, sobre los mismos 420 RFCs y para encontrar los mismos faltantes:

piso de load_dts	candidatos	requests a ELK
ninguno (90 días)	2.205.278	~2.200
2026-07-23 (el que se usó)	216.993	~217
2026-07-24	209.850	~210

Se eligió el 23 y no el 24 porque cuesta apenas 7.143 candidatos más (3%) y da un día completo de holgura sobre la última corrida limpia.

Es válido si el incidente fue indisponibilidad de escritura (ELK dejó de recibir pero no perdió lo que tenía). NO es válido si ELK perdió documentos preexistentes: ahí hay faltantes con load_dts viejo y el filtro los vuelve invisibles, en silencio.

Cómo distinguirlo, con el chequeo real que se hizo: elk_total_docs bajó 81.258 entre la corrida del 24 y la del 27, lo que a primera vista parece pérdida. Pero la ventana es rolling de 90 días y el corrimiento de 3 días saca 86.515 uuids (los de issuedat entre el 25 y el 28 de abril). Los dos números coinciden dentro del 6% ⇒ la caída la explica la ventana, no una pérdida ⇒ el piso era seguro.

Cerrar siempre con un barrido sin piso en dry-run: es el cinturón de seguridad del razonamiento. Si el piso fue correcto, da cero.

Esto es para un reproceso puntual con ventana de incidente conocida. No es un filtro para el job diario — lo convertiría en incremental y le sacaría lo que lo hace una red de seguridad. Es primo del filtro de load_dts que ya se probó y se revirtió en fetch_bq() (ese era un techo, peor todavía).

webhook_uuid = <tag>:<uuid4>, aleatorio y no derivado del uuid del CFDI

Tiene que ser distinto en cada publicación de cada mensaje, incluso para la misma factura, porque Pub/Sub es at-least-once y un bundle de Dataflow puede reintentarse. Con un webhook_uuid determinístico (<tag>:<uuid del CFDI>), una reentrega escribe dos filas en syntage.invoices con el mismo webhook_uuid — que es exactamente lo que la dedup posterior no espera.

El prefijo <tag>: mantiene la trazabilidad (WHERE webhook_uuid LIKE 'rp0727:%') y el uuid4 garantiza unicidad. Largo total 43 chars, bajo el VARCHAR(50) que declara el SchemaManager.

El tag se deriva del run_at de la corrida de capa 1, no de la fecha de hoy: un reproceso grande cruza la medianoche UTC, y si el tag cambiara de día, cada corrida nueva estrenaría checkpoint (republicando todo) y la auditoría por prefijo solo vería una parte de las filas.

Costo en BigQuery: 3 queries, ~USD 0,009 por corrida

Medido en INFORMATION_SCHEMA.JOBS durante el incidente: target_from_last_run (0,02 GB) + detect_uuid_column (0,01 GB) + start_query (1,42 GB) = USD 0,0089. La paginación de los 2,2M candidatos no genera queries nuevas (job.result(page_size=…) lee de la tabla temporal del resultado vía la API, sin facturar bytes).

Dato para no volver a optimizar esto de memoria: el filtro de RFC en investigate_rfc.fetch_rfc_uuids está después del ROW_NUMBER(), así que el escaneo no depende del tamaño del RFC — cuesta 0,956 GB para uno de 92 facturas y para uno de 700k. Lo único que movía la aguja era hacer una query en vez de una por RFC (420 corridas ≈ USD 2,46 contra USD 0,009). Y el cluster pruning por rfc solo rinde con pocos RFCs: con uno baja de 0,956 a 0,411 GB, con 60 sube a 1,358 GB.

Evidencia: incidente ELK del fin de semana 2026-07-25/26

ELK estuvo caído el sábado y el domingo. El job de capa 1 no corrió ninguno de los dos días.

Corrida (UTC)	faltantes_uuid	RFCs descuadran	pct_uuid_ok
24-jul 21:39 (previa)	0	1	100%
27-jul 19:50 (diagnóstico)	149.609	420	95,04%
27-jul 22:49 (post 1ª pasada)	2.271	12	99,92%
27-jul 23:05 (post 2ª pasada)	0	4	100%

• Concentración: 3 RFCs (rot7802026za, tne920818nj6, tec130927pd9) eran el 38% del volumen; los 246 más chicos, el 5,7%.
• 37 RFCs con elk_count = 0 — sin un solo documento en ELK. Eran onboardings nuevos cuya carga inicial completa cayó durante la caída, y ninguno aparecía en la corrida del 24.
• Throughput real: 149.498 filas en 58 minutos = 43 filas/s, con worker_num_workers=1 y API_RATE_LIMIT=3. Muy por encima de lo estimado (~2 req/s) porque GlobalRateLimiter no coordina entre workers ni threads a pesar del nombre: cada instancia de APIClient tiene su propio contador.
• Segunda pasada necesaria: los 2.271 restantes eran 2.231 de TSM121009QE5 (un onboarding que empezó a cargar a las 20:46, después del diagnóstico, así que nunca estuvo en el alcance) más 40 de residuo. Se resolvieron con --tag rp0727b para estrenar checkpoint.

Al reprocesar en tandas, usar un --tag nuevo en cada pasada

El checkpoint registra "publicado a Pub/Sub", no "llegó a ELK". Como el Dataflow no tiene DLQ, un mensaje que falla en el fetch se descarta y el checkpoint lo deja marcado como hecho: una segunda corrida con el mismo tag lo saltearía para siempre. Con --tag nuevo se re-evalúa todo contra ELK, que es la fuente de verdad real.

Efectos colaterales conocidos de un reproceso masivo

• Duplicados en syntage.invoices: esperado y correcto. BQ es append (STREAMING_INSERTS, ignore_insert_ids=True), ELK es upsert por uuid. La asimetría es lo que hace seguro republicar: BQ acumula historia, ELK converge.
• 73 filas con webhook_uuid repetido (0,049% de 149.498): mismo id, mismo rfc, escrituras separadas por 1-3 segundos, máximo 2 repeticiones, cero casos emisor/receptor. Firma de reentrega de Pub/Sub. No es evitable desde el publicador — el webhook_uuid viaja dentro del mensaje, así que una reentrega lo duplica; la única forma sería generarlo en el worker.
• Sin riesgo para dbt, verificado: dbt-etls/models/advanced_analytics/invoices.sql es incremental_strategy='merge' con unique_key='id', pero ya dedupea antes del MERGE — CTE deduped, row_number() over (partition by id order by load_dts desc, updatedat desc nulls last, …) con where rn = 1. El source entrega una fila por id, así que las reentregas no pueden producir must match at most one source row. (El único fan-out vivo en ese modelo es el LEFT JOIN a analiticas.exchange_rates, que ocurre después del row_number(): si esa tabla tuviera más de una fila por fecha, ahí sí fallaría el MERGE — y no tiene nada que ver con los reprocesos.)
• Competencia por ELK con la ingesta en vivo: TSM121009QE5 cargó entre 20:46 y 21:38 y el reproceso empezó 21:43 con 149k upserts. No se solapan del todo, pero quedó la hipótesis (a confirmar en los logs de invoices-webhook-prod) de que escrituras del webhook se perdieran por throttling. Para el próximo reproceso masivo: correrlo en ventana de baja ingesta.

Caso de validación real: dte130830ix6, incidente ELK del 2026-06-25

Los 47 faltantes de dte130830ix6 (ver arriba) resultaron ser un caso real, no ruido: los 47 uuids tienen issuedat entre el 24-jun 14:00 y 21:18, y load_dts clavado en una ventana de 4 segundos el 25-jun a las 19:15:45-49 — la firma de un único proceso batch tocando esas filas exactas, coincidiendo con un error conocido en un proceso de ELK ese día. Como nunca se volvieron a tocar en BQ desde entonces, iban a seguir apareciendo como faltante en cada corrida para siempre (correctamente — la card de persistencia los seguiría marcando). Primer caso real donde se usó investigate_rfc.py para diagnosticar y reindex_missing.py para remediar.

Cómo se adapta este patrón a capa 3 (API ↔ BigQuery) y a items

El patrón reutilizable es lista dirigida → Pub/Sub → Dataflow → sinks, y sirve para las dos capas. Lo que cambia es de dónde sale la lista, y esa diferencia no es un detalle de implementación:

	Capa 1 (BQ ↔ ELK)	Capa 3 (API ↔ BQ)
Dónde está el dato	en BQ, correcto	en BQ no está; en ELK probablemente tampoco
Fuente de la lista	BQ (start_query)	la API de Syntage (paginar por RFC)
Costo de armar la lista	1 query, ~USD 0,009	~1 request por página por RFC
Qué escribe el reproceso	BQ y ELK (la escritura a BQ es redundante: agrega filas al log de eventos)	BQ y ELK
Estado	implementado y validado	pendiente

Para capa 3, publish_reprocess.py sirve de la mitad hacia adelante (checkpoint, batching, preflight del topic, shape del mensaje, chunking) pero start_query no aplica: no se puede pedirle a BQ los uuids que a BQ le faltan. Hay que reemplazar esa función por una que pagine la API de Syntage sobre los RFCs que capa 3 marcó, y quedarse con los id que no estén en BQ. investigate_rfc_syntage.py ya hace exactamente esa comparación por RFC — la lista de ids que devuelve es el insumo.

items solo escribe a BigQuery — no tiene capa 1

templates-dataflow/items-api tiene ElkWriter, pero su parámetro --elk_enabled viene en "false": los items no van a ELK. Sin dos sinks no hay nada que cuadrar a nivel ingesta, así que para items la única reconciliación posible es contra la fuente (capa 3).

consistencia/items_consistency/ es una herramienta separada, con su propio Cloud Run y publish_container.sh, que compara syntage.items contra la API. Estaba desactualizada al momento de revisarla; si se retoma, auditarla aparte antes de confiar en sus números.

Enlaces

• 📦 Job: consistencia — código desplegable en cloud_run/: consistencia_elk_bq_syntage.py (capa 1), consistencia_bq_syntage.py (capa 3), main.py, config.py. Solo buildea/publica la imagen (./publish_container.sh prod); el Cloud Run y los schedulers viven en terraform-gcp-infrastructure.
• 🔎 Diagnóstico manual capa 1: consistencia/investigate_rfc.py, consistencia/inspect_elk_fields.py
• 🔎 Diagnóstico manual capa 3 (fase B): consistencia/investigate_rfc_syntage.py
• 🛠️ Remediación manual capa 1 — faltantes sueltos: consistencia/reindex_missing.py
• 🛠️ Remediación manual capa 1 — a escala: consistencia/publish_reprocess.py (BQ → Pub/Sub → invoices-api-prod). Requiere prender el Dataflow; ver el runbook arriba. Capa 3: pendiente.

• 📜 Antecesor histórico (referencia, no usar): reprocesos/invoices/invoices-prod-publisher.py — publicaba por ventana de fechas desde la API. Su log de dic-2025 es la evidencia de por qué se cambió a lista dirigida. Tiene la API key de Syntage hardcodeada en texto plano, igual que count_invoices.py en el mismo repo.

  Los scripts manuales viven en la raíz del repo (no en cloud_run/, no se despliegan) y se corren desde la raíz, porque importan de cloud_run.*. Si el import falla con cannot import name 'GCP_PROJECT_ID' from 'config', es un paquete de PyPI llamado config haciendo shadow: los módulos de cloud_run/ prueban cloud_run.config primero y caen a config como fallback (ese orden es a propósito), así que el síntoma solo aparece con versiones viejas del repo. - 🧾 Diagnóstico del incidente 25-jul con los 420 RFCs y su clasificación: consistencia/diagnostico_elk_27jul2026.xlsx - 📋 Spec de cards Metabase: consistencia/METABASE_CARDS.md - 📊 Dashboard Metabase 917 "Panel de inconsistencias V2" (colección Consistencia) - 📊 Estado en vivo de la ingesta (no de esta cuadratura): dashboard "Data Processes" (Cloud Monitoring)

---

Última actualización: 2026-07-27 (implementada y validada la REMEDIACIÓN DE CAPA 1 A ESCALA: publish_reprocess.py, BQ → Pub/Sub → invoices-api-prod. Incidente ELK del 25-26/jul resuelto end-to-end: 149.609 faltantes → 0 en ~3,5 h. invoices-api-prod reactivado puntualmente con el fix de índices dedicados y secretos desde Secret Manager).

Pendientes abiertos:

• Rotar API_KEY de Syntage y ELK_PASSWORD. El cambio a Secret Manager corta la exposición futura, no la que ya existe: los valores estuvieron en templates-dataflow/invoices-api/prod.env y dev.env commiteados (la misma credencial en dev y en prod), embebidos en los artefactos de template en GCS, y hay una API key distinta hardcodeada en reprocesos/invoices/invoices-prod-publisher.py y count_invoices.py. Al rotar, darle a dev credenciales propias. Conviene que Compliance/Seguridad revise el alcance antes de cerrarlo.
• Migrar los demás templates de Dataflow a Secret Manager. Solo se hizo invoices-api (patrón en resolve_secret(), resuelto en el worker — InvoiceProcessor.setup() para la API key y ElkWriter._get_elk_connector() para la de ELK — y nunca en main(), porque en un classic template main() corre al construir y hornearía el valor en el artefacto. Consecuencia para debuggear: si el secreto de ELK no resuelve, el error aparece al procesar el primer elemento, no al arrancar el DoFn). Quedan 11, de los cuales solo 4 tienen credencial literal en su .env: buro, insights, items-api e invoices-webhook — este último es prioritario, porque es el pipeline vivo de producción y es el único que además tiene la password de ELK en texto plano.
• Importar a Terraform lo que ya existe fuera de él: el servicio consistencia-elk-bq-syntage y el job consistencia-elk-ingestion. Sin eso, el primer apply falla con already exists. Comandos exactos en los comentarios de terraform.tfvars.
• Borrar a mano consistencia-elk-bq-syntage-historical — sigue disparando cada día a las 7:30 contra un endpoint que responde 410. No está en Terraform a propósito (no puede borrar lo que nunca manejó).
• Configurar alerta en Metabase, no solo dashboard: la ventana de 30 días de retención de logs es lo que hace debuggeable un descuadre.
• Sincronizar los docstrings desactualizados de consistencia/reindex_missing.py e investigate_rfc_syntage.py: el primero declara invoices-api-prod "DETENIDO" y su código "desactualizado" (ya no), y el segundo sugiere como diagnóstico "revisar si la DLQ tiene mensajes colgados" (no hay DLQ, va a mandar a alguien a buscar en un topic vacío).
• Reprocesar las 44 facturas faltantes detectadas por capa 3 el 2026-07-24 (sin urgencia: sus logs ya expiraron; ver runbook).
• Implementar la remediación de capa 3 — reutilizar publish_reprocess.py reemplazando start_query por una función que pagine la API de Syntage. Ver la sección "Cómo se adapta este patrón a capa 3".
• Decidir si se implementa la DLQ de verdad o se borra el --dlq_topic muerto del template. Su ausencia es lo que hace que un fetch fallido se pierda sin rastro y que el checkpoint del publicador pueda mentir.
• Capa 2 (RAW vs las tablas de dbt/AA) sigue sin implementar.
• Auditar consistencia/items_consistency/ si se retoma (estaba desactualizado). Recordar que items no va a ELK.
• Corregido de paso: prod.env/dev.env de invoices-api tenían ELK_UER (typo). ELK_UER no llegaba por ningún camino y funcionaba de casualidad porque el default del argumento es elastic. (ELK_USER sí habría llegado, pero por os.getenv en el default, no por el flag: deploy_templates.sh mapea VAR → --var, y el argumento se llama --elk_username.) Ahora es ELK_USERNAME.

Si se agrega la capa 2, se cambia el estado de invoices-api-prod, o se agrega/quita un índice dedicado en ELK, actualizar esta ficha en el mismo MR.