fix(metrics): fix empty-attrs bind split and bind/collect eviction race

- Make no_attribute_tracker an Arc<TrackerEntry> so bind(&[]) returns a
  handle to the same tracker that measure(&[]) uses, preventing duplicate
  data points for empty attribute sets
- Re-check bound_count under write lock in collect_and_reset() to prevent
  a TOCTOU race where bind() increments bound_count after collect() reads
  it as 0 under a concurrent read lock
- Use Ordering::Release for has_been_updated stores in bound handles,
  matching the unbound measure() path

Author: bryantbiggs

opentelemetry-sdk/src/metrics/internal/histogram.rs

@@ -99,7 +99,7 @@ impl<T: Number> BoundMeasure<T> for BoundHistogramHandle<T> {
                 let f = measurement.into_float();
                 let index = bounds.partition_point(|&x| x < f);
                 tracker.aggregator.update((measurement, index));
-                tracker.has_been_updated.store(true, Ordering::Relaxed);
+                tracker.has_been_updated.store(true, Ordering::Release);
             }
             BoundHistogramInner::Fallback { measure, attrs } => {
                 measure.call(measurement, attrs);

opentelemetry-sdk/src/metrics/internal/mod.rs

@@ -97,7 +97,7 @@ where
     /// Number of different attribute set stored in the `trackers` map.
     count: AtomicUsize,
     /// Tracker for values with no attributes attached.
-    no_attribute_tracker: TrackerEntry<A>,
+    no_attribute_tracker: Arc<TrackerEntry<A>>,
     /// Configuration for an Aggregator
     config: A::InitConfig,
     cardinality_limit: usize,
@@ -116,7 +116,7 @@ where
             trackers: RwLock::new(HashMap::with_capacity(
                 1 + min(DEFAULT_CARDINALITY_LIMIT, cardinality_limit),
             )),
-            no_attribute_tracker: TrackerEntry::new(&config),
+            no_attribute_tracker: Arc::new(TrackerEntry::new(&config)),
             count: AtomicUsize::new(0),
             config,
             cardinality_limit,
@@ -205,8 +205,10 @@ where
     #[cfg(feature = "experimental_metrics_bound_instruments")]
     fn bind(&self, attributes: &[KeyValue]) -> Option<Arc<TrackerEntry<A>>> {
         if attributes.is_empty() {
-            let sorted_attrs: Vec<KeyValue> = vec![];
-            return self.bind_sorted(sorted_attrs);
+            self.no_attribute_tracker
+                .bound_count
+                .fetch_add(1, Ordering::Relaxed);
+            return Some(Arc::clone(&self.no_attribute_tracker));
         }
 
         let sorted_attrs = sort_and_dedup(attributes);
@@ -338,10 +340,13 @@ where
 
         if !stale_entries.is_empty() {
             if let Ok(mut trackers) = self.trackers.write() {
-                // Re-check under write lock to avoid TOCTOU race: a measure() call between
-                // dropping the read lock and acquiring the write lock could have updated
-                // an entry we marked as stale.
-                stale_entries.retain(|entry| !entry.has_been_updated.load(Ordering::Acquire));
+                // Re-check under write lock to avoid TOCTOU race: a measure() or bind() call
+                // between dropping the read lock and acquiring the write lock could have
+                // updated an entry or bound a handle to one we marked as stale.
+                stale_entries.retain(|entry| {
+                    !entry.has_been_updated.load(Ordering::Acquire)
+                        && entry.bound_count.load(Ordering::Acquire) == 0
+                });
 
                 if !stale_entries.is_empty() {
                     let stale_pointers: HashSet<*const TrackerEntry<A>> =

opentelemetry-sdk/src/metrics/internal/sum.rs

@@ -70,7 +70,7 @@ impl<T: Number> BoundMeasure<T> for BoundSumHandle<T> {
         match &self.inner {
             BoundSumInner::Direct { tracker } => {
                 tracker.aggregator.update(measurement);
-                tracker.has_been_updated.store(true, Ordering::Relaxed);
+                tracker.has_been_updated.store(true, Ordering::Release);
             }
             BoundSumInner::Fallback { measure, attrs } => {
                 measure.call(measurement, attrs);

opentelemetry-sdk/src/metrics/mod.rs

@@ -5337,4 +5337,31 @@ mod tests {
         assert_eq!(sum.data_points.len(), 1);
         assert_eq!(sum.data_points[0].value, 40);
     }
+
+    #[cfg(feature = "experimental_metrics_bound_instruments")]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn bound_counter_empty_attributes_shares_with_unbound() {
+        let mut test_context = TestContext::new(Temporality::Cumulative);
+        let counter = test_context.u64_counter("test", "my_counter", None);
+        let bound = counter.bind(&[]);
+
+        // Mix bound and unbound calls with empty attributes — they must share
+        // the same data point (both route to no_attribute_tracker).
+        counter.add(10, &[]);
+        bound.add(20);
+        counter.add(30, &[]);
+        bound.add(40);
+        test_context.flush_metrics();
+
+        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
+            unreachable!()
+        };
+
+        assert_eq!(
+            sum.data_points.len(),
+            1,
+            "Bound and unbound with empty attributes must share the same data point"
+        );
+        assert_eq!(sum.data_points[0].value, 100);
+    }
 }