feat(flags): skip events writes for flag evaluations only orgs

nodejs/src/ingestion/common/flag-evaluations/flag-evaluations-service.test.ts

@@ -1,4 +1,5 @@
 import { IngestionLane } from '~/ingestion/config'
+import { FlagEvaluationsMode } from '~/types'
 
 import { FlagEvaluationsEnvConfig, createFlagEvaluationsService } from './flag-evaluations-service'
 
@@ -14,6 +15,7 @@ describe('FlagEvaluationsService', () => {
         INGESTION_FLAG_EVALUATIONS_MODE: mode,
         INGESTION_FLAG_EVALUATIONS_TEAMS: teams,
         INGESTION_FLAG_EVALUATIONS_EXCLUDED_TEAMS: excludedTeams,
+        INGESTION_FLAG_EVALUATIONS_ONLY_DISABLED: false,
         INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC: topic,
     })
 
@@ -52,4 +54,23 @@ describe('FlagEvaluationsService', () => {
             expect(service?.isEnabledForTeam(teamId)).toBe(expected)
         })
     })
+
+    describe('stopsEventsWritesFor', () => {
+        it.each([
+            [false, true],
+            [true, false],
+        ])(
+            'INGESTION_FLAG_EVALUATIONS_ONLY_DISABLED=%s -> %s for a FLAG_EVALUATIONS_ONLY team',
+            (onlyDisabled, expected) => {
+                const service = createFlagEvaluationsService({
+                    ...envConfig('dual_write'),
+                    INGESTION_FLAG_EVALUATIONS_ONLY_DISABLED: onlyDisabled,
+                })
+
+                expect(
+                    service?.stopsEventsWritesFor({ flag_evaluations_mode: FlagEvaluationsMode.FlagEvaluationsOnly })
+                ).toBe(expected)
+            }
+        )
+    })
 })

nodejs/src/ingestion/common/flag-evaluations/flag-evaluations-service.ts

@@ -7,21 +7,24 @@ import {
     IngestionOutputsConfig,
     REALTIME_INGESTION_LANES,
 } from '~/ingestion/config'
-import { ValueMatcher } from '~/types'
+import { FlagEvaluationsMode, Team, ValueMatcher } from '~/types'
 
 export interface FlagEvaluationsConfig {
     /** '*' for all teams, or an explicit allowlist of team IDs. */
     teams: number[] | '*'
     /** Escape hatch: teams never forked, even when `teams` is '*'. */
     excludedTeams: number[]
+    /** Keeps the events writes for teams on FLAG_EVALUATIONS_ONLY. */
+    flagEvaluationsOnlyDisabled: boolean
 }
 
 /**
- * Gate for the $feature_flag_called fork that shadow-writes flag evaluations to
- * the ClickHouse flag_evaluations table (via the clickhouse_flag_evaluations
- * topic) while the event continues to the events table unchanged.
+ * Gate for the $feature_flag_called fork that writes flag evaluations to the
+ * ClickHouse flag_evaluations table (via the clickhouse_flag_evaluations
+ * topic). The event continues to the events table unless the team's
+ * organization is on FLAG_EVALUATIONS_ONLY.
  *
- * The shadow write is never load-bearing for an individual event, but the batch
+ * The fork write is never load-bearing for a dual-written event, but the batch
  * does not commit its offsets until the broker answers. See
  * createForkFlagEvaluationsStep for the ack contract and
  * flagEvaluationsPendingAcks for the stall it can cause.
@@ -34,14 +37,22 @@ export interface FlagEvaluationsConfig {
  */
 export class FlagEvaluationsService {
     private isEnabled: ValueMatcher<number>
+    private flagEvaluationsOnlyDisabled: boolean
 
     constructor(config: FlagEvaluationsConfig) {
         this.isEnabled = buildTeamGate(config.teams, config.excludedTeams)
+        this.flagEvaluationsOnlyDisabled = config.flagEvaluationsOnlyDisabled
     }
 
     isEnabledForTeam(teamId: number): boolean {
         return this.isEnabled(teamId)
     }
+
+    stopsEventsWritesFor(team: Pick<Team, 'flag_evaluations_mode'>): boolean {
+        return (
+            team.flag_evaluations_mode === FlagEvaluationsMode.FlagEvaluationsOnly && !this.flagEvaluationsOnlyDisabled
+        )
+    }
 }
 
 export type FlagEvaluationsEnvConfig = Pick<
@@ -50,6 +61,7 @@ export type FlagEvaluationsEnvConfig = Pick<
     | 'INGESTION_FLAG_EVALUATIONS_MODE'
     | 'INGESTION_FLAG_EVALUATIONS_TEAMS'
     | 'INGESTION_FLAG_EVALUATIONS_EXCLUDED_TEAMS'
+    | 'INGESTION_FLAG_EVALUATIONS_ONLY_DISABLED'
 > &
     Pick<IngestionOutputsConfig, 'INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC'>
 
@@ -104,5 +116,9 @@ export function createFlagEvaluationsService(envConfig: FlagEvaluationsEnvConfig
         logger.warn('INGESTION_FLAG_EVALUATIONS_MODE is set but INGESTION_FLAG_EVALUATIONS_TEAMS is empty, not forking')
         return undefined
     }
-    return new FlagEvaluationsService({ teams, excludedTeams })
+    return new FlagEvaluationsService({
+        teams,
+        excludedTeams,
+        flagEvaluationsOnlyDisabled: envConfig.INGESTION_FLAG_EVALUATIONS_ONLY_DISABLED,
+    })
 }

nodejs/src/ingestion/common/steps/event-processing/create-event-step.test.ts

@@ -6,17 +6,20 @@ import { castTimestampOrNow } from '~/common/utils/utils'
 import { isOkResult } from '~/ingestion/framework/results'
 import { createTestEventHeaders } from '~/tests/helpers/event-headers'
 import { createTestMessage } from '~/tests/helpers/kafka-message'
-import { Person, PersonMode, PreIngestionEvent, ProjectId, TimestampFormat } from '~/types'
+import { createTestTeam } from '~/tests/helpers/team'
+import { Person, PersonMode, PreIngestionEvent, ProjectId, Team, TimestampFormat } from '~/types'
 
 import { CreateEventStepInput, createCreateEventStep } from './create-event-step'
 
 describe('create-event-step', () => {
     let mockPerson: Person
     let mockPreparedEvent: PreIngestionEvent
     let mockMessage: Message
+    let mockTeam: Team
 
     beforeEach(() => {
         mockMessage = createTestMessage()
+        mockTeam = createTestTeam({ id: 1 })
         mockPerson = {
             team_id: 1,
             properties: { email: 'test@example.com', name: 'Test User' },
@@ -40,6 +43,7 @@ describe('create-event-step', () => {
         it('should create event with processPerson=true', async () => {
             const step = createCreateEventStep(EVENTS_OUTPUT)
             const input = {
+                team: mockTeam,
                 person: mockPerson,
                 preparedEvent: mockPreparedEvent,
                 processPerson: true,
@@ -75,6 +79,7 @@ describe('create-event-step', () => {
         it('should create event with processPerson=false', async () => {
             const step = createCreateEventStep(EVENTS_OUTPUT)
             const input = {
+                team: mockTeam,
                 person: mockPerson,
                 preparedEvent: mockPreparedEvent,
                 processPerson: false,
@@ -105,6 +110,7 @@ describe('create-event-step', () => {
 
             const step = createCreateEventStep(EVENTS_OUTPUT)
             const input = {
+                team: mockTeam,
                 person: personWithForceUpgrade,
                 preparedEvent: mockPreparedEvent,
                 processPerson: true,
@@ -134,6 +140,7 @@ describe('create-event-step', () => {
 
             const step = createCreateEventStep(EVENTS_OUTPUT)
             const input = {
+                team: mockTeam,
                 person: mockPerson,
                 preparedEvent: eventWithSetProperties,
                 processPerson: true,
@@ -159,6 +166,7 @@ describe('create-event-step', () => {
         it('should preserve event properties as native object', async () => {
             const step = createCreateEventStep(EVENTS_OUTPUT)
             const input = {
+                team: mockTeam,
                 person: mockPerson,
                 preparedEvent: mockPreparedEvent,
                 processPerson: true,
@@ -191,6 +199,7 @@ describe('create-event-step', () => {
 
             const step = createCreateEventStep(EVENTS_OUTPUT)
             const input = {
+                team: mockTeam,
                 person: mockPerson,
                 preparedEvent: eventWithElements,
                 processPerson: true,
@@ -217,6 +226,7 @@ describe('create-event-step', () => {
 
             const step = createCreateEventStep<typeof EVENTS_OUTPUT, CustomInput>(EVENTS_OUTPUT)
             const input: CustomInput = {
+                team: mockTeam,
                 person: mockPerson,
                 preparedEvent: mockPreparedEvent,
                 processPerson: true,
@@ -238,6 +248,7 @@ describe('create-event-step', () => {
         it('should set correct timestamps', async () => {
             const step = createCreateEventStep(EVENTS_OUTPUT)
             const input = {
+                team: mockTeam,
                 person: mockPerson,
                 preparedEvent: mockPreparedEvent,
                 processPerson: true,
@@ -269,6 +280,7 @@ describe('create-event-step', () => {
 
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const input = {
+                    team: mockTeam,
                     person: mockPerson,
                     preparedEvent: eventWithType,
                     processPerson: true,
@@ -292,6 +304,7 @@ describe('create-event-step', () => {
             it('duplicates multivariate $feature_flag_called events for allowlisted teams with a deterministic uuid', async () => {
                 const step = createCreateEventStep(EVENTS_OUTPUT, '2')
                 const input = {
+                    team: mockTeam,
                     person: mockPerson,
                     preparedEvent: {
                         ...mockPreparedEvent,
@@ -340,6 +353,7 @@ describe('create-event-step', () => {
             ])('does not duplicate %s', async (_, teamId, event, response) => {
                 const step = createCreateEventStep(EVENTS_OUTPUT, '2')
                 const result = await step({
+                    team: mockTeam,
                     person: mockPerson,
                     preparedEvent: {
                         ...mockPreparedEvent,
@@ -363,6 +377,7 @@ describe('create-event-step', () => {
             it('does not duplicate when no teams are configured', async () => {
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const result = await step({
+                    team: mockTeam,
                     person: mockPerson,
                     preparedEvent: {
                         ...mockPreparedEvent,
@@ -388,6 +403,7 @@ describe('create-event-step', () => {
             it('should include historical_migration in event when historicalMigration=true', async () => {
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const input = {
+                    team: mockTeam,
                     person: mockPerson,
                     preparedEvent: mockPreparedEvent,
                     processPerson: true,
@@ -409,6 +425,7 @@ describe('create-event-step', () => {
             it('should not include historical_migration in event when historicalMigration=false', async () => {
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const input = {
+                    team: mockTeam,
                     person: mockPerson,
                     preparedEvent: mockPreparedEvent,
                     processPerson: true,
@@ -441,6 +458,7 @@ describe('create-event-step', () => {
 
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const input = {
+                    team: mockTeam,
                     person,
                     preparedEvent: mockPreparedEvent,
                     processPerson: config.processPerson,
@@ -464,6 +482,7 @@ describe('create-event-step', () => {
             it('should generate deterministic person_id from distinct_id when person is undefined', async () => {
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const input = {
+                    team: mockTeam,
                     person: undefined,
                     preparedEvent: mockPreparedEvent,
                     processPerson: true,
@@ -486,6 +505,7 @@ describe('create-event-step', () => {
             it('should return empty person_properties when person is undefined', async () => {
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const input = {
+                    team: mockTeam,
                     person: undefined,
                     preparedEvent: mockPreparedEvent,
                     processPerson: true,
@@ -507,6 +527,7 @@ describe('create-event-step', () => {
             it('should set person_created_at to null when person is undefined', async () => {
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const input = {
+                    team: mockTeam,
                     person: undefined,
                     preparedEvent: mockPreparedEvent,
                     processPerson: true,
@@ -528,6 +549,7 @@ describe('create-event-step', () => {
             it('should set person_mode to full when person is undefined and processPerson=true', async () => {
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const input = {
+                    team: mockTeam,
                     person: undefined,
                     preparedEvent: mockPreparedEvent,
                     processPerson: true,
@@ -549,6 +571,7 @@ describe('create-event-step', () => {
             it('should set person_mode to propertyless when person is undefined and processPerson=false', async () => {
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const input = {
+                    team: mockTeam,
                     person: undefined,
                     preparedEvent: mockPreparedEvent,
                     processPerson: false,

nodejs/src/ingestion/common/steps/event-processing/create-event-step.ts

@@ -7,7 +7,7 @@ import { createEvent } from '~/ingestion/common/steps/event-processing/create-ev
 import { EventUsageRecord } from '~/ingestion/common/steps/usage-records-steps'
 import { ok } from '~/ingestion/framework/results'
 import { ProcessingStep } from '~/ingestion/framework/steps'
-import { EventHeaders, Person, PreIngestionEvent } from '~/types'
+import { EventHeaders, Person, PreIngestionEvent, Team } from '~/types'
 
 import { EventToEmit } from './emit-event-step'
 
@@ -34,6 +34,7 @@ function isMultivariateFeatureFlagCalledEvent(event: PreIngestionEvent): boolean
 }
 
 export interface CreateEventStepInput {
+    team: Team
     person?: Person
     preparedEvent: PreIngestionEvent
     processPerson: boolean
@@ -45,6 +46,7 @@ export interface CreateEventStepInput {
 }
 
 export interface CreateEventStepResult<O extends string> {
+    team: Team
     eventsToEmit: EventToEmit<O>[]
     teamId: number
     headers: EventHeaders
@@ -61,6 +63,7 @@ export function createCreateEventStep<O extends string, T extends CreateEventSte
 
     return function createEventStep(input) {
         const {
+            team,
             person,
             preparedEvent,
             processPerson,
@@ -95,6 +98,7 @@ export function createCreateEventStep<O extends string, T extends CreateEventSte
         }
 
         const result: CreateEventStepResult<O> = {
+            team,
             eventsToEmit,
             teamId: preparedEvent.teamId,
             headers,

nodejs/src/ingestion/common/steps/event-processing/emit-event-step.ts

@@ -8,7 +8,7 @@ import { MessageSizeTooLarge } from '~/common/utils/db/error'
 import { safeClickhouseString } from '~/common/utils/db/utils'
 import { castTimestampOrNow, castTimestampToClickhouseFormat } from '~/common/utils/utils'
 import { isAiEventName } from '~/ingestion/common/ai-event-types'
-import { emitIngestionWarning } from '~/ingestion/common/ingestion-warnings'
+import { IngestionWarning, emitIngestionWarning } from '~/ingestion/common/ingestion-warnings'
 import { eventProcessedAndIngestedCounter } from '~/ingestion/common/metrics'
 import { EventUsageRecord } from '~/ingestion/common/steps/usage-records-steps'
 import { ok } from '~/ingestion/framework/results'
@@ -97,15 +97,11 @@ export function createEmitEventStep<O extends string, T extends EmitEventStepInp
                     // Some messages end up significantly larger than the original
                     // after plugin processing, person & group enrichment, etc.
                     if (error instanceof MessageSizeTooLarge) {
-                        await emitIngestionWarning(outputs, serialized.team_id, {
-                            type: 'message_size_too_large',
-                            details: {
-                                eventUuid: serialized.uuid,
-                                distinctId: serialized.distinct_id,
-                                personId: serialized.person_id,
-                            },
-                            pipelineStep: 'emit-event',
-                        })
+                        await emitIngestionWarning(
+                            outputs,
+                            serialized.team_id,
+                            eventTooLargeWarning(serialized, 'emit-event')
+                        )
                         // The event was not ingested, so there is no info to resolve with
                         return null
                     } else {
@@ -129,6 +125,17 @@ export function createEmitEventStep<O extends string, T extends EmitEventStepInp
     }
 }
 
+export function eventTooLargeWarning(
+    event: { uuid: string; distinct_id: string; person_id?: string },
+    pipelineStep: string
+): IngestionWarning {
+    return {
+        type: 'message_size_too_large',
+        details: { eventUuid: event.uuid, distinctId: event.distinct_id, personId: event.person_id },
+        pipelineStep,
+    }
+}
+
 export function serializeEvent(event: ProcessedEvent): RawKafkaEvent {
     return {
         uuid: event.uuid,