CFDIs XML · Bucket → PostgreSQL → BigQuery

Cuando se sube un XML a un bucket GCS, una notificación llega por Pub/Sub (push) a un Cloud Run que parsea el CFDI y escribe sus campos en PostgreSQL (public.xml_data); de ahí Airbyte lo replica a solvento-data-prod.svc_master_public.xml_data (cada ~20 min).

¿Llegaste hasta acá desde…?

Si ves…	Eso sale de…	La lógica está en…
Datos de CFDIs (emisor/receptor/total/timbre) en BigQuery	`solvento-data-prod.svc_master_public.xml_data`	Airbyte (réplica) ← PostgreSQL `public.xml_data` ← `bucket-xml-parser` → `bucket_xml_parser.py`
Filas con muchos campos NULL	CFDI bien formado pero sin campos esperados → se inserta con NULL (no es error)	`bucket_xml_parser.py` (`retrieve_xml_data`)
POST 500 con ~130 s de latencia	Timeout contra PostgreSQL (`XML_ERROR_TYPE:POSTGRES_ERROR`)	`bucket_xml_parser.py` (`save_xml_data_to_postgresql`); Pub/Sub reintenta

Resumen

Campo	Valor
Origen	Objeto nuevo (`OBJECT_FINALIZE`) en bucket GCS → notificación Pub/Sub (push)
Patrón	Event-driven (GCS → Pub/Sub push → Cloud Run → PostgreSQL) + réplica Airbyte a BigQuery
Destino intermedio	PostgreSQL `public.xml_data`
Destino final (BigQuery)	`solvento-data-prod.svc_master_public.xml_data`
Frecuencia	Streaming (por objeto) · réplica Airbyte cada ~20 min
Criticidad	🟡 Media (baja-media)
Dueño	Martin (data eng) · Airbyte: Gero
Repos	`bucket-xml-parser`

Diagrama de linaje

graph LR
    GCS[["🪣 Bucket GCS<br/>OBJECT_FINALIZE (XML)"]]
    TOP(["Pub/Sub<br/>notificación (push)"])
    CR["Cloud Run<br/>bucket-xml-parser-prod<br/>POST /save-xml-data"]
    PG[("PostgreSQL producto<br/>public.xml_data")]
    AB["Airbyte<br/>(cada ~20 min)"]
    BQ[("solvento-data-prod<br/>svc_master_public.xml_data")]

    GCS -->|notifica| TOP -->|push| CR
    CR -->|"parsea CFDI + INSERT"| PG
    PG -->|réplica| AB --> BQ

    classDef src fill:#fde68a,stroke:#b45309,color:#000;
    classDef store fill:#bfdbfe,stroke:#1e40af,color:#000;
    class GCS src;
    class PG,BQ store;

Paso a paso

Bucket GCS: al finalizar la subida de un objeto (OBJECT_FINALIZE) se emite una notificación de Storage a Pub/Sub. El bucket y topic productivos no están en el repo (en dev el ejemplo es gs://solvento_xml_test / topic bucket-xml-parser); confirmar nombres productivos.
Pub/Sub push → POST /save-xml-data de Cloud Run bucket-xml-parser-prod (bucket-xml-parser, bucket_xml_parser.py). Decodifica message.data, saca bucket y object:
- Archivo no-XML (no termina en .xml) → se omite con éxito (HTTP 200, skipped:true, METRIC:XML_SKIPPED:NON_XML). No es error.
- XML válido: retrieve_xml_data detecta versión CFDI (3.3 / 4.x) y extrae atributos del Comprobante, Emisor, Receptor y TimbreFiscalDigital. CFDI bien formado pero sin algún campo → ese campo entra como NULL (se inserta igual; no es error).
- XML no parseable (XML_PARSE_ERROR) → no recuperable, responde 200 (va a DLQ), y guarda los bytes crudos en gs://syntage-ingestion/xml_data/<objeto> (defaults FAILED_XML_BUCKET/FAILED_XML_PREFIX).
INSERT a PostgreSQL public.xml_data (save_xml_data_to_postgresql, psycopg2, RETURNING id). 29 columnas: file_url, datos del comprobante (comprobante_total, comprobante_subtotal, comprobante_moneda, comprobante_fecha, comprobante_tipodecomprobante, …), emisor (comprobante_emisor_rfc/nombre/regimenfiscal), receptor (comprobante_receptor_rfc/nombre/usocfdi/…) y timbre (comprobante_timbre_uuid, comprobante_timbre_fechatimbrado, comprobante_timbre_nocertificadosat, …).
Airbyte replica public.xml_data → solvento-data-prod.svc_master_public.xml_data cada ~20 min (referente Gero).

Comportamiento de errores: los no recuperables (XML_PARSE_ERROR, MISSING_DATA_FIELD, INVALID_STORAGE_DATA, VALIDATION_ERROR, UNEXPECTED_ERROR) devuelven HTTP 200 a propósito (van directo a DLQ, sin reintento). Los recuperables (POSTGRES_ERROR y timeouts) devuelven HTTP 500 para que Pub/Sub reintente. Los POST 500 con ~130 s de latencia son timeouts contra PostgreSQL (METRIC:XML_ERROR_TYPE:POSTGRES_ERROR, recoverable=true) → no son fallo real del proceso, Pub/Sub reintenta.

Tablas que alimenta

Tabla	Contenido	Job que la carga
PostgreSQL `public.xml_data`	CFDI parseado (comprobante / emisor / receptor / timbre)	Cloud Run `bucket-xml-parser-prod` (INSERT)
`solvento-data-prod.svc_master_public.xml_data`	Réplica del anterior en el warehouse	Airbyte (cada ~20 min)

Consumo aguas abajo

Análisis de comprobantes fiscales (CFDI) sobre solvento-data-prod.svc_master_public.xml_data. Criticidad baja-media; cualquier modelado dbt (si existe) vive en dbt-etls.

Si falla 🛠️

Impacto: medio. Si no se parsean/insertan los CFDIs, xml_data deja de actualizarse (y por ende su réplica en BigQuery). Pub/Sub reintenta los recuperables.

Síntoma	Dónde mirar	Acción
POST 500 con ~130 s de latencia	Métrica `xml_parser/xml_errors{error_type="POSTGRES_ERROR"}` / logs `METRIC:XML_ERROR_TYPE:POSTGRES_ERROR`	Timeout a PostgreSQL; Pub/Sub reintenta. Revisar salud/conectividad del Postgres (`POSTGRES_*`) si es persistente
Muchos `xml_skipped`	Métrica `xml_parser/xml_skipped` / `METRIC:XML_SKIPPED:NON_XML`	Normal: archivos no-XML en el bucket. No es error
Filas con campos NULL	`bucket_xml_parser.py` (`retrieve_xml_data`)	CFDI sin esos campos; no es error
`XML_PARSE_ERROR` recurrente	Métrica `xml_parser/xml_errors{error_type="XML_PARSE_ERROR"}`; bucket `gs://syntage-ingestion/xml_data/`	XML corrupto; va a DLQ. Revisar el origen del archivo
BigQuery no actualiza pero Postgres sí	Conexión Airbyte (PostgreSQL → BigQuery)	Revisar el sync de Airbyte (~20 min)

Escalación: SQL / Metabase → Diego · Airbyte → Gero (réplica a BigQuery) · resto → Martin (parser / PostgreSQL).

Enlaces

📊 Estado en vivo: dashboard "Data Processes" (Cloud Monitoring). Métricas custom: custom.googleapis.com/xml_parser/xml_processed, …/xml_errors (labels error_type, recoverable), …/xml_skipped. Ver CLOUD_MONITORING.md en el repo.
📦 Servicio: bucket-xml-parser — main bucket_xml_parser.py

Última actualización: 2026-06-21 · Si cambia el flujo, el bucket, el esquema de xml_data o el sync de Airbyte, actualizar esta ficha en el mismo MR.