Ethereum eth_subscribe: Suscripciones de Logs y Cabeceras vs Filtros de Sondeo

El modelo push: cómo funciona eth_subscribe

Ethereum JSON-RPC expone un mecanismo de publicar-suscribir (pubsub) a través del transporte WebSocket, documentado en las páginas oficiales de execution-apis de Ethereum y Eventos en tiempo real de geth. A diferencia de las llamadas de solicitud-respuesta, eth_subscribe abre un canal push persistente: el cliente envía una solicitud de suscripción, el nodo responde con un ID de suscripción (por ejemplo, 0x9cef478923ff2bf24f5f1e1a1b3f8f7b), y a partir de entonces el nodo envía objetos de notificación cada vez que ocurre el evento suscrito.

Cada notificación es un mensaje JSON-RPC con jsonrpc: "2.0", method: "eth_subscription" y un objeto params que contiene el ID de subscription y el payload result. El cliente nunca envía una solicitud por evento; simplemente lee mensajes del socket. Esto es fundamentalmente diferente del sondeo, donde el cliente pregunta repetidamente al nodo por nuevos datos.

La suscripción está vinculada a la conexión WebSocket. Si la conexión se cae, la suscripción del lado del servidor se destruye (o se vuelve inalcanzable). Esta es la causa raíz de la mayoría de los fallos de suscripción en el mundo real, y dicta la estrategia de reconexión que cubriremos más adelante.

• Transporte: WebSocket (WS o WSS). eth_subscribe no está disponible sobre HTTP simple.
• Iniciación: eth_subscribe(channel, params) devuelve un ID de suscripción.
• Notificación: el servidor envía mensajes eth_subscription con params.subscription y params.result.
• Terminación: eth_unsubscribe(subscriptionId) detiene el flujo; cerrar el socket también lo finaliza.

Los cuatro canales de suscripción estándar

La especificación JSON-RPC de Ethereum define cuatro canales estándar. Cada uno tiene un payload de resultado y un caso de uso distintos.

• newHeads: Se dispara cada vez que se añade una nueva cabecera canónica a la cadena. El resultado es un objeto de cabecera (similar a eth_getBlockByNumber con false para transacciones). Ten en cuenta que durante una reorganización de la cadena, puedes recibir cabeceras que luego quedan huérfanas; tu cliente debe manejar las reorganizaciones comparando los hashes de los bloques y revirtiendo el estado si es necesario.
• logs: Se dispara por cada log que coincida con el filtro que proporciones. La gramática del filtro es idéntica a eth_getLogs: puedes especificar address (una dirección única o un array) y topics (un array donde cada posición puede ser null para cualquiera, o un array de valores de topic alternativos). El resultado es un objeto de log con address, topics, data, blockNumber, transactionHash, logIndex, etc.
• newPendingTransactions: Se dispara cuando se añade una nueva transacción pendiente al pool de transacciones del nodo. El resultado es el hash de la transacción (por defecto) o el objeto de transacción completo, dependiendo de la configuración del nodo (por ejemplo, --rpc.evmtimeout de geth o una bandera como --ws.fulltx). Este canal es útil para monitorear el mempool, pero puede ser ruidoso.
• syncing: Se dispara cuando el estado de sincronización del nodo cambia. El resultado es un objeto de sincronización (similar a eth_sync), o false cuando el nodo está completamente sincronizado. Útil para monitorear la salud del nodo.

Ciclo de vida de la suscripción y el problema de la reconexión

El ciclo de vida de una suscripción es simple: crearla, recibir eventos y eventualmente destruirla. Pero el fallo clásico es una desconexión de WebSocket. Cuando la conexión se cae, la suscripción del lado del servidor desaparece. Si tu cliente se reconecta ingenuamente y continúa leyendo del antiguo ID de suscripción, no recibirás nada. Peor aún, si no te vuelves a suscribir, perderás eventos silenciosamente durante el intervalo.

El patrón robusto es: al reconectar, siempre crear nuevas suscripciones y luego rellenar cualquier evento perdido entre el último bloque procesado y la cabeza actual. Por eso los sistemas de producción a menudo combinan una suscripción logs con consultas periódicas eth_getLogs desde el último bloque visto. La suscripción proporciona notificaciones de baja latencia, mientras que el relleno por sondeo llena el vacío después de una desconexión.

Además, debes manejar la naturaleza asíncrona del socket. Usa un bucle de lectura dedicado (por ejemplo, una goroutine en Go o una tarea asíncrona en Python) para que el análisis y procesamiento de notificaciones nunca bloquee el envío de nuevas solicitudes de suscripción o heartbeats. Muchas bibliotecas de WebSocket almacenan mensajes en búfer, pero si bloqueas el bucle de lectura, puedes perder mensajes o causar contrapresión.

• Trata siempre una caída de WebSocket como 'resuscribir' – nunca reutilices IDs de suscripción.
• Registra el último bloque procesado (del blockNumber del log o del number de la cabecera) para habilitar el relleno.
• Después de reconectar, llama a eth_subscribe nuevamente y luego consulta eth_getLogs desde lastBlock+1 hasta currentHead para llenar el vacío.
• Usa un bucle de lectura asíncrono para evitar bloquear el socket.
• Llama a eth_unsubscribe cuando realmente hayas terminado con una suscripción para evitar fugas de recursos del lado del servidor. Los servidores comúnmente limitan el número de suscripciones concurrentes por conexión (documentado / varía según el proveedor).

Ejemplo reproducible: Suscribirse a newHeads y logs

El siguiente ejemplo en Python usa la biblioteca websockets para conectarse a un nodo público de Ethereum (por ejemplo, Cloudflare-ETH, o tu propio endpoint). Se suscribe a newHeads y a logs para una dirección de contrato y un topic específicos, imprime los IDs de suscripción y las primeras notificaciones, y luego cancela la suscripción.

• Salida esperada: dos IDs de suscripción (cadenas hexadecimales), luego una serie de objetos de notificación. El resultado de newHeads contiene una cabecera de bloque; el resultado de logs contiene una entrada de log.
• Si no ves notificaciones de log, tu filtro puede no coincidir con eventos recientes, o el nodo puede estar detrás de la cabeza de la cadena.

import asyncio
import json
import websockets

WS_URL = "wss://cloudflare-eth.com"  # reemplaza con el endpoint WSS de tu proveedor

async def main():
    async with websockets.connect(WS_URL) as ws:
        # Suscribirse a newHeads
        await ws.send(json.dumps({"jsonrpc": "2.0", "id": 1, "method": "eth_subscribe", "params": ["newHeads"]}))
        resp = json.loads(await ws.recv())
        head_sub = resp["result"]
        print("Suscripción a newHeads:", head_sub)

        # Suscribirse a logs de un contrato (por ejemplo, evento Transfer de USDC)
        # address: contrato USDC, topics: firma del evento Transfer
        filter_params = {
            "address": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
            "topics": ["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"]
        }
        await ws.send(json.dumps({"jsonrpc": "2.0", "id": 2, "method": "eth_subscribe", "params": ["logs", filter_params]}))
        resp = json.loads(await ws.recv())
        log_sub = resp["result"]
        print("Suscripción a logs:", log_sub)

        # Leer algunas notificaciones
        for _ in range(3):
            msg = json.loads(await ws.recv())
            if msg.get("method") == "eth_subscription":
                print("Notificación:", msg["params"]["subscription"], msg["params"]["result"])

        # Cancelar suscripción
        await ws.send(json.dumps({"jsonrpc": "2.0", "id": 3, "method": "eth_unsubscribe", "params": [head_sub]}))
        resp = json.loads(await ws.recv())
        print("Suscripción a cabeceras cancelada:", resp)
        await ws.send(json.dumps({"jsonrpc": "2.0", "id": 4, "method": "eth_unsubscribe", "params": [log_sub]}))
        resp = json.loads(await ws.recv())
        print("Suscripción a logs cancelada:", resp)

asyncio.run(main())

Filtros de sondeo vs suscripciones push: tabla de decisión

Elegir entre suscripciones push y filtros de sondeo depende de tus requisitos de durabilidad, manejo de vacíos, transporte y costo. La tabla a continuación resume las compensaciones.

• Durabilidad: Las suscripciones push son efímeras; si la conexión se cae, pierdes el flujo. Los filtros de sondeo (a través de eth_newFilter + eth_getFilterChanges) mantienen el estado en el nodo, pero ese estado también expira después de un tiempo de espera (típicamente 5 minutos en geth, documentado / varía según el proveedor).
• Manejo de vacíos: Con push, debes rellenar después de una reconexión. Con sondeo, puedes consultar desde un rango de bloques específico, pero debes gestionar el ciclo de vida del filtro.
• Transporte: Push requiere WebSocket; el sondeo funciona también sobre HTTP.
• Costo: Push es eficiente para eventos de alta frecuencia porque no envías solicitudes repetidas. El sondeo puede ser derrochador si sondeas con demasiada frecuencia, pero es más simple y más auditable.
• Auditabilidad: El sondeo con eth_getLogs desde un bloque conocido te da un historial verificable. Las suscripciones push son de último mensaje gana y con pérdidas a través de desconexiones, por lo que no son adecuadas como fuente de eventos auditable.

Solución de problemas: ¿Por qué no recibo eventos?

Cuando tu suscripción parece coincidir pero no recibes nada, revisa esta lista de verificación.

• Gramática incorrecta de topics: Asegúrate de que tu array topics use null para comodines y arrays para condiciones OR. Por ejemplo, ["0x...", null] coincide con cualquier segundo topic, mientras que [["0x...", "0x..."]] coincide con cualquiera de dos topics en la primera posición.
• Retraso de compromiso/cabeza: El nodo puede estar detrás de la cabeza de la cadena. Verifica eth_syncing; si devuelve un objeto de sincronización, espera hasta que sea false.
• Suscribirse a través de HTTPS en lugar de WSS: eth_subscribe solo funciona sobre WebSocket. Si usas un endpoint HTTPS, obtendrás un error o ningún evento.
• Nodo que no expone pubsub: Algunos proveedores o nodos privados deshabilitan pubsub. Consulta la documentación del nodo o prueba un endpoint WSS público.
• Suscripciones silenciosas caídas: Si la conexión WebSocket se cae silenciosamente (por ejemplo, debido a un tiempo de espera de red), tu suscripción desaparece. Implementa un heartbeat o lógica de reconexión.
• Poda de gas/logs de bloque: Los endpoints ligeros o proveedores con historial limitado pueden podar logs. Esto está documentado / varía según el proveedor. Si necesitas logs históricos, usa eth_getLogs con un rango de bloques, pero ten en cuenta los límites del proveedor.

Limitaciones y compensaciones: Cuándo no usar suscripciones

Las suscripciones push son excelentes para paneles en tiempo real, monitoreo de mempool y aplicaciones impulsadas por eventos donde la baja latencia importa. Sin embargo, son inherentemente propensas a pérdidas: si tu cliente está fuera de línea, pierdes eventos y no hay mecanismo de reproducción. El servidor no pone en cola mensajes para clientes desconectados.

Para aplicaciones que requieren un historial completo y auditable de eventos (por ejemplo, indexación, contabilidad o cumplimiento legal), confía en eth_getLogs con rangos de bloques explícitos. Los filtros de sondeo pueden ser un punto intermedio, pero también tienen límites de estado y expiración en el lado del nodo.

Diseña siempre tu sistema para manejar reconexiones con gracia. Un patrón común es ejecutar una suscripción para actualizaciones en tiempo real y un trabajo periódico de relleno que consulte eth_getLogs desde el último bloque procesado hasta la cabeza actual. Esto asegura que no se pierdan eventos, incluso si la suscripción se cae por unos segundos.

También considera el costo: cada suscripción consume recursos del servidor. Si tienes muchos clientes, puedes alcanzar los límites del proveedor en suscripciones concurrentes (documentado / varía según el proveedor). Usa eth_unsubscribe rápidamente cuando termines.

Próximos pasos y lecturas adicionales

Ahora que entiendes el modelo push, puedes aplicarlo a tus propias aplicaciones. Para una inmersión más profunda en temas relacionados, explora estos recursos:

• Manejo de desconexiones de WebSocket en Ethereum – estrategias prácticas de reconexión.

• Consultas de logs únicas con eth_getLogs y filtros de topics – para relleno y consultas históricas.

• Monitoreo de endpoints RPC y salud del nodo – rastrea el estado de sincronización y la latencia.

• Guía de RPC WebSocket de Base – patrones similares en Base.

• Elección de un nodo RPC de Ethereum (Asistente RPC) – elige un proveedor que soporte pubsub.

• Servicio API y Precios RPC – comprende tus opciones de endpoint.

• Descripción general de la red Ethereum y el centro de aprendizaje de OnFinality para más guías.