# Test that pg_rewind does not leave behind WAL segments whose timeline
# exceeds the source's.  When the source is on a lower timeline than the
# target, keepwal used to record entries with the target's (higher) TLI,
# so stray high-TLI segments survived pg_rewind, e.g.:
#
#	000000020000000000000002 (rewound)
#	000000030000000000000002 (rewound)
#	000000040000000000000002 (rewound)
#	000000080000000000000002 (kept by keepwal -- bug)
#
# The fix clamps the keepwal TLI to the source's latest, so such leftover
# segments are removed.

use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Utils;
use Test::More;

use File::Copy;
use FindBin;
use lib $FindBin::RealBin;

use RewindTest;

# Initialize primary, data checksums are mandatory.
my $node_primary = PostgreSQL::Test::Cluster->new('primary');

# Set up pg_hba.conf and pg_ident.conf for the role running pg_rewind.  This
# role is used for all the tests and has minimal permissions enough to rewind
# from an online source.
$node_primary->init(
	allows_streaming => 1,
	auth_extra => [ '--create-role' => 'rewind_user' ]);

# Set wal_keep_size to prevent WAL segment recycling after enforced
# checkpoints in the tests.
$node_primary->append_conf(
	'postgresql.conf', qq(
wal_keep_size = 320MB
allow_in_place_tablespaces = on
));

$node_primary->start;

# Create custom role which is used to run pg_rewind, and adjust its
# permissions to the minimum necessary.
$node_primary->safe_psql(
	'postgres', "
	CREATE ROLE rewind_user LOGIN;
	GRANT EXECUTE ON function pg_catalog.pg_ls_dir(text, boolean, boolean)
		TO rewind_user;
	GRANT EXECUTE ON function pg_catalog.pg_stat_file(text, boolean)
		TO rewind_user;
	GRANT EXECUTE ON function pg_catalog.pg_read_binary_file(text)
		TO rewind_user;
	GRANT EXECUTE ON function pg_catalog.pg_read_binary_file(text, bigint, bigint, boolean)
		TO rewind_user;");

# standby
my $node_standby = PostgreSQL::Test::Cluster->new('standby');
$node_primary->backup('my_backup');
$node_standby->init_from_backup($node_primary, 'my_backup');
my $connstr_primary = $node_primary->connstr();

$node_standby->append_conf(
	"postgresql.conf", qq(
primary_conninfo='$connstr_primary'
));

$node_standby->set_standby_mode();

# start standby
$node_standby->start();

# Force a few segment switches and a checkpoint so that the primary accumulates
# several timeline switches below (each promotion bumps the timeline).
$node_primary->safe_psql('postgres', "select pg_switch_wal()");
$node_primary->safe_psql('postgres', "select pg_switch_wal()");
$node_primary->safe_psql('postgres', "select pg_switch_wal()");
$node_primary->safe_psql('postgres', "select pg_switch_wal()");
$node_primary->safe_psql('postgres', "CHECKPOINT");

my $tli = $node_primary->safe_psql('postgres',
	"SELECT timeline_id FROM pg_control_checkpoint()");

$node_primary->wait_for_catchup($node_standby->name());

$node_standby->stop('immediate');

# Bump the primary's timeline several times past the divergence point. After
# this loop the primary (the future rewind target) sits on a higher timeline
# than the standby (the future rewind source), which is the scenario the bug
# manifests in.
for my $i (1 .. $tli + 2)
{
	$node_primary->stop();
	$node_primary->set_standby_mode();
	$node_primary->start();
	$node_primary->promote();
}

# Stop the (higher-timeline) primary and promote the standby so it becomes the
# rewind source.
$node_primary->stop('immediate');

$node_standby->start();
$node_standby->promote();

my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
my $primary_pgdata = $node_primary->data_dir;
my $standby_connstr = $node_standby->connstr('postgres');
$standby_connstr = "$standby_connstr user=rewind_user";
copy(
	"$primary_pgdata/postgresql.conf",
	"$tmp_folder/primary-postgresql.conf.tmp");

command_ok(
	[
		'pg_rewind',
		'--debug',
		'--source-server' => $standby_connstr,
		'--target-pgdata' => $primary_pgdata,
		'--write-recovery-conf',
		'--config-file' => "$tmp_folder/primary-postgresql.conf.tmp",
	],
	'pg_rewind');

like(slurp_file("$primary_pgdata/postgresql.auto.conf"),
	qr/dbname=postgres/m, 'recovery conf file sets dbname');

# Check that standby.signal is here as recovery configuration was requested.
ok( -e "$primary_pgdata/standby.signal",
	'standby.signal created after pg_rewind');

# Now, when pg_rewind apparently succeeded with minimal permissions, add
# REPLICATION privilege. So we could test that the new standby is able to
# connect to the new primary with the generated config.
$node_standby->safe_psql('postgres',
	"ALTER ROLE rewind_user WITH REPLICATION;");

move(
	"$tmp_folder/primary-postgresql.conf.tmp",
	"$primary_pgdata/postgresql.conf");

chmod(
	$node_primary->group_access() ? 0640 : 0600,
	"$primary_pgdata/postgresql.conf")
	or BAIL_OUT(
	"unable to set permissions for $primary_pgdata/postgresql.conf");

$node_primary->set_standby_mode();
$node_primary->start();

# Get the current WAL segment on the source node and extract the TLI from its
# name. No segment remaining in the rewound target's pg_wal may carry a higher
# timeline.
my $source_tli_hex = $node_standby->safe_psql('postgres',
	"SELECT substring(pg_walfile_name(pg_current_wal_lsn()) from 1 for 8);");

my $node_primary_wal_dir = $node_primary->data_dir . "/pg_wal";

opendir(my $dh, $node_primary_wal_dir)
	or die "Could not open directory '$node_primary_wal_dir': $!";

# Check that there are no segments with a higher TLI on the rewound node.
while (my $filename = readdir($dh))
{
	if ($filename =~ /^([0-9A-F]{8})[0-9A-F]{16}$/)
	{
		my $timeline_hex = $1;
		ok($timeline_hex le $source_tli_hex,
			"WAL segment $filename with TLI " . hex($timeline_hex)
			. " does not exceed the source TLI " . hex($source_tli_hex));
	}
}

closedir($dh);

done_testing();