Reconexión y reproducción

Estado inicial#

Sigues un trabajo, una aplicación o una organización. Las notificaciones se entregan al menos una vez, sin orden garantizado y con posibles pérdidas: el mismo evento puede llegar dos veces, tarde o nunca. La sequence guardada de cada canal es la verdad, y el trabajo continúa haya o no una pestaña abierta.

Operaciones#

Un cursor es la secuencia decimal del último evento que tienes del canal. 0 reproduce desde el primer evento retenido.

Procedimiento#

1. Suscríbete primero. Únete al canal con tu permiso y guarda en un búfer lo que llegue. Todavía no lo apliques.
2. Carga la instantánea. Responde el estado autorizado del canal en data, la sequence que ese estado refleja y el cursor correspondiente.
3. Aplica solo lo que sigue. Descarta los eventos del búfer cuya secuencia no sea mayor que la de la instantánea, y luego aplica el resto en orden de secuencia. Suscribirse antes de la instantánea es lo que hace que esta carrera no pierda nada.
4. Descarta los duplicados por id y por sequence.
5. Cierra los huecos. Cuando llega un evento cuya secuencia no es la siguiente, retenlo y reproduce desde tu cursor. Aplica la respuesta en orden, y luego los eventos que retuviste. Un puntero, el evento de difusión beyond-cdn-pointer/1, es la misma situación anunciada de forma explícita: nombra una sequence que no viajó, porque el evento era oversize o su notificación estaba stale. No es un evento. Si su sequence es mayor que tu cursor, reproduce desde tu cursor; de lo contrario, ignóralo.
6. Vuelve a empezar cuando te lo indiquen. Ante 410 CURSOR_EXPIRED, ante un evento resync.required, o cuando retengas más eventos fuera de orden de los que estás dispuesto a guardar, carga una instantánea nueva y continúa desde el paso 3.

El data de la instantánea contiene documentos de administración del ámbito del canal: job para un canal de trabajo; application, jobs y environments para un canal de aplicación; organization, jobs y credit para un canal de organización.

Una página de reproducción responde events, el nuevo cursor y more, que es true cuando siguen más eventos. Una reproducción devuelve solo eventos: los punteros nunca se guardan y nunca aparecen en ella.

Sin una suscripción#

El transporte hace que las novedades lleguen antes. Nunca es necesario para que el resultado sea correcto. Este cliente sigue un trabajo solo con los eventos guardados:

// Follow a job from persisted events only: load a snapshot, then replay after
// its cursor. A broadcast subscription makes this faster; it never replaces it.
const api = process.env.CDN_API_ORIGIN;
const token = process.env.CDN_TOKEN;
const channel = `job:${process.env.CDN_JOB}`;

const headers = { Authorization: `Bearer ${token}` };
const seen = new Set();

async function read(path) {
	const response = await fetch(new URL(`/v1/streams/${channel}/${path}`, api), { headers });
	const document = await response.json();
	if (response.ok) return document;
	if (document.error.code === 'CURSOR_EXPIRED') return undefined;
	throw new Error(`${response.status} ${document.error.code} — ${document.error.message}`);
}

let snapshot = await read('snapshot');
let cursor = snapshot.cursor;
console.log(`snapshot at sequence ${snapshot.sequence}: job is ${snapshot.data.job.state}`);

const final = ['job.succeeded', 'job.failed', 'job.cancelled', 'job.limit_exceeded'];
let ended = ['succeeded', 'failed', 'cancelled', 'limit_exceeded'].includes(snapshot.data.job.state);

while (!ended) {
	const page = await read(`events?after=${cursor}`);

	if (!page) {
		// The cursor is older than the retained events: start again from a snapshot
		snapshot = await read('snapshot');
		cursor = snapshot.cursor;
		continue;
	}

	for (const event of page.events) {
		if (seen.has(event.id)) continue; // the same event can arrive twice
		seen.add(event.id);
		console.log(`#${event.sequence} ${event.type}`);
		if (final.includes(event.type)) ended = true;
	}

	cursor = page.cursor;
	if (!page.more && !ended) await new Promise(resolve => setTimeout(resolve, 2000));
}

El mismo bucle es el que ejecutas después de una reconexión: conserva tu cursor, reproduce a partir de él y continúa.

Resultado esperado#

Después de una reconexión de cualquier duración tienes el mismo estado que un cliente que nunca se desconectó, sin haber aplicado ningún evento dos veces. Los eventos finales, como job.succeeded, job.failed y release.ready, están en la reproducción aunque nunca te haya llegado una notificación de ellos.

Límites#

• La reproducción exige el mismo acceso que la instantánea. Después de una revocación ambas responden 403 ACCESS_REVOKED o 404 NOT_FOUND.
• Los registros guardados de un trabajo tienen su propio endpoint paginado, jobs.logs, con una retención más larga que la de los eventos.