Reiniciar U_Retry al reactivar, subscription:delete y fix de grupo huérfano
- Al reactivar una suscripción (resume o detección X -> A en el dispatcher) se reinicia U_Retry de sus entregas pendientes, para que no vuelva a 'X' en el siguiente tick. - Nuevo comando subscription:delete: borra filtros, entregas fallidas y eventos de @OEVN que quedan huérfanos. - Evita TypeError en sendPendingDeliveries cuando la suscripción pasa a 'X' después de armar su grupo de envío. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
e1c8ee9015
commit
ec312d9461
3 files changed
+80
-3
No files matched your search
@@ -70,3 +70,24 @@ export async function markFailedAttempt(client: ServiceLayerClient, row: Evn1Row
|
||||
U_LastSentTM: hhmmss(now),
|
||||
});
|
||||
}
|
||||
|
||||
// Pone en 0 el contador de reintentos de las entregas pendientes de una
|
||||
// suscripción. Se usa al reactivarla (State X -> A): sin esto, las filas
|
||||
// siguen con Retry >= RETRY_LIMIT y la suscripción vuelve a X en el
|
||||
// siguiente tick sin intentar enviar nada.
|
||||
export async function resetRetries(client: ServiceLayerClient, webhookId: string): Promise<number> {
|
||||
const filter = encodeURIComponent(`U_WebhookID eq '${webhookId}' and U_Status eq 'F'`);
|
||||
const rows = (await client.getAll(`/${ENTITY}?$filter=${filter}`)).map(toRow).filter((r) => r.retry > 0);
|
||||
for (const row of rows) await client.patch(`/${ENTITY}('${row.code}')`, { U_Retry: 0 });
|
||||
return rows.length;
|
||||
}
|
||||
|
||||
// Entregas fallidas/pendientes (Status='F') de una suscripción.
|
||||
export async function listPendingForWebhook(client: ServiceLayerClient, webhookId: string): Promise<Evn1Row[]> {
|
||||
const filter = encodeURIComponent(`U_WebhookID eq '${webhookId}' and U_Status eq 'F'`);
|
||||
return (await client.getAll(`/${ENTITY}?$filter=${filter}`)).map(toRow);
|
||||
}
|
||||
|
||||
export async function deleteRow(client: ServiceLayerClient, code: string): Promise<void> {
|
||||
await client.delete(`/${ENTITY}('${code}')`);
|
||||
}
|
||||
@@ -59,6 +59,11 @@ async function registerNewEvents(client: ServiceLayerClient): Promise<number> {
|
||||
return newEvents.length;
|
||||
}
|
||||
|
||||
// Suscripciones que este proceso pasó a 'X'. Si después aparecen otra vez
|
||||
// activas (alguien las reactivó, por script o directo en Service Layer),
|
||||
// se reinicia U_Retry de sus entregas pendientes para darles otra tanda.
|
||||
const markedExceptional = new Set<string>();
|
||||
|
||||
async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number> {
|
||||
const pending = await evn1.listPending(client);
|
||||
if (pending.length === 0) return 0;
|
||||
@@ -66,11 +71,20 @@ async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number
|
||||
const activeSubscriptions = await subscriptions.listActiveSubscriptions(client);
|
||||
const byWebhookId = new Map<string, evn1.Evn1Row[]>();
|
||||
|
||||
for (const webhookId of [...markedExceptional]) {
|
||||
if (!activeSubscriptions.has(webhookId)) continue;
|
||||
markedExceptional.delete(webhookId);
|
||||
const reset = await evn1.resetRetries(client, webhookId);
|
||||
logger.info(`[dispatcher] Suscripción ${webhookId} reactivada — U_Retry reiniciado a 0 en ${reset} entrega(s).`);
|
||||
for (const row of pending) if (row.webhookId === webhookId) row.retry = 0;
|
||||
}
|
||||
|
||||
for (const row of pending) {
|
||||
if (row.retry >= config.retryLimit) {
|
||||
if (activeSubscriptions.has(row.webhookId)) {
|
||||
logger.warn(`[dispatcher] Suscripción ${row.webhookId} superó ${config.retryLimit} reintentos — pasa a State='X'.`);
|
||||
await subscriptions.markExceptional(client, row.webhookId);
|
||||
markedExceptional.add(row.webhookId);
|
||||
activeSubscriptions.delete(row.webhookId);
|
||||
}
|
||||
continue;
|
||||
@@ -83,7 +97,10 @@ async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number
|
||||
}
|
||||
|
||||
for (const [webhookId, rows] of byWebhookId) {
|
||||
await sendGroup(client, activeSubscriptions.get(webhookId)!, rows);
|
||||
// Pudo pasar a 'X' después de armar el grupo (otra fila superó el límite).
|
||||
const subscription = activeSubscriptions.get(webhookId);
|
||||
if (!subscription) continue;
|
||||
await sendGroup(client, subscription, rows);
|
||||
}
|
||||
|
||||
await recomputeEventStatuses(client, [...new Set(pending.map((r) => r.eventId))]);
|
||||
|
||||
Reference in new issue
Block a user