Save slot's restart_lsn when invalidated due to size

We put it aside as invalidated_at, which let us show "lost" in
pg_replication slot.  Prior to this change, the state value was reported
as NULL.

Backpatch to 13.

Author: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Reviewed-by: Álvaro Herrera <alvherre@alvh.no-ip.org>
Discussion: https://postgr.es/m/20200617.101707.1735599255100002667.horikyota.ntt@gmail.com
Discussion: https://postgr.es/m/20200407.120905.1507671100168805403.horikyota.ntt@gmail.com

Author: alvherre

src/backend/replication/slot.c

@@ -1226,6 +1226,7 @@ InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno)
 						(uint32) restart_lsn)));
 
 		SpinLockAcquire(&s->mutex);
+		s->data.invalidated_at = s->data.restart_lsn;
 		s->data.restart_lsn = InvalidXLogRecPtr;
 		SpinLockRelease(&s->mutex);
 		ReplicationSlotRelease();

src/backend/replication/slotfuncs.c

@@ -283,6 +283,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		bool		nulls[PG_GET_REPLICATION_SLOTS_COLS];
 		WALAvailability walstate;
 		XLogSegNo	last_removed_seg;
+		XLogRecPtr	targetLSN;
 		int			i;
 
 		if (!slot->in_use)
@@ -342,7 +343,15 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 		else
 			nulls[i++] = true;
 
-		walstate = GetWALAvailability(slot_contents.data.restart_lsn);
+		/*
+		 * Report availability from invalidated_at when the slot has been
+		 * invalidated; otherwise slots would appear as invalid without any
+		 * more clues as to what happened.
+		 */
+		targetLSN = XLogRecPtrIsInvalid(slot_contents.data.restart_lsn) ?
+			slot_contents.data.invalidated_at :
+			slot_contents.data.restart_lsn;
+		walstate = GetWALAvailability(targetLSN);
 
 		switch (walstate)
 		{

src/include/access/xlog.h

@@ -326,7 +326,7 @@ extern void ShutdownXLOG(int code, Datum arg);
 extern void InitXLOGAccess(void);
 extern void CreateCheckPoint(int flags);
 extern bool CreateRestartPoint(int flags);
-extern WALAvailability GetWALAvailability(XLogRecPtr restart_lsn);
+extern WALAvailability GetWALAvailability(XLogRecPtr targetLSN);
 extern XLogRecPtr CalculateMaxmumSafeLSN(void);
 extern void XLogPutNextOid(Oid nextOid);
 extern XLogRecPtr XLogRestorePoint(const char *rpName);

src/include/replication/slot.h

@@ -79,6 +79,9 @@ typedef struct ReplicationSlotPersistentData
 	/* oldest LSN that might be required by this replication slot */
 	XLogRecPtr	restart_lsn;
 
+	/* restart_lsn is copied here when the slot is invalidated */
+	XLogRecPtr	invalidated_at;
+
 	/*
 	 * Oldest LSN that the client has acked receipt for.  This is used as the
 	 * start_lsn point in case the client doesn't specify one, and also as a

src/test/recovery/t/019_replslot_limit.pl

@@ -186,7 +186,7 @@
 $result = $node_master->safe_psql('postgres',
 	"SELECT slot_name, active, restart_lsn IS NULL, wal_status, min_safe_lsn FROM pg_replication_slots WHERE slot_name = 'rep1'"
 );
-is($result, "rep1|f|t||", 'check that the slot became inactive');
+is($result, "rep1|f|t|lost|", 'check that the slot became inactive');
 
 # The standby no longer can connect to the master
 $logstart = get_log_size($node_standby);