From 948d581e9f458b40f07b8e469106c401c2fb5401 Mon Sep 17 00:00:00 2001 From: iho Date: Thu, 9 Jul 2026 10:56:42 +0300 Subject: [PATCH 1/5] fix: throttle chain compaction with a 1h wall-clock gap During fast sync, blocks arrive much faster than mainnet's 1/min, so the 1/DAY_HEIGHT probabilistic compaction check fires far too often and slows sync. Gate compaction triggers with MIN_COMPACTION_INTERVAL_SECS (1 hour) in addition to the existing height-based dice roll. Addresses #3594. --- core/src/global.rs | 8 ++++ servers/src/common/adapters.rs | 85 ++++++++++++++++++++++++++++++++++ 2 files changed, 93 insertions(+) diff --git a/core/src/global.rs b/core/src/global.rs index fc8b01cd28..53e2f092e4 100644 --- a/core/src/global.rs +++ b/core/src/global.rs @@ -102,6 +102,14 @@ pub const PEER_EXPIRATION_REMOVE_TIME: i64 = PEER_EXPIRATION_DAYS * 24 * 3600; /// For a node configured as "archival_mode = true" only the txhashset will be compacted. pub const COMPACTION_CHECK: u64 = DAY_HEIGHT; +/// Minimum wall-clock interval between compaction runs (seconds). +/// +/// `COMPACTION_CHECK` is height-based and assumes ~1 block/minute. During fast +/// sync many blocks arrive per second, so the probabilistic check alone would +/// compact far too often and slow sync down. Enforcing a wall-clock gap (1 hour) +/// limits that without changing post-sync average behavior (#3594). +pub const MIN_COMPACTION_INTERVAL_SECS: u64 = 60 * 60; + /// Number of blocks to reuse a txhashset zip for (automated testing and user testing). pub const TESTING_TXHASHSET_ARCHIVE_INTERVAL: u64 = 10; diff --git a/servers/src/common/adapters.rs b/servers/src/common/adapters.rs index 5bdcb5301b..14ab1cd044 100644 --- a/servers/src/common/adapters.rs +++ b/servers/src/common/adapters.rs @@ -61,6 +61,22 @@ const WORKER_CHANNEL_BUFFER_SIZE: usize = 64; const HEADER_SEGMENT_REQUEST_WINDOW_SECS: i64 = 60; const MAX_HEADER_SEGMENT_REQUESTS_PER_WINDOW: usize = 120; +/// Whether enough wall-clock time has passed since the last compaction trigger. +/// `None` means never compacted yet (always allowed). +fn compaction_wall_clock_ok( + last: Option, + now: Instant, + min_interval: std::time::Duration, +) -> bool { + match last { + None => true, + Some(t) => now + .checked_duration_since(t) + .map(|d| d >= min_interval) + .unwrap_or(true), + } +} + /// Implementation of the NetAdapter for the . Gets notified when new /// blocks and transactions are received and forwards to the chain and pool /// implementations. @@ -76,6 +92,9 @@ where config: ServerConfig, hooks: Vec>, header_segment_requests: RwLock, usize)>>, + /// Wall-clock time of last successful compaction *trigger* (not completion). + /// Used with `MIN_COMPACTION_INTERVAL_SECS` to avoid compact storms during sync. + last_compact_trigger: RwLock>, tx: mpsc::SyncSender, } @@ -708,6 +727,7 @@ where config, hooks, header_segment_requests: RwLock::new(HashMap::new()), + last_compact_trigger: RwLock::new(None), tx, }; adapter.spawn_net_adapter_worker(Arc::downgrade(&chain), rx); @@ -966,14 +986,30 @@ where } fn check_compact(&self) { + // Wall-clock throttle first (cheap). During fast sync blocks arrive much + // faster than mainnet's 1/min, so the height-based dice alone is too aggressive. + let min_interval = std::time::Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); + { + let last = self.last_compact_trigger.read(); + if !compaction_wall_clock_ok(*last, Instant::now(), min_interval) { + return; + } + } + // Roll the dice to trigger compaction at 1/COMPACTION_CHECK chance per block, // uses a different thread to avoid blocking the caller thread (likely a peer) let mut rng = thread_rng(); if 0 == rng.gen_range(0, global::COMPACTION_CHECK) { + // Record trigger time before spawn so concurrent process_block calls + // cannot start multiple compactors in the same window. + *self.last_compact_trigger.write() = Some(Instant::now()); + let chain = self.chain(); + let syncing = self.sync_state.is_syncing(); let _ = thread::Builder::new() .name("compactor".to_string()) .spawn(move || { + debug!("check_compact: starting compaction (syncing={})", syncing); if let Err(e) = chain.compact() { error!("Could not compact chain: {:?}", e); } @@ -1301,3 +1337,52 @@ impl pool::BlockChain for PoolToChainAdapter { .map_err(|_| pool::PoolError::ImmatureTransaction) } } + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + #[test] + fn compaction_wall_clock_allows_first_run() { + let now = Instant::now(); + assert!(compaction_wall_clock_ok( + None, + now, + Duration::from_secs(3600) + )); + } + + #[test] + fn compaction_wall_clock_blocks_inside_interval() { + let start = Instant::now(); + // Simulate "last" slightly in the past by sleeping a tiny amount is flaky; + // use checked path: last == now means zero elapsed < 1h. + assert!(!compaction_wall_clock_ok( + Some(start), + start, + Duration::from_secs(3600) + )); + } + + #[test] + fn compaction_wall_clock_allows_after_interval() { + let start = Instant::now(); + // Instant cannot be advanced artificially; use a zero min interval. + assert!(compaction_wall_clock_ok( + Some(start), + start + Duration::from_secs(1), + Duration::from_secs(0) + )); + assert!(compaction_wall_clock_ok( + Some(start), + start + Duration::from_secs(3600), + Duration::from_secs(3600) + )); + } + + #[test] + fn min_compaction_interval_is_one_hour() { + assert_eq!(global::MIN_COMPACTION_INTERVAL_SECS, 60 * 60); + } +} From 2b3ba7970e471cb6b1ed20d7337443ea60434d5d Mon Sep 17 00:00:00 2001 From: iho Date: Thu, 9 Jul 2026 11:07:34 +0300 Subject: [PATCH 2/5] test: prove compact at most once per hour under rapid sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Extract should_trigger_compaction and unit-test 50k same-instant process_block-equivalent checks (dice always hits) → exactly one compact per wall-clock window. - Log compact starts at info for ops verification. - Add scripts/compaction_sync_observe.sh for live mainnet observe runs. --- scripts/compaction_sync_observe.sh | 199 +++++++++++++++++++++++++++++ servers/src/common/adapters.rs | 100 +++++++++++---- 2 files changed, 276 insertions(+), 23 deletions(-) create mode 100755 scripts/compaction_sync_observe.sh diff --git a/scripts/compaction_sync_observe.sh b/scripts/compaction_sync_observe.sh new file mode 100755 index 0000000000..7870c1ea8d --- /dev/null +++ b/scripts/compaction_sync_observe.sh @@ -0,0 +1,199 @@ +#!/usr/bin/env bash +# Observe compaction cadence while a node syncs (issue #3594 / PR #3892). +# +# Runs a mainnet (or testnet) node for a fixed duration and counts how many +# times compaction is *started*. With MIN_COMPACTION_INTERVAL_SECS=3600, a +# run shorter than 1 hour must log at most ONE "check_compact: starting +# compaction" line after the first trigger (typically 0 or 1 total). +# +# Usage (from repo root): +# ./scripts/compaction_sync_observe.sh +# DURATION_SECS=300 ./scripts/compaction_sync_observe.sh +# NETWORK=testnet DURATION_SECS=180 ./scripts/compaction_sync_observe.sh +# +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +cd "$ROOT" + +GRIN_BIN="${GRIN_BIN:-$ROOT/target/debug/grin}" +DURATION_SECS="${DURATION_SECS:-240}" +NETWORK="${NETWORK:-mainnet}" # mainnet | testnet +WORKDIR="$(mktemp -d "${TMPDIR:-/tmp}/grin-compact-observe.XXXXXX")" +LOG="$WORKDIR/grin.stdout.log" +FILE_LOG="$WORKDIR/grin-server.log" +RESULTS_DIR="${RESULTS_DIR:-/tmp/grin-compact-observe-result}" +PID="" + +cleanup() { + if [[ -n "${PID}" ]] && kill -0 "$PID" 2>/dev/null; then + echo "==> Stopping grin (pid $PID)" + kill -TERM "$PID" 2>/dev/null || true + for _ in $(seq 1 40); do + kill -0 "$PID" 2>/dev/null || break + sleep 0.25 + done + kill -KILL "$PID" 2>/dev/null || true + wait "$PID" 2>/dev/null || true + fi + # Preserve logs for inspection + mkdir -p "$RESULTS_DIR" + cp -f "$LOG" "$RESULTS_DIR/grin.stdout.log" 2>/dev/null || true + cp -f "$FILE_LOG" "$RESULTS_DIR/grin-server.log" 2>/dev/null || true + # Prefer file logger (Debug) when present + if [[ -f "$FILE_LOG" ]]; then + ANALYSIS_LOG="$FILE_LOG" + else + ANALYSIS_LOG="$LOG" + fi + export ANALYSIS_LOG +} +trap cleanup EXIT + +echo "==> Workdir: $WORKDIR" +echo "==> Network: $NETWORK duration: ${DURATION_SECS}s" + +if [[ ! -x "$GRIN_BIN" ]]; then + echo "==> Building grin" + cargo build -p grin +fi + +mkdir -p "$WORKDIR" +NET_ARGS=() +case "$NETWORK" in + mainnet) ;; + testnet) NET_ARGS+=(--testnet) ;; + usernet) NET_ARGS+=(--usernet) ;; + *) echo "Unknown NETWORK=$NETWORK" >&2; exit 1 ;; +esac + +( + cd "$WORKDIR" + if ((${#NET_ARGS[@]})); then + "$GRIN_BIN" "${NET_ARGS[@]}" server config >/dev/null + else + "$GRIN_BIN" server config >/dev/null + fi +) + +CFG="$WORKDIR/grin-server.toml" +python3 - "$CFG" <<'PY' +import re, sys +path = sys.argv[1] +text = open(path).read() + +def set_key(text, key, value): + pat = re.compile(rf'(?m)^(\s*{re.escape(key)}\s*=\s*).*$') + if pat.search(text): + return pat.sub(rf'\g<1>{value}', text) + return text + +text = set_key(text, 'run_tui', 'false') +text = set_key(text, 'skip_sync_wait', 'true') +# Ensure file logging is on and detailed enough to catch compact lines. +text = set_key(text, 'log_to_file', 'true') +text = set_key(text, 'log_to_stdout', 'true') +text = set_key(text, 'stdout_log_level', '"Info"') +text = set_key(text, 'file_log_level', '"Debug"') +open(path, 'w').write(text) +print("config patched") +PY + +echo "==> Starting grin --no-tui server run (${NETWORK})" +( + cd "$WORKDIR" + if ((${#NET_ARGS[@]})); then + exec "$GRIN_BIN" "${NET_ARGS[@]}" --no-tui server run + else + exec "$GRIN_BIN" --no-tui server run + fi +) >"$LOG" 2>&1 & +PID=$! +echo " pid=$PID log=$LOG" + +echo "==> Observing for ${DURATION_SECS}s (expect ≤1 compact start if duration < 1h)..." +END=$((SECONDS + DURATION_SECS)) +while (( SECONDS < END )); do + if ! kill -0 "$PID" 2>/dev/null; then + echo "ERROR: grin exited early" >&2 + tail -40 "$LOG" >&2 || true + exit 1 + fi + sleep 5 + # Progress hint + if grep -q "synchronized\|HeaderSync\|BodySync\|TxHashset\|PIBD\|sync" "$LOG" 2>/dev/null; then + : + fi +done + +# Prefer file logger (Debug) over stdout capture. +if [[ -f "$FILE_LOG" ]]; then + ANALYSIS_LOG="$FILE_LOG" +else + ANALYSIS_LOG="$LOG" +fi + +# Count compaction starts from our instrumented log line and txhashset compact. +STARTS=$(grep -c "check_compact: starting compaction" "$ANALYSIS_LOG" 2>/dev/null || true) +STARTS=${STARTS:-0} +TXHS=$(grep -c "txhashset: starting compaction" "$ANALYSIS_LOG" 2>/dev/null || true) +TXHS=${TXHS:-0} +SKIP=$(grep -c "compact: skipping" "$ANALYSIS_LOG" 2>/dev/null || true) +SKIP=${SKIP:-0} +HEADERS=$(grep -c "Received .* block headers" "$ANALYSIS_LOG" 2>/dev/null || true) +HEADERS=${HEADERS:-0} +BLOCKS=$(grep -cE "Received block |going to process" "$ANALYSIS_LOG" 2>/dev/null || true) +BLOCKS=${BLOCKS:-0} + +echo "" +echo "==> Results" +echo " duration_secs=$DURATION_SECS" +echo " check_compact starts: $STARTS" +echo " txhashset compact starts: $TXHS" +echo " compact skips (height gate): $SKIP" +echo " header batches received: $HEADERS" +echo " full-block process lines: $BLOCKS" +echo " analysis_log: $ANALYSIS_LOG" +echo " results_dir: $RESULTS_DIR" + +# Also show timestamps of any compact starts +if [[ "$STARTS" -gt 0 ]]; then + echo " compact start lines:" + grep "check_compact: starting compaction" "$ANALYSIS_LOG" | head -20 +fi + +if [[ "$BLOCKS" -eq 0 ]]; then + echo " note: node was likely still in header sync (process_block/check_compact not yet active)." + echo " Wall-clock gate is proven by unit test rapid_sync_compacts_at_most_once_per_wall_clock_window" + echo " (50k dice-always-hit blocks → exactly 1 trigger per hour window)." +fi + +FAIL=0 +# Production min interval is 1 hour. For runs shorter than 1 hour, at most 1 start. +if (( DURATION_SECS < 3600 )); then + if (( STARTS > 1 )); then + echo "FAIL: saw $STARTS compact starts in ${DURATION_SECS}s (< 1h window allows at most 1)" >&2 + FAIL=1 + else + echo "OK: compact starts ($STARTS) ≤ 1 for run shorter than 1 hour" + fi +else + # Allow roughly one per hour (+1 slack for boundary) + MAX_ALLOWED=$(( DURATION_SECS / 3600 + 1 )) + if (( STARTS > MAX_ALLOWED )); then + echo "FAIL: $STARTS starts > allowed ~$MAX_ALLOWED for ${DURATION_SECS}s" >&2 + FAIL=1 + else + echo "OK: compact starts ($STARTS) within ~once/hour budget ($MAX_ALLOWED)" + fi +fi + +if [[ "$FAIL" -ne 0 ]]; then + tail -80 "$LOG" >&2 || true + exit 1 +fi + +echo "" +echo "PASS: compaction cadence within wall-clock bound during observe window" +echo " (Unit test rapid_sync_compacts_at_most_once_per_wall_clock_window covers the" +echo " 50k-block same-instant worst case; this script observes a real syncing node.)" diff --git a/servers/src/common/adapters.rs b/servers/src/common/adapters.rs index 14ab1cd044..0a316d160a 100644 --- a/servers/src/common/adapters.rs +++ b/servers/src/common/adapters.rs @@ -77,6 +77,17 @@ fn compaction_wall_clock_ok( } } +/// Combined wall-clock + probabilistic gate used by `check_compact`. +/// Returns true if a compact thread should be started (caller updates `last`). +fn should_trigger_compaction( + last: Option, + now: Instant, + min_interval: std::time::Duration, + dice_hit: bool, +) -> bool { + compaction_wall_clock_ok(last, now, min_interval) && dice_hit +} + /// Implementation of the NetAdapter for the . Gets notified when new /// blocks and transactions are received and forwards to the chain and pool /// implementations. @@ -986,35 +997,36 @@ where } fn check_compact(&self) { - // Wall-clock throttle first (cheap). During fast sync blocks arrive much - // faster than mainnet's 1/min, so the height-based dice alone is too aggressive. + // Wall-clock throttle + height-based dice. During fast sync blocks arrive much + // faster than mainnet's 1/min, so the dice alone is too aggressive (#3594). let min_interval = std::time::Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); + let now = Instant::now(); + let dice_hit = { + let mut rng = thread_rng(); + 0 == rng.gen_range(0, global::COMPACTION_CHECK) + }; + + // Hold the write lock across check+stamp so concurrent process_block + // calls cannot double-trigger in the same window. { - let last = self.last_compact_trigger.read(); - if !compaction_wall_clock_ok(*last, Instant::now(), min_interval) { + let mut last = self.last_compact_trigger.write(); + if !should_trigger_compaction(*last, now, min_interval, dice_hit) { return; } + *last = Some(now); } - // Roll the dice to trigger compaction at 1/COMPACTION_CHECK chance per block, - // uses a different thread to avoid blocking the caller thread (likely a peer) - let mut rng = thread_rng(); - if 0 == rng.gen_range(0, global::COMPACTION_CHECK) { - // Record trigger time before spawn so concurrent process_block calls - // cannot start multiple compactors in the same window. - *self.last_compact_trigger.write() = Some(Instant::now()); - - let chain = self.chain(); - let syncing = self.sync_state.is_syncing(); - let _ = thread::Builder::new() - .name("compactor".to_string()) - .spawn(move || { - debug!("check_compact: starting compaction (syncing={})", syncing); - if let Err(e) = chain.compact() { - error!("Could not compact chain: {:?}", e); - } - }); - } + let chain = self.chain(); + let syncing = self.sync_state.is_syncing(); + let _ = thread::Builder::new() + .name("compactor".to_string()) + .spawn(move || { + // info: visible at default log level for ops verification of #3594 + info!("check_compact: starting compaction (syncing={})", syncing); + if let Err(e) = chain.compact() { + error!("Could not compact chain: {:?}", e); + } + }); } fn request_transaction(&self, h: Hash, peer_info: &PeerInfo) { @@ -1385,4 +1397,46 @@ mod tests { fn min_compaction_interval_is_one_hour() { assert_eq!(global::MIN_COMPACTION_INTERVAL_SECS, 60 * 60); } + + /// Simulate fast sync: many blocks, dice always hits, wall clock fixed. + /// Compaction must trigger at most once per min_interval window. + #[test] + fn rapid_sync_compacts_at_most_once_per_wall_clock_window() { + let min = Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); + let t0 = Instant::now(); + let mut last: Option = None; + let mut triggers = 0u32; + + // 50k "blocks" in the same wall-clock instant (worst-case sync). + for _ in 0..50_000 { + if should_trigger_compaction(last, t0, min, true) { + triggers += 1; + last = Some(t0); + } + } + assert_eq!( + triggers, 1, + "expected exactly one compact trigger in a single wall-clock window" + ); + + // Still inside the window → no more triggers. + let t_mid = t0 + Duration::from_secs(min.as_secs() / 2); + for _ in 0..10_000 { + if should_trigger_compaction(last, t_mid, min, true) { + triggers += 1; + last = Some(t_mid); + } + } + assert_eq!(triggers, 1, "half-interval must not allow another compact"); + + // After full interval → one more trigger allowed. + let t_next = t0 + min; + assert!(should_trigger_compaction(last, t_next, min, true)); + last = Some(t_next); + triggers += 1; + assert_eq!(triggers, 2); + + // Dice miss even after interval → no trigger. + assert!(!should_trigger_compaction(last, t_next + min, min, false)); + } } From 4799a59e83ba9780d42bcb4ae16dc2d4e314b3ef Mon Sep 17 00:00:00 2001 From: iho Date: Sat, 11 Jul 2026 18:35:29 +0300 Subject: [PATCH 3/5] Address review: fix race in compaction gate, handle spawn failure - Read `now` under the lock (not before) so a slower thread's stale timestamp can no longer be judged against a newer trigger and move last_compact_trigger backwards. - Only stamp last_compact_trigger after the compactor thread actually spawns, so a spawn failure doesn't suppress compaction for an hour. - Switch last_compact_trigger from RwLock to Mutex since it's never read-locked. - Add a multi-threaded test exercising the real lock/check/stamp gate, not just the pure helper. - Drop the ad-hoc compaction_sync_observe.sh script in favor of the deterministic unit tests. --- scripts/compaction_sync_observe.sh | 199 ----------------------------- servers/src/common/adapters.rs | 75 ++++++++--- 2 files changed, 60 insertions(+), 214 deletions(-) delete mode 100755 scripts/compaction_sync_observe.sh diff --git a/scripts/compaction_sync_observe.sh b/scripts/compaction_sync_observe.sh deleted file mode 100755 index 7870c1ea8d..0000000000 --- a/scripts/compaction_sync_observe.sh +++ /dev/null @@ -1,199 +0,0 @@ -#!/usr/bin/env bash -# Observe compaction cadence while a node syncs (issue #3594 / PR #3892). -# -# Runs a mainnet (or testnet) node for a fixed duration and counts how many -# times compaction is *started*. With MIN_COMPACTION_INTERVAL_SECS=3600, a -# run shorter than 1 hour must log at most ONE "check_compact: starting -# compaction" line after the first trigger (typically 0 or 1 total). -# -# Usage (from repo root): -# ./scripts/compaction_sync_observe.sh -# DURATION_SECS=300 ./scripts/compaction_sync_observe.sh -# NETWORK=testnet DURATION_SECS=180 ./scripts/compaction_sync_observe.sh -# -set -euo pipefail - -ROOT="$(cd "$(dirname "$0")/.." && pwd)" -cd "$ROOT" - -GRIN_BIN="${GRIN_BIN:-$ROOT/target/debug/grin}" -DURATION_SECS="${DURATION_SECS:-240}" -NETWORK="${NETWORK:-mainnet}" # mainnet | testnet -WORKDIR="$(mktemp -d "${TMPDIR:-/tmp}/grin-compact-observe.XXXXXX")" -LOG="$WORKDIR/grin.stdout.log" -FILE_LOG="$WORKDIR/grin-server.log" -RESULTS_DIR="${RESULTS_DIR:-/tmp/grin-compact-observe-result}" -PID="" - -cleanup() { - if [[ -n "${PID}" ]] && kill -0 "$PID" 2>/dev/null; then - echo "==> Stopping grin (pid $PID)" - kill -TERM "$PID" 2>/dev/null || true - for _ in $(seq 1 40); do - kill -0 "$PID" 2>/dev/null || break - sleep 0.25 - done - kill -KILL "$PID" 2>/dev/null || true - wait "$PID" 2>/dev/null || true - fi - # Preserve logs for inspection - mkdir -p "$RESULTS_DIR" - cp -f "$LOG" "$RESULTS_DIR/grin.stdout.log" 2>/dev/null || true - cp -f "$FILE_LOG" "$RESULTS_DIR/grin-server.log" 2>/dev/null || true - # Prefer file logger (Debug) when present - if [[ -f "$FILE_LOG" ]]; then - ANALYSIS_LOG="$FILE_LOG" - else - ANALYSIS_LOG="$LOG" - fi - export ANALYSIS_LOG -} -trap cleanup EXIT - -echo "==> Workdir: $WORKDIR" -echo "==> Network: $NETWORK duration: ${DURATION_SECS}s" - -if [[ ! -x "$GRIN_BIN" ]]; then - echo "==> Building grin" - cargo build -p grin -fi - -mkdir -p "$WORKDIR" -NET_ARGS=() -case "$NETWORK" in - mainnet) ;; - testnet) NET_ARGS+=(--testnet) ;; - usernet) NET_ARGS+=(--usernet) ;; - *) echo "Unknown NETWORK=$NETWORK" >&2; exit 1 ;; -esac - -( - cd "$WORKDIR" - if ((${#NET_ARGS[@]})); then - "$GRIN_BIN" "${NET_ARGS[@]}" server config >/dev/null - else - "$GRIN_BIN" server config >/dev/null - fi -) - -CFG="$WORKDIR/grin-server.toml" -python3 - "$CFG" <<'PY' -import re, sys -path = sys.argv[1] -text = open(path).read() - -def set_key(text, key, value): - pat = re.compile(rf'(?m)^(\s*{re.escape(key)}\s*=\s*).*$') - if pat.search(text): - return pat.sub(rf'\g<1>{value}', text) - return text - -text = set_key(text, 'run_tui', 'false') -text = set_key(text, 'skip_sync_wait', 'true') -# Ensure file logging is on and detailed enough to catch compact lines. -text = set_key(text, 'log_to_file', 'true') -text = set_key(text, 'log_to_stdout', 'true') -text = set_key(text, 'stdout_log_level', '"Info"') -text = set_key(text, 'file_log_level', '"Debug"') -open(path, 'w').write(text) -print("config patched") -PY - -echo "==> Starting grin --no-tui server run (${NETWORK})" -( - cd "$WORKDIR" - if ((${#NET_ARGS[@]})); then - exec "$GRIN_BIN" "${NET_ARGS[@]}" --no-tui server run - else - exec "$GRIN_BIN" --no-tui server run - fi -) >"$LOG" 2>&1 & -PID=$! -echo " pid=$PID log=$LOG" - -echo "==> Observing for ${DURATION_SECS}s (expect ≤1 compact start if duration < 1h)..." -END=$((SECONDS + DURATION_SECS)) -while (( SECONDS < END )); do - if ! kill -0 "$PID" 2>/dev/null; then - echo "ERROR: grin exited early" >&2 - tail -40 "$LOG" >&2 || true - exit 1 - fi - sleep 5 - # Progress hint - if grep -q "synchronized\|HeaderSync\|BodySync\|TxHashset\|PIBD\|sync" "$LOG" 2>/dev/null; then - : - fi -done - -# Prefer file logger (Debug) over stdout capture. -if [[ -f "$FILE_LOG" ]]; then - ANALYSIS_LOG="$FILE_LOG" -else - ANALYSIS_LOG="$LOG" -fi - -# Count compaction starts from our instrumented log line and txhashset compact. -STARTS=$(grep -c "check_compact: starting compaction" "$ANALYSIS_LOG" 2>/dev/null || true) -STARTS=${STARTS:-0} -TXHS=$(grep -c "txhashset: starting compaction" "$ANALYSIS_LOG" 2>/dev/null || true) -TXHS=${TXHS:-0} -SKIP=$(grep -c "compact: skipping" "$ANALYSIS_LOG" 2>/dev/null || true) -SKIP=${SKIP:-0} -HEADERS=$(grep -c "Received .* block headers" "$ANALYSIS_LOG" 2>/dev/null || true) -HEADERS=${HEADERS:-0} -BLOCKS=$(grep -cE "Received block |going to process" "$ANALYSIS_LOG" 2>/dev/null || true) -BLOCKS=${BLOCKS:-0} - -echo "" -echo "==> Results" -echo " duration_secs=$DURATION_SECS" -echo " check_compact starts: $STARTS" -echo " txhashset compact starts: $TXHS" -echo " compact skips (height gate): $SKIP" -echo " header batches received: $HEADERS" -echo " full-block process lines: $BLOCKS" -echo " analysis_log: $ANALYSIS_LOG" -echo " results_dir: $RESULTS_DIR" - -# Also show timestamps of any compact starts -if [[ "$STARTS" -gt 0 ]]; then - echo " compact start lines:" - grep "check_compact: starting compaction" "$ANALYSIS_LOG" | head -20 -fi - -if [[ "$BLOCKS" -eq 0 ]]; then - echo " note: node was likely still in header sync (process_block/check_compact not yet active)." - echo " Wall-clock gate is proven by unit test rapid_sync_compacts_at_most_once_per_wall_clock_window" - echo " (50k dice-always-hit blocks → exactly 1 trigger per hour window)." -fi - -FAIL=0 -# Production min interval is 1 hour. For runs shorter than 1 hour, at most 1 start. -if (( DURATION_SECS < 3600 )); then - if (( STARTS > 1 )); then - echo "FAIL: saw $STARTS compact starts in ${DURATION_SECS}s (< 1h window allows at most 1)" >&2 - FAIL=1 - else - echo "OK: compact starts ($STARTS) ≤ 1 for run shorter than 1 hour" - fi -else - # Allow roughly one per hour (+1 slack for boundary) - MAX_ALLOWED=$(( DURATION_SECS / 3600 + 1 )) - if (( STARTS > MAX_ALLOWED )); then - echo "FAIL: $STARTS starts > allowed ~$MAX_ALLOWED for ${DURATION_SECS}s" >&2 - FAIL=1 - else - echo "OK: compact starts ($STARTS) within ~once/hour budget ($MAX_ALLOWED)" - fi -fi - -if [[ "$FAIL" -ne 0 ]]; then - tail -80 "$LOG" >&2 || true - exit 1 -fi - -echo "" -echo "PASS: compaction cadence within wall-clock bound during observe window" -echo " (Unit test rapid_sync_compacts_at_most_once_per_wall_clock_window covers the" -echo " 50k-block same-instant worst case; this script observes a real syncing node.)" diff --git a/servers/src/common/adapters.rs b/servers/src/common/adapters.rs index 0a316d160a..5a697999f0 100644 --- a/servers/src/common/adapters.rs +++ b/servers/src/common/adapters.rs @@ -15,7 +15,7 @@ //! Adapters connecting new block, new transaction, and accepted transaction //! events to consumers of those events. -use crate::util::RwLock; +use crate::util::{Mutex, RwLock}; use std::collections::HashMap; use std::fs::File; use std::net::SocketAddr; @@ -105,7 +105,7 @@ where header_segment_requests: RwLock, usize)>>, /// Wall-clock time of last successful compaction *trigger* (not completion). /// Used with `MIN_COMPACTION_INTERVAL_SECS` to avoid compact storms during sync. - last_compact_trigger: RwLock>, + last_compact_trigger: Mutex>, tx: mpsc::SyncSender, } @@ -738,7 +738,7 @@ where config, hooks, header_segment_requests: RwLock::new(HashMap::new()), - last_compact_trigger: RwLock::new(None), + last_compact_trigger: Mutex::new(None), tx, }; adapter.spawn_net_adapter_worker(Arc::downgrade(&chain), rx); @@ -1000,33 +1000,34 @@ where // Wall-clock throttle + height-based dice. During fast sync blocks arrive much // faster than mainnet's 1/min, so the dice alone is too aggressive (#3594). let min_interval = std::time::Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); - let now = Instant::now(); let dice_hit = { let mut rng = thread_rng(); 0 == rng.gen_range(0, global::COMPACTION_CHECK) }; - // Hold the write lock across check+stamp so concurrent process_block - // calls cannot double-trigger in the same window. - { - let mut last = self.last_compact_trigger.write(); - if !should_trigger_compaction(*last, now, min_interval, dice_hit) { - return; - } - *last = Some(now); + // Hold the lock across check+stamp+spawn so concurrent process_block calls + // cannot double-trigger in the same window, `now` is read under the lock so + // it can't be superseded by a newer trigger recorded between check and stamp, + // and `last` is only stamped once the compactor thread actually started. + let mut last = self.last_compact_trigger.lock(); + let now = Instant::now(); + if !should_trigger_compaction(*last, now, min_interval, dice_hit) { + return; } let chain = self.chain(); let syncing = self.sync_state.is_syncing(); - let _ = thread::Builder::new() + match thread::Builder::new() .name("compactor".to_string()) .spawn(move || { - // info: visible at default log level for ops verification of #3594 info!("check_compact: starting compaction (syncing={})", syncing); if let Err(e) = chain.compact() { error!("Could not compact chain: {:?}", e); } - }); + }) { + Ok(_) => *last = Some(now), + Err(e) => error!("Could not spawn compactor thread: {:?}", e), + } } fn request_transaction(&self, h: Hash, peer_info: &PeerInfo) { @@ -1439,4 +1440,48 @@ mod tests { // Dice miss even after interval → no trigger. assert!(!should_trigger_compaction(last, t_next + min, min, false)); } + + /// Exercises the shared lock-then-check-then-stamp path from multiple threads, + /// mirroring `check_compact`'s locking so the gate itself (not just the pure + /// helper) is proven to serialize concurrent triggers. + #[test] + fn concurrent_check_compact_gate_triggers_at_most_once() { + use std::sync::atomic::{AtomicU32, Ordering}; + use std::sync::Barrier; + + let min = Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); + let gate: Arc>> = Arc::new(Mutex::new(None)); + let triggers = Arc::new(AtomicU32::new(0)); + let num_threads = 64; + let barrier = Arc::new(Barrier::new(num_threads)); + + let handles: Vec<_> = (0..num_threads) + .map(|_| { + let gate = gate.clone(); + let triggers = triggers.clone(); + let barrier = barrier.clone(); + thread::spawn(move || { + barrier.wait(); + // Same order as check_compact: acquire the lock, then read `now` + // under it, then stamp before releasing. + let mut last = gate.lock(); + let now = Instant::now(); + if should_trigger_compaction(*last, now, min, true) { + triggers.fetch_add(1, Ordering::SeqCst); + *last = Some(now); + } + }) + }) + .collect(); + + for h in handles { + h.join().unwrap(); + } + + assert_eq!( + triggers.load(Ordering::SeqCst), + 1, + "concurrent callers sharing the gate must trigger compaction at most once per window" + ); + } } From b1923c7ccec8571532bbb0757f96f7c6f4ba03e2 Mon Sep 17 00:00:00 2001 From: iho Date: Sun, 12 Jul 2026 11:02:56 +0300 Subject: [PATCH 4/5] Address review: share the compaction gate between check_compact and tests Fold the wall-clock helper and dice wrapper into a single try_trigger_compaction() used by both production code and tests, reduce the test module to one interval test and one concurrency test, and drop the issue references from comments. --- core/src/global.rs | 2 +- servers/src/common/adapters.rs | 217 ++++++++++++++------------------- 2 files changed, 92 insertions(+), 127 deletions(-) diff --git a/core/src/global.rs b/core/src/global.rs index 53e2f092e4..27d46750af 100644 --- a/core/src/global.rs +++ b/core/src/global.rs @@ -107,7 +107,7 @@ pub const COMPACTION_CHECK: u64 = DAY_HEIGHT; /// `COMPACTION_CHECK` is height-based and assumes ~1 block/minute. During fast /// sync many blocks arrive per second, so the probabilistic check alone would /// compact far too often and slow sync down. Enforcing a wall-clock gap (1 hour) -/// limits that without changing post-sync average behavior (#3594). +/// limits that without changing post-sync average behavior. pub const MIN_COMPACTION_INTERVAL_SECS: u64 = 60 * 60; /// Number of blocks to reuse a txhashset zip for (automated testing and user testing). diff --git a/servers/src/common/adapters.rs b/servers/src/common/adapters.rs index 5a697999f0..596395a00f 100644 --- a/servers/src/common/adapters.rs +++ b/servers/src/common/adapters.rs @@ -61,33 +61,43 @@ const WORKER_CHANNEL_BUFFER_SIZE: usize = 64; const HEADER_SEGMENT_REQUEST_WINDOW_SECS: i64 = 60; const MAX_HEADER_SEGMENT_REQUESTS_PER_WINDOW: usize = 120; -/// Whether enough wall-clock time has passed since the last compaction trigger. -/// `None` means never compacted yet (always allowed). -fn compaction_wall_clock_ok( - last: Option, - now: Instant, +/// Wall-clock + probabilistic gate for chain compaction. If the dice hit and +/// at least `min_interval` has passed since the last recorded trigger (`None` +/// means never, always allowed), runs `start` and returns its result; the +/// trigger time is only recorded if `start` reports success. +/// +/// The lock is held across check+start+stamp so concurrent callers cannot +/// double-trigger in the same window, and `now` is read under the lock so an +/// older caller cannot move the timestamp backwards. +fn try_trigger_compaction( + gate: &Mutex>, min_interval: std::time::Duration, -) -> bool { - match last { + dice_hit: bool, + start: F, +) -> bool +where + F: FnOnce() -> bool, +{ + if !dice_hit { + return false; + } + let mut last = gate.lock(); + let now = Instant::now(); + let wall_clock_ok = match *last { None => true, Some(t) => now .checked_duration_since(t) .map(|d| d >= min_interval) .unwrap_or(true), + }; + if wall_clock_ok && start() { + *last = Some(now); + true + } else { + false } } -/// Combined wall-clock + probabilistic gate used by `check_compact`. -/// Returns true if a compact thread should be started (caller updates `last`). -fn should_trigger_compaction( - last: Option, - now: Instant, - min_interval: std::time::Duration, - dice_hit: bool, -) -> bool { - compaction_wall_clock_ok(last, now, min_interval) && dice_hit -} - /// Implementation of the NetAdapter for the . Gets notified when new /// blocks and transactions are received and forwards to the chain and pool /// implementations. @@ -998,36 +1008,31 @@ where fn check_compact(&self) { // Wall-clock throttle + height-based dice. During fast sync blocks arrive much - // faster than mainnet's 1/min, so the dice alone is too aggressive (#3594). + // faster than mainnet's 1/min, so the dice alone is too aggressive. let min_interval = std::time::Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); let dice_hit = { let mut rng = thread_rng(); 0 == rng.gen_range(0, global::COMPACTION_CHECK) }; - // Hold the lock across check+stamp+spawn so concurrent process_block calls - // cannot double-trigger in the same window, `now` is read under the lock so - // it can't be superseded by a newer trigger recorded between check and stamp, - // and `last` is only stamped once the compactor thread actually started. - let mut last = self.last_compact_trigger.lock(); - let now = Instant::now(); - if !should_trigger_compaction(*last, now, min_interval, dice_hit) { - return; - } - - let chain = self.chain(); - let syncing = self.sync_state.is_syncing(); - match thread::Builder::new() - .name("compactor".to_string()) - .spawn(move || { - info!("check_compact: starting compaction (syncing={})", syncing); - if let Err(e) = chain.compact() { - error!("Could not compact chain: {:?}", e); + try_trigger_compaction(&self.last_compact_trigger, min_interval, dice_hit, || { + let chain = self.chain(); + let syncing = self.sync_state.is_syncing(); + match thread::Builder::new() + .name("compactor".to_string()) + .spawn(move || { + info!("check_compact: starting compaction (syncing={})", syncing); + if let Err(e) = chain.compact() { + error!("Could not compact chain: {:?}", e); + } + }) { + Ok(_) => true, + Err(e) => { + error!("Could not spawn compactor thread: {:?}", e); + false } - }) { - Ok(_) => *last = Some(now), - Err(e) => error!("Could not spawn compactor thread: {:?}", e), - } + } + }); } fn request_transaction(&self, h: Hash, peer_info: &PeerInfo) { @@ -1354,98 +1359,62 @@ impl pool::BlockChain for PoolToChainAdapter { #[cfg(test)] mod tests { use super::*; + use std::cell::Cell; use std::time::Duration; + /// Simulate fast sync through the real gate: many "blocks" with the dice + /// always hitting must trigger compaction exactly once per wall-clock window. + /// Also covers dice misses and a failed start not consuming the window. #[test] - fn compaction_wall_clock_allows_first_run() { - let now = Instant::now(); - assert!(compaction_wall_clock_ok( - None, - now, - Duration::from_secs(3600) - )); - } - - #[test] - fn compaction_wall_clock_blocks_inside_interval() { - let start = Instant::now(); - // Simulate "last" slightly in the past by sleeping a tiny amount is flaky; - // use checked path: last == now means zero elapsed < 1h. - assert!(!compaction_wall_clock_ok( - Some(start), - start, - Duration::from_secs(3600) - )); - } - - #[test] - fn compaction_wall_clock_allows_after_interval() { - let start = Instant::now(); - // Instant cannot be advanced artificially; use a zero min interval. - assert!(compaction_wall_clock_ok( - Some(start), - start + Duration::from_secs(1), - Duration::from_secs(0) - )); - assert!(compaction_wall_clock_ok( - Some(start), - start + Duration::from_secs(3600), - Duration::from_secs(3600) - )); - } - - #[test] - fn min_compaction_interval_is_one_hour() { - assert_eq!(global::MIN_COMPACTION_INTERVAL_SECS, 60 * 60); - } - - /// Simulate fast sync: many blocks, dice always hits, wall clock fixed. - /// Compaction must trigger at most once per min_interval window. - #[test] - fn rapid_sync_compacts_at_most_once_per_wall_clock_window() { + fn compaction_gate_triggers_at_most_once_per_wall_clock_window() { let min = Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); - let t0 = Instant::now(); - let mut last: Option = None; - let mut triggers = 0u32; + let gate: Mutex> = Mutex::new(None); + let triggers = Cell::new(0u32); + let start = || { + triggers.set(triggers.get() + 1); + true + }; - // 50k "blocks" in the same wall-clock instant (worst-case sync). + // 50k "blocks" in one wall-clock window (worst-case sync). for _ in 0..50_000 { - if should_trigger_compaction(last, t0, min, true) { - triggers += 1; - last = Some(t0); - } + try_trigger_compaction(&gate, min, true, start); } assert_eq!( - triggers, 1, + triggers.get(), + 1, "expected exactly one compact trigger in a single wall-clock window" ); - // Still inside the window → no more triggers. - let t_mid = t0 + Duration::from_secs(min.as_secs() / 2); - for _ in 0..10_000 { - if should_trigger_compaction(last, t_mid, min, true) { - triggers += 1; - last = Some(t_mid); - } - } - assert_eq!(triggers, 1, "half-interval must not allow another compact"); - - // After full interval → one more trigger allowed. - let t_next = t0 + min; - assert!(should_trigger_compaction(last, t_next, min, true)); - last = Some(t_next); - triggers += 1; - assert_eq!(triggers, 2); - - // Dice miss even after interval → no trigger. - assert!(!should_trigger_compaction(last, t_next + min, min, false)); + // Dice miss never triggers, even on a fresh gate. + assert!(!try_trigger_compaction( + &Mutex::new(None), + min, + false, + start + )); + assert_eq!(triggers.get(), 1); + + // A failed start must not consume the window: the next successful + // start is still allowed. + let gate = Mutex::new(None); + assert!(!try_trigger_compaction(&gate, min, true, || false)); + assert!(try_trigger_compaction(&gate, min, true, start)); + assert_eq!(triggers.get(), 2); + + // Once the interval has elapsed the gate opens again (zero interval + // stands in for elapsed time, since `Instant` cannot be advanced). + assert!(try_trigger_compaction( + &gate, + Duration::from_secs(0), + true, + start + )); + assert_eq!(triggers.get(), 3); } - /// Exercises the shared lock-then-check-then-stamp path from multiple threads, - /// mirroring `check_compact`'s locking so the gate itself (not just the pure - /// helper) is proven to serialize concurrent triggers. + /// Concurrent callers racing on the shared gate must trigger at most once. #[test] - fn concurrent_check_compact_gate_triggers_at_most_once() { + fn concurrent_compaction_gate_triggers_at_most_once() { use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::Barrier; @@ -1462,14 +1431,10 @@ mod tests { let barrier = barrier.clone(); thread::spawn(move || { barrier.wait(); - // Same order as check_compact: acquire the lock, then read `now` - // under it, then stamp before releasing. - let mut last = gate.lock(); - let now = Instant::now(); - if should_trigger_compaction(*last, now, min, true) { + try_trigger_compaction(&gate, min, true, || { triggers.fetch_add(1, Ordering::SeqCst); - *last = Some(now); - } + true + }); }) }) .collect(); From ceca6dffae971292aa64e53666bbd0ae3a96b1bf Mon Sep 17 00:00:00 2001 From: iho Date: Sat, 1 Aug 2026 13:26:44 +0300 Subject: [PATCH 5/5] Address review: enforce compaction interval in Chain::compact Move the wall-clock gate to the shared compact entry point so sync dice, NoSync, and owner API share one policy. Document restart behavior, simplify duration_since, and replace the large loop test with boundary timestamps plus a concurrency test. --- chain/src/chain.rs | 106 +++++++++++++++++++- core/src/global.rs | 9 +- servers/src/common/adapters.rs | 176 +++------------------------------ 3 files changed, 124 insertions(+), 167 deletions(-) diff --git a/chain/src/chain.rs b/chain/src/chain.rs index 0cad711e90..33d3eff9a9 100644 --- a/chain/src/chain.rs +++ b/chain/src/chain.rs @@ -32,7 +32,7 @@ use crate::types::{ BlockStatus, ChainAdapter, CommitPos, NoStatus, Options, Tip, TxHashsetWriteStatus, }; use crate::util::secp::pedersen::{Commitment, RangeProof}; -use crate::util::RwLock; +use crate::util::{Mutex, RwLock}; use crate::{ core::core::hash::{Hash, Hashed}, store::Batch, @@ -159,6 +159,10 @@ pub struct Chain { denylist: Arc>>, archive_mode: bool, genesis: Block, + /// Wall-clock of last compaction start. Process-local: a restart clears it so + /// the wall-clock gate allows compact immediately; the height threshold in + /// `compact()` still limits compacting across frequent restarts. + last_compact_trigger: Mutex>, } impl Chain { @@ -209,6 +213,7 @@ impl Chain { denylist: Arc::new(RwLock::new(vec![])), archive_mode, genesis: genesis, + last_compact_trigger: Mutex::new(None), }; chain.log_heads()?; @@ -1328,6 +1333,9 @@ impl Chain { /// * compacts the txhashset based on current prune_list /// * removes historical blocks and associated data from the db (unless archive mode) /// + /// Shared entry point for all compaction triggers (sync dice, sync→NoSync, + /// owner API). Enforces `MIN_COMPACTION_INTERVAL_SECS` so callers cannot + /// compact more often than that wall-clock gap. pub fn compact(&self) -> Result<(), Error> { // A node may be restarted multiple times in a short period of time. // We compact at most once per 60 blocks in this situation by comparing @@ -1346,6 +1354,14 @@ impl Chain { } } + // Wall-clock throttle (see MIN_COMPACTION_INTERVAL_SECS). Stamp under the + // lock before work so concurrent callers cannot double-trigger. + let min_interval = Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); + if !try_record_compact_trigger(&self.last_compact_trigger, min_interval, Instant::now()) { + debug!("compact: skipping, within min wall-clock interval"); + return Ok(()); + } + // Retrieve archive header here, so as not to attempt a read // lock while removing historical blocks let archive_header = self.txhashset_archive_header()?; @@ -1862,3 +1878,91 @@ fn setup_head( batch.commit()?; Ok(()) } + +/// If the wall-clock gate allows a compact start, records `now` and returns true. +/// Lock is held across check+stamp so concurrent callers cannot double-trigger. +fn try_record_compact_trigger( + last: &Mutex>, + min_interval: Duration, + now: Instant, +) -> bool { + let mut last = last.lock(); + if let Some(t) = *last { + if now.duration_since(t) < min_interval { + return false; + } + } + *last = Some(now); + true +} + +#[cfg(test)] +mod compact_gate_tests { + use super::try_record_compact_trigger; + use crate::util::Mutex; + use std::sync::atomic::{AtomicU32, Ordering}; + use std::sync::{Arc, Barrier}; + use std::thread; + use std::time::{Duration, Instant}; + + #[test] + fn compact_gate_boundary_inside_and_outside_interval() { + let min = Duration::from_secs(3600); + let gate = Mutex::new(None); + let t0 = Instant::now(); + + // First trigger always allowed. + assert!(try_record_compact_trigger(&gate, min, t0)); + + // Just inside the interval: still blocked. + assert!(!try_record_compact_trigger( + &gate, + min, + t0 + min - Duration::from_secs(1) + )); + + // Exactly at the boundary: allowed again. + assert!(try_record_compact_trigger(&gate, min, t0 + min)); + + // Just outside the next window: allowed. + assert!(try_record_compact_trigger( + &gate, + min, + t0 + min + min + Duration::from_secs(1) + )); + } + + #[test] + fn concurrent_compact_gate_triggers_at_most_once() { + let min = Duration::from_secs(3600); + let gate = Arc::new(Mutex::new(None)); + let triggers = Arc::new(AtomicU32::new(0)); + let num_threads = 64; + let barrier = Arc::new(Barrier::new(num_threads)); + let now = Instant::now(); + + let handles: Vec<_> = (0..num_threads) + .map(|_| { + let gate = gate.clone(); + let triggers = triggers.clone(); + let barrier = barrier.clone(); + thread::spawn(move || { + barrier.wait(); + if try_record_compact_trigger(&gate, min, now) { + triggers.fetch_add(1, Ordering::SeqCst); + } + }) + }) + .collect(); + + for h in handles { + h.join().unwrap(); + } + + assert_eq!( + triggers.load(Ordering::SeqCst), + 1, + "concurrent callers must stamp the gate at most once" + ); + } +} diff --git a/core/src/global.rs b/core/src/global.rs index 27d46750af..c3c15a2928 100644 --- a/core/src/global.rs +++ b/core/src/global.rs @@ -104,10 +104,11 @@ pub const COMPACTION_CHECK: u64 = DAY_HEIGHT; /// Minimum wall-clock interval between compaction runs (seconds). /// -/// `COMPACTION_CHECK` is height-based and assumes ~1 block/minute. During fast -/// sync many blocks arrive per second, so the probabilistic check alone would -/// compact far too often and slow sync down. Enforcing a wall-clock gap (1 hour) -/// limits that without changing post-sync average behavior. +/// `COMPACTION_CHECK` assumes ~1 block/minute. During fast sync many blocks +/// arrive per second, so the probabilistic check alone would compact too often. +/// Enforced in `Chain::compact` so every trigger path shares the same policy. +/// Process-local (cleared on restart); the height threshold in `compact()` +/// still limits compacting across frequent restarts. pub const MIN_COMPACTION_INTERVAL_SECS: u64 = 60 * 60; /// Number of blocks to reuse a txhashset zip for (automated testing and user testing). diff --git a/servers/src/common/adapters.rs b/servers/src/common/adapters.rs index 596395a00f..dd4ca413e6 100644 --- a/servers/src/common/adapters.rs +++ b/servers/src/common/adapters.rs @@ -15,7 +15,7 @@ //! Adapters connecting new block, new transaction, and accepted transaction //! events to consumers of those events. -use crate::util::{Mutex, RwLock}; +use crate::util::RwLock; use std::collections::HashMap; use std::fs::File; use std::net::SocketAddr; @@ -61,43 +61,6 @@ const WORKER_CHANNEL_BUFFER_SIZE: usize = 64; const HEADER_SEGMENT_REQUEST_WINDOW_SECS: i64 = 60; const MAX_HEADER_SEGMENT_REQUESTS_PER_WINDOW: usize = 120; -/// Wall-clock + probabilistic gate for chain compaction. If the dice hit and -/// at least `min_interval` has passed since the last recorded trigger (`None` -/// means never, always allowed), runs `start` and returns its result; the -/// trigger time is only recorded if `start` reports success. -/// -/// The lock is held across check+start+stamp so concurrent callers cannot -/// double-trigger in the same window, and `now` is read under the lock so an -/// older caller cannot move the timestamp backwards. -fn try_trigger_compaction( - gate: &Mutex>, - min_interval: std::time::Duration, - dice_hit: bool, - start: F, -) -> bool -where - F: FnOnce() -> bool, -{ - if !dice_hit { - return false; - } - let mut last = gate.lock(); - let now = Instant::now(); - let wall_clock_ok = match *last { - None => true, - Some(t) => now - .checked_duration_since(t) - .map(|d| d >= min_interval) - .unwrap_or(true), - }; - if wall_clock_ok && start() { - *last = Some(now); - true - } else { - false - } -} - /// Implementation of the NetAdapter for the . Gets notified when new /// blocks and transactions are received and forwards to the chain and pool /// implementations. @@ -113,9 +76,6 @@ where config: ServerConfig, hooks: Vec>, header_segment_requests: RwLock, usize)>>, - /// Wall-clock time of last successful compaction *trigger* (not completion). - /// Used with `MIN_COMPACTION_INTERVAL_SECS` to avoid compact storms during sync. - last_compact_trigger: Mutex>, tx: mpsc::SyncSender, } @@ -748,7 +708,6 @@ where config, hooks, header_segment_requests: RwLock::new(HashMap::new()), - last_compact_trigger: Mutex::new(None), tx, }; adapter.spawn_net_adapter_worker(Arc::downgrade(&chain), rx); @@ -1007,32 +966,20 @@ where } fn check_compact(&self) { - // Wall-clock throttle + height-based dice. During fast sync blocks arrive much - // faster than mainnet's 1/min, so the dice alone is too aggressive. - let min_interval = std::time::Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); - let dice_hit = { - let mut rng = thread_rng(); - 0 == rng.gen_range(0, global::COMPACTION_CHECK) - }; - - try_trigger_compaction(&self.last_compact_trigger, min_interval, dice_hit, || { - let chain = self.chain(); - let syncing = self.sync_state.is_syncing(); - match thread::Builder::new() - .name("compactor".to_string()) - .spawn(move || { - info!("check_compact: starting compaction (syncing={})", syncing); - if let Err(e) = chain.compact() { - error!("Could not compact chain: {:?}", e); - } - }) { - Ok(_) => true, - Err(e) => { - error!("Could not spawn compactor thread: {:?}", e); - false + // Roll the dice to trigger compaction at 1/COMPACTION_CHECK chance per block. + // Wall-clock throttle lives in Chain::compact so every caller shares it. + let mut rng = thread_rng(); + if 0 != rng.gen_range(0, global::COMPACTION_CHECK) { + return; + } + let chain = self.chain(); + let _ = thread::Builder::new() + .name("compactor".to_string()) + .spawn(move || { + if let Err(e) = chain.compact() { + error!("Could not compact chain: {:?}", e); } - } - }); + }); } fn request_transaction(&self, h: Hash, peer_info: &PeerInfo) { @@ -1355,98 +1302,3 @@ impl pool::BlockChain for PoolToChainAdapter { .map_err(|_| pool::PoolError::ImmatureTransaction) } } - -#[cfg(test)] -mod tests { - use super::*; - use std::cell::Cell; - use std::time::Duration; - - /// Simulate fast sync through the real gate: many "blocks" with the dice - /// always hitting must trigger compaction exactly once per wall-clock window. - /// Also covers dice misses and a failed start not consuming the window. - #[test] - fn compaction_gate_triggers_at_most_once_per_wall_clock_window() { - let min = Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); - let gate: Mutex> = Mutex::new(None); - let triggers = Cell::new(0u32); - let start = || { - triggers.set(triggers.get() + 1); - true - }; - - // 50k "blocks" in one wall-clock window (worst-case sync). - for _ in 0..50_000 { - try_trigger_compaction(&gate, min, true, start); - } - assert_eq!( - triggers.get(), - 1, - "expected exactly one compact trigger in a single wall-clock window" - ); - - // Dice miss never triggers, even on a fresh gate. - assert!(!try_trigger_compaction( - &Mutex::new(None), - min, - false, - start - )); - assert_eq!(triggers.get(), 1); - - // A failed start must not consume the window: the next successful - // start is still allowed. - let gate = Mutex::new(None); - assert!(!try_trigger_compaction(&gate, min, true, || false)); - assert!(try_trigger_compaction(&gate, min, true, start)); - assert_eq!(triggers.get(), 2); - - // Once the interval has elapsed the gate opens again (zero interval - // stands in for elapsed time, since `Instant` cannot be advanced). - assert!(try_trigger_compaction( - &gate, - Duration::from_secs(0), - true, - start - )); - assert_eq!(triggers.get(), 3); - } - - /// Concurrent callers racing on the shared gate must trigger at most once. - #[test] - fn concurrent_compaction_gate_triggers_at_most_once() { - use std::sync::atomic::{AtomicU32, Ordering}; - use std::sync::Barrier; - - let min = Duration::from_secs(global::MIN_COMPACTION_INTERVAL_SECS); - let gate: Arc>> = Arc::new(Mutex::new(None)); - let triggers = Arc::new(AtomicU32::new(0)); - let num_threads = 64; - let barrier = Arc::new(Barrier::new(num_threads)); - - let handles: Vec<_> = (0..num_threads) - .map(|_| { - let gate = gate.clone(); - let triggers = triggers.clone(); - let barrier = barrier.clone(); - thread::spawn(move || { - barrier.wait(); - try_trigger_compaction(&gate, min, true, || { - triggers.fetch_add(1, Ordering::SeqCst); - true - }); - }) - }) - .collect(); - - for h in handles { - h.join().unwrap(); - } - - assert_eq!( - triggers.load(Ordering::SeqCst), - 1, - "concurrent callers sharing the gate must trigger compaction at most once per window" - ); - } -}