Fix WAL segment fd leaks with the WAL reading facility (xlogreader.c/h)

In most of the backend code, XLogReaders open a WAL segment file with
BasicOpenFile() (xlogutils.c, WAL summarizer and WAL sender).  On ERROR,
the opened file descriptor could leak while reading WAL segments, piling
them over time if a session can spawn XLogReaders at will, leading to
EMFILE with various consequences, the worst being a backend-side PANIC.

The problem has been reported as reachable with pg_walinspect (EXECUTE
privilege revoked from PUBLIC by default), or replication slot functions
(REPLICATION privilege required).

XLogReaderState gains a memory context reset callback, registered if a
segment_open callback has the idea to open a WAL segment with
BasicOpenFile().  XLogReaderRegisterResetCallback() can be used by a
segment_open callback to ensure the cleanup of any opened file
descriptor.

No backpatch is done.  This breaks the ABI of XLogReaderState (sizeof),
and MemoryContextUnregisterResetCallback() is new in v19.  There may be
an argument for fixing this issue in v19, but the release is close
enough that this change does not seem worth taking a risk for.

Author: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
Reviewed-by: Bertrand Drouvot <bertranddrouvot.pg@gmail.com>
Reviewed-by: Sami Imseih <samimseih.pg@gmail.com>
Reviewed-by: Chao Li <li.evan.chao@gmail.com>
Reviewed-by: Michael Paquier <michael@paquier.xyz>
Discussion: https://postgr.es/m/CALj2ACVwDuOXXDjj2cVdnTKoxsgTSLDin4XoL63AnM6aUgMQaA@mail.gmail.com

src/backend/access/transam/xlogreader.c

@@ -56,9 +56,12 @@ static bool ValidXLogRecord(XLogReaderState *state, XLogRecord *record,
 static void ResetDecoder(XLogReaderState *state);
 static void WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt,
 							   int segsize, const char *waldir);
-#if defined(USE_ZSTD) && !defined(FRONTEND)
+#ifndef FRONTEND
+static void xlogreader_memory_context_reset_cb(void *arg);
+#ifdef USE_ZSTD
 static void XLogReaderFreeZstdContext(void *arg);
 #endif
+#endif
 
 /* size of the buffer allocated for error message. */
 #define MAX_ERRORMSG_LEN 1000
@@ -166,6 +169,12 @@ XLogReaderAllocate(int wal_segment_size, const char *waldir,
 void
 XLogReaderFree(XLogReaderState *state)
 {
+#ifndef FRONTEND
+	if (state->reset_cb_registered)
+		MemoryContextUnregisterResetCallback(GetMemoryChunkContext(state),
+											 &state->reset_cb);
+#endif
+
 	if (state->seg.ws_file != -1)
 		state->routine.segment_close(state);
 
@@ -189,7 +198,43 @@ XLogReaderFree(XLogReaderState *state)
 	pfree(state);
 }
 
-#if defined(USE_ZSTD) && !defined(FRONTEND)
+#ifndef FRONTEND
+
+/*
+ * Memory context reset callback for an XLogReader.
+ */
+static void
+xlogreader_memory_context_reset_cb(void *arg)
+{
+	XLogReaderState *state = (XLogReaderState *) arg;
+
+	/* Close the WAL segment file that was left open */
+	if (state->seg.ws_file >= 0)
+		state->routine.segment_close(state);
+}
+
+/*
+ * Register a memory reset callback, closing a segment, if necessary.
+ *
+ * This is useful when opening a WAL segment file with BasicOpenFile(), which
+ * doesn't have the automatic file descriptor cleanup that OpenTransientFile()
+ * provides, to guarantee that the segment is closed before XLogReaderFree() is
+ * reached (on ERRORs, for example).
+ */
+void
+XLogReaderRegisterResetCallback(XLogReaderState *state)
+{
+	if (state->reset_cb_registered)
+		return;
+
+	state->reset_cb.func = xlogreader_memory_context_reset_cb;
+	state->reset_cb.arg = state;
+	MemoryContextRegisterResetCallback(GetMemoryChunkContext(state),
+									   &state->reset_cb);
+	state->reset_cb_registered = true;
+}
+
+#ifdef USE_ZSTD
 /* Release the zstd decomoression context. */
 static void
 XLogReaderFreeZstdContext(void *arg)
@@ -198,6 +243,8 @@ XLogReaderFreeZstdContext(void *arg)
 }
 #endif
 
+#endif							/* FRONTEND */
+
 /*
  * Allocate readRecordBuf to fit a record of at least the given length.
  *

src/backend/access/transam/xlogutils.c

@@ -836,7 +836,10 @@ wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo,
 	XLogFilePath(path, tli, nextSegNo, state->segcxt.ws_segsize);
 	state->seg.ws_file = BasicOpenFile(path, O_RDONLY | PG_BINARY);
 	if (state->seg.ws_file >= 0)
+	{
+		XLogReaderRegisterResetCallback(state);
 		return;
+	}
 
 	if (errno == ENOENT)
 		ereport(ERROR,

src/backend/postmaster/walsummarizer.c

@@ -1608,6 +1608,7 @@ summarizer_wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo,
 		state->seg.ws_file = BasicOpenFile(path, O_RDONLY | PG_BINARY);
 		if (state->seg.ws_file >= 0)
 		{
+			XLogReaderRegisterResetCallback(state);
 			*tli_p = tli;
 			return;
 		}

src/backend/replication/walsender.c

@@ -3346,7 +3346,10 @@ WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo,
 	XLogFilePath(path, *tli_p, nextSegNo, state->segcxt.ws_segsize);
 	state->seg.ws_file = BasicOpenFile(path, O_RDONLY | PG_BINARY);
 	if (state->seg.ws_file >= 0)
+	{
+		XLogReaderRegisterResetCallback(state);
 		return;
+	}
 
 	/*
 	 * If the file is not found, assume it's because the standby asked for a

src/include/access/xlogreader.h

@@ -109,7 +109,9 @@ typedef struct XLogReaderRoutine
 
 	/*
 	 * WAL segment close callback.  ->seg.ws_file shall be set to a negative
-	 * number.
+	 * number.  This shall not raise an error, as it may be called in a memory
+	 * context reset callback if XLogReaderRegisterResetCallback() has been
+	 * used.
 	 */
 	WALSegmentCloseCB segment_close;
 } XLogReaderRoutine;
@@ -324,6 +326,12 @@ struct XLogReaderState
 	 * data.
 	 */
 	bool		nonblocking;
+
+#ifndef FRONTEND
+	/* Reset callback for the memory context holding this reader. */
+	MemoryContextCallback reset_cb;
+	bool		reset_cb_registered;
+#endif
 };
 
 /*
@@ -344,6 +352,11 @@ extern XLogReaderState *XLogReaderAllocate(int wal_segment_size,
 /* Free an XLogReader */
 extern void XLogReaderFree(XLogReaderState *state);
 
+#ifndef FRONTEND
+/* Register a memory context reset callback */
+extern void XLogReaderRegisterResetCallback(XLogReaderState *state);
+#endif
+
 /* Optionally provide a circular decoding buffer to allow readahead. */
 extern void XLogReaderSetDecodeBuffer(XLogReaderState *state,
 									  void *buffer,