CPU corruption injector: randomize non-preset db_stress flags (#14867)

Summary:

Adds an opt-in --randomize_stress_flags mode to the CPU corruption injection runner.

The fixed DB_STRESS_PRESET keeps an injected corruption isolated and observable, but every run then exercises the same stress test flags (default values for non DB_STRESS_PRESET flags). To widen what a runner covers, we want every other db_stress knob to vary run to run.

Therefore this PR makes each run's db_stress flags are a fresh db_crashtest.py roll (its full blackbox parameter space) with DB_STRESS_PRESET pinned on top. A small CLOSURE gate set stops db_crashtest's finalize_and_sanitize() from un-pinning preset flags, and an assertion fails the run fast if any preset flag is changed. This is considered as a work-around until we can refactor the flag generation of db_crashtest.py into a isolated function to take a set of flags to honor.  This PR also adds some randomization into `max_keys` and local vs remote compaction as they are subjected to some special constraint unique to the cpu corruption injection framework. 

The per-run roll is seeded by the run's seed, so randomized runs' flags are reproducible with --seed (though the actual db stress test behavior may still have some difference under the same flags).

**Test plan:**
Functional test
- build_runs(..., randomize=True) produces a full randomized flag set (~286 flags vs ~49 for the fixed preset) and the preset survives the db_crashtest roll (the _assert_preset_survived check never fires).
- randomized runs are reproducible: same base_seed -> same flags; run i seeded base_seed+i.

Integration test 
```
for op in write flush compaction; do
    python3 tools/cpu_corruption_injector/runner.py --op $op --runs 30 --randomize_stress_flags --parallel 8 --stress_cmd "$BIN" --report_dir /tmp/icc-rand/$op
done
```
- every run still carries the pinned DB_STRESS_PRESET values (preset-survived assertion never fires across the campaign).
- the workload genuinely varies run to run (e.g. max_keys, different compaction_style / memtablerep / *_one_in values across runs), confirming randomization is active.
- SDC still appears across write / flush / compaction -- randomization widens coverage without losing the SDC signal; NO_INJECTION and ERROR stay ~0.

**Runs' Outcomes (`summary.json`):**

| SDC | CORRUPTION | CRASH | NO_EFFECT | NO_INJECTION | ERROR |
| --- | --- | --- | --- | --- | --- |
| 2 | 7 | 7 | 43 | 1 | 0 |

Local vs remote compaction. With `--randomize_stress_flags` a compaction run may flip to the remote compaction service (`remote_compaction_worker_threads=1`), so the injector breaks on `CompactionServiceCompactionJob::Run` instead of the local `CompactionJob::Run`. Splitting the runs by injection entry_fn:

| where | runs | SDC | CORRUPTION | CRASH | NO_EFFECT | NO_INJECTION |
| --- | --- | --- | --- | --- | --- | --- |
| local | 25 | 1 | 1 | 3 | 20 | 0 |
| remote | 35 | 1 | 6 | 4 | 23 | 1 |

Reviewed By: pdillinger

Differential Revision: D108551605

tools/cpu_corruption_injector/runner.py

@@ -23,7 +23,12 @@ if _HERE not in sys.path:
     sys.path.append(_HERE)
 
 from runner_execute import _get_gdb, execute  # noqa: E402
-from runner_build import build_runs, INJECTION_SITES, Run  # noqa: E402
+from runner_build import (  # noqa: E402
+    build_runs,
+    INJECTION_SITES,
+    REMOTE_COMPACTION_ENTRY_FN,
+    Run,
+)
 from runner_report import report  # noqa: E402
 
 logger: logging.Logger = logging.getLogger("runner")
@@ -62,7 +67,9 @@ def main(argv: list[str] | None = None) -> int:
     # replays with --seed=base_seed, and a single run i with --seed=<base_seed+i> --runs 1.
     logger.info("reproduce with --seed=%d (run i used seed %d+i)", base_seed, base_seed)
     injector_py = os.path.join(_HERE, "injector.py")
-    runs = build_runs(args.op, args.runs, report_dir, base_seed)
+    runs = build_runs(
+        args.op, args.runs, report_dir, base_seed, args.randomize_stress_flags
+    )
     execute_all(runs, injector_py, stress_cmd, parallel_runs)
     report(runs, args.op, report_dir, base_seed)
     return 0
@@ -107,6 +114,13 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         "invocation (logged as base_seed=N); pass that N back via --seed to reproduce "
         "the same runs. Run i uses seed N+i.",
     )
+    parser.add_argument(
+        "--randomize_stress_flags",
+        action="store_true",
+        help="randomize every db_stress flag the runner does not pin, varying it run to "
+        "run to widen coverage; the pinned flags (what makes an injected CPU corruption "
+        "reachable and observable) stay fixed. Off by default.",
+    )
     return parser.parse_args(argv)
 
 
@@ -129,7 +143,13 @@ def verify_injection_site(stress_cmd: str, op: str) -> None:
     # source file:line. Checking entry_fn's source line alone catches a gdb too old to
     # read this build's debug info (which makes the telemetry's source come out "@ ?:0").
     site = INJECTION_SITES[op]
-    functions = list(dict.fromkeys([site.entry_fn, *site.target_fns]))
+    functions = [site.entry_fn, *site.target_fns]
+    # A randomize-mode compaction run may inject into remote compaction, breaking on the
+    # remote job instead of site.entry_fn; verify that symbol resolves too (harmless to
+    # check even for non-randomized runs).
+    if op == "compaction":
+        functions.append(REMOTE_COMPACTION_ENTRY_FN)
+    functions = list(dict.fromkeys(functions))
     gdb_output = _run_gdb_check(stress_cmd, functions, site.entry_fn)
 
     missing = [fn for fn in functions if _function_not_found(gdb_output, fn)]

tools/cpu_corruption_injector/runner_build.py

@@ -12,6 +12,12 @@ import random
 from dataclasses import dataclass, field
 from typing import Optional
 
+from runner_randomize_stress_flags import (
+    CLOSURE,
+    randomize_stress_flags,
+    REMOTE_COMPACTION_CLOSURE,
+)
+
 
 @dataclass(frozen=True)
 class InjectionSite:
@@ -44,6 +50,16 @@ INJECTION_SITES: dict[str, InjectionSite] = {
     ),
 }
 
+# When a compaction run randomly injects into REMOTE compaction, the work runs on the
+# compaction service's worker thread, so the injector must break here instead of the local
+# CompactionJob::Run entry_fn (see build_runs).
+REMOTE_COMPACTION_ENTRY_FN = "rocksdb::CompactionServiceCompactionJob::Run"
+
+# Small SDC-safe domain for the randomized max_key. db_crashtest's own range is far too
+# large -- a single injected flip in a huge keyspace rarely lands on a multi-version key,
+# washing out the SDC signal. Every choice here still produces a multi-version layout.
+MAX_KEY_CHOICES: list[int] = [10, 100, 1000, 10000]
+
 
 # The complete db_stress config every run uses (the groups below merged). Together they
 # keep one injected CPU corruption isolated and observable, so they are load-bearing for
@@ -84,11 +100,6 @@ _SERIALIZED: dict[str, object] = {
     "num_dbs": 1,
     "disable_auto_compactions": 1,  # no surprise background compaction
     "write_buffer_size": 1 << 30,  # memtable never auto-flushes
-    # Compaction runs locally, not on the simulated remote compaction service's worker
-    # thread (db_stress default remote_compaction_worker_threads=2). Remote runs the work
-    # in CompactionServiceCompactionJob::Run, where the injector's CompactionJob::Run
-    # entry_fn is off-stack, so the op is never reached. Injecting into remote compaction
-    # is a planned follow-up (a second entry_fn, pinned to one worker).
     "remote_compaction_worker_threads": 0,
     "subcompactions": 1,  # one compaction thread, no parallel subcompactions
 }
@@ -180,6 +191,7 @@ def build_runs(
     run_count: int,
     report_dir: str,
     base_seed: int,
+    randomize: bool = False,
 ) -> list[Run]:
     site = INJECTION_SITES[op]
     # op_index picks which op instance the injector targets. Warmup single-steps that
@@ -197,16 +209,32 @@ def build_runs(
         run_dir = os.path.join(report_dir, f"run_{index:05d}")
         os.makedirs(run_dir, exist_ok=True)
         per_run_stress_flags = _per_run_stress_flags(run_dir, seed)
+        # Randomize per run for coverage -- e.g. max_key (below) exercises a different
+        # key space, others vary LSM shape and version -- and a compaction run may flip
+        # to remote compaction, a distinct code path from local with its own potential
+        # CPU-corruption vulnerabilities.
+        entry_fn = site.entry_fn
+        if randomize:
+            preset = dict(DB_STRESS_PRESET)
+            closure = CLOSURE
+            preset["max_key"] = rng.choice(MAX_KEY_CHOICES)
+            if op == "compaction" and rng.choice([0, 1]):
+                preset["remote_compaction_worker_threads"] = 1
+                closure = {**CLOSURE, **REMOTE_COMPACTION_CLOSURE}
+                entry_fn = REMOTE_COMPACTION_ENTRY_FN
+            stress_flags = randomize_stress_flags(per_run_stress_flags, preset, closure)
+        else:
+            stress_flags = _fixed_stress_flags(per_run_stress_flags)
         runs.append(
             Run(
                 index=index,
                 op=op,
                 op_index=op_index,
-                entry_fn=site.entry_fn,
+                entry_fn=entry_fn,
                 target_fn=target_fn,
                 seed=seed,
                 dir=run_dir,
-                stress_flags=_fixed_stress_flags(per_run_stress_flags),
+                stress_flags=stress_flags,
             )
         )
     return runs

tools/cpu_corruption_injector/runner_randomize_stress_flags.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+#  Copyright (c) Meta Platforms, Inc. and affiliates.
+#  This source code is licensed under both the GPLv2 (found in the COPYING file in the root directory)
+#  and the Apache 2.0 License (found in the LICENSE.Apache file in the root directory).
+
+# pyre-unsafe -- interops with db_crashtest (an untyped sibling script: gen_cmd,
+# default_params, blackbox_default_params), whose calls/attributes pyre-strict would
+# reject as untyped.
+
+from __future__ import annotations
+
+import os
+import random
+import sys
+
+
+# db_crashtest attributes/functions this file couples to. Assert they still exist so an
+# upstream rename or refactor fails loudly HERE, naming the coupling, instead of as a bare
+# AttributeError deep inside randomize_stress_flags.
+_REQUIRED_DB_CRASHTEST_API = ("default_params", "blackbox_default_params", "gen_cmd")
+
+
+def _db_crashtest():
+    # Lazy import: db_crashtest lives in the parent tools/ dir and has import-time side
+    # effects (prints a seed, resolves some randoms), so load it only when randomizing.
+    tools = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    if tools not in sys.path:
+        sys.path.append(tools)  # append so it can't shadow stdlib
+    import db_crashtest  # noqa: PLC0415
+
+    missing = [
+        name for name in _REQUIRED_DB_CRASHTEST_API if not hasattr(db_crashtest, name)
+    ]
+    if missing:
+        raise AssertionError(
+            f"db_crashtest is missing {missing}: its API changed. This tool pins a "
+            "db_stress preset by feeding db_crashtest.gen_cmd; update "
+            "runner_randomize_stress_flags.py to match."
+        )
+    return db_crashtest
+
+
+# Each entry pins a db_crashtest flag whose randomized value would otherwise make
+# finalize_and_sanitize() un-pin one of the preset flags; see
+# db_crashtest.finalize_and_sanitize for the exact couplings. _assert_preset_survived()
+# fails the run fast if this list is ever incomplete.
+CLOSURE: dict[str, object] = {
+    "memtablerep": "skip_list",
+    "enable_compaction_filter": 0,
+    "use_multiscan": 0,
+    "test_batches_snapshots": 0,
+    "inplace_update_support": 0,
+    "user_timestamp_size": 0,
+}
+
+REMOTE_COMPACTION_CLOSURE: dict[str, object] = {
+    "enable_blob_files": 0,
+    "enable_blob_garbage_collection": 0,
+    "allow_setting_blob_options_dynamically": 0,
+    "remote_compaction_failure_fall_back_to_local": "false",
+}
+
+
+def randomize_stress_flags(
+    per_run_stress_flags: dict[str, object],
+    preset: dict[str, object],
+    closure: dict[str, object] = CLOSURE,
+) -> list[str]:
+    # TODO: ideally db_crashtest.py would expose a flag generator that takes a pinned
+    # preset, so we could drop the closure and the trial-and-error preset check below.
+    db_crashtest = _db_crashtest()
+    random.seed(int(per_run_stress_flags["seed"]))
+
+    flags: dict[str, object] = dict(db_crashtest.default_params)
+    flags.update(db_crashtest.blackbox_default_params)
+    flags.update(preset)
+    flags.update(closure)
+    flags.update(per_run_stress_flags)
+
+    cmd, finalized = db_crashtest.gen_cmd(flags, [])
+    _assert_preset_survived(finalized, preset)
+    return cmd[1:]  # cmd[0] is db_crashtest's own db_stress path; keep only the flags
+
+
+def _assert_preset_survived(
+    finalized: dict[str, object], preset: dict[str, object]
+) -> None:
+    for flag, value in preset.items():
+        got = finalized.get(flag)
+        if str(got) != str(value):
+            raise AssertionError(
+                f"preset flag {flag} changed {value!r} -> {got!r} under randomization; "
+                "pin its gate in the closure."
+            )