Cooperative cancellation for metric collection during bounded PeriodicExportingMetricReader shutdown #4746

--- a/sdk/metric/periodic_reader.go
+++ b/sdk/metric/periodic_reader.go
@@ -323,6 +324,8 @@ func (r *PeriodicReader) ForceFlush(ctx context.Context) error {
 func (r *PeriodicReader) Shutdown(ctx context.Context) error {
 	err := ErrReaderShutdown
 	r.shutdownOnce.Do(func() {
+		ctx, cancel := context.WithTimeout(ctx, r.timeout)
+		defer cancel()
 		// Stop the run loop.
 		r.cancel()
 		<-r.done

--- a/sdk/metric/periodic_reader_test.go
+++ b/sdk/metric/periodic_reader_test.go
@@ -337,6 +337,46 @@ func TestPeriodicReaderFlushesPending(t *testing.T) {
...
+	t.Run("Shutdown timeout on producer", func(t *testing.T) {
+		exp, called := expFunc(t)
+		timeout := time.Millisecond
+		r := NewPeriodicReader(exp, WithTimeout(timeout))
+		r.register(testSDKProducer{
+			produceFunc: func(ctx context.Context, rm *metricdata.ResourceMetrics) error {
+				select {
+				case <-time.After(timeout + time.Second):
+					*rm = testResourceMetricsA
+				case <-ctx.Done():
+					// we timed out before we could collect metrics
+					return ctx.Err()
+				}
+				return nil
+			}})
+		r.RegisterProducer(testExternalProducer{})
+		assert.Equal(t, context.DeadlineExceeded, r.Shutdown(context.Background()), "timeout error not returned")
+		assert.False(t, *called, "exporter Export method called when it should have failed before export")
+	})
...

diff --git a/sdk/include/opentelemetry/sdk/metrics/export/collection_cancellation.h b/sdk/include/opentelemetry/sdk/metrics/export/collection_cancellation.h
new file mode 100644
--- /dev/null
+++ b/sdk/include/opentelemetry/sdk/metrics/export/collection_cancellation.h
@@ -0,0 +1,52 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <atomic>
+
+#include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+/**
+ * Lets code running inside a PeriodicExportingMetricReader collection cycle (a MetricProducer,
+ * an observable instrument callback or PushMetricExporter::Export) learn that a shutdown is
+ * waiting for the cycle, so it can return early, for example with MetricProducer::Status::kTimeout
+ * or a partial batch.
+ *
+ * Cancellation is cooperative: code that does not check it runs to completion, and Shutdown() then
+ * waits for it regardless of its timeout.
+ */
+class CollectionCancellation
+{
+public:
+  /**
+   * @return true if the calling thread is running a collection cycle whose reader has stopped
+   * waiting for it. Always false on other threads, so work handed off to another thread must
+   * capture the result itself.
+   */
+  static bool IsRequested() noexcept;
+
+  /** Publishes stop_requested to IsRequested() on this thread for the lifetime of the scope. */
+  class Scope
+  {
+  public:
+    explicit Scope(const std::atomic<bool> &stop_requested) noexcept;
+    ~Scope() noexcept;
+
+    Scope(const Scope &)            = delete;
+    Scope &operator=(const Scope &) = delete;
+
+  private:
+    const std::atomic<bool> *previous_;
+  };
+};
+
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
diff --git a/sdk/src/metrics/CMakeLists.txt b/sdk/src/metrics/CMakeLists.txt
--- a/sdk/src/metrics/CMakeLists.txt
+++ b/sdk/src/metrics/CMakeLists.txt
@@ -14,6 +14,7 @@ add_library(
   metric_reader.cc
   multi_observer_result.cc
   instrument_metadata_validator.cc
+  export/collection_cancellation.cc
   export/periodic_exporting_metric_reader.cc
   export/periodic_exporting_metric_reader_factory.cc
   export/periodic_exporting_metric_reader_options.cc
diff --git a/sdk/src/metrics/export/collection_cancellation.cc b/sdk/src/metrics/export/collection_cancellation.cc
new file mode 100644
--- /dev/null
+++ b/sdk/src/metrics/export/collection_cancellation.cc
@@ -0,0 +1,39 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include <atomic>
+
+#include "opentelemetry/sdk/metrics/export/collection_cancellation.h"
+#include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+namespace
+{
+thread_local const std::atomic<bool> *current_stop_requested = nullptr;
+}  // namespace
+
+bool CollectionCancellation::IsRequested() noexcept
+{
+  return current_stop_requested != nullptr &&
+         current_stop_requested->load(std::memory_order_acquire);
+}
+
+CollectionCancellation::Scope::Scope(const std::atomic<bool> &stop_requested) noexcept
+    : previous_(current_stop_requested)
+{
+  current_stop_requested = &stop_requested;
+}
+
+CollectionCancellation::Scope::~Scope() noexcept
+{
+  current_stop_requested = previous_;
+}
+
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
diff --git a/sdk/src/metrics/export/periodic_exporting_metric_reader.cc b/sdk/src/metrics/export/periodic_exporting_metric_reader.cc
--- a/sdk/src/metrics/export/periodic_exporting_metric_reader.cc
+++ b/sdk/src/metrics/export/periodic_exporting_metric_reader.cc
@@ -14,6 +14,7 @@
 #include "opentelemetry/common/timestamp.h"
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/sdk/common/global_log_handler.h"
+#include "opentelemetry/sdk/metrics/export/collection_cancellation.h"
 #include "opentelemetry/sdk/metrics/export/metric_producer.h"
 #include "opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader.h"
 #include "opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader_options.h"
@@ -161,6 +162,8 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
 bool PeriodicExportingMetricReader::CollectAndExportOnce()
 {
   std::uint64_t notify_force_flush = force_flush_pending_sequence_.load(std::memory_order_acquire);
+  // is_stop_requested_ is set once OnShutDown() has used up its flush budget.
+  const CollectionCancellation::Scope cancellation_scope(is_stop_requested_);
 #if OPENTELEMETRY_HAVE_EXCEPTIONS
   try
   {