diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index ecd11e9508..9ab91c3197 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -1392,14 +1392,22 @@ advance_conflict_slot_xmin(FullTransactionId new_xmin) * xmin of the replication slot. This handles the case where transaction * ID wraparound has occurred. */ +elog(LOG,"XXX: launcher new_xid= %u",next_xid); +elog(LOG,"XXX: launcher current slot xmin = %u",MyReplicationSlot->data.xmin); + if (next_xid < MyReplicationSlot->data.xmin) xmin_epoch--; full_xmin = FullTransactionIdFromEpochAndXid(xmin_epoch, MyReplicationSlot->data.xmin); +elog(LOG,"XXX: launcher full_xmin= %u",XidFromFullTransactionId(full_xmin)); if (FullTransactionIdPrecedesOrEquals(new_xmin, full_xmin)) + { + elog(LOG,"XXX: launcher returned without updating xmin!"); return false; + } + SpinLockAcquire(&MyReplicationSlot->mutex); MyReplicationSlot->data.xmin = XidFromFullTransactionId(new_xmin); @@ -1411,6 +1419,8 @@ advance_conflict_slot_xmin(FullTransactionId new_xmin) ReplicationSlotSave(); elog(DEBUG1, "updated xmin: %u", MyReplicationSlot->data.xmin); +elog(LOG,"XXX: launcher updated xmin = %u",MyReplicationSlot->data.xmin); + /* * Now the new xmin is safely on disk, we can let the global value * advance. We do not take ProcArrayLock or similar since we only advance diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index afd31e1525..d0d3b38433 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -4209,13 +4209,20 @@ wait_for_publisher_status(RetainConflictInfoData *data) if (!FullTransactionIdIsValid(data->last_phase_at)) data->last_phase_at = FullTransactionIdFromEpochAndXid(data->remote_epoch, data->remote_nextxid); - +elog(LOG,"XXX: last_phase_at = %u",XidFromFullTransactionId(data->last_phase_at)); +elog(LOG,"XXX: remote_oldestxid = %u",data->remote_oldestxid); +elog(LOG,"XXX: remote_nextxid = %u",data->remote_nextxid); /* Compute the epoch of the remote oldest running transaction ID */ if (data->remote_oldestxid > data->remote_nextxid) + { + elog(LOG,"XXX: xid WRAPAROUND happend on Publisher!!!"); remote_epoch--; + } + remote_full_xid = FullTransactionIdFromEpochAndXid(remote_epoch, data->remote_oldestxid); +elog(LOG,"XXX: remote_full_xid = %u",XidFromFullTransactionId(remote_full_xid)); /* * Check if all remote concurrent transactions that were active at the @@ -4225,7 +4232,10 @@ wait_for_publisher_status(RetainConflictInfoData *data) */ if (FullTransactionIdPrecedesOrEquals(data->last_phase_at, remote_full_xid)) + { data->phase = RCI_WAIT_FOR_LOCAL_FLUSH; + elog(LOG,"XXX: remote concurrent txn completed"); + } else data->phase = RCI_REQUEST_PUBLISHER_STATUS;