Stress/crash test improvement to remote compaction with resumable compaction (facebook#14041)

Summary:
**Context/Summary:**
- Add resumable compaction to stress test with adaptive progress cancellation
- Add fault injection to remote compaction
- Fix a real minor bug in a couple testing framework bugs with remote compaction

Pull Request resolved: facebook#14041

Test Plan: - Rehearsal stress test, finding bugs for facebook#13984 effectively and did not create new failures.

Reviewed By: jaykorean

Differential Revision: D84524194

Pulled By: hx235

fbshipit-source-id: 42b4264e428c6739631ed9aa5eb02723367510bc

Author: hx235

BUCK

@@ -425,6 +425,7 @@ rocks_cpp_library_wrapper(name="rocksdb_stress_lib", srcs=[
         "db_stress_tool/batched_ops_stress.cc",
         "db_stress_tool/cf_consistency_stress.cc",
         "db_stress_tool/db_stress_common.cc",
+        "db_stress_tool/db_stress_compaction_service.cc",
         "db_stress_tool/db_stress_compression_manager.cc",
         "db_stress_tool/db_stress_driver.cc",
         "db_stress_tool/db_stress_filters.cc",

db/compaction/compaction_job.cc

@@ -950,7 +950,8 @@ void CompactionJob::FinalizeCompactionRun(
     UpdateCompactionJobInputStatsFromInternalStats(internal_stats_,
                                                    num_input_range_del);
   }
-  UpdateCompactionJobOutputStatsFromInternalStats(internal_stats_);
+  UpdateCompactionJobOutputStatsFromInternalStats(input_status,
+                                                  internal_stats_);
   RecordCompactionIOStats();
 
   LogFlush(db_options_.info_log);
@@ -2527,6 +2528,7 @@ void CompactionJob::UpdateCompactionJobInputStatsFromInternalStats(
 }
 
 void CompactionJob::UpdateCompactionJobOutputStatsFromInternalStats(
+    const Status& status,
     const InternalStats::CompactionStatsFull& internal_stats) const {
   assert(job_stats_);
   job_stats_->elapsed_micros = internal_stats.output_level_stats.micros;
@@ -2557,7 +2559,7 @@ void CompactionJob::UpdateCompactionJobOutputStatsFromInternalStats(
         internal_stats.proximal_level_stats.num_output_files_blob;
   }
 
-  if (job_stats_->num_output_files > 0) {
+  if (status.ok() && job_stats_->num_output_files > 0) {
     CopyPrefix(compact_->SmallestUserKey(),
                CompactionJobStats::kMaxPrefixLength,
                &job_stats_->smallest_output_key_prefix);

db/compaction/compaction_job.h

@@ -208,6 +208,7 @@ class CompactionJob {
 
  protected:
   void UpdateCompactionJobOutputStatsFromInternalStats(
+      const Status& status,
       const InternalStats::CompactionStatsFull& internal_stats) const;
 
   void LogCompaction();

db/compaction/compaction_service_job.cc

@@ -411,7 +411,7 @@ Status CompactionServiceCompactionJob::Run() {
   // 2. Update job-level output stats with the aggregated internal_stats_
   // Please note that input stats will be updated by primary host when all
   // subcompactions are finished
-  UpdateCompactionJobOutputStatsFromInternalStats(internal_stats_);
+  UpdateCompactionJobOutputStatsFromInternalStats(status, internal_stats_);
   // and set fields that are not propagated as part of the update
   compaction_result_->stats.is_manual_compaction = c->is_manual_compaction();
   compaction_result_->stats.is_full_compaction = c->is_full_compaction();

db_stress_tool/CMakeLists.txt

@@ -2,6 +2,7 @@ add_executable(db_stress${ARTIFACT_SUFFIX}
   batched_ops_stress.cc
   cf_consistency_stress.cc
   db_stress.cc
+  db_stress_compaction_service.cc
   db_stress_compression_manager.cc
   db_stress_common.cc
   db_stress_driver.cc