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-cache.service.ts b/src/modules/segments/services/deleted-contacts-cache.service.ts index c90cd31..588f384 100644 --- a/src/modules/segments/services/deleted-contacts-cache.service.ts +++ b/src/modules/segments/services/deleted-contacts-cache.service.ts @@ -18,9 +18,13 @@ 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 + // 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,37 @@ 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, now).finally( + () => { + if (this.inFlight === pending) this.inFlight = null; + }, + ); + this.inFlight = pending; + return pending; + } + + private async loadDeletedContacts( + generation: number, + startedAt: 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) { + this.cached = deletedContacts; + // 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 = + startedAt < this.bypassCacheUntil + ? this.bypassCacheUntil + : startedAt + 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 +101,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'); 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..1f2a575 --- /dev/null +++ b/src/modules/segments/services/deleted-contacts-signal.relay.spec.ts @@ -0,0 +1,235 @@ +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'; + +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); + }); + + disconnect = jest.fn(); +} + +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, without waiting on a Redis that may be down', () => { + const worker = bootProcess(); + const { publisher, subscriber } = worker.relay as any; + + worker.relay.onModuleDestroy(); + + 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/deleted-contacts-signal.relay.ts b/src/modules/segments/services/deleted-contacts-signal.relay.ts new file mode 100644 index 0000000..4086ec2 --- /dev/null +++ b/src/modules/segments/services/deleted-contacts-signal.relay.ts @@ -0,0 +1,161 @@ +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. + */ +@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); + } + } + + // 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) + 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, + }); + // 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; + } +} + +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, + }; +} 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..c0005e1 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,98 @@ 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('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); + 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); + }); + }); });