#!/usr/bin/env bash
# repro-standby.sh: rows 1-5 and 9 of coverage-matrix.txt (standby promotion / replay).
# Primary + standby (wal_level=replica on both), one backend on the standby stuck in ClientWrite
# (stall.py), then pg_promote(false).  Prints one line of key=value results.
# usage: PGBIN=/path/to/install/bin ./repro-standby.sh CASE [PORT]
#   CASE: failover-at-start | local-at-start | failover-in-recovery | local-in-recovery |
#         toggled | replay-stuck | crash-then-promote          (needs contrib/test_decoding)
# Optional env: EXTRA_CONF (extra postgresql.conf lines, e.g. "procsignal_barrier_timeout = 2s"),
#   WINDOW (s to wait for promotion, default 12), RWAIT (s to wait for replay in replay-stuck,
#   default 15), CENSUS=1 (gdb census of stale caches, see coverage-matrix.txt).
# For tcp_user_timeout use EXTRA_CONF="tcp_user_timeout = 10s" WINDOW=40 RWAIT=40 and read
#   since_block_ms (time from the backend blocking to promotion / replay passing it).
set -u
CASE=$1 PP=${2:-61500}; SP=$((PP + 1))
B=${PGBIN:?set PGBIN to the bin directory of the build under test}
HERE=$(cd "$(dirname "$0")" && pwd); STALL=${STALL:-$HERE/stall.py}
RUN=$(mktemp -d); WINDOW=${WINDOW:-12}
trap 'down; rm -rf "$RUN"; exit 143' TERM INT
now_ms() { date +%s%3N; }
q() { "$B/psql" -X -h 127.0.0.1 -p "$1" -d postgres -Atq -c "$2"; }
wait_until() { local dl=$(( $(now_ms) + $1 * 1000 ))
  while [ "$(now_ms)" -lt "$dl" ]; do
    [ "$(q "$2" "$3" 2>/dev/null)" = "$4" ] && return 0; sleep 0.05; done
  return 1; }
down() { for p in ${KIDS:-}; do kill -9 "$p" 2>/dev/null; wait "$p" 2>/dev/null; done
  for d in "$RUN/standby" "$RUN/primary"; do
    [ -f "$d/postmaster.pid" ] && "$B/pg_ctl" -D "$d" -m immediate -w stop >/dev/null 2>&1; done; }
fail() { echo "case=$CASE result=SETUP_FAIL($1)"; down; rm -rf "$RUN"; exit 1; }
conf() { printf "port = %s\nlisten_addresses = '127.0.0.1'\n" "$2"
  printf "unix_socket_directories = '%s'\n" "$1"
  printf "wal_level = replica\nmax_wal_senders = 10\nhot_standby = on\nfsync = off\n"
  [ -n "${EXTRA_CONF:-}" ] && printf "%b\n" "$EXTRA_CONF"; } >> "$1/postgresql.conf"
caught_up() { local l; l=$(q "$PP" "select pg_current_wal_insert_lsn()")
  wait_until "$1" "$SP" "select pg_last_wal_replay_lsn() >= '$l'" t; }
primary_slot() { # $1 = failover true/false
  if [ "$1" = true ]; then
    q "$PP" "select pg_create_logical_replication_slot('fo', 'test_decoding', false, false, true)"
  else q "$PP" "select pg_create_logical_replication_slot('p1', 'test_decoding')"; fi >/dev/null; }
standby_slot() { # creating a slot on a standby waits for a running-xacts record from the primary
  rm -f "$RUN/mk"; { q "$SP" "select pg_create_logical_replication_slot('sb', 'test_decoding')" \
    > "$RUN/mk.out" 2>&1; touch "$RUN/mk"; } &
  for _ in $(seq 100); do q "$PP" "select pg_log_standby_snapshot()" >/dev/null
    [ -f "$RUN/mk" ] && break; sleep 0.1; done; }
wait_synced() { for _ in $(seq 120); do
  [ "$(q "$SP" "select count(*) from pg_replication_slots where slot_name = 'fo' and synced
       and not temporary")" = 1 ] && return 0
  q "$PP" "select pg_log_standby_snapshot()" >/dev/null
  q "$PP" "select pg_replication_slot_advance('fo', pg_current_wal_lsn())" >/dev/null 2>&1
  sleep 0.5; done; return 1; }
# Wait until the stuck backend stays blocked: its socket stops moving, and a wakeup no longer
# lets it reach CHECK_FOR_INTERRUPTS() (it would log its memory contexts if it did).
settle() { local last="" cur same=0 off
  for _ in $(seq 60); do
    cur=$(ss -tinH "( sport = :$SP and dport = :$CLPORT )" | grep -o 'bytes_acked:[0-9]*')
    if [ -n "$cur" ] && [ "$cur" = "$last" ]; then same=$((same + 1)); else same=0; fi
    [ $same -ge 4 ] && break; last=$cur; sleep 0.5; done
  for _ in $(seq 20); do off=$(stat -c %s "$RUN/standby.log")
    q "$SP" "select pg_log_backend_memory_contexts($HPID)" >/dev/null; sleep 0.5
    tail -c +$((off + 1)) "$RUN/standby.log" |
      grep -q "logging memory contexts of PID $HPID" || return 0
    wait_until 10 "$SP" "select wait_event from pg_stat_activity where pid = $HPID" ClientWrite
  done; }
holder() {
  python3 -I "$STALL" "$SP" postgres \
    "COPY (select generate_series(1, 200000000), repeat('x', 200)) TO STDOUT" > "$RUN/cl.out" 2>&1 &
  CPID=$!; KIDS="${KIDS:-} $CPID"
  for _ in $(seq 1000); do grep -q '^SENT' "$RUN/cl.out" && break; sleep 0.01; done
  HPID=$(awk '/^SENT/{print $2}' "$RUN/cl.out"); CLPORT=$(awk '/^SENT/{print $3}' "$RUN/cl.out")
  wait_until 20 "$SP" "select wait_event from pg_stat_activity where pid = $HPID" ClientWrite ||
    return 1
  TBLK=$(now_ms); settle; }
census() { # stale = processes whose XLogLogicalInfo differs from 1 (needs gdb and debug symbols)
  [ "${CENSUS:-0}" = 1 ] || { echo "-"; return; }
  local pm st=0 tot=0 v; pm=$(head -1 "$RUN/standby/postmaster.pid")
  for p in $(pgrep -P "$pm"); do [ "$p" = "${HPID:-x}" ] && [ "$1" = after ] && continue
    v=$(gdb -p "$p" -batch -ex 'printf "XLI=%d\n", XLogLogicalInfo' 2>/dev/null |
      grep -o 'XLI=[0-9]')
    [ -z "$v" ] && continue; tot=$((tot + 1)); [ "$v" != XLI=1 ] && st=$((st + 1)); done
  echo "$st/$tot"; }
case $CASE in failover-*|crash-then-promote) SYN=true;; *) SYN=false;; esac
"$B/initdb" -D "$RUN/primary" -A trust --no-sync >/dev/null 2>&1 || fail initdb
conf "$RUN/primary" "$PP"
{ echo "host replication all 127.0.0.1/32 trust"; cat "$RUN/primary/pg_hba.conf"; } > "$RUN/h"
mv "$RUN/h" "$RUN/primary/pg_hba.conf"
"$B/pg_ctl" -D "$RUN/primary" -l "$RUN/primary.log" -w start >/dev/null || fail primary_start
q "$PP" "create table t(i int)" >/dev/null
[ "$SYN" = true ] && q "$PP" "select pg_create_physical_replication_slot('phys')" >/dev/null
case $CASE in *-at-start|toggled|crash-then-promote) primary_slot "$SYN";; esac
"$B/pg_basebackup" -d "host=127.0.0.1 port=$PP dbname=postgres" -D "$RUN/standby" -R -X stream \
  --no-sync >/dev/null 2>&1 || fail basebackup
sed -i -e "s/^port = .*/port = $SP/" \
  -e "s#^unix_socket_directories = .*#unix_socket_directories = '$RUN/standby'#" \
  "$RUN/standby/postgresql.conf"
[ "$SYN" = true ] && printf "%s\n" "primary_slot_name = 'phys'" "hot_standby_feedback = on" \
  "sync_replication_slots = on" >> "$RUN/standby/postgresql.conf"
"$B/pg_ctl" -D "$RUN/standby" -l "$RUN/standby.log" -w start >/dev/null || fail standby_start
wait_until 30 "$PP" "select count(*) from pg_stat_replication where state = 'streaming'" 1 ||
  fail streaming
REPLAY=-
case $CASE in
  *-in-recovery) primary_slot "$SYN"; caught_up 20 || fail catchup;;
  toggled)   # status on at start, off (last slot dropped), then on again, all during recovery
    q "$PP" "select pg_drop_replication_slot('p1')" >/dev/null; q "$PP" "checkpoint" >/dev/null
    wait_until 20 "$PP" "show effective_wal_level" replica || fail no_disable
    caught_up 20 || fail catchup; primary_slot false; caught_up 20 || fail catchup2;;
  replay-stuck)   # backend stuck first, then the primary enables logical decoding
    holder || fail holder
    primary_slot false; L=$(q "$PP" "select pg_current_wal_insert_lsn()"); T=$(now_ms)
    if wait_until "${RWAIT:-15}" "$SP" "select pg_last_wal_replay_lsn() >= '$L'" t; then
      REPLAY="caught_up_$(( $(now_ms) - T ))ms(since_block_$(( $(now_ms) - TBLK ))ms)"
    else REPLAY="stalled_${RWAIT:-15}s"
      q "$SP" "select pg_terminate_backend($HPID)" >/dev/null; T=$(now_ms)
      wait_until 15 "$SP" "select pg_last_wal_replay_lsn() >= '$L'" t \
        && REPLAY="$REPLAY;after_terminate_$(( $(now_ms) - T ))ms"; fi
    kill -9 "$CPID" 2>/dev/null; wait "$CPID" 2>/dev/null; HPID="";;
esac
case $CASE in
  failover-*) wait_synced || fail not_synced;;
  local-*|toggled|replay-stuck) standby_slot;;
  crash-then-promote)   # synced slot plus a local one, then kill -9 a backend: crash restart
    wait_synced || fail not_synced; standby_slot
    "$B/psql" -X -h 127.0.0.1 -p "$SP" -d postgres -Atqc "select pg_sleep(600)" >/dev/null 2>&1 &
    KIDS="${KIDS:-} $!"; sleep 0.5
    kill -9 "$(q "$SP" "select pid from pg_stat_activity where query like 'select pg_sleep(600)%'")"
    sleep 1; wait_until 30 "$SP" "select 1" 1 || fail restart
    grep -q 'all server processes terminated; reinitializing' "$RUN/standby.log" || fail no_crash;;
esac
SLOT=$(q "$SP" "select slot_name from pg_replication_slots where slot_type = 'logical'
                and invalidation_reason is null order by slot_name limit 1")
[ -n "$SLOT" ] || fail no_standby_slot
holder || fail holder2
BEFORE=$(census before)
T=$(now_ms); q "$SP" "select pg_promote(false)" >/dev/null; RES=hung; PMS=-; SINCE=-; REL=-
while [ "$(now_ms)" -lt $((T + WINDOW * 1000)) ]; do
  [ "$(q "$SP" "select pg_is_in_recovery()" 2>/dev/null)" = f ] && {
    RES=promoted; PMS=$(( $(now_ms) - T )); SINCE=$(( $(now_ms) - TBLK )); break; }
  sleep 0.01; done
if [ $RES = hung ]; then q "$SP" "select pg_terminate_backend($HPID)" >/dev/null; T=$(now_ms)
  wait_until 15 "$SP" "select pg_is_in_recovery()" f && REL=$(( $(now_ms) - T )); fi
sleep 1; AFTER=$(census after)
q "$SP" "select pg_replication_slot_advance('$SLOT', pg_current_wal_lsn())" >/dev/null 2>&1
q "$SP" "insert into t select generate_series(1, 5)" >/dev/null
DEC=$(q "$SP" "select count(*) from pg_logical_slot_get_changes('$SLOT', null, null)
               where data like 'table public.t: INSERT%'" 2>&1 | head -1)
echo "case=$CASE result=$RES promote_ms=$PMS since_block_ms=$SINCE" \
  "release_after_terminate_ms=$REL" \
  "replay=$REPLAY stale_before=$BEFORE stale_after=$AFTER decoded_rows=$DEC" \
  "still_waiting_lines=$(grep -c 'still waiting for backend with PID' "$RUN/standby.log")" \
  "tcp_timed_out=$(grep -c 'could not send data to client: Connection timed out' \
                    "$RUN/standby.log")"
down; rm -rf "$RUN"