[PBCKP-150] Reading buffer is flushed each time we verify the checksum.

The race condition is covered with a unit-test, the buffer is flushed
now so each of 300 reads requests the data from the disc.

Author: Daniel Shelepanov

src/data.c

@@ -301,6 +301,8 @@ prepare_page(pgFile *file, XLogRecPtr prev_backup_start_lsn,
 	{
 		/* read the block */
 		int read_len = fio_pread(in, page, blknum * BLCKSZ);
+		/* avoid re-reading once buffered data, see PBCKP-150 */
+		fflush(in);
 
 		/* The block could have been truncated. It is fine. */
 		if (read_len == 0)

tests/__init__.py

@@ -7,7 +7,7 @@
     compression, page, ptrack, archive, exclude, cfs_backup, cfs_restore, \
     cfs_validate_backup, auth_test, time_stamp, logging, \
     locking, remote, external, config, checkdb, set_backup, incr_restore, \
-    catchup, CVE_2018_1058
+    catchup, CVE_2018_1058, pbckp150
 
 
 def load_tests(loader, tests, pattern):
@@ -21,6 +21,12 @@ def load_tests(loader, tests, pattern):
         if os.environ['PG_PROBACKUP_PTRACK'] == 'ON':
             suite.addTests(loader.loadTestsFromModule(ptrack))
 
+    # PG_PROBACKUP_LONG section for tests that are long
+    # by design e.g. they contain loops, sleeps and so on
+    if 'PG_PROBACKUP_LONG' in os.environ:
+        if os.environ['PG_PROBACKUP_LONG'] == 'ON':
+            suite.addTests(loader.loadTestsFromModule(pbckp150))
+
 #    suite.addTests(loader.loadTestsFromModule(auth_test))
     suite.addTests(loader.loadTestsFromModule(archive))
     suite.addTests(loader.loadTestsFromModule(backup))

tests/pbckp150.py

@@ -0,0 +1,76 @@
+import os
+import unittest
+from .helpers.ptrack_helpers import ProbackupTest
+import subprocess
+from time import sleep
+
+module_name = 'pbckp150'
+
+class TestPageCorruptionDueToRace(ProbackupTest, unittest.TestCase):
+    def test_pbckp150(self):
+        """
+        https://jira.postgrespro.ru/browse/PBCKP-150
+        create a node filled with pgbench
+        create FULL backup followed by PTRACK backup
+        run pgbench, vacuum VERBOSE FULL and ptrack backups in parallel
+        """
+        # init node
+        fname = self.id().split('.')[3]
+        node = self.make_simple_node(
+            base_dir=os.path.join(module_name, fname, 'node'),
+            set_replication=True,
+            initdb_params=['--data-checksums'])
+        node.append_conf('postgresql.conf',
+        """
+            max_connections = 100
+            wal_keep_size = 16000
+            ptrack.map_size = 1
+            shared_preload_libraries='ptrack'
+            log_statement = 'none'
+            fsync = off
+            log_checkpoints = on
+            autovacuum = off
+        """)
+
+        # init probackup and add an instance
+        backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
+        self.init_pb(backup_dir)
+        self.add_instance(backup_dir, 'node', node)
+
+        # run the node and init ptrack
+        node.slow_start()
+        node.safe_psql("postgres", "CREATE EXTENSION ptrack")
+        # populate it with pgbench
+        node.pgbench_init(scale=5)
+
+        # FULL backup followed by PTRACK backup
+        self.backup_node(backup_dir, 'node', node, options=['--stream'])
+        self.backup_node(backup_dir, 'node', node, backup_type='ptrack', options=['--stream'])
+
+        # run ordinary pgbench scenario to imitate some activity and another pgbench for vacuuming in parallel
+        nBenchDuration = 30
+        pgbench = node.pgbench(options=['-c', '20', '-j', '8', '-T', str(nBenchDuration)])
+        with open('/tmp/pbckp150vacuum.sql', 'w') as f:
+            f.write('VACUUM (FULL) pgbench_accounts, pgbench_tellers, pgbench_history; SELECT pg_sleep(1);\n')
+        pgbenchval = node.pgbench(options=['-c', '1', '-f', '/tmp/pbckp150vacuum.sql', '-T', str(nBenchDuration)])
+
+        # several PTRACK backups
+        for i in range(nBenchDuration):
+            print("[{}] backing up PTRACK diff...".format(i+1))
+            self.backup_node(backup_dir, 'node', node, backup_type='ptrack', options=['--stream', '--log-level-console', 'VERBOSE'])
+            sleep(0.1)
+            # if the activity pgbench has finished, stop backing up
+            if pgbench.poll() is not None:
+                break
+
+        pgbench.kill()
+        pgbenchval.kill()
+        pgbench.wait()
+        pgbenchval.wait()
+
+        backups = self.show_pb(backup_dir, 'node')
+        for b in backups:
+            self.assertEqual("OK", b['status'])
+
+        # Clean after yourself
+        self.del_test_dir(module_name, fname)