JITSU-157 bulker: make aborted-tx recovery in ensureTable possible on Postgres by absorbb · Pull Request #1440 · jitsucom/jitsu

absorbb · 2026-08-04T09:51:17Z

Fixes JITSU-157.

The bug

18 connections of one destination fan into a single public.events table, each consumer holding its own cached schema. When a new property appears they race the same ALTER TABLE ADD COLUMN; losers get 42701 column already exists.

ensureTable already handles that — it re-reads the schema from the db and patches again. On Postgres that recovery could never work: the failed ALTER aborts the whole transaction, so the recovery's get_table came back 25P02 and took the batch with it. The original error was swallowed, only the aftermath was logged, and the doomed batch ground on for 25–31h per batch, republishing ~500k messages to the retry topic, pushing offsets past retention and fencing the transactional producer.

The fix

1. Savepoints (tx_wrapper.go). Statements that are expected to sometimes lose a race — ADD COLUMN, and the PK statements alongside it — go through execIsolated, which wraps them in a savepoint and rolls back to it on failure. The transaction stays usable, so the existing recovery works and a lost race costs milliseconds instead of hours. Opt-in per dialect, currently Postgres only: MySQL and Snowflake don't abort the transaction, and Redshift/ClickHouse/BigQuery/DuckDB don't run these statements in one at all.

2. Recovery for every path, and the cause in the logs (table_helper.go). The re-read-and-retry was gated on actualSchema.Cached; it now runs for the non-cached path too. The first patch error is logged before the retry — until now nothing reported it.

Note this deviates slightly from the ticket's item 2 ("treat 42701 as success"). Swallowing it would keep the cached column type we tried to add, while the column that actually exists may have a different type. Going through the db re-read costs one extra SELECT and keeps the db the source of truth.

3. Backstop (batch_consumer.go). A batch that dies on 25P02 no longer sets nextBatch. The failure belongs to the destination, not to the messages, so every following batch fails identically — one failed batch per run is enough, the next run retries. This is what caps the retry amplification.

Test

TestConcurrentAddColumn drives the production sequence without concurrency: a batch stream reads the schema when it starts and patches it when it commits, so another instance adding the same column in that window is enough. With savepoints disabled it fails exactly the way production did — get_table with 25P02. TestExecIsolated pins both sides: a savepoint keeps the transaction usable, and without one the next statement is 25P02.

Full bulkerlib/implementations/sql suite green against Postgres.

Unrelated, please review separately

The last commit stops logging the Postgres destination password. It was logged verbatim at info level on every connect, so live customer database passwords are in Datadog and should be treated as compromised. Drop the commit if you'd rather track it as its own ticket.

🤖 Generated with Claude Code

…ry works Bulker instances writing to the same table each keep their own cached view of its schema, so when a new property appears they race to ALTER TABLE ADD COLUMN it. The losers get "column already exists" (42701), which ensureTable recovers from by re-reading the schema from the db and patching again. On Postgres that recovery could never work: a failed statement aborts the whole transaction, so the recovery query itself failed with 25P02 and took the batch with it. The real error was invisible - only the aftermath got logged - and the batch ground on for hours republishing every message to the retry topic. Statements that are expected to sometimes lose a race now run inside a savepoint that is rolled back on failure, which leaves the transaction usable. The first patch error is also logged now, so the actual cause is visible in production. JITSU-157

Drives the production sequence without concurrency: a batch stream reads the table schema when it starts and patches it when it commits, so another instance adding the same column in that window is enough to make its ALTER lose. With savepoints disabled the test fails exactly the way production did, on get_table with SQLSTATE 25P02. JITSU-157

…rted A batch that dies on an aborted transaction (25P02) fails for a reason that belongs to the destination, not to its messages, so every following batch is going to fail identically. Continuing to the next batch only republishes the whole backlog to the retry topic - a single 31h run scheduled half a million retries that way, pushed consumer offsets past retention and fenced the transactional producer. Backstop for the savepoint fix: one failed batch per run is enough, the next run retries. JITSU-157

The connection string was logged verbatim at info level on every connect, so live customer database passwords are sitting in Datadog. The connect string is now built twice - once for the driver, once with the password masked for the log. Found while investigating JITSU-157, unrelated to it. The already-logged credentials should be treated as compromised.

jitsu-code-review

Reviewed the Postgres transaction/savepoint fix, retry behavior in batch consumer, and the new concurrency regression tests. I left two inline questions on edge-case correctness around savepoint lifecycle and SQLSTATE detection breadth.

Keying it off the db type was wrong for exactly one adapter: Redshift classic is built by NewRedshiftClassic calling NewPostgres, so its SQLAdapterBase.typeId is "postgres" even though Redshift.Type() returns "redshift" - and Redshift has no SAVEPOINT. Nothing was broken in practice, because Redshift.OpenTx returns a DbWrapper with no transaction and savepoints need a real one. But the line that would have broken it sits commented out right below (//return p.openTx(ctx, p)), so the safety was an accident rather than a decision. The adapter now carries the flag: Postgres opts in, Redshift opts back out after inheriting it. Every other adapter leaves it at false. TestSavepointsAreOptedIn pins both ends. JITSU-157

jitsu-code-review

Reviewed the transaction/savepoint handling and batch retry flow changes in Bulker. I found a couple of non-blocking correctness concerns and left them inline: savepoint cleanup after rollback, and making aborted-transaction detection stricter to avoid accidental matches.

…25P02 match Both from PR review. ROLLBACK TO SAVEPOINT keeps the savepoint established, so every lost ALTER race left one behind for the rest of the transaction - and each attempt costs a subtransaction on the customer's database. Releasing on the rollback path too keeps at most one outstanding. The aborted-transaction check matched a bare "25P02", which can appear in an error for unrelated reasons: messages quote table names and event payloads. It now matches "SQLSTATE 25P02" as the driver renders it - still the code and not the message text, since Postgres localizes messages but never SQLSTATEs. JITSU-157

jitsu-code-review

Reviewed the Bulker transaction/savepoint changes, aborted-transaction handling, and related tests. The overall fix direction looks good, and I left one follow-up question inline about remaining credential redaction risk in connection logging.

…ting From PR review: masking the password wasn't enough, connection parameters are free-form and can carry one too (sslpassword, options, ...). Postgres now logs host:port, database and schema and nothing else. The redshift-data adapter had the same problem one step further: it logged the whole dsn, which carries sessionToken always, plus accessKeyId/secretAccessKey whenever Sanitize didn't clear them for IAM auth. Same treatment - cluster or workgroup, database, schema, region.

…e adapter Audit of the other destinations after the connection-logging review. ClickHouse was the only remaining one at risk: its dsn is built as proto://user:password@host/db and the line logging it was left commented out, so re-enabling it for a debugging session would have leaked passwords. Replaced with the same safe form as the others - hosts and database, logged at the call site where the destination id is known. MySQL and Snowflake build a dsn with credentials but never log it, DuckDB logs only the database name, and BigQuery has no connection string at all.

jitsu-code-review

Reviewed the transaction-isolation/savepoint changes in sql_adapter_base, tx_wrapper, table_helper, and adapter wiring (postgres, redshift, redshift_iam), plus the batch-consumer guard and new regression tests.

I did not find new correctness or security regressions in this patch set. The savepoint lifecycle handling and aborted-transaction guard look consistent with the intended recovery path for Postgres schema patch races.

jitsu-code-review

Reviewed the PR diff focused on transactional recovery, schema patch retry behavior, savepoint opt-in across adapters, and connection-string logging redaction.

I also checked the added regression tests around concurrent ADD COLUMN/savepoint behavior and the aborted-transaction batch backstop.

I didn’t find additional actionable bugs or security regressions in this change set beyond the existing open discussion threads.

…ement The isolation only has to hold for the patch as a whole: when it fails the caller re-reads the real schema and redoes the work from there, so rolling back the columns that did land costs nothing - it is about to redo them anyway. Per-statement savepoints made every schema change pay 2 extra round trips per column, and each savepoint is a subtransaction on the destination. Postgres degrades cluster-wide past 64 of them in one transaction, which a batch introducing that many new properties would have reached. Now a patch costs 2 extra statements regardless of how many columns it adds, and one subtransaction. Batches that don't change the schema are unaffected either way - patchTableIfNeeded returns before any of this when there is no diff. JITSU-157

jitsu-code-review

Reviewed the transaction-abort/savepoint handling changes in bulker plus the related tests and adapter wiring. Overall this looks solid; I left one inline question about error propagation when RELEASE SAVEPOINT fails so we do not potentially mask a broken transaction state.

From PR review. On the success path fn has just run, so the savepoint existed a statement ago: releasing it can only fail if the transaction itself is gone - connection dropped, context cancelled. That is not a leaked savepoint, it is a dead transaction, and the caller is about to run statements that depend on it. Returning the error reports it where it happened rather than somewhere further along with less context. The rollback path keeps logging its release failure: there the original error is the one worth returning. JITSU-157

jitsu-code-review

Reviewed the savepoint isolation changes, aborted-transaction batch handling, and connection logging redaction updates. Overall this looks solid; I left one inline question about a rollback-path savepoint edge case.

Second review round asking about the same asymmetry, so it belongs in the code.

jitsu-code-review

Reviewed the PR diff (batch-consumer aborted-transaction handling, savepoint-based schema patch isolation, adapter savepoint flags, and related tests). I did not find any new actionable bugs, security issues, or correctness regressions beyond the already-discussed review threads.

absorbb added 4 commits August 4, 2026 13:50