diff --git a/chain/src/chain.rs b/chain/src/chain.rs index 0cad711e9..799785447 100644 --- a/chain/src/chain.rs +++ b/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>, + ) -> 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. diff --git a/chain/src/txhashset/txhashset.rs b/chain/src/txhashset/txhashset.rs index 4f8446b2a..8645bf044 100644 --- a/chain/src/txhashset/txhashset.rs +++ b/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( + &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"); diff --git a/servers/src/common/adapters.rs b/servers/src/common/adapters.rs index 5bdcb5301..8945b4599 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::{RwLock, StopState}; use std::collections::HashMap; use std::fs::File; use std::net::SocketAddr; @@ -76,6 +76,8 @@ where config: ServerConfig, hooks: Vec>, header_segment_requests: RwLock, usize)>>, + /// Shared shutdown flag so compaction can abort cleanly (#3842). + stop_state: Arc, tx: mpsc::SyncSender, } @@ -698,6 +700,7 @@ where tx_pool: Arc>>, config: ServerConfig, hooks: Vec>, + stop_state: Arc, ) -> 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); } }); diff --git a/servers/src/grin/server.rs b/servers/src/grin/server.rs index 976d2081a..48246badc 100644 --- a/servers/src/grin/server.rs +++ b/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. diff --git a/servers/src/grin/sync/syncer.rs b/servers/src/grin/sync/syncer.rs index ed34cad85..413ee80ed 100644 --- a/servers/src/grin/sync/syncer.rs +++ b/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 diff --git a/store/src/pmmr.rs b/store/src/pmmr.rs index 53c6871e1..c073b1dee 100644 --- a/store/src/pmmr.rs +++ b/store/src/pmmr.rs @@ -414,8 +414,28 @@ impl PMMRBackend { /// 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 { + 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( + &mut self, + cutoff_pos: u64, + rewind_rm_pos: &Bitmap, + should_abort: F, + ) -> io::Result + 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 PMMRBackend { 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. { diff --git a/store/src/types.rs b/store/src/types.rs index 70cde93a2..b3e0696d0 100644 --- a/store/src/types.rs +++ b/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<()> { diff --git a/store/tests/pmmr.rs b/store/tests/pmmr.rs index 935b39a6b..4965dfe4d 100644 --- a/store/tests/pmmr.rs +++ b/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");