fix(profiling): reject inconsistent Python 3.14+ asyncio task lists

ddtrace/internal/datadog/profiling/stack/echion/echion/threads.h

@@ -33,6 +33,7 @@
 #include <echion/timing.h>
 
 class EchionSampler;
+class ThreadInfoTaskTraversalTest;
 
 class ThreadInfo
 {
@@ -119,6 +120,8 @@ class ThreadInfo
     };
 
   private:
+    friend class ThreadInfoTaskTraversalTest;
+
     void reset_cycle_state() noexcept;
     void render_unwound_stacks(EchionSampler&);
     [[nodiscard]] Result<void> unwind_tasks(EchionSampler&, PyThreadState*, microsecond_t wall_time_us);

ddtrace/internal/datadog/profiling/stack/src/echion/threads.cc

@@ -453,57 +453,55 @@ ThreadInfo::get_tasks_from_linked_list(EchionSampler& echion, uintptr_t head_add
         return ErrorKind::TaskInfoError;
     }
 
-    // Copy head node struct from remote memory to local memory
-    struct llist_node head_node_local;
-    if (copy_type(reinterpret_cast<void*>(head_addr), head_node_local)) {
+    const size_t tasks_start = tasks.size();
+    // This traversal only appends to tasks. On structural failure, remove its partial results while preserving entries
+    // from earlier sources.
+    auto fail = [&tasks, tasks_start]() -> Result<void> {
+        tasks.resize(tasks_start);
         return ErrorKind::TaskInfoError;
-    }
+    };
 
-    // Check if list is empty (head points to itself in circular list)
-    uintptr_t head_addr_uint = head_addr;
-    uintptr_t next_as_uint = reinterpret_cast<uintptr_t>(head_node_local.next);
-    uintptr_t prev_as_uint = reinterpret_cast<uintptr_t>(head_node_local.prev);
-    if (next_as_uint == head_addr_uint && prev_as_uint == head_addr_uint) {
-        return Result<void>::ok();
+    struct llist_node head_node;
+    if (copy_type(reinterpret_cast<void*>(head_addr), head_node)) {
+        return fail();
     }
+    llist_node current_node = head_node;
 
-    struct llist_node current_node = head_node_local; // Start with head node
-
-    // Copied from CPython's _remote_debugging_module.c: MAX_ITERATIONS
-    const size_t MAX_ITERATIONS = 1 << 16;
+    constexpr size_t max_iterations = 1 << 16;
     size_t iteration_count = 0;
+    uintptr_t current_node_addr = head_addr;
+    std::unordered_set<uintptr_t> visited;
 
-    // Iterate over linked-list. The linked list is circular, so we stop
-    // when we're back at head.
-    while (reinterpret_cast<uintptr_t>(current_node.next) != head_addr_uint) {
-        // Safety: prevent infinite loops
-        if (++iteration_count > MAX_ITERATIONS) {
-            return ErrorKind::TaskInfoError;
+    // A valid circular list must return to the expected head within the hard bound without null, repeated, unreadable,
+    // or backward-inconsistent nodes. Any violation rolls back this source.
+    while (reinterpret_cast<uintptr_t>(current_node.next) != head_addr) {
+        if (++iteration_count > max_iterations || current_node.next == nullptr) {
+            return fail();
         }
 
-        if (current_node.next == nullptr) {
-            return ErrorKind::TaskInfoError; // nullptr pointer - invalid list
+        const uintptr_t next_node_addr = reinterpret_cast<uintptr_t>(current_node.next);
+        if (!visited.insert(next_node_addr).second) {
+            return fail();
         }
 
-        uintptr_t next_node_addr = reinterpret_cast<uintptr_t>(current_node.next);
-
-        // Calculate task_addr from current_node.next
-        size_t task_node_offset_val = offsetof(TaskObj, task_node);
-        uintptr_t task_addr_uint = next_node_addr - task_node_offset_val;
-
-        // Create TaskInfo for the task
-        auto maybe_task_info = TaskInfo::create(echion, reinterpret_cast<TaskObj*>(task_addr_uint));
-        if (maybe_task_info) {
-            auto& task_info = *maybe_task_info;
-            if (task_info->loop == reinterpret_cast<PyObject*>(this->asyncio_loop)) {
-                tasks.push_back(std::move(task_info));
-            }
+        struct llist_node next_node;
+        if (copy_type(reinterpret_cast<void*>(next_node_addr), next_node) ||
+            reinterpret_cast<uintptr_t>(next_node.prev) != current_node_addr) {
+            return fail();
         }
 
-        // Read next node from current_node.next into current_node
-        if (copy_type(reinterpret_cast<void*>(next_node_addr), current_node)) {
-            return ErrorKind::TaskInfoError; // Failed to read next node
+        const uintptr_t task_addr = next_node_addr - offsetof(TaskObj, task_node);
+        auto maybe_task = TaskInfo::create(echion, reinterpret_cast<TaskObj*>(task_addr));
+        if (maybe_task && (*maybe_task)->loop == reinterpret_cast<PyObject*>(this->asyncio_loop)) {
+            tasks.push_back(std::move(*maybe_task));
         }
+
+        current_node_addr = next_node_addr;
+        current_node = next_node;
+    }
+
+    if (reinterpret_cast<uintptr_t>(head_node.prev) != current_node_addr) {
+        return fail();
     }
 
     return Result<void>::ok();
@@ -516,24 +514,19 @@ ThreadInfo::get_all_tasks(EchionSampler& echion, PyThreadState* tstate)
     if (this->asyncio_loop == 0)
         return tasks;
 
-    // Python 3.14+: Native tasks are in linked-list per thread AND per interpreter
-    // CPython iterates over both:
-    // 1. Per-thread list: tstate->asyncio_tasks_head (active tasks)
-    // 2. Per-interpreter list: interp->asyncio_tasks_head (lingering tasks)
-    // First, get tasks from this thread's linked-list (if tstate_addr is set)
-    // Note: We continue processing even if one source fails to maximize partial results
+    // Python 3.14+ task discovery combines four sources:
+    // - per-thread linked lists for active native Tasks;
+    // - the per-interpreter linked list for native Tasks surviving thread-state clearing;
+    // - _scheduled_tasks for third-party Task implementations;
+    // - _eager_tasks for Tasks executing their first eager step.
+    // The stack sampler reads Python threads without acquiring the GIL or stopping them. It can therefore observe a
+    // Task moving between sources, so deduplicate tasks by address below.
+    // Continue after one source fails to preserve results from the other sources.
     if (tstate != nullptr && this->tstate_addr != 0) {
         (void)get_tasks_from_thread_linked_list(echion, tasks);
-
-        // Second, get tasks from interpreter's linked-list (lingering tasks)
         (void)get_tasks_from_interpreter_linked_list(echion, tstate, tasks);
     }
 
-    // Handle third-party tasks from Python _scheduled_tasks WeakSet
-    // In Python 3.14+, _scheduled_tasks is a Python-level weakref.WeakSet() that only contains
-    // tasks that don't inherit from asyncio.Task. Native asyncio.Task instances are stored
-    // in linked-lists (handled above) and are NOT added to _scheduled_tasks.
-    // This is typically empty in practice, but we handle it for completeness.
     auto asyncio_scheduled_tasks = echion.asyncio_scheduled_tasks();
     if (asyncio_scheduled_tasks != nullptr) {
         if (auto maybe_scheduled_tasks_set = MirrorSet::create(asyncio_scheduled_tasks)) {
@@ -577,6 +570,10 @@ ThreadInfo::get_all_tasks(EchionSampler& echion, PyThreadState* tstate)
         }
     }
 
+    // A Task may appear in multiple sources. Keep the earliest snapshot because it is closest to the thread stack
+    // captured for this sample.
+    std::unordered_set<PyObject*> seen;
+    std::erase_if(tasks, [&seen](const TaskInfo::Ptr& task) { return !seen.insert(task->origin).second; });
     return tasks;
 }
 #else

ddtrace/internal/datadog/profiling/stack/test/CMakeLists.txt

@@ -102,6 +102,9 @@ configure_stack_internal_test(test_sample_lifecycle)
 dd_wrapper_add_test(test_sampling_cycle_state test_sampling_cycle_state.cpp)
 configure_stack_internal_test(test_sampling_cycle_state)
 
+dd_wrapper_add_test(test_task_traversal test_task_traversal.cpp)
+configure_stack_internal_test(test_task_traversal)
+
 dd_wrapper_add_test(test_alt_stack_ownership test_alt_stack_ownership.cpp)
 # ThreadAltStack lives in the vendored echion header tree.
 target_include_directories(test_alt_stack_ownership PRIVATE ../echion)

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

@@ -0,0 +1,129 @@
+#include "echion/echion_sampler.h"
+#include "echion/threads.h"
+
+#include <gtest/gtest.h>
+
+class ThreadInfoTaskTraversalTest : public ::testing::Test
+{
+  protected:
+#if PY_VERSION_HEX >= 0x030e0000
+    // Keep the production traversal private while allowing deterministic linked-list topologies in this test.
+    static Result<void> traverse(ThreadInfo& thread,
+                                 EchionSampler& echion,
+                                 uintptr_t head,
+                                 std::vector<TaskInfo::Ptr>& tasks)
+    {
+        return thread.get_tasks_from_linked_list(echion, head, tasks);
+    }
+
+    static Result<std::vector<TaskInfo::Ptr>> get_all_tasks(ThreadInfo& thread,
+                                                            EchionSampler& echion,
+                                                            PyThreadState* tstate)
+    {
+        return thread.get_all_tasks(echion, tstate);
+    }
+#endif
+};
+
+#if PY_VERSION_HEX >= 0x030e0000
+TEST_F(ThreadInfoTaskTraversalTest, RejectsTaskMovedToAnotherList)
+{
+    // A real asyncio.Task ensures TaskInfo::create follows the same coroutine and name-reading path as production.
+    Py_Initialize();
+    PyObject* globals = PyDict_New();
+    ASSERT_NE(globals, nullptr);
+    ASSERT_EQ(PyDict_SetItemString(globals, "__builtins__", PyEval_GetBuiltins()), 0);
+
+    PyObject* result = PyRun_String(R"(
+import asyncio
+loop = asyncio.new_event_loop()
+async def wait_forever():
+    await asyncio.Event().wait()
+task = loop.create_task(wait_forever())
+)",
+                                    Py_file_input,
+                                    globals,
+                                    globals);
+    ASSERT_NE(result, nullptr);
+    Py_DECREF(result);
+
+    auto* loop = PyDict_GetItemString(globals, "loop");
+    auto* task = reinterpret_cast<TaskObj*>(PyDict_GetItemString(globals, "task"));
+    ASSERT_NE(loop, nullptr);
+    ASSERT_NE(task, nullptr);
+
+    EchionSampler echion;
+#if defined PL_LINUX
+    ThreadInfo thread(1, 1, "test-thread", CLOCK_THREAD_CPUTIME_ID);
+#elif defined PL_DARWIN
+    ThreadInfo thread(1, 1, "test-thread", mach_thread_self());
+#endif
+    thread.asyncio_loop = reinterpret_cast<uintptr_t>(loop);
+
+    // Seed the output to verify a failed source preserves tasks previously found by another source.
+    std::vector<TaskInfo::Ptr> tasks;
+    auto maybe_task = TaskInfo::create(echion, task);
+    ASSERT_TRUE(maybe_task);
+    tasks.push_back(std::move(*maybe_task));
+
+    // Echion copied head A while A <-> T. CPython then moved task T under head B, leaving the copied A.next stale:
+    //
+    //   copied head: A -> T
+    //   live list:   B <-> T
+    //
+    // Following live links from T would cycle through T -> B -> T without ever returning to A.
+    const llist_node original_task_node = task->task_node;
+    llist_node expected_head{};
+    llist_node moved_head{};
+    expected_head.next = expected_head.prev = &task->task_node;
+    moved_head.next = moved_head.prev = &task->task_node;
+    task->task_node.next = task->task_node.prev = &moved_head;
+
+    result = nullptr;
+    auto traversal = traverse(thread, echion, reinterpret_cast<uintptr_t>(&expected_head), tasks);
+
+    // Reject the malformed source and roll back only the entries it appended.
+    EXPECT_FALSE(traversal);
+    EXPECT_EQ(tasks.size(), 1);
+
+    task->task_node = original_task_node;
+    tasks.clear();
+
+    // Expose the same Task through a valid thread list and the eager-task set. Cross-source discovery must still
+    // return one TaskInfo because downstream accounting and wall-time scaling operate on this result.
+    PyObject* eager_tasks = PySet_New(nullptr);
+    ASSERT_NE(eager_tasks, nullptr);
+    ASSERT_EQ(PySet_Add(eager_tasks, reinterpret_cast<PyObject*>(task)), 0);
+    echion.init_asyncio(nullptr, eager_tasks);
+
+    _PyThreadStateImpl remote_tstate{};
+    remote_tstate.asyncio_tasks_head.next = remote_tstate.asyncio_tasks_head.prev = &task->task_node;
+    task->task_node.next = task->task_node.prev = &remote_tstate.asyncio_tasks_head;
+    thread.tstate_addr = reinterpret_cast<uintptr_t>(&remote_tstate);
+    PyThreadState local_tstate{};
+
+    auto all_tasks = get_all_tasks(thread, echion, &local_tstate);
+
+    // Restore CPython's real links before cancellation or object destruction can inspect them.
+    task->task_node = original_task_node;
+    Py_DECREF(eager_tasks);
+    ASSERT_TRUE(all_tasks);
+    EXPECT_EQ(all_tasks->size(), 1);
+
+    // Process cancellation and close the loop so the real Task does not remain pending at process exit.
+    result = PyRun_String(R"(
+task.cancel()
+try:
+    loop.run_until_complete(task)
+except asyncio.CancelledError:
+    pass
+loop.close()
+)",
+                          Py_file_input,
+                          globals,
+                          globals);
+    EXPECT_NE(result, nullptr);
+    Py_XDECREF(result);
+    Py_DECREF(globals);
+}
+#endif

releasenotes/notes/fix-profiler-asyncio-task-traversal-754a147329fc90cd.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    profiling: Fixes an issue where asyncio tasks can be duplicated in profiles on Python 3.14, causing inflated task counts and wall time and increased profiler CPU usage.