refactor: S3 batch export with SPMC abstractions

[tomasfarias]

Problem

Builds on top of #26575 to refactor S3 batch export to use fully-async SPMC abstractions. Big part of the work was done in the linked PR, so there isn't much to do here.

Changes

1. Implement new S3Consumer class
2. Move flush function to flush method of S3Consumer.
3. Call S3Consumer with run_consumer_loop in S3 batch export.
4. Implement the new events_recent view in spmc, as that was missing from my other PR.

The implementation does have a potential concurrency conflict when computing the next_part_number that I think to have resolved by using a new pending_parts.

👉 Stay up-to-date with PostHog coding conventions for a smoother review.

Does this work well for both Cloud and self-hosted?

How did you test this code?

[tomasfarias]

Without this, we could run into concurrency problems: As we only append a part to self.parts after it has been sent. Imagine the following (very realistic) sequence of events with just 2 consumers:

1. Consumer 1 starts, gets assigned part 1 by self.part_number+1 as self.parts is empty.
2. Consumer 1 relinquishes control back to main loop on awaiting self.upload_part_retryable.
3. Consumer 2 starts, gets assigned part 1 by self.part_number+1 as self.parts is still empty.
4. Consumer 2 relinquishes control back to main loop on awaiting self.upload_part_retryable.
5. Consumer 1 resumes, appends part 1 to self.parts, and finishes.
6. Consumer 2 resumes, appends part 1 to self.parts, and finishes.

We have ended with two parts with part number 1, which means whichever consumer uploaded last would have overridden the other in the upload.

With the addition of self.pending_parts we can "lock" any part numbers in progress so that they don't get repeated when doing self.part_number+1.

[rossgray]

thanks for the explanation 👍

[rossgray]

LGTM

[sentry-io]

Suspect Issues

This pull request was deployed and Sentry observed the following issues:

• ‼️ RecordBatchTaskError: The record batch consumer encountered an error during execution posthog.temporal.batch_exports.spmc in raise_on... View Issue
• ‼️ RecordBatchConsumerRetryableExceptionGroup: At least one unhandled retryable errors in a RecordBatch consumer TaskGroup posthog.temporal.batch_exports.spmc in run_cons... View Issue
• ‼️ RecordBatchTaskError: The record batch consumer encountered an error during execution posthog.temporal.batch_exports.spmc in raise_on... View Issue
• ‼️ RecordBatchTaskError: The record batch consumer encountered an error during execution posthog.temporal.batch_exports.spmc in raise_on... View Issue

_{Did you find this useful? React with a 👍 or 👎}