perf(nodejs): move the ingestion e2e suite to the parallel lane

nodejs/jest.config.js

@@ -3,4 +3,8 @@ const config = require('./jest.config.shared')
 module.exports = {
     ...config,
     testMatch: ['<rootDir>/tests/**/!(*.serial).test.ts', '<rootDir>/src/**/!(*.serial).test.ts'],
+    // Only applies to `test.concurrent` bodies (the ingestion e2e harness). Those spend most of
+    // their time waiting on ClickHouse's Kafka engine flush, so overlapping more of them per
+    // worker shortens the file far more than it costs in CPU. Jest's default is 5.
+    maxConcurrency: 15,
 }

nodejs/tests/helpers/ingestion-e2e.ts

@@ -50,7 +50,7 @@ import { EncryptedFields } from '../../src/cdp/utils/encryption-utils'
 import { PipelineEvent, PluginsServerConfig, ProjectId, RawClickHouseEvent, RedisPool, Team } from '../../src/types'
 import { Clickhouse } from './clickhouse'
 import { waitForExpect } from './expectations'
-import { ensureKafkaTopics } from './kafka'
+import { TEST_KAFKA_TOPICS, ensureKafkaTopics } from './kafka'
 import { createUserTeamAndOrganization, uniqueTestId } from './sql'
 
 export const DEFAULT_TEAM: Team = {
@@ -493,6 +493,29 @@ export async function createIngestionTestInfra(
  * test. The caller supplies the `buildIngester` function that constructs the
  * consumer under test — different pipelines have different deps.
  */
+let infraReady: Promise<void> | undefined
+
+/**
+ * Shared infra every e2e ingester needs: the output topics and a consuming ClickHouse Kafka
+ * engine. Memoized per worker process and awaited inside each test, because jest-circus starts
+ * `test.concurrent` bodies as soon as a nested `beforeAll` hits its first await, so a hook alone
+ * cannot guarantee the topics exist before the first consumer's startup check runs.
+ */
+export function ensureIngestionE2EInfraReady(): Promise<void> {
+    if (!infraReady) {
+        infraReady = (async () => {
+            await ensureKafkaTopics(TEST_KAFKA_TOPICS)
+            const clickhouse = Clickhouse.create()
+            try {
+                await waitForClickHouseKafkaConsumer(clickhouse)
+            } finally {
+                clickhouse.close()
+            }
+        })()
+    }
+    return infraReady
+}
+
 export function createTestWithTeamIngester<T extends IngesterLike>(
     baseConfig: Partial<IngestionTestConfig>,
     buildIngester: BuildIngester<T>
@@ -502,7 +525,8 @@ export function createTestWithTeamIngester<T extends IngesterLike>(
         config: TeamIngesterTestConfig = {},
         testFn: (ctx: TeamIngesterTestContext<T>) => Promise<void>
     ) => {
-        test(name, async () => {
+        test.concurrent(name, async () => {
+            await ensureIngestionE2EInfraReady()
             const infra = await createIngestionTestInfra({
                 ...baseConfig,
                 ...config.pluginServerConfig,

nodejs/tests/ingestion/flag-evaluations-e2e.test.ts

@@ -16,13 +16,12 @@ import {
     EventBuilder,
     createKafkaMessages,
     createTestWithTeamIngester,
+    ensureIngestionE2EInfraReady,
     fetchEvents,
     fetchFlagEvaluations,
-    waitForClickHouseKafkaConsumer,
     waitForKafkaMessages,
 } from '~/tests/helpers/ingestion-e2e'
 import { createTestIngestionOutputs, createTestMonitoringOutputs } from '~/tests/helpers/ingestion-outputs'
-import { TEST_KAFKA_TOPICS, ensureKafkaTopics } from '~/tests/helpers/kafka'
 
 jest.mock('~/common/utils/logger')
 
@@ -44,8 +43,7 @@ describe('Flag evaluations shadow-routing E2E', () => {
 
     beforeAll(async () => {
         clickhouse = Clickhouse.create()
-        await ensureKafkaTopics(TEST_KAFKA_TOPICS)
-        await waitForClickHouseKafkaConsumer(clickhouse)
+        await ensureIngestionE2EInfraReady()
     })
 
     afterAll(() => {

nodejs/tests/ingestion/ingestion-e2e.serial.test.ts → nodejs/tests/ingestion/ingestion-e2e.test.ts

@@ -1,4 +1,3 @@
-// Serial until the startup topic check tolerates a transient metadata gap (see the follow-up PR).
 import { DateTime } from 'luxon'
 
 import { createHogTransformerService } from '~/cdp/hog-transformations/hog-transformer.service'
@@ -13,13 +12,12 @@ import {
     EventBuilder,
     createKafkaMessages,
     createTestWithTeamIngester,
+    ensureIngestionE2EInfraReady,
     fetchEvents,
     fetchIngestionWarnings,
-    waitForClickHouseKafkaConsumer,
     waitForKafkaMessages,
 } from '~/tests/helpers/ingestion-e2e'
 import { createTestIngestionOutputs, createTestMonitoringOutputs } from '~/tests/helpers/ingestion-outputs'
-import { TEST_KAFKA_TOPICS, ensureKafkaTopics } from '~/tests/helpers/kafka'
 import { createUserTeamAndOrganization, fetchPostgresPersons, uniqueTestId } from '~/tests/helpers/sql'
 import { GroupTypeIndex, InternalPerson } from '~/types'
 
@@ -97,8 +95,7 @@ describe.each([
     beforeAll(async () => {
         console.log('Creating Clickhouse client')
         clickhouse = Clickhouse.create()
-        await ensureKafkaTopics(TEST_KAFKA_TOPICS)
-        await waitForClickHouseKafkaConsumer(clickhouse)
+        await ensureIngestionE2EInfraReady()
         process.env.SITE_URL = 'https://example.com'
     })
 

nodejs/tests/ingestion/person-properties-metadata.e2e.test.ts

@@ -16,9 +16,12 @@ import { KafkaProducerWrapper } from '~/common/kafka/producer'
 import { UUIDT } from '~/common/utils/utils'
 import { IngestionConsumer } from '~/ingestion/ingestion-consumer'
 import { waitForExpect } from '~/tests/helpers/expectations'
-import { IngestionTestInfra, createIngestionTestInfra } from '~/tests/helpers/ingestion-e2e'
+import {
+    IngestionTestInfra,
+    createIngestionTestInfra,
+    ensureIngestionE2EInfraReady,
+} from '~/tests/helpers/ingestion-e2e'
 import { createTestIngestionOutputs, createTestMonitoringOutputs } from '~/tests/helpers/ingestion-outputs'
-import { TEST_KAFKA_TOPICS, ensureKafkaTopics } from '~/tests/helpers/kafka'
 import { createUserTeamAndOrganization, fetchPostgresPersons, uniqueTestId } from '~/tests/helpers/sql'
 import { PipelineEvent, PluginsServerConfig, ProjectId, Team } from '~/types'
 
@@ -211,7 +214,7 @@ describe('Person properties_last_updated_at and properties_last_operation behavi
     const testWithTeamIngester = createTestWithTeamIngester()
 
     beforeAll(async () => {
-        await ensureKafkaTopics(TEST_KAFKA_TOPICS)
+        await ensureIngestionE2EInfraReady()
         process.env.SITE_URL = 'https://example.com'
     })