Subject: Re: Why is src/test/modules/committs/t/002_standby.pl flaky?

From: Tom Lane <[email-masked]>
Date: Sat, 08 Jan 2022 18:41:56 -0500

Andrew Dunstan <[email-masked]> writes:
> On 12/30/21 15:01, Thomas Munro wrote:
>> There's a wait for replay that is open coded (instead of using the
>> wait_for_catchup() routine), and sometimes the second of two such
>> waits at line 51 (in master) times out after 3 minutes with "standby
>> never caught up".  It's happening on three particular Windows boxes,
>> but once also happened on the AIX box "tern".

> FYI, drongo and fairywren are run on the same AWS/EC2 Windows Server
> 2019 instance. Nothing else runs on it.

I spent a little time looking into this just now.  There are similar
failures in both 002_standby.pl and 003_standby_2.pl, which is
unsurprising because there are essentially-identical test sequences
in both.  What I've realized is that the issue is triggered by
this sequence:

$standby->start;
...
$primary->restart;
$primary->safe_psql('postgres', 'checkpoint');
my $primary_lsn =
  $primary->safe_psql('postgres', 'select pg_current_wal_lsn()');
$standby->poll_query_until('postgres',
	qq{SELECT '$primary_lsn'::pg_lsn <= pg_last_wal_replay_lsn()})
  or die "standby never caught up";

(the failing poll_query_until is at line 51 in 002_standby.pl, or
line 37 in 003_standby_2.pl).  That is, we have forced a primary
restart since the standby first connected to the primary, and
now we have to wait for the standby to reconnect and catch up.

*These two tests seem to be the only TAP tests that do that*.
So I think there's not really anything specific to commit_ts testing
involved, it's just a dearth of primary restarts elsewhere.

Looking at the logs in the failing cases, there's no evidence
that the standby has even detected the primary's disconnection,
which explains why it hasn't attempted to reconnect.  For
example, in the most recent HEAD failure,

https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=fairywren&dt=2022-01-03%2018%3A04%3A41

the standby reports successful connection:

2022-01-03 18:58:04.920 UTC [179700:1] LOG:  started streaming WAL from primary at 0/3000000 on timeline 1

(which we can also see in the primary's log), but after that
there's no log traffic at all except the test script's vain
checks of pg_last_wal_replay_lsn().  In the same animal's
immediately preceding successful run,

https://buildfarm.postgresql.org/cgi-bin/show_stage_log.pl?nm=fairywren&dt=2022-01-03%2015%3A04%3A41&stg=module-commit_ts-check

we see:

2022-01-03 15:59:24.186 UTC [176664:1] LOG:  started streaming WAL from primary at 0/3000000 on timeline 1
2022-01-03 15:59:25.003 UTC [176664:2] LOG:  replication terminated by primary server
2022-01-03 15:59:25.003 UTC [176664:3] DETAIL:  End of WAL reached on timeline 1 at 0/3030CB8.
2022-01-03 15:59:25.003 UTC [176664:4] FATAL:  could not send end-of-streaming message to primary: server closed the connection unexpectedly
		This probably means the server terminated abnormally
		before or while processing the request.
	no COPY in progress
2022-01-03 15:59:25.005 UTC [177092:5] LOG:  invalid record length at 0/3030CB8: wanted 24, got 0
...
2022-01-03 15:59:25.564 UTC [177580:1] LOG:  started streaming WAL from primary at 0/3000000 on timeline 1

So for some reason, on these machines detection of walsender-initiated
connection close is unreliable ... or maybe, the walsender didn't close
the connection, but is somehow still hanging around?  Don't have much idea
where to dig beyond that, but maybe someone else will.  I wonder in
particular if this could be related to our recent discussions about
whether to use shutdown(2) on Windows --- could we need to do the
equivalent of 6051857fc/ed52c3707 on walsender connections?

			regards, tom lane

---

From: Tom Lane <[email-masked]>
Date: Sat, 08 Jan 2022 20:17:16 -0500

> equivalent of 6051857fc/ed52c3707 on walsender connections?

... wait a minute.  After some more study of the buildfarm logs,
it was brought home to me that these failures started happening
just after 6051857fc went in:

https://buildfarm.postgresql.org/cgi-bin/show_failures.pl?max_days=90&branch=&member=&stage=module-commit_tsCheck&filter=Submit

The oldest matching failure is jacana's on 2021-12-03.
(The above sweep finds an unrelated-looking failure on 2021-11-11,
but no others before 6051857fc went in on 2021-12-02.  Also, it
looks likely that ed52c3707 on 2021-12-07 made the failure more
probable, because jacana's is the only matching failure before 12-07.)

So I'm now thinking it's highly likely that those commits are
causing it somehow, but how?

			regards, tom lane

---

From: Alexander Lakhin <[email-masked]>
Date: Sun, 9 Jan 2022 14:00:00 +0300

Hello Tom,
> causing it somehow, but how?
I've managed to reproduce this failure too.
Removing "shutdown(MyProcPort->sock, SD_SEND);" doesn't help here, so
the culprit is exactly "closesocket(MyProcPort->sock);".
I've added `system("netstat -ano");` before die() in 002_standby.pl and see:
# Postmaster PID for node "primary" is 944
  Proto  Local Address          Foreign Address        State           PID
...
  TCP    127.0.0.1:58545        127.0.0.1:61995        FIN_WAIT_2      944
...
  TCP    127.0.0.1:61995        127.0.0.1:58545        CLOSE_WAIT      1352

(Replacing SD_SEND with SD_BOTH doesn't change the behaviour.)

Looking at the libpqwalreceiver.c:
        /* Now that we've consumed some input, try again */
        rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
here we get -1 on the primary disconnection.
Then we get COMMAND_OK here:
        res = libpqrcv_PQgetResult(conn->streamConn);
        if (PQresultStatus(res) == PGRES_COMMAND_OK)
and finally just hang at:
            /* Verify that there are no more results. */
            res = libpqrcv_PQgetResult(conn->streamConn);
until the standby gets interrupted by the TAP test. (That call can also
return NULL and then the test completes successfully.)
Going down through the call chain, I see that at the end of it
WaitForMultipleObjects() hangs while waiting for the primary connection
socket event. So it looks like the socket, that is closed by the
primary, can get into a state unsuitable for WaitForMultipleObjects().
I tried to check the socket state with the WSAPoll() function and
discovered that it returns POLLHUP for the "problematic" socket.
The following draft addition in latch.c:
int
WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock,
                  long timeout, uint32 wait_event_info)
{
    int            ret = 0;
    int            rc;
    WaitEvent    event;

#ifdef WIN32
    if (wakeEvents & WL_SOCKET_MASK) {
        WSAPOLLFD pollfd;
        pollfd.fd = sock;
        pollfd.events = POLLRDNORM | POLLWRNORM;
        pollfd.revents = 0;
        int rc = WSAPoll(&pollfd, 1, 0);
        if ((rc == 1) && (pollfd.revents & POLLHUP)) {
            elog(WARNING, "WaitLatchOrSocket: A stream-oriented
connection was either disconnected or aborted.");
            return WL_SOCKET_MASK;
        }
    }
#endif

makes the test 002_stanby.pl pass (100 of 100 iterations, while without
the fix I get failures roughly on each third). I'm not sure where to
place this check, maybe it's better to move it up to
libpqrcv_PQgetResult() to minimize it's footprint or to find less
Windows-specific approach, but I'd prefer a client-side fix anyway, as
graceful closing a socket by a server seems a legitimate action.

Best regards,
Alexander

---

From: Tom Lane <[email-masked]>
Date: Sun, 09 Jan 2022 11:49:30 -0500

> the culprit is exactly "closesocket(MyProcPort->sock);".

Ugh.  Did you try removing the closesocket and keeping shutdown?
I don't recall if we tried that combination before.

> ...  I'm not sure where to

What concerns me here is whether this implies that other clients
(libpq, jdbc, etc) are going to need changes as well.  Maybe
libpq is okay, because we've not seen failures of the isolation
tests that use pg_cancel_backend(), but still it's worrisome.
I'm not entirely sure whether the isolationtester would notice
that a connection that should have died didn't.

			regards, tom lane

---

From: Thomas Munro <[email-masked]>
Date: Mon, 10 Jan 2022 08:06:19 +1300

> primary, can get into a state unsuitable for WaitForMultipleObjects().

I wonder if FD_CLOSE is edge-triggered, and it's already told us once.
I think that's what these Python Twisted guys are saying:

https://stackoverflow.com/questions/7598936/how-can-a-disconnected-tcp-socket-be-reliably-detected-using-msgwaitformultipleo

> discovered that it returns POLLHUP for the "problematic" socket.

Good discovery.  I guess if the above theory is right, there's a
memory somewhere that makes this level-triggered as expected by users
of poll().

---

From: Thomas Munro <[email-masked]>
Date: Mon, 10 Jan 2022 15:00:19 +1300

> I wonder if FD_CLOSE is edge-triggered, and it's already told us once.

Can you reproduce it with this patch?
From d3cfff53911ce068d4a31fe7ac7933958936d1b0 Mon Sep 17 00:00:00 2001
From: Thomas Munro <[email-masked]>
Date: Mon, 10 Jan 2022 14:45:05 +1300
Subject: [PATCH] Make Windows FD_CLOSE reporting sticky.

XXX Just testing an idea...
---
 src/backend/storage/ipc/latch.c | 16 ++++++++++++++++
 src/include/storage/latch.h     |  1 +
 2 files changed, 17 insertions(+)

diff --git a/src/backend/storage/ipc/latch.c b/src/backend/storage/ipc/latch.c
index 1d893cf863..4a61b31cd5 100644
--- a/src/backend/storage/ipc/latch.c
+++ b/src/backend/storage/ipc/latch.c
@@ -899,6 +899,7 @@ AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch,
 	event->user_data = user_data;
 #ifdef WIN32
 	event->reset = false;
+	event->closed = false;
 #endif
 
 	if (events == WL_LATCH_SET)
@@ -1882,6 +1883,20 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
 				return 1;
 			}
 		}
+
+		/*
+		 * XXX: HYPOTHESIS TO TEST
+		 * Windows only reports FD_CLOSE once.  If we've seen that already,
+		 * continue to report it.
+		 */
+		if ((cur_event & WL_SOCKET_MASK) && cur_event->closed)
+		{
+			occurred_events->pos = cur_event->pos;
+			occurred_events->user_data = cur_event->user_data;
+			occurred_events->events = (cur_event->events & WL_SOCKET_MASK);
+			occurred_events->fd = cur_event->fd;
+			return 1;
+		}
 	}
 
 	/*
@@ -2002,6 +2017,7 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
 		{
 			/* EOF/error, so signal all caller-requested socket flags */
 			occurred_events->events |= (cur_event->events & WL_SOCKET_MASK);
+			cur_event->closed = true;
 		}
 
 		if (occurred_events->events != 0)
diff --git a/src/include/storage/latch.h b/src/include/storage/latch.h
index 44f9368c64..c24f46dc37 100644
--- a/src/include/storage/latch.h
+++ b/src/include/storage/latch.h
@@ -147,6 +147,7 @@ typedef struct WaitEvent
 	void	   *user_data;		/* pointer provided in AddWaitEventToSet */
 #ifdef WIN32
 	bool		reset;			/* Is reset of the event required? */
+	bool		closed;			/* Has FD_CLOSE event been reported? */
 #endif
 } WaitEvent;
 
-- 
2.33.1

---

From: Alexander Lakhin <[email-masked]>
Date: Mon, 10 Jan 2022 10:00:00 +0300

> Can you reproduce it with this patch?
Unfortunately, this fix (with the correction "(cur_event &
WL_SOCKET_MASK)" -> "(cur_event->events & WL_SOCKET_MASK") doesn't work,
>             res = libpqrcv_PQgetResult(conn->streamConn);
The libpqrcv_PQgetResult function, in turn, invokes WaitLatchOrSocket()
where WaitEvents are defined locally, and the closed flag set on the
first invocation but expected to be checked on second.
> I don't recall if we tried that combination before.
Even with shutdown() only I still observe WaitForMultipleObjects()
hanging (and WSAPoll() returns POLLHUP for the socket).

As to your concern regarding other clients, I suspect that this issue is
caused by libpqwalreceiver' specific call pattern and may be other
clients just don't do that. I need some more time to analyze this.

Best regards,
Alexander

---

From: Thomas Munro <[email-masked]>
Date: Mon, 10 Jan 2022 22:20:14 +1300

> first invocation but expected to be checked on second.

D'oh, right.  There's also a WaitLatchOrSocket call in walreceiver.c.
We'd need a long-lived WaitEventSet common across all of these sites,
which is hard here (because the socket might change under you, as
discussed in other threads that introduced long lived WaitEventSets to
other places but not here).

/me wonders if it's possible that graceful FD_CLOSE is reported only
once, but abortive/error FD_CLOSE is reported multiple times...

---

From: Thomas Munro <[email-masked]>
Date: Mon, 10 Jan 2022 22:40:27 +1300

> other places but not here).

This is super quick-and-dirty code (and doesn't handle some errors or
socket changes correctly), but does it detect the closed socket?
From 3590a8c9b3e8992a65dea7a9d6aecdca2f25582d Mon Sep 17 00:00:00 2001
From: Thomas Munro <[email-masked]>
Date: Mon, 10 Jan 2022 14:45:05 +1300
Subject: [PATCH v2 1/2] Make Windows FD_CLOSE reporting sticky.

index 1d893cf863..8f3176a00f 100644
 
+		if ((cur_event->events & WL_SOCKET_MASK) && cur_event->closed)
 
 	/*

-- 
2.33.1

From 755ee3da7934d33a706d59eb0abc0e4f1829acb0 Mon Sep 17 00:00:00 2001
From: Thomas Munro <[email-masked]>
Date: Mon, 10 Jan 2022 22:02:21 +1300
Subject: [PATCH v2 2/2] XXX Quick hack to use persistent WaitEventSet in
 walreceiver

XXX There are several things wrong with this, it's just experimental code.
---
 .../libpqwalreceiver/libpqwalreceiver.c       | 97 ++++++++++++-------
 src/backend/replication/walreceiver.c         | 12 +--
 src/include/replication/walreceiver.h         |  8 ++
 3 files changed, 75 insertions(+), 42 deletions(-)

diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index c08e599eef..f8dd76e2d0 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -40,6 +40,7 @@ void		_PG_init(void);
 
 struct WalReceiverConn
 {
+	WaitEventSet *wes;
 	/* Current connection to the primary, if any */
 	PGconn	   *streamConn;
 	/* Used to remember if the connection is logical or physical */
@@ -82,6 +83,7 @@ static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
 									   const int nRetTypes,
 									   const Oid *retTypes);
 static void libpqrcv_disconnect(WalReceiverConn *conn);
+static int libpqrcv_wait(WalReceiverConn *conn, int timeout, int wait_event);
 
 static WalReceiverFunctionsType PQWalReceiverFunctions = {
 	libpqrcv_connect,
@@ -98,12 +100,13 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
 	libpqrcv_create_slot,
 	libpqrcv_get_backend_pid,
 	libpqrcv_exec,
-	libpqrcv_disconnect
+	libpqrcv_disconnect,
+	libpqrcv_wait
 };
 
 /* Prototypes for private functions */
-static PGresult *libpqrcv_PQexec(PGconn *streamConn, const char *query);
-static PGresult *libpqrcv_PQgetResult(PGconn *streamConn);
+static PGresult *libpqrcv_PQexec(WalReceiverConn *conn, const char *query);
+static PGresult *libpqrcv_PQgetResult(WalReceiverConn *conn);
 static char *stringlist_to_identifierstr(PGconn *conn, List *strings);
 
 /*
@@ -182,6 +185,15 @@ libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
 		return NULL;
 	}
 
+	conn->wes = CreateWaitEventSet(TopMemoryContext, 3);
+	AddWaitEventToSet(conn->wes, WL_SOCKET_WRITEABLE,
+					  PQsocket(conn->streamConn), NULL, NULL);
+	AddWaitEventToSet(conn->wes, WL_LATCH_SET, PGINVALID_SOCKET,
+					  MyLatch, NULL);
+	AddWaitEventToSet(conn->wes, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET,
+					  NULL, NULL);
+
+
 	/*
 	 * Poll connection until we have OK or FAILED status.
 	 *
@@ -191,7 +203,7 @@ libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
 	do
 	{
 		int			io_flag;
-		int			rc;
+		WaitEvent	event;
 
 		if (status == PGRES_POLLING_READING)
 			io_flag = WL_SOCKET_READABLE;
@@ -203,21 +215,20 @@ libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
 		else
 			io_flag = WL_SOCKET_WRITEABLE;
 
-		rc = WaitLatchOrSocket(MyLatch,
-							   WL_EXIT_ON_PM_DEATH | WL_LATCH_SET | io_flag,
-							   PQsocket(conn->streamConn),
-							   0,
-							   WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
+		ModifyWaitEvent(conn->wes, 0, io_flag, NULL);
+		if (WaitEventSetWait(conn->wes, -1, &event, 1,
+							 WAIT_EVENT_LIBPQWALRECEIVER_CONNECT) < 1)
+			continue;
 
 		/* Interrupted? */
-		if (rc & WL_LATCH_SET)
+		if (event.events & WL_LATCH_SET)
 		{
 			ResetLatch(MyLatch);
 			ProcessWalRcvInterrupts();
 		}
 
 		/* If socket is ready, advance the libpq state machine */
-		if (rc & io_flag)
+		if (event.events & io_flag)
 			status = PQconnectPoll(conn->streamConn);
 	} while (status != PGRES_POLLING_OK && status != PGRES_POLLING_FAILED);
 
@@ -231,7 +242,7 @@ libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
 	{
 		PGresult   *res;
 
-		res = libpqrcv_PQexec(conn->streamConn,
+		res = libpqrcv_PQexec(conn,
 							  ALWAYS_SECURE_SEARCH_PATH_SQL);
 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
 		{
@@ -359,7 +370,7 @@ libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
 	 * Get the system identifier and timeline ID as a DataRow message from the
 	 * primary server.
 	 */
-	res = libpqrcv_PQexec(conn->streamConn, "IDENTIFY_SYSTEM");
+	res = libpqrcv_PQexec(conn, "IDENTIFY_SYSTEM");
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
 		PQclear(res);
@@ -482,7 +493,7 @@ libpqrcv_startstreaming(WalReceiverConn *conn,
 						 options->proto.physical.startpointTLI);
 
 	/* Start streaming. */
-	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
+	res = libpqrcv_PQexec(conn, cmd.data);
 	pfree(cmd.data);
 
 	if (PQresultStatus(res) == PGRES_COMMAND_OK)
@@ -532,7 +543,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
 	 * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
 	 * also possible in case we aborted the copy in mid-stream.
 	 */
-	res = libpqrcv_PQgetResult(conn->streamConn);
+	res = libpqrcv_PQgetResult(conn);
 	if (PQresultStatus(res) == PGRES_TUPLES_OK)
 	{
 		/*
@@ -547,7 +558,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
 		PQclear(res);
 
 		/* the result set should be followed by CommandComplete */
-		res = libpqrcv_PQgetResult(conn->streamConn);
+		res = libpqrcv_PQgetResult(conn);
 	}
 	else if (PQresultStatus(res) == PGRES_COPY_OUT)
 	{
@@ -561,7 +572,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
 							pchomp(PQerrorMessage(conn->streamConn)))));
 
 		/* CommandComplete should follow */
-		res = libpqrcv_PQgetResult(conn->streamConn);
+		res = libpqrcv_PQgetResult(conn);
 	}
 
 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
@@ -572,7 +583,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
 	PQclear(res);
 
 	/* Verify that there are no more results */
-	res = libpqrcv_PQgetResult(conn->streamConn);
+	res = libpqrcv_PQgetResult(conn);
 	if (res != NULL)
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
@@ -597,7 +608,7 @@ libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
 	 * Request the primary to send over the history file for given timeline.
 	 */
 	snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
-	res = libpqrcv_PQexec(conn->streamConn, cmd);
+	res = libpqrcv_PQexec(conn, cmd);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
 		PQclear(res);
@@ -641,8 +652,9 @@ libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
  * May return NULL, rather than an error result, on failure.
  */
 static PGresult *
-libpqrcv_PQexec(PGconn *streamConn, const char *query)
+libpqrcv_PQexec(WalReceiverConn *conn, const char *query)
 {
+	PGconn *streamConn = conn->streamConn;
 	PGresult   *lastResult = NULL;
 
 	/*
@@ -665,7 +677,7 @@ libpqrcv_PQexec(PGconn *streamConn, const char *query)
 		/* Wait for, and collect, the next PGresult. */
 		PGresult   *result;
 
-		result = libpqrcv_PQgetResult(streamConn);
+		result = libpqrcv_PQgetResult(conn);
 		if (result == NULL)
 			break;				/* query is complete, or failure */
 
@@ -690,30 +702,31 @@ libpqrcv_PQexec(PGconn *streamConn, const char *query)
  * Perform the equivalent of PQgetResult(), but watch for interrupts.
  */
 static PGresult *
-libpqrcv_PQgetResult(PGconn *streamConn)
+libpqrcv_PQgetResult(WalReceiverConn *conn)
 {
+	PGconn *streamConn = conn->streamConn;
+
 	/*
 	 * Collect data until PQgetResult is ready to get the result without
 	 * blocking.
 	 */
 	while (PQisBusy(streamConn))
 	{
-		int			rc;
+		WaitEvent	event;
 
 		/*
 		 * We don't need to break down the sleep into smaller increments,
 		 * since we'll get interrupted by signals and can handle any
 		 * interrupts here.
 		 */
-		rc = WaitLatchOrSocket(MyLatch,
-							   WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
-							   WL_LATCH_SET,
-							   PQsocket(streamConn),
-							   0,
-							   WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
+		ModifyWaitEvent(conn->wes, 0, WL_SOCKET_READABLE, NULL);
+
+		if (WaitEventSetWait(conn->wes, -1, &event, 1,
+							 WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE) < 1)
+			continue;
 
 		/* Interrupted? */
-		if (rc & WL_LATCH_SET)
+		if (event.events & WL_LATCH_SET)
 		{
 			ResetLatch(MyLatch);
 			ProcessWalRcvInterrupts();
@@ -737,12 +750,28 @@ libpqrcv_PQgetResult(PGconn *streamConn)
 static void
 libpqrcv_disconnect(WalReceiverConn *conn)
 {
+	FreeWaitEventSet(conn->wes);
 	PQfinish(conn->streamConn);
 	if (conn->recvBuf != NULL)
 		PQfreemem(conn->recvBuf);
 	pfree(conn);
 }
 
+/*
+ * Wait for data, latch, or timeout.
+ */
+static int
+libpqrcv_wait(WalReceiverConn *conn, int timeout, int wait_event)
+{
+	WaitEvent event;
+
+	if (WaitEventSetWait(conn->wes, timeout, &event, 1, wait_event) == 1)
+		return event.events;
+
+	return 0;
+}
+
+
 /*
  * Receive a message available from XLOG stream.
  *
@@ -793,13 +822,13 @@ libpqrcv_receive(WalReceiverConn *conn, char **buffer,
 	{
 		PGresult   *res;
 
-		res = libpqrcv_PQgetResult(conn->streamConn);
+		res = libpqrcv_PQgetResult(conn);
 		if (PQresultStatus(res) == PGRES_COMMAND_OK)
 		{
 			PQclear(res);
 
 			/* Verify that there are no more results. */
-			res = libpqrcv_PQgetResult(conn->streamConn);
+			res = libpqrcv_PQgetResult(conn);
 			if (res != NULL)
 			{
 				PQclear(res);
@@ -941,7 +970,7 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
 			appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
 	}
 
-	res = libpqrcv_PQexec(conn->streamConn, cmd.data);
+	res = libpqrcv_PQexec(conn, cmd.data);
 	pfree(cmd.data);
 
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
@@ -1068,7 +1097,7 @@ libpqrcv_exec(WalReceiverConn *conn, const char *query,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 				 errmsg("the query interface requires a database connection")));
 
-	pgres = libpqrcv_PQexec(conn->streamConn, query);
+	pgres = libpqrcv_PQexec(conn, query);
 
 	switch (PQresultStatus(pgres))
 	{
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index 7a7eb3784e..f308ea9c24 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -507,13 +507,9 @@ WalReceiverMain(void)
 				 * could add and remove just the socket each time, potentially
 				 * avoiding some system calls.
 				 */
-				Assert(wait_fd != PGINVALID_SOCKET);
-				rc = WaitLatchOrSocket(MyLatch,
-									   WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
-									   WL_TIMEOUT | WL_LATCH_SET,
-									   wait_fd,
-									   NAPTIME_PER_CYCLE,
-									   WAIT_EVENT_WAL_RECEIVER_MAIN);
+				//Assert(wait_fd != PGINVALID_SOCKET);
+				rc = walrcv_wait(wrconn, NAPTIME_PER_CYCLE,
+								 WAIT_EVENT_WAL_RECEIVER_MAIN);
 				if (rc & WL_LATCH_SET)
 				{
 					ResetLatch(MyLatch);
@@ -532,7 +528,7 @@ WalReceiverMain(void)
 						XLogWalRcvSendReply(true, false);
 					}
 				}
-				if (rc & WL_TIMEOUT)
+				if (rc == 0)
 				{
 					/*
 					 * We didn't receive anything new. If we haven't heard
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 0b607ed777..3e80ff63ce 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -380,6 +380,11 @@ typedef WalRcvExecResult *(*walrcv_exec_fn) (WalReceiverConn *conn,
  */
 typedef void (*walrcv_disconnect_fn) (WalReceiverConn *conn);
 
+/*
+ * XXX Wait for readable data, timeout, or latch.
+ */
+typedef int (*walrcv_wait_fn) (WalReceiverConn *conn, int timeout, int wait_event);
+
 typedef struct WalReceiverFunctionsType
 {
 	walrcv_connect_fn walrcv_connect;
@@ -397,6 +402,7 @@ typedef struct WalReceiverFunctionsType
 	walrcv_get_backend_pid_fn walrcv_get_backend_pid;
 	walrcv_exec_fn walrcv_exec;
 	walrcv_disconnect_fn walrcv_disconnect;
+	walrcv_wait_fn walrcv_wait;
 } WalReceiverFunctionsType;
 
 extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
@@ -431,6 +437,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
 	WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)
 #define walrcv_disconnect(conn) \
 	WalReceiverFunctions->walrcv_disconnect(conn)
+#define walrcv_wait(conn, timeout, wait_event) \
+	WalReceiverFunctions->walrcv_wait(conn, timeout, wait_event)
 
 static inline void
 walrcv_clear_result(WalRcvExecResult *walres)
-- 
2.33.1

---

From: Alexander Lakhin <[email-masked]>
Date: Mon, 10 Jan 2022 20:00:00 +0300

> socket changes correctly), but does it detect the closed socket?
Yes, it fixes the behaviour and makes the 002_standby test pass (100 of
100 iterations). I'm yet to find out whether the other
WaitLatchOrSocket' users (e. g. postgres_fdw) can suffer from the
disconnected socket state, but this approach definitely works for
walreceiver.

Best regards,
Alexander

---

From: Thomas Munro <[email-masked]>
Date: Tue, 11 Jan 2022 09:52:01 +1300

> 100 iterations).

Thanks for testing.  That result does seem to confirm the hypothesis
that FD_CLOSE is reported only once for the socket on graceful
shutdown (that is, it's edge-triggered and incidentally you won't get
FD_READ), so you need to keep track of it carefully.  Incidentally,
another observation is that your WSAPoll() test appears to be
returning POLLHUP where at least Linux, FreeBSD and Solaris would not:
a socket that is only half shut down (the primary shut down its end
gracefully, but walreceiver did not), so I suspect Windows' POLLHUP
might have POLLRDHUP semantics.

> I'm yet to find out whether the other

I see where you're going: there might be safe call sequences and
unsafe call sequences, and maybe walreceiver is asking for trouble by
double-polling.  I'm not sure about that; I got the impression
recently that it's possible to get FD_CLOSE while you still have
buffered data to read, so then the next recv() will return > 0 and
then we don't have any state left anywhere to remember that we saw
FD_CLOSE, even if you're careful to poll and read in the ideal
sequence.  I could be wrong, and it would be nice if there is an easy
fix along those lines...  The documentation around FD_CLOSE is
unclear.

I do plan to make a higher quality patch like the one I showed
(material from earlier unfinished work[1] that needs a bit more
infrastructure), but to me that's new feature/efficiency work, not
something we'd want to back-patch.

Hmm, one thing I'm still unclear on: did this problem really start
with 6051857fc/ed52c3707?  My initial email in this thread lists
similar failures going back further, doesn't it?  (And what's tern
doing mixed up in this mess?)

[1] https://www.postgresql.org/message-id/flat/CA%2BhUKGJPaygh-6WHEd0FnH89GrkTpVyN_ew9ckv3%2BnwjmLcSeg%40mail.gmail.com#aa33ec3e7ad85499f35dd1434a139c3f

---

From: Andrew Dunstan <[email-masked]>
Date: Mon, 10 Jan 2022 16:21:39 -0500

> doing mixed up in this mess?)

Your list contains at least some false positives. e.g.
<https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=jacana&dt=2019-12-22%2014:19:02>
which has a different script failing.

cheers

andrew

--
Andrew Dunstan
EDB: https://www.enterprisedb.com

---

From: Tom Lane <[email-masked]>
Date: Mon, 10 Jan 2022 16:25:27 -0500

> doing mixed up in this mess?)

Well, those earlier ones may be committs failures, but a lot of
them contain different-looking symptoms, eg pg_ctl failures.

tern's failure at
https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=tern&dt=2021-07-17+10%3A57%3A49
does look similar, but we can see in its log that the standby
*did* notice the primary disconnection and then reconnect:

2021-07-17 16:29:08.248 UTC [17498380:2] LOG:  replication terminated by primary server
2021-07-17 16:29:08.248 UTC [17498380:3] DETAIL:  End of WAL reached on timeline 1 at 0/30378F8.
2021-07-17 16:29:08.248 UTC [17498380:4] FATAL:  could not send end-of-streaming message to primary: no COPY in progress
2021-07-17 16:29:08.248 UTC [25166230:5] LOG:  invalid record length at 0/30378F8: wanted 24, got 0
2021-07-17 16:29:08.350 UTC [16318578:1] FATAL:  could not connect to the primary server: server closed the connection unexpectedly
		This probably means the server terminated abnormally
		before or while processing the request.
2021-07-17 16:29:36.369 UTC [7077918:1] FATAL:  could not connect to the primary server: FATAL:  the database system is starting up
2021-07-17 16:29:36.380 UTC [11338028:1] FATAL:  could not connect to the primary server: FATAL:  the database system is starting up
...
2021-07-17 16:29:36.881 UTC [17367092:1] LOG:  started streaming WAL from primary at 0/3000000 on timeline 1

So I'm not sure what happened there, but it's not an instance
of this problem.  One thing that looks a bit suspicious is
this in the primary's log:

2021-07-17 16:26:47.832 UTC [12386550:1] LOG:  using stale statistics instead of current ones because stats collector is not responding

which makes me wonder if the timeout is down to out-of-date
pg_stats data.  The loop in 002_standby.pl doesn't appear to
depend on the stats collector:

my $primary_lsn =

but maybe I'm missing the connection.

Apropos of that, it's worth noting that wait_for_catchup *is*
dependent on up-to-date stats, and here's a recent run where
it sure looks like the timeout cause is AWOL stats collector:

https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=sungazer&dt=2022-01-10%2004%3A51%3A34

I wonder if we should refactor wait_for_catchup to probe the
standby directly instead of relying on the upstream's view.

			regards, tom lane

---

From: Alexander Lakhin <[email-masked]>
Date: Tue, 11 Jan 2022 18:00:01 +0300

> fix along those lines...  The documentation around FD_CLOSE is
I had no strong opinion regarding unsafe sequence, though initially I
suspected that exactly the second libpqrcv_PQgetResult call could cause
the issue. But after digging into WaitLatchOrSocket I'd inclined to put
the fix deeper to satisfy all possible callers.
At the other hand, I've shared Tom's concerns regarding other clients,
that can stuck on WaitForMultipleObjects() just as walreceiver does, and
hoped that only walreceiver suffer from a graceful server socket closing.
So to get these doubts cleared, I've made a simple test for postgres_fdw
(please look at the attachment; you can put it into
contrib/postgres_fdw/t and run `vcregress taptest contrib\postgres_fdw`).
This test shows for me:
===
...
t/001_disconnection.pl .. # 12:13:39.481084 executing query...
# 12:13:43.245277 result:       0
# 0|0

# 12:13:43.246342 executing query...
# 12:13:46.525924 result:       0
# 0|0

# 12:13:46.527097 executing query...
# 12:13:47.745176 result:       3
#
# psql:<stdin>:1: WARNING:  no connection to the server
# psql:<stdin>:1: ERROR:  FATAL:  terminating connection due to
administrator co
mmand
# server closed the connection unexpectedly
#       This probably means the server terminated abnormally
#       before or while processing the request.
# CONTEXT:  remote SQL command: FETCH 100 FROM c1
# 12:13:47.794612 executing query...
# 12:13:51.073318 result:       0
# 0|0

# 12:13:51.074347 executing query...
===

With the simple logging added to connection.c:
                /* Sleep until there's something to do */
elog(LOG, "pgfdw_get_result before WaitLatchOrSocket");
                wc = WaitLatchOrSocket(MyLatch,
                                       WL_LATCH_SET | WL_SOCKET_READABLE |
                                       WL_EXIT_ON_PM_DEATH,
                                       PQsocket(conn),
                                       -1L, PG_WAIT_EXTENSION);
elog(LOG, "pgfdw_get_result after WaitLatchOrSocket");

I see in 001_disconnection_local.log:
...
2022-01-11 15:13:52.875 MSK|Administrator|postgres|61dd747f.5e4|LOG: 
pgfdw_get_result after WaitLatchOrSocket
2022-01-11 15:13:52.875
MSK|Administrator|postgres|61dd747f.5e4|STATEMENT:  SELECT * FROM large
WHERE a = fx2(a)
2022-01-11 15:13:52.875 MSK|Administrator|postgres|61dd747f.5e4|LOG: 
pgfdw_get_result before WaitLatchOrSocket
2022-01-11 15:13:52.875
MSK|Administrator|postgres|61dd747f.5e4|STATEMENT:  SELECT * FROM large
WHERE a = fx2(a)
2022-01-11 15:14:36.976 MSK|||61dd74ac.840|DEBUG:  autovacuum:
processing database "postgres"
2022-01-11 15:14:51.088 MSK|Administrator|postgres|61dd747f.5e4|LOG: 
pgfdw_get_result after WaitLatchOrSocket
2022-01-11 15:14:51.088
MSK|Administrator|postgres|61dd747f.5e4|STATEMENT:  SELECT * FROM large
WHERE a = fx2(a)
2022-01-11 15:14:51.089 MSK|Administrator|postgres|61dd747f.5e4|LOG: 
pgfdw_get_result before WaitLatchOrSocket
2022-01-11 15:14:51.089
MSK|Administrator|postgres|61dd747f.5e4|STATEMENT:  SELECT * FROM large
WHERE a = fx2(a)
2022-01-11 15:15:37.006 MSK|||61dd74e9.9e8|DEBUG:  autovacuum:
processing database "postgres"
2022-01-11 15:16:37.116 MSK|||61dd7525.ad0|DEBUG:  autovacuum:
processing database "postgres"
2022-01-11 15:17:37.225 MSK|||61dd7561.6a0|DEBUG:  autovacuum:
processing database "postgres"
2022-01-11 15:18:36.916 MSK|||61dd7470.704|LOG:  checkpoint starting: time
...
2022-01-11 15:36:38.225 MSK|||61dd79d6.2a0|DEBUG:  autovacuum:
processing database "postgres"
...

So here we get similar hanging on WaitLatchOrSocket().
Just to make sure that it's indeed the same issue, I've removed socket
shutdown&close and the test executed to the end (several times). Argh.

Best regards,
Alexander

---

From: Thomas Munro <[email-masked]>
Date: Wed, 12 Jan 2022 09:10:42 +1300

> shutdown&close and the test executed to the end (several times). Argh.

Ouch.  I think our options at this point are:
1.  Revert 6051857fc (and put it back when we have a working
long-lived WES as I showed).  This is not very satisfying, now that we
understand the bug, because even without that change I guess you must
be able to reach the hanging condition by using Windows postgres_fdw
to talk to a non-Windows server (ie a normal TCP stack with graceful
shutdown/linger on process exit).
2.  Put your poll() check into the READABLE side.  There's some
precedent for that sort of kludge on the WRITEABLE side (and a
rejection of the fragile idea that clients of latch.c should only
perform "safe" sequences):

        /*
         * Windows does not guarantee to log an FD_WRITE network event
         * indicating that more data can be sent unless the previous send()
         * failed with WSAEWOULDBLOCK.  While our caller might well have made
         * such a call, we cannot assume that here.  Therefore, if waiting for
         * write-ready, force the issue by doing a dummy send().  If the dummy
         * send() succeeds, assume that the socket is in fact write-ready, and
         * return immediately.  Also, if it fails with something other than
         * WSAEWOULDBLOCK, return a write-ready indication to let our caller
         * deal with the error condition.
         */

---

From: Tom Lane <[email-masked]>
Date: Tue, 11 Jan 2022 15:16:50 -0500

> shutdown/linger on process exit).

It'd be worth checking, perhaps.  One thing I've been wondering all
along is how much of this behavior is specific to the local-loopback
case where Windows can see both ends of the connection.  You'd think
that they couldn't long get away with such blatant violations of the
TCP specs when talking to external servers, because the failures
would be visible to everyone with a web browser.

			regards, tom lane

---

From: Alexander Lakhin <[email-masked]>
Date: Wed, 12 Jan 2022 10:00:00 +0300

> 11.01.2022 23:16, Tom Lane wrote:
I've split my test (both parts attached) and run it on two virtual
machines with clean builds from master (ac7c8075) on both (just the
debugging output added to connection.c). I provide probably redundant
info (also see attached screenshot) just to make sure that I didn't make
a mistake.
The excerpt from 001_disconnection1_local.log:
...
2022-01-12 09:29:48.099 MSK|Administrator|postgres|61de755a.a54|LOG: 
pgfdw_get_result: before WaitLatchOrSocket
2022-01-12 09:29:48.099
MSK|Administrator|postgres|61de755a.a54|STATEMENT:  SELECT * FROM large
WHERE a = fx2(a)
2022-01-12 09:29:48.100 MSK|Administrator|postgres|61de755a.a54|LOG: 
pgfdw_get_result: after WaitLatchOrSocket
2022-01-12 09:29:48.100
MSK|Administrator|postgres|61de755a.a54|STATEMENT:  SELECT * FROM large
WHERE a = fx2(a)
2022-01-12 09:29:48.100 MSK|Administrator|postgres|61de755a.a54|LOG: 
pgfdw_get_result: before WaitLatchOrSocket
2022-01-12 09:29:48.100
MSK|Administrator|postgres|61de755a.a54|STATEMENT:  SELECT * FROM large
WHERE a = fx2(a)
2022-01-12 09:29:48.100 MSK|Administrator|postgres|61de755a.a54|LOG: 
pgfdw_get_result: after WaitLatchOrSocket
2022-01-12 09:29:48.100
MSK|Administrator|postgres|61de755a.a54|STATEMENT:  SELECT * FROM large
WHERE a = fx2(a)
2022-01-12 09:29:48.100 MSK|Administrator|postgres|61de755a.a54|ERROR: 
FATAL:  terminating connection due to administrator command
    server closed the connection unexpectedly
        This probably means the server terminated abnormally
        before or while processing the request.
2022-01-12 09:29:48.100
MSK|Administrator|postgres|61de755a.a54|CONTEXT:  remote SQL command:
FETCH 100 FROM c1
2022-01-12 09:29:48.100
MSK|Administrator|postgres|61de755a.a54|WARNING:  no connection to the
server
2022-01-12 09:29:48.100
MSK|Administrator|postgres|61de755a.a54|CONTEXT:  remote SQL command:
ABORT TRANSACTION
2022-01-12 09:29:48.107 MSK|Administrator|postgres|61de755a.a54|LOG: 
disconnection: session time: 0:00:01.577 user=Administrator
database=postgres host=127.0.0.1 port=49752
2022-01-12 09:29:48.257 MSK|[unknown]|[unknown]|61de755c.a4c|LOG: 
connection received: host=127.0.0.1 port=49754
2022-01-12 09:29:48.261 MSK|Administrator|postgres|61de755c.a4c|LOG: 
connection authenticated: identity="WIN-FCPSOVMM1JC\Administrator"
method=sspi
(C:/src/postgrespro/contrib/postgres_fdw/tmp_check/t_001_disconnection1_local_data/pgdata/pg_hba.conf:2)
2022-01-12 09:29:48.261 MSK|Administrator|postgres|61de755c.a4c|LOG: 
connection authorized: user=Administrator database=postgres
application_name=001_disconnection1.pl
2022-01-12 09:29:48.263 MSK|Administrator|postgres|61de755c.a4c|LOG: 
statement: SELECT * FROM large WHERE a = fx2(a)
2022-01-12 09:29:48.285 MSK|Administrator|postgres|61de755c.a4c|LOG: 
pgfdw_get_result: before WaitLatchOrSocket
2022-01-12 09:29:48.285
MSK|Administrator|postgres|61de755c.a4c|STATEMENT:  SELECT * FROM large
WHERE a = fx2(a)
...

By the look of things, you are right and this is the localhost-only issue.
I've rechecked that the test 001_disconnection.pl (local-loopback
version) hangs on both machines while 001_disconnection1.pl performs
successfully in both directions. I'm not sure whether the Windows client
and non-Windows server or reverse combinations are of interest in light
of the above.

Best regards,
Alexander

---

From: Thomas Munro <[email-masked]>
Date: Thu, 13 Jan 2022 19:36:01 +1300

> By the look of things, you are right and this is the localhost-only issue.

But can't that be explained with timing races?  You change some stuff
around and it becomes less likely that you get a FIN to arrive in a
super narrow window, which I'm guessing looks something like: recv ->
EWOULDBLOCK, [receive FIN], wait -> FD_CLOSE, wait [hangs].  Note that
it's not happening on several Windows BF animals, and the ones it is
happening on only do it only every few weeks.

Here's a draft attempt at a fix.  First I tried to use recv(fd, &c, 1,
MSG_PEEK) == 0 to detect EOF, which seemed to me to be a reasonable
enough candidate, but somehow it corrupts the stream (!?), so I used
Alexander's POLLHUP idea, except I pushed it down to a more principled
place IMHO.  Then I suppressed it after the initial check because then
the logic from my earlier patch takes over, so stuff like FeBeWaitSet
doesn't suffer from extra calls, only these two paths that haven't
been converted to long-lived WESes yet.  Does this pass the test?

I wonder if this POLLHUP technique is reliable enough (I know that
wouldn't work on other systems[1], which is why I was trying to make
MSG_PEEK work...).

What about environment variable PG_TEST_USE_UNIX_SOCKETS=1, does it
reproduce with that set, and does the patch fix it?  I'm hoping that
explains some Windows CI failures from a nearby thread[2].

[1] https://illumos.topicbox.com/groups/developer/T5576767e764aa26a-Maf8f3460c2866513b0ac51bf
[2] https://www.postgresql.org/message-id/flat/CALT9ZEG%3DC%3DJSypzt2gz6NoNtx-ew2tYHbwiOfY_xNo%2ByBY_%3Djw%40mail.gmail.com
From a2cf2b016afc03d384a1682cd02aab760273d6fb Mon Sep 17 00:00:00 2001
From: Thomas Munro <[email-masked]>
Date: Thu, 13 Jan 2022 15:48:14 +1300
Subject: [PATCH] Fix handling of socket FD_CLOSE events on Windows.

In some situations we could hang waiting for an FD_CLOSE event that has
already been reported.  Fix this by adding some state to WaitEvent to
remember that we've already been told the socket is closed.  To handle
client code that creates and destroys multiple temporary WaitEventSet
objects and thus would lose that state, check for EOF explicitly the
first time through.
---
 src/backend/storage/ipc/latch.c | 41 +++++++++++++++++++++++++++++++++
 src/include/storage/latch.h     |  1 +
 2 files changed, 42 insertions(+)

diff --git a/src/backend/storage/ipc/latch.c b/src/backend/storage/ipc/latch.c
index 61c876beff..86b86e71c4 100644
--- a/src/backend/storage/ipc/latch.c
+++ b/src/backend/storage/ipc/latch.c
@@ -899,6 +899,7 @@ AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch,
 	event->user_data = user_data;
 #ifdef WIN32
 	event->reset = false;
+	event->eof = -1;			/* unknown */
 #endif
 
 	if (events == WL_LATCH_SET)
@@ -1851,6 +1852,45 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
 			cur_event->reset = false;
 		}
 
+		/*
+		 * Windows does not report FD_CLOSE repeatedly, and does not report
+		 * FD_READ on EOF.  Therefore we need to remember if we've seen
+		 * FD_CLOSE, to defend against callers that wait twice in a row
+		 * without a recv() that fails with WSAEWOULDBLOCK in between.
+		 */
+		if (cur_event->events & WL_SOCKET_READABLE)
+		{
+			/*
+			 * Check for EOF explicitly the first time through, to bridge the
+			 * gap between temporary WaitEventSet objects (such as those
+			 * created by WaitLatchOrSocket()).
+			 */
+			if (cur_event->eof == -1)
+			{
+				WSAPOLLFD	pollfd;
+				int			rc;
+
+				pollfd.fd = cur_event->fd;
+				pollfd.events = POLLRDNORM;
+				pollfd.revents = 0;
+				rc = WSAPoll(&pollfd, 1, 0);
+				if (rc == 1 && pollfd.revents & POLLHUP)
+					cur_event->eof = 1; /* EOF */
+				else
+					cur_event->eof = 0; /* data or error available */
+			}
+
+			/* If we've seen EOF or FD_CLOSE, keep reporting readiness. */
+			if (cur_event->eof == 1)
+			{
+				occurred_events->pos = cur_event->pos;
+				occurred_events->user_data = cur_event->user_data;
+				occurred_events->events = WL_SOCKET_READABLE;
+				occurred_events->fd = cur_event->fd;
+				return 1;
+			}
+		}
+
 		/*
 		 * Windows does not guarantee to log an FD_WRITE network event
 		 * indicating that more data can be sent unless the previous send()
@@ -2002,6 +2042,7 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
 		{
 			/* EOF/error, so signal all caller-requested socket flags */
 			occurred_events->events |= (cur_event->events & WL_SOCKET_MASK);
+			cur_event->eof = 1;
 		}
 
 		if (occurred_events->events != 0)
diff --git a/src/include/storage/latch.h b/src/include/storage/latch.h
index 3aa7b33834..41a857c0c3 100644
--- a/src/include/storage/latch.h
+++ b/src/include/storage/latch.h
@@ -147,6 +147,7 @@ typedef struct WaitEvent
 	void	   *user_data;		/* pointer provided in AddWaitEventToSet */
 #ifdef WIN32
 	bool		reset;			/* Is reset of the event required? */
+	int8		eof;			/* Has EOF been reached on a socket? */
 #endif
 } WaitEvent;
 
-- 
2.33.1

---

From: Alexander Lakhin <[email-masked]>
Date: Thu, 13 Jan 2022 10:00:00 +0300

> happening on only do it only every few weeks.
But I still see the issue when I run both test parts on a single
machine: first instance is `vcregress taptest src\test\restart` and the
second `set NO_TEMP_INSTALL=1 & vcregress taptest contrib/postgres_fdw`
(see attachment).

I'll try new tests and continue investigation later today/tomorrow. Thanks!

Best regards,
Alexander

---

From: Thomas Munro <[email-masked]>
Date: Fri, 14 Jan 2022 20:31:22 +1300

On Thu, Jan 13, 2022 at 7:36 PM Thomas Munro <[email-masked]> wrote:
> ... First I tried to use recv(fd, &c, 1,
> enough candidate, but somehow it corrupts the stream (!?),

Ahh, that'd be because recv() and friends are redirected to our
wrappers in socket.c, where we use the overlapped Winsock API (that
is, async network IO), which is documented as not supporting MSG_PEEK.
OK then.

Andres and I chatted about this stuff off list and he pointed out
something else about the wrappers in socket.c: there are more paths in
there that work with socket events, which means more ways to lose the
precious FD_CLOSE event.  I don't know if any of those paths are
reachable in the relevant cases, but it does look a little bit like
the lack of graceful shutdown might have been hiding a whole class of
event tracking bug.

---

From: Alexander Lakhin <[email-masked]>
Date: Fri, 14 Jan 2022 11:00:00 +0300

> been converted to long-lived WESes yet.  Does this pass the test?
Yes, this fix eliminates the flakiness for me. The test commit_ts (with
002_standby and 003_standby_2) passed 2x200 iterations.
It also makes my test postgres_fdw/001_disconnection pass reliably.
> explains some Windows CI failures from a nearby thread[2].
With PG_TEST_USE_UNIX_SOCKETS=1 the test commit_ts/002_standby fails on
the unpatched HEAD:
t/002_standby.pl .... 1/4 # poll_query_until timed out executing this query:
# SELECT '0/303C628'::pg_lsn <= pg_last_wal_replay_lsn()
# expecting this output:
# t
# last actual query output:
# f
# with stderr:
# Looks like your test exited with 25 just after 1.
t/002_standby.pl .... Dubious, test returned 25 (wstat 6400, 0x1900)

002_standby_primary.log contains:
2022-01-13 18:57:32.925 PST [1448] LOG:  starting PostgreSQL 15devel,
compiled by Visual C++ build 1928, 64-bit
2022-01-13 18:57:32.926 PST [1448] LOG:  listening on Unix socket
"C:/Users/1/AppData/Local/Temp/yOKQPH1FoO/.s.PGSQL.62733"

The same with my postgres_fdw test:
# 03:41:12.533246 result:       0
# 0|0
# 03:41:12.534758 executing query...
# 03:41:14.267594 result:       3
#
# psql:<stdin>:1: WARNING:  no connection to the server
# psql:<stdin>:1: ERROR:  FATAL:  terminating connection due to
administrator command
# server closed the connection unexpectedly
#       This probably means the server terminated abnormally
#       before or while processing the request.
# CONTEXT:  remote SQL command: FETCH 100 FROM c1
# 03:41:14.270449 executing query...
# 03:41:14.334437 result:       3
#
# psql:<stdin>:1: ERROR:  could not connect to server "fpg"
# DETAIL:  connection to server on socket
"C:/Users/1/AppData/Local/Temp/hJWD9mzPHM/.s.PGSQL.57414" failed:
Connection refused (0x0000274D/10061)
#       Is the server running locally and accepting connections on that
socket?
# 03:41:14.336918 executing query...
# 03:41:14.422381 result:       3
#
# psql:<stdin>:1: ERROR:  could not connect to server "fpg"
# DETAIL:  connection to server on socket
"C:/Users/1/AppData/Local/Temp/hJWD9mzPHM/.s.PGSQL.57414" failed:
Connection refused (0x0000274D/10061)
#       Is the server running locally and accepting connections on that
socket?
# 03:41:14.425628 executing query...
...hang...

With the patch these tests pass successfully.

I can also confirm that on Windows 10 20H2 (previous tests were
performed on Windows Server 2012) the unpatched HEAD +
PG_TEST_USE_UNIX_SOCKETS=1 hangs on src\test\recovery\001_stream_rep (on
iterations 1, 1, 4 for me).
(v9-0001-Add-option-for-amcheck-and-pg_amcheck-to-check-un.patch [1] not
required to see that.)
001_stream_rep_primary.log contains:
...
2022-01-13 19:46:48.287 PST [1364] LOG:  listening on Unix socket
"C:/Users/1/AppData/Local/Temp/EWzapwiXjV/.s.PGSQL.58248"
2022-01-13 19:46:48.317 PST [6736] LOG:  database system was shut down
at 2022-01-13 19:46:37 PST
2022-01-13 19:46:48.331 PST [1364] LOG:  database system is ready to
accept connections
2022-01-13 19:46:49.536 PST [1088] 001_stream_rep.pl LOG:  statement:
CREATE TABLE tab_int AS SELECT generate_series(1,1002) AS a
2022-01-13 19:46:49.646 PST [3028] 001_stream_rep.pl LOG:  statement:
SELECT pg_current_wal_insert_lsn()
2022-01-13 19:46:49.745 PST [3360] 001_stream_rep.pl LOG:  statement:
SELECT '0/3023268' <= replay_lsn AND state = 'streaming' FROM
pg_catalog.pg_stat_replication WHERE application_name = 'standby_1';
...
2022-01-13 19:50:39.755 PST [4924] 001_stream_rep.pl LOG:  statement:
SELECT '0/3023268' <= replay_lsn AND state = 'streaming' FROM
pg_catalog.pg_stat_replication WHERE application_name = 'standby_1';
2022-01-13 19:50:39.928 PST [5924] 001_stream_rep.pl LOG:  statement:
SELECT '0/3023268' <= replay_lsn AND state = 'streaming' FROM
pg_catalog.pg_stat_replication WHERE application_name = 'standby_1';

Without PG_TEST_USE_UNIX_SOCKETS=1 and without the fix the
001_stream_rep hangs too (but on iterations 3, 8, 2). So it seems that
using unix sockets increases the fail rate.

With the fix 100 iterations with PG_TEST_USE_UNIX_SOCKETS=1 and 40 (and
still counting) iterations without PG_TEST_USE_UNIX_SOCKETS pass.

So the fix looks as absolutely working to me with the tests that we have
for now.

[1]
https://www.postgresql.org/message-id/CALT9ZEHx2%2B9rqAeAANkUXJCsTueQqdx2Tt6ypaig9tozJkWvkQ%40mail.gmail.com

---

From: Andres Freund <[email-masked]>
Date: Fri, 14 Jan 2022 12:28:48 -0800

Hi,

On 2022-01-14 20:31:22 +1300, Thomas Munro wrote:
> precious FD_CLOSE event.

I think it doesn't even need to touch socket.c to cause breakage. Using two
different WaitEventSets is enough.

https://docs.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-wsaeventselect
says:

> The FD_CLOSE network event is recorded when a close indication is received
> for the virtual circuit corresponding to the socket. In TCP terms, this
> means that the FD_CLOSE is recorded when the connection goes into the TIME
> WAIT or CLOSE WAIT states. This results from the remote end performing a
> shutdown on the send side or a closesocket. FD_CLOSE being posted after all
> data is read from a socket

So FD_CLOSE is *recorded* internally when the connection is closed. But only
posted to the visible event once all data is read. All good so far. But
combine that with:

> Issuing a WSAEventSelect for a socket cancels any previous WSAAsyncSelect or
> WSAEventSelect for the same socket and clears the internal network event
> record.

Note the bit about clearing the internal network event record. Which seems to
pretty precisely explain why we're loosing FD_CLOSEs?

And it does also explain why this is more likely after the shutdown changes:
It's more likely the network stack knows it has readable data *and* that the
connection closed. Which is recorded in the "internal network event
record". But once all the data is read, walsender.c will do another
WaitLatchOrSocket(), which does WSAEventSelect(), clearing the "internal event
record" and loosing the FD_CLOSE.

My first inclination was that we ought to wrap the socket created for windows
in pgwin32_socket() in a custom type with some additional data - including
information about already received events, an EVENT, etc. I think that might
help to remove a lot of the messy workarounds we have in socket.c etc. But: It
wouldn't do much good here, because the socket is not a socket created by
socket.c but by libpq :(.

Greetings,

Andres Freund

---

From: Andres Freund <[email-masked]>
Date: Fri, 14 Jan 2022 12:47:26 -0800

Hi,

On 2022-01-14 12:28:48 -0800, Andres Freund wrote:
> But once all the data is read, walsender.c will do another

Walreceiver only started using WES in
2016-03-29 [314cbfc5d] Add new replication mode synchronous_commit = 'remote_ap

With that came the following comment:

                /*
                 * Ideally we would reuse a WaitEventSet object repeatedly
                 * here to avoid the overheads of WaitLatchOrSocket on epoll
                 * systems, but we can't be sure that libpq (or any other
                 * walreceiver implementation) has the same socket (even if
                 * the fd is the same number, it may have been closed and
                 * reopened since the last time).  In future, if there is a
                 * function for removing sockets from WaitEventSet, then we
                 * could add and remove just the socket each time, potentially
                 * avoiding some system calls.
                 */
                Assert(wait_fd != PGINVALID_SOCKET);
                rc = WaitLatchOrSocket(MyLatch,
                                       WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
                                       WL_TIMEOUT | WL_LATCH_SET,
                                       wait_fd,
                                       NAPTIME_PER_CYCLE,
                                       WAIT_EVENT_WAL_RECEIVER_MAIN);

I don't really see how libpq could have changed the socket underneath us, as
long as we get it the first time after the connection successfully was
established?  I mean, there's a running command that we're processing the
result of?  Nor do I understand what "any other walreceiver implementation"
refers to?

Thomas, I think you wrote that?

Greetings,

Andres Freund

---

From: Thomas Munro <[email-masked]>
Date: Sat, 15 Jan 2022 10:59:00 +1300

> different WaitEventSets is enough.

Right.  I was interested in your observation because so far we'd
*only* been considering the two-consecutive-WaitEventSets case, which
we could grok experimentally.  The patch Alexander tested most
recently uses a tri-state eof flag, so (1) the WES event starts out in
"unknown" state and polls with WSAPoll() to figure out whether the
socket was already closed when it wasn't looking, and then (2) it
switches to believing that we'll definitely get an FD_CLOSE so we
don't need to make the extra call on every wait.  That does seem to do
the job, but if there are *other* places that can eat FD_CLOSE event
once we've switched to believing that it will come, that might be
fatal to the second part of that idea, and we might have to assume
"unknown" all the time, which would be somewhat similar to the way we
do a dummy WSASend() every time when waiting for WRITEABLE.

(That patch is assuming that we're looking for something simple to
back-patch, in the event that we decide not to just revert the
graceful shutdown patch from back branches.  Reverting might be a
better idea for now, and then we could fix all this stuff going
forward.)

> pretty precisely explain why we're loosing FD_CLOSEs?

Indeed.

> record" and loosing the FD_CLOSE.

Yeah.

---

From: Thomas Munro <[email-masked]>
Date: Sat, 15 Jan 2022 11:01:44 +1300

> better idea for now, and then we could fix all this stuff going

(Though, as mentioned already, reverting isn't really enough either,
because other OSes that Windows might be talking to use lingering
sockets...  and there may still be ways for this to break...)

---

From: Andres Freund <[email-masked]>
Date: Fri, 14 Jan 2022 14:44:20 -0800

Hi,

On 2022-01-15 10:59:00 +1300, Thomas Munro wrote:
> we could grok experimentally.

There likely are further problems in other parts, but I think socket.c is
unlikely to be involved in walreceiver case - there shouldn't be any socket.c
style socket in walreceiver itself, nor do I think we are doing a
send/recv/select backed by socket.c.

> The patch Alexander tested most recently uses a tri-state eof flag [...]

What about instead giving WalReceiverConn an internal WaitEventSet, and using
that consistently? I've attached a draft for that.

Alexander, could you test with that patch applied?

Greetings,

Andres Freund
>From 8139d324687364164c9abbd3e2b2a54b5d5bbcad Mon Sep 17 00:00:00 2001
From: Andres Freund <[email-masked]>
Date: Fri, 14 Jan 2022 14:39:57 -0800
Subject: [PATCH] WIP: use long-lived WaitEventSet for libpqwalreceiver
 connections.

---
 src/include/replication/walreceiver.h         |  29 +++--
 .../libpqwalreceiver/libpqwalreceiver.c       | 112 ++++++++++++------
 src/backend/replication/logical/tablesync.c   |  11 +-
 src/backend/replication/logical/worker.c      |  13 +-
 src/backend/replication/walreceiver.c         |  26 +---
 5 files changed, 112 insertions(+), 79 deletions(-)

diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 92f73a55b8d..07a16470d48 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -253,6 +253,17 @@ typedef void (*walrcv_check_conninfo_fn) (const char *conninfo);
  */
 typedef char *(*walrcv_get_conninfo_fn) (WalReceiverConn *conn);
 
+/*
+ * walrcv_wait_fn
+ *
+ * Waits for socket to become readable, or a latch interrupt.
+ *
+ * timeout is passed on to WaitEventSetWait(), return value is
+ * like WaitLatchOrSocket's.
+ */
+typedef int (*walrcv_wait_fn) (WalReceiverConn *conn, long timeout,
+							   uint32 wait_event_info);
+
 /*
  * walrcv_get_senderinfo_fn
  *
@@ -317,14 +328,13 @@ typedef void (*walrcv_endstreaming_fn) (WalReceiverConn *conn,
 /*
  * walrcv_receive_fn
  *
- * Receive a message available from the WAL stream.  'buffer' is a pointer
- * to a buffer holding the message received.  Returns the length of the data,
- * 0 if no data is available yet ('wait_fd' is a socket descriptor which can
- * be waited on before a retry), and -1 if the cluster ended the COPY.
+ * Receive a message available from the WAL stream.  'buffer' is a pointer to
+ * a buffer holding the message received.  Returns the length of the data, 0
+ * if no data is available yet (see walrcv_wait()), and -1 if the cluster
+ * ended the COPY.
  */
 typedef int (*walrcv_receive_fn) (WalReceiverConn *conn,
-								  char **buffer,
-								  pgsocket *wait_fd);
+								  char **buffer);
 
 /*
  * walrcv_send_fn
@@ -385,6 +395,7 @@ typedef struct WalReceiverFunctionsType
 	walrcv_connect_fn walrcv_connect;
 	walrcv_check_conninfo_fn walrcv_check_conninfo;
 	walrcv_get_conninfo_fn walrcv_get_conninfo;
+	walrcv_wait_fn walrcv_wait;
 	walrcv_get_senderinfo_fn walrcv_get_senderinfo;
 	walrcv_identify_system_fn walrcv_identify_system;
 	walrcv_server_version_fn walrcv_server_version;
@@ -407,6 +418,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
 	WalReceiverFunctions->walrcv_check_conninfo(conninfo)
 #define walrcv_get_conninfo(conn) \
 	WalReceiverFunctions->walrcv_get_conninfo(conn)
+#define walrcv_wait(conn, timeout, wait_event_info) \
+	WalReceiverFunctions->walrcv_wait(conn, timeout, wait_event_info)
 #define walrcv_get_senderinfo(conn, sender_host, sender_port) \
 	WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)
 #define walrcv_identify_system(conn, primary_tli) \
@@ -419,8 +432,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
 	WalReceiverFunctions->walrcv_startstreaming(conn, options)
 #define walrcv_endstreaming(conn, next_tli) \
 	WalReceiverFunctions->walrcv_endstreaming(conn, next_tli)
-#define walrcv_receive(conn, buffer, wait_fd) \
-	WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
+#define walrcv_receive(conn, buffer) \
+	WalReceiverFunctions->walrcv_receive(conn, buffer)
 #define walrcv_send(conn, buffer, nbytes) \
 	WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
 #define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn) \
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 0d89db4e6a6..24f062f779c 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -46,6 +46,8 @@ struct WalReceiverConn
 	bool		logical;
 	/* Buffer for currently read records */
 	char	   *recvBuf;
+
+	WaitEventSet *wes;
 };
 
 /* Prototypes for interface functions */
@@ -54,6 +56,8 @@ static WalReceiverConn *libpqrcv_connect(const char *conninfo,
 										 char **err);
 static void libpqrcv_check_conninfo(const char *conninfo);
 static char *libpqrcv_get_conninfo(WalReceiverConn *conn);
+static int libpqrcv_wait(WalReceiverConn *conn, long timeout,
+						 uint32 wait_event_info);
 static void libpqrcv_get_senderinfo(WalReceiverConn *conn,
 									char **sender_host, int *sender_port);
 static char *libpqrcv_identify_system(WalReceiverConn *conn,
@@ -66,8 +70,7 @@ static bool libpqrcv_startstreaming(WalReceiverConn *conn,
 									const WalRcvStreamOptions *options);
 static void libpqrcv_endstreaming(WalReceiverConn *conn,
 								  TimeLineID *next_tli);
-static int	libpqrcv_receive(WalReceiverConn *conn, char **buffer,
-							 pgsocket *wait_fd);
+static int	libpqrcv_receive(WalReceiverConn *conn, char **buffer);
 static void libpqrcv_send(WalReceiverConn *conn, const char *buffer,
 						  int nbytes);
 static char *libpqrcv_create_slot(WalReceiverConn *conn,
@@ -87,6 +90,7 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
 	libpqrcv_connect,
 	libpqrcv_check_conninfo,
 	libpqrcv_get_conninfo,
+	libpqrcv_wait,
 	libpqrcv_get_senderinfo,
 	libpqrcv_identify_system,
 	libpqrcv_server_version,
@@ -102,8 +106,8 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
 };

 /*
@@ -203,6 +207,11 @@ libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
 		else
 			io_flag = WL_SOCKET_WRITEABLE;
 
+		/*
+		 * NB: cannot yet use/create conn->wes, during connection
+		 * establishment the socket can change.
+		 */
+
 		rc = WaitLatchOrSocket(MyLatch,
 							   WL_EXIT_ON_PM_DEATH | WL_LATCH_SET | io_flag,
 							   PQsocket(conn->streamConn),
@@ -227,11 +236,20 @@ libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
 		return NULL;
 	}
 
+	conn->wes = CreateWaitEventSet(CurrentMemoryContext, 3);
+	AddWaitEventToSet(conn->wes, WL_LATCH_SET, PGINVALID_SOCKET,
+					  MyLatch, NULL);
+	/* XXX: This isn't necessarily nice to do unconditionally? */
+	AddWaitEventToSet(conn->wes, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET,
+					  NULL, NULL);
+	AddWaitEventToSet(conn->wes, WL_SOCKET_READABLE, PQsocket(conn->streamConn),
+					  NULL, NULL);
+
 	if (logical)
 	{
 		PGresult   *res;
 
@@ -359,7 +377,7 @@ libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
@@ -482,7 +500,7 @@ libpqrcv_startstreaming(WalReceiverConn *conn,
 
 	/* Start streaming. */
 
 	if (PQresultStatus(res) == PGRES_COMMAND_OK)
@@ -532,7 +550,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
 	 * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
 	 * also possible in case we aborted the copy in mid-stream.
 	 */
-	res = libpqrcv_PQgetResult(conn->streamConn);
+	res = libpqrcv_PQgetResult(conn);
 	if (PQresultStatus(res) == PGRES_TUPLES_OK)
 	{
 		/*
@@ -547,7 +565,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
 		PQclear(res);
 
@@ -561,7 +579,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)

 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
@@ -572,7 +590,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
 	PQclear(res);
 
@@ -597,7 +615,7 @@ libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
@@ -641,7 +659,7 @@ libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
 
@@ -657,7 +675,7 @@ libpqrcv_PQexec(PGconn *streamConn, const char *query)
 	 * theoretically block.  In practice, since we don't send very long query
 	 * strings, the risk seems negligible.
 	 */
-	if (!PQsendQuery(streamConn, query))
+	if (!PQsendQuery(conn->streamConn, query))
 		return NULL;
 
 	for (;;)
@@ -665,7 +683,7 @@ libpqrcv_PQexec(PGconn *streamConn, const char *query)
 		/* Wait for, and collect, the next PGresult. */
 		PGresult   *result;

@@ -679,24 +697,49 @@ libpqrcv_PQexec(PGconn *streamConn, const char *query)
 		if (PQresultStatus(lastResult) == PGRES_COPY_IN ||
 			PQresultStatus(lastResult) == PGRES_COPY_OUT ||
 			PQresultStatus(lastResult) == PGRES_COPY_BOTH ||
-			PQstatus(streamConn) == CONNECTION_BAD)
+			PQstatus(conn->streamConn) == CONNECTION_BAD)
 			break;
 	}
 
 	return lastResult;
 }
 
+/*
+ * XXX: Currently the WES is configured to only check if socket is readable,
+ * not writeable. That might need to be configurable in the future.
+ */
+static int
+libpqrcv_wait(WalReceiverConn *conn, long timeout, uint32 wait_event_info)
+{
+	int			nevents;
+	WaitEvent	event;
+	int			ret = 0;
+
+	nevents = WaitEventSetWait(conn->wes, timeout, &event, 1, wait_event_info);
+
+	if (nevents == 0)
+		ret |= WL_TIMEOUT;
+	else
+	{
+		ret |= event.events & (WL_LATCH_SET |
+							   WL_POSTMASTER_DEATH |
+							   WL_SOCKET_MASK);
+	}
+
+	return ret;
+}
+
 /*
  * Perform the equivalent of PQgetResult(), but watch for interrupts.
  */
 static PGresult *
-libpqrcv_PQgetResult(PGconn *streamConn)
+libpqrcv_PQgetResult(WalReceiverConn *conn)
 {
 	/*
 	 * Collect data until PQgetResult is ready to get the result without
 	 * blocking.
 	 */
-	while (PQisBusy(streamConn))
+	while (PQisBusy(conn->streamConn))
 	{
 		int			rc;
 
@@ -705,12 +748,7 @@ libpqrcv_PQgetResult(PGconn *streamConn)
 		 * since we'll get interrupted by signals and can handle any
 		 * interrupts here.
 		 */
-		rc = WaitLatchOrSocket(MyLatch,
-							   WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
-							   WL_LATCH_SET,
-							   PQsocket(streamConn),
-							   0,
-							   WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
+		rc = libpqrcv_wait(conn, -1, WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
 
 		/* Interrupted? */
 		if (rc & WL_LATCH_SET)
@@ -720,7 +758,7 @@ libpqrcv_PQgetResult(PGconn *streamConn)
 		}
 
 		/* Consume whatever data is available from the socket */
-		if (PQconsumeInput(streamConn) == 0)
+		if (PQconsumeInput(conn->streamConn) == 0)
 		{
 			/* trouble; return NULL */
 			return NULL;
@@ -728,7 +766,7 @@ libpqrcv_PQgetResult(PGconn *streamConn)
 	}
 
 	/* Now we can collect and return the next PGresult */
-	return PQgetResult(streamConn);
+	return PQgetResult(conn->streamConn);
 }
 
 /*
@@ -737,6 +775,12 @@ libpqrcv_PQgetResult(PGconn *streamConn)
 static void
 libpqrcv_disconnect(WalReceiverConn *conn)
 {
+	if (conn->wes != NULL)
+	{
+		FreeWaitEventSet(conn->wes);
+		conn->wes = NULL;
+	}
+
 	PQfinish(conn->streamConn);
 	if (conn->recvBuf != NULL)
 		PQfreemem(conn->recvBuf);
@@ -752,16 +796,15 @@ libpqrcv_disconnect(WalReceiverConn *conn)
  *	 point to a buffer holding the received message. The buffer is only valid
  *	 until the next libpqrcv_* call.
  *
- *	 If no data was available immediately, returns 0, and *wait_fd is set to a
- *	 socket descriptor which can be waited on before trying again.
+ *	 If no data was available immediately, returns 0. In that case walrcv_wait()
+ *	 can be used to wait for socket readiness or interrupts before trying again.
  *
  *	 -1 if the server ended the COPY.
  *
  * ereports on error.
  */
 static int
-libpqrcv_receive(WalReceiverConn *conn, char **buffer,
-				 pgsocket *wait_fd)
+libpqrcv_receive(WalReceiverConn *conn, char **buffer)
 {
 	int			rawlen;
 
@@ -785,7 +828,6 @@ libpqrcv_receive(WalReceiverConn *conn, char **buffer,
 		if (rawlen == 0)
 		{
 			/* Tell caller to try again when our socket is ready. */
-			*wait_fd = PQsocket(conn->streamConn);
 			return 0;
 		}
 	}
@@ -793,13 +835,13 @@ libpqrcv_receive(WalReceiverConn *conn, char **buffer,
 	{
 		PGresult   *res;

@@ -941,7 +983,7 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,

 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
@@ -1068,7 +1110,7 @@ libpqrcv_exec(WalReceiverConn *conn, const char *query,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 				 errmsg("the query interface requires a database connection")));

 	switch (PQresultStatus(pgres))
 	{
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index e596b69d466..9eb78d1faab 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -637,14 +637,13 @@ copy_read_data(void *outbuf, int minread, int maxread)
 
 	while (maxread > 0 && bytesread < minread)
 	{
-		pgsocket	fd = PGINVALID_SOCKET;
 		int			len;
 		char	   *buf = NULL;
 
 		for (;;)
 		{
 			/* Try read the data. */
-			len = walrcv_receive(LogRepWorkerWalRcvConn, &buf, &fd);
+			len = walrcv_receive(LogRepWorkerWalRcvConn, &buf);
 
 			CHECK_FOR_INTERRUPTS();
 
@@ -676,11 +675,8 @@ copy_read_data(void *outbuf, int minread, int maxread)
 		/*
 		 * Wait for more data or latch.
 		 */
-		(void) WaitLatchOrSocket(MyLatch,
-								 WL_SOCKET_READABLE | WL_LATCH_SET |
-								 WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
-								 fd, 1000L, WAIT_EVENT_LOGICAL_SYNC_DATA);
-
+		(void) walrcv_wait(LogRepWorkerWalRcvConn, 1000L,
+						   WAIT_EVENT_LOGICAL_SYNC_DATA);
 		ResetLatch(MyLatch);
 	}
 
@@ -967,6 +963,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 	 * application_name, so that it is different from the main apply worker,
 	 * so that synchronous replication can distinguish them.
 	 */
+	Assert(LogRepWorkerWalRcvConn == NULL);
 	LogRepWorkerWalRcvConn =
 		walrcv_connect(MySubscription->conninfo, true, slotname, &err);
 	if (LogRepWorkerWalRcvConn == NULL)
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index c9af775bc18..95d4d0c7d1f 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -2597,7 +2597,6 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 	/* This outer loop iterates once per wait. */
 	for (;;)
 	{
-		pgsocket	fd = PGINVALID_SOCKET;
 		int			rc;
 		int			len;
 		char	   *buf = NULL;
@@ -2608,7 +2607,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 
 		MemoryContextSwitchTo(ApplyMessageContext);
 
-		len = walrcv_receive(LogRepWorkerWalRcvConn, &buf, &fd);
+		len = walrcv_receive(LogRepWorkerWalRcvConn, &buf);
 
 		if (len != 0)
 		{
@@ -2688,7 +2687,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 					MemoryContextReset(ApplyMessageContext);
 				}
 
-				len = walrcv_receive(LogRepWorkerWalRcvConn, &buf, &fd);
+				len = walrcv_receive(LogRepWorkerWalRcvConn, &buf);
 			}
 		}
 
@@ -2729,11 +2728,8 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 		else
 			wait_time = NAPTIME_PER_CYCLE;
 
-		rc = WaitLatchOrSocket(MyLatch,
-							   WL_SOCKET_READABLE | WL_LATCH_SET |
-							   WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
-							   fd, wait_time,
-							   WAIT_EVENT_LOGICAL_APPLY_MAIN);
+		rc = walrcv_wait(LogRepWorkerWalRcvConn, wait_time,
+						 WAIT_EVENT_LOGICAL_APPLY_MAIN);
 
 		if (rc & WL_LATCH_SET)
 		{
@@ -3541,6 +3537,7 @@ ApplyWorkerMain(Datum main_arg)
 		origin_startpos = replorigin_session_get_progress(false);
 		CommitTransactionCommand();
 
+		Assert(LogRepWorkerWalRcvConn == NULL);
 		LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo, true,
 												MySubscription->name, &err);
 		if (LogRepWorkerWalRcvConn == NULL)
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index b39fce8c23c..d70fc462bea 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -423,7 +423,6 @@ WalReceiverMain(void)
 				char	   *buf;
 				int			len;
 				bool		endofwal = false;
-				pgsocket	wait_fd = PGINVALID_SOCKET;
 				int			rc;
 
 				/*
@@ -446,7 +445,7 @@ WalReceiverMain(void)
 				}
 
 				/* See if we can read data immediately */
-				len = walrcv_receive(wrconn, &buf, &wait_fd);
+				len = walrcv_receive(wrconn, &buf);
 				if (len != 0)
 				{
 					/*
@@ -478,7 +477,7 @@ WalReceiverMain(void)
 							endofwal = true;
 							break;
 						}
-						len = walrcv_receive(wrconn, &buf, &wait_fd);
+						len = walrcv_receive(wrconn, &buf);
 					}
 
 					/* Let the primary know that we received some data. */
@@ -496,24 +495,9 @@ WalReceiverMain(void)
 				if (endofwal)
 					break;
 
-				/*
-				 * Ideally we would reuse a WaitEventSet object repeatedly
-				 * here to avoid the overheads of WaitLatchOrSocket on epoll
-				 * systems, but we can't be sure that libpq (or any other
-				 * walreceiver implementation) has the same socket (even if
-				 * the fd is the same number, it may have been closed and
-				 * reopened since the last time).  In future, if there is a
-				 * function for removing sockets from WaitEventSet, then we
-				 * could add and remove just the socket each time, potentially
-				 * avoiding some system calls.
-				 */
-				Assert(wait_fd != PGINVALID_SOCKET);
-				rc = WaitLatchOrSocket(MyLatch,
-									   WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
-									   WL_TIMEOUT | WL_LATCH_SET,
-									   wait_fd,
-									   NAPTIME_PER_CYCLE,
-									   WAIT_EVENT_WAL_RECEIVER_MAIN);
+				rc = walrcv_wait(wrconn, NAPTIME_PER_CYCLE,
+								 WAIT_EVENT_WAL_RECEIVER_MAIN);
+
 				if (rc & WL_LATCH_SET)
 				{
 					ResetLatch(MyLatch);
-- 
2.34.0

---

From: Tom Lane <[email-masked]>
Date: Fri, 14 Jan 2022 17:51:52 -0500

> better idea for now, and then we could fix all this stuff going

FWIW, I'm just fine with reverting, particularly in the back branches.
It seems clear that this dank corner of Windows contains even more
creepy-crawlies than we thought.

			regards, tom lane

---

From: Thomas Munro <[email-masked]>
Date: Sat, 15 Jan 2022 13:19:42 +1300

> Alexander, could you test with that patch applied?

Isn't your patch nearly identical to one that I already posted, that
Alexander tested and reported success with here?

https://www.postgresql.org/message-id/5d507424-13ce-d19f-2f5d-ab4c6a987316%40gmail.com

I can believe that fixes walreceiver (if we're sure that there isn't a
libpq-changes-the-socket problem), but AFAICS the same problem exists
for postgres_fdw and async append.  That's why I moved to trying to
fix the multiple-WES thing (though of course I agree we should be
using long lived WESes wherever possible, I just didn't think that
seemed back-patchable, so it's more of a feature patch for the
future).

---

From: Thomas Munro <[email-masked]>
Date: Sat, 15 Jan 2022 13:40:59 +1300

On Sat, Jan 15, 2022 at 9:47 AM Andres Freund <[email-masked]> wrote:
> result of?

Erm, I didn't analyse the situation much back then, I just knew that
libpq could reconnect in early phases.  I can see that once you reach
that stage you can count on socket stability though, so yeah that
should work as long as you can handle it correctly in the earlier
connection phase (probably using the other patch I posted and
Alexander tested), it should all work nicely.  You'd probably want to
formalise the interface/documentation on that point.

> Nor do I understand what "any other walreceiver implementation"
> refers to?

I think I meant that it goes via function pointers to talk to
libpqwalreceiver.c, but I know now that we don't actually support
using that to switch to different code, it's just a solution to a
backend/frontend linking problem.  The comment was probably just
paranoia based on the way the interface works.

---

From: Andres Freund <[email-masked]>
Date: Fri, 14 Jan 2022 16:51:01 -0800

Hi,

> Alexander tested and reported success with here?

Sorry, somehow I missed that across the many patches in the thread... And yes,
it does look remarkably similar.

> I can believe that fixes walreceiver

One thing that still bothers me around this is that we didn't detect the
problem of the dead walreceiver connection, even after missing the
FD_CLOSE. There's plenty other ways that a connection can get stalled for
prolonged was, that we'd not get notified about either. That's why there's
wal_receiver_timeout, after all.

But from what I can see wal_receiver_timeout doesn't work even halfway
reliably, because of all the calls down to libpqrcv_PQgetResult, where we just
block indefinitely?

This actually seems like a significant issue to me, and not just on windows.

> (if we're sure that there isn't a libpq-changes-the-socket problem)

I just don't see what that problem could be, once connection is
established. The only way a the socket fd could change is a reconnect, which
doesn't happen automatically.

I actually was searching the archives for threads on it, but I didn't find
much besides the references around [1]. And I didn't see a concrete risk
explained there?

> but AFAICS the same problem exists for postgres_fdw and async append.

Perhaps - but I suspect it'll matter far less with them than with walreceiver.

> That's why I moved to trying to
> using long lived WESes wherever possible

That approach seems like a very very leaky bandaid, with a decent potential
for unintended consequences. Perhaps there's nothing better that we can do,
but I'd rather try to fix the problem closer to the root...

> I just didn't think that seemed back-patchable, so it's more of a feature
> patch for the future).

Hm, it doesn't seem crazy invasive to me. But I guess we might be looking at a
revert of the shutdown changes for now anyway? In that case we should be
fixing this anyway, but we might be able to afford doing it in master only?

Greetings,

Andres Freund

[1] https://www.postgresql.org/message-id/CA%2BhUKG%2BzCNJZBXcURPdQvdY-tjyD0y7Li2wZEC6XChyUej1S5w%40mail.gmail.com

---

From: Alexander Lakhin <[email-masked]>
Date: Sun, 16 Jan 2022 14:00:00 +0300

Hello Andres,
> Alexander, could you test with that patch applied?
Unfortunately, this patch doesn't fix the issue. The test
commit_ts/002_standby still fails (on iterations 3, 1, 8 for me).
With the debugging logging added:
...
elog(LOG, "libpqrcv_wait() before WaitEventSetWait(%p)", conn->wes);
    nevents = WaitEventSetWait(conn->wes, timeout, &event, 1,
wait_event_info);
elog(LOG, "libpqrcv_wait() after WaitEventSetWait");
...
elog(LOG, "WaitEventSetWaitBlock before WaitForMultipleObjects(%p)",
set->handles);
    rc = WaitForMultipleObjects(set->nevents + 1, set->handles, FALSE,
                                cur_timeout);
elog(LOG, "WaitEventSetWaitBlock aftet WaitForMultipleObjects");
...
and so on.

I see in 002_standby_standby.log
...
2022-01-16 13:31:36.244 MSK [1336] LOG:  libpqrcv_receive PQgetCopyData
returned: 145
2022-01-16 13:31:36.244 MSK [1336] LOG:  libpqrcv_receive PQgetCopyData
returned: 0
2022-01-16 13:31:36.244 MSK [1336] LOG:  libpqrcv_wait() before
WaitEventSetWait(000000000063ABA8)
2022-01-16 13:31:36.244 MSK [1336] LOG:  WaitEventSetWait before
WaitEventSetWaitBlock(000000000063ABA8)
2022-01-16 13:31:36.244 MSK [1336] LOG:  WaitEventSetWaitBlock before
WaitForMultipleObjects(000000000063AC30)
2022-01-16 13:31:36.244 MSK [2820] LOG:  WaitEventSetWaitBlock aftet
WaitForMultipleObjects
2022-01-16 13:31:36.244 MSK [2820] LOG:  WaitEventSetWait after
WaitEventSetWaitBlock
2022-01-16 13:31:36.244 MSK [1336] LOG:  WaitEventSetWaitBlock aftet
WaitForMultipleObjects
2022-01-16 13:31:36.244 MSK [1336] LOG:  WaitEventSetWait after
WaitEventSetWaitBlock
2022-01-16 13:31:36.244 MSK [1336] LOG:  libpqrcv_wait() after
WaitEventSetWait
2022-01-16 13:31:36.244 MSK [1336] LOG:  libpqrcv_receive PQgetCopyData
returned: 0
2022-01-16 13:31:36.244 MSK [1336] LOG:  libpqrcv_wait() before
WaitEventSetWait(000000000063ABA8)
2022-01-16 13:31:36.244 MSK [1336] LOG:  WaitEventSetWait before
WaitEventSetWaitBlock(000000000063ABA8)
2022-01-16 13:31:36.244 MSK [1336] LOG:  WaitEventSetWaitBlock before
WaitForMultipleObjects(000000000063AC30)
2022-01-16 13:31:36.244 MSK [2820] LOG:  WaitEventSetWait before
WaitEventSetWaitBlock(0000000000649FB8)
2022-01-16 13:31:36.244 MSK [2820] LOG:  WaitEventSetWaitBlock before
WaitForMultipleObjects(000000000064A020)
2022-01-16 13:31:36.247 MSK [1336] LOG:  WaitEventSetWaitBlock aftet
WaitForMultipleObjects
2022-01-16 13:31:36.247 MSK [1336] LOG:  WaitEventSetWait after
WaitEventSetWaitBlock
2022-01-16 13:31:36.247 MSK [1336] LOG:  libpqrcv_wait() after
WaitEventSetWait
2022-01-16 13:31:36.247 MSK [1336] LOG:  libpqrcv_receive PQgetCopyData
returned: -1
2022-01-16 13:31:36.247 MSK [1336] LOG:  libpqrcv_receive before
libpqrcv_PQgetResult(1)
2022-01-16 13:31:36.247 MSK [1336] LOG:  libpqrcv_receive
libpqrcv_PQgetResult(1) returned 0000000000692400
2022-01-16 13:31:36.247 MSK [1336] LOG:  libpqrcv_receive before
libpqrcv_PQgetResult(2)
2022-01-16 13:31:36.247 MSK [1336] LOG:  libpqrcv_wait() before
WaitEventSetWait(000000000063ABA8)
2022-01-16 13:31:36.247 MSK [1336] LOG:  WaitEventSetWait before
WaitEventSetWaitBlock(000000000063ABA8)
2022-01-16 13:31:36.247 MSK [1336] LOG:  WaitEventSetWaitBlock before
WaitForMultipleObjects(000000000063AC30)
2022-01-16 13:31:36.368 MSK [984] LOG:  WaitEventSetWaitBlock aftet
WaitForMultipleObjects
2022-01-16 13:31:36.368 MSK [984] LOG:  WaitEventSetWait after
WaitEventSetWaitBlock
...
After that, the process 1336 hangs till shutdown.

Best regards,
Alexander

---

From: Andres Freund <[email-masked]>
Date: Mon, 24 Jan 2022 12:02:22 -0800

Hi,

> creepy-crawlies than we thought.

Seems we should revert now-ish? There's a minor release coming up and I think
it'd be bad to ship these changes to users.

Greetings,

Andres Freund

---

From: Tom Lane <[email-masked]>
Date: Mon, 24 Jan 2022 15:35:25 -0500

> it'd be bad to ship these changes to users.

Sure.  Do we want to revert in HEAD too?

			regards, tom lane

---

From: Andres Freund <[email-masked]>
Date: Mon, 24 Jan 2022 13:28:13 -0800

Hi,

> Sure.  Do we want to revert in HEAD too?

Not sure. I'm also OK with trying to go with Thomas' patch to walreceiver and
try a bit longer to get all this working. Thomas?

Greetings,

Andres Freund

---

From: Noah Misch <[email-masked]>
Date: Sat, 19 Mar 2022 01:47:04 -0700

> standby directly instead of relying on the upstream's view.

It would be nice.  For logical replication tests, do we have a monitoring API
independent of the stats collector?  If not and we don't want to add one, a
hacky alternative might be for wait_for_catchup to run a WAL-writing command
every ~20s.  That way, if the stats collector misses the datagram about the
standby reaching a certain LSN, the stats collector would have more chances.

---

From: Thomas Munro <[email-masked]>
Date: Tue, 22 Mar 2022 15:16:06 +1300

I have a new socket abstraction patch that should address the known
Windows socket/event bugs, but it's a little bigger than I thought it
would be, not quite ready, and now too late to expect people to review
for 15, so I think it should go into the next cycle.  I've bounced
https://commitfest.postgresql.org/37/3523/ into the next CF.  We'll
need to do something like 75674c7e for master.

---

From: Tom Lane <[email-masked]>
Date: Mon, 21 Mar 2022 23:13:31 -0400

> need to do something like 75674c7e for master.

OK.  You want me to push 75674c7e to HEAD?

			regards, tom lane