set -uo pipefail

ROOT=${1:-/tmp/toast-leak-check/build/tmp_install_root/usr/local/pgsql}
WORK=${2:-$HOME/toast-leak-repro}
BINDIR=$ROOT/bin
PORT=55678
export PGPORT=$PORT PGHOST=localhost
export LD_LIBRARY_PATH="$ROOT/lib64:$ROOT/lib:${LD_LIBRARY_PATH:-}"

B="$BINDIR"
PSQL="$B/psql -U postgres -X -q -v ON_ERROR_STOP=1"

echo "== binaries: $BINDIR/psql; workdir: $WORK"

rm -rf "$WORK" /tmp/vf_pid_repro
mkdir -p "$WORK"
"$B/initdb" -D "$WORK/data" -U postgres > "$WORK/initdb.log" 2>&1
cat >> "$WORK/data/postgresql.conf" <<EOF
autovacuum = off
port = $PORT
listen_addresses = 'localhost'
log_error_verbosity = verbose
EOF
"$B/pg_ctl" -D "$WORK/data" -l "$WORK/server.log" -w start
trap '"$B/pg_ctl" -D "$WORK/data" -m immediate stop > /dev/null 2>&1' EXIT

$PSQL -d postgres -c "CREATE DATABASE holder_db"

echo "== setup: 1000 rows x 2 x 100kB toasted; drop front half to free early blocks"
$PSQL -d postgres <<'SQL'
CREATE TABLE leak_demo (id int PRIMARY KEY, a text, b text)
    WITH (autovacuum_enabled = off);
ALTER TABLE leak_demo ALTER COLUMN a SET STORAGE EXTERNAL;
ALTER TABLE leak_demo ALTER COLUMN b SET STORAGE EXTERNAL;
INSERT INTO leak_demo SELECT g, repeat('x', 100000), repeat('y', 100000)
  FROM generate_series(1, 1000) g;
DELETE FROM leak_demo WHERE id <= 500;
VACUUM leak_demo;
CHECKPOINT;
SQL

TOAST=$($PSQL -d postgres -Atc \
  "SELECT 'pg_toast.' || relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname='leak_demo')")
echo "== toast table: $TOAST"

echo "== start global xmin holder (holder_db) + local horizon holder"
"$B/psql" -U postgres -X -q -d holder_db > "$WORK/holder.out" 2>&1 <<'SQL' &
BEGIN;
CREATE TEMP TABLE xid_holder(x int);
SELECT pg_sleep(600);
SQL
HOLDER=$!
"$B/psql" -U postgres -X -q -d postgres > "$WORK/local.out" 2>&1 <<'SQL' &
BEGIN ISOLATION LEVEL REPEATABLE READ;
SELECT 1;
SELECT pg_sleep(20);
COMMIT;
SQL
LOCALH=$!
sleep 2

echo "== UPDATE only column b (old b chunks die; a chunks stay shared with live rows)"
$PSQL -d postgres -c "UPDATE leak_demo SET b = repeat('z', 100000)"
echo "== VACUUM main only (old versions kept: local holder pins)"
$PSQL -d postgres -c "VACUUM (PROCESS_TOAST false) leak_demo"
wait $LOCALH
echo "== VACUUM toast table (old b chunks removed; a chunks live)"
$PSQL -d postgres -c "VACUUM $TOAST"
$PSQL -d postgres -Atc "SELECT count(*) FROM $TOAST" | sed 's/^/toast chunks remaining: /'

echo "== running VACUUM FULL in dedicated backend, sampling contexts + RSS"
"$B/psql" -U postgres -X -q -d postgres > "$WORK/vf.out" 2>&1 <<'SQL' &
SELECT pg_backend_pid() AS vf_pid \g /tmp/vf_pid_repro
SELECT pg_sleep(3);
VACUUM (FULL, VERBOSE) leak_demo;
SELECT 'VF DONE' AS done;
SELECT pg_sleep(600);
SQL
VF=$!
for i in $(seq 1 100); do [ -s /tmp/vf_pid_repro ] && break; sleep 0.1; done
VFPID=$(grep -oE '[0-9]+' /tmp/vf_pid_repro | head -1)
echo "== VACUUM FULL backend pid: $VFPID"
awk '/VmRSS/{print "RSS before VF:", $2, "kB"}' /proc/$VFPID/status

maxrss=0
while ! grep -q "VF DONE" "$WORK/vf.out" 2>/dev/null; do
    "$B/psql" -U postgres -X -q -d postgres \
        -c "SELECT pg_log_backend_memory_contexts($VFPID)" > /dev/null 2>&1
    rss=$(awk '/VmRSS/{print $2}' /proc/$VFPID/status 2>/dev/null || echo 0)
    [ "${rss:-0}" -gt "$maxrss" ] && maxrss=$rss
    sleep 0.1
done
echo "== max RSS during VF: $maxrss kB (NB: includes touched shared_buffers)"
kill $VF $HOLDER 2>/dev/null
wait 2>/dev/null

echo
echo "== VACUUM FULL output:"
grep -E "INFO|ERROR" "$WORK/vf.out" || true
echo
echo "== LEAKDBG evidence (server log):"
grep -c "fails at attr" "$WORK/server.log" | sed 's/^/failures in toast_tuple_init: /'
grep "fails at attr" "$WORK/server.log" | head -2 | sed 's/.*00000: /  /'
grep "fails at attr" "$WORK/server.log" | grep -oE "[0-9]+ bytes" |
    awk '{s+=$1} END {printf "total bytes detoasted then abandoned/freed: %d (%.1f MB)\n", s, s/1048576}'
echo
echo "== 'Table rewrite' context samples (pg_log_backend_memory_contexts):"
grep "Table rewrite" "$WORK/server.log" | sed 's/.*00000: /  /' | tail -4
echo
echo "== COPYDBG classification counts:"
grep "COPYDBG" "$WORK/server.log" | grep -oE "recently_dead=[01]" | sort | uniq -c
echo
echo "DONE.  Workdir: $WORK"