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

@@ -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

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.

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 {

src/dispatcher/evn1-repository.ts

@@ -26,6 +26,10 @@ function toRow(raw: Record<string, unknown>): Evn1Row {
 export async function findExisting(client: ServiceLayerClient, eventId: string, webhookId: string): Promise<Evn1Row | null> {
   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<string, unknown>[]) ?? [];
   return rows.length > 0 ? toRow(rows[0]) : null;
 }

src/dispatcher/process-tick.ts

@@ -27,42 +27,67 @@ async function registerNewEvents(client: ServiceLayerClient): Promise<number> {
   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<string>();
+const PROGRESS_EVERY = 100;
+
+async function registerEvent(
+  client: ServiceLayerClient,
+  event: oevn.OevnRow,
+  filters: Awaited<ReturnType<typeof subscriptions.listFilters>>,
+): Promise<void> {
+  // 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<number> {
   const pending = await evn1.listPending(client);
@@ -71,20 +96,11 @@ 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;
@@ -96,11 +112,16 @@ async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number
     byWebhookId.set(row.webhookId, group);
   }
 
+  logger.info(`[dispatcher] Entrega: ${pending.length} fila(s) pendiente(s) en @EVN1, ${byWebhookId.size} suscripción(es) a despachar.`);
   for (const [webhookId, rows] of byWebhookId) {
-    // 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);
+    const batchSize = config.maxEventsPerBatch;
+    const batchCount = Math.ceil(rows.length / batchSize);
+    logger.info(`[dispatcher] Preparando ${rows.length} evento(s) para ${webhookId} en ${batchCount} lote(s) de hasta ${batchSize}.`);
+    // Cada lote se marca entregado/fallido por separado: uno rechazado no
+    // arrastra a los demás.
+    for (let i = 0; i < rows.length; i += batchSize) {
+      await sendGroup(client, activeSubscriptions.get(webhookId)!, rows.slice(i, i + batchSize));
+    }
   }
 
   await recomputeEventStatuses(client, [...new Set(pending.map((r) => r.eventId))]);
@@ -152,7 +173,11 @@ async function sendGroup(
 }
 
 async function recomputeEventStatuses(client: ServiceLayerClient, eventIds: string[]): Promise<void> {
-  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');

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<void> {
+  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;
   }
 }
 

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

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;
   }