Repository navigation
Check the stored cursor before each SQL sink flush (PostgreSQL) - #1012
Conversation
sduchesneau
left a comment
sduchesneau
left a comment
There was a problem hiding this comment.
Good, and performance impact does not seem too bad since it's just one per flush.
I just dislike the 'setExpectedCursor' inside a Get, please take it out and make it explicit from the SQLSinker.Run after it loads the cursor.
Other than that, good to go when rebased.
| activeCursor, found := cursors[outputModuleHash] | ||
| if found { | ||
| return activeCursor, false, err | ||
| l.setExpectedCursor(rows[outputModuleHash].Cursor) |
There was a problem hiding this comment.
setting this here is not pretty (side-effect of activating a check inside a "Get"...)
add the rawCursor to the output of GetCursor and call it like this in sinker.go:
cursor, rawCursor, mismatchDetected, err := s.loader.GetCursor(ctx, s.OutputModuleHash())
// error checking (...)
s.loader.SetExpectedCursor(rawCursor)
There was a problem hiding this comment.
Done: GetCursor now returns the stored cursor as is (rawCursor), and SQLSinker.Run calls SetExpectedCursor once it has read or inserted the cursor. InsertCursor no longer sets it either, so the loader never turns the check on by itself.
The call sits after the two InsertCursor branches rather than right after the error check: on a module hash mismatch, GetCursor returns no rawCursor (the row found belongs to another module) and Run inserts cursor.String() under its own module, so that branch sets rawCursor to what it wrote.
|
I also noted that this does not cover the 'from-proto' sink. that sink has more interfaces that were added when optimizing for (insert-only) performance, so it might be trickier to add there -- so I'm fine with this PR only covering the 'databaseChanges' usecase/codepath. |
The final flush when the stream ends (stopped, or at the stop block) treated the stream's latest block as final. It pruned the undo history of blocks that were still reversible, so a reorg across a sink restart left orphaned rows in the database, and with an undo buffer it saved a cursor past the blocks still in the buffer, which were then skipped. It now writes the handled blocks with the last handled block's cursor and the stream's final block height, writes nothing if a block failed to be handled, and the CLI waits for it before exiting and reports a failure that happens while stopping. Sink library: a handler error wrapping io.EOF (a lost database connection, for example) ended the run as if the stop block was reached, so a live sink exited with status 0 and was not restarted on failure. A handler error is never taken for the end of the stream anymore. When a handler fails while the sinker is stopping, the completion handler no longer runs on top of the block it did not finish, and the error is reported unless the stop itself caused it. The completion handler gets the cursor of the last block the handler accepted; Sinker.LastReceivedCursor still gives the stream's. An undo whose transaction failed was not rolled back: the deferred rollback checked a different error variable.
Once the sink has read or written its cursor, each flush and revert
transaction starts by locking the sink's cursor row and checking it still
holds the cursor this sink last read or wrote:
SELECT 1 FROM cursors WHERE id = '<module>' AND cursor = '<last>' COLLATE "C" FOR UPDATE
Otherwise nothing is written and the sink stops with "the stored cursor is not
the one this sink last read or wrote", to be restarted from the stored cursor.
A conflict is not retried.
This stops:
- a flush retry applying a batch twice after a COMMIT whose reply was lost
(the retry replayed the same buffer, doubling ADD fields);
- a sink writing on top of a database that lost commits, after a restore or a
failover with asynchronous replication reached through a proxy, which left
a silent hole;
- a second sink of the same module writing to the same database.
Only the sink's own row is read and locked, and the cursor UPDATE stays the
last statement of the transaction, so triggers on the cursors table are
unaffected. ClickHouse is unchanged.
9862ad1 to
0063172
Compare
|
Rebased on develop (after #1007), along with #1011 and #1013 on top. Checked end to end with two sinks of one module resuming from the same cursor: one stops at its first flush with the cursor conflict, and with the new Agreed on from-proto: this only covers the DatabaseChanges path, as the changelog entry says. |
Builds on #1007 and #1011. Until those are merged, this PR also shows their commits.
The sink saves its cursor with a plain
UPDATE cursors SET … WHERE id = '<module>', without checking that the stored cursor is still the one it last wrote. So a few things can write on top of the wrong state:A flush retry can apply a batch twice. If a COMMIT goes through but its reply is lost (a restart, a network blip),
flushWithRetryreplays the same buffer in a new transaction. Inserts fail on duplicate keys, butADDfields are applied twice and the history records a wrong previous value.A failover can leave a silent hole. With asynchronous replication, the new primary can be missing the last commits. A sink connected through a proxy reconnects and keeps writing from where it was. In a test with a primary, an async replica and a proxy, the stock sink ended with 1,481 transactions missing and exited 0.
Two sinks of the same module overwrite each other's work and move the cursor back.
Fix (PostgreSQL): once the sink has read or written its cursor, each flush and revert transaction starts with
If no row matches, the transaction is rolled back before anything is written. The sink stops with
the stored cursor is not the one this sink last read or wrote, and on restart resumes from the stored cursor. A conflict is not retried. A concurrent writer waits on the row lock, then fails the same check.Only the sink's own row is read and locked (it already reads and updates that row). The cursor
UPDATEstays the last statement, so triggers on the cursors table see the same thing as before. ClickHouse is unchanged.Added
TestSinker_CursorCheck. With real Postgres (PG_DSN), added:TestCursorCheck_Postgres: a second writer, a replayed commit, other modules' rows, a restart;TestLostCommitReply_Postgres: a proxy in front of Postgres drops the reply to a COMMIT after the server applied it. The retry is refused and theADDisn't doubled.In the same failover test as above, the sink stops, replays from the new primary's cursor, and ends with tables identical to a clean run.