v14-0011-pg_verifybackup-Read-tar-files-and-verify-its-co.patch

From 68a85869b9a8c3b872035661df85976c890116f4 Mon Sep 17 00:00:00 2001
From: Amul Sul <amul.sul@enterprisedb.com>
Date: Wed, 21 Aug 2024 12:49:04 +0530
Subject: [PATCH v14 11/12] pg_verifybackup: Read tar files and verify its
 contents

This patch implements TAR format backup verification.

For progress reporting support, we perform this verification in two
passes: the first pass calculates total_size, and the second pass
updates done_size as verification progresses.

For the verification, in the first pass, we call precheck_tar_backup_file(),
which performs basic verification by expecting only base.tar, pg_wal.tar, or
<tablespaceoid>.tar files and raises an error for any other files.  It
also determines the compression type of the archive file. All this
information is stored in a newly added tarFile struct, which is
appended to a list that will be used in the second pass for the final
verification. In the second pass, the tar archives are read,
decompressed, and the required verification is carried out.

For reading and decompression, fe_utils/astreamer.h is used. For
verification, a new archive streamer has been added in
astreamer_verify.c to handle TAR member files and their contents; see
astreamer_verify_content() for details. The stack of astreamers will
be set up for each TAR file in verify_tar_content(), depending on its
compression type which is detected in the first pass.

When information about a TAR member file (i.e., ASTREAMER_MEMBER_HEADER)
is received, we first verify its entry against the backup manifest. We
then decide if further checks are needed, such as checksum
verification and control data verification (if it is a pg_control
file), once the member file contents are received. Although this
decision could be made when the contents are received, it is more
efficient to make it earlier since the member file contents are
received in multiple iterations. In short, we process
ASTREAMER_MEMBER_CONTENTS multiple times but only once for other
ASTREAMER_MEMBER_* cases. We maintain this information in the
astreamer_verify structure for each member file, which is reset when
the file ends.

Unlike in a plain backup, checksum verification here occurs in two
steps. First, as the contents are received, the checksum is computed
incrementally (see member_compute_checksum). Then, at the end of
processing the member file, the final verification is performed (see
member_verify_checksum).

Similarly, during the content receiving stage, if the file is
pg_control, the data will be copied into a local buffer (see
member_copy_control_data).  The verification will then be carried out
at the end of the member file processing (see member_verify_control_data)
---
 src/bin/pg_verifybackup/Makefile           |   2 +
 src/bin/pg_verifybackup/astreamer_verify.c | 358 +++++++++++++++++++++
 src/bin/pg_verifybackup/meson.build        |   1 +
 src/bin/pg_verifybackup/pg_verifybackup.c  | 326 ++++++++++++++++++-
 src/bin/pg_verifybackup/pg_verifybackup.h  |   6 +
 src/tools/pgindent/typedefs.list           |   2 +
 6 files changed, 692 insertions(+), 3 deletions(-)
 create mode 100644 src/bin/pg_verifybackup/astreamer_verify.c

diff --git a/src/bin/pg_verifybackup/Makefile b/src/bin/pg_verifybackup/Makefile
index 7c045f142e8..374d4a8afd1 100644
--- a/src/bin/pg_verifybackup/Makefile
+++ b/src/bin/pg_verifybackup/Makefile
@@ -17,10 +17,12 @@ top_builddir = ../../..
 include $(top_builddir)/src/Makefile.global
 
 # We need libpq only because fe_utils does.
+override CPPFLAGS := -I$(libpq_srcdir) $(CPPFLAGS)
 LDFLAGS_INTERNAL += -L$(top_builddir)/src/fe_utils -lpgfeutils $(libpq_pgport)
 
 OBJS = \
 	$(WIN32RES) \
+	astreamer_verify.o \
 	pg_verifybackup.o
 
 all: pg_verifybackup
diff --git a/src/bin/pg_verifybackup/astreamer_verify.c b/src/bin/pg_verifybackup/astreamer_verify.c
new file mode 100644
index 00000000000..b496e9320ea
--- /dev/null
+++ b/src/bin/pg_verifybackup/astreamer_verify.c
@@ -0,0 +1,358 @@
+/*-------------------------------------------------------------------------
+ *
+ * astreamer_verify.c
+ *
+ * Extend fe_utils/astreamer.h archive streaming facility to verify TAR
+ * format backup.
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ *
+ * src/bin/pg_verifybackup/astreamer_verify.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres_fe.h"
+
+#include "pg_verifybackup.h"
+
+typedef struct astreamer_verify
+{
+	astreamer	base;
+	verifier_context *context;
+	char	   *archive_name;
+	Oid			tblspc_oid;
+	pg_checksum_context *checksum_ctx;
+
+	/* Hold information for a member file verification */
+	manifest_file *mfile;
+	int64		received_bytes;
+	bool		verify_checksum;
+	bool		verify_control_data;
+} astreamer_verify;
+
+static void astreamer_verify_content(astreamer *streamer,
+									 astreamer_member *member,
+									 const char *data, int len,
+									 astreamer_archive_context context);
+static void astreamer_verify_finalize(astreamer *streamer);
+static void astreamer_verify_free(astreamer *streamer);
+
+static void member_verify_header(astreamer *streamer, astreamer_member *member);
+static void member_compute_checksum(astreamer *streamer,
+									astreamer_member *member,
+									const char *data, int len);
+static void member_verify_checksum(astreamer *streamer);
+static void member_copy_control_data(astreamer *streamer,
+									 astreamer_member *member,
+									 const char *data, int len);
+static void member_verify_control_data(astreamer *streamer);
+static void member_reset_info(astreamer *streamer);
+
+static const astreamer_ops astreamer_verify_ops = {
+	.content = astreamer_verify_content,
+	.finalize = astreamer_verify_finalize,
+	.free = astreamer_verify_free
+};
+
+/*
+ * Create a astreamer that can verifies content of a TAR file.
+ */
+astreamer *
+astreamer_verify_content_new(astreamer *next, verifier_context *context,
+							 char *archive_name, Oid tblspc_oid)
+{
+	astreamer_verify *streamer;
+
+	streamer = palloc0(sizeof(astreamer_verify));
+	*((const astreamer_ops **) &streamer->base.bbs_ops) =
+		&astreamer_verify_ops;
+
+	streamer->base.bbs_next = next;
+	streamer->context = context;
+	streamer->archive_name = archive_name;
+	streamer->tblspc_oid = tblspc_oid;
+	initStringInfo(&streamer->base.bbs_buffer);
+
+	if (!context->skip_checksums)
+		streamer->checksum_ctx = pg_malloc(sizeof(pg_checksum_context));
+
+	return &streamer->base;
+}
+
+/*
+ * The main entry point of the archive streamer for verifying tar members.
+ */
+static void
+astreamer_verify_content(astreamer *streamer, astreamer_member *member,
+						 const char *data, int len,
+						 astreamer_archive_context context)
+{
+	astreamer_verify *mystreamer = (astreamer_verify *) streamer;
+
+	Assert(context != ASTREAMER_UNKNOWN);
+
+	switch (context)
+	{
+		case ASTREAMER_MEMBER_HEADER:
+
+			/*
+			 * Perform the initial check and setup verification steps.
+			 */
+			member_verify_header(streamer, member);
+			break;
+
+		case ASTREAMER_MEMBER_CONTENTS:
+
+			/*
+			 * Since we are receiving the member content in chunks, it must be
+			 * processed according to the flags set by the member header
+			 * processing routine. This includes performing incremental
+			 * checksum computations and copying control data to the local
+			 * buffer.
+			 */
+			if (mystreamer->verify_checksum)
+				member_compute_checksum(streamer, member, data, len);
+
+			if (mystreamer->verify_control_data)
+				member_copy_control_data(streamer, member, data, len);
+			break;
+
+		case ASTREAMER_MEMBER_TRAILER:
+
+			/*
+			 * We have reached the end of the member file. By this point, we
+			 * should have successfully computed the checksum of the received
+			 * content and copied the entire pg_control file data into our
+			 * local buffer. We can now proceed with the final verification.
+			 */
+			if (mystreamer->verify_checksum)
+				member_verify_checksum(streamer);
+
+			if (mystreamer->verify_control_data)
+				member_verify_control_data(streamer);
+
+			/*
+			 * Reset the temporary information stored for the verification.
+			 */
+			member_reset_info(streamer);
+			break;
+
+		case ASTREAMER_ARCHIVE_TRAILER:
+			break;
+
+		default:
+			/* Shouldn't happen. */
+			pg_fatal("unexpected state while parsing tar file");
+	}
+}
+
+/*
+ * End-of-stream processing for a astreamer_verify stream.
+ */
+static void
+astreamer_verify_finalize(astreamer *streamer)
+{
+	Assert(streamer->bbs_next == NULL);
+}
+
+/*
+ * Free memory associated with a astreamer_verify stream.
+ */
+static void
+astreamer_verify_free(astreamer *streamer)
+{
+	astreamer_verify *mystreamer = (astreamer_verify *) streamer;
+
+	if (mystreamer->checksum_ctx)
+		pfree(mystreamer->checksum_ctx);
+
+	pfree(streamer->bbs_buffer.data);
+	pfree(streamer);
+}
+
+/*
+ * Verifies whether the tar member entry exists in the backup manifest.
+ *
+ * If the archive being processed is a tablespace, it prepares the necessary
+ * file path first. If a valid entry is found in the backup manifest, it then
+ * determines whether checksum and control data verification should be
+ * performed during file content processing.
+ */
+static void
+member_verify_header(astreamer *streamer, astreamer_member *member)
+{
+	astreamer_verify *mystreamer = (astreamer_verify *) streamer;
+	manifest_file *m;
+	char		pathname[MAXPGPATH];
+
+	/* We are only interested in normal files. */
+	if (member->is_directory || member->is_link)
+		return;
+
+	/*
+	 * The backup manifest stores a relative path to the base directory for
+	 * files belonging to a tablespace, while the tablespace backup tar
+	 * archive does not include this path. Ensure the required path is
+	 * prepared; otherwise, the manifest entry verification will fail.
+	 */
+	if (OidIsValid(mystreamer->tblspc_oid))
+		snprintf(pathname, MAXPGPATH, "%s/%u/%s",
+				 "pg_tblspc", mystreamer->tblspc_oid, member->pathname);
+	else
+		memcpy(pathname, member->pathname, MAXPGPATH);
+
+
+	/* Ignore any files that are listed in the ignore list. */
+	if (should_ignore_relpath(mystreamer->context, pathname))
+		return;
+
+	/* Check the manifest entry */
+	m = verify_manifest_entry(mystreamer->context, pathname,
+							  member->size);
+	mystreamer->mfile = m;
+
+	/* Prepare for checksum and control data verification. */
+	mystreamer->verify_checksum =
+		(!mystreamer->context->skip_checksums && should_verify_checksum(m));
+	mystreamer->verify_control_data =
+		should_verify_control_data(mystreamer->context->manifest, m);
+
+	/* Initialize the context required for checksum verification. */
+	if (mystreamer->verify_checksum &&
+		pg_checksum_init(mystreamer->checksum_ctx, m->checksum_type) < 0)
+	{
+		report_backup_error(mystreamer->context,
+							"%s: could not initialize checksum of file \"%s\"",
+							mystreamer->archive_name, m->pathname);
+
+		/*
+		 * Checksum verification cannot be performed without proper context
+		 * initialization.
+		 */
+		mystreamer->verify_checksum = false;
+	}
+}
+
+/*
+ * Computes the checksum incrementally for the received file content.
+ *
+ * Should have a correctly initialized checksum_ctx, which will be used for
+ * incremental checksum computation.
+ */
+static void
+member_compute_checksum(astreamer *streamer, astreamer_member *member,
+						const char *data, int len)
+{
+	astreamer_verify *mystreamer = (astreamer_verify *) streamer;
+	pg_checksum_context *checksum_ctx = mystreamer->checksum_ctx;
+	manifest_file *m = mystreamer->mfile;
+
+	Assert(mystreamer->verify_checksum);
+
+	/*
+	 * Should have been applied to the correct file. Note that strcmp() cannot
+	 * be used because the member pathname (if it belongs to a tablespace) is
+	 * not relative to the base directory, unlike the backup manifest. For
+	 * more details, see member_verify_header().
+	 */
+	Assert(should_verify_checksum(m));
+	Assert(m->checksum_type == checksum_ctx->type);
+	Assert(strstr(m->pathname, member->pathname));
+
+	/*
+	 * Update the total count of computed checksum bytes for cross-checking
+	 * with the file size in the final verification stage.
+	 */
+	mystreamer->received_bytes += len;
+
+	if (pg_checksum_update(checksum_ctx, (uint8 *) data, len) < 0)
+	{
+		report_backup_error(mystreamer->context,
+							"could not update checksum of file \"%s\"",
+							m->pathname);
+		mystreamer->verify_checksum = false;
+	}
+}
+
+/*
+ * Perform the final computation and checksum verification after the entire
+ * file content has been processed.
+ */
+static void
+member_verify_checksum(astreamer *streamer)
+{
+	astreamer_verify *mystreamer = (astreamer_verify *) streamer;
+
+	Assert(mystreamer->verify_checksum);
+
+	verify_checksum(mystreamer->context, mystreamer->mfile,
+					mystreamer->checksum_ctx, mystreamer->received_bytes);
+}
+
+/*
+ * Stores the pg_control file contents into a local buffer; we need the entire
+ * control file data for verification.
+ */
+static void
+member_copy_control_data(astreamer *streamer, astreamer_member *member,
+						 const char *data, int len)
+{
+	/* Should be here only for control file */
+	Assert(strcmp(member->pathname, "global/pg_control") == 0);
+	Assert(((astreamer_verify *) streamer)->verify_control_data);
+
+	/* Copy enough control file data needed for verification. */
+	astreamer_buffer_until(streamer, &data, &len, sizeof(ControlFileData));
+}
+
+/*
+ * Performs the CRC calculation of pg_control data and then calls the routines
+ * that execute the final verification of the control file information.
+ */
+static void
+member_verify_control_data(astreamer *streamer)
+{
+	astreamer_verify *mystreamer = (astreamer_verify *) streamer;
+	manifest_data *manifest = mystreamer->context->manifest;
+	ControlFileData *control_file;
+	pg_crc32c	crc;
+	bool		crc_ok;
+
+	/* Should be here only for control file */
+	Assert(strcmp(mystreamer->mfile->pathname, "global/pg_control") == 0);
+	Assert(mystreamer->verify_control_data);
+
+	/* Should have enough control file data needed for verification. */
+	if (streamer->bbs_buffer.len != sizeof(ControlFileData))
+		report_fatal_error("%s: unexpected control file size: %d, should be %zu",
+						   mystreamer->archive_name, streamer->bbs_buffer.len,
+						   sizeof(ControlFileData));
+
+	control_file = (ControlFileData *) streamer->bbs_buffer.data;
+
+	/* Check the CRC. */
+	INIT_CRC32C(crc);
+	COMP_CRC32C(crc, (char *) (control_file), offsetof(ControlFileData, crc));
+	FIN_CRC32C(crc);
+
+	crc_ok = EQ_CRC32C(crc, control_file->crc);
+
+	/* Do the final control data verification. */
+	verify_control_data(control_file, mystreamer->mfile->pathname, crc_ok,
+						manifest->system_identifier);
+}
+
+/*
+ * Reset flags and free memory allocations for member file verification.
+ */
+static void
+member_reset_info(astreamer *streamer)
+{
+	astreamer_verify *mystreamer = (astreamer_verify *) streamer;
+
+	mystreamer->mfile = NULL;
+	mystreamer->received_bytes = 0;
+	mystreamer->verify_checksum = false;
+	mystreamer->verify_control_data = false;
+}
diff --git a/src/bin/pg_verifybackup/meson.build b/src/bin/pg_verifybackup/meson.build
index 7c7d31a0350..0e09d1379d1 100644
--- a/src/bin/pg_verifybackup/meson.build
+++ b/src/bin/pg_verifybackup/meson.build
@@ -1,6 +1,7 @@
 # Copyright (c) 2022-2024, PostgreSQL Global Development Group
 
 pg_verifybackup_sources = files(
+  'astreamer_verify.c',
   'pg_verifybackup.c'
 )
 
diff --git a/src/bin/pg_verifybackup/pg_verifybackup.c b/src/bin/pg_verifybackup/pg_verifybackup.c
index c1542983b93..e63a0ed0798 100644
--- a/src/bin/pg_verifybackup/pg_verifybackup.c
+++ b/src/bin/pg_verifybackup/pg_verifybackup.c
@@ -22,6 +22,7 @@
 #include "common/parse_manifest.h"
 #include "fe_utils/simple_list.h"
 #include "getopt_long.h"
+#include "limits.h"
 #include "pg_verifybackup.h"
 #include "pgtime.h"
 
@@ -44,6 +45,16 @@
  */
 #define READ_CHUNK_SIZE				(128 * 1024)
 
+/*
+ * Tar file information needed for content verification.
+ */
+typedef struct tar_file
+{
+	char	   *relpath;
+	Oid			tblspc_oid;
+	pg_compress_algorithm compress_algorithm;
+} tar_file;
+
 static manifest_data *parse_manifest_file(char *manifest_path);
 static void verifybackup_version_cb(JsonManifestParseContext *context,
 									int manifest_version);
@@ -63,10 +74,16 @@ static void report_manifest_error(JsonManifestParseContext *context,
 			pg_attribute_printf(2, 3) pg_attribute_noreturn();
 
 static char find_backup_format(verifier_context *context);
+static void verify_plain_backup(verifier_context *context);
+static void verify_tar_backup(verifier_context *context);
 static void verify_backup_directory(verifier_context *context,
 									char *relpath, char *fullpath);
-static void verify_plain_backup_file(verifier_context *context,
-									 char *relpath, char *fullpath);
+static void verify_plain_backup_file(verifier_context *context, char *relpath,
+									 char *fullpath);
+static void precheck_tar_backup_file(verifier_context *context, char *relpath,
+									 char *fullpath, SimplePtrList *tarfiles);
+static void verify_tar_file(verifier_context *context, char *relpath,
+							char *fullpath, astreamer *streamer);
 static void report_extra_backup_files(verifier_context *context);
 static void verify_backup_checksums(verifier_context *context);
 static void verify_file_checksum(verifier_context *context,
@@ -75,6 +92,10 @@ static void verify_file_checksum(verifier_context *context,
 static void parse_required_wal(verifier_context *context,
 							   char *pg_waldump_path,
 							   char *wal_directory);
+static astreamer *create_archive_verifier(verifier_context *context,
+										  char *archive_name,
+										  Oid tblspc_oid,
+										  pg_compress_algorithm compress_algo);
 
 static void progress_report(bool finished);
 static void usage(void);
@@ -294,7 +315,10 @@ main(int argc, char **argv)
 	 * match. We also set the "matched" flag on every manifest entry that
 	 * corresponds to a file on disk.
 	 */
-	verify_backup_directory(&context, NULL, context.backup_directory);
+	if (context.format == 'p')
+		verify_plain_backup(&context);
+	else
+		verify_tar_backup(&context);
 
 	/*
 	 * The "matched" flag should now be set on every entry in the hash table.
@@ -546,6 +570,16 @@ verifybackup_per_wal_range_cb(JsonManifestParseContext *context,
 	manifest->last_wal_range = range;
 }
 
+/*
+ * Verify plain backup.
+ */
+static void
+verify_plain_backup(verifier_context *context)
+{
+	Assert(context->format == 'p');
+	verify_backup_directory(context, NULL, context->backup_directory);
+}
+
 /*
  * Verify one directory.
  *
@@ -682,6 +716,257 @@ verify_plain_backup_file(verifier_context *context, char *relpath,
 		total_size += m->size;
 }
 
+/*
+ * Verify tar backup.
+ *
+ * Unlike plan backup verification, tar backup verification carried out in two
+ * passes; in the first pass this would simply sanity check on expected tar file
+ * to be present in the backup directory and it's compression type and collect
+ * these information is list. In the second pass, the tar archives are read,
+ * decompressed, and the required verification is carried out.
+ */
+static void
+verify_tar_backup(verifier_context *context)
+{
+	DIR		   *dir;
+	struct dirent *dirent;
+	SimplePtrList tarfiles = {NULL, NULL};
+	SimplePtrListCell *cell;
+	char	   *fullpath;
+
+	Assert(context->format == 't');
+
+	progress_report(false);
+
+	/*
+	 * If the backup directory cannot be found, treat this as a fatal error.
+	 */
+	fullpath = context->backup_directory;
+	dir = opendir(fullpath);
+	if (dir == NULL)
+		report_fatal_error("could not open directory \"%s\": %m", fullpath);
+
+	while (errno = 0, (dirent = readdir(dir)) != NULL)
+	{
+		char	   *filename = dirent->d_name;
+		char	   *newfullpath = psprintf("%s/%s", fullpath, filename);
+
+		/* Skip "." and ".." */
+		if (filename[0] == '.' && (filename[1] == '\0'
+								   || strcmp(filename, "..") == 0))
+			continue;
+
+		/* First pass: Collect valid tar files from the backup. */
+		if (!should_ignore_relpath(context, filename))
+			precheck_tar_backup_file(context, filename, newfullpath,
+									 &tarfiles);
+
+		pfree(newfullpath);
+	}
+
+	if (closedir(dir))
+	{
+		report_backup_error(context,
+							"could not close directory \"%s\": %m", fullpath);
+		return;
+	}
+
+	/* Second pass: Perform the final verification of the tar contents. */
+	for (cell = tarfiles.head; cell != NULL; cell = cell->next)
+	{
+		tar_file   *tar = (tar_file *) cell->ptr;
+		astreamer  *streamer;
+
+		/*
+		 * Prepares the archive streamer stack according to the tar
+		 * compression format.
+		 */
+		streamer = create_archive_verifier(context,
+										   tar->relpath,
+										   tar->tblspc_oid,
+										   tar->compress_algorithm);
+
+		/* Compute the full pathname to the target file. */
+		fullpath = psprintf("%s/%s", context->backup_directory,
+							tar->relpath);
+
+		/* Invoke the streamer for reading, decompressing, and verifying. */
+		verify_tar_file(context, tar->relpath, fullpath, streamer);
+
+		/* Cleanup. */
+		pfree(tar->relpath);
+		pfree(tar);
+		pfree(fullpath);
+
+		astreamer_finalize(streamer);
+		astreamer_free(streamer);
+	}
+	simple_ptr_list_destroy(&tarfiles);
+
+	progress_report(true);
+}
+
+/*
+ * Preparatory steps for verifying files in tar format backups.
+ *
+ * Carries out basic validation of the tar format backup file, detects the
+ * compression type, and appends that information to the tarfiles list. An
+ * error will be reported if the tar file is inaccessible, or if the file type,
+ * name, or compression type is not as expected.
+ *
+ * The arguments to this function are mostly the same as the
+ * verify_plain_backup_file. The additional argument outputs a list of valid
+ * tar files.
+ */
+static void
+precheck_tar_backup_file(verifier_context *context, char *relpath,
+						 char *fullpath, SimplePtrList *tarfiles)
+{
+	struct stat sb;
+	Oid			tblspc_oid = InvalidOid;
+	pg_compress_algorithm compress_algorithm;
+	tar_file   *tar;
+	char	   *suffix = NULL;
+
+	/* Should be tar format backup */
+	Assert(context->format == 't');
+
+	/* Get file information */
+	if (stat(fullpath, &sb) != 0)
+	{
+		report_backup_error(context,
+							"could not stat file or directory \"%s\": %m",
+							relpath);
+		return;
+	}
+
+	/* In a tar format backup, we expect only plain files. */
+	if (!S_ISREG(sb.st_mode))
+	{
+		report_backup_error(context,
+							"\"%s\" is not a plain file",
+							relpath);
+		return;
+	}
+
+	/*
+	 * We expect tar files for backing up the main directory, tablespace, and
+	 * pg_wal directory.
+	 *
+	 * pg_basebackup writes the main data directory to an archive file named
+	 * base.tar, the pg_wal directory to pg_wal.tar, and the tablespace
+	 * directory to <tablespaceoid>.tar, each followed by a compression type
+	 * extension such as .gz, .lz4, or .zst.
+	 */
+	if (strncmp("base", relpath, 4) == 0)
+		suffix = relpath + 4;
+	else if (strncmp("pg_wal", relpath, 6) == 0)
+		suffix = relpath + 6;
+	else
+	{
+		/* Expected a <tablespaceoid>.tar file here. */
+		uint64		num = strtoul(relpath, &suffix, 10);
+
+		/*
+		 * Report an error if we didn't consume at least one character, if the
+		 * result is 0, or if the value is too large to be a valid OID.
+		 */
+		if (suffix == NULL || num <= 0 || num > OID_MAX)
+			report_backup_error(context,
+								"file \"%s\" is not expected in a tar format backup",
+								relpath);
+		tblspc_oid = (Oid) num;
+	}
+
+	/* Now, check the compression type of the tar */
+	if (strcmp(suffix, ".tar") == 0)
+		compress_algorithm = PG_COMPRESSION_NONE;
+	else if (strcmp(suffix, ".tgz") == 0)
+		compress_algorithm = PG_COMPRESSION_GZIP;
+	else if (strcmp(suffix, ".tar.gz") == 0)
+		compress_algorithm = PG_COMPRESSION_GZIP;
+	else if (strcmp(suffix, ".tar.lz4") == 0)
+		compress_algorithm = PG_COMPRESSION_LZ4;
+	else if (strcmp(suffix, ".tar.zst") == 0)
+		compress_algorithm = PG_COMPRESSION_ZSTD;
+	else
+	{
+		report_backup_error(context,
+							"file \"%s\" is not expected in a tar format backup",
+							relpath);
+		return;
+	}
+
+	/*
+	 * Ignore WALs, as reading and verification will be handled through
+	 * pg_waldump.
+	 */
+	if (strncmp("pg_wal", relpath, 6) == 0)
+		return;
+
+	/*
+	 * Append the information to the list for complete verification at a later
+	 * stage.
+	 */
+	tar = pg_malloc(sizeof(tar_file));
+	tar->relpath = pstrdup(relpath);
+	tar->tblspc_oid = tblspc_oid;
+	tar->compress_algorithm = compress_algorithm;
+
+	simple_ptr_list_append(tarfiles, tar);
+
+	/* Update statistics for progress report, if necessary */
+	if (show_progress)
+		total_size += sb.st_size;
+}
+
+/*
+ * Verification of a single tar file content.
+ *
+ * It reads a given tar archive in predefined chunks and passes it to the
+ * streamer, which initiates routines for decompression (if necessary) and then
+ * verifies each member within the tar file.
+ */
+static void
+verify_tar_file(verifier_context *context, char *relpath, char *fullpath,
+				astreamer *streamer)
+{
+	int			fd;
+	int			rc;
+	char	   *buffer;
+
+	pg_log_debug("reading \"%s\"", fullpath);
+
+	/* Open the target file. */
+	if ((fd = open(fullpath, O_RDONLY | PG_BINARY, 0)) < 0)
+	{
+		report_backup_error(context, "could not open file \"%s\": %m",
+							relpath);
+		return;
+	}
+
+	buffer = pg_malloc(READ_CHUNK_SIZE * sizeof(uint8));
+
+	/* Perform the reads */
+	while ((rc = read(fd, buffer, READ_CHUNK_SIZE)) > 0)
+	{
+		astreamer_content(streamer, NULL, buffer, rc, ASTREAMER_UNKNOWN);
+
+		/* Report progress */
+		done_size += rc;
+		progress_report(false);
+	}
+
+	if (rc < 0)
+		report_backup_error(context, "could not read file \"%s\": %m",
+							relpath);
+
+	/* Close the file. */
+	if (close(fd) != 0)
+		report_backup_error(context, "could not close file \"%s\": %m",
+							relpath);
+}
+
 /*
  * Verify file and its size entry in the manifest.
  */
@@ -1044,6 +1329,41 @@ find_backup_format(verifier_context *context)
 	return result;
 }
 
+/*
+ * Identifies the necessary steps for verifying the contents of the
+ * provided tar file.
+ */
+static astreamer *
+create_archive_verifier(verifier_context *context, char *archive_name,
+						Oid tblspc_oid, pg_compress_algorithm compress_algo)
+{
+	astreamer  *streamer = NULL;
+
+	/* Should be here only for tar backup */
+	Assert(context->format == 't');
+
+	/*
+	 * To verify the contents of the tar, the initial step is to parse its
+	 * content.
+	 */
+	streamer = astreamer_verify_content_new(streamer, context, archive_name,
+											tblspc_oid);
+	streamer = astreamer_tar_parser_new(streamer);
+
+	/*
+	 * If the tar is compressed, we must perform the appropriate decompression
+	 * operation before proceeding with the verification of its contents.
+	 */
+	if (compress_algo == PG_COMPRESSION_GZIP)
+		streamer = astreamer_gzip_decompressor_new(streamer);
+	else if (compress_algo == PG_COMPRESSION_LZ4)
+		streamer = astreamer_lz4_decompressor_new(streamer);
+	else if (compress_algo == PG_COMPRESSION_ZSTD)
+		streamer = astreamer_zstd_decompressor_new(streamer);
+
+	return streamer;
+}
+
 /*
  * Print a progress report based on the global variables.
  *
diff --git a/src/bin/pg_verifybackup/pg_verifybackup.h b/src/bin/pg_verifybackup/pg_verifybackup.h
index 80031ad4dbc..be7438af346 100644
--- a/src/bin/pg_verifybackup/pg_verifybackup.h
+++ b/src/bin/pg_verifybackup/pg_verifybackup.h
@@ -18,6 +18,7 @@
 #include "common/hashfn_unstable.h"
 #include "common/logging.h"
 #include "common/parse_manifest.h"
+#include "fe_utils/astreamer.h"
 #include "fe_utils/simple_list.h"
 
 /*
@@ -123,4 +124,9 @@ extern void report_fatal_error(const char *pg_restrict fmt,...)
 extern bool should_ignore_relpath(verifier_context *context,
 								  const char *relpath);
 
+extern astreamer *astreamer_verify_content_new(astreamer *next,
+											   verifier_context *context,
+											   char *archive_name,
+											   Oid tblspc_oid);
+
 #endif							/* PG_VERIFYBACKUP_H */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index e9ebddde24d..2b155586f8c 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -3330,6 +3330,7 @@ astreamer_plain_writer
 astreamer_recovery_injector
 astreamer_tar_archiver
 astreamer_tar_parser
+astreamer_verify
 astreamer_zstd_frame
 bgworker_main_type
 bh_node_type
@@ -3951,6 +3952,7 @@ substitute_phv_relids_context
 subxids_array_status
 symbol
 tablespaceinfo
+tar_file
 td_entry
 teSection
 temp_tablespaces_extra
-- 
2.18.0