fix(mysql): avoid unsafe inactive offset checkpoints

[dtunikov]

Problem

We always advance the offset when there is no records and we've got a timeout from MySQL binlog reader. It's wrong, because there could be schema changes that we haven't persisted yet.

Solution

• advance offset when there is no activity on the pipe other than DDLs ONLY after we applied DDLs to destination
• avoid persisting an inactive MySQL offset while a transaction is in progress (not directly related to the problem, but still a valid fix as it doesn't make sense to advance the offset while tx is in progress)

[ilidemi]

Reviewing something like this requires rebuilding a full mental model of how pull-sync-normalize works, and when are batches considered done to checkpoint on the source vs in PeerDB bookkeeping. This is at the core of CDC processing so needs to be approached carefully and holistically. Could you add a refresher/description for before vs after into the PR description?

[jgao54]

Thinking about this bug:

• postgres is safe because DDLs get buffered on the server-side and only get delivered to us with the next corresponding change event
• mongo is safe because it doesn't have to worry about DDLs

For mysql, in the happy path, we never close the pullRecord batch until we receive at least one message, however if there is a context cancellation caused by an error or user pausing the pipe, it certainly sounds feasible that we could lose the DDL as we'd have advanced the offset.

So the bug is specific to mysql. We should add an e2e test to reproduce the bug, example: execute add column ddl -> wait for sync interval to pass -> pause the pipe -> resume the pipe -> insert a row with the added column -> check if the added column got persisted to clickhouse.

And given this edge case only applies to MySQL, i wonder if we should handle the offset progression inside mysql's PullRecord instead (similar to how we progress the no-record scenario). (Edit: nvm, i mis-remembered here, both ReplayTableSchemaDeltas to clickhouse and applying it to catalog happen at the end of an empty batch in the shared code, so offset should also happen here as you have implemented.