database/postgres
database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-05-03 22:24:49 +03:00
Code

Fix remote position tracking in logical replication

Browse Source 
We need to set the origin remote position to end_lsn, not commit_lsn, as
commit_lsn is the start of commit record, and we use the origin remote
position as start position when restarting replication stream.  If we'd
use commit_lsn, we could request data that we already received from the
remote server after a crash of a downstream server.

Author: Petr Jelinek <petr.jelinek@2ndquadrant.com>
This commit is contained in:
Peter Eisentraut 2017-04-04 08:24:32 -04:00
parent b38006ef6d
commit fe7bbc4ddb
1 changed files with 7 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
src/backend/replication/logical/worker.c
View File

@ -421,9 +421,6 @@ apply_handle_begin(StringInfo s)
logicalrep_read_begin(s, &begin_data); logicalrep_read_begin(s, &begin_data);
replorigin_session_origin_timestamp = begin_data.committime;
replorigin_session_origin_lsn = begin_data.final_lsn;
remote_final_lsn = begin_data.final_lsn; remote_final_lsn = begin_data.final_lsn;
in_remote_transaction = true; in_remote_transaction = true;
@ -443,14 +440,18 @@ apply_handle_commit(StringInfo s)
logicalrep_read_commit(s, &commit_data); logicalrep_read_commit(s, &commit_data);
Assert(commit_data.commit_lsn == replorigin_session_origin_lsn);
Assert(commit_data.committime == replorigin_session_origin_timestamp);
Assert(commit_data.commit_lsn == remote_final_lsn); Assert(commit_data.commit_lsn == remote_final_lsn);
/* The synchronization worker runs in single transaction. */ /* The synchronization worker runs in single transaction. */
if (IsTransactionState() && !am_tablesync_worker()) if (IsTransactionState() && !am_tablesync_worker())
{ {
/*
* Update origin state so we can restart streaming from correct
* position in case of crash.
*/
replorigin_session_origin_lsn = commit_data.end_lsn;
replorigin_session_origin_timestamp = commit_data.committime;
CommitTransactionCommand(); CommitTransactionCommand();
store_flush_position(commit_data.end_lsn); store_flush_position(commit_data.end_lsn);