Fixes

PostHog · Dec 25, 2024 · aaadad3 · aaadad3
1 parent 209685f
commit aaadad3
Show file tree

Hide file tree

Showing 8 changed files with 16 additions and 16 deletions.
diff --git a/plugin-server/functional_tests/scheduled-tasks-runner.test.ts b/plugin-server/functional_tests/scheduled-tasks-runner.test.ts
@@ -69,7 +69,6 @@ describe('dlq handling', () => {
             topic: 'scheduled_tasks',
             message: Buffer.from(JSON.stringify({ taskType: 'invalidTaskType', pluginConfigId: 1 })),
             key,
-    ,
         })
 
         await waitForExpect(() => {
@@ -85,7 +84,6 @@ describe('dlq handling', () => {
             topic: 'scheduled_tasks',
             message: Buffer.from(JSON.stringify({ taskType: 'runEveryMinute', pluginConfigId: 'asdf' })),
             key,
-    ,
         })
 
         await waitForExpect(() => {

diff --git a/plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer.ts b/plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer.ts
@@ -12,9 +12,9 @@ import {
 } from '../../../config/kafka-topics'
 import { BatchConsumer, startBatchConsumer } from '../../../kafka/batch-consumer'
 import { createRdConnectionConfigFromEnvVars } from '../../../kafka/config'
+import { KafkaProducerWrapper } from '../../../kafka/producer'
 import { PluginServerService, PluginsServerConfig, RedisPool, TeamId, ValueMatcher } from '../../../types'
 import { BackgroundRefresher } from '../../../utils/background-refresher'
-import { KafkaProducerWrapper } from '../../../kafka/producer'
 import { PostgresRouter } from '../../../utils/db/postgres'
 import { createRedisPool } from '../../../utils/db/redis'
 import { status } from '../../../utils/status'

diff --git a/plugin-server/src/main/ingestion-queues/session-recording/utils.ts b/plugin-server/src/main/ingestion-queues/session-recording/utils.ts
@@ -4,8 +4,8 @@ import { KafkaConsumer, Message, MessageHeader, PartitionMetadata } from 'node-r
 import path from 'path'
 import { Counter } from 'prom-client'
 
-import { PipelineEvent, RawEventMessage, RRWebEvent } from '../../../types'
 import { KafkaProducerWrapper } from '../../../kafka/producer'
+import { PipelineEvent, RawEventMessage, RRWebEvent } from '../../../types'
 import { status } from '../../../utils/status'
 import { captureIngestionWarning } from '../../../worker/ingestion/utils'
 import { eventDroppedCounter } from '../metrics'

diff --git a/plugin-server/src/main/pluginsServer.ts b/plugin-server/src/main/pluginsServer.ts
@@ -18,9 +18,9 @@ import {
     CdpProcessedEventsConsumer,
 } from '../cdp/cdp-consumers'
 import { defaultConfig } from '../config/config'
+import { KafkaProducerWrapper } from '../kafka/producer'
 import { Hub, PluginServerCapabilities, PluginServerService, PluginsServerConfig } from '../types'
 import { closeHub, createHub, createKafkaClient } from '../utils/db/hub'
-import { KafkaProducerWrapper } from '../kafka/producer'
 import { PostgresRouter } from '../utils/db/postgres'
 import { createRedisClient } from '../utils/db/redis'
 import { cancelAllScheduledJobs } from '../utils/node-schedule'

diff --git a/plugin-server/src/types.ts b/plugin-server/src/types.ts
@@ -19,10 +19,10 @@ import { DateTime } from 'luxon'
 import { VM } from 'vm2'
 
 import { EncryptedFields } from './cdp/encryption-utils'
+import { KafkaProducerWrapper } from './kafka/producer'
 import { ObjectStorage } from './main/services/object_storage'
 import { Celery } from './utils/db/celery'
 import { DB } from './utils/db/db'
-import { KafkaProducerWrapper } from './kafka/producer'
 import { PostgresRouter } from './utils/db/postgres'
 import { UUID } from './utils/utils'
 import { ActionManager } from './worker/ingestion/action-manager'

diff --git a/plugin-server/src/utils/db/db.ts b/plugin-server/src/utils/db/db.ts
@@ -1099,12 +1099,12 @@ export class DB {
         return queryResult.data as PluginLogEntry[]
     }
 
-    public queuePluginLogEntry(entry: LogEntryPayload): void {
+    public queuePluginLogEntry(entry: LogEntryPayload): Promise<void> {
         const { pluginConfig, source, message, type, timestamp, instanceId } = entry
         const configuredLogLevel = pluginConfig.plugin?.log_level || this.pluginsDefaultLogLevel
 
         if (!shouldStoreLog(configuredLogLevel, type)) {
-            return
+            return Promise.resolve()
         }
 
         const parsedEntry = {
@@ -1122,7 +1122,7 @@ export class DB {
         if (parsedEntry.message.length > 50_000) {
             const { message, ...rest } = parsedEntry
             status.warn('⚠️', 'Plugin log entry too long, ignoring.', rest)
-            return
+            return Promise.resolve()
         }
 
         pluginLogEntryCounter.labels({ plugin_id: String(pluginConfig.plugin_id), source }).inc()
@@ -1144,9 +1144,13 @@ export class DB {
                         entry: parsedEntry,
                     })
                 })
+
+            // TRICKY: We don't want to block the caller, so we return a promise that resolves immediately.
+            return Promise.resolve()
         } catch (e) {
             captureException(e, { tags: { team_id: entry.pluginConfig.team_id } })
             console.error('Failed to produce message', e, parsedEntry)
+            return Promise.resolve()
         }
     }
 

diff --git a/plugin-server/src/utils/db/error.ts b/plugin-server/src/utils/db/error.ts
@@ -25,7 +25,7 @@ export class MessageSizeTooLarge extends Error {
     readonly isRetriable = false
 }
 
-export function processError(
+export async function processError(
     server: Hub,
     pluginConfig: PluginConfig | null,
     error: Error | string,
@@ -60,16 +60,14 @@ export function processError(
                   event: event,
               }
 
-    server.db.queuePluginLogEntry({
+    await server.db.queuePluginLogEntry({
         pluginConfig,
         source: PluginLogEntrySource.Plugin,
         type: PluginLogEntryType.Error,
         message: pluginError.stack ?? pluginError.message,
         instanceId: server.instanceId,
         timestamp: pluginError.time,
     })
-
-    return Promise.resolve()
 }
 
 export function cleanErrorStackTrace(stack: string | undefined): string | undefined {

diff --git a/plugin-server/src/worker/plugins/teardown.ts b/plugin-server/src/worker/plugins/teardown.ts
@@ -14,7 +14,7 @@ export async function teardownPlugins(server: Hub, pluginConfig?: PluginConfig):
                     (async () => {
                         try {
                             await teardownPlugin()
-                            server.db.queuePluginLogEntry({
+                            await server.db.queuePluginLogEntry({
                                 pluginConfig,
                                 source: PluginLogEntrySource.System,
                                 type: PluginLogEntryType.Debug,
@@ -23,7 +23,7 @@ export async function teardownPlugins(server: Hub, pluginConfig?: PluginConfig):
                             })
                         } catch (error) {
                             await processError(server, pluginConfig, error)
-                            server.db.queuePluginLogEntry({
+                            await server.db.queuePluginLogEntry({
                                 pluginConfig,
                                 source: PluginLogEntrySource.System,
                                 type: PluginLogEntryType.Error,
@@ -34,7 +34,7 @@ export async function teardownPlugins(server: Hub, pluginConfig?: PluginConfig):
                     })()
                 )
             } else {
-                server.db.queuePluginLogEntry({
+                await server.db.queuePluginLogEntry({
                     pluginConfig,
                     source: PluginLogEntrySource.System,
                     type: PluginLogEntryType.Debug,