#!/bin/bash
# Variant for an old summary: everything up to the promotion runs with PGBIN,
# the build without the fix, so the standby writes a timeline 1 summary that
# crosses the switch point.  Then the promoted server is restarted with
# PGBIN_NEW (a minor update), which takes the incremental backup, combines
# and compares.  PGBIN_NEW=PGBIN is the control (a restart, no update).
#
# Incremental backup after promoting a streaming standby at a recovery target
# that falls inside a WAL summary.  The standby's summarizer follows the
# walreceiver, so it has already written an old-timeline summary that runs
# past the switch point, including WAL the promoted server never replayed.
# That summary is used for the old timeline, and what the never-replayed WAL
# did to a relation (here by default a VACUUM truncation) is applied to the
# incremental backup.
#
# usage: PGBIN=/path/to/bin bash repro-C.sh WORKDIR [PORT]
# knobs (environment):
#   KIND=lsn       recovery_target_lsn = an insert LSN, inclusive = false [default]
#   KIND=lsn-incl  the same LSN, recovery_target_inclusive = true
#   KIND=xid       recovery_target_xid = the xid of the replayed UPDATE
#   KIND=time      recovery_target_time = a timestamp taken after it
#   KIND=name      recovery_target_name = a restore point created after it
#   ACTION=pause   recovery_target_action, then pg_ctl promote   [default]
#   ACTION=promote recovery_target_action = promote (no wait for summaries)
#   JUNK=vacuum    never-replayed WAL: DELETE + VACUUM truncates t 8 -> 4 blocks [default]
#   JUNK=drop | truncate | vacuumfull | cluster   on t
#   JUNK=dropdb    DROP DATABASE d2 (d2.t2 was updated before the target)
#   JUNK=none      nothing destructive after the target (control)
# WAIT=summary (default) waits for a timeline 1 summary past the target;
# WAIT=receive only for the walreceiver, then for the summarizer to reach replay.
# exit: 0 restored data matches the source, 1 it does not, 2 setup problem,
#       3 the incremental backup was refused
set -u
: "${PGBIN:?set PGBIN to the bin directory of the build under test}"
: "${PGBIN_NEW:?set PGBIN_NEW to the bin directory used after the update}"
W=${1:?usage: PGBIN=... bash repro-C.sh WORKDIR [PORT]}
P=${2:-5440}; P1=$((P + 1)); P2=$((P + 2))
KIND=${KIND:-lsn}; ACTION=${ACTION:-pause}; JUNK=${JUNK:-vacuum}
die() { echo "SETUP: $*"; exit 2; }
q() { "$PGBIN/psql" -X -h "$W" -p "$1" -d "${3:-postgres}" -Atq -c "$2"; }
poll() {  # poll PORT SQL: wait up to 60 s for SQL to return t
  for _ in $(seq 600); do [ "$(q "$1" "$2" 2>/dev/null)" = t ] && return 0; sleep 0.1; done
  return 1
}
stopall() { for d in src standby restored; do
  [ -f "$W/$d/postmaster.pid" ] && "$PGBIN/pg_ctl" -D "$W/$d" -m immediate -w stop >/dev/null 2>&1
done; }
trap stopall EXIT

rm -rf "$W"; mkdir -p "$W" || die "mkdir $W"
"$PGBIN/initdb" -D "$W/src" -A trust --no-locale >/dev/null || die initdb
cat >> "$W/src/postgresql.conf" <<EOC
port = $P
unix_socket_directories = '$W'
listen_addresses = ''
summarize_wal = on
autovacuum = off
EOC
"$PGBIN/pg_ctl" -D "$W/src" -l "$W/src.log" -w start >/dev/null || die "start primary"

# One row per page: t has 8 rows in 8 blocks; d2.t2 the same.
q $P "CREATE TABLE t (id int, pad text STORAGE PLAIN)"
q $P "INSERT INTO t SELECT g, repeat('x', 4096) FROM generate_series(1, 8) g"
q $P "CREATE INDEX t_id ON t (id)"
q $P "VACUUM t"
q $P "CREATE DATABASE d2"
q $P "CREATE TABLE t2 (id int, pad text STORAGE PLAIN)" d2
q $P "INSERT INTO t2 SELECT g, repeat('y', 4096) FROM generate_series(1, 8) g" d2
q $P "VACUUM t2" d2
"$PGBIN/pg_basebackup" -h "$W" -p $P -D "$W/full" -c fast || die "full backup"
cp -a "$W/full" "$W/standby"
touch "$W/standby/standby.signal"

# A WAL summary starts at this checkpoint's redo LSN; the target falls after it.
q $P "CHECKPOINT"
SUMSTART=$(q $P "SELECT redo_lsn FROM pg_control_checkpoint()")
# Replayed by the standby: new versions of rows 5..8 (t blocks 4..11).
UPDXID=$(q $P "BEGIN; UPDATE t SET pad = pad || 'o' WHERE id > 4;
               SELECT pg_current_xact_id(); COMMIT;")
q $P "UPDATE t2 SET pad = pad || 'o' WHERE id > 4" d2
sleep 1
TS=$(q $P "SELECT clock_timestamp()")
sleep 1
q $P "SELECT pg_create_restore_point('c_target')" >/dev/null
q $P "CREATE TABLE marker ()"
LSN=$(q $P "SELECT pg_current_wal_insert_lsn()")
case $KIND in
  lsn)      TGT="recovery_target_lsn = '$LSN'
recovery_target_inclusive = false" ;;
  lsn-incl) TGT="recovery_target_lsn = '$LSN'
recovery_target_inclusive = true" ;;
  xid)      TGT="recovery_target_xid = '$UPDXID'" ;;
  time)     TGT="recovery_target_time = '$TS'" ;;
  name)     TGT="recovery_target_name = 'c_target'" ;;
  *) die "KIND=$KIND" ;;
esac

# Received by the standby, never replayed.
case $JUNK in
  vacuum)     q $P "DELETE FROM t WHERE id > 4"; q $P "VACUUM t" ;;
  drop)       q $P "DROP TABLE t" ;;
  truncate)   q $P "TRUNCATE t" ;;
  vacuumfull) q $P "VACUUM FULL t" ;;
  cluster)    q $P "CLUSTER t USING t_id" ;;
  dropdb)     q $P "DROP DATABASE d2" ;;
  none)       q $P "CREATE TABLE marker2 ()" ;;
  *) die "JUNK=$JUNK" ;;
esac
q $P "CHECKPOINT"
X=$(q $P "SELECT redo_lsn FROM pg_control_checkpoint()")
echo "KIND=$KIND ACTION=$ACTION JUNK=$JUNK; summary start $SUMSTART, last primary redo $X"

cat >> "$W/standby/postgresql.conf" <<EOC
port = $P1
primary_conninfo = 'host=$W port=$P'
$TGT
recovery_target_action = '$ACTION'
EOC
"$PGBIN/pg_ctl" -D "$W/standby" -l "$W/standby.log" -w start >/dev/null || die "start standby"
if [ "$ACTION" = pause ]; then
  poll $P1 "SELECT pg_get_wal_replay_pause_state() = 'paused'" || die "standby did not pause"
  # The summarizer follows the walreceiver, not replay.
  if [ "${WAIT:-summary}" = summary ]; then
  poll $P1 "SELECT EXISTS (SELECT FROM pg_available_wal_summaries()
            WHERE tli = 1 AND end_lsn >= '$X')" || die "no timeline 1 summary up to $X"
  else
    # WAIT=receive: the standby has received everything; summarization has
    # caught up with replay (a summarizer that follows replay stops there).
    poll $P1 "SELECT pg_last_wal_receive_lsn() >= '$X'" || die "standby did not receive up to $X"
    poll $P1 "SELECT pending_lsn >= pg_last_wal_replay_lsn() FROM pg_get_wal_summarizer_state()" \
      || die "summarizer did not reach replay"
    sleep 2
    echo "summarizer pending $(q $P1 "SELECT pending_lsn FROM pg_get_wal_summarizer_state()"), replay $(q $P1 "SELECT pg_last_wal_replay_lsn()")"
  fi
  "$PGBIN/pg_ctl" -D "$W/src" -w stop >/dev/null
  "$PGBIN/pg_ctl" -D "$W/standby" -w promote >/dev/null || die promote
else
  poll $P1 "SELECT NOT pg_is_in_recovery()" || die "standby did not promote"
  "$PGBIN/pg_ctl" -D "$W/src" -w stop >/dev/null
fi
SW=$(awk '$1 == 1 {print $2}' "$W/standby/pg_wal/00000002.history")
echo "switch point $SW; timeline 1 summaries on the standby:"
q $P1 "SELECT '  ' || start_lsn || ' .. ' || end_lsn ||
       CASE WHEN start_lsn < '$SW' AND end_lsn > '$SW' THEN '   <- crosses the switch point'
            WHEN start_lsn = '$SW' THEN '   <- starts at the switch point' ELSE '' END
       FROM pg_available_wal_summaries() WHERE tli = 1 ORDER BY start_lsn"
q $P1 "CHECKPOINT"
# Minor update: restart the promoted server with the new binaries.
"$PGBIN/pg_ctl" -D "$W/standby" -w stop >/dev/null || die "stop before update"
PGBIN=$PGBIN_NEW
"$PGBIN/pg_ctl" -D "$W/standby" -l "$W/standby-new.log" -w start >/dev/null || die "start after update"
echo "restarted with $PGBIN ($("$PGBIN/postgres" -V))"
q $P1 "SELECT '  after update: ' || start_lsn || ' .. ' || end_lsn FROM pg_available_wal_summaries() WHERE tli = 1 AND start_lsn < '$SW' AND end_lsn > '$SW'"

# Don't read the tables before the backup: with data checksums, setting hint
# bits WAL-logs those blocks again on the new timeline and hides the loss.
if ! "$PGBIN/pg_basebackup" -h "$W" -p $P1 -D "$W/incr" -c fast \
  --incremental "$W/full/backup_manifest"; then echo "RESULT: REFUSED"; exit 3; fi
FP="SELECT count(*) || '|' || coalesce(max(id), 0) || '|' || md5(coalesce(string_agg(id
    || ':' || md5(pad), ',' ORDER BY id), ''))"
fp() { echo "t $(q $1 "$FP FROM t")  t2 $(q $1 "$FP FROM t2" d2)"; }
REL=$(q $P1 "SELECT pg_relation_filepath('t')")
SRC=$(fp $P1)
"$PGBIN/pg_ctl" -D "$W/standby" -w stop >/dev/null
echo "timeline 1 summaries with an entry for t ($REL):"
for f in $(ls "$W/standby/pg_wal/summaries" | grep '^00000001'); do
  s=$("$PGBIN/pg_walsummary" "$W/standby/pg_wal/summaries/$f" \
      | grep "REL ${REL##*/}, FORK main" | sed 's/.*FORK main: //' | tr '\n' ' ')
  echo "  $f: ${s:--}"
done
"$PGBIN/pg_verifybackup" -q "$W/full" && "$PGBIN/pg_verifybackup" -q "$W/incr" \
  && echo "pg_verifybackup: both backups OK"
"$PGBIN/pg_combinebackup" -o "$W/restored" "$W/full" "$W/incr" || die pg_combinebackup
sed -i "s/^port = $P1\$/port = $P2/; /^recovery_target/d; /^primary_conninfo/d" \
  "$W/restored/postgresql.conf"
"$PGBIN/pg_ctl" -D "$W/restored" -l "$W/restored.log" -w start >/dev/null || die "start restored"
DST=$(fp $P2)
"$PGBIN/pg_ctl" -D "$W/restored" -w stop >/dev/null
echo "source:   $SRC   (count|max(id)|md5 of rows)"
echo "restored: $DST"
if [ "$SRC" = "$DST" ]; then echo "RESULT: OK"; exit 0; fi
echo "RESULT: DATA LOSS"; exit 1