#!/usr/bin/env bash
#
# T9 - crash rate at the .ibb boundary, AND what happens to the transaction.
#
# WHY THIS MATTERS
#   The investigation started from a customer report about a binlog damaged
#   when the filesystem filled (MDEV-41194). For the file binlog that was
#   reproduced: a statement timeout breaks the ENOSPC wait inside my_write(),
#   leaving a torn event group with valid transactions written behind it, no
#   crash involved.
#
#   12.3 with binlog_storage_engine=innodb is structurally different. .ibb
#   files are preallocated at full max_binlog_size, so a transaction write
#   never meets ENOSPC inside a file; only creating the next file asks the
#   filesystem for space, and a spare is kept ahead. Measured outcomes when
#   that spare cannot be replaced:
#
#     survives:  [ERROR] InnoDB: Unable to allocate file ./binlog-000003.ibb:
#                        "No space left on device". Waiting 60 seconds...
#                client: COMMIT returned OK
#     crashes:   mariadbd got signal 11, exit 139, in
#                  trx_t::write_serialisation_history   (trx0trx.cc:1194)
#                  -> mtr_t::commit -> log_t::append    (mtr0mtr.cc:1224)
#                  -> memcpy
#                with  binlog_ctx = <Cannot access memory at address 0x0>,
#                while innodb_binlog_prealloc_thread sat in
#                  fsp_binlog_tablespace_close(file_no=1) -> flush_up_to
#
#   Unmeasured until now: after such a crash, is the transaction rolled back,
#   committed, or inconsistent with the binlog? That decides whether this is
#   an availability bug or a data bug - the customer's actual question.
#
# WHY A HOST FILESYSTEM
#   Earlier versions put the datadir on tmpfs (vanishes with the container,
#   so the aftermath cannot be examined) or on a loop mount inside the
#   container (which silently failed because the image runs as uid mysql,
#   leaving the fill step to write 200 GB onto the host root filesystem).
#
#   So the size limit now lives on the host, where it is visible and
#   verifiable, and `run` refuses to start unless the target really is a
#   separate, small filesystem. Freeing space after a crash is a plain rm.
#
# Usage
#   sudo ./t9-after-the-segv.sh setup            # create and mount the image
#        ./t9-after-the-segv.sh run [N] [opts]   # N iterations, default 5
#   sudo ./t9-after-the-segv.sh teardown         # unmount and delete
#
#        ./t9-after-the-segv.sh exhaust [opts]   # see below
#
#   opts:  --binlog-cache-size=N     default 268435456 (256M)
#          --max-statement-time=N    default 20, used by `exhaust`
#          --debug                   capture a core and a gdb backtrace
#
# THE `exhaust` MODE - the question `run` never reaches
#   `run` crosses into the already preallocated spare, so the user session
#   never waits: the failed allocation of the file AFTER that is handled by
#   the background thread, which logs
#     Unable to allocate file ./binlog-000003.ibb: "No space left on device".
#     Waiting 60 seconds before trying again...
#   and the transaction commits normally.
#
#   But that only holds while a spare exists. `exhaust` keeps writing until
#   the spare itself is full while no replacement can be created. At that
#   point there is no next .ibb and a user session must wait for one - which
#   is the first and only place in 12.3 where a transaction can block on a
#   full filesystem, and therefore the first place where max_statement_time
#   can break it. That is the direct analogue of MDEV-41194, where a timeout
#   breaking wait_for_free_space() left a torn event group behind.
#
#   max_statement_time defaults to 20s here, deliberately below the 60s
#   retry interval, so the timeout fires while the allocation is still
#   waiting.
#
#   The crash rate depends on binlog_cache_size: the 32K default produced no
#   crashes in 10 runs, 256M produced 1/5, and earlier runs of the same shape
#   produced 3/4 and 4/5. Hence the option.
#
set -uo pipefail

CMD="${1:-help}"; shift 2>/dev/null || true
ITER=5; DEBUG=0; BCACHE="268435456"; MST=20
for a in "$@"; do
  case "$a" in
    --debug) DEBUG=1 ;;
    --binlog-cache-size=*) BCACHE="${a#*=}" ;;
    --max-statement-time=*) MST="${a#*=}" ;;
    ''|*[!0-9]*) echo "unknown argument: $a"; exit 2 ;;
    *) ITER="$a" ;;
  esac
done

VER="${MARIADB_VERSION:-12.3}"
PW=secret
N=binlog-t9
HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
LOGDIR="$HERE/t9-logs"
COREDIR="$HERE/t9-cores"
BASE="${T9_BASE:-/srv/t9}"
IMGFILE="$BASE/data.img"
DATADIR="$BASE/mnt"
IMGFILE_MB="${IMGFILE_MB:-900}"
MAX_SANE_MB=5000                 # a datadir larger than this is not the test's
BINLOG_MB="${BINLOG_MB:-32}"
KEEP_MB="${KEEP_MB:-20}"
BLOB_MB="${BLOB_MB:-4}"
IMG="mariadb:$VER"; [ "$DEBUG" = 1 ] && IMG="mariadb:${VER}-dbg"

G=$'\033[32m'; R=$'\033[31m'; Y=$'\033[33m'; D=$'\033[2m'; Z=$'\033[m'
say(){ printf '\n%s==> %s%s\n' "$G" "$*" "$Z"; }
note(){ printf '%s    %s%s\n' "$D" "$*" "$Z"; }
warn(){ printf '%s !! %s%s\n' "$Y" "$*" "$Z"; }
bad(){ printf '%s XX %s%s\n' "$R" "$*" "$Z"; }
SUDO=""; [ "$(id -u)" = 0 ] || SUDO=sudo

########################################################################
do_setup() {
  say "creating a $IMGFILE_MB MB filesystem at $DATADIR"
  mkdir -p "$BASE" "$DATADIR"
  if mountpoint -q "$DATADIR"; then
    note "already mounted"; df -h "$DATADIR" | tail -1 | sed 's/^/    /'; return 0
  fi
  [ -f "$IMGFILE" ] || { truncate -s "${IMGFILE_MB}M" "$IMGFILE"; mkfs.ext4 -q -F "$IMGFILE"; }
  mount -o loop "$IMGFILE" "$DATADIR"
  chown 999:999 "$DATADIR"          # uid of mysql in the official image
  chmod 755 "$DATADIR"
  df -h "$DATADIR" | tail -1 | sed 's/^/    /'
  note "ready. Run without sudo:  $0 run 5"
}

do_teardown() {
  say "removing"
  docker rm -f "$N" >/dev/null 2>&1 || true
  mountpoint -q "$DATADIR" && umount "$DATADIR"
  rm -f "$IMGFILE"; rmdir "$DATADIR" "$BASE" 2>/dev/null
  note "gone"
}

# Nothing runs until the target is proven to be a small, separate filesystem.
# This check is the reason the script exists in this shape.
guard() {
  local src size_mb
  mountpoint -q "$DATADIR" || {
    bad "$DATADIR is not a mount point."
    bad "Run:  sudo $0 setup"
    bad "Without this the fill step would write onto your root filesystem."
    exit 1; }
  src="$(findmnt -no SOURCE --target "$DATADIR")"
  size_mb=$(( $(df -B1M --output=size "$DATADIR" | tail -1) ))
  case "$src" in /dev/loop*) ;; *)
    bad "$DATADIR is backed by $src, not a loop device. Refusing."; exit 1 ;; esac
  [ "$size_mb" -le "$MAX_SANE_MB" ] || {
    bad "$DATADIR is ${size_mb} MB, over the ${MAX_SANE_MB} MB sanity limit. Refusing."; exit 1; }
  note "datadir: $DATADIR on $src, ${size_mb} MB  (verified)"
}

wipe_datadir() { $SUDO find "$DATADIR" -mindepth 1 -delete 2>/dev/null; $SUDO chown 999:999 "$DATADIR"; }
free_space()   { $SUDO rm -f "$DATADIR/filler"; df -h "$DATADIR" | tail -1 | sed 's/^/      /'; }

########################################################################
alive(){ [ "$(docker inspect -f '{{.State.Running}}' "$N" 2>/dev/null)" = true ]; }
dex(){ alive || return 1; timeout 240 docker exec "$N" "$@" </dev/null; }
q(){ alive || return 1; timeout 240 docker exec "$N" mariadb -uroot -p"$PW" "$@" </dev/null; }
qn(){ q -N -B "$@" 2>/dev/null; }
pos(){ qn -e "SHOW MASTER STATUS" | awk '{print $1" "$2}'; }

OLD_PATTERN=""
cleanup_all(){
  docker rm -f "$N" >/dev/null 2>&1 || true
  [ -n "$OLD_PATTERN" ] && { $SUDO sysctl -q -w kernel.core_pattern="$OLD_PATTERN" 2>/dev/null || true
                             note "core_pattern restored"; }
}

build_debug_image() {
  docker image inspect "$IMG" >/dev/null 2>&1 && return 0
  say "building $IMG"
  docker build -t "$IMG" -f - "$HERE" <<DOCKERFILE
FROM mariadb:${VER}
USER root
RUN set -eux; export DEBIAN_FRONTEND=noninteractive; \\
    apt-get update -qq; apt-get install -y -qq --no-install-recommends gdb binutils; \\
    (apt-get install -y -qq --no-install-recommends ubuntu-dbgsym-keyring || true); \\
    . /etc/os-release; \\
    printf 'deb http://ddebs.ubuntu.com %s main restricted universe multiverse\\n' \\
      "\$VERSION_CODENAME" > /etc/apt/sources.list.d/ddebs.list; \\
    apt-get update -qq 2>/dev/null || true; \\
    for p in mariadb-server-core-dbgsym mariadb-server-dbgsym; do \\
      apt-get install -y -qq --no-install-recommends "\$p" 2>/dev/null \\
        && { echo "INSTALLED: \$p"; touch /tmp/.ok; }; done; \\
    [ -f /tmp/.ok ] && echo '=== debug symbols installed ===' \\
                    || echo '=== no debug symbols; gdb on a core still beats the built-in trace ==='; \\
    apt-get clean; rm -rf /var/lib/apt/lists/*
RUN mkdir -p /cores && chmod 1777 /cores
USER mysql
DOCKERFILE
}

boot() {
  local extra=(); [ "$DEBUG" = 1 ] && extra=( --ulimit core=-1 -v "$COREDIR":/cores )
  docker run -d --name "$N" -e MARIADB_ROOT_PASSWORD="$PW" \
    -v "$DATADIR":/var/lib/mysql -v "$LOGDIR":/logs "${extra[@]}" \
    "$IMG" \
    --log-bin --binlog-storage-engine=innodb \
    --max-binlog-size=$((BINLOG_MB*1024*1024)) \
    --binlog-format=ROW --binlog-row-image=FULL \
    --sync-binlog=1 --binlog-checksum=CRC32 \
    $([ "$DEBUG" = 1 ] && echo --core-file) \
    --binlog-cache-size="$BCACHE" --binlog-stmt-cache-size="$BCACHE" \
    --max-statement-time="${BOOT_MST:-0}" \
    --max-allowed-packet=256M --innodb-buffer-pool-size=64M \
    --tmpdir=/tmp --log-error=/logs/error-$ITN.log >/dev/null 2>&1 || return 1
  printf '    server '
  local i
  for i in $(seq 1 90); do
    docker exec "$N" mariadb -uroot -p"$PW" -e "SELECT 1" >/dev/null 2>&1 && { echo 'up'; return 0; }
    alive || { echo; bad "container died while starting"
               tail -12 "$LOGDIR/error-$ITN.log" 2>/dev/null | sed 's/^/      /'; return 1; }
    printf '.'; sleep 2
  done
  echo; bad "no connection"; return 1
}

restart_same() {
  docker start "$N" >/dev/null 2>&1
  printf '    restarting '
  local i
  for i in $(seq 1 60); do
    docker exec "$N" mariadb -uroot -p"$PW" -e "SELECT 1" >/dev/null 2>&1 && { echo 'up'; return 0; }
    if grep -qiE "Aborting" "$LOGDIR/error-$ITN.log" 2>/dev/null; then
      echo; bad "the server will not start:"
      grep -iE "ERROR|Aborting|corrupt|in the future|insufficient|space" \
        "$LOGDIR/error-$ITN.log" | tail -10 | sed 's/^/      /'
      return 1
    fi
    printf '.'; sleep 2
  done
  echo; bad "did not come back"; return 1
}

fill_to() {
  local target="$1" g=0 p f s; s="$(qn -e 'SHOW MASTER STATUS' | awk '{print $1}')"
  while :; do
    p="$(qn -e 'SHOW MASTER STATUS' | awk '{print $2}')"
    f="$(qn -e 'SHOW MASTER STATUS' | awk '{print $1}')"
    [ -z "$p" ] && { echo; return 1; }
    [ "$f" != "$s" ] && { echo; return 0; }
    [ "$p" -ge "$target" ] 2>/dev/null && { echo; return 0; }
    q -e "UPDATE t.pump SET pad=REPEAT(CHAR(65+FLOOR(RAND()*26)),${BLOB_MB}*1048576)
          WHERE id=1" >/dev/null 2>&1 || { echo; return 1; }
    g=$((g+1)); printf '\r%s    %s %s -> %s%-15s%s' "$D" "$f" "$p" "$target" "" "$Z"
    [ "$g" -gt 60 ] && { echo; return 1; }
  done
}

stream_check() {
  dex bash -c "
    log=\$(mariadb -uroot -p$PW -N -B -e 'SHOW BINARY LOGS' | tail -1 | cut -f1)
    gtid=\$(mariadb -uroot -p$PW -N -B -e 'SELECT @@gtid_binlog_state')
    echo \"      newest=\$log  gtid=\$gtid\"
    for how in file gtid; do
      if [ \$how = file ]; then
        out=\$(mariadb-binlog --read-from-remote-server --host=127.0.0.1 --user=root \
                --password=$PW --verify-binlog-checksum \"\$log\" 2>&1)
      else
        [ -n \"\$gtid\" ] || continue
        out=\$(mariadb-binlog --read-from-remote-server --host=127.0.0.1 --user=root \
                --password=$PW --verify-binlog-checksum --start-position=\"\$gtid\" \"\$log\" 2>&1)
      fi
      echo \"      [\$how] Xid events: \$(printf '%s' \"\$out\" | grep -c 'Xid =')\"
      printf '%s' \"\$out\" | grep -iE 'ERROR|truncat|corrupt' | head -2 | sed 's/^/      /'
    done
    true"
}

########################################################################
do_run() {
  guard
  trap cleanup_all EXIT
  mkdir -p "$LOGDIR"; chmod 777 "$LOGDIR"

  if [ "$DEBUG" = 1 ]; then
    mkdir -p "$COREDIR"; chmod 777 "$COREDIR"
    build_debug_image || exit 1
    OLD_PATTERN="$(cat /proc/sys/kernel/core_pattern)"
    $SUDO sysctl -q -w kernel.core_pattern='/cores/core.%e.%p'
    note "core_pattern: $OLD_PATTERN  ->  $(cat /proc/sys/kernel/core_pattern)"
  fi

  say "T9 on MariaDB $VER"
  printf '    binlog_cache_size : %s%s%s\n' "$G" "$BCACHE" "$Z"
  printf '    binlog_stmt_cache : %s\n' "$BCACHE"
  printf '    max_binlog_size   : %s MB   (.ibb size)\n' "$BINLOG_MB"
  printf '    free space left   : %s MB   (room to write, none for a new .ibb)\n' "$KEEP_MB"
  printf '    iterations        : %s\n' "$ITER"
  printf '    core capture      : %s\n' "$([ "$DEBUG" = 1 ] && echo yes || echo no)"
  note "crash rate tracks binlog_cache_size: 32K gave 0/10, 256M gave 1/5,"
  note "and earlier runs of the same shape gave 3/4 and 4/5"

  local crashed=0 survived=0 rolled_back=0 committed_anyway=0 unexaminable=0

  for ITN in $(seq 1 "$ITER"); do
    say "iteration $ITN/$ITER"
    docker rm -f "$N" >/dev/null 2>&1 || true
    wipe_datadir
    : > "$LOGDIR/error-$ITN.log"; chmod 666 "$LOGDIR/error-$ITN.log"
    boot || { unexaminable=$((unexaminable+1)); continue; }
    q -e "SELECT @@binlog_cache_size, @@max_binlog_size, @@binlog_storage_engine\G" 2>&1 | sed 's/^/      /'

    q -e "INSTALL SONAME 'ha_sequence'" >/dev/null 2>&1
    q -e "CREATE DATABASE IF NOT EXISTS t;
          CREATE TABLE t.pump (id INT PRIMARY KEY, pad LONGBLOB) ENGINE=InnoDB;
          CREATE TABLE t.log (id INT AUTO_INCREMENT PRIMARY KEY, note VARCHAR(64)) ENGINE=InnoDB;
          INSERT INTO t.pump VALUES (1, REPEAT('a',${BLOB_MB}*1048576));" >/dev/null 2>&1

    note "consuming the preallocated spare"
    fill_to $((BINLOG_MB*1024*1024 - 2*1024*1024)) || true
    note "at $(pos)"

    note "filling the datadir, leaving ${KEEP_MB} MB"
    dex bash -c "
      free=\$(df -B1 --output=avail /var/lib/mysql | tail -1)
      sz=\$((free - ${KEEP_MB}*1024*1024)); [ \"\$sz\" -lt 0 ] && sz=0
      echo \"      writing \$((sz/1048576)) MB of filler\"
      dd if=/dev/zero of=/var/lib/mysql/filler bs=1M count=\$((sz/1048576)) status=none
      df -h /var/lib/mysql | tail -1 | sed 's/^/      /'" 2>/dev/null

    note "approaching the boundary"
    fill_to $((BINLOG_MB*1024*1024 - 6*1024*1024)) || true

    local BEFORE_MARKERS BEFORE_GTID BEFORE_POS RC
    BEFORE_MARKERS="$(qn -e 'SELECT COUNT(*) FROM t.log')"
    BEFORE_GTID="$(qn -e 'SELECT @@gtid_binlog_state')"
    BEFORE_POS="$(pos)"
    note "before: markers=$BEFORE_MARKERS gtid=$BEFORE_GTID pos=$BEFORE_POS"

    note "crossing transaction, marker 'doomed-$ITN'"
    q -e "BEGIN;
          UPDATE t.pump SET pad=REPEAT('X',${BLOB_MB}*1048576) WHERE id=1;
          UPDATE t.pump SET pad=REPEAT('Y',${BLOB_MB}*1048576) WHERE id=1;
          INSERT INTO t.log (note) VALUES ('doomed-$ITN');
          COMMIT;
          SELECT 'COMMIT RETURNED OK' AS verdict;" >/tmp/t9-$ITN.log 2>&1
    RC=$?
    sleep 5

    if alive; then
      survived=$((survived+1))
      printf '    %sSURVIVED%s (healthy path), client rc=%s\n' "$G" "$Z" "$RC"
      grep -q "COMMIT RETURNED OK" /tmp/t9-$ITN.log \
        && note "the client was told the commit succeeded" \
        || { note "client output:"; sed 's/^/      /' /tmp/t9-$ITN.log; }
      note "markers now $(qn -e 'SELECT COUNT(*) FROM t.log'), at $(pos)"
      grep -iE "Unable to allocate" "$LOGDIR/error-$ITN.log" | tail -1 | sed 's/^/      /'
      continue
    fi

    crashed=$((crashed+1))
    bad "CRASHED exit=$(docker inspect -f '{{.State.ExitCode}}' "$N" 2>/dev/null) $(grep -oE 'got signal [0-9]+' "$LOGDIR/error-$ITN.log" | tail -1)"
    note "client rc=$RC; the COMMIT never returned"

    note "freeing space on the host before restarting"
    free_space
    restart_same || { unexaminable=$((unexaminable+1)); continue; }

    note "recovery messages:"
    grep -iE "recover|rollback|binlog|truncat|InnoDB: " "$LOGDIR/error-$ITN.log" | tail -10 | sed 's/^/      /'

    local AFTER_MARKERS AFTER_GTID AFTER_LIST
    AFTER_MARKERS="$(qn -e 'SELECT COUNT(*) FROM t.log')"
    AFTER_GTID="$(qn -e 'SELECT @@gtid_binlog_state')"
    AFTER_LIST="$(qn -e "SELECT IFNULL(GROUP_CONCAT(note),'(none)') FROM t.log")"

    say "verdict for iteration $ITN"
    printf '      markers  before=%s  after=%s\n' "$BEFORE_MARKERS" "$AFTER_MARKERS"
    printf '      gtid     before=%s  after=%s\n' "$BEFORE_GTID" "$AFTER_GTID"
    printf '      position before=%s  after=%s\n' "$BEFORE_POS" "$(pos)"
    printf '      markers present: %s\n' "$AFTER_LIST"
    if [ "$BEFORE_MARKERS" = "$AFTER_MARKERS" ]; then
      printf '      %sROLLED BACK%s - doomed-%s is absent\n' "$G" "$Z" "$ITN"
      rolled_back=$((rolled_back+1))
    else
      printf '      %sCOMMITTED ANYWAY%s - doomed-%s survived its own crash\n' "$Y" "$Z" "$ITN"
      committed_anyway=$((committed_anyway+1))
    fi

    note "binlog stream:"; stream_check
    note "can it write again?"
    q -e "INSERT INTO t.log (note) VALUES ('post-restart-$ITN'); SELECT 'WRITE OK' AS p" \
      2>&1 | tail -2 | sed 's/^/      /'

    [ "$DEBUG" = 1 ] && break
  done

  say "summary"
  printf '  binlog_cache_size : %s\n' "$BCACHE"
  printf '  iterations        : %s\n' "$ITER"
  printf '  crashed           : %s\n' "$crashed"
  printf '  survived          : %s\n' "$survived"
  printf '  unexaminable      : %s\n' "$unexaminable"
  printf '  after a crash: rolled back = %s, committed anyway = %s\n' "$rolled_back" "$committed_anyway"
  note "logs: $LOGDIR"

  if [ "$DEBUG" = 1 ]; then
    say "cores"; ls -lh "$COREDIR" 2>/dev/null | sed 's/^/  /'
    local CORE; CORE="$(ls -t "$COREDIR"/core.* 2>/dev/null | head -1)"
    if [ -n "$CORE" ]; then
      say "backtrace from $(basename "$CORE")"
      docker run --rm -v "$COREDIR":/cores --entrypoint gdb "$IMG" -q -batch \
        -ex 'set pagination off' -ex 'thread apply all bt full' \
        /usr/sbin/mariadbd "/cores/$(basename "$CORE")" 2>&1 \
        | tee "$HERE/t9-backtrace.txt" | tail -30
      note "full trace in $HERE/t9-backtrace.txt"
    else
      warn "no core written; check /proc/sys/kernel/core_pattern"
    fi
  fi

  cat <<'EOF'

HOW TO READ IT

  ROLLED BACK everywhere, binlog readable, writes resume
    Availability bug, nothing more. The transaction is gone from both the
    table and the binlog, consistently. Expected: the crash happens while
    writing the serialisation history, before it completes, and the binlog
    lives in the same engine, so the two cannot disagree.

  COMMITTED ANYWAY
    The transaction survived the crash that occurred during its own commit.
    Check whether the binlog carries it too - table yes, stream no means
    silent replica divergence, and the ticket becomes a data-loss ticket.

  Xid count disagreeing with the markers, or a read error on the stream
    Then the engine binlog does not prevent a torn stream after all, and
    MDEV-41194's failure mode has an analogue here.

  SERVER WILL NOT RESTART even after space is freed
    A separate third finding. Rule out this test's own redo-log pressure
    first by grepping for "insufficient innodb_log_file_size".

  NOTHING CRASHES
    Raise --binlog-cache-size or the iteration count.
EOF
}


########################################################################
# exhaust: keep writing until the spare .ibb is full while no replacement
# can be allocated, so a USER SESSION has to wait for a file that cannot be
# created. Then see whether max_statement_time breaks that wait, and what it
# leaves behind.
########################################################################
timed_write() {          # $1 = marker; prints: rc seconds, writes /tmp/t9x-$1.log
  local tag="$1" t0 t1 rc
  t0=$(date +%s)
  ( q -e "SET SESSION max_statement_time=$MST;
          BEGIN;
          UPDATE t.pump SET pad=REPEAT(CHAR(65+FLOOR(RAND()*26)),${BLOB_MB}*1048576) WHERE id=1;
          INSERT INTO t.log (note) VALUES ('$tag');
          COMMIT;
          SELECT 'COMMIT RETURNED OK' AS verdict;" >/tmp/t9x-$tag.log 2>&1 ) &
  local p=$! waited=0
  while kill -0 $p 2>/dev/null; do
    sleep 2; waited=$((waited+2))
    if [ $((waited % 10)) -eq 0 ]; then
      printf '%s      t+%02ds  %s%s\n' "$D" "$waited" \
        "$(docker exec "$N" mariadb -uroot -p"$PW" -N -B \
            -e "SELECT CONCAT(ID,':',TIME,'s:',IFNULL(STATE,'-'))
                FROM information_schema.PROCESSLIST
                WHERE COMMAND<>'Sleep' AND ID<>CONNECTION_ID()" 2>/dev/null | tr '\n' ' ')" "$Z"
    fi
    alive || break
    [ "$waited" -gt 180 ] && { warn "still running after 180s, moving on"; break; }
  done
  wait $p 2>/dev/null; rc=$?
  t1=$(date +%s)
  echo "$rc $((t1-t0))"
}

do_exhaust() {
  guard
  trap cleanup_all EXIT
  mkdir -p "$LOGDIR"; chmod 777 "$LOGDIR"
  ITN=exhaust
  BOOT_MST="$MST"

  say "T9 exhaust - no next .ibb, so a user session must wait"
  printf '    binlog_cache_size   : %s\n' "$BCACHE"
  printf '    max_binlog_size     : %s MB   (.ibb size)\n' "$BINLOG_MB"
  printf '    max_statement_time  : %s%s s%s   (retry interval is 60 s)\n' "$G" "$MST" "$Z"
  printf '    free space left     : %s MB\n' "$KEEP_MB"

  docker rm -f "$N" >/dev/null 2>&1 || true
  wipe_datadir
  : > "$LOGDIR/error-$ITN.log"; chmod 666 "$LOGDIR/error-$ITN.log"
  boot || return 1
  q -e "SELECT @@max_statement_time, @@binlog_cache_size, @@max_binlog_size\G" 2>&1 | sed 's/^/      /'

  q -e "INSTALL SONAME 'ha_sequence'" >/dev/null 2>&1
  q -e "CREATE DATABASE IF NOT EXISTS t;
        CREATE TABLE t.pump (id INT PRIMARY KEY, pad LONGBLOB) ENGINE=InnoDB;
        CREATE TABLE t.log (id INT AUTO_INCREMENT PRIMARY KEY, note VARCHAR(64)) ENGINE=InnoDB;
        INSERT INTO t.pump VALUES (1, REPEAT('a',${BLOB_MB}*1048576));" >/dev/null 2>&1

  note "consuming the preallocated spare"
  BOOT_MST_SAVE="$MST"; MST=0           # setup writes must not time out
  fill_to $((BINLOG_MB*1024*1024 - 2*1024*1024)) || true
  note "at $(pos)"

  note "filling the datadir, leaving ${KEEP_MB} MB"
  dex bash -c "
    free=\$(df -B1 --output=avail /var/lib/mysql | tail -1)
    sz=\$((free - ${KEEP_MB}*1024*1024)); [ \"\$sz\" -lt 0 ] && sz=0
    echo \"      writing \$((sz/1048576)) MB of filler\"
    dd if=/dev/zero of=/var/lib/mysql/filler bs=1M count=\$((sz/1048576)) status=none
    df -h /var/lib/mysql | tail -1 | sed 's/^/      /'" 2>/dev/null

  note "crossing into the spare, which makes the allocation of the next one fail"
  fill_to $((BINLOG_MB*1024*1024 - 4*1024*1024)) || true
  q -e "SET SESSION max_statement_time=0;
        UPDATE t.pump SET pad=REPEAT('C',${BLOB_MB}*1048576) WHERE id=1;
        UPDATE t.pump SET pad=REPEAT('D',${BLOB_MB}*1048576) WHERE id=1;" >/dev/null 2>&1
  MST="$BOOT_MST_SAVE"
  note "now at $(pos)"
  if grep -qi "Unable to allocate" "$LOGDIR/error-$ITN.log"; then
    grep -i "Unable to allocate" "$LOGDIR/error-$ITN.log" | tail -1 | sed 's/^/      /'
    note "good: no replacement .ibb can be created"
  else
    warn "no allocation failure logged yet - the premise is not established."
    warn "Lower KEEP_MB or raise the fill, otherwise what follows means nothing."
  fi
  dex bash -c 'ls -la /var/lib/mysql/*.ibb | sed "s/^/      /"'

  say "writing on, until the spare itself is exhausted"
  note "each write is timed; the one that needs a file which cannot exist is the test"
  local BEFORE_MARKERS BEFORE_GTID
  BEFORE_MARKERS="$(qn -e 'SELECT COUNT(*) FROM t.log')"
  BEFORE_GTID="$(qn -e 'SELECT @@gtid_binlog_state')"
  note "before: markers=$BEFORE_MARKERS gtid=$BEFORE_GTID"

  local i out rc secs blocked=0
  for i in $(seq 1 8); do
    printf '    write %d  at %s\n' "$i" "$(pos)"
    out="$(timed_write "w$i")"; rc="${out%% *}"; secs="${out##* }"
    if [ "$secs" -ge $((MST > 5 ? MST - 5 : 3)) ]; then blocked=$((blocked+1)); fi
    printf '      -> rc=%s after %ss : %s\n' "$rc" "$secs" \
      "$(head -3 /tmp/t9x-w$i.log | tr '\n' ' ' | cut -c1-110)"
    alive || { bad "the server went down during write $i"; break; }
  done

  say "state after the writes"
  alive && {
    q -e "SHOW MASTER STATUS" 2>&1 | sed 's/^/      /'
    note "markers: $(qn -e 'SELECT COUNT(*) FROM t.log') (before: $BEFORE_MARKERS)"
    note "gtid   : $(qn -e 'SELECT @@gtid_binlog_state') (before: $BEFORE_GTID)"
    note "present: $(qn -e "SELECT IFNULL(GROUP_CONCAT(note),'(none)') FROM t.log")"
  }
  note "allocation retries logged:"
  grep -ci "Unable to allocate" "$LOGDIR/error-$ITN.log" | sed 's/^/      /'
  grep -iE "got signal|Waiting 60|Unable to allocate|ERROR" "$LOGDIR/error-$ITN.log" | tail -8 | sed 's/^/      /'
  q -e "SHOW GLOBAL STATUS LIKE 'Max_statement_time_exceeded'" 2>&1 | sed 's/^/      /'

  say "binlog stream while still short of space"
  alive && stream_check

  say "freeing space - does it recover?"
  free_space
  sleep 8
  if alive; then
    q -e "INSERT INTO t.log (note) VALUES ('after-free'); SELECT 'WRITE OK' AS p" 2>&1 | tail -2 | sed 's/^/      /'
    note "at $(pos)"
    dex bash -c 'ls -la /var/lib/mysql/*.ibb | sed "s/^/      /"'
    stream_check
  else
    bad "server is down; restarting"
    restart_same && { q -e "SELECT COUNT(*) AS markers FROM t.log" 2>&1 | sed 's/^/      /'; stream_check; }
  fi

  cat <<'XEOF'

WHAT TO LOOK FOR

  THE PREMISE
    "Unable to allocate file ./binlog-00000N.ibb" must appear before the
    timed writes start. Without it there is still a spare and nothing can
    block; the run says nothing.

  THE WRITE THAT BLOCKS
    Early writes should finish in a second or two. The one that needs a file
    which cannot be created is the interesting one. Three outcomes:

      a) it blocks and then fails at roughly max_statement_time
         -> the timeout breaks a wait on binlog allocation, the same shape
            as MDEV-41194. Then check the stream and the marker count: if
            anything is torn or missing, 12.3 has the same class of problem.

      b) it blocks for the full 60s retry and then succeeds or fails cleanly
         -> the wait is not interruptible by a statement timeout. Good.

      c) it fails immediately with a clear error
         -> best case: no wait at all, nothing to interrupt.

  AFTERWARDS
    Marker count against Xid count in the stream. A marker in the table that
    is missing from the binlog is silent divergence and the worst outcome.

  RECOVERY
    After freeing space, a new .ibb should appear and writes resume. If the
    server needs a restart, or will not restart, that is a separate finding.
XEOF
}

case "$CMD" in
  setup)    [ "$(id -u)" = 0 ] || { echo "setup needs root: sudo $0 setup"; exit 1; }; do_setup ;;
  teardown) [ "$(id -u)" = 0 ] || { echo "teardown needs root: sudo $0 teardown"; exit 1; }; do_teardown ;;
  run)      do_run ;;
  exhaust)  do_exhaust ;;
  *) sed -n '/^# Usage/,/^#$/p' "$0" | sed 's/^# \{0,1\}//' ;;
esac