From 5a2e149825b57d3ee79030a2fa861c468b901014 Mon Sep 17 00:00:00 2001 From: "george.calle" Date: Tue, 20 Oct 2026 15:26:39 -0300 Subject: [PATCH] Dispatcher: evitar ticks solapados, enviar en lotes y mejorar logs - index.ts: guard tickRunning; un tick largo ya no se solapa con el siguiente de node-cron (causaba -2035 en @EVN1 y -1116 cruzados por requests concurrentes sobre la misma sesion de Service Layer). - process-tick.ts: un evento con error ya no corta el tick; progreso cada 100 eventos; envios en lotes de MAX_EVENTS_PER_BATCH (default 30) para no superar el tamano de body del receptor (HTTP 413). - evn1-repository.ts: findExisting falla si el GET falla, en vez de tratarlo como "no existe" y crear un duplicado. - service-layer-client.ts: errores incluyen metodo, ruta y body. - logger.ts: no poder escribir el archivo de log no tumba el proceso. - CLAUDE.md: notas operativas y mejoras pendientes. --- .env.example | 4 ++ CLAUDE.md | 42 +++++++++++ src/config.ts | 3 + src/dispatcher/evn1-repository.ts | 4 ++ src/dispatcher/process-tick.ts | 111 ++++++++++++++++++------------ src/index.ts | 13 ++++ src/lib/logger.ts | 13 +++- src/service-layer-client.ts | 9 ++- 8 files changed, 154 insertions(+), 45 deletions(-) create mode 100644 CLAUDE.md diff --git a/.env.example b/.env.example index 909e04a..1b3d318 100644 --- a/.env.example +++ b/.env.example @@ -10,6 +10,10 @@ POLL_INTERVAL_SECONDS=30 # Reintentos por suscripción antes de pasarla a State='X' (Exceptional) RETRY_LIMIT=5 +# Máximo de eventos por POST a una suscripción (un backlog grande se parte +# en varios envíos, para no superar el tamaño de body del receptor) +MAX_EVENTS_PER_BATCH=30 + # El destino y la autenticación de cada envío NO se configuran acá: salen # de @OEVC (WebhookURL/AuthType/AuthCred), administrado con # `scripts/manage-subscriptions.ts` — una suscripción puede apuntar a diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..c179b0c --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,42 @@ +# CLAUDE.md + +Notas operativas para trabajar en este repo. El diseño completo está en +`docs/DISENO.md`. + +## Entorno de este servidor + +- El `.env` apunta a **producción** (`SBOPRDLIBROS`, SAP B1 sobre HANA). + Cualquier `npm run run-once` / `npm start` escribe en `@OEVN`/`@EVN1` y + envía webhooks reales a la suscripción activa en `@OEVC`. +- El dispatcher corre como servicio Windows NSSM `sap-legacy-webhook` + (cuenta LocalSystem). **Detener el servicio antes de correrlo a mano** + (`Stop-Service sap-legacy-webhook`) y volver a iniciarlo después: dos + procesos en paralelo comparten eventos y duplican entregas. +- Los logs diarios (`logs/dispatcher-YYYY-MM-DD.log`) los crea SYSTEM; un + usuario normal no puede escribirlos. El logger cae a solo-consola en ese + caso (no es un error). + +## Lecciones aprendidas (2026-10-20) + +- **Ticks solapados**: con backlog grande un tick dura más que + `POLL_INTERVAL_SECONDS`, y node-cron lanzaba otro en paralelo sobre la + misma sesión de Service Layer → `-2035 already exists` en `@EVN1` y + `-1116` (a veces "Unknown error", a veces violación de `KEVN1_NAME` con + la clave de *otro* request). Corregido con el guard `tickRunning` en + `src/index.ts`. Si reaparecen `-1116` cruzados, sospechar concurrencia + sobre la misma sesión antes que datos. +- **Tamaño de lote**: el receptor rechaza bodies grandes (HTTP 413). Los + envíos se parten en lotes de `MAX_EVENTS_PER_BATCH` (30 por defecto). + +## Mejoras pendientes + +- **Lock de instancia única**: archivo de bloqueo (ej. `logs/dispatcher.lock` + con el PID) para que una segunda instancia —servicio o ejecución manual— + se cierre al arrancar. Hoy solo lo evita la regla operativa de arriba. +- **Paralelizar llamadas al Service Layer** dentro de un lote (GET de + eventos / PATCH de `@EVN1` de a 5–10) para vaciar backlogs grandes más + rápido. Cuidado: misma sesión de SL — validar que tolera concurrencia o + usar sesiones separadas (ver lección de ticks solapados). +- **Causa del `-1116` aislado**: tras el fix de solapamiento no se volvió a + ver; si aparece con un solo tick en curso, el log ahora indica método, + ruta y body exactos. diff --git a/src/config.ts b/src/config.ts index cc42d7a..e93e84f 100644 --- a/src/config.ts +++ b/src/config.ts @@ -19,6 +19,9 @@ export const config = { // cualquier URL, no solo a una instancia fija de sap-webhook-api. pollIntervalSeconds: Number(process.env.POLL_INTERVAL_SECONDS ?? 30), retryLimit: Number(process.env.RETRY_LIMIT ?? 5), + // Tope de eventos por POST a una suscripción — sin tope, un backlog grande + // se manda entero en un solo body y el receptor lo rechaza (HTTP 413). + maxEventsPerBatch: Number(process.env.MAX_EVENTS_PER_BATCH ?? 30), }; function requireEnv(name: string): string { diff --git a/src/dispatcher/evn1-repository.ts b/src/dispatcher/evn1-repository.ts index e736578..4f06557 100644 --- a/src/dispatcher/evn1-repository.ts +++ b/src/dispatcher/evn1-repository.ts @@ -26,6 +26,10 @@ function toRow(raw: Record): Evn1Row { export async function findExisting(client: ServiceLayerClient, eventId: string, webhookId: string): Promise { const filter = encodeURIComponent(`U_EventID eq '${eventId}' and U_WebhookID eq '${webhookId}'`); const res = await client.get(`/${ENTITY}?$filter=${filter}`); + // Un GET fallido no es "no existe": tratarlo así lleva a un POST duplicado. + if (res.status !== 200) { + throw new Error(`Service Layer respondió HTTP ${res.status} buscando @EVN1 de ${eventId}: ${JSON.stringify(res.data)}`); + } const rows = (res.data?.value as Record[]) ?? []; return rows.length > 0 ? toRow(rows[0]) : null; } diff --git a/src/dispatcher/process-tick.ts b/src/dispatcher/process-tick.ts index 14152b6..bb89f0f 100644 --- a/src/dispatcher/process-tick.ts +++ b/src/dispatcher/process-tick.ts @@ -27,42 +27,67 @@ async function registerNewEvents(client: ServiceLayerClient): Promise { if (newEvents.length === 0) return 0; const filters = await subscriptions.listFilters(client); + logger.info(`[dispatcher] Registrando ${newEvents.length} evento(s) nuevo(s) (${filters.length} filtro(s) de suscripción).`); - for (const event of newEvents) { - // Eco de una escritura nuestra sobre @OEVC/@EVC1/@OEVN/@EVN1 (ver - // own-table-guard.ts) — nunca se toca, ni se resuelve ni se borra, - // para no perpetuar la cascada. - if (isEchoOfOwnTable(event.objectType)) continue; - - let bizObject: string; + let failed = 0; + for (const [index, event] of newEvents.entries()) { + // Con miles de eventos por tick, sin un rastro intermedio el proceso + // parece colgado hasta el log final. + if (index > 0 && index % PROGRESS_EVERY === 0) { + logger.info(`[dispatcher] Registro: ${index}/${newEvents.length} evento(s) procesado(s), ${failed} con error.`); + } + // Un evento que falla no corta el tick: se registra con su contexto y + // queda en Status='N' para reintentarse en el próximo tick, sin + // bloquear a todos los que vienen detrás. try { - bizObject = bizObjectFor(event.objectType); - } catch { - logger.warn(`[dispatcher] ObjectType desconocido '${event.objectType}' (evento ${event.code}) — se descarta.`); - await oevn.deleteEvent(client, event.code); - continue; + await registerEvent(client, event, filters); + } catch (err) { + failed++; + logger.error( + `[dispatcher] Error registrando evento ${event.code} (ObjectType='${event.objectType}', TransType='${event.transType}', ` + + `${event.fieldNames}=${event.fieldValue}):`, + err, + ); } - - const matches = filters.filter((f) => f.bizObject === bizObject && f.transType === event.transType); - if (matches.length === 0) { - logger.info(`[dispatcher] Sin suscripciones para ${bizObject}/${event.transType} (evento ${event.code}) — se descarta.`); - await oevn.deleteEvent(client, event.code); - continue; - } - - for (const match of matches) { - const existing = await evn1.findExisting(client, event.code, match.webhookId); - if (!existing) await evn1.create(client, event.code, match.webhookId); - } - await oevn.setEventStatus(client, event.code, 'P'); } + if (failed > 0) logger.warn(`[dispatcher] Registro: ${failed}/${newEvents.length} evento(s) con error, quedan para el próximo tick.`); 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(); +const PROGRESS_EVERY = 100; + +async function registerEvent( + client: ServiceLayerClient, + event: oevn.OevnRow, + filters: Awaited>, +): Promise { + // Eco de una escritura nuestra sobre @OEVC/@EVC1/@OEVN/@EVN1 (ver + // own-table-guard.ts) — nunca se toca, ni se resuelve ni se borra, + // para no perpetuar la cascada. + if (isEchoOfOwnTable(event.objectType)) return; + + let bizObject: string; + try { + bizObject = bizObjectFor(event.objectType); + } catch { + logger.warn(`[dispatcher] ObjectType desconocido '${event.objectType}' (evento ${event.code}) — se descarta.`); + await oevn.deleteEvent(client, event.code); + return; + } + + const matches = filters.filter((f) => f.bizObject === bizObject && f.transType === event.transType); + if (matches.length === 0) { + logger.info(`[dispatcher] Sin suscripciones para ${bizObject}/${event.transType} (evento ${event.code}) — se descarta.`); + await oevn.deleteEvent(client, event.code); + return; + } + + for (const match of matches) { + const existing = await evn1.findExisting(client, event.code, match.webhookId); + if (!existing) await evn1.create(client, event.code, match.webhookId); + } + await oevn.setEventStatus(client, event.code, 'P'); +} async function sendPendingDeliveries(client: ServiceLayerClient): Promise { const pending = await evn1.listPending(client); @@ -71,20 +96,11 @@ async function sendPendingDeliveries(client: ServiceLayerClient): Promise(); - 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; @@ -96,11 +112,16 @@ async function sendPendingDeliveries(client: ServiceLayerClient): Promise r.eventId))]); @@ -152,7 +173,11 @@ async function sendGroup( } async function recomputeEventStatuses(client: ServiceLayerClient, eventIds: string[]): Promise { - for (const eventId of eventIds) { + logger.info(`[dispatcher] Recalculando Status de ${eventIds.length} evento(s) en @OEVN.`); + for (const [index, eventId] of eventIds.entries()) { + if (index > 0 && index % PROGRESS_EVERY === 0) { + logger.info(`[dispatcher] Recalculo: ${index}/${eventIds.length} evento(s).`); + } const rows = await evn1.listForEvent(client, eventId); if (rows.length === 0) continue; const allDelivered = rows.every((r) => r.status === 'D'); diff --git a/src/index.ts b/src/index.ts index f72f3e4..8818c62 100644 --- a/src/index.ts +++ b/src/index.ts @@ -6,11 +6,24 @@ import { logger } from './lib/logger.ts'; const client = new ServiceLayerClient(config.serviceLayer); +// Un tick con backlog grande dura más que POLL_INTERVAL_SECONDS: sin este +// guard, cron lanza otro en paralelo sobre los mismos eventos y la misma +// sesión de Service Layer (duplicados -2035 y errores -1116 cruzados entre +// requests concurrentes). +let tickRunning = false; + async function runOnce(): Promise { + if (tickRunning) { + logger.info('[dispatcher] El tick anterior sigue en curso — se omite este.'); + return; + } + tickRunning = true; try { await processTick(client); } catch (err) { logger.error('[dispatcher] Error procesando el tick:', err); + } finally { + tickRunning = false; } } diff --git a/src/lib/logger.ts b/src/lib/logger.ts index f2573a9..412f6e8 100644 --- a/src/lib/logger.ts +++ b/src/lib/logger.ts @@ -34,9 +34,20 @@ function logFilePath(date: Date): string { return `${LOGS_DIR}dispatcher-${day}.log`; } +// No poder escribir el archivo de log (ej. creado por el servicio como +// SYSTEM y corrido a mano con otro usuario: EPERM) nunca debe tumbar el +// dispatcher a mitad de un tick — se avisa una vez y se sigue solo por consola. +let fileWriteFailed = false; + function write(level: string, message: string): void { + if (fileWriteFailed) return; const now = new Date(); - appendFileSync(logFilePath(now), `${localTimestamp(now)} [${level}] ${message}\n`); + try { + appendFileSync(logFilePath(now), `${localTimestamp(now)} [${level}] ${message}\n`); + } catch (err) { + fileWriteFailed = true; + console.warn(`[logger] No se pudo escribir ${logFilePath(now)} (${err instanceof Error ? err.message : String(err)}) — se sigue solo por consola.`); + } } // Errores de red de axios (ECONNREFUSED, timeout, DNS...): primero se diff --git a/src/service-layer-client.ts b/src/service-layer-client.ts index b9d15a2..44b9427 100644 --- a/src/service-layer-client.ts +++ b/src/service-layer-client.ts @@ -83,7 +83,14 @@ export class ServiceLayerClient { // una fila que "parece" creada/actualizada pero no lo está. private assertOk(res: AxiosResponse): AxiosResponse { if (res.status < 200 || res.status >= 300) { - throw new Error(`Service Layer respondió HTTP ${res.status}: ${JSON.stringify(res.data)}`); + // Método + ruta + body enviado: sin esto el error no dice sobre qué + // fila/entidad falló, y con miles de eventos por tick no hay forma de + // ubicarlo. + const method = res.config.method?.toUpperCase() ?? '?'; + throw new Error( + `Service Layer respondió HTTP ${res.status} en ${method} ${res.config.url}` + + ` (body: ${res.config.data ?? '-'}): ${JSON.stringify(res.data)}`, + ); } return res; }