Streaming Reliability Contract

Qué garantiza el stream, qué no garantiza deliberadamente, y exactamente qué debe implementar tu cliente para consumirlo correctamente. Esta página es el contrato; las referencias de SSE y WebSocket documentan los formatos de mensajes individuales.

El principio de diseño: frescura sobre completitud. Las cuotas son perecederas — un delta entregado tarde es peor que un snapshot fresco, porque un precio desactualizado luce idéntico a uno en vivo. Cada política a continuación (reanudación limitada, omisión de mensajes bajo contrapresión, desconexión de clientes que no pueden mantener el ritmo) se deriva de ese principio. Construye tu cliente para tratar desconexión → reconexión → re-baseline como operación normal, no como un camino de error.

1. Snapshot, luego deltasPermalink for this section

Cada conexión sigue el mismo ciclo de vida en ambos transportes:

1. connected — confirmación con tu tier, filtros y el descriptor resume (abajo).
2. Snapshot inicial — el estado actual completo para tus canales y filtros suscritos (SSE: chunks snapshot; WS: opportunities_snapshot + mensajes initial por libro).
3. snapshot:complete — la línea base está lista; siguen los deltas en vivo.
4. Deltas — actualizaciones incrementales que fusionas en el estado local.

Los deltas se indexan por un identificador estable por canal. Fusiónalo por esa clave — nunca por una clave compuesta que construyas tú mismo (consulta la advertencia sobre id):

2. La reanudación es de mejor esfuerzo — dilo claramentePermalink for this section

Reconectar con un cursor (Last-Event-ID en SSE, from_seq en WS) permite al servidor intentar reproducir lo que te perdiste en lugar de reenviar un snapshot completo. Esto funciona bien para caídas breves. La ventana es por transporte: aproximadamente los últimos 5 minutos en WebSocket, aproximadamente los últimos 2 minutos en SSE. Ambas también están limitadas por el recuento de entradas (WebSocket adicionalmente por bytes), por lo que el volumen sostenido las acorta aún más.

No es un registro duradero sin huecos:

• La ventana de reproducción vive en la memoria de la instancia del servidor que atendió tu conexión. Una reconexión que aterriza en una instancia diferente, o que cruza un deploy o reinicio del servidor, no puede reanudar y establece una nueva base con un snapshot completo.
• El fallback es explícito, nunca silencioso. Siempre sabes el resultado a través del ack connected y el snapshot:complete que sigue — el servidor nunca finge que una reconexión con pérdida fue sin huecos.

Diseña para el snapshot completo como camino de recuperación rutinario y trata una reanudación exitosa como una optimización. Si tu caso de uso no puede tolerar el re-baselining, concilia contra REST /api/v1/odds después de cualquier reconexión.

3. El descriptor resumePermalink for this section

Cada ack connected — SSE y WS, conexión nueva o reconexión — lleva un objeto resume autodescriptivo para que tu cliente pueda descubrir el contrato de antemano:

"resume": {
  "durable": false,
  "enabled": true,
  "scope": "process_local",
  "resumable_channels": ["odds"]
}

La cobertura difiere por transporte:

La reanudación por WebSocket cubre estrictamente más canales — prefiere WS para consumidores sensibles a la corrección de oportunidades, gamestate o datos de closing line.

4. Los dos resultados de recuperación que DEBES manejarPermalink for this section

Una reconexión con cursor se resuelve en uno de dos caminos. El ack connected te dice cuál (resumed: true/false), y el snapshot:complete que sigue etiqueta el modo de recuperación:

Manejar solo un camino es el error de implementación clásico: un cliente que solo maneja full_resync mantiene silenciosamente filas desactualizadas después de una reanudación con hueco; un cliente que solo maneja resume mezcla un snapshot fresco en un mapa desactualizado. Maneja ambos.

En un fallback, connected lleva resumed: false más un fallback_reason que explica el porqué (informativo — la acción de recuperación es la misma en cualquier caso: aceptar el snapshot completo etiquetado). Los valores incluyen:

Trata esto como un conjunto abierto — pueden aparecer nuevos valores; los valores desconocidos significan lo mismo (sigue un snapshot completo).

5. Política de consumidor lentoPermalink for this section

El servidor nunca deja que un cliente lento retenga el feed — ni le sirva precios desactualizados. La entrega se degrada en tres etapas explícitas:

1. Buffering. Cada conexión tiene un buffer de envío limitado que absorbe ráfagas normales.
2. Omisión. Bajo contrapresión sostenida, los mensajes delta en vivo se omiten en lugar de encolarse hacia la obsolescencia. Cuando la presión cede, el servidor envía un mensaje de control resync_required antes del siguiente mensaje de datos, para que sepas que tu estado está ahora incompleto: recarga via REST /odds o reconéctate. La fase de snapshot inicial nunca se omite — solo los deltas en vivo.
3. Desconexión. Un cliente que permanece demasiado lento se desconecta — cierre WS 1008 (motivo "sustained backpressure — reconnect" o "sustained skip rate — reconnect"), teardown del stream SSE. Reconéctate y toma un snapshot fresco; entregarte un backlog de cuotas desactualizadas sería peor.

{
  "type": "resync_required",
  "reason": "backpressure",
  "message": "Deltas were dropped due to slow consumption. Request /api/v1/odds for a full snapshot or reconnect."
}

Si alcanzas la etapa 2 o 3 regularmente: reduce tu suscripción (canales, filtros sport, league, sportsbook, market), mueve el parseo de JSON fuera del bucle de recepción, o consume desde una posición de red más rápida.

Detectar un stream bloqueado. Los heartbeats llegan cada 30 segundos en ambos transportes y llevan el seq / global_seq actual. Heartbeats fluyendo mientras global_seq permanece plano durante partidos en vivo significa que tu suscripción está congelada — reconéctate proactivamente. Los heartbeats SSE llevan adicionalmente book_updated_ms (relojes de última emisión por libro) para que puedas detectar un libro que se queda en silencio mientras otros siguen haciendo streaming. Sin heartbeat por más de 60 segundos significa que la conexión está muerta — reconéctate.

Etiqueta de reconexión. Usa backoff exponencial con jitter (ej. 1s → 2s → 4s … con tope en 30s), restablece en una connected exitosa. EventSource reintenta automáticamente (el servidor sugiere retry: 3000); los clientes WebSocket implementan su propio bucle.

6. Una conexión por clave — el desplazamiento es explícitoPermalink for this section

Cada clave API tiene un slot de stream por defecto, compartido entre SSE y WS, con semántica newer-wins: una segunda conexión desplaza la primera, y la nueva conexión siempre tiene éxito. La conexión desplazada es notificada explícitamente:

• WS: código de cierre 4001 con motivo "displaced by newer session".
• SSE: un evento final displaced (code: "too_many_streams", reconnect: false), luego el stream se cierra.

No reconectes automáticamente después de un desplazamiento — el slot lo tiene tu sesión más nueva, y reconectar la desplazaría directamente de vuelta (una tormenta de reconexiones autoinfligida). Una conexión bien filtrada cubre efectivamente todo — consulta Una Conexión, Muchos Temas. Las flotas que genuinamente necesitan streams paralelos solicitan un límite superior por clave (hello@sharpapi.io) o usan una clave por proceso.

7. Los permisos se vuelven a verificar mientras haces streamingPermalink for this section

La autorización no es solo en el momento de la conexión: el servidor verifica periódicamente los permisos de cada conexión de streaming, y de inmediato tras un cambio de suscripción. Si tu clave pierde acceso en medio del stream (downgrade, clave revocada, eliminación de add-on), el stream se cierra explícitamente en lugar de continuar con el grant desactualizado:

• WS: código de cierre 4003 (el motivo indica qué cambió, ej. tier o acceso a streaming).
• SSE: un evento error con código tier_restricted o invalid_api_key, luego el stream se cierra.

Reconéctate para obtener tus permisos actuales. Las actualizaciones funcionan igual — una conexión mantiene su grant del momento de conexión, así que reconéctate después de actualizar para obtener el nuevo acceso.

Lista de verificación del clientePermalink for this section

• Aplicar el snapshot inicial, luego fusionar deltas por la clave estable para cada canal (§1); eliminar en odds:removed / *:expired.
• Esperar a snapshot:complete antes de tratar el estado como completo.
• Persistir tu cursor (Last-Event-ID / último global_seq) y presentarlo al reconectar.
• Al reconectar, ramificar según el resultado: full_resync → descartar estado; resume + gap_detected → recargar via REST /odds; resume limpio → continuar (§4).
• Manejar resync_required → recargar o reconectar (§5).
• Reconectar con backoff + jitter en cualquier cierre excepto displaced / 4001 displaced (§6).
• En 4003 / SSE error (tier_restricted, invalid_api_key) → reconectar para reautorizar (§7).
• Vigilar los heartbeats: ninguno por 60s, o global_seq plano durante partidos en vivo → reconectar (§5).

Implementaciones de referenciaPermalink for this section

SSE — reconexión con Last-Event-ID (navegador)Permalink for this section

EventSource reenvía Last-Event-ID automáticamente; tu trabajo es solo ramificar según el resultado de recuperación:

const oddsMap = new Map();
const es = new EventSource(
  'https://api.sharpapi.io/api/v1/stream?channel=odds&league=nba&api_key=YOUR_KEY'
);
 
es.addEventListener('connected', (e) => {
  const ack = JSON.parse(e.data);
  // ack.resume describes the contract: { durable, enabled, scope, resumable_channels }
  if (ack.resumed === false) {
    // Resume fell back — a full snapshot follows. Discard stale state NOW,
    // before the snapshot chunks arrive. Do NOT clear when ack.resumed is
    // true: no snapshot follows a successful resume, only replayed deltas.
    oddsMap.clear();
    console.log('re-baselining:', ack.fallback_reason);
  }
});
 
es.addEventListener('snapshot', (e) => {
  for (const odd of JSON.parse(e.data).odds) oddsMap.set(odd.id, odd);
});
 
es.addEventListener('snapshot:complete', (e) => {
  const { mode } = JSON.parse(e.data);
  // mode "full_resync": the snapshot above is the new baseline.
  // mode "resume": replayed deltas already merged — state is continuous.
  // No mode key: fresh connect. Either way, state is complete from here.
  console.log('ready', mode ?? 'fresh');
});
 
es.addEventListener('odds:update', (e) => {
  for (const delta of JSON.parse(e.data).odds) {
    const row = oddsMap.get(delta.id);
    if (row) Object.assign(row, delta);
  }
});
 
es.addEventListener('odds:removed', (e) => {
  for (const id of JSON.parse(e.data).ids) oddsMap.delete(id);
});
 
es.addEventListener('resync_required', () => {
  // Deltas were skipped while we were slow — state is incomplete.
  es.close();            // simplest recovery: reconnect for a fresh snapshot
  reconnectWithBackoff();
});
 
es.addEventListener('displaced', () => {
  es.close();            // newer session on this key took the slot —
  // do NOT reconnect automatically (§6)
});

WebSocket — reanudación con from_seqPermalink for this section

let lastGlobalSeq = null;
let delay = 1000;
 
function connect() {
  const params = new URLSearchParams({
    api_key: 'YOUR_KEY',
    channels: 'odds,ev',
  });
  if (lastGlobalSeq) params.set('from_seq', lastGlobalSeq); // resume attempt
  const ws = new WebSocket(`wss://ws.sharpapi.io?${params}`);
 
  ws.onmessage = (event) => {
    const msg = JSON.parse(event.data);
    if (msg.global_seq) lastGlobalSeq = msg.global_seq; // string — keep as-is
 
    switch (msg.type) {
      case 'connected':
        delay = 1000; // reset backoff
        if (msg.resumed === false) {
          oddsMap.clear(); // full snapshot follows — discard stale state
          console.log('resume fell back:', msg.fallback_reason);
        }
        break;
      case 'snapshot:complete':
        if (msg.mode === 'resume' && msg.gap_detected) {
          refetchFromRest(); // GET /api/v1/odds for your filter scope
        }
        break;
      case 'initial':
        for (const odd of msg.data) oddsMap.set(odd.id, odd);
        break;
      case 'odds:update':
        for (const odd of msg.data) oddsMap.set(odd.id, odd);
        break;
      case 'odds:removed':
        for (const id of msg.ids) oddsMap.delete(id);
        break;
      case 'resync_required':
        refetchFromRest(); // deltas were skipped — reconcile
        break;
    }
  };
 
  ws.onclose = (event) => {
    if (event.code === 4001 && event.reason.includes('displaced')) {
      return; // newer session owns the slot — do not reconnect (§6)
    }
    // 4003 (entitlements changed), 1008 (too slow), 1012 (restart), network
    // drops: reconnect with backoff — from_seq makes it a resume attempt.
    setTimeout(connect, delay);
    delay = Math.min(delay * 2, 30000);
  };
}
 
connect();

Ver tambiénPermalink for this section

• Descripción General del Streaming — protocolos, canales, quick start
• Referencia de la API SSE — payloads de eventos y parámetros
• Referencia de la API WebSocket — esquemas de mensajes y códigos de cierre
• Una Conexión, Muchos Temas — patrones de filtros que hacen que una sola conexión sea suficiente