Fix concurrency, overflow, and validation issues in reference-chain tracking

Addresses review findings from PR #644 (reference chains for surviving
live-heap samples): FrontierTable's shared-lock mutation race, a
single-byte JFR event size prefix that silently truncates chains longer
than 255 bytes, blocking sample-lock retries in writeReferenceChain()
now bounded by a shared per-batch deadline with a drop counter,
unvalidated referencechains sub-option values, a startThread()/
pthread_kill() race publishing _running before the thread handle is
initialized, a null-JNIEnv leak in threadLoop(), releaseSearchTags()
now surfacing GetObjectsWithTags() failures so restartSearch() never
resets tag state prematurely, resolveLoadedClasses() skipping its
per-class scan only when the loaded-class count is unchanged (not just
non-decreasing), a spurious COMPLETED state after a failed first-pass
FollowReferences call, and removal of a leftover debug helper in
ExternalProcessReferenceChainTest. Adds regression test coverage for
release-failure and negative-value option paths.

Verified via the full ddprof-lib gtestDebug suite (149/149 tasks,
57/57 referenceChains_ut tests).

Environment: Datadog workspace

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: jbachorik

ddprof-lib/src/main/cpp/arguments.cpp

@@ -18,6 +18,7 @@
 #include "arguments.h"
 #include "vmEntry.h"
 
+#include <algorithm>
 #include <errno.h>
 #include <limits.h>
 #include <stdio.h>
@@ -473,18 +474,42 @@ Error Arguments::parse(const char *args) {
           char *eq = strchr(cursor, '=');
           if (eq) {
             *(eq++) = 0;
+            // Floor every sub-option at the parse boundary rather than
+            // trusting a downstream cast/clamp to make an operator-supplied
+            // negative value safe: a negative hops value in particular gets
+            // compared as `depth >= (u32)ctx->hop_cap` (referenceChains.cpp),
+            // so an unclamped negative wraps to ~4e9 and silently disables
+            // the hop cap entirely - the opposite of the flag's intent, and
+            // it removes the one guard that otherwise bounds how long a
+            // single reference chain (and therefore its
+            // datadog.ReferenceChain JFR event) can grow. A negative budget
+            // similarly collapses ReferenceChainTracker::_effective_budget
+            // to 0 (updatePacing()'s own PID-clamp logic), which truncates
+            // every pass immediately and leaves the search RUNNING
+            // (re-walking the whole graph each cadence) until TTL instead of
+            // making progress. A negative framecap is handed straight to
+            // FrontierTable's constructor, which floors it to a
+            // zero-capacity table (that class's own std::max(max_cap, 0)),
+            // silently disabling tracking rather than erroring. ttl/
+            // pausetarget/painbudget already have incidental downstream
+            // clamps (runPass()'s `_ttl_ms > 0` gate, this class's own
+            // PidController/PainBudget std::max(..., 0) calls) but are
+            // floored here too so every sub-option's validation lives at one
+            // boundary instead of being split between here and several
+            // unrelated call sites.
             if (strcasecmp(cursor, "hops") == 0) {
-              _reference_chains_hop_cap = atoi(eq);
+              _reference_chains_hop_cap = std::max(atoi(eq), 1);
             } else if (strcasecmp(cursor, "budget") == 0) {
-              _reference_chains_budget = atoi(eq);
+              _reference_chains_budget = std::max(atoi(eq), 1);
             } else if (strcasecmp(cursor, "ttl") == 0) {
-              _reference_chains_ttl_ms = atol(eq);
+              _reference_chains_ttl_ms = std::max(atol(eq), 0L);
             } else if (strcasecmp(cursor, "framecap") == 0) {
-              _reference_chains_frontier_cap = atoi(eq);
+              _reference_chains_frontier_cap = std::max(atoi(eq), 1);
             } else if (strcasecmp(cursor, "pausetarget") == 0) {
-              _reference_chains_pause_target_ms = atol(eq);
+              _reference_chains_pause_target_ms = std::max(atol(eq), 0L);
             } else if (strcasecmp(cursor, "painbudget") == 0) {
-              _reference_chains_pain_budget_percent = atoi(eq);
+              _reference_chains_pain_budget_percent =
+                  std::min(std::max(atoi(eq), 0), 100);
             }
           }
           cursor = next;

ddprof-lib/src/main/cpp/counters.h

@@ -129,7 +129,18 @@
    * any dump() drained it - permanently lost, since _emitted_target_tags     \
    * marks its tag as emitted the moment it is queued, not when it is         \
    * actually written. See MAX_PENDING_CHAIN_EVENTS' own comment. */          \
-  X(REFERENCE_CHAIN_EVENTS_DROPPED, "reference_chain_events_dropped")
+  X(REFERENCE_CHAIN_EVENTS_DROPPED, "reference_chain_events_dropped")         \
+  /* ReferenceChainTracker::releaseSearchTags() (referenceChains.cpp) failed  \
+   * to call GetObjectsWithTags() for at least one batch - the search's tag   \
+   * release is retried on a later call rather than proceeding, but this     \
+   * counts how often that retry path is taken. */                           \
+  X(REFERENCE_CHAIN_TAG_RELEASE_FAILED, "reference_chain_tag_release_failed") \
+  /* Profiler::writeReferenceChain() (profiler.cpp) could not acquire a       \
+   * sample-record lock within its bounded retry budget and dropped the      \
+   * already-dequeued datadog.ReferenceChain event - permanently lost, same  \
+   * as REFERENCE_CHAIN_EVENTS_DROPPED above but from the write side rather  \
+   * than the pending-queue side. */                                         \
+  X(REFERENCE_CHAIN_WRITE_DROPPED, "reference_chain_write_dropped")
 #define X_ENUM(a, b) a,
 typedef enum CounterId : int {
   DD_COUNTER_TABLE(X_ENUM) DD_NUM_COUNTERS

ddprof-lib/src/main/cpp/flightRecorder.cpp

@@ -1949,19 +1949,43 @@ void Recording::recordHeapLiveObject(Buffer *buf, int tid, u64 call_trace_id,
 }
 
 void Recording::recordReferenceChain(Buffer *buf, ReferenceChainEvent *event) {
-  int start = buf->skip(1);
+  // event->_chain's length is bounded only by FrontierTable::maxCapacity()
+  // (tens of thousands of entries, referenceChains.h) - NOT by
+  // MAX_JFR_EVENT_SIZE, so this event cannot use writeEventSizePrefix()'s
+  // single-byte size field (its assert(size < MAX_JFR_EVENT_SIZE) is
+  // compiled out in release builds, making an oversize chain a silent
+  // corrupt size byte rather than a caught bug) nor rely on the trailing
+  // flushIfNeeded(buf) every fixed-size event above uses (that only flushes
+  // *after* already writing past the buffer). Truncate to
+  // MAX_REFERENCE_CHAIN_EVENT_HOPS (that constant's own comment) and
+  // reserve room for the truncated worst case up front instead.
+  u32 chain_size = (u32)event->_chain.size();
+  u32 emitted_size = chain_size < (u32)MAX_REFERENCE_CHAIN_EVENT_HOPS
+                          ? chain_size
+                          : (u32)MAX_REFERENCE_CHAIN_EVENT_HOPS;
+
+  flushIfNeeded(
+      buf, RECORDING_BUFFER_LIMIT -
+               (MAX_VAR32_LENGTH /* multi-byte size prefix, below */ +
+                3 * MAX_VAR64_LENGTH /* type id, start_time, target_tag */ +
+                2 * MAX_VAR32_LENGTH /* depth, chain count */ +
+                (int)emitted_size * MAX_VAR32_LENGTH));
+  // Multi-byte size prefix (like writeDatadogSetting() above), not
+  // writeEventSizePrefix()'s single byte - this event's size can exceed
+  // MAX_JFR_EVENT_SIZE (255) once the chain is more than a few dozen hops.
+  int start = buf->skip(MAX_VAR32_LENGTH);
   buf->putVar64(T_REFERENCE_CHAIN);
   buf->putVar64(event->_start_time);
   buf->putVar64(event->_target_tag);
   buf->putVar32(event->_depth);
   // T_CLASS array field (F_CPOOL|F_ARRAY, jfrMetadata.cpp) - each entry is a
   // StringDictionary class id, same encoding as a scalar objectClass field
   // (e.g. recordAllocation() above), just repeated `count` times.
-  buf->putVar32((u32)event->_chain.size());
-  for (u32 klass_id : event->_chain) {
-    buf->putVar32(klass_id);
+  buf->putVar32(emitted_size);
+  for (u32 i = 0; i < emitted_size; i++) {
+    buf->putVar32(event->_chain[i]);
   }
-  writeEventSizePrefix(buf, start);
+  buf->putVar32(start, (u32)(buf->offset() - start));
   flushIfNeeded(buf);
 }
 

ddprof-lib/src/main/cpp/flightRecorder.h

@@ -40,6 +40,20 @@ const int JFR_EVENT_FLUSH_THRESHOLD = RECORDING_BUFFER_LIMIT;
 const int MAX_VAR64_LENGTH = 10;
 const int MAX_VAR32_LENGTH = 5;
 
+// Chain length Recording::recordReferenceChain() (flightRecorder.cpp) will
+// actually serialize per datadog.ReferenceChain event, independent of
+// ReferenceChainTracker's own _hop_cap/frontier-table cap - the frontier
+// table's own defensive walk bound (FrontierTable::reconstructChain(),
+// referenceChains.h) is maxCapacity(), which can run into the tens of
+// thousands of entries, and neither that cap nor _hop_cap is itself
+// range-validated against a buffer-safe maximum (see arguments.cpp's own
+// sub-option parsing). recordReferenceChain() truncates event->_chain to
+// this many entries before writing, so its own worst-case size never
+// depends on trusting either of those upstream caps to stay small - a chain
+// longer than this is still truncated defense-in-depth even if a caller
+// changes those caps later.
+const int MAX_REFERENCE_CHAIN_EVENT_HOPS = 4096;
+
 #ifndef CONCURRENCY_LEVEL
 const int CONCURRENCY_LEVEL = 16;
 #endif

ddprof-lib/src/main/cpp/profiler.cpp

@@ -809,37 +809,54 @@ void Profiler::writeReferenceChainAbandoned(ReferenceChainAbandonedEvent *event)
 // Unlike writeReferenceChainAbandoned() above (mirroring CPU/wall's signal-handler-safe
 // non-blocking pattern out of caution, even though its own call site - Profiler::dump(),
 // profiler.cpp - isn't a signal handler either), this call site genuinely cannot be one:
-// pollWatchedTargets() (referenceChains.cpp) only ever runs on ReferenceChainTracker's own
-// agent thread, never from a signal handler. A single bare 3-slot tryLock() sweep with no
-// wait - correct for a signal handler, which must never block - was found, by running
-// PROF-15341's end-to-end integration test
+// this is called from Profiler::dump()'s drain loop, on dump()'s own calling thread, once
+// per event drained from ReferenceChainTracker::_pending_chain_events (up to
+// MAX_PENDING_CHAIN_EVENTS per dump) - never from pollWatchedTargets() or any other call on
+// ReferenceChainTracker's own BFS agent thread, and never from a signal handler. A single
+// bare 3-slot tryLock() sweep with no wait - correct for a signal handler, which must never
+// block - was found, by running PROF-15341's end-to-end integration test
 // (ddprof-test's ReferenceChainTrackingTest.shouldReconstructReferrerChainToGcRoot) for real,
 // to drop this event under perfectly ordinary contention: the same _locks[] pool is shared
 // with every other sample type (recordJVMTISample() et al.), and any nontrivial allocation
 // throughput keeps enough of CONCURRENCY_LEVEL's slots busy that 3 immediate, back-to-back
 // attempts routinely all miss. A bounded retry with a short sleep between sweeps costs
-// nothing this thread cannot afford (it already sleeps for its own cadence, 10ms-4s between
-// passes, referenceChains.h's _effective_cadence_ns) and fixes the drop without touching the
-// signal-handler-safe call sites that must stay non-blocking.
-void Profiler::writeReferenceChain(ReferenceChainEvent *event) {
+// nothing the dump()-thread cannot afford, but the retry budget below is a single deadline
+// shared across the *entire* drain batch (see the caller in dump()) rather than per event:
+// with up to MAX_PENDING_CHAIN_EVENTS events queued, a fresh per-event budget could stall
+// the dump/JFR-flush thread for seconds under contention. Once the shared deadline has
+// passed this degrades to the same single non-blocking 3-slot sweep as
+// writeReferenceChainAbandoned() above for the remainder of the batch.
+void Profiler::writeReferenceChain(ReferenceChainEvent *event, u64 deadline_ns) {
   int tid = ProfiledThread::currentTid();
   if (tid < 0) {
     return;
   }
-  const int kMaxLockAttempts = 50; // ~50ms worst case at 1ms/attempt - see comment above
   u32 lock_index;
   bool locked = false;
-  for (int attempt = 0; attempt < kMaxLockAttempts; attempt++) {
+  for (;;) {
     lock_index = getLockIndex(tid);
     if (_locks[lock_index].tryLock() ||
         _locks[lock_index = (lock_index + 1) % CONCURRENCY_LEVEL].tryLock() ||
         _locks[lock_index = (lock_index + 2) % CONCURRENCY_LEVEL].tryLock()) {
       locked = true;
       break;
     }
+    if (OS::nanotime() >= deadline_ns) {
+      // Shared batch budget exhausted - the sweep just above was already a
+      // single non-blocking attempt, so stop retrying rather than sleeping
+      // again.
+      break;
+    }
     usleep(1000);
   }
   if (!locked) {
+    // Already dequeued from _pending_chain_events and its tag already
+    // recorded in _emitted_target_tags (drainPendingChainEvents(), before
+    // this call) - pollWatchedTargets() will never regenerate this event, so
+    // the drop is permanent. Count it like every other counted-drop path
+    // (REFERENCE_CHAIN_EVENTS_DROPPED's own comment) rather than dropping it
+    // silently.
+    Counters::increment(REFERENCE_CHAIN_WRITE_DROPPED);
     return;
   }
   _jfr.recordReferenceChain(lock_index, event);
@@ -1787,8 +1804,15 @@ Error Profiler::dump(const char *path, const int length) {
     std::vector<ReferenceChainEvent> pending_chain_events;
     ReferenceChainTracker::instance()->drainPendingChainEvents(
         &pending_chain_events);
+    // One ~50ms retry budget for the *whole* batch, not per event -
+    // writeReferenceChain()'s own comment for why: up to
+    // MAX_PENDING_CHAIN_EVENTS events can be queued, and a fresh per-event
+    // budget would let this dump()-thread stall for seconds under ordinary
+    // _locks[] contention.
+    const u64 kChainDrainBudgetNs = 50 * 1000000ULL;
+    u64 chain_drain_deadline_ns = OS::nanotime() + kChainDrainBudgetNs;
     for (auto &rc_event : pending_chain_events) {
-      writeReferenceChain(&rc_event);
+      writeReferenceChain(&rc_event, chain_drain_deadline_ns);
     }
 
     Libraries::instance()->refresh();

ddprof-lib/src/main/cpp/profiler.h

@@ -433,16 +433,15 @@ class alignas(alignof(SpinLock)) Profiler {
   // the same way LivenessTracker::flush() is called from dump().
   void writeReferenceChainAbandoned(ReferenceChainAbandonedEvent *event);
   // Unlike writeReferenceChainAbandoned() above, this is NOT a bare 3-slot
-  // tryLock() sweep - it retries with a bounded, sleeping loop (up to ~50ms
-  // worst case) because its call site, ReferenceChainTracker::
-  // pollWatchedTargets() (referenceChains.cpp), runs only on that tracker's
-  // own agent thread and can tolerate blocking, unlike a signal handler; see
-  // this method's own comment in profiler.cpp for why that retry exists.
-  // Called for each chain event discovered this poll cycle, rather than from
-  // dump() - see FlightRecorder::recordReferenceChain()'s own comment for why
-  // chain events need their own call site instead of piggybacking on dump()'s
-  // flush-on-dump pattern.
-  void writeReferenceChain(ReferenceChainEvent *event);
+  // tryLock() sweep - it retries with a bounded, sleeping loop because its
+  // call site is dump()'s drain loop (profiler.cpp), on dump()'s own calling
+  // thread, which can tolerate blocking, unlike a signal handler; see this
+  // method's own comment in profiler.cpp for why that retry exists.
+  // `deadline_ns` is a single retry budget shared across dump()'s *entire*
+  // drain batch (not reset per event) - see the caller in dump() and this
+  // method's own comment in profiler.cpp for why a per-event budget would be
+  // unbounded across a large batch.
+  void writeReferenceChain(ReferenceChainEvent *event, u64 deadline_ns);
   int eventMask() const { return _event_mask; }
   bool isRemoteSymbolication() const { return _remote_symbolication; }