diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 2180a38063..a142855bd3 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -1454,6 +1454,11 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, { int last_signaled_pid = 0; bool released_lock = false; + bool terminated = false; + XLogRecPtr initial_effective_xmin = InvalidXLogRecPtr; + XLogRecPtr initial_catalog_effective_xmin = InvalidXLogRecPtr; + XLogRecPtr initial_restart_lsn = InvalidXLogRecPtr; + ReplicationSlotInvalidationCause conflict_prev PG_USED_FOR_ASSERTS_ONLY = RS_INVAL_NONE; for (;;) { @@ -1488,11 +1493,24 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, */ if (s->data.invalidated == RS_INVAL_NONE) { + /* + * The slot's mutex will be released soon, and it is possible that + * those values change since the process holding the slot has been + * terminated (if any), so record them here to ensure that we + * would report the correct conflict cause. + */ + if (!terminated) + { + initial_restart_lsn = s->data.restart_lsn; + initial_effective_xmin = s->effective_xmin; + initial_catalog_effective_xmin = s->effective_catalog_xmin; + } + switch (cause) { case RS_INVAL_WAL_REMOVED: - if (s->data.restart_lsn != InvalidXLogRecPtr && - s->data.restart_lsn < oldestLSN) + if (initial_restart_lsn != InvalidXLogRecPtr && + initial_restart_lsn < oldestLSN) conflict = cause; break; case RS_INVAL_HORIZON: @@ -1501,12 +1519,12 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, /* invalid DB oid signals a shared relation */ if (dboid != InvalidOid && dboid != s->data.database) break; - if (TransactionIdIsValid(s->effective_xmin) && - TransactionIdPrecedesOrEquals(s->effective_xmin, + if (TransactionIdIsValid(initial_effective_xmin) && + TransactionIdPrecedesOrEquals(initial_effective_xmin, snapshotConflictHorizon)) conflict = cause; - else if (TransactionIdIsValid(s->effective_catalog_xmin) && - TransactionIdPrecedesOrEquals(s->effective_catalog_xmin, + else if (TransactionIdIsValid(initial_catalog_effective_xmin) && + TransactionIdPrecedesOrEquals(initial_catalog_effective_xmin, snapshotConflictHorizon)) conflict = cause; break; @@ -1519,6 +1537,13 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, } } + /* + * The conflict cause recorded previously should not change while the + * process owning the slot (if any) has been terminated. + */ + Assert(!(conflict_prev != RS_INVAL_NONE && terminated && + conflict_prev != conflict)); + /* if there's no conflict, we're done */ if (conflict == RS_INVAL_NONE) { @@ -1601,6 +1626,8 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, (void) kill(active_pid, SIGTERM); last_signaled_pid = active_pid; + terminated = true; + conflict_prev = conflict; } /* Wait until the slot is released. */