From bbdcd2825fd907c8014ba0995dca58880cb476fe Mon Sep 17 00:00:00 2001 From: Xingbo Wang Date: Mon, 17 Aug 2026 19:30:37 -0700 Subject: [PATCH] Add remote storage liveness crash test signal (#15102) Summary: Pull Request resolved: https://github.com/facebook/rocksdb/pull/15102 Add a dedicated Skycastle signal that invokes liveness through the existing remote-storage test-driver path. For remote DB runs, double default wrapper/no-progress timeouts while preserving explicit overrides. Default remote liveness `expected_values_dir` to local storage so `UniqueIdVerifier` never treats a remote DB path as a local path; local liveness behavior remains unchanged. Normalize both equals and spaced `--env_uri`/`--fs_uri` forms once so remote detection, workload execution, and cleanup share the same arguments. Track the crash harness and internal release scripts as CI source triggers, scoped so release-script changes select only the remote-storage signal. Add command-level regression coverage and failure-safe expected-values cleanup. Use generic remote DB/storage terminology throughout public RocksDB code and comments. Reviewed By: pdillinger Differential Revision: D115936164 fbshipit-source-id: e4997bf19eec18e4914626d479690b78c6cd3ad2 --- crash_test.mk | 4 + db/deletefile_test.cc | 4 +- tools/db_crashtest.py | 90 ++++++++-- tools/db_crashtest_test.py | 212 +++++++++++++++++++++++- utilities/checkpoint/checkpoint_impl.cc | 3 +- 5 files changed, 291 insertions(+), 22 deletions(-) diff --git a/crash_test.mk b/crash_test.mk index 07bf2ff66a..604b099cf7 100644 --- a/crash_test.mk +++ b/crash_test.mk @@ -25,6 +25,7 @@ CRASHTEST_PY=$(PYTHON) -u tools/db_crashtest.py --stress_cmd=$(DB_STRESS_CMD) -- crash_test_with_multiops_wup_txn \ crash_test_with_optimistic_txn \ crash_test_with_tiered_storage \ + liveness_crash_test \ blackbox_crash_test blackbox_crash_test_with_atomic_flush \ blackbox_crash_test_with_wc_txn blackbox_crash_test_with_wp_txn \ blackbox_crash_test_with_wup_txn \ @@ -94,6 +95,9 @@ crash_test_with_multiops_wp_txn: $(DB_STRESS_CMD) crash_test_with_multiops_wup_txn: $(DB_STRESS_CMD) $(CRASHTEST_MAKE) blackbox_crash_test_with_multiops_wup_txn +liveness_crash_test: $(DB_STRESS_CMD) + $(CRASHTEST_PY) liveness $(CRASH_TEST_EXT_ARGS) + blackbox_crash_test: $(DB_STRESS_CMD) $(CRASHTEST_PY) --simple blackbox $(CRASH_TEST_EXT_ARGS) $(CRASHTEST_PY) blackbox $(CRASH_TEST_EXT_ARGS) diff --git a/db/deletefile_test.cc b/db/deletefile_test.cc index 49aa2fa49c..cb0096b06d 100644 --- a/db/deletefile_test.cc +++ b/db/deletefile_test.cc @@ -262,8 +262,8 @@ TEST_F(DeleteFileTest, BackgroundPurgeIteratorTest) { // Regression test for a use-after-free where an obsolete-file purge started // DURING DB close -- after CloseHelper's early purge drain -- continued using // DBImpl after ~DBImpl destroyed mutex_. Seen as a vhost-user-blk SIGABRT when -// many RocksDB instances closed concurrently under a Warm Storage fault, with -// avoid_unnecessary_blocking_io=true on a shared, process-wide Env. +// many RocksDB instances closed concurrently under a remote file-system fault +// while using avoid_unnecessary_blocking_io=true on a shared, process-wide Env. // // Here we make the straggler deterministic: keep a dropped column family handle // alive, delete it only after CloseHelper has passed its early purge drain, diff --git a/tools/db_crashtest.py b/tools/db_crashtest.py index da6b9c0379..a80d95e6d8 100644 --- a/tools/db_crashtest.py +++ b/tools/db_crashtest.py @@ -44,7 +44,10 @@ _TSAN_SUPPRESSIONS_FILE = os.path.abspath( os.path.join(os.path.dirname(__file__), "tsan_suppressions.txt") ) DEFAULT_LIVENESS_TIMEOUT_SEC = 3600 +DEFAULT_LIVENESS_NO_PROGRESS_TIMEOUT_SEC = 300 +REMOTE_DB_LIVENESS_TIMEOUT_MULTIPLIER = 2 _FAULT_INJECTION_LOG_DIR_NAME = "fault_injection_logs" +_REMOTE_DB_URI_FLAGS = ("--env_uri", "--fs_uri") def get_random_seed(override): @@ -73,6 +76,43 @@ def stress_cmd_env(): return env +def normalize_remote_db_args(args): + """Normalize supported remote URI flags to --flag=value form.""" + normalized_args = [] + index = 0 + while index < len(args): + arg = args[index] + if ( + arg in _REMOTE_DB_URI_FLAGS + and index + 1 < len(args) + and args[index + 1] + and not args[index + 1].startswith("-") + ): + normalized_args.append(arg + "=" + args[index + 1]) + index += 2 + continue + normalized_args.append(arg) + index += 1 + return normalized_args + + +def parse_remote_db_uri_arg(arg): + flag, separator, value = arg.partition("=") + if flag not in _REMOTE_DB_URI_FLAGS or not separator: + return None + return flag, value + + +def remote_db_enabled(args): + effective_values = {} + for arg in args: + parsed_arg = parse_remote_db_uri_arg(arg) + if parsed_arg is not None: + flag, value = parsed_arg + effective_values[flag] = value + return any(effective_values.values()) + + def early_argument_parsing_before_main(): parser = argparse.ArgumentParser() parser.add_argument( @@ -93,16 +133,14 @@ def early_argument_parsing_before_main(): global remain_args args, remain_args = parser.parse_known_args() + remain_args = normalize_remote_db_args(remain_args) init_random_seed = get_random_seed(args.initial_random_seed_override) global per_iteration_random_seed_override per_iteration_random_seed_override = args.per_iteration_random_seed_override global is_remote_db - # Set is_remote_db if remain_args has a non-empty --env_uri= or --fs_uri= argument - for arg in remain_args: - parts = arg.split("=", 1) - if parts[0] in ["--env_uri", "--fs_uri"] and len(parts) > 1 and parts[1]: - is_remote_db = True - break + # Remote URI arguments are normalized above so all later consumers see + # the same representation. + is_remote_db = remote_db_enabled(remain_args) print(f"Start with random seed {init_random_seed}") random.seed(init_random_seed) @@ -699,12 +737,34 @@ liveness_default_params = { "duration": DEFAULT_LIVENESS_TIMEOUT_SEC, "enable_thread_tracking": 1, "liveness_check_interval_sec": 1, - "liveness_no_progress_timeout_sec": 300, + "liveness_no_progress_timeout_sec": DEFAULT_LIVENESS_NO_PROGRESS_TIMEOUT_SEC, "ops_per_thread": 100000000, "progress_reports": 1, } liveness_default_params.update(liveness_fault_injection_params) + +def liveness_default_timeout_sec(): + multiplier = REMOTE_DB_LIVENESS_TIMEOUT_MULTIPLIER if is_remote_db else 1 + return DEFAULT_LIVENESS_TIMEOUT_SEC * multiplier + + +def apply_liveness_remote_defaults(params, args): + if not is_remote_db: + return + + # Keep the wrapper duration in the same ratio as the no-progress timeout so + # the longer remote DB timeout still has enough run time to exercise + # liveness. + if getattr(args, "duration", None) is None: + params["duration"] = liveness_default_timeout_sec() + if getattr(args, "liveness_no_progress_timeout_sec", None) is None: + params["liveness_no_progress_timeout_sec"] = ( + DEFAULT_LIVENESS_NO_PROGRESS_TIMEOUT_SEC + * REMOTE_DB_LIVENESS_TIMEOUT_MULTIPLIER + ) + + simple_default_params = { "allow_concurrent_memtable_write": lambda: random.randint(0, 1), "column_families": 1, @@ -1033,10 +1093,10 @@ def finalize_and_sanitize(src_params): # Blob direct write requires concurrent read visibility of files still open # for writing (BlobFileReader calls GetFileSize() on active partition files). - # Remote file systems such as Warm Storage do not guarantee that writes from - # a WritableFile are visible to a separate RandomAccessFile until the writer - # is closed, causing "Malformed blob file" corruption. Disable BDW when a - # remote --env_uri / --fs_uri is in use. + # Some remote file systems do not guarantee that writes from a WritableFile + # are visible to a separate RandomAccessFile until the writer is closed, + # causing "Malformed blob file" corruption. Disable BDW when a remote + # --env_uri / --fs_uri is in use. if is_remote_db: dest_params["enable_blob_direct_write"] = 0 @@ -1736,6 +1796,7 @@ def gen_cmd_params(args): if v is not None: params[k] = v if args.test_type == "liveness": + apply_liveness_remote_defaults(params, args) params.update(liveness_fault_injection_params) return params @@ -2152,8 +2213,7 @@ def cleanup_after_success(db_arg, num_dbs=1): ] # Pass through relevant arguments for remote DB access for arg in remain_args: - parts = arg.split("=", 1) - if parts[0] in ["--env_uri", "--fs_uri"]: + if parse_remote_db_uri_arg(arg) is not None: cleanup_cmd_parts.append(arg) print("Running DB cleanup command - %s\n" % " ".join(cleanup_cmd_parts)) ret = subprocess.call(cleanup_cmd_parts, env=stress_cmd_env()) @@ -2202,7 +2262,7 @@ def print_fault_injection_log(pid): def liveness_timeout(cmd_params): duration = cmd_params.get("duration", 0) if duration is None or duration <= 0: - return DEFAULT_LIVENESS_TIMEOUT_SEC + return liveness_default_timeout_sec() return duration @@ -2213,6 +2273,8 @@ def liveness_main(args, unknown_args): if not cmd_params.get("db"): cmd_params["db"] = db_parent_dir + if is_remote_db and not cmd_params.get("expected_values_dir"): + cmd_params["expected_values_dir"] = get_ev_parent_dir() apply_random_seed_per_iteration() wrapper_timeout = liveness_timeout(cmd_params) diff --git a/tools/db_crashtest_test.py b/tools/db_crashtest_test.py index 2d8c44c2f1..d6d581ba21 100644 --- a/tools/db_crashtest_test.py +++ b/tools/db_crashtest_test.py @@ -13,6 +13,7 @@ import tempfile import unittest from contextlib import redirect_stdout from types import SimpleNamespace +from unittest import mock _DB_CRASHTEST_PATH = os.path.join(os.path.dirname(__file__), "db_crashtest.py") @@ -24,14 +25,14 @@ _TEST_EXPECTED_DIR_ENV_VAR = "TEST_TMPDIR_EXPECTED" _TSAN_OPTIONS_ENV_VAR = "TSAN_OPTIONS" -def load_db_crashtest_module(): +def load_db_crashtest_module(args=None): spec = importlib.util.spec_from_file_location( "db_crashtest_under_test", _DB_CRASHTEST_PATH ) module = importlib.util.module_from_spec(spec) old_argv = sys.argv[:] try: - sys.argv = [_DB_CRASHTEST_PATH] + sys.argv = [_DB_CRASHTEST_PATH] + list(args or []) spec.loader.exec_module(module) finally: sys.argv = old_argv @@ -77,8 +78,18 @@ class DBCrashTestTest(unittest.TestCase): shutil.rmtree(self.test_tmpdir) - def load_db_crashtest(self): - return load_db_crashtest_module() + def load_db_crashtest(self, args=None): + return load_db_crashtest_module(args) + + def register_expected_values_dir_cleanup(self, db_crashtest): + def cleanup_expected_values_dir(): + if db_crashtest.ev_parent_dir_global: + shutil.rmtree( + db_crashtest.ev_parent_dir_global, + ignore_errors=True, + ) + + self.addCleanup(cleanup_expected_values_dir) def load_fault_injection_log_parser(self): return load_fault_injection_log_parser_module() @@ -540,6 +551,33 @@ class DBCrashTestTest(unittest.TestCase): ) self.assertEqual(30, db_crashtest.liveness_timeout({"duration": 30})) + def test_liveness_remote_db_uses_longer_timeouts(self): + db_crashtest = self.load_db_crashtest() + db_crashtest.is_remote_db = True + + params = db_crashtest.gen_cmd_params(self.build_mode_args()) + + self.assertEqual( + db_crashtest.DEFAULT_LIVENESS_TIMEOUT_SEC + * db_crashtest.REMOTE_DB_LIVENESS_TIMEOUT_MULTIPLIER, + params["duration"], + ) + self.assertEqual( + db_crashtest.DEFAULT_LIVENESS_NO_PROGRESS_TIMEOUT_SEC + * db_crashtest.REMOTE_DB_LIVENESS_TIMEOUT_MULTIPLIER, + params["liveness_no_progress_timeout_sec"], + ) + self.assertEqual( + params["duration"], db_crashtest.liveness_timeout({"duration": 0}) + ) + + params = db_crashtest.gen_cmd_params( + self.build_mode_args(duration=17, liveness_no_progress_timeout_sec=7) + ) + + self.assertEqual(17, params["duration"]) + self.assertEqual(7, params["liveness_no_progress_timeout_sec"]) + def test_liveness_wrapper_timeout_is_successful_end_of_run(self): db_crashtest = self.load_db_crashtest() execute_calls = [] @@ -567,8 +605,174 @@ class DBCrashTestTest(unittest.TestCase): self.assertEqual(17, execute_calls[0][1]) self.assertFalse(execute_calls[0][2]) self.assertFalse(execute_calls[0][3]) + self.assertIn("--expected_values_dir=", execute_calls[0][0]) self.assertEqual(1, len(cleanups)) + def test_remote_liveness_uses_local_expected_values_dir(self): + db_crashtest = self.load_db_crashtest() + os.environ[_TEST_DIR_ENV_VAR] = "/dev_test/rocksdb_crash_test/job123" + db_crashtest.is_remote_db = True + self.register_expected_values_dir_cleanup(db_crashtest) + execute_calls = [] + + def fake_execute_cmd( + cmd, timeout=None, timeout_pstack=False, expected_to_timeout=True + ): + execute_calls.append(cmd) + return ( + True, + -15, + "Received signal 15 (Terminated)\n", + "", + 123, + ) + + def expected_values_dir_from(cmd): + values = [ + arg.split("=", 1)[1] + for arg in cmd + if arg.startswith("--expected_values_dir=") + ] + self.assertEqual(1, len(values)) + return values[0] + + db_crashtest.execute_cmd = fake_execute_cmd + db_crashtest.print_fault_injection_log = lambda pid: None + db_crashtest.cleanup_after_success = lambda db_arg, num_dbs=1: None + + db_crashtest.liveness_main(self.build_mode_args(duration=17), []) + + self.assertEqual(1, len(execute_calls)) + expected_values_dir = expected_values_dir_from(execute_calls[0]) + self.assertTrue(expected_values_dir) + self.assertEqual( + os.path.realpath(tempfile.gettempdir()), + os.path.dirname(os.path.realpath(expected_values_dir)), + ) + self.assertTrue(os.path.isdir(expected_values_dir)) + + explicit_dir = os.path.join(self.test_tmpdir, "explicit_expected") + db_crashtest.get_ev_parent_dir = lambda: self.fail( + "explicit expected_values_dir should not allocate a default" + ) + db_crashtest.liveness_main( + self.build_mode_args(duration=17, expected_values_dir=explicit_dir), [] + ) + + self.assertEqual(2, len(execute_calls)) + self.assertEqual(explicit_dir, expected_values_dir_from(execute_calls[1])) + self.assertTrue(os.path.isdir(explicit_dir)) + + def test_remote_uri_forms_are_normalized_for_workload_and_cleanup(self): + os.environ[_TEST_DIR_ENV_VAR] = "/dev_test/rocksdb_crash_test/job123" + uri = "remote://example/db" + + for flag in ("--env_uri", "--fs_uri"): + for form in ("equals", "spaced"): + with self.subTest(flag=flag, form=form): + remote_args = ( + [flag + "=" + uri] if form == "equals" else [flag, uri] + ) + db_crashtest = self.load_db_crashtest( + [ + "liveness", + "--duration=17", + "--stress_cmd=/bin/db_stress", + ] + + remote_args + ) + self.register_expected_values_dir_cleanup(db_crashtest) + workload_calls = [] + cleanup_calls = [] + expected_values_dirs = [] + + def fake_execute_cmd( + cmd, + timeout=None, + timeout_pstack=False, + expected_to_timeout=True, + ): + workload_calls.append(cmd) + expected_values_dirs.extend( + arg.split("=", 1)[1] + for arg in cmd + if arg.startswith("--expected_values_dir=") + ) + return ( + True, + -15, + "Received signal 15 (Terminated)\n", + "", + 123, + ) + + def fake_cleanup_call(cmd, env=None): + cleanup_calls.append(cmd) + return 0 + + db_crashtest.execute_cmd = fake_execute_cmd + db_crashtest.print_fault_injection_log = lambda pid: None + with mock.patch.object( + db_crashtest.subprocess, + "call", + side_effect=fake_cleanup_call, + ): + db_crashtest.main() + + normalized_arg = flag + "=" + uri + self.assertTrue(db_crashtest.is_remote_db) + self.assertEqual(1, db_crashtest.remain_args.count(normalized_arg)) + self.assertEqual(1, len(workload_calls)) + self.assertEqual(1, workload_calls[0].count(normalized_arg)) + self.assertNotIn(flag, workload_calls[0]) + self.assertEqual(1, len(cleanup_calls)) + self.assertEqual(1, cleanup_calls[0].count(normalized_arg)) + self.assertNotIn(flag, cleanup_calls[0]) + self.assertEqual(1, len(expected_values_dirs)) + self.assertTrue(expected_values_dirs[0]) + self.assertEqual( + os.path.realpath(tempfile.gettempdir()), + os.path.dirname(os.path.realpath(expected_values_dirs[0])), + ) + + def test_remote_uri_missing_value_does_not_consume_following_option(self): + db_crashtest = self.load_db_crashtest() + + for flag in ("--env_uri", "--fs_uri"): + for following_option in ("-x", "--duration=17"): + with self.subTest(flag=flag, following_option=following_option): + args = [flag, following_option] + self.assertEqual( + args, + db_crashtest.normalize_remote_db_args(args), + ) + self.assertFalse(db_crashtest.remote_db_enabled(args)) + + def test_remote_uri_last_value_wins_and_cleanup_preserves_order(self): + cases = ( + (["--env_uri=first", "--env_uri="], False), + (["--env_uri=", "--env_uri=last"], True), + (["--env_uri=first", "--fs_uri="], True), + (["--env_uri=", "--fs_uri="], False), + ) + + for remote_args, expected_remote in cases: + with self.subTest(remote_args=remote_args): + db_crashtest = self.load_db_crashtest( + ["liveness"] + remote_args + ) + db_crashtest.stress_cmd = "/bin/db_stress" + + self.assertEqual(expected_remote, db_crashtest.is_remote_db) + self.assertEqual(remote_args, db_crashtest.remain_args[1:]) + with mock.patch.object( + db_crashtest.subprocess, + "call", + return_value=0, + ) as cleanup_call: + db_crashtest.cleanup_after_success("/remote/db") + self.assertEqual(remote_args, cleanup_call.call_args.args[0][4:]) + def test_liveness_can_target_transaction_lock_manager(self): db_crashtest = self.load_db_crashtest() diff --git a/utilities/checkpoint/checkpoint_impl.cc b/utilities/checkpoint/checkpoint_impl.cc index 703b6a3679..2df84fc769 100644 --- a/utilities/checkpoint/checkpoint_impl.cc +++ b/utilities/checkpoint/checkpoint_impl.cc @@ -163,8 +163,7 @@ class SerialFileMover : public CheckpointFileMover { // Enqueues link/copy work on the CopyEngine pool; Finish() awaits it and does // the link->copy fallback for FileSystems that cannot hard-link (which costs an -// extra round-trip per file; the intended warm-storage targets support -// linking). +// extra round-trip per file; the intended remote file systems support linking). class ParallelFileMover : public CheckpointFileMover { public: ParallelFileMover(CopyEngine* engine, Env* env, bool use_link, bool use_fsync,