fix: throttle chain compaction with a 1h wall-clock gap

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;
 

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;
@@ -61,6 +61,33 @@ 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<Instant>,
+	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),
+	}
+}
+
+/// 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<Instant>,
+	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.
@@ -76,6 +103,9 @@ where
 	config: ServerConfig,
 	hooks: Vec<Box<dyn NetEvents + Send + Sync>>,
 	header_segment_requests: RwLock<HashMap<SocketAddr, (DateTime<Utc>, 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<Option<Instant>>,
 	tx: mpsc::SyncSender<NetAdapterWorkerMessage>,
 }
 
@@ -708,6 +738,7 @@ 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);
@@ -966,18 +997,36 @@ where
 	}
 
 	fn check_compact(&self) {
-		// 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) {
-			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);
-					}
-				});
+		// 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 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);
+				}
+			}) {
+			Ok(_) => *last = Some(now),
+			Err(e) => error!("Could not spawn compactor thread: {:?}", e),
 		}
 	}
 
@@ -1301,3 +1350,138 @@ 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);
+	}
+
+	/// 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<Instant> = 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));
+	}
+
+	/// 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<Mutex<Option<Instant>>> = 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"
+		);
+	}
+}