From 5d8eccd701012fa27b1e89101d30db997ec17680 Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Tue, 11 Aug 2026 22:04:50 -0400 Subject: [PATCH 1/6] feat(personhog): route ingestion person writes by store mode PERSONS_STORE_MODE selects the world person writes land in: pg keeps today's path untouched, personhog routes allowlisted teams to the identity and leader RPCs, and shadow runs the personhog verb after the authoritative Postgres call, counting failures without failing the batch. Shadowed writes re-resolve the personhog world's own person by distinct id, because Postgres row ids mean nothing there. Merge execution, deletes, and lifecycle marks stay on Postgres in every mode until the merge saga owns them, as does any verb under a Postgres transaction. A bad mode or a missing endpoint fails boot loudly. --- nodejs/src/common/config.ts | 9 + nodejs/src/common/personhog/client.ts | 26 +- .../src/common/personhog/identity-clients.ts | 22 + nodejs/src/common/personhog/index.ts | 3 + nodejs/src/common/persons/metrics.ts | 12 + .../persons/routing-persons-store.test.ts | 241 +++++++++ .../common/persons/routing-persons-store.ts | 503 ++++++++++++++++++ nodejs/src/servers/ingestion-api-server.ts | 58 +- 8 files changed, 868 insertions(+), 6 deletions(-) create mode 100644 nodejs/src/common/personhog/identity-clients.ts create mode 100644 nodejs/src/ingestion/common/persons/routing-persons-store.test.ts create mode 100644 nodejs/src/ingestion/common/persons/routing-persons-store.ts diff --git a/nodejs/src/common/config.ts b/nodejs/src/common/config.ts index 018e45a1932b..dcd52ccf874d 100644 --- a/nodejs/src/common/config.ts +++ b/nodejs/src/common/config.ts @@ -112,6 +112,12 @@ export type CommonConfig = BaseServerConfig & { // PersonHog gRPC PERSONHOG_ENABLED: boolean + /** Which world the ingestion persons store writes: 'pg' (default), 'personhog', or 'shadow' (pg authoritative, personhog best-effort). */ + PERSONS_STORE_MODE: string + /** Comma-separated team ids the non-pg mode applies to; empty applies it to every team. */ + PERSONS_STORE_MODE_TEAMS: string + /** Host and port of the personhog identity server. */ + PERSONHOG_IDENTITY_ADDR: string PERSONHOG_ADDR: string PERSONHOG_GROUPS_ROLLOUT_PERCENTAGE: number PERSONHOG_GROUPS_ROLLOUT_TEAM_IDS: string @@ -293,6 +299,9 @@ export function getDefaultCommonConfig(): CommonConfig { // PersonHog gRPC PERSONHOG_ENABLED: false, PERSONHOG_ADDR: '', + PERSONS_STORE_MODE: 'pg', + PERSONS_STORE_MODE_TEAMS: '', + PERSONHOG_IDENTITY_ADDR: '', PERSONHOG_GROUPS_ROLLOUT_PERCENTAGE: 0, PERSONHOG_GROUPS_ROLLOUT_TEAM_IDS: '', PERSONHOG_PERSONS_ROLLOUT_PERCENTAGE: 0, diff --git a/nodejs/src/common/personhog/client.ts b/nodejs/src/common/personhog/client.ts index a255a8a090c3..0eaf72dfc68a 100644 --- a/nodejs/src/common/personhog/client.ts +++ b/nodejs/src/common/personhog/client.ts @@ -194,6 +194,26 @@ export class PersonHogClient { } static fromConfig(config: PersonHogClientConfig): PersonHogClient { + const { transport, stateMonitor } = createPersonhogTransport(config) + return new PersonHogClient(transport, stateMonitor) + } + + close(): void { + this.stateMonitor?.close() + } +} + +/** + * The transport every personhog gRPC client shares: caller headers, + * consistency stamping, and an HTTP/2 session kept alive and monitored. + * The identity server's clients build on it too, so the wire behavior + * cannot drift between endpoints. + */ +export function createPersonhogTransport(config: PersonHogClientConfig): { + transport: Transport + stateMonitor: SessionStateMonitor +} { + { const scheme = config.useTls ? 'https' : 'http' const interceptors: Interceptor[] = [] if (config.clientName) { @@ -238,10 +258,6 @@ export class PersonHogClient { sessionManager: stateMonitor, interceptors, }) - return new PersonHogClient(transport, stateMonitor) - } - - close(): void { - this.stateMonitor?.close() + return { transport, stateMonitor } } } diff --git a/nodejs/src/common/personhog/identity-clients.ts b/nodejs/src/common/personhog/identity-clients.ts new file mode 100644 index 000000000000..afeaa26578a7 --- /dev/null +++ b/nodejs/src/common/personhog/identity-clients.ts @@ -0,0 +1,22 @@ +import { createClient } from '@connectrpc/connect' + +import { PersonHogIdentity } from '~/common/generated/personhog/personhog/identity/v1/identity_pb' + +import { PersonHogClientConfig, createPersonhogTransport } from './client' +import { PersonhogIdentityOperations } from './identity' + +/** + * Client for the identity server; a separate factory from + * PersonHogClient because it answers on a different address than the + * router. + */ +export function createIdentityClients(config: PersonHogClientConfig): { + identity: PersonhogIdentityOperations + close: () => void +} { + const { transport, stateMonitor } = createPersonhogTransport(config) + return { + identity: new PersonhogIdentityOperations(createClient(PersonHogIdentity, transport)), + close: () => stateMonitor.close(), + } +} diff --git a/nodejs/src/common/personhog/index.ts b/nodejs/src/common/personhog/index.ts index 3c7e69849f92..da91f5fee3c5 100644 --- a/nodejs/src/common/personhog/index.ts +++ b/nodejs/src/common/personhog/index.ts @@ -17,6 +17,9 @@ export type PersonHogConfig = Pick< CommonConfig, | 'PERSONHOG_ENABLED' | 'PERSONHOG_ADDR' + | 'PERSONHOG_IDENTITY_ADDR' + | 'PERSONS_STORE_MODE' + | 'PERSONS_STORE_MODE_TEAMS' | 'PERSONHOG_GROUPS_ROLLOUT_PERCENTAGE' | 'PERSONHOG_GROUPS_ROLLOUT_TEAM_IDS' | 'PERSONHOG_PERSONS_ROLLOUT_PERCENTAGE' diff --git a/nodejs/src/common/persons/metrics.ts b/nodejs/src/common/persons/metrics.ts index f6d4b11aa622..bb115e911864 100644 --- a/nodejs/src/common/persons/metrics.ts +++ b/nodejs/src/common/persons/metrics.ts @@ -160,6 +160,18 @@ export function observeLatencyByVersion(person: InternalPerson | undefined, star personOperationLatencyByVersionSummary.labels(operation, versionBucket).observe(performance.now() - start) } +export const personhogStoreShadowSkipsCounter = new Counter({ + name: 'personhog_store_shadow_skips_total', + help: 'Shadowed writes skipped because the person does not exist in the personhog world yet', + labelNames: ['verb'], +}) + +export const personhogStoreShadowErrorsCounter = new Counter({ + name: 'personhog_store_shadow_errors_total', + help: 'Personhog shadow-side store verb failures by verb; shadow errors never fail the batch', + labelNames: ['verb'], +}) + export const personProfileUpdateOutcomeCounter = new Counter({ name: 'person_profile_update_outcome_total', help: 'Outcome of person profile update operations at event level', diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts new file mode 100644 index 000000000000..7465b4b8c7f7 --- /dev/null +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts @@ -0,0 +1,241 @@ +import { personhogStoreShadowErrorsCounter, personhogStoreShadowSkipsCounter } from '~/common/persons/metrics' +import { InternalPerson } from '~/types' + +import { RoutingPersonsStore, assertPersonsStoreModeConfig, parsePersonsStoreMode } from './routing-persons-store' + +jest.mock('~/common/persons/metrics', () => ({ + personhogStoreShadowErrorsCounter: { labels: jest.fn().mockReturnValue({ inc: jest.fn() }) }, + personhogStoreShadowSkipsCounter: { labels: jest.fn().mockReturnValue({ inc: jest.fn() }) }, +})) + +describe('RoutingPersonsStore', () => { + const person = (teamId: number): InternalPerson => + ({ id: '1', team_id: teamId, properties: {}, is_identified: false }) as unknown as InternalPerson + + const ops = { set: {}, setOnce: {}, unset: [], denied: false, shouldForceUpdate: false, eventName: '$set' } + + const makeStores = () => { + const pg = { + fetchForChecking: jest.fn().mockResolvedValue(null), + fetchForUpdate: jest.fn().mockResolvedValue(person(1)), + applyEventOps: jest.fn().mockResolvedValue([person(1), []]), + createPerson: jest.fn().mockResolvedValue({ success: true }), + deletePerson: jest.fn().mockResolvedValue([]), + moveDistinctIds: jest.fn().mockResolvedValue({ success: true }), + prefetchPersons: jest.fn().mockResolvedValue(undefined), + flush: jest.fn().mockResolvedValue([]), + releaseBatch: jest.fn(), + shutdown: jest.fn().mockResolvedValue(undefined), + } as any + const personhog = { + fetchForChecking: jest.fn().mockResolvedValue(null), + fetchForUpdate: jest.fn().mockResolvedValue(person(1)), + applyEventOps: jest.fn().mockResolvedValue([person(1), []]), + createPerson: jest.fn().mockResolvedValue({ success: true }), + deletePerson: jest.fn().mockResolvedValue([]), + prefetchPersons: jest.fn().mockResolvedValue(undefined), + flush: jest.fn().mockResolvedValue([]), + releaseBatch: jest.fn(), + removeDistinctIdFromCache: jest.fn(), + shutdown: jest.fn().mockResolvedValue(undefined), + } as any + return { pg, personhog } + } + + it('rejects an unknown mode at parse time', () => { + expect(() => parsePersonsStoreMode('both')).toThrow('PERSONS_STORE_MODE') + expect(parsePersonsStoreMode('shadow')).toBe('shadow') + }) + + it.each([ + ['shadow', '', 'id:1', 'PERSONHOG_ADDR'], + ['personhog', 'router:1', '', 'PERSONHOG_IDENTITY_ADDR'], + ['shadow', '', '', 'PERSONHOG_ADDR and PERSONHOG_IDENTITY_ADDR'], + ] as const)('%s mode without endpoints fails at boot naming the knob', (mode, routerAddr, identityAddr, named) => { + expect(() => assertPersonsStoreModeConfig(mode, { routerAddr, identityAddr })).toThrow(named) + }) + + it('pg mode needs no endpoints', () => { + expect(() => assertPersonsStoreModeConfig('pg', { routerAddr: '', identityAddr: '' })).not.toThrow() + }) + + describe('personhog mode with a team allowlist', () => { + it('routes allowlisted teams to personhog and the rest to pg', async () => { + const { pg, personhog } = makeStores() + const store = new RoutingPersonsStore(pg, personhog, 'personhog', new Set([1])) + + await store.fetchForUpdate(1, 'a', 0) + expect(personhog.fetchForUpdate).toHaveBeenCalledWith(1, 'a', 0) + expect(pg.fetchForUpdate).not.toHaveBeenCalled() + + await store.fetchForUpdate(2, 'b', 0) + expect(pg.fetchForUpdate).toHaveBeenCalledWith(2, 'b', 0) + }) + + it('a personhog flush failure propagates, because the store is authoritative', async () => { + const { pg, personhog } = makeStores() + personhog.flush.mockRejectedValue(new Error('leader down')) + const store = new RoutingPersonsStore(pg, personhog, 'personhog', null) + store.forBatch(7) + await expect(store.flush()).rejects.toThrow('leader down') + }) + }) + + describe('shadow mode', () => { + it('returns the pg result and runs the personhog verb after it', async () => { + const { pg, personhog } = makeStores() + const pgPerson = person(1) + pg.applyEventOps.mockResolvedValue([pgPerson, []]) + const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + + const [updated] = await store.applyEventOps(person(1), ops as any, 'a', 0) + expect(updated).toBe(pgPerson) + expect(personhog.applyEventOps).toHaveBeenCalled() + }) + + it('a personhog failure is counted and never fails the batch', async () => { + const { pg, personhog } = makeStores() + personhog.applyEventOps.mockRejectedValue(new Error('identity down')) + const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + + await expect(store.applyEventOps(person(1), ops as any, 'a', 0)).resolves.toBeDefined() + expect(personhogStoreShadowErrorsCounter.labels).toHaveBeenCalledWith({ verb: 'applyEventOps' }) + }) + + it('a pg failure still fails the batch', async () => { + const { pg, personhog } = makeStores() + pg.applyEventOps.mockRejectedValue(new Error('pg down')) + const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + await expect(store.applyEventOps(person(1), ops as any, 'a', 0)).rejects.toThrow('pg down') + expect(personhog.applyEventOps).not.toHaveBeenCalled() + }) + + it('a personhog flush failure is swallowed', async () => { + const { pg, personhog } = makeStores() + personhog.flush.mockRejectedValue(new Error('leader down')) + const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + store.forBatch(7) + await expect(store.flush()).resolves.toEqual([]) + expect(personhog.flush).toHaveBeenCalledWith(7) + }) + }) + + describe('verbs that never route', () => { + it('merge execution stays on pg for an allowlisted team', async () => { + const { pg, personhog } = makeStores() + const store = new RoutingPersonsStore(pg, personhog, 'personhog', new Set([1])) + const tx = {} as any + await store.moveDistinctIds(person(1), person(1), 'a', undefined, tx, 0) + expect(pg.moveDistinctIds).toHaveBeenCalled() + }) + + it('a transactional create stays on pg for an allowlisted team', async () => { + const { pg, personhog } = makeStores() + const store = new RoutingPersonsStore(pg, personhog, 'personhog', new Set([1])) + const tx = {} as any + await store.createPerson( + null as any, + {}, + {}, + {}, + 1, + null, + false, + 'uuid', + { distinctId: 'a' }, + undefined, + tx, + 0 + ) + expect(pg.createPerson).toHaveBeenCalled() + expect(personhog.createPerson).not.toHaveBeenCalled() + }) + }) + + it('prefetch splits entries by route', async () => { + const { pg, personhog } = makeStores() + const store = new RoutingPersonsStore(pg, personhog, 'personhog', new Set([1])) + await store.prefetchPersons([ + { teamId: 1, distinctId: 'a', batchId: 0 }, + { teamId: 2, distinctId: 'b', batchId: 0 }, + ]) + expect(pg.prefetchPersons).toHaveBeenCalledWith([{ teamId: 2, distinctId: 'b', batchId: 0 }]) + expect(personhog.prefetchPersons).toHaveBeenCalledWith([{ teamId: 1, distinctId: 'a', batchId: 0 }]) + }) + + it('a released batch is no longer flushed on the personhog side', async () => { + const { pg, personhog } = makeStores() + const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + store.forBatch(7) + store.releaseBatch(7) + await store.flush() + expect(personhog.flush).not.toHaveBeenCalled() + expect(personhog.releaseBatch).toHaveBeenCalledWith(7) + }) + + describe('shadow writes resolve the personhog world person', () => { + const pgPerson = { id: '7', team_id: 1, properties: {} } as unknown as InternalPerson + const shadowPerson = { id: '99', team_id: 1, properties: {} } as unknown as InternalPerson + + it.each([ + [ + 'applyEventOps', + (store: RoutingPersonsStore) => store.applyEventOps(pgPerson, ops as any, 'd1', 0), + (personhog: any) => personhog.applyEventOps, + ], + [ + 'updatePersonWithPropertiesDiffForUpdate', + (store: RoutingPersonsStore) => + store.updatePersonWithPropertiesDiffForUpdate(pgPerson, { a: '1' }, [], {}, 'd1', 0), + (personhog: any) => personhog.updatePersonWithPropertiesDiffForUpdate, + ], + ] as const)('%s ships the shadow world id, not the pg id', async (_verb, call, personhogFn) => { + const { pg, personhog } = makeStores() + personhog.fetchForUpdate.mockResolvedValue(shadowPerson) + personhog.updatePersonWithPropertiesDiffForUpdate = jest.fn().mockResolvedValue([shadowPerson, [], true]) + pg.updatePersonWithPropertiesDiffForUpdate = jest.fn().mockResolvedValue([pgPerson, [], true]) + const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + + await call(store) + + // The pg call keeps the caller's person; the shadowed call must + // re-resolve, because pg ids mean nothing in the personhog world. + const shadowArgs = personhogFn(personhog).mock.calls[0] + expect(shadowArgs[0].id).toBe('99') + }) + + it('skips the shadow write, counted, when the person does not exist in the personhog world', async () => { + const { pg, personhog } = makeStores() + pg.applyEventOps.mockResolvedValue([pgPerson, []]) + personhog.fetchForUpdate.mockResolvedValue(null) + const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + + const [result] = await store.applyEventOps(pgPerson, ops as any, 'd1', 0) + + expect(result.id).toBe('7') + expect(personhog.applyEventOps).not.toHaveBeenCalled() + expect(personhogStoreShadowSkipsCounter.labels).toHaveBeenCalledWith({ verb: 'applyEventOps' }) + }) + + it('a failing shadow resolution is swallowed like any shadow error', async () => { + const { pg, personhog } = makeStores() + pg.applyEventOps.mockResolvedValue([pgPerson, []]) + personhog.fetchForUpdate.mockRejectedValue(new Error('identity down')) + const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + + const [result] = await store.applyEventOps(pgPerson, ops as any, 'd1', 0) + + expect(result.id).toBe('7') + expect(personhogStoreShadowErrorsCounter.labels).toHaveBeenCalledWith({ verb: 'applyEventOps' }) + }) + }) + + it('shutdown closes the personhog side even when pg shutdown fails', async () => { + const { pg, personhog } = makeStores() + pg.shutdown.mockRejectedValue(new Error('pg teardown failed')) + const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + + await expect(store.shutdown()).rejects.toThrow('pg teardown failed') + expect(personhog.shutdown).toHaveBeenCalled() + }) +}) diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.ts new file mode 100644 index 000000000000..24f9e33352c8 --- /dev/null +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.ts @@ -0,0 +1,503 @@ +import { DateTime } from 'luxon' + +import { personhogStoreShadowErrorsCounter, personhogStoreShadowSkipsCounter } from '~/common/persons/metrics' +import { PersonMessage } from '~/common/persons/person-message' +import { InternalPersonWithDistinctId, LifecycleMarkPerson } from '~/common/persons/repositories/person-repository' +import { PersonRepositoryTransaction } from '~/common/persons/repositories/person-repository-transaction' +import { CreatePersonResult, MoveDistinctIdsResult } from '~/common/utils/db/db' +import { logger } from '~/common/utils/logger' +import { BatchWritingStoreFlushStats } from '~/ingestion/common/stores/batch-writing-store' +import { Properties } from '~/plugin-scaffold' +import { InternalPerson, PropertiesLastOperation, PropertiesLastUpdatedAt, Team } from '~/types' + +import { EventOps } from './person-update' +import { PersonhogPersonsStore } from './personhog-persons-store' +import { FlushResult, PersonsStore } from './persons-store' +import { BatchBoundPersonsStore, PersonsStoreForBatch } from './persons-store-for-batch' +import { PersonsStoreTransaction } from './persons-store-transaction' + +export type PersonsStoreMode = 'pg' | 'personhog' | 'shadow' + +export function parsePersonsStoreMode(raw: string): PersonsStoreMode { + if (raw === 'pg' || raw === 'personhog' || raw === 'shadow') { + return raw + } + throw new Error(`PERSONS_STORE_MODE must be pg, personhog, or shadow; got ${JSON.stringify(raw)}`) +} + +/** + * Fails startup when a non-pg mode is missing the endpoints it dials, + * so a misconfiguration is one loud boot error instead of every write + * failing at its first RPC. + */ +export function assertPersonsStoreModeConfig( + mode: PersonsStoreMode, + addrs: { routerAddr: string; identityAddr: string } +): void { + if (mode === 'pg') { + return + } + const missing = [ + ...(addrs.routerAddr ? [] : ['PERSONHOG_ADDR']), + ...(addrs.identityAddr ? [] : ['PERSONHOG_IDENTITY_ADDR']), + ] + if (missing.length > 0) { + throw new Error(`PERSONS_STORE_MODE=${mode} requires ${missing.join(' and ')} to be set`) + } +} + +/** + * Routes person-store verbs between the Postgres world and the personhog + * world by team. The mode applies to the teams in `teams` (every team + * when null); everything else stays on Postgres. In shadow the Postgres + * result is authoritative and the personhog side runs the same verb + * afterwards, its failures counted and logged but never failing the + * batch. + * + * Two verb families do not route: + * + * - Merge execution — including deleting its losing persons — runs on + * Postgres in every mode (the merge saga will own it on personhog), as + * does any verb invoked under a Postgres transaction (see `route`). + * - `personPropertiesSize` routes by mode but is not shadowed: the + * personhog store answers it with a constant because the leader + * enforces the size ceiling at admission. + */ +export class RoutingPersonsStore implements PersonsStore { + /** + * Batches with personhog lanes that may hold unflushed folds. The + * personhog store flushes per batch, so the store-level flush needs + * the live set. + */ + private activeBatches = new Set() + + constructor( + private pg: PersonsStore, + private personhog: PersonhogPersonsStore, + private mode: 'personhog' | 'shadow', + private teams: ReadonlySet | null + ) {} + + private modeFor(teamId: number): PersonsStoreMode { + return this.teams === null || this.teams.has(teamId) ? this.mode : 'pg' + } + + /** + * Run the personhog side of a shadowed verb: sequential, awaited, and + * never allowed to fail the batch. + */ + private async shadowed(verb: string, run: () => Promise): Promise { + try { + await run() + } catch (error) { + personhogStoreShadowErrorsCounter.labels({ verb }).inc() + logger.warn('personhog shadow verb failed', { verb, error: String(error) }) + } + } + + /** + * The whole mode semantics, once: pg teams run the pg call, personhog + * teams the personhog call, shadow teams run pg as the authoritative + * result and the personhog call after it, swallowed. A verb invoked + * under a Postgres transaction is the merge flow's and stays on pg + * regardless of team. The two lambdas are each verb's signature + * adapter — the stores disagree on tx and batchId parameters. + */ + private async route( + verb: string, + teamId: number, + pg: () => Promise, + personhog: () => Promise, + opts?: { tx?: unknown; shadow?: () => Promise } + ): Promise { + const mode = opts?.tx ? 'pg' : this.modeFor(teamId) + if (mode === 'personhog') { + return personhog() + } + const result = await pg() + if (mode === 'shadow') { + await this.shadowed(verb, opts?.shadow ?? personhog) + } + return result + } + + /** + * Resolve the personhog world's own person for a shadowed write. The + * caller holds the Postgres row, whose numeric id means nothing in + * the personhog world — the two id sequences are independent — so a + * shadow write must re-resolve by distinct id and skip, counted, + * when the person does not exist there yet. The fetch memoizes per + * batch, so repeated writes to one person cost one resolution. + */ + private async withShadowPerson( + verb: string, + teamId: number, + distinctId: string, + batchId: number, + run: (person: InternalPerson) => Promise + ): Promise { + const shadowPerson = await this.personhog.fetchForUpdate(teamId, distinctId, batchId) + if (shadowPerson === null) { + personhogStoreShadowSkipsCounter.labels({ verb }).inc() + return + } + await run(shadowPerson) + } + + forBatch(batchId: number): PersonsStoreForBatch { + this.activeBatches.add(batchId) + return new BatchBoundPersonsStore(this, batchId) + } + + inTransaction(description: string, transaction: (tx: PersonsStoreTransaction) => Promise): Promise { + return this.pg.inTransaction(description, transaction) + } + + fetchForChecking(teamId: number, distinctId: string, batchId: number): Promise { + return this.route( + 'fetchForChecking', + teamId, + () => this.pg.fetchForChecking(teamId, distinctId, batchId), + () => this.personhog.fetchForChecking(teamId, distinctId, batchId) + ) + } + + fetchForUpdate(teamId: number, distinctId: string, batchId: number): Promise { + return this.route( + 'fetchForUpdate', + teamId, + () => this.pg.fetchForUpdate(teamId, distinctId, batchId), + () => this.personhog.fetchForUpdate(teamId, distinctId, batchId) + ) + } + + fetchPersonsForUpdateByDistinctIds( + teamId: number, + distinctIds: string[], + batchId: number + ): Promise { + // Merge-fold pre-lock: merge flows stay whole on Postgres. + return this.pg.fetchPersonsForUpdateByDistinctIds(teamId, distinctIds, batchId) + } + + createPerson( + createdAt: DateTime, + properties: Properties, + propertiesLastUpdatedAt: PropertiesLastUpdatedAt, + propertiesLastOperation: PropertiesLastOperation, + teamId: number, + isUserId: number | null, + isIdentified: boolean, + uuid: string, + primaryDistinctId: { distinctId: string; version?: number }, + extraDistinctIds: { distinctId: string; version?: number }[] | undefined, + tx: PersonRepositoryTransaction | undefined, + batchId: number + ): Promise { + return this.route( + 'createPerson', + teamId, + () => + this.pg.createPerson( + createdAt, + properties, + propertiesLastUpdatedAt, + propertiesLastOperation, + teamId, + isUserId, + isIdentified, + uuid, + primaryDistinctId, + extraDistinctIds, + tx, + batchId + ), + () => + this.personhog.createPerson( + createdAt, + properties, + propertiesLastUpdatedAt, + propertiesLastOperation, + teamId, + isUserId, + isIdentified, + uuid, + primaryDistinctId, + extraDistinctIds, + batchId + ), + { tx } + ) + } + + applyEventOps( + person: InternalPerson, + ops: EventOps, + distinctId: string, + batchId: number + ): Promise<[InternalPerson, PersonMessage[]]> { + return this.route( + 'applyEventOps', + person.team_id, + () => this.pg.applyEventOps(person, ops, distinctId, batchId), + () => this.personhog.applyEventOps(person, ops, distinctId, batchId), + { + shadow: () => + this.withShadowPerson('applyEventOps', person.team_id, distinctId, batchId, (shadowPerson) => + this.personhog.applyEventOps(shadowPerson, ops, distinctId, batchId) + ), + } + ) + } + + updatePersonWithPropertiesDiffForUpdate( + person: InternalPerson, + propertiesToSet: Properties, + propertiesToUnset: string[], + otherUpdates: Partial, + distinctId: string, + batchId: number, + forceUpdate?: boolean, + tx?: PersonRepositoryTransaction + ): Promise<[InternalPerson, PersonMessage[], boolean]> { + return this.route( + 'updatePersonWithPropertiesDiffForUpdate', + person.team_id, + () => + this.pg.updatePersonWithPropertiesDiffForUpdate( + person, + propertiesToSet, + propertiesToUnset, + otherUpdates, + distinctId, + batchId, + forceUpdate, + tx + ), + () => + this.personhog.updatePersonWithPropertiesDiffForUpdate( + person, + propertiesToSet, + propertiesToUnset, + otherUpdates, + distinctId, + forceUpdate + ), + { + tx, + shadow: () => + this.withShadowPerson( + 'updatePersonWithPropertiesDiffForUpdate', + person.team_id, + distinctId, + batchId, + (shadowPerson) => + this.personhog.updatePersonWithPropertiesDiffForUpdate( + shadowPerson, + propertiesToSet, + propertiesToUnset, + otherUpdates, + distinctId, + forceUpdate + ) + ), + } + ) + } + + // Merge execution runs on Postgres in every mode, deletes included: + // they exist only to destroy a merge's losing persons, which the + // merge saga will own on personhog. + + deletePerson( + person: InternalPerson, + distinctId: string, + tx?: PersonRepositoryTransaction + ): Promise { + return this.pg.deletePerson(person, distinctId, tx) + } + + claimLifecycleMarks( + opId: string, + teamId: number, + persons: LifecycleMarkPerson[], + distinctId: string, + tx?: PersonRepositoryTransaction + ): Promise { + return this.pg.claimLifecycleMarks(opId, teamId, persons, distinctId, tx) + } + + releaseLifecycleMarks( + opId: string, + teamId: number, + distinctId: string, + tx?: PersonRepositoryTransaction + ): Promise { + return this.pg.releaseLifecycleMarks(opId, teamId, distinctId, tx) + } + + isPersonLive(person: InternalPerson, distinctId: string, tx?: PersonRepositoryTransaction): Promise { + return this.pg.isPersonLive(person, distinctId, tx) + } + + updatePersonForMerge( + person: InternalPerson, + update: Partial, + distinctId: string, + batchId: number, + tx?: PersonRepositoryTransaction + ): Promise<[InternalPerson, PersonMessage[], boolean]> { + return this.pg.updatePersonForMerge(person, update, distinctId, batchId, tx) + } + + addDistinctId( + person: InternalPerson, + distinctId: string, + version: number, + tx: PersonRepositoryTransaction | undefined, + batchId: number + ): Promise { + return this.pg.addDistinctId(person, distinctId, version, tx, batchId) + } + + moveDistinctIds( + source: InternalPerson, + target: InternalPerson, + distinctId: string, + limit: number | undefined, + tx: PersonRepositoryTransaction, + batchId: number + ): Promise { + return this.pg.moveDistinctIds(source, target, distinctId, limit, tx, batchId) + } + + moveDistinctIdsFromPersons( + sources: InternalPerson[], + target: InternalPerson, + distinctId: string, + tx: PersonRepositoryTransaction, + batchId: number + ): Promise { + return this.pg.moveDistinctIdsFromPersons(sources, target, distinctId, tx, batchId) + } + + deletePersons( + persons: InternalPerson[], + distinctId: string, + tx?: PersonRepositoryTransaction + ): Promise { + return this.pg.deletePersons(persons, distinctId, tx) + } + + countDistinctIdsForPersons( + teamId: Team['id'], + personIds: InternalPerson['id'][], + distinctId: string, + tx: PersonRepositoryTransaction + ): Promise> { + return this.pg.countDistinctIdsForPersons(teamId, personIds, distinctId, tx) + } + + updateCohortsAndFeatureFlagsForMerge( + teamID: Team['id'], + sourcePersonID: InternalPerson['id'], + targetPersonID: InternalPerson['id'], + distinctId: string, + tx?: PersonRepositoryTransaction + ): Promise { + return this.pg.updateCohortsAndFeatureFlagsForMerge(teamID, sourcePersonID, targetPersonID, distinctId, tx) + } + + updateCohortsAndFeatureFlagsForMergeBatch( + teamID: Team['id'], + sourcePersonIDs: InternalPerson['id'][], + targetPersonID: InternalPerson['id'], + distinctId: string, + tx?: PersonRepositoryTransaction + ): Promise { + return this.pg.updateCohortsAndFeatureFlagsForMergeBatch( + teamID, + sourcePersonIDs, + targetPersonID, + distinctId, + tx + ) + } + + fetchPersonDistinctIds( + person: InternalPerson, + distinctId: string, + limit: number | undefined, + tx: PersonRepositoryTransaction + ): Promise { + return this.pg.fetchPersonDistinctIds(person, distinctId, limit, tx) + } + + personPropertiesSize(personId: string, teamId: number): Promise { + // The personhog world has no size query: the leader enforces the + // ceiling at admission, so a personhog-mode team reports zero, + // mirroring the personhog store's own batch-bound answer. + return this.modeFor(teamId) === 'personhog' + ? Promise.resolve(0) + : this.pg.personPropertiesSize(personId, teamId) + } + + removeDistinctIdFromCache(teamId: number, distinctId: string): void { + this.pg.removeDistinctIdFromCache(teamId, distinctId) + this.personhog.removeDistinctIdFromCache(teamId, distinctId) + } + + async prefetchPersons(teamDistinctIds: { teamId: number; distinctId: string; batchId: number }[]): Promise { + const pgEntries = teamDistinctIds.filter((entry) => this.modeFor(entry.teamId) !== 'personhog') + const personhogEntries = teamDistinctIds.filter((entry) => this.modeFor(entry.teamId) !== 'pg') + if (pgEntries.length > 0) { + await this.pg.prefetchPersons(pgEntries) + } + if (personhogEntries.length > 0) { + await this.shadowedOrDirect('prefetchPersons', () => this.personhog.prefetchPersons(personhogEntries)) + } + } + + /** + * The personhog side of a fan-out: swallowed in shadow, propagated in + * personhog mode, where the store is authoritative and redelivery is + * the retry. + */ + private async shadowedOrDirect(verb: string, run: () => Promise): Promise { + if (this.mode === 'shadow') { + await this.shadowed(verb, run) + } else { + await run() + } + } + + getFlushStats(): BatchWritingStoreFlushStats { + return this.pg.getFlushStats() + } + + async flush(): Promise { + // Flushes every active batch's personhog lanes; assumes the + // consumer processes one batch at a time, as it does today — + // concurrent batches would ship each other's half-folded lanes + // early (correct, just less folded). + const results = await this.pg.flush() + for (const batchId of this.activeBatches) { + await this.shadowedOrDirect('flush', () => this.personhog.flush(batchId)) + } + return results + } + + releaseBatch(batchId: number): void { + this.activeBatches.delete(batchId) + this.pg.releaseBatch(batchId) + this.personhog.releaseBatch(batchId) + } + + async shutdown(): Promise { + try { + await this.pg.shutdown() + } finally { + await this.personhog.shutdown() + } + } +} diff --git a/nodejs/src/servers/ingestion-api-server.ts b/nodejs/src/servers/ingestion-api-server.ts index 9d83eec26ace..585c40bc032e 100644 --- a/nodejs/src/servers/ingestion-api-server.ts +++ b/nodejs/src/servers/ingestion-api-server.ts @@ -14,6 +14,9 @@ import { ClickhouseGroupRepository } from '~/common/groups/repositories/clickhou import { PostgresGroupRepository } from '~/common/groups/repositories/postgres-group-repository' import { KafkaProducerRegistry } from '~/common/outputs/kafka-producer-registry' import { PersonHogConfig, buildGroupRepository, buildPersonRepository, createPersonHogClient } from '~/common/personhog' +import { PersonHogClient, parseRolloutTeamIds } from '~/common/personhog/client' +import { createIdentityClients } from '~/common/personhog/identity-clients' +import { PersonHogPersonWriteRepository } from '~/common/personhog/personhog-person-write-repository' import { PostgresPersonRepository } from '~/common/persons/repositories/postgres-person-repository' import { PostgresRouter } from '~/common/utils/db/postgres' import { createRedisPoolFromConfig } from '~/common/utils/db/redis' @@ -37,7 +40,13 @@ import { } from '~/ingestion/common/outputs/producers' import { BatchWritingPersonsStore } from '~/ingestion/common/persons/batch-writing-person-store' import { effectivePersonMergeEventsEnabled } from '~/ingestion/common/persons/person-merge-event' +import { PersonhogPersonsStore } from '~/ingestion/common/persons/personhog-persons-store' import { PersonsStore } from '~/ingestion/common/persons/persons-store' +import { + RoutingPersonsStore, + assertPersonsStoreModeConfig, + parsePersonsStoreMode, +} from '~/ingestion/common/persons/routing-persons-store' import { FlushBatchStoresOutputs, createGroupProducePromises, @@ -171,6 +180,8 @@ export class IngestionApiServer implements NodeServer { private cookielessManager?: CookielessManager private pubsub?: PubSub private personsStore?: BatchWritingPersonsStore + private personhogStore?: PersonhogPersonsStore + private personhogClientClosers: Array<() => void> = [] private groupStore?: BatchWritingGroupStore // Held so shutdown cleanup can produce ClickHouse messages returned by a // bare groupStore.flush() — the store itself no longer holds outputs @@ -363,7 +374,48 @@ export class IngestionApiServer implements NodeServer { optimisticUpdateRetryInterval: this.config.PERSON_BATCH_WRITING_OPTIMISTIC_UPDATE_RETRY_INTERVAL_MS, updateAllProperties: this.config.PERSON_PROPERTIES_UPDATE_ALL, }) - const personsStore: PersonsStore = this.personsStore + // Which world person writes land in: pg (default) builds nothing + // new; the other modes construct the personhog store and route + // per team, shadow keeping pg authoritative. + const personsStoreMode = parsePersonsStoreMode(this.config.PERSONS_STORE_MODE) + assertPersonsStoreModeConfig(personsStoreMode, { + routerAddr: this.config.PERSONHOG_ADDR, + identityAddr: this.config.PERSONHOG_IDENTITY_ADDR, + }) + let personsStore: PersonsStore = this.personsStore + if (personsStoreMode !== 'pg') { + const routerClient = PersonHogClient.fromConfig({ + addr: this.config.PERSONHOG_ADDR, + useTls: this.config.PERSONHOG_TLS, + timeoutMs: this.config.PERSONHOG_TIMEOUT_MS, + readMaxBytes: this.config.PERSONHOG_READ_MAX_BYTES, + writeMaxBytes: this.config.PERSONHOG_WRITE_MAX_BYTES, + clientName: 'ingestion-persons-store', + }) + const identityClients = createIdentityClients({ + addr: this.config.PERSONHOG_IDENTITY_ADDR, + useTls: this.config.PERSONHOG_TLS, + timeoutMs: this.config.PERSONHOG_TIMEOUT_MS, + clientName: 'ingestion-persons-store', + }) + this.personhogClientClosers = [() => routerClient.close(), identityClients.close] + const writeRepository = new PersonHogPersonWriteRepository( + routerClient, + identityClients.identity, + 'ingestion-persons-store' + ) + this.personhogStore = new PersonhogPersonsStore(writeRepository, { + maxConcurrentUpdates: this.config.PERSONHOG_STORE_MAX_CONCURRENT_UPDATES, + updateAllProperties: this.config.PERSON_PROPERTIES_UPDATE_ALL, + }) + const teams = parseRolloutTeamIds(this.config.PERSONS_STORE_MODE_TEAMS) + personsStore = new RoutingPersonsStore( + this.personsStore, + this.personhogStore, + personsStoreMode, + teams.size > 0 ? teams : null + ) + } this.groupStore = new BatchWritingGroupStore(groupRepository, clickhouseGroupRepository, { useBatchUpdates: this.config.GROUP_BATCH_WRITING_USE_BATCH_UPDATES, @@ -584,6 +636,10 @@ export class IngestionApiServer implements NodeServer { await this.personsStore.flushAndProduceMessages() await this.personsStore.shutdown() } + if (this.personhogStore) { + await this.personhogStore.shutdown() + } + this.personhogClientClosers.forEach((close) => close()) if (this.groupStore) { const groupFlushResults = await this.groupStore.flush() // flush() returns messages for the caller to produce (it no From 5de8facb26bd568aab0c4d59ca003dfdea737225 Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Tue, 11 Aug 2026 23:02:12 -0400 Subject: [PATCH 2/6] fix(personhog): keep merge execution inside one person world MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review found that a personhog-routed team's merge fetched persons from the personhog world and then mutated Postgres rows keyed on those ids, whose sequences are independent — wrong-row writes on collision. Two mechanisms allowed it: the routing store delegated inTransaction to the Postgres store, whose transaction wrapper wraps itself, so nothing inside a merge transaction ever routed again; and the personhog store implemented no interface, so its missing merge members were invisible and routing silently delegated them to Postgres. The personhog store now declares PersonsStore, with the compiler enforcing completeness: merge mutations are the merge saga's loud placeholders, cohort bookkeeping keeps its deliberate no-ops, and the vestigial batch-bound class the pipeline never reached is deleted. The routing store builds the transaction wrapper around itself, so transactional verbs re-enter routing, and merge verbs route to the team's world and nowhere else — a routed team's merge fails at the store placeholder before any Postgres mutation. Store mocks are now compile-checked against the interface, so drift breaks the build instead of surviving as stale mocks. --- .../persons/personhog-persons-store.test.ts | 8 +- .../common/persons/personhog-persons-store.ts | 395 +++++++----------- .../persons/routing-persons-store.test.ts | 374 ++++++++++------- .../common/persons/routing-persons-store.ts | 106 +++-- nodejs/src/servers/ingestion-api-server.ts | 3 +- 5 files changed, 441 insertions(+), 445 deletions(-) diff --git a/nodejs/src/ingestion/common/persons/personhog-persons-store.test.ts b/nodejs/src/ingestion/common/persons/personhog-persons-store.test.ts index 1e0261e65cba..ebb19efcae06 100644 --- a/nodejs/src/ingestion/common/persons/personhog-persons-store.test.ts +++ b/nodejs/src/ingestion/common/persons/personhog-persons-store.test.ts @@ -361,13 +361,9 @@ describe('PersonhogPersonsStore', () => { expect(results).toEqual([]) }) - it('runs transactions as a passthrough over the store itself', async () => { + it('has no transactions of its own; the routing store owns them', () => { const bound = store.forBatch(0) - const result = await bound.inTransaction('test', (tx) => { - expect(tx).toBe(bound) - return Promise.resolve('done') - }) - expect(result).toBe('done') + expect(() => bound.inTransaction('test', () => Promise.resolve('done'))).toThrow('no personhog RPC') }) it.each([ diff --git a/nodejs/src/ingestion/common/persons/personhog-persons-store.ts b/nodejs/src/ingestion/common/persons/personhog-persons-store.ts index bd48697e3a95..54d696d26b3e 100644 --- a/nodejs/src/ingestion/common/persons/personhog-persons-store.ts +++ b/nodejs/src/ingestion/common/persons/personhog-persons-store.ts @@ -10,12 +10,14 @@ import { PersonRepositoryTransaction } from '~/common/persons/repositories/perso import { CreatePersonResult, MoveDistinctIdsResult } from '~/common/utils/db/db' import { logger } from '~/common/utils/logger' import { NoRowsUpdatedError } from '~/common/utils/utils' +import { BatchWritingStoreFlushStats } from '~/ingestion/common/stores/batch-writing-store' import { Properties } from '~/plugin-scaffold' import { InternalPerson, PropertiesLastOperation, PropertiesLastUpdatedAt, Team } from '~/types' import { EventOps, applyEventPropertyUpdates, computeOpsScalarUpdates, foldOps, refineEventOps } from './person-update' -import { FlushResult } from './persons-store' -import { PersonsStoreForBatch, PersonsStoreTransactionForBatch } from './persons-store-for-batch' +import { FlushResult, PersonsStore } from './persons-store' +import { BatchBoundPersonsStore, PersonsStoreForBatch } from './persons-store-for-batch' +import { PersonsStoreTransaction } from './persons-store-transaction' export const personhogStoreFlushCounter = new Counter({ name: 'personhog_store_flush_ops_total', @@ -103,7 +105,7 @@ interface OpsLaneEntry { * identity service; the uuid argument to createPerson is advisory and * the returned person carries the authoritative one. */ -export class PersonhogPersonsStore { +export class PersonhogPersonsStore implements PersonsStore { private options: PersonhogPersonsStoreOptions /** Folded ops per batch, keyed by `${teamId}:${personId}`. */ private lanes: Map> = new Map() @@ -130,7 +132,7 @@ export class PersonhogPersonsStore { } forBatch(batchId: number): PersonsStoreForBatch { - return new BatchBoundPersonhogStore(this, batchId) + return new BatchBoundPersonsStore(this, batchId) } /** @@ -258,6 +260,7 @@ export class PersonhogPersonsStore { _uuid: string, primaryDistinctId: { distinctId: string; version?: number }, extraDistinctIds: { distinctId: string; version?: number }[] | undefined, + _tx: PersonRepositoryTransaction | undefined, batchId: number ): Promise { const { person, created } = await this.repository.getOrCreatePersonByDistinctId( @@ -348,11 +351,136 @@ export class PersonhogPersonsStore { // the shadow gates merge events off this store; once merges move to // personhog, the merge saga owns those deletions end to end, so no // store-level delete path will ever be needed here. - deletePersons(_persons: InternalPerson[], _distinctId: string, _batchId?: number): Promise { + /** + * The personhog world has no Postgres transactions; transaction + * semantics for routed deployments live in the routing store, which + * never delegates this member. Reaching it is a wiring bug. + */ + inTransaction(_description: string, _transaction: (tx: PersonsStoreTransaction) => Promise): Promise { + throw new PersonhogPendingRpcError('inTransaction', 'merge saga') + } + + // Merge execution is the merge saga's once it lands; until then every + // mutation in the family is a loud placeholder. + + updatePersonForMerge( + _person: InternalPerson, + _update: Partial, + _distinctId: string, + _batchId: number, + _tx?: PersonRepositoryTransaction + ): Promise<[InternalPerson, PersonMessage[], boolean]> { + throw new PersonhogPendingRpcError('updatePersonForMerge', 'merge saga') + } + + claimLifecycleMarks( + _opId: string, + _teamId: number, + _persons: LifecycleMarkPerson[], + _distinctId: string, + _tx?: PersonRepositoryTransaction + ): Promise { + throw new PersonhogPendingRpcError('claimLifecycleMarks', 'merge saga') + } + + releaseLifecycleMarks( + _opId: string, + _teamId: number, + _distinctId: string, + _tx?: PersonRepositoryTransaction + ): Promise { + throw new PersonhogPendingRpcError('releaseLifecycleMarks', 'merge saga') + } + + isPersonLive(_person: InternalPerson, _distinctId: string, _tx?: PersonRepositoryTransaction): Promise { + throw new PersonhogPendingRpcError('isPersonLive', 'merge saga') + } + + addDistinctId( + _person: InternalPerson, + _distinctId: string, + _version: number, + _tx: PersonRepositoryTransaction | undefined, + _batchId: number + ): Promise { + throw new PersonhogPendingRpcError('addDistinctId', 'merge saga') + } + + moveDistinctIds( + _source: InternalPerson, + _target: InternalPerson, + _distinctId: string, + _limit: number | undefined, + _tx: PersonRepositoryTransaction, + _batchId: number + ): Promise { + throw new PersonhogPendingRpcError('moveDistinctIds', 'merge saga') + } + + moveDistinctIdsFromPersons( + _sources: InternalPerson[], + _target: InternalPerson, + _distinctId: string, + _tx: PersonRepositoryTransaction, + _batchId: number + ): Promise { + throw new PersonhogPendingRpcError('moveDistinctIdsFromPersons', 'merge saga') + } + + // Postgres bookkeeping with nothing to answer in this world: shadow + // teams are fresh, so no cohort rows or hash-key overrides exist to + // fix up. + + updateCohortsAndFeatureFlagsForMerge( + _teamID: Team['id'], + _sourcePersonID: InternalPerson['id'], + _targetPersonID: InternalPerson['id'], + _distinctId: string, + _tx?: PersonRepositoryTransaction + ): Promise { + return Promise.resolve() + } + + updateCohortsAndFeatureFlagsForMergeBatch( + _teamID: Team['id'], + _sourcePersonIDs: InternalPerson['id'][], + _targetPersonID: InternalPerson['id'], + _distinctId: string, + _tx?: PersonRepositoryTransaction + ): Promise { + return Promise.resolve() + } + + /** The leader enforces the size ceiling at admission; there is nothing to measure here. */ + personPropertiesSize(_personId: string, _teamId: number): Promise { + return Promise.resolve(0) + } + + getFlushStats(): BatchWritingStoreFlushStats { + let dirtyEntryCount = 0 + for (const lane of this.lanes.values()) { + dirtyEntryCount += lane.size + } + let cacheEntryCount = 0 + for (const memo of this.personState.values()) { + cacheEntryCount += memo.size + } + return { dirtyEntryCount, referencedBatchCount: this.lanes.size, cacheEntryCount } + } + + deletePersons( + _persons: InternalPerson[], + _distinctId: string, + _tx?: PersonRepositoryTransaction + ): Promise { throw new PersonhogPendingRpcError('deletePersons', 'merge saga') } - deletePerson(_person: InternalPerson, _distinctId: string, _batchId?: number): Promise { + deletePerson( + _person: InternalPerson, + _distinctId: string, + _tx?: PersonRepositoryTransaction + ): Promise { throw new PersonhogPendingRpcError('deletePerson', 'merge saga') } @@ -369,7 +497,9 @@ export class PersonhogPersonsStore { propertiesToUnset: string[], otherUpdates: Partial, _distinctId: string, - _forceUpdate?: boolean + _batchId: number, + _forceUpdate?: boolean, + _tx?: PersonRepositoryTransaction ): Promise<[InternalPerson, PersonMessage[], boolean]> { const unsupported = Object.keys(otherUpdates).filter((key) => key !== 'is_identified' && key !== 'last_seen_at') if (unsupported.length > 0) { @@ -407,13 +537,20 @@ export class PersonhogPersonsStore { */ async countDistinctIdsForPersons( teamId: Team['id'], - personIds: InternalPerson['id'][] + personIds: InternalPerson['id'][], + _distinctId: string, + _tx: PersonRepositoryTransaction ): Promise> { const byPerson = await this.repository.getDistinctIdsForPersons(teamId, personIds, undefined, CALLER_TAG) return new Map(personIds.map((id) => [id, byPerson[id]?.length ?? 0])) } - async fetchPersonDistinctIds(person: InternalPerson, limit?: number): Promise { + async fetchPersonDistinctIds( + person: InternalPerson, + _distinctId: string, + limit: number | undefined, + _tx: PersonRepositoryTransaction + ): Promise { const byPerson = await this.repository.getDistinctIdsForPersons(person.team_id, [person.id], limit, CALLER_TAG) return byPerson[person.id] ?? [] } @@ -490,7 +627,12 @@ export class PersonhogPersonsStore { * rather than shipped, the same no-op classification the Postgres * store applies at its flush. The leader never sees the noise. */ - async flush(batchId: number): Promise { + async flush(): Promise { + const results = await Promise.all([...this.lanes.keys()].map((batchId) => this.flushBatch(batchId))) + return results.flat() + } + + private async flushBatch(batchId: number): Promise { const lane = this.lanes.get(batchId) if (!lane) { return [] @@ -603,236 +745,3 @@ export class PersonhogPersonsStore { return memo } } - -/** - * The batch-bound view: batchId curried. It doubles as its own - * transaction view: `inTransaction` is a passthrough, because there are - * no client-side transactions here. Merge safety comes from the merge - * flow's own progress tracking over idempotent leader verbs, and the - * merge-execution verbs throw, so the passthrough cannot silently - * half-merge. - */ -class BatchBoundPersonhogStore implements PersonsStoreForBatch, PersonsStoreTransactionForBatch { - constructor( - private readonly store: PersonhogPersonsStore, - public readonly batchId: number - ) {} - - fetchForChecking(teamId: number, distinctId: string): Promise { - return this.store.fetchForChecking(teamId, distinctId, this.batchId) - } - - fetchForUpdate(teamId: number, distinctId: string): Promise { - return this.store.fetchForUpdate(teamId, distinctId, this.batchId) - } - - fetchPersonsForUpdateByDistinctIds(teamId: number, distinctIds: string[]): Promise { - return this.store.fetchPersonsForUpdateByDistinctIds(teamId, distinctIds, this.batchId) - } - - applyEventOps( - person: InternalPerson, - ops: EventOps, - distinctId: string - ): Promise<[InternalPerson, PersonMessage[]]> { - return this.store.applyEventOps(person, ops, distinctId, this.batchId) - } - - createPerson( - createdAt: DateTime, - properties: Properties, - propertiesLastUpdatedAt: PropertiesLastUpdatedAt, - propertiesLastOperation: PropertiesLastOperation, - teamId: number, - isUserId: number | null, - isIdentified: boolean, - uuid: string, - primaryDistinctId: { distinctId: string; version?: number }, - extraDistinctIds?: { distinctId: string; version?: number }[] - ): Promise { - return this.store.createPerson( - createdAt, - properties, - propertiesLastUpdatedAt, - propertiesLastOperation, - teamId, - isUserId, - isIdentified, - uuid, - primaryDistinctId, - extraDistinctIds, - this.batchId - ) - } - - deletePersons(persons: InternalPerson[], distinctId: string): Promise { - return this.store.deletePersons(persons, distinctId, this.batchId) - } - - deletePerson( - person: InternalPerson, - distinctId: string, - _tx?: PersonRepositoryTransaction - ): Promise { - return this.store.deletePerson(person, distinctId, this.batchId) - } - - // Lifecycle marks exist to serialize merges against lifecycle - // operations, and merge events are gated off this store. - - claimLifecycleMarks( - _opId: string, - _teamId: number, - _persons: LifecycleMarkPerson[], - _distinctId: string, - _tx?: PersonRepositoryTransaction - ): Promise { - throw new PersonhogPendingRpcError('claimLifecycleMarks', 'merge saga') - } - - releaseLifecycleMarks( - _opId: string, - _teamId: number, - _distinctId: string, - _tx?: PersonRepositoryTransaction - ): Promise { - throw new PersonhogPendingRpcError('releaseLifecycleMarks', 'merge saga') - } - - isPersonLive(_person: InternalPerson, _distinctId: string, _tx?: PersonRepositoryTransaction): Promise { - throw new PersonhogPendingRpcError('isPersonLive', 'merge saga') - } - - inTransaction( - _description: string, - transaction: (tx: PersonsStoreTransactionForBatch) => Promise - ): Promise { - return transaction(this) - } - - updatePersonWithPropertiesDiffForUpdate( - person: InternalPerson, - propertiesToSet: Properties, - propertiesToUnset: string[], - otherUpdates: Partial, - distinctId: string, - forceUpdate?: boolean, - _tx?: PersonRepositoryTransaction - ): Promise<[InternalPerson, PersonMessage[], boolean]> { - return this.store.updatePersonWithPropertiesDiffForUpdate( - person, - propertiesToSet, - propertiesToUnset, - otherUpdates, - distinctId, - forceUpdate - ) - } - - countDistinctIdsForPersons( - teamId: Team['id'], - personIds: InternalPerson['id'][], - _distinctId: string, - _tx?: PersonRepositoryTransaction - ): Promise> { - return this.store.countDistinctIdsForPersons(teamId, personIds) - } - - fetchPersonDistinctIds( - person: InternalPerson, - _distinctId: string, - limit?: number, - _tx?: PersonRepositoryTransaction - ): Promise { - return this.store.fetchPersonDistinctIds(person, limit) - } - - // Merge execution is unsupported: each verb throws, naming the RPC - // it lacks, and shadow processing gates merge events off this store. - // - // An implementation must clear the source persons' fold lanes as its - // last step: the lanes' pending content already traveled to the - // merge target through the memo projection, so flushing them after - // the merge would manufacture not_found outcomes and drown the - // cross-batch race signal that counter exists to carry. - - updatePersonForMerge( - _person: InternalPerson, - _update: Partial, - _distinctId: string, - _tx?: PersonRepositoryTransaction - ): Promise<[InternalPerson, PersonMessage[], boolean]> { - throw new PersonhogPendingRpcError( - 'updatePersonForMerge', - 'created_at min-merge and merge version semantics on UpdatePersonProperties' - ) - } - - addDistinctId(_person: InternalPerson, _distinctId: string, _version: number): Promise { - throw new PersonhogPendingRpcError('addDistinctId', 'an idempotent AddDistinctId RPC') - } - - moveDistinctIds( - _source: InternalPerson, - _target: InternalPerson, - _distinctId: string, - _limit?: number, - _tx?: PersonRepositoryTransaction - ): Promise { - throw new PersonhogPendingRpcError('moveDistinctIds', 'an idempotent MoveDistinctIds RPC') - } - - moveDistinctIdsFromPersons( - _sources: InternalPerson[], - _target: InternalPerson, - _distinctId: string, - _tx?: PersonRepositoryTransaction - ): Promise { - throw new PersonhogPendingRpcError('moveDistinctIdsFromPersons', 'an idempotent MoveDistinctIds RPC') - } - - // Postgres bookkeeping with nothing to answer in this world: shadow - // teams are fresh, so no cohort rows or hash-key overrides exist to - // fix up. - - updateCohortsAndFeatureFlagsForMerge( - _teamID: Team['id'], - _sourcePersonID: InternalPerson['id'], - _targetPersonID: InternalPerson['id'], - _distinctId: string, - _tx?: PersonRepositoryTransaction - ): Promise { - return Promise.resolve() - } - - updateCohortsAndFeatureFlagsForMergeBatch( - _teamID: Team['id'], - _sourcePersonIDs: InternalPerson['id'][], - _targetPersonID: InternalPerson['id'], - _distinctId: string, - _tx?: PersonRepositoryTransaction - ): Promise { - return Promise.resolve() - } - - /** The leader enforces the properties-size ceiling at admission. */ - personPropertiesSize(_personId: string, _teamId: number): Promise { - return Promise.resolve(0) - } - - removeDistinctIdFromCache(teamId: number, distinctId: string): void { - this.store.removeDistinctIdFromCache(teamId, distinctId) - } - - prefetchPersons(teamDistinctIds: { teamId: number; distinctId: string; batchId: number }[]): Promise { - return this.store.prefetchPersons(teamDistinctIds) - } - - flush(): Promise { - return this.store.flush(this.batchId) - } - - shutdown(): Promise { - return this.store.shutdown() - } -} diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts index 7465b4b8c7f7..24debc752528 100644 --- a/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts @@ -1,6 +1,11 @@ import { personhogStoreShadowErrorsCounter, personhogStoreShadowSkipsCounter } from '~/common/persons/metrics' +import { PersonRepository } from '~/common/persons/repositories/person-repository' +import { PersonRepositoryTransaction } from '~/common/persons/repositories/person-repository-transaction' import { InternalPerson } from '~/types' +import { EventOps } from './person-update' +import { PersonhogPersonsStore } from './personhog-persons-store' +import { PersonsStore } from './persons-store' import { RoutingPersonsStore, assertPersonsStoreModeConfig, parsePersonsStoreMode } from './routing-persons-store' jest.mock('~/common/persons/metrics', () => ({ @@ -8,40 +13,78 @@ jest.mock('~/common/persons/metrics', () => ({ personhogStoreShadowSkipsCounter: { labels: jest.fn().mockReturnValue({ inc: jest.fn() }) }, })) +/** + * A complete, compile-checked PersonsStore mock: the annotation forces + * every interface member to exist, so an interface change breaks this + * factory at compile time instead of leaving stale mocks that only fail + * when a newly routed method runs. + */ +function mockStore(): jest.Mocked { + return { + inTransaction: jest.fn(), + fetchForChecking: jest.fn().mockResolvedValue(null), + fetchForUpdate: jest.fn().mockResolvedValue(null), + fetchPersonsForUpdateByDistinctIds: jest.fn().mockResolvedValue([]), + createPerson: jest.fn().mockResolvedValue({ success: true }), + updatePersonForMerge: jest.fn(), + applyEventOps: jest.fn(), + updatePersonWithPropertiesDiffForUpdate: jest.fn(), + deletePerson: jest.fn().mockResolvedValue([]), + claimLifecycleMarks: jest.fn().mockResolvedValue(undefined), + releaseLifecycleMarks: jest.fn().mockResolvedValue(undefined), + isPersonLive: jest.fn().mockResolvedValue(true), + addDistinctId: jest.fn().mockResolvedValue([]), + moveDistinctIds: jest.fn().mockResolvedValue({ success: true }), + moveDistinctIdsFromPersons: jest.fn().mockResolvedValue({ success: true }), + deletePersons: jest.fn().mockResolvedValue([]), + countDistinctIdsForPersons: jest.fn().mockResolvedValue(new Map()), + updateCohortsAndFeatureFlagsForMerge: jest.fn().mockResolvedValue(undefined), + updateCohortsAndFeatureFlagsForMergeBatch: jest.fn().mockResolvedValue(undefined), + personPropertiesSize: jest.fn().mockResolvedValue(0), + fetchPersonDistinctIds: jest.fn().mockResolvedValue([]), + shutdown: jest.fn().mockResolvedValue(undefined), + removeDistinctIdFromCache: jest.fn(), + prefetchPersons: jest.fn().mockResolvedValue(undefined), + flush: jest.fn().mockResolvedValue([]), + releaseBatch: jest.fn(), + getFlushStats: jest.fn().mockReturnValue({ dirtyEntryCount: 0, referencedBatchCount: 0, cacheEntryCount: 0 }), + } +} + describe('RoutingPersonsStore', () => { - const person = (teamId: number): InternalPerson => - ({ id: '1', team_id: teamId, properties: {}, is_identified: false }) as unknown as InternalPerson + const fakeTx = {} as PersonRepositoryTransaction + + const person = (teamId: number, id = '1'): InternalPerson => + ({ id, team_id: teamId, properties: {}, is_identified: false }) as unknown as InternalPerson - const ops = { set: {}, setOnce: {}, unset: [], denied: false, shouldForceUpdate: false, eventName: '$set' } + const ops: EventOps = { + set: {}, + setOnce: {}, + unset: [], + denied: false, + shouldForceUpdate: false, + eventName: '$set', + } as unknown as EventOps const makeStores = () => { - const pg = { - fetchForChecking: jest.fn().mockResolvedValue(null), - fetchForUpdate: jest.fn().mockResolvedValue(person(1)), - applyEventOps: jest.fn().mockResolvedValue([person(1), []]), - createPerson: jest.fn().mockResolvedValue({ success: true }), - deletePerson: jest.fn().mockResolvedValue([]), - moveDistinctIds: jest.fn().mockResolvedValue({ success: true }), - prefetchPersons: jest.fn().mockResolvedValue(undefined), - flush: jest.fn().mockResolvedValue([]), - releaseBatch: jest.fn(), - shutdown: jest.fn().mockResolvedValue(undefined), - } as any - const personhog = { - fetchForChecking: jest.fn().mockResolvedValue(null), - fetchForUpdate: jest.fn().mockResolvedValue(person(1)), - applyEventOps: jest.fn().mockResolvedValue([person(1), []]), - createPerson: jest.fn().mockResolvedValue({ success: true }), - deletePerson: jest.fn().mockResolvedValue([]), - prefetchPersons: jest.fn().mockResolvedValue(undefined), - flush: jest.fn().mockResolvedValue([]), - releaseBatch: jest.fn(), - removeDistinctIdFromCache: jest.fn(), - shutdown: jest.fn().mockResolvedValue(undefined), - } as any - return { pg, personhog } + const pg = mockStore() + // The personhog store implements PersonsStore, so the same + // compile-checked factory serves; the cast to the concrete class + // is the constructor's requirement, not an escape from checking. + const personhogMock = mockStore() + const personhog = personhogMock as unknown as PersonhogPersonsStore + const personRepository: Pick = { + inTransaction: jest.fn((_description, cb) => cb(fakeTx)) as PersonRepository['inTransaction'], + } + return { pg, personhogMock, personhog, personRepository } } + const makeStore = ( + stores: ReturnType, + mode: 'personhog' | 'shadow', + teams: ReadonlySet | null + ) => new RoutingPersonsStore(stores.pg, stores.personhog, mode, teams, stores.personRepository) + it('rejects an unknown mode at parse time', () => { expect(() => parsePersonsStoreMode('both')).toThrow('PERSONS_STORE_MODE') expect(parsePersonsStoreMode('shadow')).toBe('shadow') @@ -59,183 +102,208 @@ describe('RoutingPersonsStore', () => { expect(() => assertPersonsStoreModeConfig('pg', { routerAddr: '', identityAddr: '' })).not.toThrow() }) - describe('personhog mode with a team allowlist', () => { + describe('personhog mode', () => { it('routes allowlisted teams to personhog and the rest to pg', async () => { - const { pg, personhog } = makeStores() - const store = new RoutingPersonsStore(pg, personhog, 'personhog', new Set([1])) + const stores = makeStores() + const store = makeStore(stores, 'personhog', new Set([1])) await store.fetchForUpdate(1, 'a', 0) - expect(personhog.fetchForUpdate).toHaveBeenCalledWith(1, 'a', 0) - expect(pg.fetchForUpdate).not.toHaveBeenCalled() + expect(stores.personhogMock.fetchForUpdate).toHaveBeenCalledWith(1, 'a', 0) + expect(stores.pg.fetchForUpdate).not.toHaveBeenCalled() await store.fetchForUpdate(2, 'b', 0) - expect(pg.fetchForUpdate).toHaveBeenCalledWith(2, 'b', 0) - }) - - it('a personhog flush failure propagates, because the store is authoritative', async () => { - const { pg, personhog } = makeStores() - personhog.flush.mockRejectedValue(new Error('leader down')) - const store = new RoutingPersonsStore(pg, personhog, 'personhog', null) - store.forBatch(7) - await expect(store.flush()).rejects.toThrow('leader down') - }) - }) - - describe('shadow mode', () => { - it('returns the pg result and runs the personhog verb after it', async () => { - const { pg, personhog } = makeStores() - const pgPerson = person(1) - pg.applyEventOps.mockResolvedValue([pgPerson, []]) - const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) - - const [updated] = await store.applyEventOps(person(1), ops as any, 'a', 0) - expect(updated).toBe(pgPerson) - expect(personhog.applyEventOps).toHaveBeenCalled() - }) - - it('a personhog failure is counted and never fails the batch', async () => { - const { pg, personhog } = makeStores() - personhog.applyEventOps.mockRejectedValue(new Error('identity down')) - const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) - - await expect(store.applyEventOps(person(1), ops as any, 'a', 0)).resolves.toBeDefined() - expect(personhogStoreShadowErrorsCounter.labels).toHaveBeenCalledWith({ verb: 'applyEventOps' }) - }) - - it('a pg failure still fails the batch', async () => { - const { pg, personhog } = makeStores() - pg.applyEventOps.mockRejectedValue(new Error('pg down')) - const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) - await expect(store.applyEventOps(person(1), ops as any, 'a', 0)).rejects.toThrow('pg down') - expect(personhog.applyEventOps).not.toHaveBeenCalled() - }) - - it('a personhog flush failure is swallowed', async () => { - const { pg, personhog } = makeStores() - personhog.flush.mockRejectedValue(new Error('leader down')) - const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) - store.forBatch(7) - await expect(store.flush()).resolves.toEqual([]) - expect(personhog.flush).toHaveBeenCalledWith(7) - }) - }) - - describe('verbs that never route', () => { - it('merge execution stays on pg for an allowlisted team', async () => { - const { pg, personhog } = makeStores() - const store = new RoutingPersonsStore(pg, personhog, 'personhog', new Set([1])) - const tx = {} as any - await store.moveDistinctIds(person(1), person(1), 'a', undefined, tx, 0) - expect(pg.moveDistinctIds).toHaveBeenCalled() + expect(stores.pg.fetchForUpdate).toHaveBeenCalledWith(2, 'b', 0) }) it('a transactional create stays on pg for an allowlisted team', async () => { - const { pg, personhog } = makeStores() - const store = new RoutingPersonsStore(pg, personhog, 'personhog', new Set([1])) - const tx = {} as any + const stores = makeStores() + const store = makeStore(stores, 'personhog', new Set([1])) await store.createPerson( - null as any, + undefined as never, {}, {}, {}, 1, null, false, - 'uuid', - { distinctId: 'a' }, + 'u', + { distinctId: 'd' }, undefined, - tx, + fakeTx, 0 ) - expect(pg.createPerson).toHaveBeenCalled() - expect(personhog.createPerson).not.toHaveBeenCalled() + expect(stores.pg.createPerson).toHaveBeenCalled() + expect(stores.personhogMock.createPerson).not.toHaveBeenCalled() }) - }) - it('prefetch splits entries by route', async () => { - const { pg, personhog } = makeStores() - const store = new RoutingPersonsStore(pg, personhog, 'personhog', new Set([1])) - await store.prefetchPersons([ - { teamId: 1, distinctId: 'a', batchId: 0 }, - { teamId: 2, distinctId: 'b', batchId: 0 }, - ]) - expect(pg.prefetchPersons).toHaveBeenCalledWith([{ teamId: 2, distinctId: 'b', batchId: 0 }]) - expect(personhog.prefetchPersons).toHaveBeenCalledWith([{ teamId: 1, distinctId: 'a', batchId: 0 }]) + it('a personhog flush failure propagates, because the store is authoritative', async () => { + const stores = makeStores() + stores.personhogMock.flush.mockRejectedValue(new Error('leader down')) + const store = makeStore(stores, 'personhog', null) + await expect(store.flush()).rejects.toThrow('leader down') + }) }) - it('a released batch is no longer flushed on the personhog side', async () => { - const { pg, personhog } = makeStores() - const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) - store.forBatch(7) - store.releaseBatch(7) - await store.flush() - expect(personhog.flush).not.toHaveBeenCalled() - expect(personhog.releaseBatch).toHaveBeenCalledWith(7) + describe('shadow mode', () => { + it('pg is authoritative and the personhog verb runs shadowed', async () => { + const stores = makeStores() + stores.pg.fetchForUpdate.mockResolvedValue(person(1, '7')) + stores.personhogMock.fetchForUpdate.mockResolvedValue(person(1, '99')) + const store = makeStore(stores, 'shadow', null) + + const result = await store.fetchForUpdate(1, 'a', 0) + + expect(result?.id).toBe('7') + expect(stores.personhogMock.fetchForUpdate).toHaveBeenCalled() + }) + + it('a shadow failure is swallowed and counted, never failing the batch', async () => { + const stores = makeStores() + stores.pg.fetchForUpdate.mockResolvedValue(person(1, '7')) + stores.personhogMock.fetchForUpdate.mockRejectedValue(new Error('identity down')) + const store = makeStore(stores, 'shadow', null) + + const result = await store.fetchForUpdate(1, 'a', 0) + + expect(result?.id).toBe('7') + expect(personhogStoreShadowErrorsCounter.labels).toHaveBeenCalledWith({ verb: 'fetchForUpdate' }) + }) + + it('a shadow flush failure is swallowed', async () => { + const stores = makeStores() + stores.personhogMock.flush.mockRejectedValue(new Error('leader down')) + const store = makeStore(stores, 'shadow', null) + await expect(store.flush()).resolves.toEqual([]) + }) }) describe('shadow writes resolve the personhog world person', () => { - const pgPerson = { id: '7', team_id: 1, properties: {} } as unknown as InternalPerson - const shadowPerson = { id: '99', team_id: 1, properties: {} } as unknown as InternalPerson - it.each([ [ 'applyEventOps', - (store: RoutingPersonsStore) => store.applyEventOps(pgPerson, ops as any, 'd1', 0), - (personhog: any) => personhog.applyEventOps, + (store: RoutingPersonsStore) => store.applyEventOps(person(1, '7'), ops, 'd1', 0), + (m: jest.Mocked) => m.applyEventOps, ], [ 'updatePersonWithPropertiesDiffForUpdate', (store: RoutingPersonsStore) => - store.updatePersonWithPropertiesDiffForUpdate(pgPerson, { a: '1' }, [], {}, 'd1', 0), - (personhog: any) => personhog.updatePersonWithPropertiesDiffForUpdate, + store.updatePersonWithPropertiesDiffForUpdate(person(1, '7'), { a: '1' }, [], {}, 'd1', 0), + (m: jest.Mocked) => m.updatePersonWithPropertiesDiffForUpdate, ], - ] as const)('%s ships the shadow world id, not the pg id', async (_verb, call, personhogFn) => { - const { pg, personhog } = makeStores() - personhog.fetchForUpdate.mockResolvedValue(shadowPerson) - personhog.updatePersonWithPropertiesDiffForUpdate = jest.fn().mockResolvedValue([shadowPerson, [], true]) - pg.updatePersonWithPropertiesDiffForUpdate = jest.fn().mockResolvedValue([pgPerson, [], true]) - const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + ] as const)('%s ships the shadow world id, not the pg id', async (_verb, call, member) => { + const stores = makeStores() + stores.pg.applyEventOps.mockResolvedValue([person(1, '7'), []]) + stores.pg.updatePersonWithPropertiesDiffForUpdate.mockResolvedValue([person(1, '7'), [], true]) + stores.personhogMock.fetchForUpdate.mockResolvedValue(person(1, '99')) + stores.personhogMock.applyEventOps.mockResolvedValue([person(1, '99'), []]) + stores.personhogMock.updatePersonWithPropertiesDiffForUpdate.mockResolvedValue([person(1, '99'), [], true]) + const store = makeStore(stores, 'shadow', null) await call(store) - // The pg call keeps the caller's person; the shadowed call must - // re-resolve, because pg ids mean nothing in the personhog world. - const shadowArgs = personhogFn(personhog).mock.calls[0] - expect(shadowArgs[0].id).toBe('99') + const shadowArgs = member(stores.personhogMock).mock.calls[0] + expect((shadowArgs[0] as InternalPerson).id).toBe('99') }) it('skips the shadow write, counted, when the person does not exist in the personhog world', async () => { - const { pg, personhog } = makeStores() - pg.applyEventOps.mockResolvedValue([pgPerson, []]) - personhog.fetchForUpdate.mockResolvedValue(null) - const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + const stores = makeStores() + stores.pg.applyEventOps.mockResolvedValue([person(1, '7'), []]) + stores.personhogMock.fetchForUpdate.mockResolvedValue(null) + const store = makeStore(stores, 'shadow', null) - const [result] = await store.applyEventOps(pgPerson, ops as any, 'd1', 0) + const [result] = await store.applyEventOps(person(1, '7'), ops, 'd1', 0) expect(result.id).toBe('7') - expect(personhog.applyEventOps).not.toHaveBeenCalled() + expect(stores.personhogMock.applyEventOps).not.toHaveBeenCalled() expect(personhogStoreShadowSkipsCounter.labels).toHaveBeenCalledWith({ verb: 'applyEventOps' }) }) + }) - it('a failing shadow resolution is swallowed like any shadow error', async () => { - const { pg, personhog } = makeStores() - pg.applyEventOps.mockResolvedValue([pgPerson, []]) - personhog.fetchForUpdate.mockRejectedValue(new Error('identity down')) - const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + describe('merge execution routes to the team world, never across it', () => { + it.each([ + ['deletePersons', (s: RoutingPersonsStore) => s.deletePersons([person(1)], 'd'), 'deletePersons'], + [ + 'addDistinctId', + (s: RoutingPersonsStore) => s.addDistinctId(person(1), 'd', 0, undefined, 0), + 'addDistinctId', + ], + [ + 'updatePersonForMerge', + (s: RoutingPersonsStore) => s.updatePersonForMerge(person(1), {}, 'd', 0), + 'updatePersonForMerge', + ], + [ + 'claimLifecycleMarks', + (s: RoutingPersonsStore) => s.claimLifecycleMarks('op', 1, [], 'd'), + 'claimLifecycleMarks', + ], + ] as const)('%s reaches the personhog store for a routed team', async (_name, call, member) => { + const stores = makeStores() + const store = makeStore(stores, 'personhog', new Set([1])) - const [result] = await store.applyEventOps(pgPerson, ops as any, 'd1', 0) + await call(store) - expect(result.id).toBe('7') - expect(personhogStoreShadowErrorsCounter.labels).toHaveBeenCalledWith({ verb: 'applyEventOps' }) + expect(stores.personhogMock[member as keyof PersonsStore]).toHaveBeenCalled() + expect(stores.pg[member as keyof PersonsStore]).not.toHaveBeenCalled() + }) + + it.each([ + ['a pg-routed team in personhog mode', 'personhog', new Set([99])], + ['shadow mode', 'shadow', null], + ] as const)('%s runs merges on pg, unshadowed', async (_name, mode, teams) => { + const stores = makeStores() + const store = makeStore(stores, mode, teams as ReadonlySet | null) + + await store.deletePersons([person(1)], 'd') + + expect(stores.pg.deletePersons).toHaveBeenCalled() + expect(stores.personhogMock.deletePersons).not.toHaveBeenCalled() + }) + }) + + describe('inTransaction wraps this store, so transactional verbs still route', () => { + it('a routed team merge inside a transaction reaches the personhog store', async () => { + const stores = makeStores() + const store = makeStore(stores, 'personhog', new Set([1])) + + await store.inTransaction('merge', async (tx) => { + await tx.deletePerson(person(1), 'd') + }) + + // The wrapper re-entered routing: the personhog store saw the + // delete, and pg never received a mutation keyed on a + // personhog-world row id. + expect(stores.personhogMock.deletePerson).toHaveBeenCalledWith(person(1), 'd', fakeTx) + expect(stores.pg.deletePerson).not.toHaveBeenCalled() + }) + + it('an unrouted team merge inside a transaction reaches pg with the transaction', async () => { + const stores = makeStores() + const store = makeStore(stores, 'personhog', new Set([99])) + + await store.inTransaction('merge', async (tx) => { + await tx.deletePerson(person(1), 'd') + }) + + expect(stores.pg.deletePerson).toHaveBeenCalledWith(person(1), 'd', fakeTx) + expect(stores.personhogMock.deletePerson).not.toHaveBeenCalled() }) }) it('shutdown closes the personhog side even when pg shutdown fails', async () => { - const { pg, personhog } = makeStores() - pg.shutdown.mockRejectedValue(new Error('pg teardown failed')) - const store = new RoutingPersonsStore(pg, personhog, 'shadow', null) + const stores = makeStores() + stores.pg.shutdown.mockRejectedValue(new Error('pg teardown failed')) + const store = makeStore(stores, 'shadow', null) await expect(store.shutdown()).rejects.toThrow('pg teardown failed') - expect(personhog.shutdown).toHaveBeenCalled() + expect(stores.personhogMock.shutdown).toHaveBeenCalled() + }) + + it('releaseBatch releases both worlds', () => { + const stores = makeStores() + const store = makeStore(stores, 'shadow', null) + store.releaseBatch(4) + expect(stores.pg.releaseBatch).toHaveBeenCalledWith(4) + expect(stores.personhogMock.releaseBatch).toHaveBeenCalledWith(4) }) }) diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.ts index 24f9e33352c8..c51a20bbf199 100644 --- a/nodejs/src/ingestion/common/persons/routing-persons-store.ts +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.ts @@ -3,6 +3,7 @@ import { DateTime } from 'luxon' import { personhogStoreShadowErrorsCounter, personhogStoreShadowSkipsCounter } from '~/common/persons/metrics' import { PersonMessage } from '~/common/persons/person-message' import { InternalPersonWithDistinctId, LifecycleMarkPerson } from '~/common/persons/repositories/person-repository' +import { PersonRepository } from '~/common/persons/repositories/person-repository' import { PersonRepositoryTransaction } from '~/common/persons/repositories/person-repository-transaction' import { CreatePersonResult, MoveDistinctIdsResult } from '~/common/utils/db/db' import { logger } from '~/common/utils/logger' @@ -56,26 +57,25 @@ export function assertPersonsStoreModeConfig( * * Two verb families do not route: * - * - Merge execution — including deleting its losing persons — runs on - * Postgres in every mode (the merge saga will own it on personhog), as - * does any verb invoked under a Postgres transaction (see `route`). + * - Merge execution has no personhog support until the merge saga + * lands: pg and shadow teams run it on Postgres as before, and a + * personhog-routed team fails loudly at the first merge mutation — + * its reads and writes live in the personhog world, whose row ids + * mean nothing to Postgres, so quietly running the merge there would + * mutate whatever rows happen to share the numbers. Any non-merge + * verb invoked under a Postgres transaction still goes to Postgres + * (see `route`). * - `personPropertiesSize` routes by mode but is not shadowed: the * personhog store answers it with a constant because the leader * enforces the size ceiling at admission. */ export class RoutingPersonsStore implements PersonsStore { - /** - * Batches with personhog lanes that may hold unflushed folds. The - * personhog store flushes per batch, so the store-level flush needs - * the live set. - */ - private activeBatches = new Set() - constructor( private pg: PersonsStore, private personhog: PersonhogPersonsStore, private mode: 'personhog' | 'shadow', - private teams: ReadonlySet | null + private teams: ReadonlySet | null, + private personRepository: Pick ) {} private modeFor(teamId: number): PersonsStoreMode { @@ -145,12 +145,21 @@ export class RoutingPersonsStore implements PersonsStore { } forBatch(batchId: number): PersonsStoreForBatch { - this.activeBatches.add(batchId) return new BatchBoundPersonsStore(this, batchId) } + /** + * The transaction wrapper is built around this store, not the + * Postgres one, so every verb inside a merge transaction re-enters + * routing with the transaction attached. Delegating to the Postgres + * store here would hand the callback a wrapper around that store, + * and nothing inside the transaction would route again — which is + * how personhog-world person ids could reach Postgres row keys. + */ inTransaction(description: string, transaction: (tx: PersonsStoreTransaction) => Promise): Promise { - return this.pg.inTransaction(description, transaction) + return this.personRepository.inTransaction(description, (tx: PersonRepositoryTransaction) => + transaction(new PersonsStoreTransaction(this, tx)) + ) } fetchForChecking(teamId: number, distinctId: string, batchId: number): Promise { @@ -224,6 +233,7 @@ export class RoutingPersonsStore implements PersonsStore { uuid, primaryDistinctId, extraDistinctIds, + tx, batchId ), { tx } @@ -281,7 +291,9 @@ export class RoutingPersonsStore implements PersonsStore { propertiesToUnset, otherUpdates, distinctId, - forceUpdate + batchId, + forceUpdate, + tx ), { tx, @@ -298,6 +310,7 @@ export class RoutingPersonsStore implements PersonsStore { propertiesToUnset, otherUpdates, distinctId, + batchId, forceUpdate ) ), @@ -305,16 +318,26 @@ export class RoutingPersonsStore implements PersonsStore { ) } - // Merge execution runs on Postgres in every mode, deletes included: - // they exist only to destroy a merge's losing persons, which the - // merge saga will own on personhog. + /** + * Merge execution routes to the team's world and nowhere else: pg + * and shadow teams run it on Postgres, and a personhog-routed team + * reaches the personhog store, whose members are the merge saga's + * loud placeholders until it lands. Shadowing is deliberately absent + * — a merge mirrored into the other world would key on foreign row + * ids. Transaction context is no exemption: the transaction wrapper + * re-enters this store, and a routed team's merge must fail before + * any Postgres mutation, not run inside one. + */ + private mergeWorld(teamId: number): PersonsStore { + return this.modeFor(teamId) === 'personhog' ? this.personhog : this.pg + } deletePerson( person: InternalPerson, distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.pg.deletePerson(person, distinctId, tx) + return this.mergeWorld(person.team_id).deletePerson(person, distinctId, tx) } claimLifecycleMarks( @@ -324,7 +347,7 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.pg.claimLifecycleMarks(opId, teamId, persons, distinctId, tx) + return this.mergeWorld(teamId).claimLifecycleMarks(opId, teamId, persons, distinctId, tx) } releaseLifecycleMarks( @@ -333,11 +356,11 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.pg.releaseLifecycleMarks(opId, teamId, distinctId, tx) + return this.mergeWorld(teamId).releaseLifecycleMarks(opId, teamId, distinctId, tx) } isPersonLive(person: InternalPerson, distinctId: string, tx?: PersonRepositoryTransaction): Promise { - return this.pg.isPersonLive(person, distinctId, tx) + return this.mergeWorld(person.team_id).isPersonLive(person, distinctId, tx) } updatePersonForMerge( @@ -347,7 +370,7 @@ export class RoutingPersonsStore implements PersonsStore { batchId: number, tx?: PersonRepositoryTransaction ): Promise<[InternalPerson, PersonMessage[], boolean]> { - return this.pg.updatePersonForMerge(person, update, distinctId, batchId, tx) + return this.mergeWorld(person.team_id).updatePersonForMerge(person, update, distinctId, batchId, tx) } addDistinctId( @@ -357,7 +380,7 @@ export class RoutingPersonsStore implements PersonsStore { tx: PersonRepositoryTransaction | undefined, batchId: number ): Promise { - return this.pg.addDistinctId(person, distinctId, version, tx, batchId) + return this.mergeWorld(person.team_id).addDistinctId(person, distinctId, version, tx, batchId) } moveDistinctIds( @@ -368,7 +391,7 @@ export class RoutingPersonsStore implements PersonsStore { tx: PersonRepositoryTransaction, batchId: number ): Promise { - return this.pg.moveDistinctIds(source, target, distinctId, limit, tx, batchId) + return this.mergeWorld(source.team_id).moveDistinctIds(source, target, distinctId, limit, tx, batchId) } moveDistinctIdsFromPersons( @@ -378,7 +401,7 @@ export class RoutingPersonsStore implements PersonsStore { tx: PersonRepositoryTransaction, batchId: number ): Promise { - return this.pg.moveDistinctIdsFromPersons(sources, target, distinctId, tx, batchId) + return this.mergeWorld(target.team_id).moveDistinctIdsFromPersons(sources, target, distinctId, tx, batchId) } deletePersons( @@ -386,7 +409,10 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.pg.deletePersons(persons, distinctId, tx) + if (persons.length === 0) { + return this.pg.deletePersons(persons, distinctId, tx) + } + return this.mergeWorld(persons[0].team_id).deletePersons(persons, distinctId, tx) } countDistinctIdsForPersons( @@ -395,7 +421,7 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx: PersonRepositoryTransaction ): Promise> { - return this.pg.countDistinctIdsForPersons(teamId, personIds, distinctId, tx) + return this.mergeWorld(teamId).countDistinctIdsForPersons(teamId, personIds, distinctId, tx) } updateCohortsAndFeatureFlagsForMerge( @@ -405,7 +431,13 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.pg.updateCohortsAndFeatureFlagsForMerge(teamID, sourcePersonID, targetPersonID, distinctId, tx) + return this.mergeWorld(teamID).updateCohortsAndFeatureFlagsForMerge( + teamID, + sourcePersonID, + targetPersonID, + distinctId, + tx + ) } updateCohortsAndFeatureFlagsForMergeBatch( @@ -415,7 +447,7 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.pg.updateCohortsAndFeatureFlagsForMergeBatch( + return this.mergeWorld(teamID).updateCohortsAndFeatureFlagsForMergeBatch( teamID, sourcePersonIDs, targetPersonID, @@ -430,15 +462,12 @@ export class RoutingPersonsStore implements PersonsStore { limit: number | undefined, tx: PersonRepositoryTransaction ): Promise { - return this.pg.fetchPersonDistinctIds(person, distinctId, limit, tx) + return this.mergeWorld(person.team_id).fetchPersonDistinctIds(person, distinctId, limit, tx) } personPropertiesSize(personId: string, teamId: number): Promise { - // The personhog world has no size query: the leader enforces the - // ceiling at admission, so a personhog-mode team reports zero, - // mirroring the personhog store's own batch-bound answer. return this.modeFor(teamId) === 'personhog' - ? Promise.resolve(0) + ? this.personhog.personPropertiesSize(personId, teamId) : this.pg.personPropertiesSize(personId, teamId) } @@ -476,19 +505,12 @@ export class RoutingPersonsStore implements PersonsStore { } async flush(): Promise { - // Flushes every active batch's personhog lanes; assumes the - // consumer processes one batch at a time, as it does today — - // concurrent batches would ship each other's half-folded lanes - // early (correct, just less folded). const results = await this.pg.flush() - for (const batchId of this.activeBatches) { - await this.shadowedOrDirect('flush', () => this.personhog.flush(batchId)) - } + await this.shadowedOrDirect('flush', () => this.personhog.flush()) return results } releaseBatch(batchId: number): void { - this.activeBatches.delete(batchId) this.pg.releaseBatch(batchId) this.personhog.releaseBatch(batchId) } diff --git a/nodejs/src/servers/ingestion-api-server.ts b/nodejs/src/servers/ingestion-api-server.ts index 585c40bc032e..415d9a03e30f 100644 --- a/nodejs/src/servers/ingestion-api-server.ts +++ b/nodejs/src/servers/ingestion-api-server.ts @@ -413,7 +413,8 @@ export class IngestionApiServer implements NodeServer { this.personsStore, this.personhogStore, personsStoreMode, - teams.size > 0 ? teams : null + teams.size > 0 ? teams : null, + personRepository ) } From e66c3ae6fabd6ee7cfeb4b3ec88ebc75aa14a28c Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Tue, 11 Aug 2026 23:19:09 -0400 Subject: [PATCH 3/6] fix(personhog): route merges through the placeholders, mode-wide MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review-driven simplifications. Merge verbs stop being a special family: they route through the same combinator as every other verb, so personhog mode reaches the store's pending-saga placeholders, shadow mode shadows them like anything else (the placeholder throw lands in the existing swallow-and-count machinery), and implementing the saga changes zero routing lines. The per-team allowlist is gone: PERSONS_STORE_MODE applies to the whole deployment. Per-team sampling coupled unrelated worlds inside one Postgres transaction; with a deployment-wide mode, inTransaction routes by mode like any verb — pg and shadow run the real Postgres transaction, personhog mode answers with the store's placeholder — and the repository-backed transaction wrapper machinery deletes. --- nodejs/src/common/config.ts | 3 - nodejs/src/common/personhog/index.ts | 1 - .../persons/routing-persons-store.test.ts | 84 ++++----- .../common/persons/routing-persons-store.ts | 170 +++++++++++------- nodejs/src/servers/ingestion-api-server.ts | 11 +- 5 files changed, 145 insertions(+), 124 deletions(-) diff --git a/nodejs/src/common/config.ts b/nodejs/src/common/config.ts index dcd52ccf874d..1bf722e08ea2 100644 --- a/nodejs/src/common/config.ts +++ b/nodejs/src/common/config.ts @@ -114,8 +114,6 @@ export type CommonConfig = BaseServerConfig & { PERSONHOG_ENABLED: boolean /** Which world the ingestion persons store writes: 'pg' (default), 'personhog', or 'shadow' (pg authoritative, personhog best-effort). */ PERSONS_STORE_MODE: string - /** Comma-separated team ids the non-pg mode applies to; empty applies it to every team. */ - PERSONS_STORE_MODE_TEAMS: string /** Host and port of the personhog identity server. */ PERSONHOG_IDENTITY_ADDR: string PERSONHOG_ADDR: string @@ -300,7 +298,6 @@ export function getDefaultCommonConfig(): CommonConfig { PERSONHOG_ENABLED: false, PERSONHOG_ADDR: '', PERSONS_STORE_MODE: 'pg', - PERSONS_STORE_MODE_TEAMS: '', PERSONHOG_IDENTITY_ADDR: '', PERSONHOG_GROUPS_ROLLOUT_PERCENTAGE: 0, PERSONHOG_GROUPS_ROLLOUT_TEAM_IDS: '', diff --git a/nodejs/src/common/personhog/index.ts b/nodejs/src/common/personhog/index.ts index da91f5fee3c5..9ce5fd3f3db0 100644 --- a/nodejs/src/common/personhog/index.ts +++ b/nodejs/src/common/personhog/index.ts @@ -19,7 +19,6 @@ export type PersonHogConfig = Pick< | 'PERSONHOG_ADDR' | 'PERSONHOG_IDENTITY_ADDR' | 'PERSONS_STORE_MODE' - | 'PERSONS_STORE_MODE_TEAMS' | 'PERSONHOG_GROUPS_ROLLOUT_PERCENTAGE' | 'PERSONHOG_GROUPS_ROLLOUT_TEAM_IDS' | 'PERSONHOG_PERSONS_ROLLOUT_PERCENTAGE' diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts index 24debc752528..e611ae5f6e3a 100644 --- a/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts @@ -1,5 +1,4 @@ import { personhogStoreShadowErrorsCounter, personhogStoreShadowSkipsCounter } from '~/common/persons/metrics' -import { PersonRepository } from '~/common/persons/repositories/person-repository' import { PersonRepositoryTransaction } from '~/common/persons/repositories/person-repository-transaction' import { InternalPerson } from '~/types' @@ -73,17 +72,11 @@ describe('RoutingPersonsStore', () => { // is the constructor's requirement, not an escape from checking. const personhogMock = mockStore() const personhog = personhogMock as unknown as PersonhogPersonsStore - const personRepository: Pick = { - inTransaction: jest.fn((_description, cb) => cb(fakeTx)) as PersonRepository['inTransaction'], - } - return { pg, personhogMock, personhog, personRepository } + return { pg, personhogMock, personhog } } - const makeStore = ( - stores: ReturnType, - mode: 'personhog' | 'shadow', - teams: ReadonlySet | null - ) => new RoutingPersonsStore(stores.pg, stores.personhog, mode, teams, stores.personRepository) + const makeStore = (stores: ReturnType, mode: 'personhog' | 'shadow') => + new RoutingPersonsStore(stores.pg, stores.personhog, mode) it('rejects an unknown mode at parse time', () => { expect(() => parsePersonsStoreMode('both')).toThrow('PERSONS_STORE_MODE') @@ -103,21 +96,19 @@ describe('RoutingPersonsStore', () => { }) describe('personhog mode', () => { - it('routes allowlisted teams to personhog and the rest to pg', async () => { + it('routes every team to personhog', async () => { const stores = makeStores() - const store = makeStore(stores, 'personhog', new Set([1])) + const store = makeStore(stores, 'personhog') await store.fetchForUpdate(1, 'a', 0) - expect(stores.personhogMock.fetchForUpdate).toHaveBeenCalledWith(1, 'a', 0) - expect(stores.pg.fetchForUpdate).not.toHaveBeenCalled() - await store.fetchForUpdate(2, 'b', 0) - expect(stores.pg.fetchForUpdate).toHaveBeenCalledWith(2, 'b', 0) + expect(stores.personhogMock.fetchForUpdate).toHaveBeenCalledTimes(2) + expect(stores.pg.fetchForUpdate).not.toHaveBeenCalled() }) it('a transactional create stays on pg for an allowlisted team', async () => { const stores = makeStores() - const store = makeStore(stores, 'personhog', new Set([1])) + const store = makeStore(stores, 'personhog') await store.createPerson( undefined as never, {}, @@ -139,7 +130,7 @@ describe('RoutingPersonsStore', () => { it('a personhog flush failure propagates, because the store is authoritative', async () => { const stores = makeStores() stores.personhogMock.flush.mockRejectedValue(new Error('leader down')) - const store = makeStore(stores, 'personhog', null) + const store = makeStore(stores, 'personhog') await expect(store.flush()).rejects.toThrow('leader down') }) }) @@ -149,7 +140,7 @@ describe('RoutingPersonsStore', () => { const stores = makeStores() stores.pg.fetchForUpdate.mockResolvedValue(person(1, '7')) stores.personhogMock.fetchForUpdate.mockResolvedValue(person(1, '99')) - const store = makeStore(stores, 'shadow', null) + const store = makeStore(stores, 'shadow') const result = await store.fetchForUpdate(1, 'a', 0) @@ -161,7 +152,7 @@ describe('RoutingPersonsStore', () => { const stores = makeStores() stores.pg.fetchForUpdate.mockResolvedValue(person(1, '7')) stores.personhogMock.fetchForUpdate.mockRejectedValue(new Error('identity down')) - const store = makeStore(stores, 'shadow', null) + const store = makeStore(stores, 'shadow') const result = await store.fetchForUpdate(1, 'a', 0) @@ -172,7 +163,7 @@ describe('RoutingPersonsStore', () => { it('a shadow flush failure is swallowed', async () => { const stores = makeStores() stores.personhogMock.flush.mockRejectedValue(new Error('leader down')) - const store = makeStore(stores, 'shadow', null) + const store = makeStore(stores, 'shadow') await expect(store.flush()).resolves.toEqual([]) }) }) @@ -197,7 +188,7 @@ describe('RoutingPersonsStore', () => { stores.personhogMock.fetchForUpdate.mockResolvedValue(person(1, '99')) stores.personhogMock.applyEventOps.mockResolvedValue([person(1, '99'), []]) stores.personhogMock.updatePersonWithPropertiesDiffForUpdate.mockResolvedValue([person(1, '99'), [], true]) - const store = makeStore(stores, 'shadow', null) + const store = makeStore(stores, 'shadow') await call(store) @@ -209,7 +200,7 @@ describe('RoutingPersonsStore', () => { const stores = makeStores() stores.pg.applyEventOps.mockResolvedValue([person(1, '7'), []]) stores.personhogMock.fetchForUpdate.mockResolvedValue(null) - const store = makeStore(stores, 'shadow', null) + const store = makeStore(stores, 'shadow') const [result] = await store.applyEventOps(person(1, '7'), ops, 'd1', 0) @@ -239,7 +230,7 @@ describe('RoutingPersonsStore', () => { ], ] as const)('%s reaches the personhog store for a routed team', async (_name, call, member) => { const stores = makeStores() - const store = makeStore(stores, 'personhog', new Set([1])) + const store = makeStore(stores, 'personhog') await call(store) @@ -247,53 +238,46 @@ describe('RoutingPersonsStore', () => { expect(stores.pg[member as keyof PersonsStore]).not.toHaveBeenCalled() }) - it.each([ - ['a pg-routed team in personhog mode', 'personhog', new Set([99])], - ['shadow mode', 'shadow', null], - ] as const)('%s runs merges on pg, unshadowed', async (_name, mode, teams) => { + it('shadow mode runs merges on pg and swallows the personhog placeholder', async () => { const stores = makeStores() - const store = makeStore(stores, mode, teams as ReadonlySet | null) + stores.personhogMock.deletePersons.mockRejectedValue(new Error('no personhog RPC: merge saga')) + const store = makeStore(stores, 'shadow') await store.deletePersons([person(1)], 'd') expect(stores.pg.deletePersons).toHaveBeenCalled() - expect(stores.personhogMock.deletePersons).not.toHaveBeenCalled() + expect(personhogStoreShadowErrorsCounter.labels).toHaveBeenCalledWith({ verb: 'deletePersons' }) }) }) - describe('inTransaction wraps this store, so transactional verbs still route', () => { - it('a routed team merge inside a transaction reaches the personhog store', async () => { + describe('inTransaction routes by mode', () => { + it('personhog mode reaches the personhog store, whose placeholder answers', () => { const stores = makeStores() - const store = makeStore(stores, 'personhog', new Set([1])) + const store = makeStore(stores, 'personhog') + const cb = () => Promise.resolve('x') - await store.inTransaction('merge', async (tx) => { - await tx.deletePerson(person(1), 'd') - }) + void store.inTransaction('merge', cb) - // The wrapper re-entered routing: the personhog store saw the - // delete, and pg never received a mutation keyed on a - // personhog-world row id. - expect(stores.personhogMock.deletePerson).toHaveBeenCalledWith(person(1), 'd', fakeTx) - expect(stores.pg.deletePerson).not.toHaveBeenCalled() + expect(stores.personhogMock.inTransaction).toHaveBeenCalledWith('merge', cb) + expect(stores.pg.inTransaction).not.toHaveBeenCalled() }) - it('an unrouted team merge inside a transaction reaches pg with the transaction', async () => { + it('shadow mode runs the transaction on pg exactly once, unshadowed', () => { const stores = makeStores() - const store = makeStore(stores, 'personhog', new Set([99])) + const store = makeStore(stores, 'shadow') + const cb = () => Promise.resolve('x') - await store.inTransaction('merge', async (tx) => { - await tx.deletePerson(person(1), 'd') - }) + void store.inTransaction('merge', cb) - expect(stores.pg.deletePerson).toHaveBeenCalledWith(person(1), 'd', fakeTx) - expect(stores.personhogMock.deletePerson).not.toHaveBeenCalled() + expect(stores.pg.inTransaction).toHaveBeenCalledWith('merge', cb) + expect(stores.personhogMock.inTransaction).not.toHaveBeenCalled() }) }) it('shutdown closes the personhog side even when pg shutdown fails', async () => { const stores = makeStores() stores.pg.shutdown.mockRejectedValue(new Error('pg teardown failed')) - const store = makeStore(stores, 'shadow', null) + const store = makeStore(stores, 'shadow') await expect(store.shutdown()).rejects.toThrow('pg teardown failed') expect(stores.personhogMock.shutdown).toHaveBeenCalled() @@ -301,7 +285,7 @@ describe('RoutingPersonsStore', () => { it('releaseBatch releases both worlds', () => { const stores = makeStores() - const store = makeStore(stores, 'shadow', null) + const store = makeStore(stores, 'shadow') store.releaseBatch(4) expect(stores.pg.releaseBatch).toHaveBeenCalledWith(4) expect(stores.personhogMock.releaseBatch).toHaveBeenCalledWith(4) diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.ts index c51a20bbf199..bad8e50906c4 100644 --- a/nodejs/src/ingestion/common/persons/routing-persons-store.ts +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.ts @@ -3,7 +3,6 @@ import { DateTime } from 'luxon' import { personhogStoreShadowErrorsCounter, personhogStoreShadowSkipsCounter } from '~/common/persons/metrics' import { PersonMessage } from '~/common/persons/person-message' import { InternalPersonWithDistinctId, LifecycleMarkPerson } from '~/common/persons/repositories/person-repository' -import { PersonRepository } from '~/common/persons/repositories/person-repository' import { PersonRepositoryTransaction } from '~/common/persons/repositories/person-repository-transaction' import { CreatePersonResult, MoveDistinctIdsResult } from '~/common/utils/db/db' import { logger } from '~/common/utils/logger' @@ -73,15 +72,9 @@ export class RoutingPersonsStore implements PersonsStore { constructor( private pg: PersonsStore, private personhog: PersonhogPersonsStore, - private mode: 'personhog' | 'shadow', - private teams: ReadonlySet | null, - private personRepository: Pick + private mode: 'personhog' | 'shadow' ) {} - private modeFor(teamId: number): PersonsStoreMode { - return this.teams === null || this.teams.has(teamId) ? this.mode : 'pg' - } - /** * Run the personhog side of a shadowed verb: sequential, awaited, and * never allowed to fail the batch. @@ -110,7 +103,7 @@ export class RoutingPersonsStore implements PersonsStore { personhog: () => Promise, opts?: { tx?: unknown; shadow?: () => Promise } ): Promise { - const mode = opts?.tx ? 'pg' : this.modeFor(teamId) + const mode = opts?.tx ? 'pg' : this.mode if (mode === 'personhog') { return personhog() } @@ -149,17 +142,16 @@ export class RoutingPersonsStore implements PersonsStore { } /** - * The transaction wrapper is built around this store, not the - * Postgres one, so every verb inside a merge transaction re-enters - * routing with the transaction attached. Delegating to the Postgres - * store here would hand the callback a wrapper around that store, - * and nothing inside the transaction would route again — which is - * how personhog-world person ids could reach Postgres row keys. + * Routes by mode like every other verb; the callback runs exactly + * once, so shadow mode delegates to Postgres alone rather than + * executing it a second time against the personhog store. In + * personhog mode the store's own placeholder answers until the + * merge saga lands. */ inTransaction(description: string, transaction: (tx: PersonsStoreTransaction) => Promise): Promise { - return this.personRepository.inTransaction(description, (tx: PersonRepositoryTransaction) => - transaction(new PersonsStoreTransaction(this, tx)) - ) + return this.mode === 'personhog' + ? this.personhog.inTransaction(description, transaction) + : this.pg.inTransaction(description, transaction) } fetchForChecking(teamId: number, distinctId: string, batchId: number): Promise { @@ -318,26 +310,17 @@ export class RoutingPersonsStore implements PersonsStore { ) } - /** - * Merge execution routes to the team's world and nowhere else: pg - * and shadow teams run it on Postgres, and a personhog-routed team - * reaches the personhog store, whose members are the merge saga's - * loud placeholders until it lands. Shadowing is deliberately absent - * — a merge mirrored into the other world would key on foreign row - * ids. Transaction context is no exemption: the transaction wrapper - * re-enters this store, and a routed team's merge must fail before - * any Postgres mutation, not run inside one. - */ - private mergeWorld(teamId: number): PersonsStore { - return this.modeFor(teamId) === 'personhog' ? this.personhog : this.pg - } - deletePerson( person: InternalPerson, distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.mergeWorld(person.team_id).deletePerson(person, distinctId, tx) + return this.route( + 'deletePerson', + person.team_id, + () => this.pg.deletePerson(person, distinctId, tx), + () => this.personhog.deletePerson(person, distinctId, tx) + ) } claimLifecycleMarks( @@ -347,7 +330,12 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.mergeWorld(teamId).claimLifecycleMarks(opId, teamId, persons, distinctId, tx) + return this.route( + 'claimLifecycleMarks', + teamId, + () => this.pg.claimLifecycleMarks(opId, teamId, persons, distinctId, tx), + () => this.personhog.claimLifecycleMarks(opId, teamId, persons, distinctId, tx) + ) } releaseLifecycleMarks( @@ -356,11 +344,21 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.mergeWorld(teamId).releaseLifecycleMarks(opId, teamId, distinctId, tx) + return this.route( + 'releaseLifecycleMarks', + teamId, + () => this.pg.releaseLifecycleMarks(opId, teamId, distinctId, tx), + () => this.personhog.releaseLifecycleMarks(opId, teamId, distinctId, tx) + ) } isPersonLive(person: InternalPerson, distinctId: string, tx?: PersonRepositoryTransaction): Promise { - return this.mergeWorld(person.team_id).isPersonLive(person, distinctId, tx) + return this.route( + 'isPersonLive', + person.team_id, + () => this.pg.isPersonLive(person, distinctId, tx), + () => this.personhog.isPersonLive(person, distinctId, tx) + ) } updatePersonForMerge( @@ -370,7 +368,12 @@ export class RoutingPersonsStore implements PersonsStore { batchId: number, tx?: PersonRepositoryTransaction ): Promise<[InternalPerson, PersonMessage[], boolean]> { - return this.mergeWorld(person.team_id).updatePersonForMerge(person, update, distinctId, batchId, tx) + return this.route( + 'updatePersonForMerge', + person.team_id, + () => this.pg.updatePersonForMerge(person, update, distinctId, batchId, tx), + () => this.personhog.updatePersonForMerge(person, update, distinctId, batchId, tx) + ) } addDistinctId( @@ -380,7 +383,12 @@ export class RoutingPersonsStore implements PersonsStore { tx: PersonRepositoryTransaction | undefined, batchId: number ): Promise { - return this.mergeWorld(person.team_id).addDistinctId(person, distinctId, version, tx, batchId) + return this.route( + 'addDistinctId', + person.team_id, + () => this.pg.addDistinctId(person, distinctId, version, tx, batchId), + () => this.personhog.addDistinctId(person, distinctId, version, tx, batchId) + ) } moveDistinctIds( @@ -391,7 +399,12 @@ export class RoutingPersonsStore implements PersonsStore { tx: PersonRepositoryTransaction, batchId: number ): Promise { - return this.mergeWorld(source.team_id).moveDistinctIds(source, target, distinctId, limit, tx, batchId) + return this.route( + 'moveDistinctIds', + source.team_id, + () => this.pg.moveDistinctIds(source, target, distinctId, limit, tx, batchId), + () => this.personhog.moveDistinctIds(source, target, distinctId, limit, tx, batchId) + ) } moveDistinctIdsFromPersons( @@ -401,7 +414,12 @@ export class RoutingPersonsStore implements PersonsStore { tx: PersonRepositoryTransaction, batchId: number ): Promise { - return this.mergeWorld(target.team_id).moveDistinctIdsFromPersons(sources, target, distinctId, tx, batchId) + return this.route( + 'moveDistinctIdsFromPersons', + target.team_id, + () => this.pg.moveDistinctIdsFromPersons(sources, target, distinctId, tx, batchId), + () => this.personhog.moveDistinctIdsFromPersons(sources, target, distinctId, tx, batchId) + ) } deletePersons( @@ -412,7 +430,12 @@ export class RoutingPersonsStore implements PersonsStore { if (persons.length === 0) { return this.pg.deletePersons(persons, distinctId, tx) } - return this.mergeWorld(persons[0].team_id).deletePersons(persons, distinctId, tx) + return this.route( + 'deletePersons', + persons[0].team_id, + () => this.pg.deletePersons(persons, distinctId, tx), + () => this.personhog.deletePersons(persons, distinctId, tx) + ) } countDistinctIdsForPersons( @@ -421,7 +444,12 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx: PersonRepositoryTransaction ): Promise> { - return this.mergeWorld(teamId).countDistinctIdsForPersons(teamId, personIds, distinctId, tx) + return this.route( + 'countDistinctIdsForPersons', + teamId, + () => this.pg.countDistinctIdsForPersons(teamId, personIds, distinctId, tx), + () => this.personhog.countDistinctIdsForPersons(teamId, personIds, distinctId, tx) + ) } updateCohortsAndFeatureFlagsForMerge( @@ -431,12 +459,18 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.mergeWorld(teamID).updateCohortsAndFeatureFlagsForMerge( + return this.route( + 'updateCohortsAndFeatureFlagsForMerge', teamID, - sourcePersonID, - targetPersonID, - distinctId, - tx + () => this.pg.updateCohortsAndFeatureFlagsForMerge(teamID, sourcePersonID, targetPersonID, distinctId, tx), + () => + this.personhog.updateCohortsAndFeatureFlagsForMerge( + teamID, + sourcePersonID, + targetPersonID, + distinctId, + tx + ) ) } @@ -447,12 +481,25 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - return this.mergeWorld(teamID).updateCohortsAndFeatureFlagsForMergeBatch( + return this.route( + 'updateCohortsAndFeatureFlagsForMergeBatch', teamID, - sourcePersonIDs, - targetPersonID, - distinctId, - tx + () => + this.pg.updateCohortsAndFeatureFlagsForMergeBatch( + teamID, + sourcePersonIDs, + targetPersonID, + distinctId, + tx + ), + () => + this.personhog.updateCohortsAndFeatureFlagsForMergeBatch( + teamID, + sourcePersonIDs, + targetPersonID, + distinctId, + tx + ) ) } @@ -462,11 +509,16 @@ export class RoutingPersonsStore implements PersonsStore { limit: number | undefined, tx: PersonRepositoryTransaction ): Promise { - return this.mergeWorld(person.team_id).fetchPersonDistinctIds(person, distinctId, limit, tx) + return this.route( + 'fetchPersonDistinctIds', + person.team_id, + () => this.pg.fetchPersonDistinctIds(person, distinctId, limit, tx), + () => this.personhog.fetchPersonDistinctIds(person, distinctId, limit, tx) + ) } personPropertiesSize(personId: string, teamId: number): Promise { - return this.modeFor(teamId) === 'personhog' + return this.mode === 'personhog' ? this.personhog.personPropertiesSize(personId, teamId) : this.pg.personPropertiesSize(personId, teamId) } @@ -477,14 +529,10 @@ export class RoutingPersonsStore implements PersonsStore { } async prefetchPersons(teamDistinctIds: { teamId: number; distinctId: string; batchId: number }[]): Promise { - const pgEntries = teamDistinctIds.filter((entry) => this.modeFor(entry.teamId) !== 'personhog') - const personhogEntries = teamDistinctIds.filter((entry) => this.modeFor(entry.teamId) !== 'pg') - if (pgEntries.length > 0) { - await this.pg.prefetchPersons(pgEntries) - } - if (personhogEntries.length > 0) { - await this.shadowedOrDirect('prefetchPersons', () => this.personhog.prefetchPersons(personhogEntries)) + if (this.mode === 'shadow') { + await this.pg.prefetchPersons(teamDistinctIds) } + await this.shadowedOrDirect('prefetchPersons', () => this.personhog.prefetchPersons(teamDistinctIds)) } /** diff --git a/nodejs/src/servers/ingestion-api-server.ts b/nodejs/src/servers/ingestion-api-server.ts index 415d9a03e30f..f849b33d6d82 100644 --- a/nodejs/src/servers/ingestion-api-server.ts +++ b/nodejs/src/servers/ingestion-api-server.ts @@ -14,7 +14,7 @@ import { ClickhouseGroupRepository } from '~/common/groups/repositories/clickhou import { PostgresGroupRepository } from '~/common/groups/repositories/postgres-group-repository' import { KafkaProducerRegistry } from '~/common/outputs/kafka-producer-registry' import { PersonHogConfig, buildGroupRepository, buildPersonRepository, createPersonHogClient } from '~/common/personhog' -import { PersonHogClient, parseRolloutTeamIds } from '~/common/personhog/client' +import { PersonHogClient } from '~/common/personhog/client' import { createIdentityClients } from '~/common/personhog/identity-clients' import { PersonHogPersonWriteRepository } from '~/common/personhog/personhog-person-write-repository' import { PostgresPersonRepository } from '~/common/persons/repositories/postgres-person-repository' @@ -408,14 +408,7 @@ export class IngestionApiServer implements NodeServer { maxConcurrentUpdates: this.config.PERSONHOG_STORE_MAX_CONCURRENT_UPDATES, updateAllProperties: this.config.PERSON_PROPERTIES_UPDATE_ALL, }) - const teams = parseRolloutTeamIds(this.config.PERSONS_STORE_MODE_TEAMS) - personsStore = new RoutingPersonsStore( - this.personsStore, - this.personhogStore, - personsStoreMode, - teams.size > 0 ? teams : null, - personRepository - ) + personsStore = new RoutingPersonsStore(this.personsStore, this.personhogStore, personsStoreMode) } this.groupStore = new BatchWritingGroupStore(groupRepository, clickhouseGroupRepository, { From e9a6be861daec7dbfb33f236d21f93dfc1e176a4 Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Tue, 11 Aug 2026 23:21:44 -0400 Subject: [PATCH 4/6] chore(personhog): describe the mode-wide routing as it is The routing comments still described the removed per-team allowlist, the merge special-casing, and signature drift the interface work eliminated. --- .../persons/routing-persons-store.test.ts | 2 +- .../common/persons/routing-persons-store.ts | 39 +++++++------------ nodejs/src/servers/ingestion-api-server.ts | 6 +-- 3 files changed, 19 insertions(+), 28 deletions(-) diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts index e611ae5f6e3a..26bd3175991a 100644 --- a/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts @@ -106,7 +106,7 @@ describe('RoutingPersonsStore', () => { expect(stores.pg.fetchForUpdate).not.toHaveBeenCalled() }) - it('a transactional create stays on pg for an allowlisted team', async () => { + it('a transactional create stays on pg in personhog mode', async () => { const stores = makeStores() const store = makeStore(stores, 'personhog') await store.createPerson( diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.ts index bad8e50906c4..72c4444ebeda 100644 --- a/nodejs/src/ingestion/common/persons/routing-persons-store.ts +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.ts @@ -48,25 +48,17 @@ export function assertPersonsStoreModeConfig( /** * Routes person-store verbs between the Postgres world and the personhog - * world by team. The mode applies to the teams in `teams` (every team - * when null); everything else stays on Postgres. In shadow the Postgres - * result is authoritative and the personhog side runs the same verb - * afterwards, its failures counted and logged but never failing the - * batch. + * world. The mode applies to the whole deployment: personhog sends every + * verb to the personhog store, and shadow runs the Postgres call as the + * authoritative result with the personhog call after it, its failures + * counted and logged but never failing the batch. Verbs the personhog + * store cannot serve yet answer with its pending-saga placeholders, so + * unsupported paths fail loudly in personhog mode and surface as counted + * shadow errors in shadow mode. * - * Two verb families do not route: - * - * - Merge execution has no personhog support until the merge saga - * lands: pg and shadow teams run it on Postgres as before, and a - * personhog-routed team fails loudly at the first merge mutation — - * its reads and writes live in the personhog world, whose row ids - * mean nothing to Postgres, so quietly running the merge there would - * mutate whatever rows happen to share the numbers. Any non-merge - * verb invoked under a Postgres transaction still goes to Postgres - * (see `route`). - * - `personPropertiesSize` routes by mode but is not shadowed: the - * personhog store answers it with a constant because the leader - * enforces the size ceiling at admission. + * `personPropertiesSize` routes by mode but is not shadowed: the + * personhog store answers it with a constant because the leader enforces + * the size ceiling at admission. */ export class RoutingPersonsStore implements PersonsStore { constructor( @@ -89,12 +81,11 @@ export class RoutingPersonsStore implements PersonsStore { } /** - * The whole mode semantics, once: pg teams run the pg call, personhog - * teams the personhog call, shadow teams run pg as the authoritative - * result and the personhog call after it, swallowed. A verb invoked - * under a Postgres transaction is the merge flow's and stays on pg - * regardless of team. The two lambdas are each verb's signature - * adapter — the stores disagree on tx and batchId parameters. + * The whole mode semantics, once: personhog mode runs the personhog + * call, shadow runs pg as the authoritative result and the personhog + * call after it, swallowed. A verb invoked under a live Postgres + * transaction stays on pg — its work is already inside that + * transaction's world. */ private async route( verb: string, diff --git a/nodejs/src/servers/ingestion-api-server.ts b/nodejs/src/servers/ingestion-api-server.ts index f849b33d6d82..a3f2eb3ce631 100644 --- a/nodejs/src/servers/ingestion-api-server.ts +++ b/nodejs/src/servers/ingestion-api-server.ts @@ -374,9 +374,9 @@ export class IngestionApiServer implements NodeServer { optimisticUpdateRetryInterval: this.config.PERSON_BATCH_WRITING_OPTIMISTIC_UPDATE_RETRY_INTERVAL_MS, updateAllProperties: this.config.PERSON_PROPERTIES_UPDATE_ALL, }) - // Which world person writes land in: pg (default) builds nothing - // new; the other modes construct the personhog store and route - // per team, shadow keeping pg authoritative. + // Which world person writes land in, deployment-wide: pg (the + // default) builds nothing new; the other modes construct the + // personhog store, shadow keeping pg authoritative. const personsStoreMode = parsePersonsStoreMode(this.config.PERSONS_STORE_MODE) assertPersonsStoreModeConfig(personsStoreMode, { routerAddr: this.config.PERSONHOG_ADDR, From b22853b7d5e5ce42b9cf90ec644718da2dad40f2 Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Tue, 11 Aug 2026 23:24:22 -0400 Subject: [PATCH 5/6] chore(personhog): route the last special-cased verbs fetchPersonsForUpdateByDistinctIds still pinned to Postgres with a stale merge comment, deletePersons kept an empty-array short-circuit that existed only for the removed per-team guard, and flush stats reported the Postgres world alone. All three now follow the uniform routing, and route() drops the team parameter the allowlist removal left dead. --- .../common/persons/routing-persons-store.ts | 37 ++++++------------- 1 file changed, 12 insertions(+), 25 deletions(-) diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.ts index 72c4444ebeda..dc601df0020a 100644 --- a/nodejs/src/ingestion/common/persons/routing-persons-store.ts +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.ts @@ -89,7 +89,6 @@ export class RoutingPersonsStore implements PersonsStore { */ private async route( verb: string, - teamId: number, pg: () => Promise, personhog: () => Promise, opts?: { tx?: unknown; shadow?: () => Promise } @@ -148,7 +147,6 @@ export class RoutingPersonsStore implements PersonsStore { fetchForChecking(teamId: number, distinctId: string, batchId: number): Promise { return this.route( 'fetchForChecking', - teamId, () => this.pg.fetchForChecking(teamId, distinctId, batchId), () => this.personhog.fetchForChecking(teamId, distinctId, batchId) ) @@ -157,7 +155,6 @@ export class RoutingPersonsStore implements PersonsStore { fetchForUpdate(teamId: number, distinctId: string, batchId: number): Promise { return this.route( 'fetchForUpdate', - teamId, () => this.pg.fetchForUpdate(teamId, distinctId, batchId), () => this.personhog.fetchForUpdate(teamId, distinctId, batchId) ) @@ -168,8 +165,11 @@ export class RoutingPersonsStore implements PersonsStore { distinctIds: string[], batchId: number ): Promise { - // Merge-fold pre-lock: merge flows stay whole on Postgres. - return this.pg.fetchPersonsForUpdateByDistinctIds(teamId, distinctIds, batchId) + return this.route( + 'fetchPersonsForUpdateByDistinctIds', + () => this.pg.fetchPersonsForUpdateByDistinctIds(teamId, distinctIds, batchId), + () => this.personhog.fetchPersonsForUpdateByDistinctIds(teamId, distinctIds, batchId) + ) } createPerson( @@ -188,7 +188,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise { return this.route( 'createPerson', - teamId, () => this.pg.createPerson( createdAt, @@ -231,7 +230,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise<[InternalPerson, PersonMessage[]]> { return this.route( 'applyEventOps', - person.team_id, () => this.pg.applyEventOps(person, ops, distinctId, batchId), () => this.personhog.applyEventOps(person, ops, distinctId, batchId), { @@ -255,7 +253,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise<[InternalPerson, PersonMessage[], boolean]> { return this.route( 'updatePersonWithPropertiesDiffForUpdate', - person.team_id, () => this.pg.updatePersonWithPropertiesDiffForUpdate( person, @@ -308,7 +305,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise { return this.route( 'deletePerson', - person.team_id, () => this.pg.deletePerson(person, distinctId, tx), () => this.personhog.deletePerson(person, distinctId, tx) ) @@ -323,7 +319,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise { return this.route( 'claimLifecycleMarks', - teamId, () => this.pg.claimLifecycleMarks(opId, teamId, persons, distinctId, tx), () => this.personhog.claimLifecycleMarks(opId, teamId, persons, distinctId, tx) ) @@ -337,7 +332,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise { return this.route( 'releaseLifecycleMarks', - teamId, () => this.pg.releaseLifecycleMarks(opId, teamId, distinctId, tx), () => this.personhog.releaseLifecycleMarks(opId, teamId, distinctId, tx) ) @@ -346,7 +340,6 @@ export class RoutingPersonsStore implements PersonsStore { isPersonLive(person: InternalPerson, distinctId: string, tx?: PersonRepositoryTransaction): Promise { return this.route( 'isPersonLive', - person.team_id, () => this.pg.isPersonLive(person, distinctId, tx), () => this.personhog.isPersonLive(person, distinctId, tx) ) @@ -361,7 +354,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise<[InternalPerson, PersonMessage[], boolean]> { return this.route( 'updatePersonForMerge', - person.team_id, () => this.pg.updatePersonForMerge(person, update, distinctId, batchId, tx), () => this.personhog.updatePersonForMerge(person, update, distinctId, batchId, tx) ) @@ -376,7 +368,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise { return this.route( 'addDistinctId', - person.team_id, () => this.pg.addDistinctId(person, distinctId, version, tx, batchId), () => this.personhog.addDistinctId(person, distinctId, version, tx, batchId) ) @@ -392,7 +383,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise { return this.route( 'moveDistinctIds', - source.team_id, () => this.pg.moveDistinctIds(source, target, distinctId, limit, tx, batchId), () => this.personhog.moveDistinctIds(source, target, distinctId, limit, tx, batchId) ) @@ -407,7 +397,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise { return this.route( 'moveDistinctIdsFromPersons', - target.team_id, () => this.pg.moveDistinctIdsFromPersons(sources, target, distinctId, tx, batchId), () => this.personhog.moveDistinctIdsFromPersons(sources, target, distinctId, tx, batchId) ) @@ -418,12 +407,8 @@ export class RoutingPersonsStore implements PersonsStore { distinctId: string, tx?: PersonRepositoryTransaction ): Promise { - if (persons.length === 0) { - return this.pg.deletePersons(persons, distinctId, tx) - } return this.route( 'deletePersons', - persons[0].team_id, () => this.pg.deletePersons(persons, distinctId, tx), () => this.personhog.deletePersons(persons, distinctId, tx) ) @@ -437,7 +422,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise> { return this.route( 'countDistinctIdsForPersons', - teamId, () => this.pg.countDistinctIdsForPersons(teamId, personIds, distinctId, tx), () => this.personhog.countDistinctIdsForPersons(teamId, personIds, distinctId, tx) ) @@ -452,7 +436,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise { return this.route( 'updateCohortsAndFeatureFlagsForMerge', - teamID, () => this.pg.updateCohortsAndFeatureFlagsForMerge(teamID, sourcePersonID, targetPersonID, distinctId, tx), () => this.personhog.updateCohortsAndFeatureFlagsForMerge( @@ -474,7 +457,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise { return this.route( 'updateCohortsAndFeatureFlagsForMergeBatch', - teamID, () => this.pg.updateCohortsAndFeatureFlagsForMergeBatch( teamID, @@ -502,7 +484,6 @@ export class RoutingPersonsStore implements PersonsStore { ): Promise { return this.route( 'fetchPersonDistinctIds', - person.team_id, () => this.pg.fetchPersonDistinctIds(person, distinctId, limit, tx), () => this.personhog.fetchPersonDistinctIds(person, distinctId, limit, tx) ) @@ -540,7 +521,13 @@ export class RoutingPersonsStore implements PersonsStore { } getFlushStats(): BatchWritingStoreFlushStats { - return this.pg.getFlushStats() + const pg = this.pg.getFlushStats() + const personhog = this.personhog.getFlushStats() + return { + dirtyEntryCount: pg.dirtyEntryCount + personhog.dirtyEntryCount, + referencedBatchCount: pg.referencedBatchCount + personhog.referencedBatchCount, + cacheEntryCount: pg.cacheEntryCount + personhog.cacheEntryCount, + } } async flush(): Promise { From d925ad6d91db5771cbce86a0f288bae3d2af536f Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Wed, 12 Aug 2026 00:13:57 -0400 Subject: [PATCH 6/6] fix(personhog): make store-wide flush safe under concurrent batches flush() serializes passes and claims a snapshot of every lane up front, so ops folded mid-pass ship on the next pass instead of mutating an entry already in flight. A failed ship restores its entry, so a sibling batch's unacked ops survive another batch's flush failure. A no-change lane ships instead of being suppressed when a sibling batch holds ops for the same person, whose baseline the verdict never saw. Also: pending-saga placeholders reject instead of throwing synchronously, the routing store drops its dead transaction rule and routes personPropertiesSize/prefetchPersons/flush uniformly, and getFlushStats no longer double-counts batches in shadow mode. --- nodejs/src/common/config.ts | 7 +- .../persons/personhog-persons-store.test.ts | 85 ++++++- .../common/persons/personhog-persons-store.ts | 219 +++++++++++------- .../persons/routing-persons-store.test.ts | 52 +++-- .../common/persons/routing-persons-store.ts | 71 +++--- 5 files changed, 280 insertions(+), 154 deletions(-) diff --git a/nodejs/src/common/config.ts b/nodejs/src/common/config.ts index 1bf722e08ea2..31a774e803d5 100644 --- a/nodejs/src/common/config.ts +++ b/nodejs/src/common/config.ts @@ -112,7 +112,12 @@ export type CommonConfig = BaseServerConfig & { // PersonHog gRPC PERSONHOG_ENABLED: boolean - /** Which world the ingestion persons store writes: 'pg' (default), 'personhog', or 'shadow' (pg authoritative, personhog best-effort). */ + /** + * Which world the ingestion persons store writes: 'pg' (default), + * 'personhog', or 'shadow' (pg authoritative, personhog best-effort). + * Until the merge saga lands, merge events fail loudly in personhog + * mode, so it is only safe for traffic that produces none. + */ PERSONS_STORE_MODE: string /** Host and port of the personhog identity server. */ PERSONHOG_IDENTITY_ADDR: string diff --git a/nodejs/src/ingestion/common/persons/personhog-persons-store.test.ts b/nodejs/src/ingestion/common/persons/personhog-persons-store.test.ts index ebb19efcae06..a39ee01ae907 100644 --- a/nodejs/src/ingestion/common/persons/personhog-persons-store.test.ts +++ b/nodejs/src/ingestion/common/persons/personhog-persons-store.test.ts @@ -226,9 +226,9 @@ describe('PersonhogPersonsStore', () => { expect(fetched?.id).toBe('7') }) - it('deletes have no personhog path and fail loudly', () => { + it('deletes have no personhog path and fail loudly', async () => { const bound = store.forBatch(0) - expect(() => bound.deletePerson(person, 'd1')).toThrow('no personhog RPC') + await expect(bound.deletePerson(person, 'd1')).rejects.toThrow('no personhog RPC') }) it('creation resolves through identity and memoizes every distinct id it mapped', async () => { @@ -280,6 +280,79 @@ describe('PersonhogPersonsStore', () => { await expect(bound.flush()).rejects.toThrow('leader unreachable') }) + describe('concurrent batches share one flush', () => { + const personB: () => InternalPerson = () => ({ ...person, id: '8', uuid: 'person-uuid-8' }) + + it("a sibling entry's failure leaves it in its lane for the next pass", async () => { + const bound0 = store.forBatch(0) + const bound1 = store.forBatch(1) + await bound0.applyEventOps(person, ops({ $set: { a: '1' } }), 'd1') + await bound1.applyEventOps(personB(), ops({ $set: { b: '2' } }), 'd2') + + repository.updatePersonProperties.mockImplementation((request) => + request.personId === '8' + ? Promise.reject(new Error('leader unreachable')) + : Promise.resolve({ person: { ...person, version: 2 }, updated: true }) + ) + await expect(store.flush()).rejects.toThrow('leader unreachable') + + repository.updatePersonProperties.mockClear() + repository.updatePersonProperties.mockResolvedValue({ person: { ...person, version: 2 }, updated: true }) + await store.flush() + // Only the failed entry survives to re-ship; the succeeded + // one was consumed by the first pass. + expect(repository.updatePersonProperties).toHaveBeenCalledTimes(1) + expect(repository.updatePersonProperties.mock.calls[0][0].personId).toBe('8') + }) + + it('flush passes serialize, so ops folded mid-pass ship strictly after it', async () => { + let releaseFirst!: () => void + const firstGate = new Promise((resolve) => { + releaseFirst = resolve + }) + const callOrder: string[] = [] + repository.updatePersonProperties + .mockImplementationOnce(async () => { + callOrder.push('first:start') + await firstGate + callOrder.push('first:end') + return { person: { ...person, version: 2 }, updated: true } + }) + .mockImplementation(() => { + callOrder.push('second') + return Promise.resolve({ person: { ...person, version: 3 }, updated: true }) + }) + + const bound = store.forBatch(0) + await bound.applyEventOps(person, ops({ $set: { a: '1' } }), 'd1') + const firstFlush = store.flush() + // Spin microtasks until the first pass is mid-ship, blocked + // on the gate, so the next fold genuinely lands mid-pass. + for (let i = 0; i < 100 && !callOrder.includes('first:start'); i++) { + await Promise.resolve() + } + await bound.applyEventOps(person, ops({ $set: { a: '2' } }), 'd1') + const secondFlush = store.flush() + releaseFirst() + await Promise.all([firstFlush, secondFlush]) + + expect(callOrder).toEqual(['first:start', 'first:end', 'second']) + }) + + it('a no-change lane ships anyway when a sibling batch holds ops for the person', async () => { + person.properties = { $browser: 'Firefox' } + const bound0 = store.forBatch(0) + const bound1 = store.forBatch(1) + // Batch 0 folds a filtered-only no-change; batch 1 holds real + // ops for the same person, so batch 0's verdict may be stale + // and the leader judges it instead. + await bound0.applyEventOps(person, ops({ $set: { $browser: 'Chrome' } }, 'pageview'), 'd1') + await bound1.applyEventOps(person, ops({ $set: { plan: 'pro' } }, 'pageview'), 'd1') + await store.flush() + expect(repository.updatePersonProperties).toHaveBeenCalledTimes(2) + }) + }) + it('publishes nothing when the leader reports no change', async () => { repository.updatePersonProperties.mockResolvedValue({ person, updated: false }) const bound = store.forBatch(0) @@ -361,9 +434,9 @@ describe('PersonhogPersonsStore', () => { expect(results).toEqual([]) }) - it('has no transactions of its own; the routing store owns them', () => { + it('has no transactions of its own; the routing store owns them', async () => { const bound = store.forBatch(0) - expect(() => bound.inTransaction('test', () => Promise.resolve('done'))).toThrow('no personhog RPC') + await expect(bound.inTransaction('test', () => Promise.resolve('done'))).rejects.toThrow('no personhog RPC') }) it.each([ @@ -386,9 +459,9 @@ describe('PersonhogPersonsStore', () => { (b: ReturnType, p: InternalPerson) => b.moveDistinctIdsFromPersons([p], p, 'd1', undefined as any), ], - ])('%s fails loudly while the leader RPC is pending', (_method, call) => { + ])('%s fails loudly while the leader RPC is pending', async (_method, call) => { const bound = store.forBatch(0) - expect(() => call(bound, person)).toThrow(PersonhogPendingRpcError) + await expect(call(bound, person)).rejects.toThrow(PersonhogPendingRpcError) }) it('maps a direct diff update onto the folded RPC', async () => { diff --git a/nodejs/src/ingestion/common/persons/personhog-persons-store.ts b/nodejs/src/ingestion/common/persons/personhog-persons-store.ts index 54d696d26b3e..d09b3be7a370 100644 --- a/nodejs/src/ingestion/common/persons/personhog-persons-store.ts +++ b/nodejs/src/ingestion/common/persons/personhog-persons-store.ts @@ -123,6 +123,8 @@ export class PersonhogPersonsStore implements PersonsStore { * service state. */ private personState: Map> = new Map() + /** Serializes flush passes; see flush(). */ + private flushChain: Promise = Promise.resolve() constructor( private repository: PersonHogPersonWriteRepository, @@ -357,7 +359,7 @@ export class PersonhogPersonsStore implements PersonsStore { * never delegates this member. Reaching it is a wiring bug. */ inTransaction(_description: string, _transaction: (tx: PersonsStoreTransaction) => Promise): Promise { - throw new PersonhogPendingRpcError('inTransaction', 'merge saga') + return Promise.reject(new PersonhogPendingRpcError('inTransaction', 'merge saga')) } // Merge execution is the merge saga's once it lands; until then every @@ -370,7 +372,7 @@ export class PersonhogPersonsStore implements PersonsStore { _batchId: number, _tx?: PersonRepositoryTransaction ): Promise<[InternalPerson, PersonMessage[], boolean]> { - throw new PersonhogPendingRpcError('updatePersonForMerge', 'merge saga') + return Promise.reject(new PersonhogPendingRpcError('updatePersonForMerge', 'merge saga')) } claimLifecycleMarks( @@ -380,7 +382,7 @@ export class PersonhogPersonsStore implements PersonsStore { _distinctId: string, _tx?: PersonRepositoryTransaction ): Promise { - throw new PersonhogPendingRpcError('claimLifecycleMarks', 'merge saga') + return Promise.reject(new PersonhogPendingRpcError('claimLifecycleMarks', 'merge saga')) } releaseLifecycleMarks( @@ -389,11 +391,11 @@ export class PersonhogPersonsStore implements PersonsStore { _distinctId: string, _tx?: PersonRepositoryTransaction ): Promise { - throw new PersonhogPendingRpcError('releaseLifecycleMarks', 'merge saga') + return Promise.reject(new PersonhogPendingRpcError('releaseLifecycleMarks', 'merge saga')) } isPersonLive(_person: InternalPerson, _distinctId: string, _tx?: PersonRepositoryTransaction): Promise { - throw new PersonhogPendingRpcError('isPersonLive', 'merge saga') + return Promise.reject(new PersonhogPendingRpcError('isPersonLive', 'merge saga')) } addDistinctId( @@ -403,7 +405,7 @@ export class PersonhogPersonsStore implements PersonsStore { _tx: PersonRepositoryTransaction | undefined, _batchId: number ): Promise { - throw new PersonhogPendingRpcError('addDistinctId', 'merge saga') + return Promise.reject(new PersonhogPendingRpcError('addDistinctId', 'merge saga')) } moveDistinctIds( @@ -414,7 +416,7 @@ export class PersonhogPersonsStore implements PersonsStore { _tx: PersonRepositoryTransaction, _batchId: number ): Promise { - throw new PersonhogPendingRpcError('moveDistinctIds', 'merge saga') + return Promise.reject(new PersonhogPendingRpcError('moveDistinctIds', 'merge saga')) } moveDistinctIdsFromPersons( @@ -424,7 +426,7 @@ export class PersonhogPersonsStore implements PersonsStore { _tx: PersonRepositoryTransaction, _batchId: number ): Promise { - throw new PersonhogPendingRpcError('moveDistinctIdsFromPersons', 'merge saga') + return Promise.reject(new PersonhogPendingRpcError('moveDistinctIdsFromPersons', 'merge saga')) } // Postgres bookkeeping with nothing to answer in this world: shadow @@ -473,7 +475,7 @@ export class PersonhogPersonsStore implements PersonsStore { _distinctId: string, _tx?: PersonRepositoryTransaction ): Promise { - throw new PersonhogPendingRpcError('deletePersons', 'merge saga') + return Promise.reject(new PersonhogPendingRpcError('deletePersons', 'merge saga')) } deletePerson( @@ -481,7 +483,7 @@ export class PersonhogPersonsStore implements PersonsStore { _distinctId: string, _tx?: PersonRepositoryTransaction ): Promise { - throw new PersonhogPendingRpcError('deletePerson', 'merge saga') + return Promise.reject(new PersonhogPendingRpcError('deletePerson', 'merge saga')) } /** @@ -610,7 +612,7 @@ export class PersonhogPersonsStore implements PersonsStore { } /** - * Ships the batch's folded lanes to the leader, one entry per + * Ships every batch's folded lanes to the leader, one entry per * person, segments in order. There is deliberately no Postgres * fallback. A missing person (deleted or merged mid-batch) and the * leader's size rejection are counted and skipped, since neither can @@ -619,87 +621,142 @@ export class PersonhogPersonsStore implements PersonsStore { * response's no-change flag: a retried call whose lost first attempt * landed replays into the leader's no-change fast path, and the * version still tells the truth. Any other failure fails the flush - * so the batch retries whole; lanes are batch-scoped, so one batch's - * failure never discards a sibling's shipped results. + * so the batch retries whole. * - * A lane with no update-worthy change — every refined change + * Passes serialize, one at a time, with later calls queueing behind + * the running one. A pass claims a synchronous snapshot of every + * lane before shipping, so ops folded mid-pass land in fresh entries + * and ship on the next pass, never into an entry already in flight. + * A failed ship restores its entry, ahead of anything folded since: + * the failing flush call fails its own batch, but the entry may + * belong to a sibling batch that never acked its events, and folds + * are idempotent, so re-shipping on a later pass is safe. + * + * A lane entry with no update-worthy change — every refined change * filtered, nothing forced, no scalar movement — is suppressed here * rather than shipped, the same no-op classification the Postgres - * store applies at its flush. The leader never sees the noise. + * store applies at its flush. The exception is a person a sibling + * batch also holds ops for: the no-change verdict was judged against + * this batch's own baseline, so the entry ships instead and the + * leader's refinement no-ops it if it truly is one. */ async flush(): Promise { - const results = await Promise.all([...this.lanes.keys()].map((batchId) => this.flushBatch(batchId))) - return results.flat() + const run = this.flushChain.then(() => this.flushPass()) + this.flushChain = run.then( + () => undefined, + () => undefined + ) + return run } - private async flushBatch(batchId: number): Promise { - const lane = this.lanes.get(batchId) - if (!lane) { - return [] + private async flushPass(): Promise { + const captured: { batchId: number; personKey: string; entry: OpsLaneEntry }[] = [] + for (const [batchId, lane] of this.lanes) { + for (const [personKey, entry] of lane) { + captured.push({ batchId, personKey, entry }) + } + } + // One entry per person per lane, so the number of captured + // entries for a person is the number of batches holding it. + const entriesPerPerson = new Map() + for (const { personKey } of captured) { + entriesPerPerson.set(personKey, (entriesPerPerson.get(personKey) ?? 0) + 1) + } + for (const { batchId, personKey } of captured) { + this.lanes.get(batchId)?.delete(personKey) } - this.lanes.delete(batchId) const limit = pLimit(this.options.maxConcurrentUpdates) - const entries = [...lane.values()] - - const results = await Promise.all( - entries.map((entry) => - limit(async (): Promise => { - if (!entry.triggersUpdate) { - personhogStoreFlushCounter.inc({ outcome: 'filtered' }) - return [] - } - let finalPerson: InternalPerson | null = null - try { - for (const ops of entry.segments) { - const { person } = await this.repository.updatePersonProperties( - { - teamId: entry.teamId, - personId: entry.personId, - eventName: ops.eventName, - setProperties: ops.set, - setOnceProperties: ops.setOnce, - unsetProperties: ops.unset, - isIdentified: ops.isIdentified, - lastSeenAtMs: ops.lastSeenAtMs, - }, - CALLER_TAG - ) - finalPerson = person ?? finalPerson - } - personhogStoreFlushCounter.inc({ outcome: 'success' }) - } catch (error) { - if (error instanceof NoRowsUpdatedError) { - // The person was merged or deleted since - // the fold. In-batch that is fine: the - // merge carried the pending projection to - // its target. Merge events are gated off - // this store, so this counter firing means - // a bug, not a cross-batch race. - personhogStoreFlushCounter.inc({ outcome: 'not_found' }) - } else if (error instanceof PersonhogPropertiesSizeError) { - // Counted only: the store holds no outputs - // handle, so the size-violation ingestion - // warning the Postgres store emits has no - // path from here. - personhogStoreFlushCounter.inc({ outcome: 'size_violation' }) - } else { - personhogStoreFlushCounter.inc({ outcome: 'error' }) - logger.error('Failed to flush folded update to personhog', { - teamId: entry.teamId, - personId: entry.personId, - error, - }) - throw error - } - } - // No FlushResult: the leader's changelog is the - // ClickHouse person feed, so a flush publishes - // nothing — shipping the segments is the whole job. - return [] - }) + const outcomes = await Promise.allSettled( + captured.map(({ batchId, personKey, entry }) => + limit(() => this.shipEntry(batchId, personKey, entry, (entriesPerPerson.get(personKey) ?? 0) > 1)) ) ) - return results.flat() + for (const [batchId, lane] of this.lanes) { + if (lane.size === 0) { + this.lanes.delete(batchId) + } + } + const failed = outcomes.find((outcome): outcome is PromiseRejectedResult => outcome.status === 'rejected') + if (failed) { + throw failed.reason + } + // No FlushResults: the leader's changelog is the ClickHouse + // person feed, so a flush publishes nothing — shipping the + // segments is the whole job. + return [] + } + + private async shipEntry( + batchId: number, + personKey: string, + entry: OpsLaneEntry, + siblingPending: boolean + ): Promise { + if (!entry.triggersUpdate && !siblingPending) { + personhogStoreFlushCounter.inc({ outcome: 'filtered' }) + return + } + try { + for (const ops of entry.segments) { + await this.repository.updatePersonProperties( + { + teamId: entry.teamId, + personId: entry.personId, + eventName: ops.eventName, + setProperties: ops.set, + setOnceProperties: ops.setOnce, + unsetProperties: ops.unset, + isIdentified: ops.isIdentified, + lastSeenAtMs: ops.lastSeenAtMs, + }, + CALLER_TAG + ) + } + personhogStoreFlushCounter.inc({ outcome: 'success' }) + } catch (error) { + if (error instanceof NoRowsUpdatedError) { + // The person was merged or deleted since the fold. + // In-batch that is fine: the merge carried the pending + // projection to its target. Merge events are gated off + // this store, so this counter firing means a bug, not a + // cross-batch race. + personhogStoreFlushCounter.inc({ outcome: 'not_found' }) + } else if (error instanceof PersonhogPropertiesSizeError) { + // Counted only: the store holds no outputs handle, so + // the size-violation ingestion warning the Postgres + // store emits has no path from here. + personhogStoreFlushCounter.inc({ outcome: 'size_violation' }) + } else { + personhogStoreFlushCounter.inc({ outcome: 'error' }) + logger.error('Failed to flush folded update to personhog', { + teamId: entry.teamId, + personId: entry.personId, + error, + }) + this.restoreEntry(batchId, personKey, entry) + throw error + } + } + } + + /** + * Puts a failed entry back in its lane, its segments ahead of any + * folded since the pass claimed it, preserving order for the next + * pass. A released batch stays released: its redelivery re-folds + * these ops. + */ + private restoreEntry(batchId: number, personKey: string, entry: OpsLaneEntry): void { + const lane = this.lanes.get(batchId) + if (!lane) { + return + } + const newer = lane.get(personKey) + if (!newer) { + lane.set(personKey, entry) + return + } + newer.segments = [...entry.segments, ...newer.segments] + newer.triggersUpdate = newer.triggersUpdate || entry.triggersUpdate } /** diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts index 26bd3175991a..407b83b8d597 100644 --- a/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.test.ts @@ -1,5 +1,4 @@ import { personhogStoreShadowErrorsCounter, personhogStoreShadowSkipsCounter } from '~/common/persons/metrics' -import { PersonRepositoryTransaction } from '~/common/persons/repositories/person-repository-transaction' import { InternalPerson } from '~/types' import { EventOps } from './person-update' @@ -51,8 +50,6 @@ function mockStore(): jest.Mocked { } describe('RoutingPersonsStore', () => { - const fakeTx = {} as PersonRepositoryTransaction - const person = (teamId: number, id = '1'): InternalPerson => ({ id, team_id: teamId, properties: {}, is_identified: false }) as unknown as InternalPerson @@ -96,7 +93,7 @@ describe('RoutingPersonsStore', () => { }) describe('personhog mode', () => { - it('routes every team to personhog', async () => { + it('routes every verb to personhog, never touching pg', async () => { const stores = makeStores() const store = makeStore(stores, 'personhog') @@ -106,32 +103,19 @@ describe('RoutingPersonsStore', () => { expect(stores.pg.fetchForUpdate).not.toHaveBeenCalled() }) - it('a transactional create stays on pg in personhog mode', async () => { + it('a personhog flush failure propagates, because the store is authoritative', async () => { const stores = makeStores() + stores.personhogMock.flush.mockRejectedValue(new Error('leader down')) const store = makeStore(stores, 'personhog') - await store.createPerson( - undefined as never, - {}, - {}, - {}, - 1, - null, - false, - 'u', - { distinctId: 'd' }, - undefined, - fakeTx, - 0 - ) - expect(stores.pg.createPerson).toHaveBeenCalled() - expect(stores.personhogMock.createPerson).not.toHaveBeenCalled() + await expect(store.flush()).rejects.toThrow('leader down') }) - it('a personhog flush failure propagates, because the store is authoritative', async () => { + it('flush never runs the pg side, and returns the personhog results', async () => { const stores = makeStores() - stores.personhogMock.flush.mockRejectedValue(new Error('leader down')) const store = makeStore(stores, 'personhog') - await expect(store.flush()).rejects.toThrow('leader down') + await expect(store.flush()).resolves.toEqual([]) + expect(stores.personhogMock.flush).toHaveBeenCalled() + expect(stores.pg.flush).not.toHaveBeenCalled() }) }) @@ -166,6 +150,26 @@ describe('RoutingPersonsStore', () => { const store = makeStore(stores, 'shadow') await expect(store.flush()).resolves.toEqual([]) }) + + it('prefetch warms both worlds', async () => { + const stores = makeStores() + const store = makeStore(stores, 'shadow') + await store.prefetchPersons([{ teamId: 1, distinctId: 'd', batchId: 0 }]) + expect(stores.pg.prefetchPersons).toHaveBeenCalled() + expect(stores.personhogMock.prefetchPersons).toHaveBeenCalled() + }) + + it('getFlushStats counts a batch once when both worlds reference it', () => { + const stores = makeStores() + stores.pg.getFlushStats.mockReturnValue({ dirtyEntryCount: 2, referencedBatchCount: 1, cacheEntryCount: 3 }) + stores.personhogMock.getFlushStats.mockReturnValue({ + dirtyEntryCount: 1, + referencedBatchCount: 1, + cacheEntryCount: 2, + }) + const store = makeStore(stores, 'shadow') + expect(store.getFlushStats()).toEqual({ dirtyEntryCount: 3, referencedBatchCount: 1, cacheEntryCount: 5 }) + }) }) describe('shadow writes resolve the personhog world person', () => { diff --git a/nodejs/src/ingestion/common/persons/routing-persons-store.ts b/nodejs/src/ingestion/common/persons/routing-persons-store.ts index dc601df0020a..22bdd39567f7 100644 --- a/nodejs/src/ingestion/common/persons/routing-persons-store.ts +++ b/nodejs/src/ingestion/common/persons/routing-persons-store.ts @@ -55,10 +55,6 @@ export function assertPersonsStoreModeConfig( * store cannot serve yet answer with its pending-saga placeholders, so * unsupported paths fail loudly in personhog mode and surface as counted * shadow errors in shadow mode. - * - * `personPropertiesSize` routes by mode but is not shadowed: the - * personhog store answers it with a constant because the leader enforces - * the size ceiling at admission. */ export class RoutingPersonsStore implements PersonsStore { constructor( @@ -83,24 +79,22 @@ export class RoutingPersonsStore implements PersonsStore { /** * The whole mode semantics, once: personhog mode runs the personhog * call, shadow runs pg as the authoritative result and the personhog - * call after it, swallowed. A verb invoked under a live Postgres - * transaction stays on pg — its work is already inside that - * transaction's world. + * call after it, swallowed. Verbs carrying a live Postgres + * transaction never arrive here, because the transaction wrapper + * pg's inTransaction hands out pins every tx-scoped call to the + * store that opened it. */ private async route( verb: string, pg: () => Promise, personhog: () => Promise, - opts?: { tx?: unknown; shadow?: () => Promise } + opts?: { shadow?: () => Promise } ): Promise { - const mode = opts?.tx ? 'pg' : this.mode - if (mode === 'personhog') { + if (this.mode === 'personhog') { return personhog() } const result = await pg() - if (mode === 'shadow') { - await this.shadowed(verb, opts?.shadow ?? personhog) - } + await this.shadowed(verb, opts?.shadow ?? personhog) return result } @@ -217,8 +211,7 @@ export class RoutingPersonsStore implements PersonsStore { extraDistinctIds, tx, batchId - ), - { tx } + ) ) } @@ -276,7 +269,6 @@ export class RoutingPersonsStore implements PersonsStore { tx ), { - tx, shadow: () => this.withShadowPerson( 'updatePersonWithPropertiesDiffForUpdate', @@ -490,9 +482,11 @@ export class RoutingPersonsStore implements PersonsStore { } personPropertiesSize(personId: string, teamId: number): Promise { - return this.mode === 'personhog' - ? this.personhog.personPropertiesSize(personId, teamId) - : this.pg.personPropertiesSize(personId, teamId) + return this.route( + 'personPropertiesSize', + () => this.pg.personPropertiesSize(personId, teamId), + () => this.personhog.personPropertiesSize(personId, teamId) + ) } removeDistinctIdFromCache(teamId: number, distinctId: string): void { @@ -500,24 +494,12 @@ export class RoutingPersonsStore implements PersonsStore { this.personhog.removeDistinctIdFromCache(teamId, distinctId) } - async prefetchPersons(teamDistinctIds: { teamId: number; distinctId: string; batchId: number }[]): Promise { - if (this.mode === 'shadow') { - await this.pg.prefetchPersons(teamDistinctIds) - } - await this.shadowedOrDirect('prefetchPersons', () => this.personhog.prefetchPersons(teamDistinctIds)) - } - - /** - * The personhog side of a fan-out: swallowed in shadow, propagated in - * personhog mode, where the store is authoritative and redelivery is - * the retry. - */ - private async shadowedOrDirect(verb: string, run: () => Promise): Promise { - if (this.mode === 'shadow') { - await this.shadowed(verb, run) - } else { - await run() - } + prefetchPersons(teamDistinctIds: { teamId: number; distinctId: string; batchId: number }[]): Promise { + return this.route( + 'prefetchPersons', + () => this.pg.prefetchPersons(teamDistinctIds), + () => this.personhog.prefetchPersons(teamDistinctIds) + ) } getFlushStats(): BatchWritingStoreFlushStats { @@ -525,15 +507,20 @@ export class RoutingPersonsStore implements PersonsStore { const personhog = this.personhog.getFlushStats() return { dirtyEntryCount: pg.dirtyEntryCount + personhog.dirtyEntryCount, - referencedBatchCount: pg.referencedBatchCount + personhog.referencedBatchCount, + // Both stores see the same batches in shadow mode, so batch + // references overlap rather than add; entries and cache + // slots are per-store and sum. + referencedBatchCount: Math.max(pg.referencedBatchCount, personhog.referencedBatchCount), cacheEntryCount: pg.cacheEntryCount + personhog.cacheEntryCount, } } - async flush(): Promise { - const results = await this.pg.flush() - await this.shadowedOrDirect('flush', () => this.personhog.flush()) - return results + flush(): Promise { + return this.route( + 'flush', + () => this.pg.flush(), + () => this.personhog.flush() + ) } releaseBatch(batchId: number): void {