diff --git a/dash-spv-bench/.gitignore b/dash-spv-bench/.gitignore index fb948c9f4..31fa05194 100644 --- a/dash-spv-bench/.gitignore +++ b/dash-spv-bench/.gitignore @@ -1,6 +1,5 @@ bench-results/ bench-storage/ -profiles/ *.lock @@ -17,8 +16,5 @@ chain-data/ results/ .bin/ -# FlameGraph tooling clone (run.sh --flame), fetched on demand. -.flamegraph/ - # Wallet mnemonics (one BIP39 phrase per line) — secrets, never committed. wallets.txt diff --git a/dash-spv-bench/Cargo.toml b/dash-spv-bench/Cargo.toml index 0cf11a8b9..58aa104b0 100644 --- a/dash-spv-bench/Cargo.toml +++ b/dash-spv-bench/Cargo.toml @@ -20,6 +20,15 @@ tracing = "0.1" tracing-subscriber = { version = "0.3.20", features = ["env-filter"] } indicatif = "0.18" +[target.'cfg(target_os = "linux")'.dependencies] +pprof = { version = "0.14", features = ["flamegraph"], optional = true } +tikv-jemallocator = { version = "0.7", features = ["profiling"], optional = true } +jemalloc_pprof = { version = "0.9", features = ["flamegraph", "symbolize"], optional = true } + +[features] +cpu-profile = ["dep:pprof"] +heap-profile = ["dep:tikv-jemallocator", "dep:jemalloc_pprof"] + [[bin]] name = "dash-spv-bench" path = "src/main.rs" diff --git a/dash-spv-bench/run.sh b/dash-spv-bench/run.sh index 09758e204..79e60be35 100755 --- a/dash-spv-bench/run.sh +++ b/dash-spv-bench/run.sh @@ -1,115 +1,68 @@ #!/bin/bash # -# dash-spv benchmark driver — runs ONE scenario end to end (build, bring up peers, sync, report). +# dash-spv benchmark driver: builds the client, brings up local peers when the +# scenario has them, syncs inside the client container and archives the results. # -# Usage: -# ./run.sh Run the scenario -# ./run.sh --flame Same, under the sampler (perf on Linux, sample on macOS) -# -> profiles/flamegraph.svg -# --wallets Wallets for this run: a file with one BIP39 mnemonic per -# line. No file => the run has no wallet. +# Usage: ./run.sh ... [--flame] [--memory-snapshot] [--wallets ] # -# ./run.sh 'scenarios/mainnet.*' Several scenarios: any argument that is not a file is -# ./run.sh scenarios/local.*.yml treated as a glob (quote it to let run.sh expand it, or -# ./run.sh scenarios/*.yml let your shell do it). Patterns resolve against the -# current directory and then against scenarios/, so -# 'mainnet.*' works from anywhere. +# --flame CPU flamegraph sampled by the client -> flamegraph.svg +# --memory-snapshot live-heap flamegraph at the RSS peak -> heap-peak.svg +# --wallets one BIP39 mnemonic per line (default: wallets.txt) # -# Every invocation, one scenario or twenty, writes results// containing per scenario -# .log stdout of the run (build, bring-up, summary) -# .run.log the sync trace, kept because bench-storage/ is wiped by the next run -# .summary.txt the metrics block -# plus results.tsv and report.md over all of them. A scenario that fails is recorded and the -# rest still run. -# -# RUST_LOG can be set to tweak logging, e.g. RUST_LOG=info ./run.sh scenarios/local.1ideal.yml -# It is the tracing filter for BOTH log sinks and overrides the defaults, which are -# terminal = "warn,dash_spv_bench=info" (kept light so the live bars stay readable) -# file = "warn,dash_spv=debug,dash_spv_bench=debug" (debug, for offline analysis) -# -# Outputs land in bench-storage/ (gitignored, wiped at the start of each run): -# run.log full trace of the sync (debug by default) for offline analysis -# summary.txt metrics + per-wallet tx/balance fingerprint (also printed to stdout) -# -# bench-storage/ is wiped at the start of every run, so both are archived into results/ too. -# plus the SPV storage the run produced (block_headers/, filters/, ...) +# An argument that is not a file is a glob, tried against the current directory +# and then scenarios/. Every invocation writes results// with, per +# scenario, .log, .run.log, .summary.txt and any flamegraph, plus +# results.tsv and report.md. RUST_LOG overrides the client's log filters. set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" -COMPOSE_FILE="" # local mode generates one here; deleted on exit -DASH_VERSION="23.1.7" -IMAGE="dash-spv-bench/dashd:${DASH_VERSION}" +SELF="${SCRIPT_DIR}/$(basename "${BASH_SOURCE[0]}")" +INVOCATION_DIR="${PWD}" +IMAGE="dash-spv-bench/dashd:23.1.7" CLIENT_IMAGE="dash-spv-bench/client:1" -# Rust image for the Linux build, taken from the workspace's own pin so the two -# cannot drift: with a mismatched tag the image spends every run having rustup -# fetch the pinned toolchain before it can compile anything. RUST_CHANNEL="$(sed -n 's/^channel *= *"\(.*\)"/\1/p' "${REPO_ROOT}/rust-toolchain.toml" 2>/dev/null || true)" RUST_IMAGE="rust:${RUST_CHANNEL:-1.89}-bookworm" -# Separate from the host's `target/`: a different triple, and sharing one -# directory across both would make every switch a full rebuild. PROJECT="spv-bench" STATE="${SCRIPT_DIR}/.clonedir" -FLAME_SVG="${SCRIPT_DIR}/profiles/flamegraph.svg" -FLAMEGRAPH_DIR="${SCRIPT_DIR}/.flamegraph" # FlameGraph tooling clone (gitignored, inside the package) CHAIN_DIR="${SCRIPT_DIR}/chain-data" -# Absolute, because the script `cd`s to its own directory below: a relative `$0` -# stops resolving after that, which broke both `--help` and re-invoking self. -SELF="${SCRIPT_DIR}/$(basename "${BASH_SOURCE[0]}")" - -INVOCATION_DIR="${PWD}" abspath() { case "$1" in /*) printf '%s\n' "$1" ;; *) printf '%s\n' "${INVOCATION_DIR%/}/$1" ;; esac; } - cd "${SCRIPT_DIR}" -FLAME=0 +FEATURES="" +FLAGS=() SCN_ARGS=() WALLETS_ARG="" while [ $# -gt 0 ]; do case "$1" in - --flame) FLAME=1; shift ;; - --wallets) WALLETS_ARG="${2:?--wallets needs a file path}"; shift 2 ;; - -h | --help) sed -n '3,24p' "${SELF}"; exit 0 ;; + --flame) FEATURES="${FEATURES} cpu-profile"; FLAGS+=("$1"); shift ;; + --memory-snapshot) FEATURES="${FEATURES} heap-profile"; FLAGS+=("$1"); shift ;; + --wallets) WALLETS_ARG="$(abspath "${2:?--wallets needs a file path}")"; FLAGS+=("$1" "${WALLETS_ARG}"); shift 2 ;; + -h | --help) sed -n '3,/^set /p' "${SELF}" | sed '$d'; exit 0 ;; -*) echo "unknown flag: $1" >&2; exit 1 ;; *) SCN_ARGS+=("$1"); shift ;; esac done -[ "${#SCN_ARGS[@]}" -gt 0 ] || { echo "usage: $0 ... [--flame] [--wallets ]" >&2; exit 1; } +[ "${#SCN_ARGS[@]}" -gt 0 ] || { echo "usage: $0 ... [--flame] [--memory-snapshot] [--wallets ]" >&2; exit 1; } -# Resolve each argument to scenario files. -# -# A bare path stays a bare path, so the original single-file invocation is -# untouched. Anything that is not a file is treated as a glob — which covers -# both a quoted pattern (`'scenarios/mainnet.*'`, expanded here) and one the -# caller's shell already expanded into several arguments. Patterns are tried -# against the invocation directory first and then against `scenarios/`, so -# `mainnet.*` works from anywhere. SCN_FILES=() +add_matches() { + local m + while IFS= read -r m; do + case "${m}" in /*) ;; *) m="$1/${m}" ;; esac + if [ -f "${m}" ]; then SCN_FILES+=("${m}"); fi + done < <(cd "$1" && compgen -G "$2") +} for arg in "${SCN_ARGS[@]}"; do - if [ -f "${arg}" ] || [ -f "$(abspath "${arg}")" ]; then - SCN_FILES+=("$(abspath "${arg}")") - continue - fi - matched=0 - while IFS= read -r hit; do - [ -f "${hit}" ] || continue - SCN_FILES+=("${hit}") - matched=1 - done < <(cd "${INVOCATION_DIR}" 2>/dev/null && shopt -s nullglob && printf '%s\n' ${arg} | while IFS= read -r m; do abspath "${m}"; done - cd "${SCRIPT_DIR}/scenarios" 2>/dev/null && shopt -s nullglob && printf '%s\n' ${arg} | while IFS= read -r m; do printf '%s\n' "${SCRIPT_DIR}/scenarios/${m}"; done) - [ "${matched}" -eq 1 ] || { echo "Error: no scenario matched: ${arg}" >&2; exit 1; } + n="${#SCN_FILES[@]}" + add_matches "${INVOCATION_DIR}" "${arg}" + [ "${#SCN_FILES[@]}" -gt "${n}" ] || add_matches "${SCRIPT_DIR}/scenarios" "${arg}" + [ "${#SCN_FILES[@]}" -gt "${n}" ] || { echo "Error: no scenario matched: ${arg}" >&2; exit 1; } done -# Run each scenario through a fresh invocation of this script and collect the -# numbers. Re-invoking rather than looping in place keeps the per-scenario path -# — exports, compose, traps, teardown — byte for byte what a single-file run has -# always done, so the extension cannot change the thing it is measuring. -# -# Taken for one scenario as much as for twenty: naming a file and naming a glob -# that happens to match one file are the same request, and having them leave -# their results in different shapes is a trap for anyone scripting on top. The -# child is marked so it runs the scenario instead of wrapping it again. +# Each scenario runs in a fresh invocation of this script, so one that fails +# cannot take the others with it. if [ -z "${BENCH_BATCH:-}" ]; then RUN_TS="$(date +%Y%m%d-%H%M%S)" OUT_DIR="${SCRIPT_DIR}/results/${RUN_TS}" @@ -117,11 +70,6 @@ if [ -z "${BENCH_BATCH:-}" ]; then TSV="${OUT_DIR}/results.tsv" REPORT="${OUT_DIR}/report.md" printf 'scenario\tcompleted\ttotal_ms\tblock_headers_ms\tfilter_headers_ms\tfilters_ms\ttransactions\tconfirmed_sat\tpeak_rss_mib\n' >"${TSV}" - - child_flags=() - [ "${FLAME}" -eq 1 ] && child_flags+=(--flame) - [ -n "${WALLETS_ARG}" ] && child_flags+=(--wallets "${WALLETS_ARG}") - metric() { awk -F':[[:space:]]*' -v k="$2" '$1==k {gsub(/[[:space:]]+$/,"",$2); print $2; exit}' "$1"; } echo "==> ${#SCN_FILES[@]} scenario(s); results in ${OUT_DIR}" @@ -129,37 +77,25 @@ if [ -z "${BENCH_BATCH:-}" ]; then name="$(basename "${f}" .yml)" log="${OUT_DIR}/${name}.log" echo "===== ${name} =====" - # A scenario that fails must not take the batch with it: record it and move - # on, or one bad run costs every result after it. BENCH_BATCH=1 BENCH_ARCHIVE_DIR="${OUT_DIR}" BENCH_ARCHIVE_NAME="${name}" \ - "${SELF}" "${f}" "${child_flags[@]+"${child_flags[@]}"}" 2>&1 | tee "${log}" || true - # Read the metrics from the summary the child archived, not from its stdout: - # the summary is written by the binary and is plain text either way, while a - # pty-captured log carries the bars' escape codes and carriage returns. - src="${log}" - if [ -s "${OUT_DIR}/${name}.summary.txt" ]; then src="${OUT_DIR}/${name}.summary.txt"; fi - wallet_line="$(grep -m1 -o 'confirmed_sat=[0-9]*' "${src}" || true)" - printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \ - "${name}" \ - "$(metric "${src}" completed)" \ - "$(metric "${src}" total_ms)" \ - "$(metric "${src}" block_headers_ms)" \ - "$(metric "${src}" filter_headers_ms)" \ - "$(metric "${src}" filters_ms)" \ - "$(metric "${src}" transactions)" \ - "$(sed -n 's/.*confirmed_sat=\([0-9]*\).*/\1/p' <<<"${wallet_line}")" \ - "$(metric "${src}" peak_rss_mib)" \ - >>"${TSV}" + "${SELF}" "${f}" ${FLAGS[@]+"${FLAGS[@]}"} 2>&1 | tee "${log}" || true + src="${OUT_DIR}/${name}.summary.txt" + [ -s "${src}" ] || src="${log}" + row="${name}" + for key in completed total_ms block_headers_ms filter_headers_ms filters_ms transactions; do + row+=$'\t'"$(metric "${src}" "${key}")" + done + row+=$'\t'"$(grep -m1 -o 'confirmed_sat=[0-9]*' "${src}" | cut -d= -f2 || true)" + row+=$'\t'"$(metric "${src}" peak_rss_mib)" + echo "${row}" >>"${TSV}" done { echo "# dash-spv bench — ${RUN_TS}" echo - echo "| scenario | completed | total_ms | headers_ms | filter_headers_ms | filters_ms | transactions | confirmed_sat | peak_rss_mib |" - echo "|---|---|---|---|---|---|---|---|---|" - tail -n +2 "${TSV}" | awk -F'\t' '{printf "| %s | %s | %s | %s | %s | %s | %s | %s | %s |\n", $1,$2,$3,$4,$5,$6,$7,$8,$9}' + awk -F'\t' '{ line = "|"; for (i = 1; i <= NF; i++) line = line " " $i " |"; print line + if (NR == 1) { line = "|"; for (i = 1; i <= NF; i++) line = line "---|"; print line } }' "${TSV}" } >"${REPORT}" - echo echo "==> report: ${REPORT}" cat "${REPORT}" @@ -167,33 +103,27 @@ if [ -z "${BENCH_BATCH:-}" ]; then fi SCN_FILE="${SCN_FILES[0]}" -[ -f "${SCN_FILE}" ] || { echo "Error: scenario file not found: ${SCN_FILE}" >&2; exit 1; } -YQ_VERSION="v4.44.6" -ensure_yq() { - if command -v yq >/dev/null 2>&1; then YQ=yq; return 0; fi - local bin="${SCRIPT_DIR}/.bin/yq" - if [ ! -x "${bin}" ]; then +if command -v yq >/dev/null 2>&1; then + YQ=yq +else + YQ="${SCRIPT_DIR}/.bin/yq" + if [ ! -x "${YQ}" ]; then + os="$(uname -s | tr '[:upper:]' '[:lower:]')" + case "$(uname -m)" in x86_64 | amd64) arch=amd64 ;; aarch64 | arm64) arch=arm64 ;; *) arch="$(uname -m)" ;; esac + echo "==> fetching yq into .bin/yq" mkdir -p "${SCRIPT_DIR}/.bin" - local os arch - os="$(uname -s | tr '[:upper:]' '[:lower:]')"; arch="$(uname -m)" - case "${arch}" in x86_64 | amd64) arch=amd64 ;; aarch64 | arm64) arch=arm64 ;; esac - echo "==> fetching yq ${YQ_VERSION} (${os}/${arch}) into .bin/yq" - curl -fsSL "https://github.com/mikefarah/yq/releases/download/${YQ_VERSION}/yq_${os}_${arch}" \ - -o "${bin}" || { echo "Error: could not download yq; install it manually." >&2; exit 1; } - chmod +x "${bin}" + curl -fsSL "https://github.com/mikefarah/yq/releases/download/v4.44.6/yq_${os}_${arch}" -o "${YQ}" \ + || { echo "Error: could not download yq; install it manually." >&2; exit 1; } + chmod +x "${YQ}" fi - YQ="${bin}" -} -ensure_yq -scn() { "${YQ}" "$1" "${SCN_FILE}"; } # evaluate a yq expression against the scenario file - -_peer_group() { - "${YQ}" ".peers[$1] | [.count, .latency_ms // 0, .jitter_ms // 0, .loss_pct // 0, .rate_kbit // 0, .corrupt_pct // 0, .reorder_pct // 0] | @tsv" "${SCN_FILE}" -} +fi +scn() { "${YQ}" "$1" "${SCN_FILE}"; } -build_netem() { - local lat="$1" jit="$2" loss="$3" rate="$4" corrupt="$5" reorder="$6" a="" +# `tc netem` arguments for the link described at yq path $1. +netem_args() { + local lat jit loss rate corrupt reorder a="" + read -r lat jit loss rate corrupt reorder <<<"$(scn "$1 | [.latency_ms // 0, .jitter_ms // 0, .loss_pct // 0, .rate_kbit // 0, .corrupt_pct // 0, .reorder_pct // 0] | @tsv")" [ "${lat}" != 0 ] && { a="delay ${lat}ms"; [ "${jit}" != 0 ] && a="${a} ${jit}ms"; } [ "${loss}" != 0 ] && a="${a} loss ${loss}%" [ "${rate}" != 0 ] && a="${a} rate ${rate}kbit" @@ -202,38 +132,8 @@ build_netem() { echo "${a# }" } -# The measured client's own link shaping, if the scenario asks for one -client_netem() { - local lat jit loss rate corrupt reorder - read -r lat jit loss rate corrupt reorder <<<"$("${YQ}" \ - '[.client.latency_ms // 0, .client.jitter_ms // 0, .client.loss_pct // 0, .client.rate_kbit // 0, .client.corrupt_pct // 0, .client.reorder_pct // 0] | @tsv' \ - "${SCN_FILE}")" - build_netem "${lat}" "${jit}" "${loss}" "${rate}" "${corrupt}" "${reorder}" -} - -# The client's bandwidth cap on its own, so the container can mirror it onto -# ingress -client_rate_kbit() { - "${YQ}" '.client.rate_kbit // 0' "${SCN_FILE}" -} - -peers_summary() { - local ng g count lat jit loss rate corrupt reorder tag out="" - ng="$(scn '.peers | length')" - case "${ng}" in ''|null|*[!0-9]*) ng=0 ;; esac - for ((g = 0; g < ng; g++)); do - read -r count lat jit loss rate corrupt reorder <<<"$(_peer_group "${g}")" - tag="${lat}ms" - [ "${jit}" != 0 ] && tag="${tag}±${jit}" - [ "${loss}" != 0 ] && tag="${tag}/${loss}%loss" - [ "${rate}" != 0 ] && tag="${tag}/${rate}kbit" - out="${out}, ${count}×${tag}" - done - echo "${out#, }" -} - emit_compose() { - local out="$1" peer_cpus="$2" + local out="$1" groups g count n peer=0 cat >"${out}" <
>"${out}" <>"${out}" </dev/null \ + if [ "\$\${INGRESS_RATE_KBIT:-0}" != 0 ]; then + ip link add ifb0 type ifb \ && ip link set ifb0 up \ && tc qdisc add dev eth0 handle ffff: ingress \ && tc filter add dev eth0 parent ffff: protocol all prio 1 u32 \ match u32 0 0 action mirred egress redirect dev ifb0 \ - && tc qdisc add dev ifb0 root netem rate \$\${INGRESS_RATE_KBIT}kbit; then - echo "client netem (ingress): rate \$\${INGRESS_RATE_KBIT}kbit via ifb0" - else - tc qdisc del dev eth0 ingress 2>/dev/null || true - if tc qdisc add dev eth0 handle ffff: ingress 2>/dev/null \ - && tc filter add dev eth0 parent ffff: protocol all prio 1 u32 \ - match u32 0 0 action police rate \$\${INGRESS_RATE_KBIT}kbit \ - burst \$\${INGRESS_BURST_K}k mtu 64k conform-exceed drop; then - echo "client police (ingress): rate \$\${INGRESS_RATE_KBIT}kbit burst \$\${INGRESS_BURST_K}k" - echo "NOTE: no ifb, so ingress is POLICED (drops) not shaped (queues) — throughput lands within ~3% but the loss pattern differs; do not compare against ifb runs" - else - echo "WARNING: ingress shaping failed (NET_ADMIN? no ifb and no act_police?)" - echo "WARNING: DOWNLOADS ARE UNSHAPED — rate_kbit is NOT enforced and this run is not comparable" - fi - fi + && tc qdisc add dev ifb0 root netem rate \$\${INGRESS_RATE_KBIT}kbit \ + || { echo "ERROR: cannot shape the client's downloads (no ifb in this kernel?)"; exit 1; } + echo "client netem (ingress): rate \$\${INGRESS_RATE_KBIT}kbit via ifb0" fi exec /usr/local/bin/dash-spv-bench SERVICE } - MODE="$(scn '.mode // "local"')" case "${MODE}" in local | testnet | mainnet) ;; *) echo "Error: mode must be 'local', 'testnet' or 'mainnet' (got '${MODE}')" >&2; exit 1 ;; esac -export BENCH_MODE="${MODE}" export CLONE_DIR="${CLONE_DIR:-/nonexistent}" -CLIENT_NETEM="$(client_netem)" -CLIENT_RATE_KBIT="$(client_rate_kbit)" +CLIENT_NETEM="$(netem_args .client)" +CLIENT_RATE_KBIT="$(scn '.client.rate_kbit // 0')" case "${CLIENT_RATE_KBIT}" in '' | *[!0-9]*) echo "Error: client.rate_kbit must be a whole number of kbit (got '${CLIENT_RATE_KBIT}')" >&2; exit 1 ;; esac -# Burst for the policer fallback: ~0.5s of the rate (kbit/16 => kbytes), floored -# so a very small rate still gets a workable bucket. Unused on the ifb path. -CLIENT_INGRESS_BURST_K=$(( CLIENT_RATE_KBIT / 16 )) -if [ "${CLIENT_INGRESS_BURST_K}" -lt 32 ]; then CLIENT_INGRESS_BURST_K=32; fi BENCH_CPUS="$(scn '.cpus // ""')" -BENCH_MAX_PEERS="$(scn '.max_peers // ""')" # only if set; else the ClientConfig default +BENCH_MAX_PEERS="$(scn '.max_peers // ""')" export BENCH_MAX_PEERS - -BENCH_WALLET_FILE="${SCRIPT_DIR}/wallets.txt" -[ -n "${WALLETS_ARG}" ] && BENCH_WALLET_FILE="$(abspath "${WALLETS_ARG}")" -export BENCH_WALLET_FILE +BENCH_WALLET_FILE="${WALLETS_ARG:-${SCRIPT_DIR}/wallets.txt}" export BENCH_STORAGE_DIR="${SCRIPT_DIR}/bench-storage" DESC="$(scn '.description // ""')" -[ -n "${DESC}" ] && echo "==> description: ${DESC}" +[ -z "${DESC}" ] || echo "==> description: ${DESC}" -BLOCKS="" if [ "${MODE}" = local ]; then - BLOCKS="$(scn '.blocks // 1000000')" - export BENCH_HEIGHT="${BLOCKS}" - unset BENCH_START_HEIGHT # local always syncs from genesis + BENCH_HEIGHT="$(scn '.blocks // 1000000')" + export BENCH_HEIGHT + unset BENCH_START_HEIGHT else BENCH_PEERS="$(scn '.peers // [] | join(",")')" export BENCH_PEERS - sh="$(scn '.start_height // ""')"; [ -n "${sh}" ] && export BENCH_START_HEIGHT="${sh}" + start_height="$(scn '.start_height // ""')" + [ -z "${start_height}" ] || export BENCH_START_HEIGHT="${start_height}" fi -expand_cpus() { - local part lo hi - local IFS=, - for part in $1; do - case "${part}" in - *-*) lo="${part%-*}"; hi="${part#*-}"; seq "${lo}" "${hi}" ;; - *) echo "${part}" ;; - esac - done -} - -CPU_PREFIX=() +# The client gets BENCH_CPUS, docker peers get every other core. +BENCH_PEER_CPUS="" if [ -n "${BENCH_CPUS}" ]; then ncpu="$(getconf _NPROCESSORS_ONLN 2>/dev/null || sysctl -n hw.ncpu 2>/dev/null || nproc 2>/dev/null || echo 0)" - bench_cores="$(expand_cpus "${BENCH_CPUS}" | sort -nu)" - if [ "${ncpu}" -gt 0 ]; then - peer_cores="" - for i in $(seq 0 $((ncpu - 1))); do - grep -qxF "${i}" <<<"${bench_cores}" || peer_cores="${peer_cores},${i}" - done - BENCH_PEER_CPUS="${peer_cores#,}" - fi - - if [ -n "${CLIENT_NETEM}" ]; then - # The client is a container: `cpuset` on the service does the pinning that - # `taskset` does for a host run, and it works on macOS where taskset does - # not exist at all. Emitted by `emit_client_service` from BENCH_CPUS. - echo "==> pinning the measured client container to CPUs ${BENCH_CPUS}${BENCH_PEER_CPUS:+; docker peers to ${BENCH_PEER_CPUS}}" - elif command -v taskset >/dev/null 2>&1; then - CPU_PREFIX=(taskset -c "${BENCH_CPUS}") - echo "==> pinning the measured run to CPUs ${BENCH_CPUS}${BENCH_PEER_CPUS:+; docker peers to ${BENCH_PEER_CPUS}}" - else - echo "==> note: 'taskset' not found (e.g. macOS); running the measured binary UNPINNED${BENCH_PEER_CPUS:+ (docker peers still pinned to ${BENCH_PEER_CPUS})}" >&2 - fi -fi - -# Where the measured binary will be, decided before the compose that mounts -# it is written. Everything lands in the workspace `target/`: a cross build -# gets cargo's own per-triple subdirectory, so it cannot collide with a host -# build, and neither duplicates the dependency graph. -if [ -n "${CLIENT_NETEM}" ] && [ "$(uname -s)" != Linux ]; then - CROSS_TRIPLE="$(docker run --rm "${RUST_IMAGE}" rustc -vV | sed -n 's/^host: //p')" - [ -n "${CROSS_TRIPLE}" ] || { echo "could not read the build image's target triple" >&2; exit 1; } - BIN="${REPO_ROOT}/target/${CROSS_TRIPLE}/release/dash-spv-bench" -else - CROSS_TRIPLE="" - BIN="${REPO_ROOT}/target/release/dash-spv-bench" -fi - -# A compose file is needed for the peers (local mode) and for the client -# container (any mode with a `client:` section) — testnet/mainnet shape the -# client against the real network, with no peer services at all. -if [ "${MODE}" = local ] || [ -n "${CLIENT_NETEM}" ]; then - COMPOSE_FILE="$(mktemp "${SCRIPT_DIR}/.scenario.XXXXXX")" - mv "${COMPOSE_FILE}" "${COMPOSE_FILE}.yml" - COMPOSE_FILE="${COMPOSE_FILE}.yml" - npeers="$(emit_compose "${COMPOSE_FILE}" "${BENCH_PEER_CPUS:-}")" - if [ "${MODE}" = local ]; then - echo "==> scenario '$(basename "${SCN_FILE}" .yml)': ${npeers} peers [$(peers_summary)], cpus=${BENCH_CPUS:-}, blocks=${BLOCKS}" - fi + bench_cores="$(IFS=,; for part in ${BENCH_CPUS}; do case "${part}" in *-*) seq "${part%-*}" "${part#*-}" ;; *) echo "${part}" ;; esac; done)" + for ((i = 0; i < ncpu; i++)); do + grep -qxF "${i}" <<<"${bench_cores}" || BENCH_PEER_CPUS="${BENCH_PEER_CPUS},${i}" + done + BENCH_PEER_CPUS="${BENCH_PEER_CPUS#,}" + echo "==> pinning the client container to CPUs ${BENCH_CPUS}${BENCH_PEER_CPUS:+; docker peers to ${BENCH_PEER_CPUS}}" fi -[ -n "${CLIENT_NETEM}" ] && echo "==> client link shaped: ${CLIENT_NETEM}" +# Built in the Rust image so the binary links against the client image's glibc. +BIN="${REPO_ROOT}/target/bench/release/dash-spv-bench" +echo "==> building bench binary in ${RUST_IMAGE}" +docker run --rm -v "${REPO_ROOT}:/src" -w /src --user "$(id -u):$(id -g)" \ + -e CARGO_HOME=/src/target/bench/cargo-home -e CARGO_TARGET_DIR=/src/target/bench \ + -e CARGO_NET_GIT_FETCH_WITH_CLI=true -e CARGO_PROFILE_RELEASE_DEBUG=line-tables-only \ + "${RUST_IMAGE}" cargo build --release -p dash-spv-bench ${FEATURES:+--features "${FEATURES# }"} +[ -x "${BIN}" ] || { echo "build produced no binary at ${BIN}" >&2; exit 1; } + +COMPOSE_FILE="$(mktemp "${SCRIPT_DIR}/.scenario.XXXXXX")" +mv "${COMPOSE_FILE}" "${COMPOSE_FILE}.yml" +COMPOSE_FILE="${COMPOSE_FILE}.yml" +emit_compose "${COMPOSE_FILE}" +[ -z "${CLIENT_NETEM}" ] || echo "==> client link shaped: ${CLIENT_NETEM}" compose() { docker compose -p "${PROJECT}" -f "${COMPOSE_FILE}" "$@"; } +teardown() { + compose down --remove-orphans >/dev/null 2>&1 || true + [ ! -f "${STATE}" ] || rm -rf "$(cat "${STATE}")" "${STATE}" 2>/dev/null || true + rm -rf "${SCRIPT_DIR}"/.bench-clones.* 2>/dev/null || true +} +trap 'teardown; rm -f "${COMPOSE_FILE}"' EXIT +trap 'exit 130' INT +trap 'exit 143' TERM +trap 'exit 129' HUP + wait_loaded() { - local c logs + local c logs all for _ in $(seq 1 400); do - local all=1 + all=1 for c in "$@"; do logs="$(docker logs "${c}" 2>&1)" || { all=0; break; } case "${logs}" in *"init message: Done loading"*) ;; *) all=0; break ;; esac done - [ "${all}" -eq 1 ] && return 0 + [ "${all}" -eq 0 ] || return 0 echo -n "."; sleep 3 done return 1 } -teardown() { - [ -n "${COMPOSE_FILE}" ] && compose down --remove-orphans >/dev/null 2>&1 || true - [ -f "${STATE}" ] && { rm -rf "$(cat "${STATE}")" 2>/dev/null || true; rm -f "${STATE}"; } - rm -rf "${SCRIPT_DIR}"/.bench-clones.* 2>/dev/null || true -} - -arm_teardown() { - # On exit: tear down, THEN delete the generated compose (down needs it to still exist). - trap 'teardown; [ -n "${COMPOSE_FILE}" ] && rm -f "${COMPOSE_FILE}"' EXIT - trap 'exit 130' INT - trap 'exit 143' TERM - trap 'exit 129' HUP -} - -bring_up() { +if [ "${MODE}" = local ]; then + bash "${SCRIPT_DIR}/snapshot-chain.sh" # builds ./chain-data to BENCH_HEIGHT via docker echo "==> ensuring a clean network" teardown docker image inspect "${IMAGE}" >/dev/null 2>&1 || { echo "==> building peer image"; compose build; } - - # The client is in the same compose file but is not a peer: it must not be - # counted, started here, or waited on for "Done loading". - local services; services="$(compose config --services | grep -v '^client$' || true)" - local n; n="$(echo ${services} | wc -w | tr -d ' ')" - # BENCH_PEERS is derived from the generated peers (host ports 19401..). - local peers_csv="" - for svc in ${services}; do peers_csv="${peers_csv},127.0.0.1:$((19400 + ${svc#dashd}))"; done - export BENCH_PEERS="${peers_csv#,}" - - echo "==> CoW-cloning ${CHAIN_DIR} for ${n} peers (instant)" - local clone_dir; clone_dir="$(mktemp -d "${SCRIPT_DIR}/.bench-clones.XXXXXX")" - echo "${clone_dir}" > "${STATE}" + services="$(compose config --services | grep -v '^client$' || true)" + npeers="$(echo ${services} | wc -w | tr -d ' ')" + peers_summary="$(scn '[.peers[] | (.count | tostring) + "×" + ([to_entries[] | select(.key != "count") | .key + "=" + (.value | tostring)] | join(" "))] | join(", ")')" + echo "==> scenario '$(basename "${SCN_FILE}" .yml)': ${npeers} peers [${peers_summary}], cpus=${BENCH_CPUS:-}, blocks=${BENCH_HEIGHT}" + echo "==> CoW-cloning ${CHAIN_DIR} for ${npeers} peers" + clone_dir="$(mktemp -d "${SCRIPT_DIR}/.bench-clones.XXXXXX")" + echo "${clone_dir}" >"${STATE}" for svc in ${services}; do - local dst="${clone_dir}/peer${svc#dashd}" + dst="${clone_dir}/peer${svc#dashd}" cp -c -R "${CHAIN_DIR}" "${dst}" 2>/dev/null \ || cp --reflink=auto -R "${CHAIN_DIR}" "${dst}" 2>/dev/null \ || cp -R "${CHAIN_DIR}" "${dst}" done - - local batch=4 - echo "==> starting ${n} peers in batches of ${batch}" - local started="" count=0 + echo "==> starting ${npeers} peers in batches of 4" + started="" + count=0 for svc in ${services}; do CLONE_DIR="${clone_dir}" compose up -d "${svc}" >/dev/null 2>&1 - started="${started} spv-bench-${svc}"; count=$((count + 1)) - if [ $((count % batch)) -eq 0 ]; then - echo -n " loaded ${count}/${n} " + started="${started} spv-bench-${svc}" + count=$((count + 1)) + if [ $((count % 4)) -eq 0 ] || [ "${count}" -eq "${npeers}" ]; then + echo -n " loaded ${count}/${npeers} " wait_loaded ${started} || { echo " timeout loading batch" >&2; exit 1; } echo " ok" fi done - if [ $((count % batch)) -ne 0 ]; then # wait on the trailing partial batch too - echo -n " loaded ${count}/${n} " - wait_loaded ${started} || { echo " timeout loading batch" >&2; exit 1; } - echo " ok" - fi - - echo "==> ${n} peers started" - - # A containerised client shares the compose network with the peers, so it must - # reach them at their container addresses; the published host ports only exist - # for a client running on the host. Read the addresses back from the running - # containers rather than pinning a subnet in the compose file: a pinned subnet - # is one more thing that can collide with whatever else the machine has up, - # VPNs included, and it buys nothing a lookup does not. - if [ -n "${CLIENT_NETEM}" ]; then - local ip csv="" - for svc in ${services}; do - ip="$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "spv-bench-${svc}")" - [ -n "${ip}" ] || { echo "Error: could not resolve address of spv-bench-${svc}" >&2; exit 1; } - csv="${csv},${ip}:19400" - done - export BENCH_PEERS="${csv#,}" - echo "==> client will reach peers at ${BENCH_PEERS}" - fi -} - -build_bin() { - if [ -n "${CROSS_TRIPLE}" ]; then - # The client runs in a Linux container, so the binary has to be a Linux - # one. On a Linux host the ordinary build already is — same triple, and - # glibc is forward compatible — so only a non-Linux host comes here. - # Bind-mounts the workspace and builds into its `target/`, which cargo - # keeps under the triple, so this stays incremental — an image layer build - # would replay the whole workspace on every source change, which makes A/B - # runs unusable. - echo "==> cross-building bench binary (${RUST_IMAGE}, ${CROSS_TRIPLE})" - # As the invoking user, with CARGO_HOME inside the workspace: writing - # into the shared `target/` as root leaves artifacts the host build then - # cannot overwrite. - docker run --rm \ - -v "${REPO_ROOT}:/src" \ - -w /src \ - --user "$(id -u):$(id -g)" \ - -e CARGO_HOME=/src/target/.cross-cargo-home \ - -e CARGO_TARGET_DIR=/src/target \ - -e CARGO_NET_GIT_FETCH_WITH_CLI=true \ - -e CARGO_PROFILE_RELEASE_DEBUG=line-tables-only \ - "${RUST_IMAGE}" \ - cargo build --release --target "${CROSS_TRIPLE}" -p dash-spv-bench - else - echo "==> building bench binary (release + line-table symbols)" - ( cd "${REPO_ROOT}" && CARGO_PROFILE_RELEASE_DEBUG=line-tables-only \ - cargo build --release -p dash-spv-bench ) - fi - [ -x "${BIN}" ] || { echo "build produced no binary at ${BIN}" >&2; exit 1; } -} - -FLAME_TOOL="" -if [ "${FLAME}" -eq 1 ]; then # fail fast on missing profiler deps, before building - if command -v perf >/dev/null; then FLAME_TOOL=perf # Linux - elif command -v /usr/bin/sample >/dev/null; then FLAME_TOOL=sample # macOS - else echo "flame mode needs 'perf' (Linux) or '/usr/bin/sample' (macOS)" >&2; exit 1; fi - [ -f "${FLAMEGRAPH_DIR}/flamegraph.pl" ] || \ - git clone --depth 1 https://github.com/brendangregg/FlameGraph "${FLAMEGRAPH_DIR}" -fi - -build_bin - -if [ "${MODE}" = local ]; then - arm_teardown - bash "${SCRIPT_DIR}/snapshot-chain.sh" # builds ./chain-data to BENCH_HEIGHT via docker - bring_up + # Container addresses, read back rather than pinned to a subnet that could + # collide with whatever else the machine has up. + csv="" + for svc in ${services}; do + ip="$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "spv-bench-${svc}")" + [ -n "${ip}" ] || { echo "Error: could not resolve address of spv-bench-${svc}" >&2; exit 1; } + csv="${csv},${ip}:19400" + done + export BENCH_PEERS="${csv#,}" + echo "==> ${npeers} peers started, reachable at ${BENCH_PEERS}" else echo "==> ${MODE} mode, peers: ${BENCH_PEERS:-}" fi -# Run the sync with its live output straight on the terminal instead of through -# the batch wrapper's pipe -run_live() { - # Not `[ -w /dev/tty ]`: the device node is writable even with no - # controlling terminal, and the redirect then fails with ENXIO. Only - # actually opening it answers the question. - if { : >/dev/tty; } 2>/dev/null; then - "$@" >/dev/tty 2>&1 - else - "$@" - fi -} - -if [ -n "${CLIENT_NETEM}" ]; then - # `run` rather than `up`: it streams the client's own stdout and gives back - # its exit code, which is what the report is read from. Profiling is not - # wired through the container, so `--flame` is refused rather than silently - # producing a graph of the host doing nothing. - [ "${FLAME}" -eq 0 ] || { echo "Error: --flame is not supported with a containerised client" >&2; exit 1; } - arm_teardown - echo "==> running sync in the client container" - run_live compose run --rm --build client -elif [ "${FLAME}" -eq 0 ]; then - echo "==> running sync" - run_live "${CPU_PREFIX[@]+"${CPU_PREFIX[@]}"}" "${BIN}" -elif [ "${FLAME_TOOL}" = perf ]; then - echo "==> running sync under perf" - mkdir -p "$(dirname "${FLAME_SVG}")" - perf record -F 499 -g -o /tmp/bench-perf.data -- "${CPU_PREFIX[@]+"${CPU_PREFIX[@]}"}" "${BIN}" - perf script -i /tmp/bench-perf.data \ - | "${FLAMEGRAPH_DIR}/stackcollapse-perf.pl" \ - | "${FLAMEGRAPH_DIR}/flamegraph.pl" --title "dash-spv sync" --colors hot > "${FLAME_SVG}" - echo "==> wrote ${FLAME_SVG}" +echo "==> running sync in the client container" +if { : >/dev/tty; } 2>/dev/null; then + compose run --rm --build client >/dev/tty 2>&1 else - echo "==> running sync under the sampler" - "${CPU_PREFIX[@]+"${CPU_PREFIX[@]}"}" "${BIN}" & bpid=$! - sleep 3 - /usr/bin/sample "${bpid}" 2000 1 -file /tmp/bench.sample.txt -mayDie >/dev/null 2>&1 || true - wait "${bpid}" 2>/dev/null || true - mkdir -p "$(dirname "${FLAME_SVG}")" - "${FLAMEGRAPH_DIR}/stackcollapse-sample.awk" /tmp/bench.sample.txt \ - | sed -E 's/^Thread_[^;]*;//' \ - | "${FLAMEGRAPH_DIR}/flamegraph.pl" --title "dash-spv sync" --colors hot > "${FLAME_SVG}" - echo "==> wrote ${FLAME_SVG}" + compose run --rm --build client fi -# Keep this run's outputs, which `bench-storage/` does not: it is wiped at the -# start of every run, so without this the trace of anything but the most recent -# scenario is gone — and in a batch that meant 15 of 16 syncs left nothing to -# look at afterwards. -# -# The batch passes its own directory in, so a child archives straight into it -# and there is one copy, named after the scenario. A lone run makes its own. -ARCHIVE_DIR="${BENCH_ARCHIVE_DIR:-${SCRIPT_DIR}/results/$(date +%Y%m%d-%H%M%S)-$(basename "${SCN_FILE}" .yml)}" -# Prefixed with the scenario only inside a batch, where one directory holds -# every scenario. A lone run's directory is already named after it. -ARCHIVE_NAME="${BENCH_ARCHIVE_NAME:-}" +ARCHIVE_DIR="${BENCH_ARCHIVE_DIR:?BENCH_ARCHIVE_DIR is set by the batch wrapper}" mkdir -p "${ARCHIVE_DIR}" -for out in run.log summary.txt; do +for out in run.log summary.txt flamegraph.svg heap-peak.svg; do [ -s "${BENCH_STORAGE_DIR}/${out}" ] || continue - cp "${BENCH_STORAGE_DIR}/${out}" "${ARCHIVE_DIR}/${ARCHIVE_NAME:+${ARCHIVE_NAME}.}${out}" + cp "${BENCH_STORAGE_DIR}/${out}" "${ARCHIVE_DIR}/${BENCH_ARCHIVE_NAME}.${out}" done echo "==> logs archived to ${ARCHIVE_DIR}" diff --git a/dash-spv-bench/src/main.rs b/dash-spv-bench/src/main.rs index 03b34da7c..35544319b 100644 --- a/dash-spv-bench/src/main.rs +++ b/dash-spv-bench/src/main.rs @@ -1,5 +1,7 @@ mod dashboard; mod metrics; +#[cfg(target_os = "linux")] +mod profile; use std::net::SocketAddr; use std::path::PathBuf; @@ -40,11 +42,11 @@ fn load_mnemonics() -> Vec { .unwrap_or_default() } -fn peak_rss_kb() -> Option { +fn proc_status_kb(field: &str) -> Option { std::fs::read_to_string("/proc/self/status") .ok()? .lines() - .find_map(|line| line.strip_prefix("VmHWM:"))? + .find_map(|line| line.strip_prefix(field))? .split_whitespace() .next()? .parse() @@ -89,6 +91,9 @@ async fn main() -> Result<()> { ) .init(); + #[cfg(target_os = "linux")] + let profilers = profile::start(); + let mode = env_or("BENCH_MODE", "local").trim().to_ascii_lowercase(); let (network, remote) = match mode.as_str() { "local" => (Network::Testnet, false), @@ -197,9 +202,13 @@ async fn main() -> Result<()> { run_handle.abort(); let _ = run_handle.await; + let peak_rss_kb = proc_status_kb("VmHWM:"); + #[cfg(target_os = "linux")] + profilers.finish(&output_dir); + use std::fmt::Write as _; let mut report = format!("{m}\n"); - if let Some(kb) = peak_rss_kb() { + if let Some(kb) = peak_rss_kb { let _ = writeln!(report, "peak_rss_mib: {}", kb / 1024); } diff --git a/dash-spv-bench/src/profile.rs b/dash-spv-bench/src/profile.rs new file mode 100644 index 000000000..05735115a --- /dev/null +++ b/dash-spv-bench/src/profile.rs @@ -0,0 +1,91 @@ +use std::path::Path; + +#[cfg(feature = "heap-profile")] +#[global_allocator] +static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + +#[cfg(feature = "heap-profile")] +#[export_name = "_rjem_malloc_conf"] +pub static MALLOC_CONF: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:19\0"; + +#[cfg(feature = "heap-profile")] +type HeapPeak = std::sync::Arc>>; + +pub(crate) struct Profilers { + #[cfg(feature = "cpu-profile")] + cpu: Option>, + #[cfg(feature = "heap-profile")] + heap_peak: HeapPeak, +} + +pub(crate) fn start() -> Profilers { + let profilers = Profilers { + #[cfg(feature = "cpu-profile")] + cpu: pprof::ProfilerGuardBuilder::default() + .frequency(99) + .blocklist(&["libc", "libgcc", "pthread", "vdso"]) + .build() + .inspect_err(|e| tracing::warn!("cpu profiler not started: {e}")) + .ok(), + #[cfg(feature = "heap-profile")] + heap_peak: HeapPeak::default(), + }; + #[cfg(feature = "heap-profile")] + { + let peak = profilers.heap_peak.clone(); + std::thread::spawn(move || { + let Some(ctl) = jemalloc_pprof::PROF_CTL.as_ref() else { + tracing::warn!("jemalloc heap profiling is not enabled"); + return; + }; + let mut dumped_kb = 0; + while let Some(rss_kb) = crate::proc_status_kb("VmRSS:") { + if rss_kb > dumped_kb + dumped_kb / 20 { + match ctl.blocking_lock().dump_profile() { + Ok(profile) => { + dumped_kb = rss_kb; + if let Ok(mut slot) = peak.lock() { + *slot = Some((rss_kb, profile)); + } + } + Err(e) => { + tracing::warn!("heap snapshot failed: {e}"); + return; + } + } + } + std::thread::sleep(std::time::Duration::from_millis(500)); + } + }); + } + profilers +} + +impl Profilers { + #[allow(unused_variables)] + pub(crate) fn finish(self, dir: &Path) { + #[cfg(feature = "heap-profile")] + if let Some((rss_kb, profile)) = self.heap_peak.lock().ok().and_then(|mut slot| slot.take()) + { + let mut opts = jemalloc_pprof::FlamegraphOptions::default(); + opts.title = format!("dash-spv live heap at the RSS peak ({} MiB)", rss_kb / 1024); + opts.count_name = "bytes".to_string(); + let written = + profile.to_flamegraph(&mut opts).map_err(|e| e.to_string()).and_then(|svg| { + std::fs::write(dir.join("heap-peak.svg"), svg).map_err(|e| e.to_string()) + }); + if let Err(e) = written { + tracing::warn!("could not write heap-peak.svg: {e}"); + } + } + #[cfg(feature = "cpu-profile")] + if let Some(report) = self.cpu.as_ref().and_then(|guard| guard.report().build().ok()) { + let written = std::fs::File::create(dir.join("flamegraph.svg")) + .map_err(|e| e.to_string()) + .and_then(|file| report.flamegraph(file).map_err(|e| e.to_string())); + if let Err(e) = written { + tracing::warn!("could not write flamegraph.svg: {e}"); + } + } + } +} diff --git a/dash-spv/src/storage/blocks.rs b/dash-spv/src/storage/blocks.rs index 34b90a76f..27324d5c9 100644 --- a/dash-spv/src/storage/blocks.rs +++ b/dash-spv/src/storage/blocks.rs @@ -33,17 +33,6 @@ pub trait BlockStorage: Send + Sync + 'static { /// A crash between `truncate_above` and `persist` may leave orphaned segment /// files on disk and cause the storage to reopen at the pre-truncation tip. async fn truncate_above(&mut self, target_height: CoreBlockHeight) -> StorageResult<()>; - - /// Declare the highest block height that has been applied to every - /// interested wallet, allowing the storage to stop holding those block - /// bodies in memory. - /// - /// Blocks at or below this height stay readable: `load_block` reloads them - /// from disk on demand. The watermark may move backwards when a rescan - /// re-processes lower heights. - /// - /// Defaults to a no-op for implementations that hold no in-memory cache. - async fn set_committed_height(&mut self, _height: CoreBlockHeight) {} } /// Persistent storage for full blocks using segmented files. @@ -92,10 +81,6 @@ impl BlockStorage for PersistentBlockStorage { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.blocks.write().await.truncate_above(target_height).await } - - async fn set_committed_height(&mut self, height: u32) { - self.blocks.write().await.set_committed_height(height); - } } #[cfg(test)] @@ -158,49 +143,6 @@ mod tests { assert_eq!(storage.load_block(3).await.unwrap(), None); } - /// Block bodies are the dominant memory consumer during a long backfill. - /// Once applied, they must leave memory while staying loadable from disk — - /// `handle_sync_event` re-reads stored blocks by height on resume/rescan. - #[tokio::test] - async fn test_committed_blocks_are_released_but_still_loadable() { - let temp_dir = TempDir::new().unwrap(); - let mut storage = PersistentBlockStorage::open(temp_dir.path()).await.unwrap(); - - // One block in each of segments 0, 1 and 2 (50_000 heights per segment). - // Carry real transaction payloads so the reload proves the block bodies - // round-trip, not merely that a slot is occupied. - let txs = vec![dashcore::Transaction::dummy_empty()]; - let low = HashedBlock::dummy(10, txs.clone()); - let mid = HashedBlock::dummy(50_010, txs.clone()); - let tip = HashedBlock::dummy(100_010, txs); - - storage.store_block(10, low.clone()).await.unwrap(); - storage.store_block(50_010, mid.clone()).await.unwrap(); - storage.store_block(100_010, tip.clone()).await.unwrap(); - storage.persist(temp_dir.path()).await.unwrap(); - - assert_eq!(storage.blocks.read().await.resident_segment_ids(), vec![0, 1, 2]); - - // Everything below segment 2 has been applied to the wallets. - storage.set_committed_height(99_999).await; - storage.persist(temp_dir.path()).await.unwrap(); - - // Segments 0 and 1 are gone from memory; the frontier segment stays. - assert_eq!( - storage.blocks.read().await.resident_segment_ids(), - vec![2], - "applied block segments must be released" - ); - - // All three blocks still load, byte-identically, via the disk fallback. - assert_eq!(storage.load_block(10).await.unwrap(), Some(low)); - assert_eq!(storage.load_block(50_010).await.unwrap(), Some(mid)); - assert_eq!(storage.load_block(100_010).await.unwrap(), Some(tip)); - - // Gaps inside a released segment still report absent, not sentinel data. - assert_eq!(storage.load_block(11).await.unwrap(), None); - } - #[tokio::test] async fn test_returns_none_for_gaps() { let temp_dir = TempDir::new().unwrap(); diff --git a/dash-spv/src/storage/filters.rs b/dash-spv/src/storage/filters.rs index 249b966b9..3578a26ea 100644 --- a/dash-spv/src/storage/filters.rs +++ b/dash-spv/src/storage/filters.rs @@ -51,17 +51,6 @@ pub trait FilterStorage: Send + Sync + 'static { /// A crash between `truncate_above` and `persist` may leave orphaned segment /// files on disk and cause the storage to reopen at the pre-truncation tip. async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()>; - - /// Declare the highest filter height that has been scanned and committed - /// for every wallet, allowing the storage to stop holding those filters in - /// memory. - /// - /// Filters at or below this height stay readable: `load_filters` reloads - /// them from disk on demand. The watermark may move backwards when a - /// rescan rolls the scan position back. - /// - /// Defaults to a no-op for implementations that hold no in-memory cache. - async fn set_committed_height(&mut self, _height: u32) {} } pub struct PersistentFilterStorage { @@ -121,10 +110,6 @@ impl FilterStorage for PersistentFilterStorage { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.filters.write().await.truncate_above(target_height).await } - - async fn set_committed_height(&mut self, height: u32) { - self.filters.write().await.set_committed_height(height); - } } #[cfg(test)] diff --git a/dash-spv/src/storage/io.rs b/dash-spv/src/storage/io.rs index 9854d5ab0..c965fbd30 100644 --- a/dash-spv/src/storage/io.rs +++ b/dash-spv/src/storage/io.rs @@ -3,7 +3,8 @@ use std::path::{Path, PathBuf}; use crate::error::{StorageError, StorageResult}; -use tokio::io::AsyncWriteExt; +use dashcore::consensus::Encodable; +use tokio::io::{AsyncWriteExt, BufWriter}; /// Get the temporary file path for atomic writes. /// Uses process ID and a counter to ensure uniqueness even with concurrent writes. @@ -22,6 +23,38 @@ fn get_temp_path(path: &Path) -> PathBuf { /// Atomically write data to a file. /// Uses temporary file + sync + rename pattern for crash resilience. pub(crate) async fn atomic_write(path: &Path, data: &[u8]) -> StorageResult<()> { + atomic_write_with(path, |mut file| async move { + file.write_all(data).await?; + file.sync_all().await + }) + .await +} + +/// Like [`atomic_write`], but encodes `items` into the file one at a time +/// instead of into a buffer holding all of them. +pub(crate) async fn atomic_write_items( + path: &Path, + items: &[I], +) -> StorageResult<()> { + atomic_write_with(path, |file| async move { + let mut writer = BufWriter::with_capacity(1 << 20, file); + let mut encoded = Vec::new(); + for item in items { + encoded.clear(); + item.consensus_encode(&mut encoded)?; + writer.write_all(&encoded).await?; + } + writer.flush().await?; + writer.into_inner().sync_all().await + }) + .await +} + +async fn atomic_write_with(path: &Path, write: F) -> StorageResult<()> +where + F: FnOnce(tokio::fs::File) -> Fut, + Fut: std::future::Future>, +{ // Ensure parent directory exists if let Some(parent) = path.parent() { tokio::fs::create_dir_all(parent) @@ -32,14 +65,7 @@ pub(crate) async fn atomic_write(path: &Path, data: &[u8]) -> StorageResult<()> let temp_path = get_temp_path(path); // Write to temporary file - let write_result = async { - let mut file = tokio::fs::File::create(&temp_path).await?; - file.write_all(data).await?; - file.sync_all().await?; - - Ok::<(), std::io::Error>(()) - } - .await; + let write_result = async { write(tokio::fs::File::create(&temp_path).await?).await }.await; // Clean up temp file on error if let Err(e) = write_result { @@ -62,6 +88,18 @@ mod tests { use std::fs; use tempfile::TempDir; + #[tokio::test] + async fn test_atomic_write_items_matches_encoding_all_at_once() { + let temp_dir = TempDir::new().unwrap(); + let path = temp_dir.path().join("items.dat"); + let items = vec![vec![1u8, 2, 3], vec![], vec![4u8; 300]]; + + atomic_write_items(&path, &items).await.unwrap(); + + let expected: Vec = items.iter().flat_map(dashcore::consensus::serialize).collect(); + assert_eq!(fs::read(&path).unwrap(), expected); + } + #[test] fn test_get_temp_path_uniqueness() { let path = Path::new("some").join("path").join("file.dat"); diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs index 70a851acb..248ab10d0 100644 --- a/dash-spv/src/storage/mod.rs +++ b/dash-spv/src/storage/mod.rs @@ -386,10 +386,6 @@ impl filters::FilterStorage for DiskStorageManager { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.filters.write().await.truncate_above(target_height).await } - - async fn set_committed_height(&mut self, height: u32) { - self.filters.write().await.set_committed_height(height).await; - } } #[async_trait] @@ -405,10 +401,6 @@ impl BlockStorage for DiskStorageManager { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.blocks.write().await.truncate_above(target_height).await } - - async fn set_committed_height(&mut self, height: u32) { - self.blocks.write().await.set_committed_height(height).await; - } } #[async_trait] diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index ec900899c..4788bb0c7 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -19,14 +19,15 @@ use dashcore_hashes::Hash; use crate::{ error::StorageResult, - storage::io::atomic_write, + storage::io::atomic_write_items, types::{HashedBlock, HashedBlockHeader}, StorageError, }; -pub trait Persistable: Sized + Encodable + Decodable + PartialEq + Clone { +pub(super) trait Persistable: Sized + Encodable + Decodable + PartialEq + Clone { const SEGMENT_PREFIX: &'static str = "segment"; const DATA_FILE_EXTENSION: &'static str = "dat"; + const ITEMS_PER_SEGMENT: u32; fn segment_file_name(segment_id: u32) -> String { format!("{}_{:04}.{}", Self::SEGMENT_PREFIX, segment_id, Self::DATA_FILE_EXTENSION) @@ -36,12 +37,16 @@ pub trait Persistable: Sized + Encodable + Decodable + PartialEq + Clone { } impl Persistable for Vec { + const ITEMS_PER_SEGMENT: u32 = 2_000; + fn sentinel() -> Self { vec![] } } impl Persistable for HashedBlockHeader { + const ITEMS_PER_SEGMENT: u32 = 10_000; + fn sentinel() -> Self { let header = BlockHeader { version: Version::from_consensus(i32::MAX), // Invalid version @@ -57,12 +62,16 @@ impl Persistable for HashedBlockHeader { } impl Persistable for FilterHeader { + const ITEMS_PER_SEGMENT: u32 = 50_000; + fn sentinel() -> Self { FilterHeader::from_byte_array([0u8; 32]) } } impl Persistable for HashedBlock { + const ITEMS_PER_SEGMENT: u32 = 1_000; + fn sentinel() -> Self { let block = Block { header: *HashedBlockHeader::sentinel().header(), @@ -76,37 +85,26 @@ impl Persistable for HashedBlock { #[derive(Debug)] pub struct SegmentCache { segments: HashMap>, - evicted: HashMap>, tip_height: Option, start_height: Option, segments_dir: PathBuf, /// Segment ids whose backing files must be removed on the next `persist`. /// Populated by `truncate_above` for segments that are dropped entirely. to_delete: HashSet, - /// Highest height whose items are durably persisted *and* fully consumed - /// by every reader, as declared by the owning sync manager through - /// [`SegmentCache::set_committed_height`]. - /// - /// `None` — the default — disables committed-height release entirely, so a - /// cache whose owner never declares a watermark keeps exactly the residency - /// behavior it had before this field existed. - committed_height: Option, } impl SegmentCache { - const MAX_ACTIVE_SEGMENTS: usize = 10; + const MAX_ACTIVE_SEGMENTS: usize = 2; pub async fn load_or_new(segments_dir: impl Into) -> StorageResult { let segments_dir = segments_dir.into(); let mut cache = Self { segments: HashMap::with_capacity(Self::MAX_ACTIVE_SEGMENTS), - evicted: HashMap::new(), tip_height: None, start_height: None, segments_dir: segments_dir.clone(), to_delete: HashSet::new(), - committed_height: None, }; // Building the metadata @@ -186,6 +184,10 @@ impl SegmentCache { let segments_len = self.segments.len(); if self.segments.contains_key(segment_id) { + tracing::trace!( + "SegmentCache<{}>: segment {segment_id} cache hit", + std::any::type_name::() + ); let segment = self.segments.get_mut(segment_id).expect("We already checked that it exists"); return Ok(segment); @@ -193,28 +195,34 @@ impl SegmentCache { if segments_len >= Self::MAX_ACTIVE_SEGMENTS { let key_to_evict = - self.segments.iter_mut().min_by_key(|(_, s)| s.last_accessed).map(|(k, v)| (*k, v)); + self.segments.iter().min_by_key(|(_, s)| s.last_accessed).map(|(k, _)| *k); - if let Some((key, _)) = key_to_evict { - if let Some(segment) = self.segments.remove(&key) { - if segment.state == SegmentState::Dirty { - self.evicted.insert(key, segment); - } + if let Some(key) = key_to_evict { + if let Some(segment) = self.segments.get_mut(&key) { + segment.persist(&self.segments_dir).await?; } + self.segments.remove(&key); } } - // If the segment is already in the to_persist map, load it from there. // If the segment is queued for deletion, return a fresh empty segment. // The next `persist` will atomically overwrite the stale file. // Otherwise, load it from disk. - let segment = if let Some(segment) = self.evicted.remove(segment_id) { - segment - } else if self.to_delete.remove(segment_id) { - Segment::new(*segment_id, vec![], SegmentState::Dirty) + let (segment, source) = if self.to_delete.remove(segment_id) { + (Segment::new(*segment_id, vec![], SegmentState::Dirty), "new") } else { - Segment::load(&self.segments_dir, *segment_id).await? + let segment = Segment::load(&self.segments_dir, *segment_id).await?; + let source = if segment.state == SegmentState::Clean { + "disk" + } else { + "new" + }; + (segment, source) }; + tracing::trace!( + "SegmentCache<{}>: segment {segment_id} cache miss ({source})", + std::any::type_name::() + ); let segment = self.segments.entry(*segment_id).or_insert(segment); Ok(segment) @@ -448,21 +456,11 @@ impl SegmentCache { for segment_id in (boundary_segment_id + 1)..=max_segment_id { self.segments.remove(&segment_id); - self.evicted.remove(&segment_id); self.to_delete.insert(segment_id); } self.tip_height = Some(target_height); - // The watermark must never claim heights that no longer exist: a - // rescan re-reads this range, and a stale high watermark would keep - // releasing segments it is actively refilling. - if let Some(committed) = self.committed_height { - if committed > target_height { - self.committed_height = Some(target_height); - } - } - Ok(()) } @@ -499,20 +497,16 @@ impl SegmentCache { Err(e) => return Err(StorageError::Io(e)), } - // Scan succeeded — now it is safe to mutate cache state. Resident and - // evicted segments may be dirty and not persisted yet, so the scan - // above cannot see them. Queue their ids too; `persist` ignores - // missing files. + // Scan succeeded — now it is safe to mutate cache state. Resident + // segments may be dirty and not persisted yet, so the scan above + // cannot see them. Queue their ids too; `persist` ignores missing + // files. to_delete.extend(self.segments.keys().copied()); - to_delete.extend(self.evicted.keys().copied()); self.to_delete.extend(to_delete); self.segments.clear(); - self.evicted.clear(); self.tip_height = None; self.start_height = None; - // Nothing is stored, so nothing is committed. - self.committed_height = None; Ok(()) } @@ -534,136 +528,11 @@ impl SegmentCache { } self.to_delete.extend(failed); - for (id, segments) in self.evicted.iter_mut() { - if let Err(e) = segments.persist(&segments_dir).await { - tracing::error!("Failed to persist segment with id {id}: {e}"); - } - } - - self.evicted.clear(); - for (id, segments) in self.segments.iter_mut() { if let Err(e) = segments.persist(&segments_dir).await { tracing::error!("Failed to persist segment with id {id}: {e}"); } } - - // After the writes above, so segments cleaned by *this* pass are - // eligible immediately. A segment whose persist failed is still Dirty - // and is therefore skipped. - let released = self.release_committed_segments(); - if released > 0 { - tracing::debug!( - "SegmentCache: released {} committed segment(s) below height {:?}; {} resident", - released, - self.committed_height, - self.segments.len(), - ); - } - } - - /// Declare the highest height whose items are durably persisted and fully - /// consumed, so the cache may stop holding them in memory. - /// - /// Segments lying *entirely* at or below `height` are released from memory - /// by the next [`SegmentCache::persist`], once they are `Clean`. The - /// on-disk file remains the source of truth and a later read of a released - /// height transparently reloads the segment through the same lazy path used - /// for any never-resident segment — the data a reader observes is - /// unchanged, only the resident-set size is. - /// - /// The watermark may move **backwards**: a wallet rescan rolls it back - /// before re-reading lower heights. A regression only ever narrows the - /// release window, so it is always safe. Callers therefore assign rather - /// than accumulate, and no monotonicity is enforced here. - pub fn set_committed_height(&mut self, height: u32) { - self.committed_height = Some(height); - } - - /// The committed-height watermark, or `None` when the owner has never - /// declared one (in which case no committed-height release occurs). - /// - /// Test-only for now — nothing reads the watermark back in production. - /// Promote it to an unconditional accessor when a caller needs it. - #[cfg(test)] - #[inline] - pub fn committed_height(&self) -> Option { - self.committed_height - } - - /// Sorted ids of the segments currently held in memory. Test-only hook for - /// asserting residency from the storage wrappers. - #[cfg(test)] - pub(crate) fn resident_segment_ids(&self) -> Vec { - let mut ids: Vec = self.segments.keys().copied().collect(); - ids.sort_unstable(); - ids - } - - /// Release the in-memory items of every fully-committed clean segment, - /// keeping the on-disk file as the source of truth. - /// - /// This is what stops a long backfill from pinning every segment it ever - /// touched. `MAX_ACTIVE_SEGMENTS` alone cannot: a scan spanning fewer than - /// ten segments never trips the LRU, so every downloaded item stays - /// resident until the process dies. - /// - /// Releasing a `Clean` segment is invisible to readers. `Clean` is reachable - /// only via [`Segment::load`] from an existing file or a successful - /// [`Segment::persist`], so a clean segment's items are byte-identical to - /// its backing file, and `get_segment_mut` reloads exactly those bytes on - /// the next access. - /// - /// Three classes are deliberately kept: - /// - `Dirty` segments — their contents exist *only* in memory, so dropping - /// them would lose data outright. - /// - The segment holding the sync frontier (`tip_height`) — it is still - /// being written, and releasing it would force a reload on the very next - /// store. - /// - Any segment not entirely at or below the watermark, so a partially - /// committed segment is never released out from under an active reader. - /// - /// Segments queued in `to_delete` cannot appear here: both `truncate_above` - /// and `clear` remove a segment from `self.segments` in the same step that - /// queues it, so the two sets are disjoint by construction. - /// - /// Returns the number of segments released. - fn release_committed_segments(&mut self) -> usize { - let Some(committed) = self.committed_height else { - return 0; - }; - - // Still-written frontier; never release it. - let frontier_segment = self.tip_height.map(Self::height_to_segment_id); - - let items_per_segment = Segment::::ITEMS_PER_SEGMENT as u64; - let committed = committed as u64; - - // Widened to u64 so a segment id near u32::MAX cannot overflow the - // range arithmetic; such a segment simply fails the comparison. - let releasable: Vec = self - .segments - .iter() - .filter(|(id, segment)| { - if segment.state != SegmentState::Clean { - return false; - } - - if Some(**id) == frontier_segment { - return false; - } - - let last_height = (**id as u64) * items_per_segment + items_per_segment - 1; - last_height <= committed - }) - .map(|(id, _)| *id) - .collect(); - - for id in &releasable { - self.segments.remove(id); - } - - releasable.len() } #[inline] @@ -702,7 +571,7 @@ pub struct Segment { } impl Segment { - const ITEMS_PER_SEGMENT: u32 = 50_000; + const ITEMS_PER_SEGMENT: u32 = I::ITEMS_PER_SEGMENT; fn new(segment_id: u32, mut items: Vec, state: SegmentState) -> Self { debug_assert!(items.len() <= Self::ITEMS_PER_SEGMENT as usize); @@ -788,15 +657,7 @@ impl Segment { return Err(StorageError::WriteFailed(format!("Failed to persist segment: {}", e))); } - let mut buffer = Vec::new(); - - for item in self.items.iter() { - item.consensus_encode(&mut buffer).map_err(|e| { - StorageError::WriteFailed(format!("Failed to encode segment item: {}", e)) - })?; - } - - atomic_write(&path, &buffer).await?; + atomic_write_items(&path, &self.items).await?; self.state = SegmentState::Clean; Ok(()) @@ -904,11 +765,14 @@ mod tests { segment.insert(FilterHeader::dummy(i), 0); } + assert!(tmp_dir.path().join(FilterHeader::segment_file_name(0)).exists()); + for i in 0..=MAX_SEGMENTS { assert_eq!(cache.segments.len(), MAX_SEGMENTS as usize); let segment = cache.get_segment_mut(&i).await.expect("Failed to load segment"); + assert!(segment.state == SegmentState::Clean); assert_eq!(segment.get(0..1), [FilterHeader::dummy(i)]); } } @@ -1444,206 +1308,6 @@ mod tests { ); } - /// The headline guarantee: a committed, clean segment is dropped from - /// memory on persist, and a later read transparently reloads it from disk - /// byte-identically. - /// - /// This is the leak that pinned a 350k-block backfill in RAM: the scan - /// spans fewer than `MAX_ACTIVE_SEGMENTS`, so the LRU never fires and every - /// item stays resident until the process dies. - #[tokio::test] - async fn test_release_committed_segments_reloads_from_disk() { - let tmp_dir = TempDir::new().unwrap(); - - const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; - - // Dense across segments 0 and 1, plus a few items into segment 2 so the - // frontier lives above the range we expect to be released. - let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT * 2 + 5); - - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - cache.store_items_at_height(&items, 0).await.unwrap(); - cache.persist(tmp_dir.path()).await; - - // Baseline: without a watermark nothing is released, which is exactly - // the pre-fix behavior. - assert_eq!(cache.committed_height(), None); - assert_eq!(cache.segments.len(), 3, "no watermark declared, so nothing is released"); - - // Segments 0 and 1 are now fully committed (their last heights are - // ITEMS_PER_SEGMENT-1 and 2*ITEMS_PER_SEGMENT-1). - cache.set_committed_height(ITEMS_PER_SEGMENT * 2 - 1); - cache.persist(tmp_dir.path()).await; - - assert!(!cache.segments.contains_key(&0), "committed clean segment 0 must be released"); - assert!(!cache.segments.contains_key(&1), "committed clean segment 1 must be released"); - assert!(cache.segments.contains_key(&2), "frontier segment must stay resident"); - assert_eq!(cache.segments.len(), 1); - - // The cache-level watermarks are unaffected by residency. - assert_eq!(cache.start_height(), Some(0)); - assert_eq!(cache.tip_height(), Some(ITEMS_PER_SEGMENT * 2 + 4)); - - // Reading the released range reloads both segments from disk and - // returns exactly the bytes that were written. - let reread = cache.get_items(0..ITEMS_PER_SEGMENT * 2).await.unwrap(); - assert_eq!(reread, items[0..(ITEMS_PER_SEGMENT * 2) as usize]); - - // Single-item reads across the boundary agree too. - assert_eq!(cache.get_item(0).await.unwrap(), Some(items[0])); - assert_eq!( - cache.get_item(ITEMS_PER_SEGMENT).await.unwrap(), - Some(items[ITEMS_PER_SEGMENT as usize]) - ); - } - - /// Never release a segment that is dirty (its items exist only in memory), - /// the frontier segment (still being written), or a segment only partially - /// below the watermark. - #[tokio::test] - async fn test_release_skips_dirty_frontier_and_partial_segments() { - let tmp_dir = TempDir::new().unwrap(); - - const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; - - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - - // Sparse writes into segments 0, 1 and 2. - cache.store_items_at_height(&FilterHeader::dummy_batch(0..1), 10).await.unwrap(); - cache - .store_items_at_height(&FilterHeader::dummy_batch(1..2), ITEMS_PER_SEGMENT + 10) - .await - .unwrap(); - cache - .store_items_at_height(&FilterHeader::dummy_batch(2..3), ITEMS_PER_SEGMENT * 2 + 10) - .await - .unwrap(); - - // A watermark above everything, but nothing has been persisted yet, so - // all three segments are Dirty and none may be released. - cache.set_committed_height(u32::MAX); - let released = cache.release_committed_segments(); - assert_eq!(released, 0, "dirty segments must never be released"); - assert_eq!(cache.segments.len(), 3); - - cache.persist(tmp_dir.path()).await; - // persist() cleaned all three, then released every non-frontier one. - assert_eq!(cache.segments.keys().copied().collect::>(), vec![2]); - - // Now verify the partial-coverage rule: a watermark inside segment 0 - // does not release it, because its upper heights are not yet committed. - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - let _ = cache.get_segment_mut(&0).await.unwrap(); - assert!(cache.segments.contains_key(&0)); - - cache.set_committed_height(ITEMS_PER_SEGMENT - 2); // one short of the segment's last height - assert_eq!(cache.release_committed_segments(), 0, "partially committed segment must stay"); - assert!(cache.segments.contains_key(&0)); - - cache.set_committed_height(ITEMS_PER_SEGMENT - 1); // exactly the last height - assert_eq!(cache.release_committed_segments(), 1, "fully committed segment is releasable"); - assert!(!cache.segments.contains_key(&0)); - } - - /// The rescan path re-reads heights far below the watermark. Those segments - /// were released, so this exercises the lazy reload under the access - /// pattern `reset_for_rescan` / `start_download` produce. - #[tokio::test] - async fn test_released_segments_serve_a_rescan_reread() { - let tmp_dir = TempDir::new().unwrap(); - - const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; - - let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT * 2 + 5); - - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - cache.store_items_at_height(&items, 0).await.unwrap(); - cache.set_committed_height(ITEMS_PER_SEGMENT * 2 - 1); - cache.persist(tmp_dir.path()).await; - assert_eq!(cache.segments.len(), 1); - - // A wallet appears behind the scan: the manager rolls the watermark - // back and re-reads from a low height. - cache.set_committed_height(0); - let rescanned = cache.get_items(5..ITEMS_PER_SEGMENT + 5).await.unwrap(); - assert_eq!(rescanned, items[5..(ITEMS_PER_SEGMENT + 5) as usize]); - - // With the watermark rolled back, the re-read segments are retained - // rather than being dropped again underneath the rescan. - cache.persist(tmp_dir.path()).await; - assert!(cache.segments.contains_key(&0), "rolled-back watermark must stop re-release"); - - // And the data still round-trips after a full reopen. - let mut reloaded = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - assert_eq!(reloaded.tip_height(), Some(ITEMS_PER_SEGMENT * 2 + 4)); - assert_eq!( - reloaded.get_items(0..ITEMS_PER_SEGMENT * 2 + 5).await.unwrap(), - items, - "released-then-reloaded data must survive a process restart byte-identically" - ); - } - - /// The watermark must never outlive the data it refers to: `truncate_above` - /// clamps it and `clear` drops it, so a later scan of the same range is not - /// released out from under itself. - #[tokio::test] - async fn test_committed_height_follows_truncate_and_clear() { - let tmp_dir = TempDir::new().unwrap(); - - let items = FilterHeader::dummy_batch(0..30); - - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - cache.store_items_at_height(&items, 0).await.unwrap(); - - cache.set_committed_height(25); - assert_eq!(cache.committed_height(), Some(25)); - - // Truncating above a height below the watermark clamps it. - cache.truncate_above(10).await.unwrap(); - assert_eq!(cache.committed_height(), Some(10)); - - // Truncating above the watermark leaves it alone. - cache.set_committed_height(5); - cache.truncate_above(8).await.unwrap(); - assert_eq!(cache.committed_height(), Some(5)); - - cache.clear().unwrap(); - assert_eq!(cache.committed_height(), None, "an empty cache has nothing committed"); - } - - /// A released segment must still accept new writes into its unused slots, - /// reloading the existing contents first rather than silently starting from - /// a blank segment (which would drop the persisted items on the next write). - #[tokio::test] - async fn test_store_into_released_segment_preserves_existing_items() { - let tmp_dir = TempDir::new().unwrap(); - - const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; - - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - cache.store_items_at_height(&FilterHeader::dummy_batch(0..10), 0).await.unwrap(); - cache - .store_items_at_height(&FilterHeader::dummy_batch(50..55), ITEMS_PER_SEGMENT) - .await - .unwrap(); - - cache.set_committed_height(ITEMS_PER_SEGMENT - 1); - cache.persist(tmp_dir.path()).await; - assert!(!cache.segments.contains_key(&0), "segment 0 released"); - - // Write into a gap in the released segment. - cache.store_items_at_height(&FilterHeader::dummy_batch(90..95), 100).await.unwrap(); - - // Both the reloaded originals and the new items are present. - assert_eq!(cache.get_items(0..10).await.unwrap(), FilterHeader::dummy_batch(0..10)); - assert_eq!(cache.get_items(100..105).await.unwrap(), FilterHeader::dummy_batch(90..95)); - - cache.persist(tmp_dir.path()).await; - let mut reloaded = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - assert_eq!(reloaded.get_items(0..10).await.unwrap(), FilterHeader::dummy_batch(0..10)); - assert_eq!(reloaded.get_items(100..105).await.unwrap(), FilterHeader::dummy_batch(90..95)); - } - #[test] fn test_segment_insert_get() { let segment_id = 10; diff --git a/dash-spv/src/sync/blocks/manager.rs b/dash-spv/src/sync/blocks/manager.rs index e4a1464d8..feb428da6 100644 --- a/dash-spv/src/sync/blocks/manager.rs +++ b/dash-spv/src/sync/blocks/manager.rs @@ -77,9 +77,6 @@ impl BlocksManager SyncResult> { let mut events = Vec::new(); - // Highest height applied in this drain, used below to advance the - // storage's committed watermark exactly once. - let mut last_applied: Option = None; // Process blocks in height order using pipeline's ordering logic while let Some((block, height, interested)) = self.pipeline.take_next_ordered_block() { @@ -127,7 +124,6 @@ impl BlocksManager BlocksManager= batch_end { @@ -759,13 +754,6 @@ impl