Add the ability to enable deferred writes (off by default.)

PostHog · Dec 6, 2023 · 347c425 · 347c425
1 parent e5e803a
commit 347c425
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 3 deletions.
diff --git a/plugin-server/src/config/config.ts b/plugin-server/src/config/config.ts
@@ -128,6 +128,7 @@ export function getDefaultConfig(): PluginsServerConfig {
         EXTERNAL_REQUEST_TIMEOUT_MS: 10 * 1000, // 10 seconds
         DROP_EVENTS_BY_TOKEN_DISTINCT_ID: '',
         DROP_EVENTS_BY_TOKEN: '',
+        POE_DEFERRED_WRITES_ENABLED: false,
         POE_EMBRACE_JOIN_FOR_TEAMS: '',
         RELOAD_PLUGIN_JITTER_MAX_MS: 60000,
 

diff --git a/plugin-server/src/types.ts b/plugin-server/src/types.ts
@@ -198,6 +198,7 @@ export interface PluginsServerConfig {
     DROP_EVENTS_BY_TOKEN_DISTINCT_ID: string
     DROP_EVENTS_BY_TOKEN: string
     POE_EMBRACE_JOIN_FOR_TEAMS: string
+    POE_DEFERRED_WRITES_ENABLED: boolean
     RELOAD_PLUGIN_JITTER_MAX_MS: number
     SKIP_UPDATE_EVENT_AND_PROPERTIES_STEP: boolean
 

diff --git a/plugin-server/src/worker/ingestion/event-pipeline/processPersonsStep.ts b/plugin-server/src/worker/ingestion/event-pipeline/processPersonsStep.ts
@@ -2,9 +2,10 @@ import { PluginEvent } from '@posthog/plugin-scaffold'
 import { DateTime } from 'luxon'
 import { Person } from 'types'
 
+import { defaultConfig } from '../../../config/config'
 import { normalizeEvent } from '../../../utils/event'
 import { status } from '../../../utils/status'
-import { PersonOverrideWriter, PersonState } from '../person-state'
+import { DeferredPersonOverrideWriter, PersonOverrideWriter, PersonState } from '../person-state'
 import { parseEventTimestamp } from '../timestamps'
 import { EventPipelineRunner } from './runner'
 
@@ -22,13 +23,22 @@ export async function processPersonsStep(
         throw error
     }
 
+    let overridesWriter: PersonOverrideWriter | DeferredPersonOverrideWriter | undefined = undefined
+    if (runner.poEEmbraceJoin) {
+        if (defaultConfig.POE_DEFERRED_WRITES_ENABLED) {
+            overridesWriter = new DeferredPersonOverrideWriter(runner.hub.db.postgres)
+        } else {
+            overridesWriter = new PersonOverrideWriter(runner.hub.db.postgres)
+        }
+    }
+
     const person = await new PersonState(
         event,
         event.team_id,
         String(event.distinct_id),
         timestamp,
         runner.hub.db,
-        runner.poEEmbraceJoin ? new PersonOverrideWriter(runner.hub.db.postgres) : undefined
+        overridesWriter
     ).update()
 
     return [event, person]

diff --git a/plugin-server/src/worker/ingestion/person-state.ts b/plugin-server/src/worker/ingestion/person-state.ts
@@ -712,7 +712,7 @@ export class DeferredPersonOverrideWriter {
      * @param lockId the lock identifier/key used to ensure that only one
      *               process is updating the overrides at a time
      */
-    constructor(private postgres: PostgresRouter, private lockId: number) {}
+    constructor(private postgres: PostgresRouter, private lockId: number = 567) {}
 
     /**
      * Enqueue an override for deferred processing.