Enable reading un-synced data in db stress test (#12752)

Summary:
**Context/Summary:**
There are a few blockers to enabling reading un-synced data in db stress test
(1) GetFileSize() will always return 0 for file written under direct IO because we don't track the last flushed position for `TestFSWritableFile` under direct IO. So it will surface as
```
Verification failed: VerifyChecksum failed: Corruption: file is too short (0 bytes) to be an sstable: /tmp/rocksdb_crashtest_blackbox4deg_c5e/000009.sst
db_stress: db_stress_tool/db_stress_test_base.cc:518: void rocksdb::StressTest::ProcessStatus(rocksdb::SharedState*, std::string, const rocksdb::Status&, bool) const: Assertion `false' failed.
Received signal 6 (Aborted)
Invoking GDB for stack trace...
```
(2) A couple minor FIXME in left in https://github.com/facebook/rocksdb/pull/12729.

This PR fixed (1) and (2) and enabled reading un-synced data in stress test.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/12752

Test Plan:
- The following command failed before this PR and passed after.

```
./db_stress --WAL_size_limit_MB=1 --WAL_ttl_seconds=0 --acquire_snapshot_one_in=100 --adaptive_readahead=0 --adm_policy=0 --advise_random_on_open=1 --allow_concurrent_memtable_write=0 --allow_data_in_errors=True --allow_fallocate=1 --async_io=1 --atomic_flush=1 --auto_readahead_size=1 --avoid_flush_during_recovery=0 --avoid_flush_during_shutdown=1 --avoid_unnecessary_blocking_io=0 --backup_max_size=104857600 --backup_one_in=100000 --batch_protection_bytes_per_key=0 --bgerror_resume_retry_interval=10000 --block_align=0 --block_protection_bytes_per_key=4 --block_size=16384 --bloom_before_level=2147483647 --bloom_bits=37.92024930098943 --bottommost_compression_type=disable --bottommost_file_compaction_delay=0 --bytes_per_sync=0 --cache_index_and_filter_blocks=1 --cache_index_and_filter_blocks_with_high_priority=0 --cache_size=8388608 --cache_type=auto_hyper_clock_cache --charge_compression_dictionary_building_buffer=1 --charge_file_metadata=0 --charge_filter_construction=0 --charge_table_reader=1 --check_multiget_consistency=0 --check_multiget_entity_consistency=0 --checkpoint_one_in=1000000 --checksum_type=kXXH3 --clear_column_family_one_in=0 --column_families=1 --compact_files_one_in=1000 --compact_range_one_in=1000 --compaction_pri=3 --compaction_readahead_size=0 --compaction_ttl=10 --compress_format_version=2 --compressed_secondary_cache_size=8388608 --compression_checksum=1 --compression_max_dict_buffer_bytes=0 --compression_max_dict_bytes=0 --compression_parallel_threads=1 --compression_type=zlib --compression_use_zstd_dict_trainer=0 --compression_zstd_max_train_bytes=0 --continuous_verification_interval=0 --daily_offpeak_time_utc= --data_block_index_type=0 --db=/tmp/rocksdb_crashtest_blackbox4deg_c5e --db_write_buffer_size=0 --default_temperature=kWarm --default_write_temperature=kHot --delete_obsolete_files_period_micros=30000000 --delpercent=4 --delrangepercent=1 --destroy_db_initially=0 --detect_filter_construct_corruption=0 --disable_file_deletions_one_in=1000000 --disable_manual_compaction_one_in=10000 --disable_wal=1 --dump_malloc_stats=0 --enable_checksum_handoff=0 --enable_compaction_filter=0 --enable_custom_split_merge=1 --enable_do_not_compress_roles=0 --enable_index_compression=1 --enable_memtable_insert_with_hint_prefix_extractor=0 --enable_pipelined_write=0 --enable_sst_partitioner_factory=1 --enable_thread_tracking=0 --enable_write_thread_adaptive_yield=0 --expected_values_dir=/tmp/rocksdb_crashtest_expected_8whyhdxm --fail_if_options_file_error=0 --fifo_allow_compaction=1 --file_checksum_impl=xxh64 --fill_cache=1 --flush_one_in=1000 --format_version=4 --get_all_column_family_metadata_one_in=10000 --get_current_wal_file_one_in=0 --get_live_files_apis_one_in=1000000 --get_properties_of_all_tables_one_in=100000 --get_property_one_in=100000 --get_sorted_wal_files_one_in=0 --hard_pending_compaction_bytes_limit=274877906944 --high_pri_pool_ratio=0.5 --index_block_restart_interval=9 --index_shortening=0 --index_type=0 --ingest_external_file_one_in=0 --initial_auto_readahead_size=0 --inplace_update_support=0 --iterpercent=10 --key_len_percent_dist=1,30,69 --key_may_exist_one_in=100000 --last_level_temperature=kUnknown --level_compaction_dynamic_level_bytes=1 --lock_wal_one_in=100 --log_file_time_to_roll=0 --log_readahead_size=0 --long_running_snapshots=0 --low_pri_pool_ratio=0 --lowest_used_cache_tier=0 --manifest_preallocation_size=5120 --manual_wal_flush_one_in=0 --mark_for_compaction_one_file_in=0 --max_auto_readahead_size=0 --max_background_compactions=1 --max_bytes_for_level_base=67108864 --max_key=100000 --max_key_len=3 --max_log_file_size=0 --max_manifest_file_size=1073741824 --max_sequential_skip_in_iterations=2 --max_total_wal_size=0 --max_write_batch_group_size_bytes=1048576 --max_write_buffer_number=3 --max_write_buffer_size_to_maintain=2097152 --memtable_insert_hint_per_batch=1 --memtable_max_range_deletions=100 --memtable_prefix_bloom_size_ratio=0.001 --memtable_protection_bytes_per_key=0 --memtable_whole_key_filtering=0 --memtablerep=skip_list --metadata_charge_policy=1 --min_write_buffer_number_to_merge=2 --mmap_read=0 --mock_direct_io=False --nooverwritepercent=1 --num_file_reads_for_auto_readahead=1 --open_files=-1 --open_metadata_write_fault_one_in=0 --open_read_fault_one_in=0 --open_write_fault_one_in=16 --ops_per_thread=100000000 --optimize_filters_for_hits=1 --optimize_filters_for_memory=0 --optimize_multiget_for_io=0 --paranoid_file_checks=1 --partition_filters=0 --partition_pinning=1 --pause_background_one_in=1000000 --periodic_compaction_seconds=1000 --prefix_size=5 --prefixpercent=5 --prepopulate_block_cache=1 --preserve_internal_time_seconds=0 --progress_reports=0 --promote_l0_one_in=0 --read_amp_bytes_per_bit=0 --read_fault_one_in=1000 --readahead_size=16384 --readpercent=45 --recycle_log_file_num=0 --reopen=0 --report_bg_io_stats=0 --reset_stats_one_in=1000000 --sample_for_compression=5 --secondary_cache_fault_one_in=32 --secondary_cache_uri= --set_options_one_in=0 --skip_stats_update_on_db_open=0 --snapshot_hold_ops=100000 --soft_pending_compaction_bytes_limit=68719476736 --sst_file_manager_bytes_per_sec=104857600 --sst_file_manager_bytes_per_truncate=1048576 --stats_dump_period_sec=0 --stats_history_buffer_size=0 --strict_bytes_per_sync=0 --subcompactions=2 --sync=0 --sync_fault_injection=1 --table_cache_numshardbits=0 --target_file_size_base=16777216 --target_file_size_multiplier=1 --test_batches_snapshots=0 --top_level_index_pinning=0 --uncache_aggressiveness=1 --universal_max_read_amp=-1 --unpartitioned_pinning=0 --use_adaptive_mutex=0 --use_adaptive_mutex_lru=0 --use_attribute_group=1 --use_delta_encoding=1 --use_direct_io_for_flush_and_compaction=1 --use_direct_reads=0 --use_full_merge_v1=0 --use_get_entity=0 --use_merge=0 --use_multi_cf_iterator=0 --use_multi_get_entity=1 --use_multiget=0 --use_put_entity_one_in=5 --use_timed_put_one_in=0 --use_write_buffer_manager=0 --user_timestamp_size=0 --value_size_mult=32 --verification_only=0 --verify_checksum=1 --verify_checksum_one_in=10 --verify_compression=1 --verify_db_one_in=10000 --verify_file_checksums_one_in=10 --verify_iterator_with_expected_state_one_in=5 --verify_sst_unique_id_in_manifest=1 --wal_bytes_per_sync=0 --wal_compression=zstd --write_buffer_size=33554432 --write_dbid_to_manifest=0 --write_fault_one_in=0 --writepercent=35

Verification failed: VerifyChecksum failed: Corruption: file is too short (0 bytes) to be an sstable: /tmp/rocksdb_crashtest_blackbox4deg_c5e/000009.sst
db_stress: db_stress_tool/db_stress_test_base.cc:518: void rocksdb::StressTest::ProcessStatus(rocksdb::SharedState*, std::string, const rocksdb::Status&, bool) const: Assertion `false' failed.
Received signal 6 (Aborted)
Invoking GDB for stack trace...
```
- Run python3 tools/db_crashtest.py --simple blackbox --lock_wal_one_in=10 --backup_one_in=10 --sync_fault_injection=0 --use_direct_io_for_flush_and_compaction=0 for 1 hour
- Monitor stress test CI

Reviewed By: pdillinger

Differential Revision: D58395807

Pulled By: hx235

fbshipit-source-id: 7d4b321acc0a0af3501b62dc417a7f6e2d318265

2 files changed, 66 insertions, 24 deletions

diff --git a/utilities/fault_injection_fs.cc b/utilities/fault_injection_fs.cc
index 1cdc1256f..397adfca3 100644
--- a/utilities/fault_injection_fs.cc
+++ b/utilities/fault_injection_fs.cc
@@ -151,7 +151,7 @@ TestFSWritableFile::TestFSWritableFile(const std::string& fname,
       writable_file_opened_(true),
       fs_(fs) {
   assert(target_ != nullptr);
-  state_.pos_ = 0;
+  state_.pos_at_last_append_ = 0;
 }
 
 TestFSWritableFile::~TestFSWritableFile() {
@@ -166,14 +166,19 @@ IOStatus TestFSWritableFile::Append(const Slice& data, const IOOptions& options,
   if (!fs_->IsFilesystemActive()) {
     return fs_->GetError();
   }
+  IOStatus io_s;
   if (target_->use_direct_io()) {
-    target_->Append(data, options, dbg).PermitUncheckedError();
+    // TODO(hx235): buffer data for direct IO write to simulate data loss like
+    // non-direct IO write
+    io_s = target_->Append(data, options, dbg);
   } else {
     state_.buffer_.append(data.data(), data.size());
-    state_.pos_ += data.size();
+  }
+  if (io_s.ok()) {
+    state_.pos_at_last_append_ += data.size();
     fs_->WritableFileAppended(state_);
+    io_s = fs_->InjectWriteError(state_.filename_);
   }
-  IOStatus io_s = fs_->InjectWriteError(state_.filename_);
   return io_s;
 }
 
@@ -202,14 +207,46 @@ IOStatus TestFSWritableFile::Append(
         "current data checksum: " + Slice(checksum).ToString(true);
     return IOStatus::Corruption(msg);
   }
+  IOStatus io_s;
   if (target_->use_direct_io()) {
-    target_->Append(data, options, dbg).PermitUncheckedError();
+    // TODO(hx235): buffer data for direct IO write to simulate data loss like
+    // non-direct IO write
+    io_s = target_->Append(data, options, dbg);
   } else {
     state_.buffer_.append(data.data(), data.size());
-    state_.pos_ += data.size();
+  }
+  if (io_s.ok()) {
+    state_.pos_at_last_append_ += data.size();
+    fs_->WritableFileAppended(state_);
+    io_s = fs_->InjectWriteError(state_.filename_);
+  }
+  return io_s;
+}
+
+IOStatus TestFSWritableFile::Truncate(uint64_t size, const IOOptions& options,
+                                      IODebugContext* dbg) {
+  MutexLock l(&mutex_);
+  // TODO(hx235): inject error
+  IOStatus io_s = target_->Truncate(size, options, dbg);
+  if (io_s.ok()) {
+    state_.pos_at_last_append_ = size;
+  }
+  return io_s;
+}
+
+IOStatus TestFSWritableFile::PositionedAppend(const Slice& data,
+                                              uint64_t offset,
+                                              const IOOptions& options,
+                                              IODebugContext* dbg) {
+  MutexLock l(&mutex_);
+  // TODO(hx235): buffer data for direct IO write to simulate data loss like
+  // non-direct IO write
+  // TODO(hx235): inject error
+  IOStatus io_s = target_->PositionedAppend(data, offset, options, dbg);
+  if (io_s.ok()) {
+    state_.pos_at_last_append_ = offset + data.size();
     fs_->WritableFileAppended(state_);
   }
-  IOStatus io_s = fs_->InjectWriteError(state_.filename_);
   return io_s;
 }
 
@@ -236,8 +273,14 @@ IOStatus TestFSWritableFile::PositionedAppend(
         "current data checksum: " + Slice(checksum).ToString(true);
     return IOStatus::Corruption(msg);
   }
-  target_->PositionedAppend(data, offset, options, dbg);
-  IOStatus io_s = fs_->InjectWriteError(state_.filename_);
+  // TODO(hx235): buffer data for direct IO write to simulate data loss like
+  // non-direct IO write
+  IOStatus io_s = target_->PositionedAppend(data, offset, options, dbg);
+  if (io_s.ok()) {
+    state_.pos_at_last_append_ = offset + data.size();
+    fs_->WritableFileAppended(state_);
+    io_s = fs_->InjectWriteError(state_.filename_);
+  }
   return io_s;
 }
 
@@ -292,7 +335,7 @@ IOStatus TestFSWritableFile::Sync(const IOOptions& options,
   state_.buffer_.resize(0);
   // Ignore sync errors
   target_->Sync(options, dbg).PermitUncheckedError();
-  state_.pos_at_last_sync_ = state_.pos_;
+  state_.pos_at_last_sync_ = state_.pos_at_last_append_;
   fs_->WritableFileSynced(state_);
   return io_s;
 }
@@ -493,10 +536,11 @@ void FaultInjectionTestFS::AddUnsyncedToRead(const std::string& fname,
   auto it = db_file_state_.find(fname);
   if (it != db_file_state_.end()) {
     auto& st = it->second;
-    if (st.pos_ > static_cast<ssize_t>(pos_after)) {
+    if (st.pos_at_last_append_ > static_cast<ssize_t>(pos_after)) {
       size_t remaining_requested = n - result->size();
-      size_t to_copy = std::min(remaining_requested,
-                                static_cast<size_t>(st.pos_) - pos_after);
+      size_t to_copy =
+          std::min(remaining_requested,
+                   static_cast<size_t>(st.pos_at_last_append_) - pos_after);
       size_t buffer_offset = pos_after - static_cast<size_t>(std::max(
                                              st.pos_at_last_sync_, ssize_t{0}));
       // Data might have been dropped from buffer
@@ -805,7 +849,7 @@ IOStatus FaultInjectionTestFS::GetFileSize(const std::string& f,
     MutexLock l(&mutex_);
     auto it = db_file_state_.find(f);
     if (it != db_file_state_.end()) {
-      *file_size = it->second.pos_;
+      *file_size = it->second.pos_at_last_append_;
     }
   }
   return io_s;
diff --git a/utilities/fault_injection_fs.h b/utilities/fault_injection_fs.h
index ea76bf538..5335567f8 100644
--- a/utilities/fault_injection_fs.h
+++ b/utilities/fault_injection_fs.h
@@ -34,16 +34,18 @@ class FaultInjectionTestFS;
 
 struct FSFileState {
   std::string filename_;
-  ssize_t pos_;
+  ssize_t pos_at_last_append_;
   ssize_t pos_at_last_sync_;
   std::string buffer_;
 
   explicit FSFileState(const std::string& filename)
-      : filename_(filename), pos_(-1), pos_at_last_sync_(-1) {}
+      : filename_(filename), pos_at_last_append_(-1), pos_at_last_sync_(-1) {}
 
-  FSFileState() : pos_(-1), pos_at_last_sync_(-1) {}
+  FSFileState() : pos_at_last_append_(-1), pos_at_last_sync_(-1) {}
 
-  bool IsFullySynced() const { return pos_ <= 0 || pos_ == pos_at_last_sync_; }
+  bool IsFullySynced() const {
+    return pos_at_last_append_ <= 0 || pos_at_last_append_ == pos_at_last_sync_;
+  }
 
   IOStatus DropUnsyncedData();
 
@@ -65,9 +67,7 @@ class TestFSWritableFile : public FSWritableFile {
                   const DataVerificationInfo& verification_info,
                   IODebugContext* dbg) override;
   IOStatus Truncate(uint64_t size, const IOOptions& options,
-                    IODebugContext* dbg) override {
-    return target_->Truncate(size, options, dbg);
-  }
+                    IODebugContext* dbg) override;
   IOStatus Close(const IOOptions& options, IODebugContext* dbg) override;
   IOStatus Flush(const IOOptions&, IODebugContext*) override;
   IOStatus Sync(const IOOptions& options, IODebugContext* dbg) override;
@@ -76,9 +76,7 @@ class TestFSWritableFile : public FSWritableFile {
   bool IsSyncThreadSafe() const override { return true; }
   IOStatus PositionedAppend(const Slice& data, uint64_t offset,
                             const IOOptions& options,
-                            IODebugContext* dbg) override {
-    return target_->PositionedAppend(data, offset, options, dbg);
-  }
+                            IODebugContext* dbg) override;
   IOStatus PositionedAppend(const Slice& data, uint64_t offset,
                             const IOOptions& options,
                             const DataVerificationInfo& verification_info,