Add db_stress operation breadcrumbs (#15194)

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

## Why
Intermittent db_stress and db_crashtest failures often report only the terminal assertion, after the operations that established the failing state have scrolled out of logs. The diagnostics must retain the latest operations even when the stress process is killed before it can run a failure handler.

## Why the existing recovery trace is not enough
The existing expected-state trace is a correctness oracle for recovery: after an expected-state checkpoint it records ordered write batches, and recovery replays the prefix implied by the database sequence number. Reads and iterator seeks are deliberately filtered out, trace-write failures are fatal, and the trace is machine-consumed rather than intended as a concise failure narrative. It therefore cannot show which read, scan, maintenance operation, or operation phase immediately preceded an assertion. Breadcrumbs fill that observability gap with a bounded, human-readable, per-thread history. They are best effort, never replayed, and remain independent of the recovery-validation contract.

## Previous behavior
db_stress did not retain a bounded per-thread operation history. db_crashtest did not automatically provision a local directory for these artifacts. The first version of this diff kept the ring in process-local memory and periodically wrote a complete temp-file snapshot, which could lose every operation since the last snapshot when the process crashed.

## New behavior
An opt-in per-thread memory-mapped breadcrumb ring records recent operation begin, context, and end events directly in a local file. Committed records therefore remain in the kernel page cache when only the db_stress process exits or is killed; no periodic flush is required. db_crashtest enables breadcrumbs by default, assigns a local diagnostics directory, retains the newest eight process groups after failed child runs, and removes managed artifacts after success.

## Implementation
Each ThreadState lazily creates a uniquely named, preallocated file and maps it through Env::NewMemoryMappedFileBuffer(). The file contains a fixed header and 512-byte text records. A record slot is marked I before its body is replaced and V only after the complete body is published; E denotes an unused slot. Each valid record carries a monotonically increasing sequence number so wrapped records can be ordered. Oversized context is bounded and ends with details_truncated=1. Lazy initialization avoids empty files for auxiliary ThreadState objects, and a process-wide file ID prevents collisions between ThreadState objects that share a logical thread ID.

The periodic-flush flag, temp-file/rename snapshot path, and verification-failure fallback flush are removed. Enabled operation-context formatting continues to reuse a per-thread string and uses std::to_chars for integers; disabled call sites avoid formatting. Diagnostic directory setup and mapping failures are reported but do not change the workload result. Auto-directory pruning and cleanup accept only generated filename schemas, reject symlink roots, preserve unrelated files and explicit user directories, and retain at most eight PID groups.

## Safety and unsupported cases
This is stress-tool-only diagnostic behavior. Raw db_stress keeps breadcrumbs disabled by default. The expected-state recovery trace and scan-witness contracts are unchanged. The mapping deliberately does not use msync or fsync: it is intended to survive a process crash while the kernel remains running, not power loss, host reset, or kernel failure. Filesystems or Env implementations without shared writable mappings disable breadcrumbs non-fatally. Replacing or truncating a live mapped file is unsupported. Records are fixed width, so unusually large operation context is explicitly truncated. Concurrent live readers are not promised a consistent snapshot; post-termination readers must ignore I and E records and order V records by sequence number. Nested operation scopes remain unsupported.

## Tests and limitations
Validated with all 47 tools/db_crashtest_test.py tests, exact integrated debug and release db_stress builds, complete-stack arc lint, lower/upper/combined clang-format checks, normal ring wrap, distinct files for four worker threads, an oversized 256-column-family context, and process-crash tests. A release process killed with SIGKILL retained all 16 committed ring records without an explicit flush. A debugger-forced SIGKILL after the I store but before V publication left the interrupted slot marked I. In five CPU-pinned alternating release pairs of 50,000 operations, enabled and disabled medians were both 2.15 seconds. This remains a bounded latest history rather than a complete log and provides no host-crash or power-loss durability.

## Rebase update
Rebased onto fbsource master dc59f986b964. The logical diff is unchanged; the rebased stack passes debug and release db_stress builds, all 47 db_crashtest Python tests, and arc lint.

## Review follow-up
The mapped-slot invalidation now uses an acquire-release atomic exchange so later body stores cannot become visible before the I marker on weakly ordered architectures.

Reviewed By: anand1976

Differential Revision: D119160910

fbshipit-source-id: ffe172af9ebc75c617196592806dc9641a539a25

db_stress_tool/db_stress_common.h

@@ -107,6 +107,9 @@ DECLARE_bool(destroy_db_and_exit);
 DECLARE_string(delete_dir_and_exit);
 DECLARE_bool(verbose);
 DECLARE_bool(progress_reports);
+DECLARE_string(stress_diagnostics_dir);
+DECLARE_bool(stress_diagnostics_breadcrumbs);
+DECLARE_uint64(stress_diagnostics_breadcrumb_entries);
 DECLARE_uint64(db_write_buffer_size);
 DECLARE_int32(write_buffer_size);
 DECLARE_int32(max_write_buffer_number);

db_stress_tool/db_stress_gflags.cc

@@ -22,6 +22,17 @@ static bool ValidateUint32Range(const char* flagname, uint64_t value) {
   return true;
 }
 
+static bool ValidateStressDiagnosticEntryCount(const char* flagname,
+                                               uint64_t value) {
+  constexpr uint64_t kMaxStressDiagnosticEntries = 64 * 1024;
+  if (value > kMaxStressDiagnosticEntries) {
+    fprintf(stderr, "Invalid value for --%s: %lu, maximum is %lu\n", flagname,
+            (unsigned long)value, (unsigned long)kMaxStressDiagnosticEntries);
+    return false;
+  }
+  return true;
+}
+
 DEFINE_uint64(seed, 2341234,
               "Seed for PRNG. When --nooverwritepercent is "
               "nonzero and --expected_values_dir is nonempty, this value "
@@ -166,6 +177,24 @@ DEFINE_uint64(
     "If non-zero with --liveness_check_interval_sec, terminate db_stress when "
     "no operation completes for this many seconds during the operation phase.");
 
+DEFINE_string(stress_diagnostics_dir, "",
+              "Directory where db_stress writes optional diagnostic artifacts. "
+              "If empty, diagnostics that require file output are disabled.");
+
+DEFINE_bool(stress_diagnostics_breadcrumbs, false,
+            "If true, keep a bounded per-thread memory-mapped ring of recent "
+            "stress operations in --stress_diagnostics_dir when set.");
+
+DEFINE_uint64(stress_diagnostics_breadcrumb_entries, 256,
+              "Number of recent operation breadcrumbs retained per stress "
+              "thread when --stress_diagnostics_breadcrumbs is true. Maximum "
+              "is 65536.");
+// NOLINTNEXTLINE(cert-err58-cpp)
+static const bool FLAGS_stress_diagnostics_breadcrumb_entries_dummy
+    __attribute__((__unused__)) =
+        RegisterFlagValidator(&FLAGS_stress_diagnostics_breadcrumb_entries,
+                              &ValidateStressDiagnosticEntryCount);
+
 DEFINE_uint64(db_write_buffer_size,
               ROCKSDB_NAMESPACE::Options().db_write_buffer_size,
               "Number of bytes to buffer in all memtables before compacting");

db_stress_tool/db_stress_shared_state.cc

@@ -11,10 +11,79 @@
 #ifdef GFLAGS
 #include "db_stress_tool/db_stress_shared_state.h"
 
+#include <algorithm>
+#include <atomic>
+#include <charconv>
+#include <cstring>
+#include <new>
+
 #include "db_stress_tool/db_stress_test_base.h"
+#include "port/port.h"
 #include "rocksdb/env.h"
 
 namespace ROCKSDB_NAMESPACE {
+namespace {
+
+using BreadcrumbRecordState = std::atomic<char>;
+
+constexpr size_t kOperationBreadcrumbRecordBytes = 512;
+constexpr size_t kOperationBreadcrumbHeaderBytes =
+    kOperationBreadcrumbRecordBytes;
+constexpr size_t kOperationBreadcrumbRecordsPerInitializationWrite = 128;
+
+std::atomic<uint64_t> next_operation_breadcrumb_file_id{0};
+
+std::string SanitizePathComponent(const std::string& value) {
+  std::string result;
+  result.reserve(value.size());
+  for (char c : value) {
+    if ((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
+        (c >= '0' && c <= '9') || c == '_' || c == '-' || c == '.') {
+      result.push_back(c);
+    } else {
+      result.push_back('_');
+    }
+  }
+  return result;
+}
+
+template <typename Integer>
+void AppendInteger(Integer value, std::string* output) {
+  char buffer[32];
+  const auto result = std::to_chars(buffer, buffer + sizeof(buffer), value);
+  assert(result.ec == std::errc());
+  output->append(buffer, result.ptr);
+}
+
+void AppendSanitizedRecordValue(const std::string& value, size_t length,
+                                std::string* output) {
+  assert(length <= value.size());
+  for (size_t i = 0; i < length; ++i) {
+    const char c = value[i];
+    output->push_back(c == '\n' || c == '\r' || c == '\t' ? ' ' : c);
+  }
+}
+
+std::string OperationBreadcrumbFilePath(const ThreadState& thread,
+                                        uint64_t file_id) {
+  std::string path = FLAGS_stress_diagnostics_dir;
+  if (!path.empty() && path.back() != '/') {
+    path.push_back('/');
+  }
+  const StressTest* const stress_test = thread.shared->GetStressTest();
+  const std::string db_label =
+      stress_test ? stress_test->GetDbLabel() : "unknown_db";
+  path.append(SanitizePathComponent(db_label));
+  path.append(".pid_");
+  path.append(std::to_string(port::GetProcessID()));
+  path.append(".thread_");
+  path.append(std::to_string(file_id));
+  path.append(".breadcrumbs.txt");
+  return path;
+}
+
+}  // namespace
+
 thread_local bool SharedState::ignore_read_error;
 
 SharedState::SharedState(Env* env, StressTest* stress_test)
@@ -134,6 +203,260 @@ bool SharedState::BeginOperation(uint32_t tid, StressOperationType type) {
   return true;
 }
 
+ThreadState::ThreadState(uint32_t index, SharedState* _shared)
+    : tid(index),
+      rand(1000 + index + _shared->GetSeed()),
+      shared(_shared),
+      operation_breadcrumb_pos(0),
+      operation_breadcrumb_sequence(0),
+      operation_breadcrumb_file_id(0),
+      operation_breadcrumb_setup_attempted(false),
+      diagnostic_io_disabled(false),
+      operation_ordinal(0),
+      current_operation_ordinal(0) {}
+
+ThreadState::~ThreadState() {
+  if (operation_breadcrumb_mapping == nullptr) {
+    return;
+  }
+  const size_t entry_count = (operation_breadcrumb_mapping->GetLen() -
+                              kOperationBreadcrumbHeaderBytes) /
+                             kOperationBreadcrumbRecordBytes;
+  char* const base =
+      static_cast<char*>(operation_breadcrumb_mapping->GetBase());
+  for (size_t i = 0; i < entry_count; ++i) {
+    char* const slot = base + kOperationBreadcrumbHeaderBytes +
+                       i * kOperationBreadcrumbRecordBytes;
+    BreadcrumbRecordState* const state =
+        std::launder(reinterpret_cast<BreadcrumbRecordState*>(slot));
+    state->~BreadcrumbRecordState();
+  }
+}
+
+bool ThreadState::OperationBreadcrumbsEnabled() const {
+  return FLAGS_stress_diagnostics_breadcrumbs &&
+         FLAGS_stress_diagnostics_breadcrumb_entries > 0 &&
+         !FLAGS_stress_diagnostics_dir.empty() && !diagnostic_io_disabled &&
+         (!operation_breadcrumb_setup_attempted ||
+          operation_breadcrumb_mapping != nullptr);
+}
+
+bool ThreadState::InitializeOperationBreadcrumbs() {
+  if (operation_breadcrumb_mapping != nullptr) {
+    return true;
+  }
+  if (!OperationBreadcrumbsEnabled()) {
+    return false;
+  }
+  operation_breadcrumb_setup_attempted = true;
+
+  if (sizeof(BreadcrumbRecordState) != sizeof(char) ||
+      kOperationBreadcrumbRecordBytes % alignof(BreadcrumbRecordState) != 0 ||
+      !BreadcrumbRecordState{}.is_lock_free()) {
+    fprintf(stdout,
+            "Operation breadcrumbs require a lock-free one-byte atomic "
+            "record marker\n");
+    return false;
+  }
+
+  Env* const env = Env::Default();
+  Status s = env->CreateDirIfMissing(FLAGS_stress_diagnostics_dir);
+  if (!s.ok()) {
+    // Some Env implementations report an error when another thread creates
+    // the directory between their existence check and create call.
+    Status exists_status = env->FileExists(FLAGS_stress_diagnostics_dir);
+    if (!exists_status.ok()) {
+      fprintf(stdout, "Failed to create stress diagnostics directory %s: %s\n",
+              FLAGS_stress_diagnostics_dir.c_str(), s.ToString().c_str());
+      diagnostic_io_disabled = true;
+      return false;
+    }
+  }
+
+  operation_breadcrumb_file_id =
+      next_operation_breadcrumb_file_id.fetch_add(1, std::memory_order_relaxed);
+  const std::string path =
+      OperationBreadcrumbFilePath(*this, operation_breadcrumb_file_id);
+  std::unique_ptr<WritableFile> file;
+  s = env->NewWritableFile(path, &file, EnvOptions());
+  if (!s.ok()) {
+    fprintf(stdout, "Failed to create operation breadcrumb file %s: %s\n",
+            path.c_str(), s.ToString().c_str());
+    return false;
+  }
+
+  std::string header =
+      "# mmap_breadcrumbs_v1 record_state=V(valid),I(incomplete),E(empty) "
+      "order_by=seq power_loss_durable=0 pid=";
+  AppendInteger(port::GetProcessID(), &header);
+  header.append(" tid=");
+  AppendInteger(tid, &header);
+  header.append(" file_id=");
+  AppendInteger(operation_breadcrumb_file_id, &header);
+  header.append(" seed=");
+  AppendInteger(shared->GetSeed(), &header);
+  assert(header.size() < kOperationBreadcrumbHeaderBytes);
+  header.resize(kOperationBreadcrumbHeaderBytes - 1, ' ');
+  header.push_back('\n');
+
+  Status write_status = file->Append(Slice(header));
+  std::string empty_records(kOperationBreadcrumbRecordsPerInitializationWrite *
+                                kOperationBreadcrumbRecordBytes,
+                            ' ');
+  for (size_t i = 0; i < kOperationBreadcrumbRecordsPerInitializationWrite;
+       ++i) {
+    empty_records[i * kOperationBreadcrumbRecordBytes] = 'E';
+    empty_records[(i + 1) * kOperationBreadcrumbRecordBytes - 1] = '\n';
+  }
+
+  const size_t entry_count = FLAGS_stress_diagnostics_breadcrumb_entries;
+  for (size_t first = 0; write_status.ok() && first < entry_count;
+       first += kOperationBreadcrumbRecordsPerInitializationWrite) {
+    const size_t count = std::min(
+        kOperationBreadcrumbRecordsPerInitializationWrite, entry_count - first);
+    write_status = file->Append(
+        Slice(empty_records.data(), count * kOperationBreadcrumbRecordBytes));
+  }
+
+  Status close_status = file->Close();
+  file.reset();
+  if (!write_status.ok()) {
+    s = write_status;
+  } else if (!close_status.ok()) {
+    s = close_status;
+  }
+  if (!s.ok()) {
+    env->DeleteFile(path).PermitUncheckedError();
+    fprintf(stdout, "Failed to initialize operation breadcrumb file %s: %s\n",
+            path.c_str(), s.ToString().c_str());
+    return false;
+  }
+
+  std::unique_ptr<MemoryMappedFileBuffer> mapping;
+  s = env->NewMemoryMappedFileBuffer(path, &mapping);
+  const size_t expected_size = kOperationBreadcrumbHeaderBytes +
+                               entry_count * kOperationBreadcrumbRecordBytes;
+  if (s.ok() && (mapping == nullptr || mapping->GetBase() == nullptr ||
+                 mapping->GetLen() != expected_size)) {
+    s = Status::IOError("Unexpected operation breadcrumb mapping size");
+  }
+  if (!s.ok()) {
+    mapping.reset();
+    env->DeleteFile(path).PermitUncheckedError();
+    fprintf(stdout, "Failed to map operation breadcrumb file %s: %s\n",
+            path.c_str(), s.ToString().c_str());
+    return false;
+  }
+
+  char* const base = static_cast<char*>(mapping->GetBase());
+  for (size_t i = 0; i < entry_count; ++i) {
+    char* const slot = base + kOperationBreadcrumbHeaderBytes +
+                       i * kOperationBreadcrumbRecordBytes;
+    new (slot) BreadcrumbRecordState('E');
+  }
+  operation_breadcrumb_mapping = std::move(mapping);
+  operation_breadcrumb_details.reserve(kOperationBreadcrumbRecordBytes);
+  operation_breadcrumb_record.reserve(kOperationBreadcrumbRecordBytes);
+  return true;
+}
+
+void ThreadState::AppendOperationBreadcrumb(StressOperationType type,
+                                            const char* phase,
+                                            const std::string& details) {
+  if (!InitializeOperationBreadcrumbs()) {
+    return;
+  }
+
+  const uint64_t sequence = ++operation_breadcrumb_sequence;
+  operation_breadcrumb_record.clear();
+  operation_breadcrumb_record.append(" seq=");
+  AppendInteger(sequence, &operation_breadcrumb_record);
+  operation_breadcrumb_record.append(" time_micros=");
+  AppendInteger(shared->GetEnv()->NowMicros(), &operation_breadcrumb_record);
+  operation_breadcrumb_record.append(" tid=");
+  AppendInteger(tid, &operation_breadcrumb_record);
+  operation_breadcrumb_record.append(" ordinal=");
+  AppendInteger(current_operation_ordinal, &operation_breadcrumb_record);
+  operation_breadcrumb_record.append(" op=");
+  operation_breadcrumb_record.append(StressOperationTypeName(type));
+  operation_breadcrumb_record.append(" phase=");
+  operation_breadcrumb_record.append(phase);
+
+  constexpr size_t kMaxRecordContentBytes = kOperationBreadcrumbRecordBytes - 2;
+  assert(operation_breadcrumb_record.size() < kMaxRecordContentBytes);
+  if (!details.empty()) {
+    operation_breadcrumb_record.push_back(' ');
+    const size_t available =
+        kMaxRecordContentBytes - operation_breadcrumb_record.size();
+    if (details.size() <= available) {
+      AppendSanitizedRecordValue(details, details.size(),
+                                 &operation_breadcrumb_record);
+    } else {
+      static constexpr char kTruncatedSuffix[] = " details_truncated=1";
+      constexpr size_t kTruncatedSuffixBytes = sizeof(kTruncatedSuffix) - 1;
+      assert(available >= kTruncatedSuffixBytes);
+      AppendSanitizedRecordValue(details, available - kTruncatedSuffixBytes,
+                                 &operation_breadcrumb_record);
+      operation_breadcrumb_record.append(kTruncatedSuffix);
+    }
+  }
+
+  const size_t entry_count = (operation_breadcrumb_mapping->GetLen() -
+                              kOperationBreadcrumbHeaderBytes) /
+                             kOperationBreadcrumbRecordBytes;
+  char* const base =
+      static_cast<char*>(operation_breadcrumb_mapping->GetBase());
+  char* const slot = base + kOperationBreadcrumbHeaderBytes +
+                     operation_breadcrumb_pos * kOperationBreadcrumbRecordBytes;
+  BreadcrumbRecordState* const state =
+      std::launder(reinterpret_cast<BreadcrumbRecordState*>(slot));
+  // The acquire half prevents the following body stores from becoming visible
+  // before the slot is invalidated on weakly ordered architectures.
+  state->exchange('I', std::memory_order_acq_rel);
+  std::memset(slot + 1, ' ', kOperationBreadcrumbRecordBytes - 2);
+  slot[kOperationBreadcrumbRecordBytes - 1] = '\n';
+  std::memcpy(slot + 1, operation_breadcrumb_record.data(),
+              operation_breadcrumb_record.size());
+  state->store('V', std::memory_order_release);
+
+  operation_breadcrumb_pos = (operation_breadcrumb_pos + 1) % entry_count;
+}
+
+std::string* ThreadState::PrepareOperationEvent() {
+  if (!InitializeOperationBreadcrumbs()) {
+    return nullptr;
+  }
+  operation_breadcrumb_details.clear();
+  return &operation_breadcrumb_details;
+}
+
+void ThreadState::CommitOperationEvent(StressOperationType type) {
+  AppendOperationBreadcrumb(type, "event", operation_breadcrumb_details);
+}
+
+bool ThreadState::BeginOperation(StressOperationType type) {
+  ++operation_ordinal;
+  current_operation_ordinal = operation_ordinal;
+
+  if (OperationBreadcrumbsEnabled()) {
+    AppendOperationBreadcrumb(type, "begin", "");
+  }
+
+  if (LivenessTrackingEnabled()) {
+    return shared->BeginOperation(tid, type);
+  }
+  return false;
+}
+
+void ThreadState::RecordOperationEnd(StressOperationType type) {
+  if (!OperationBreadcrumbsEnabled()) {
+    return;
+  }
+  const bool verification_failed = shared->HasVerificationFailedYet();
+  AppendOperationBreadcrumb(
+      type, "end", verification_failed ? "verification_failure=1" : "");
+}
+
 bool SharedState::ShouldVerifyAtBeginning() const {
   return !stress_test_->GetExpectedValuesDir().empty();
 }

db_stress_tool/db_stress_shared_state.h

@@ -14,6 +14,8 @@
 #include <atomic>
 #include <cstddef>
 #include <cstdint>
+#include <memory>
+#include <string>
 
 #include "db_stress_tool/db_stress_stat.h"
 #include "db_stress_tool/expected_state.h"
@@ -37,6 +39,9 @@ DECLARE_bool(error_recovery_with_no_fault_injection);
 DECLARE_bool(sync_fault_injection);
 DECLARE_uint64(liveness_check_interval_sec);
 DECLARE_uint64(liveness_no_progress_timeout_sec);
+DECLARE_string(stress_diagnostics_dir);
+DECLARE_bool(stress_diagnostics_breadcrumbs);
+DECLARE_uint64(stress_diagnostics_breadcrumb_entries);
 DECLARE_int32(range_deletion_width);
 DECLARE_bool(disable_wal);
 DECLARE_int32(manual_wal_flush_one_in);
@@ -56,6 +61,7 @@ DECLARE_bool(disable_auto_compactions);
 DECLARE_bool(enable_compaction_filter);
 
 namespace ROCKSDB_NAMESPACE {
+class MemoryMappedFileBuffer;
 class StressTest;
 
 enum class StressOperationType : uint32_t {
@@ -284,6 +290,8 @@ class SharedState {
 
   bool HasVerificationFailedYet() const { return verification_failure_.load(); }
 
+  Env* GetEnv() const { return env_; }
+
   void IncFinishedOps() {
     finished_ops_.fetch_add(1, std::memory_order_relaxed);
   }
@@ -707,6 +715,16 @@ struct ThreadState {
   Random rand;   // Has different seeds for different threads
   SharedState* shared;
   Stats stats;
+  std::unique_ptr<MemoryMappedFileBuffer> operation_breadcrumb_mapping;
+  std::string operation_breadcrumb_details;
+  std::string operation_breadcrumb_record;
+  size_t operation_breadcrumb_pos;
+  uint64_t operation_breadcrumb_sequence;
+  uint64_t operation_breadcrumb_file_id;
+  bool operation_breadcrumb_setup_attempted;
+  bool diagnostic_io_disabled;
+  uint64_t operation_ordinal;
+  uint64_t current_operation_ordinal;
   struct SnapshotState {
     const Snapshot* snapshot;
     // The cf from which we did a Get at this snapshot
@@ -726,20 +744,15 @@ struct ThreadState {
   };
   std::queue<std::pair<uint64_t, SnapshotState>> snapshot_queue;
 
-  ThreadState(uint32_t index, SharedState* _shared)
-      : tid(index), rand(1000 + index + _shared->GetSeed()), shared(_shared) {}
+  ThreadState(uint32_t index, SharedState* _shared);
+  ~ThreadState();
 
   bool LivenessTrackingEnabled() const {
     return FLAGS_liveness_check_interval_sec > 0 &&
            FLAGS_liveness_no_progress_timeout_sec > 0;
   }
 
-  bool BeginOperation(StressOperationType type) {
-    if (LivenessTrackingEnabled()) {
-      return shared->BeginOperation(tid, type);
-    }
-    return false;
-  }
+  bool BeginOperation(StressOperationType type);
 
   bool EndOperation(StressOperationType type) {
     if (LivenessTrackingEnabled()) {
@@ -754,6 +767,14 @@ struct ThreadState {
     }
   }
 
+  bool OperationBreadcrumbsEnabled() const;
+
+  std::string* PrepareOperationEvent();
+
+  void CommitOperationEvent(StressOperationType type);
+
+  void RecordOperationEnd(StressOperationType type);
+
   void CompletedOpForDiagnostics(StressOperationType type) {
     if (LivenessTrackingEnabled()) {
       shared->IncCompletedOpForDiagnostics(type);
@@ -771,6 +792,11 @@ struct ThreadState {
     RecordSingleOpFinished();
     ClearOperation();
   }
+
+ private:
+  bool InitializeOperationBreadcrumbs();
+  void AppendOperationBreadcrumb(StressOperationType type, const char* phase,
+                                 const std::string& details);
 };
 }  // namespace ROCKSDB_NAMESPACE
 #endif  // GFLAGS

db_stress_tool/db_stress_test_base.cc

@@ -9,12 +9,15 @@
 //
 
 #include <algorithm>
+#include <charconv>
 #include <cstdlib>
+#include <initializer_list>
 #include <iomanip>
 #include <ios>
 #include <memory>
 #include <mutex>
 #include <optional>
+#include <sstream>
 #include <thread>
 #include <unordered_set>
 
@@ -78,6 +81,7 @@ class ScopedThreadOperation {
     if (tracking_) {
       thread_->CompletedOpForDiagnostics(type_);
     }
+    thread_->RecordOperationEnd(type_);
     switch (finish_action_) {
       case FinishAction::kPop:
         break;
@@ -97,6 +101,67 @@ class ScopedThreadOperation {
   bool tracking_;
 };
 
+template <typename Integer>
+void AppendOperationContextInteger(Integer value, std::string* output) {
+  char buffer[32];
+  const auto result = std::to_chars(buffer, buffer + sizeof(buffer), value);
+  assert(result.ec == std::errc());
+  output->append(buffer, result.ptr);
+}
+
+template <typename ColumnFamilies, typename Keys>
+void FormatOperationContext(const ColumnFamilies& column_families,
+                            const Keys& keys, std::string* output) {
+  output->clear();
+  output->append("cfs=[");
+  bool first = true;
+  for (int column_family : column_families) {
+    if (!first) {
+      output->push_back(',');
+    }
+    first = false;
+    AppendOperationContextInteger(column_family, output);
+  }
+  output->append("] keys=[");
+  first = true;
+  for (int64_t key : keys) {
+    if (!first) {
+      output->push_back(',');
+    }
+    first = false;
+    AppendOperationContextInteger(key, output);
+  }
+  output->push_back(']');
+}
+
+template <typename ColumnFamilies, typename Keys>
+void MaybeRecordOperationContextImpl(ThreadState* thread,
+                                     StressOperationType type,
+                                     const ColumnFamilies& column_families,
+                                     const Keys& keys) {
+  if (!thread->OperationBreadcrumbsEnabled()) {
+    return;
+  }
+  std::string* details = thread->PrepareOperationEvent();
+  if (details == nullptr) {
+    return;
+  }
+  FormatOperationContext(column_families, keys, details);
+  thread->CommitOperationEvent(type);
+}
+
+void MaybeRecordOperationContext(ThreadState* thread, StressOperationType type,
+                                 const std::vector<int>& column_families,
+                                 const std::vector<int64_t>& keys) {
+  MaybeRecordOperationContextImpl(thread, type, column_families, keys);
+}
+
+void MaybeRecordOperationContext(ThreadState* thread, StressOperationType type,
+                                 std::initializer_list<int> column_families,
+                                 std::initializer_list<int64_t> keys) {
+  MaybeRecordOperationContextImpl(thread, type, column_families, keys);
+}
+
 class StressReadScopedBlockBufferProvider
     : public ReadScopedBlockBufferProvider {
  public:
@@ -2359,6 +2424,8 @@ void StressTest::OperateDb(ThreadState* thread) {
 
       if (thread->rand.OneInOpt(FLAGS_compact_files_one_in)) {
         ScopedThreadOperation op(thread, StressOperationType::kCompactFiles);
+        MaybeRecordOperationContext(thread, StressOperationType::kCompactFiles,
+                                    {rand_column_family}, {});
         TestCompactFiles(thread, column_family);
       }
 
@@ -2368,6 +2435,8 @@ void StressTest::OperateDb(ThreadState* thread) {
 
       if (thread->rand.OneInOpt(FLAGS_compact_range_one_in)) {
         ScopedThreadOperation op(thread, StressOperationType::kCompactRange);
+        MaybeRecordOperationContext(thread, StressOperationType::kCompactRange,
+                                    {rand_column_family}, {rand_key});
         TestCompactRange(thread, rand_key, key, column_family);
         if (thread->shared->HasVerificationFailedYet()) {
           break;
@@ -2383,6 +2452,8 @@ void StressTest::OperateDb(ThreadState* thread) {
 
       if (thread->rand.OneInOpt(FLAGS_flush_one_in)) {
         ScopedThreadOperation op(thread, StressOperationType::kFlush);
+        MaybeRecordOperationContext(thread, StressOperationType::kFlush,
+                                    rand_column_families, {});
         TestFlush(thread, rand_column_families);
       }
 
@@ -2501,11 +2572,16 @@ void StressTest::OperateDb(ThreadState* thread) {
       if (thread->rand.OneInOpt(FLAGS_ingest_external_file_one_in)) {
         ScopedThreadOperation op(thread,
                                  StressOperationType::kIngestExternalFile);
+        MaybeRecordOperationContext(thread,
+                                    StressOperationType::kIngestExternalFile,
+                                    rand_column_families, rand_keys);
         TestIngestExternalFile(thread, rand_column_families, rand_keys);
       }
 
       if (thread->rand.OneInOpt(FLAGS_backup_one_in)) {
         ScopedThreadOperation op(thread, StressOperationType::kBackup);
+        MaybeRecordOperationContext(thread, StressOperationType::kBackup,
+                                    rand_column_families, rand_keys);
         // Beyond a certain DB size threshold, this test becomes heavier than
         // it's worth.
         uint64_t total_size = 0;
@@ -2533,18 +2609,25 @@ void StressTest::OperateDb(ThreadState* thread) {
 
       if (thread->rand.OneInOpt(FLAGS_checkpoint_one_in)) {
         ScopedThreadOperation op(thread, StressOperationType::kCheckpoint);
+        MaybeRecordOperationContext(thread, StressOperationType::kCheckpoint,
+                                    rand_column_families, rand_keys);
         Status s = TestCheckpoint(thread, rand_column_families, rand_keys);
         ProcessStatus(shared, "Checkpoint", s);
       }
 
       if (thread->rand.OneInOpt(FLAGS_approximate_size_one_in)) {
         ScopedThreadOperation op(thread, StressOperationType::kApproximateSize);
+        MaybeRecordOperationContext(thread,
+                                    StressOperationType::kApproximateSize,
+                                    rand_column_families, rand_keys);
         Status s =
             TestApproximateSize(thread, i, rand_column_families, rand_keys);
         ProcessStatus(shared, "ApproximateSize", s);
       }
       if (thread->rand.OneInOpt(FLAGS_acquire_snapshot_one_in)) {
         ScopedThreadOperation op(thread, StressOperationType::kSnapshot);
+        MaybeRecordOperationContext(thread, StressOperationType::kSnapshot,
+                                    {rand_column_family}, {rand_key});
         TestAcquireSnapshot(thread, rand_column_family, keystr, i);
       }
 
@@ -2565,6 +2648,8 @@ void StressTest::OperateDb(ThreadState* thread) {
 
       if (thread->rand.OneInOpt(FLAGS_key_may_exist_one_in)) {
         ScopedThreadOperation op(thread, StressOperationType::kKeyMayExist);
+        MaybeRecordOperationContext(thread, StressOperationType::kKeyMayExist,
+                                    rand_column_families, rand_keys);
         TestKeyMayExist(thread, read_opts, rand_column_families, rand_keys);
       }
       // Historical expected-state restore replays exactly
@@ -2592,12 +2677,16 @@ void StressTest::OperateDb(ThreadState* thread) {
           assert(i + batch_size <= ops_per_open);
 
           rand_keys = GenerateNKeys(thread, static_cast<int>(batch_size), i);
+          MaybeRecordOperationContext(thread, StressOperationType::kRead,
+                                      rand_column_families, rand_keys);
           ThreadStatusUtil::SetThreadOperation(
               ThreadStatus::OperationType::OP_MULTIGETENTITY);
           TestMultiGetEntity(thread, read_opts, rand_column_families,
                              rand_keys);
           i += batch_size - 1;
         } else if (FLAGS_use_get_entity) {
+          MaybeRecordOperationContext(thread, StressOperationType::kRead,
+                                      rand_column_families, rand_keys);
           ThreadStatusUtil::SetThreadOperation(
               ThreadStatus::OperationType::OP_GETENTITY);
           TestGetEntity(thread, read_opts, rand_column_families, rand_keys);
@@ -2611,11 +2700,15 @@ void StressTest::OperateDb(ThreadState* thread) {
           // If its the last iteration, ensure that multiget_batch_size is 1
           multiget_batch_size = std::max(multiget_batch_size, 1);
           rand_keys = GenerateNKeys(thread, multiget_batch_size, i);
+          MaybeRecordOperationContext(thread, StressOperationType::kRead,
+                                      rand_column_families, rand_keys);
           ThreadStatusUtil::SetThreadOperation(
               ThreadStatus::OperationType::OP_MULTIGET);
           TestMultiGet(thread, read_opts, rand_column_families, rand_keys);
           i += multiget_batch_size - 1;
         } else {
+          MaybeRecordOperationContext(thread, StressOperationType::kRead,
+                                      rand_column_families, rand_keys);
           ThreadStatusUtil::SetThreadOperation(
               ThreadStatus::OperationType::OP_GET);
           TestGet(thread, read_opts, rand_column_families, rand_keys);
@@ -2627,6 +2720,8 @@ void StressTest::OperateDb(ThreadState* thread) {
         ScopedThreadOperation op(
             thread, StressOperationType::kPrefixScan,
             ScopedThreadOperation::FinishAction::kFinishSingleOp);
+        MaybeRecordOperationContext(thread, StressOperationType::kPrefixScan,
+                                    rand_column_families, rand_keys);
         // keys are 8 bytes long, prefix size is FLAGS_prefix_size. There are
         // (8 - FLAGS_prefix_size) bytes besides the prefix. So there will
         // be 2 ^ ((8 - FLAGS_prefix_size) * 8) possible keys with the same
@@ -2638,6 +2733,8 @@ void StressTest::OperateDb(ThreadState* thread) {
         ScopedThreadOperation op(
             thread, StressOperationType::kWrite,
             ScopedThreadOperation::FinishAction::kFinishSingleOp);
+        MaybeRecordOperationContext(thread, StressOperationType::kWrite,
+                                    rand_column_families, rand_keys);
         if (disable_fault_injection_during_user_write) {
           db_fault_injection_fs_->DisableAllThreadLocalErrorInjection();
         }
@@ -2652,6 +2749,8 @@ void StressTest::OperateDb(ThreadState* thread) {
         ScopedThreadOperation op(
             thread, StressOperationType::kDelete,
             ScopedThreadOperation::FinishAction::kFinishSingleOp);
+        MaybeRecordOperationContext(thread, StressOperationType::kDelete,
+                                    rand_column_families, rand_keys);
         if (disable_fault_injection_during_user_write) {
           db_fault_injection_fs_->DisableAllThreadLocalErrorInjection();
         }
@@ -2665,6 +2764,8 @@ void StressTest::OperateDb(ThreadState* thread) {
         ScopedThreadOperation op(
             thread, StressOperationType::kDeleteRange,
             ScopedThreadOperation::FinishAction::kFinishSingleOp);
+        MaybeRecordOperationContext(thread, StressOperationType::kDeleteRange,
+                                    rand_column_families, rand_keys);
         if (disable_fault_injection_during_user_write) {
           db_fault_injection_fs_->DisableAllThreadLocalErrorInjection();
         }
@@ -2686,6 +2787,8 @@ void StressTest::OperateDb(ThreadState* thread) {
           // and an upper bound
           rand_keys = GenerateNKeys(thread, num_seeks * 2, i);
           i += num_seeks - 1;
+          MaybeRecordOperationContext(thread, StressOperationType::kIterate,
+                                      rand_column_families, rand_keys);
           ThreadStatusUtil::SetEnableTracking(FLAGS_enable_thread_tracking);
           ThreadStatusUtil::SetThreadOperation(
               ThreadStatus::OperationType::OP_DBITERATOR);
@@ -2697,6 +2800,8 @@ void StressTest::OperateDb(ThreadState* thread) {
                    thread->rand.OneInOpt(
                        FLAGS_verify_iterator_with_expected_state_one_in)) {
           ThreadStatusUtil::SetEnableTracking(FLAGS_enable_thread_tracking);
+          MaybeRecordOperationContext(thread, StressOperationType::kIterate,
+                                      rand_column_families, rand_keys);
           ThreadStatusUtil::SetThreadOperation(
               ThreadStatus::OperationType::OP_DBITERATOR);
           TestIterateAgainstExpected(thread, read_opts, rand_column_families,
@@ -2710,6 +2815,8 @@ void StressTest::OperateDb(ThreadState* thread) {
                        static_cast<uint64_t>(1))));
           rand_keys = GenerateNKeys(thread, num_seeks, i);
           i += num_seeks - 1;
+          MaybeRecordOperationContext(thread, StressOperationType::kIterate,
+                                      rand_column_families, rand_keys);
           ThreadStatusUtil::SetEnableTracking(FLAGS_enable_thread_tracking);
           ThreadStatusUtil::SetThreadOperation(
               ThreadStatus::OperationType::OP_DBITERATOR);
@@ -2729,6 +2836,8 @@ void StressTest::OperateDb(ThreadState* thread) {
         ScopedThreadOperation op(
             thread, StressOperationType::kCustom,
             ScopedThreadOperation::FinishAction::kFinishSingleOp);
+        MaybeRecordOperationContext(thread, StressOperationType::kCustom,
+                                    rand_column_families, rand_keys);
         TestCustomOperations(thread, rand_column_families);
       }
     }

tools/db_crashtest.py

@@ -2,6 +2,7 @@
 # Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved.
 
 import argparse
+import errno
 import glob
 import math
 import os
@@ -9,6 +10,7 @@ import random
 import re
 import shlex
 import shutil
+import stat
 import subprocess
 import sys
 import tempfile
@@ -52,6 +54,15 @@ 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"
+_STRESS_DIAGNOSTICS_DIR_NAME = "rocksdb_crashtest_diagnostics"
+_MAX_STRESS_DIAGNOSTIC_PROCESS_GROUPS = 8
+_STRESS_DIAGNOSTIC_FILE_RE = re.compile(
+    r"^[A-Za-z0-9_.-]*\.pid_([0-9]+)\.thread_[0-9]+(?:"
+    r"\.breadcrumbs\.txt(?:\.tmp)?|"
+    r"\."
+    r"(?:multi_scan|prefix_scan|prefix_scan_batched|"
+    r"prefix_scan_cf_consistency)\.witness\.txt(?:\.tmp)?)$"
+)
 _REMOTE_DB_URI_FLAGS = ("--env_uri", "--fs_uri")
 _MIN_WRITE_BUFFER_SIZE = 64 * 1024
 
@@ -260,6 +271,9 @@ default_params = {
     "db": "",
     "destroy_db_initially": 0,
     "expected_values_dir": "",
+    "stress_diagnostics_dir": "",
+    "stress_diagnostics_breadcrumbs": 1,
+    "stress_diagnostics_breadcrumb_entries": 256,
     "num_dbs": 1,
     "enable_pipelined_write": lambda: random.randint(0, 1),
     "enable_compaction_filter": lambda: random.choice([0, 0, 0, 1]),
@@ -692,6 +706,127 @@ def setup_multiops_txn_key_spaces_file():
     return multiops_txn_key_spaces_file
 
 
+diagnostics_dir_global = None
+
+
+def is_real_directory(path):
+    try:
+        return stat.S_ISDIR(os.lstat(path).st_mode)
+    except OSError:
+        return False
+
+
+def create_stress_diagnostics_dir(test_tmpdir):
+    if not test_tmpdir or not os.path.isdir(test_tmpdir):
+        return tempfile.mkdtemp(prefix=_STRESS_DIAGNOSTICS_DIR_NAME)
+
+    diagnostics_dir = os.path.join(test_tmpdir, _STRESS_DIAGNOSTICS_DIR_NAME)
+    if not os.path.lexists(diagnostics_dir):
+        os.mkdir(diagnostics_dir)
+        return diagnostics_dir
+    if not is_real_directory(diagnostics_dir):
+        raise OSError(
+            errno.ENOTDIR,
+            "stress diagnostics path is not a real directory",
+            diagnostics_dir,
+        )
+    return diagnostics_dir
+
+
+def get_diagnostics_dir():
+    # Diagnostics are local process artifacts, independent of whether the DB
+    # itself uses a local or remote filesystem.
+    global diagnostics_dir_global
+    if diagnostics_dir_global is not None:
+        return diagnostics_dir_global
+
+    try:
+        diagnostics_dir_global = create_stress_diagnostics_dir(
+            os.environ.get(_TEST_DIR_ENV_VAR)
+        )
+    except OSError as error:
+        print(f"Failed to create stress diagnostics directory: {error}")
+        diagnostics_dir_global = None
+        return ""
+    return diagnostics_dir_global
+
+
+def stress_diagnostics_enabled(params):
+    return params.get("stress_diagnostics_breadcrumbs", 0) == 1
+
+
+def set_default_stress_diagnostics_dir(params):
+    if stress_diagnostics_enabled(params) and not params.get("stress_diagnostics_dir"):
+        params["stress_diagnostics_dir"] = get_diagnostics_dir()
+
+
+def cleanup_stress_diagnostics_dir():
+    global diagnostics_dir_global
+    if diagnostics_dir_global is None:
+        return
+
+    try:
+        if is_real_directory(diagnostics_dir_global):
+            with os.scandir(diagnostics_dir_global) as entries:
+                for entry in entries:
+                    if entry.is_file(
+                        follow_symlinks=False
+                    ) and _STRESS_DIAGNOSTIC_FILE_RE.fullmatch(entry.name):
+                        os.remove(entry.path)
+            try:
+                os.rmdir(diagnostics_dir_global)
+            except OSError as error:
+                if error.errno not in (errno.ENOTEMPTY, errno.EEXIST):
+                    raise
+    except OSError as error:
+        print(
+            "Failed to clean up stress diagnostics directory "
+            f"{diagnostics_dir_global}: {error}"
+        )
+    finally:
+        diagnostics_dir_global = None
+
+
+def prune_stress_diagnostics_dir(current_pid):
+    if diagnostics_dir_global is None or not is_real_directory(
+        diagnostics_dir_global
+    ):
+        return
+
+    try:
+        process_files = {}
+        with os.scandir(diagnostics_dir_global) as entries:
+            for entry in entries:
+                if not entry.is_file(follow_symlinks=False):
+                    continue
+                match = _STRESS_DIAGNOSTIC_FILE_RE.fullmatch(entry.name)
+                if match is None:
+                    continue
+                pid = int(match.group(1))
+                process_files.setdefault(pid, []).append(
+                    (entry.stat(follow_symlinks=False).st_mtime_ns, entry.path)
+                )
+
+        if len(process_files) <= _MAX_STRESS_DIAGNOSTIC_PROCESS_GROUPS:
+            return
+
+        groups_by_age = sorted(
+            process_files.items(),
+            key=lambda item: max(mtime for mtime, _ in item[1]),
+        )
+        groups_to_remove = len(process_files) - _MAX_STRESS_DIAGNOSTIC_PROCESS_GROUPS
+        for pid, files in groups_by_age:
+            if groups_to_remove == 0:
+                break
+            if pid == current_pid:
+                continue
+            for _, path in files:
+                os.remove(path)
+            groups_to_remove -= 1
+    except OSError as error:
+        print(f"Failed to prune stress diagnostics directory: {error}")
+
+
 def is_direct_io_supported(dbname):
     if is_remote_db:
         return False
@@ -2137,6 +2272,7 @@ def diagnostic_paths(finalized_params):
     return [
         finalized_params.get("db"),
         finalized_params.get("expected_values_dir"),
+        finalized_params.get("stress_diagnostics_dir"),
     ]
 
 
@@ -2176,6 +2312,7 @@ def execute_cmd(cmd, timeout=None, timeout_pstack=False, expected_to_timeout=Tru
             print("KILLED %d (SIGTERM did not work)\n" % child.pid)
             outs, errs = child.communicate()
 
+    prune_stress_diagnostics_dir(pid)
     return (
         hit_timeout,
         child.returncode,
@@ -2333,6 +2470,7 @@ def liveness_main(args, unknown_args):
         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()
+    set_default_stress_diagnostics_dir(cmd_params)
 
     apply_random_seed_per_iteration()
     wrapper_timeout = liveness_timeout(cmd_params)
@@ -2386,6 +2524,7 @@ def blackbox_crash_main(args, unknown_args):
         cmd_params["db"] = db_parent_dir
     if not cmd_params.get("expected_values_dir"):
         cmd_params["expected_values_dir"] = ev_parent_dir
+    set_default_stress_diagnostics_dir(cmd_params)
 
     exit_time = time.time() + cmd_params["duration"]
 
@@ -2457,6 +2596,7 @@ def whitebox_crash_main(args, unknown_args):
         cmd_params["db"] = db_parent_dir
     if not cmd_params.get("expected_values_dir"):
         cmd_params["expected_values_dir"] = ev_parent_dir
+    set_default_stress_diagnostics_dir(cmd_params)
 
     cur_time = time.time()
     exit_time = cur_time + cmd_params["duration"]
@@ -2717,6 +2857,7 @@ def main():
             shutil.rmtree(ev_parent_dir_global)
     if multiops_txn_key_spaces_file is not None:
         os.remove(multiops_txn_key_spaces_file)
+    cleanup_stress_diagnostics_dir()
 
 
 if __name__ == "__main__":

tools/db_crashtest_test.py

@@ -91,6 +91,16 @@ class DBCrashTestTest(unittest.TestCase):
 
         self.addCleanup(cleanup_expected_values_dir)
 
+    def register_diagnostics_dir_cleanup(self, db_crashtest):
+        def cleanup_diagnostics_dir():
+            if db_crashtest.diagnostics_dir_global:
+                shutil.rmtree(
+                    db_crashtest.diagnostics_dir_global,
+                    ignore_errors=True,
+                )
+
+        self.addCleanup(cleanup_diagnostics_dir)
+
     def load_fault_injection_log_parser(self):
         return load_fault_injection_log_parser_module()
 
@@ -153,6 +163,213 @@ class DBCrashTestTest(unittest.TestCase):
             any(arg.startswith("--use_coro_db_api=") for arg in command)
         )
 
+    def test_stress_diagnostics_directory_creation_failure_is_non_fatal(self):
+        db_crashtest = self.load_db_crashtest()
+        diagnostics_path = os.path.join(
+            self.test_tmpdir, "rocksdb_crashtest_diagnostics"
+        )
+        with open(diagnostics_path, "w", encoding="utf-8"):
+            pass
+        params = {
+            "stress_diagnostics_breadcrumbs": 1,
+            "stress_diagnostics_dir": "",
+        }
+        output = io.StringIO()
+
+        with redirect_stdout(output):
+            db_crashtest.set_default_stress_diagnostics_dir(params)
+
+        self.assertEqual("", params["stress_diagnostics_dir"])
+        self.assertIsNone(db_crashtest.diagnostics_dir_global)
+        self.assertIn(
+            "Failed to create stress diagnostics directory", output.getvalue()
+        )
+
+    def test_stress_diagnostics_cleanup_failure_is_non_fatal(self):
+        db_crashtest = self.load_db_crashtest()
+        db_crashtest.get_diagnostics_dir()
+        output = io.StringIO()
+
+        with mock.patch.object(
+            db_crashtest.os,
+            "rmdir",
+            side_effect=OSError("injected cleanup failure"),
+        ):
+            with redirect_stdout(output):
+                db_crashtest.cleanup_stress_diagnostics_dir()
+
+        self.assertIsNone(db_crashtest.diagnostics_dir_global)
+        self.assertIn(
+            "Failed to clean up stress diagnostics directory", output.getvalue()
+        )
+
+    def test_stress_diagnostics_cleanup_preserves_unrelated_files(self):
+        db_crashtest = self.load_db_crashtest()
+        diagnostics_dir = db_crashtest.get_diagnostics_dir()
+        unrelated_path = os.path.join(diagnostics_dir, "keep-me.txt")
+        whitespace_path = os.path.join(
+            diagnostics_dir, "keep me.pid_1.thread_0.breadcrumbs.txt"
+        )
+        unicode_path = os.path.join(
+            diagnostics_dir, "keep_me_\N{SNOWMAN}.pid_1.thread_0.breadcrumbs.txt"
+        )
+        diagnostic_path = os.path.join(
+            diagnostics_dir, "db.pid_1.thread_0.breadcrumbs.txt"
+        )
+        witness_path = os.path.join(
+            diagnostics_dir,
+            "db.pid_1.thread_0.prefix_scan_batched.witness.txt",
+        )
+        witness_temp_path = witness_path + ".tmp"
+        for path in (unrelated_path, whitespace_path, unicode_path):
+            with open(path, "w", encoding="utf-8"):
+                pass
+        for path in (diagnostic_path, witness_path, witness_temp_path):
+            with open(path, "w", encoding="utf-8"):
+                pass
+
+        db_crashtest.cleanup_stress_diagnostics_dir()
+
+        self.assertTrue(os.path.exists(unrelated_path))
+        self.assertTrue(os.path.exists(whitespace_path))
+        self.assertTrue(os.path.exists(unicode_path))
+        self.assertFalse(os.path.exists(diagnostic_path))
+        self.assertFalse(os.path.exists(witness_path))
+        self.assertFalse(os.path.exists(witness_temp_path))
+        self.assertIsNone(db_crashtest.diagnostics_dir_global)
+
+    def test_stress_diagnostics_rejects_symlinked_auto_directory(self):
+        db_crashtest = self.load_db_crashtest()
+        diagnostics_path = os.path.join(
+            self.test_tmpdir, "rocksdb_crashtest_diagnostics"
+        )
+        target_dir = os.path.join(self.test_tmpdir, "unrelated")
+        os.mkdir(target_dir)
+        target_file = os.path.join(
+            target_dir, "db.pid_1.thread_0.breadcrumbs.txt"
+        )
+        with open(target_file, "w", encoding="utf-8"):
+            pass
+        os.symlink(target_dir, diagnostics_path)
+        params = {
+            "stress_diagnostics_breadcrumbs": 1,
+            "stress_diagnostics_dir": "",
+        }
+
+        db_crashtest.set_default_stress_diagnostics_dir(params)
+        db_crashtest.prune_stress_diagnostics_dir(current_pid=1)
+        db_crashtest.cleanup_stress_diagnostics_dir()
+
+        self.assertEqual("", params["stress_diagnostics_dir"])
+        self.assertTrue(os.path.exists(target_file))
+        self.assertIsNone(db_crashtest.diagnostics_dir_global)
+
+    def test_stress_diagnostics_pruning_keeps_recent_processes(self):
+        db_crashtest = self.load_db_crashtest()
+        diagnostics_dir = db_crashtest.get_diagnostics_dir()
+        self.register_diagnostics_dir_cleanup(db_crashtest)
+        unrelated_path = os.path.join(diagnostics_dir, "keep-me.txt")
+        misleading_path = os.path.join(diagnostics_dir, "notes.pid_1.txt")
+        with open(unrelated_path, "w", encoding="utf-8"):
+            pass
+        with open(misleading_path, "w", encoding="utf-8"):
+            pass
+
+        for pid in range(10):
+            path = os.path.join(
+                diagnostics_dir, f"db.pid_{pid}.thread_0.breadcrumbs.txt"
+            )
+            with open(path, "w", encoding="utf-8"):
+                pass
+            os.utime(path, ns=(pid + 1, pid + 1))
+
+        db_crashtest.prune_stress_diagnostics_dir(current_pid=9)
+
+        self.assertTrue(os.path.exists(unrelated_path))
+        self.assertTrue(os.path.exists(misleading_path))
+        self.assertFalse(
+            os.path.exists(
+                os.path.join(
+                    diagnostics_dir, "db.pid_0.thread_0.breadcrumbs.txt"
+                )
+            )
+        )
+        self.assertFalse(
+            os.path.exists(
+                os.path.join(
+                    diagnostics_dir, "db.pid_1.thread_0.breadcrumbs.txt"
+                )
+            )
+        )
+        for pid in range(2, 10):
+            self.assertTrue(
+                os.path.exists(
+                    os.path.join(
+                        diagnostics_dir, f"db.pid_{pid}.thread_0.breadcrumbs.txt"
+                    )
+                )
+            )
+
+    def test_stress_diagnostics_pruning_ignores_explicit_directory(self):
+        db_crashtest = self.load_db_crashtest()
+        explicit_dir = os.path.join(self.test_tmpdir, "explicit_diagnostics")
+        os.mkdir(explicit_dir)
+        path = os.path.join(explicit_dir, "db.pid_1.thread_0.breadcrumbs.txt")
+        with open(path, "w", encoding="utf-8"):
+            pass
+
+        db_crashtest.prune_stress_diagnostics_dir(current_pid=1)
+
+        self.assertTrue(os.path.exists(path))
+
+    def test_stress_diagnostics_defaults_to_local_artifact_dir(self):
+        db_crashtest = self.load_db_crashtest()
+        self.register_diagnostics_dir_cleanup(db_crashtest)
+        params = db_crashtest.gen_cmd_params(self.build_mode_args("blackbox"))
+        params["db"] = self.test_tmpdir
+
+        db_crashtest.set_default_stress_diagnostics_dir(params)
+        command, finalized = db_crashtest.gen_cmd(params, [])
+
+        self.assertEqual(1, finalized["stress_diagnostics_breadcrumbs"])
+        self.assertEqual(
+            os.path.join(self.test_tmpdir, "rocksdb_crashtest_diagnostics"),
+            finalized["stress_diagnostics_dir"],
+        )
+        self.assertTrue(os.path.isdir(finalized["stress_diagnostics_dir"]))
+        self.assertIn("--stress_diagnostics_breadcrumbs=1", command)
+        self.assertIn(
+            "--stress_diagnostics_dir=" + finalized["stress_diagnostics_dir"],
+            command,
+        )
+
+    def test_stress_diagnostics_disabled_does_not_allocate_dir(self):
+        db_crashtest = self.load_db_crashtest()
+        params = {
+            "stress_diagnostics_breadcrumbs": 0,
+            "stress_diagnostics_dir": "",
+        }
+
+        db_crashtest.set_default_stress_diagnostics_dir(params)
+
+        self.assertEqual("", params["stress_diagnostics_dir"])
+        self.assertIsNone(db_crashtest.diagnostics_dir_global)
+
+    def test_stress_diagnostics_paths_include_artifact_dir(self):
+        db_crashtest = self.load_db_crashtest()
+        diagnostics_dir = os.path.join(self.test_tmpdir, "diagnostics")
+
+        self.assertEqual(
+            ["/db", "/expected", diagnostics_dir],
+            db_crashtest.diagnostic_paths(
+                {
+                    "db": "/db",
+                    "expected_values_dir": "/expected",
+                    "stress_diagnostics_dir": diagnostics_dir,
+                }
+            ),
+        )
+
     def test_cache_and_write_buffer_size_multiplier_preserves_randomization(self):
         db_crashtest = self.load_db_crashtest()
         cache_sizes = iter([8 * 1024 * 1024, 32 * 1024 * 1024])
@@ -748,6 +965,7 @@ class DBCrashTestTest(unittest.TestCase):
         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)
+        self.register_diagnostics_dir_cleanup(db_crashtest)
         execute_calls = []
 
         def fake_execute_cmd(