Fix RuntimeError on dictionary size change during iteration

CHANGELOG.md

@@ -38,6 +38,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4862](https://github.com/open-telemetry/opentelemetry-python/pull/4862))
 - `opentelemetry-exporter-otlp-proto-http`: fix retry logic and error handling for connection failures in trace, metric, and log exporters
   ([#4709](https://github.com/open-telemetry/opentelemetry-python/pull/4709))
+- `opentelemetry-sdk`: avoid RuntimeError during iteration of view instrument match dictionary in MetricReaderStorage.collect()
+  ([#4891](https://github.com/open-telemetry/opentelemetry-python/pull/4891))
 
 ## Version 1.39.0/0.60b0 (2025-12-03)
 

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/metric_reader_storage.py

@@ -143,10 +143,14 @@ def collect(self) -> Optional[MetricsData]:
                 InstrumentationScope, ScopeMetrics
             ] = {}
 
+            instrument_matches_snapshot = list(
+                self._instrument_view_instrument_matches.items()
+            )
+
             for (
                 instrument,
                 view_instrument_matches,
-            ) in self._instrument_view_instrument_matches.items():
+            ) in instrument_matches_snapshot:
                 aggregation_temporality = self._instrument_class_temporality[
                     instrument.__class__
                 ]

opentelemetry-sdk/tests/metrics/test_metric_reader_storage.py

@@ -278,6 +278,44 @@ def send_measurement():
         # _ViewInstrumentMatch constructor should have only been called once
         self.assertEqual(mock_view_instrument_match_ctor.call_count, 1)
 
+    def test_race_collect_with_new_instruments(self):
+        storage = MetricReaderStorage(
+            SdkConfiguration(
+                exemplar_filter=Mock(),
+                resource=Mock(),
+                metric_readers=(),
+                views=(View(instrument_name="test"),),
+            ),
+            MagicMock(
+                **{
+                    "__getitem__.return_value": AggregationTemporality.CUMULATIVE
+                }
+            ),
+            MagicMock(**{"__getitem__.return_value": DefaultAggregation()}),
+        )
+
+        counter = _Counter("counter", Mock(), Mock())
+        storage.consume_measurement(
+            Measurement(1, time_ns(), counter, Context())
+        )
+
+        view_instrument_match = storage._instrument_view_instrument_matches[
+            counter
+        ][0]
+        original_collect = view_instrument_match.collect
+
+        new_counter = _Counter("new_counter", Mock(), Mock())
+
+        # Patch collect() to add a new counter during iteration
+        def collect_with_modification(*args, **kwargs):
+            storage._instrument_view_instrument_matches[new_counter] = []
+            return original_collect(*args, **kwargs)
+
+        view_instrument_match.collect = collect_with_modification
+        storage.collect()
+
+        self.assertIn(new_counter, storage._instrument_view_instrument_matches)
+
     @patch(
         "opentelemetry.sdk.metrics._internal."
         "metric_reader_storage._ViewInstrumentMatch"