TEST: run same branch without uv

.github/workflows/benchmark.yml

@@ -59,8 +59,8 @@ jobs:
                   cache-dependency-path: '**/requirements*.txt'
                   token: ${{ secrets.POSTHOG_BOT_GITHUB_TOKEN }}
 
-            # uv is a fast pip alternative: https://github.com/astral-sh/uv/
-            - run: pip install uv
+            - uses: syphar/restore-virtualenv@v1
+              id: cache-benchmark-tests
 
             - name: Install SAML (python3-saml) dependencies
               shell: bash
@@ -69,12 +69,13 @@ jobs:
                   sudo apt-get install libxml2-dev libxmlsec1-dev libxmlsec1-openssl
 
             - name: Install python dependencies
+              if: steps.cache-benchmark-tests.outputs.cache-hit != 'true'
               run: |
-                  uv pip install --system -r requirements-dev.txt
-                  uv pip install --system -r requirements.txt
+                  python -m pip install -r requirements-dev.txt
+                  python -m pip install -r requirements.txt
 
             - name: Install asv
-              run: uv pip install --system asv==0.5.1 virtualenv
+              run: python -m pip install asv==0.5.1 virtualenv
 
             - name: Set up PostHog
               run: |

.github/workflows/build-hogql-parser.yml

@@ -84,11 +84,8 @@ jobs:
               if: matrix.os == 'ubuntu-22.04' # Only build the sdist once
               run: cd hogql_parser && python setup.py sdist
 
-            # uv is a fast pip alternative: https://github.com/astral-sh/uv/
-            - run: pip install uv
-
             - name: Install cibuildwheel
-              run: uv pip install --system cibuildwheel==2.16.*
+              run: python -m pip install cibuildwheel==2.16.*
 
             - name: Build wheels
               run: cd hogql_parser && python -m cibuildwheel --output-dir dist

.github/workflows/ci-backend.yml

@@ -112,17 +112,20 @@ jobs:
                   cache-dependency-path: '**/requirements*.txt'
                   token: ${{ secrets.POSTHOG_BOT_GITHUB_TOKEN }}
 
-            # uv is a fast pip alternative: https://github.com/astral-sh/uv/
-            - run: pip install uv
+            - uses: syphar/restore-virtualenv@v1
+              id: cache-backend-tests
+              with:
+                  custom_cache_key_element: v2-
 
             - name: Install SAML (python3-saml) dependencies
               run: |
                   sudo apt-get update
                   sudo apt-get install libxml2-dev libxmlsec1 libxmlsec1-dev libxmlsec1-openssl
 
             - name: Install Python dependencies
+              if: steps.cache-backend-tests.outputs.cache-hit != 'true'
               run: |
-                  uv pip install --system -r requirements.txt -r requirements-dev.txt
+                  python -m pip install -r requirements.txt -r requirements-dev.txt
 
             - name: Check for syntax errors, import sort, and code style violations
               run: |
@@ -189,17 +192,20 @@ jobs:
                   cache-dependency-path: '**/requirements*.txt'
                   token: ${{ secrets.POSTHOG_BOT_GITHUB_TOKEN }}
 
-            # uv is a fast pip alternative: https://github.com/astral-sh/uv/
-            - run: pip install uv
+            - uses: syphar/restore-virtualenv@v1
+              id: cache-backend-tests
+              with:
+                  custom_cache_key_element: v1-
 
             - name: Install SAML (python3-saml) dependencies
               run: |
                   sudo apt-get update
                   sudo apt-get install libxml2-dev libxmlsec1-dev libxmlsec1-openssl
 
             - name: Install python dependencies
+              if: steps.cache-backend-tests.outputs.cache-hit != 'true'
               run: |
-                  uv pip install --system -r requirements.txt -r requirements-dev.txt
+                  python -m pip install -r requirements.txt -r requirements-dev.txt
 
             - uses: actions/checkout@v3
               with:
@@ -209,7 +215,7 @@ jobs:
               run: |
                   # We need to ensure we have requirements for the master branch
                   # now also, so we can run migrations up to master.
-                  uv pip install --system -r requirements.txt -r requirements-dev.txt
+                  python -m pip install -r requirements.txt -r requirements-dev.txt
                   python manage.py migrate
 
             - uses: actions/checkout@v3
@@ -338,18 +344,21 @@ jobs:
                   cache-dependency-path: '**/requirements*.txt'
                   token: ${{ secrets.POSTHOG_BOT_GITHUB_TOKEN }}
 
-            # uv is a fast pip alternative: https://github.com/astral-sh/uv/
-            - run: pip install uv
+            - uses: syphar/restore-virtualenv@v1
+              id: cache-backend-tests
+              with:
+                  custom_cache_key_element: v2-
 
             - name: Install SAML (python3-saml) dependencies
               run: |
                   sudo apt-get update
                   sudo apt-get install libxml2-dev libxmlsec1-dev libxmlsec1-openssl
 
             - name: Install python dependencies
+              if: steps.cache-backend-tests.outputs.cache-hit != 'true'
               shell: bash
               run: |
-                  uv pip install --system -r requirements.txt -r requirements-dev.txt
+                  python -m pip install -r requirements.txt -r requirements-dev.txt
 
             - name: Add kafka host to /etc/hosts for kafka connectivity
               run: sudo echo "127.0.0.1 kafka" | sudo tee -a /etc/hosts

.github/workflows/ci-plugin-server.yml

@@ -122,8 +122,11 @@ jobs:
                   cache-dependency-path: '**/requirements*.txt'
                   token: ${{ secrets.POSTHOG_BOT_GITHUB_TOKEN }}
 
-            # uv is a fast pip alternative: https://github.com/astral-sh/uv/
-            - run: pip install uv
+            - uses: syphar/restore-virtualenv@v1
+              if: needs.changes.outputs.plugin-server == 'true'
+              id: cache-backend-tests
+              with:
+                  custom_cache_key_element: v1-
 
             - name: Install SAML (python3-saml) dependencies
               if: needs.changes.outputs.plugin-server == 'true'
@@ -132,10 +135,10 @@ jobs:
                   sudo apt-get install libxml2-dev libxmlsec1-dev libxmlsec1-openssl
 
             - name: Install python dependencies
-              if: needs.changes.outputs.plugin-server == 'true'
+              if: needs.changes.outputs.plugin-server == 'true' && steps.cache-backend-tests.outputs.cache-hit != 'true'
               run: |
-                  uv pip install --system -r requirements-dev.txt
-                  uv pip install --system -r requirements.txt
+                  python -m pip install -r requirements-dev.txt
+                  python -m pip install -r requirements.txt
 
             - name: Install pnpm
               if: needs.changes.outputs.plugin-server == 'true'
@@ -216,18 +219,22 @@ jobs:
                   cache-dependency-path: '**/requirements*.txt'
                   token: ${{ secrets.POSTHOG_BOT_GITHUB_TOKEN }}
 
-            # uv is a fast pip alternative: https://github.com/astral-sh/uv/
-            - run: pip install uv
+            - uses: syphar/restore-virtualenv@v1
+              id: cache-backend-tests
+              with:
+                  custom_cache_key_element: v1-
 
             - name: Install SAML (python3-saml) dependencies
               run: |
                   sudo apt-get update
                   sudo apt-get install libxml2-dev libxmlsec1-dev libxmlsec1-openssl
+              if: steps.cache-backend-tests.outputs.cache-hit != 'true'
 
             - name: Install python dependencies
+              if: steps.cache-backend-tests.outputs.cache-hit != 'true'
               run: |
-                  uv pip install --system -r requirements-dev.txt
-                  uv pip install --system -r requirements.txt
+                  python -m pip install -r requirements-dev.txt
+                  python -m pip install -r requirements.txt
 
             - name: Install pnpm
               uses: pnpm/action-setup@v2

plugin-server/src/capabilities.ts

@@ -19,6 +19,7 @@ export function getPluginServerCapabilities(config: PluginsServerConfig): Plugin
                 processAsyncOnEventHandlers: true,
                 processAsyncWebhooksHandlers: true,
                 sessionRecordingBlobIngestion: true,
+                sessionRecordingBlobOverflowIngestion: config.SESSION_RECORDING_OVERFLOW_ENABLED,
                 personOverrides: true,
                 appManagementSingleton: true,
                 preflightSchedules: true,
@@ -55,7 +56,11 @@ export function getPluginServerCapabilities(config: PluginsServerConfig): Plugin
                 sessionRecordingBlobIngestion: true,
                 ...sharedCapabilities,
             }
-
+        case PluginServerMode.recordings_blob_ingestion_overflow:
+            return {
+                sessionRecordingBlobOverflowIngestion: true,
+                ...sharedCapabilities,
+            }
         case PluginServerMode.recordings_ingestion_v3:
             return {
                 sessionRecordingV3Ingestion: true,

plugin-server/src/config/kafka-topics.ts

@@ -29,6 +29,8 @@ export const KAFKA_PERSON_OVERRIDE = `${prefix}clickhouse_person_override${suffi
 
 // read session recording snapshot items
 export const KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_EVENTS = `${prefix}session_recording_snapshot_item_events${suffix}`
+export const KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_OVERFLOW = `${prefix}session_recording_snapshot_item_overflow${suffix}`
+
 // write session recording and replay events to ClickHouse
 export const KAFKA_CLICKHOUSE_SESSION_RECORDING_EVENTS = `${prefix}clickhouse_session_recording_events${suffix}`
 export const KAFKA_CLICKHOUSE_SESSION_REPLAY_EVENTS = `${prefix}clickhouse_session_replay_events${suffix}`

plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer.ts

@@ -6,7 +6,10 @@ import { CODES, features, KafkaConsumer, librdkafkaVersion, Message, TopicPartit
 import { Counter, Gauge, Histogram } from 'prom-client'
 
 import { sessionRecordingConsumerConfig } from '../../../config/config'
-import { KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_EVENTS } from '../../../config/kafka-topics'
+import {
+    KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_EVENTS,
+    KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_OVERFLOW,
+} from '../../../config/kafka-topics'
 import { BatchConsumer, startBatchConsumer } from '../../../kafka/batch-consumer'
 import { createRdConnectionConfigFromEnvVars, createRdProducerConfigFromEnvVars } from '../../../kafka/config'
 import { createKafkaProducer } from '../../../kafka/producer'
@@ -43,6 +46,7 @@ require('@sentry/tracing')
 
 // WARNING: Do not change this - it will essentially reset the consumer
 const KAFKA_CONSUMER_GROUP_ID = 'session-recordings-blob'
+const KAFKA_CONSUMER_GROUP_ID_OVERFLOW = 'session-recordings-blob-overflow'
 const KAFKA_CONSUMER_SESSION_TIMEOUT_MS = 30000
 const SHUTDOWN_FLUSH_TIMEOUT_MS = 30000
 const CAPTURE_OVERFLOW_REDIS_KEY = '@posthog/capture-overflow/replay'
@@ -141,7 +145,8 @@ export class SessionRecordingIngester {
     teamsRefresher: BackgroundRefresher<Record<string, TeamIDWithConfig>>
     latestOffsetsRefresher: BackgroundRefresher<Record<number, number | undefined>>
     config: PluginsServerConfig
-    topic = KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_EVENTS
+    topic: string
+    consumerGroupId: string
     totalNumPartitions = 0
     isStopping = false
 
@@ -156,11 +161,16 @@ export class SessionRecordingIngester {
         private globalServerConfig: PluginsServerConfig,
         private postgres: PostgresRouter,
         private objectStorage: ObjectStorage,
+        private consumeOverflow: boolean,
         captureRedis: Redis | undefined
     ) {
         this.debugPartition = globalServerConfig.SESSION_RECORDING_DEBUG_PARTITION
             ? parseInt(globalServerConfig.SESSION_RECORDING_DEBUG_PARTITION)
             : undefined
+        this.topic = consumeOverflow
+            ? KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_OVERFLOW
+            : KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_EVENTS
+        this.consumerGroupId = this.consumeOverflow ? KAFKA_CONSUMER_GROUP_ID_OVERFLOW : KAFKA_CONSUMER_GROUP_ID
 
         // NOTE: globalServerConfig contains the default pluginServer values, typically not pointing at dedicated resources like kafka or redis
         // We still connect to some of the non-dedicated resources such as postgres or the Replay events kafka.
@@ -169,7 +179,7 @@ export class SessionRecordingIngester {
 
         this.realtimeManager = new RealtimeManager(this.redisPool, this.config)
 
-        if (globalServerConfig.SESSION_RECORDING_OVERFLOW_ENABLED && captureRedis) {
+        if (globalServerConfig.SESSION_RECORDING_OVERFLOW_ENABLED && captureRedis && !consumeOverflow) {
             this.overflowDetection = new OverflowManager(
                 globalServerConfig.SESSION_RECORDING_OVERFLOW_BUCKET_CAPACITY,
                 globalServerConfig.SESSION_RECORDING_OVERFLOW_BUCKET_REPLENISH_RATE,
@@ -207,7 +217,7 @@ export class SessionRecordingIngester {
         this.latestOffsetsRefresher = new BackgroundRefresher(async () => {
             const results = await Promise.all(
                 this.assignedTopicPartitions.map(({ partition }) =>
-                    queryWatermarkOffsets(this.connectedBatchConsumer, partition).catch((err) => {
+                    queryWatermarkOffsets(this.connectedBatchConsumer, this.topic, partition).catch((err) => {
                         // NOTE: This can error due to a timeout or the consumer being disconnected, not stop the process
                         // as it is currently only used for reporting lag.
                         captureException(err)
@@ -259,8 +269,6 @@ export class SessionRecordingIngester {
 
         const { team_id, session_id } = event
         const key = `${team_id}-${session_id}`
-        // TODO: use this for session key too if it's safe to do so
-        const overflowKey = `${team_id}:${session_id}`
 
         const { partition, highOffset } = event.metadata
         const isDebug = this.debugPartition === partition
@@ -285,11 +293,7 @@ export class SessionRecordingIngester {
 
         // Check that we are not below the high-water mark for this partition (another consumer may have flushed further than us when revoking)
         if (
-            await this.persistentHighWaterMarker.isBelowHighWaterMark(
-                event.metadata,
-                KAFKA_CONSUMER_GROUP_ID,
-                highOffset
-            )
+            await this.persistentHighWaterMarker.isBelowHighWaterMark(event.metadata, this.consumerGroupId, highOffset)
         ) {
             dropEvent('high_water_mark_partition')
             return
@@ -318,7 +322,7 @@ export class SessionRecordingIngester {
 
         await Promise.allSettled([
             this.sessions[key]?.add(event),
-            this.overflowDetection?.observe(overflowKey, event.metadata.rawSize, event.metadata.timestamp),
+            this.overflowDetection?.observe(session_id, event.metadata.rawSize, event.metadata.timestamp),
         ])
     }
 
@@ -486,8 +490,8 @@ export class SessionRecordingIngester {
         const replayClusterConnectionConfig = createRdConnectionConfigFromEnvVars(this.config)
         this.batchConsumer = await startBatchConsumer({
             connectionConfig: replayClusterConnectionConfig,
-            groupId: KAFKA_CONSUMER_GROUP_ID,
-            topic: KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_EVENTS,
+            groupId: this.consumerGroupId,
+            topic: this.topic,
             autoCommit: false,
             sessionTimeout: KAFKA_CONSUMER_SESSION_TIMEOUT_MS,
             maxPollIntervalMs: this.config.KAFKA_CONSUMPTION_MAX_POLL_INTERVAL_MS,
@@ -510,7 +514,7 @@ export class SessionRecordingIngester {
             debug: this.config.SESSION_RECORDING_KAFKA_DEBUG,
         })
 
-        this.totalNumPartitions = (await getPartitionsForTopic(this.connectedBatchConsumer)).length
+        this.totalNumPartitions = (await getPartitionsForTopic(this.connectedBatchConsumer, this.topic)).length
 
         addSentryBreadcrumbsEventListeners(this.batchConsumer.consumer)
 
@@ -820,7 +824,7 @@ export class SessionRecordingIngester {
                 })
 
                 // Store the committed offset to the persistent store to avoid rebalance issues
-                await this.persistentHighWaterMarker.add(tp, KAFKA_CONSUMER_GROUP_ID, highestOffsetToCommit)
+                await this.persistentHighWaterMarker.add(tp, this.consumerGroupId, highestOffsetToCommit)
                 // Clear all session offsets below the committed offset (as we know they have been flushed)
                 await this.sessionHighWaterMarker.clear(tp, highestOffsetToCommit)
                 gaugeOffsetCommitted.set({ partition }, highestOffsetToCommit)

plugin-server/src/main/ingestion-queues/session-recording/utils.ts

@@ -4,7 +4,6 @@ import { KafkaConsumer, Message, MessageHeader, PartitionMetadata, TopicPartitio
 import path from 'path'
 import { Counter } from 'prom-client'
 
-import { KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_EVENTS } from '../../../config/kafka-topics'
 import { PipelineEvent, RawEventMessage, RRWebEvent } from '../../../types'
 import { KafkaProducerWrapper } from '../../../utils/db/kafka-producer-wrapper'
 import { status } from '../../../utils/status'
@@ -36,6 +35,7 @@ export const bufferFileDir = (root: string) => path.join(root, 'session-buffer-f
 
 export const queryWatermarkOffsets = (
     kafkaConsumer: KafkaConsumer | undefined,
+    topic: string,
     partition: number,
     timeout = 10000
 ): Promise<[number, number]> => {
@@ -44,20 +44,15 @@ export const queryWatermarkOffsets = (
             return reject('Not connected')
         }
 
-        kafkaConsumer.queryWatermarkOffsets(
-            KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_EVENTS,
-            partition,
-            timeout,
-            (err, offsets) => {
-                if (err) {
-                    captureException(err)
-                    status.error('🔥', 'Failed to query kafka watermark offsets', err)
-                    return reject(err)
-                }
-
-                resolve([partition, offsets.highOffset])
+        kafkaConsumer.queryWatermarkOffsets(topic, partition, timeout, (err, offsets) => {
+            if (err) {
+                captureException(err)
+                status.error('🔥', 'Failed to query kafka watermark offsets', err)
+                return reject(err)
             }
-        )
+
+            resolve([partition, offsets.highOffset])
+        })
     })
 }
 
@@ -89,7 +84,7 @@ export const queryCommittedOffsets = (
 
 export const getPartitionsForTopic = (
     kafkaConsumer: KafkaConsumer | undefined,
-    topic = KAFKA_SESSION_RECORDING_SNAPSHOT_ITEM_EVENTS
+    topic: string
 ): Promise<PartitionMetadata[]> => {
     return new Promise<PartitionMetadata[]>((resolve, reject) => {
         if (!kafkaConsumer) {