test(profiling): prove native-heap ownership handoff in CI via gotter test-hook

Add a deterministic, cluster-independent test that proves the Phase 2
producer-side ownership handoff in a single process, replacing the flaky
staging A/B dedup signal. With the gotter armed and the size partition on, a
>512B managed allocation must be owned by exactly one producer: dropped by the
in-process _memalloc sampler AND captured by the native gotter (its
process-global hook-hit counter advances). A <=512B control stays pool-served
and is still sampled in-process.

The proof uses the gotter's built-in test-support hook-hit counter
(ddtrace_heap_gotter_test_hook_hits), which increments on every intercepted raw
malloc/free (not sampling-gated), so no eBPF/Full-Host attach is needed:

- heap_gotter activator: add a defensive, test-only test_hook_hits() that binds
  the counter symbol if present and returns None otherwise (absent cdylib,
  non-Linux, or a non-test-support build).
- setup.py: add env-gated DD_PROFILING_NATIVE_HEAP_TEST_SUPPORT=1 to build the
  cdylib with --features test-support. Additive, never set for shipped wheels.
- The test is @subprocess (install() patches the GOT permanently; the partition
  flag is process-global) and skips unless a Linux 64-bit test-support gotter
  build is loaded. It therefore skips in the standard CI wheel (no gotter) and
  runs in a dedicated test-support build.

Author: vlad-scherbich

ddtrace/internal/datadog/profiling/heap_gotter/__init__.py

@@ -53,6 +53,12 @@
 # Stays False when the cdylib is absent or was built allocation-only.
 _live_heap_available: bool = False
 
+# Whether the loaded cdylib exports the test-only hook-hit counter symbol
+# (``ddtrace_heap_gotter_test_hook_hits``). Only true for a ``test-support``
+# build (never a shipped wheel). Lets deterministic CI tests prove the patched
+# GOT actually ran without a live eBPF attach; see ``test_hook_hits`` below.
+_test_hook_available: bool = False
+
 _lib: ctypes.CDLL | None = None  # kept alive for process lifetime; never dlclose'd
 
 
@@ -95,6 +101,16 @@ def _library_path() -> str:
     except AttributeError:
         _live_heap_available = False
 
+    # Bind the test-only hook-hit counter defensively: it exists ONLY on a
+    # `test-support` build (never a shipped wheel). A missing symbol simply
+    # leaves the counter reported as unavailable rather than failing the load.
+    try:
+        _lib.ddtrace_heap_gotter_test_hook_hits.argtypes = []
+        _lib.ddtrace_heap_gotter_test_hook_hits.restype = ctypes.c_uint64
+        _test_hook_available = True
+    except AttributeError:
+        _test_hook_available = False
+
 except Exception as e:
     failure_msg = str(e)
     _lib = None
@@ -134,3 +150,26 @@ def live_heap_enabled() -> bool:
     does not change over the process lifetime.
     """
     return _live_heap_available
+
+
+def test_hook_hits() -> int | None:
+    """Test-only: number of times the patched GOT hooks have run in this process.
+
+    Returns the process-global ``gotter_malloc``/``gotter_free`` hit counter,
+    which increments on every intercepted raw (glibc) ``malloc``/``free`` — it is
+    NOT sampling-gated — so a deterministic single-process test can prove the
+    native gotter actually captured the raw-domain allocations that the
+    in-process ``_memalloc`` sampler dropped under the ownership partition,
+    without needing a live eBPF/Full-Host attach.
+
+    Returns ``None`` when the counter is unavailable: on a non-Linux platform,
+    when the cdylib is absent, or (the common CI case) when the shipped cdylib
+    was NOT built with the ``test-support`` cargo feature. Tests must treat
+    ``None`` as "skip: no test-support gotter build".
+    """
+    if not is_available or _lib is None or not _test_hook_available:
+        return None
+    try:
+        return int(_lib.ddtrace_heap_gotter_test_hook_hits())
+    except Exception:
+        return None

setup.py

@@ -138,6 +138,19 @@
 # staging A/B harness sets this to bake the artifact into its custom wheels;
 # runtime install is separately gated by DD_PROFILING_NATIVE_HEAP_ENABLED.
 BUILD_NATIVE_HEAP_GOTTER: bool = os.getenv("DD_PROFILING_NATIVE_HEAP_BUILD", "0").lower() in ("1", "yes", "on", "true")
+# Opt-in: also compile the gotter cdylib with the `test-support` cargo feature so
+# it exports `ddtrace_heap_gotter_test_hook_hits()` (a process-global hook-hit
+# counter). This lets a deterministic, cluster-independent CI test prove the
+# producer-side ownership handoff (the native gotter actually captures the raw
+# glibc-malloc tail the in-process sampler drops) without a live eBPF attach.
+# Never set for shipped wheels — it is strictly a test build knob and only has
+# any effect when BUILD_NATIVE_HEAP_GOTTER is also on.
+BUILD_NATIVE_HEAP_GOTTER_TEST_SUPPORT = os.getenv("DD_PROFILING_NATIVE_HEAP_TEST_SUPPORT", "0").lower() in (
+    "1",
+    "yes",
+    "on",
+    "true",
+)
 
 CURRENT_OS = platform.system()
 SERVERLESS_BUILD = os.getenv("DD_SERVERLESS_BUILD", "0").lower() in ("1", "yes", "on", "true")
@@ -1062,6 +1075,14 @@ def build_heap_gotter(self) -> None:
         # `live-heap` is a default cargo feature (see Cargo.toml), so the built
         # cdylib always emits both the `ddheap:alloc` and `ddheap:free` USDTs
         # (verifiable via `readelf -n`) and stamps per-allocation retain flags.
+        #
+        # Test builds may also opt into `test-support`, which forwards to the
+        # upstream crate and exports `ddtrace_heap_gotter_test_hook_hits()` so a
+        # deterministic single-process test can prove the patched GOT actually
+        # ran. This is additive (default features stay on) and is never enabled
+        # for shipped wheels.
+        if BUILD_NATIVE_HEAP_GOTTER_TEST_SUPPORT:
+            cargo_cmd += ["--features", "test-support"]
         proc: subprocess.CompletedProcess[str] = subprocess.run(
             cargo_cmd, check=True, stdout=subprocess.PIPE, text=True
         )

tests/profiling/test_native_heap_gotter.py

@@ -19,6 +19,22 @@
 import pytest
 
 
+# Evaluated in the PARENT interpreter (subprocess bodies cannot express a skip:
+# an in-body ``pytest.skip`` would surface as a non-zero exit and FAIL the outer
+# test). ``test_hook_hits()`` is a read-only counter query with no side effects —
+# it does NOT install the gotter — so it is safe to call at import time. It
+# returns ``None`` unless the loaded cdylib was built with the ``test-support``
+# cargo feature (Linux 64-bit, ``DD_PROFILING_NATIVE_HEAP_TEST_SUPPORT=1``),
+# which the standard CI wheel is not, so the end-to-end handoff proof below skips
+# everywhere except a dedicated test-support build.
+try:
+    from ddtrace.internal.datadog.profiling import heap_gotter as _heap_gotter
+
+    _GOTTER_TEST_HOOK_AVAILABLE: bool = _heap_gotter.test_hook_hits() is not None
+except Exception:
+    _GOTTER_TEST_HOOK_AVAILABLE = False
+
+
 @pytest.mark.skipif(sys.platform != "linux", reason="native heap gotter is Linux-only")
 @pytest.mark.subprocess
 def test_native_heap_gotter_smoke() -> None:
@@ -228,3 +244,129 @@ def test_profiler_keeps_managed_heap_when_native_heap_disabled() -> None:
                 set_partition.assert_called_once_with(False)
             finally:
                 prof.stop(flush=False)
+
+
+# ---------------------------------------------------------------------------
+# End-to-end producer-side ownership handoff (Phase 2 de-dup)
+#
+# The tests above prove the *wiring* (arming turns the partition on) and
+# tests/profiling/collector/test_memalloc.py proves the *in-process* half of the
+# partition (> 512B managed allocations are dropped, <= 512B kept, and with the
+# partition off everything is sampled). What neither can prove without a live
+# eBPF/Full-Host attach is the *other* half of the handoff: that the native
+# gotter actually captures the > 512B raw glibc-malloc tail the in-process
+# sampler drops — i.e. that exactly one producer owns each allocation.
+#
+# The test below closes that gap deterministically in a single process using the
+# gotter's built-in ``test-support`` hook-hit counter, replacing the flaky
+# staging A/B dedup signal with an in-CI assertion. It requires a Linux 64-bit
+# ``test-support`` gotter build (see the module-level skip note); it skips in the
+# standard CI wheel, which ships no gotter at all.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.skipif(
+    sys.platform != "linux" or not _GOTTER_TEST_HOOK_AVAILABLE,
+    reason=(
+        "needs a Linux 64-bit test-support gotter build exposing "
+        "ddtrace_heap_gotter_test_hook_hits() (build with "
+        "DD_PROFILING_NATIVE_HEAP_BUILD=1 DD_PROFILING_NATIVE_HEAP_TEST_SUPPORT=1); "
+        "the standard CI wheel ships no gotter"
+    ),
+)
+@pytest.mark.subprocess
+def test_native_heap_ownership_handoff_end_to_end() -> None:
+    """Deterministic, cluster-independent proof of the Phase 2 ownership handoff.
+
+    With the gotter armed and the producer-side size partition on, a > 512B
+    managed OBJ allocation must be owned by *exactly one* producer:
+
+      (a) it is NOT sampled by the in-process ``_memalloc`` heap profiler (the
+          partition drops the > 512B tail), AND
+      (b) it IS seen by the native gotter — the process-global hook-hit counter
+          advances by at least one per large allocation, proving the patched GOT
+          captured the raw glibc ``malloc`` the in-process sampler dropped.
+
+    A <= 512B control allocation stays pymalloc-pool-served and is still sampled
+    in-process, confirming the partition splits by size rather than dropping
+    everything. Runs in a subprocess because ``install()`` patches the process
+    GOT permanently and the partition flag is process-global.
+    """
+    import os
+    import tempfile
+
+    from ddtrace.internal.datadog.profiling import ddup
+    from ddtrace.internal.datadog.profiling import heap_gotter
+    from ddtrace.profiling.collector import memalloc
+    from tests.profiling.collector import pprof_utils
+    from tests.profiling.collector.test_memalloc import _PARTITION_LARGE_ALLOC_COUNT
+    from tests.profiling.collector.test_memalloc import _allocate_large_buffers
+    from tests.profiling.collector.test_memalloc import _allocate_small_objects
+    from tests.profiling.collector.test_memalloc import _count_heap_samples_with_function
+
+    # Defensive: the module-level skipif already gated on these, but assert so a
+    # mis-configured skip can never let this test pass vacuously.
+    assert heap_gotter.is_available, "test requires the gotter cdylib to be present"
+    assert heap_gotter.test_hook_hits() is not None, "test requires a test-support gotter build"
+
+    # Arm the native producer (permanent + process-global; hence @subprocess).
+    assert heap_gotter.install() is True
+    assert heap_gotter.is_installed() is True
+
+    prefix = os.path.join(tempfile.mkdtemp(), "handoff")
+    output_filename = prefix + "." + str(os.getpid())
+    ddup.config(
+        service="test_native_heap_ownership_handoff",
+        version="test",
+        env="test",
+        output_filename=prefix,
+    )
+    ddup.start()
+
+    store: list[object] = []
+    mc = memalloc.MemoryCollector(heap_sample_size=64 * 1024)
+    memalloc.set_native_heap_partition(True)
+    try:
+        with mc:
+            # Measure the native counter strictly around the > 512B allocations.
+            # The counter is process-global and increments on EVERY intercepted
+            # raw malloc (it is NOT sampling-gated), so background allocations
+            # can only inflate the delta — never shrink it below the number of
+            # large buffers we deliberately allocate.
+            hits_before = heap_gotter.test_hook_hits()
+            _allocate_large_buffers(store)
+            hits_after = heap_gotter.test_hook_hits()
+
+            _allocate_small_objects(store)
+            mc.snapshot()
+        ddup.upload()
+
+        profile = pprof_utils.parse_newest_profile(output_filename)
+        heap_samples = pprof_utils.get_samples_with_value_type(profile, "heap-space")
+
+        # (a) In-process producer dropped the > 512B tail ...
+        large_count = _count_heap_samples_with_function(profile, heap_samples, "_allocate_large_buffers")
+        assert large_count == 0, (
+            f"partition ON: > 512B managed allocations must NOT be sampled in-process (got {large_count})"
+        )
+
+        # (b) ... and the native producer captured it. Each > 512B bytes object
+        # is a single raw malloc routed through the patched GOT, so the hook-hit
+        # counter must advance by at least the number of large buffers.
+        assert hits_before is not None and hits_after is not None
+        delta = hits_after - hits_before
+        assert delta >= _PARTITION_LARGE_ALLOC_COUNT, (
+            "native gotter must capture the > 512B raw-malloc tail the in-process sampler dropped "
+            f"(hook-hit delta {delta} < {_PARTITION_LARGE_ALLOC_COUNT} large allocations)"
+        )
+
+        # Control: <= 512B pool-served allocations are invisible to the gotter
+        # and must still be sampled in-process — the partition splits by size.
+        small_count = _count_heap_samples_with_function(profile, heap_samples, "_allocate_small_objects")
+        assert small_count > 0, "partition ON: <= 512B managed allocations must still be sampled in-process"
+    finally:
+        # Reset the process-global flag so it cannot bleed into other tests
+        # sharing this interpreter (belt-and-braces; the subprocess exits anyway).
+        memalloc.set_native_heap_partition(False)
+
+    del store