#!/usr/bin/env bash
# CTR empirical test B: crash-recovery UNDO-PHASE duration vs in-flight chain.
#
# README.undo:119-127 says a transaction that wrote UNDO but was caught
# in-flight by a crash (neither COMMIT nor ABORT) has its UNDO chain "walked
# and applied during this recovery pass". This build logs that pass explicitly:
#
#   LOG: starting undo phase for incomplete transactions
#   LOG: UNDO recovery complete: 1 transactions rolled back, <N> records applied
#   LOG: undo phase complete
#   LOG: database system is ready to accept connections   (AFTER the undo phase)
#
# So the undo phase runs synchronously before the cluster opens. We measure its
# duration (from the server log's millisecond timestamps) as a function of the
# in-flight chain length N, and confirm "records applied" == N (the chain
# survived the crash and was fully applied).
#
# Protocol per N: fresh start; a background psql builds an in-flight txn of N
# chmods, touches a readiness file, holds the txn open (pg_sleep); the driver
# waits for the file, then immediate-stops (crash); restart; parse the undo
# phase from the log.
set -u
PORT=${PORT:-5467}
BIN=$HOME/undo-v2-inst/bin
DATA=$HOME/undo-v2-data
SRVLOG=$DATA/server.log
PSQL="$BIN/psql -p $PORT -U postgres -X -q -t -A"
REPEATS=${REPEATS:-2}
NS=${NS:-"2000 8000 32000 128000"}
SIG=/tmp/ctr_ready
LOGDIR=/home/manu/Proyectos/yggdrasil/aportes/postgres-undo-ctr/logs_v2
mkdir -p "$LOGDIR"
OUT=${OUT:-$LOGDIR/ctr_recovery_time.tsv}

ensure_up() { $BIN/pg_ctl -D "$DATA" -l "$SRVLOG" -w start >/dev/null 2>&1; }
median() { sort -n | awk '{a[NR]=$1} END{ if(NR==0){print "NA"} else if(NR%2){print a[(NR+1)/2]} else {printf "%.3f\n",(a[NR/2]+a[NR/2+1])/2} }'; }
# ms between two "HH:MM:SS.mmm" server-log stamps on the given grep patterns,
# taken from the TAIL of the log (this restart).
phase_ms() {
  awk '
    /starting undo phase/        {t0=stamp($0)}
    /database system is ready/   {t1=stamp($0); print (t1-t0)*1000; exit}
    function stamp(l,  a,hh,mm,ss){ split(l,a," "); split(a[2],h,":"); return h[1]*3600+h[2]*60+h[3] }
  ' "$SRVLOG" 2>/dev/null | tail -1
}

ensure_up
$PSQL -c "CREATE EXTENSION IF NOT EXISTS test_fileops;" >/dev/null 2>&1
F=$($PSQL -c "SELECT test_fileops_create_tempfile('ctr_rec.dat');")

# one trial: returns "undo_ms<TAB>records_applied<TAB>ready_after_undo(0/1)"
one_trial() {
  local n=$1
  ensure_up
  rm -f "$SIG"
  ( $BIN/psql -p "$PORT" -U postgres -X -q -t -A >/dev/null 2>&1 <<SQL
BEGIN;
SELECT count(test_fileops_chmod('$F', (0400 + (g%2)*0200))) FROM generate_series(1,$n) g;
\! touch $SIG
SELECT pg_sleep(120);
SQL
  ) &
  local waited=0
  while [ ! -f "$SIG" ]; do sleep 0.3; waited=$((waited+1)); [ $waited -gt 1200 ] && { echo "NA\tNA\tNA"; return; }; done
  sleep 0.3
  # mark the log so phase_ms only sees this restart
  local mark_lines
  mark_lines=$(wc -l < "$SRVLOG")
  $BIN/pg_ctl -D "$DATA" -m immediate -w stop >/dev/null 2>&1
  $BIN/pg_ctl -D "$DATA" -l "$SRVLOG" -w start >/dev/null 2>&1
  # parse only lines from this restart onward
  local seg; seg=$(tail -n +"$mark_lines" "$SRVLOG")
  local ums rec order
  ums=$(printf '%s\n' "$seg" | awk '
    /starting undo phase/      {split($2,h,":"); t0=h[1]*3600+h[2]*60+h[3]}
    /undo phase complete/      {split($2,h,":"); t1=h[1]*3600+h[2]*60+h[3]; print (t1-t0)*1000; exit}')
  rec=$(printf '%s\n' "$seg" | grep -oE '[0-9]+ records applied' | grep -oE '^[0-9]+' | tail -1)
  # ordering check: does "undo phase complete" appear BEFORE "ready to accept"?
  local lc_undo lc_ready
  lc_undo=$(printf '%s\n' "$seg" | grep -n "undo phase complete" | head -1 | cut -d: -f1)
  lc_ready=$(printf '%s\n' "$seg" | grep -n "ready to accept" | head -1 | cut -d: -f1)
  if [ -n "$lc_undo" ] && [ -n "$lc_ready" ] && [ "$lc_undo" -lt "$lc_ready" ]; then order=1; else order=0; fi
  printf '%s\t%s\t%s\n' "${ums:-NA}" "${rec:-NA}" "$order"
}

echo -e "N_inflight\tundo_phase_ms_median\trecords_applied\tundo_before_open\ttrials" | tee "$OUT"
for n in $NS; do
  mss=""; rec=""; ord=""
  for r in $(seq 1 $REPEATS); do
    res=$(one_trial "$n")
    m=$(printf '%s' "$res" | cut -f1); a=$(printf '%s' "$res" | cut -f2); o=$(printf '%s' "$res" | cut -f3)
    [ "$m" != "NA" ] && mss="${mss}${m}"$'\n'
    rec="$a"; ord="$o"
  done
  med=$(printf '%s' "$mss" | median)
  echo -e "$n\t$med\t$rec\t$ord\t$REPEATS" | tee -a "$OUT"
done
echo "=== DONE ctr_recovery_time ==="