refactor(cohorts): isolate person_merge_events topic producer

[matheus-vb]

Problem

person_merge_events is a shadow topic consumed only by the cohort stream processor, yet its producer was bundled into the same kafkaAck Promise.all as the critical person messages, so a failed or slow shadow produce could crash or stall analytics ingestion.
Before flipping the producer gate on, producing to it must not be able to affect any service that does not consume it.

Changes

Stacked on #69514.

• producePersonMergeEvent (person-context.ts) is now best effort.
  It wraps message building and the produce in try/catch, never throws, logs and drops on failure, and counts only broker acked produces.

• The produce is detached from the ack chain (person-merge-service.ts).
  kafkaAck now carries only the person messages and the merge event is fire and forget, so a shadow produce failure or slowness can no longer reject or delay offset commits.

• A new effectivePersonMergeEventsEnabled helper (person-merge-event.ts) also requires a configured output topic, so flipping the flag on before wiring the topic env silently does nothing rather than warning on every gated merge.

• Delivery stays at most once. Tightening it is deferred to the actual cut over from the shadow mode.

How did you test this code?

• Added a person-context.test.ts case where a rejecting produce still resolves and does not increment the counter, pinning the never throws contract.
• Added a person-state-batch.test.ts case that hangs the shadow produce and asserts kafkaAck still resolves.
• Because the producer never throws now, only a stuck produce reveals a kafkaAck bundled back together, so a rejecting produce would not catch the regression.

Docs update

No.

[trunk-io]

     

View Full Report ↗︎ ⋅ Docs

[deployment-status-posthog]

Deploy status

Environment	Status	Deployed At	Workflow
dev	✅ Deployed	2026-07-13 19:16 UTC	Run
prod-us	✅ Deployed	2026-07-13 19:28 UTC	Run
prod-eu	✅ Deployed	2026-07-13 19:31 UTC	Run

[haacked]

Nice fix! I was late to this one, but some comments for possible follow-up?

[haacked]

suggestion: No test covers effectivePersonMergeEventsEnabled, the on/off gate for the whole feature. person-merge-event.test.ts only tests buildPersonMergeEventMessage, so a regression here (the feature silently disabling itself, or enabled with an unwired topic that swallows produce failures) ships uncaught.

Add the test cases below. The topic-empty case needs jest.isolateModules (or an exported reset) around warnedUnconfiguredMergeEventsTopic, since that module-level flag never resets and test order would otherwise decide whether the assertion passes.

describe('effectivePersonMergeEventsEnabled', () => {
    it('returns false when the gate is off, regardless of topic', () => {
        expect(effectivePersonMergeEventsEnabled({ PERSON_MERGE_EVENTS_ENABLED: false, INGESTION_OUTPUT_PERSON_MERGE_EVENTS_TOPIC: 'topic' })).toBe(false)
    })
    it('returns true when the gate is on and the topic is configured', () => {
        expect(effectivePersonMergeEventsEnabled({ PERSON_MERGE_EVENTS_ENABLED: true, INGESTION_OUTPUT_PERSON_MERGE_EVENTS_TOPIC: 'topic' })).toBe(true)
    })
    it('returns false when the gate is on but the topic is unset', () => {
        expect(effectivePersonMergeEventsEnabled({ PERSON_MERGE_EVENTS_ENABLED: true, INGESTION_OUTPUT_PERSON_MERGE_EVENTS_TOPIC: '' })).toBe(false)
    })
})

[haacked]

nit: producePersonMergeEvent now keeps the full sourcePerson/targetPerson objects alive for the entire produce, since it's detached from kafkaAck; buildPersonMergeEventMessage and the catch-block log only need the two UUIDs. During a broker backlog, produces can stay pending for a while (bounded by message timeout and the shared queue buffer), so this keeps that many full InternalPerson objects, including properties, in memory instead of just UUID strings.

const sourceUuid = sourcePerson.uuid
const targetUuid = targetPerson.uuid
try {
    const { key, partition, value } = buildPersonMergeEventMessage(
        this.team.id,
        sourceUuid,
        targetUuid,
        Date.now(),
        this.mergeEventsConfig.partitionCount
    )
    ...
} catch (error) {
    logger.warn('person_merge_events produce failed, dropping', {
        team_id: this.team.id,
        source_person_uuid: sourceUuid,
        target_person_uuid: targetUuid,
        error,
    })
}

[haacked]

question: Before flipping this gate on, does INGESTION_OUTPUT_PERSON_MERGE_EVENTS_PRODUCER resolve to the same producer instance as the person-message producers? If it does, detaching the merge produce from kafkaAck removes the backpressure that used to slow down merges, letting uncapped merge operations compete with critical person messages for the same queue.buffering.max.messages buffer during a broker slowdown. Pointing the shadow topic at its own producer slot would keep the isolation this PR is going for.