qa: add fd-guard, the close() guard and regtest harness that found the double close

shim/ is an LD_PRELOAD guard around close/fclose/dup2/dup3:
- before each close it resolves /proc/self/fd/N;
- it reports, with a backtrace, a non-owner close of wallet.dat,
  database/log.* or blocks/*.dat, and every close that fails with EBADF
  (the second half of a double close);
- FDGUARD_ABORT=1 aborts on the first hit;
- run_tests.sh self-tests it (51 checks).

repro/ runs a shimmed dragonxd on isolated regtest. Each run gets its
own unprivileged network namespace (loopback only). The loads are:
- rpc: a hammer with impatient/RST/half-open clients, work-queue
  bursts and long-polls;
- p2p: TLS peers churned by disconnectnode, ban and kill, plus a raw
  TCP/TLS misbehaver;
- tor: a fake control port;
- mine: mining with blocknotify and autoshield.
tools/tally.py groups the reports by closer. tools/scan_bdb.py finds BDB
pages written into block files.

With the wolfSSL CloseSocket macro in effect:
- p2pmin (300 s) gives one EBADF double close per disconnect;
- the full load (1200 s) gives 930 EBADF closes, 186 non-owner closes
  (37 of wallet/log/blk files) and a 632 KB db.log with BDB0060 PANIC;
- RPC-only and Tor-only loads give none.
After the fix, every count is 0 and db.log stays empty. The README has
the commands and the numbers.

Paths come from the environment (GOLDEN, BIN, RUNS, SHIM, ASMAP and the
Sapling params). Run output goes to $RUNS, outside the tree.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

qa/fd-guard/.gitignore

@@ -0,0 +1,4 @@
+shim/libfdguard.so
+shim/fdguard_test
+*.o
+shim/t/

qa/fd-guard/README.md

@@ -0,0 +1,55 @@
+# fd-guard: catch stray close() of files a thread doesn't own
+
+This is the harness that found the wolfSSL `CloseSocket` macro bug. With that bug, P2P disconnects closed their fd number a second time, after another thread had reused it, and BDB then wrote wallet pages into `blocks/*.dat`.
+
+It has two parts:
+
+- `shim/`: an `LD_PRELOAD` guard around `close`, `fclose`, `dup2` and `dup3`.
+  - Before each close it resolves `/proc/self/fd/N`.
+  - It logs a backtrace when a non-owner closes `wallet.dat`, `database/log.*` or `blocks/*.dat`, and for every close that fails with `EBADF` (the second half of a double close).
+  - `FDGUARD_ABORT=1` makes the first hit abort.
+- `repro/`: isolated regtest runs that put a shimmed `dragonxd` under load:
+  - **rpc:** impatient and RST clients, half-open connections, work-queue bursts and long-polls;
+  - **p2p:** TLS peers churned by `disconnectnode`/ban/kill, plus misbehaving raw TCP/TLS clients;
+  - **tor:** a fake control port;
+  - **mining:** with `blocknotify` and autoshield.
+
+  Each run executes in its own unprivileged network namespace: loopback only, no route.
+
+## Requirements
+
+- Linux.
+- `unshare -rn` (unprivileged user namespaces), `nsenter`, `ip`, python3, gcc, and `addr2line` for symbolizing.
+- A `dragonxd`/`dragonx-cli` built with `-g`. Adding `-rdynamic` makes backtraces readable without `addr2line`.
+- A regtest datadir with a grown wallet to copy from (`GOLDEN`). It is never modified.
+
+## Running
+
+```sh
+qa/fd-guard/shim/build.sh                      # builds libfdguard.so
+qa/fd-guard/shim/run_tests.sh                  # self-test: double close, foreign close of a guarded file
+cd qa/fd-guard/repro
+export GOLDEN=/path/to/regtest/datadir BIN=/path/to/build/src   # RUNS defaults to ~/fd-guard-runs
+LOADS=p2pmin,mine DISC_INT=5 GUARD=all ./go.sh p2pmin 300      # 1 peer, disconnectnode every 5 s
+LOADS=rpc,p2p,tor,mine WQ=64 RT=8 GUARD=all ./go.sh full 1200  # everything, production-like work queue
+python3 tools/tally.py "$RUNS/<name>/fdguard.log"              # who closed what
+python3 ../shim/symbolize.py --summary "$RUNS/<name>/fdguard.log"
+python3 tools/scan_bdb.py "$RUNS/<name>"/target/regtest/blocks/*.dat   # BDB pages inside block files
+```
+
+Each run writes `run.txt` (full log plus summary), `fdguard.log` and the target's `debug.log` and `db.log` under `$RUNS/<name>/`.
+
+## What to expect
+
+Measured on a 2041-note regtest wallet, before and after the fix (`net: stop wolfSSL's CloseSocket macro ...`), under the same loads:
+
+| Run | Binary | EBADF closes | Non-owner closes (wallet/log/blk) | libevent `Epoll` failures | `db.log` |
+|---|---|---|---|---|---|
+| `p2pmin`, 300 s | before | 51 (≈ 1 per disconnect) | 1 | 0 | empty |
+| | after | 0 | 0 | 0 | empty |
+| `full`, 1200 s, `WQ=64 RT=8` | before | 930 | 186 (37) | 111 | 632 KB: 7500 × `BDB0060 PANIC`, `write failed for page N` |
+| | after | 0 | 0 | 0 | empty |
+
+- RPC-only and Tor-only loads give 0 reports, even before the fix: P2P disconnects are the only origin.
+- `scan_bdb.py` found no BDB pages in block files in any run. The landing of a page in a block file needs the reused number to be a writable block file at the moment BDB writes. Production did hit that (blk00011/blk00020), but it is rare.
+- `SSL_write ... closing connection` lines are expected under `tcpchurn`. They are how the load triggers disconnects, not a symptom.

qa/fd-guard/repro/common.sh

@@ -0,0 +1,72 @@
+# shellcheck shell=bash
+# Shared config + helpers for the fd-double-close REPRODUCTION harness.
+# Regtest only. Meant to run INSIDE an unprivileged network namespace (see ns.sh),
+# so every port is private to the namespace and nothing can reach a real network.
+set +e
+
+REPRO=${REPRO:-$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)}
+SRCROOT=$(cd "$REPRO/../../.." && pwd)
+BIN=${BIN:-$SRCROOT/src}                    # dir with dragonxd + dragonx-cli (build with -g -rdynamic for backtraces)
+DRAGONXD=$BIN/dragonxd
+CLI=$BIN/dragonx-cli
+RUNS=${RUNS:-$HOME/fd-guard-runs}           # run output (block files, logs): keep it out of the repo
+GOLDEN=${GOLDEN:?set GOLDEN to a regtest datadir with a grown wallet (it is only ever copied)}
+ASMAP=${ASMAP:-$SRCROOT/asmap.dat}
+SP=${SP:-$HOME/.zcash-params/sapling-spend.params}
+OP=${OP:-$HOME/.zcash-params/sapling-output.params}
+SHIM=${SHIM:-$REPRO/../shim/libfdguard.so}
+
+RPCUSER=ab
+RPCPASS=abpass
+RPC_T=18901 ; P2P_T=18911
+RPC_H1=18902; P2P_H1=18912
+RPC_H2=18903; P2P_H2=18913
+TORPORT=18950
+
+clin() { local dd=$1 rp=$2; shift 2; "$CLI" -regtest -datadir="$dd" -rpcuser=$RPCUSER -rpcpassword=$RPCPASS -rpcport=$rp "$@"; }
+
+write_conf() { printf 'rpcuser=%s\nrpcpassword=%s\nrpcport=%s\n' "$RPCUSER" "$RPCPASS" "$2" > "$1/DRAGONX.conf"; }
+
+# prep_golden <dd> : copy of the golden wallet datadir (2041 notes, txindex=1, height 286)
+prep_golden() {
+  local dd=$1
+  rm -rf "$dd"; cp -a "$GOLDEN" "$dd"
+  cp -f "$ASMAP" "$dd/asmap.dat"; cp -f "$SP" "$dd/sapling-spend.params"; cp -f "$OP" "$dd/sapling-output.params"
+  write_conf "$dd" "$RPC_T"
+  : > "$dd/regtest/debug.log"; rm -f "$dd/regtest/db.log"
+}
+
+# prep_fresh <dd> <rpcport> : empty regtest datadir (helpers)
+prep_fresh() {
+  local dd=$1 rp=$2
+  rm -rf "$dd"; mkdir -p "$dd"
+  cp -f "$ASMAP" "$dd/asmap.dat"; cp -f "$SP" "$dd/sapling-spend.params"; cp -f "$OP" "$dd/sapling-output.params"
+  write_conf "$dd" "$rp"
+}
+
+start_helper() {   # <dd> <p2p> <rpc> <logfile>
+  local dd=$1 p2p=$2 rp=$3 lg=$4
+  ( cd "$dd" && exec nice -n 10 "$DRAGONXD" -regtest -datadir="$dd" -testnode=1 -checkpoints=0 \
+      -listen=1 -bind=127.0.0.1 -port=$p2p -connect=127.0.0.1:1 -addnode=127.0.0.1:$P2P_T \
+      -dnsseed=0 -dns=0 -asmap="$dd/asmap.dat" -autoshield=0 \
+      -rpcuser=$RPCUSER -rpcpassword=$RPCPASS -rpcport=$rp -listenonion=0 -maxinboundfromip=100 \
+      -debug=net -debug=tls -daemon ) >> "$lg" 2>&1
+}
+
+waitrpc() {        # <dd> <rpc> [secs]
+  local dd=$1 rp=$2 n=${3:-300} i
+  for i in $(seq 1 "$n"); do clin "$dd" "$rp" getblockcount >/dev/null 2>&1 && return 0; sleep 1; done
+  return 1
+}
+
+stop_daemon() {    # <dd> <rpc>   (helpers, started with -daemon)
+  local dd=$1 rp=$2 pid i
+  pid=$(cat "$dd/regtest/hushd.pid" 2>/dev/null)
+  clin "$dd" "$rp" stop >/dev/null 2>&1
+  for i in $(seq 1 90); do { [ -z "$pid" ] || ! kill -0 "$pid" 2>/dev/null; } && return 0; sleep 1; done
+  [ -n "$pid" ] && kill -9 "$pid" 2>/dev/null
+}
+
+MARKERS=('libevent: Epoll' 'Socket close failed' 'Bad file descriptor' 'Operation not permitted' 'Illegal seek'
+         'SSL_write' 'Unable to seek' 'socket select error' 'write failed for page' 'PANIC' 'error scanning DB'
+         'terminate' 'Assertion' 'EXCEPTION')

qa/fd-guard/repro/go.sh

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+# go.sh <name> <seconds>  -- create a private network namespace (loopback only, no route),
+# run run.sh inside it, then tear the namespace down. Env knobs pass through to run.sh.
+export REPRO=${REPRO:-$(cd "$(dirname "$0")" && pwd)}
+export RUNS=${RUNS:-$HOME/fd-guard-runs}
+OUT=$RUNS/$1; mkdir -p "$OUT"
+setsid unshare -rn sh -c "ip link set lo up && echo \$\$ > $OUT/ns.pid && exec sleep infinity" </dev/null >/dev/null 2>&1 &
+for _ in 1 2 3 4 5 6 7 8 9 10; do [ -s "$OUT/ns.pid" ] && break; sleep 0.3; done
+NSP=$(cat "$OUT/ns.pid")
+cp "$REPRO/run.sh" "$OUT/run.sh.snapshot"   # run a frozen copy: editing run.sh cannot disturb a live run
+nsenter -t "$NSP" -U -n --preserve-credentials bash "$OUT/run.sh.snapshot" "$@" > "$OUT/go.txt" 2>&1
+rc=$?
+kill "$NSP" 2>/dev/null
+echo "go.sh: run.sh rc=$rc, namespace holder $NSP killed" >> "$OUT/go.txt"

qa/fd-guard/repro/run.sh

@@ -0,0 +1,156 @@
+#!/usr/bin/env bash
+# run.sh <name> <seconds>   -- run INSIDE the network namespace (go.sh does that).
+# Env knobs:
+#   LOADS=rpc,p2p,tor,mine   which load generators run (default all four); p2pmin = 1 helper + disconnectnode every DISC_INT s
+#   WQ=4 RT=2                -rpcworkqueue / -rpcthreads of the target
+#   GUARD=all                FDGUARD_GUARD (db,blk | all)
+#   SHIMON=1                 run the target under LD_PRELOAD=libfdguard.so
+#   STRACE=0                 1 = also wrap the target in strace (failed fd syscalls only)
+#   TRACE=0                  1 = FDGUARD_TRACE_ALL (one line per open/close; big)
+#   MINE_INT=3               seconds between generate 1
+#   HAMMER_ARGS=...          override the hammer thread mix
+#   STOP_ON_HIT=0            1 = end the run 60 s after the first NON-OWNER db/blk close
+#   EXTRA=...                extra target args
+source "$REPRO/common.sh"
+NAME=$1; DUR=${2:-600}
+OUT=$RUNS/$NAME
+LOADS=${LOADS:-rpc,p2p,tor,mine}; WQ=${WQ:-4}; RT=${RT:-2}; GUARD=${GUARD:-all}
+SHIMON=${SHIMON:-1}; STRACE=${STRACE:-0}; TRACE=${TRACE:-0}; MINE_INT=${MINE_INT:-3}
+STOP_ON_HIT=${STOP_ON_HIT:-0}
+HAMMER_ARGS=${HAMMER_ARGS:---normal 8 --impatient 8 --slowfire 2 --halfopen 3 --pipeline 3 --garbage 2 --slowread 2 --longpoll 4 --burst-size 16 --burst-every 2}
+has() { case ",$LOADS," in *",$1,"*) return 0;; esac; return 1; }
+
+mkdir -p "$OUT/logs"; rm -f "$OUT/STOP"
+DD_T=$OUT/target; DD_H1=$OUT/helper1; DD_H2=$OUT/helper2
+exec > >(tee -a "$OUT/run.txt") 2>&1
+echo "=== repro run $NAME  $(date -u +%FT%TZ)  dur=${DUR}s loads=$LOADS wq=$WQ rt=$RT guard=$GUARD shim=$SHIMON strace=$STRACE"
+echo "netns check (must be ENETUNREACH=101): $(python3 -c 'import socket;print(socket.socket().connect_ex(("1.1.1.1",80)))')"
+echo "binary: $DRAGONXD  build-id: $(readelf -n "$DRAGONXD" | awk '/Build ID/{print $3}')"
+echo "dragonxd already running in this namespace? $(pgrep -a -x dragonxd | wc -l) (other agents' nodes live in other namespaces)"
+
+PIDS=()
+cleanup() {
+  echo "--- cleanup $(date -u +%T)"
+  touch "$OUT/STOP"
+  for p in "${PIDS[@]}"; do kill "$p" 2>/dev/null; done
+  stop_daemon "$DD_H2" "$RPC_H2"; stop_daemon "$DD_H1" "$RPC_H1"
+  if [ -n "$TPID" ] && kill -0 "$TPID" 2>/dev/null; then
+    timeout 20 "$CLI" -regtest -datadir="$DD_T" -rpcuser=$RPCUSER -rpcpassword=$RPCPASS -rpcport=$RPC_T stop >/dev/null 2>&1
+    for _ in $(seq 1 60); do kill -0 "$TPID" 2>/dev/null || break; sleep 1; done
+    # abandoned longpoll getblocktemplate calls can hold every -rpcthreads worker, so RPC stop may never run:
+    # fall back to SIGTERM (same clean StartShutdown path), SIGKILL only as a last resort
+    kill -0 "$TPID" 2>/dev/null && { echo "RPC stop did not finish in 60 s -> SIGTERM"; kill -TERM "$TPID"; }
+    for _ in $(seq 1 180); do kill -0 "$TPID" 2>/dev/null || break; sleep 1; done
+    kill -0 "$TPID" 2>/dev/null && { echo "target did not stop -> SIGKILL"; kill -9 "$TPID"; }
+  fi
+  [ -n "$WPID" ] && wait "$WPID" 2>/dev/null; echo "target exit status: $?"
+  for p in "${PIDS[@]}"; do kill -9 "$p" 2>/dev/null; done
+  summarize
+}
+summarize() {
+  local DL=$DD_T/regtest/debug.log
+  echo "=== markers in target debug.log"
+  for m in "${MARKERS[@]}"; do printf '  [%6d] %s\n' "$(grep -c -- "$m" "$DL" 2>/dev/null)" "$m"; done
+  echo "  db.log bytes: $(stat -c %s "$DD_T/regtest/db.log" 2>/dev/null || echo absent)"
+  grep -q 'Shutdown: done' "$DL" && echo "  clean shutdown: yes" || echo "  clean shutdown: NO"
+  if [ "$SHIMON" = 1 ]; then
+    echo "=== fdguard reports"; grep -o '=== FDGUARD #[0-9]* [A-Z-]*' "$OUT/fdguard.log" | awk '{print $4}' | sort | uniq -c
+    grep '^S ' "$OUT/fdguard.log" | tail -1
+    python3 "$REPRO/../shim/symbolize.py" --summary "$OUT/fdguard.log" > "$OUT/fdguard.summary.txt" 2>&1
+    head -60 "$OUT/fdguard.summary.txt"
+  fi
+  echo "=== BDB-page-in-blockfile detector"
+  python3 "$REPRO/tools/scan_bdb.py" "$DD_T"/regtest/blocks/blk*.dat "$DD_T"/regtest/blocks/rev*.dat 2>&1 | tail -15
+  echo "=== end $(date -u +%FT%TZ)"
+}
+trap cleanup EXIT
+trap 'exit 130' INT TERM
+
+echo "--- prep datadirs"
+prep_golden "$DD_T"
+prep_fresh "$DD_H1" "$RPC_H1"; prep_fresh "$DD_H2" "$RPC_H2"
+cat > "$OUT/blocknotify.sh" <<EOF
+#!/bin/sh
+# pool-like blocknotify: a couple of dragonx-cli calls per block
+$CLI -regtest -datadir=$DD_T -rpcuser=$RPCUSER -rpcpassword=$RPCPASS -rpcport=$RPC_T getblock "\$1" >/dev/null 2>&1
+$CLI -regtest -datadir=$DD_T -rpcuser=$RPCUSER -rpcpassword=$RPCPASS -rpcport=$RPC_T getblocktemplate >/dev/null 2>&1
+echo "\$(date +%s) \$1" >> $OUT/logs/blocknotify.log
+EOF
+chmod +x "$OUT/blocknotify.sh"
+
+if has tor; then
+  nice -n 10 python3 "$REPRO/tools/faketor.py" --port $TORPORT > "$OUT/logs/faketor.txt" 2>&1 & PIDS+=($!)
+  TORARGS="-listenonion=1 -torcontrol=127.0.0.1:$TORPORT -debug=tor"
+else
+  TORARGS="-listenonion=0"
+fi
+if has p2p || has p2pmin; then P2PARGS="-listen=1 -bind=127.0.0.1 -port=$P2P_T -allowlist=127.0.0.0/8 -maxinboundfromip=100"
+else P2PARGS="-listen=0 -port=$P2P_T"; fi
+
+TARGS=(-regtest -datadir="$DD_T" -testnode=1 -checkpoints=0 $P2PARGS -connect=127.0.0.1:1 -dnsseed=0 -dns=0
+       -asmap="$DD_T/asmap.dat" -rpcuser=$RPCUSER -rpcpassword=$RPCPASS -rpcport=$RPC_T
+       -txindex=1 -autoshield=1 -rpcworkqueue=$WQ -rpcthreads=$RT
+       "-blocknotify=$OUT/blocknotify.sh %s" $TORARGS -debug=net -debug=tls -printtoconsole=0 -daemon=0 $EXTRA)
+echo "target args: ${TARGS[*]}" | sed "s/$RPCPASS/***/g"
+ENVV=()
+if [ "$SHIMON" = 1 ]; then
+  ENVV=(FDGUARD_LOG="$OUT/fdguard.log" FDGUARD_GUARD="$GUARD" FDGUARD_ABORT=0 FDGUARD_STATS_SEC=60 FDGUARD_MAXREPORTS=2000 LD_PRELOAD="$SHIM")
+  [ "$TRACE" = 1 ] && ENVV+=(FDGUARD_TRACE_ALL=1)
+fi
+WRAP=()
+[ "$STRACE" = 1 ] && WRAP=(strace -f -tt -qq -e trace=close,openat,accept4,socket,dup2,dup3 -e status=failed -o "$OUT/strace.txt")
+( cd "$DD_T" && exec env "${ENVV[@]}" nice -n 10 "${WRAP[@]}" "$DRAGONXD" "${TARGS[@]}" ) > "$OUT/logs/target_stdout.txt" 2>&1 &
+WPID=$!
+echo "--- target launched (wrapper pid $WPID); waiting for RPC"
+waitrpc "$DD_T" "$RPC_T" 600 || { echo "target never answered RPC"; exit 1; }
+TPID=$(cat "$DD_T/regtest/hushd.pid" 2>/dev/null); [ -z "$TPID" ] && TPID=$WPID
+echo "target up pid=$TPID height=$(clin "$DD_T" "$RPC_T" getblockcount) $(date -u +%T)"
+
+T0=$(date +%s)
+if has p2p; then
+  start_helper "$DD_H1" "$P2P_H1" "$RPC_H1" "$OUT/logs/helper1.txt"
+  start_helper "$DD_H2" "$P2P_H2" "$RPC_H2" "$OUT/logs/helper2.txt"
+  waitrpc "$DD_H1" "$RPC_H1" 120; waitrpc "$DD_H2" "$RPC_H2" 120
+  nice -n 10 python3 "$REPRO/tools/tcpchurn.py" --port $P2P_T --duration "$DUR" --concurrency 8 > "$OUT/logs/tcpchurn.txt" 2>&1 & PIDS+=($!)
+  nice -n 10 bash "$REPRO/tools/rpcchurn.sh" "$OUT" "$DUR" > "$OUT/logs/rpcchurn.txt" 2>&1 & PIDS+=($!)
+fi
+if has p2pmin; then
+  # minimal P2P: ONE well-behaved helper peer, and an ordinary "disconnectnode" every DISC_INT s
+  # (the helper's -addnode re-dials the target). No raw TCP/TLS churn, no bans, no kills.
+  start_helper "$DD_H1" "$P2P_H1" "$RPC_H1" "$OUT/logs/helper1.txt"; waitrpc "$DD_H1" "$RPC_H1" 120
+  ( k=0; while [ ! -e "$OUT/STOP" ]; do sleep "${DISC_INT:-5}"; for a in $(clin "$DD_T" "$RPC_T" getpeerinfo 2>/dev/null | python3 -c 'import json,sys; [print(p["addr"]) for p in json.load(sys.stdin)]' 2>/dev/null); do
+      clin "$DD_T" "$RPC_T" disconnectnode "$a" >/dev/null 2>&1 && { k=$((k+1)); echo "$(date -u +%T.%N) disconnectnode $a" >> "$OUT/logs/disconnects.log"; }; done
+    sleep 1; clin "$DD_H1" "$RPC_H1" addnode "127.0.0.1:$P2P_T" onetry >/dev/null 2>&1   # helper re-dials now (its -addnode retry is ~2 min)
+    echo "$k" > "$OUT/logs/disconnects.txt"; done ) & PIDS+=($!)
+fi
+if has rpc; then
+  MINEARG=""; has mine && MINEARG="--mine --mine-interval $MINE_INT"
+  nice -n 10 python3 "$REPRO/tools/hammer.py" --port $RPC_T --user $RPCUSER --password $RPCPASS --duration "$DUR" \
+      $HAMMER_ARGS $MINEARG --pid "$TPID" --logfile "$OUT/logs/hammer.log" > "$OUT/logs/hammer_stdout.txt" 2>&1 & PIDS+=($!)
+elif has mine; then
+  ( while [ ! -e "$OUT/STOP" ]; do clin "$DD_T" "$RPC_T" generate 1 >/dev/null 2>&1; sleep "$MINE_INT"; done ) & PIDS+=($!)
+fi
+
+# ---- poll
+end=$(( T0 + DUR )); n=0; hit_at=0
+while [ "$(date +%s)" -lt "$end" ] && [ ! -e "$OUT/STOP" ]; do
+  sleep 30; n=$((n+1))
+  alive=no; kill -0 "$TPID" 2>/dev/null && alive=yes
+  DL=$DD_T/regtest/debug.log
+  h=$(clin "$DD_T" "$RPC_T" getblockcount 2>/dev/null); c=$(clin "$DD_T" "$RPC_T" getconnectioncount 2>/dev/null)
+  rep=$(grep -c '^=== FDGUARD' "$OUT/fdguard.log" 2>/dev/null); noc=$(grep -c 'NON-OWNER-CLOSE' "$OUT/fdguard.log" 2>/dev/null)
+  ebf=$(grep -c 'EBADF-CLOSE' "$OUT/fdguard.log" 2>/dev/null); epl=$(grep -c '=== FDGUARD #[0-9]* EPOLL' "$OUT/fdguard.log" 2>/dev/null)
+  dbnoc=$(grep -A3 'NON-OWNER-CLOSE' "$OUT/fdguard.log" 2>/dev/null | grep -c 'class=\(db\|blk\)')
+  printf '[poll %3d %s +%4ds] alive=%s h=%s conns=%s | fdguard reports=%s nonowner=%s(db/blk %s) ebadf=%s epoll=%s | Epoll=%s SockCloseFail=%s EBADF=%s EPERM=%s SSL_write=%s seek=%s sel=%s | db.log=%s\n' \
+    $n "$(date -u +%T)" $(( $(date +%s) - T0 )) $alive "$h" "$c" "$rep" "$noc" "$dbnoc" "$ebf" "$epl" \
+    "$(grep -c 'libevent: Epoll' "$DL")" "$(grep -c 'Socket close failed' "$DL")" "$(grep -c 'Bad file descriptor' "$DL")" \
+    "$(grep -c 'Operation not permitted' "$DL")" "$(grep -c 'SSL_write' "$DL")" "$(grep -c 'Unable to seek' "$DL")" \
+    "$(grep -c 'socket select error' "$DL")" "$(stat -c %s "$DD_T/regtest/db.log" 2>/dev/null)" | tee -a "$OUT/poll.txt"
+  [ "$alive" = no ] && { echo "!!! target died"; break; }
+  if [ "$STOP_ON_HIT" = 1 ] && [ "${dbnoc:-0}" -gt 0 ]; then
+    [ "$hit_at" = 0 ] && hit_at=$(date +%s)
+    [ $(( $(date +%s) - hit_at )) -ge 60 ] && { echo "--- stop-on-hit"; break; }
+  fi
+done
+cat "$OUT/logs/tcpchurn.txt" "$OUT/logs/rpcchurn.txt" 2>/dev/null
+grep -E 'TOTAL|totals' -A12 "$OUT/logs/hammer.log" 2>/dev/null | tail -14

qa/fd-guard/repro/tools/faketor.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+"""
+Fake Tor control port for the fd-bug harness.
+
+dragonxd's torcontrol.cpp opens a libevent bufferevent to -torcontrol=<host:port>,
+speaks the control protocol (PROTOCOLINFO / AUTHCHALLENGE / AUTHENTICATE / ADD_ONION),
+and on EOF/ERROR reconnects with exponential backoff. We give it a server on a
+loopback port that behaves badly in several ways so the reconnect/teardown loop
+(bufferevent_free with BEV_OPT_CLOSE_ON_FREE, the event_base, the reconnect timer)
+runs constantly. Each accepted connection picks a behaviour at random:
+
+  close   : accept, then immediately close (EOF right away)
+  rst     : accept, set SO_LINGER 0, close -> sends RST
+  stall   : accept, read nothing, hold the socket open then drop it
+  garbage : send non-control-protocol bytes then close
+  halfok  : answer PROTOCOLINFO plausibly, then drop before AUTHENTICATE completes
+
+Loopback only. Run with --port. Ctrl-C / SIGTERM stops it.
+"""
+import argparse, random, socket, struct, threading, time, sys, signal
+
+STOP = threading.Event()
+
+def handle(conn, mode):
+    try:
+        conn.settimeout(5.0)
+        if mode == "close":
+            conn.close(); return
+        if mode == "rst":
+            conn.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
+                            struct.pack("ii", 1, 0))
+            conn.close(); return
+        if mode == "stall":
+            # read whatever the controller sends but never reply; hold 2-4s
+            try:
+                conn.recv(4096)
+            except Exception:
+                pass
+            time.sleep(random.uniform(2.0, 4.0))
+            conn.close(); return
+        if mode == "garbage":
+            try:
+                conn.recv(4096)
+            except Exception:
+                pass
+            conn.sendall(b"\x00\xff\x01garbage not a tor reply\r\n" * 4)
+            conn.close(); return
+        if mode == "halfok":
+            # read the PROTOCOLINFO line, give a plausible 250 reply advertising
+            # NULL auth, then drop the connection before AUTHENTICATE finishes.
+            try:
+                conn.recv(4096)
+            except Exception:
+                pass
+            conn.sendall(b"250-PROTOCOLINFO 1\r\n"
+                         b"250-AUTH METHODS=NULL\r\n"
+                         b'250-VERSION Tor="0.4.7.13"\r\n'
+                         b"250 OK\r\n")
+            time.sleep(random.uniform(0.2, 1.0))
+            conn.close(); return
+    except Exception:
+        try: conn.close()
+        except Exception: pass
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--port", type=int, required=True)
+    ap.add_argument("--host", default="127.0.0.1")
+    args = ap.parse_args()
+
+    modes = ["close", "rst", "stall", "garbage", "halfok"]
+    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    srv.bind((args.host, args.port))
+    srv.listen(16)
+    srv.settimeout(1.0)
+    print(f"faketor: listening on {args.host}:{args.port}", flush=True)
+
+    def onsig(*_): STOP.set()
+    signal.signal(signal.SIGTERM, onsig)
+    signal.signal(signal.SIGINT, onsig)
+
+    n = 0
+    while not STOP.is_set():
+        try:
+            conn, _ = srv.accept()
+        except socket.timeout:
+            continue
+        except Exception:
+            break
+        mode = modes[n % len(modes)]
+        n += 1
+        threading.Thread(target=handle, args=(conn, mode), daemon=True).start()
+    srv.close()
+    print(f"faketor: stopped after {n} connections", flush=True)
+
+if __name__ == "__main__":
+    main()

qa/fd-guard/repro/tools/hammer.py

@@ -0,0 +1,796 @@
+#!/usr/bin/env python3
+# hammer.py -- JSON-RPC robustness/load harness for a LOCAL regtest dragonxd.
+#
+# Scope: connects only to the --host/--port you pass (meant to be 127.0.0.1 and a
+# regtest RPC port). Its job is to reproduce a known file-descriptor bug in the
+# libevent-2.1.8 HTTP RPC server (src/httpserver.cpp) by exercising the same
+# request patterns the production pool box sees: many concurrent clients, clients
+# that disconnect mid-request/mid-response (the explorer indexer's 120 s client
+# timeout), work-queue overflow, keep-alive pipelining, and malformed input.
+# stdlib only.
+#
+# Correctness oracle: every request carries a unique JSON-RPC id; if a reply comes
+# back with a different id on the same socket, that is cross-talk (the server wrote
+# one client's reply onto another client's fd) and is logged + counted.
+#
+# See README.txt for the exact invocation.
+
+import argparse
+import base64
+import collections
+import json
+import os
+import random
+import socket
+import struct
+import threading
+import time
+
+# ------------------------------------------------------------------ config/state
+
+class C:
+    host = "127.0.0.1"
+    port = 19461
+    auth = ""
+    ctimeout = 15.0
+
+CFG = C()
+STOP = threading.Event()
+SHARED = {"hashes": [], "lpid": None, "zaddrs": [], "rawblock": None}
+SHARED_LOCK = threading.Lock()
+_ID = [random.SystemRandom().randrange(1 << 40)]
+_ID_LOCK = threading.Lock()
+LOGLOCK = threading.Lock()
+LOGF = [None]
+
+def next_id():
+    with _ID_LOCK:
+        _ID[0] += 1
+        return _ID[0]
+
+def log(msg):
+    line = "%s %s" % (time.strftime("%H:%M:%S"), msg)
+    with LOGLOCK:
+        print(line, flush=True)
+        if LOGF[0]:
+            LOGF[0].write(line + "\n"); LOGF[0].flush()
+
+# ------------------------------------------------------------------ stats
+
+KEYS = ["sent", "ok", "rpcerr", "wq500", "http401", "http4xx", "httpother",
+        "early_rst", "early_fin", "srvreset", "srveof", "ctimeout", "connfail",
+        "idmismatch", "malformed", "mined", "half_open", "half_srvclose"]
+
+class Stats:
+    def __init__(self):
+        self.lk = threading.Lock()
+        self.total = collections.Counter()
+        self.win = collections.Counter()
+        self.anom = []
+    def inc(self, k, n=1):
+        with self.lk:
+            self.total[k] += n; self.win[k] += n
+    def anomaly(self, kind, detail):
+        with self.lk:
+            if len(self.anom) < 300:
+                self.anom.append((time.time(), kind, detail))
+        log("ANOMALY %s %s" % (kind, detail))
+    def swap(self):
+        with self.lk:
+            w = self.win; self.win = collections.Counter(); return w
+
+ST = Stats()
+
+# ------------------------------------------------------------------ socket helpers
+
+def auth_header(u, p):
+    return "Basic " + base64.b64encode(("%s:%s" % (u, p)).encode()).decode()
+
+def mkconn(timeout=None, rcvbuf=None):
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    if rcvbuf:
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, rcvbuf)
+    s.settimeout(timeout if timeout is not None else CFG.ctimeout)
+    s.connect((CFG.host, CFG.port))
+    return s
+
+def end_rst(s):
+    # linger 0 -> TCP RST on close (models a client that dies / resets abruptly)
+    try:
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
+    except OSError:
+        pass
+    try:
+        s.close()
+    except OSError:
+        pass
+
+def end_fin(s):
+    try:
+        s.shutdown(socket.SHUT_RDWR)
+    except OSError:
+        pass
+    try:
+        s.close()
+    except OSError:
+        pass
+
+def end_early(s, how=None):
+    how = how or random.choice(("rst", "fin"))
+    if how == "rst":
+        end_rst(s); ST.inc("early_rst")
+    else:
+        end_fin(s); ST.inc("early_fin")
+
+def req_bytes(method, params, rid, keepalive=True, auth=None, uri="/",
+              httpmethod="POST", body=None, clen=None):
+    if body is None:
+        body = json.dumps({"jsonrpc": "1.0", "id": rid,
+                           "method": method, "params": params}).encode()
+    head = [
+        "%s %s HTTP/1.1" % (httpmethod, uri),
+        "Host: %s:%d" % (CFG.host, CFG.port),
+        "Authorization: %s" % (auth if auth is not None else CFG.auth),
+        "Content-Type: application/json",
+        "Content-Length: %d" % (clen if clen is not None else len(body)),
+        "Connection: %s" % ("keep-alive" if keepalive else "close"),
+        "", "",
+    ]
+    return "\r\n".join(head).encode() + body
+
+class Reader:
+    """Incremental HTTP/1.1 reply reader over one socket."""
+    def __init__(self, s):
+        self.s = s; self.buf = b""
+    def _fill(self):
+        d = self.s.recv(65536)
+        if not d:
+            raise EOFError()
+        self.buf += d
+    def read(self):
+        while b"\r\n\r\n" not in self.buf:
+            self._fill()
+            if len(self.buf) > (1 << 20):
+                raise ValueError("oversize header")
+        head, self.buf = self.buf.split(b"\r\n\r\n", 1)
+        lines = head.split(b"\r\n")
+        p = lines[0].split(b" ", 2)
+        if len(p) < 2 or not p[0].startswith(b"HTTP/1."):
+            raise ValueError("bad status line %r" % lines[0][:80])
+        status = int(p[1])
+        hdr = {}
+        for ln in lines[1:]:
+            if b":" in ln:
+                k, v = ln.split(b":", 1)
+                hdr[k.strip().lower()] = v.strip()
+        if hdr.get(b"transfer-encoding", b"").lower() == b"chunked":
+            body = b""
+            while True:
+                while b"\r\n" not in self.buf:
+                    self._fill()
+                ln, self.buf = self.buf.split(b"\r\n", 1)
+                n = int(ln.split(b";")[0], 16)
+                while len(self.buf) < n + 2:
+                    self._fill()
+                body += self.buf[:n]; self.buf = self.buf[n + 2:]
+                if n == 0:
+                    break
+        elif b"content-length" in hdr:
+            n = int(hdr[b"content-length"])
+            while len(self.buf) < n:
+                self._fill()
+            body, self.buf = self.buf[:n], self.buf[n:]
+        else:
+            try:
+                while True:
+                    self._fill()
+            except EOFError:
+                pass
+            body, self.buf = self.buf, b""
+        return status, hdr, body
+
+def account(status, body, rid, method, expect401=False):
+    if status == 401:
+        ST.inc("http401")
+        if not expect401:
+            ST.anomaly("unexpected401", (method,))
+        return None
+    if status == 500 and b"Work queue depth exceeded" in body:
+        ST.inc("wq500"); return None
+    j = None
+    if body[:1] in (b"{", b"["):
+        try:
+            j = json.loads(body)
+        except ValueError:
+            j = None
+    if j is None:
+        if 400 <= status < 500:
+            ST.inc("http4xx")
+        else:
+            ST.inc("httpother")
+            if status == 200:
+                ST.anomaly("non-json-200", (method, body[:120].decode("latin1")))
+        return None
+    if isinstance(j, dict) and rid is not None and j.get("id") != rid:
+        ST.inc("idmismatch")
+        ST.anomaly("idmismatch", {"method": method, "sent": rid, "got": j.get("id"),
+                                  "body": body[:200].decode("latin1")})
+    if isinstance(j, dict) and j.get("error"):
+        ST.inc("rpcerr")
+    else:
+        ST.inc("ok")
+    return j
+
+def call(s, rd, method, params, keepalive=True, expect401=False, auth=None):
+    rid = next_id()
+    try:
+        s.sendall(req_bytes(method, params, rid, keepalive=keepalive, auth=auth))
+        ST.inc("sent")
+        status, hdr, body = rd.read()
+    except socket.timeout:
+        ST.inc("ctimeout"); return None, False
+    except (ConnectionResetError, BrokenPipeError):
+        ST.inc("srvreset"); return None, False
+    except EOFError:
+        ST.inc("srveof"); return None, False
+    except ValueError as e:
+        ST.inc("malformed"); ST.anomaly("malformed", (method, str(e))); return None, False
+    except OSError:
+        ST.inc("srvreset"); return None, False
+    j = account(status, body, rid, method, expect401)
+    alive = keepalive and hdr.get(b"connection", b"").lower() != b"close"
+    return j, alive
+
+# ------------------------------------------------------------------ menus
+
+def a_hash():
+    with SHARED_LOCK:
+        return random.choice(SHARED["hashes"]) if SHARED["hashes"] else None
+
+def a_zaddr():
+    with SHARED_LOCK:
+        return random.choice(SHARED["zaddrs"]) if SHARED["zaddrs"] else None
+
+def normal_menu():
+    # Mostly fast calls (sub-ms) so a well-behaved client interleaves with mining.
+    # Heavy wallet calls (z_gettotalbalance ~2.3s on the golden wallet) appear but
+    # are rare, so they don't permanently peg the small worker pool on their own.
+    fast = [("getinfo", []), ("getblockcount", []), ("getbestblockhash", []),
+            ("getrawmempool", []), ("getmininginfo", []), ("getdifficulty", []),
+            ("getblockchaininfo", []), ("getpeerinfo", []),
+            ("getblocktemplate", []), ("getnetworkinfo", [])]
+    h = a_hash()
+    if h:
+        fast.append(("getblock", [h, 2]))
+        fast.append(("getblockheader", [h]))
+    # "slow" here is only mildly-larger fast calls; the genuinely heavy wallet calls
+    # live in the paced slow-fire class so a normal dashboard poller never pegs the
+    # worker pool.
+    slow = [("coinsupply", []), ("listtransactions", ["*", 10]),
+            ("listunspent", [0]), ("getblock", [a_hash() or "", 2])]
+    return random.choice(slow) if random.random() < 0.1 else random.choice(fast)
+
+def medium_menu():
+    # Fast to compute (~4 ms on this node) but a non-trivial reply to write, so they
+    # are good targets for a mid-response disconnect. The fd-reuse race fires on the
+    # disconnect, not on call duration, so firing these at a HIGH rate drives the
+    # race hard while the worker frees in milliseconds (mining keeps its slots).
+    m = [("listtransactions", ["*", 10000]), ("getrawmempool", []),
+         ("getrawmempool", [True]), ("getblockchaininfo", []),
+         ("getblocktemplate", []), ("getchaintips", []), ("getinfo", []),
+         ("coinsupply", []), ("listunspent", [0])]
+    h = a_hash()
+    if h:
+        m.append(("getblock", [h, 2]))
+        m.append(("getblock", [h, 1]))
+    return random.choice(m)
+
+def heavy_menu():
+    # Genuinely slow calls (z_gettotalbalance ~1.7 s, z_listunspent ~5.7 s on the
+    # golden wallet). Fired only by the small, PACED slow-fire class so a worker
+    # keeps running a multi-second call long after the client disconnected (the
+    # production indexer/coinsupply scenario that widens the fd-reuse window),
+    # without the fire-and-disconnect backlog pinning the whole pool.
+    m = [("z_gettotalbalance", [0]), ("z_listunspent", [0])]
+    z = a_zaddr()
+    if z:
+        m.append(("z_listreceivedbyaddress", [z, 0]))
+    return random.choice(m)
+
+# ------------------------------------------------------------------ clients
+
+def client_normal(cid):
+    while not STOP.is_set():
+        keepalive = random.random() < 0.7
+        try:
+            s = mkconn()
+        except OSError:
+            ST.inc("connfail"); time.sleep(0.05); continue
+        rd = Reader(s)
+        n = random.randint(1, 8) if keepalive else 1
+        try:
+            for _ in range(n):
+                if STOP.is_set():
+                    break
+                method, params = normal_menu()
+                j, alive = call(s, rd, method, params, keepalive=keepalive)
+                if not alive:
+                    break
+        finally:
+            (end_fin if random.random() < 0.8 else end_rst)(s)
+        time.sleep(random.uniform(0.01, 0.15))
+
+def client_impatient(cid):
+    # High-rate fast-disconnect churn: send a (medium) request, then disconnect
+    # after 1-500 ms at a random phase. This is the main driver of the fd-reuse
+    # race -- a disconnect concurrent with the server writing the reply / re-arming
+    # the socket -- and the worker frees in ms so mining is unaffected.
+    while not STOP.is_set():
+        method, params = medium_menu()
+        rid = next_id()
+        body = json.dumps({"jsonrpc": "1.0", "id": rid, "method": method,
+                           "params": params}).encode()
+        phase = random.choice(("mid_request", "after_send", "mid_response"))
+        try:
+            s = mkconn(timeout=2.0)
+        except OSError:
+            ST.inc("connfail"); continue
+        try:
+            if phase == "mid_request":
+                full = req_bytes(method, params, rid, keepalive=False, body=body)
+                cut = random.randint(len(full) - len(body), len(full) - 1)
+                s.sendall(full[:cut])
+                time.sleep(random.uniform(0.001, 0.5))
+                end_early(s); continue
+            s.sendall(req_bytes(method, params, rid, keepalive=False, body=body))
+            ST.inc("sent")
+            time.sleep(random.uniform(0.001, 0.5))
+            if phase == "mid_response":
+                try:
+                    s.recv(random.randint(0, 64))  # read a few bytes, then bail
+                except OSError:
+                    pass
+            end_early(s)
+        except (OSError, socket.timeout):
+            ST.inc("srvreset")
+            end_rst(s)
+        time.sleep(random.uniform(0.05, 0.4))
+
+def client_slowfire(cid, pace):
+    # Paced indexer/coinsupply client: fire ONE genuinely-slow request, disconnect
+    # after 1-500 ms (so the worker runs the multi-second call with no client left
+    # == the production fd-reuse window), then wait `pace` s before the next so a
+    # fire-and-disconnect backlog cannot pin the whole worker pool. Phase: cut the
+    # request mid-body, or send it whole and abort while the worker is busy, or read
+    # a few bytes of the reply then abort.
+    while not STOP.is_set():
+        method, params = heavy_menu()
+        rid = next_id()
+        body = json.dumps({"jsonrpc": "1.0", "id": rid, "method": method,
+                           "params": params}).encode()
+        phase = random.choice(("mid_request", "after_send", "mid_response"))
+        try:
+            s = mkconn(timeout=2.0)
+        except OSError:
+            ST.inc("connfail"); time.sleep(pace); continue
+        try:
+            if phase == "mid_request":
+                full = req_bytes(method, params, rid, keepalive=False, body=body)
+                cut = random.randint(len(full) - len(body), len(full) - 1)
+                s.sendall(full[:cut])
+                time.sleep(random.uniform(0.001, 0.5)); end_early(s)
+            else:
+                s.sendall(req_bytes(method, params, rid, keepalive=False, body=body))
+                ST.inc("sent")
+                time.sleep(random.uniform(0.001, 0.5))
+                if phase == "mid_response":
+                    try:
+                        s.recv(random.randint(0, 64))
+                    except OSError:
+                        pass
+                end_early(s)
+        except (OSError, socket.timeout):
+            ST.inc("srvreset"); end_rst(s)
+        time.sleep(random.uniform(pace * 0.5, pace * 1.5))
+
+def client_half_open(cid):
+    # connect, send nothing / partial headers, trickle, then vanish
+    while not STOP.is_set():
+        try:
+            s = mkconn(timeout=1.0)
+        except OSError:
+            ST.inc("connfail"); time.sleep(0.1); continue
+        ST.inc("half_open")
+        style = random.choice(("silent", "partial_headers", "full_headers_no_body"))
+        try:
+            if style == "partial_headers":
+                s.sendall(b"POST / HTTP/1.1\r\nHost: x\r\n")
+            elif style == "full_headers_no_body":
+                s.sendall(b"POST / HTTP/1.1\r\nHost: x\r\nAuthorization: " +
+                          CFG.auth.encode() + b"\r\nContent-Length: 1000\r\n\r\n")
+            deadline = time.monotonic() + random.uniform(0.1, 20.0)
+            while time.monotonic() < deadline and not STOP.is_set():
+                time.sleep(random.uniform(0.2, 1.5))
+                try:
+                    s.sendall(b" ")  # one byte, keep it half-alive
+                except OSError:
+                    ST.inc("half_srvclose"); break
+                try:
+                    s.settimeout(0.01)
+                    if s.recv(1) == b"":
+                        ST.inc("half_srvclose"); break
+                except socket.timeout:
+                    pass
+                except OSError:
+                    ST.inc("half_srvclose"); break
+        finally:
+            end_early(s)
+
+def client_burst(cid, size, every):
+    while not STOP.is_set():
+        batch = []
+        for _ in range(size):
+            try:
+                s = mkconn(timeout=3.0)
+            except OSError:
+                ST.inc("connfail"); continue
+            batch.append(s)
+        for s in batch:
+            try:
+                method, params = medium_menu()
+                s.sendall(req_bytes(method, params, next_id(), keepalive=False))
+                ST.inc("sent")
+            except OSError:
+                pass
+        # most close immediately (before the 500/reply is written); a few read it
+        keep = random.sample(batch, max(0, len(batch) // 4)) if batch else []
+        for s in batch:
+            if s in keep:
+                continue
+            end_early(s)
+        for s in keep:
+            try:
+                rd = Reader(s); st, hd, bd = rd.read(); account(st, bd, None, "burst")
+            except Exception:
+                ST.inc("srvreset")
+            finally:
+                end_fin(s)
+        for _ in range(int(every * 10)):
+            if STOP.is_set():
+                break
+            time.sleep(0.1)
+
+def client_pipeline(cid):
+    while not STOP.is_set():
+        try:
+            s = mkconn()
+        except OSError:
+            ST.inc("connfail"); time.sleep(0.05); continue
+        n = random.randint(2, 8)
+        ids = []
+        blob = b""
+        for _ in range(n):
+            rid = next_id(); ids.append(rid)
+            method, params = normal_menu()
+            blob += req_bytes(method, params, rid, keepalive=True)
+        try:
+            s.sendall(blob); ST.inc("sent", n)
+            rd = Reader(s)
+            toread = random.randint(0, n)  # sometimes leave replies unread, then close
+            for k in range(toread):
+                st, hd, bd = rd.read()
+                account(st, bd, ids[k], "pipeline")
+        except (OSError, EOFError, ValueError):
+            ST.inc("srvreset")
+        finally:
+            end_early(s)
+        time.sleep(random.uniform(0, 0.03))
+
+def client_garbage(cid):
+    probes = [
+        ("submitblock", lambda: ["ab" * random.randint(1, 4000)]),
+        ("submitblock", lambda: [os.urandom(random.randint(1, 200)).hex()]),
+        ("submitblock", lambda: ["zzzz"]),
+        ("submitblock", lambda: [""]),
+    ]
+    while not STOP.is_set():
+        pick = random.random()
+        try:
+            s = mkconn(timeout=5.0)
+        except OSError:
+            ST.inc("connfail"); continue
+        try:
+            if pick < 0.5:
+                method, mk = random.choice(probes)
+                j, _ = call(s, Reader(s), method, mk(), keepalive=False)
+            elif pick < 0.65:
+                # malformed JSON body
+                s.sendall(req_bytes("x", [], next_id(), keepalive=False,
+                                    body=b'{"jsonrpc":"1.0","method":'))
+                ST.inc("sent")
+                try:
+                    account(*((lambda r: (r[0], r[2]))(Reader(s).read())), rid=None, method="badjson")
+                except Exception:
+                    ST.inc("srvreset")
+            elif pick < 0.8:
+                # unknown method / wrong URI
+                s.sendall(req_bytes("getinfo", [], next_id(), keepalive=False, uri="/nope"))
+                ST.inc("sent")
+                try:
+                    st, hd, bd = Reader(s).read(); account(st, bd, None, "wronguri")
+                except Exception:
+                    ST.inc("srvreset")
+            else:
+                # Content-Length larger than the body we send, then bail
+                body = b'{"id":1}'
+                s.sendall(req_bytes("getinfo", [], 1, keepalive=False, body=body, clen=len(body) + 5000))
+                ST.inc("sent")
+                time.sleep(random.uniform(0.001, 0.3))
+                end_early(s)
+                continue
+        except (OSError, socket.timeout):
+            ST.inc("srvreset")
+        finally:
+            end_early(s)
+        time.sleep(random.uniform(0.02, 0.2))
+
+def client_slowread(cid):
+    # ask for a big reply with a tiny recv buffer and do not read it, stalling the
+    # server write side, then close abruptly
+    while not STOP.is_set():
+        try:
+            s = mkconn(timeout=3.0, rcvbuf=512)
+        except OSError:
+            ST.inc("connfail"); continue
+        try:
+            s.sendall(req_bytes("listtransactions", ["*", 10000], next_id(), keepalive=False))
+            ST.inc("sent")
+            time.sleep(random.uniform(0.05, 1.0))
+        except OSError:
+            ST.inc("srvreset")
+        finally:
+            end_early(s)
+        time.sleep(random.uniform(0, 0.05))
+
+def _stale_lpid():
+    # A longpollid whose hash != the current tip makes getblocktemplate return at
+    # once (the server's wait loop condition is immediately false), so it exercises
+    # the long-poll request path (lpval parsing, cs_main leave/enter) and the
+    # mid-response disconnect WITHOUT parking a worker. On a tiny -rpcthreads pool a
+    # genuinely parking long-poll + RPC-driven mining self-deadlocks (both workers
+    # wait for a block that only a third worker could mine); that is a pool-size
+    # artifact, not the fd bug, so the harness avoids it by default.
+    with SHARED_LOCK:
+        lpid = SHARED["lpid"]
+    if lpid and len(lpid) > 64:
+        h = lpid[:64]
+        flip = "0" if h[0] != "0" else "1"
+        return flip + h[1:] + lpid[64:]
+    return ("0" * 64) + "0"
+
+def client_longpoll(cid):
+    while not STOP.is_set():
+        try:
+            s = mkconn(timeout=10.0)
+        except OSError:
+            ST.inc("connfail"); time.sleep(0.2); continue
+        rd = Reader(s)
+        params = [{"longpollid": _stale_lpid()}]
+        rid = next_id()
+        try:
+            s.sendall(req_bytes("getblocktemplate", params, rid, keepalive=False))
+            ST.inc("sent")
+            if random.random() < 0.5:
+                # impatient pool worker: abandon the request after 1-500 ms
+                time.sleep(random.uniform(0.001, 0.5))
+                end_early(s); continue
+            st, hd, bd = rd.read()
+            j = account(st, bd, rid, "getblocktemplate(lp)")
+            if j and isinstance(j, dict) and j.get("result"):
+                with SHARED_LOCK:
+                    SHARED["lpid"] = j["result"].get("longpollid")
+        except (OSError, socket.timeout, EOFError, ValueError):
+            ST.inc("srvreset")
+        finally:
+            end_fin(s)
+        time.sleep(random.uniform(0.02, 0.2))
+
+def miner(interval):
+    # Mine a block every `interval` seconds. Retry through work-queue rejections so
+    # blocks still land under heavy load (the production box mines steadily too).
+    while not STOP.is_set():
+        got = False
+        deadline = time.monotonic() + interval + 30
+        while not got and not STOP.is_set() and time.monotonic() < deadline:
+            try:
+                s = mkconn(timeout=60.0)
+                j, _ = call(s, Reader(s), "generate", [1], keepalive=False)
+                end_fin(s)
+                if j and not j.get("error") and j.get("result"):
+                    ST.inc("mined"); got = True
+                    h = j["result"][0] if isinstance(j["result"], list) else None
+                    if h:
+                        with SHARED_LOCK:
+                            SHARED["hashes"].append(h)
+                            SHARED["hashes"] = SHARED["hashes"][-200:]
+                else:
+                    time.sleep(0.2)  # likely work-queue 500; back off and retry
+            except Exception:
+                time.sleep(0.2)
+        for _ in range(int(interval * 10)):
+            if STOP.is_set():
+                break
+            time.sleep(0.1)
+
+# ------------------------------------------------------------------ bootstrap shared state
+
+def bootstrap():
+    try:
+        s = mkconn(timeout=30.0); rd = Reader(s)
+        j, _ = call(s, rd, "getbestblockhash", [], keepalive=True)
+        if j and j.get("result"):
+            with SHARED_LOCK:
+                SHARED["hashes"].append(j["result"])
+        j, _ = call(s, rd, "getblockcount", [], keepalive=True)
+        cnt = j.get("result") if j else None
+        if isinstance(cnt, int):
+            for hh in range(max(0, cnt - 30), cnt + 1):
+                jj, _ = call(s, rd, "getblockhash", [hh], keepalive=True)
+                if jj and jj.get("result"):
+                    with SHARED_LOCK:
+                        SHARED["hashes"].append(jj["result"])
+        j, _ = call(s, rd, "listaddresses", [], keepalive=True)
+        j, _ = call(s, rd, "z_listaddresses", [], keepalive=True)
+        if j and isinstance(j.get("result"), list):
+            with SHARED_LOCK:
+                SHARED["zaddrs"] = j["result"][:16]
+        j, _ = call(s, rd, "getblocktemplate", [{}], keepalive=True)
+        if j and isinstance(j.get("result"), dict):
+            with SHARED_LOCK:
+                SHARED["lpid"] = j["result"].get("longpollid")
+        end_fin(s)
+        log("bootstrap: %d hashes, %d zaddrs, lpid=%s" %
+            (len(SHARED["hashes"]), len(SHARED["zaddrs"]), bool(SHARED["lpid"])))
+    except Exception as e:
+        log("bootstrap failed: %s" % e)
+
+# ------------------------------------------------------------------ fd accounting
+
+def count_fds(pid):
+    if not pid:
+        return (None, None, None)
+    base = "/proc/%d/fd" % pid
+    try:
+        fds = os.listdir(base)
+    except OSError:
+        return (None, None, None)
+    sock = files = 0
+    for fd in fds:
+        try:
+            tgt = os.readlink(os.path.join(base, fd))
+        except OSError:
+            continue
+        if tgt.startswith("socket:"):
+            sock += 1
+        elif tgt.startswith("/"):
+            files += 1
+    return (len(fds), sock, files)
+
+def pid_alive(pid):
+    if not pid:
+        return True
+    try:
+        os.kill(pid, 0); return True
+    except OSError:
+        return False
+
+# ------------------------------------------------------------------ main
+
+def reporter(pid):
+    t0 = time.monotonic()
+    while not STOP.is_set():
+        for _ in range(10):
+            if STOP.is_set():
+                break
+            time.sleep(0.1)
+        w = ST.swap()
+        tot, sk, fl = count_fds(pid)
+        fdstr = ""
+        if tot is not None:
+            fdstr = " | fds=%d sock=%d file=%d" % (tot, sk, fl)
+        log("t=%4ds ok=%d rpcerr=%d wq500=%d 4xx=%d other=%d | "
+            "earlyRST=%d earlyFIN=%d srvRST=%d srvEOF=%d ctmo=%d connfail=%d | "
+            "idmis=%d malf=%d | mined=%d halfopen=%d%s" % (
+                int(time.monotonic() - t0),
+                w["ok"], w["rpcerr"], w["wq500"], w["http4xx"], w["httpother"],
+                w["early_rst"], w["early_fin"], w["srvreset"], w["srveof"],
+                w["ctimeout"], w["connfail"], w["idmismatch"], w["malformed"],
+                w["mined"], w["half_open"], fdstr))
+        if pid and not pid_alive(pid):
+            log("!!! NODE PROCESS %d IS GONE -- stopping harness" % pid)
+            STOP.set()
+            break
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--host", default="127.0.0.1")
+    ap.add_argument("--port", type=int, default=19461)
+    ap.add_argument("--user", default="ab")
+    ap.add_argument("--password", default="abpass")
+    ap.add_argument("--duration", type=int, default=300)
+    ap.add_argument("--ctimeout", type=float, default=15.0)
+    ap.add_argument("--normal", type=int, default=24)
+    ap.add_argument("--impatient", type=int, default=10)
+    ap.add_argument("--slowfire", type=int, default=3,
+                    help="paced clients that fire a genuinely-slow call then disconnect")
+    ap.add_argument("--slowfire-pace", type=float, default=2.0)
+    ap.add_argument("--halfopen", type=int, default=6)
+    ap.add_argument("--pipeline", type=int, default=6)
+    ap.add_argument("--garbage", type=int, default=4)
+    ap.add_argument("--slowread", type=int, default=2)
+    ap.add_argument("--longpoll", type=int, default=6)
+    # burst-size must exceed (rpcthreads + rpcworkqueue) to overflow the queue even
+    # when calls are fast; with -rpcthreads=8 -rpcworkqueue=4 that threshold is 12.
+    ap.add_argument("--burst-threads", type=int, default=1)
+    ap.add_argument("--burst-size", type=int, default=24)
+    ap.add_argument("--burst-every", type=float, default=1.5)
+    ap.add_argument("--mine", action="store_true", help="run a miner thread")
+    ap.add_argument("--mine-interval", type=float, default=4.0)
+    ap.add_argument("--pid", type=int, default=0, help="node pid, for /proc fd + liveness")
+    ap.add_argument("--logfile", default="")
+    args = ap.parse_args()
+
+    CFG.host = args.host; CFG.port = args.port
+    CFG.auth = auth_header(args.user, args.password); CFG.ctimeout = args.ctimeout
+    if args.logfile:
+        LOGF[0] = open(args.logfile, "a")
+
+    log("hammer start host=%s port=%d duration=%ds pid=%s" %
+        (CFG.host, CFG.port, args.duration, args.pid or "-"))
+    bootstrap()
+
+    threads = []
+    def spawn(fn, n, *a):
+        for i in range(n):
+            t = threading.Thread(target=fn, args=(i,) + a, daemon=True)
+            t.start(); threads.append(t)
+
+    spawn(client_normal, args.normal)
+    spawn(client_impatient, args.impatient)
+    spawn(client_slowfire, args.slowfire, args.slowfire_pace)
+    spawn(client_half_open, args.halfopen)
+    spawn(client_pipeline, args.pipeline)
+    spawn(client_garbage, args.garbage)
+    spawn(client_slowread, args.slowread)
+    spawn(client_longpoll, args.longpoll)
+    for i in range(args.burst_threads):
+        t = threading.Thread(target=client_burst, args=(i, args.burst_size, args.burst_every), daemon=True)
+        t.start(); threads.append(t)
+    if args.mine:
+        t = threading.Thread(target=miner, args=(args.mine_interval,), daemon=True)
+        t.start(); threads.append(t)
+
+    rt = threading.Thread(target=reporter, args=(args.pid,), daemon=True)
+    rt.start()
+
+    end = time.monotonic() + args.duration
+    try:
+        while time.monotonic() < end and not STOP.is_set():
+            time.sleep(0.2)
+    except KeyboardInterrupt:
+        pass
+    STOP.set()
+    time.sleep(1.0)
+
+    log("=== TOTALS ===")
+    for k in KEYS:
+        if ST.total[k]:
+            log("  %-14s %d" % (k, ST.total[k]))
+    log("anomalies: %d" % len(ST.anom))
+    for t, kind, detail in ST.anom[:40]:
+        log("  %s %s %s" % (time.strftime("%H:%M:%S", time.localtime(t)), kind, detail))
+    if args.pid:
+        log("node alive at end: %s" % pid_alive(args.pid))
+
+if __name__ == "__main__":
+    main()

qa/fd-guard/repro/tools/rpcchurn.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+# RPC-driven P2P peer churn on the target (same mix as fdbug/p2p/rpcchurn.sh):
+# disconnectnode / addnode onetry / setban+clearbanned / kill -9 + restart of helper2.
+# usage: rpcchurn.sh <run dir> <seconds>
+source "$REPRO/common.sh"
+OUT=$1; DUR=${2:-300}
+DD_T=$OUT/target; DD_H1=$OUT/helper1; DD_H2=$OUT/helper2
+end=$(( $(date +%s) + DUR )); i=0; restarts=0
+while [ "$(date +%s)" -lt "$end" ] && [ ! -e "$OUT/STOP" ]; do
+  i=$((i+1))
+  case $(( i % 5 )) in
+    0) clin "$DD_T" "$RPC_T" disconnectnode "127.0.0.1:$P2P_H1" >/dev/null 2>&1 ;;
+    1) clin "$DD_T" "$RPC_T" addnode "127.0.0.1:$P2P_H1" onetry >/dev/null 2>&1 ;;
+    2) clin "$DD_T" "$RPC_T" setban "127.0.0.1" add 1 >/dev/null 2>&1; sleep 0.3
+       clin "$DD_T" "$RPC_T" clearbanned >/dev/null 2>&1
+       clin "$DD_H2" "$RPC_H2" addnode "127.0.0.1:$P2P_T" onetry >/dev/null 2>&1 ;;
+    3) pid=$(cat "$DD_H2/regtest/hushd.pid" 2>/dev/null)
+       if [ -n "$pid" ] && kill -0 "$pid" 2>/dev/null; then
+         kill -9 "$pid" 2>/dev/null
+         for _ in $(seq 1 20); do kill -0 "$pid" 2>/dev/null || break; sleep 0.5; done
+         start_helper "$DD_H2" "$P2P_H2" "$RPC_H2" "$OUT/logs/helper2.txt"
+         waitrpc "$DD_H2" "$RPC_H2" 120 >/dev/null 2>&1
+         restarts=$((restarts+1))
+       fi ;;
+    4) clin "$DD_T" "$RPC_T" addnode "127.0.0.1:$P2P_H2" onetry >/dev/null 2>&1 ;;
+  esac
+  sleep 0.7
+done
+echo "rpcchurn: $i cycles, $restarts helper2 restarts"

qa/fd-guard/repro/tools/scan_bdb.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""
+scan_bdb.py -- find Berkeley DB pages written into files that must not contain them
+(blocks/blk*.dat, blocks/rev*.dat).  Stand-in for the production incident tool.
+
+BDB writes page P of a database at file offset P*pagesize through pwrite(fd, ...).  If
+fd has been closed and its number reused by a block file, the page lands in the block
+file at exactly P*pagesize.  So at every pagesize-aligned offset we decode the PAGE
+header (db_page.h, all little-endian on x86):
+    0  u32 lsn.file     4  u32 lsn.offset   8  u32 pgno   12 u32 prev_pgno
+   16  u32 next_pgno   20  u16 entries     22  u16 hf_offset
+   24  u8  level       25  u8  type
+and a metadata page (pgno 0 / any meta type) also carries magic at 12 and pagesize at 20.
+
+A hit needs ALL of:
+  * pgno == offset // pagesize                       (the self-locating property)
+  * type is a real BDB page type (1..16 except 0)
+  * level consistent with type (overflow/meta/hash: 0; btree leaf: 1; internal: 2..32)
+  * hf_offset <= pagesize; entries <= pagesize/2
+  * for overflow (7): entries (refcount) >= 1 and 0 < hf_offset (data length) <= pagesize-26
+Strength: "STRONG" if lsn == (0,1) (pages rewritten by lsn_reset -- the production
+signature) or a meta page with a valid magic, else "weak".
+Also reports any aligned BDB meta magic (btree 0x053162, hash 0x061561, queue 0x042253,
+heap 0x074582) even when pgno does not match.
+
+usage: scan_bdb.py [--pagesize 4096] [--all-offsets] files...
+exit 0 = clean, 2 = hits.
+"""
+import argparse, struct, sys
+
+TYPES = {1: "duplicate", 2: "hash_unsorted", 3: "ibtree", 4: "irecno", 5: "lbtree", 6: "lrecno",
+         7: "overflow", 8: "hashmeta", 9: "btreemeta", 10: "qammeta", 11: "qamdata", 12: "ldup",
+         13: "hash", 14: "heapmeta", 15: "heap", 16: "iheap"}
+MAGIC = {0x053162: "btree", 0x061561: "hash", 0x042253: "queue", 0x074582: "heap"}
+META = {8, 9, 10, 14}
+
+
+def level_ok(t, lvl):
+    if t in (5, 6, 12):           # leaf pages
+        return lvl == 1
+    if t in (3, 4):               # internal pages
+        return 2 <= lvl <= 32
+    return lvl == 0               # overflow / meta / hash / queue / heap
+
+
+def check(buf, off, ps):
+    if len(buf) < 26:
+        return None
+    lf, lo, pgno, prv, nxt, ent, hfo, lvl, typ = struct.unpack_from("<IIIIIHHBB", buf, 0)
+    if typ not in TYPES or pgno != off // ps or not level_ok(typ, lvl):
+        return None
+    if typ in META:
+        magic, ver, psz = struct.unpack_from("<III", buf, 12)
+        if magic not in MAGIC:
+            return None
+        return ("STRONG", f"{TYPES[typ]} magic={MAGIC[magic]} ver={ver} pagesize={psz} lsn=({lf},{lo})")
+    if hfo > ps or ent > ps // 2:
+        return None
+    if typ == 7 and not (ent >= 1 and 0 < hfo <= ps - 26):
+        return None
+    strength = "STRONG" if (lf, lo) == (0, 1) else "weak"
+    return (strength, f"{TYPES[typ]} lsn=({lf},{lo}) prev={prv} next={nxt} entries={ent} hf_offset={hfo} level={lvl}")
+
+
+def scan(path, ps):
+    hits, magic_only = [], []
+    with open(path, "rb") as f:
+        off = 0
+        while True:
+            buf = f.read(ps)
+            if not buf:
+                break
+            r = check(buf, off, ps)
+            if r:
+                hits.append((off, off // ps) + r)
+            elif len(buf) >= 16:
+                m = struct.unpack_from("<I", buf, 12)[0]
+                if m in MAGIC and buf[25] in META:
+                    magic_only.append((off, MAGIC[m]))
+            off += len(buf)
+    return hits, magic_only, off
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--pagesize", type=int, default=4096)
+    ap.add_argument("--quiet", action="store_true")
+    ap.add_argument("files", nargs="+")
+    a = ap.parse_args()
+    total = strong = 0
+    for p in a.files:
+        try:
+            hits, mo, size = scan(p, a.pagesize)
+        except OSError as e:
+            print(f"{p}: ERROR {e}")
+            continue
+        s = sum(1 for h in hits if h[2] == "STRONG")
+        total += len(hits); strong += s
+        if hits or mo:
+            print(f"{p}: {len(hits)} BDB page(s) ({s} strong), {len(mo)} stray meta magic, size={size}")
+            if not a.quiet:
+                for h in hits[:20]:
+                    print(f"    off={h[0]} pgno={h[1]} {h[2]} {h[3]}")
+                for m in mo[:5]:
+                    print(f"    meta-magic off={m[0]} {m[1]}")
+        else:
+            print(f"{p}: clean ({size} bytes)")
+    print(f"TOTAL BDB pages found: {total} ({strong} strong)")
+    return 2 if total else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

qa/fd-guard/repro/tools/tally.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+"""tally.py <fdguard.log> -- attribute every fdguard report to (type, closer function, target class)
+and, for EBADF/NON-OWNER reports, to the code that made the previous close of that fd number.
+
+The point: show that the stray closes ORIGINATE in CNode code (net.cpp CloseSocketDisconnect /
+~CNode closing a number they already closed) and that libevent / BDB / CreateNodeFromAcceptedSocket /
+GetTransaction only appear as victims-turned-amplifiers downstream of that.
+"""
+import collections, re, sys
+
+ORIGIN = ("_ZN5CNode21CloseSocketDisconnectEv", "_ZN5CNodeD1Ev", "_ZN5CNodeD2Ev", "_ZN11CNetCleanupD2Ev")
+NAMES = {"_ZN5CNode21CloseSocketDisconnectEv": "CNode::CloseSocketDisconnect (net.cpp:572)",
+         "_ZN5CNodeD1Ev": "CNode::~CNode (net.cpp:2819)",
+         "_ZN11CNetCleanupD2Ev": "CNetCleanup::~CNetCleanup (net.cpp:2465, shutdown)",
+         "evhttp_connection_free": "libevent evhttp_connection_free (http.c:1244)",
+         "__os_closehandle": "BDB __os_closehandle",
+         "_Z28CreateNodeFromAcceptedSocketjbRK8CAddressS1_": "CreateNodeFromAcceptedSocket (net.cpp:1336)",
+         "_Z14GetTransactionRK7uint256R12CTransactionRS_b": "GetTransaction fclose (main.cpp)",
+         "epoll_nochangelist_add": "libevent epoll ADD", "epoll_nochangelist_del": "libevent epoll DEL/MOD"}
+
+
+def fn(frame_line):
+    m = re.search(r"\+0x[0-9a-f]+ (\S+?)\+0x", frame_line)
+    return m.group(1) if m else "?"
+
+
+def main(path):
+    reps, cur = [], None
+    for line in open(path, errors="replace"):
+        if line.startswith("=== FDGUARD #"):
+            cur = {"hdr": line.strip(), "lines": []}
+            reps.append(cur)
+        elif cur is not None:
+            if line.startswith("=== END"):
+                cur = None
+            else:
+                cur["lines"].append(line.rstrip("\n"))
+    tally = collections.Counter()
+    prev_tally = collections.Counter()
+    db_blk = []
+    for r in reps:
+        typ = r["hdr"].split()[3]
+        L = r["lines"]
+        cls = next((re.search(r"class=(\S+)", l).group(1) for l in L if " target=" in l and "class=" in l), "?")
+        try:
+            i = next(k for k, l in enumerate(L) if l.strip() == "closer stack:")
+            closer = fn(L[i + 1])
+        except StopIteration:
+            closer = "?"
+        prev = "-"
+        for k, l in enumerate(L):
+            if "previous successful close of this fd number" in l:
+                # first frame line after the target= line
+                for l2 in L[k + 1:k + 4]:
+                    if l2.strip().startswith("#0"):
+                        prev = fn(l2)
+                        break
+                break
+        tally[(typ, NAMES.get(closer, closer), cls)] += 1
+        if typ in ("EBADF-CLOSE", "NON-OWNER-CLOSE"):
+            prev_tally[NAMES.get(prev, prev)] += 1
+        if typ == "NON-OWNER-CLOSE" and cls in ("db", "blk"):
+            tgt = next((l.split("target=")[1].split()[0] for l in L if " target=" in l), "?")
+            db_blk.append((r["hdr"].split()[2], r["hdr"].split()[4], NAMES.get(closer, closer), tgt.rsplit("/", 2)[-2:]))
+    print(f"{len(reps)} reports in {path}")
+    print("\n count  type              class  closer")
+    for (typ, closer, cls), n in tally.most_common():
+        print(f"{n:6d}  {typ:17s} {cls:6s} {closer}")
+    print("\nEBADF + NON-OWNER reports by the code that made the PREVIOUS close of the same number:")
+    for k, n in prev_tally.most_common():
+        print(f"{n:6d}  {k}")
+    o = sum(n for (typ, c, cls), n in tally.items() if typ == "NON-OWNER-CLOSE" and cls in ("db", "blk") and c.startswith(("CNode", "CNetCleanup")))
+    a = sum(n for (typ, c, cls), n in tally.items() if typ == "NON-OWNER-CLOSE" and cls in ("db", "blk")) - o
+    print(f"\nNON-OWNER closes of wallet/BDB-log/block files: {o + a}  (by CNode code directly: {o}; by an amplifier: {a})")
+    for d in db_blk[:40]:
+        print("   ", d[0], d[1], "closer=" + d[2], "target=" + "/".join(d[3]))
+
+
+if __name__ == "__main__":
+    main(sys.argv[1])

qa/fd-guard/repro/tools/tcpchurn.py

@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+"""
+Raw TCP / TLS client churn against the target node's P2P port.
+
+dragonxd P2P is TLS 1.3 only (wolfSSL), self-signed, no cert verification. The
+accept path is net.cpp CreateNodeFromAcceptedSocket -> tlsmanager.accept()
+(wolfSSL_accept), and teardown is CloseSocketDisconnect / ~CNode
+(wolfSSL_shutdown + wolfSSL_free + CloseSocket). This driver pokes the accept and
+TLS-teardown/error paths (SSL_read/SSL_write at net.cpp ~1005 and tlsmanager.cpp
+~660-680) by opening many short-lived connections that each misbehave:
+
+  rst      : connect, then abort with RST (SO_LINGER 0) before any TLS bytes
+  fin      : connect, then close cleanly (FIN) before any TLS bytes
+  garbage  : connect, send non-ClientHello bytes, then drop
+  stall    : connect, send a few bytes, hold mid-handshake, drop
+  tlsjunk  : complete a real TLS 1.3 handshake (verify off), send junk, drop
+  tlshalf  : start TLS, send a partial ClientHello record, drop
+
+Loopback only, against a port this harness started. --duration seconds.
+"""
+import argparse, random, socket, ssl, struct, sys, time, os
+
+def mk():
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.settimeout(3.0)
+    return s
+
+def rst_close(s):
+    try:
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
+    except Exception:
+        pass
+    try: s.close()
+    except Exception: pass
+
+def do_rst(host, port):
+    s = mk()
+    try:
+        s.connect((host, port)); rst_close(s)
+    except Exception:
+        rst_close(s)
+
+def do_fin(host, port):
+    s = mk()
+    try:
+        s.connect((host, port)); s.close()
+    except Exception:
+        try: s.close()
+        except Exception: pass
+
+def do_garbage(host, port):
+    s = mk()
+    try:
+        s.connect((host, port))
+        s.sendall(os.urandom(random.randint(16, 400)))
+        time.sleep(random.uniform(0.0, 0.3))
+    except Exception:
+        pass
+    rst_close(s)
+
+def do_stall(host, port):
+    s = mk()
+    try:
+        s.connect((host, port))
+        # a couple of bytes that look like the start of a TLS record, then wait
+        s.sendall(b"\x16\x03\x01")
+        time.sleep(random.uniform(0.5, 2.0))
+    except Exception:
+        pass
+    rst_close(s)
+
+def _ctx():
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    try:
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_3
+    except Exception:
+        pass
+    return ctx
+
+def do_tlsjunk(host, port):
+    raw = mk()
+    try:
+        raw.connect((host, port))
+        tls = _ctx().wrap_socket(raw, server_hostname=None)
+        # handshake completed; the node now expects a P2P version message.
+        # send junk inside the TLS channel, then tear down.
+        try:
+            tls.sendall(os.urandom(random.randint(8, 128)))
+        except Exception:
+            pass
+        time.sleep(random.uniform(0.0, 0.4))
+        try:
+            tls.close()
+        except Exception:
+            rst_close(raw)
+    except Exception:
+        rst_close(raw)
+
+def do_tlshalf(host, port):
+    s = mk()
+    try:
+        s.connect((host, port))
+        # a well-formed TLS record header claiming a long handshake, but we send
+        # only part of it, then drop -> server sits in wolfSSL_accept WANT_READ.
+        body = os.urandom(40)
+        rec = b"\x16\x03\x03" + struct.pack(">H", 512) + body
+        s.sendall(rec)
+        time.sleep(random.uniform(0.3, 1.5))
+    except Exception:
+        pass
+    rst_close(s)
+
+HANDLERS = [do_rst, do_fin, do_garbage, do_stall, do_tlsjunk, do_tlshalf]
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--host", default="127.0.0.1")
+    ap.add_argument("--port", type=int, required=True)
+    ap.add_argument("--duration", type=float, default=300.0)
+    ap.add_argument("--concurrency", type=int, default=6)
+    ap.add_argument("--rate", type=float, default=0.03, help="sleep between bursts")
+    args = ap.parse_args()
+
+    end = time.time() + args.duration
+    counts = {h.__name__: 0 for h in HANDLERS}
+    import threading
+    lock = threading.Lock()
+
+    def worker():
+        while time.time() < end:
+            h = random.choice(HANDLERS)
+            h(args.host, args.port)
+            with lock:
+                counts[h.__name__] += 1
+            time.sleep(random.uniform(0.0, args.rate))
+
+    ts = [threading.Thread(target=worker, daemon=True) for _ in range(args.concurrency)]
+    for t in ts: t.start()
+    for t in ts: t.join()
+
+    total = sum(counts.values())
+    print(f"tcpchurn: {total} connections in {args.duration:.0f}s -> {counts}", flush=True)
+
+if __name__ == "__main__":
+    main()

qa/fd-guard/shim/build.sh

@@ -0,0 +1,15 @@
+#!/bin/sh
+# Build the fdguard LD_PRELOAD shim and its self-test program.
+#   ./build.sh            -> libfdguard.so, fdguard_test
+set -e
+cd "$(dirname "$0")"
+CC=${CC:-gcc}
+$CC -O2 -g -fPIC -shared -Wall -Wextra -Wno-unused-parameter -fvisibility=default \
+    -o libfdguard.so.tmp fdguard.c -ldl -lpthread
+# rename, never overwrite in place: a running process may have the old one mapped
+mv -f libfdguard.so.tmp libfdguard.so
+# the test program keeps its symbols (no -rdynamic on purpose: like dragonxd,
+# owner detection must work from .symtab alone) and frame pointers for clarity
+$CC -O1 -g -fno-omit-frame-pointer -fno-inline -Wall -o fdguard_test.tmp fdguard_test.c -lpthread
+mv -f fdguard_test.tmp fdguard_test
+echo "built: $(pwd)/libfdguard.so $(pwd)/fdguard_test"

qa/fd-guard/shim/fdguard_test.c

@@ -0,0 +1,206 @@
+/*
+ * fdguard_test.c -- exercises libfdguard.so.
+ *
+ * The function names imitate the real owners inside dragonxd so the shim's
+ * built-in domain patterns apply unchanged:
+ *   __os_openhandle / __os_closehandle   Berkeley DB os layer   (domain bdb)
+ *   OpenDiskFile / FlushBlockFile        block-file code         (domain blk)
+ *   ConnectSocketDirectly / CloseSocket  P2P sockets             (domain p2p)
+ *   foreign_stale_close / foreign_*      somebody who holds a stale number
+ *
+ * usage: fdguard_test <dir> <case>   (see main() for the list)
+ */
+#define _GNU_SOURCE
+#include <errno.h>
+#include <fcntl.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/epoll.h>
+#include <sys/socket.h>
+#include <sys/syscall.h>
+#include <time.h>
+#include <unistd.h>
+
+#define NI __attribute__((noinline))
+static char D[512];
+static char *P(const char *rel) { static char b[4][700]; static int i; i = (i + 1) & 3; snprintf(b[i], sizeof b[i], "%s/%s", D, rel); return b[i]; }
+static void say(const char *m) { fprintf(stderr, "[test] %s\n", m); }
+
+/* --- "Berkeley DB" --- */
+NI int __os_openhandle(const char *path) { return open(path, O_RDWR | O_CREAT, 0600); }
+NI int __os_closehandle(int fd) { return close(fd); }
+/* --- "block files" --- */
+NI FILE *OpenDiskFile(const char *path) { FILE *f = fopen(path, "rb+"); if (!f) f = fopen(path, "wb+"); return f; }
+NI int FlushBlockFile(FILE *f) { fflush(f); return fclose(f); }
+/* --- "P2P" --- */
+NI int ConnectSocketDirectly(void) { return socket(AF_INET, SOCK_STREAM, 0); }
+NI int CloseSocket(int *s) { int r = close(*s); if (r) fprintf(stderr, "[test] Socket close failed: %d. Error: %s\n", *s, strerror(errno)); return r; }
+/* --- foreigners --- */
+NI int foreign_stale_close(int fd) { return close(fd); }
+NI int foreign_stale_fclose(FILE *f) { return fclose(f); }
+NI int foreign_dup2(int fd) { int dn = open("/dev/null", O_RDWR); int r = dup2(dn, fd); close(dn); return r; }
+NI int foreign_close_range(int fd) { return close_range((unsigned)fd, (unsigned)fd, 0); }
+NI void foreign_closefrom(int fd) { closefrom(fd); }
+/* --- plain code that opens and closes in the same function --- */
+NI int copy_like(const char *src) { int fd = open(src, O_RDONLY); char b[16]; if (fd >= 0) { (void)!read(fd, b, sizeof b); close(fd); } return fd; }
+NI int plain_open(const char *p) { return open(p, O_RDWR | O_CREAT, 0600); }
+NI int plain_close(int fd) { return close(fd); }
+
+static int fd_valid(int fd) { return fcntl(fd, F_GETFD) != -1; }
+
+static int c_double(void)
+{
+    int fd = plain_open(P("plain.txt"));
+    plain_close(fd);
+    int r = plain_close(fd);  /* EBADF */
+    return r == -1 && errno == EBADF ? 0 : 1;
+}
+static int c_dbowner(void) { int fd = __os_openhandle(P("regtest/wallet.dat")); int r = __os_closehandle(fd); return r; }
+static int c_dbforeign(void)
+{
+    int fd = __os_openhandle(P("regtest/wallet.dat"));
+    foreign_stale_close(fd);              /* HIT */
+    int r = __os_closehandle(fd);         /* BDB's own close then fails: EBADF (unless deny) */
+    (void)r;
+    return 0;
+}
+static int c_deny(void)
+{
+    int fd = __os_openhandle(P("regtest/wallet.dat"));
+    foreign_stale_close(fd);              /* HIT + DENIED */
+    if (!fd_valid(fd)) { say("deny FAILED: fd closed"); return 1; }
+    if (write(fd, "x", 1) != 1) { say("deny FAILED: write"); return 1; }
+    return __os_closehandle(fd);
+}
+static int c_blkowner(void) { FILE *f = OpenDiskFile(P("regtest/blocks/blk00000.dat")); fputs("blk", f); return FlushBlockFile(f); }
+static int c_blkraw(void)
+{
+    FILE *f = OpenDiskFile(P("regtest/blocks/blk00001.dat"));
+    foreign_stale_close(fileno(f));       /* HIT: raw close under a live FILE */
+    FlushBlockFile(f);                     /* EBADF from fclose */
+    return 0;
+}
+static int c_stalefile(void)
+{
+    FILE *f = OpenDiskFile(P("regtest/blocks/blk00002.dat"));
+    int n = fileno(f);
+    foreign_stale_close(n);               /* HIT #1 */
+    int w = __os_openhandle(P("regtest/wallet.dat"));
+    if (w != n) fprintf(stderr, "[test] note: wallet got fd %d, FILE had %d\n", w, n);
+    FlushBlockFile(f);                    /* HIT #2: fclose of a stale FILE whose number is now wallet.dat */
+    if (w == n) __os_closehandle(w);      /* EBADF: BDB lost its fd */
+    return 0;
+}
+static int c_reuse(void)
+{
+    int s = ConnectSocketDirectly();
+    int keep = s;
+    CloseSocket(&s);                      /* legit */
+    int w = __os_openhandle(P("regtest/wallet.dat"));   /* reuses the number */
+    if (w != keep) fprintf(stderr, "[test] note: wallet got fd %d, socket was %d\n", w, keep);
+    s = keep;
+    CloseSocket(&s);                      /* HIT: stale socket number now = wallet.dat */
+    __os_closehandle(w);                  /* EBADF */
+    return 0;
+}
+static int c_dup2(void) { int w = __os_openhandle(P("regtest/wallet.dat")); foreign_dup2(w); __os_closehandle(w); return 0; }
+static int c_range(void)
+{
+    int w = __os_openhandle(P("regtest/wallet.dat"));
+    foreign_close_range(w);                /* HIT */
+    w = __os_openhandle(P("regtest/database/log.0000000001"));
+    int hi = fcntl(w, F_DUPFD, 600);       /* move it high so closefrom only hits it */
+    close(w);
+    foreign_closefrom(hi);                /* HIT (domain mismatch: opened by test main via fcntl dup? -> fallback db) */
+    return 0;
+}
+static int c_owner_misc(void)
+{
+    /* nothing here may be flagged */
+    copy_like(P("regtest/wallet.dat"));           /* common-caller */
+    int fd = __os_openhandle(P("regtest/blocks/blk00003.dat"));
+    FILE *f = fdopen(fd, "rb+");                   /* ownership moves to FILE */
+    FlushBlockFile(f);
+    int s = ConnectSocketDirectly(); CloseSocket(&s);
+    fd = (int)syscall(SYS_openat, AT_FDCWD, P("regtest/wallet.dat"), O_RDWR);  /* untracked open */
+    __os_closehandle(fd);                          /* fallback:bdb */
+    return 0;
+}
+static int c_fallback_hit(void)
+{
+    int fd = (int)syscall(SYS_openat, AT_FDCWD, P("regtest/wallet.dat"), O_RDWR);  /* untracked open */
+    foreign_stale_close(fd);                       /* HIT via fallback rule */
+    return 0;
+}
+static int c_epoll(void)
+{
+    int ep = epoll_create1(0);
+    int fd = plain_open(P("plain2.txt"));
+    struct epoll_event ev = { .events = EPOLLIN };
+    int r = epoll_ctl(ep, EPOLL_CTL_ADD, fd, &ev);  /* EPERM: regular file */
+    close(fd);
+    r = epoll_ctl(ep, EPOLL_CTL_ADD, fd, &ev);      /* EBADF */
+    close(ep);
+    return r == -1 ? 0 : 1;
+}
+static int c_selfclose(void)
+{
+    /* close every fd from 3 to 1100 by number, like a buggy daemonizer */
+    for (int i = 3; i < 1100; i++) close(i);
+    return 0;
+}
+
+/* concurrency + micro-benchmark: open/close churn in many threads */
+static int nthr = 8, iters = 20000;
+static void *worker(void *arg)
+{
+    long id = (long)arg;
+    char p[700]; snprintf(p, sizeof p, "%s/thr%ld.txt", D, id);
+    for (int i = 0; i < iters; i++) {
+        int s = ConnectSocketDirectly(); CloseSocket(&s);
+        int fd = plain_open(p); plain_close(fd);
+        FILE *f = fopen(p, "r"); if (f) fclose(f);
+        if ((i & 255) == 0) { int w = __os_openhandle(P("regtest/wallet.dat")); __os_closehandle(w); }
+    }
+    return NULL;
+}
+static int c_threads(void)
+{
+    const char *e = getenv("T_ITERS"); if (e) iters = atoi(e);
+    pthread_t t[64];
+    struct timespec a, b; clock_gettime(CLOCK_MONOTONIC, &a);
+    for (long i = 0; i < nthr; i++) pthread_create(&t[i], NULL, worker, (void *)i);
+    for (int i = 0; i < nthr; i++) pthread_join(t[i], NULL);
+    clock_gettime(CLOCK_MONOTONIC, &b);
+    double ns = (b.tv_sec - a.tv_sec) * 1e9 + (b.tv_nsec - a.tv_nsec);
+    long ops = (long)nthr * iters * 3;
+    printf("threads=%d iters=%d ops=%ld (socket/open/fopen + close each) wall=%.3fs  %.0f ns/op-pair/thread\n",
+           nthr, iters, ops, ns / 1e9, ns * nthr / ops);
+    return 0;
+}
+
+int main(int argc, char **argv)
+{
+    if (argc < 3) { fprintf(stderr, "usage: %s <dir> <case>\n", argv[0]); return 2; }
+    snprintf(D, sizeof D, "%s", argv[1]);
+    const char *c = argv[2];
+    if (!strcmp(c, "double")) return c_double();
+    if (!strcmp(c, "dbowner")) return c_dbowner();
+    if (!strcmp(c, "dbforeign")) return c_dbforeign();
+    if (!strcmp(c, "deny")) return c_deny();
+    if (!strcmp(c, "blkowner")) return c_blkowner();
+    if (!strcmp(c, "blkraw")) return c_blkraw();
+    if (!strcmp(c, "stalefile")) return c_stalefile();
+    if (!strcmp(c, "reuse")) return c_reuse();
+    if (!strcmp(c, "dup2")) return c_dup2();
+    if (!strcmp(c, "range")) return c_range();
+    if (!strcmp(c, "ownermisc")) return c_owner_misc();
+    if (!strcmp(c, "fallback")) return c_fallback_hit();
+    if (!strcmp(c, "epoll")) return c_epoll();
+    if (!strcmp(c, "selfclose")) return c_selfclose();
+    if (!strcmp(c, "threads")) return c_threads();
+    fprintf(stderr, "unknown case %s\n", c);
+    return 2;
+}

qa/fd-guard/shim/run_tests.sh

@@ -0,0 +1,97 @@
+#!/bin/bash
+# Self-test for libfdguard.so.  Every case lists the report types it must
+# produce (exactly, in order) -- an empty list means "no report at all".
+#   ./run_tests.sh            (builds first)
+set -u
+cd "$(dirname "$0")"
+./build.sh >/dev/null || { echo "build failed"; exit 1; }
+SO=$PWD/libfdguard.so
+T=$PWD/t
+rm -rf "$T"; mkdir -p "$T/regtest/blocks" "$T/regtest/database"
+pass=0; fail=0
+
+run() {  # run <case> [ENV=VAL ...]  -> sets RC, LOG
+  local c=$1; shift
+  LOG=$T/$c${SUFFIX:-}.log; rm -f "$LOG"
+  env FDGUARD_EXE=fdguard_test FDGUARD_LOG="$LOG" "$@" LD_PRELOAD="$SO" ./fdguard_test "$T" "$c" >"$T/$c${SUFFIX:-}.out" 2>&1
+  RC=$?
+}
+types() { grep '^=== FDGUARD' "$LOG" 2>/dev/null | awk '{print $4}' | tr '\n' ' ' | sed 's/ $//'; }
+reasons() { grep -o 'reason=[^ ]*' "$LOG" 2>/dev/null | sed 's/reason=//' | tr '\n' ' ' | sed 's/ $//'; }
+check() {  # check <name> <cond-description> <actual> <expected>
+  if [ "$3" == "$4" ]; then pass=$((pass+1)); printf 'PASS %-12s %s\n' "$1" "$2";
+  else fail=$((fail+1)); printf 'FAIL %-12s %s\n      got:  [%s]\n      want: [%s]\n' "$1" "$2" "$3" "$4"; fi
+}
+
+# case | expected report types | expected reasons
+while IFS='|' read -r c want wantr; do
+  [ -z "$c" ] && continue
+  run "$c"
+  check "$c" "reports" "$(types)" "$want"
+  check "$c" "reasons" "$(reasons)" "$wantr"
+  check "$c" "exit 0" "$RC" "0"
+done <<'EOF'
+double|EBADF-CLOSE|
+dbowner||
+dbforeign|NON-OWNER-CLOSE EBADF-CLOSE|DOMAIN-MISMATCH
+blkowner||
+blkraw|NON-OWNER-CLOSE EBADF-CLOSE|stdio:RAW-CLOSE-UNDER-LIVE-FILE
+stalefile|NON-OWNER-CLOSE NON-OWNER-CLOSE EBADF-CLOSE|stdio:RAW-CLOSE-UNDER-LIVE-FILE stdio:STALE-FILE
+reuse|NON-OWNER-CLOSE EBADF-CLOSE|DOMAIN-MISMATCH
+dup2|NON-OWNER-CLOSE|DOMAIN-MISMATCH
+range|NON-OWNER-CLOSE NON-OWNER-CLOSE NON-OWNER-CLOSE|DOMAIN-MISMATCH DOMAIN-MISMATCH fallback:DB-FILE-CLOSED-OUTSIDE-BDB
+ownermisc||
+fallback|NON-OWNER-CLOSE|fallback:DB-FILE-CLOSED-OUTSIDE-BDB
+epoll|EPOLL-CTL-EPERM EPOLL-CTL-EBADF|epoll_ctl(ADD)-failed epoll_ctl(ADD)-failed
+EOF
+
+# the previous close of the reused number must be the first CloseSocket (proves the double close)
+run reuse
+check reuse "history shows 1st CloseSocket on a socket" \
+  "$(grep -A2 'previous successful close' "$LOG" | head -3 | grep -c -e 'class=sock dom=p2p' -e 'CloseSocket+')" "2"
+# the stale FILE report must name wallet.dat as the victim and OpenDiskFile as the FILE's opener
+run stalefile
+check stalefile "victim=wallet.dat, FILE opened in OpenDiskFile" \
+  "$(sed -n '/#2 NON-OWNER/,/=== END #2/p' "$LOG" | grep -m1 -c 'target=.*wallet.dat class=db') $(sed -n '/#2 NON-OWNER/,/=== END #2/p' "$LOG" | grep -A3 'which had been opened by: fopen' | grep -c 'OpenDiskFile+')" "1 1"
+
+# the log fd survives a close-everything loop and the attempt is reported
+run selfclose
+check selfclose "log-fd close refused + reported" "$(grep -c 'CLOSE-OF-FDGUARD-LOG-FD' "$LOG")" "1"
+check selfclose "log still writable afterwards (final stats)" "$(grep -c '^S .* final' "$LOG")" "1"
+
+# abort mode: first non-owner close -> SIGABRT (rc 134), report already written
+SUFFIX=_abort run dbforeign FDGUARD_ABORT=1
+check abort "rc=134 (SIGABRT)" "$RC" "134"
+check abort "report written before abort" "$(types)" "NON-OWNER-CLOSE"
+# abort=2 also aborts on EBADF
+SUFFIX=_abort2 run double FDGUARD_ABORT=2
+check abort2 "rc=134 on EBADF" "$RC" "134"
+
+# deny mode: wallet fd survives the foreign close; no EBADF afterwards
+SUFFIX=_deny run deny FDGUARD_DENY=1
+check deny "exit 0 (fd still valid + writable)" "$RC" "0"
+check deny "reported as DENIED, no EBADF" "$(types) $(grep -c 'ACTION=DENIED' "$LOG")" "NON-OWNER-CLOSE 1"
+
+# extra classes: guard=all with the owner cases must still be silent
+SUFFIX=_all run ownermisc FDGUARD_GUARD=all
+check guard-all "ownermisc silent with FDGUARD_GUARD=all" "$(types)" ""
+
+# trace-all writes one T line per open/close
+SUFFIX=_trace run dbowner FDGUARD_TRACE_ALL=1
+check trace "T lines for open+close" "$(grep -c '^T ' "$LOG")" "2"
+
+# passive in non-matching processes
+LOG=$T/passive.log; env FDGUARD_EXE=dragonxd FDGUARD_LOG="$LOG" LD_PRELOAD="$SO" ./fdguard_test "$T" dbforeign >/dev/null 2>&1
+check passive "no log when exe does not match" "$( [ -e "$LOG" ] && echo exists || echo none)" "none"
+
+# concurrency: 8 threads x 5000 x (socket, open, fopen) churn, guard=all -> no reports
+SUFFIX=_thr run threads T_ITERS=5000 FDGUARD_GUARD=all
+check threads "no reports under 8-thread churn (guard=all)" "$(types)" ""
+check threads "stats: HITS=0 EBADF=0" "$(grep '^S .*final' "$LOG" | grep -o 'HITS=[0-9]* EBADF=[0-9]*')" "HITS=0 EBADF=0"
+
+# symbolizer resolves file:line for the test binary
+check symbolize "addr2line resolved CloseSocket to fdguard_test.c" \
+  "$(./symbolize.py "$T/reuse.log" | grep -m1 -o 'in CloseSocket at .*fdguard_test.c:[0-9]*' | sed 's/.*fdguard_test.c:[0-9]*/ok/')" "ok"
+
+echo "== $pass passed, $fail failed"
+[ $fail -eq 0 ]

qa/fd-guard/shim/symbolize.py

@@ -0,0 +1,212 @@
+#!/usr/bin/env python3
+"""
+symbolize.py -- turn fdguard log frames into function + file:line.
+
+Every stack frame in an fdguard report looks like
+    #3 0x5b96b433c991 /path/to/module+0x1991 c_dbforeign+0x2d [bdb]
+where "+0x1991" is the ELF virtual address inside that module (runtime address
+minus the module's load bias, so PIE executables and shared libraries need no
+further adjustment; for a non-PIE executable it equals the absolute address).
+Frames are return addresses, so we ask addr2line about vaddr-1 (the call
+instruction), with -f -C -i (function, demangle, inline chain).
+
+backtrace_symbols_fd lines "module(+0x1991)[0x...]" are symbolized too.
+
+usage:
+  symbolize.py [options] LOG [LOG...]          annotated log on stdout
+  symbolize.py --summary LOG                    one line per distinct report
+options:
+  --map OLD=NEW      read module OLD from path NEW instead (repeatable),
+                     e.g. --map /path/to/deployed/dragonxd=./dragonxd.unstripped
+  --addr2line PATH   addr2line binary (default: addr2line)
+  --only-reports     drop T/S lines (trace + stats) from the output
+"""
+import os
+import re
+import subprocess
+import sys
+from collections import defaultdict, OrderedDict
+
+FRAME_RE = re.compile(r'^(\s*#\d+ 0x[0-9a-f]+ )(\S+)\+0x([0-9a-f]+)(.*)$')
+BSFD_RE = re.compile(r'^(\S+)\(\+0x([0-9a-f]+)\)\[0x[0-9a-f]+\]\s*$')
+HDR_RE = re.compile(r'^=== FDGUARD #(\d+) (\S+)')
+
+
+def parse_args(argv):
+    opts = {'map': {}, 'addr2line': 'addr2line', 'summary': False, 'only': False, 'logs': []}
+    it = iter(argv)
+    for a in it:
+        if a == '--map':
+            old, new = next(it).split('=', 1)
+            opts['map'][old] = new
+        elif a.startswith('--map='):
+            old, new = a[6:].split('=', 1)
+            opts['map'][old] = new
+        elif a == '--addr2line':
+            opts['addr2line'] = next(it)
+        elif a == '--summary':
+            opts['summary'] = True
+        elif a == '--only-reports':
+            opts['only'] = True
+        elif a in ('-h', '--help'):
+            print(__doc__)
+            sys.exit(0)
+        else:
+            opts['logs'].append(a)
+    if not opts['logs']:
+        opts['logs'] = ['-']
+    return opts
+
+
+def read_lines(paths):
+    out = []
+    for p in paths:
+        f = sys.stdin if p == '-' else open(p, errors='replace')
+        out.extend(f.read().splitlines())
+        if f is not sys.stdin:
+            f.close()
+    return out
+
+
+def resolve_module(mod, opts, known):
+    if mod in opts['map']:
+        return opts['map'][mod]
+    if os.path.exists(mod):
+        return mod
+    base = os.path.basename(mod)
+    for k in known:
+        if os.path.basename(k) == base and os.path.exists(opts['map'].get(k, k)):
+            return opts['map'].get(k, k)
+    return None
+
+
+def run_addr2line(binary, addrs, a2l):
+    """addrs: list of int vaddrs (already -1'd). returns {addr: [(func, loc), ...]}"""
+    res = {}
+    if not addrs:
+        return res
+    uniq = sorted(set(addrs))
+    try:
+        p = subprocess.run([a2l, '-a', '-f', '-C', '-i', '-e', binary] + ['0x%x' % a for a in uniq],
+                           stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, check=False)
+    except FileNotFoundError:
+        return res
+    cur = None
+    lines = p.stdout.splitlines()
+    i = 0
+    while i < len(lines):
+        ln = lines[i]
+        if re.match(r'^0x[0-9a-f]+$', ln):
+            cur = int(ln, 16)
+            res[cur] = []
+            i += 1
+            continue
+        func = ln
+        loc = lines[i + 1] if i + 1 < len(lines) else '??:0'
+        if cur is not None:
+            res[cur].append((func, loc))
+        i += 2
+    return res
+
+
+def main():
+    opts = parse_args(sys.argv[1:])
+    lines = read_lines(opts['logs'])
+
+    if opts['summary']:
+        summary(lines)
+        return
+
+    # pass 1: collect modules + addresses
+    want = defaultdict(set)
+    known = OrderedDict()
+    for ln in lines:
+        m = FRAME_RE.match(ln)
+        if m:
+            known[m.group(2)] = 1
+            want[m.group(2)].add(max(int(m.group(3), 16) - 1, 0))
+    for ln in lines:
+        m = BSFD_RE.match(ln)
+        if m:
+            want[m.group(1)].add(max(int(m.group(2), 16) - 1, 0))
+
+    table = {}
+    for mod, addrs in want.items():
+        path = resolve_module(mod, opts, known)
+        if not path:
+            continue
+        table[mod] = run_addr2line(path, list(addrs), opts['addr2line'])
+
+    def fmt(mod, vaddr):
+        r = table.get(mod, {}).get(vaddr)
+        if not r:
+            return []
+        out = []
+        for j, (func, loc) in enumerate(r):
+            if func == '??' and loc.startswith('??'):
+                continue
+            # addr2line -i lists the innermost (inlined) frame first
+            tag = 'inlined:' if j < len(r) - 1 else 'in'
+            out.append('%s %s at %s' % (tag, func, loc))
+        return out
+
+    w = sys.stdout.write
+    for ln in lines:
+        if opts['only'] and (ln.startswith('T ') or ln.startswith('S ')):
+            continue
+        w(ln + '\n')
+        m = FRAME_RE.match(ln)
+        if m:
+            v = max(int(m.group(3), 16) - 1, 0)
+            for s in fmt(m.group(2), v):
+                w('         -> ' + s + '\n')
+            continue
+        m = BSFD_RE.match(ln)
+        if m:
+            v = max(int(m.group(2), 16) - 1, 0)
+            for s in fmt(m.group(1), v):
+                w('         -> ' + s + '\n')
+
+
+def summary(lines):
+    """one line per (type, reason, class, first exe closer frame)"""
+    groups = OrderedDict()
+    cur = None
+    in_closer = False
+    for ln in lines:
+        h = HDR_RE.match(ln)
+        if h:
+            cur = {'type': h.group(2), 'reason': '', 'cls': '', 'top': '', 'call': ''}
+            in_closer = False
+            continue
+        if cur is None:
+            continue
+        s = ln.strip()
+        if s.startswith('call:'):
+            cur['call'] = s.split('(')[0][6:]
+        elif s.startswith('target=') and ' class=' in s and not cur['cls']:
+            cur['cls'] = s.rsplit('class=', 1)[-1]
+        elif s.startswith('verdict='):
+            cur['reason'] = s.split('reason=', 1)[-1].split(' ')[0]
+        elif s == 'closer stack:':
+            in_closer = True
+        elif in_closer and FRAME_RE.match(ln):
+            parts = s.split(' ')
+            if len(parts) >= 4 and not cur['top'] and '/libc.so' not in parts[2] and '/libstdc++' not in parts[2]:
+                cur['top'] = parts[3]
+        elif in_closer and not s.startswith('#'):
+            in_closer = False
+        if s.startswith('=== END'):
+            k = (cur['type'], cur['call'], cur['reason'], cur['cls'], cur['top'])
+            groups[k] = groups.get(k, 0) + 1
+            cur = None
+    print('%6s  %-22s %-12s %-6s %-40s %s' % ('count', 'type', 'call', 'class', 'reason', 'first closer frame'))
+    for (t, c, r, cl, top), n in groups.items():
+        print('%6d  %-22s %-12s %-6s %-40s %s' % (n, t[:22], c, cl, r[:40], top))
+    stats = [ln for ln in lines if ln.startswith('S ')]
+    if stats:
+        print('\nlast stats line:\n  ' + stats[-1])
+
+
+if __name__ == '__main__':
+    main()

src/Makefile.gtest.include

@@ -15,7 +15,10 @@ hush_gtest_SOURCES = \
 	gtest/test_hdtransparent.cpp \
 	gtest/test_mnemonic_compat.cpp \
 	gtest/test_stratum_jobid.cpp \
-	gtest/test_wallet_notedata.cpp
+	gtest/test_wallet_notedata.cpp \
+	gtest/test_net_closesocket.cpp \
+	gtest/test_fd_defense.cpp \
+	gtest/test_wallet_dbenv.cpp
 
 hush_gtest_CPPFLAGS = $(AM_CPPFLAGS) -DMULTICORE -fopenmp -DBINARY_OUTPUT -DCURVE_ALT_BN128 -DSTATIC $(BITCOIN_INCLUDES)
 hush_gtest_CXXFLAGS = $(AM_CXXFLAGS) $(PIE_FLAGS)

src/gtest/test_fd_defense.cpp

@@ -0,0 +1,83 @@
+// Copyright (c) 2016-2024 The Hush developers
+// Distributed under the GPLv3 software license, see the accompanying
+// file COPYING or https://www.gnu.org/licenses/gpl-3.0.en.html
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <errno.h>
+#include <stdexcept>
+#include <stdio.h>
+#include <unistd.h>
+
+#include <boost/filesystem.hpp>
+#include <boost/thread.hpp>
+
+#include "chainparams.h"
+#include "init.h"
+#include "main.h"
+#include "noui.h"
+#include "util.h"
+#include "validationinterface.h"
+
+extern std::atomic<bool> fRequestShutdown;
+
+// A block file opened for reading must not be writable through its fd: a stray write through a reused fd number
+// (the wolfSSL CloseSocket macro let the wallet's BDB fd be closed behind its back) used to land in it.
+TEST(FdDefense, ReadOnlyBlockFileIsNotWritable) {
+    SelectParams(CBaseChainParams::MAIN);
+    const boost::filesystem::path dir = boost::filesystem::temp_directory_path() / boost::filesystem::unique_path("fddefense-%%%%%%%%");
+    const bool hadDatadir = mapArgs.count("-datadir") > 0;
+    const std::string oldDatadir = hadDatadir ? mapArgs["-datadir"] : "";
+    boost::filesystem::create_directories(dir);
+    mapArgs["-datadir"] = dir.string();
+    ClearDatadirCache();
+
+    const CDiskBlockPos pos(0, 0);
+    const boost::filesystem::path blk = GetBlockPosFilename(pos, "blk");
+    boost::filesystem::create_directories(blk.parent_path());
+    {
+        FILE* f = fopen(blk.string().c_str(), "wb");
+        ASSERT_TRUE(f != nullptr);
+        fputs("block bytes", f);
+        fclose(f);
+    }
+
+    FILE* ro = OpenBlockFile(pos, true);
+    ASSERT_TRUE(ro != nullptr);
+    char c = 0;
+    EXPECT_EQ(1u, fread(&c, 1, 1, ro)) << "still readable";
+    errno = 0;
+    EXPECT_EQ(-1, pwrite(fileno(ro), "X", 1, 0)) << "a write through a read-only block file's fd must fail";
+    EXPECT_EQ(EBADF, errno);
+    fclose(ro);
+
+    FILE* rw = OpenBlockFile(pos, false);
+    ASSERT_TRUE(rw != nullptr);
+    EXPECT_EQ(1, pwrite(fileno(rw), "b", 1, 0)) << "the writable open still writes";
+    fclose(rw);
+
+    if (hadDatadir) mapArgs["-datadir"] = oldDatadir; else mapArgs.erase("-datadir");
+    ClearDatadirCache();
+    boost::filesystem::remove_all(dir);
+}
+
+// A wallet notification that throws (a wallet DB failure) requests a clean shutdown instead of escaping the
+// notifier thread into std::terminate; a successful one doesn't; thread interruption still propagates.
+TEST(FdDefense, WalletNotifyFailureShutsDownCleanly) {
+    noui_connect();   // the daemon connects these UI signals at startup; the error path shows a message box
+    fRequestShutdown = false;
+    EXPECT_TRUE(NotifyWalletsOrShutdown([] {}));
+    EXPECT_FALSE(ShutdownRequested());
+
+    EXPECT_FALSE(NotifyWalletsOrShutdown([] { throw std::runtime_error("CWalletDB::ListAccountCreditDebit(): error scanning DB"); }));
+    EXPECT_TRUE(ShutdownRequested());
+    fRequestShutdown = false;
+
+    EXPECT_FALSE(NotifyWalletsOrShutdown([] { throw 42; }));
+    EXPECT_TRUE(ShutdownRequested());
+    fRequestShutdown = false;
+
+    EXPECT_THROW(NotifyWalletsOrShutdown([] { throw boost::thread_interrupted(); }), boost::thread_interrupted);
+    EXPECT_FALSE(ShutdownRequested());
+}

src/gtest/test_net_closesocket.cpp

@@ -0,0 +1,50 @@
+// Copyright (c) 2016-2024 The Hush developers
+// Distributed under the GPLv3 software license, see the accompanying
+// file COPYING or https://www.gnu.org/licenses/gpl-3.0.en.html
+
+#include <gtest/gtest.h>
+
+#include <fcntl.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include "net.h"
+#include "util/sock.h"
+
+// Regression test for the wolfSSL CloseSocket macro (see net.h): closing a CNode's socket must invalidate
+// hSocket, so neither a repeated CloseSocketDisconnect nor ~CNode can close that fd number again once another
+// owner (the wallet's BDB files, a block file, an RPC socket) has been handed it.
+TEST(NetCloseSocket, ClosedNodeNeverClosesAReusedFdNumber) {
+    int pair[2];
+    ASSERT_EQ(0, socketpair(AF_UNIX, SOCK_STREAM, 0, pair));
+    const int number = pair[0];
+
+    CNode* node = new CNode(number, CAddress(), "", true);   // inbound: sends no version message
+    node->CloseSocketDisconnect();
+    EXPECT_EQ(INVALID_SOCKET, node->hSocket) << "CloseSocketDisconnect left the closed fd number in hSocket";
+    EXPECT_EQ(-1, fcntl(number, F_GETFD)) << "the node's socket wasn't closed";
+
+    // Another owner gets the freed number, as the kernel's lowest-free-fd rule does in a busy node.
+    const int other = open("/dev/null", O_RDONLY);
+    ASSERT_GE(other, 0);
+    ASSERT_EQ(number, dup2(other, number));
+
+    node->CloseSocketDisconnect();   // e.g. a send error after the disconnect was already handled
+    delete node;                     // ~CNode
+    EXPECT_NE(-1, fcntl(number, F_GETFD)) << "a closed CNode closed the fd number's new owner";
+
+    close(number);
+    close(other);
+    close(pair[1]);
+}
+
+// util/sock.cpp's CloseSocket is the one that must be called: it closes once and invalidates.
+TEST(NetCloseSocket, CloseSocketInvalidates) {
+    int pair[2];
+    ASSERT_EQ(0, socketpair(AF_UNIX, SOCK_STREAM, 0, pair));
+    SOCKET s = pair[0];
+    EXPECT_TRUE(CloseSocket(s));
+    EXPECT_EQ(INVALID_SOCKET, s);
+    EXPECT_FALSE(CloseSocket(s)) << "a second close of an invalidated handle must be a no-op";
+    close(pair[1]);
+}

src/gtest/test_wallet_dbenv.cpp

@@ -0,0 +1,58 @@
+// Copyright (c) 2016-2024 The Hush developers
+// Distributed under the GPLv3 software license, see the accompanying
+// file COPYING or https://www.gnu.org/licenses/gpl-3.0.en.html
+
+#include <gtest/gtest.h>
+
+#include <boost/filesystem.hpp>
+
+#include "wallet/db.h"
+
+namespace {
+
+// A fresh environment in a temp dir with one committed record in "wallet.dat", so database/ holds a log file.
+class WalletDbEnv : public ::testing::Test {
+protected:
+    boost::filesystem::path dir;
+    CDBEnv env;
+
+    void SetUp() override {
+        dir = boost::filesystem::temp_directory_path() / boost::filesystem::unique_path("dbenv-%%%%%%%%");
+        boost::filesystem::create_directories(dir);
+        ASSERT_TRUE(env.Open(dir));
+        Db db(env.dbenv, DB_CXX_NO_EXCEPTIONS);
+        ASSERT_EQ(0, db.open(NULL, "wallet.dat", "main", DB_BTREE, DB_CREATE | DB_AUTO_COMMIT, 0));
+        char k[] = "key", v[] = "value";
+        Dbt key(k, sizeof k), value(v, sizeof v);
+        ASSERT_EQ(0, db.put(NULL, &key, &value, 0));
+        ASSERT_EQ(0, db.close(0));
+        ASSERT_TRUE(boost::filesystem::exists(Log()));
+    }
+    void TearDown() override {
+        env.Close();
+        boost::system::error_code ec;
+        boost::filesystem::remove_all(dir, ec);
+    }
+    boost::filesystem::path Log() const { return dir / "database" / "log.0000000001"; }
+};
+
+} // namespace
+
+TEST_F(WalletDbEnv, CleanShutdownRemovesTheLogs) {
+    env.Flush(true);
+    EXPECT_FALSE(boost::filesystem::exists(dir / "database"));
+}
+
+// BDB0060 PANIC (a page write failed, e.g. through an fd closed behind BDB's back): the logs may hold committed
+// changes wallet.dat never got, and the next start's DB_RECOVER needs them.
+TEST_F(WalletDbEnv, PanickedEnvironmentKeepsTheLogs) {
+    ASSERT_EQ(0, env.dbenv->set_flags(DB_PANIC_ENVIRONMENT, 1));
+    env.Flush(true);
+    EXPECT_TRUE(boost::filesystem::exists(Log()));
+}
+
+TEST_F(WalletDbEnv, FailedCheckpointKeepsTheLogs) {
+    env.CheckpointLSN("no-such-file.dat");   // lsn_reset fails
+    env.Flush(true);
+    EXPECT_TRUE(boost::filesystem::exists(Log()));
+}

src/main.cpp

@@ -5922,7 +5922,10 @@ FILE* OpenDiskFile(const CDiskBlockPos &pos, const char *prefix, bool fReadOnly)
         return NULL;
     boost::filesystem::path path = GetBlockPosFilename(pos, prefix);
     boost::filesystem::create_directories(path.parent_path());
-    FILE* file = fopen(path.string().c_str(), "rb+");
+    // Read-only opens get a read-only FILE (every fReadOnly caller only reads). With "rb+", a write through a
+    // reused fd number landed here: when the wolfSSL CloseSocket macro (see net.h) closed the wallet's BDB fd
+    // behind its back, BDB's next page write went to whatever reopened that number -- a block file opened here.
+    FILE* file = fopen(path.string().c_str(), fReadOnly ? "rb" : "rb+");
     if (!file && !fReadOnly)
         file = fopen(path.string().c_str(), "wb+");
     if (!file) {

src/net.cpp

@@ -45,6 +45,13 @@
 #include <wolfssl/options.h>
 #include <wolfssl/ssl.h>
 #include <hush/tlsmanager.cpp>
+
+// CloseSocket must be util/sock.cpp's CloseSocket(SOCKET&), which invalidates the handle it closes. If
+// wolfSSL's `#define CloseSocket(s) close(s)` survived into this file (see net.h), each close here would leave
+// its stale fd number behind to be closed again after another thread reused it.
+#ifdef CloseSocket
+#error "CloseSocket is a macro here (wolfssl/wolfio.h): it would bypass CloseSocket(SOCKET&) in util/sock.cpp"
+#endif
 using namespace hush;
 extern int32_t HUSH_TESTNODE;
 extern uint8_t ASSETCHAINS_CLEARNET;

src/net.h

@@ -57,6 +57,14 @@
 #define byte unsigned char
 #include <wolfssl/ssl.h>
 #undef byte
+// wolfssl/wolfio.h (pulled in by ssl.h) does `#ifndef CloseSocket / #define CloseSocket(s) close(s)`. Left
+// defined, it turns every CloseSocket(hSocket) after this point -- all 15 calls in net.cpp, including the ones in
+// CloseSocketDisconnect and ~CNode -- into a bare close() that never sets hSocket = INVALID_SOCKET, so util/sock.cpp's
+// CloseSocket(SOCKET&) is never called. Every P2P disconnect then closed its fd number again later (a repeated
+// CloseSocketDisconnect, then ~CNode), by which time another thread often owned that number: wallet.dat or a
+// database/log.* fd (BDB then wrote wallet pages into whatever got the number next, e.g. a blocks/*.dat file
+// opened "rb+"), a block file, or an HTTP RPC socket. wolfio.h is include-guarded, so it can't come back.
+#undef CloseSocket
 
 class CAddrMan;
 class CBlockIndex;

src/validationinterface.cpp

@@ -86,6 +86,30 @@ struct CachedBlockData {
         pindex(pindex), oldTrees(oldTrees), txConflicted(txConflicted) {}
 };
 
+bool NotifyWalletsOrShutdown(const std::function<void()>& notify)
+{
+    // The wallet can throw here on a database failure, e.g. CWalletDB::ListAccountCreditDebit's "error scanning
+    // DB" after BDB lost its file. Escaping this thread, that std::terminate()d the node (TraceThread logs and
+    // rethrows) with nothing flushed. Stop cleanly instead: going on would leave the wallet behind the chain,
+    // missing the block it failed on.
+    try {
+        notify();
+        return true;
+    } catch (const boost::thread_interrupted&) {
+        throw;
+    } catch (const std::exception& e) {
+        PrintExceptionContinue(&e, "ThreadNotifyWallets()");
+    } catch (...) {
+        PrintExceptionContinue(NULL, "ThreadNotifyWallets()");
+    }
+    LogPrintf("*** %s\n", "Wallet failed to process a block notification; shutting down");
+    StartShutdown();
+    uiInterface.ThreadSafeMessageBox(
+        _("Error: A fatal internal error occurred, see debug.log for details"),
+        "", CClientUIInterface::MSG_ERROR);
+    return false;
+}
+
 void ThreadNotifyWallets(CBlockIndex *pindexLastTip)
 {
     // If pindexLastTip == nullptr, the wallet is at genesis.
@@ -208,13 +232,16 @@ void ThreadNotifyWallets(CBlockIndex *pindexLastTip)
                 break;
             }
 
-            // Let wallets know transactions went from 1-confirmed to
-            // 0-confirmed or conflicted:
-            for (const CTransaction &tx : block.vtx) {
-                SyncWithWallets(tx, NULL);
-            }
-            // Update cached incremental witnesses
-            GetMainSignals().ChainTip(pindexLastTip, &block, boost::none);
+            if (!NotifyWalletsOrShutdown([&] {
+                // Let wallets know transactions went from 1-confirmed to
+                // 0-confirmed or conflicted:
+                for (const CTransaction &tx : block.vtx) {
+                    SyncWithWallets(tx, NULL);
+                }
+                // Update cached incremental witnesses
+                GetMainSignals().ChainTip(pindexLastTip, &block, boost::none);
+            }))
+                return;
 
             // On to the next block!
             pindexLastTip = pindexLastTip->pprev;
@@ -246,17 +273,20 @@ void ThreadNotifyWallets(CBlockIndex *pindexLastTip)
                 break;
             }
 
-            // Tell wallet about transactions that went from mempool
-            // to conflicted:
-            for (const CTransaction &tx : blockData.txConflicted) {
-                SyncWithWallets(tx, NULL);
-            }
-            // ... and about transactions that got confirmed:
-            for (const CTransaction &tx : block.vtx) {
-                SyncWithWallets(tx, &block);
-            }
-            // Update cached incremental witnesses
-            GetMainSignals().ChainTip(blockData.pindex, &block, blockData.oldTrees);
+            if (!NotifyWalletsOrShutdown([&] {
+                // Tell wallet about transactions that went from mempool
+                // to conflicted:
+                for (const CTransaction &tx : blockData.txConflicted) {
+                    SyncWithWallets(tx, NULL);
+                }
+                // ... and about transactions that got confirmed:
+                for (const CTransaction &tx : block.vtx) {
+                    SyncWithWallets(tx, &block);
+                }
+                // Update cached incremental witnesses
+                GetMainSignals().ChainTip(blockData.pindex, &block, blockData.oldTrees);
+            }))
+                return;
 
             // This block is done!
             pindexLastTip = blockData.pindex;

src/validationinterface.h

@@ -8,6 +8,7 @@
 #define HUSH_VALIDATIONINTERFACE_H
 
 #include <boost/signals2/signal.hpp>
+#include <functional>
 #include "zcash/IncrementalMerkleTree.hpp"
 
 class CBlock;
@@ -76,4 +77,8 @@ CMainSignals& GetMainSignals();
 
 void ThreadNotifyWallets(CBlockIndex *pindexLastTip);
 
+/** Runs one block's wallet notifications for ThreadNotifyWallets. Returns false, after logging the error and
+ *  requesting a shutdown, if they threw (e.g. a wallet database failure); thread interruption propagates. */
+bool NotifyWalletsOrShutdown(const std::function<void()>& notify);
+
 #endif // HUSH_VALIDATIONINTERFACE_H

src/wallet/db.cpp

@@ -61,6 +61,7 @@ void CDBEnv::Reset()
     dbenv = new DbEnv(DB_CXX_NO_EXCEPTIONS);
     fDbEnvInit = false;
     fMockDb = false;
+    fKeepLogs = false;
 }
 
 CDBEnv::CDBEnv() : dbenv(NULL)
@@ -276,10 +277,12 @@ bool CDBEnv::Salvage(const std::string& strFile, bool fAggressive, std::vector<C
 
 void CDBEnv::CheckpointLSN(const std::string& strFile)
 {
-    dbenv->txn_checkpoint(0, 0, 0);
+    if (dbenv->txn_checkpoint(0, 0, 0) != 0)
+        fKeepLogs = true;
     if (fMockDb)
         return;
-    dbenv->lsn_reset(strFile.c_str(), 0);
+    if (dbenv->lsn_reset(strFile.c_str(), 0) != 0)
+        fKeepLogs = true;
 }
 
 
@@ -498,10 +501,16 @@ void CDBEnv::Flush(bool fShutdown)
                 // Move log data to the dat file
                 CloseDb(strFile);
                 LogPrint("db", "CDBEnv::Flush: %s checkpoint\n", strFile);
-                dbenv->txn_checkpoint(0, 0, 0);
+                int ret = dbenv->txn_checkpoint(0, 0, 0);
+                if (ret != 0) {
+                    LogPrintf("CDBEnv::Flush: %s checkpoint failed: %s\n", strFile, DbEnv::strerror(ret));
+                    fKeepLogs = true;
+                }
                 LogPrint("db", "CDBEnv::Flush: %s detach\n", strFile);
-                if (!fMockDb)
-                    dbenv->lsn_reset(strFile.c_str(), 0);
+                if (!fMockDb && (ret = dbenv->lsn_reset(strFile.c_str(), 0)) != 0) {
+                    LogPrintf("CDBEnv::Flush: %s lsn_reset failed: %s\n", strFile, DbEnv::strerror(ret));
+                    fKeepLogs = true;
+                }
                 LogPrint("db", "CDBEnv::Flush: %s closed\n", strFile);
                 mapFileUseCount.erase(mi++);
             } else
@@ -511,10 +520,24 @@ void CDBEnv::Flush(bool fShutdown)
         if (fShutdown) {
             char** listp;
             if (mapFileUseCount.empty()) {
-                dbenv->log_archive(&listp, DB_ARCH_REMOVE);
-                Close();
-                if (!fMockDb)
-                    boost::filesystem::remove_all(boost::filesystem::path(strPath) / "database");
+                // After a failed flush, or once the environment has panicked (BDB0060, e.g. after a page
+                // write failed), wallet.dat can be missing committed changes that only the logs hold.
+                // Removing them would turn a wallet the next start recovers (DB_RECOVER) into one that
+                // silently lost them. If the logs are unusable, the next start moves database/ aside.
+                // (get_flags itself fails with DB_RUNRECOVERY once the environment has panicked.)
+                u_int32_t envFlags = 0;
+                if (dbenv->get_flags(&envFlags) != 0 || (envFlags & DB_PANIC_ENVIRONMENT))
+                    fKeepLogs = true;
+                if (fKeepLogs && !fMockDb) {
+                    LogPrintf("CDBEnv::Flush: the wallet database was not flushed cleanly; keeping %s for recovery at the next start\n",
+                              (boost::filesystem::path(strPath) / "database").string());
+                    Close();
+                } else {
+                    dbenv->log_archive(&listp, DB_ARCH_REMOVE);
+                    Close();
+                    if (!fMockDb)
+                        boost::filesystem::remove_all(boost::filesystem::path(strPath) / "database");
+                }
             }
         }
     }

src/wallet/db.h

@@ -28,6 +28,7 @@
 #include "sync.h"
 #include "version.h"
 
+#include <atomic>
 #include <map>
 #include <string>
 #include <vector>
@@ -53,6 +54,9 @@ class CDBEnv
 private:
     bool fDbEnvInit;
     bool fMockDb;
+    // A checkpoint or lsn_reset failed: the logs in database/ may hold the only copy of committed
+    // wallet changes, so the shutdown flush must leave them for DB_RECOVER at the next start.
+    std::atomic<bool> fKeepLogs;
     // Don't change into boost::filesystem::path, as that can result in
     // shutdown problems/crashes caused by a static initialized internal pointer.
     std::string strPath;