diff --git a/.changelog/persist-parallel-invariant-corpus.md b/.changelog/persist-parallel-invariant-corpus.md new file mode 100644 index 0000000000000..b0e82b981667f --- /dev/null +++ b/.changelog/persist-parallel-invariant-corpus.md @@ -0,0 +1,6 @@ +--- +forge: patch +foundry-evm: patch +--- + +Fixed parallel invariant campaigns losing their corpus when externally terminated. diff --git a/crates/evm/evm/src/executors/corpus.rs b/crates/evm/evm/src/executors/corpus.rs index c82626ad5e8e6..bd7c06c6e54b4 100644 --- a/crates/evm/evm/src/executors/corpus.rs +++ b/crates/evm/evm/src/executors/corpus.rs @@ -148,14 +148,19 @@ impl CorpusEntry { } fn write_to_disk_in(&self, dir: &Path, can_gzip: bool) -> foundry_common::fs::Result { - let file_name = self.file_name(can_gzip); + let should_gzip = self.should_gzip(can_gzip); + let file_name = self.file_name(should_gzip); let path = dir.join(&file_name); let temp_path = dir.join(format!(".{file_name}.{}.tmp", Uuid::new_v4())); - if self.should_gzip(can_gzip) { - foundry_common::fs::write_json_gzip_file(&temp_path, &self.tx_seq)?; + let write_result = if should_gzip { + foundry_common::fs::write_json_gzip_file(&temp_path, &self.tx_seq) } else { - foundry_common::fs::write_json_file(&temp_path, &self.tx_seq)?; + foundry_common::fs::write_json_file(&temp_path, &self.tx_seq) + }; + if let Err(err) = write_result { + let _ = foundry_common::fs::remove_file(&temp_path); + return Err(err); } if let Err(err) = std::fs::rename(&temp_path, &path) { @@ -166,11 +171,11 @@ impl CorpusEntry { Ok(path) } - fn file_name(&self, can_gzip: bool) -> String { + fn file_name(&self, gzip: bool) -> String { if let Some(name) = &self.persisted_file_name { return name.clone(); } - let ext = if self.should_gzip(can_gzip) { ".json.gz" } else { ".json" }; + let ext = if gzip { ".json.gz" } else { ".json" }; format!("{}-{}{ext}", self.uuid, self.timestamp) } @@ -183,13 +188,6 @@ impl CorpusEntry { } } -/// Corpus entry selected by a worker and returned for logical-campaign persistence. -#[derive(Debug, Clone)] -pub(crate) struct CampaignCorpusEntry { - tx_seq: Vec, - dedupe_by_coverage: bool, -} - /// Persists one call sequence as a corpus seed in the canonical worker0 corpus directory. pub fn persist_corpus_seed( config: &FuzzCorpusConfig, @@ -294,7 +292,6 @@ fn accept_synced_corpus_file( #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum CorpusInsertionMode { Live, - Deferred, MemoryOnly, } @@ -467,9 +464,9 @@ impl WorkerCorpusSeed { let mut seen_entries = seed.in_memory_corpus.iter().map(|entry| entry.uuid).collect::>(); for entry in unique_corpus_entries(&replay_dirs, &mut seen_entries) { - // A corrupt or truncated corpus file (e.g. a process killed mid-write, since entries - // are persisted non-atomically) must not abort the whole campaign startup: skip it - // and keep loading the rest of the corpus. + // A corrupt or truncated corpus file must not abort the whole campaign startup: skip + // it and keep loading the rest of the corpus. Canonical entries are atomically + // published, but malformed files may come from older versions or manual edits. let tx_seq = match entry.read_tx_seq() { Ok(tx_seq) => tx_seq, Err(err) => { @@ -511,50 +508,6 @@ impl WorkerCorpusSeed { Ok(seed) } - - /// Filters and persists logical-campaign corpus entries after worker results have merged. - /// - /// This consumes the deferred entries and writes each retained entry as soon as replay proves - /// it contributes new coverage. Keeping this path streaming avoids building a second filtered - /// copy of every campaign entry during invariant finalization. - pub(crate) fn persist_filtered_campaign_outputs( - &self, - config: &FuzzCorpusConfig, - entries: impl IntoIterator, - executor: &Executor, - target: ReplayTarget<'_>, - optimization_best: Option<(I256, &[BasicTxDetails])>, - ) -> Result<()> { - let mut history_map = self.history_map.clone(); - let mut edge_indices = self.edge_indices.clone(); - let mut sancov_history_map = self.sancov_history_map.clone(); - - let mut output_dir_ready = false; - for entry in entries { - if entry.dedupe_by_coverage { - let coverage = ReplayCoverage { - history_map: &mut history_map, - edge_indices: &mut edge_indices, - sancov_history_map: &mut sancov_history_map, - metrics: None, - }; - let ReplayOutcome { keep_entry, new_coverage, .. } = - replay_corpus_sequence(&entry.tx_seq, executor, target, coverage)?; - if !keep_entry || !new_coverage { - continue; - } - } - - if !output_dir_ready { - prepare_campaign_output_dir(config); - output_dir_ready = true; - } - persist_campaign_entry(config, entry); - } - - persist_optimization_output(config, optimization_best); - Ok(()) - } } #[derive(Default)] @@ -702,6 +655,8 @@ pub struct WorkerCorpus { /// Worker Dir /// corpus_dir/worker1/ worker_dir: Option, + /// Whether this worker already warned that a live corpus entry could not be persisted. + warned_persistence_failure: bool, /// Metrics at last sync - used to calculate deltas while syncing with global metrics last_sync_metrics: CorpusMetrics, /// Optimization mode: the best value found so far (loaded from disk or discovered in-run). @@ -930,17 +885,19 @@ impl WorkerCorpus { mut seed: WorkerCorpusSeed, ) -> Result { let initial_export_dirs = if id == 0 { seed.replay_dirs.take() } else { None }; - let worker_dir = config.corpus_dir.as_ref().map(|corpus_dir| { + let worker_dir = if let Some(corpus_dir) = &config.corpus_dir { let worker_dir = corpus_dir.join(format!("{WORKER}{id}")); let worker_corpus = worker_dir.join(CORPUS_DIR); let sync_dir = worker_dir.join(SYNC_DIR); // Create the necessary directories for the worker. - let _ = foundry_common::fs::create_dir_all(&worker_corpus); - let _ = foundry_common::fs::create_dir_all(&sync_dir); + foundry_common::fs::create_dir_all(&worker_corpus)?; + foundry_common::fs::create_dir_all(&sync_dir)?; - worker_dir - }); + Some(worker_dir) + } else { + None + }; Ok(Self { id, @@ -957,6 +914,7 @@ impl WorkerCorpus { new_entry_indices: Default::default(), initial_export_dirs, worker_dir, + warned_persistence_failure: false, last_sync_metrics: Default::default(), optimization_best_value: seed.optimization_best_value, optimization_best_sequence: seed.optimization_best_sequence, @@ -975,11 +933,20 @@ impl WorkerCorpus { new_coverage: bool, optimization: Option<(I256, Vec)>, ) { - let _ = self.process_inputs_inner(inputs, cmp_seq, new_coverage, optimization, true); + self.process_inputs_inner( + inputs, + cmp_seq, + new_coverage, + optimization, + CorpusInsertionMode::Live, + true, + ); } - /// Updates worker-local corpus state and returns any corpus entry to persist after the - /// logical campaign has merged worker outputs. + /// Updates worker-local corpus state and persists interesting inputs immediately, while + /// leaving campaign-wide optimization persistence to the coordinator. Entries are retained + /// per worker rather than globally filtered so abrupt termination cannot discard discoveries + /// that have not yet reached the coordinator. #[instrument(skip_all)] pub fn process_inputs_for_campaign( &mut self, @@ -987,8 +954,15 @@ impl WorkerCorpus { cmp_seq: &[Vec], new_coverage: bool, optimization: Option<(I256, Vec)>, - ) -> Option { - self.process_inputs_inner(inputs, cmp_seq, new_coverage, optimization, false) + ) { + self.process_inputs_inner( + inputs, + cmp_seq, + new_coverage, + optimization, + CorpusInsertionMode::Live, + false, + ); } fn process_inputs_inner( @@ -997,8 +971,9 @@ impl WorkerCorpus { cmp_seq: &[Vec], new_coverage: bool, optimization: Option<(I256, Vec)>, - persist_now: bool, - ) -> Option { + insertion_mode: CorpusInsertionMode, + persist_optimization: bool, + ) { // Check if this run improved the optimization value. let improved_optimization = optimization.as_ref().is_some_and(|(value, _)| { self.optimization_best_value.is_none_or(|best| *value > best) @@ -1029,18 +1004,18 @@ impl WorkerCorpus { { self.optimization_best_value = Some(value); self.optimization_best_sequence = best_seq; - if persist_now { + if persist_optimization { self.persist_optimization_state(); } } if !self.config.is_coverage_guided() { - return None; + return; } // Collect inputs if current run produced new coverage or improved optimization. if !new_coverage && !improved_optimization { - return None; + return; } // When the run is interesting only because of optimization (no new coverage), @@ -1055,7 +1030,7 @@ impl WorkerCorpus { inputs.to_vec() }; if corpus_inputs.is_empty() { - return None; + return; } let corpus_cmp_seq = cmp_seq .iter() @@ -1066,28 +1041,24 @@ impl WorkerCorpus { .collect(); let corpus = CorpusEntry::new_with_cmp(corpus_inputs, corpus_cmp_seq, Uuid::new_v4()); - self.insert_corpus_entry( - corpus, - if persist_now { CorpusInsertionMode::Live } else { CorpusInsertionMode::Deferred }, - new_coverage, - ) + self.insert_corpus_entry(corpus, insertion_mode) } - fn insert_corpus_entry( - &mut self, - corpus: CorpusEntry, - insertion_mode: CorpusInsertionMode, - dedupe_by_coverage: bool, - ) -> Option { - let campaign_entry = matches!(insertion_mode, CorpusInsertionMode::Deferred) - .then(|| CampaignCorpusEntry { tx_seq: corpus.tx_seq.clone(), dedupe_by_coverage }); - + fn insert_corpus_entry(&mut self, corpus: CorpusEntry, insertion_mode: CorpusInsertionMode) { if matches!(insertion_mode, CorpusInsertionMode::Live) && let Some(worker_dir) = &self.worker_dir { let worker_corpus = worker_dir.join(CORPUS_DIR); let write_result = corpus.write_to_disk_in(&worker_corpus, self.config.corpus_gzip); if let Err(err) = write_result { + if !self.warned_persistence_failure { + let _ = sh_warn!( + "Failed to persist coverage corpus entries for worker {} in {}: {err}", + self.id, + worker_corpus.display() + ); + self.warned_persistence_failure = true; + } debug!(target: "corpus", %err, "failed to record call sequence {:?}", corpus.tx_seq); } else { trace!( @@ -1100,7 +1071,6 @@ impl WorkerCorpus { } self.push_corpus_entry(corpus); - campaign_entry } fn push_corpus_entry(&mut self, corpus: CorpusEntry) { @@ -1122,25 +1092,7 @@ impl WorkerCorpus { let optimization_best = self .optimization_best_value .map(|value| (value, self.optimization_best_sequence.as_slice())); - Self::persist_campaign_outputs(&self.config, Vec::new(), optimization_best); - } - - /// Persists logical-campaign corpus and optimization outputs after worker results have merged. - pub(crate) fn persist_campaign_outputs( - config: &FuzzCorpusConfig, - entries: impl IntoIterator, - optimization_best: Option<(I256, &[BasicTxDetails])>, - ) { - let mut output_dir_ready = false; - for entry in entries { - if !output_dir_ready { - prepare_campaign_output_dir(config); - output_dir_ready = true; - } - persist_campaign_entry(config, entry); - } - - persist_optimization_output(config, optimization_best); + persist_optimization_output(&self.config, optimization_best); } /// Collects EVM and sancov coverage from call result and updates metrics. @@ -1183,9 +1135,9 @@ impl WorkerCorpus { parent_tx: &BasicTxDetails, targeted_contracts: &FuzzRunIdentifiedContracts, insertion_mode: CorpusInsertionMode, - ) -> Option { + ) { if !self.config.is_coverage_guided() || observed.is_empty() { - return None; + return; } let tx_seq = { @@ -1261,7 +1213,7 @@ impl WorkerCorpus { CorpusInsertionMode::MemoryOnly }; let len_before = self.in_memory_corpus.len(); - let _ = self.push_observed_sequence(seq, insertion_mode); + self.push_observed_sequence(seq, insertion_mode); if self.in_memory_corpus.len() > len_before { debug!(target: "corpus", test = %func.name, "seeded corpus sequence from test trace"); added += 1; @@ -1275,14 +1227,14 @@ impl WorkerCorpus { &mut self, tx_seq: Vec, insertion_mode: CorpusInsertionMode, - ) -> Option { + ) { if !self.config.is_coverage_guided() || tx_seq.is_empty() { - return None; + return; } let corpus = CorpusEntry::new(tx_seq); - self.insert_corpus_entry(corpus, insertion_mode, false) + self.insert_corpus_entry(corpus, insertion_mode) } /// Flush the oldest corpus mutated more than configured max mutations unless it is favored /// or pending synchronization. @@ -1506,7 +1458,7 @@ impl WorkerCorpus { delivered.insert(index); continue; }; - let file_name = corpus.file_name(self.config.corpus_gzip); + let file_name = corpus.file_name(corpus.should_gzip(self.config.corpus_gzip)); let file_path = corpus_dir.join(&file_name); if !file_path.is_file() && let Err(err) = corpus.write_to_disk_in(&corpus_dir, self.config.corpus_gzip) @@ -1553,7 +1505,8 @@ impl WorkerCorpus { delivered.insert(index); continue; }; - let path = master_corpus_dir.join(corpus.file_name(self.config.corpus_gzip)); + let path = master_corpus_dir + .join(corpus.file_name(corpus.should_gzip(self.config.corpus_gzip))); if !path.is_file() && let Err(err) = corpus.write_to_disk_in(&master_corpus_dir, self.config.corpus_gzip) @@ -1791,35 +1744,6 @@ fn sequence_from_observed( .collect() } -fn prepare_campaign_output_dir(config: &FuzzCorpusConfig) { - let Some(root) = &config.corpus_dir else { - return; - }; - let corpus_dir = root.join(format!("{WORKER}0")).join(CORPUS_DIR); - if let Err(err) = foundry_common::fs::create_dir_all(&corpus_dir) { - debug!(target: "corpus", %err, "failed to create campaign corpus dir"); - } -} - -fn persist_campaign_entry(config: &FuzzCorpusConfig, entry: CampaignCorpusEntry) { - let Some(root) = &config.corpus_dir else { - return; - }; - let corpus_dir = root.join(format!("{WORKER}0")).join(CORPUS_DIR); - let corpus = CorpusEntry::new(entry.tx_seq); - let write_result = corpus.write_to_disk_in(&corpus_dir, config.corpus_gzip); - if let Err(err) = write_result { - debug!(target: "corpus", %err, "failed to record call sequence {:?}", corpus.tx_seq); - } else { - trace!( - target: "corpus", - "persisted {} inputs for new coverage for {} corpus", - corpus.tx_seq.len(), - corpus.uuid, - ); - } -} - fn persist_optimization_output( config: &FuzzCorpusConfig, optimization_best: Option<(I256, &[BasicTxDetails])>, @@ -1844,6 +1768,14 @@ fn persist_optimization_output( } } +pub(crate) fn persist_campaign_optimization( + config: &FuzzCorpusConfig, + value: Option, + sequence: &[BasicTxDetails], +) { + persist_optimization_output(config, value.map(|value| (value, sequence))); +} + fn has_legacy_invariant_corpus_dirs(path: &Path) -> bool { std::fs::read_dir(path).is_ok_and(|entries| { entries.flatten().any(|entry| { @@ -1973,6 +1905,19 @@ mod tests { worker_corpus(id, corpus_root, WorkerCorpusSeed::default()) } + #[test] + fn worker_initialization_fails_when_corpus_directories_cannot_be_created() { + let corpus_root = temp_corpus_dir().join("not-a-directory"); + fs::write(&corpus_root, b"blocked").unwrap(); + let config = corpus_config(corpus_root); + let generator = + test_sequence(&config, TxGenerator::from_strategy(Just(basic_tx()).boxed())); + + assert!( + WorkerCorpus::from_seed(0, config, generator, WorkerCorpusSeed::default()).is_err() + ); + } + fn sync_test_executor(corpus_root: PathBuf, target: Address) -> Executor { let mut executor = ExecutorBuilder::::default().gas_limit(1 << 24).build( EvmEnvFor::::default(), @@ -2057,10 +2002,9 @@ mod tests { FuzzRunIdentifiedContracts::new(targets, false) } - // A corrupt/truncated corpus file (valid name, unparsable content — e.g. a process killed - // mid-write, since entries are persisted non-atomically) must surface as a per-entry read - // error rather than break directory scanning, so the load/sync loops can skip it instead of - // aborting the whole campaign. + // A corrupt/truncated corpus file (valid name, unparsable content) must surface as a per-entry + // read error rather than break directory scanning, so the load/sync loops can skip malformed + // files from older versions or manual edits instead of aborting the whole campaign. #[test] fn corrupt_corpus_file_surfaces_as_error_for_load_to_skip() { let dir = temp_corpus_dir(); @@ -2423,18 +2367,16 @@ mod tests { } #[test] - fn campaign_processing_returns_corpus_without_writing_worker_file() { + fn campaign_processing_writes_worker_file_immediately() { let corpus_root = temp_corpus_dir(); let worker_subdir = corpus_root.join("worker1"); let mut manager = empty_worker_corpus(1, corpus_root); - let record = manager.process_inputs_for_campaign(&[basic_tx()], &[], true, None); + manager.process_inputs_for_campaign(&[basic_tx()], &[], true, None); - let record = record.unwrap(); - assert!(record.dedupe_by_coverage); assert_eq!(manager.in_memory_corpus.len(), 1); assert_eq!(manager.metrics.corpus_count, 1); - assert_eq!(read_corpus_dir(&worker_subdir.join(CORPUS_DIR)).count(), 0); + assert_eq!(read_corpus_dir(&worker_subdir.join(CORPUS_DIR)).count(), 1); } /// `RawCallResult` carrying a single edge hit, to drive `merge_edge_coverage` without the EVM. @@ -2488,9 +2430,8 @@ mod tests { let worker_subdir = corpus_root.join("worker1"); let mut manager = empty_worker_corpus(1, corpus_root); - let record = manager.process_inputs_for_campaign(&[], &[], true, None); + manager.process_inputs_for_campaign(&[], &[], true, None); - assert!(record.is_none()); assert_eq!(manager.in_memory_corpus.len(), 0); assert_eq!(manager.metrics.corpus_count, 0); assert_eq!(read_corpus_dir(&worker_subdir.join(CORPUS_DIR)).count(), 0); @@ -2502,33 +2443,32 @@ mod tests { } #[test] - fn merged_campaign_outputs_write_corpus_and_optimization_to_master_dir() { + fn campaign_processing_defers_only_optimization_persistence() { let corpus_root = temp_corpus_dir(); let mut manager = empty_worker_corpus(1, corpus_root.clone()); let sequence = vec![basic_tx()]; - let record = manager - .process_inputs_for_campaign( - &sequence, - &[], - false, - Some((I256::try_from(7).unwrap(), sequence.clone())), - ) - .unwrap(); - let inputs = vec![record]; - WorkerCorpus::persist_campaign_outputs( - &corpus_config(corpus_root.clone()), - inputs, - Some((I256::try_from(7).unwrap(), &sequence)), + manager.process_inputs_for_campaign( + &sequence, + &[], + false, + Some((I256::try_from(7).unwrap(), sequence.clone())), ); - let master_corpus_dir = corpus_root.join("worker0").join(CORPUS_DIR); - let entries = read_corpus_dir(&master_corpus_dir).collect::>(); + let worker_corpus_dir = corpus_root.join("worker1").join(CORPUS_DIR); + let entries = read_corpus_dir(&worker_corpus_dir).collect::>(); assert_eq!(entries.len(), 1); let persisted_sequence = entries[0].read_tx_seq().unwrap(); assert_eq!(persisted_sequence.len(), sequence.len()); assert_eq!(persisted_sequence[0].sender, sequence[0].sender); assert_eq!(persisted_sequence[0].call_details.target, sequence[0].call_details.target); assert_eq!(persisted_sequence[0].call_details.calldata, sequence[0].call_details.calldata); + assert!(!corpus_root.join(OPTIMIZATION_BEST_FILE).exists()); + + persist_campaign_optimization( + &corpus_config(corpus_root.clone()), + Some(I256::try_from(7).unwrap()), + &sequence, + ); let state: OptimizationState = foundry_common::fs::read_json_file(&corpus_root.join(OPTIMIZATION_BEST_FILE)).unwrap(); @@ -2617,22 +2557,21 @@ mod tests { .unwrap(); let worse_sequence = vec![basic_tx()]; - let worse = manager.process_inputs_for_campaign( + manager.process_inputs_for_campaign( &worse_sequence, &[], false, Some((I256::try_from(50).unwrap(), worse_sequence.clone())), ); - assert!(worse.is_none()); let better_sequence = vec![basic_tx()]; - let better = manager.process_inputs_for_campaign( + manager.process_inputs_for_campaign( &better_sequence, &[], false, Some((I256::try_from(150).unwrap(), better_sequence.clone())), ); - assert!(better.is_some()); + assert_eq!(manager.optimization_best_value, Some(I256::try_from(150).unwrap())); } #[test] @@ -2885,14 +2824,13 @@ mod tests { }; let mut manager = empty_worker_corpus(0, temp_corpus_dir()); - let campaign_entry = manager.hoist_observed_calls( + manager.hoist_observed_calls( &observed, &parent_tx, &targeted_contracts, CorpusInsertionMode::Live, ); - assert!(campaign_entry.is_none()); assert_eq!(manager.in_memory_corpus.len(), 1); assert_eq!(manager.metrics.corpus_count, 1); @@ -2916,7 +2854,7 @@ mod tests { } #[test] - fn hoist_observed_calls_returns_deferred_campaign_entry_without_persisting() { + fn hoist_observed_calls_persists_immediately() { let target = Address::from([0x42; 20]); let foo = Function::parse("foo()").unwrap(); let selector = foo.selector(); @@ -2932,18 +2870,15 @@ mod tests { let worker_corpus_dir = corpus_root.join("worker1").join(CORPUS_DIR); let mut manager = empty_worker_corpus(1, corpus_root); - let campaign_entry = manager.hoist_observed_calls( + manager.hoist_observed_calls( &observed, &basic_tx(), &targeted_contracts, - CorpusInsertionMode::Deferred, + CorpusInsertionMode::Live, ); - let campaign_entry = campaign_entry.expect("deferred hoist should return campaign entry"); - assert!(!campaign_entry.dedupe_by_coverage); - assert_eq!(campaign_entry.tx_seq.len(), 1); assert_eq!(manager.in_memory_corpus.len(), 1); - assert_eq!(read_corpus_dir(&worker_corpus_dir).count(), 0); + assert_eq!(read_corpus_dir(&worker_corpus_dir).count(), 1); } #[test] @@ -2967,28 +2902,20 @@ mod tests { let mut manager = WorkerCorpus::from_seed(0, no_corpus_config, generator, WorkerCorpusSeed::default()) .unwrap(); - assert!( - manager - .hoist_observed_calls( - &observed, - &basic_tx(), - &targeted_contracts, - CorpusInsertionMode::Live - ) - .is_none() + manager.hoist_observed_calls( + &observed, + &basic_tx(), + &targeted_contracts, + CorpusInsertionMode::Live, ); assert!(manager.in_memory_corpus.is_empty()); let mut manager = empty_worker_corpus(0, temp_corpus_dir()); - assert!( - manager - .hoist_observed_calls( - &[], - &basic_tx(), - &targeted_contracts, - CorpusInsertionMode::Live - ) - .is_none() + manager.hoist_observed_calls( + &[], + &basic_tx(), + &targeted_contracts, + CorpusInsertionMode::Live, ); assert!(manager.in_memory_corpus.is_empty()); } @@ -3055,19 +2982,13 @@ mod tests { let worker0_corpus_dir = corpus_root.join("worker0").join(CORPUS_DIR); let mut manager = empty_worker_corpus(0, corpus_root.clone()); - assert!( - manager.push_observed_sequence(vec![basic_tx()], CorpusInsertionMode::Live).is_none() - ); + manager.push_observed_sequence(vec![basic_tx()], CorpusInsertionMode::Live); assert_eq!(manager.in_memory_corpus.len(), 1); assert_eq!(read_corpus_dir(&worker0_corpus_dir).count(), 1); let mut manager = empty_worker_corpus(1, corpus_root.clone()); let worker1_corpus_dir = corpus_root.join("worker1").join(CORPUS_DIR); - assert!( - manager - .push_observed_sequence(vec![basic_tx()], CorpusInsertionMode::MemoryOnly) - .is_none() - ); + manager.push_observed_sequence(vec![basic_tx()], CorpusInsertionMode::MemoryOnly); assert_eq!(manager.in_memory_corpus.len(), 1); assert_eq!(read_corpus_dir(&worker1_corpus_dir).count(), 0); } diff --git a/crates/evm/evm/src/executors/invariant/campaign.rs b/crates/evm/evm/src/executors/invariant/campaign.rs index 963c4fc98e631..00d2af541131e 100644 --- a/crates/evm/evm/src/executors/invariant/campaign.rs +++ b/crates/evm/evm/src/executors/invariant/campaign.rs @@ -2,7 +2,7 @@ use super::{ FailureKey, InvariantFailureMetrics, InvariantFailures, InvariantFuzzError, InvariantFuzzTestResult, InvariantMetrics, }; -use crate::executors::{EarlyExit, EvmExecutionCancellation, corpus::CampaignCorpusEntry}; +use crate::executors::{EarlyExit, EvmExecutionCancellation}; use alloy_primitives::{Address, I256, Selector}; use eyre::{Result, ensure}; use foundry_evm_coverage::HitMaps; @@ -206,13 +206,12 @@ impl InvariantCampaignState { pub struct InvariantWorkerOutput { pub plan: InvariantWorkerPlan, pub result: InvariantFuzzTestResult, - pub corpus_entries: Vec, } impl InvariantWorkerOutput { #[cfg(test)] pub const fn new(plan: InvariantWorkerPlan, result: InvariantFuzzTestResult) -> Self { - Self { plan, result, corpus_entries: Vec::new() } + Self { plan, result } } } @@ -245,14 +244,10 @@ impl InvariantCampaignAggregator { /// Validates the collected worker ranges and folds them into one logical campaign result. #[cfg(test)] pub fn finish(self) -> Result { - Ok(self.finish_with_corpus_entries()?.0) + self.finish_campaign() } - /// Validates the collected worker ranges and folds them into one logical campaign result with - /// corpus artifacts selected in logical worker order. - pub fn finish_with_corpus_entries( - mut self, - ) -> Result<(InvariantFuzzTestResult, Vec)> { + pub fn finish_campaign(mut self) -> Result { ensure!(!self.outputs.is_empty(), "missing invariant worker output"); self.outputs.sort_by_key(|output| output.plan.first_global_run); @@ -266,9 +261,7 @@ impl InvariantCampaignAggregator { /// worker may have completed fewer than its assigned runs, so the original static ranges can /// contain gaps. The merge still validates worker identity and preserves deterministic worker /// order, but final run count is derived from the completed worker counters. - pub fn finish_partial_with_corpus_entries( - mut self, - ) -> Result<(InvariantFuzzTestResult, Vec)> { + pub fn finish_partial(mut self) -> Result { ensure!(!self.outputs.is_empty(), "missing invariant worker output"); self.outputs.sort_by_key(|output| output.plan.first_global_run); @@ -277,9 +270,7 @@ impl InvariantCampaignAggregator { } } -fn fold_outputs( - outputs: Vec, -) -> Result<(InvariantFuzzTestResult, Vec)> { +fn fold_outputs(outputs: Vec) -> Result { let workers = outputs.len(); let mut errors = HashMap::default(); let mut handler_errors = HashMap::default(); @@ -290,11 +281,10 @@ fn fold_outputs( let mut gas_report_traces = Vec::new(); let mut line_coverage = None; let mut metrics = HashMap::default(); - let mut corpus_entries = Vec::new(); let mut failed_corpus_replays = 0; let mut optimization_best = None; - for InvariantWorkerOutput { plan, result, corpus_entries: worker_entries } in outputs { + for InvariantWorkerOutput { plan, result } in outputs { if plan.worker_id == 0 { failed_corpus_replays = result.failed_corpus_replays; } @@ -302,7 +292,6 @@ fn fold_outputs( errors.entry(invariant).or_insert(error); } merge_handler_errors(&mut handler_errors, result.handler_errors); - corpus_entries.extend(worker_entries); runs += result.runs; calls += result.calls; reverts += result.reverts; @@ -320,23 +309,20 @@ fn fold_outputs( } let (optimization_best_value, optimization_best_sequence) = optimization_best.map(|(value, sequence)| (Some(value), sequence)).unwrap_or_default(); - Ok(( - InvariantFuzzTestResult::new( - errors, - handler_errors, - runs, - calls, - reverts, - last_run_inputs, - gas_report_traces, - line_coverage, - metrics, - failed_corpus_replays, - workers, - optimization_best_value, - optimization_best_sequence, - ), - corpus_entries, + Ok(InvariantFuzzTestResult::new( + errors, + handler_errors, + runs, + calls, + reverts, + last_run_inputs, + gas_report_traces, + line_coverage, + metrics, + failed_corpus_replays, + workers, + optimization_best_value, + optimization_best_sequence, )) } @@ -833,12 +819,11 @@ mod tests { result_with_counts(1, 10, true, 0), )); - let (result, corpus_entries) = partial.finish_partial_with_corpus_entries().unwrap(); + let result = partial.finish_partial().unwrap(); assert_eq!(result.runs, 3); assert_eq!(result.calls, 30); assert_eq!(result.failed_corpus_replays, 5); - assert!(corpus_entries.is_empty()); } #[test] diff --git a/crates/evm/evm/src/executors/invariant/mod.rs b/crates/evm/evm/src/executors/invariant/mod.rs index 2ccfaf55cf4c9..0f52faedf09ee 100644 --- a/crates/evm/evm/src/executors/invariant/mod.rs +++ b/crates/evm/evm/src/executors/invariant/mod.rs @@ -3,6 +3,7 @@ use crate::{ DURATION_BETWEEN_METRICS_REPORT, EarlyExit, EvmError, Executor, RawCallResult, corpus::{ CorpusInsertionMode, DynamicTargetCtx, ReplayTarget, WorkerCorpus, WorkerCorpusSeed, + persist_campaign_optimization, }, }, inspectors::Fuzzer, @@ -426,20 +427,6 @@ fn invariant_focus_seed( }) } -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum InvariantCorpusPersistence { - /// Preserve the legacy single-worker behavior: each interesting input is written immediately. - Live, - /// Parallel workers return interesting inputs to the campaign coordinator for merged writes. - Deferred, -} - -impl InvariantCorpusPersistence { - const fn is_deferred(self) -> bool { - matches!(self, Self::Deferred) - } -} - /// Converts a cumulative campaign total into an average per-second rate. /// /// Returns `0.0` during the initial zero-elapsed startup window to avoid @@ -945,11 +932,6 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { dynamic: Some(&dynamic), }, )?; - let corpus_persistence = if actual_worker_count > 1 { - InvariantCorpusPersistence::Deferred - } else { - InvariantCorpusPersistence::Live - }; let mut runner = self.runner.clone(); let config = self.config.clone(); let setup_contracts = self.setup_contracts; @@ -959,7 +941,7 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { let campaign_state = Arc::new(InvariantCampaignState::new(early_exit.clone(), self.config.timeout)); - let worker_outputs = if corpus_persistence.is_deferred() { + let worker_outputs = if actual_worker_count > 1 { let worker_jobs = worker_plans .into_iter() .map(|worker_plan| { @@ -1007,10 +989,12 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { &campaign_state, worker_campaign_seed, worker_corpus_seed, - corpus_persistence, actual_worker_count, gas_report_samples, ); + if output.is_err() { + campaign_state.request_terminal_stop(); + } debug!("finished in {:?}", timer.elapsed()); output }) @@ -1048,7 +1032,6 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { &campaign_state, worker_campaign_seed, worker_corpus_seed, - corpus_persistence, actual_worker_count, gas_report_samples, )?] @@ -1058,27 +1041,16 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { for worker_output in worker_outputs { aggregator.push(worker_output); } - let (result, corpus_entries) = if campaign_state.is_timed_campaign() { - aggregator.finish_partial_with_corpus_entries()? + let result = if campaign_state.is_timed_campaign() { + aggregator.finish_partial()? } else { - aggregator.finish_with_corpus_entries()? + aggregator.finish_campaign()? }; - if corpus_persistence.is_deferred() { - let dynamic_target_ctx = self.dynamic_target_ctx(); - corpus_seed.persist_filtered_campaign_outputs( - &self.config.corpus, - corpus_entries, - &self.executor, - ReplayTarget { - stateless: None, - fuzzed_contracts: Some(&replay_targets), - dynamic: Some(&dynamic_target_ctx), - }, - result - .optimization_best_value - .map(|value| (value, result.optimization_best_sequence.as_slice())), - )?; - } + persist_campaign_optimization( + &self.config.corpus, + result.optimization_best_value, + &result.optimization_best_sequence, + ); Ok(result) } @@ -1098,7 +1070,6 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { campaign_state: &InvariantCampaignState, campaign_seed: InvariantCampaignSeed, corpus_seed: WorkerCorpusSeed, - corpus_persistence: InvariantCorpusPersistence, worker_count: usize, gas_report_samples: usize, ) -> Result { @@ -1120,8 +1091,6 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { &campaign_seed, corpus_seed, )?; - let mut corpus_entries = Vec::new(); - let mut runs = 0; campaign_state.sync_handler_failures(&invariant_test.test_data.failures); @@ -1480,20 +1449,13 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { campaign_state.sync_handler_failures(&invariant_test.test_data.failures); } - let insertion_mode = if corpus_persistence.is_deferred() { - CorpusInsertionMode::Deferred - } else { - CorpusInsertionMode::Live - }; for (observed_calls, parent_tx) in observed_call_entries { - if let Some(entry) = corpus_manager.hoist_observed_calls( + corpus_manager.hoist_observed_calls( &observed_calls, &parent_tx, &invariant_test.targeted_contracts, - insertion_mode, - ) { - corpus_entries.push(entry); - } + CorpusInsertionMode::Live, + ); } // Extend corpus only after the run and its optional hook have completed. @@ -1501,15 +1463,13 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { let prefix = current_run.inputs[..current_run.optimization_prefix_len].to_vec(); (v, prefix) }); - if corpus_persistence.is_deferred() { - if let Some(input) = corpus_manager.process_inputs_for_campaign( + if worker_count > 1 { + corpus_manager.process_inputs_for_campaign( ¤t_run.inputs, ¤t_run.cmp_seq, current_run.new_coverage, optimization, - ) { - corpus_entries.push(input); - } + ); } else { corpus_manager.process_inputs( ¤t_run.inputs, @@ -1580,11 +1540,8 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { } msg.push_str(&format!("⚠ {handler_bugs} handler bug(s)")); } - let msg = if corpus_persistence.is_deferred() { - format!("[w{}] {msg}", plan.worker_id) - } else { - msg - }; + let msg = + if worker_count > 1 { format!("[w{}] {msg}", plan.worker_id) } else { msg }; progress.set_message(msg); } } else if edge_coverage_enabled @@ -1661,7 +1618,7 @@ impl<'a, FEN: FoundryEvmNetwork> InvariantExecutor<'a, FEN> { // `first_global_run` offsets were computed from the original partition. plan }; - Ok(InvariantWorkerOutput { plan: reported_plan, result: worker_result, corpus_entries }) + Ok(InvariantWorkerOutput { plan: reported_plan, result: worker_result }) } fn shrink_handler_failures( @@ -2851,7 +2808,6 @@ mod tests { &campaign_state, campaign_seed, WorkerCorpusSeed::default(), - InvariantCorpusPersistence::Live, 1, 1, ); @@ -2878,7 +2834,6 @@ mod tests { assert!(output.result.line_coverage.is_none()); assert!(output.result.metrics.is_empty()); assert!(output.result.optimization_best_value.is_none()); - assert!(output.corpus_entries.is_empty()); } #[test] diff --git a/crates/forge/tests/cli/test_cmd/invariant/mod.rs b/crates/forge/tests/cli/test_cmd/invariant/mod.rs index a66daf034b544..a00bff881b2c3 100644 --- a/crates/forge/tests/cli/test_cmd/invariant/mod.rs +++ b/crates/forge/tests/cli/test_cmd/invariant/mod.rs @@ -2,6 +2,11 @@ use alloy_primitives::U256; use foundry_test_utils::{ TestCommand, forgetest_init, snapbox::cmd::OutputAssert, str, util::OutputExt, }; +use std::{ + process::Stdio, + thread, + time::{Duration, Instant}, +}; mod common; mod handler; @@ -1347,10 +1352,12 @@ Ran 3 test suites [ELAPSED]: 6 tests passed, 0 failed, 0 skipped (6 total tests) ); }); -forgetest_init!(contract_level_invariant_corpus_dir, |prj, cmd| { +forgetest_init!(parallel_invariant_corpus_uses_worker_dirs, |prj, cmd| { prj.update_config(|config| { config.invariant.runs = 2; config.invariant.depth = 2; + config.invariant.workers = + foundry_config::InvariantWorkers::Fixed(std::num::NonZeroUsize::new(2).unwrap()); config.invariant.corpus.corpus_dir = Some("invariant_corpus".into()); }); prj.add_test( @@ -1387,6 +1394,193 @@ Ran 1 test for test/ContractCorpusTest.t.sol:ContractCorpusTest assert!(contract_dir.exists()); assert!(!contract_dir.join("invariant_a").exists()); assert!(!contract_dir.join("invariant_b").exists()); + for worker in ["worker0", "worker1"] { + let worker_corpus = contract_dir.join(worker).join("corpus"); + assert!( + std::fs::read_dir(&worker_corpus).is_ok_and(|mut entries| entries.next().is_some()), + "expected {worker} to persist corpus entries during the campaign" + ); + } +}); + +forgetest_init!(parallel_invariant_corpus_survives_external_termination, |prj, cmd| { + prj.update_config(|config| { + config.invariant.runs = u32::MAX; + config.invariant.depth = 64; + config.invariant.workers = + foundry_config::InvariantWorkers::Fixed(std::num::NonZeroUsize::new(2).unwrap()); + config.invariant.corpus.corpus_dir = Some("invariant_corpus".into()); + config.invariant.corpus.corpus_gzip = false; + }); + prj.add_test( + "InterruptedCorpusTest.t.sol", + r#" +import {Test} from "forge-std/Test.sol"; + +contract InterruptedCorpusHandler { + uint256 public value; + + function set(uint256 next) external { + value = next; + } +} + +contract InterruptedCorpusTest is Test { + InterruptedCorpusHandler handler; + + function setUp() public { + handler = new InterruptedCorpusHandler(); + targetContract(address(handler)); + } + + function invariant_ok() public pure {} +} + "#, + ); + + // Finish compilation before starting the process that will be terminated. + cmd.args(["build", "-q"]).assert_success(); + cmd.forge_fuse().args([ + "test", + "--mc", + "InterruptedCorpusTest", + "--mt", + "invariant_ok", + "--fuzz-seed", + "0x574", + "-q", + ]); + cmd.cmd().stdout(Stdio::null()).stderr(Stdio::null()); + let mut child = cmd.cmd().spawn().unwrap(); + + let contract_dir = prj.root().join("invariant_corpus").join("InterruptedCorpusTest"); + let worker_dirs = + [contract_dir.join("worker0").join("corpus"), contract_dir.join("worker1").join("corpus")]; + let deadline = Instant::now() + Duration::from_secs(30); + let mut persisted = false; + while Instant::now() < deadline { + persisted = worker_dirs.iter().any(|dir| { + std::fs::read_dir(dir).is_ok_and(|mut entries| { + entries.any(|entry| { + entry.is_ok_and(|entry| { + entry.path().extension().is_some_and(|ext| ext == "json") + }) + }) + }) + }); + if persisted { + break; + } + thread::sleep(Duration::from_millis(25)); + } + + child.kill().unwrap(); + child.wait().unwrap(); + assert!(persisted, "parallel invariant campaign did not persist corpus before termination"); + + let entries = worker_dirs + .iter() + .flat_map(|dir| std::fs::read_dir(dir).into_iter().flatten().flatten()) + .map(|entry| entry.path()) + .filter(|path| path.extension().is_some_and(|ext| ext == "json")) + .collect::>(); + assert!(!entries.is_empty()); + for entry in entries { + let contents = std::fs::read_to_string(&entry).unwrap(); + serde_json::from_str::>(&contents).unwrap(); + } + + // Exercise normal startup discovery and EVM replay against the corpus retained by the killed + // campaign, rather than only checking that the files contain valid JSON. + prj.update_config(|config| { + config.invariant.runs = 1; + config.invariant.depth = 1; + }); + let replay = cmd + .forge_fuse() + .args([ + "test", + "--mc", + "InterruptedCorpusTest", + "--mt", + "invariant_ok", + "--fuzz-seed", + "0x574", + ]) + .assert_success(); + let stdout = replay.get_output().stdout_lossy(); + assert!(!stdout.contains("failed corpus replays"), "{stdout}"); +}); + +forgetest_init!(parallel_invariant_worker_error_stops_campaign, |prj, cmd| { + prj.update_config(|config| { + config.invariant.runs = u32::MAX; + config.invariant.depth = 64; + config.invariant.workers = + foundry_config::InvariantWorkers::Fixed(std::num::NonZeroUsize::new(2).unwrap()); + config.invariant.corpus.corpus_dir = Some("invariant_corpus".into()); + }); + prj.add_test( + "CorpusSetupFailureTest.t.sol", + r#" +import {Test} from "forge-std/Test.sol"; + +contract CorpusSetupFailureHandler { + uint256 public value; + + function set(uint256 next) external { + value = next; + } +} + +contract CorpusSetupFailureTest is Test { + CorpusSetupFailureHandler handler; + + function setUp() public { + handler = new CorpusSetupFailureHandler(); + targetContract(address(handler)); + } + + function invariant_ok() public pure {} +} + "#, + ); + + // Finish compilation before timing how promptly a worker setup error stops its sibling. + cmd.args(["build", "-q"]).assert_success(); + let worker1 = + prj.root().join("invariant_corpus").join("CorpusSetupFailureTest").join("worker1"); + std::fs::create_dir_all(worker1.parent().unwrap()).unwrap(); + std::fs::write(&worker1, b"not a directory").unwrap(); + + cmd.forge_fuse().args([ + "test", + "--mc", + "CorpusSetupFailureTest", + "--mt", + "invariant_ok", + "--fuzz-seed", + "0x574", + "-q", + ]); + cmd.cmd().stdout(Stdio::null()).stderr(Stdio::null()); + let mut child = cmd.cmd().spawn().unwrap(); + + let deadline = Instant::now() + Duration::from_secs(10); + let status = loop { + if let Some(status) = child.try_wait().unwrap() { + break Some(status); + } + if Instant::now() >= deadline { + child.kill().unwrap(); + child.wait().unwrap(); + break None; + } + thread::sleep(Duration::from_millis(25)); + }; + + let status = status.expect("worker corpus setup error did not stop the parallel campaign"); + assert!(!status.success(), "worker corpus setup error unexpectedly succeeded"); }); forgetest_init!(optimization_invariants_use_function_level_corpus_dir, |prj, cmd| {