fix(profiling): clear finished physical span links (#19529)

## Description

This is the first PR in the span-attribution stack. It fixes stale endpoint and span attribution for physical thread stacks before the later PRs introduce logical asyncio and gevent targets.

### Problem

The stack profiler samples threads independently of application execution, so it keeps a native per-thread copy of the span metadata needed to label profile samples. That copy must follow the lifetime of the tracing context. Otherwise, work that runs after a span finishes can still be labeled with the finished span ID, local-root span ID, and endpoint.

Context activation events are necessary, but they are not sufficient to keep these copies accurate. Consider this sequence:

1. A worker thread activates a span and publishes a native span link.
2. Another thread finishes that span.
3. The worker continues running without performing another context activation.

Finishing the span does not produce an activation event on the worker. Without a separate finish-time cleanup path, its copied link remains stale and subsequent samples are incorrectly attributed to the finished trace.

A local root can also finish before one of its children. In that case the child is still valid and can continue producing profile samples, so cleanup for the root must not remove mappings whose current span is that unfinished child.

### Solution

This PR makes the native `ThreadSpanLinks` store own both directions of each physical association under one mutex:

- thread ID to copied span metadata;
- current span ID to every thread carrying a copy of that exact span.

It then uses two complementary lifecycle signals:

- `ddtrace.context_provider.activate` handles immediate state replacement on the current thread. Activating a span or propagated `Context` publishes its mapping, restoring a parent replaces it, and activating `None` or a `Context` without a span ID clears it.
- `trace.span_finish` performs global reverse invalidation for the exact span that finished. It removes every copied physical-thread mapping for that span, including mappings published on another thread that may never receive another activation. It deliberately preserves mappings for unfinished children when their local root finishes first.

Updating the forward and reverse indexes atomically prevents the two views from diverging. Re-linking a thread removes its previous reverse-index entries first, so a later finish event for an older span cannot erase the thread's newer attribution.

Listener registration occurs only after the collector completes fallible initialization, and both listeners are removed during shutdown. Fork handling reconstructs the native mutex and inherited indexes without traversing potentially inconsistent parent state.

### Review follow-up

Native lifecycle bookkeeping now records in-flight physical span publications before releasing the GIL. If a span finishes while publication is waiting on the native mutex, cleanup is deferred until the final publication completes. This preserves the GIL release introduced in #14852 to avoid thread-pool deadlocks.

## Testing

Added coverage for:

- immediate physical-link clearing on context deactivation;
- finished child-span cleanup across physical threads;
- cross-thread finish cleanup before the linked worker performs another activation;
- preservation of active child links when the local root finishes first;
- reverse-index replacement when a thread advances to a new span;
- nested child-to-parent restoration;
- cleanup on reused thread-pool workers.

Validation completed on the reordered physical-only PR:

- Python 3.13 profiling suite in normal and fast-copy modes: `359 passed, 35 skipped, 2 xfailed` per mode
- Python 3.13 focused cross-thread finish regression: `1 passed` in each mode
- Python 3.11 profiling suite after hardening the reused-worker assertion: `309 passed, 85 skipped, 2 xfailed` in both normal and fast-copy modes
- Reused-worker regression repeated 11 times in both modes: `22 passed`
- Control experiment with the span-finish listener disabled: the cross-thread regression failed in both modes because post-finish samples retained the endpoint, confirming activation alone is insufficient
- `scripts/lint checks`
- targeted style and typing checks
- `scripts/lint cformat`
- `scripts/lint profiling-native-check`
- rebased onto current `main` at `3efdcbe3af`
- `git diff --check`

## Risks

- Physical activation updates perform expected O(1) forward and reverse-index mutations under the existing span-link mutex.
- Finished-span cleanup visits only threads indexed by the exact span that finished.
- The reverse index adds bounded memory proportional to currently linked physical threads.
- The rollback is to disable stack profiling or revert this PR.

## Additional Notes

- Next, common logical stack helpers: #19420
- Gevent integration: #19379
- Asyncio integration: #19346
- Includes a customer-facing profiling fix release note.

Before

<img width="1732" height="905" alt="Screenshot 2026-08-06 at 10 15 34 AM" src="https://github.com/user-attachments/assets/d6a69ad3-1324-4328-b241-c3dc2fdf7cd5" />

After

<img width="1749" height="908" alt="Screenshot 2026-08-06 at 10 15 24 AM" src="https://github.com/user-attachments/assets/cef79a9e-ddc9-4299-b3ff-492126e13ad2" />








Co-authored-by: taegyun.kim <taegyun.kim@datadoghq.com>

Author: taegyunkim

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

@@ -35,6 +35,12 @@ def link_span(span: typing.Optional[typing.Union[context.Context, ddspan.Span]])
         elif isinstance(span, context.Context) and span.span_id is not None:
             local_root_span_id, span_type = context_meta.read_profiler_link(span)
             _stack.link_span(span.span_id, local_root_span_id, span_type)
+        else:
+            _stack.clear_span()
+
+    def _unlink_finished_span(span: ddspan.Span) -> None:
+        """Remove physical-thread attribution derived from a finished span."""
+        _stack.unlink_finished_span(span.span_id)
 
     def link_origin_task(task_id: int, task_name: str) -> None:
         """

ddtrace/internal/datadog/profiling/stack/__init__.pyi

@@ -80,6 +80,7 @@ def resume_sampling() -> None: ...
 
 # span <-> profile association
 def link_span(span: Optional[Union[context.Context, ddspan.Span]]) -> None: ...
+def _unlink_finished_span(span: ddspan.Span) -> None: ...
 
 # Thread management
 def register_thread(python_thread_id: int, native_id: int, name: str) -> None: ...

ddtrace/internal/datadog/profiling/stack/_stack.pyi

@@ -41,6 +41,7 @@ def link_span(
 ) -> None: ...
 def unlink_span(expected_span_id: int) -> None: ...
 def clear_span() -> None: ...
+def unlink_finished_span(span_id: int) -> None: ...
 
 # executor worker thread <-> originating asyncio task association
 def link_origin_task(

ddtrace/internal/datadog/profiling/stack/include/thread_span_links.hpp

@@ -1,11 +1,13 @@
 #pragma once
 
-#include <memory>
+#include <cstddef>
 #include <mutex>
 #include <optional>
 #include <stdint.h>
 #include <string>
 #include <unordered_map>
+#include <unordered_set>
+#include <utility>
 
 namespace Datadog {
 
@@ -47,13 +49,35 @@ class ThreadSpanLinks
     const std::optional<Span> get_active_span_from_thread_id(uint64_t thread_id);
     void unlink_span(uint64_t thread_id);
     void unlink_span(uint64_t thread_id, uint64_t expected_span_id);
+    void unlink_finished_span(uint64_t span_id);
     void reset();
 
+    // These lifecycle methods run with the GIL held, before or after a native map mutation that releases it.
+    void on_link_start(uint64_t span_id);
+    bool on_link_end(uint64_t span_id);
+    bool on_span_finish(uint64_t span_id);
+
     static void postfork_child();
 
   private:
+    using ThreadIdSet = std::unordered_set<uint64_t>;
+    using ThreadIdToSpanMap = std::unordered_map<uint64_t, Span>;
+    using SpanToThreadMap = std::unordered_map<uint64_t, ThreadIdSet>;
+
+    struct PendingSpanLink
+    {
+        size_t count = 0;
+        bool finished = false;
+    };
+
+    void remove_thread_locked(uint64_t thread_id);
+
     std::mutex mtx;
-    std::unordered_map<uint64_t, std::unique_ptr<Span>> thread_id_to_span;
+    ThreadIdToSpanMap thread_id_to_span;
+    SpanToThreadMap span_to_threads;
+
+    // Protected by the GIL. This bridges lifecycle callback order to mutations that release it above.
+    std::unordered_map<uint64_t, PendingSpanLink> pending_span_links;
 
     // Private Constructor/Destructor
     ThreadSpanLinks() = default;

ddtrace/internal/datadog/profiling/stack/src/stack.cpp

@@ -162,10 +162,19 @@ stack_link_span_impl(PyObject* self, PyObject* args, PyObject* kwargs)
         span_type = empty_string.c_str();
     }
 
+    auto& links = ThreadSpanLinks::get_instance();
+    links.on_link_start(span_id);
+
     Py_BEGIN_ALLOW_THREADS;
-    ThreadSpanLinks::get_instance().link_span(thread_id, span_id, local_root_span_id, std::string(span_type));
+    links.link_span(thread_id, span_id, local_root_span_id, std::string(span_type));
     Py_END_ALLOW_THREADS;
 
+    if (links.on_link_end(span_id)) {
+        Py_BEGIN_ALLOW_THREADS;
+        links.unlink_finished_span(span_id);
+        Py_END_ALLOW_THREADS;
+    }
+
     Py_RETURN_NONE;
 }
 
@@ -214,6 +223,26 @@ stack_clear_span(PyObject* self, PyObject* args)
     Py_RETURN_NONE;
 }
 
+static PyObject*
+stack_unlink_finished_span(PyObject* self, PyObject* args)
+{
+    (void)self;
+    uint64_t span_id;
+
+    if (!PyArg_ParseTuple(args, "K", &span_id)) {
+        return nullptr;
+    }
+
+    auto& links = ThreadSpanLinks::get_instance();
+    if (links.on_span_finish(span_id)) {
+        Py_BEGIN_ALLOW_THREADS;
+        links.unlink_finished_span(span_id);
+        Py_END_ALLOW_THREADS;
+    }
+
+    Py_RETURN_NONE;
+}
+
 // Records the asyncio task that offloaded work to the current (worker) thread.
 // The thread id is derived from the calling thread's state (this runs on the
 // worker thread), matching how stack_link_span_impl resolves it.
@@ -1010,6 +1039,10 @@ static PyMethodDef stack_methods[] = {
       METH_VARARGS,
       "Clear the span linked to the current thread if its ID matches the expected span ID" },
     { "clear_span", stack_clear_span, METH_NOARGS, "Clear the span linked to the current thread" },
+    { "unlink_finished_span",
+      stack_unlink_finished_span,
+      METH_VARARGS,
+      "Clear every physical-thread link derived from a finished span" },
     { "link_origin_task",
       reinterpret_cast<PyCFunction>(stack_link_origin_task),
       METH_VARARGS | METH_KEYWORDS,

ddtrace/internal/datadog/profiling/stack/src/thread_span_links.cpp

@@ -4,42 +4,59 @@
 #include <optional>
 #include <stdint.h>
 #include <string>
+#include <utility>
 
 namespace Datadog {
+
+void
+ThreadSpanLinks::remove_thread_locked(uint64_t thread_id)
+{
+    auto thread_it = thread_id_to_span.find(thread_id);
+    if (thread_it == thread_id_to_span.end()) {
+        return;
+    }
+
+    const auto& span = thread_it->second;
+    auto span_it = span_to_threads.find(span.span_id);
+    if (span_it != span_to_threads.end()) {
+        span_it->second.erase(thread_id);
+        if (span_it->second.empty()) {
+            span_to_threads.erase(span_it);
+        }
+    }
+    thread_id_to_span.erase(thread_it);
+}
+
 void
 ThreadSpanLinks::link_span(uint64_t thread_id, uint64_t span_id, uint64_t local_root_span_id, std::string span_type)
 {
     std::lock_guard<std::mutex> lock(mtx);
 
-    auto it = thread_id_to_span.find(thread_id);
-    if (it == thread_id_to_span.end()) {
-        thread_id_to_span[thread_id] = std::make_unique<Span>(span_id, local_root_span_id, std::move(span_type));
-    } else {
-        it->second->span_id = span_id;
-        it->second->local_root_span_id = local_root_span_id;
-        it->second->span_type = std::move(span_type);
-    }
+    remove_thread_locked(thread_id);
+    Span span(span_id, local_root_span_id, std::move(span_type));
+    thread_id_to_span.try_emplace(thread_id, std::move(span));
+    // Index only the current span. A local root can finish before an active child, and finishing it must not remove the
+    // child's attribution before that child finishes.
+    span_to_threads[span_id].insert(thread_id);
 }
 
 const std::optional<Span>
 ThreadSpanLinks::get_active_span_from_thread_id(uint64_t thread_id)
 {
     std::lock_guard<std::mutex> lock(mtx);
 
-    std::optional<Span> span;
     auto it = thread_id_to_span.find(thread_id);
-    if (it != thread_id_to_span.end()) {
-        span = *(it->second);
+    if (it == thread_id_to_span.end()) {
+        return std::nullopt;
     }
-    return span;
+    return it->second;
 }
 
 void
 ThreadSpanLinks::unlink_span(uint64_t thread_id)
 {
     std::lock_guard<std::mutex> lock(mtx);
-
-    thread_id_to_span.erase(thread_id); // This is a no-op if the key is not found
+    remove_thread_locked(thread_id);
 }
 
 void
@@ -48,16 +65,64 @@ ThreadSpanLinks::unlink_span(uint64_t thread_id, uint64_t expected_span_id)
     std::lock_guard<std::mutex> lock(mtx);
 
     auto it = thread_id_to_span.find(thread_id);
-    if (it != thread_id_to_span.end() && it->second->span_id == expected_span_id) {
-        thread_id_to_span.erase(it);
+    if (it != thread_id_to_span.end() && it->second.span_id == expected_span_id) {
+        remove_thread_locked(thread_id);
+    }
+}
+
+void
+ThreadSpanLinks::unlink_finished_span(uint64_t span_id)
+{
+    std::lock_guard<std::mutex> lock(mtx);
+
+    auto span_it = span_to_threads.find(span_id);
+    if (span_it == span_to_threads.end()) {
+        return;
     }
+
+    for (const auto thread_id : span_it->second) {
+        thread_id_to_span.erase(thread_id);
+    }
+    span_to_threads.erase(span_it);
 }
 
 void
 ThreadSpanLinks::reset()
 {
     std::lock_guard<std::mutex> lock(mtx);
     thread_id_to_span.clear();
+    span_to_threads.clear();
+}
+
+void
+ThreadSpanLinks::on_link_start(uint64_t span_id)
+{
+    ++pending_span_links[span_id].count;
+}
+
+bool
+ThreadSpanLinks::on_link_end(uint64_t span_id)
+{
+    auto pending_span = pending_span_links.find(span_id);
+    if (--pending_span->second.count != 0) {
+        return false;
+    }
+
+    const bool finished = pending_span->second.finished;
+    pending_span_links.erase(pending_span);
+    return finished;
+}
+
+bool
+ThreadSpanLinks::on_span_finish(uint64_t span_id)
+{
+    auto pending_span = pending_span_links.find(span_id);
+    if (pending_span == pending_span_links.end()) {
+        return true;
+    }
+
+    pending_span->second.finished = true;
+    return false;
 }
 
 void
@@ -66,16 +131,14 @@ ThreadSpanLinks::postfork_child()
     auto& instance = get_instance();
     // NB placement-new to re-init and leak the mutex because doing anything else is UB
     new (&instance.mtx) std::mutex();
-    // thread_id_to_span may be in a mid-mutation state if fork raced with
-    // link_span/unlink_span. Its inherited pointers may be inconsistent,
-    // so calling clear (or letting the destructor run) would traverse the same
-    // corrupted linked-list state, which is UB. Instead, reconstruct the map in
-    // place with placement-new without inspecting its contents. This intentionally
-    // leaks the old map's heap allocations (nodes/buckets), but that memory belonged
-    // to the parent's address space snapshot and is a bounded, one-time leak per
-    // fork in the child; freeing it safely is impossible given the possible
-    // corruption, so leaking is the correct trade-off.
-    new (&instance.thread_id_to_span) std::unordered_map<uint64_t, std::unique_ptr<Span>>();
+    // Either map may be in a mid-mutation state if fork raced with a link or unlink. Its inherited pointers may be
+    // inconsistent, so calling clear (or letting the destructor run) would traverse corrupted linked-list state,
+    // which is UB. Reconstruct the maps in place without inspecting their contents. This intentionally leaks the old
+    // maps' heap allocations, because their possibly corrupted pointers cannot be safely traversed or freed in the
+    // child.
+    new (&instance.thread_id_to_span) std::unordered_map<uint64_t, Span>();
+    new (&instance.span_to_threads) SpanToThreadMap();
+    new (&instance.pending_span_links) std::unordered_map<uint64_t, PendingSpanLink>();
 }
 
 } // namespace Datadog

ddtrace/internal/datadog/profiling/stack/test/test_thread_span_links.cpp

@@ -100,6 +100,59 @@ TEST(ThreadSpanLinks, ClearFinished)
     }
 }
 
+TEST(ThreadSpanLinks, UnlinkFinishedSpanAcrossThreads)
+{
+    auto& links = Datadog::ThreadSpanLinks::get_instance();
+    links.reset();
+
+    links.link_span(100, 1000, 1000, "web");
+    links.link_span(101, 1001, 1000, "web");
+    links.link_span(102, 1001, 1000, "web");
+    links.link_span(103, 2001, 2000, "worker");
+
+    links.unlink_finished_span(1000);
+    EXPECT_EQ(links.get_active_span_from_thread_id(100), std::nullopt);
+    EXPECT_EQ(links.get_active_span_from_thread_id(101), Datadog::Span(1001, 1000, "web"));
+    EXPECT_EQ(links.get_active_span_from_thread_id(102), Datadog::Span(1001, 1000, "web"));
+    EXPECT_EQ(links.get_active_span_from_thread_id(103), Datadog::Span(2001, 2000, "worker"));
+
+    links.unlink_finished_span(1001);
+    EXPECT_EQ(links.get_active_span_from_thread_id(101), std::nullopt);
+    EXPECT_EQ(links.get_active_span_from_thread_id(102), std::nullopt);
+    EXPECT_EQ(links.get_active_span_from_thread_id(103), Datadog::Span(2001, 2000, "worker"));
+}
+
+TEST(ThreadSpanLinks, RelinkUpdatesFinishedSpanIndex)
+{
+    auto& links = Datadog::ThreadSpanLinks::get_instance();
+    links.reset();
+
+    links.link_span(201, 3001, 3000, "old");
+    links.link_span(201, 4001, 4000, "new");
+
+    links.unlink_finished_span(3000);
+    EXPECT_EQ(links.get_active_span_from_thread_id(201), Datadog::Span(4001, 4000, "new"));
+    links.unlink_finished_span(4001);
+    EXPECT_EQ(links.get_active_span_from_thread_id(201), std::nullopt);
+}
+
+TEST(ThreadSpanLinks, FinishingWhileLinkingDefersCleanup)
+{
+    auto& links = Datadog::ThreadSpanLinks::get_instance();
+    constexpr uint64_t thread_id = 301;
+    constexpr uint64_t span_id = 3001;
+    links.reset();
+
+    links.on_link_start(span_id);
+    EXPECT_FALSE(links.on_span_finish(span_id));
+
+    links.link_span(thread_id, span_id, span_id, "web");
+    EXPECT_TRUE(links.on_link_end(span_id));
+
+    links.unlink_finished_span(span_id);
+    EXPECT_EQ(links.get_active_span_from_thread_id(thread_id), std::nullopt);
+}
+
 int
 main(int argc, char** argv)
 {