[SVLS-6907] feat: Support span dedup (#939)

## TL;DR
- Support deduping spans by span_id
- which fixes trace stats overcount issue for Node.js

## Problem
Node.js tracer sometimes sends duplicate traces to the extension,
causing over-count of trace stats. In our tests, the over-count is
usually 1–4 for 50, 500 or 5000 invocations. This only happens when the
"default" flushing strategy (which uses the "continuous" strategy) is
used, and doesn't happen when the "end" strategy is used.

## Cause of problem
I think it's similar to a known problem of continuous flushing.
- The known problem is, when Lambda runtime is frozen, the connection
between **extension** and **DD endpoint** can time out, causing data
flush failure.
- In the trace stats case, the problem is, when Lambda runtime is
frozen, the connection between **tracer** and **extension** can time
out. Extension receives the request from tracer, then freezes before
sending a response, causing the tracer's request to time out, which
makes the tracer resend the trace. This doesn't happen with the END
flush strategy because in that case, after the Lambda handler finishes,
extension still needs to flush the data and doesn't freeze so fast, and
it has enough time to respond to tracer.

## Testing 

### Steps
- Build a test extension layer
- Run e2e tests, including:
  - Install it on various Lambda functions
  - Invoke these functions with various traffic pattern
  - Check the trace stats result

### Result
**Before:**
- Over-count happens for (1) Node.js runtime + (2) "Sampling" test,
which uses the default flush strategy
- (vs expected 50)

<img width="1336" height="277" alt="Screenshot 2025-11-20 at 9 30 00 PM"
src="https://github.com/user-attachments/assets/3bf33c26-8996-43e4-8df5-7313247926b3"
/>

**After:**
- No over-count (vs expected 5000)
<img width="1447" height="421" alt="Screenshot 2025-11-20 at 9 25 32 PM"
src="https://github.com/user-attachments/assets/21576483-dbdc-499f-acc8-1801d3927b0b"
/>

## Options considered
1. **At most once**: Disable retry in tracer, at least for Lambda.
2. **At least once**
    - 2.1 Do nothing. Call out this as a known limitation.
- 2.2 Treat tracer as VIP. Before calling /next, make sure tracer's
requests have been responded. This may cause regression on invocation
duration, especially when volume is high.
3. **Exactly once**: Implement dedup in extension, by trace_id.

This PR chooses 3 because it's the easiest.

## Notes
Thanks @astuyve @rochdev @purple4reina for discussion.
Thanks Cursor for writing most of the code.
The under-count issue will be addressed separately.

Related issue:
#688

Author: lym953

bottlecap/Cargo.lock

@@ -61,12 +61,12 @@ indexmap = {version = "2.11.0", default-features = false}
 # be found in the clippy.toml file adjacent to this Cargo.toml.
 datadog-protos = { version = "0.1.0", default-features = false, git = "https://github.com/DataDog/saluki/", rev = "c89b58e5784b985819baf11f13f7d35876741222"}
 ddsketch-agent = { version = "0.1.0", default-features = false, git = "https://github.com/DataDog/saluki/", rev = "c89b58e5784b985819baf11f13f7d35876741222"}
-libdd-common = { git = "https://github.com/DataDog/libdatadog", rev = "c98467eb286c61b4483b5af5a33b268a55ccc6ff" }
-libdd-trace-protobuf = { git = "https://github.com/DataDog/libdatadog", rev = "c98467eb286c61b4483b5af5a33b268a55ccc6ff"  }
-libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "c98467eb286c61b4483b5af5a33b268a55ccc6ff" , features = ["mini_agent"] }
-libdd-trace-normalization = { git = "https://github.com/DataDog/libdatadog",  rev = "c98467eb286c61b4483b5af5a33b268a55ccc6ff" }
-datadog-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "c98467eb286c61b4483b5af5a33b268a55ccc6ff"  }
-libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog", rev = "c98467eb286c61b4483b5af5a33b268a55ccc6ff"  }
+libdd-common = { git = "https://github.com/DataDog/libdatadog", rev = "7540423559548ce049c5424599d28ee1731378e8" }
+libdd-trace-protobuf = { git = "https://github.com/DataDog/libdatadog", rev = "7540423559548ce049c5424599d28ee1731378e8"  }
+libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "7540423559548ce049c5424599d28ee1731378e8" , features = ["mini_agent"] }
+libdd-trace-normalization = { git = "https://github.com/DataDog/libdatadog",  rev = "7540423559548ce049c5424599d28ee1731378e8" }
+datadog-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "7540423559548ce049c5424599d28ee1731378e8"  }
+libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog", rev = "7540423559548ce049c5424599d28ee1731378e8"  }
 dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "502f005c56b8d51dee95424a9c1404df46e2aae4", default-features = false }
 datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "502f005c56b8d51dee95424a9c1404df46e2aae4", default-features = false }
 libddwaf = { version = "1.28.1", git = "https://github.com/DataDog/libddwaf-rust", rev = "d1534a158d976bd4f747bf9fcc58e0712d2d17fc", default-features = false, features = ["serde"] }

bottlecap/Cargo.toml

@@ -62,6 +62,7 @@ use bottlecap::{
         propagation::DatadogCompositePropagator,
         proxy_aggregator,
         proxy_flusher::Flusher as ProxyFlusher,
+        span_dedup_service,
         stats_aggregator::StatsAggregator,
         stats_concentrator_service::{StatsConcentratorHandle, StatsConcentratorService},
         stats_flusher::{self, StatsFlusher},
@@ -1085,6 +1086,9 @@ fn start_trace_agent(
         obfuscation_config: Arc::new(obfuscation_config),
     });
 
+    let (span_dedup_service, span_dedup_handle) = span_dedup_service::DedupService::new();
+    tokio::spawn(span_dedup_service.run());
+
     // Proxy
     let proxy_aggregator = Arc::new(TokioMutex::new(proxy_aggregator::Aggregator::default()));
     let proxy_flusher = Arc::new(ProxyFlusher::new(
@@ -1105,6 +1109,7 @@ fn start_trace_agent(
         appsec_processor,
         Arc::clone(tags_provider),
         stats_concentrator_handle.clone(),
+        span_dedup_handle,
     );
     let trace_agent_channel = trace_agent.get_sender_copy();
     let shutdown_token = trace_agent.shutdown_token();

bottlecap/src/bin/bottlecap/main.rs

@@ -5,6 +5,8 @@ pub mod context;
 pub mod propagation;
 pub mod proxy_aggregator;
 pub mod proxy_flusher;
+pub mod span_dedup;
+pub mod span_dedup_service;
 pub mod span_pointers;
 pub mod stats_aggregator;
 pub mod stats_concentrator_service;

bottlecap/src/traces/mod.rs

@@ -0,0 +1,185 @@
+// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+use std::collections::{HashSet, VecDeque};
+
+const DEFAULT_CAPACITY: usize = 100;
+
+/// A key for span deduplication consisting of `trace_id` and `span_id`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct DedupKey {
+    pub trace_id: u64,
+    pub span_id: u64,
+}
+
+impl DedupKey {
+    /// Creates a new `SpanDedupKey`.
+    #[must_use]
+    pub fn new(trace_id: u64, span_id: u64) -> Self {
+        Self { trace_id, span_id }
+    }
+}
+
+/// A deduplicator that maintains a fixed-size cache of recently seen span keys.
+/// When the capacity is reached, the oldest key is evicted (FIFO).
+pub struct Deduper {
+    /// Set for O(1) existence checks
+    seen: HashSet<DedupKey>,
+    /// Queue to maintain insertion order for FIFO eviction
+    order: VecDeque<DedupKey>,
+    /// Maximum number of keys to keep
+    capacity: usize,
+}
+
+impl Default for Deduper {
+    fn default() -> Self {
+        Self::new(DEFAULT_CAPACITY)
+    }
+}
+
+impl Deduper {
+    /// Creates a new `Deduper` with the specified capacity.
+    ///
+    /// # Arguments
+    ///
+    /// * `capacity` - Maximum number of IDs to track
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bottlecap::traces::span_dedup::Deduper;
+    ///
+    /// let deduper = Deduper::new(100);
+    /// ```
+    #[must_use]
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            seen: HashSet::with_capacity(capacity),
+            order: VecDeque::with_capacity(capacity),
+            capacity,
+        }
+    }
+
+    /// Checks if a span key exists and adds it if it doesn't.
+    ///
+    /// # Arguments
+    ///
+    /// * `key` - The span key to check and potentially add
+    ///
+    /// # Returns
+    ///
+    /// `true` if the key was added (didn't exist before), `false` if it already existed
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bottlecap::traces::span_dedup::{Deduper, DedupKey};
+    ///
+    /// let mut deduper = Deduper::default();
+    /// let key = DedupKey::new(123, 456);
+    /// assert!(deduper.check_and_add(key)); // Returns true, key was added
+    /// assert!(!deduper.check_and_add(key)); // Returns false, key already exists
+    /// ```
+    pub fn check_and_add(&mut self, key: DedupKey) -> bool {
+        // If the key already exists, return false
+        if self.seen.contains(&key) {
+            return false;
+        }
+
+        // If we're at capacity, evict the oldest entry
+        if self.order.len() >= self.capacity {
+            if let Some(oldest) = self.order.pop_front() {
+                self.seen.remove(&oldest);
+            }
+        }
+
+        // Add the new key
+        self.seen.insert(key);
+        self.order.push_back(key);
+
+        // Return true to indicate the key was added
+        true
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_new_deduper() {
+        let deduper = Deduper::new(10);
+        assert_eq!(deduper.seen.len(), 0);
+    }
+
+    #[test]
+    fn test_default_deduper() {
+        let deduper = Deduper::default();
+        assert_eq!(deduper.capacity, DEFAULT_CAPACITY);
+    }
+
+    #[test]
+    fn test_check_and_add() {
+        let mut deduper = Deduper::new(3);
+
+        let key1 = DedupKey::new(100, 123);
+        let key2 = DedupKey::new(100, 456);
+
+        // First call should return true (key was added)
+        assert!(deduper.check_and_add(key1));
+
+        // Second call should return false (key already exists)
+        assert!(!deduper.check_and_add(key1));
+
+        // Different key should return true
+        assert!(deduper.check_and_add(key2));
+
+        // Calling again should return false
+        assert!(!deduper.check_and_add(key2));
+    }
+
+    #[test]
+    fn test_check_and_add_with_eviction() {
+        let mut deduper = Deduper::new(2);
+
+        let key1 = DedupKey::new(1, 10);
+        let key2 = DedupKey::new(2, 20);
+        let key3 = DedupKey::new(3, 30);
+
+        assert!(deduper.check_and_add(key1));
+        assert!(deduper.check_and_add(key2));
+
+        // Adding 3rd should evict key1
+        assert!(deduper.check_and_add(key3));
+
+        // Now key1 should be addable again (was evicted)
+        assert!(deduper.check_and_add(key1));
+
+        // But key2 should have been evicted
+        assert!(deduper.check_and_add(key2));
+    }
+
+    #[test]
+    fn test_same_trace_id_different_span_id() {
+        let mut deduper = Deduper::new(3);
+
+        let key1 = DedupKey::new(100, 1);
+        let key2 = DedupKey::new(100, 2);
+
+        // Both should be added even though they have the same trace_id
+        assert!(deduper.check_and_add(key1));
+        assert!(deduper.check_and_add(key2));
+    }
+
+    #[test]
+    fn test_same_span_id_different_trace_id() {
+        let mut deduper = Deduper::new(3);
+
+        let key1 = DedupKey::new(1, 100);
+        let key2 = DedupKey::new(2, 100);
+
+        // Both should be added even though they have the same span_id
+        assert!(deduper.check_and_add(key1));
+        assert!(deduper.check_and_add(key2));
+    }
+}