#!/bin/bash
#
# pg_lion: automated crash-recovery and hot-standby tests.
#
# USAGE
#
#   test/recovery/run.sh --prefix <pg install prefix> [--iters N] [--keep]
#   make recovery-check PG_CONFIG=<prefix>/bin/pg_config RECOVERY_PREFIX=<prefix>
#
#   --prefix   PostgreSQL installation to use (bin/, lib/, share/).  REQUIRED,
#              here or as RECOVERY_PREFIX in the environment: there is no
#              default, because the extension is rebuilt from a clean copy of
#              this tree and INSTALLED into that prefix (unless
#              RECOVERY_SKIP_INSTALL=1), and a guessed prefix is someone
#              else's server.  "make install" is never run in the tree, and
#              nothing in it is written but test/recovery/log, where the run's
#              log goes (and, on failure, copies of the server logs).
#   --iters    crash/recover iterations in phase 1 (default 8).
#   --keep     leave the clusters and their logs in place on exit.
#   --mode     generic (the default) or rmgr: rmgr preloads the library so
#              that the custom WAL resource manager of DESIGN.md §25 is
#              registered and every index built here is written through it.
#              Phase 2 then expects the standby to TRUST the visibility map,
#              which is the property §25 buys.
#   --phases   which phases to run, e.g. "3" or "2 3" (default: all of
#              "1 1b 1c 1d 1e 1f 1g 1h 2 3 4 5").  For development, and for a
#              server without injection points, which can run only "1 1f 1h
#              2 4 5" (CI's packaged-server jobs do; phase 5 then leaves out
#              its one case that needs them); `make recovery-check` always
#              runs everything.
#   --conf     an extra postgresql.conf line for the primary (repeatable),
#              appended last so it wins.  This is how
#              wal_consistency_checking = 'pg_lion' is turned on:
#                  --mode rmgr --conf "wal_consistency_checking = 'pg_lion'"
#              which makes the standby (and every crash recovery in phase 1)
#              compare the page replay produced against the page the primary
#              had, for every page of every lion record.
#
# ENVIRONMENT
#
#   RECOVERY_PREFIX   the same as --prefix (the option wins).
#   RECOVERY_RUN_AS   an unprivileged OS user to run initdb, pg_ctl and
#                     pg_basebackup as (through runuser).  initdb and the
#                     server refuse to run as root, so a root shell - a
#                     container, a CI image - needs this; the run refuses to
#                     start as root without it.  That user must be able to read
#                     the prefix, and it gets ownership of the run's private
#                     directory below.  psql, pgbench, pg_waldump and the
#                     extension build still run as the caller.
#   RECOVERY_TMPDIR   where the run's private directory is made (default
#                     $TMPDIR, else /tmp).
#   RECOVERY_SKIP_INSTALL=1
#                     do not build and install the extension: use the pg_lion
#                     already installed in the prefix, which must be there.
#                     For an installation the caller cannot write to (CI's
#                     packaged servers, where the job installs it with sudo
#                     beforehand).  A pg_lion.so in the tree that differs from
#                     the installed one is reported, since the installed one
#                     is what gets tested.
#   INJECTION_POINTS=1
#                     phases 1b-1e, 1g and 3 park a backend on an injection point
#                     and are skipped, with a line in the summary, on a server
#                     without the injection_points extension.  With this set
#                     (as CI's source jobs set it) that is a failure instead:
#                     a server that should have them and does not must not
#                     end in "ALL ... PASSED" having run none of them.
#   PGVECTOR=1        the same for phase 5, which needs pgvector (and
#                     pageinspect) and is otherwise skipped without it; CI
#                     sets it wherever it installed pgvector, as the Makefile
#                     takes it for lionvec's regression files and specs.
#
# Exit status is 0 only if every check passed.  A summary is printed at the
# end; the full log is written to test/recovery/log/run.log, and on failure the
# server logs of both clusters are copied next to it.
#
# WHAT IT DOES
#
#   Phase 1  N rounds of: run a mixed write load (pgbench, 4 clients,
#            synchronous_commit = on) for ~2 s against a table with one column
#            per operator class kind, kill the server without a chance to
#            flush (pg_ctl -m immediate and kill -9, alternating; twice while a
#            VACUUM is in flight), restart, and check that the indexes came
#            back intact: lion_index_verify(idx, true) for each, single-key
#            counts through the index / a forced seqscan / lion_index_count()
#            / the count pushdown all equal, GROUP BY results equal as
#            multisets (EXCEPT ALL both ways), and lion_index_stats().ntids
#            consistent with the heap - exactly, after a VACUUM.  Each round
#            also proves the restart really replayed generic WAL.
#
#   Phase 2  pg_basebackup -R a hot standby, and check that it answers exactly
#            what the primary answers, that lion_index_count_stats() shows
#            it trusting the visibility map for nothing (DESIGN.md section 9,
#            "Hot standby"), and that a REPEATABLE READ standby snapshot whose
#            rows the primary deletes and vacuums is either preserved or
#            cancelled with a recovery conflict - never silently answered with
#            a different number; and that a position tree damaged in the
#            standby's copy alone is reported there.  Then promote it and
#            verify every index again.
#
#   Phase 4  lion_index_verify() in a loop on a hot standby while the primary
#            inserts into the posting sets it walks, with a watchdog that fails
#            the phase once replay stops moving: a verify() that held a
#            directory leaf while it locked posting pages, which replay locks
#            the other way round, would deadlock with replay for good.
#
#   Phase 5  lionvec's locator of vectors in TOAST (DESIGN.md §44.13): crash
#            rounds under a write load of its own, an insert cut between its
#            items and its LOC entry, and a hot standby, each checked by
#            lionvec.sql's lv_rec_check() - the locator against the heap and
#            the TOAST relation, and the scans' rows read by place against
#            the heap's and a sort's.  Needs pgvector; skipped without it.
#
# See test/recovery/README.md for what is deliberately not covered.
#
# The cluster discipline is bench/lib.sh's: a private socket directory and port
# that nothing else uses, refuse to start if either is already in use, check
# SHOW data_directory before the first statement, and remove everything on exit
# through a trap.  These clusters are created here and are never shared with
# the dev cluster or the benchmark clusters.
#
# "Private" is literal: everything the run creates - both data directories,
# their logs, the socket directory, the build copy - lives in ONE directory
# that mktemp -d makes for this run (mode 0700, a name nobody else has), and
# the exit trap removes that directory and nothing else.  There is no fixed
# path in a shared /tmp that a second run, another user or a stale directory
# could already occupy, and nothing the run did not create is ever removed.

set -euo pipefail

HERE=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
PROJECT=$(cd "$HERE/../.." && pwd)

PREFIX=${RECOVERY_PREFIX:-}
RUN_AS=${RECOVERY_RUN_AS:-}
TMPROOT=${RECOVERY_TMPDIR:-${TMPDIR:-/tmp}}
SKIP_INSTALL=${RECOVERY_SKIP_INSTALL:-0}
REQUIRE_INJECTION=${INJECTION_POINTS:-0}
REQUIRE_PGVECTOR=${PGVECTOR:-0}
ITERS=8
KEEP=0
MODE=generic
PHASES="1 1b 1c 1d 1e 1f 1g 1h 2 3 4 5"
EXTRA_CONF=()

# Set by make_private_dirs(), which the main section calls once the arguments
# have been checked; empty until then, which cleanup() relies on.
BASE=""
SOCKDIR=""
SOCKDIR_OWN=""		# a socket directory of its own, when BASE is too long for one
PRIMARY_PORT=54340
STANDBY_PORT=54341
DBNAME=postgres

# The dev cluster's default port (dev.sh).  The socket directory is private, so
# no other cluster can be listening in it; this only stops an edit of the two
# ports above from pointing the run at the endpoint a developer's psql uses.
FORBIDDEN_PORTS="54329"

LOGDIR=$HERE/log
RUNLOG=$LOGDIR/run.log

PRIMARY_DATA=""
STANDBY_DATA=""
PRIMARY_LOG=""
STANDBY_LOG=""
BUILDDIR=""
VACLOOP=""
VACPID=""
RRPID=""
P4PIDS=""
GENERIC_TOTAL=0
SUMMARY=()
PHASE3_FAILS=()

nap() { command sleep "$1"; }
die() { echo "FAIL: $*" >&2; exit 1; }
log() { echo "$*" | tee -a "$RUNLOG"; }
now_ms() { date +%s%3N; }

# no_injection_points <phase> <summary line>: the phase needs the
# injection_points extension, and this server does not have it.  A skip, in
# the log and the summary - or, under INJECTION_POINTS=1, a failure.
no_injection_points() {
	if [ "$REQUIRE_INJECTION" = 1 ]; then
		die "$1 needs the injection_points extension, which this server does not have (INJECTION_POINTS=1 makes that a failure rather than a skip)"
	fi
	log "$1 skipped: this server has no injection_points extension"
	SUMMARY+=("$2")
}

# ---------------------------------------------------------------- arguments

while [ $# -gt 0 ]; do
	case $1 in
		--prefix) PREFIX=${2:-}; shift 2 ;;
		--iters)  ITERS=${2:-}; shift 2 ;;
		--keep)   KEEP=1; shift ;;
		--mode)   MODE=${2:-}; shift 2 ;;
		--conf)   EXTRA_CONF+=("${2:-}"); shift 2 ;;
		--phases) PHASES=${2:-}; shift 2 ;;
		-h|--help) sed -n '/^# USAGE/,/^# Exit status/p' "${BASH_SOURCE[0]}" |
			sed '$d; s/^#\{0,1\} \{0,1\}//'; exit 0 ;;
		*) echo "unknown argument: $1" >&2; exit 2 ;;
	esac
done

case $MODE in
	generic) ;;
	rmgr)
		# DESIGN.md §25: the custom resource manager can only be registered
		# from shared_preload_libraries, and an index built on such a server
		# is written through it.
		EXTRA_CONF+=("shared_preload_libraries = 'pg_lion'")
		;;
	*) die "--mode wants generic or rmgr" ;;
esac

# wal_consistency_checking puts a full-page image in every record it names,
# and one crash round of phase 1 then replays 600-800 MB of WAL, more than
# the 512 MB kept otherwise: recovery_evidence() would find the round's first
# segment recycled and count nothing.  So when a --conf line turns it on,
# the primary keeps (and lets checkpoints grow to) 2 GB; a --conf line
# setting either of these still wins.
WAL_KEEP_SIZE=512MB
MAX_WAL_SIZE=1GB
for line in "${EXTRA_CONF[@]}"; do
	case $line in
		*wal_consistency_checking*) WAL_KEEP_SIZE=2GB; MAX_WAL_SIZE=2GB ;;
	esac
done

[ -n "$PREFIX" ] ||
	die "no PostgreSQL installation given: pass --prefix <prefix> or set RECOVERY_PREFIX (make recovery-check RECOVERY_PREFIX=<prefix>).  The extension is built and INSTALLED into it, so there is no default."
PREFIX=$(cd "$PREFIX" 2>/dev/null && pwd) || die "prefix $PREFIX does not exist"
PGBIN=$PREFIX/bin
for prog in initdb pg_ctl psql pgbench pg_basebackup pg_waldump pg_config; do
	[ -x "$PGBIN/$prog" ] || die "$PGBIN/$prog is missing; is $PREFIX a PostgreSQL install?"
done
case $ITERS in ''|*[!0-9]*) die "--iters wants a number" ;; esac
[ "$ITERS" -ge 1 ] || die "--iters must be at least 1"
case $SKIP_INSTALL in 0|1) ;; *) die "RECOVERY_SKIP_INSTALL wants 0 or 1, not '$SKIP_INSTALL'" ;; esac

if [ -n "$RUN_AS" ]; then
	id -u "$RUN_AS" >/dev/null 2>&1 || die "RECOVERY_RUN_AS: no such user '$RUN_AS'"
	[ "$(id -u "$RUN_AS")" != 0 ] || die "RECOVERY_RUN_AS must be an unprivileged user, not root"
	[ "$(id -u)" = 0 ] || die "RECOVERY_RUN_AS needs root (runuser) to switch to '$RUN_AS'"
	command -v runuser >/dev/null 2>&1 || die "RECOVERY_RUN_AS needs runuser(1), which is missing"
elif [ "$(id -u)" = 0 ]; then
	die "initdb and postgres refuse to run as root: set RECOVERY_RUN_AS=<unprivileged user> to run the clusters as that user"
fi

# ---------------------------------------------------------------- safety

for p in $FORBIDDEN_PORTS; do
	[ "$PRIMARY_PORT" = "$p" ] && die "port $p belongs to another cluster"
	[ "$STANDBY_PORT" = "$p" ] && die "port $p belongs to another cluster"
done

# as_server <program> [args]: run a program that must not run as root - initdb,
# pg_ctl, and pg_basebackup, whose output is a data directory the server will
# own - as RECOVERY_RUN_AS when that is set, else as the caller.  From / so that
# a working directory the other user cannot read is never an issue.
as_server() {
	if [ -n "$RUN_AS" ]; then
		(cd / && runuser -u "$RUN_AS" -- "$@")
	else
		"$@"
	fi
}

# Make the run's private directory, and give it to RECOVERY_RUN_AS if set: the
# clusters, their logs and the socket directory are all created inside it by
# that user.  A Unix-domain socket path is limited to ~107 bytes, so when the
# private directory's path is too long for "<dir>/.s.PGSQL.<port>" (a deep
# TMPDIR) the socket directory is a second mktemp -d under /tmp; it is removed
# with the first.
make_private_dirs() {
	mkdir -p "$TMPROOT" || die "cannot create $TMPROOT"
	BASE=$(mktemp -d "$TMPROOT/lion_recovery.XXXXXX") ||
		die "mktemp -d in $TMPROOT failed"
	BASE=$(cd "$BASE" && pwd)
	SOCKDIR=$BASE/sock
	if [ "${#SOCKDIR}" -gt 80 ]; then
		SOCKDIR=$(mktemp -d /tmp/lion_recovery_sock.XXXXXX) ||
			die "mktemp -d for the socket directory failed"
		SOCKDIR_OWN=$SOCKDIR
	else
		mkdir "$SOCKDIR"
	fi
	PRIMARY_DATA=$BASE/primary
	STANDBY_DATA=$BASE/standby
	PRIMARY_LOG=$BASE/primary.log
	STANDBY_LOG=$BASE/standby.log
	if [ -n "$RUN_AS" ]; then
		chown "$RUN_AS" "$BASE" "$SOCKDIR" || die "cannot give $BASE to $RUN_AS"
	fi
}

# rm -rf only a directory this run made: the path mktemp returned, still of
# the shape it was made with.
remove_private_dir() {
	local d=$1 pattern=$2
	[ -n "$d" ] && [ -d "$d" ] || return 0
	case $d in
		$pattern) rm -rf "$d" ;;
		*) echo "not removing $d: not a directory this run created" >&2 ;;
	esac
}

# ---------------------------------------------------------------- shutdown

# A cluster that is not running is left as it is: pg_ctl stop fails on one,
# and without the `|| true` set -e would end the whole run there, silently -
# which is what phase 4 met, calling this on the standby phase 3 had stopped.
stop_hard() {
	local d=$1
	[ -n "$d" ] && [ -d "$d" ] || return 0
	as_server "$PGBIN/pg_ctl" -D "$d" stop -m immediate -w -t 30 >/dev/null 2>&1 || true
	return 0
}

kill_tree() {
	local pid=$1 k
	[ -n "$pid" ] || return 0
	kill -0 "$pid" 2>/dev/null || return 0
	for k in $(pgrep -P "$pid" 2>/dev/null || true); do kill -9 "$k" 2>/dev/null || true; done
	kill -9 "$pid" 2>/dev/null || true
	wait "$pid" 2>/dev/null || true
	return 0
}

cleanup() {
	local rc=$?
	trap - EXIT
	set +e
	kill_tree "$RRPID"
	for p in $P4PIDS; do kill_tree "$p"; done
	kill_tree "$VACPID"
	kill_tree "$VACLOOP"
	stop_hard "$STANDBY_DATA"
	stop_hard "$PRIMARY_DATA"
	if [ "$rc" != 0 ]; then
		mkdir -p "$LOGDIR"
		for f in "$PRIMARY_LOG" "$STANDBY_LOG"; do
			[ -f "$f" ] && cp "$f" "$LOGDIR/$(basename "$f")"
		done
		echo "server logs copied to $LOGDIR" >&2
	fi
	if [ "$KEEP" = 1 ]; then
		echo "--keep: clusters left in $BASE (prefix $PREFIX)"
	else
		# BUILDDIR is inside BASE; the socket directory may not be.
		remove_private_dir "$BASE" '*/lion_recovery.??????'
		remove_private_dir "$SOCKDIR_OWN" '/tmp/lion_recovery_sock.??????'
	fi
	exit $rc
}

# ---------------------------------------------------------------- psql

psql_p() {
	"$PGBIN/psql" -X -q -v ON_ERROR_STOP=1 -h "$SOCKDIR" -p "$PRIMARY_PORT" \
		-U postgres -d "$DBNAME" "$@"
}
# Planner settings for a reference count that must come from the heap, not
# from a lion index: no index plan of any kind and no count pushdown.
HEAPONLY="-c pg_lion.enable_count_pushdown=off -c enable_indexscan=off -c enable_indexonlyscan=off -c enable_bitmapscan=off"
psql_s() {
	"$PGBIN/psql" -X -q -v ON_ERROR_STOP=1 -h "$SOCKDIR" -p "$STANDBY_PORT" \
		-U postgres -d "$DBNAME" "$@"
}

# wait_true <psql fn> <boolean sql> <seconds> <what>
wait_true() {
	local fn=$1 sql=$2 limit=$3 what=$4 i=0 v=""
	while [ "$i" -lt "$((limit * 10))" ]; do
		v=$("$fn" -tAc "$sql" 2>/dev/null || true)
		[ "$v" = "t" ] && return 0
		i=$((i + 1))
		nap 0.1
	done
	die "timed out after ${limit}s waiting for: $what"
}

# ---------------------------------------------------------------- build

build_extension() {
	log "-- building the extension into $PREFIX"
	BUILDDIR=$BASE/build
	mkdir "$BUILDDIR"
	cp -r "$PROJECT/src" "$PROJECT/Makefile" "$PROJECT"/pg_lion*.control \
		"$PROJECT"/pg_lion*--*.sql "$BUILDDIR/"
	# Never reuse objects built against another server, nor the header
	# dependencies the compiler wrote beside them (src/*.d), which name that
	# server's headers.
	rm -f "$BUILDDIR"/src/*.o "$BUILDDIR"/src/*.d "$BUILDDIR"/src/*.bc "$BUILDDIR"/*.so
	mkdir -p "$BUILDDIR/test/sql" "$BUILDDIR/test/isolation"
	(
		cd "$BUILDDIR" &&
		make -s PG_CONFIG="$PGBIN/pg_config" &&
		make -s PG_CONFIG="$PGBIN/pg_config" install
	) >>"$RUNLOG" 2>&1 || die "extension build failed (see $RUNLOG, build dir $BUILDDIR)"
}

# RECOVERY_SKIP_INSTALL=1: the extension was installed into the prefix
# beforehand (by someone who can write to it) and is used as it is.  Both
# control files and the library must be there; a library in the tree that is
# not the installed one is reported, because the installed one is what the
# clusters load.
check_installed_extension() {
	local libdir sharedir lib c
	libdir=$("$PGBIN/pg_config" --pkglibdir) || die "pg_config --pkglibdir failed"
	sharedir=$("$PGBIN/pg_config" --sharedir) || die "pg_config --sharedir failed"
	for c in pg_lion pg_lion_citext; do
		[ -f "$sharedir/extension/$c.control" ] ||
			die "RECOVERY_SKIP_INSTALL=1, but $sharedir/extension/$c.control is missing: install the extension into $PREFIX first"
	done
	if [ -f "$libdir/pg_lion.so" ]; then
		lib=$libdir/pg_lion.so
	elif [ -f "$libdir/pg_lion.dylib" ]; then
		lib=$libdir/pg_lion.dylib
	else
		die "RECOVERY_SKIP_INSTALL=1, but there is no pg_lion library in $libdir: install the extension into $PREFIX first"
	fi
	log "-- RECOVERY_SKIP_INSTALL=1: using the extension installed in $PREFIX ($lib)"
	if [ -f "$PROJECT/$(basename "$lib")" ] && ! cmp -s "$PROJECT/$(basename "$lib")" "$lib"; then
		log "   WARNING: $PROJECT/$(basename "$lib") differs from the installed $lib; the installed one is tested"
	fi
}

# ---------------------------------------------------------------- clusters

write_primary_conf() {
	cat >>"$PRIMARY_DATA/postgresql.conf" <<-EOF

		# ---- test/recovery/run.sh
		listen_addresses = ''
		port = $PRIMARY_PORT
		unix_socket_directories = '$SOCKDIR'
		# The crash tests are about durability: a commit pgbench saw must be on
		# disk before the server is taken away.
		fsync = on
		synchronous_commit = on
		wal_level = replica
		max_wal_senders = 5
		max_wal_size = $MAX_WAL_SIZE
		# Enough that the range each restart replays is still on disk when
		# recovery_evidence() runs pg_waldump over it.
		wal_keep_size = $WAL_KEEP_SIZE
		hot_standby = on
		# Explicit VACUUMs only, so the crash points and the recovery-conflict
		# test are the ones this script chose.
		autovacuum = off
		max_connections = 30
		# So pg_lion.enable_count_pushdown exists in every session,
		# including the ones the read-only helpers run in on a standby.
		session_preload_libraries = 'pg_lion'
		log_line_prefix = '%m [%p] '
		log_checkpoints = on
	EOF

	# --mode and --conf, last so that they win.
	local line
	for line in "${EXTRA_CONF[@]}"; do
		printf '%s\n' "$line" >>"$PRIMARY_DATA/postgresql.conf"
	done
}

write_standby_conf() {
	local feedback=$1 delay=$2
	cat >"$STANDBY_DATA/standby_extra.conf" <<-EOF
		# ---- test/recovery/run.sh (included last, so it wins)
		hot_standby = on
		hot_standby_feedback = $feedback
		max_standby_streaming_delay = $delay
		# The default 10s would make the feedback-on case depend on luck.
		wal_receiver_status_interval = 1s
	EOF
	# Written by the caller, read by a server that may run as someone else.
	if [ -n "$RUN_AS" ]; then
		chown "$RUN_AS" "$STANDBY_DATA/standby_extra.conf"
	fi
}

start_node() {
	local d=$1 port=$2 logf=$3 tries=0
	while :; do
		if as_server "$PGBIN/pg_ctl" -D "$d" -l "$logf" \
			-o "-p $port -k $SOCKDIR -c listen_addresses=''" -w -t 120 start \
			>>"$RUNLOG" 2>&1; then
			break
		fi
		tries=$((tries + 1))
		[ "$tries" -ge 3 ] && die "could not start the cluster at $d (see $logf)"
		# A SIGKILLed postmaster can leave a backend attached for a moment.
		nap 1
	done
}

verify_node() {
	local fn=$1 d=$2 actual
	actual=$("$fn" -tAc "show data_directory" 2>/dev/null || true)
	[ -n "$actual" ] || die "could not ask the cluster at $d which data directory it serves"
	[ "$(cd "$actual" 2>/dev/null && pwd)" = "$(cd "$d" && pwd)" ] ||
		die "connected to a cluster with data_directory '$actual', expected '$d'"
}

init_primary() {
	log "-- initdb $PRIMARY_DATA"
	as_server "$PGBIN/initdb" -D "$PRIMARY_DATA" -U postgres --auth=trust --no-sync \
		--data-checksums -E UTF8 --locale=C >>"$RUNLOG" 2>&1 ||
		die "initdb failed (see $RUNLOG)"
	write_primary_conf
	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"
	[ "$(psql_p -tAc 'select pg_is_in_recovery()')" = "f" ] ||
		die "the fresh primary is in recovery"
}

# ---------------------------------------------------------------- checks

NCHECKS=0

# run_check <label> <psql fn> <sql returning (ok boolean, detail text)>
#
# Fails the run on any row with ok = false, and on an empty result (a check
# battery that produced nothing has checked nothing).  Sets NCHECKS.  Never
# call this in a command substitution: it must be able to exit the script.
run_check() {
	local label=$1 fn=$2 sql=$3 out fails rc=0 errf=$BASE/check.err
	out=$("$fn" -tA -F'|' -c "$sql" 2>"$errf") || rc=$?
	{ echo "---- $label"; printf '%s\n' "$out"; echo "---- $label stderr"; cat "$errf"; } >>"$RUNLOG"
	grep '^WARNING' "$errf" >>"$BASE/warnings.txt" 2>/dev/null || true
	[ "$rc" = 0 ] || die "$label: the check query itself failed: $(tail -3 "$errf")"
	fails=$(printf '%s\n' "$out" | grep -v '^t|' || true)
	if [ -n "$fails" ]; then
		echo "FAILED checks ($label):" >&2
		printf '%s\n' "$fails" >&2
		die "$label: $(printf '%s\n' "$fails" | wc -l) check(s) failed (full log: $RUNLOG)"
	fi
	NCHECKS=$(printf '%s\n' "$out" | wc -l)
}

# vacuum_page_counts <label> <table> <index>
#
# VACUUM (VERBOSE) the table and read what it reports about the index's free
# pages (DESIGN.md §18, "Page counts") into NEWLY, CURRENT and REUSABLE: the
# pages this VACUUM freed, the free pages the index holds, and those of them
# the next allocation could take already.  Only a shell can read them - no SQL
# function sees an IndexBulkDeleteResult - which is why this lives here and
# not in the regression suite.  Never call it in a command substitution.
NEWLY=0
CURRENT=0
REUSABLE=0
vacuum_page_counts() {
	local label=$1 table=$2 index=$3 line
	line=$(psql_p -c "VACUUM (VERBOSE, INDEX_CLEANUP ON) $table" 2>&1 |
		tee -a "$RUNLOG" | grep "index \"$index\": pages:" || true)
	[ -n "$line" ] || die "$label: VACUUM VERBOSE printed no page counts for $index"
	NEWLY=$(printf '%s' "$line" | sed -n 's/.* \([0-9]*\) newly deleted.*/\1/p')
	CURRENT=$(printf '%s' "$line" | sed -n 's/.* \([0-9]*\) currently deleted.*/\1/p')
	REUSABLE=$(printf '%s' "$line" | sed -n 's/.* \([0-9]*\) reusable.*/\1/p')
	[ -n "$NEWLY" ] && [ -n "$CURRENT" ] && [ -n "$REUSABLE" ] ||
		die "$label: could not read the page counts in: $line"
	log "$label: $line"
}

# ---------------------------------------------------------------- phase 1

crash_immediate() {
	as_server "$PGBIN/pg_ctl" -D "$PRIMARY_DATA" stop -m immediate -w -t 60 >>"$RUNLOG" 2>&1 ||
		die "pg_ctl stop -m immediate failed"
}

crash_kill9() {
	local pm kids k i=0
	pm=$(head -1 "$PRIMARY_DATA/postmaster.pid")
	[ -n "$pm" ] || die "no postmaster.pid to kill"
	kids=$(pgrep -P "$pm" 2>/dev/null | tr '\n' ' ' || true)
	kill -9 "$pm" 2>/dev/null || true
	# Backends normally exit on their own when the postmaster dies
	# (WL_EXIT_ON_PM_DEATH); make sure none is left attached to the shared
	# memory segment, or the restart would refuse to run.
	for k in $kids; do kill -9 "$k" 2>/dev/null || true; done
	while kill -0 "$pm" 2>/dev/null; do
		i=$((i + 1))
		[ "$i" -gt 300 ] && die "postmaster $pm survived SIGKILL"
		nap 0.1
	done
	shm_detach_all
}

# A child the postmaster forked after `kids` was listed - or one busy enough
# not to notice its death for a while - is still attached to the cluster's
# System V shared memory interlock segment, and while anything is, the next
# start refuses to run ("pre-existing shared memory block ... is still in
# use"): on master's CI runner one outlived start_node()'s retries.  Find
# every process with the segment mapped, by the key postmaster.pid records
# on its seventh line, and SIGKILL it too, until none is left.
shm_detach_all() {
	local key pat pids k i=0
	key=$(sed -n 7p "$PRIMARY_DATA/postmaster.pid" 2>/dev/null | awk '{print $1}')
	case $key in '' | 0 | *[!0-9]*) return 0 ;; esac
	pat=$(printf 'SYSV%08x' "$key")
	while :; do
		# grep finds nothing (status 1) when all is well, and cannot read
		# some processes' maps (status 2): neither may end the run under -e.
		pids=$({ grep -l "$pat" /proc/[0-9]*/maps 2>/dev/null || true; } |
			cut -d/ -f3 | tr '\n' ' ')
		[ -n "${pids// /}" ] || break
		for k in $pids; do kill -9 "$k" 2>/dev/null || true; done
		i=$((i + 1))
		[ "$i" -gt 300 ] && die "still attached to the shared memory of $PRIMARY_DATA: $pids"
		nap 0.1
	done
}

# Everything the log says about the recovery that just happened, so an
# iteration that replayed nothing cannot pass silently.  Prints
# "<redo start LSN> <generic WAL records replayed>".
recovery_evidence() {
	local off=$1 txt redo_start redo_end generic segs startseg endseg
	txt=$(tail -c "+$((off + 1))" "$PRIMARY_LOG")
	printf '%s\n' "$txt" >>"$RUNLOG"
	# Not `printf | grep -q`: grep -q exits at the first match, the writer can
	# then die of SIGPIPE, and under pipefail that failed the check about once
	# in a thousand runs although the line was there.
	grep -q "database system was not properly shut down" <<<"$txt" ||
		die "the restart did not report an unclean shutdown: the crash did not take effect"
	redo_start=$(printf '%s' "$txt" |
		sed -n 's/.*redo starts at \([0-9A-F]*\/[0-9A-F]*\).*/\1/p' | head -1)
	[ -n "$redo_start" ] ||
		die "the restart replayed no WAL ('redo starts at' is missing): nothing was tested"
	redo_end=$(printf '%s' "$txt" |
		sed -n 's/.*redo done at \([0-9A-F]*\/[0-9A-F]*\).*/\1/p' | head -1)
	[ -n "$redo_end" ] ||
		die "the restart printed no 'redo done at': recovery did not finish normally"
	# The write paths of this AM emit nothing but records of ONE resource
	# manager, so a range that replayed none of them replayed no index change
	# at all.  Both names are counted because which one it is depends on
	# --mode: generic WAL is "Generic", and in rmgr mode pg_waldump prints
	# "customNNN", since it does not load the module and cannot ask it for a
	# name (DESIGN.md §25).
	#
	# pg_waldump reads the range off DISK, so a range longer than
	# wal_keep_size is no longer there to be counted; that is a limit of the
	# EVIDENCE, not of the test, so it is reported and skipped rather than
	# treated as a failure.  wal_consistency_checking makes every record carry
	# a full-page image and is what makes a range that long, so the primary
	# keeps more when it is on (WAL_KEEP_SIZE).
	#
	# The segments the range starts and ends in are named explicitly.  Given
	# only a directory, pg_waldump learns the segment size from the header of
	# the first WAL file readdir() happens to return, and a segment the
	# server has preallocated but not yet written has an all-zero header:
	# "invalid WAL segment size ... (0 bytes)", intermittently, and the
	# round's evidence was lost.  (Naming the start alone is not enough:
	# pg_waldump then requires the end to lie in that same segment.)  The
	# primary is up and not in recovery here, so pg_walfile_name() can name
	# both, on the timeline crash recovery kept.
	segs=$(psql_p -tAc "select pg_walfile_name('$redo_start') || ' ' || pg_walfile_name('$redo_end')") ||
		die "could not name the WAL segments of $redo_start..$redo_end"
	startseg=${segs% *}
	endseg=${segs#* }
	generic=$("$PGBIN/pg_waldump" -p "$PRIMARY_DATA/pg_wal" \
		-s "$redo_start" -e "$redo_end" "$startseg" "$endseg" 2>/dev/null |
		grep -cE 'rmgr: (Generic|custom[0-9]+)' || true)
	if [ "$generic" -eq 0 ]; then
		if "$PGBIN/pg_waldump" -p "$PRIMARY_DATA/pg_wal" -s "$redo_start" \
			-e "$redo_end" "$startseg" "$endseg" >/dev/null 2>>"$RUNLOG"; then
			die "the range $redo_start..$redo_end was replayed but holds no index WAL record: this round tested no index change"
		fi
		# To stderr: stdout is the one line the caller parses.
		log "note: pg_waldump could not read $redo_start..$redo_end (its error is in $RUNLOG); the replay happened, the count did not" >&2
	fi
	echo "$redo_start $generic"
}

start_vacuum_loop() {
	(
		while :; do
			"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres \
				-d "$DBNAME" -c "VACUUM (INDEX_CLEANUP ON) lion_rec" >/dev/null 2>&1 || true
			command sleep 0.6
		done
	) &
	VACLOOP=$!
}

stop_vacuum_loop() {
	kill_tree "$VACLOOP"
	VACLOOP=""
}

phase1() {
	local it style vacuum_crash off ev t0 t1 nrows ck
	log ""
	log "=== phase 1: crash recovery under load ($ITERS iterations) ==="
	for it in $(seq 1 "$ITERS"); do
		t0=$(now_ms)
		if [ $((it % 2)) -eq 1 ]; then style=immediate; else style=kill9; fi
		vacuum_crash=0
		case $it in 5|6) vacuum_crash=1 ;; esac

		if [ "$vacuum_crash" = 0 ]; then start_vacuum_loop; fi
		PGOPTIONS='-c synchronous_commit=on' \
			"$PGBIN/pgbench" -n -r -T 2 -c 4 -j 2 -f "$HERE/writer.sql" \
			-h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres "$DBNAME" \
			>>"$RUNLOG" 2>&1 || die "iteration $it: the pgbench writer failed (see $RUNLOG)"
		stop_vacuum_loop

		off=$(stat -c %s "$PRIMARY_LOG")
		if [ "$vacuum_crash" = 1 ]; then
			"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres \
				-d "$DBNAME" -c "VACUUM (INDEX_CLEANUP ON, VERBOSE) lion_rec" \
				>>"$RUNLOG" 2>&1 &
			VACPID=$!
			nap 0.35
		fi

		"crash_$style"

		if [ "$vacuum_crash" = 1 ]; then
			wait "$VACPID" 2>/dev/null || true
			VACPID=""
		fi

		start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
		verify_node psql_p "$PRIMARY_DATA"
		[ "$(psql_p -tAc 'select pg_is_in_recovery()')" = "f" ] ||
			die "iteration $it: the primary is still in recovery after pg_ctl -w start"
		ev=$(recovery_evidence "$off") || exit 1
		# Anything but "<lsn> <count>" here would make the arithmetic below
		# an expansion error, and bash abandons the whole top-level command
		# on one - this phase - and carries on with the next as if it had
		# passed.  So check the shape first and fail loudly.
		case ${ev#* } in
			''|*[!0-9]*) die "iteration $it: unreadable recovery evidence: '$ev'" ;;
		esac
		GENERIC_TOTAL=$((GENERIC_TOTAL + ${ev#* }))

		# Before VACUUM the index may still hold TIDs of dead tuples, so ntids
		# is only required to be >= what the heap needs; after it, equal.
		run_check "iter $it post-recovery" psql_p "select * from lion_rec_check(true, false)"
		ck=$NCHECKS
		psql_p -c "VACUUM (INDEX_CLEANUP ON) lion_rec" >>"$RUNLOG" 2>&1
		run_check "iter $it post-vacuum" psql_p "select * from lion_rec_check(false, true)"

		nrows=$(psql_p -tAc "select count(*) from lion_rec")
		t1=$(now_ms)
		log "iter $it: crash=$style vacuum_in_flight=$vacuum_crash redo_from=${ev% *} index_wal_replayed=${ev#* } rows=$nrows checks=$((ck + NCHECKS)) $(( (t1 - t0) / 1000 ))s"
		SUMMARY+=("phase1 iter $it  crash=$style vacuum_in_flight=$vacuum_crash index_wal_replayed=${ev#* } rows=$nrows checks=$((ck + NCHECKS))")
	done
	[ "$GENERIC_TOTAL" -gt 0 ] ||
		die "no generic WAL record was ever replayed: the crash phase proved nothing"
}

# ------------------------------------------------------------- phase 1b

# A crash that lands exactly between the two steps of a whole-chain free
# (DESIGN.md §18): the entries are deleted and WAL-logged, their container
# pages are unreachable but not yet marked LION_PAGE_DELETED, and the server
# dies.  The pages are then leaked, which is harmless by construction -
# nothing references them - and the leak sweep at the end of the next
# ambulkdelete is what turns them into free pages again.
#
# The crash point is deterministic: an injection point parks the VACUUM in
# that window and the server is pulled out from under it.  A dedicated table
# is used so that the keys that must empty out are exactly the ones this test
# deletes, and so that nothing of the phase 1 fixture is disturbed.
phase1b() {
	local before after freed parked vacpid off warn
	log ""
	log "=== phase 1b: a crash between entry deletion and page marking ==="

	psql_p -c "CREATE EXTENSION IF NOT EXISTS injection_points" >>"$RUNLOG" 2>&1 ||
		{ no_injection_points "phase 1b" "phase1b            skipped (no injection points)"; return 0; }

	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1b: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS lion_leak;
		CREATE TABLE lion_leak (id int, k int NOT NULL);
		CREATE INDEX lion_leak_k ON lion_leak USING lion (k)
			WITH (inline_limit = 64);
		INSERT INTO lion_leak SELECT i, i % 6 FROM generate_series(1, 120000) i;
	SQL
	psql_p -c "VACUUM (ANALYZE) lion_leak" >>"$RUNLOG" 2>&1
	before=$(psql_p -tAc "select container_pages from lion_index_stats('lion_leak_k')")
	psql_p -c "SET synchronous_commit = on; DELETE FROM lion_leak WHERE k IN (0,1,2,3)" \
		>>"$RUNLOG" 2>&1 || die "phase 1b: delete failed"

	# Park the VACUUM in the window and crash the server under it.
	psql_p -c "SELECT injection_points_attach('lion-vacuum-entries-deleted', 'wait')" \
		>>"$RUNLOG" 2>&1 || die "phase 1b: could not attach the injection point"

	off=$(stat -c %s "$PRIMARY_LOG")
	"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres -d "$DBNAME" \
		-c "VACUUM (INDEX_CLEANUP ON) lion_leak" >>"$RUNLOG" 2>&1 &
	vacpid=$!

	wait_true psql_p \
		"select count(*) > 0 from pg_stat_activity where wait_event = 'lion-vacuum-entries-deleted'" \
		60 "the VACUUM to park between the entry deletion and the page marking"

	# ambulkdelete's records belong to no transaction, so nothing has flushed
	# them: a crash here would simply lose the whole index vacuum and VACUUM
	# would redo it (correct, but it would test nothing).  A checkpoint makes
	# the first of the two steps durable, so the crash really does land
	# BETWEEN them.
	psql_p -c "CHECKPOINT" >>"$RUNLOG" 2>&1
	parked=$(psql_p -tAc "select entries from lion_index_stats('lion_leak_k')")
	log "phase 1b: parked with $parked of 6 entries left"
	[ "$parked" -lt 6 ] ||
		die "phase 1b: the VACUUM parked before it had deleted an entry"

	crash_immediate
	wait "$vacpid" 2>/dev/null || true

	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"

	# recovery_evidence() is deliberately NOT used here.  The checkpoint above
	# is what makes this crash land between the two steps, and it also leaves
	# redo almost nothing to do; what this round proves is the leak and the
	# sweep, and phase 1 is what proves replay.
	tail -c "+$((off + 1))" "$PRIMARY_LOG" >>"$RUNLOG"
	grep -q "database system was not properly shut down" \
		<<<"$(tail -c "+$((off + 1))" "$PRIMARY_LOG")" ||
		die "phase 1b: the restart did not report an unclean shutdown"

	# The entries really are gone, the pages really are leaked (verify()
	# tolerates unreferenced EMPTY container pages, with a WARNING), and the
	# index still answers correctly.
	after=$(psql_p -tAc "select entries from lion_index_stats('lion_leak_k')")
	[ "$after" = "$parked" ] ||
		die "phase 1b: the entry deletion did not survive the crash ($after entries, $parked before)"

	# The pages of the chain that entry owned are now LEAKED: unreferenced,
	# and not yet marked free.  verify() says so (a WARNING, not an error) and
	# the index still answers correctly.
	warn=$(psql_p -c "SELECT lion_index_verify('lion_leak_k', true)" 2>&1 >>"$RUNLOG" |
		grep -c 'unused and unreachable' || true)
	[ "$warn" -gt 0 ] ||
		die "phase 1b: no page was leaked, so the crash did not land between the two steps"
	run_check "phase 1b post-crash" psql_p \
		"select (select entries from lion_index_stats('lion_leak_k')) < 6 as ok,
				format('entries left: %s of 6, %s pages leaked',
					   (select entries from lion_index_stats('lion_leak_k')), $warn) as detail
		 union all
		 select (select count(*) from lion_leak where k = 4) = 20000,
				'k = 4 still answers 20000 rows'
		 union all
		 select (select count(*) from lion_leak where k = 0) = 0,
				'k = 0 answers no rows'"

	# The sweep recovers them: the next ambulkdelete reads the blocks its walk
	# did not account for, turns the leaked ones into DELETED pages and
	# records them in the free space map.  verify() then has nothing left to
	# warn about, which is the assertion that the leak is gone.
	psql_p -c "SET synchronous_commit = on; DELETE FROM lion_leak WHERE k = 4" \
		>>"$RUNLOG" 2>&1
	vacuum_page_counts "phase 1b" lion_leak lion_leak_k
	freed=$(psql_p -tAc "select deleted_pages from lion_index_stats('lion_leak_k')")
	[ "$freed" -gt 0 ] ||
		die "phase 1b: the leak sweep recovered no pages (deleted_pages = $freed)"

	# What VACUUM reported about it (DESIGN.md §18, "Page counts").  The index
	# held no DELETED page before, so every one it holds now is one this
	# VACUUM freed - the leaked pages its sweep found and k = 4's chain - and
	# none of them is reusable yet; the other free pages, if a split left any,
	# are all-zero and reusable.  A page freed with its chain is counted once,
	# not again by the sweep, and a page freed now is not reusable.
	[ "$NEWLY" = "$freed" ] ||
		die "phase 1b: VACUUM reported $NEWLY pages newly deleted, the index holds $freed DELETED pages"
	[ "$((CURRENT - NEWLY))" = "$REUSABLE" ] ||
		die "phase 1b: VACUUM reported $CURRENT pages currently deleted and $REUSABLE reusable for $NEWLY newly deleted"
	warn=$(psql_p -c "SELECT lion_index_verify('lion_leak_k', true)" 2>&1 >>"$RUNLOG" |
		grep -c 'unused and unreachable' || true)
	[ "$warn" = 0 ] ||
		die "phase 1b: $warn page(s) are still leaked after the sweep"

	psql_p -c "SELECT injection_points_detach('lion-vacuum-entries-deleted')" \
		>>"$RUNLOG" 2>&1 || true
	psql_p -c "DROP TABLE lion_leak" >>"$RUNLOG" 2>&1

	log "phase 1b: the entry deletion survived the crash and the sweep recovered $freed pages (of $before container pages)"
	SUMMARY+=("phase1b            crash between the two steps; sweep recovered $freed pages")
}

# ------------------------------------------------------------- phase 1c

# A crash that lands exactly between the two records of a DIRECTORY SPLIT
# (DESIGN.md §21): the leaf has been split and flagged LION_PAGE_INCOMPLETE_
# SPLIT, its right sibling exists and is linked in, but the parent has no
# downlink for it yet, and the server dies.  Readers never noticed (the right
# link gets them there), and the first writer that descends to that page
# finishes the split before touching it.
#
# The crash point is deterministic: an injection point parks the inserting
# backend in that window and the server is pulled out from under it.
phase1c() {
	local off before after warn parked
	log ""
	log "=== phase 1c: a crash between a directory split and its downlink ==="

	psql_p -c "CREATE EXTENSION IF NOT EXISTS injection_points" >>"$RUNLOG" 2>&1 ||
		{ no_injection_points "phase 1c" "phase1c            skipped (no injection points)"; return 0; }

	# A table whose index is created EMPTY and grown by inserts, so that the
	# splits really happen through aminsert and not through ambuild.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1c: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS lion_split;
		CREATE TABLE lion_split (id int, k text NOT NULL);
		CREATE INDEX lion_split_k ON lion_split USING lion (k);
		INSERT INTO lion_split SELECT i, 'k' || lpad(i::text, 40, '0')
			FROM generate_series(1, 4000) i;
	SQL
	psql_p -c "VACUUM (ANALYZE) lion_split" >>"$RUNLOG" 2>&1
	before=$(psql_p -tAc "select leaf_pages from lion_index_stats('lion_split_k')")
	[ "${before:-0}" -gt 1 ] ||
		die "phase 1c: the fixture did not split the directory at all"

	psql_p -c "SELECT injection_points_attach('lion-dir-split-incomplete', 'wait')" \
		>>"$RUNLOG" 2>&1 || die "phase 1c: could not attach the injection point"

	off=$(stat -c %s "$PRIMARY_LOG")
	"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres -d "$DBNAME" \
		-c "SET synchronous_commit = on;
		    INSERT INTO lion_split SELECT 100000 + i, 'k' || lpad((100000 + i)::text, 40, '0')
		      FROM generate_series(1, 4000) i" >>"$RUNLOG" 2>&1 &
	parked=$!

	wait_true psql_p \
		"select count(*) > 0 from pg_stat_activity where wait_event = 'lion-dir-split-incomplete'" \
		60 "an insert to park between the split and its downlink"

	# The split's record has been inserted but not flushed, and the crash would
	# otherwise simply lose it (and with it the whole split, which would test
	# nothing).  A COMMIT from another session with synchronous_commit = on
	# flushes the WAL stream up to its own commit LSN, which is past the split
	# record.  A CHECKPOINT would be the obvious way and is NOT usable here:
	# the parked backend holds the leaf's content lock EXCLUSIVE and the
	# checkpointer would block writing that very buffer.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL
		SET synchronous_commit = on;
		CREATE TABLE IF NOT EXISTS lion_split_flush (i int);
		INSERT INTO lion_split_flush VALUES (1);
	SQL

	crash_immediate
	wait "$parked" 2>/dev/null || true

	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"

	tail -c "+$((off + 1))" "$PRIMARY_LOG" >>"$RUNLOG"
	grep -q "database system was not properly shut down" \
		<<<"$(tail -c "+$((off + 1))" "$PRIMARY_LOG")" ||
		die "phase 1c: the restart did not report an unclean shutdown"

	# The page really came back flagged: verify() says so, as a WARNING and
	# not an error, and the index still answers correctly through the right
	# link that makes an unfinished split invisible to readers.
	warn=$(psql_p -c "SELECT lion_index_verify('lion_split_k')" 2>&1 >>"$RUNLOG" |
		grep -c 'unfinished split' || true)
	[ "$warn" -gt 0 ] ||
		die "phase 1c: no page came back with an unfinished split, so the crash missed the window"
	log "phase 1c: $warn page(s) came back with an unfinished split"
	run_check "phase 1c post-crash" psql_p \
		"select (select count(*) from lion_split where k = 'k' || lpad('1', 40, '0')) = 1,
				'the first key still answers'
		 union all
		 select (select count(*) from lion_split where k = 'k' || lpad('4000', 40, '0')) = 1,
				'the last committed key still answers'"

	# A WRITER's descent is what repairs it.  One insert is enough: the
	# descent that lands on the flagged page finishes its split first.
	psql_p -c "SET synchronous_commit = on;
	           INSERT INTO lion_split SELECT 900000 + i, 'k' || lpad((900000 + i)::text, 40, '0')
	             FROM generate_series(1, 400) i" >>"$RUNLOG" 2>&1 ||
		die "phase 1c: the repairing insert failed"

	warn=$(psql_p -c "SELECT lion_index_verify('lion_split_k', true)" 2>&1 >>"$RUNLOG" |
		grep -c 'unfinished split' || true)
	[ "$warn" = 0 ] ||
		die "phase 1c: $warn page(s) still have an unfinished split after a writer descended"

	after=$(psql_p -tAc "select leaf_pages from lion_index_stats('lion_split_k')")
	run_check "phase 1c repaired" psql_p \
		"select (select count(*) from lion_split where k = 'k' || lpad('900001', 40, '0')) = 1,
				'the repairing insert is indexed'
		 union all
		 select (select count(*) from lion_split) = (select count(*) from lion_split where k like 'k%'),
				'every row is still reachable through the index'"

	psql_p -c "SELECT injection_points_detach('lion-dir-split-incomplete')" \
		>>"$RUNLOG" 2>&1 || true
	psql_p -c "DROP TABLE lion_split" >>"$RUNLOG" 2>&1
	psql_p -c "DROP TABLE IF EXISTS lion_split_flush" >>"$RUNLOG" 2>&1

	log "phase 1c: the unfinished split survived the crash and a writer's descent repaired it ($before -> $after leaves)"
	SUMMARY+=("phase1c            crash between split and downlink; repaired on the next descent")
}


# ------------------------------------------------------------- phase 1d

# A crash that lands exactly between the two records of a POSTING-TREE LEAF
# SPLIT (DESIGN.md §22): the leaf has been split and flagged
# LION_PAGE_INCOMPLETE_SPLIT, its right sibling exists and is linked in, but
# the parent has no downlink for it yet, and the server dies.  Readers do not
# notice (the right link gets them there, and a sequential walk of the leaves
# is all they do), and the next writer finishes the split: an APPEND - the
# split was of the set's last leaf, so its right half is the entry's tail -
# and, after it, inserts that descend into every leaf's range.
#
# The crash point is deterministic: an injection point parks the inserting
# backend in that window and the server is pulled out from under it.
phase1d() {
	local off before after warn parked
	log ""
	log "=== phase 1d: a crash between a posting-tree split and its downlink ==="

	psql_p -c "CREATE EXTENSION IF NOT EXISTS injection_points" >>"$RUNLOG" 2>&1 ||
		{ no_injection_points "phase 1d" "phase1d            skipped (no injection points)"; return 0; }

	# Two keys taking alternate TIDs of wide rows, 65 to a heap page, built at
	# posting_fillfactor 10, which makes every container a leaf of its own:
	# each set has more than the 16 leaves a set without an internal level
	# may have (DESIGN.md §22), a tree, whose leaf splits take two records.
	# The split is aminsert's: the insert below fills the last leaf with
	# containers and splits it.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1d: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS lion_psplit;
		CREATE TABLE lion_psplit (id int, k int NOT NULL,
								  pad char(80) NOT NULL DEFAULT 'x');
		INSERT INTO lion_psplit SELECT i, i % 2 FROM generate_series(1, 90000) i;
		CREATE INDEX lion_psplit_k ON lion_psplit USING lion (k)
			WITH (posting_fillfactor = 10);
	SQL
	psql_p -c "VACUUM (ANALYZE) lion_psplit" >>"$RUNLOG" 2>&1
	before=$(psql_p -tAc "select container_pages from lion_index_stats('lion_psplit_k')")
	[ "${before:-0}" -gt 32 ] ||
		die "phase 1d: the fixture's posting sets have $before leaves, not a leaf a container"
	[ "$(psql_p -tAc "select max_posting_height from lion_index_stats('lion_psplit_k')")" = 1 ] ||
		die "phase 1d: the fixture's posting sets are not trees"

	psql_p -c "SELECT injection_points_attach('lion-posting-split-incomplete', 'wait')" \
		>>"$RUNLOG" 2>&1 || die "phase 1d: could not attach the injection point"

	off=$(stat -c %s "$PRIMARY_LOG")
	"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres -d "$DBNAME" \
		-c "SET synchronous_commit = on;
		    INSERT INTO lion_psplit SELECT 100000 + i, i % 2
		      FROM generate_series(1, 60000) i" >>"$RUNLOG" 2>&1 &
	parked=$!

	wait_true psql_p \
		"select count(*) > 0 from pg_stat_activity where wait_event = 'lion-posting-split-incomplete'" \
		60 "an insert to park between the posting split and its downlink"

	# The split's record has been inserted but not flushed, and the crash would
	# otherwise simply lose it (and with it the whole split, which would test
	# nothing).  A COMMIT from another session with synchronous_commit = on
	# flushes the WAL stream up to its own commit LSN, which is past the split
	# record.  A CHECKPOINT would be the obvious way and is NOT usable here:
	# the parked backend holds the leaf's content lock EXCLUSIVE and the
	# checkpointer would block writing that very buffer.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL
		SET synchronous_commit = on;
		CREATE TABLE IF NOT EXISTS lion_psplit_flush (i int);
		INSERT INTO lion_psplit_flush VALUES (1);
	SQL

	crash_immediate
	wait "$parked" 2>/dev/null || true

	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"

	tail -c "+$((off + 1))" "$PRIMARY_LOG" >>"$RUNLOG"
	grep -q "database system was not properly shut down" \
		<<<"$(tail -c "+$((off + 1))" "$PRIMARY_LOG")" ||
		die "phase 1d: the restart did not report an unclean shutdown"

	# The page really came back flagged: verify() says so, as a WARNING and
	# not an error, and the index still answers correctly - a reader walks the
	# leaves by their right links and an unfinished split is invisible to it.
	warn=$(psql_p -c "SELECT lion_index_verify('lion_psplit_k')" 2>&1 >>"$RUNLOG" |
		grep -c 'unfinished split' || true)
	[ "$warn" -gt 0 ] ||
		die "phase 1d: no posting page came back with an unfinished split, so the crash missed the window"
	log "phase 1d: $warn posting page(s) came back with an unfinished split"
	run_check "phase 1d post-crash" psql_p \
		"select (select count(*) from lion_psplit where k = 0) =
				(select count(*) from lion_psplit where id % 2 = 0),
				'k = 0 still answers exactly'
		 union all
		 select (select count(*) from lion_psplit where k = 1) =
				(select count(*) from lion_psplit where id % 2 = 1),
				'k = 1 still answers exactly'"

	# An APPEND, which is what an append-only table does next.  The append
	# hint names the set's last leaf - the right half of the unfinished split,
	# which has no downlink - and a split of THAT page would find no downlink
	# for it, nor would any later split of the tail (DESIGN.md §22 addendum,
	# "an unfinished split and the writers that do not descend").  An insert
	# that may split the page descends instead: the descent routes to the
	# flagged left half and finishes it on the way.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1d: appending after the crash failed"
		SET synchronous_commit = on;
		INSERT INTO lion_psplit SELECT 200000 + i, i % 2
		  FROM generate_series(1, 20000) i;
	SQL

	warn=$(psql_p -c "SELECT lion_index_verify('lion_psplit_k', true)" 2>&1 >>"$RUNLOG" |
		grep -c 'unfinished split' || true)
	[ "$warn" = 0 ] ||
		die "phase 1d: $warn posting page(s) still have an unfinished split after the append"
	run_check "phase 1d appended" psql_p \
		"select (select count(*) from lion_psplit where k = 0) =
				(select count(*) from lion_psplit where id % 2 = 0),
				'k = 0 answers exactly after the append'
		 union all
		 select (select count(*) from lion_psplit where k = 1) =
				(select count(*) from lion_psplit where id % 2 = 1),
				'k = 1 answers exactly after the append'
		 union all
		 select true, 'verify() is clean after the append'
		   from (select lion_index_verify('lion_psplit_k', true)) v"

	# Then writers that DESCEND into every leaf's range, the repair path this
	# phase was first written for: a third of the rows is deleted, VACUUM frees
	# those line pointers on every page, and the inserts that refill them have
	# container keys in every leaf's range.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1d: the repairing insert failed"
		SET synchronous_commit = on;
		DELETE FROM lion_psplit WHERE id % 3 = 0;
		VACUUM lion_psplit;
		INSERT INTO lion_psplit SELECT 900000 + i, i % 2
		  FROM generate_series(1, 20000) i;
	SQL

	warn=$(psql_p -c "SELECT lion_index_verify('lion_psplit_k', true)" 2>&1 >>"$RUNLOG" |
		grep -c 'unfinished split' || true)
	[ "$warn" = 0 ] ||
		die "phase 1d: $warn posting page(s) still have an unfinished split after a writer descended"

	after=$(psql_p -tAc "select container_pages from lion_index_stats('lion_psplit_k')")
	run_check "phase 1d repaired" psql_p \
		"select (select count(*) from lion_psplit) =
				(select count(*) from lion_psplit where k in (0, 1)),
				'every row is still reachable through the index'
		 union all
		 select (select ntids from lion_index_stats('lion_psplit_k')) =
				(select count(*) from lion_psplit),
				'ntids agrees with the heap'"

	psql_p -c "SELECT injection_points_detach('lion-posting-split-incomplete')" \
		>>"$RUNLOG" 2>&1 || true
	psql_p -c "DROP TABLE lion_psplit" >>"$RUNLOG" 2>&1
	psql_p -c "DROP TABLE IF EXISTS lion_psplit_flush" >>"$RUNLOG" 2>&1

	log "phase 1d: the unfinished posting split survived the crash and the next append repaired it ($before -> $after leaves)"
	SUMMARY+=("phase1d            crash between posting split and downlink; repaired by the next append")
}

# ------------------------------------------------------------- phase 1e

# A crash in the middle of a MULTI-LEAF INLINE spill (DESIGN.md §18, "The
# spill").  VACUUM's filtering turns one key's INLINE payload - eleven RUN
# containers over 150,000 clustered rows - into eleven NARROWs of 2,056 bytes
# (DESIGN.md §38), three to a leaf: four leaves, a set of no more than 16
# leaves, which has no internal level and whose first leaf is its head
# (DESIGN.md §22).  The spill logs the head first, as
# an internal page with no downlink, writes a record per other leaf, and dies
# before its last record: the head, made the first leaf, together with the
# rewritten entry.  Then:
#
#  * replay brings the leaves back and the entry stays INLINE with every TID
#    it had, the dead ones included: nothing is lost and nothing is half-CHAIN;
#  * lion_index_verify() calls the leaves a leak (a WARNING), not corruption;
#  * the next VACUUM spills the set properly, and its sweep frees the orphan
#    leaves, which are FULL: it frees an unreferenced leaf whose root is not a
#    live root of its key whatever it holds.  Its VERBOSE page counts
#    (DESIGN.md §18, "Page counts") are checked here, where a shell can read
#    them: the three orphans and the head their spill logged before them (with
#    no downlink) newly deleted, and never counted twice;
#  * and a second crash replays that VACUUM, the whole multi-leaf spill
#    included (under wal_consistency_checking when --conf asks for it).
#
# The crash point is deterministic: an injection point parks the VACUUM after
# its last leaf record and the server is pulled out from under it.
phase1e() {
	local off parked warn ev
	log ""
	log "=== phase 1e: a crash between a spill's leaves and its head ==="

	psql_p -c "CREATE EXTENSION IF NOT EXISTS injection_points" >>"$RUNLOG" 2>&1 ||
		{ no_injection_points "phase 1e" "phase1e            skipped (no injection points)"; return 0; }

	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1e: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS lion_mspill;
		CREATE TABLE lion_mspill (id int NOT NULL, k int NOT NULL);
		INSERT INTO lion_mspill SELECT i, 0 FROM generate_series(1, 150000) i;
		CREATE INDEX lion_mspill_k ON lion_mspill USING lion (k);
		DELETE FROM lion_mspill WHERE id % 10 = 3;
	SQL
	[ "$(psql_p -tAc "select inline_entries from lion_index_stats('lion_mspill_k')")" = 1 ] ||
		die "phase 1e: the fixture's posting set is not INLINE"

	psql_p -c "SELECT injection_points_attach('lion-spill-leaves-written', 'wait')" \
		>>"$RUNLOG" 2>&1 || die "phase 1e: could not attach the injection point"

	off=$(stat -c %s "$PRIMARY_LOG")
	"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres -d "$DBNAME" \
		-c "VACUUM (INDEX_CLEANUP ON) lion_mspill" >>"$RUNLOG" 2>&1 &
	parked=$!

	wait_true psql_p \
		"select count(*) > 0 from pg_stat_activity where wait_event = 'lion-spill-leaves-written'" \
		60 "the VACUUM to park between the spill's leaves and its head"

	# The leaf records are inserted and not flushed: ambulkdelete's records
	# belong to no transaction.  A COMMIT from another session flushes the
	# stream past them.  Not a CHECKPOINT: the parked backend holds the
	# directory leaf's cleanup lock and the checkpointer could block on it.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL
		SET synchronous_commit = on;
		CREATE TABLE IF NOT EXISTS lion_mspill_flush (i int);
		INSERT INTO lion_mspill_flush VALUES (1);
	SQL

	crash_immediate
	wait "$parked" 2>/dev/null || true

	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"
	ev=$(recovery_evidence "$off")
	GENERIC_TOTAL=$((GENERIC_TOTAL + ${ev#* }))

	# The leaves were replayed and nothing references them; the entry is the
	# INLINE entry it was, dead TIDs and all.
	warn=$(psql_p -c "SELECT lion_index_verify('lion_mspill_k', true)" 2>&1 >>"$RUNLOG" |
		grep -c 'unused and unreachable' || true)
	[ "$warn" -ge 4 ] ||
		die "phase 1e: $warn unreachable page(s) after the crash, expected the three orphan leaves and their head"
	run_check "phase 1e post-crash" psql_p \
		"select inline_entries = 1 and ntids = 150000,
				format('the entry is still INLINE with every TID (%s entries INLINE, ntids %s)',
					   inline_entries, ntids)
		   from lion_index_stats('lion_mspill_k')
		 union all
		 select container_pages = 3 and posting_internal_pages = 1,
				format('the three leaves and their head came back unreferenced (%s container pages, %s internal)',
					   container_pages, posting_internal_pages)
		   from lion_index_stats('lion_mspill_k')
		 union all
		 select lion_index_count('lion_mspill_k', 0) = 135000,
				'the index still counts all 135000 rows of k = 0'"

	psql_p -c "SELECT injection_points_detach('lion-spill-leaves-written')" \
		>>"$RUNLOG" 2>&1 || true

	# The next VACUUM spills the set and sweeps the orphans away, and says so:
	# the three orphans and their head - logged first, as an internal page with
	# no downlink, so that its block is never handed out again (DESIGN.md
	# §18) - are what it newly deleted, and they are not reusable yet.  Any
	# other free page is all-zero and reusable, so the two differences agree
	# exactly unless a page is counted twice or a fresh one is called
	# reusable.
	vacuum_page_counts "phase 1e" lion_mspill lion_mspill_k
	[ "$NEWLY" = 4 ] ||
		die "phase 1e: VACUUM reported $NEWLY pages newly deleted, expected the three orphan leaves and their head"
	[ "$((CURRENT - NEWLY))" = "$REUSABLE" ] ||
		die "phase 1e: VACUUM reported $CURRENT pages currently deleted and $REUSABLE reusable for $NEWLY newly deleted"

	warn=$(psql_p -c "SELECT lion_index_verify('lion_mspill_k', true)" 2>&1 >>"$RUNLOG" |
		grep -c 'unused and unreachable' || true)
	[ "$warn" -le 1 ] ||
		die "phase 1e: $warn page(s) are still leaked after the sweep"
	run_check "phase 1e swept" psql_p \
		"select inline_entries = 0 and posting_internal_pages = 0 and container_pages = 4,
				format('spilled: %s INLINE, %s internal, %s leaves',
					   inline_entries, posting_internal_pages, container_pages)
		   from lion_index_stats('lion_mspill_k')
		 union all
		 select deleted_pages = 4, format('%s DELETED pages, the three orphans and their head', deleted_pages)
		   from lion_index_stats('lion_mspill_k')
		 union all
		 select ntids = (select count(*) from lion_mspill), 'ntids agrees with the heap'
		   from lion_index_stats('lion_mspill_k')"

	# Replay that VACUUM too: the leaves, the head with the entry, the sweep.
	off=$(stat -c %s "$PRIMARY_LOG")
	psql_p >>"$RUNLOG" 2>&1 <<-SQL
		SET synchronous_commit = on;
		INSERT INTO lion_mspill_flush VALUES (2);
	SQL
	crash_immediate
	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"
	ev=$(recovery_evidence "$off")
	GENERIC_TOTAL=$((GENERIC_TOTAL + ${ev#* }))
	run_check "phase 1e replayed" psql_p \
		"select lion_index_verify('lion_mspill_k', true) is not null, 'verify'
		 union all
		 select inline_entries = 0 and ntids = (select count(*) from lion_mspill),
				'the spilled set replayed with every TID'
		   from lion_index_stats('lion_mspill_k')
		 union all
		 select lion_index_count('lion_mspill_k', 0) = 135000,
				'the index counts all 135000 rows of k = 0'"

	psql_p -c "DROP TABLE lion_mspill, lion_mspill_flush" >>"$RUNLOG" 2>&1

	log "phase 1e: the crash left the entry INLINE and three orphan leaves; the next VACUUM spilled it and freed them"
	SUMMARY+=("phase1e            crash inside a 4-leaf spill; nothing lost, orphans swept")
}

# ------------------------------------------------------------- phase 1f

# VACUUM of an rmgr-mode index on a server WITHOUT the resource manager
# (DESIGN.md §25, "VACUUM is the one path that could write without
# lion_wal_begin()").  The index is built while the library is preloaded,
# and the server is then restarted without it.  A VACUUM that removes nothing
# from the index - a partial index none of whose rows died - walks it all the
# same, and must write nothing under rmid 128, a manager this server does not
# have: not even the standby barrier of the pages it visited, which past
# pg_lion.vacuum_barrier_ranges goes out in records of its own.  Crash
# recovery would stop with a FATAL at the first such record, so nothing of
# the kind may be in the WAL, and the crash after it must recover.
#
# Both modes run it: generic mode preloads the library for the build, rmgr
# mode drops the preload (and wal_consistency_checking, which names the
# manager) for the VACUUM.
#
# An UNLOGGED rmgr-mode index rides along: it writes no WAL at all, so the
# server without the manager must let it be written - inserted into and
# vacuumed - and not refuse the write with the preload hint
# (lion_wal_begin()).  A LOGGED rmgr-mode index built with
# wal_mode = auto must still refuse an insert, and a VACUUM with dead rows to
# remove from it, with a hint naming both the preload and a REINDEX in
# generic mode - and that REINDEX must make it writable again.
#
# Where the server has injection points, one more index comes with work that
# any VACUUM finds, removing or not: phase 1b's crash, with the manager,
# leaves it with entries deleted and the pages of their chains leaked.  The
# VACUUM without the manager has nothing to remove from it and must go
# through, leaving the leaks for a VACUUM with the manager, which in rmgr
# mode frees them at the end of the phase.  That VACUUM is given dead rows
# to remove: one with none calls no ambulkdelete, and only ambulkdelete's
# walk sweeps the index for leaks (lionvacuumcleanup()), as only nbtree's
# btvacuumscan() finds its half-dead pages.
phase1f() {
	local before after custom off out leak=0 leaked parked vacpid
	log ""
	log "=== phase 1f: VACUUM of an rmgr-mode index on a server without the manager ==="

	as_server "$PGBIN/pg_ctl" -D "$PRIMARY_DATA" stop -m fast -w >>"$RUNLOG" 2>&1 ||
		die "phase 1f: could not stop the primary"
	as_server "$PGBIN/pg_ctl" -D "$PRIMARY_DATA" -l "$PRIMARY_LOG" \
		-o "-p $PRIMARY_PORT -k $SOCKDIR -c listen_addresses='' -c shared_preload_libraries=pg_lion" \
		-w -t 120 start >>"$RUNLOG" 2>&1 || die "phase 1f: could not start the primary with the preload"
	verify_node psql_p "$PRIMARY_DATA"

	# An rmgr-mode index with work left over for any VACUUM, removing or
	# not: phase 1b's crash between a whole-chain free's two steps, with the
	# manager, leaves its entries deleted and the pages of their chains
	# unreferenced.  A VACUUM without the manager has nothing to remove from
	# it - the entries and their TIDs are gone - and must leave the leaks
	# for one with the manager, not stop at the record that would free them.
	if psql_p -c "CREATE EXTENSION IF NOT EXISTS injection_points" >>"$RUNLOG" 2>&1; then
		leak=1
		psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1f: leak fixture failed"
			SET synchronous_commit = on;
			DROP TABLE IF EXISTS lion_norm_l;
			CREATE TABLE lion_norm_l (id int, k int NOT NULL);
			CREATE INDEX lion_norm_l_k ON lion_norm_l USING lion (k)
				WITH (wal_mode = rmgr, inline_limit = 64);
			INSERT INTO lion_norm_l SELECT i, i % 6 FROM generate_series(1, 120000) i;
			VACUUM (ANALYZE) lion_norm_l;
			DELETE FROM lion_norm_l WHERE k IN (0, 1, 2, 3);
			SELECT injection_points_attach('lion-vacuum-entries-deleted', 'wait');
		SQL
		"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres -d "$DBNAME" \
			-c "VACUUM (INDEX_CLEANUP ON) lion_norm_l" >>"$RUNLOG" 2>&1 &
		vacpid=$!
		wait_true psql_p \
			"select count(*) > 0 from pg_stat_activity where wait_event = 'lion-vacuum-entries-deleted'" \
			60 "the VACUUM of lion_norm_l to park between the entry deletion and the page marking"
		# what makes the first step durable, as in phase 1b
		psql_p -c "CHECKPOINT" >>"$RUNLOG" 2>&1
		parked=$(psql_p -tAc "select entries from lion_index_stats('lion_norm_l_k')")
		[ "$parked" -lt 6 ] ||
			die "phase 1f: the VACUUM of lion_norm_l parked before it had deleted an entry"
		crash_immediate
		wait "$vacpid" 2>/dev/null || true
		as_server "$PGBIN/pg_ctl" -D "$PRIMARY_DATA" -l "$PRIMARY_LOG" \
			-o "-p $PRIMARY_PORT -k $SOCKDIR -c listen_addresses='' -c shared_preload_libraries=pg_lion" \
			-w -t 120 start >>"$RUNLOG" 2>&1 || die "phase 1f: could not restart the primary with the preload"
		verify_node psql_p "$PRIMARY_DATA"
		leaked=$(psql_p -c "SELECT lion_index_verify('lion_norm_l_k', true)" 2>&1 >>"$RUNLOG" |
			grep -c 'unused and unreachable' || true)
		[ "$leaked" -gt 0 ] ||
			die "phase 1f: the crash leaked no page of lion_norm_l"
	elif [ "$REQUIRE_INJECTION" = 1 ]; then
		die "phase 1f's leaked index needs the injection_points extension, which this server does not have"
	else
		log "phase 1f: no injection_points extension, so no index with leaks to vacuum without the manager"
	fi

	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1f: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS lion_norm;
		CREATE TABLE lion_norm (id int NOT NULL, k int NOT NULL);
		INSERT INTO lion_norm SELECT i, i % 500 FROM generate_series(1, 100000) i;
		CREATE INDEX lion_norm_k ON lion_norm USING lion (k)
			WITH (wal_mode = rmgr, inline_limit = 64) WHERE id % 2 = 0;
		DROP TABLE IF EXISTS lion_norm_u;
		CREATE UNLOGGED TABLE lion_norm_u (id int NOT NULL, k int NOT NULL);
		INSERT INTO lion_norm_u SELECT i, i % 500 FROM generate_series(1, 20000) i;
		CREATE INDEX lion_norm_u_k ON lion_norm_u USING lion (k)
			WITH (wal_mode = rmgr, inline_limit = 64);
		DROP TABLE IF EXISTS lion_norm_w;
		CREATE TABLE lion_norm_w (id int NOT NULL, k int NOT NULL);
		INSERT INTO lion_norm_w SELECT i, i % 50 FROM generate_series(1, 20000) i;
		CREATE INDEX lion_norm_w_k ON lion_norm_w USING lion (k);
	SQL
	[ "$(psql_p -tAc "select lion_index_wal_mode('lion_norm_w_k'::regclass)")" = rmgr ] ||
		die "phase 1f: the index built under the preload with wal_mode = auto is not in rmgr mode"
	[ "$(psql_p -tAc "select lion_index_wal_mode('lion_norm_k'::regclass)")" = rmgr ] ||
		die "phase 1f: the partial index was not built in rmgr mode"
	[ "$(psql_p -tAc "select lion_index_wal_mode('lion_norm_u_k'::regclass)")" = rmgr ] ||
		die "phase 1f: the unlogged index was not built in rmgr mode"

	# A fast stop checkpoints, so nothing of the build is left to replay.
	as_server "$PGBIN/pg_ctl" -D "$PRIMARY_DATA" stop -m fast -w >>"$RUNLOG" 2>&1 ||
		die "phase 1f: could not stop the primary"
	as_server "$PGBIN/pg_ctl" -D "$PRIMARY_DATA" -l "$PRIMARY_LOG" \
		-o "-p $PRIMARY_PORT -k $SOCKDIR -c listen_addresses='' -c shared_preload_libraries='' -c wal_consistency_checking=''" \
		-w -t 120 start >>"$RUNLOG" 2>&1 || die "phase 1f: could not start the primary without the preload"
	verify_node psql_p "$PRIMARY_DATA"
	[ "$(psql_p -tAc "select count(*) from pg_get_wal_resource_managers() where rm_name = 'pg_lion'")" = 0 ] ||
		die "phase 1f: the resource manager is still registered"

	# The unlogged index, written without the manager: inserts that spill and
	# split its entries, and a VACUUM that removes TIDs from them.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1f: the unlogged rmgr-mode index could not be written without the manager"
		INSERT INTO lion_norm_u SELECT i, i % 500 FROM generate_series(20001, 40000) i;
		DELETE FROM lion_norm_u WHERE id % 3 = 0;
		VACUUM (INDEX_CLEANUP ON) lion_norm_u;
	SQL
	# The heap's own answer, read with every index plan and the count
	# pushdown off - otherwise LionCount answers it through the very index
	# under test.
	nu=$(PGOPTIONS="$HEAPONLY" psql_p -tAc "select count(*) from lion_norm_u where k = 4")
	run_check "phase 1f unlogged" psql_p \
		"select lion_index_verify('lion_norm_u_k', true) is not null, 'verify the unlogged index'
		 union all
		 select lion_index_count('lion_norm_u_k', 4) = $nu,
				'the unlogged index counts k = 4 as the heap does'"

	# The deleted rows are not in the partial index, so the VACUUM removes
	# nothing from it and only visits its pages; one visited range per
	# stand-alone record is the most records it could write.
	psql_p -c "SET synchronous_commit = on; DELETE FROM lion_norm WHERE id % 2 = 1" \
		>>"$RUNLOG" 2>&1 || die "phase 1f: delete failed"
	before=$(psql_p -tAc "select pg_current_wal_insert_lsn()")
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1f: the VACUUM failed"
		SET pg_lion.vacuum_barrier_ranges = 1;
		VACUUM (INDEX_CLEANUP ON) lion_norm;
	SQL
	if [ "$leak" = 1 ]; then
		psql_p -c "VACUUM (INDEX_CLEANUP ON) lion_norm_l" >>"$RUNLOG" 2>&1 ||
			die "phase 1f: the VACUUM of the index with leaks failed without the manager"
		[ "$(psql_p -c "SELECT lion_index_verify('lion_norm_l_k', true)" 2>&1 >>"$RUNLOG" |
			grep -c 'unused and unreachable' || true)" = "$leaked" ] ||
			die "phase 1f: the VACUUM without the manager changed the $leaked leaked pages of lion_norm_l"
		nu=$(PGOPTIONS="$HEAPONLY" psql_p -tAc "select count(*) from lion_norm_l where k = 4")
		run_check "phase 1f leaks left" psql_p \
			"select lion_index_count('lion_norm_l_k', 4) = $nu and $nu = 20000,
					'the index with leaks counts the 20000 rows of k = 4'
			 union all
			 select (select entries from lion_index_stats('lion_norm_l_k')) = $parked,
					'its entries are the $parked the crash left'"
	fi
	off=$(stat -c %s "$PRIMARY_LOG")
	psql_p >>"$RUNLOG" 2>&1 <<-SQL
		SET synchronous_commit = on;
		CREATE TABLE lion_norm_flush (i int);
	SQL
	after=$(psql_p -tAc "select pg_current_wal_flush_lsn()")
	custom=$("$PGBIN/pg_waldump" -p "$PRIMARY_DATA/pg_wal" -s "$before" -e "$after" 2>>"$RUNLOG" |
		grep -c 'rmgr: custom' || true)
	[ "$custom" = 0 ] ||
		die "phase 1f: the VACUUM wrote $custom record(s) of a resource manager this server does not have"

	# A logged rmgr-mode index, without the manager: an insert and a VACUUM
	# with dead rows to remove from it both fail, with a hint that names the
	# preload and REINDEX in generic mode; that REINDEX makes it writable.
	out=$(psql_p -c "INSERT INTO lion_norm_w VALUES (0, 1)" 2>&1) &&
		die "phase 1f: an insert into an rmgr-mode index succeeded without the manager"
	grep -q 'HINT:.*shared_preload_libraries.*REINDEX the index to rebuild it with wal_mode = generic' <<<"$out" ||
		die "phase 1f: the insert's error has no hint naming both fixes: $out"
	psql_p -c "DELETE FROM lion_norm_w WHERE id % 2 = 0" >>"$RUNLOG" 2>&1 ||
		die "phase 1f: the delete from lion_norm_w failed"
	out=$(psql_p -c "VACUUM (INDEX_CLEANUP ON) lion_norm_w" 2>&1) &&
		die "phase 1f: a VACUUM of an rmgr-mode index with dead rows succeeded without the manager"
	grep -q 'HINT:.*shared_preload_libraries.*REINDEX the index to rebuild it with wal_mode = generic' <<<"$out" ||
		die "phase 1f: the VACUUM's error has no hint naming both fixes: $out"
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1f: lion_norm_w was not writable after REINDEX in generic mode"
		ALTER INDEX lion_norm_w_k SET (wal_mode = generic);
		REINDEX INDEX lion_norm_w_k;
		INSERT INTO lion_norm_w VALUES (0, 1);
		VACUUM (INDEX_CLEANUP ON) lion_norm_w;
	SQL
	run_check "phase 1f reindexed" psql_p \
		"select lion_index_wal_mode('lion_norm_w_k') = 'generic', 'REINDEX rebuilt it in generic mode'
		 union all
		 select lion_index_count('lion_norm_w_k', 1) = 401, 'it counts the 401 rows of k = 1'"

	# And recovery goes through that range.
	crash_immediate
	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"
	tail -c "+$((off + 1))" "$PRIMARY_LOG" >>"$RUNLOG"
	grep -q "database system was not properly shut down" \
		<<<"$(tail -c "+$((off + 1))" "$PRIMARY_LOG")" ||
		die "phase 1f: the restart did not report an unclean shutdown"
	run_check "phase 1f recovered" psql_p \
		"select lion_index_verify('lion_norm_k', true) is not null, 'verify'
		 union all
		 select lion_index_count('lion_norm_k', 4) = 200, 'the index counts the 200 rows of k = 4'"

	# The primary goes back to what --mode says for the phases after this one.
	as_server "$PGBIN/pg_ctl" -D "$PRIMARY_DATA" stop -m fast -w >>"$RUNLOG" 2>&1 ||
		die "phase 1f: could not stop the primary"
	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"

	# With the manager back, the next VACUUM that walks the index frees the
	# leaks.  The VACUUM without the manager took the dead rows the crash
	# left in the heap, so this one is given 17 of its own to remove: without
	# them it would call no ambulkdelete, whose walk is what sweeps for them.
	if [ "$leak" = 1 ] && [ "$MODE" = rmgr ]; then
		psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1f: the VACUUM of lion_norm_l failed with the manager"
			SET synchronous_commit = on;
			DELETE FROM lion_norm_l WHERE k = 4 AND id <= 100;
			VACUUM (INDEX_CLEANUP ON) lion_norm_l;
		SQL
		[ "$(psql_p -c "SELECT lion_index_verify('lion_norm_l_k', true)" 2>&1 >>"$RUNLOG" |
			grep -c 'unused and unreachable' || true)" = 0 ] ||
			die "phase 1f: the VACUUM with the manager left pages of lion_norm_l leaked"
		nu=$(PGOPTIONS="$HEAPONLY" psql_p -tAc "select count(*) from lion_norm_l where k = 4")
		run_check "phase 1f leaks freed" psql_p \
			"select lion_index_count('lion_norm_l_k', 4) = $nu and $nu = 19983,
					'the index counts the 19983 rows of k = 4 left'"
	fi
	psql_p -c "DROP TABLE IF EXISTS lion_norm_l" >>"$RUNLOG" 2>&1
	psql_p -c "DROP TABLE lion_norm, lion_norm_flush, lion_norm_u, lion_norm_w" >>"$RUNLOG" 2>&1

	log "phase 1f: no record of an unregistered manager in $before..$after, and recovery went through it"
	SUMMARY+=("phase1f            VACUUM without the manager wrote nothing it could not replay")
	if [ "$leak" = 1 ]; then
		SUMMARY+=("phase1f            ... and left the $leaked pages a crash leaked for a VACUUM with it")
	fi
	SUMMARY+=("phase1f            an unlogged rmgr-mode index was written without the manager")
	SUMMARY+=("phase1f            a logged one refused insert and VACUUM with the hint, until REINDEXed generic")
}

# ------------------------------------------------------------- phase 1g

# A crash between a PROMOTION and the split that needed it (DESIGN.md §22, "A
# small set has no internal level").  A posting set of up to 16 leaves is a
# chain under its head, with no level above; the split that would make a 17th
# leaf first pushes the head down under a new root that links every leaf -
# one record - and then splits the leaf as a leaf of the tree, in records of
# its own.  The insert parks in between, on an append to the last leaf, and
# the server dies.  Then:
#
#  * the promotion came back: the head is an internal page over every leaf,
#    nothing is flagged, verify() is clean and both keys answer exactly;
#  * the next inserts split the full last leaf as a tree's, and post its
#    downlink in the root the promotion made;
#  * and a second crash replays all of it (under wal_consistency_checking
#    when --conf asks for it).
#
# The crash point is deterministic: an injection point parks the inserting
# backend after the promotion's record and the server is pulled out from
# under it.
phase1g() {
	local off parked warn ev
	log ""
	log "=== phase 1g: a crash between a posting set's promotion and its split ==="

	psql_p -c "CREATE EXTENSION IF NOT EXISTS injection_points" >>"$RUNLOG" 2>&1 ||
		{ no_injection_points "phase 1g" "phase1g            skipped (no injection points)"; return 0; }

	# Two keys on alternate wide rows, 65 to a heap page: 66000 rows are 16
	# container keys, which a build at posting_fillfactor 10 makes a leaf
	# each, so each key has 16 leaves and no internal level.  The insert below
	# fills the last leaf with containers, and its split is the 17th.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1g: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS lion_promo;
		CREATE TABLE lion_promo (id int, k int NOT NULL,
								 pad char(80) NOT NULL DEFAULT 'x');
		INSERT INTO lion_promo SELECT i, i % 2 FROM generate_series(1, 66000) i;
		CREATE INDEX lion_promo_k ON lion_promo USING lion (k)
			WITH (posting_fillfactor = 10);
	SQL
	[ "$(psql_p -tAc "select max_posting_height || '/' || posting_internal_pages from lion_index_stats('lion_promo_k')")" = "0/0" ] ||
		die "phase 1g: the fixture's posting sets are not without an internal level"

	psql_p -c "SELECT injection_points_attach('lion-posting-promoted', 'wait')" \
		>>"$RUNLOG" 2>&1 || die "phase 1g: could not attach the injection point"

	off=$(stat -c %s "$PRIMARY_LOG")
	"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres -d "$DBNAME" \
		-c "SET synchronous_commit = on;
		    INSERT INTO lion_promo SELECT 100000 + i, i % 2
		      FROM generate_series(1, 30000) i" >>"$RUNLOG" 2>&1 &
	parked=$!

	wait_true psql_p \
		"select count(*) > 0 from pg_stat_activity where wait_event = 'lion-posting-promoted'" \
		60 "an insert to park between the promotion and the split"

	# Flush the promotion's record with a COMMIT from another session, as
	# phase 1d does, and not with a CHECKPOINT: the parked backend holds the
	# leaf it is about to split EXCLUSIVE.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL
		SET synchronous_commit = on;
		CREATE TABLE IF NOT EXISTS lion_promo_flush (i int);
		INSERT INTO lion_promo_flush VALUES (1);
	SQL

	crash_immediate
	wait "$parked" 2>/dev/null || true

	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"
	ev=$(recovery_evidence "$off")
	GENERIC_TOTAL=$((GENERIC_TOTAL + ${ev#* }))

	# One set was promoted, and nothing is half done: the promotion is one
	# record, and the split it was for had not begun.
	warn=$(psql_p -c "SELECT lion_index_verify('lion_promo_k', true)" 2>&1 >>"$RUNLOG" |
		grep -c 'unfinished split' || true)
	[ "$warn" = 0 ] ||
		die "phase 1g: $warn posting page(s) came back with an unfinished split"
	run_check "phase 1g post-crash" psql_p \
		"select max_posting_height = 1 and posting_internal_pages = 1,
				format('one set came back promoted (height %s, %s internal pages)',
					   max_posting_height, posting_internal_pages)
		   from lion_index_stats('lion_promo_k')
		 union all
		 select true, 'verify() is clean after the crash'
		   from (select lion_index_verify('lion_promo_k', true)) v
		 union all
		 select (select count(*) from lion_promo where k = 0) =
				(select count(*) from lion_promo where id % 2 = 0),
				'k = 0 still answers exactly'
		 union all
		 select (select count(*) from lion_promo where k = 1) =
				(select count(*) from lion_promo where id % 2 = 1),
				'k = 1 still answers exactly'"

	psql_p -c "SELECT injection_points_detach('lion-posting-promoted')" \
		>>"$RUNLOG" 2>&1 || true

	# The next inserts split the full leaf as a leaf of the tree, and promote
	# the other key's set on the way.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1g: inserting after the crash failed"
		SET synchronous_commit = on;
		INSERT INTO lion_promo SELECT 200000 + i, i % 2
		  FROM generate_series(1, 20000) i;
		VACUUM lion_promo;
	SQL
	run_check "phase 1g inserted" psql_p \
		"select max_posting_height = 1 and posting_internal_pages = 2,
				format('both sets are trees (height %s, %s internal pages)',
					   max_posting_height, posting_internal_pages)
		   from lion_index_stats('lion_promo_k')
		 union all
		 select true, 'verify() is clean after the inserts'
		   from (select lion_index_verify('lion_promo_k', true)) v
		 union all
		 select (select count(*) from lion_promo where k = 0) = 43000,
				'k = 0 answers 43000 rows'
		 union all
		 select (select count(*) from lion_promo where k = 1) = 43000,
				'k = 1 answers 43000 rows'
		 union all
		 select ntids = (select count(*) from lion_promo), 'ntids agrees with the heap'
		   from lion_index_stats('lion_promo_k')"

	# Replay it all: the promotion, the split after it, the second promotion.
	off=$(stat -c %s "$PRIMARY_LOG")
	psql_p >>"$RUNLOG" 2>&1 <<-SQL
		SET synchronous_commit = on;
		INSERT INTO lion_promo_flush VALUES (2);
	SQL
	crash_immediate
	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"
	ev=$(recovery_evidence "$off")
	GENERIC_TOTAL=$((GENERIC_TOTAL + ${ev#* }))
	run_check "phase 1g replayed" psql_p \
		"select true, 'verify'
		   from (select lion_index_verify('lion_promo_k', true)) v
		 union all
		 select (select count(*) from lion_promo where k = 0) = 43000 and
				(select count(*) from lion_promo where k = 1) = 43000,
				'both keys answer 43000 rows after the replay'
		 union all
		 select ntids = (select count(*) from lion_promo), 'ntids agrees with the heap'
		   from lion_index_stats('lion_promo_k')"

	psql_p -c "DROP TABLE lion_promo, lion_promo_flush" >>"$RUNLOG" 2>&1

	log "phase 1g: the promotion survived the crash on its own, and the next inserts split the leaf it was for"
	SUMMARY+=("phase1g            crash between a promotion and its split; the next insert split the leaf")
}

# ---------------------------------------------------------------- phase 1h

# The index free space map is not WAL-logged (DESIGN.md §18, "The free space
# map holds only pages that may be handed out"): with data checksums a change
# to one of its pages is logged only as the full-page image of its first
# change after a checkpoint, so a crash brings back entries for pages that
# allocations had taken and that are in use again.  An allocation offered
# one must refuse it and NOT record it again, or the entry outlives every
# later VACUUM - which records only free pages and never clears an entry -
# and every allocation pays for it.
phase1h() {
	local off ev live
	log ""
	log "=== phase 1h: free space map entries a crash left on pages in use ==="

	if [ "$(psql_p -tAc "select count(*) from pg_available_extensions where name in ('pg_freespacemap', 'pageinspect')" 2>>"$RUNLOG" || true)" != 2 ]; then
		log "phase 1h skipped: this server has no pg_freespacemap or no pageinspect"
		SUMMARY+=("phase1h            skipped (no pg_freespacemap or no pageinspect)")
		return 0
	fi

	# lion_fsm_live(idx): how many blocks the map lists as free that hold a
	# page in use - neither all-zero, nor DELETED (flag 8), nor an UNLINKED
	# posting leaf (512); the flags are the uint16 at offset 26 of the
	# special area (lion_page.h).
	#
	# The fixture: 100000 unique keys, half of them deleted and vacuumed;
	# the second VACUUM, after an xid has been spent, records the pages the
	# first one freed.  The checkpoint after it is what makes the map's
	# next changes reach the WAL as nothing but one image per map page.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1h: fixture failed"
		SET synchronous_commit = on;
		CREATE EXTENSION IF NOT EXISTS pg_freespacemap;
		CREATE EXTENSION IF NOT EXISTS pageinspect;
		CREATE OR REPLACE FUNCTION lion_fsm_live(idx regclass) RETURNS bigint
		LANGUAGE sql AS \$\$
			SELECT count(*)
			  FROM pg_freespace(idx) f,
				   LATERAL get_raw_page(idx::text, f.blkno::int) r,
				   LATERAL page_header(r) h
			 WHERE f.avail > 0 AND f.blkno > 0 AND h.lower > 0
			   AND (get_byte(r, h.special + 26) +
					256 * get_byte(r, h.special + 27)) & (8 | 512) = 0
		\$\$;
		DROP TABLE IF EXISTS lion_fsm;
		CREATE TABLE lion_fsm (id int, k int) WITH (autovacuum_enabled = off);
		INSERT INTO lion_fsm SELECT i, i FROM generate_series(1, 100000) i;
		CREATE INDEX lion_fsm_k ON lion_fsm USING lion (k);
		DELETE FROM lion_fsm WHERE id BETWEEN 10001 AND 60000;
		VACUUM lion_fsm;
		SELECT txid_current();
		VACUUM lion_fsm;
		CHECKPOINT;
	SQL
	run_check "phase 1h before" psql_p \
		"select (select count(*) from pg_freespace('lion_fsm_k') where avail > 0) > 100,
				'the map holds the pages the delete freed'
		 union all
		 select lion_fsm_live('lion_fsm_k') = 0, 'and no page in use'"

	# The inserts take the pages the map lists; their commit flushes the
	# map pages' images, each with its first change only.
	off=$(stat -c %s "$PRIMARY_LOG")
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1h: insert failed"
		SET synchronous_commit = on;
		INSERT INTO lion_fsm SELECT i, i FROM generate_series(200001, 250000) i;
	SQL
	crash_immediate
	start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
	verify_node psql_p "$PRIMARY_DATA"
	ev=$(recovery_evidence "$off")
	GENERIC_TOTAL=$((GENERIC_TOTAL + ${ev#* }))

	# The precondition: without it nothing below tests anything.
	live=$(psql_p -tAc "select lion_fsm_live('lion_fsm_k')") ||
		die "phase 1h: could not read the free space map"
	[ "$live" -gt 0 ] ||
		die "phase 1h: the crash left no map entry on a page in use: nothing was tested"

	# Allocations are offered those pages, refuse each and drop its entry.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 1h: insert after the crash failed"
		SET synchronous_commit = on;
		INSERT INTO lion_fsm SELECT i, i FROM generate_series(300001, 330000) i;
	SQL
	run_check "phase 1h after" psql_p \
		"select lion_fsm_live('lion_fsm_k') = 0,
				format('the inserts dropped the %s entries on pages in use',
					   $live)
		 union all
		 select true, 'verify() is clean'
		   from (select lion_index_verify('lion_fsm_k', true)) v
		 union all
		 select (select count(*) from lion_fsm) = 130000 and
				(select count(*) from lion_fsm where k between 200001 and 330000) = 80000,
				'the rows answer exactly'
		 union all
		 select ntids = 130000, 'ntids agrees with the heap'
		   from lion_index_stats('lion_fsm_k')"

	psql_p -c "DROP TABLE lion_fsm; DROP FUNCTION lion_fsm_live(regclass)" >>"$RUNLOG" 2>&1

	log "phase 1h: $live map entries the crash left on pages in use were dropped by the next allocations"
	SUMMARY+=("phase1h            $live stale map entries after a crash; the next inserts dropped them")
}

# ---------------------------------------------------------------- phase 2

basebackup_standby() {
	log "-- pg_basebackup -R into $STANDBY_DATA"
	as_server "$PGBIN/pg_basebackup" -D "$STANDBY_DATA" -R -X stream -c fast --no-sync \
		-h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres >>"$RUNLOG" 2>&1 ||
		die "pg_basebackup failed (see $RUNLOG)"
	[ -f "$STANDBY_DATA/standby.signal" ] ||
		die "pg_basebackup -R did not write standby.signal"
	write_standby_conf on 2s
	echo "include 'standby_extra.conf'" >>"$STANDBY_DATA/postgresql.conf"
	start_node "$STANDBY_DATA" "$STANDBY_PORT" "$STANDBY_LOG"
	verify_node psql_s "$STANDBY_DATA"
	[ "$(psql_s -tAc 'select pg_is_in_recovery()')" = "t" ] ||
		die "the standby is not in recovery"
}

restart_standby() {
	as_server "$PGBIN/pg_ctl" -D "$STANDBY_DATA" stop -m fast -w -t 60 >>"$RUNLOG" 2>&1 ||
		die "could not stop the standby"
	start_node "$STANDBY_DATA" "$STANDBY_PORT" "$STANDBY_LOG"
	verify_node psql_s "$STANDBY_DATA"
}

# The standby has replayed everything the primary had written when it reaches
# a mark written here: a transactional logical message, committed with
# synchronous_commit = on, so that the commit flushes - and so sends - every
# record inserted before it, whoever inserted it; the standby replays the
# message as nothing.  The target is the LSN pg_logical_emit_message()
# returns, the end of the message, which is a record's end, as the replay
# position is, and which no record written later moves.
#
# Not pg_current_wal_insert_lsn(): that is where the next record would START,
# and once a record ends exactly at a page boundary it is past the next page's
# header, where no replay position ever is - replay reports the end of the
# last record, the boundary.  Only more WAL moves it, and an idle primary may
# write none: its last record is often the bgwriter's running-xacts record,
# which the bgwriter logs only after other WAL, so one that itself ends on a
# boundary leaves the insert position out of reach until the next checkpoint.
# catchup_mark() writes the mark and prints its LSN; wait_catchup() waits for
# the primary's view of replay to reach it.
catchup_mark() {
	psql_p -tAc "set synchronous_commit = on; select pg_logical_emit_message(true, 'lion_recovery', 'catchup')"
}

wait_catchup() {
	local target
	target=$(catchup_mark) || die "could not write the catch-up mark on the primary"
	wait_true psql_p \
		"select coalesce(bool_or(replay_lsn >= '$target'::pg_lsn), false) from pg_stat_replication" \
		120 "the standby to replay everything the primary has written (to $target)"
}

# rr_variant <hot_standby_feedback on|off> <k4 value>
#
# DESIGN.md section 9: a standby count must never trust the visibility map, so
# a REPEATABLE READ snapshot that counted key K must keep seeing that number
# even after the primary deletes those rows and vacuums them away.  With
# feedback on the primary cannot remove them and the count must be unchanged;
# with feedback off it can, and the only other legal outcome is that the
# session is cancelled with a recovery conflict.
rr_variant() {
	local feedback=$1 key=$2 before after n1 n2 rc=0 out err fresh seqcnt
	log ""
	log "-- recovery conflict: hot_standby_feedback = $feedback, k4 = $key"
	write_standby_conf "$feedback" 2s
	restart_standby
	wait_catchup

	before=$(psql_p -tAc "select count(*) from lion_rec where k4 = $key")
	[ "$before" -gt 0 ] || die "k4 = $key has no rows on the primary; pick another key"

	out=$BASE/rr_$feedback.out
	err=$BASE/rr_$feedback.err
	"$PGBIN/psql" -X -q -At -v ON_ERROR_STOP=1 -h "$SOCKDIR" -p "$STANDBY_PORT" \
		-U postgres -d "$DBNAME" -v key="$key" -v hold=6 \
		-f "$HERE/standby_rr.sql" >"$out" 2>"$err" &
	RRPID=$!

	# Do not touch the primary until that session holds its snapshot and has
	# already counted the key, or the test would prove nothing.
	wait_true psql_s \
		"select count(*) > 0 from pg_stat_activity where pid <> pg_backend_pid() and backend_xmin is not null and state = 'active' and query ilike 'select pg_sleep%'" \
		30 "the standby REPEATABLE READ session to hold its snapshot"
	if [ "$feedback" = on ]; then
		wait_true psql_p \
			"select coalesce(bool_or(backend_xmin is not null), false) from pg_stat_replication" \
			30 "the standby's xmin to reach the primary (hot_standby_feedback = on)"
	else
		wait_true psql_p \
			"select coalesce(bool_and(backend_xmin is null), true) from pg_stat_replication" \
			30 "the primary to forget the standby's xmin (hot_standby_feedback = off)"
	fi

	psql_p -c "DELETE FROM lion_rec WHERE k4 = $key" >>"$RUNLOG" 2>&1
	psql_p -c "VACUUM (INDEX_CLEANUP ON) lion_rec" >>"$RUNLOG" 2>&1

	# Neither outcome below means anything unless the primary's VACUUM did
	# what this variant is about: with feedback on the standby's xmin must
	# have held the deleted TIDs in the index, with feedback off they must be
	# gone - that removal is what the recovery conflict exists to protect the
	# standby's snapshot from.
	local ntids live
	ntids=$(psql_p -tAc "select ntids from lion_index_stats('lion_rec_k4')")
	live=$(psql_p -tAc "select count(*) from lion_rec")
	if [ "$feedback" = on ]; then
		[ "$ntids" -gt "$live" ] ||
			die "hot_standby_feedback = on did not hold the primary's horizon back: the index has $ntids TIDs for $live live rows, so VACUUM removed the deleted ones and the test below proves nothing"
		log "   primary VACUUM was held back: index still has $ntids TIDs for $live live rows"
	else
		[ "$ntids" = "$live" ] ||
			die "hot_standby_feedback = off but the primary's VACUUM did not remove the deleted TIDs ($ntids in the index, $live live rows), so the conflict path was not reachable"
		log "   primary VACUUM removed the deleted TIDs: $ntids = $live"
	fi

	wait "$RRPID" || rc=$?
	RRPID=""
	n1=$(sed -n 's/^n1 //p' "$out" | head -1)
	n2=$(sed -n 's/^n2 //p' "$out" | head -1)
	{ echo "---- standby RR session (feedback=$feedback, exit $rc)"; cat "$out" "$err"; } >>"$RUNLOG"

	if [ "$rc" = 0 ]; then
		{ [ -n "$n1" ] && [ -n "$n2" ]; } ||
			die "the standby session exited 0 but did not print both counts (see $RUNLOG)"
		[ "$n1" = "$before" ] ||
			die "the standby's first count of k4 = $key was $n1, the primary had $before"
		[ "$n1" = "$n2" ] ||
			die "BUG: a standby REPEATABLE READ snapshot counted k4 = $key as $n1 and then as $n2 after the primary deleted and vacuumed those rows (hot_standby_feedback = $feedback)"
		log "   snapshot preserved: $n1 -> $n2, no conflict"
		SUMMARY+=("phase2 conflict feedback=$feedback  snapshot preserved, $n1 = $n2")
	else
		grep -qi "conflict with recovery" "$err" ||
			die "the standby session failed with something other than a recovery conflict (exit $rc): $(tail -3 "$err")"
		[ "$feedback" = off ] ||
			die "BUG: hot_standby_feedback is on, yet the standby session hit a recovery conflict: $(tail -2 "$err")"
		{ [ -z "$n2" ] || [ "$n2" = "$n1" ]; } ||
			die "BUG: the standby printed a second, different count ($n1 then $n2) before the conflict"
		log "   cancelled by a recovery conflict after counting $n1 (legal outcome)"
		SUMMARY+=("phase2 conflict feedback=$feedback  cancelled by a recovery conflict after counting $n1")
	fi

	# A fresh standby session, once caught up, must see the new count.
	wait_catchup
	after=$(psql_p -tAc "select count(*) from lion_rec where k4 = $key")
	fresh=$(psql_s -tAc "select lion_index_count('lion_rec_k4'::regclass, $key::int4)")
	seqcnt=$(psql_s -tAc "set enable_seqscan = on; set enable_bitmapscan = off; set enable_indexscan = off; set pg_lion.enable_count_pushdown = off; select count(*) from lion_rec where k4 = $key")
	[ "$fresh" = "$after" ] ||
		die "BUG: after catching up, a fresh standby session counts k4 = $key as $fresh; the primary says $after"
	[ "$seqcnt" = "$after" ] ||
		die "BUG: the standby's sequential scan counts k4 = $key as $seqcnt; the primary says $after"
	log "   fresh standby session sees the new count: $fresh (primary $after)"
	SUMMARY+=("phase2 catchup feedback=$feedback  fresh standby count of k4=$key is $fresh, primary $after")
}

# Chain free and page reuse, replayed on a standby while a reader works
# through the very pages being recycled (DESIGN.md §18).
#
# On a standby, generic WAL cannot raise a recovery conflict, so replay may
# reuse a page under a reader's feet.  What keeps that reader honest is the
# owner in each container page's special area: a page that no longer claims
# the chain a reader came from ends that reader's walk.  Here the primary
# empties two keys (entries deleted, chains freed, pages recorded free) and
# then fills two new ones, which take those pages back; a standby reader runs
# against the index throughout, and afterwards the standby must answer exactly
# what the primary answers for every key.
standby_chain_reuse() {
	local loop flag errf=$BASE/standby_reuse.err k p_out s_out freed
	log ""
	log "-- standby: chain free and page reuse under a reader"

	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "standby_chain_reuse: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS lion_reuse;
		CREATE TABLE lion_reuse (id int, k int NOT NULL);
		CREATE INDEX lion_reuse_k ON lion_reuse USING lion (k)
			WITH (inline_limit = 64);
		INSERT INTO lion_reuse SELECT i, i % 4 FROM generate_series(1, 40000) i;
	SQL
	psql_p -c "VACUUM (ANALYZE) lion_reuse" >>"$RUNLOG" 2>&1
	wait_catchup

	# A standby reader that keeps asking the index questions for as long as
	# the primary is freeing and recycling pages underneath it.  One short
	# query every 200 ms, not a tight loop: it has to coexist with replay,
	# not starve it.  The flag file is what stops it, so that nothing is left
	# spinning if a step below fails.
	flag=$BASE/standby_reuse.run
	: >"$flag"
	: >"$errf"
	(
		while [ -f "$flag" ]; do
			"$PGBIN/psql" -X -q -v ON_ERROR_STOP=1 -h "$SOCKDIR" \
				-p "$STANDBY_PORT" -U postgres -d "$DBNAME" -tAc \
				"SET enable_seqscan = off;
				 SELECT count(*) FROM lion_reuse WHERE k = 2;
				 SELECT count(*) FROM lion_reuse WHERE k = 3;
				 SELECT lion_index_count('lion_reuse_k'::regclass, 2::int4)" \
				>/dev/null 2>>"$errf" || true
			command sleep 0.2
		done
	) &
	loop=$!

	psql_p -c "SET synchronous_commit = on; DELETE FROM lion_reuse WHERE k IN (0, 1)" \
		>>"$RUNLOG" 2>&1 || { rm -f "$flag"; die "standby_chain_reuse: delete failed"; }
	# hot_standby_feedback is on here, so the primary's removal horizon is the
	# standby's oldest snapshot as the standby last REPORTED it, and that
	# report is only sent every wal_receiver_status_interval.  So the VACUUM
	# may find nothing removable for a few seconds after the delete; retry it
	# until the chains really are freed.
	wait_catchup
	freed=0
	for i in $(seq 1 40); do
		psql_p -c "VACUUM (INDEX_CLEANUP ON) lion_reuse" >>"$RUNLOG" 2>&1
		freed=$(psql_p -tAc "select deleted_pages from lion_index_stats('lion_reuse_k')")
		[ "${freed:-0}" -gt 0 ] && break
		nap 0.5
	done
	psql_p -c "SET synchronous_commit = on;
			   INSERT INTO lion_reuse SELECT i, 4 + (i % 2)
				 FROM generate_series(100001, 120000) i" >>"$RUNLOG" 2>&1 ||
		{ rm -f "$flag"; die "standby_chain_reuse: the reusing insert failed"; }
	psql_p -c "VACUUM (ANALYZE) lion_reuse" >>"$RUNLOG" 2>&1
	wait_catchup

	rm -f "$flag"
	wait "$loop" 2>/dev/null || true

	[ "${freed:-0}" -gt 0 ] ||
		die "standby_chain_reuse: the primary freed no pages (deleted_pages = $freed)"
	if grep -q . "$errf" 2>/dev/null; then
		head -10 "$errf" >&2
		die "standby_chain_reuse: the standby reader hit errors (see $errf)"
	fi

	# Every key, primary against standby, through the index on both sides.
	for k in 0 1 2 3 4 5; do
		p_out=$(psql_p -tAc "set enable_seqscan=off;
							 select count(*) from lion_reuse where k = $k")
		s_out=$(psql_s -tAc "set enable_seqscan=off;
							 select count(*) from lion_reuse where k = $k")
		[ "$p_out" = "$s_out" ] ||
			die "standby_chain_reuse: k = $k is $p_out on the primary and $s_out on the standby"
	done

	run_check "standby reuse" psql_s \
		"select (select count(*) from lion_reuse where k = 0) = 0,
				'freed key 0 answers no rows on the standby'
		 union all
		 select (select count(*) from lion_reuse where k = 4) = 10000,
				'recycled pages answer key 4 correctly on the standby'"

	psql_p -c "DROP TABLE lion_reuse" >>"$RUNLOG" 2>&1
	wait_catchup

	log "standby: $freed freed pages were recycled and replayed with a reader running"
	SUMMARY+=("phase2 standby      chain free + reuse replayed under a reader ($freed pages)")
}

# Index-only scans on the standby (DESIGN.md §9 "Hot standby", §29.9).  An
# index-only scan trusts the visibility map for every TID it returns, which a
# standby may only do where replay waits for the pin the scan keeps: rmgr
# mode.  In generic mode liongettuple() looks every TID up in the heap first,
# or it could return rows replay has made dead.  Either way
# the standby's index-only scan must answer what its sequential scan answers,
# before and after the primary deletes rows and vacuums them away.  lion plans
# an index-only scan only for a query that needs no column, which a partial
# index whose predicate is the WHERE clause gives.
standby_index_only() {
	local plan ios seq want round
	local ioset="set enable_seqscan = off; set enable_bitmapscan = off; set pg_lion.enable_count_pushdown = off;"
	local seqset="set enable_indexscan = off; set enable_indexonlyscan = off; set enable_bitmapscan = off; set pg_lion.enable_count_pushdown = off;"
	log ""
	log "-- standby: index-only scans"

	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "standby_index_only: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS lion_ios;
		CREATE TABLE lion_ios (id int NOT NULL, k int NOT NULL, flag bool NOT NULL);
		INSERT INTO lion_ios SELECT i, i % 50, i % 2 = 0 FROM generate_series(1, 40000) i;
		CREATE INDEX lion_ios_k ON lion_ios USING lion (k) WHERE flag;
	SQL
	psql_p -c "VACUUM (ANALYZE) lion_ios" >>"$RUNLOG" 2>&1
	wait_catchup

	plan=$(psql_s -tAc "$ioset explain (costs off) select count(*) from lion_ios where flag")
	grep -q "Index Only Scan using lion_ios_k" <<<"$plan" ||
		die "standby_index_only: the standby plans no index-only scan of lion_ios_k: $plan"

	for round in before after; do
		if [ "$round" = after ]; then
			psql_p -c "SET synchronous_commit = on; DELETE FROM lion_ios WHERE id % 4 = 0" \
				>>"$RUNLOG" 2>&1 || die "standby_index_only: delete failed"
			psql_p -c "VACUUM (INDEX_CLEANUP ON) lion_ios" >>"$RUNLOG" 2>&1
			wait_catchup
		fi
		want=$(psql_p -tAc "select count(*) from lion_ios where flag")
		ios=$(psql_s -tAc "$ioset select count(*) from lion_ios where flag")
		seq=$(psql_s -tAc "$seqset select count(*) from lion_ios where flag")
		{ [ "$ios" = "$seq" ] && [ "$seq" = "$want" ]; } ||
			die "BUG: standby_index_only ($round the delete): the standby's index-only scan counts $ios, its sequential scan $seq, the primary $want"
	done

	psql_p -c "DROP TABLE lion_ios" >>"$RUNLOG" 2>&1
	wait_catchup

	log "standby: index-only scans answer $ios rows, as the sequential scan does, before and after a VACUUM"
	SUMMARY+=("phase2 standby      index-only scans agree with the heap ($MODE mode)")
}

# An index BUILD over existing rows, replayed on the standby.  Everything else
# a standby replays here is written by aminsert and VACUUM, and phase 1 crashes
# only after the checkpoint that follows its builds, so this is where the
# build's own WAL - the bulk writer's page images, which on 16 lion_build.c's
# copy of 17's bulk-write API writes - is read back: two indexes built on the
# primary, one of a few dense keys and one of many text keys, each verified on
# the standby and counted there against the primary's heap.
standby_build_replay() {
	local nk nt i
	log ""
	log "-- standby: index builds replayed"

	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "standby_build_replay: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS lion_built;
		CREATE TABLE lion_built (id int NOT NULL, k int NOT NULL, t text NOT NULL);
		INSERT INTO lion_built
		SELECT i, i % 7, 'key-' || (i % 5000) FROM generate_series(1, 300000) i;
		CREATE INDEX lion_built_k ON lion_built USING lion (k);
		CREATE INDEX lion_built_t ON lion_built USING lion (t) WITH (inline_limit = 64);
	SQL
	nk=$(PGOPTIONS="$HEAPONLY" psql_p -tAc "select count(*) from lion_built where k = 3")
	nt=$(PGOPTIONS="$HEAPONLY" psql_p -tAc "select count(*) from lion_built where t = 'key-17'")
	wait_catchup

	run_check "standby build replay" psql_s "
		select lion_index_verify('lion_built_k'::regclass, true) is not null, 'verify lion_built_k'
		union all
		select lion_index_verify('lion_built_t'::regclass, true) is not null, 'verify lion_built_t'
		union all
		select lion_index_count('lion_built_k'::regclass, 3) = $nk, 'k = 3 counted through the replayed build'
		union all
		select lion_index_count('lion_built_t'::regclass, 'key-17'::text) = $nt, 't = key-17 counted through the replayed build'"

	# DESIGN.md §33: the key counts on the meta page, as the build wrote them
	# and as an ANALYZE and a VACUUM of the primary write them again, each in
	# a META record: the standby reads what the primary does.  The ANALYZE's
	# are exact - 1000 new keys of t and one of k, every row live, and a
	# directory well inside what ANALYZE may walk.  The VACUUM's are counted
	# by ambulkdelete's own walk, so the VACUUM that counts is one that
	# removes the rows: with hot_standby_feedback on, the primary's horizon is
	# the standby's last report, and the VACUUM is retried until it has
	# removed them (standby_chain_reuse does the same) - until the primary
	# counts t's and k's keys back at 5000 and 7.
	standby_ndistinct_check "standby key counts after the build" 5000 7
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "standby_build_replay: ANALYZE failed"
		SET synchronous_commit = on;
		INSERT INTO lion_built SELECT i, 7, 'new-' || i FROM generate_series(1, 1000) i;
		ANALYZE lion_built;
	SQL
	standby_ndistinct_check "standby key counts after ANALYZE" 6000 8
	psql_p -c "SET synchronous_commit = on; DELETE FROM lion_built WHERE t LIKE 'new-%'" \
		>>"$RUNLOG" 2>&1 || die "standby_build_replay: DELETE failed"
	wait_catchup
	for i in $(seq 1 40); do
		psql_p -c "VACUUM (INDEX_CLEANUP ON) lion_built" >>"$RUNLOG" 2>&1 ||
			die "standby_build_replay: VACUUM failed"
		[ "$(psql_p -tAc "select ndistinct from lion_index_stats('lion_built_t')")" = 5000 ] &&
			break
		nap 0.5
	done
	standby_ndistinct_check "standby key counts after VACUUM" 5000 7

	psql_p -c "DROP TABLE lion_built" >>"$RUNLOG" 2>&1
	wait_catchup

	log "standby: two index builds replayed, verified and counted ($nk and $nt rows), and their key counts read as the primary wrote them"
	SUMMARY+=("phase2 standby      two index builds replayed, verified and counted")
	SUMMARY+=("phase2 standby      key counts of a build, an ANALYZE and a VACUUM replayed (§33)")
}

# A position tree damaged in the standby's copy of an index (DESIGN.md §7,
# "Position trees"): a standby compares a position tree with its posting set
# only from a walk that read every page as it was at one replay position, so
# a quiet standby must report damage the primary's check would.  The index is
# verify.sql's seventeenth case - 600 rows of 'alpha:5 beta:6', keys that are
# CHAIN entries with a position tree of one leaf, one chunk of three bytes a
# row after a 10-byte header - built on the primary and replayed, checked on
# the standby while it is sound, and then, with the standby stopped, the first
# member's delta (at 10 of 'alpha''s chunk) goes from 1 to 2 in the standby's
# file: every member moves one offset on, and the first row of each of the
# five heap blocks has no positions.  The checkpoint on the primary and the
# restartpoint on the standby put the replayed build on disk and replay's
# starting point after it, so nothing replays the leaf back over the damage.
# The page's checksum is wrong after that, which ignore_checksum_failure lets
# through.
standby_position_damage() {
	local pos rel out rc=0
	log ""
	log "-- standby: a damaged position tree"

	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "standby_position_damage: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS lion_posdmg;
		CREATE TABLE lion_posdmg (id int NOT NULL, d tsvector NOT NULL)
		  WITH (autovacuum_enabled = off);
		INSERT INTO lion_posdmg SELECT i, 'alpha:5 beta:6' FROM generate_series(1, 600) i;
		CREATE INDEX lion_posdmg_d ON lion_posdmg USING lion (d)
		  WITH (store_positions = true, inline_limit = 64);
		CHECKPOINT;
	SQL
	wait_catchup
	run_check "standby position trees, sound" psql_s "
		select lion_index_verify('lion_posdmg_d'::regclass, true) is not null,
			   'verify lion_posdmg_d before the damage'"

	# Where the member is, read from the primary's file: the standby's pages
	# are the build's, replayed.  The leaf is the position page (flag 128,
	# the low byte of the flags at 26 of the special area, the last 32 bytes)
	# at level 0 (24) whose first chunk starts with 'alpha''s first member.
	rel=$(psql_p -tAc "select pg_relation_filepath('lion_posdmg_d')")
	pos=$(psql_p -tAc "
		select b * c.bs + l.lp + 10
		  from (select current_setting('block_size')::int8 as bs,
					   pg_relation_filepath('lion_posdmg_d') as path) c,
			   generate_series(1, pg_relation_size('lion_posdmg_d') /
								  current_setting('block_size')::int8 - 1) b,
			   lateral (select get_byte(x, 0) + (get_byte(x, 1) & 127) * 256 as lp
						  from pg_read_binary_file(c.path, b * c.bs + 24, 4) x) l
		 where get_byte(pg_read_binary_file(c.path, b * c.bs + c.bs - 32 + 26, 1), 0) & 128 <> 0
		   and pg_read_binary_file(c.path, b * c.bs + c.bs - 32 + 24, 2) = '\x0000'
		   and pg_read_binary_file(c.path, b * c.bs + l.lp + 10, 3) = '\x010114'")
	[ -n "$rel" ] && [ -n "$pos" ] && [ "$(wc -l <<<"$pos")" = 1 ] ||
		die "standby_position_damage: no single position leaf of 'alpha' in $rel (found: $pos)"

	psql_s -c "CHECKPOINT" >>"$RUNLOG" 2>&1 ||
		die "standby_position_damage: the restartpoint failed"
	as_server "$PGBIN/pg_ctl" -D "$STANDBY_DATA" stop -m fast -w -t 60 >>"$RUNLOG" 2>&1 ||
		die "could not stop the standby"
	printf '\002' | as_server dd of="$STANDBY_DATA/$rel" bs=1 seek="$pos" conv=notrunc \
		2>>"$RUNLOG" || die "standby_position_damage: could not damage $STANDBY_DATA/$rel"
	start_node "$STANDBY_DATA" "$STANDBY_PORT" "$STANDBY_LOG"
	verify_node psql_s "$STANDBY_DATA"

	out=$(psql_s -tAc "
		set ignore_checksum_failure = on;
		set client_min_messages = error;
		select lion_index_verify('lion_posdmg_d'::regclass)" 2>&1) || rc=$?
	echo "---- standby_position_damage: $out" >>"$RUNLOG"
	[ "$rc" != 0 ] ||
		die "BUG: standby_position_damage: the standby's check passed a position tree without positions for five of its TIDs"
	grep -q 'lion index "lion_posdmg_d": 5 TIDs of chain entry [0-9]* on block [0-9]* have no positions, the first (0,1)' <<<"$out" ||
		die "standby_position_damage: the standby's check failed with something else: $out"

	psql_p -c "DROP TABLE lion_posdmg" >>"$RUNLOG" 2>&1
	wait_catchup

	log "standby: a position tree damaged in the standby's copy reported there"
	SUMMARY+=("phase2 standby      a damaged position tree reported on the standby ($MODE mode)")
}

# standby_ndistinct_check <label> <t's count> <k's count>
#
# The ndistinct of lion_built_t and lion_built_k (lion_index_stats(), DESIGN.md
# §33) on the standby, once it has replayed everything, against the primary's
# - and against the counts given, when they are.
standby_ndistinct_check() {
	local label=$1 want_t=$2 want_k=$3 nd_t nd_k
	nd_t=$(psql_p -tAc "select ndistinct from lion_index_stats('lion_built_t')")
	nd_k=$(psql_p -tAc "select ndistinct from lion_index_stats('lion_built_k')")
	if [ -z "$nd_t" ] || [ -z "$nd_k" ]; then
		die "$label: the primary records no key count"
	fi
	if [ -n "$want_t" ] && { [ "$nd_t" != "$want_t" ] || [ "$nd_k" != "$want_k" ]; }; then
		die "$label: the primary counted $nd_t and $nd_k keys, not $want_t and $want_k"
	fi
	wait_catchup
	run_check "$label" psql_s "
		select ndistinct = $nd_t, 'lion_built_t: $nd_t keys, as the primary counted them'
		  from lion_index_stats('lion_built_t')
		union all
		select ndistinct = $nd_k, 'lion_built_k: $nd_k keys, as the primary counted them'
		  from lion_index_stats('lion_built_k')"
}

phase2() {
	local probe_p probe_s ck
	log ""
	log "=== phase 2: hot standby ==="

	psql_p -c "VACUUM (ANALYZE, INDEX_CLEANUP ON) lion_rec" >>"$RUNLOG" 2>&1
	psql_p -c "CHECKPOINT" >>"$RUNLOG" 2>&1
	run_check "primary pre-basebackup" psql_p "select * from lion_rec_check(true, true)"
	ck=$NCHECKS
	run_check "primary count_stats" psql_p "select * from lion_rec_check_count_stats(false)"
	log "primary: $ck checks ok, and it does skip heap blocks via the visibility map"

	probe_p=$BASE/probe_primary.txt
	probe_s=$BASE/probe_standby.txt
	psql_p -tAc "select * from lion_rec_probe()" >"$probe_p"
	[ -s "$probe_p" ] || die "the primary probe produced nothing"

	basebackup_standby
	wait_catchup

	run_check "standby" psql_s "select * from lion_rec_check(true, false)"
	log "standby: $NCHECKS checks ok (verify, counts, GROUP BY, ntids)"
	SUMMARY+=("phase2 standby      $NCHECKS checks ok in recovery")

	if [ "$MODE" = rmgr ]; then
		# DESIGN.md §25: replay of an rmgr-mode removal takes the cleanup
		# lock the §9 interlock needs, so the standby may use the visibility
		# map again - and this is where that is proved to HAPPEN, not merely
		# to be allowed.
		run_check "standby count_stats (rmgr: VM trusted)" psql_s \
			"select * from lion_rec_check_count_stats(true, true)"
		log "standby: counts skip heap blocks via the visibility map (rmgr mode)"
		SUMMARY+=("phase2 standby      counts skip blocks via the VM (rmgr mode, §25)")
	else
		run_check "standby count_stats" psql_s "select * from lion_rec_check_count_stats(true)"
		log "standby: all $NCHECKS counts skipped 0 heap blocks and rechecked every TID"
		SUMMARY+=("phase2 standby      $NCHECKS counts skipped 0 blocks, rechecked every TID")
	fi

	psql_s -tAc "select * from lion_rec_probe()" >"$probe_s"
	if ! diff -u "$probe_p" "$probe_s" >/dev/null 2>&1; then
		diff -u "$probe_p" "$probe_s" | tee -a "$RUNLOG" | head -40 >&2
		die "BUG: the standby does not answer what the primary answers (see $RUNLOG)"
	fi
	log "standby: all $(wc -l <"$probe_s") index probes identical to the primary"
	SUMMARY+=("phase2 standby      $(wc -l <"$probe_s") index probes identical to the primary")

	standby_chain_reuse
	standby_index_only
	standby_build_replay
	standby_position_damage

	rr_variant on 11
	rr_variant off 7

	log ""
	log "-- promoting the standby"
	as_server "$PGBIN/pg_ctl" -D "$STANDBY_DATA" promote -w -t 120 >>"$RUNLOG" 2>&1 ||
		die "pg_ctl promote failed"
	wait_true psql_s "select not pg_is_in_recovery()" 120 "the standby to finish promotion"
	run_check "promoted standby" psql_s "select * from lion_rec_check(true, false)"
	ck=$NCHECKS
	psql_s -c "VACUUM (INDEX_CLEANUP ON) lion_rec" >>"$RUNLOG" 2>&1
	run_check "promoted standby post-vacuum" psql_s "select * from lion_rec_check(false, true)"
	run_check "promoted standby count_stats" psql_s "select * from lion_rec_check_count_stats(false)"
	log "promoted standby: $ck checks ok, and it trusts the visibility map again"
	SUMMARY+=("phase2 promoted     $ck checks ok after promotion, VM path back in use")
}

# ---------------------------------------------------------------- phase 3

# The standby half of the §9/§11 interlock, attacked where it is thinnest
# (DESIGN.md §9 "Hot standby", §11 "On a standby", §25 "Standby").
#
# A standby reader copies containers out of an index page, keeps the PIN, and
# only then asks the visibility map about their heap blocks - exactly as on a
# primary.  In rmgr mode it trusts the map, on the argument that replay of
# every removal takes a CLEANUP lock.  Each case below parks such a reader at
# the injection point 'lion-count-containers-pinned' (containers copied, page
# pinned, map not yet consulted) and then has the PRIMARY move the TIDs the
# reader copied onto a page it is not pinning, and VACUUM them there:
#
#   split     an insert splits the leaf the reader pins; the containers with
#             the dead TIDs move to the new right sibling, and the VACUUM
#             removes them there.  The pinned leaf itself has nothing left to
#             remove, so the primary's VACUUM cleanup-locks it but writes no
#             record for it.
#   pushdown  the reader pins the HEAD of a posting set of 16 leaves with no
#             internal level, its first leaf; an insert needs a 17th leaf and
#             pushes the head down (§22), so the containers move to a new
#             child and the head becomes an internal page.
#   spill     the reader pins a directory leaf holding an INLINE posting set;
#             the VACUUM's own filtering makes the payload outgrow the entry
#             and spill onto container pages, which rewrites the entry on the
#             pinned leaf.
#
# In every case the primary's VACUUM ends by marking the deleted rows' heap
# pages all-visible, and replay of those heap records follows replay of the
# index records.  So unless replay of the index side WAITS for the reader's
# pin, the reader - woken once the standby has caught up - counts the dead
# rows from its copy.  The legal outcomes are: replay blocks on the pin (the
# startup process shows up waiting for a cleanup lock - the barrier of §25,
# carried by a stand-alone VACUUM_VISIT record in the split case and riding on
# the removal record in the other two) and the reader, released
# then, answers the right number; or the standby rechecks every TID (generic
# mode) and answers the right number without blocking anything.  A standby
# that catches up while the reader is parked, and a reader that then answers
# anything else, is the bug.
#
# A fourth case, setaba, parks a standby's lion_index_verify() between two
# leaves of a posting set instead (standby_verify_aba()).

# PHASE3_CASES (environment, a development aid) runs a subset of the cases.
want_case() { case " ${PHASE3_CASES:-split pushdown spill setaba $PHASE3_VERIFY_CASES} " in *" $1 "*) return 0 ;; esac; return 1; }
PHASE3_VERIFY_CASES="vunlink vfree vreach vdownlink vleftlink vsummary vsplitdir vsplitpost"

# standby_pin_case <label> <index> <key> <expected> <sql to run on the primary>
standby_pin_case() {
	local label=$1 idx=$2 key=$3 expected=$4 sql=$5 out err rc=0 n i state="" rpid plsn
	out=$BASE/pin_$label.out
	err=$BASE/pin_$label.err

	"$PGBIN/psql" -X -q -At -v ON_ERROR_STOP=1 -h "$SOCKDIR" -p "$STANDBY_PORT" \
		-U postgres -d "$DBNAME" >"$out" 2>"$err" <<-SQL &
		SET pg_lion.enable_count_pushdown = off;
		SELECT injection_points_set_local();
		SELECT injection_points_attach('lion-count-containers-pinned', 'wait');
		SELECT 'count ' || lion_index_count('$idx'::regclass, $key::int4);
	SQL
	rpid=$!
	RRPID=$rpid
	wait_true psql_s \
		"select count(*) > 0 from pg_stat_activity where wait_event = 'lion-count-containers-pinned'" \
		30 "$label: the standby reader to park with its page pinned"

	# The trailing xid makes a commit record, whose synchronous flush takes the
	# VACUUM's own records (which commit nothing) to the standby with it.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "$label: the primary's writes failed"
		SET synchronous_commit = on;
		$sql
		SELECT pg_current_xact_id();
	SQL

	# Either replay blocks on the reader's pin, or it catches up, to the mark
	# (catchup_mark()).  Both are asked of the standby itself: the primary's
	# view of replay_lsn is only as fresh as the last status message.
	plsn=$(catchup_mark) || die "$label: could not write the catch-up mark on the primary"
	for i in $(seq 1 300); do
		if [ "$(psql_s -tAc "select count(*) > 0 from pg_stat_activity where backend_type = 'startup' and wait_event in ('BufferPin', 'BufferCleanup')")" = t ]; then
			state=blocked
			break
		fi
		if [ "$(psql_s -tAc "select pg_last_wal_replay_lsn() >= '$plsn'::pg_lsn")" = t ]; then
			state=caughtup
			break
		fi
		nap 0.1
	done
	if [ -z "$state" ]; then
		psql_s -c "select backend_type, state, wait_event_type, wait_event, left(query, 60) from pg_stat_activity" >&2 || true
		die "$label: replay neither blocked nor caught up within 30s"
	fi

	psql_s -tAc "select injection_points_detach('lion-count-containers-pinned')" >/dev/null 2>&1 || true
	for i in $(seq 1 100); do
		n=$(psql_s -tAc "select count(*) from pg_stat_activity where wait_event = 'lion-count-containers-pinned'")
		[ "$n" = 0 ] && break
		psql_s -tAc "select injection_points_wakeup('lion-count-containers-pinned')" >/dev/null 2>&1 || true
		nap 0.1
	done
	wait "$rpid" || rc=$?
	RRPID=""
	{ echo "---- standby pin case $label (exit $rc, replay $state)"; cat "$out" "$err"; } >>"$RUNLOG"
	[ "$rc" = 0 ] || die "$label: the standby reader failed: $(tail -3 "$err")"
	n=$(sed -n 's/^count //p' "$out" | head -1)

	# A failure is recorded and the other cases still run, so that one run
	# says which of the three shapes is open.
	wait_catchup
	{ echo "---- $label: count_stats and ntids after catching up, standby then primary";
	  psql_s -tAc "select *, (select ntids from lion_index_stats('$idx')) from lion_index_count_stats('$idx'::regclass, $key::int4)";
	  psql_p -tAc "select *, (select ntids from lion_index_stats('$idx')) from lion_index_count_stats('$idx'::regclass, $key::int4)"; } >>"$RUNLOG" 2>&1
	if [ "$n" != "$expected" ]; then
		log "   BUG ($label): a standby count parked with its page pinned answered $n, the right answer for its snapshot is $expected (replay $state while it was parked, mode $MODE)"
		PHASE3_FAILS+=("$label")
	elif [ "$MODE" = rmgr ] && [ "$state" != blocked ]; then
		log "   BUG ($label): replay caught up past the primary's VACUUM while a standby reader held a pin on the page its containers came from (the answer happened to be right: $n)"
		PHASE3_FAILS+=("$label")
	else
		log "   $label: parked reader answered $n (expected $expected), replay $state while it was parked"
		SUMMARY+=("phase3 $label  standby reader $n = $expected, replay $state")
	fi
}

# A standby's check of a posting set that replay changes under it, leaving
# its entry as it was (DESIGN.md §7, "Posting sets on a standby").  Replay
# applies VACUUM, which the primary's check locks out: a TID a VACUUM takes
# out of the set's first leaf and another an INSERT puts into its last leave
# the entry's header - ntids, containers, head and tail - as it was, and a
# walk that read the first leaf before the one and the last after the other
# counts a TID more than the entry says.  Agreeing with the entry is not
# what such a walk must be judged by: it is kept for what it checked of the
# pages, and its totals are counted again, a pass that reads the set's
# leaves alone, and the pass that read every page as it was at one replay
# position settles them (lion_verify_set_count()).
#
# k = 1 is every other row of 2020 wide ones, heap blocks 0..134: one CHAIN
# set of three containers, which posting_fillfactor 10 spreads over two
# leaves; the other rows have a key each, INLINE.  Row 1, in heap block 0
# and the first leaf, is deleted before the standby's check
# takes its snapshot - so that removing it is no recovery conflict - and not
# vacuumed.  The check parks after the set's first leaf
# ('lion-verify-set-leaf-read'); the primary inserts a row of k = 1, which
# the free space map puts on the last heap page and the set into its last
# container, on the second leaf, and a VACUUM takes row 1 out; the standby
# replays both, and the check goes on.  It must pass, the totals of its one
# walk settled by a counting pass.
standby_verify_aba() {
	local out=$BASE/setaba.out err=$BASE/setaba.err rpid rc=0 before after plsn i n state=""
	log ""
	log "-- standby: a posting set changed under a check, its entry as it was"

	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "standby_verify_aba: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS vfy_aba;
		CREATE TABLE vfy_aba (id int, k int NOT NULL, pad char(480) NOT NULL DEFAULT '')
		  WITH (autovacuum_enabled = off);
		INSERT INTO vfy_aba (id, k)
		SELECT i, CASE WHEN i % 2 = 1 THEN 1 ELSE 2 + i END FROM generate_series(1, 2020) i;
		CREATE INDEX vfy_aba_k ON vfy_aba USING lion (k)
		  WITH (inline_limit = 64, posting_fillfactor = 10);
		VACUUM (FREEZE) vfy_aba;
		DELETE FROM vfy_aba WHERE id = 1;
	SQL
	[ "$(psql_p -tAc "select container_pages from lion_index_stats('vfy_aba_k')")" = 2 ] ||
		die "standby_verify_aba: the set does not have two leaves"
	before=$(psql_p -tAc "select ntids || ' TIDs, ' || containers || ' containers, ' || container_pages || ' leaves' from lion_index_stats('vfy_aba_k')")
	wait_catchup

	"$PGBIN/psql" -X -q -At -v ON_ERROR_STOP=1 -h "$SOCKDIR" -p "$STANDBY_PORT" \
		-U postgres -d "$DBNAME" >"$out" 2>"$err" <<-SQL &
		SELECT injection_points_set_local();
		SELECT injection_points_attach('lion-verify-set-leaf-read', 'wait');
		SET client_min_messages = debug1;
		SELECT 'verified ' || (lion_index_verify('vfy_aba_k'::regclass) IS NOT NULL);
	SQL
	rpid=$!
	RRPID=$rpid
	wait_true psql_s \
		"select count(*) > 0 from pg_stat_activity where wait_event = 'lion-verify-set-leaf-read'" \
		30 "standby_verify_aba: the standby's check to park after the set's first leaf"

	# The VACUUM is retried until it has removed row 1: with
	# hot_standby_feedback on, the primary's horizon is the standby's last
	# report, which can predate the delete (standby_build_replay does the
	# same).
	psql_p -c "SET synchronous_commit = on; INSERT INTO vfy_aba (id, k) VALUES (100000, 1)" \
		>>"$RUNLOG" 2>&1 || die "standby_verify_aba: the insert failed"
	for i in $(seq 1 60); do
		psql_p -c "VACUUM (INDEX_CLEANUP ON) vfy_aba" >>"$RUNLOG" 2>&1 ||
			die "standby_verify_aba: VACUUM failed"
		after=$(psql_p -tAc "select ntids || ' TIDs, ' || containers || ' containers, ' || container_pages || ' leaves' from lion_index_stats('vfy_aba_k')")
		[ "$after" = "$before" ] && break
		nap 0.5
	done
	[ "$after" = "$before" ] ||
		die "standby_verify_aba: the set's entry did not come back as it was ($before, then $after)"
	psql_p -c "SET synchronous_commit = on; SELECT pg_current_xact_id()" >>"$RUNLOG" 2>&1

	plsn=$(catchup_mark) || die "standby_verify_aba: could not write the catch-up mark on the primary"
	for i in $(seq 1 300); do
		if [ "$(psql_s -tAc "select pg_last_wal_replay_lsn() >= '$plsn'::pg_lsn")" = t ]; then
			state=caughtup
			break
		fi
		nap 0.1
	done
	if [ -z "$state" ]; then
		psql_s -c "select backend_type, state, wait_event_type, wait_event, left(query, 60) from pg_stat_activity" >&2 || true
		die "standby_verify_aba: replay did not catch up within 30s while the check was parked"
	fi

	psql_s -tAc "select injection_points_detach('lion-verify-set-leaf-read')" >/dev/null 2>&1 || true
	for i in $(seq 1 100); do
		n=$(psql_s -tAc "select count(*) from pg_stat_activity where wait_event = 'lion-verify-set-leaf-read'")
		[ "$n" = 0 ] && break
		psql_s -tAc "select injection_points_wakeup('lion-verify-set-leaf-read')" >/dev/null 2>&1 || true
		nap 0.1
	done
	wait "$rpid" || rc=$?
	RRPID=""
	{ echo "---- standby_verify_aba (exit $rc)"; cat "$out" "$err"; } >>"$RUNLOG"
	[ "$rc" = 0 ] && grep -q '^verified true$' "$out" ||
		die "BUG: standby_verify_aba: the standby's check reported damage replay left it: $(grep -v '^DEBUG' "$err" | tail -3)"
	grep -q '1 posting-set walks, 0 thrown away, 0 with the leaf held, 1 counting passes, 0 kept uncompared' "$err" ||
		die "standby_verify_aba: the check did not count the set again: $(grep 'posting-set walks' "$err")"

	psql_p -c "DROP TABLE vfy_aba" >>"$RUNLOG" 2>&1
	wait_catchup
	log "   setaba: a set replay changed under the check, its entry as it was ($after), counted again and passed"
	SUMMARY+=("phase3 setaba  a standby's check counted a set again that replay changed, its entry unchanged")
}

# ---- a standby's check beside the states replay passes through
#
# The rest of phase 3 sets up, one at a time and at an injection point, the
# states a standby's lion_index_verify() meets and a primary's never does
# (DESIGN.md §7, "verify() on a standby"; R-S1 to R-S5 of the 2026-10-23
# review).  A primary's check locks VACUUM out and waits for the writers
# beside it; a standby's can do neither: replay applies VACUUM's records
# beside it, and every operation the primary writes in more than one record
# - a split, a row's key and its summary - shows the states between its
# records to the standby's readers for as long as the primary takes.  Every
# case has the check pass, and none has it warn of an unfinished split that
# is not abandoned; the last has an abandoned one, which it must warn of.
#
#   vunlink    the check parks after a posting set's first leaf, and a VACUUM
#              takes the emptied second leaf out of the set (UNLINKED) under
#              it: the walk reads it so, past the replay position it began
#              at, and walks again (lion_verify_tree_page()).
#   vfree      ... and a VACUUM that removed the key's every row frees the
#              set and its entry: the check of the set ends there
#              (lion_verify_refind()).
#   vreach     the check parks before its reachability pass, and a split of
#              a directory leaf takes a page out of the free space map that
#              no walk reached: its LSN is past the position the check began
#              at, and it is not reported (lion_verify_reachable()).
#   vdownlink  the check parks after the directory's leaves, and the split's
#              downlink, on a parent past that position, names a leaf the
#              walk did not reach (lion_verify_filter_downlinks()).
#   vleftlink  the check parks on the first leaf, and the split of that leaf
#              changes the left link of the page after it, which the walk
#              reads past the position (lion_verify_walk_level()).
#   vsummary   an INSERT parks on the primary between its row's key and its
#              summary (lion-insert-before-summary): its transaction is in
#              progress, and its row is no candidate (lion_verify_settle()).
#   vsplitdir  a directory leaf split parks on the primary between its first
#              record and its downlink: the flag the check sees is a split in
#              progress, and it is not warned of; the check parks before it
#              looks at the flagged page again, the split finishes, and it is
#              not warned of either (lion_verify_flags_recheck()); then a
#              split that fails there (an error) is warned of.
#   vsplitpost the same for a split of a posting leaf, without the re-look.
#
# Each table is the case's own and is dropped after it.

# sv_sync: the standby has replayed everything the primary wrote.  The commit
# is what flushes - and so sends - WAL that a writer parked in the middle of
# its transaction has only inserted.
sv_sync() {
	psql_p -c "SET synchronous_commit = on; SELECT pg_current_xact_id()" >>"$RUNLOG" 2>&1 ||
		die "sv_sync: the primary's commit failed"
	wait_catchup
}

# sv_park <label> <point> <index>: a standby lion_index_verify() of <index>
# in the background, parked at <point>; it prints "verified true" in
# $BASE/sv_<label>.out when it returns, and its messages go to .err.
sv_park() {
	local label=$1 point=$2 idx=$3
	SV_LABEL=$label
	SV_OUT=$BASE/sv_$label.out
	SV_ERR=$BASE/sv_$label.err
	"$PGBIN/psql" -X -q -At -v ON_ERROR_STOP=1 -h "$SOCKDIR" -p "$STANDBY_PORT" \
		-U postgres -d "$DBNAME" >"$SV_OUT" 2>"$SV_ERR" <<-SQL &
		SELECT injection_points_set_local();
		SELECT injection_points_attach('$point', 'wait');
		SET client_min_messages = debug1;
		SELECT 'verified ' || (lion_index_verify('$idx'::regclass) IS NOT NULL);
	SQL
	SV_PID=$!
	RRPID=$SV_PID
	wait_true psql_s \
		"select count(*) > 0 from pg_stat_activity where wait_event = '$point'" \
		30 "$label: the standby's check to park at $point"
}

# sv_release <point>: let the parked check go for good, and wait for it.
sv_release() {
	local point=$1 i n
	SV_RC=0
	psql_s -tAc "select injection_points_detach('$point')" >/dev/null 2>&1 || true
	for i in $(seq 1 100); do
		n=$(psql_s -tAc "select count(*) from pg_stat_activity where wait_event = '$point'")
		[ "$n" = 0 ] && break
		psql_s -tAc "select injection_points_wakeup('$point')" >/dev/null 2>&1 || true
		nap 0.1
	done
	wait "$SV_PID" || SV_RC=$?
	RRPID=""
}

# sv_now <label> <index> [heapallindexed]: a standby lion_index_verify() of
# <index>, run to its end, into the same files.
sv_now() {
	local label=$1 idx=$2 hai=${3:-false}
	SV_LABEL=$label
	SV_OUT=$BASE/sv_$label.out
	SV_ERR=$BASE/sv_$label.err
	SV_RC=0
	psql_s -At -c "SET client_min_messages = debug1" \
		-c "SELECT 'verified ' || (lion_index_verify('$idx'::regclass, $hai) IS NOT NULL)" \
		>"$SV_OUT" 2>"$SV_ERR" || SV_RC=$?
}

# sv_judge <warn>: the last check passed, and with <warn> "none" gave no
# WARNING, with "split" warned of an unfinished split.
sv_judge() {
	local warn=$1
	{ echo "---- standby case $SV_LABEL (exit $SV_RC)"; cat "$SV_OUT" "$SV_ERR"; } >>"$RUNLOG"
	[ "$SV_RC" = 0 ] && grep -q '^verified true$' "$SV_OUT" ||
		die "BUG: $SV_LABEL: the standby's check reported damage that replay left it: $(grep -v '^DEBUG' "$SV_ERR" | tail -3)"
	case $warn in
	none)
		! grep -q '^WARNING' "$SV_ERR" ||
			die "BUG: $SV_LABEL: the standby's check warned of a state replay passes through: $(grep '^WARNING' "$SV_ERR" | head -2)" ;;
	split)
		grep -q '^WARNING: .* has an unfinished split$' "$SV_ERR" ||
			die "$SV_LABEL: the standby's check did not warn of an abandoned split: $(grep -v '^DEBUG' "$SV_ERR" | tail -3)" ;;
	esac
}

# pw_park <label> <point> <sql>: <sql> on the primary in the background, its
# writer parked at <point>.
pw_park() {
	local label=$1 point=$2 sql=$3
	PW_POINT=$point
	"$PGBIN/psql" -X -q -At -v ON_ERROR_STOP=1 -h "$SOCKDIR" -p "$PRIMARY_PORT" \
		-U postgres -d "$DBNAME" >"$BASE/pw_$label.out" 2>&1 <<-SQL &
		SELECT injection_points_set_local();
		SELECT injection_points_attach('$point', 'wait');
		$sql
	SQL
	PW_PID=$!
	wait_true psql_p \
		"select count(*) > 0 from pg_stat_activity where wait_event = '$point'" \
		120 "$label: the primary's writer to park at $point"
}

# pw_release: let the parked writer go for good, and wait for it to commit.
pw_release() {
	local i n rc=0
	psql_p -tAc "select injection_points_detach('$PW_POINT')" >/dev/null 2>&1 || true
	for i in $(seq 1 100); do
		n=$(psql_p -tAc "select count(*) from pg_stat_activity where wait_event = '$PW_POINT'")
		[ "$n" = 0 ] && break
		psql_p -tAc "select injection_points_wakeup('$PW_POINT')" >/dev/null 2>&1 || true
		nap 0.1
	done
	wait "$PW_PID" || rc=$?
	[ "$rc" = 0 ] || die "the primary's parked writer failed: $(tail -3 "$BASE"/pw_*.out)"
}

# sv_quiet <label> <index>: both nodes, caught up, verify <index>.
sv_quiet() {
	local label=$1 idx=$2
	sv_sync
	run_check "phase 3 $label quiesced" psql_s "
		select lion_index_verify('$idx'::regclass, true) is not null, 'standby verify $idx'"
	run_check "phase 3 $label quiesced" psql_p "
		select lion_index_verify('$idx'::regclass, true) is not null, 'primary verify $idx'"
}

# standby_verify_vacuum <unlink|free>: R-S1.  k = 1 is every other row of
# 6000 wide ones, a CHAIN set whose six leaves (posting_fillfactor 10) are
# heap blocks 0..63, 64..127 and so on; the rest have a key each.  The rows
# of the second leaf - or every row of k = 1 - are deleted before the
# check takes its snapshot, and vacuumed after it has read the first leaf.
standby_verify_vacuum() {
	local variant=$1 label=v$1 i n want
	log ""
	log "-- standby: VACUUM replayed under a check's walk of a posting set ($variant)"
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "$label: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS vfy_vac;
		CREATE TABLE vfy_vac (id int, k int NOT NULL, pad char(480) NOT NULL DEFAULT '')
		  WITH (autovacuum_enabled = off);
		INSERT INTO vfy_vac (id, k)
		SELECT i, CASE WHEN i % 2 = 1 THEN 1 ELSE 2 + i END FROM generate_series(1, 6000) i;
		CREATE INDEX vfy_vac_k ON vfy_vac USING lion (k)
		  WITH (inline_limit = 64, posting_fillfactor = 10);
		VACUUM (FREEZE) vfy_vac;
	SQL
	[ "$(psql_p -tAc "select container_pages || ' ' || max_posting_height from lion_index_stats('vfy_vac_k')")" = "6 0" ] ||
		die "$label: the set is not six leaves without an internal level"
	if [ "$variant" = free ]; then
		psql_p -c "SET synchronous_commit = on; DELETE FROM vfy_vac WHERE k = 1" >>"$RUNLOG" 2>&1
		want="0 "
	else
		psql_p -c "SET synchronous_commit = on; DELETE FROM vfy_vac WHERE k = 1 AND (ctid::text::point)[0] BETWEEN 64 AND 127" >>"$RUNLOG" 2>&1
		want="5 1 "
	fi
	sv_sync
	sv_park "$label" lion-verify-set-leaf-read vfy_vac_k

	# Retried: with hot_standby_feedback on, the primary's horizon is the
	# standby's last report, which can predate the delete.
	for i in $(seq 1 60); do
		psql_p -c "VACUUM (INDEX_CLEANUP ON) vfy_vac" >>"$RUNLOG" 2>&1 || die "$label: VACUUM failed"
		n=$(psql_p -tAc "select container_pages || ' ' || deleted_pages || ' ' from lion_index_stats('vfy_vac_k')")
		case "$n" in "$want"*) break ;; esac
		nap 0.5
	done
	case "$n" in "$want"*) ;; *) die "$label: VACUUM did not take the set apart (pages, deleted: $n)" ;; esac
	sv_sync
	sv_release lion-verify-set-leaf-read
	sv_judge none
	sv_quiet "$label" vfy_vac_k
	psql_p -c "DROP TABLE vfy_vac" >>"$RUNLOG" 2>&1
	log "   $label: a check whose set VACUUM's replay took apart under it passed ($(grep -o '[0-9]* posting-set walks, [0-9]* thrown away' "$SV_ERR"))"
	SUMMARY+=("phase3 $label  a standby's check passed beside the replay of a VACUUM that took its set apart")
}

# standby_verify_dirsplit <reach|downlink|leftlink>: R-S2.  The table of
# standby_verify_vacuum(), whose 3000 keys fill a few directory leaves under
# a root; new keys among the first leaf's split it.  For vreach the emptied
# second leaf of k = 1 is taken out of its set and handed to the free space
# map first, so that the split's new page is a block the index had when the
# check began, which the reachability pass looks at.
standby_verify_dirsplit() {
	local variant=$1 label=v$1 point i n
	log ""
	log "-- standby: a directory leaf split replayed under a check ($variant)"
	if [ "$variant" = reach ] &&
		[ "$(psql_p -tAc "select count(*) from pg_available_extensions where name = 'pg_freespacemap'" 2>>"$RUNLOG" || true)" != 1 ]; then
		log "   $label skipped: this server has no pg_freespacemap"
		SUMMARY+=("phase3 $label  skipped (no pg_freespacemap)")
		return 0
	fi
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "$label: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS vfy_dir;
		CREATE TABLE vfy_dir (id int, k int NOT NULL, pad char(480) NOT NULL DEFAULT '')
		  WITH (autovacuum_enabled = off);
		INSERT INTO vfy_dir (id, k)
		SELECT i, CASE WHEN i % 2 = 1 THEN 1 ELSE 2 + i END FROM generate_series(1, 6000) i;
		CREATE INDEX vfy_dir_k ON vfy_dir USING lion (k)
		  WITH (inline_limit = 64, posting_fillfactor = 10);
		VACUUM (FREEZE) vfy_dir;
	SQL
	[ "$(psql_p -tAc "select directory_height from lion_index_stats('vfy_dir_k')")" = 1 ] ||
		die "$label: the directory is not leaves under a root"
	if [ "$variant" = reach ]; then
		psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "$label: fixture failed"
			CREATE EXTENSION IF NOT EXISTS pg_freespacemap;
			SET synchronous_commit = on;
			DELETE FROM vfy_dir WHERE k = 1 AND (ctid::text::point)[0] BETWEEN 64 AND 127;
		SQL
		# The leaf goes once the delete is behind the standby's horizon, and
		# into the map once its safexid is, which a later VACUUM sees.
		for i in $(seq 1 60); do
			nap 1.2
			psql_p -c "VACUUM (INDEX_CLEANUP ON) vfy_dir" >>"$RUNLOG" 2>&1 || die "$label: VACUUM failed"
			[ "$(psql_p -tAc "select deleted_pages from lion_index_stats('vfy_dir_k')")" = 1 ] && break
		done
		[ "$(psql_p -tAc "select deleted_pages from lion_index_stats('vfy_dir_k')")" = 1 ] ||
			die "$label: VACUUM did not take the emptied leaf out of its set"
		for i in $(seq 1 60); do
			psql_p -c "SET synchronous_commit = on; DELETE FROM vfy_dir WHERE id = (SELECT min(id) FROM vfy_dir WHERE k > 1)" >>"$RUNLOG" 2>&1
			nap 1.2
			psql_p -c "VACUUM (INDEX_CLEANUP ON) vfy_dir" >>"$RUNLOG" 2>&1 || die "$label: VACUUM failed"
			[ "$(psql_p -tAc "select count(*) from pg_freespace('vfy_dir_k') where avail > 0")" = 1 ] && break
		done
		[ "$(psql_p -tAc "select count(*) from pg_freespace('vfy_dir_k') where avail > 0")" = 1 ] ||
			die "$label: the free leaf did not reach the free space map"
	fi
	sv_sync
	case $variant in
	reach) point=lion-verify-reachable ;;
	downlink) point=lion-verify-dir-level-walked ;;
	leftlink) point=lion-verify-dir-page-read ;;
	esac
	sv_park "$label" "$point" vfy_dir_k
	n=$(psql_p -tAc "select leaf_pages from lion_index_stats('vfy_dir_k')")
	psql_p -c "SET synchronous_commit = on; INSERT INTO vfy_dir (id, k) SELECT 200000 + i, 2 * i + 3 FROM generate_series(1, 300) i" \
		>>"$RUNLOG" 2>&1 || die "$label: the inserts failed"
	[ "$(psql_p -tAc "select leaf_pages from lion_index_stats('vfy_dir_k')")" -gt "$n" ] ||
		die "$label: the inserts split no directory leaf"
	[ "$variant" != reach ] ||
		[ "$(psql_p -tAc "select count(*) from pg_freespace('vfy_dir_k') where avail > 0")" = 0 ] ||
		die "$label: the split did not take the free page"
	sv_sync
	sv_release "$point"
	sv_judge none
	sv_quiet "$label" vfy_dir_k
	psql_p -c "DROP TABLE vfy_dir" >>"$RUNLOG" 2>&1
	log "   $label: a check that a directory split was replayed under at $point passed"
	SUMMARY+=("phase3 $label  a standby's check passed beside the replay of a directory split ($point)")
}

# standby_verify_summary: R-S3.  Keys 1..4000, a row each, in summaries of
# 1000 TIDs; a new row of key 2500 parks between its key and its summary.
standby_verify_summary() {
	local label=vsummary
	log ""
	log "-- standby: a check beside an INSERT between its key and its summary"
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "$label: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS vfy_sum;
		CREATE TABLE vfy_sum (id int, k int NOT NULL) WITH (autovacuum_enabled = off);
		INSERT INTO vfy_sum SELECT i, i FROM generate_series(1, 4000) i;
		CREATE INDEX vfy_sum_k ON vfy_sum USING lion (k) WITH (summaries = on, summary_tids = 1000);
		VACUUM (FREEZE) vfy_sum;
	SQL
	pw_park "$label" lion-insert-before-summary "INSERT INTO vfy_sum VALUES (99999, 2500);"
	sv_sync
	sv_now "$label" vfy_sum_k
	sv_judge none
	sv_now "${label}_hai" vfy_sum_k true
	sv_judge none
	pw_release
	sv_quiet "$label" vfy_sum_k
	psql_p -c "DROP TABLE vfy_sum" >>"$RUNLOG" 2>&1
	log "   $label: a check beside an INSERT between its key and its summary passed"
	SUMMARY+=("phase3 $label  a standby's check passed beside an INSERT between its key and its summary")
}

# standby_verify_split <dir|post>: R-S5.  vsplitdir: the table of
# standby_verify_vacuum(), whose first directory leaf the new keys split;
# vsplitpost: 90000 rows of two keys at posting_fillfactor 10, each key's
# set a tree (verify_incomplete_split.spec), whose last leaves new rows of
# the two keys, taking alternate TIDs, fill and split.
standby_verify_split() {
	local variant=$1 label=vsplit$1 point idx ins ins2
	log ""
	log "-- standby: a check beside a $variant split the primary is in the middle of"
	if [ "$variant" = dir ]; then
		psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "$label: fixture failed"
			SET synchronous_commit = on;
			DROP TABLE IF EXISTS vfy_split;
			CREATE TABLE vfy_split (id int, k int NOT NULL, pad char(480) NOT NULL DEFAULT '')
			  WITH (autovacuum_enabled = off);
			INSERT INTO vfy_split (id, k)
			SELECT i, CASE WHEN i % 2 = 1 THEN 1 ELSE 2 + i END FROM generate_series(1, 6000) i;
			CREATE INDEX vfy_split_k ON vfy_split USING lion (k)
			  WITH (inline_limit = 64, posting_fillfactor = 10);
			VACUUM (FREEZE) vfy_split;
		SQL
		point=lion-dir-split-incomplete
		ins="INSERT INTO vfy_split (id, k) SELECT 200000 + i, 2 * i + 3 FROM generate_series(1, 300) i;"
		ins2="INSERT INTO vfy_split (id, k) SELECT 300000 + i, 1000000 + i FROM generate_series(1, 600) i;"
	else
		psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "$label: fixture failed"
			SET synchronous_commit = on;
			DROP TABLE IF EXISTS vfy_split;
			CREATE TABLE vfy_split (id int, k int NOT NULL, pad char(80) NOT NULL DEFAULT 'x')
			  WITH (autovacuum_enabled = off);
			INSERT INTO vfy_split SELECT i, i % 2 FROM generate_series(1, 90000) i;
			CREATE INDEX vfy_split_k ON vfy_split USING lion (k) WITH (posting_fillfactor = 10);
			VACUUM (FREEZE) vfy_split;
		SQL
		[ "$(psql_p -tAc "select max_posting_height from lion_index_stats('vfy_split_k')")" -ge 1 ] ||
			die "$label: the posting sets are not trees"
		point=lion-posting-split-incomplete
		ins="INSERT INTO vfy_split SELECT 100000 + i, i % 2 FROM generate_series(1, 60000) i;"
		ins2="INSERT INTO vfy_split SELECT 200000 + i, i % 2 FROM generate_series(1, 60000) i;"
	fi
	idx=vfy_split_k

	# The split in progress, seen by the whole check.
	pw_park "$label" "$point" "$ins"
	sv_sync
	if [ "$variant" = dir ]; then
		sv_now "$label" "$idx"
		sv_judge none
		# ... and finished while the check is about to look at it again.
		sv_park "${label}_done" lion-verify-flags-recheck "$idx"
		pw_release
		sv_sync
		sv_release lion-verify-flags-recheck
		sv_judge none
	else
		sv_now "$label" "$idx"
		sv_judge none
		pw_release
	fi

	# A split that fails between its records is abandoned, and warned of: new
	# keys past the last for the directory, which split its last leaf.
	psql_p -At >"$BASE/pw_${label}_err.out" 2>&1 <<-SQL || true
		SELECT injection_points_set_local();
		SELECT injection_points_attach('$point', 'error');
		$ins2
	SQL
	grep -q "error triggered for injection point $point" "$BASE/pw_${label}_err.out" ||
		die "$label: the failing split did not fail at $point: $(tail -3 "$BASE/pw_${label}_err.out")"
	sv_sync
	sv_now "${label}_abandoned" "$idx"
	sv_judge split
	sv_quiet "$label" "$idx"
	psql_p -c "DROP TABLE vfy_split" >>"$RUNLOG" 2>&1
	log "   $label: a split in progress was not warned of, an abandoned one was"
	SUMMARY+=("phase3 $label  a standby's check warned of an abandoned $variant split, not of one in progress")
}

phase3() {
	local exp pages
	log ""
	log "=== phase 3: standby barriers for pinned readers (§9/§11/§25) ==="
	# Not `psql | grep -q`: under pipefail a psql that grep -q has stopped
	# reading from can fail the pipeline, which would read as "no injection
	# points" (recovery_evidence() has the story).
	[ "$(psql_p -tAc "select count(*) from pg_available_extensions where name = 'injection_points'" 2>>"$RUNLOG" || true)" = 1 ] ||
		{ no_injection_points "phase 3" "phase3             skipped (no injection points)"; return 0; }

	# A standby of its own: phase 2 promotes the one it made.
	stop_hard "$STANDBY_DATA"
	rm -rf "$STANDBY_DATA"
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 3: fixture failed"
		SET synchronous_commit = on;
		CREATE EXTENSION IF NOT EXISTS injection_points;

		-- split: some thirty ARRAY containers on a full leaf
		-- (count_split_race.spec)
		DROP TABLE IF EXISTS pin_split;
		-- 15 rows to a heap page, two of them k = 1's: 128 members of k = 1
		-- in a container, a 264-byte ARRAY, which the split case's inserts
		-- grow into a NARROW of width 1 (DESIGN.md section 38), 520 bytes.
		-- The build packs the leaves full (posting_fillfactor 100): k = 1's
		-- first leaf has 96 bytes to spare.
		CREATE TABLE pin_split (id int, k int NOT NULL, pad char(480));
		INSERT INTO pin_split SELECT i, CASE WHEN i % 15 IN (1, 8) THEN 1 ELSE 0 END, ''
		  FROM generate_series(1, 38400) i;
		CREATE INDEX pin_split_k ON pin_split USING lion (k) WITH (posting_fillfactor = 100);

		-- pushdown: count_root_pushdown_race.spec's fixture.  k = 1's first
		-- container on its head, and then wide rows, every other one k = 1,
		-- 16 container keys in all, which a build at posting_fillfactor 10
		-- makes a leaf each: 16 leaves and no internal level
		DROP TABLE IF EXISTS pin_push;
		CREATE TABLE pin_push (id int, k int NOT NULL, pad char(80) NOT NULL DEFAULT 'x');
		INSERT INTO pin_push SELECT i, i % 4 FROM generate_series(1, 4000) i;
		INSERT INTO pin_push SELECT 10000 + i, CASE WHEN i % 2 = 0 THEN 1 ELSE 5 END
		  FROM generate_series(1, 60000) i;
		CREATE INDEX pin_push_k ON pin_push USING lion (k)
			WITH (inline_limit = 64, posting_fillfactor = 10);

		-- spill: k = 1 is 200 consecutive offsets of heap block 0, one RUN
		-- container of a few bytes, INLINE; every other one deleted, it is a
		-- 100-member ARRAY that no longer fits inline_limit
		DROP TABLE IF EXISTS pin_spill;
		CREATE TABLE pin_spill (id int, k int NOT NULL);
		INSERT INTO pin_spill SELECT i, CASE WHEN i <= 200 THEN 1 ELSE 2 + i % 50 END
		  FROM generate_series(1, 3000) i;
		CREATE INDEX pin_spill_k ON pin_spill USING lion (k) WITH (inline_limit = 64);
	SQL
	psql_p -c "VACUUM (FREEZE, ANALYZE) pin_split" >>"$RUNLOG" 2>&1
	psql_p -c "VACUUM (FREEZE, ANALYZE) pin_push" >>"$RUNLOG" 2>&1
	psql_p -c "VACUUM (FREEZE, ANALYZE) pin_spill" >>"$RUNLOG" 2>&1
	# Free line pointers in heap blocks 0..63 only, so that the inserts of the
	# split case land there and grow the FIRST container of the first leaf.
	psql_p -c "SET synchronous_commit = on; DELETE FROM pin_split WHERE k = 0 AND (ctid::text::point)[0] < 64" >>"$RUNLOG" 2>&1
	# INDEX_CLEANUP ON: with fewer than 2% of the pages affected VACUUM would
	# otherwise bypass index vacuuming, leave the line pointers dead and never
	# tell the free space map about those pages.
	psql_p -c "VACUUM (INDEX_CLEANUP ON) pin_split" >>"$RUNLOG" 2>&1

	[ "$(psql_p -tAc "select max_posting_height from lion_index_stats('pin_push_k')")" = 0 ] ||
		die "phase 3: the pushdown fixture's posting set has an internal level"
	[ "$(psql_p -tAc "select inline_entries from lion_index_stats('pin_spill_k')")" = 1 ] ||
		die "phase 3: the spill fixture does not have exactly one INLINE entry"

	# Rows dead to everyone - deleted before the standby reader takes its
	# snapshot - and not vacuumed yet.
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 3: deletes failed"
		SET synchronous_commit = on;
		-- containers 1 and 2 of the first leaf (heap blocks 70..190)
		DELETE FROM pin_split WHERE k = 1 AND (ctid::text::point)[0] BETWEEN 70 AND 190;
		DELETE FROM pin_push WHERE k = 1 AND id <= 2000;
		DELETE FROM pin_spill WHERE k = 1 AND id % 2 = 0;
	SQL

	basebackup_standby
	write_standby_conf on -1
	restart_standby
	wait_catchup

	if want_case split; then
		exp=$(psql_p -tAc "select count(*) from pin_split where k = 1")
		pages=$(psql_p -tAc "select container_pages from lion_index_stats('pin_split_k')")
		# One range per stand-alone VACUUM_VISIT record, so that the barrier
		# this case depends on travels in that record type and not riding on
		# a removal record, which the other two cases cover.
		standby_pin_case split pin_split_k 1 "$exp" "
			INSERT INTO pin_split SELECT 1000000 + i, 1, '' FROM generate_series(1, 700) i;
			SET pg_lion.vacuum_barrier_ranges = 1;
			VACUUM (INDEX_CLEANUP ON) pin_split;
			RESET pg_lion.vacuum_barrier_ranges;"
		[ "$(psql_p -tAc "select container_pages from lion_index_stats('pin_split_k')")" -gt "$pages" ] ||
			die "phase 3: the split case did not split a leaf"
	fi

	if want_case pushdown; then
		exp=$(psql_p -tAc "select count(*) from pin_push where k = 1")
		standby_pin_case pushdown pin_push_k 1 "$exp" "
			INSERT INTO pin_push SELECT 100000 + i, CASE WHEN i % 2 = 0 THEN 1 ELSE 5 END
			  FROM generate_series(1, 40000) i;
			VACUUM (INDEX_CLEANUP ON) pin_push;"
		[ "$(psql_p -tAc "select max_posting_height from lion_index_stats('pin_push_k')")" -ge 1 ] ||
			die "phase 3: the pushdown case did not push the head down"
	fi

	if want_case spill; then
		exp=$(psql_p -tAc "select count(*) from pin_spill where k = 1")
		standby_pin_case spill pin_spill_k 1 "$exp" "
			VACUUM (INDEX_CLEANUP ON) pin_spill;"
		[ "$(psql_p -tAc "select inline_entries from lion_index_stats('pin_spill_k')")" = 0 ] ||
			die "phase 3: the spill case did not spill"
	fi

	want_case setaba && standby_verify_aba
	want_case vunlink && standby_verify_vacuum unlink
	want_case vfree && standby_verify_vacuum free
	want_case vreach && standby_verify_dirsplit reach
	want_case vdownlink && standby_verify_dirsplit downlink
	want_case vleftlink && standby_verify_dirsplit leftlink
	want_case vsummary && standby_verify_summary
	want_case vsplitdir && standby_verify_split dir
	want_case vsplitpost && standby_verify_split post

	run_check "phase 3 standby" psql_s "
		select lion_index_verify(i::regclass, true) is not null, 'verify ' || i
		  from unnest(array['pin_split_k', 'pin_push_k', 'pin_spill_k']) i
		union all
		select (select count(*) from pin_split where k = 1) = $(psql_p -tAc "select count(*) from pin_split where k = 1"), 'pin_split count'
		union all
		select (select count(*) from pin_push where k = 1) = $(psql_p -tAc "select count(*) from pin_push where k = 1"), 'pin_push count'
		union all
		select (select count(*) from pin_spill where k = 1) = $(psql_p -tAc "select count(*) from pin_spill where k = 1"), 'pin_spill count'"

	stop_hard "$STANDBY_DATA"
	psql_p -c "DROP TABLE pin_split, pin_push, pin_spill" >>"$RUNLOG" 2>&1
	[ "${#PHASE3_FAILS[@]}" = 0 ] ||
		die "phase 3: a standby reader was overtaken by replay in: ${PHASE3_FAILS[*]}"
}

# ---------------------------------------------------------------- phase 4

# lion_index_verify() on a hot standby while replay writes the posting sets it
# walks (DESIGN.md §7, §11, §25).  Redo takes the blocks of a record in the
# order the writer registered them and holds every one to the end of the
# record, and every record that changes a posting set registers the set's
# pages BEFORE the entry's directory leaf - an insert's, a split's, a
# push-down's, VACUUM's - so replay locks container page, then leaf: the
# reverse of the primary's order.  A verify() that walked a set with that
# leaf held SHARE, or whose heapallindexed lookups descended a set with it
# held, would lock leaf, then container page, and the startup process and the
# backend would wait for each other on buffer content locks, which have no
# deadlock detector and ignore cancel and terminate: replay would stop for
# good, `pg_ctl stop -m fast` would hang, and only an immediate shutdown would
# get the standby back.  So on a standby verify() walks a set with nothing
# held, and its lookups let the entry's leaf go before they descend the set
# (DESIGN.md §7).
#
# So: inserts into three hot keys on the primary, and verify() in two loops on
# the standby - structural, and with heapallindexed - for PHASE4_SECS
# seconds (default 12).  A watchdog fails the phase once replay has not moved
# for PHASE4_STALL seconds (default 10) while the standby holds WAL it has
# received and not replayed: that deadlock never ends, so it cannot hang the
# run, and statement_timeout could not have ended it - a backend waiting for
# a buffer lock never looks at it.  Every verify() call must return, and
# return without an error.  Then, caught up, verify() and the counts once more
# on the quiet standby.

# p4_verify_loop <label> <heapallindexed> <deadline ms>: verify() on the standby
# until the deadline; "<calls ok> <calls failed>" goes to $BASE/p4_<label>.n.
p4_verify_loop() {
	local label=$1 hai=$2 deadline=$3 ok=0 bad=0
	while [ "$(now_ms)" -lt "$deadline" ]; do
		if psql_s -c "select lion_index_verify('sv_k'::regclass, $hai)" \
			>>"$BASE/p4_$label.out" 2>>"$BASE/p4_$label.err"; then
			ok=$((ok + 1))
		else
			bad=$((bad + 1))
		fi
	done
	echo "$ok $bad" >"$BASE/p4_$label.n"
}

# p4_bounded <program> [args]: through timeout(1) where there is one, so that
# not even a standby that stopped answering can hang the watchdog.
p4_bounded() {
	if command -v timeout >/dev/null 2>&1; then
		timeout 10 "$@"
	else
		"$@"
	fi
}

# The standby's replay position, and whether it has received WAL beyond it.
p4_replay_state() {
	p4_bounded "$PGBIN/psql" -X -q -tA -F' ' -h "$SOCKDIR" -p "$STANDBY_PORT" \
		-U postgres -d "$DBNAME" -c \
		"select pg_last_wal_replay_lsn(), pg_last_wal_receive_lsn() > pg_last_wal_replay_lsn()" \
		2>/dev/null || echo "? ?"
}

phase4() {
	local secs=${PHASE4_SECS:-12} stall=${PHASE4_STALL:-10} deadline cap
	local wpid spid hpid t lsn behind lastlsn="" lastmove longest=0 alive
	local lsn0 lsn1 n_s n_h bad_s bad_h nwarn p
	log ""
	log "=== phase 4: lion_index_verify() on a hot standby beside replay (${secs}s) ==="

	# A standby of its own: phase 2 promotes the one it made, phase 3 stops
	# its own.
	stop_hard "$STANDBY_DATA"
	rm -rf "$STANDBY_DATA"
	psql_p >>"$RUNLOG" 2>&1 <<-SQL || die "phase 4: fixture failed"
		SET synchronous_commit = on;
		DROP TABLE IF EXISTS sv;
		CREATE TABLE sv (k int NOT NULL, r int);
		-- three keys, each a posting tree of about twenty leaves and a root
		-- (a set of up to 16 has none, DESIGN.md §22)
		INSERT INTO sv SELECT g % 3, g FROM generate_series(1, 900000) g;
		CREATE INDEX sv_k ON sv USING lion (k);
	SQL
	psql_p -c "VACUUM (FREEZE, ANALYZE) sv" >>"$RUNLOG" 2>&1 ||
		die "phase 4: VACUUM of the fixture failed"
	[ "$(psql_p -tAc "select max_posting_height > 0 and inline_entries = 0 from lion_index_stats('sv_k')")" = t ] ||
		die "phase 4: the fixture's sets are not posting trees"

	basebackup_standby
	wait_catchup

	cat >"$BASE/p4_writer.sql" <<-'EOF2'
		\set k random(0, 2)
		INSERT INTO sv SELECT :k, g FROM generate_series(1, 5) g;
	EOF2
	: >"$BASE/p4_s.out"; : >"$BASE/p4_s.err"; : >"$BASE/p4_h.out"; : >"$BASE/p4_h.err"
	rm -f "$BASE/p4_s.n" "$BASE/p4_h.n"
	lsn0=$(psql_p -tAc "select pg_current_wal_insert_lsn()")
	t=$(now_ms)
	deadline=$((t + secs * 1000))
	cap=$((deadline + 60000))

	PGOPTIONS='-c synchronous_commit=off' \
		"$PGBIN/pgbench" -n -T "$secs" -c 4 -j 2 -f "$BASE/p4_writer.sql" \
		-h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres "$DBNAME" \
		>"$BASE/p4_writer.log" 2>&1 &
	wpid=$!
	p4_verify_loop s false "$deadline" &
	spid=$!
	p4_verify_loop h true "$deadline" &
	hpid=$!
	P4PIDS="$wpid $spid $hpid"

	# The watchdog, until the writer and both loops are done.
	lastmove=$(now_ms)
	while :; do
		alive=0
		for p in $wpid $spid $hpid; do
			kill -0 "$p" 2>/dev/null && alive=1
		done
		[ "$alive" = 0 ] && break
		t=$(now_ms)
		read -r lsn behind <<<"$(p4_replay_state)"
		if [ "$lsn" != "$lastlsn" ] || [ "$behind" != t ]; then
			[ $((t - lastmove)) -gt "$longest" ] && longest=$((t - lastmove))
			lastlsn=$lsn
			lastmove=$t
		elif [ $((t - lastmove)) -ge $((stall * 1000)) ]; then
			{ echo "---- phase 4: the standby's backends when replay stopped";
			  p4_bounded "$PGBIN/psql" -X -h "$SOCKDIR" -p "$STANDBY_PORT" -U postgres -d "$DBNAME" -c \
				"select pid, backend_type, state, wait_event_type, wait_event, left(query, 60) as query from pg_stat_activity where backend_type in ('startup', 'client backend') and pid <> pg_backend_pid()" || true; } |
				tee -a "$RUNLOG" >&2
			die "phase 4: replay on the standby has not moved from $lsn for ${stall}s while WAL it received waits to be replayed: it is deadlocked against lion_index_verify() (DESIGN.md §25)"
		fi
		[ "$t" -lt "$cap" ] ||
			die "phase 4: the writer or a standby verify() loop is still running $(( (t - deadline) / 1000 ))s after its deadline"
		nap 0.5
	done
	P4PIDS=""
	wait "$wpid" || die "phase 4: the pgbench writer failed: $(tail -3 "$BASE/p4_writer.log")"
	wait "$spid" "$hpid" || true
	lsn1=$(psql_p -tAc "select pg_current_wal_insert_lsn()")
	cat "$BASE/p4_writer.log" "$BASE/p4_s.err" "$BASE/p4_h.err" >>"$RUNLOG"

	read -r n_s bad_s <"$BASE/p4_s.n" || die "phase 4: the structural verify() loop left no count"
	read -r n_h bad_h <"$BASE/p4_h.n" || die "phase 4: the heapallindexed verify() loop left no count"
	[ "$bad_s" = 0 ] && [ "$bad_h" = 0 ] ||
		die "phase 4: $bad_s structural and $bad_h heapallindexed verify() calls on the standby failed beside replay: $(grep -h ERROR "$BASE/p4_s.err" "$BASE/p4_h.err" | head -3)"
	[ "$n_s" -gt 0 ] && [ "$n_h" -gt 0 ] ||
		die "phase 4: a verify() loop made no call at all ($n_s structural, $n_h heapallindexed)"
	nwarn=$(cat "$BASE/p4_s.err" "$BASE/p4_h.err" | grep -c '^WARNING' || true)

	wait_catchup
	run_check "phase 4 standby" psql_s "
		select lion_index_verify('sv_k'::regclass, true) is not null, 'verify sv_k'
		union all
		select lion_index_count('sv_k'::regclass, k) = n, 'count k = ' || k
		  from (select k, count(*) as n from sv group by k) s
		union all
		select (select count(*) from sv) = $(psql_p -tAc "select count(*) from sv"), 'rows'"

	log "phase 4: $n_s structural and $n_h heapallindexed verify() calls on the standby beside replay of $(psql_p -tAc "select pg_size_pretty(pg_wal_lsn_diff('$lsn1', '$lsn0'))") of WAL; replay never stalled (longest pause $((longest / 1000)).$(( (longest % 1000) / 100 ))s); $nwarn warnings"
	SUMMARY+=("phase4 standby      $n_s + $n_h verify() calls beside replay, none failed, replay never stalled")

	stop_hard "$STANDBY_DATA"
	psql_p -c "DROP TABLE sv" >>"$RUNLOG" 2>&1
}

# ---------------------------------------------------------------- phase 5

# lionvec's locator (DESIGN.md §44.13) through crash recovery and on a hot
# standby.  An index over vectors in TOAST keeps, for each row, where its
# value's chunks lie - LOC entries in heap order under a LOCDIR, and the same
# place in the row's REFINE item - written by the build (log_newpage_range()),
# by every insert (generic WAL, after the row's REFINE and CODE items, under
# the meta page's lock) and removed by VACUUM.  None of it may come back
# wrong: a committed row's entry must be there, name its value and its
# places, and a scan must read the same rows by place as from the heap.
#
#  * Three rounds of a mixed write load (lionvec_writer.sql: inserts,
#    updates HOT and not, of the vector and of other columns, DELETEs,
#    rollbacks, INSERT ... SELECT) beside a loop of VACUUMs, of the table
#    and of its TOAST relation alone, killed as phase 1 kills it, the last
#    with a VACUUM in flight; each restart must have replayed generic WAL,
#    and lv_rec_check() (lionvec.sql) passes before and after a VACUUM.  The
#    writers set wal_consistency_checking = 'generic', so every record they
#    write carries the page as the primary had it, and recovery compares.
#  * A crash with an insert parked between its CODE items and its LOC entry
#    (lionvec-insert-unlocated, injection points only): its row comes back
#    aborted, with its items and without its entry, which the scans pass
#    over until VACUUM removes them.
#  * A hot standby of the result, under the same load: lv_rec_check() on the
#    standby, its answers the primary's, and after a VACUUM FULL of the
#    TOAST relation alone - every place stale, the scans falling back - the
#    same again.
start_lv_vacuum_loop() {
	(
		n=0
		while :; do
			if [ $((n % 2)) = 0 ]; then w="VACUUM lv_rec"; else w="VACUUM (PROCESS_MAIN false) lv_rec"; fi
			"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres \
				-d "$DBNAME" -c "SET wal_consistency_checking = 'generic'" -c "$w" >/dev/null 2>&1 || true
			n=$((n + 1))
			command sleep 0.5
		done
	) &
	VACLOOP=$!
}

lv_writer() {
	PGOPTIONS="-c synchronous_commit=on -c wal_consistency_checking=generic" \
		"$PGBIN/pgbench" -n -T "$1" -c 4 -j 2 -f "$HERE/lionvec_writer.sql" \
		-h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres "$DBNAME" \
		>>"$RUNLOG" 2>&1 || die "phase 5: the lionvec writer failed (see $RUNLOG)"
}

phase5() {
	local it style off ev nrows ck pid i probe_p probe_s
	log ""
	log "=== phase 5: lionvec's TOAST locator through crash recovery and a hot standby ==="
	if [ "$(psql_p -tAc "select count(*) from pg_available_extensions where name in ('vector', 'pageinspect')" 2>>"$RUNLOG" || true)" != 2 ]; then
		if [ "$REQUIRE_PGVECTOR" = 1 ]; then
			die "phase 5 needs pgvector and pageinspect, and this server lacks one of them (PGVECTOR=1 makes that a failure rather than a skip)"
		fi
		log "phase 5 skipped: this server has no pgvector or no pageinspect"
		SUMMARY+=("phase5              skipped (no pgvector or no pageinspect)")
		return 0
	fi
	psql_p -f "$HERE/lionvec.sql" >>"$RUNLOG" 2>&1 || die "phase 5: lionvec.sql failed (see $RUNLOG)"
	psql_p -c "CHECKPOINT" >>"$RUNLOG" 2>&1
	run_check "phase 5 baseline" psql_p "select * from lv_rec_check(true, true)"
	log "phase 5 baseline: $NCHECKS checks ok"

	for it in 1 2 3; do
		if [ $((it % 2)) -eq 1 ]; then style=immediate; else style=kill9; fi
		[ "$it" = 3 ] || start_lv_vacuum_loop
		lv_writer 2
		stop_vacuum_loop
		off=$(stat -c %s "$PRIMARY_LOG")
		if [ "$it" = 3 ]; then
			"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres -d "$DBNAME" \
				-c "SET wal_consistency_checking = 'generic'" -c "VACUUM (VERBOSE) lv_rec" >>"$RUNLOG" 2>&1 &
			VACPID=$!
			nap 0.3
		fi
		"crash_$style"
		if [ "$it" = 3 ]; then
			wait "$VACPID" 2>/dev/null || true
			VACPID=""
		fi
		start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
		verify_node psql_p "$PRIMARY_DATA"
		ev=$(recovery_evidence "$off") || exit 1
		case ${ev#* } in
			''|*[!0-9]*) die "phase 5 round $it: unreadable recovery evidence: '$ev'" ;;
		esac
		GENERIC_TOTAL=$((GENERIC_TOTAL + ${ev#* }))
		run_check "phase 5 round $it post-recovery" psql_p "select * from lv_rec_check(true, false)"
		ck=$NCHECKS
		psql_p -c "VACUUM (INDEX_CLEANUP ON) lv_rec" >>"$RUNLOG" 2>&1
		run_check "phase 5 round $it post-vacuum" psql_p "select * from lv_rec_check(true, true)"
		nrows=$(psql_p -tAc "select count(*) from lv_rec")
		log "phase 5 round $it: crash=$style index_wal_replayed=${ev#* } rows=$nrows checks=$((ck + NCHECKS))"
		SUMMARY+=("phase5 round $it      crash=$style index_wal_replayed=${ev#* } rows=$nrows checks=$((ck + NCHECKS))")
	done

	# The insert parked between its items and its entry, and the crash.
	if [ "$(psql_p -tAc "select count(*) from pg_available_extensions where name = 'injection_points'" 2>>"$RUNLOG" || true)" = 1 ]; then
		psql_p -c "CREATE EXTENSION IF NOT EXISTS injection_points" >>"$RUNLOG" 2>&1
		"$PGBIN/psql" -X -q -h "$SOCKDIR" -p "$PRIMARY_PORT" -U postgres -d "$DBNAME" \
			-c "SELECT injection_points_set_local()" \
			-c "SELECT injection_points_attach('lionvec-insert-unlocated', 'wait')" \
			-c "INSERT INTO lv_rec VALUES (-1000, 3, 'p', lv_rec_vec(-1000))" >>"$RUNLOG" 2>&1 &
		pid=$!
		wait_true psql_p "select count(*) > 0 from pg_stat_activity where wait_event = 'lionvec-insert-unlocated'" \
			60 "the insert to park at lionvec-insert-unlocated"
		# A commit after it flushes its records, so that replay brings back
		# its items (into another table: the parked insert holds lv_rec_v's
		# meta page).
		psql_p -c "SET synchronous_commit = on" -c "CREATE TABLE lv_rec_flush (i int)" >>"$RUNLOG" 2>&1 ||
			die "phase 5: the flushing commit failed"
		off=$(stat -c %s "$PRIMARY_LOG")
		crash_immediate
		wait "$pid" 2>/dev/null || true
		start_node "$PRIMARY_DATA" "$PRIMARY_PORT" "$PRIMARY_LOG"
		verify_node psql_p "$PRIMARY_DATA"
		ev=$(recovery_evidence "$off") || exit 1
		run_check "phase 5 unlocated" psql_p "select * from lv_rec_unlocated(-1000)"
		run_check "phase 5 unlocated, the rest" psql_p "select * from lv_rec_check(true, false)"
		psql_p -c "VACUUM (INDEX_CLEANUP ON) lv_rec" >>"$RUNLOG" 2>&1
		run_check "phase 5 unlocated, vacuumed" psql_p "select * from lv_rec_check(true, true)"
		log "phase 5: the insert parked before its entry came back aborted, without it, and the scans passed it over"
		SUMMARY+=("phase5 unlocated    an insert cut before its LOC entry: its row aborted, the scans the same")
	else
		no_injection_points "phase 5 (the unlocated insert)" "phase5 unlocated    skipped (no injection points)"
	fi

	# A hot standby of its own, under the same load.
	stop_hard "$STANDBY_DATA"
	rm -rf "$STANDBY_DATA"
	basebackup_standby
	lv_writer 2
	psql_p -c "SET wal_consistency_checking = 'generic'" -c "VACUUM (PROCESS_MAIN false) lv_rec" >>"$RUNLOG" 2>&1
	lv_writer 1
	wait_catchup
	run_check "phase 5 standby" psql_s "select * from lv_rec_check(true, false)"
	ck=$NCHECKS
	probe_p=$BASE/lv_probe_primary.txt
	probe_s=$BASE/lv_probe_standby.txt
	psql_p -tAc "select * from lv_rec_check(true, false)" >"$probe_p"
	psql_s -tAc "select * from lv_rec_check(true, false)" >"$probe_s"
	diff -u "$probe_p" "$probe_s" >>"$RUNLOG" 2>&1 ||
		die "phase 5: the standby's checks do not read as the primary's (see $RUNLOG)"
	# Every place stale: the TOAST relation rewritten on its own.
	psql_p -c "SET wal_consistency_checking = 'generic'" \
		-c "VACUUM FULL $(psql_p -tAc "select reltoastrelid::regclass from pg_class where oid = 'lv_rec'::regclass")" \
		>>"$RUNLOG" 2>&1 || die "phase 5: VACUUM FULL of the TOAST relation failed"
	lv_writer 1
	wait_catchup
	run_check "phase 5 standby, TOAST rewritten" psql_s "select * from lv_rec_check(false, false)"
	run_check "phase 5 primary, TOAST rewritten" psql_p "select * from lv_rec_check(false, false)"
	stop_hard "$STANDBY_DATA"
	log "phase 5 standby: $ck checks ok, the primary's answers, and the same after the TOAST relation's rewrite"
	SUMMARY+=("phase5 standby      $ck checks ok in recovery, and after a rewrite of the TOAST relation")
	psql_p -c "DROP TABLE IF EXISTS lv_rec, lv_rec_flush" >>"$RUNLOG" 2>&1
}

# ---------------------------------------------------------------- main

mkdir -p "$LOGDIR"
: >"$RUNLOG"
trap cleanup EXIT
make_private_dirs
: >"$BASE/warnings.txt"

START=$(now_ms)
log "pg_lion recovery tests"
log "prefix     $PREFIX"
log "server     $("$PGBIN/pg_config" --version)"
log "clusters   $PRIMARY_DATA (port $PRIMARY_PORT), $STANDBY_DATA (port $STANDBY_PORT)"
log "socket dir $SOCKDIR"
[ -z "$RUN_AS" ] || log "run as     $RUN_AS (initdb, pg_ctl, pg_basebackup)"
log ""

if [ "$SKIP_INSTALL" = 1 ]; then
	check_installed_extension
else
	build_extension
fi
init_primary

log "-- loading the fixture (test/recovery/schema.sql)"
psql_p -f "$HERE/schema.sql" >>"$RUNLOG" 2>&1 || die "schema.sql failed (see $RUNLOG)"
psql_p -c "VACUUM (ANALYZE, INDEX_CLEANUP ON) lion_rec" >>"$RUNLOG" 2>&1
# Without this the first crash replays everything since initdb, and the
# end-of-recovery checkpoint recycles the older half of that range before
# recovery_evidence() can read it.
psql_p -c "CHECKPOINT" >>"$RUNLOG" 2>&1
log "-- index shapes after the load and before any crash"
psql_p -tA -F'|' -c "select idx, ntids, entries, inline_entries, containers, sparse_segments, container_pages, leaf_pages, internal_pages, null_tids, empty_tids, nullelem_tids from lion_rec_indexes(), lion_index_stats(idx::regclass)" |
	tee -a "$RUNLOG"
run_check "baseline" psql_p "select * from lion_rec_check(true, true)"
log "-- baseline: $NCHECKS checks ok"

want_phase() { case " $PHASES " in *" $1 "*) return 0 ;; esac; return 1; }
want_phase 1 && phase1
want_phase 1b && phase1b
want_phase 1c && phase1c
want_phase 1d && phase1d
want_phase 1e && phase1e
want_phase 1f && phase1f
want_phase 1g && phase1g
want_phase 1h && phase1h
want_phase 2 && phase2
want_phase 3 && phase3
want_phase 4 && phase4
want_phase 5 && phase5

END=$(now_ms)
NWARN=$(wc -l <"$BASE/warnings.txt")
log ""
log "=== summary ==="
for s in "${SUMMARY[@]}"; do log "  $s"; done
log ""
log "  phases run: $PHASES"
log "  index WAL records replayed across all crashes ($MODE mode): $GENERIC_TOTAL"
log "  verify() warnings (unreferenced/empty pages, harmless by design): $NWARN"
log "  total time: $(( (END - START) / 1000 ))s"
log ""
log "ALL RECOVERY AND HOT-STANDBY CHECKS PASSED"