From e82e488ede1743f43f078edc7777512053ad5615 Mon Sep 17 00:00:00 2001 From: Nickolas Oliveira Date: Fri, 9 Oct 2026 13:01:37 -0300 Subject: [PATCH 1/7] =?UTF-8?q?fix(segments):=20leva=20a=20exclus=C3=A3o?= =?UTF-8?q?=20de=20contato=20ao=20cache=20de=20todos=20os=20processos=20vi?= =?UTF-8?q?a=20Redis=20(CRM-245)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit O sinal de contato excluído saía só pelo EventEmitter2 do processo que recebe o /events/identify. No split api/worker, o segment-worker, que recalcula os segmentos, nunca o recebia e seguia com a lista velha até o TTL de 5 minutos. O DeletedContactsSignalRelay publica o sinal num canal Redis (evo-flow:db:segments:contact-deleted; pub/sub ignora o índice do banco) e reemite como evento local, marcado como relayed, o que chega dos outros processos. Cada processo ignora a própria origem. Redis fora do ar não segura o boot nem a ingestão: os caches voltam ao TTL. Co-Authored-By: Claude Code <[EMAIL_REDACTED]> --- src/modules/segments/segments-cache.module.ts | 2 + .../services/deleted-contacts-signal.relay.ts | 151 ++++++++++++++++++ 2 files changed, 153 insertions(+) create mode 100644 src/modules/segments/services/deleted-contacts-signal.relay.ts diff --git a/src/modules/segments/segments-cache.module.ts b/src/modules/segments/segments-cache.module.ts index c155753..610eeda 100644 --- a/src/modules/segments/segments-cache.module.ts +++ b/src/modules/segments/segments-cache.module.ts @@ -11,6 +11,7 @@ import { SegmentQueryExecutionService } from './services/segment-query-execution import { SegmentAssignmentService } from './services/segment-assignment.service'; import { SegmentEventsService } from './services/segment-events.service'; import { DeletedContactsCacheService } from './services/deleted-contacts-cache.service'; +import { DeletedContactsSignalRelay } from './services/deleted-contacts-signal.relay'; import { SegmentCircuitBreakerService } from './services/segment-circuit-breaker.service'; import { SegmentMetricsService } from './metrics/segment-metrics.service'; import { SegmentQueueService } from './services/segment-queue.service'; @@ -38,6 +39,7 @@ import { ProcessingModule } from '../processing/processing.module'; SegmentAssignmentService, SegmentEventsService, DeletedContactsCacheService, + DeletedContactsSignalRelay, SegmentCircuitBreakerService, SegmentMetricsService, SegmentQueueService, diff --git a/src/modules/segments/services/deleted-contacts-signal.relay.ts b/src/modules/segments/services/deleted-contacts-signal.relay.ts new file mode 100644 index 0000000..c81a61b --- /dev/null +++ b/src/modules/segments/services/deleted-contacts-signal.relay.ts @@ -0,0 +1,151 @@ +import { randomUUID } from 'crypto'; +import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +import { EventEmitter2, OnEvent } from '@nestjs/event-emitter'; +import Redis from 'ioredis'; +import { CustomLoggerService } from 'src/common/services/custom-logger.service'; +import { getProcessingConfig } from '../../processing/config/processing.config'; +import { CONTACT_DELETED_INGESTED_EVENT } from '../queries/contact-event-names'; + +export interface ContactDeletedSignal { + contactId?: string; + relayed?: boolean; +} + +interface RelayMessage { + origin: string; + contactId?: string; +} + +/** + * Carries the deleted-contact signal across processes. The ingest that emits it runs + * in the api process, while segments are recomputed by the segment-worker: without this + * relay the worker's DeletedContactsCacheService keeps a stale set until its TTL. + * Best effort: if Redis is down the caches fall back to that TTL, as before. + */ +@Injectable() +export class DeletedContactsSignalRelay + implements OnModuleInit, OnModuleDestroy +{ + private readonly logger = new CustomLoggerService( + DeletedContactsSignalRelay.name, + ); + private readonly origin = randomUUID(); + private readonly channel: string; + private publisher: Redis | null = null; + private subscriber: Redis | null = null; + + constructor(private readonly eventEmitter: EventEmitter2) { + // Pub/sub channels ignore the Redis db index, so the db goes into the name. + const db = getProcessingConfig().redis?.db ?? 5; + this.channel = `evo-flow:db${db}:segments:contact-deleted`; + } + + // Never awaits Redis: an outage at boot must not hold up any run mode. + onModuleInit(): void { + this.publisher = this.createClient({ enableOfflineQueue: false }); + this.subscriber = this.createClient({}); + + const subscriber = this.subscriber; + subscriber.on('message', (channel: string, raw: string) => + this.onRelayMessage(channel, raw), + ); + // Subscribing on every 'ready' also covers a boot where Redis was still down. + subscriber.on('ready', () => { + subscriber + .subscribe(this.channel) + .then(() => + this.logger.log( + `Listening for deleted-contact signals on ${this.channel}`, + ), + ) + .catch((error: Error) => + this.logger.warn( + `Could not subscribe to ${this.channel}; caches fall back to their TTL: ${error.message}`, + ), + ); + }); + + for (const client of [this.publisher, subscriber]) { + client.connect().catch(() => undefined); + } + } + + async onModuleDestroy(): Promise { + await Promise.allSettled( + [this.subscriber, this.publisher].map((client) => client?.quit()), + ); + } + + @OnEvent(CONTACT_DELETED_INGESTED_EVENT) + async onContactDeletedIngested(signal?: ContactDeletedSignal): Promise { + if (signal?.relayed || !this.publisher) return; + + const message: RelayMessage = { + origin: this.origin, + contactId: signal?.contactId, + }; + try { + await this.publisher.publish(this.channel, JSON.stringify(message)); + } catch (error) { + this.logger.warn( + `Could not publish the deleted-contact signal; other processes fall back to their TTL: ${(error as Error).message}`, + ); + } + } + + private onRelayMessage(channel: string, raw: string): void { + if (channel !== this.channel) return; + + const message = parseRelayMessage(raw); + if (!message) { + this.logger.warn(`Ignoring malformed deleted-contact signal: ${raw}`); + return; + } + if (message.origin === this.origin) return; + + this.logger.debug( + `Deleted-contact signal from another process (contact ${message.contactId ?? 'unknown'})`, + ); + const relayed: ContactDeletedSignal = { + contactId: message.contactId, + relayed: true, + }; + this.eventEmitter.emit(CONTACT_DELETED_INGESTED_EVENT, relayed); + } + + private createClient(options: { enableOfflineQueue?: boolean }): Redis { + const redis = getProcessingConfig().redis; + const client = new Redis({ + host: redis?.host || 'localhost', + port: redis?.port || 6379, + password: redis?.password, + db: redis?.db ?? 5, + ...(redis?.tls ? { tls: redis.tls } : {}), + lazyConnect: true, + maxRetriesPerRequest: 3, + connectTimeout: 10000, + ...options, + }); + client.on('error', (error) => + this.logger.error(`Deleted-contact signal Redis error: ${error.message}`), + ); + return client; + } +} + +function parseRelayMessage(raw: string): RelayMessage | null { + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return null; + } + if (typeof parsed !== 'object' || parsed === null) return null; + + const { origin, contactId } = parsed as Record; + if (typeof origin !== 'string' || !origin) return null; + return { + origin, + contactId: typeof contactId === 'string' ? contactId : undefined, + }; +} From 400105a0a4f81f73679b3c8f666955b99f1e5318 Mon Sep 17 00:00:00 2001 From: Nickolas Oliveira Date: Fri, 9 Oct 2026 13:01:37 -0300 Subject: [PATCH 2/7] fix(segments): busca de deletados compartilhada e sem gravar conjunto velho (CRM-245) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Na janela de 15 s depois de uma exclusão o cache não responde, e cada chamada ia ao ClickHouse: um recálculo de N segmentos disparava N buscas iguais. Chamadas concorrentes agora compartilham a busca em andamento. Cada invalidação incrementa uma geração e descarta a busca em voo, então um resultado que começou antes da exclusão volta para quem pediu mas não é gravado por 5 minutos. O expiresAt passa a ser calculado quando a busca termina. Co-Authored-By: Claude Code <[EMAIL_REDACTED]> --- .../deleted-contacts-cache.service.ts | 32 ++++++++++++++++--- 1 file changed, 28 insertions(+), 4 deletions(-) diff --git a/src/modules/segments/services/deleted-contacts-cache.service.ts b/src/modules/segments/services/deleted-contacts-cache.service.ts index c90cd31..ac53950 100644 --- a/src/modules/segments/services/deleted-contacts-cache.service.ts +++ b/src/modules/segments/services/deleted-contacts-cache.service.ts @@ -21,6 +21,10 @@ export class DeletedContactsCacheService { // this window every call queries ClickHouse and nothing is cached (CRM-215). private readonly BYPASS_AFTER_DELETE_MS = 15000; private bypassCacheUntil = 0; + // Concurrent callers share one ClickHouse fetch. Each invalidation bumps the + // generation, so a fetch that started before it never re-caches the old set. + private inFlight: Promise> | null = null; + private generation = 0; constructor(private readonly clickhouseService: ClickHouseService) {} @@ -33,14 +37,32 @@ export class DeletedContactsCacheService { return this.cached; } + if (this.inFlight) { + return this.inFlight; + } + this.logger.debug('Deleted contacts cache miss, fetching from ClickHouse'); + const pending = this.loadDeletedContacts(this.generation).finally(() => { + if (this.inFlight === pending) this.inFlight = null; + }); + this.inFlight = pending; + return pending; + } + + private async loadDeletedContacts(generation: number): Promise> { try { const deletedContacts = await this.fetchDeletedContactsFromClickHouse(); - this.cached = deletedContacts; - // A set fetched inside the bypass window may be incomplete: let it expire with the window. - this.expiresAt = bypass ? this.bypassCacheUntil : now + this.CACHE_TTL; - this.logger.debug(`Cached ${deletedContacts.size} deleted contacts`); + if (generation === this.generation) { + const now = Date.now(); + this.cached = deletedContacts; + // A set fetched inside the bypass window may be incomplete: let it expire with the window. + this.expiresAt = + now < this.bypassCacheUntil + ? this.bypassCacheUntil + : now + this.CACHE_TTL; + this.logger.debug(`Cached ${deletedContacts.size} deleted contacts`); + } return deletedContacts; } catch (error) { this.logger.error('Failed to fetch deleted contacts:', error); @@ -74,6 +96,8 @@ export class DeletedContactsCacheService { } invalidateCache(): void { + this.generation += 1; + this.inFlight = null; this.cached = null; this.expiresAt = 0; this.logger.debug('Deleted contacts cache invalidated'); From 6554e1c726cbbcc73defcd660550b7acd75776c7 Mon Sep 17 00:00:00 2001 From: Nickolas Oliveira Date: Fri, 9 Oct 2026 13:07:06 -0300 Subject: [PATCH 3/7] test(segments): sinal entre processos e busca compartilhada do cache de deletados (CRM-245) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit O spec do relay simula dois processos sobre um Redis em memória: a exclusão ingerida na api invalida o cache do worker, a origem não recebe o sinal duas vezes, o que chega de fora não é republicado, mensagem malformada ou de outro canal é ignorada e falha de publish não lança. No cache: chamadas concorrentes fazem uma só busca, dentro e fora da janela de bypass, e uma busca iniciada antes do sinal não grava o conjunto velho. Co-Authored-By: Claude Code <[EMAIL_REDACTED]> --- .../deleted-contacts-signal.relay.spec.ts | 181 ++++++++++++++++++ .../segment-canonical-event-names.spec.ts | 70 +++++++ 2 files changed, 251 insertions(+) create mode 100644 src/modules/segments/services/deleted-contacts-signal.relay.spec.ts diff --git a/src/modules/segments/services/deleted-contacts-signal.relay.spec.ts b/src/modules/segments/services/deleted-contacts-signal.relay.spec.ts new file mode 100644 index 0000000..7a296b8 --- /dev/null +++ b/src/modules/segments/services/deleted-contacts-signal.relay.spec.ts @@ -0,0 +1,181 @@ +import { EventEmitter2 } from '@nestjs/event-emitter'; +import { CONTACT_DELETED_INGESTED_EVENT } from '../queries/contact-event-names'; +import { DeletedContactsCacheService } from './deleted-contacts-cache.service'; +import { DeletedContactsSignalRelay } from './deleted-contacts-signal.relay'; + +type Handler = (...args: any[]) => void; + +/** An in-memory Redis server shared by every fake client: pub/sub crosses "processes". */ +const bus = { + subscribers: new Set(), + failPublish: false, +}; + +class FakeRedis { + handlers = new Map(); + channels = new Set(); + + on(event: string, handler: Handler) { + this.handlers.set(event, [...(this.handlers.get(event) ?? []), handler]); + return this; + } + + emit(event: string, ...args: any[]) { + (this.handlers.get(event) ?? []).forEach((h) => h(...args)); + } + + connect = jest.fn(() => { + this.emit('ready'); + return Promise.resolve(); + }); + + subscribe = jest.fn((channel: string) => { + this.channels.add(channel); + bus.subscribers.add(this); + return Promise.resolve(1); + }); + + publish = jest.fn((channel: string, message: string) => { + if (bus.failPublish) return Promise.reject(new Error('connection lost')); + bus.subscribers.forEach((client) => { + if (client.channels.has(channel)) + client.emit('message', channel, message); + }); + return Promise.resolve(bus.subscribers.size); + }); + + quit = jest.fn(() => Promise.resolve('OK')); +} + +jest.mock('ioredis', () => ({ + __esModule: true, + default: jest.fn().mockImplementation(() => new FakeRedis()), +})); + +/** One evo-flow process: its own emitter, cache and relay, wired as Nest would. */ +function bootProcess() { + const emitter = new EventEmitter2(); + const cache = new DeletedContactsCacheService({} as any); + const relay = new DeletedContactsSignalRelay(emitter); + const onCacheSignal = jest.spyOn(cache, 'onContactDeletedIngested'); + + emitter.on(CONTACT_DELETED_INGESTED_EVENT, () => + cache.onContactDeletedIngested(), + ); + emitter.on(CONTACT_DELETED_INGESTED_EVENT, (signal) => { + void relay.onContactDeletedIngested(signal); + }); + relay.onModuleInit(); + + return { emitter, cache, relay, onCacheSignal }; +} + +const flush = () => new Promise((resolve) => setImmediate(resolve)); + +describe('DeletedContactsSignalRelay', () => { + beforeEach(() => { + bus.subscribers.clear(); + bus.failPublish = false; + }); + + it('invalidates the cache of another process (api ingests, segment-worker recomputes)', async () => { + const api = bootProcess(); + const worker = bootProcess(); + (worker.cache as any).cached = new Set(['stale']); + (worker.cache as any).expiresAt = Number.MAX_SAFE_INTEGER; + + api.emitter.emit(CONTACT_DELETED_INGESTED_EVENT, { contactId: 'c-1' }); + await flush(); + + expect(worker.onCacheSignal).toHaveBeenCalledTimes(1); + expect((worker.cache as any).cached).toBeNull(); + }); + + it('does not signal the originating process twice', async () => { + const api = bootProcess(); + bootProcess(); + + api.emitter.emit(CONTACT_DELETED_INGESTED_EVENT, { contactId: 'c-1' }); + await flush(); + + expect(api.onCacheSignal).toHaveBeenCalledTimes(1); + }); + + it('does not publish again what it received from another process', async () => { + const api = bootProcess(); + const worker = bootProcess(); + + api.emitter.emit(CONTACT_DELETED_INGESTED_EVENT, { contactId: 'c-1' }); + await flush(); + + const workerPublisher = (worker.relay as any).publisher as FakeRedis; + expect(workerPublisher.publish).not.toHaveBeenCalled(); + }); + + it('carries the contact id across processes', async () => { + const api = bootProcess(); + const worker = bootProcess(); + const seen: unknown[] = []; + worker.emitter.on(CONTACT_DELETED_INGESTED_EVENT, (signal) => + seen.push(signal), + ); + + api.emitter.emit(CONTACT_DELETED_INGESTED_EVENT, { contactId: 'c-1' }); + await flush(); + + expect(seen).toEqual([{ contactId: 'c-1', relayed: true }]); + }); + + it('ignores malformed messages and messages without an origin', () => { + const worker = bootProcess(); + const subscriber = (worker.relay as any).subscriber as FakeRedis; + const channel = [...subscriber.channels][0]; + + subscriber.emit('message', channel, 'not json'); + subscriber.emit('message', channel, JSON.stringify({ contactId: 'c-1' })); + subscriber.emit('message', channel, 'null'); + + expect(worker.onCacheSignal).not.toHaveBeenCalled(); + }); + + it('ignores other channels', () => { + const worker = bootProcess(); + const subscriber = (worker.relay as any).subscriber as FakeRedis; + + subscriber.emit( + 'message', + 'some-other-channel', + JSON.stringify({ origin: 'elsewhere', contactId: 'c-1' }), + ); + + expect(worker.onCacheSignal).not.toHaveBeenCalled(); + }); + + it('never throws when Redis fails to publish: the ingest goes on and caches fall back to the TTL', async () => { + const api = bootProcess(); + bus.failPublish = true; + + await expect( + api.relay.onContactDeletedIngested({ contactId: 'c-1' }), + ).resolves.toBeUndefined(); + }); + + it('namespaces the channel with the Redis db, since pub/sub ignores the db index', () => { + const worker = bootProcess(); + const subscriber = (worker.relay as any).subscriber as FakeRedis; + + expect([...subscriber.channels]).toEqual([ + expect.stringMatching(/^evo-flow:db\d+:segments:contact-deleted$/), + ]); + }); + + it('closes both connections on shutdown', async () => { + const worker = bootProcess(); + const { publisher, subscriber } = worker.relay as any; + + await worker.relay.onModuleDestroy(); + + expect((publisher as FakeRedis).quit).toHaveBeenCalled(); + expect((subscriber as FakeRedis).quit).toHaveBeenCalled(); + }); +}); diff --git a/src/modules/segments/services/segment-canonical-event-names.spec.ts b/src/modules/segments/services/segment-canonical-event-names.spec.ts index 886471e..207a0fd 100644 --- a/src/modules/segments/services/segment-canonical-event-names.spec.ts +++ b/src/modules/segments/services/segment-canonical-event-names.spec.ts @@ -251,4 +251,74 @@ describe('CRM-215 deleted-contacts cache never trades correctness for speed', () expect((cache as any).cached).toBeNull(); }); + + describe('concurrent fetches', () => { + function deferredFetches(cache: DeletedContactsCacheService) { + const pending: Array<(set: Set) => void> = []; + const fetch = jest.fn( + () => new Promise>((resolve) => pending.push(resolve)), + ); + (cache as any).fetchDeletedContactsFromClickHouse = fetch; + return { fetch, pending }; + } + + it('shares one ClickHouse query among concurrent calls inside the bypass window', async () => { + const cache = new DeletedContactsCacheService({} as any); + const { fetch, pending } = deferredFetches(cache); + cache.onContactDeletedIngested(); + + const calls = Array.from({ length: 5 }, () => cache.getDeletedContacts()); + pending[0](new Set(['gone'])); + const results = await Promise.all(calls); + + expect(fetch).toHaveBeenCalledTimes(1); + results.forEach((set) => expect(set).toEqual(new Set(['gone']))); + }); + + it('shares one query among concurrent cache misses outside the window too', async () => { + const cache = new DeletedContactsCacheService({} as any); + const { fetch, pending } = deferredFetches(cache); + + const calls = [cache.getDeletedContacts(), cache.getDeletedContacts()]; + pending[0](new Set(['gone'])); + await Promise.all(calls); + + expect(fetch).toHaveBeenCalledTimes(1); + expect(await cache.getDeletedContacts()).toEqual(new Set(['gone'])); // cache hit + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it('does not re-cache a set fetched before the deletion signal arrived', async () => { + const cache = new DeletedContactsCacheService({} as any); + const { fetch, pending } = deferredFetches(cache); + + const beforeSignal = cache.getDeletedContacts(); + cache.onContactDeletedIngested(); + const afterSignal = cache.getDeletedContacts(); + + expect(fetch).toHaveBeenCalledTimes(2); // the old fetch is not shared after the signal + pending[0](new Set(['stale'])); + expect(await beforeSignal).toEqual(new Set(['stale'])); + expect((cache as any).cached).toBeNull(); + + pending[1](new Set(['stale', 'fresh'])); + expect(await afterSignal).toEqual(new Set(['stale', 'fresh'])); + expect((cache as any).cached).toEqual(new Set(['stale', 'fresh'])); + }); + + it('starts a new query once the shared one settles', async () => { + const cache = new DeletedContactsCacheService({} as any); + const { fetch, pending } = deferredFetches(cache); + cache.onContactDeletedIngested(); + + const first = cache.getDeletedContacts(); + pending[0](new Set(['a'])); + await first; + const second = cache.getDeletedContacts(); // still inside the window: no cache hit + pending[1](new Set(['a', 'b'])); + + expect(await second).toEqual(new Set(['a', 'b'])); + expect(fetch).toHaveBeenCalledTimes(2); + }); + }); }); From 379fbbd9d15d624f46c469070b0a3e67dc192f5a Mon Sep 17 00:00:00 2001 From: Nickolas Oliveira Date: Fri, 9 Oct 2026 15:47:35 -0300 Subject: [PATCH 4/7] =?UTF-8?q?fix(segments):=20a=20janela=20de=20bypass?= =?UTF-8?q?=20vale=20pelo=20in=C3=ADcio=20da=20busca,=20n=C3=A3o=20pelo=20?= =?UTF-8?q?fim=20(CRM-245)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Achado da review: com o expiresAt calculado ao fim da busca, uma busca que começava dentro da janela de 15 s e respondia depois dela gravava por 5 minutos um conjunto lido dentro da janela, possivelmente sem a exclusão. É a corrida que a janela existe para evitar, e a develop já decidia pelo início. Co-Authored-By: Claude Code <[EMAIL_REDACTED]> --- .../deleted-contacts-cache.service.ts | 21 ++++++++++++------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/src/modules/segments/services/deleted-contacts-cache.service.ts b/src/modules/segments/services/deleted-contacts-cache.service.ts index ac53950..c5df8c0 100644 --- a/src/modules/segments/services/deleted-contacts-cache.service.ts +++ b/src/modules/segments/services/deleted-contacts-cache.service.ts @@ -43,24 +43,29 @@ export class DeletedContactsCacheService { this.logger.debug('Deleted contacts cache miss, fetching from ClickHouse'); - const pending = this.loadDeletedContacts(this.generation).finally(() => { - if (this.inFlight === pending) this.inFlight = null; - }); + const pending = this.loadDeletedContacts(this.generation, now).finally( + () => { + if (this.inFlight === pending) this.inFlight = null; + }, + ); this.inFlight = pending; return pending; } - private async loadDeletedContacts(generation: number): Promise> { + private async loadDeletedContacts( + generation: number, + startedAt: number, + ): Promise> { try { const deletedContacts = await this.fetchDeletedContactsFromClickHouse(); if (generation === this.generation) { - const now = Date.now(); this.cached = deletedContacts; - // A set fetched inside the bypass window may be incomplete: let it expire with the window. + // A set fetched inside the bypass window may be incomplete: let it expire with the + // window. Judged by when the fetch started, not when it returned. this.expiresAt = - now < this.bypassCacheUntil + startedAt < this.bypassCacheUntil ? this.bypassCacheUntil - : now + this.CACHE_TTL; + : startedAt + this.CACHE_TTL; this.logger.debug(`Cached ${deletedContacts.size} deleted contacts`); } return deletedContacts; From 48cd7062b0916c31b0dd460491152cc6e68765e9 Mon Sep 17 00:00:00 2001 From: Nickolas Oliveira Date: Fri, 9 Oct 2026 15:47:35 -0300 Subject: [PATCH 5/7] =?UTF-8?q?fix(segments):=20relay=20fecha=20as=20conex?= =?UTF-8?q?=C3=B5es=20e=20loga=20uma=20vez=20por=20queda=20do=20Redis=20(C?= =?UTF-8?q?RM-245)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Com o Redis fora, o quit() rejeitava e os clientes seguiam reconectando depois do shutdown; disconnect() encerra na hora. O erro de conexão era logado a cada tentativa de reconexão (40 linhas em 10 s): agora sai uma linha por queda e outra quando a conexão volta. Co-Authored-By: Claude Code <[EMAIL_REDACTED]> --- .../services/deleted-contacts-signal.relay.ts | 24 +++++++++++++------ 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/src/modules/segments/services/deleted-contacts-signal.relay.ts b/src/modules/segments/services/deleted-contacts-signal.relay.ts index c81a61b..8675207 100644 --- a/src/modules/segments/services/deleted-contacts-signal.relay.ts +++ b/src/modules/segments/services/deleted-contacts-signal.relay.ts @@ -70,10 +70,12 @@ export class DeletedContactsSignalRelay } } - async onModuleDestroy(): Promise { - await Promise.allSettled( - [this.subscriber, this.publisher].map((client) => client?.quit()), - ); + // disconnect(), not quit(): with Redis down, quit() rejects and the client keeps + // reconnecting, leaving an open handle behind. + onModuleDestroy(): void { + for (const client of [this.subscriber, this.publisher]) { + client?.disconnect(); + } } @OnEvent(CONTACT_DELETED_INGESTED_EVENT) @@ -126,9 +128,17 @@ export class DeletedContactsSignalRelay connectTimeout: 10000, ...options, }); - client.on('error', (error) => - this.logger.error(`Deleted-contact signal Redis error: ${error.message}`), - ); + // One line per outage, not one per reconnect attempt. + let down = false; + client.on('error', (error) => { + if (down) return; + down = true; + this.logger.error(`Deleted-contact signal Redis error: ${error.message}`); + }); + client.on('ready', () => { + if (down) this.logger.log('Deleted-contact signal Redis connection back'); + down = false; + }); return client; } } From 76041c7f1ba71a4b0b49baa4837130e17b17cb2a Mon Sep 17 00:00:00 2001 From: Nickolas Oliveira Date: Fri, 9 Oct 2026 15:47:35 -0300 Subject: [PATCH 6/7] =?UTF-8?q?test(segments):=20janela=20pelo=20in=C3=ADc?= =?UTF-8?q?io=20da=20busca=20e=20o=20@OnEvent=20real=20do=20relay=20(CRM-2?= =?UTF-8?q?45)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cobre os achados da review: a busca que começa dentro da janela e responde depois dela não é servida do cache; um TestingModule com o EventEmitterModule real prova que o @OnEvent publica a exclusão local uma vez e não republica a reemitida; shutdown usa disconnect; erro de conexão sai uma vez por queda. Co-Authored-By: Claude Code <[EMAIL_REDACTED]> --- .../deleted-contacts-signal.relay.spec.ts | 66 +++++++++++++++++-- .../segment-canonical-event-names.spec.ts | 24 +++++++ 2 files changed, 84 insertions(+), 6 deletions(-) diff --git a/src/modules/segments/services/deleted-contacts-signal.relay.spec.ts b/src/modules/segments/services/deleted-contacts-signal.relay.spec.ts index 7a296b8..1f2a575 100644 --- a/src/modules/segments/services/deleted-contacts-signal.relay.spec.ts +++ b/src/modules/segments/services/deleted-contacts-signal.relay.spec.ts @@ -1,4 +1,6 @@ -import { EventEmitter2 } from '@nestjs/event-emitter'; +import { Test } from '@nestjs/testing'; +import { EventEmitter2, EventEmitterModule } from '@nestjs/event-emitter'; +import { ClickHouseService } from '../../processing/clickhouse/clickhouse.service'; import { CONTACT_DELETED_INGESTED_EVENT } from '../queries/contact-event-names'; import { DeletedContactsCacheService } from './deleted-contacts-cache.service'; import { DeletedContactsSignalRelay } from './deleted-contacts-signal.relay'; @@ -44,7 +46,7 @@ class FakeRedis { return Promise.resolve(bus.subscribers.size); }); - quit = jest.fn(() => Promise.resolve('OK')); + disconnect = jest.fn(); } jest.mock('ioredis', () => ({ @@ -169,13 +171,65 @@ describe('DeletedContactsSignalRelay', () => { ]); }); - it('closes both connections on shutdown', async () => { + it('closes both connections on shutdown, without waiting on a Redis that may be down', () => { const worker = bootProcess(); const { publisher, subscriber } = worker.relay as any; - await worker.relay.onModuleDestroy(); + worker.relay.onModuleDestroy(); - expect((publisher as FakeRedis).quit).toHaveBeenCalled(); - expect((subscriber as FakeRedis).quit).toHaveBeenCalled(); + expect((publisher as FakeRedis).disconnect).toHaveBeenCalled(); + expect((subscriber as FakeRedis).disconnect).toHaveBeenCalled(); + }); + + it('logs one error per outage, not one per reconnect attempt', () => { + const worker = bootProcess(); + const subscriber = (worker.relay as any).subscriber as FakeRedis; + const logError = jest.spyOn((worker.relay as any).logger, 'error'); + + subscriber.emit('error', new Error('ECONNREFUSED')); + subscriber.emit('error', new Error('ECONNREFUSED')); + subscriber.emit('error', new Error('ECONNREFUSED')); + expect(logError).toHaveBeenCalledTimes(1); + + subscriber.emit('ready'); + subscriber.emit('error', new Error('ECONNREFUSED')); + expect(logError).toHaveBeenCalledTimes(2); + }); +}); + +describe('DeletedContactsSignalRelay wired by Nest', () => { + beforeEach(() => bus.subscribers.clear()); + + it('the real @OnEvent publishes a local deletion once and never a relayed one', async () => { + const moduleRef = await Test.createTestingModule({ + imports: [EventEmitterModule.forRoot()], + providers: [ + DeletedContactsSignalRelay, + DeletedContactsCacheService, + { provide: ClickHouseService, useValue: {} }, + ], + }).compile(); + await moduleRef.init(); + + const emitter = moduleRef.get(EventEmitter2); + const relay = moduleRef.get(DeletedContactsSignalRelay); + const cache = moduleRef.get(DeletedContactsCacheService); + const publisher = (relay as any).publisher as FakeRedis; + (cache as any).cached = new Set(['stale']); + + emitter.emit(CONTACT_DELETED_INGESTED_EVENT, { contactId: 'c-1' }); + emitter.emit(CONTACT_DELETED_INGESTED_EVENT, { + contactId: 'c-2', + relayed: true, + }); + await flush(); + + expect(publisher.publish).toHaveBeenCalledTimes(1); + expect(JSON.parse(publisher.publish.mock.calls[0][1])).toMatchObject({ + contactId: 'c-1', + }); + expect((cache as any).cached).toBeNull(); + + await moduleRef.close(); }); }); diff --git a/src/modules/segments/services/segment-canonical-event-names.spec.ts b/src/modules/segments/services/segment-canonical-event-names.spec.ts index 207a0fd..c0005e1 100644 --- a/src/modules/segments/services/segment-canonical-event-names.spec.ts +++ b/src/modules/segments/services/segment-canonical-event-names.spec.ts @@ -306,6 +306,30 @@ describe('CRM-215 deleted-contacts cache never trades correctness for speed', () expect((cache as any).cached).toEqual(new Set(['stale', 'fresh'])); }); + it('does not cache for 5 minutes a set whose fetch started inside the window and ended after it', async () => { + const t0 = 1_000_000; + const clock = jest.spyOn(Date, 'now').mockReturnValue(t0); + try { + const cache = new DeletedContactsCacheService({} as any); + const { fetch, pending } = deferredFetches(cache); + cache.onContactDeletedIngested(); // window open until t0 + 15 s + + clock.mockReturnValue(t0 + 10_000); + const insideWindow = cache.getDeletedContacts(); + clock.mockReturnValue(t0 + 16_000); // ClickHouse answers after the window closed + pending[0](new Set(['maybe-incomplete'])); + await insideWindow; + + clock.mockReturnValue(t0 + 17_000); + const next = cache.getDeletedContacts(); + expect(fetch).toHaveBeenCalledTimes(2); // refetched, not served from cache + pending[1](new Set(['maybe-incomplete', 'complete'])); + expect(await next).toEqual(new Set(['maybe-incomplete', 'complete'])); + } finally { + clock.mockRestore(); + } + }); + it('starts a new query once the shared one settles', async () => { const cache = new DeletedContactsCacheService({} as any); const { fetch, pending } = deferredFetches(cache); From 33f9a38de7c831ac7c52623d16092fe0dd089ccc Mon Sep 17 00:00:00 2001 From: Guilherme Gomes Date: Fri, 9 Oct 2026 17:14:48 -0300 Subject: [PATCH 7/7] docs(segments): keep the deleted-contacts comments true to the shared fetch The bypass-window comment still said every call queries ClickHouse; concurrent calls now share one fetch. Also drops a history note from the relay header. --- src/modules/segments/services/deleted-contacts-cache.service.ts | 2 +- src/modules/segments/services/deleted-contacts-signal.relay.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/modules/segments/services/deleted-contacts-cache.service.ts b/src/modules/segments/services/deleted-contacts-cache.service.ts index c5df8c0..588f384 100644 --- a/src/modules/segments/services/deleted-contacts-cache.service.ts +++ b/src/modules/segments/services/deleted-contacts-cache.service.ts @@ -18,7 +18,7 @@ export class DeletedContactsCacheService { private readonly CACHE_TTL = 300000; // 5 minutes // Ingest → Kafka → ClickHouse MV is async: a fetch right after the deleted-contact // signal may not see the row yet and would re-cache a stale set for CACHE_TTL. During - // this window every call queries ClickHouse and nothing is cached (CRM-215). + // this window the cache is never served. private readonly BYPASS_AFTER_DELETE_MS = 15000; private bypassCacheUntil = 0; // Concurrent callers share one ClickHouse fetch. Each invalidation bumps the diff --git a/src/modules/segments/services/deleted-contacts-signal.relay.ts b/src/modules/segments/services/deleted-contacts-signal.relay.ts index 8675207..4086ec2 100644 --- a/src/modules/segments/services/deleted-contacts-signal.relay.ts +++ b/src/modules/segments/services/deleted-contacts-signal.relay.ts @@ -20,7 +20,7 @@ interface RelayMessage { * Carries the deleted-contact signal across processes. The ingest that emits it runs * in the api process, while segments are recomputed by the segment-worker: without this * relay the worker's DeletedContactsCacheService keeps a stale set until its TTL. - * Best effort: if Redis is down the caches fall back to that TTL, as before. + * Best effort: if Redis is down the caches fall back to that TTL. */ @Injectable() export class DeletedContactsSignalRelay