Re: Minimal logical decoding on standbys

Hi,

On 3/30/23 9:04 AM, Andres Freund wrote:
> Hi,
>
> On 2023-03-04 12:19:57 +0100, Drouvot, Bertrand wrote:
>> Subject: [PATCH v52 1/6] Add info in WAL records in preparation for logical
>> slot conflict handling.
>
> This is a very nice commit message.

Thanks! Melanie and Robert did provide great feedback/input to help make it
as it is now.

> I think this commit is ready to go. Unless somebody thinks differently, I
> think I might push it tomorrow.

Great! Once done, I'll submit a new patch so that GlobalVisTestFor() can make
use of the heap relation in vacuumRedirectAndPlaceholder() (which will be possible
once 0001 is committed).

>
>> Subject: [PATCH v52 2/6] Handle logical slot conflicts on standby.
>
>
>> @@ -6807,7 +6808,8 @@ CreateCheckPoint(int flags)
>> */
>> XLByteToSeg(RedoRecPtr, _logSegNo, wal_segment_size);
>> KeepLogSeg(recptr, &_logSegNo);
>> - if (InvalidateObsoleteReplicationSlots(_logSegNo))
>> + InvalidateObsoleteReplicationSlots(_logSegNo, &invalidated, InvalidOid, NULL);
>> + if (invalidated)
>> {
>> /*
>> * Some slots have been invalidated; recalculate the old-segment
>
> I don't really understand why you changed InvalidateObsoleteReplicationSlots
> to return void instead of bool, and then added an output boolean argument via
> a pointer?
>
>

I gave a second thought and it looks like I over complicated that part. I removed the
pointer parameter in V53 attached (and it now returns bool as before).

>
>> @@ -7964,6 +7968,22 @@ xlog_redo(XLogReaderState *record)
>> /* Update our copy of the parameters in pg_control */
>> memcpy(&xlrec, XLogRecGetData(record), sizeof(xl_parameter_change));
>>
>> + /*
>> + * Invalidate logical slots if we are in hot standby and the primary does not
>> + * have a WAL level sufficient for logical decoding. No need to search
>> + * for potentially conflicting logically slots if standby is running
>> + * with wal_level lower than logical, because in that case, we would
>> + * have either disallowed creation of logical slots or invalidated existing
>> + * ones.
>> + */
>> + if (InRecovery && InHotStandby &&
>> + xlrec.wal_level < WAL_LEVEL_LOGICAL &&
>> + wal_level >= WAL_LEVEL_LOGICAL)
>> + {
>> + TransactionId ConflictHorizon = InvalidTransactionId;
>> + InvalidateObsoleteReplicationSlots(InvalidXLogRecPtr, NULL, InvalidOid, &ConflictHorizon);
>> + }
>> +
>
> Are there races around changing wal_level?
>

Humm, not that I can think of right now. Do you have one/some in mind?

>
>> @@ -855,8 +855,10 @@ ReplicationSlotsComputeRequiredXmin(bool already_locked)
>> SpinLockAcquire(&s->mutex);
>> effective_xmin = s->effective_xmin;
>> effective_catalog_xmin = s->effective_catalog_xmin;
>> - invalidated = (!XLogRecPtrIsInvalid(s->data.invalidated_at) &&
>> - XLogRecPtrIsInvalid(s->data.restart_lsn));
>> + invalidated = ((!XLogRecPtrIsInvalid(s->data.invalidated_at) &&
>> + XLogRecPtrIsInvalid(s->data.restart_lsn))
>> + || (!TransactionIdIsValid(s->data.xmin) &&
>> + !TransactionIdIsValid(s->data.catalog_xmin)));
>> SpinLockRelease(&s->mutex);
>>
>> /* invalidated slots need not apply */
>
> I still would like a wrapper function to determine whether a slot has been
> invalidated. This This is too complicated to be repeated in other places.
>
>

Agree, so adding ObsoleteSlotIsInvalid() and SlotIsInvalid() in V53 attached.

ObsoleteSlotIsInvalid() could also be done in a dedicated patch outside this patch series, though.

>> @@ -1224,20 +1226,21 @@ ReplicationSlotReserveWal(void)
>> }
>>
>> /*
>> - * Helper for InvalidateObsoleteReplicationSlots -- acquires the given slot
>> - * and mark it invalid, if necessary and possible.
>> + * Helper for InvalidateObsoleteReplicationSlots
>> + *
>> + * Acquires the given slot and mark it invalid, if necessary and possible.
>> *
>> * Returns whether ReplicationSlotControlLock was released in the interim (and
>> * in that case we're not holding the lock at return, otherwise we are).
>> *
>> - * Sets *invalidated true if the slot was invalidated. (Untouched otherwise.)
>> + * Sets *invalidated true if an obsolete slot was invalidated. (Untouched otherwise.)
>> *
>> * This is inherently racy, because we release the LWLock
>> * for syscalls, so caller must restart if we return true.
>> */
>> static bool
>> -InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
>> - bool *invalidated)
>> +InvalidatePossiblyObsoleteOrConflictingLogicalSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
>> + bool *invalidated, TransactionId *xid)
>
> This is too long a name. I'd probably just leave it at the old name.
>
>

Done in V53 attached.

>
>> @@ -1261,18 +1267,33 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
>> * Check if the slot needs to be invalidated. If it needs to be
>> * invalidated, and is not currently acquired, acquire it and mark it
>> * as having been invalidated. We do this with the spinlock held to
>> - * avoid race conditions -- for example the restart_lsn could move
>> - * forward, or the slot could be dropped.
>> + * avoid race conditions -- for example the restart_lsn (or the
>> + * xmin(s) could) move forward or the slot could be dropped.
>> */
>> SpinLockAcquire(&s->mutex);
>>
>> restart_lsn = s->data.restart_lsn;
>> + slot_xmin = s->data.xmin;
>> + slot_catalog_xmin = s->data.catalog_xmin;
>> +
>> + /* slot has been invalidated (logical decoding conflict case) */
>> + if ((xid &&
>> + ((LogicalReplicationSlotIsInvalid(s))
>> + ||
>>
>
> Uh, huh?
>
> That's very odd formatting.
>
>> /*
>> - * If the slot is already invalid or is fresh enough, we don't need to
>> - * do anything.
>> + * We are not forcing for invalidation because the xid is valid and
>> + * this is a non conflicting slot.
>> */
>> - if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn >= oldestLSN)
>> + (TransactionIdIsValid(*xid) && !(
>> + (TransactionIdIsValid(slot_xmin) && TransactionIdPrecedesOrEquals(slot_xmin, *xid))
>> + ||
>> + (TransactionIdIsValid(slot_catalog_xmin) && TransactionIdPrecedesOrEquals(slot_catalog_xmin, *xid))
>> + ))
>> + ))
>> + ||
>> + /* slot has been invalidated (obsolete LSN case) */
>> + (!xid && (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn >= oldestLSN)))
>> {
>> SpinLockRelease(&s->mutex);
>> if (released_lock)
>
>
> This needs some cleanup.

Added a new macro LogicalReplicationSlotXidsConflict() and reformatted a bit.
Also ran pgindent on it, hope it's cleaner now.

>
>
>> @@ -1292,9 +1313,16 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
>> {
>> MyReplicationSlot = s;
>> s->active_pid = MyProcPid;
>> - s->data.invalidated_at = restart_lsn;
>> - s->data.restart_lsn = InvalidXLogRecPtr;
>> -
>> + if (xid)
>> + {
>> + s->data.xmin = InvalidTransactionId;
>> + s->data.catalog_xmin = InvalidTransactionId;
>> + }
>> + else
>> + {
>> + s->data.invalidated_at = restart_lsn;
>> + s->data.restart_lsn = InvalidXLogRecPtr;
>> + }
>> /* Let caller know */
>> *invalidated = true;
>> }
>> @@ -1327,15 +1355,39 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
>> */
>> if (last_signaled_pid != active_pid)
>> {
>> - ereport(LOG,
>> - errmsg("terminating process %d to release replication slot \"%s\"",
>> - active_pid, NameStr(slotname)),
>> - errdetail("The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
>> - LSN_FORMAT_ARGS(restart_lsn),
>> - (unsigned long long) (oldestLSN - restart_lsn)),
>> - errhint("You might need to increase max_slot_wal_keep_size."));
>> + if (xid)
>> + {
>> + if (TransactionIdIsValid(*xid))
>> + {
>> + ereport(LOG,
>> + errmsg("terminating process %d because replication slot \"%s\" conflicts with recovery",
>> + active_pid, NameStr(slotname)),
>> + errdetail("The slot conflicted with xid horizon %u.",
>> + *xid));
>> + }
>> + else
>> + {
>> + ereport(LOG,
>> + errmsg("terminating process %d because replication slot \"%s\" conflicts with recovery",
>> + active_pid, NameStr(slotname)),
>> + errdetail("Logical decoding on standby requires wal_level to be at least logical on the primary server"));
>> + }
>> +
>> + (void) SendProcSignal(active_pid, PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT, InvalidBackendId);
>> + }
>> + else
>> + {
>> + ereport(LOG,
>> + errmsg("terminating process %d to release replication slot \"%s\"",
>> + active_pid, NameStr(slotname)),
>> + errdetail("The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
>> + LSN_FORMAT_ARGS(restart_lsn),
>> + (unsigned long long) (oldestLSN - restart_lsn)),
>> + errhint("You might need to increase max_slot_wal_keep_size."));
>> +
>> + (void) kill(active_pid, SIGTERM);
>
> I think it ought be possible to deduplicate this a fair bit. For one, two of
> the errmsg()s above are identical. But I think this could be consolidated
> further, e.g. by using the same message style for the three cases, and passing
> in a separately translated reason for the termination?
>

deduplication done in V53 so that there is a single ereport() call.
I'm not sure the translation is fine the way I did it, please advice if that's not right.

>
>> + }
>>
>> - (void) kill(active_pid, SIGTERM);
>> last_signaled_pid = active_pid;
>> }
>>
>> @@ -1369,13 +1421,33 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
>> ReplicationSlotSave();
>> ReplicationSlotRelease();
>>
>> - ereport(LOG,
>> - errmsg("invalidating obsolete replication slot \"%s\"",
>> - NameStr(slotname)),
>> - errdetail("The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
>> - LSN_FORMAT_ARGS(restart_lsn),
>> - (unsigned long long) (oldestLSN - restart_lsn)),
>> - errhint("You might need to increase max_slot_wal_keep_size."));
>> + if (xid)
>> + {
>> + pgstat_drop_replslot(s);
>
> Why is this done here now?
>
>

Oops, moved above the if() in V53.

>> + if (TransactionIdIsValid(*xid))
>> + {
>> + ereport(LOG,
>> + errmsg("invalidating slot \"%s\" because it conflicts with recovery", NameStr(slotname)),
>> + errdetail("The slot conflicted with xid horizon %u.", *xid));
>> + }
>> + else
>> + {
>> + ereport(LOG,
>> + errmsg("invalidating slot \"%s\" because it conflicts with recovery", NameStr(slotname)),
>> + errdetail("Logical decoding on standby requires wal_level to be at least logical on the primary server"));
>> + }
>> + }
>> + else
>> + {
>> + ereport(LOG,
>> + errmsg("invalidating obsolete replication slot \"%s\"",
>> + NameStr(slotname)),
>> + errdetail("The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
>> + LSN_FORMAT_ARGS(restart_lsn),
>> + (unsigned long long) (oldestLSN - restart_lsn)),
>> + errhint("You might need to increase max_slot_wal_keep_size."));
>> + }
>>
>
> I don't like all these repeated elogs...

deduplication done in V53 so that there is a single ereport() call.
I'm not sure the translation is fine the way I did it, please advice if that's not right.

>
>
>
>> @@ -3057,6 +3060,27 @@ RecoveryConflictInterrupt(ProcSignalReason reason)
>> case PROCSIG_RECOVERY_CONFLICT_LOCK:
>> case PROCSIG_RECOVERY_CONFLICT_TABLESPACE:
>> case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
>> + case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
>> +
>> + /*
>> + * For conflicts that require a logical slot to be
>> + * invalidated, the requirement is for the signal receiver to
>> + * release the slot, so that it could be invalidated by the
>> + * signal sender. So for normal backends, the transaction
>> + * should be aborted, just like for other recovery conflicts.
>> + * But if it's walsender on standby, we don't want to go
>> + * through the following IsTransactionOrTransactionBlock()
>> + * check, so break here.
>> + */
>> + if (am_cascading_walsender &&
>> + reason == PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT &&
>> + MyReplicationSlot && SlotIsLogical(MyReplicationSlot))
>> + {
>> + RecoveryConflictPending = true;
>> + QueryCancelPending = true;
>> + InterruptPending = true;
>> + break;
>> + }
>>
>> /*
>> * If we aren't in a transaction any longer then ignore.
>
> I can't see any reason for this to be mixed into the same case "body" as LOCK
> etc?
>

Oh right, nice catch. I don't know how it ended up done that way. Fixed in V53.

>
>> diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
>> index 38c6f18886..290d4b45f4 100644
>> --- a/src/backend/replication/slot.c
>> +++ b/src/backend/replication/slot.c
>> @@ -51,6 +51,7 @@
>> #include "storage/proc.h"
>> #include "storage/procarray.h"
>> #include "utils/builtins.h"
>> +#include "access/xlogrecovery.h"
>
> Add new includes in the "alphabetically" right place...

Fixed in 0003 in V53 and the other places (aka other sub-patches) where it was needed.

Regards,

--
Bertrand Drouvot
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com