psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2025-02-07 21:00:33 +03:00
Code Issues Packages Projects Releases Wiki Activity

Don't advance replication cursor when the message wasn't confirmed

Browse Source 
Fixes https://github.com/psycopg/psycopg2/issues/940
This commit is contained in:
Alexander Kukushkin 2019-07-05 08:37:24 +02:00
parent 4d10f1235f
commit 9097a5b989
3 changed files with 7 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
psycopg/pqpath.c
View File

@ -1576,7 +1576,9 @@ retry:
/* We can safely forward flush_lsn to the wal_end from the server keepalive message /* We can safely forward flush_lsn to the wal_end from the server keepalive message
* if we know that the client already processed (confirmed) the last XLogData message */ * if we know that the client already processed (confirmed) the last XLogData message */
if (repl->flush_lsn >= repl->last_msg_data_start && wal_end > repl->flush_lsn) { if (repl->explicitly_flushed_lsn >= repl->last_msg_data_start
&& wal_end > repl->explicitly_flushed_lsn
&& wal_end > repl->flush_lsn) {
repl->flush_lsn = wal_end; repl->flush_lsn = wal_end;
} }

1
psycopg/replication_cursor.h
View File

@ -52,6 +52,7 @@ typedef struct replicationCursorObject {
XLogRecPtr last_msg_data_start; /* WAL pointer to the last non-keepalive message from the server */ XLogRecPtr last_msg_data_start; /* WAL pointer to the last non-keepalive message from the server */
struct timeval last_feedback; /* timestamp of the last feedback message to the server */ struct timeval last_feedback; /* timestamp of the last feedback message to the server */
XLogRecPtr explicitly_flushed_lsn; /* the flush LSN explicitly set by the send_feedback call */
} replicationCursorObject; } replicationCursorObject;

3
psycopg/replication_cursor_type.c
View File

@ -211,6 +211,9 @@ send_feedback(replicationCursorObject *self,
if (write_lsn > self->write_lsn) if (write_lsn > self->write_lsn)
self->write_lsn = write_lsn; self->write_lsn = write_lsn;
if (flush_lsn > self->explicitly_flushed_lsn)
self->explicitly_flushed_lsn = flush_lsn;
if (flush_lsn > self->flush_lsn) if (flush_lsn > self->flush_lsn)
self->flush_lsn = flush_lsn; self->flush_lsn = flush_lsn;