fix: abort chain compaction cleanly on node shutdown

chain/src/chain.rs

@@ -1329,6 +1329,23 @@ impl Chain {
 	/// * removes historical blocks and associated data from the db (unless archive mode)
 	///
 	pub fn compact(&self) -> Result<(), Error> {
+		self.compact_with_stop(None)
+	}
+
+	/// Compact the chain, optionally aborting if `stop_state` is stopped.
+	/// Used on shutdown so long compaction does not keep rewriting files after
+	/// the node was asked to stop (#3842).
+	pub fn compact_with_stop(
+		&self,
+		stop_state: Option<Arc<crate::util::StopState>>,
+	) -> Result<(), Error> {
+		let should_abort = || stop_state.as_ref().map(|s| s.is_stopped()).unwrap_or(false);
+
+		if should_abort() {
+			info!("compact: not starting, node is stopping");
+			return Ok(());
+		}
+
 		// 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
 		// current "head" and "tail" height to our cut-through horizon and
@@ -1364,7 +1381,13 @@ impl Chain {
 			let horizon_hash = header_pmmr.get_header_hash_by_height(horizon_height)?;
 			let horizon_header = batch.get_block_header(&horizon_hash)?;
 
-			txhashset.compact(&horizon_header, &batch)?;
+			txhashset.compact_until(&horizon_header, &batch, should_abort)?;
+		}
+
+		if should_abort() {
+			info!("compact: aborted after txhashset compact (node stopping)");
+			// Do not commit further DB changes if we aborted mid-flight.
+			return Ok(());
 		}
 
 		// If we are not in archival mode remove historical blocks from the db.

chain/src/txhashset/txhashset.rs

@@ -528,21 +528,57 @@ impl TxHashSet {
 		horizon_header: &BlockHeader,
 		batch: &Batch<'_>,
 	) -> Result<(), Error> {
+		self.compact_until(horizon_header, batch, || false)
+	}
+
+	/// Compact MMR data files, honouring an abort callback so shutdown can
+	/// cancel before live files are replaced (see #3842).
+	pub fn compact_until<F>(
+		&mut self,
+		horizon_header: &BlockHeader,
+		batch: &Batch<'_>,
+		should_abort: F,
+	) -> Result<(), Error>
+	where
+		F: Fn() -> bool + Copy,
+	{
 		debug!("txhashset: starting compaction...");
 
+		if should_abort() {
+			info!("txhashset: compaction aborted (node stopping)");
+			return Ok(());
+		}
+
 		let head_header = batch.head_header()?;
 
 		let rewind_rm_pos = input_pos_to_rewind(&horizon_header, &head_header, batch)?;
 
 		debug!("txhashset: check_compact output mmr backend...");
-		self.output_pmmr_h
-			.backend
-			.check_compact(horizon_header.output_mmr_size, &rewind_rm_pos)?;
+		let output_done = self.output_pmmr_h.backend.check_compact_until(
+			horizon_header.output_mmr_size,
+			&rewind_rm_pos,
+			should_abort,
+		)?;
+		if !output_done {
+			info!("txhashset: compaction aborted during output mmr compact");
+			return Ok(());
+		}
+
+		if should_abort() {
+			info!("txhashset: compaction aborted before rangeproof mmr compact");
+			return Ok(());
+		}
 
 		debug!("txhashset: check_compact rangeproof mmr backend...");
-		self.rproof_pmmr_h
-			.backend
-			.check_compact(horizon_header.output_mmr_size, &rewind_rm_pos)?;
+		let rproof_done = self.rproof_pmmr_h.backend.check_compact_until(
+			horizon_header.output_mmr_size,
+			&rewind_rm_pos,
+			should_abort,
+		)?;
+		if !rproof_done {
+			info!("txhashset: compaction aborted during rangeproof mmr compact");
+			return Ok(());
+		}
 
 		debug!("txhashset: ... compaction finished");
 

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::{RwLock, StopState};
 use std::collections::HashMap;
 use std::fs::File;
 use std::net::SocketAddr;
@@ -76,6 +76,8 @@ where
 	config: ServerConfig,
 	hooks: Vec<Box<dyn NetEvents + Send + Sync>>,
 	header_segment_requests: RwLock<HashMap<SocketAddr, (DateTime<Utc>, usize)>>,
+	/// Shared shutdown flag so compaction can abort cleanly (#3842).
+	stop_state: Arc<StopState>,
 	tx: mpsc::SyncSender<NetAdapterWorkerMessage>,
 }
 
@@ -698,6 +700,7 @@ where
 		tx_pool: Arc<RwLock<pool::TransactionPool<B, P>>>,
 		config: ServerConfig,
 		hooks: Vec<Box<dyn NetEvents + Send + Sync>>,
+		stop_state: Arc<StopState>,
 	) -> Self {
 		let (tx, rx) = mpsc::sync_channel(WORKER_CHANNEL_BUFFER_SIZE);
 		let adapter = NetToChainAdapter {
@@ -708,6 +711,7 @@ where
 			config,
 			hooks,
 			header_segment_requests: RwLock::new(HashMap::new()),
+			stop_state,
 			tx,
 		};
 		adapter.spawn_net_adapter_worker(Arc::downgrade(&chain), rx);
@@ -966,15 +970,25 @@ where
 	}
 
 	fn check_compact(&self) {
+		// Do not start long-running compaction while the node is shutting down.
+		if self.stop_state.is_stopped() {
+			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) {
 			let chain = self.chain();
+			let stop_state = self.stop_state.clone();
 			let _ = thread::Builder::new()
 				.name("compactor".to_string())
 				.spawn(move || {
-					if let Err(e) = chain.compact() {
+					if stop_state.is_stopped() {
+						debug!("compactor: not starting, node is stopping");
+						return;
+					}
+					if let Err(e) = chain.compact_with_stop(Some(stop_state)) {
 						error!("Could not compact chain: {:?}", e);
 					}
 				});

servers/src/grin/server.rs

@@ -223,6 +223,7 @@ impl Server {
 			tx_pool.clone(),
 			config.clone(),
 			init_net_hooks(&config)?,
+			stop_state.clone(),
 		));
 
 		// Initialize our capabilities.

servers/src/grin/sync/syncer.rs

@@ -174,7 +174,9 @@ impl SyncRunner {
 					// This triggers a chain compaction to keep out local node tidy.
 					// Note: Chain compaction runs with an internal threshold
 					// so can be safely run even if the node is restarted frequently.
-					unwrap_or_restart_loop!(self.chain.compact());
+					unwrap_or_restart_loop!(self
+						.chain
+						.compact_with_stop(Some(self.stop_state.clone())));
 				}
 
 				// sleep for 10 secs but check stop signal every second

store/src/pmmr.rs

@@ -414,8 +414,28 @@ impl<T: PMMRable> PMMRBackend<T> {
 	/// will have a suitable output_pos. This is used to enforce a horizon
 	/// after which the local node should have all the data to allow rewinding.
 	pub fn check_compact(&mut self, cutoff_pos: u64, rewind_rm_pos: &Bitmap) -> io::Result<bool> {
+		self.check_compact_until(cutoff_pos, rewind_rm_pos, || false)
+	}
+
+	/// Compact backend files, optionally aborting before live files are replaced
+	/// when `should_abort` returns true (e.g. node shutdown). Aborted compaction
+	/// discards `.tmp` files and leaves the live PMMR files untouched (#3842).
+	pub fn check_compact_until<F>(
+		&mut self,
+		cutoff_pos: u64,
+		rewind_rm_pos: &Bitmap,
+		should_abort: F,
+	) -> io::Result<bool>
+	where
+		F: Fn() -> bool,
+	{
 		assert!(self.prunable, "Trying to compact a non-prunable PMMR");
 
+		if should_abort() {
+			debug!("compact: aborted before writing tmp files");
+			return Ok(false);
+		}
+
 		// Calculate the sets of leaf positions and node positions to remove based
 		// on the cutoff_pos provided.
 		let (leaves_removed, pos_to_rm) = self.pos_to_rm(cutoff_pos, rewind_rm_pos);
@@ -447,6 +467,15 @@ impl<T: PMMRable> PMMRBackend<T> {
 			self.data_file.write_tmp_pruned(&pos_to_rm)?;
 		}
 
+		// Critical section: do not replace live files if we are shutting down.
+		// Leaving a half-applied replace causes Invalid Root on next start (#3842).
+		if should_abort() {
+			info!("compact: aborted before replace; discarding tmp files");
+			self.hash_file.discard_tmp();
+			self.data_file.discard_tmp();
+			return Ok(false);
+		}
+
 		// Replace hash and data files with compact copies.
 		// Rebuild and intialize from the new files.
 		{

store/src/types.rs

@@ -18,6 +18,7 @@ use crate::grin_core::ser::{
 	self, BinWriter, DeserializationMode, ProtocolVersion, Readable, Reader, StreamingReader,
 	Writeable, Writer,
 };
+use log::warn;
 use std::fmt::Debug;
 use std::fs::{self, File, OpenOptions};
 use std::io::{self, BufReader, BufWriter, Seek, SeekFrom, Write};
@@ -160,6 +161,11 @@ where
 		self.file.write_tmp_pruned(prune_idx.as_slice())
 	}
 
+	/// Discard any temporary pruned file without applying it.
+	pub fn discard_tmp(&self) {
+		self.file.discard_tmp()
+	}
+
 	/// Replace with file at tmp path.
 	/// Rebuild and initialize from new file.
 	pub fn replace_with_tmp(&mut self) -> io::Result<()> {
@@ -521,6 +527,16 @@ where
 		Ok(())
 	}
 
+	/// Drop any `.tmp` companion file without replacing the live file.
+	pub fn discard_tmp(&self) {
+		let tmp = self.tmp_path();
+		if tmp.exists() {
+			if let Err(e) = fs::remove_file(&tmp) {
+				warn!("discard_tmp: failed to remove {:?}: {}", tmp, e);
+			}
+		}
+	}
+
 	/// Replace the underlying file with the file at tmp path.
 	/// Rebuild and initialize from the new file.
 	pub fn replace_with_tmp(&mut self) -> io::Result<()> {

store/tests/pmmr.rs

@@ -214,6 +214,36 @@ fn pmmr_compact_leaf_sibling() {
 	teardown(data_dir);
 }
 
+/// Aborting compaction before replace must leave live files intact (#3842).
+#[test]
+fn pmmr_compact_abort_before_replace() {
+	let (data_dir, elems) = setup("compact_abort");
+	{
+		let mut backend =
+			store::pmmr::PMMRBackend::new(data_dir.to_string(), true, ProtocolVersion(1), None)
+				.unwrap();
+		let mmr_size = load(0, &elems[..], &mut backend);
+		backend.sync().unwrap();
+		let root_before = {
+			let pmmr = PMMR::at(&mut backend, mmr_size);
+			pmmr.root().unwrap()
+		};
+
+		// Always abort: should not replace live files.
+		let done = backend
+			.check_compact_until(2, &Bitmap::new(), || true)
+			.unwrap();
+		assert!(!done);
+
+		let root_after = {
+			let pmmr = PMMR::at(&mut backend, mmr_size);
+			pmmr.root().unwrap()
+		};
+		assert_eq!(root_before, root_after);
+	}
+	teardown(data_dir);
+}
+
 #[test]
 fn pmmr_prune_compact() {
 	let (data_dir, elems) = setup("prune_compact");