Re: Minimal logical decoding on standbys

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.
> 
> Overall design:
> 
> 1. We want to enable logical decoding on standbys, but replay of WAL
> from the primary might remove data that is needed by logical decoding,
> causing error(s) on the standby. To prevent those errors, a new replication
> conflict scenario needs to be addressed (as much as hot standby does).
> 
> 2. Our chosen strategy for dealing with this type of replication slot
> is to invalidate logical slots for which needed data has been removed.
> 
> 3. To do this we need the latestRemovedXid for each change, just as we
> do for physical replication conflicts, but we also need to know
> whether any particular change was to data that logical replication
> might access. That way, during WAL replay, we know when there is a risk of
> conflict and, if so, if there is a conflict.
> 
> 4. We can't rely on the standby's relcache entries for this purpose in
> any way, because the startup process can't access catalog contents.
> 
> 5. Therefore every WAL record that potentially removes data from the
> index or heap must carry a flag indicating whether or not it is one
> that might be accessed during logical decoding.
> 
> Why do we need this for logical decoding on standby?
> 
> First, let's forget about logical decoding on standby and recall that
> on a primary database, any catalog rows that may be needed by a logical
> decoding replication slot are not removed.
> 
> This is done thanks to the catalog_xmin associated with the logical
> replication slot.
> 
> But, with logical decoding on standby, in the following cases:
> 
> - hot_standby_feedback is off
> - hot_standby_feedback is on but there is no a physical slot between
>   the primary and the standby. Then, hot_standby_feedback will work,
>   but only while the connection is alive (for example a node restart
>   would break it)
> 
> Then, the primary may delete system catalog rows that could be needed
> by the logical decoding on the standby (as it does not know about the
> catalog_xmin on the standby).
> 
> So, it’s mandatory to identify those rows and invalidate the slots
> that may need them if any. Identifying those rows is the purpose of
> this commit.

This is a very nice commit message.


> Implementation:
> 
> When a WAL replay on standby indicates that a catalog table tuple is
> to be deleted by an xid that is greater than a logical slot's
> catalog_xmin, then that means the slot's catalog_xmin conflicts with
> the xid, and we need to handle the conflict. While subsequent commits
> will do the actual conflict handling, this commit adds a new field
> isCatalogRel in such WAL records (and a new bit set in the
> xl_heap_visible flags field), that is true for catalog tables, so as to
> arrange for conflict handling.
> 
> The affected WAL records are the ones that already contain the
> snapshotConflictHorizon field, namely:
> 
> - gistxlogDelete
> - gistxlogPageReuse
> - xl_hash_vacuum_one_page
> - xl_heap_prune
> - xl_heap_freeze_page
> - xl_heap_visible
> - xl_btree_reuse_page
> - xl_btree_delete
> - spgxlogVacuumRedirect
> 
> Due to this new field being added, xl_hash_vacuum_one_page and
> gistxlogDelete do now contain the offsets to be deleted as a
> FLEXIBLE_ARRAY_MEMBER. This is needed to ensure correct alignement.
> It's not needed on the others struct where isCatalogRel has
> been added.
> 
> Author: Andres Freund (in an older version), Amit Khandekar, Bertrand
> Drouvot

I think you're first author on this one by now.


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


> 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?



> @@ -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?



> @@ -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.


> @@ -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.



> @@ -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.


> @@ -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?


> +				}
>  
> -				(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?


> +				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...



> @@ -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?


> 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...



Greetings,

Andres Freund