feat(personhog): route ingestion person writes by store mode

nodejs/src/common/config.ts

@@ -112,6 +112,15 @@ 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).
+     * 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
     PERSONHOG_ADDR: string
     PERSONHOG_GROUPS_ROLLOUT_PERCENTAGE: number
     PERSONHOG_GROUPS_ROLLOUT_TEAM_IDS: string
@@ -293,6 +302,8 @@ export function getDefaultCommonConfig(): CommonConfig {
         // PersonHog gRPC
         PERSONHOG_ENABLED: false,
         PERSONHOG_ADDR: '',
+        PERSONS_STORE_MODE: 'pg',
+        PERSONHOG_IDENTITY_ADDR: '',
         PERSONHOG_GROUPS_ROLLOUT_PERCENTAGE: 0,
         PERSONHOG_GROUPS_ROLLOUT_TEAM_IDS: '',
         PERSONHOG_PERSONS_ROLLOUT_PERCENTAGE: 0,

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

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(),
+    }
+}

nodejs/src/common/personhog/index.ts

@@ -17,6 +17,8 @@ export type PersonHogConfig = Pick<
     CommonConfig,
     | 'PERSONHOG_ENABLED'
     | 'PERSONHOG_ADDR'
+    | 'PERSONHOG_IDENTITY_ADDR'
+    | 'PERSONS_STORE_MODE'
     | 'PERSONHOG_GROUPS_ROLLOUT_PERCENTAGE'
     | 'PERSONHOG_GROUPS_ROLLOUT_TEAM_IDS'
     | 'PERSONHOG_PERSONS_ROLLOUT_PERCENTAGE'

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',

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<void>((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,13 +434,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', async () => {
         const bound = store.forBatch(0)
-        const result = await bound.inTransaction('test', (tx) => {
-            expect(tx).toBe(bound)
-            return Promise.resolve('done')
-        })
-        expect(result).toBe('done')
+        await expect(bound.inTransaction('test', () => Promise.resolve('done'))).rejects.toThrow('no personhog RPC')
     })
 
     it.each([
@@ -390,9 +459,9 @@ describe('PersonhogPersonsStore', () => {
             (b: ReturnType<PersonhogPersonsStore['forBatch']>, 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 () => {