public inbox for [email protected]  

help / color / mirror / Atom feed

[PATCH 1/1] pg_rewind: Fetch small files according to new size.
23+ messages / 2 participants
[nested] [flat]

* [PATCH 1/1] pg_rewind: Fetch small files according to new size.
@ 2020-11-13 09:04  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2020-11-13 09:04 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 47beba277a4..f544d9828a2 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index fa1b6e80ec3..c1ad2c49c5f 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that we size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 2bbf8e74385..fe828c9e840 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -525,10 +525,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index e87f239a47a..a8dc9fda0da 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.20.1


--------------70A2CAC3C3DFA320063B7D3B--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* [PATCH v2 1/1] pg_rewind: Fetch small files according to new size.
@ 2021-01-22 13:16  Heikki Linnakangas <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Heikki Linnakangas @ 2021-01-22 13:16 UTC (permalink / raw)

There's a race condition if a file changes in the source system after we
have collected the file list. If the file becomes larger, we only fetched
up to its original size. That can easily result in a truncated file.
That's not a problem for relation files, files in pg_xact, etc. because
any actions on them will be replayed from the WAL. However, configuration
files are affected.

This commit mitigates the race condition by fetching small files in
whole, even if they have grown. This is not a full fix: we still believe
the original file size for files larger than 1 MB. That should be enough
for configuration files, and doing more than that would require bigger
changes to the chunking logic in in libpq_source.c.

That mitigates the race condition if the file is modified between the
original scan of files and copying the file, but there's still a race
condition if a file is changed while it's being copied. That's a much
smaller window, though, and pg_basebackup has the same issue.

I ran into this while playing with pg_auto_failover, which frequently
uses ALTER SYSTEM, which update postgresql.auto.conf. Often, pg_rewind
would fail, because the postgresql.auto.conf file changed concurrently
and a partial version of it was copied to the target. The partial
file would fail to parse, preventing the server from starting up.

Reviewed-by: Cary Huang
Discussion: https://www.postgresql.org/message-id/f67feb24-5833-88cb-1020-19a4a2b83ac7%40iki.fi
---
 src/bin/pg_rewind/libpq_source.c  | 32 +++++++++++++
 src/bin/pg_rewind/local_source.c  | 76 +++++++++++++++++++++++++++----
 src/bin/pg_rewind/pg_rewind.c     |  5 +-
 src/bin/pg_rewind/rewind_source.h | 13 ++++++
 4 files changed, 112 insertions(+), 14 deletions(-)

diff --git a/src/bin/pg_rewind/libpq_source.c b/src/bin/pg_rewind/libpq_source.c
index 86d2adcaee9..ff16add16f5 100644
--- a/src/bin/pg_rewind/libpq_source.c
+++ b/src/bin/pg_rewind/libpq_source.c
@@ -63,6 +63,7 @@ static void process_queued_fetch_requests(libpq_source *src);
 /* public interface functions */
 static void libpq_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
+static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
 static void libpq_queue_fetch_range(rewind_source *source, const char *path,
 									off_t off, size_t len);
 static void libpq_finish_fetch(rewind_source *source);
@@ -88,6 +89,7 @@ init_libpq_source(PGconn *conn)
 
 	src->common.traverse_files = libpq_traverse_files;
 	src->common.fetch_file = libpq_fetch_file;
+	src->common.queue_fetch_file = libpq_queue_fetch_file;
 	src->common.queue_fetch_range = libpq_queue_fetch_range;
 	src->common.finish_fetch = libpq_finish_fetch;
 	src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
@@ -307,6 +309,36 @@ libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
 	PQclear(res);
 }
 
+/*
+ * Queue up a request to fetch a file from remote system.
+ */
+static void
+libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	/*
+	 * Truncate the target file immediately, and queue a request to fetch it
+	 * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
+	 * request fetching a full-sized chunk anyway, so that if the file has
+	 * become larger in the source system, after we scanned the source
+	 * directory, we still fetch the whole file. This only works for files up
+	 * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
+	 * and such that are changed every now and then, but not WAL-logged.
+	 * For larger files, we fetch up to the original size.
+	 *
+	 * Even with that mechanism, there is an inherent race condition if the
+	 * file is modified at the same instant that we're copying it, so that we
+	 * might copy a torn version of the file with one half from the old
+	 * version and another half from the new. But pg_basebackup has the same
+	 * problem, and it hasn't been problem in practice.
+	 *
+	 * It might seem more natural to truncate the file later, when we receive
+	 * it from the source server, but then we'd need to track which
+	 * fetch-requests are for a whole file.
+	 */
+	open_target_file(path, true);
+	libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
+}
+
 /*
  * Queue up a request to fetch a piece of a file from remote system.
  */
diff --git a/src/bin/pg_rewind/local_source.c b/src/bin/pg_rewind/local_source.c
index 9c3491c3fba..1899d1cc4ae 100644
--- a/src/bin/pg_rewind/local_source.c
+++ b/src/bin/pg_rewind/local_source.c
@@ -29,8 +29,10 @@ static void local_traverse_files(rewind_source *source,
 								 process_file_callback_t callback);
 static char *local_fetch_file(rewind_source *source, const char *path,
 							  size_t *filesize);
-static void local_fetch_file_range(rewind_source *source, const char *path,
-								   off_t off, size_t len);
+static void local_queue_fetch_file(rewind_source *source, const char *path,
+								   size_t len);
+static void local_queue_fetch_range(rewind_source *source, const char *path,
+									off_t off, size_t len);
 static void local_finish_fetch(rewind_source *source);
 static void local_destroy(rewind_source *source);
 
@@ -43,7 +45,8 @@ init_local_source(const char *datadir)
 
 	src->common.traverse_files = local_traverse_files;
 	src->common.fetch_file = local_fetch_file;
-	src->common.queue_fetch_range = local_fetch_file_range;
+	src->common.queue_fetch_file = local_queue_fetch_file;
+	src->common.queue_fetch_range = local_queue_fetch_range;
 	src->common.finish_fetch = local_finish_fetch;
 	src->common.get_current_wal_insert_lsn = NULL;
 	src->common.destroy = local_destroy;
@@ -65,12 +68,65 @@ local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
 	return slurpFile(((local_source *) source)->datadir, path, filesize);
 }
 
+/*
+ * Copy a file from source to target.
+ *
+ * 'len' is the expected length of the file.
+ */
+static void
+local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
+{
+	const char *datadir = ((local_source *) source)->datadir;
+	PGAlignedBlock buf;
+	char		srcpath[MAXPGPATH];
+	int			srcfd;
+	size_t		written_len;
+
+	snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
+
+	/* Open source file for reading. */
+	srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
+	if (srcfd < 0)
+		pg_fatal("could not open source file \"%s\": %m",
+				 srcpath);
+
+	/* Truncate and open the target file for writing. */
+	open_target_file(path, true);
+
+	written_len = 0;
+	for (;;)
+	{
+		ssize_t		read_len;
+
+		read_len = read(srcfd, buf.data, sizeof(buf));
+
+		if (read_len < 0)
+			pg_fatal("could not read file \"%s\": %m", srcpath);
+		else if (read_len == 0)
+			break;	/* EOF reached */
+
+		write_target_range(buf.data, written_len, read_len);
+		written_len += read_len;
+	}
+
+	/*
+	 * A local source is not expected to change while we're rewinding, so check
+	 * that the size of the file matches our earlier expectation.
+	 */
+	if (written_len != len)
+		pg_fatal("size of source file \"%s\" changed concurrently: " UINT64_FORMAT " bytes expected, " UINT64_FORMAT " copied",
+				 srcpath, len, written_len);
+
+	if (close(srcfd) != 0)
+		pg_fatal("could not close file \"%s\": %m", srcpath);
+}
+
 /*
  * Copy a file from source to target, starting at 'off', for 'len' bytes.
  */
 static void
-local_fetch_file_range(rewind_source *source, const char *path, off_t off,
-					   size_t len)
+local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
+						size_t len)
 {
 	const char *datadir = ((local_source *) source)->datadir;
 	PGAlignedBlock buf;
@@ -94,14 +150,14 @@ local_fetch_file_range(rewind_source *source, const char *path, off_t off,
 	while (end - begin > 0)
 	{
 		ssize_t		readlen;
-		size_t		len;
+		size_t		thislen;
 
 		if (end - begin > sizeof(buf))
-			len = sizeof(buf);
+			thislen = sizeof(buf);
 		else
-			len = end - begin;
+			thislen = end - begin;
 
-		readlen = read(srcfd, buf.data, len);
+		readlen = read(srcfd, buf.data, thislen);
 
 		if (readlen < 0)
 			pg_fatal("could not read file \"%s\": %m", srcpath);
@@ -120,7 +176,7 @@ static void
 local_finish_fetch(rewind_source *source)
 {
 	/*
-	 * Nothing to do, local_fetch_file_range() copies the ranges immediately.
+	 * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
 	 */
 }
 
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 359a6a587cb..9030a1505e3 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -537,10 +537,7 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 				break;
 
 			case FILE_ACTION_COPY:
-				/* Truncate the old file out of the way, if any */
-				open_target_file(entry->path, true);
-				source->queue_fetch_range(source, entry->path,
-										  0, entry->source_size);
+				source->queue_fetch_file(source, entry->path, entry->source_size);
 				break;
 
 			case FILE_ACTION_TRUNCATE:
diff --git a/src/bin/pg_rewind/rewind_source.h b/src/bin/pg_rewind/rewind_source.h
index 2da92dbff94..799b7c120ea 100644
--- a/src/bin/pg_rewind/rewind_source.h
+++ b/src/bin/pg_rewind/rewind_source.h
@@ -47,6 +47,19 @@ typedef struct rewind_source
 	void		(*queue_fetch_range) (struct rewind_source *, const char *path,
 									  off_t offset, size_t len);
 
+	/*
+	 * Like queue_fetch_range(), but requests replacing the whole local file
+	 * from the source system. 'len' is the expected length of the file,
+	 * although when the source is a live server, the file may change
+	 * concurrently. The implementation is not obliged to copy more than 'len'
+	 * bytes, even if the file is larger. However, to avoid copying a
+	 * truncated version of the file, which can cause trouble if e.g. a
+	 * configuration file is modified concurrently, the implementation should
+	 * try to copy the whole file, even if it's larger than expected.
+	 */
+	void		(*queue_fetch_file) (struct rewind_source *, const char *path,
+									 size_t len);
+
 	/*
 	 * Execute all requests queued up with queue_fetch_range().
 	 */
-- 
2.29.2


--------------79EE74DA33DF3081397629A9--





^ permalink  raw  reply  [nested|flat] 23+ messages in thread

* pg_usleep for multisecond delays
@ 2023-02-09 20:59  Nathan Bossart <[email protected]>
  0 siblings, 0 replies; 23+ messages in thread

From: Nathan Bossart @ 2023-02-09 20:59 UTC (permalink / raw)
  To: pgsql-hackers

I just found myself carefully counting the zeros in a call to pg_usleep().
Besides getting my eyes checked, perhaps there should be a wrapper called
pg_ssleep() than can be used for multisecond sleeps.  Or maybe the
USECS_PER_SEC macro should be used more widely.  I attached a patch for the
former approach.  I don't have a strong opinion, but I do think it's worth
improving readability a bit here.

-- 
Nathan Bossart
Amazon Web Services: https://aws.amazon.com


Attachments:

  [text/x-diff] pg_ssleep.patch (8.0K, ../../20230209205929.GA720594@nathanxps13/2-pg_ssleep.patch)
  download | inline diff:
diff --git a/src/backend/access/nbtree/nbtpage.c b/src/backend/access/nbtree/nbtpage.c
index 3feee28d19..63de896cae 100644
--- a/src/backend/access/nbtree/nbtpage.c
+++ b/src/backend/access/nbtree/nbtpage.c
@@ -2976,7 +2976,7 @@ _bt_pendingfsm_finalize(Relation rel, BTVacState *vstate)
 	 * never be effective without some other backend concurrently consuming an
 	 * XID.
 	 */
-	pg_usleep(5000000L);
+	pg_ssleep(5);
 #endif
 
 	/*
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index f9f0f6db8d..87664045d0 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -5103,7 +5103,7 @@ StartupXLOG(void)
 	/* This is just to allow attaching to startup process with a debugger */
 #ifdef XLOG_REPLAY_DELAY
 	if (ControlFile->state != DB_SHUTDOWNED)
-		pg_usleep(60000000L);
+		pg_ssleep(60);
 #endif
 
 	/*
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index ff6149a179..744adff984 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -441,7 +441,7 @@ AutoVacLauncherMain(int argc, char *argv[])
 			(errmsg_internal("autovacuum launcher started")));
 
 	if (PostAuthDelay)
-		pg_usleep(PostAuthDelay * 1000000L);
+		pg_ssleep(PostAuthDelay);
 
 	SetProcessingMode(InitProcessing);
 
@@ -561,7 +561,7 @@ AutoVacLauncherMain(int argc, char *argv[])
 		 * Sleep at least 1 second after any error.  We don't want to be
 		 * filling the error logs as fast as we can.
 		 */
-		pg_usleep(1000000L);
+		pg_ssleep(1);
 	}
 
 	/* We can now handle ereport(ERROR) */
@@ -687,7 +687,7 @@ AutoVacLauncherMain(int argc, char *argv[])
 				 * of a worker will continue to fail in the same way.
 				 */
 				AutoVacuumShmem->av_signal[AutoVacForkFailed] = false;
-				pg_usleep(1000000L);	/* 1s */
+				pg_ssleep(1);
 				SendPostmasterSignal(PMSIGNAL_START_AUTOVAC_WORKER);
 				continue;
 			}
@@ -1708,7 +1708,7 @@ AutoVacWorkerMain(int argc, char *argv[])
 				(errmsg_internal("autovacuum: processing database \"%s\"", dbname)));
 
 		if (PostAuthDelay)
-			pg_usleep(PostAuthDelay * 1000000L);
+			pg_ssleep(PostAuthDelay);
 
 		/* And do an appropriate amount of work */
 		recentXid = ReadNextTransactionId();
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 0dd22b2351..6d38dfeeba 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -759,7 +759,7 @@ StartBackgroundWorker(void)
 
 	/* Apply PostAuthDelay */
 	if (PostAuthDelay > 0)
-		pg_usleep(PostAuthDelay * 1000000L);
+		pg_ssleep(PostAuthDelay);
 
 	/*
 	 * Set up signal handlers.
diff --git a/src/backend/postmaster/bgwriter.c b/src/backend/postmaster/bgwriter.c
index 9bb47da404..147f9b1e38 100644
--- a/src/backend/postmaster/bgwriter.c
+++ b/src/backend/postmaster/bgwriter.c
@@ -196,7 +196,7 @@ BackgroundWriterMain(void)
 		 * to be repeated, and we don't want to be filling the error logs as
 		 * fast as we can.
 		 */
-		pg_usleep(1000000L);
+		pg_ssleep(1);
 
 		/*
 		 * Close all open files after any error.  This is helpful on Windows,
diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c
index aaad5c5228..12786229dd 100644
--- a/src/backend/postmaster/checkpointer.c
+++ b/src/backend/postmaster/checkpointer.c
@@ -310,7 +310,7 @@ CheckpointerMain(void)
 		 * to be repeated, and we don't want to be filling the error logs as
 		 * fast as we can.
 		 */
-		pg_usleep(1000000L);
+		pg_ssleep(1);
 
 		/*
 		 * Close all open files after any error.  This is helpful on Windows,
diff --git a/src/backend/postmaster/pgarch.c b/src/backend/postmaster/pgarch.c
index e551af2905..6460c69726 100644
--- a/src/backend/postmaster/pgarch.c
+++ b/src/backend/postmaster/pgarch.c
@@ -450,7 +450,7 @@ pgarch_ArchiverCopyLoop(void)
 				}
 
 				/* wait a bit before retrying */
-				pg_usleep(1000000L);
+				pg_ssleep(1);
 				continue;
 			}
 
@@ -482,7 +482,7 @@ pgarch_ArchiverCopyLoop(void)
 									xlog)));
 					return;		/* give up archiving for now */
 				}
-				pg_usleep(1000000L);	/* wait a bit before retrying */
+				pg_ssleep(1);	/* wait a bit before retrying */
 			}
 		}
 	}
diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c
index 2552327d90..6b80f423a8 100644
--- a/src/backend/postmaster/postmaster.c
+++ b/src/backend/postmaster/postmaster.c
@@ -4292,7 +4292,7 @@ BackendInitialize(Port *port)
 	 * is not honored until after authentication.)
 	 */
 	if (PreAuthDelay > 0)
-		pg_usleep(PreAuthDelay * 1000000L);
+		pg_ssleep(PreAuthDelay);
 
 	/* This flag will remain set until InitPostgres finishes authentication */
 	ClientAuthInProgress = true;	/* limit visibility of log messages */
diff --git a/src/backend/postmaster/walwriter.c b/src/backend/postmaster/walwriter.c
index 513e580c51..a003717efd 100644
--- a/src/backend/postmaster/walwriter.c
+++ b/src/backend/postmaster/walwriter.c
@@ -189,7 +189,7 @@ WalWriterMain(void)
 		 * to be repeated, and we don't want to be filling the error logs as
 		 * fast as we can.
 		 */
-		pg_usleep(1000000L);
+		pg_ssleep(1);
 
 		/*
 		 * Close all open files after any error.  This is helpful on Windows,
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index 2f07ca7a0e..5647795f4a 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -973,7 +973,7 @@ InitPostgres(const char *in_dbname, Oid dboid,
 
 		/* Apply PostAuthDelay as soon as we've read all options */
 		if (PostAuthDelay > 0)
-			pg_usleep(PostAuthDelay * 1000000L);
+			pg_ssleep(PostAuthDelay);
 
 		/* initialize client encoding */
 		InitializeClientEncoding();
@@ -1180,7 +1180,7 @@ InitPostgres(const char *in_dbname, Oid dboid,
 
 	/* Apply PostAuthDelay as soon as we've read all options */
 	if (PostAuthDelay > 0)
-		pg_usleep(PostAuthDelay * 1000000L);
+		pg_ssleep(PostAuthDelay);
 
 	/*
 	 * Initialize various default states that can't be set up until we've
diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c
index f3c7937a1d..ca8bfac936 100644
--- a/src/bin/pg_basebackup/pg_recvlogical.c
+++ b/src/bin/pg_basebackup/pg_recvlogical.c
@@ -991,7 +991,7 @@ main(int argc, char **argv)
 			/* translator: check source for value for %d */
 			pg_log_info("disconnected; waiting %d seconds to try again",
 						RECONNECT_SLEEP_TIME);
-			pg_usleep(RECONNECT_SLEEP_TIME * 1000000);
+			pg_ssleep(RECONNECT_SLEEP_TIME);
 		}
 	}
 }
diff --git a/src/bin/pg_upgrade/exec.c b/src/bin/pg_upgrade/exec.c
index 5b2edebe41..a1fc352eb8 100644
--- a/src/bin/pg_upgrade/exec.c
+++ b/src/bin/pg_upgrade/exec.c
@@ -151,7 +151,7 @@ exec_prog(const char *log_filename, const char *opt_log_file,
 
 		for (iter = 0; iter < 4 && log == NULL; iter++)
 		{
-			pg_usleep(1000000); /* 1 sec */
+			pg_ssleep(1);
 			log = fopen(log_file, "a");
 		}
 	}
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 44b5c8726e..610ebc777c 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -1217,7 +1217,7 @@ main(int argc, char **argv)
 				break;
 			else
 			{
-				pg_usleep(1000000L);	/* 1 second */
+				pg_ssleep(1);
 				continue;
 			}
 		}
diff --git a/src/include/port.h b/src/include/port.h
index e66193bed9..43ce931766 100644
--- a/src/include/port.h
+++ b/src/include/port.h
@@ -163,6 +163,11 @@ extern int	pg_disable_aslr(void);
 /* Portable delay handling */
 extern void pg_usleep(long microsec);
 
+static inline void pg_ssleep(long sec)
+{
+	pg_usleep(sec * 1000000L);
+}
+
 /* Portable SQL-like case-independent comparisons and conversions */
 extern int	pg_strcasecmp(const char *s1, const char *s2);
 extern int	pg_strncasecmp(const char *s1, const char *s2, size_t n);
diff --git a/src/test/regress/pg_regress.c b/src/test/regress/pg_regress.c
index 6cd5998b9d..018237d750 100644
--- a/src/test/regress/pg_regress.c
+++ b/src/test/regress/pg_regress.c
@@ -2386,7 +2386,7 @@ regression_main(int argc, char *argv[],
 				exit(2);
 			}
 
-			pg_usleep(1000000L);
+			pg_ssleep(1);
 		}
 		if (i >= wait_seconds)
 		{


^ permalink  raw  reply  [nested|flat] 23+ messages in thread

end of thread, other threads:[~2023-02-09 20:59 UTC | newest]

Thread overview: 23+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2020-11-13 09:04 [PATCH 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2021-01-22 13:16 [PATCH v2 1/1] pg_rewind: Fetch small files according to new size. Heikki Linnakangas <[email protected]>
2023-02-09 20:59 pg_usleep for multisecond delays Nathan Bossart <[email protected]>

This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox