refactor(metrics): extend direct-bind to all aggregators

Removes BoundFallbackHandle and the fallback parameter on
Measure::bind, leaving every aggregator on the same direct-bind
shape: bind() returns a BoundXxxHandle holding an Arc<TrackerEntry>
that writes directly to the aggregator without per-call attribute
lookup.

LastValue, PrecomputedSum, and ExpoHistogram now match the Sum /
Histogram pattern. PrecomputedSum's bind() is unreachable from user
code today (async-only) but the impl exists so the trait is uniform
and future Gauge / Observable bind() extensions are mechanical.

The rare RwLock-poisoned case now yields a NoopBoundMeasure that
drops measurements silently — mirroring measure()'s own poison
handling.

Test coverage added:
- Bound ExponentialHistogram via View: delta with NaN/inf filter,
  bind-at-overflow attribution, persistence across delta eviction,
  view filter applied at bind time, and cumulative accumulation.
- Cumulative bind-at-overflow tests for Counter and Histogram.
- LastValue::bind and PrecomputedSum::bind unit tests exercising
  Measure / BoundMeasure traits and bound_count Drop semantics.
- ValueMap::bind under a poisoned trackers RwLock returns None so
  the caller produces a NoopBoundMeasure rather than panicking.

Author: bryantbiggs

opentelemetry-sdk/src/metrics/instrument.rs

@@ -398,7 +398,7 @@ impl<T: Copy + 'static> SyncInstrument<T> for ResolvedMeasures<T> {
         let bound_measures: Vec<Box<dyn BoundMeasure<T>>> = self
             .measures
             .iter()
-            .map(|m| m.bind(attrs, Arc::clone(m)))
+            .map(|m| m.bind(attrs))
             .collect();
         Box::new(ResolvedBoundMeasures {
             measures: bound_measures,

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

@@ -20,7 +20,7 @@ pub(crate) trait Measure<T>: Send + Sync + 'static {
     fn call(&self, measurement: T, attrs: &[KeyValue]);
 
     #[cfg(feature = "experimental_metrics_bound_instruments")]
-    fn bind(&self, attrs: &[KeyValue], fallback: Arc<dyn Measure<T>>) -> Box<dyn BoundMeasure<T>>;
+    fn bind(&self, attrs: &[KeyValue]) -> Box<dyn BoundMeasure<T>>;
 }
 
 /// A pre-bound measurement handle that bypasses attribute lookup.
@@ -29,26 +29,28 @@ pub(crate) trait BoundMeasure<T>: Send + Sync + 'static {
     fn call(&self, measurement: T);
 }
 
-/// Fallback bound handle for aggregator types that don't support direct binding.
-/// Delegates every call to the unbound `Measure::call()` path.
+/// A bound handle that drops every measurement silently. Used when
+/// `ValueMap::bind` returns `None` because the trackers `RwLock` is poisoned —
+/// an extremely rare degenerate state in which the SDK can no longer aggregate
+/// reliably. Returning a noop here mirrors `measure()`'s own poison handling
+/// (silent drop) rather than panicking on the user's hot path.
 #[cfg(feature = "experimental_metrics_bound_instruments")]
-pub(crate) struct BoundFallbackHandle<T> {
-    measure: Arc<dyn Measure<T>>,
-    attrs: Vec<KeyValue>,
+pub(crate) struct NoopBoundMeasure<T> {
+    _marker: marker::PhantomData<T>,
 }
 
 #[cfg(feature = "experimental_metrics_bound_instruments")]
-impl<T: Send + Sync + 'static> BoundMeasure<T> for BoundFallbackHandle<T> {
-    fn call(&self, measurement: T) {
-        self.measure.call(measurement, &self.attrs);
+impl<T> NoopBoundMeasure<T> {
+    pub(crate) fn new() -> Self {
+        Self {
+            _marker: marker::PhantomData,
+        }
     }
 }
 
 #[cfg(feature = "experimental_metrics_bound_instruments")]
-impl<T> BoundFallbackHandle<T> {
-    pub(crate) fn new(measure: Arc<dyn Measure<T>>, attrs: Vec<KeyValue>) -> Self {
-        Self { measure, attrs }
-    }
+impl<T: Send + Sync + 'static> BoundMeasure<T> for NoopBoundMeasure<T> {
+    fn call(&self, _measurement: T) {}
 }
 
 /// Stores the aggregate of measurements into the aggregation and returns the number

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

@@ -1,4 +1,6 @@
 #[cfg(feature = "experimental_metrics_bound_instruments")]
+use std::sync::atomic::Ordering;
+#[cfg(feature = "experimental_metrics_bound_instruments")]
 use std::sync::Arc;
 use std::{f64::consts::LOG2_E, mem::replace, ops::DerefMut, sync::Mutex};
 
@@ -15,11 +17,41 @@ use super::{
     Aggregator, ComputeAggregation, Measure, Number, ValueMap,
 };
 #[cfg(feature = "experimental_metrics_bound_instruments")]
-use super::{BoundFallbackHandle, BoundMeasure};
+use super::{BoundMeasure, NoopBoundMeasure, TrackerEntry};
 
 pub(crate) const EXPO_MAX_SCALE: i8 = 20;
 pub(crate) const EXPO_MIN_SCALE: i8 = -10;
 
+/// Pre-bound exponential-histogram handle: writes go directly to a fixed
+/// `TrackerEntry` without per-call attribute lookup. Unlike `BoundHistogramHandle`,
+/// no bucket precomputation happens at the call site — `update()` does scale
+/// resolution and bin assignment inside the entry's Mutex. The NaN/inf filter
+/// from the unbound `call()` path is preserved here.
+#[cfg(feature = "experimental_metrics_bound_instruments")]
+struct BoundExpoHistogramHandle<T: Number> {
+    tracker: Arc<TrackerEntry<Mutex<ExpoHistogramDataPoint<T>>>>,
+}
+
+#[cfg(feature = "experimental_metrics_bound_instruments")]
+impl<T: Number> BoundMeasure<T> for BoundExpoHistogramHandle<T> {
+    fn call(&self, measurement: T) {
+        // Mirror unbound ExpoHistogram::call: ignore NaN and infinity so that
+        // ExpoHistogramDataPoint::record's invariants are preserved.
+        if !measurement.into_float().is_finite() {
+            return;
+        }
+        self.tracker.aggregator.update(measurement);
+        self.tracker.has_been_updated.store(true, Ordering::Release);
+    }
+}
+
+#[cfg(feature = "experimental_metrics_bound_instruments")]
+impl<T: Number> Drop for BoundExpoHistogramHandle<T> {
+    fn drop(&mut self) {
+        self.tracker.bound_count.fetch_sub(1, Ordering::Relaxed);
+    }
+}
+
 /// A single data point in an exponential histogram.
 #[derive(Debug, PartialEq)]
 struct ExpoHistogramDataPoint<T> {
@@ -532,8 +564,15 @@ where
     }
 
     #[cfg(feature = "experimental_metrics_bound_instruments")]
-    fn bind(&self, attrs: &[KeyValue], fallback: Arc<dyn Measure<T>>) -> Box<dyn BoundMeasure<T>> {
-        Box::new(BoundFallbackHandle::new(fallback, attrs.to_vec()))
+    fn bind(&self, attrs: &[KeyValue]) -> Box<dyn BoundMeasure<T>> {
+        let mut bound_attrs = Vec::new();
+        self.filter.apply(attrs, |filtered| {
+            bound_attrs = filtered.to_vec();
+        });
+        match self.value_map.bind(&bound_attrs) {
+            Some(tracker) => Box::new(BoundExpoHistogramHandle { tracker }),
+            None => Box::new(NoopBoundMeasure::new()),
+        }
     }
 }
 

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

@@ -14,7 +14,7 @@ use opentelemetry::KeyValue;
 use super::aggregate::{AggregateTimeInitiator, AttributeSetFilter};
 use super::{Aggregator, ComputeAggregation, Measure, Number, ValueMap};
 #[cfg(feature = "experimental_metrics_bound_instruments")]
-use super::{BoundFallbackHandle, BoundMeasure, TrackerEntry};
+use super::{BoundMeasure, NoopBoundMeasure, TrackerEntry};
 
 impl<T> Aggregator for Mutex<Buckets<T>>
 where
@@ -268,7 +268,7 @@ where
     }
 
     #[cfg(feature = "experimental_metrics_bound_instruments")]
-    fn bind(&self, attrs: &[KeyValue], fallback: Arc<dyn Measure<T>>) -> Box<dyn BoundMeasure<T>> {
+    fn bind(&self, attrs: &[KeyValue]) -> Box<dyn BoundMeasure<T>> {
         let mut bound_attrs = Vec::new();
         self.filter.apply(attrs, |filtered| {
             bound_attrs = filtered.to_vec();
@@ -278,10 +278,9 @@ where
                 tracker,
                 bounds: self.bounds.clone(),
             }),
-            // Trackers RwLock is poisoned — extremely rare. Hand back a fallback
-            // handle whose writes will silently drop (mirroring `measure()`'s
-            // own poison handling) rather than panic on the user's hot path.
-            None => Box::new(BoundFallbackHandle::new(fallback, bound_attrs)),
+            // Trackers RwLock is poisoned — return a noop handle so writes
+            // silently drop, mirroring `measure()`'s own poison handling.
+            None => Box::new(NoopBoundMeasure::new()),
         }
     }
 }

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

@@ -4,14 +4,40 @@ use crate::metrics::{
 };
 use opentelemetry::KeyValue;
 #[cfg(feature = "experimental_metrics_bound_instruments")]
+use std::sync::atomic::Ordering;
+#[cfg(feature = "experimental_metrics_bound_instruments")]
 use std::sync::Arc;
 
 use super::{
     aggregate::{AggregateTimeInitiator, AttributeSetFilter},
     Aggregator, AtomicTracker, AtomicallyUpdate, ComputeAggregation, Measure, Number, ValueMap,
 };
 #[cfg(feature = "experimental_metrics_bound_instruments")]
-use super::{BoundFallbackHandle, BoundMeasure};
+use super::{BoundMeasure, NoopBoundMeasure, TrackerEntry};
+
+/// Pre-bound gauge/last-value handle: writes go directly to a fixed
+/// `TrackerEntry` without per-call attribute lookup. The `tracker` is either
+/// a dedicated entry for the bound attribute set, or — if bind() hit the
+/// cardinality limit — the shared overflow tracker.
+#[cfg(feature = "experimental_metrics_bound_instruments")]
+struct BoundLastValueHandle<T: Number> {
+    tracker: Arc<TrackerEntry<Assign<T>>>,
+}
+
+#[cfg(feature = "experimental_metrics_bound_instruments")]
+impl<T: Number> BoundMeasure<T> for BoundLastValueHandle<T> {
+    fn call(&self, measurement: T) {
+        self.tracker.aggregator.update(measurement);
+        self.tracker.has_been_updated.store(true, Ordering::Release);
+    }
+}
+
+#[cfg(feature = "experimental_metrics_bound_instruments")]
+impl<T: Number> Drop for BoundLastValueHandle<T> {
+    fn drop(&mut self) {
+        self.tracker.bound_count.fetch_sub(1, Ordering::Relaxed);
+    }
+}
 
 /// this is reused by PrecomputedSum
 pub(crate) struct Assign<T>
@@ -148,8 +174,15 @@ where
     }
 
     #[cfg(feature = "experimental_metrics_bound_instruments")]
-    fn bind(&self, attrs: &[KeyValue], fallback: Arc<dyn Measure<T>>) -> Box<dyn BoundMeasure<T>> {
-        Box::new(BoundFallbackHandle::new(fallback, attrs.to_vec()))
+    fn bind(&self, attrs: &[KeyValue]) -> Box<dyn BoundMeasure<T>> {
+        let mut bound_attrs = Vec::new();
+        self.filter.apply(attrs, |filtered| {
+            bound_attrs = filtered.to_vec();
+        });
+        match self.value_map.bind(&bound_attrs) {
+            Some(tracker) => Box::new(BoundLastValueHandle { tracker }),
+            None => Box::new(NoopBoundMeasure::new()),
+        }
     }
 }
 
@@ -166,3 +199,73 @@ where
         (len, new.map(T::make_aggregated_metrics))
     }
 }
+
+#[cfg(all(test, feature = "experimental_metrics_bound_instruments"))]
+mod tests {
+    use super::*;
+    use crate::metrics::data::{AggregatedMetrics, Gauge, MetricData};
+
+    fn extract_gauge(agg: AggregatedMetrics) -> Gauge<u64> {
+        match agg {
+            AggregatedMetrics::U64(MetricData::Gauge(g)) => g,
+            _ => panic!("expected u64 Gauge"),
+        }
+    }
+
+    /// Direct unit coverage for `LastValue::bind`. Sync `Gauge::bind()` is not yet
+    /// exposed in the public API, so the only callers of this code path today
+    /// are Views that remap an instrument to `Aggregation::LastValue`. This test
+    /// constructs the aggregator directly and exercises the bound handle through
+    /// the `Measure` / `BoundMeasure` traits to keep the impl honest.
+    #[test]
+    fn bind_writes_through_bound_handle() {
+        let last_value = LastValue::<u64>::new(Temporality::Cumulative, AttributeSetFilter::new(None), 100);
+        let attrs = [KeyValue::new("k", "v")];
+        let bound = Measure::bind(&last_value, &attrs);
+
+        bound.call(7);
+        bound.call(42); // overwrites previous value (LastValue semantics)
+
+        let (count, agg) = ComputeAggregation::call(&last_value, None);
+        assert_eq!(count, 1);
+        let gauge = extract_gauge(agg.expect("aggregation produced"));
+        assert_eq!(gauge.data_points.len(), 1);
+        assert_eq!(gauge.data_points[0].value, 42);
+        assert_eq!(gauge.data_points[0].attributes, attrs.to_vec());
+    }
+
+    #[test]
+    fn bound_handle_drop_decrements_bound_count() {
+        let last_value = LastValue::<u64>::new(Temporality::Delta, AttributeSetFilter::new(None), 100);
+        let attrs = [KeyValue::new("k", "v")];
+
+        let bound = Measure::bind(&last_value, &attrs);
+        bound.call(5);
+
+        // While the handle exists, the entry's bound_count is 1.
+        let trackers = last_value.value_map.trackers.read().unwrap();
+        let entry = trackers
+            .values()
+            .next()
+            .expect("entry should exist after bind+call");
+        assert_eq!(
+            entry.bound_count.load(Ordering::Relaxed),
+            1,
+            "bound_count should reflect a live handle"
+        );
+        drop(trackers);
+
+        drop(bound);
+
+        let trackers = last_value.value_map.trackers.read().unwrap();
+        let entry = trackers
+            .values()
+            .next()
+            .expect("entry should still exist post-drop");
+        assert_eq!(
+            entry.bound_count.load(Ordering::Relaxed),
+            0,
+            "bound_count should drop to 0 after handle drops"
+        );
+    }
+}

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

@@ -18,7 +18,7 @@ use std::sync::{Arc, OnceLock, RwLock};
 
 pub(crate) use aggregate::{AggregateBuilder, AggregateFns, ComputeAggregation, Measure};
 #[cfg(feature = "experimental_metrics_bound_instruments")]
-pub(crate) use aggregate::{BoundFallbackHandle, BoundMeasure};
+pub(crate) use aggregate::{BoundMeasure, NoopBoundMeasure};
 pub(crate) use exponential_histogram::{EXPO_MAX_SCALE, EXPO_MIN_SCALE};
 #[cfg(feature = "experimental_metrics_bound_instruments")]
 use opentelemetry::otel_debug;
@@ -892,4 +892,43 @@ mod tests {
             "count should reach 0 after eviction"
         );
     }
+
+    /// When the trackers `RwLock` is poisoned, `bind()` cannot safely insert or
+    /// look up entries, so it returns `None` and the caller (Sum/Histogram/etc.)
+    /// hands back a `NoopBoundMeasure`. This is a defensive branch that fires
+    /// on degenerate states (a thread panicked while holding the write lock)
+    /// and is unreachable through normal traffic. The test induces poisoning
+    /// explicitly so the branch keeps coverage.
+    #[cfg(feature = "experimental_metrics_bound_instruments")]
+    #[test]
+    fn bind_returns_none_when_trackers_lock_is_poisoned() {
+        let value_map = ValueMap::<Assign<i64>>::new((), 100);
+
+        // Poison the trackers RwLock by panicking inside a write guard.
+        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let _guard = value_map.trackers.write().unwrap();
+            panic!("intentional poison");
+        }));
+
+        assert!(
+            value_map.trackers.is_poisoned(),
+            "trackers lock must be poisoned for this test to be meaningful"
+        );
+
+        // Empty attrs use the no_attribute_tracker fast path and never touch
+        // the poisoned lock — they should still succeed.
+        assert!(
+            value_map.bind(&[]).is_some(),
+            "bind(&[]) must succeed even with poisoned lock; uses no_attribute_tracker"
+        );
+
+        // Non-empty attrs go through bind_attrs which needs the trackers lock.
+        // The read-lock try succeeds (only writes poison, but read on poisoned
+        // can also fail) — fall through to write lock which fails poisoned.
+        let result = value_map.bind(&[KeyValue::new("k", 1_i64)]);
+        assert!(
+            result.is_none(),
+            "bind() with non-empty attrs must return None on poisoned lock"
+        );
+    }
 }