CPU corruption injector: gdb register flip into one db_stress op (#14858)

Summary:

This PR is the injection layer of the CPU corruption injector, runs inside gdb and randomly corrupts a register by bit flip in exactly one db_stress op (i.e, write, foreground compaction and flush) per stress test run. Detection layer is at db_stress (https://github.com/facebook/rocksdb/pull/14852); orchestration layer is coming up.

__How one run works__ 
- The orchestration layer, coming up, randomly picks which stress test `op` instance (so corruption can land at different points in the LSM shape journey) and which `target_fn` of that `op` (so to cap instructions to step under a reasonable limit; `injector.py` in this PR randomly picks which instruction within the `target_fn` to inject (so corruption can land at different points of a `target_fn`).
- Attach: gdb starts with injector.py's parameters passed via -iex and the db_stress command after --args, so db_stress runs unmodified. Example:
```
gdb --batch --nx \
  -iex "py import sys; sys.argv=['injector.py','--op','write','--op_index','42','--entry_fn','rocksdb::MemTable::Add','--target_fn','rocksdb::MemTable::Add','--corruptions_per_op','1','--seed','7','--dir','<rundir>']" \
  -x tools/cpu_corruption_injector/injector.py \
  --args <db_stress> --threads=1 --verify_cpu_corruption_dir=<rundir>  ...
```
- Navigate: The orchestration layer will pick op_index. `entry_fn` is called exactly once per stress test run's op so the op_index-th op is its op_index-th call. `injector_navigate.py` breaks on `entry_fn` and set a gdb ignore-count of op_index-1 to fast-forward to op_index-th one. It also breaks at the first `target_fn` within that `entry_fn`. 

- Warm up: `injector_critical_instruction.py` will choose "critical instruction" (those that move key/value bytes with general-purpose or vector registers or set a branch flag) uniformly within the chosen `target_fn` by the orchestration layer. In order to do that, it needs to approximate how many such instructions within the `target_fn`. Hence we have this warm-up phase. It single-steps the instruction within the first encoutering of `target_fn` to count and draw the critical instruction index, then corrupt that index at a later call. 

- Corrupt: on a later call of `target_fn`, `injector_critical_instruction.py` single-step to the m-th critical instruction and bit-flip the register through `injector_register_corruption.py`. The way to corrupt register depends on what instruction it is. If the current call of `target_fn`'s m-th instruction is not a critical instruction, we will try next `target_fn` till running out of `target_fn`.

- Record: `injector_telemetry.py` provides telemetry to capture the corruption for later analysis.


**Test plan:**
1. Isolated tests (real gdb-captured x/i fixtures): test_inject_critical_instruction

2. E2E test on navigation, inject, telemetry will be done in the later orchestration PR. Below is inject.json from such run
```
{
  "injection_result": "injected",
  "db_stress_crash_signal": null,
  "op": "write",
  "op_index": 279,
  "entry_fn": "rocksdb::MemTable::Add",
  "target_fn": "rocksdb::MemTable::Add",
  "critical_instruction_index": 37,
  "corruptions": [
    {
      "instruction": "mov    %rsi,0x8c8(%rbx)",
      "register": "rsi",
      "corruption_type": "bit_flip",
      "before": "0x7fffee4c64d8",
      "after": "0x7fffee4c64c8",
      "details": {
        "source": "rocksdb::Arena::AllocateAligned @ ./fbcode/internal_repo_rocksdb/repo/memory/arena.cc:135",
        "call_chain": [
          "rocksdb::Arena::AllocateAligned @ ./fbcode/internal_repo_rocksdb/repo/memory/arena.cc:135",
          "rocksdb::ConcurrentArena::AllocateAligned(unsigned long, unsigned long, rocksdb::Logger*)::{lambda()#1}::operator()() const @ fbcode/internal_repo_rocksdb/repo/memory/concurrent_arena.h:65",
          "rocksdb::ConcurrentArena::AllocateImpl<rocksdb::ConcurrentArena::AllocateAligned(unsigned long, unsigned long, rocksdb::Logger*)::{lambda()#1}>(unsigned long, bool, rocksdb::ConcurrentArena::AllocateAligned(unsigned long, unsigned long, rocksdb::Logger*)::{lambda()#1} const&) @ fbcode/internal_repo_rocksdb/repo/memory/concurrent_arena.h:145",
          "rocksdb::ConcurrentArena::AllocateAligned @ fbcode/internal_repo_rocksdb/repo/memory/concurrent_arena.h:63",
          "rocksdb::InlineSkipList<rocksdb::MemTableRep::KeyComparator const&>::AllocateNode @ fbcode/internal_repo_rocksdb/repo/memtable/inlineskiplist.h:868",
          "rocksdb::InlineSkipList<rocksdb::MemTableRep::KeyComparator const&>::AllocateKey @ fbcode/internal_repo_rocksdb/repo/memtable/inlineskiplist.h:855",
          "rocksdb::(anonymous namespace)::SkipListRep::Allocate @ ./fbcode/internal_repo_rocksdb/repo/memtable/skiplistrep.cc:36",
          "rocksdb::MemTable::Add @ ./fbcode/internal_repo_rocksdb/repo/db/memtable.cc:1157"
        ]
      }
    }
  ],
  "ops_seen": 279,
  "critical_instructions_seen": 38
}
```

Reviewed By: pdillinger

Differential Revision: D107999835

tools/cpu_corruption_injector/injector.py

@@ -0,0 +1,190 @@
+#!/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
+# Imports `gdb` (only inside gdb's embedded Python, not via pip/buck), so this file
+# and its sibling injector_* modules are not buck/pyre targets.
+
+from __future__ import annotations
+
+import argparse
+import os
+import random
+import sys
+import traceback
+from dataclasses import dataclass, field
+from typing import Optional
+
+import gdb  # provided by gdb's embedded Python (not pip-importable)
+
+# gdb runs this file as a plain script (not a package), so its directory is not on
+# sys.path; append it (at the END, so a sibling can never shadow a stdlib module) so
+# the injector_* imports below resolve.
+_HERE = os.path.dirname(os.path.abspath(__file__))
+if _HERE not in sys.path:
+    sys.path.append(_HERE)
+
+# These imports must follow the sys.path bootstrap above, so flake8's E402
+# ("module level import not at top of file") is expected here -- hence the noqa.
+import injector_navigate as nav  # noqa: E402
+from injector_register_corruption import CorruptedInstruction  # noqa: E402
+from injector_telemetry import write_inject_record_json  # noqa: E402
+
+
+@dataclass
+class InjectRecord:
+    # outcome of this run's injection attempt:
+    #   injected | op_not_reached | target_fn_not_reached | no_critical_instruction
+    #   | error (a bug in the injector itself, distinct from a db_stress crash)
+    injection_result: str = "op_not_reached"
+    # fatal signal db_stress took (e.g. SIGSEGV), else None; set only when
+    # injection_result == "injected" (an injection-caused db_stress crash).
+    db_stress_crash_signal: Optional[str] = None
+    op: str = ""
+    op_index: int = 0
+    entry_fn: str = ""
+    target_fn: str = ""
+    critical_instruction_index: int = 0
+    corruptions: list[CorruptedInstruction] = field(default_factory=list)
+    ops_seen: int = 0
+    critical_instructions_seen: int = 0
+
+
+@dataclass
+class Args:
+    op: str
+    op_index: int
+    entry_fn: str
+    target_fn: str
+    dir: str
+    seed: int
+
+
+def parse_args() -> Args:
+    parser = argparse.ArgumentParser(
+        description="db_stress gdb cpu-corruption injector"
+    )
+    parser.add_argument("--op", required=True, choices=["write", "flush", "compaction"])
+    parser.add_argument("--op_index", type=int, required=True)
+    parser.add_argument(
+        "--entry_fn",
+        required=True,
+        help="entry function whose op_index-th call is the op instance to corrupt",
+    )
+    parser.add_argument(
+        "--target_fn",
+        required=True,
+        # entry_fn is too large/expensive to single-step, so we step a smaller inner
+        # function and corrupt a register there.
+        help="inner function we single-step and corrupt a register in",
+    )
+    parser.add_argument("--dir", required=True, help="output directory for inject.json")
+    parser.add_argument("--seed", type=int, default=0)
+    # parse_known_args ignores anything gdb may have left in argv.
+    parsed, _ = parser.parse_known_args(sys.argv[1:])
+    return Args(
+        op=parsed.op,
+        op_index=max(parsed.op_index, 1),
+        entry_fn=parsed.entry_fn,
+        target_fn=parsed.target_fn,
+        dir=parsed.dir,
+        seed=parsed.seed,
+    )
+
+
+def run(record: InjectRecord, args: Args) -> None:
+    record.op = args.op
+    record.op_index = args.op_index
+    record.entry_fn = args.entry_fn
+    record.target_fn = args.target_fn
+
+    out_path = os.path.join(args.dir, "inject.json")
+    # Baseline record so inject.json always exists even if gdb dies mid-run; every
+    # later write_inject_record_json truncates and rewrites the file, so the final write wins.
+    write_inject_record_json(out_path, record)
+
+    rng = random.Random(args.seed)
+    nav.setup_gdb()
+    site = nav.Navigator(args.entry_fn, args.target_fn)
+
+    # Start db_stress, stopped at its very first instruction. reach_op then sets a
+    # breakpoint on the op's entry_fn and runs to its op_index-th call -- the op
+    # instance we corrupt.
+    gdb.execute("starti")
+    if not site.reach_op(args.op_index):
+        record.injection_result = "op_not_reached"
+        _finish_run_and_write(record, out_path, site)
+        return
+    if not site.reach_target_fn_call():
+        record.injection_result = "target_fn_not_reached"
+        _finish_run_and_write(record, out_path, site)
+        return
+
+    critical_instruction_index = _pick_critical_instruction_to_corrupt(site, rng)
+    if critical_instruction_index is None:
+        record.injection_result = "no_critical_instruction"
+        _finish_run_and_write(record, out_path, site)
+        return
+    record.critical_instruction_index = critical_instruction_index
+
+    record.injection_result = (
+        "no_critical_instruction"  # until a corruption lands below
+    )
+    # Corrupt the chosen critical instruction in this op type's next target_fn call; if
+    # that call has no critical instruction at that index, try the next call.
+    while site.reach_target_fn_call():
+        corruption, record.critical_instructions_seen = (
+            site.corrupt_critical_instruction(critical_instruction_index, rng)
+        )
+        if corruption is not None:
+            record.corruptions = [corruption]
+            record.injection_result = "injected"
+            write_inject_record_json(out_path, record)  # checkpoint: injection happened
+            break
+    _finish_run_and_write(record, out_path, site)
+
+
+def _pick_critical_instruction_to_corrupt(
+    site: nav.Navigator, rng: random.Random
+) -> Optional[int]:
+    """Single-step target_fn calls until one has critical instructions, then return
+    a random index among them; None if none before the op ended. (Corrupting happens
+    on a later call -- a call cannot be single-stepped twice.)"""
+    while True:
+        critical_instruction_count = site.count_critical_instructions()
+        if critical_instruction_count > 0:
+            return rng.randrange(critical_instruction_count)
+        if not site.reach_target_fn_call():
+            return None
+
+
+def _finish_run_and_write(
+    record: InjectRecord, out_path: str, site: nav.Navigator
+) -> None:
+    # The navigator owns db_stress's end-of-run lifecycle (kill-if-crashed vs
+    # run-to-exit); we only record what it reports and write the file.
+    record.ops_seen = site.ops_seen()
+    record.db_stress_crash_signal = nav.finish_db_stress()
+    write_inject_record_json(out_path, record)
+
+
+def main() -> None:
+    args = parse_args()
+    record = InjectRecord()
+    out_path = os.path.join(args.dir, "inject.json")
+    try:
+        run(record, args)
+    # Blind-except (flake8 BLE001) on purpose: a bug in the injector must never abort
+    # gdb and leave db_stress wedged. Log it, mark the run errored, and still end
+    # db_stress + write inject.json.
+    except Exception as exc:  # noqa: BLE001
+        gdb.write(f"injector.py: fatal: {exc}\n{traceback.format_exc()}\n", gdb.STDERR)
+        record.injection_result = "error"
+        # The navigator owns ending db_stress (kill if crashed, else run to exit).
+        record.db_stress_crash_signal = nav.finish_db_stress()
+        write_inject_record_json(out_path, record)
+
+
+main()

tools/cpu_corruption_injector/injector_critical_instruction.py

@@ -0,0 +1,294 @@
+#!/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
+# Runs ONLY inside the gdb python runtime (imports `gdb`); not a buck/pyre target.
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Optional
+
+import gdb  # provided by gdb's embedded Python (not pip-importable)
+
+# This decoder reads ONE disassembled x86-64 instruction and decides whether to
+# corrupt it, which register to corrupt, and how. A corruption models a CPU fault: we
+# mangle the value INSIDE a CPU register as the instruction handles it -- we do NOT
+# corrupt bytes in memory.
+#
+# classify_current_instruction() forces AT&T disassembly and fails fast on any
+# non-x86-64 target, so these assumptions always hold. AT&T writes the source operand
+# first and the destination last -- "mov src, dst" (the reverse of Intel syntax).
+#
+# We corrupt exactly two things; everything else (compute, lea, reg-reg copies,
+# immediates, stack spills) is left alone:
+#   - a cmp/test    -> flip a bit of the eflags it produces (a mis-evaluated branch)
+#   - a MEMORY move -> corrupt the register holding the value being moved
+
+# re.compile precompiles a pattern we match once per stepped instruction.
+# Any move form -- general-purpose (mov, movzx, movsxd, ...) or vector (vmovdqu,
+# movaps, movq, ...). They all start with "mov" or "vmov", so one prefix matches all
+# (no trailing \b on purpose -- we WANT every mov* form). vmovdqu is how glibc memcpy
+# copies a value, the bulk case we most want to catch.
+_MOV_RE = re.compile(r"^v?mov")
+# cmp / test set the eflags a following branch reads. [bwlq]? allows the AT&T size
+# suffixes (cmpq, testb, ...); the trailing \b stops it matching cmpxchg (an atomic
+# compare-and-exchange) or cmpsd (an SSE compare) -- different instructions whose
+# flags/semantics are not the simple branch decision we model.
+_CMP_TEST_RE = re.compile(r"^(cmp|test)[bwlq]?\b")
+
+# A register operand token, e.g. %rax, %eax, %r10d, %xmm3, %ymm12, %zmm5. The capture
+# group is the bare name (no '%'). This matches by SHAPE only -- it cannot tell a data
+# register from a control one -- so _NON_DATA_REGS below removes the control registers.
+_REG_TOK_RE = re.compile(r"%([a-z][a-z0-9]+|xmm\d+|ymm\d+|zmm\d+|r\d+[bwd]?)")
+
+# Registers we must NOT corrupt: the instruction pointer, stack/frame pointers, and
+# segment registers. They hold control state, not key/value data, so flipping them is
+# a control-flow crash rather than a value fault (and gdb types them as pointers,
+# which rejects the XOR we use). _REG_TOK_RE above already allowlists register-shaped
+# tokens; this short, fixed set just removes the few control registers that share that
+# shape. We do NOT positively allowlist data registers by regex even though the vector
+# set (xmm/ymm/zmm \d+) is clean, because the general-purpose set is NOT: data and
+# control GP registers overlap in shape -- rax is data, rsp is control, both "r..",
+# each in 4 widths -- so a positive pattern would be large and would STILL have to
+# special-case rsp/rbp/rip. Excluding the small control set is simpler and safer.
+_NON_DATA_REGS = frozenset(
+    {
+        "rip",
+        "eip",
+        "ip",
+        "rsp",
+        "esp",
+        "sp",
+        "spl",
+        "rbp",
+        "ebp",
+        "bp",
+        "bpl",
+        "fs",
+        "gs",
+        "cs",
+        "ds",
+        "es",
+        "ss",
+    }
+)
+
+
+@dataclass
+class CriticalInstruction:
+    """An x86-64 instruction we CHOOSE to corrupt to simulate a CPU fault that can
+    flow into stored data (we do not fault the CPU; we mimic one by flipping a
+    register's bits at the right moment).
+
+    reg                -- the register we corrupt, e.g. "rbx", "xmm3", or "eflags".
+    corruption_type    -- how we corrupt it (this is the corruption_type in
+                          inject.json):
+                            bit_flip  -- flip a bit of a general-purpose register
+                            flag_flip -- flip a bit of the eflags register (a cmp/test)
+                            lane_bit_flip -- flip a 64-bit lane of a vector register
+    corrupt_after_exec -- True:  corrupt AFTER the instruction runs, because it
+                          PRODUCES the value we corrupt (a load destination, or
+                          cmp/test flags). False: corrupt BEFORE it runs, because it
+                          CONSUMES the register we corrupt (a store to memory).
+    text               -- the disassembled instruction, e.g. "mov 0x8(%r8),%rbx".
+    """
+
+    reg: str
+    corruption_type: str
+    corrupt_after_exec: bool
+    text: str = ""
+
+
+def simd_int64_lanes(reg: str) -> tuple[Optional[str], int]:
+    """How to address a vector register as 64-bit integer lanes, returned as
+    (lane_expr, lane_count), or (None, 0) if gdb cannot. We corrupt vector data one
+    64-bit lane at a time.
+
+    gdb exposes a vector register as a struct with several "member" views of the same
+    bits; we use the int64 ones: `$xmm0.v2_int64` views the 128-bit xmm0 as 2 int64s,
+    `$ymm0.v4_int64` views the 256-bit ymm0 as 4, `$zmm0.v8_int64` views the 512-bit
+    zmm0 as 8. lane_expr is that member string (e.g. "xmm0.v2_int64"); lane_count is
+    its number of lanes. We probe widest-first; parse_and_eval raises gdb.error if a
+    member does not exist, which is how we find the one this register has."""
+    for member, lane_count in (("v8_int64", 8), ("v4_int64", 4), ("v2_int64", 2)):
+        try:
+            gdb.parse_and_eval(f"${reg}.{member}")  # exists? else gdb.error
+            return f"{reg}.{member}", lane_count
+        except gdb.error:
+            continue
+    return None, 0
+
+
+def _gdb_can_corrupt_int64_lane(reg: str) -> bool:
+    """Whether gdb can address `reg` as 64-bit int lanes -- i.e. whether we can corrupt
+    it. A deliberate scope choice: we only corrupt vector registers gdb exposes as
+    int64 lanes; anything else is skipped. We need only the yes/no here, so we drop the
+    lane count. (SIMD coverage is surfaced by the campaign report's corruption_type
+    breakdown: if lane_bit_flip stays near zero, we are losing SIMD coverage here. TODO:
+    widen lane support beyond the int64 view.)"""
+    lane_expr, _ = simd_int64_lanes(reg)
+    return lane_expr is not None
+
+
+_arch_verified = False
+
+
+def classify_current_instruction() -> Optional[CriticalInstruction]:
+    """Decode the instruction at the current program counter. `x/i $pc` asks gdb to
+    disassemble one instruction ('i') at $pc; we parse that single text line. The
+    first call also verifies the target is x86-64 and forces AT&T disassembly (once),
+    so callers never have to know about that low-level requirement."""
+    global _arch_verified
+    if not _arch_verified:
+        _require_x86_att()
+        _arch_verified = True
+    return decode_instruction(gdb.execute("x/i $pc", to_string=True))
+
+
+def _require_x86_att() -> None:
+    """Fail fast unless the target is x86-64, and force AT&T disassembly so this
+    decoder's syntax assumptions hold no matter how the user's gdb is configured.
+    Runs once, lazily, from classify_current_instruction (a frame exists by then)."""
+    gdb.execute("set disassembly-flavor att")
+    arch = gdb.selected_frame().architecture().name()
+    assert "x86-64" in arch, (
+        f"cpu-corruption injector supports only x86-64 (AT&T); got architecture "
+        f"{arch!r}"
+    )
+
+
+def decode_instruction(
+    raw_xi_line: str, can_corrupt_int64_lane=_gdb_can_corrupt_int64_lane
+) -> Optional[CriticalInstruction]:
+    """Decode one `x/i` line into a CriticalInstruction, or None if it is not an
+    instruction we corrupt. `can_corrupt_int64_lane(reg)` answers whether gdb can
+    address a vector register's int64 lanes; it is a parameter so unit tests can feed
+    REAL `x/i` lines without a live gdb."""
+    text = _strip_to_instruction(raw_xi_line)
+    if not text:
+        return None
+    mnemonic, operands = _split_mnemonic_operands(text)
+    if _CMP_TEST_RE.match(mnemonic):
+        # cmp/test: corrupt the eflags it produces, after it runs.
+        return CriticalInstruction("eflags", "flag_flip", True, text)
+    if _MOV_RE.match(mnemonic):
+        return _critical_instruction_for_memory_move(
+            operands, text, can_corrupt_int64_lane
+        )
+    return None
+
+
+def _critical_instruction_for_memory_move(
+    operands: list[str], text: str, can_corrupt_int64_lane
+) -> Optional[CriticalInstruction]:
+    """The CriticalInstruction for a memory move we corrupt -- a mov that carries a
+    value between a MEMORY location and a DATA register. Returns None when either side
+    of the move disqualifies it:
+      memory side   -- no memory operand at all (a reg-reg copy or immediate load: no
+                       value moves through memory), or the memory operand is the stack
+                       (rbp/rsp: a spill/reload, i.e. compiler bookkeeping of a
+                       temporary, not key/value data);
+      register side -- the register operand is a control register (rsp/rbp/rip/segment),
+                       which _data_register rejects as non-value data."""
+    if not _has_memory_operand(operands):
+        return None
+    if "(%rbp" in text or "(%rsp" in text:  # rbp/rsp = frame/stack pointer
+        return None
+
+    # Load vs store by the source operand (AT&T writes the source first):
+    #   "mov MEM, REG"  source is memory -> a load:  corrupt the dest register AFTER
+    #   "mov REG, MEM"  source is a reg  -> a store: corrupt the source register BEFORE
+    if "(" in operands[0]:
+        reg, after_exec = _data_register(operands[-1]), True
+    else:
+        reg, after_exec = _data_register(operands[0]), False
+    if (
+        reg is None
+    ):  # the register side is a control register (caught by _data_register)
+        return None
+
+    # The register decides the corruption_type -- two cases by construction: a vector
+    # register -> lane_bit_flip (only if gdb can address its int64 lanes), else a
+    # general-purpose register -> bit_flip.
+    if "xmm" in reg or "ymm" in reg or "zmm" in reg:
+        return (
+            CriticalInstruction(reg, "lane_bit_flip", after_exec, text)
+            if can_corrupt_int64_lane(reg)
+            else None
+        )
+    return CriticalInstruction(reg, "bit_flip", after_exec, text)
+
+
+def _strip_to_instruction(raw_xi_line: str) -> str:
+    """Reduce a raw `x/i` line to just the instruction text. gdb returns e.g.
+    "=> 0x4f2e <sym+20>:\tmov 0x8(%r8),%rbx  # comment"; we strip it in three steps to
+    leave "mov 0x8(%r8),%rbx". (.strip() removes surrounding whitespace incl. tabs.)"""
+    line = raw_xi_line.strip()
+    if line.startswith("=>"):
+        line = line[2:].strip()  # drop the "=>" current-pc marker
+    colon = line.find(":")
+    if colon != -1:
+        line = line[colon + 1 :].strip()  # drop the "0x4f2e <sym+20>:" address prefix
+    hashpos = line.find("#")
+    if hashpos != -1:
+        line = line[:hashpos].strip()  # drop a trailing "# ..." comment
+    return line
+
+
+def _split_mnemonic_operands(text: str) -> tuple[str, list[str]]:
+    """Split "mov 0x8(%r8),%rbx" into ("mov", ["0x8(%r8)", "%rbx"]): the mnemonic
+    (first whitespace-delimited word) and its operand list. We need the operands to
+    find the register and to tell a load from a store. An instruction with no operands
+    (ret, nop, ...) yields []; that is not an error -- it simply is not corruptible."""
+    parts = text.split(None, 1)  # split on whitespace into [mnemonic, rest], at most 2
+    mnemonic = parts[0].lower()
+    operands = _split_operands(parts[1]) if len(parts) > 1 else []
+    return mnemonic, operands
+
+
+def _split_operands(operand_text: str) -> list[str]:
+    """Split the operand text on top-level commas. We cannot use str.split(",")
+    because a memory operand carries its own commas inside parentheses, e.g.
+    0x10(%r14,%rcx,8); there is no stdlib paren-aware splitter, so we scan characters
+    and split only at commas seen outside any "(...)"."""
+    operands: list[str] = []
+    depth = 0  # how many unclosed "(" we are currently inside
+    current = ""  # the operand accumulated so far
+    for ch in operand_text:
+        if ch == "(":
+            depth += 1
+        elif ch == ")":
+            depth -= 1
+        if ch == "," and depth == 0:  # a top-level comma ends the current operand
+            operands.append(current.strip())
+            current = ""
+        else:
+            current += ch
+    if current.strip():  # flush the final operand (there is no trailing comma)
+        operands.append(current.strip())
+    return operands
+
+
+def _has_memory_operand(operands: list[str]) -> bool:
+    """True if any operand dereferences memory (has "(", e.g. 0x8(%r8)). A move with
+    no memory operand is a reg-reg copy or an immediate load -- no value travels, so
+    there is nothing for us to corrupt."""
+    return any("(" in operand for operand in operands)
+
+
+def _data_register(operand: str) -> Optional[str]:
+    """The bare data-register name of an operand, e.g. "%rbx" -> "rbx", or None if the
+    operand is a memory dereference (has "(") or a control register we must not
+    corrupt."""
+    if "(" in operand:
+        return None
+    # m is a re.Match if the WHOLE operand is a register token (e.g. "%rbx"), else None.
+    m = _REG_TOK_RE.fullmatch(operand.strip())
+    if not m:
+        return None
+    name = m.group(1)  # capture group 1: the name without the '%', e.g. "%rbx" -> "rbx"
+    return None if name in _NON_DATA_REGS else name

tools/cpu_corruption_injector/injector_navigate.py

@@ -0,0 +1,396 @@
+#!/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
+# Runs ONLY inside the gdb python runtime (imports `gdb`); not a buck/pyre target.
+
+from __future__ import annotations
+
+import random
+import re
+from typing import Optional
+
+import gdb  # provided by gdb's embedded Python (not pip-importable)
+
+from injector_critical_instruction import classify_current_instruction
+from injector_register_corruption import apply_corruption, CorruptedInstruction
+
+# When single-stepping a target_fn call (to count + corrupt its instructions) we step
+# THROUGH rocksdb's own code by default; these are the EXTRA, non-rocksdb libraries
+# we ALSO step through -- data/integrity primitives (memcpy, (de)compression,
+# checksum) that carry key/value bytes. Any other callee (jemalloc, std::, libc
+# runtime) we `finish` straight back out of, to keep corruptions on value/data movement
+# rather than allocator/runtime bookkeeping. This is a broad filter: rocksdb's OWN
+# allocator (e.g. rocksdb::Arena) is rocksdb:: code and so IS stepped -- only
+# NON-rocksdb runtime noise is excluded. The same filter runs while counting and
+# while corrupting, so the critical instructions of one target_fn call line up with
+# another's -- SIMILARLY, not identically: data-dependent branches (e.g. key/value
+# size) can make a later call run a different instruction sequence, so a counted
+# index is a best-effort corruption target (a call with fewer criticals at that index
+# corrupts nothing; the caller handles that by retrying on the next target_fn call).
+_EXTRA_ALLOWED_LIBRARY_TOKENS = (
+    "memmove",
+    "memcpy",
+    "zstd",
+    "snappy",
+    "lz4",
+    "zlib",
+    "brotli",
+    "basic_string",
+    "char_traits",
+    "xxh",
+    "xxph",
+    "crc32",
+    "hash_bytes",
+)
+
+
+# Per-call single-step cap. A single gdb `stepi` is slow (~ms), and the outer per-run
+# timeout (runner_execute RUN_TIMEOUT) would kill the WHOLE run if one pathological
+# target_fn call (a giant key loop) stepped forever. This cap lets us give up on one
+# such call and move to the next, still producing an injection within the timeout.
+_MAX_STEPS_PER_CALL = 20000
+
+
+# db_stress lifecycle / error-handling model.
+#
+# WHY this is not a simple yes/no: we corrupt a CPU register, which often makes a
+# later instruction read a bad address -> db_stress takes a fatal signal (SIGSEGV).
+# By default gdb hands that signal to db_stress, whose own handler forks a SECOND gdb
+# to print a stack trace; two gdbs on one process collide and the run hangs. So we
+# tell gdb to freeze db_stress and hand control to us WITHOUT delivering the signal
+# (handle ... stop nopass, see setup_gdb). That one choice creates a third,
+# in-between condition -- alive, but crashed -- that a two-way alive/dead check misses.
+#
+# So db_stress is always in exactly ONE of three states; db_stress_state() returns
+# it, and callers can rely on that:
+#
+#   EXITED    the process is gone -- db_stress's own exit, or our kill.
+#   HEALTHY   alive and stopped where we put it (a breakpoint, or mid single-step).
+#   <signal>  alive but frozen on a fatal signal name ("SIGSEGV", ...) = CRASHED.
+#             nopass left the signal un-delivered, so a continue/stepi would only
+#             re-fault at the same pc and never make progress -- so the only way for
+#             us to end a crashed db_stress is to kill it.
+#
+# Lifecycle of one run (db_stress runs UNDER gdb throughout); both ways end EXITED,
+# and finish_db_stress() is the one place that drives the current state to EXITED:
+#
+#   HEALTHY:  finish_db_stress lets db_stress exit on its own       --> EXITED
+#   CRASHED:  finish_db_stress kills it (the only way to end a crash) --> EXITED
+#
+# (A corruption's bad value being used is what turns HEALTHY into CRASHED.)
+EXITED = "exited"
+HEALTHY = "healthy"
+
+
+def db_stress_state() -> str:
+    """Read db_stress's current state -- the one value finish_db_stress acts on (see
+    the note above):
+      - EXITED   db_stress is gone -- it exited on its own (when we let it finish), or
+                 we killed it. A crash never lands here -- at any time, including during
+                 db_stress's finishing, handle-nopass freezes db_stress alive on the
+                 signal instead.
+      - <signal> frozen on a fatal signal -- e.g. our corruption's SIGSEGV, or db_stress's
+                 own SIGABRT (an assert/abort); the value is that signal name.
+      - HEALTHY  alive, stopped where we put it.
+    Computed fresh each call."""
+    # gdb.selected_inferior() is gdb's handle to the db_stress process; pid 0, or no
+    # process to query at all, both mean it is gone.
+    try:
+        if gdb.selected_inferior().pid == 0:
+            return EXITED
+    except gdb.error:
+        return EXITED
+    try:
+        out = gdb.execute("info program", to_string=True)
+    except gdb.error:
+        return (
+            HEALTHY  # cannot read status -> assume alive; the next step reveals a crash
+        )
+    match = re.search(r"stopped with signal (SIG[A-Z0-9]+)", out)
+    return match.group(1) if match else HEALTHY
+
+
+def kill_db_stress() -> None:
+    try:
+        gdb.execute("kill")
+    except gdb.error:
+        # Usually db_stress is already dead. If a kill ever fails for another reason,
+        # runner_execute's per-run timeout killpg's the whole gdb+db_stress process
+        # group, so nothing is left orphaned.
+        pass
+
+
+def setup_gdb() -> None:
+    # Pagination: gdb pauses long output every screenful with a "---Type <return> to
+    # continue---" prompt; in batch mode that prompt blocks forever. Off.
+    gdb.execute("set pagination off")
+    # Confirm: gdb asks interactive y/n questions ("Delete all breakpoints?"); off so
+    # scripted commands never block on a prompt.
+    gdb.execute("set confirm off")
+    # Signal handling -- the source of most of the lifecycle complexity in this file,
+    # done to avoid TWO gdbs fighting over db_stress. On a fatal signal gdb's DEFAULT
+    # is to deliver it to db_stress, whose own signal handler forks a NESTED gdb to
+    # dump a stack trace -- that nested gdb fights ours for the process and wedges the
+    # run. So for these signals we set:
+    #   stop   -- gdb takes control and returns to this script (we inspect + record).
+    #   nopass -- do NOT forward the signal to db_stress, so its handler never runs
+    #             and no nested gdb is spawned. The faulting instruction is left
+    #             un-retired, so db_stress is now ALIVE-but-stopped-on-a-signal (the
+    #             "crashed" state db_stress_state() reports); a continue/stepi would
+    #             just re-fault at the same pc, which is why we KILL rather than
+    #             resume a crashed db_stress.
+    #   print  -- gdb writes a one-line "Program received signal ..." to gdb.log
+    #             (diagnostic only; runner_execute scans gdb.log for harness deaths).
+    gdb.execute("handle SIGSEGV SIGBUS SIGABRT SIGFPE stop nopass print")
+
+
+class Navigator:
+    """Drives gdb through ONE db_stress run to the op INSTANCE we inject into (the
+    op_index-th call of entry_fn). `_target_fn_breakpoint` is the standing breakpoint
+    reused across this op instance's target_fn calls (-1 until installed)."""
+
+    def __init__(self, entry_fn: str, target_fn: str) -> None:
+        self.entry_fn = entry_fn
+        self.target_fn = target_fn
+        self._target_fn_breakpoint = -1
+        self._ops_seen = 0
+
+    def reach_op(self, op_index: int) -> bool:
+        """Run db_stress to the op_index-th op instance: break on entry_fn (which
+        db_stress calls exactly once per op), skip the first op_index-1 hits, then
+        stop on the op_index-th. True if we stopped there, False if db_stress ended
+        first. op_index is 1-based (1 = the first op).
+
+        gdb's `break` does not return the new breakpoint's number to Python, so we
+        read it back via _last_breakpoint_num(). We record ops_seen from the
+        breakpoint's hit count, then DELETE the breakpoint -- keeping it would stop on
+        every entry_fn call AFTER this op instance too, starving the rest of the run
+        (scoping to our op is instead done by _entry_fn_on_stack in
+        reach_target_fn_call). So ops_seen is the EXACT op total when the op was not
+        reached (db_stress exited first), else op_index (a lower bound)."""
+        assert op_index >= 1, f"op_index is 1-based, got {op_index}"
+        gdb.execute(f"break {self.entry_fn}")
+        breakpoint_num = _last_breakpoint_num()
+        if op_index > 1:
+            # `ignore N k` only sets a skip counter; `continue` does the running. So
+            # ignore the first op_index-1 hits, then continue runs to the next
+            # (op_index-th) hit. For op_index == 1 there is nothing to skip.
+            gdb.execute(f"ignore {breakpoint_num} {op_index - 1}")
+        reached = False
+        try:
+            gdb.execute("continue")
+            reached = db_stress_state() == HEALTHY
+        except gdb.error:
+            reached = False
+        self._ops_seen = _breakpoint_hit_count(breakpoint_num)
+        _delete_breakpoint(breakpoint_num)
+        return reached
+
+    def ops_seen(self) -> int:
+        return self._ops_seen
+
+    def reach_target_fn_call(self) -> bool:
+        """Continue until db_stress next enters target_fn under an op of OUR TYPE --
+        this op instance, or (acceptably) a later instance of the same type. Installs
+        the standing target_fn breakpoint on the first call and reuses it afterwards.
+        Returns False once our op type is no longer on the stack, or db_stress exits /
+        crashes."""
+        if db_stress_state() != HEALTHY:
+            return False  # db_stress already exited / crashed -- nothing to reach
+        if self._target_fn_breakpoint < 0:
+            gdb.execute(f"break {self.target_fn}")
+            self._target_fn_breakpoint = _last_breakpoint_num()
+        try:
+            gdb.execute("continue")
+        except gdb.error:
+            return False  # db_stress exited
+        if db_stress_state() != HEALTHY:
+            return False
+        # The target_fn breakpoint fires on EVERY target_fn call in db_stress, and
+        # target_fn is SHARED across op types (e.g. BlockBuilder::Add is called by
+        # both flush and compaction). We accept a call only if OUR entry_fn is an
+        # ancestor frame -- that scopes us to the right OP TYPE (flush's call has
+        # FlushJob::Run on the stack, compaction's has CompactionJob::Run). It does
+        # NOT pin a single op INSTANCE, and that is fine: same op TYPE is all we need,
+        # so warmup may count critical instructions in one instance and the corruption land
+        # in a later instance of that same type (same code -> a SIMILAR, best-effort
+        # instruction index space). For the write op entry_fn == target_fn, so the
+        # current frame trivially passes.
+        return _entry_fn_on_stack(self.entry_fn)
+
+    def count_critical_instructions(self) -> int:
+        """Count this target_fn call's critical instructions (no corruption)."""
+        return self._walk_target_fn_call(corrupt_index=None)[1]
+
+    def corrupt_critical_instruction(
+        self, corrupt_index: int, rng: random.Random
+    ) -> tuple[Optional[CorruptedInstruction], int]:
+        """Corrupt this call's corrupt_index-th critical instruction (None if absent)."""
+        return self._walk_target_fn_call(corrupt_index, rng)
+
+    def _walk_target_fn_call(
+        self, corrupt_index: Optional[int], rng: Optional[random.Random] = None
+    ) -> tuple[Optional[CorruptedInstruction], int]:
+        call_start_depth = _frame_depth()
+        index = 0
+        for _ in range(_MAX_STEPS_PER_CALL):
+            instruction = classify_current_instruction()
+            if instruction is not None:
+                if corrupt_index is not None and index == corrupt_index:
+                    return apply_corruption(instruction, rng, self.target_fn), index + 1
+                index += 1
+            if _step_one_instruction() != HEALTHY or _frame_depth() < call_start_depth:
+                break
+        return None, index
+
+
+def finish_db_stress() -> Optional[str]:
+    """Finish this run's db_stress and return its crash signal (None if it exited
+    cleanly). The ONE place that ends a still-running db_stress, acting on
+    db_stress_state():
+      - HEALTHY  let db_stress make its own exit (db_stress's own shutdown, not ours).
+                 Delete our breakpoints first, or the final `continue` just stops on
+                 them again -- and that exit may itself end in a crash, handled next.
+      - CRASHED  KILL it -- a crash from before we got here, OR one during the exit
+                 above. Under handle-nopass (see setup_gdb) db_stress is frozen on the
+                 un-delivered signal; a `continue` would only re-fault, so killing is
+                 the only way we can end it.
+      - EXITED   nothing to do (db_stress already exited, or we killed it earlier).
+    """
+    state = db_stress_state()
+    if state == HEALTHY:
+        try:
+            gdb.execute("delete")
+        except gdb.error:
+            pass  # no breakpoints to delete -- fine
+        try:
+            gdb.execute("continue")
+        except gdb.error:
+            pass  # db_stress made its own exit during the continue
+        # Finishing ends db_stress one of two ways: its own exit, or a crash raised
+        # inside its own handling. Still HEALTHY would mean neither happened -- a bug
+        # in our breakpoint cleanup, so fail fast.
+        state = db_stress_state()
+        assert state != HEALTHY, f"db_stress did not finish (state={state})"
+    if state == EXITED:
+        return None
+    assert state not in (EXITED, HEALTHY), f"expected a crash signal, got {state}"
+    kill_db_stress()
+    return state
+
+
+def _step_one_instruction() -> str:
+    """Advance db_stress by one machine instruction and RETURN its state afterwards
+    (EXITED / HEALTHY / a fatal signal name). Returning it -- rather than the caller
+    calling db_stress_state() again -- avoids a second `info program` per step in the
+    hot stepping loop. Beyond the bare `stepi`:
+      - if db_stress is no longer HEALTHY (it exited, or an earlier corruption made it read a bad address and it crashed), return that state at once. We do NOT
+        kill a crash here -- finish_db_stress performs the single kill later; because
+        callers stop stepping a non-HEALTHY db_stress, it never re-faults (setup_gdb).
+      - else, if the step descended INTO a callee that is NOT rocksdb or an extra
+        library, `finish` straight back out (we only corrupt value/data-movement
+        instructions, not allocator/std/runtime ones)."""
+    depth_before = _frame_depth()
+    gdb.execute("stepi")
+    state = db_stress_state()
+    if state != HEALTHY:
+        return state
+    if _frame_depth() > depth_before and not _in_allowed_module(_source_location()):
+        try:
+            gdb.execute("finish")
+        except gdb.error:
+            # `finish` runs until the current function returns, which needs gdb to
+            # UNWIND it (work out its return address from debug/unwind metadata). A
+            # frame lacking that metadata (hand-written asm, stripped library) cannot
+            # be unwound; nothing sensible to do but stay put and keep stepping.
+            pass
+    return HEALTHY
+
+
+def _in_allowed_module(source_location: str) -> bool:
+    """Whether the frame described by `source_location` ("fn @ file:line") is code we
+    step THROUGH: rocksdb's own functions, or one of a few extra libraries (memcpy,
+    (de)compression, checksum). Everything else (jemalloc, std::, libc runtime) we
+    `finish` out of. We match on the function-name PREFIX for rocksdb so a wrapper
+    like std::__atomic_base<rocksdb::...> is NOT taken for rocksdb code."""
+    # source_location is "fn @ file:line"; take the fn part (maxsplit 1 so a " @ "
+    # inside the path cannot break it), trim, lowercase for case-insensitive match.
+    # `token in fn` is a substring test.
+    fn = source_location.split(" @ ", 1)[0].strip().lower()
+    if fn.startswith("rocksdb::") or "thunk to rocksdb::" in fn:
+        return True
+    return any(token in fn for token in _EXTRA_ALLOWED_LIBRARY_TOKENS)
+
+
+def _source_location() -> str:
+    """The current frame's "fn @ file:line", read from gdb. `frame.name()` is the
+    function name; `find_sal()` is gdb's symbol-and-line lookup for the frame's pc.
+    Returns "?" placeholders where symbols are unavailable."""
+    try:
+        frame = gdb.selected_frame()
+        fn = frame.name() or "?"
+        sal = frame.find_sal()
+        filename = sal.symtab.filename if sal and sal.symtab else "?"
+        line = sal.line if sal else 0
+        return f"{fn} @ {filename}:{line}"
+    except gdb.error:
+        return "?"
+
+
+def _entry_fn_on_stack(entry_fn: str) -> bool:
+    """True if `entry_fn` is on db_stress's current call stack (an ancestor frame).
+    Read-only walk like _frame_depth. Substring match (entry_fn in the frame name) so
+    a decorated/qualified frame name still matches the plain ENTRY_FN string."""
+    try:
+        frame = gdb.newest_frame()
+        while frame is not None:
+            if entry_fn in (frame.name() or ""):
+                return True
+            frame = frame.older()
+    except gdb.error:
+        return False
+    return False
+
+
+def _frame_depth() -> int:
+    """Number of frames on db_stress's current call stack. Read-only: walking
+    gdb.newest_frame() -> .older() inspects frames, it does not change the selected
+    frame or advance execution."""
+    depth = 0
+    try:
+        frame = gdb.newest_frame()
+        while frame is not None:
+            depth += 1
+            frame = frame.older()
+    except gdb.error:
+        pass
+    return depth
+
+
+def _last_breakpoint_num() -> int:
+    """The number of the most recently created breakpoint. gdb's `break` command does
+    not return it to Python, so we read it off the breakpoint list."""
+    breakpoints = gdb.breakpoints()
+    return breakpoints[-1].number if breakpoints else -1
+
+
+def _delete_breakpoint(num: int) -> None:
+    if num > 0:  # -1 is the "not installed" sentinel
+        try:
+            gdb.execute(f"delete {num}")
+        except gdb.error:
+            pass  # already gone -- fine
+
+
+def _breakpoint_hit_count(num: int) -> int:
+    """Times breakpoint `num` was hit (gdb counts hits even while it is ignored)."""
+    for breakpoint in gdb.breakpoints() or []:
+        if breakpoint.number == num:
+            try:
+                return int(breakpoint.hit_count)
+            except (gdb.error, TypeError, ValueError):
+                return 0
+    return 0

tools/cpu_corruption_injector/injector_register_corruption.py

@@ -0,0 +1,120 @@
+#!/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
+# Runs ONLY inside the gdb python runtime (imports `gdb`); not a buck/pyre target.
+
+from __future__ import annotations
+
+import random
+from dataclasses import dataclass
+
+import gdb  # provided by gdb's embedded Python (not pip-importable)
+
+from injector_critical_instruction import CriticalInstruction, simd_int64_lanes
+from injector_telemetry import capture_corruption_details, CorruptionDetails
+
+# eflags bit positions we flip: CF, ZF, SF, OF -- the condition flags a branch reads.
+_EFLAGS_BITS = (0, 6, 7, 11)
+
+
+@dataclass
+class CorruptedInstruction:
+    """A CriticalInstruction after we corrupted it, plus what we observed: the register
+    value before and after the corruption, and the details to locate the instruction. (This
+    is what one element of inject.json's "corruptions" is serialized from.)"""
+
+    instruction: CriticalInstruction
+    before: str
+    after: str
+    details: CorruptionDetails
+
+
+def apply_corruption(
+    critical_instruction: CriticalInstruction, rng: random.Random, target_fn: str
+) -> CorruptedInstruction:
+    """Corrupt the critical_instruction we are stopped on and return a
+    CorruptedInstruction. If corrupt_after_exec, step the instruction first so we corrupt
+    the value it just produced; otherwise corrupt now, before it consumes the register.
+    A corruption that produces no observable change means the decoder mis-selected the
+    operand -> the before != after assert (fail fast). target_fn bounds the captured
+    call chain (see capture_corruption_details)."""
+    details = capture_corruption_details(
+        target_fn
+    )  # before a possible step changes $pc
+    if critical_instruction.corrupt_after_exec:
+        gdb.execute("stepi")  # run the instruction so we corrupt the value it produced
+    reg = critical_instruction.reg
+    before = read_register(reg)
+    _corrupt_register(reg, critical_instruction.corruption_type, rng)
+    after = read_register(reg)
+    assert before != after, f"corruption produced no change: {before!r} == {after!r}"
+    return CorruptedInstruction(critical_instruction, before, after, details)
+
+
+def _corrupt_register(reg: str, corruption_type: str, rng: random.Random) -> None:
+    """Corrupt `reg` the way corruption_type says. Raises gdb.error if the write fails;
+    callers fail fast on that, since the decoder only yields operands we can corrupt.
+
+    We compute the XOR mask in Python and embed it as a DECIMAL literal, rather than
+    writing the shift in gdb. The hazard is the gdb-side shift `1 << n` for large n
+    (lane_bit_flip goes up to 1 << 63): gdb types the literal `1` as a 32-bit C `int`, so a
+    shift of 32+ overflows to 0 -- e.g. gdb evaluates `1 << 40` as 0, silently a no-op.
+    A whole decimal literal is safe -- gdb types an integer constant by its magnitude
+    (C's rule), so the same mask written out, 1099511627776, is taken as a 64-bit
+    `long`."""
+    if corruption_type == "bit_flip":
+        # Flip a random bit in the low byte (bits 0-7). Every GP register is >= 8 bits,
+        # so a low-byte flip is always observable, and the low bytes hold most key/value
+        # data -- the highest-yield, simplest choice.
+        gdb.execute(f"set ${reg} = ${reg} ^ {1 << rng.randrange(8)}")
+        return
+    if corruption_type == "flag_flip":
+        gdb.execute(f"set ${reg} = ${reg} ^ {1 << rng.choice(_EFLAGS_BITS)}")
+        return
+    if corruption_type == "lane_bit_flip":
+        # gdb cannot XOR a whole vector register, only a scalar lane member -- so we
+        # MUST address a lane just to write to the register at all. We then flip ONE
+        # bit: the lane choice picks which 64-bit chunk of the value, the bit choice the
+        # bit within it. Single-bit (not a whole-lane scribble) on purpose: a bit flip
+        # is the realistic CPU fault we model and it matches bit_flip/flag_flip;
+        # addressing the lane is just gdb's required mechanism, not a reason to corrupt
+        # more.
+        lane_expr, lane_count = simd_int64_lanes(reg)
+        assert lane_expr is not None, f"no int64 lane view for vector reg {reg}"
+        lane = rng.randrange(lane_count)
+        gdb.execute(
+            f"set ${lane_expr}[{lane}] = ${lane_expr}[{lane}] ^ {1 << rng.randrange(64)}"
+        )
+        return
+    raise AssertionError(f"unknown corruption_type {corruption_type!r} (reg={reg})")
+
+
+def read_register(reg: str) -> str:
+    """The register's value as a string, for the before/after telemetry. The read
+    matches the register kind:
+      - eflags            -> hex of its low 32 bits (eflags is a 32-bit register)
+      - a vector register -> its int64 lanes as a list, e.g. "[123,456]"
+      - any other (GP)    -> hex of its low 64 bits
+    gdb.parse_and_eval returns a gdb.Value; int(...) turns it into a Python int, and
+    for a vector we index the lane members (lanes[i])."""
+    if reg == "eflags":
+        try:
+            return hex(int(gdb.parse_and_eval("$eflags")) & 0xFFFFFFFF)
+        except gdb.error:
+            return "<eflags?>"
+    if "xmm" in reg or "ymm" in reg or "zmm" in reg:
+        lane_expr, lane_count = simd_int64_lanes(reg)
+        if lane_expr is None:
+            return "<vector?>"
+        try:
+            lanes = gdb.parse_and_eval(f"${lane_expr}")
+            return "[" + ",".join(str(int(lanes[i])) for i in range(lane_count)) + "]"
+        except gdb.error:
+            return "<vector?>"
+    try:
+        return hex(int(gdb.parse_and_eval(f"${reg}")) & 0xFFFFFFFFFFFFFFFF)
+    except gdb.error:
+        return "<reg?>"

tools/cpu_corruption_injector/injector_telemetry.py

@@ -0,0 +1,145 @@
+#!/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
+# Runs ONLY inside the gdb python runtime (imports `gdb`); not a buck/pyre target.
+
+# This module is the injector's OBSERVABILITY layer: it captures where a corruption
+# landed (CorruptionDetails) and serializes the whole per-run record to inject.json.
+# It only reads the other modules' types, never the reverse, so the import graph stays
+# acyclic (everyone imports telemetry; telemetry imports none of them).
+#
+# inject.json data model -- each type is defined where its logic lives:
+#
+#   InjectRecord                (injector.py)        one per run: injection_result
+#                                                    (injected / op_not_reached / ...)
+#                                                    + op context
+#     .corruptions: list of
+#       CorruptedInstruction    (injector_register_corruption.py)  a single flip
+#         .instruction:
+#           CriticalInstruction (injector_critical_instruction.py) reg / type / text
+#         .before, .after       the register value the flip changed (hex / lane list)
+#         .details:
+#           CorruptionDetails   (this module)        where the flip landed in source
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+
+import gdb  # provided by gdb's embedded Python (not pip-importable)
+
+
+@dataclass
+class CorruptionDetails:
+    """Where a corruption landed in source -- enough to triage it afterwards.
+    source      "function @ file:line" of the corrupted instruction, e.g.
+                "rocksdb::Slice::compare @ include/rocksdb/slice.h:285".
+    call_chain  the call chain from that instruction up to and including target_fn
+                (the op's value-carrying function), innermost first -- inlined leaf
+                frames plus the physical callers above them, e.g.
+                ["rocksdb::Slice::compare @ include/rocksdb/slice.h:285",
+                 "...BytewiseComparatorImpl::Compare @ util/comparator.cc:37",
+                 "rocksdb::MemTable::Add @ db/memtable.cc:602"].
+    """
+
+    source: str
+    call_chain: list[str]
+
+
+def capture_corruption_details(target_fn: str) -> CorruptionDetails:
+    """Snapshot where the instruction at $pc is in source. Call it BEFORE a
+    corrupt-after-exec corruption steps past the instruction, so $pc still points at
+    the instruction we corrupt. `source` is the innermost frame (the exact corrupted
+    expression); `call_chain` extends up to target_fn for caller context."""
+    call_chain = _call_chain_to_target(target_fn)
+    source = call_chain[0] if call_chain else _func_offset()
+    return CorruptionDetails(source=source, call_chain=call_chain)
+
+
+def inject_record_to_json(inject_record) -> dict[str, object]:
+    return {
+        "injection_result": inject_record.injection_result,
+        "db_stress_crash_signal": inject_record.db_stress_crash_signal,
+        "op": inject_record.op,
+        "op_index": inject_record.op_index,
+        "entry_fn": inject_record.entry_fn,
+        "target_fn": inject_record.target_fn,
+        "critical_instruction_index": inject_record.critical_instruction_index,
+        "corruptions": [
+            _corrupted_instruction_to_json(c) for c in inject_record.corruptions
+        ],
+        "ops_seen": inject_record.ops_seen,
+        "critical_instructions_seen": inject_record.critical_instructions_seen,
+    }
+
+
+def write_inject_record_json(path: str, inject_record) -> None:
+    try:
+        with open(path, "w", encoding="utf-8") as f:
+            json.dump(inject_record_to_json(inject_record), f, indent=2)
+            f.write("\n")
+    except OSError as e:
+        gdb.write(f"injector.py: failed to write {path}: {e}\n", gdb.STDERR)
+
+
+def _corrupted_instruction_to_json(corrupted) -> dict[str, object]:
+    instruction = corrupted.instruction  # the CriticalInstruction we corrupted
+    return {
+        "instruction": instruction.text,
+        "register": instruction.reg,
+        "corruption_type": instruction.corruption_type,
+        "before": corrupted.before,
+        "after": corrupted.after,
+        "details": {
+            "source": corrupted.details.source,
+            "call_chain": corrupted.details.call_chain,
+        },
+    }
+
+
+def _func_offset() -> str:
+    """ "function+offset" at $pc -- the `source` fallback when the pc has no inline frame
+    (e.g. a leaf with no line info)."""
+    try:
+        pc = int(gdb.selected_frame().pc())
+        out = gdb.execute(f"info symbol {hex(pc)}", to_string=True).strip()
+    except gdb.error:
+        return "?"
+    # `info symbol` prints e.g. "rocksdb::MemTable::Add(...) + 123 in section .text of
+    # /path"; we drop the " in section ..." tail and return "rocksdb::MemTable::Add(...)
+    # + 123".
+    return out.split(" in section ", 1)[0].strip()
+
+
+def _call_chain_to_target(target_fn: str) -> list[str]:
+    """The call chain at $pc, innermost first, as ["function @ file:line", ...], up to
+    and INCLUDING target_fn -- the op's value-carrying function we corrupt inside.
+
+    It spans both the inlined leaf chain (gdb exposes each inlined function as an
+    INLINE_FRAME sharing the pc, so the leaf pins the exact corrupted expression) and
+    the physical frames above them, up to target_fn. We stop at target_fn because
+    frames above it are generic op dispatch, not part of the corrupted expression's
+    context. A compiler-generated frame (e.g. a non-virtual thunk) has no source line
+    -> "@ ?:0"; that is expected, and walking past it now reaches the real callers
+    instead of dead-ending there. Capped at 32 frames in case target_fn is not found."""
+    frames: list[str] = []
+    try:
+        frame = gdb.newest_frame()
+    except gdb.error:
+        return frames
+    while frame is not None and len(frames) < 32:
+        try:
+            function_name = frame.name() or "?"
+            sal = frame.find_sal()  # gdb's symbol-and-line lookup for this frame's pc
+            file_name = sal.symtab.filename if sal and sal.symtab else "?"
+            line = sal.line if sal else 0
+            frames.append(f"{function_name} @ {file_name}:{line}")
+            if target_fn in function_name:
+                break  # reached the op's value-carrying function -- stop here
+            frame = frame.older()
+        except gdb.error:
+            break
+    return frames

tools/cpu_corruption_injector/tests/test_injector_critical_instruction.py

@@ -0,0 +1,104 @@
+#!/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
+"""Unit tests for injector_critical_instruction.decode_instruction -- one
+representative case per decode path (REAL `x/i` lines from gdb on db_stress).
+
+decode_instruction is pure string parsing, but the module it lives in does
+`import gdb` at the top. `gdb` is not a pip package -- it only exists inside gdb's
+embedded Python interpreter, so a plain `python3` test run would fail at that import
+with ModuleNotFoundError. To test the parser standalone, we register a fake `gdb`
+module in sys.modules first (see below), so the import succeeds.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import types
+import unittest
+
+# The module under test lives one directory up; add that dir to sys.path so the import
+# below resolves. Append (not insert(0)) puts it at the END of the search path, so a
+# sibling file here can't shadow a stdlib module that happens to share its name.
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+# Register a stand-in `gdb` so the module's top-level `import gdb` finds this instead
+# of failing. The decoder never calls gdb -- it only references gdb.error in except
+# clauses -- so a single exception class is all the stub needs.
+if "gdb" not in sys.modules:
+    stub = types.ModuleType("gdb")
+    stub.error = type("error", (Exception,), {})
+    sys.modules["gdb"] = stub
+
+from injector_critical_instruction import decode_instruction  # noqa: E402
+
+# (name, x/i line, expected reg, expected corruption_type, corrupt_after_exec).
+# corruption_type "" means the decoder rejects the line. One representative case per
+# decode path.
+_CASES: list[tuple[str, str, str, str, bool]] = [
+    # memory load into a GP register -> bit_flip on the dest, corrupt AFTER the load.
+    ("load_gp_reg", "   0x6f42e4:\tmov    0x8(%r8),%rbx", "rbx", "bit_flip", True),
+    # store of a GP register to non-stack memory -> bit_flip on the source, BEFORE.
+    ("store_gp_reg", "   0x6f4400:\tmov    %rbx,0x10(%r14)", "rbx", "bit_flip", False),
+    # cmp/test -> flag_flip on eflags, corrupt AFTER it executes.
+    ("cmp_sets_flags", "   0x6f4313:\tcmp    $0x80,%ebx", "eflags", "flag_flip", True),
+    # rejected: reg-reg copy (no memory operand).
+    ("reject_reg_reg_copy", "   0x6f430a:\tmov    %ebx,%r13d", "", "", False),
+    # rejected: lea (address math, not a value move).
+    ("reject_lea", "   0x6f439c:\tlea    0x2750(%r14),%rdx", "", "", False),
+    # rejected: stack spill (memory operand is rbp-relative).
+    ("reject_stack_spill", "   0x6f42e8:\tmov    %r9,-0xf8(%rbp)", "", "", False),
+    # rejected: indexed/scaled stack spill (rbp base with index, not plain "(%rbp)").
+    (
+        "reject_stack_spill_indexed",
+        "   0x6f42ec:\tmov    %r9,0x10(%rbp,%rcx,8)",
+        "",
+        "",
+        False,
+    ),
+    # rejected: load into a control register (rsp).
+    ("reject_load_control_reg", "   0x6f4600:\tmov    0x8(%r8),%rsp", "", "", False),
+]
+
+
+class DecodeInsnTest(unittest.TestCase):
+    def test_representative_cases(self) -> None:
+        # subTest (stdlib) rather than @parameterized.expand: this test runs standalone
+        # under plain python3 (the module under test imports gdb, so it cannot be a buck
+        # target) and `parameterized` is not in the devserver stdlib. subTest still
+        # reports each case independently and keeps going after a failing case.
+        for name, line, reg, corruption_type, corrupt_after_exec in _CASES:
+            with self.subTest(name=name):
+                d = decode_instruction(line, can_corrupt_int64_lane=lambda _r: True)
+                if corruption_type == "":
+                    self.assertIsNone(d, f"expected reject for {line!r}")
+                    continue
+                self.assertIsNotNone(d, f"expected decode for {line!r}")
+                self.assertEqual(
+                    (d.reg, d.corruption_type, d.corrupt_after_exec),
+                    (reg, corruption_type, corrupt_after_exec),
+                    line,
+                )
+
+    def test_simd_load_is_lane_bit_flip_when_addressable(self) -> None:
+        d = decode_instruction(
+            "   0x6f4500:\tvmovdqu 0x0(%r14),%xmm3",
+            can_corrupt_int64_lane=lambda _r: True,
+        )
+        self.assertEqual(
+            (d.reg, d.corruption_type, d.corrupt_after_exec),
+            ("xmm3", "lane_bit_flip", True),
+        )
+
+    def test_simd_skipped_when_not_addressable(self) -> None:
+        # If gdb cannot address the lanes, the decoder skips it (the "only corruptible
+        # operands" contract).
+        d = decode_instruction(
+            "   0x6f4500:\tvmovdqu 0x0(%r14),%xmm3",
+            can_corrupt_int64_lane=lambda _r: False,
+        )
+        self.assertIsNone(d)