feat: add Kafka balance outcome consumer - #3191
Conversation
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 527d83280d
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| message, | ||
| }) => { | ||
| const partitionKey = JSON.stringify([recordTopic, partition]); | ||
| if (!initializedPartitions.has(partitionKey)) { |
There was a problem hiding this comment.
When a partition moves from this consumer to another worker and later returns, this process-wide set still marks it initialized, so the first newly assigned record skips reconciliation. If the other worker advanced the group while this worker's local SQLite remained behind, processTrackOutcomeRecord applies the newer outcome to stale state and throws OutOfOrderTrackOutcomeError, repeatedly crashing consumption instead of rewinding to the stored offset. Clear or rebuild this state on partition revocation/assignment rather than retaining it for the consumer's lifetime.
Useful? React with 👍 / 👎.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
3 issues found across 8 files
Confidence score: 3/5
apps/balance-worker/src/kafka/kafkaTrackOutcomeConsumer.tscan leave a Kafka consumer running when shutdown races with startup, causing incomplete cleanup and potentially lingering processing; track the in-flight startup sostop()awaits or cancels it before disconnecting.apps/balance-worker/src/kafka/trackOutcomeRecord.tsmay accept malformed UTF-8 in outcomes or matching keys after replacement characters are inserted, leading to incorrect validation or key matching; use fatal UTF-8 decoding for both fields and map decode failures appropriately.apps/balance-worker/tests/unit/kafka/kafka-test-fixtures.tscan leak temporary directories when SQLite initialization throws, leaving test artifacts behind; move store creation inside the cleanup-protectedtryblock.
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="apps/balance-worker/tests/unit/kafka/kafka-test-fixtures.ts">
<violation number="1" location="apps/balance-worker/tests/unit/kafka/kafka-test-fixtures.ts:79">
P3: If `openSqliteBalanceStateStore` throws (e.g., SQLite fails to open the database file), the `try`/`catch` never runs, so the freshly created temp directory from `mkdtempSync` is left behind and never removed. Move `openSqliteBalanceStateStore` inside the `try` so the error path also closes/cleans up, or wrap the whole body including the open call.</violation>
</file>
<file name="apps/balance-worker/src/kafka/trackOutcomeRecord.ts">
<violation number="1" location="apps/balance-worker/src/kafka/trackOutcomeRecord.ts:124">
P2: Malformed UTF-8 can be accepted as a valid outcome or matching key because `Buffer.toString("utf8")` silently inserts U+FFFD before validation. Decode both value and key with fatal UTF-8 handling, mapping decoder failures to `InvalidKafkaTrackOutcomeRecordError`.</violation>
</file>
<file name="apps/balance-worker/src/kafka/kafkaTrackOutcomeConsumer.ts">
<violation number="1" location="apps/balance-worker/src/kafka/kafkaTrackOutcomeConsumer.ts:139">
P2: When shutdown races with startup, `stop()` returns before the consumer is marked started and never stops or disconnects it. Track the in-flight startup and have `stop()` await or cancel it before performing shutdown.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
|
|
||
| let input: unknown; | ||
| try { | ||
| input = JSON.parse(value.toString("utf8")); |
There was a problem hiding this comment.
P2: Malformed UTF-8 can be accepted as a valid outcome or matching key because Buffer.toString("utf8") silently inserts U+FFFD before validation. Decode both value and key with fatal UTF-8 handling, mapping decoder failures to InvalidKafkaTrackOutcomeRecordError.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At apps/balance-worker/src/kafka/trackOutcomeRecord.ts, line 124:
<comment>Malformed UTF-8 can be accepted as a valid outcome or matching key because `Buffer.toString("utf8")` silently inserts U+FFFD before validation. Decode both value and key with fatal UTF-8 handling, mapping decoder failures to `InvalidKafkaTrackOutcomeRecordError`.</comment>
<file context>
@@ -0,0 +1,140 @@
+
+ let input: unknown;
+ try {
+ input = JSON.parse(value.toString("utf8"));
+ } catch (cause) {
+ throw new InvalidKafkaTrackOutcomeRecordError({ cause });
</file context>
| }; | ||
|
|
||
| const stop = async (): Promise<void> => { | ||
| if (!isStarted || isStopped) return; |
There was a problem hiding this comment.
P2: When shutdown races with startup, stop() returns before the consumer is marked started and never stops or disconnects it. Track the in-flight startup and have stop() await or cancel it before performing shutdown.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At apps/balance-worker/src/kafka/kafkaTrackOutcomeConsumer.ts, line 139:
<comment>When shutdown races with startup, `stop()` returns before the consumer is marked started and never stops or disconnects it. Track the in-flight startup and have `stop()` await or cancel it before performing shutdown.</comment>
<file context>
@@ -0,0 +1,149 @@
+ };
+
+ const stop = async (): Promise<void> => {
+ if (!isStarted || isStopped) return;
+ isStopped = true;
+ try {
</file context>
| store: SqliteBalanceStateStore; | ||
| } => { | ||
| const directory = mkdtempSync(join(tmpdir(), "autumn-kafka-consumer-")); | ||
| const store = openSqliteBalanceStateStore({ |
There was a problem hiding this comment.
P3: If openSqliteBalanceStateStore throws (e.g., SQLite fails to open the database file), the try/catch never runs, so the freshly created temp directory from mkdtempSync is left behind and never removed. Move openSqliteBalanceStateStore inside the try so the error path also closes/cleans up, or wrap the whole body including the open call.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At apps/balance-worker/tests/unit/kafka/kafka-test-fixtures.ts, line 79:
<comment>If `openSqliteBalanceStateStore` throws (e.g., SQLite fails to open the database file), the `try`/`catch` never runs, so the freshly created temp directory from `mkdtempSync` is left behind and never removed. Move `openSqliteBalanceStateStore` inside the `try` so the error path also closes/cleans up, or wrap the whole body including the open call.</comment>
<file context>
@@ -0,0 +1,102 @@
+ store: SqliteBalanceStateStore;
+} => {
+ const directory = mkdtempSync(join(tmpdir(), "autumn-kafka-consumer-"));
+ const store = openSqliteBalanceStateStore({
+ databasePath: join(directory, "balance-state.sqlite"),
+ });
</file context>
![capy-ai[bot]](https://avatars.githubusercontent.com/in/1915919?s=60&v=4){kind=small}
|
|
||
| let isStarted = false; | ||
| let isStopped = false; | ||
| const initializedPartitions = new Set<string>(); |
There was a problem hiding this comment.
High: initializedPartitions never clears on reassignment, so SQLite-vs-group reconciliation runs only once per process lifetime.
Worker A processes a partition and marks it in initializedPartitions. After a rebalance the partition moves away (another member may advance the group offset) and later returns. The first delivered record skips the storedNextOffset check because the key is still present. applyDurableTrackOutcome then treats the gap as a valid Kafka offset hole, jumps next_offset forward, and either throws OutOfOrderTrackOutcomeError (crash loop) or applies a later outcome while silently skipping the missed range. Clear initializedPartitions on GROUP_JOIN/REBALANCING (or drop the set and compare every record against readNextOffset before folding).
| }; | ||
|
|
||
| const stop = async (): Promise<void> => { | ||
| if (!isStarted || isStopped) return; |
There was a problem hiding this comment.
Medium: stop() is a no-op while start() is still in flight, so shutdown can leave a live Kafka consumer running.
Caller invokes start() (connect/subscribe/run in progress, isStarted still false) and concurrently or immediately calls stop() on SIGTERM. stop() hits if (!isStarted || isStopped) return and exits without stop/disconnect. When start() finishes, the consumer is running with no shutdown hook holding it. Track the in-flight start promise and make stop() await or cancel it, then always stop and disconnect once the consumer is connected.
|
The latest updates on your projects. Learn more about Vercel for GitHub. 2 Skipped Deployments
|
1 participant
What this adds
This is PR 3 in the balance-worker stack. PR 1 computes and validates balance outcomes; PR 2 stores applied outcomes atomically in worker-local SQLite. This slice connects that state boundary to Kafka without routing live requests to it.
The producer ACK and consumer commit remain separate. Kafka first makes the outcome durable. This consumer only advances its group offset after the outcome is present in the worker's serving state.
Normal delivery
The KafkaJS adapter runs with
autoCommit: false. It parses the record, calls the PR 2 SQLite transaction, then commits the exactnextOffsetreturned by that transaction.Crash and restore behavior
If SQLite commits but the Kafka commit fails during a rebalance, the group redelivers the record. SQLite already holds the receipt and offset, so the retry does not deduct again; it repairs the Kafka offset and seeks forward.
A replacement worker may restore SQLite at offset 100 while the consumer group remembers offset 200. The first delivered record now reconciles the group back to SQLite's offset before any outcome is folded. After that one reconciliation, ordinary Kafka offset gaps remain valid because transactional control records are not delivered to
read_committedconsumers.Code
The consumer processes directly inside KafkaJS's per-partition callback, so this adds no second in-memory queue.
partitionsConsumedConcurrentlypermits concurrency across partitions while KafkaJS preserves ordering inside each partition.Scope
Included: versioned outcomes, customer keys, partition reconciliation, atomic fold-before-commit ordering, duplicate delivery, valid offset gaps, commit failure, malformed records, and graceful shutdown.
Not included:
/trackor/checkrouting, async command decisions, state bootstrap, S3 snapshots, Postgres projection, deployment, or production authority.Before an authoritative rollout, poison records need an explicit parking/operator-recovery policy, and per-outcome offset commits need a load test to decide whether to batch them.
Stack
Verification
bun install --frozen-lockfile --ignore-scriptsbun -F @autumn/balance-worker test:unit— 19 tests, 61 expectationsbun -F @autumn/balance-worker tsbun -F @autumn/balance-engine test:unit— 25 tests, 53 expectationsbun -F @autumn/balance-engine tsnode_modules/.bin/knip --workspace @autumn/balance-worker --no-progressgit diff --checkNo live Kafka broker was used in this slice; Kafka lifecycle behavior is covered with a fake KafkaJS consumer backed by the real SQLite store.
Summary by cubic
Adds a Kafka consumer that applies TrackOutcome records to the worker's SQLite store, committing the Kafka offset only after the SQLite transaction succeeds. No live
/trackor/checkrouting is wired yet; this is shadow consumption.Delivery and recovery
autoCommit: falseand commits once per batch, only after every record in that batch is folded into SQLite.StateBehindKafkaLogStartErrorwhen SQLite progress predates the Kafka log start instead of folding records that may have been trimmed.kafkajsas a dependency.Scope
Written for commit ca99739. Summary will update on new commits.