diff --git a/google/cloud/google_cloud_cpp_common.bzl b/google/cloud/google_cloud_cpp_common.bzl
index 421c9c00a4160..5c86d6d9df7fc 100644
--- a/google/cloud/google_cloud_cpp_common.bzl
+++ b/google/cloud/google_cloud_cpp_common.bzl
@@ -55,6 +55,7 @@ google_cloud_cpp_common_hdrs = [
     "internal/future_fwd.h",
     "internal/future_impl.h",
     "internal/future_then_impl.h",
+    "internal/generic_background_threads_impl.h",
     "internal/getenv.h",
     "internal/group_options.h",
     "internal/invocation_id_generator.h",
diff --git a/google/cloud/google_cloud_cpp_common.cmake b/google/cloud/google_cloud_cpp_common.cmake
index 8b3601db39f01..a892e6db3d02c 100644
--- a/google/cloud/google_cloud_cpp_common.cmake
+++ b/google/cloud/google_cloud_cpp_common.cmake
@@ -86,6 +86,7 @@ add_library(
     internal/future_impl.h
     internal/future_then_impl.cc
     internal/future_then_impl.h
+    internal/generic_background_threads_impl.h
     internal/getenv.cc
     internal/getenv.h
     internal/group_options.h
diff --git a/google/cloud/google_cloud_cpp_rest_internal.cmake b/google/cloud/google_cloud_cpp_rest_internal.cmake
index c2fa155d7ad9e..588e1da4bda2b 100644
--- a/google/cloud/google_cloud_cpp_rest_internal.cmake
+++ b/google/cloud/google_cloud_cpp_rest_internal.cmake
@@ -298,6 +298,7 @@ if (BUILD_TESTING)
         internal/rest_lro_helpers_test.cc
         internal/rest_opentelemetry_test.cc
         internal/rest_parse_json_error_test.cc
+        internal/rest_pure_background_threads_impl_test.cc
         internal/rest_request_test.cc
         internal/rest_response_test.cc
         internal/rest_retry_loop_test.cc
diff --git a/google/cloud/google_cloud_cpp_rest_internal_unit_tests.bzl b/google/cloud/google_cloud_cpp_rest_internal_unit_tests.bzl
index 003a9fe1d2080..c0e1909d4ca6b 100644
--- a/google/cloud/google_cloud_cpp_rest_internal_unit_tests.bzl
+++ b/google/cloud/google_cloud_cpp_rest_internal_unit_tests.bzl
@@ -61,6 +61,7 @@ google_cloud_cpp_rest_internal_unit_tests = [
     "internal/rest_lro_helpers_test.cc",
     "internal/rest_opentelemetry_test.cc",
     "internal/rest_parse_json_error_test.cc",
+    "internal/rest_pure_background_threads_impl_test.cc",
     "internal/rest_request_test.cc",
     "internal/rest_response_test.cc",
     "internal/rest_retry_loop_test.cc",
diff --git a/google/cloud/internal/background_threads_impl.cc b/google/cloud/internal/background_threads_impl.cc
index d5e10b1750699..174d57311eb12 100644
--- a/google/cloud/internal/background_threads_impl.cc
+++ b/google/cloud/internal/background_threads_impl.cc
@@ -21,47 +21,7 @@
 namespace google {
 namespace cloud {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
-namespace internal {
-
-AutomaticallyCreatedBackgroundThreads::AutomaticallyCreatedBackgroundThreads(
-    std::size_t thread_count)
-    : pool_(thread_count == 0 ? 1 : thread_count) {
-  std::generate_n(pool_.begin(), pool_.size(), [this] {
-    promise<void> started;
-    auto thread = std::thread(
-        [](CompletionQueue cq, promise<void>& started) {
-          started.set_value();
-          cq.Run();
-        },
-        cq_, std::ref(started));
-    started.get_future().wait();
-    return thread;
-  });
-}
-
-AutomaticallyCreatedBackgroundThreads::
-    ~AutomaticallyCreatedBackgroundThreads() {
-  Shutdown();
-}
-
-void AutomaticallyCreatedBackgroundThreads::Shutdown() {
-  cq_.Shutdown();
-  for (auto& t : pool_) {
-#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
-    try {
-#endif
-      t.join();
-#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
-    } catch (std::system_error const& e) {
-      GCP_LOG(FATAL) << "AutomaticallyCreatedBackgroundThreads::Shutdown: "
-                     << e.what();
-    }
-#endif
-  }
-  pool_.clear();
-}
-
-}  // namespace internal
+namespace internal {}  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace cloud
 }  // namespace google
diff --git a/google/cloud/internal/background_threads_impl.h b/google/cloud/internal/background_threads_impl.h
index a307ded7976cf..b13f477658e1d 100644
--- a/google/cloud/internal/background_threads_impl.h
+++ b/google/cloud/internal/background_threads_impl.h
@@ -17,6 +17,7 @@
 
 #include "google/cloud/background_threads.h"
 #include "google/cloud/completion_queue.h"
+#include "google/cloud/internal/generic_background_threads_impl.h"
 #include "google/cloud/version.h"
 #include <thread>
 #include <vector>
@@ -40,19 +41,9 @@ class CustomerSuppliedBackgroundThreads : public BackgroundThreads {
 };
 
 /// Create a background thread to perform background operations.
-class AutomaticallyCreatedBackgroundThreads : public BackgroundThreads {
- public:
-  explicit AutomaticallyCreatedBackgroundThreads(std::size_t thread_count = 1U);
-  ~AutomaticallyCreatedBackgroundThreads() override;
-
-  CompletionQueue cq() const override { return cq_; }
-  void Shutdown();
-  std::size_t pool_size() const { return pool_.size(); }
-
- private:
-  CompletionQueue cq_;
-  std::vector<std::thread> pool_;
-};
+using AutomaticallyCreatedBackgroundThreads =
+    AutomaticallyCreatedBackgroundThreadsImpl<CompletionQueue,
+                                              BackgroundThreads>;
 
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
diff --git a/google/cloud/internal/generic_background_threads_impl.h b/google/cloud/internal/generic_background_threads_impl.h
new file mode 100644
index 0000000000000..ca1c79b71731c
--- /dev/null
+++ b/google/cloud/internal/generic_background_threads_impl.h
@@ -0,0 +1,91 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_GENERIC_BACKGROUND_THREADS_IMPL_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_GENERIC_BACKGROUND_THREADS_IMPL_H
+
+#include "google/cloud/future.h"
+#include "google/cloud/log.h"
+#include "google/cloud/version.h"
+#include <algorithm>
+#include <system_error>
+#include <thread>
+#include <vector>
+
+namespace google {
+namespace cloud {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+namespace internal {
+
+template <typename QueueType>
+struct DefaultQueueTraits {
+  static QueueType Create() { return QueueType(); }
+  static void Run(QueueType cq, promise<void>& started) {
+    started.set_value();
+    cq.Run();
+  }
+};
+
+template <typename QueueType, typename BaseInterface,
+          typename QueueTraits = DefaultQueueTraits<QueueType>>
+class AutomaticallyCreatedBackgroundThreadsImpl : public BaseInterface {
+ public:
+  explicit AutomaticallyCreatedBackgroundThreadsImpl(
+      std::size_t thread_count = 1U)
+      : cq_(QueueTraits::Create()),
+        pool_(thread_count == 0 ? 1 : thread_count) {
+    std::generate_n(pool_.begin(), pool_.size(), [this] {
+      promise<void> started;
+      auto thread = std::thread(
+          [](QueueType cq, promise<void>& started) {
+            QueueTraits::Run(std::move(cq), started);
+          },
+          cq_, std::ref(started));
+      started.get_future().wait();
+      return thread;
+    });
+  }
+
+  ~AutomaticallyCreatedBackgroundThreadsImpl() override { Shutdown(); }
+
+  QueueType cq() const override { return cq_; }
+  std::size_t pool_size() const { return pool_.size(); }
+
+  void Shutdown() {
+    cq_.Shutdown();
+    for (auto& t : pool_) {
+#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
+      try {
+#endif
+        if (t.joinable()) t.join();
+#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
+      } catch (std::system_error const& e) {
+        GCP_LOG(FATAL) << "Shutdown error: " << e.what();
+      }
+#endif
+    }
+    pool_.clear();
+  }
+
+ private:
+  QueueType cq_;
+  std::vector<std::thread> pool_;
+};
+
+}  // namespace internal
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_GENERIC_BACKGROUND_THREADS_IMPL_H
diff --git a/google/cloud/internal/rest_pure_background_threads_impl.cc b/google/cloud/internal/rest_pure_background_threads_impl.cc
index 525b92c0adb79..a830172037f80 100644
--- a/google/cloud/internal/rest_pure_background_threads_impl.cc
+++ b/google/cloud/internal/rest_pure_background_threads_impl.cc
@@ -25,48 +25,6 @@ namespace cloud {
 namespace rest_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
-AutomaticallyCreatedRestPureBackgroundThreads::
-    AutomaticallyCreatedRestPureBackgroundThreads(std::size_t thread_count)
-    : cq_(std::make_shared<RestPureCompletionQueueImpl>()),
-      pool_(thread_count == 0 ? 1 : thread_count) {
-  std::generate_n(pool_.begin(), pool_.size(), [this] {
-    promise<void> started;
-    auto thread = std::thread(
-        [](RestPureCompletionQueue cq, promise<void>& started,
-           internal::CallContext c) {
-          internal::ScopedCallContext scope(std::move(c));
-          started.set_value();
-          cq.Run();
-        },
-        cq_, std::ref(started), internal::CallContext{});
-    started.get_future().wait();
-    return thread;
-  });
-}
-
-AutomaticallyCreatedRestPureBackgroundThreads::
-    ~AutomaticallyCreatedRestPureBackgroundThreads() {
-  Shutdown();
-}
-
-void AutomaticallyCreatedRestPureBackgroundThreads::Shutdown() {
-  cq_.Shutdown();
-  for (auto& t : pool_) {
-#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
-    try {
-#endif
-      t.join();
-#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
-    } catch (std::system_error const& e) {
-      GCP_LOG(FATAL)
-          << "AutomaticallyCreatedRestPureBackgroundThreads::Shutdown: "
-          << e.what();
-    }
-#endif
-  }
-  pool_.clear();
-}
-
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace rest_internal
 }  // namespace cloud
diff --git a/google/cloud/internal/rest_pure_background_threads_impl.h b/google/cloud/internal/rest_pure_background_threads_impl.h
index 4061729dd7f63..a9c8aef7c14ca 100644
--- a/google/cloud/internal/rest_pure_background_threads_impl.h
+++ b/google/cloud/internal/rest_pure_background_threads_impl.h
@@ -15,6 +15,8 @@
 #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_REST_PURE_BACKGROUND_THREADS_IMPL_H
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_REST_PURE_BACKGROUND_THREADS_IMPL_H
 
+#include "google/cloud/internal/call_context.h"
+#include "google/cloud/internal/generic_background_threads_impl.h"
 #include "google/cloud/internal/rest_pure_completion_queue_impl.h"
 #include "google/cloud/version.h"
 #include <thread>
@@ -37,23 +39,25 @@ class RestPureBackgroundThreads {
   virtual RestPureCompletionQueue cq() const = 0;
 };
 
-/// Background threads that run on a RestPureCompletionQueue.
-class AutomaticallyCreatedRestPureBackgroundThreads
-    : public RestPureBackgroundThreads {
- public:
-  explicit AutomaticallyCreatedRestPureBackgroundThreads(
-      std::size_t thread_count = 1U);
-  ~AutomaticallyCreatedRestPureBackgroundThreads() override;
-
-  RestPureCompletionQueue cq() const override { return cq_; }
-  void Shutdown();
-  std::size_t pool_size() const { return pool_.size(); }
-
- private:
-  RestPureCompletionQueue cq_;
-  std::vector<std::thread> pool_;
+struct RestPureQueueTraits {
+  static RestPureCompletionQueue Create() {
+    return RestPureCompletionQueue(
+        std::make_shared<RestPureCompletionQueueImpl>());
+  }
+  static void Run(RestPureCompletionQueue cq, promise<void>& started) {
+    internal::CallContext context;
+    internal::ScopedCallContext scope(std::move(context));
+    started.set_value();
+    cq.Run();
+  }
 };
 
+/// Background threads that run on a RestPureCompletionQueue.
+using AutomaticallyCreatedRestPureBackgroundThreads =
+    google::cloud::internal::AutomaticallyCreatedBackgroundThreadsImpl<
+        RestPureCompletionQueue, RestPureBackgroundThreads,
+        RestPureQueueTraits>;
+
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace rest_internal
 }  // namespace cloud
diff --git a/google/cloud/internal/rest_pure_background_threads_impl_test.cc b/google/cloud/internal/rest_pure_background_threads_impl_test.cc
new file mode 100644
index 0000000000000..31ee8e68ec0c1
--- /dev/null
+++ b/google/cloud/internal/rest_pure_background_threads_impl_test.cc
@@ -0,0 +1,84 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "google/cloud/internal/rest_pure_background_threads_impl.h"
+#include <gmock/gmock.h>
+#include <set>
+#include <vector>
+
+namespace google {
+namespace cloud {
+namespace rest_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+namespace {
+
+using ::testing::Contains;
+using ::testing::Not;
+
+/// @test Verify that automatically created completion queues are usable.
+TEST(AutomaticallyCreatedRestPureBackgroundThreads, IsActive) {
+  AutomaticallyCreatedRestPureBackgroundThreads actual;
+  EXPECT_EQ(1, actual.pool_size());
+
+  promise<std::thread::id> bg;
+  actual.cq().RunAsync([&bg] { bg.set_value(std::this_thread::get_id()); });
+  EXPECT_NE(std::this_thread::get_id(), bg.get_future().get());
+}
+
+/// @test Verify that automatically created completion queues are usable.
+TEST(AutomaticallyCreatedRestPureBackgroundThreads, NoEmptyPools) {
+  AutomaticallyCreatedRestPureBackgroundThreads actual(0);
+  EXPECT_EQ(1, actual.pool_size());
+
+  promise<std::thread::id> bg;
+  actual.cq().RunAsync([&bg] { bg.set_value(std::this_thread::get_id()); });
+  EXPECT_NE(std::this_thread::get_id(), bg.get_future().get());
+}
+
+/// @test Verify that automatically created completion queues work.
+TEST(AutomaticallyCreatedRestPureBackgroundThreads, ManyThreads) {
+  auto constexpr kThreadCount = 4;
+  AutomaticallyCreatedRestPureBackgroundThreads actual(kThreadCount);
+  EXPECT_EQ(kThreadCount, actual.pool_size());
+
+  std::vector<promise<std::thread::id>> promises(100 * kThreadCount);
+  for (auto& p : promises) {
+    actual.cq().RunAsync([&p] { p.set_value(std::this_thread::get_id()); });
+  }
+  std::set<std::thread::id> ids;
+  for (auto& p : promises) ids.insert(p.get_future().get());
+  EXPECT_FALSE(ids.empty());
+  EXPECT_GE(kThreadCount, ids.size());
+  EXPECT_THAT(ids, Not(Contains(std::this_thread::get_id())));
+}
+
+/// @test Verify that automatically created completion queues work.
+TEST(AutomaticallyCreatedRestPureBackgroundThreads, ManualShutdown) {
+  auto constexpr kThreadCount = 4;
+  AutomaticallyCreatedRestPureBackgroundThreads actual(kThreadCount);
+  EXPECT_EQ(kThreadCount, actual.pool_size());
+
+  std::vector<promise<void>> promises(2 * kThreadCount);
+  for (auto& p : promises) {
+    actual.cq().RunAsync([&p] { p.set_value(); });
+  }
+  for (auto& p : promises) p.get_future().get();
+  actual.Shutdown();
+}
+
+}  // namespace
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace rest_internal
+}  // namespace cloud
+}  // namespace google
diff --git a/google/cloud/storage/client.cc b/google/cloud/storage/client.cc
index 559d2f87fca7c..ee138a3caf5be 100644
--- a/google/cloud/storage/client.cc
+++ b/google/cloud/storage/client.cc
@@ -555,7 +555,8 @@ Options DefaultOptions(Options opts) {
                                    STORAGE_CLIENT_DEFAULT_MAXIMUM_BACKOFF_DELAY,
                                    STORAGE_CLIENT_DEFAULT_BACKOFF_SCALING)
               .clone())
-      .set<IdempotencyPolicyOption>(AlwaysRetryIdempotencyPolicy().clone());
+      .set<IdempotencyPolicyOption>(AlwaysRetryIdempotencyPolicy().clone())
+      .set<storage_experimental::OTelSpanEnrichmentOption>(true);
 
   o = google::cloud::internal::MergeOptions(std::move(opts), std::move(o));
   // If the application did not set `DownloadStallTimeoutOption` then use the
diff --git a/google/cloud/storage/google_cloud_cpp_storage.bzl b/google/cloud/storage/google_cloud_cpp_storage.bzl
index ed40ddb04ab5f..de3c9b4516942 100644
--- a/google/cloud/storage/google_cloud_cpp_storage.bzl
+++ b/google/cloud/storage/google_cloud_cpp_storage.bzl
@@ -49,6 +49,7 @@ google_cloud_cpp_storage_hdrs = [
     "internal/binary_data_as_debug_string.h",
     "internal/bucket_access_control_parser.h",
     "internal/bucket_acl_requests.h",
+    "internal/bucket_metadata_cache.h",
     "internal/bucket_metadata_parser.h",
     "internal/bucket_requests.h",
     "internal/complex_option.h",
@@ -164,6 +165,7 @@ google_cloud_cpp_storage_srcs = [
     "internal/base64.cc",
     "internal/bucket_access_control_parser.cc",
     "internal/bucket_acl_requests.cc",
+    "internal/bucket_metadata_cache.cc",
     "internal/bucket_metadata_parser.cc",
     "internal/bucket_requests.cc",
     "internal/compute_engine_util.cc",
diff --git a/google/cloud/storage/google_cloud_cpp_storage.cmake b/google/cloud/storage/google_cloud_cpp_storage.cmake
index f3d70b6766817..52e106efc9e5c 100644
--- a/google/cloud/storage/google_cloud_cpp_storage.cmake
+++ b/google/cloud/storage/google_cloud_cpp_storage.cmake
@@ -74,6 +74,8 @@ add_library(
     internal/bucket_access_control_parser.h
     internal/bucket_acl_requests.cc
     internal/bucket_acl_requests.h
+    internal/bucket_metadata_cache.cc
+    internal/bucket_metadata_cache.h
     internal/bucket_metadata_parser.cc
     internal/bucket_metadata_parser.h
     internal/bucket_requests.cc
@@ -420,6 +422,7 @@ if (BUILD_TESTING)
         idempotency_policy_test.cc
         internal/base64_test.cc
         internal/bucket_acl_requests_test.cc
+        internal/bucket_metadata_cache_test.cc
         internal/bucket_requests_test.cc
         internal/complex_option_test.cc
         internal/compute_engine_util_test.cc
diff --git a/google/cloud/storage/internal/bucket_metadata_cache.cc b/google/cloud/storage/internal/bucket_metadata_cache.cc
new file mode 100644
index 0000000000000..f663eb31714fc
--- /dev/null
+++ b/google/cloud/storage/internal/bucket_metadata_cache.cc
@@ -0,0 +1,105 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "google/cloud/storage/internal/bucket_metadata_cache.h"
+#include "google/cloud/storage/bucket_metadata.h"
+#include <mutex>
+#include <utility>
+
+namespace google {
+namespace cloud {
+namespace storage_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+BucketCacheEntry BucketCacheEntry::FromMetadata(
+    storage::BucketMetadata const& m) {
+  std::string loc = m.location();
+  if (m.location_type() == "multi-region" ||
+      m.location_type() == "dual-region") {
+    loc = "global";
+  }
+  return {
+      "projects/" + std::to_string(m.project_number()) + "/buckets/" + m.name(),
+      std::move(loc)};
+}
+
+void BucketMetadataCache::MoveToFront(std::list<std::string>::iterator it) {
+  list_.splice(list_.begin(), list_, it);
+}
+
+absl::optional<BucketCacheEntry> BucketMetadataCache::Get(
+    std::string const& bucket_name) {
+  std::unique_lock<std::mutex> lk(mu_);
+  auto it = map_.find(bucket_name);
+  if (it == map_.end()) return absl::nullopt;
+
+  MoveToFront(it->second.second);
+  return it->second.first;
+}
+
+void BucketMetadataCache::Put(std::string const& bucket_name,
+                              BucketCacheEntry entry) {
+  if (max_size_ == 0) return;
+  std::unique_lock<std::mutex> lk(mu_);
+  auto it = map_.find(bucket_name);
+  if (it != map_.end()) {
+    it->second.first = std::move(entry);
+    MoveToFront(it->second.second);
+    return;
+  }
+
+  if (map_.size() >= max_size_ && !list_.empty()) {
+    auto oldest = list_.back();
+    list_.pop_back();
+    map_.erase(oldest);
+  }
+
+  list_.push_front(bucket_name);
+  map_[bucket_name] = {std::move(entry), list_.begin()};
+}
+
+void BucketMetadataCache::Invalidate(std::string const& bucket_name) {
+  std::unique_lock<std::mutex> lk(mu_);
+  auto it = map_.find(bucket_name);
+  if (it != map_.end()) {
+    list_.erase(it->second.second);
+    map_.erase(it);
+  }
+}
+
+void BucketMetadataCache::Clear() {
+  std::unique_lock<std::mutex> lk(mu_);
+  map_.clear();
+  list_.clear();
+  in_flight_fetch_.clear();
+}
+
+bool BucketMetadataCache::StartFetch(std::string const& bucket_name) {
+  std::unique_lock<std::mutex> lk(mu_);
+  if (in_flight_fetch_.find(bucket_name) != in_flight_fetch_.end()) {
+    return false;
+  }
+  in_flight_fetch_.insert(bucket_name);
+  return true;
+}
+
+void BucketMetadataCache::EndFetch(std::string const& bucket_name) {
+  std::unique_lock<std::mutex> lk(mu_);
+  in_flight_fetch_.erase(bucket_name);
+}
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace storage_internal
+}  // namespace cloud
+}  // namespace google
diff --git a/google/cloud/storage/internal/bucket_metadata_cache.h b/google/cloud/storage/internal/bucket_metadata_cache.h
new file mode 100644
index 0000000000000..172dc48772f8b
--- /dev/null
+++ b/google/cloud/storage/internal/bucket_metadata_cache.h
@@ -0,0 +1,74 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H
+
+#include "google/cloud/storage/version.h"
+#include "absl/types/optional.h"
+#include <cstddef>
+#include <list>
+#include <mutex>
+#include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+
+namespace google {
+namespace cloud {
+namespace storage {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+class BucketMetadata;
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace storage
+namespace storage_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+struct BucketCacheEntry {
+  std::string id;
+  std::string location;
+
+  static BucketCacheEntry FromMetadata(storage::BucketMetadata const& m);
+};
+
+class BucketMetadataCache {
+ public:
+  explicit BucketMetadataCache(std::size_t max_size = 10000)
+      : max_size_(max_size) {}
+
+  absl::optional<BucketCacheEntry> Get(std::string const& bucket_name);
+  void Put(std::string const& bucket_name, BucketCacheEntry entry);
+  void Invalidate(std::string const& bucket_name);
+  void Clear();
+  bool StartFetch(std::string const& bucket_name);
+  void EndFetch(std::string const& bucket_name);
+
+ private:
+  void MoveToFront(std::list<std::string>::iterator it);
+
+  std::size_t max_size_;
+  std::mutex mu_;
+  std::list<std::string> list_;
+  std::unordered_map<std::string, std::pair<BucketCacheEntry,
+                                            std::list<std::string>::iterator>>
+      map_;
+  std::unordered_set<std::string> in_flight_fetch_;
+};
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace storage_internal
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H
diff --git a/google/cloud/storage/internal/bucket_metadata_cache_test.cc b/google/cloud/storage/internal/bucket_metadata_cache_test.cc
new file mode 100644
index 0000000000000..9290e7b7d4e6f
--- /dev/null
+++ b/google/cloud/storage/internal/bucket_metadata_cache_test.cc
@@ -0,0 +1,103 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "google/cloud/storage/internal/bucket_metadata_cache.h"
+#include <gmock/gmock.h>
+
+namespace google {
+namespace cloud {
+namespace storage_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+namespace {
+
+using ::testing::Eq;
+using ::testing::IsFalse;
+using ::testing::IsTrue;
+
+TEST(BucketMetadataCacheTest, HitAndMiss) {
+  BucketMetadataCache cache(10);
+  EXPECT_FALSE(cache.Get("test-bucket").has_value());
+
+  BucketCacheEntry entry{"projects/123/buckets/test-bucket", "us-central1"};
+  cache.Put("test-bucket", entry);
+
+  auto res = cache.Get("test-bucket");
+  ASSERT_TRUE(res.has_value());
+  EXPECT_THAT(res->id, Eq("projects/123/buckets/test-bucket"));
+  EXPECT_THAT(res->location, Eq("us-central1"));
+}
+
+TEST(BucketMetadataCacheTest, PutUpdatesExisting) {
+  BucketMetadataCache cache(10);
+  BucketCacheEntry entry1{"projects/123/buckets/test-bucket", "us-central1"};
+  cache.Put("test-bucket", entry1);
+
+  BucketCacheEntry entry2{"projects/456/buckets/test-bucket", "global"};
+  cache.Put("test-bucket", entry2);
+
+  auto res = cache.Get("test-bucket");
+  ASSERT_TRUE(res.has_value());
+  EXPECT_THAT(res->id, Eq("projects/456/buckets/test-bucket"));
+  EXPECT_THAT(res->location, Eq("global"));
+}
+
+TEST(BucketMetadataCacheTest, InvalidateAndClear) {
+  BucketMetadataCache cache(10);
+  BucketCacheEntry entry{"projects/123/buckets/test-bucket", "us-central1"};
+  cache.Put("test-bucket", entry);
+  EXPECT_TRUE(cache.Get("test-bucket").has_value());
+
+  cache.Invalidate("test-bucket");
+  EXPECT_FALSE(cache.Get("test-bucket").has_value());
+
+  cache.Put("bucket1", entry);
+  cache.Put("bucket2", entry);
+  EXPECT_TRUE(cache.Get("bucket1").has_value());
+  EXPECT_TRUE(cache.Get("bucket2").has_value());
+
+  cache.Clear();
+  EXPECT_FALSE(cache.Get("bucket1").has_value());
+  EXPECT_FALSE(cache.Get("bucket2").has_value());
+}
+
+TEST(BucketMetadataCacheTest, EvictsOldest) {
+  BucketMetadataCache cache(2);
+  BucketCacheEntry entry{"id", "loc"};
+
+  cache.Put("b1", entry);
+  cache.Put("b2", entry);
+  EXPECT_TRUE(cache.Get("b1").has_value());  // b1 becomes most recent
+
+  cache.Put("b3", entry);  // pushes out b2 (oldest)
+  EXPECT_TRUE(cache.Get("b1").has_value());
+  EXPECT_FALSE(cache.Get("b2").has_value());
+  EXPECT_TRUE(cache.Get("b3").has_value());
+}
+
+TEST(BucketMetadataCacheTest, InFlightFetch) {
+  BucketMetadataCache cache(10);
+  EXPECT_THAT(cache.StartFetch("b1"), IsTrue());
+  EXPECT_THAT(cache.StartFetch("b1"), IsFalse());
+
+  EXPECT_THAT(cache.StartFetch("b2"), IsTrue());
+
+  cache.EndFetch("b1");
+  EXPECT_THAT(cache.StartFetch("b1"), IsTrue());
+}
+
+}  // namespace
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace storage_internal
+}  // namespace cloud
+}  // namespace google
diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc
index 6d7e20a2d349b..706bf839abdbd 100644
--- a/google/cloud/storage/internal/tracing_connection.cc
+++ b/google/cloud/storage/internal/tracing_connection.cc
@@ -14,8 +14,14 @@
 
 #include "google/cloud/storage/internal/tracing_connection.h"
 #include "google/cloud/storage/internal/tracing_object_read_source.h"
+#include "google/cloud/storage/options.h"
 #include "google/cloud/storage/parallel_upload.h"
 #include "google/cloud/internal/opentelemetry.h"
+#include "google/cloud/options.h"
+#if GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC
+#include "google/cloud/grpc_options.h"
+#endif
+#include <algorithm>
 #include <memory>
 #include <string>
 #include <utility>
@@ -26,11 +32,89 @@ namespace cloud {
 namespace storage_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
+namespace {
+std::size_t DefaultThreadPoolSize(Options const& options) {
+#if GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC
+  auto pool_size = options.get<GrpcBackgroundThreadPoolSizeOption>();
+  if (pool_size == 0) return 1U;
+  return pool_size;
+#else
+  (void)options;
+  return 1U;
+#endif
+}
+}  // namespace
+
 TracingConnection::TracingConnection(std::shared_ptr<StorageConnection> impl)
-    : impl_(std::move(impl)) {}
+    : impl_(std::move(impl)),
+      background_threads_(
+          std::make_unique<AutomaticallyCreatedStorageBackgroundThreads>(
+              DefaultThreadPoolSize(impl_->options()))) {}
+
+TracingConnection::~TracingConnection() = default;
+
+BucketMetadataCache& TracingConnection::cache() {
+  static BucketMetadataCache instance(10000);
+  return instance;
+}
+
+void TracingConnection::ResetCacheForTesting() { cache().Clear(); }
 
 Options TracingConnection::options() const { return impl_->options(); }
 
+void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span,
+                                   BucketCacheEntry const& entry) {
+  span.SetAttribute("gcp.resource.destination.id", entry.id);
+  span.SetAttribute("gcp.resource.destination.location", entry.location);
+}
+
+void TracingConnection::MaybeTriggerBackgroundFetch(
+    std::string const& bucket_name) {
+  if (!cache().StartFetch(bucket_name)) {
+    return;
+  }
+
+  auto current_options = google::cloud::internal::SaveCurrentOptions();
+  background_threads_->cq().RunAsync([this, bucket_name, current_options]() {
+    google::cloud::internal::OptionsSpan span(current_options);
+    storage::internal::GetBucketMetadataRequest request(bucket_name);
+    auto result = impl_->GetBucketMetadata(request);
+
+    if (result.ok()) {
+      cache().Put(bucket_name, BucketCacheEntry::FromMetadata(*result));
+    } else if (result.status().code() == StatusCode::kPermissionDenied) {
+      cache().Put(bucket_name, {"projects/_/buckets/" + bucket_name, "global"});
+    }
+
+    cache().EndFetch(bucket_name);
+  });
+}
+
+void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span,
+                                   std::string const& bucket_name) {
+  if (bucket_name.empty()) return;
+  auto const enabled =
+      options().get<storage_experimental::OTelSpanEnrichmentOption>();
+  if (!enabled) return;
+  auto entry = cache().Get(bucket_name);
+  if (entry.has_value()) {
+    EnrichSpan(span, *entry);
+  } else {
+    MaybeTriggerBackgroundFetch(bucket_name);
+  }
+}
+
+void TracingConnection::EnrichSpan(
+    opentelemetry::trace::Span& span,
+    storage::BucketMetadata const& metadata) const {
+  auto const enabled =
+      options().get<storage_experimental::OTelSpanEnrichmentOption>();
+  if (!enabled) return;
+  auto entry = BucketCacheEntry::FromMetadata(metadata);
+  EnrichSpan(span, entry);
+  cache().Put(metadata.name(), std::move(entry));
+}
+
 StatusOr<storage::internal::ListBucketsResponse> TracingConnection::ListBuckets(
     storage::internal::ListBucketsRequest const& request) {
   // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client
@@ -43,49 +127,80 @@ StatusOr<storage::BucketMetadata> TracingConnection::CreateBucket(
     storage::internal::CreateBucketRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::CreateBucket");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->CreateBucket(request));
+  auto result = impl_->CreateBucket(request);
+  if (result.ok()) EnrichSpan(*span, *result);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::BucketMetadata> TracingConnection::GetBucketMetadata(
     storage::internal::GetBucketMetadataRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::GetBucketMetadata");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->GetBucketMetadata(request));
+  auto result = impl_->GetBucketMetadata(request);
+  if (result.ok()) {
+    EnrichSpan(*span, *result);
+  } else {
+    MaybeInvalidate(result, request.bucket_name());
+  }
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::EmptyResponse> TracingConnection::DeleteBucket(
     storage::internal::DeleteBucketRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::DeleteBucket");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->DeleteBucket(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->DeleteBucket(request);
+  if (result.ok() || result.status().code() == StatusCode::kNotFound) {
+    cache().Invalidate(request.bucket_name());
+  }
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::BucketMetadata> TracingConnection::UpdateBucket(
     storage::internal::UpdateBucketRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::UpdateBucket");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->UpdateBucket(request));
+  auto result = impl_->UpdateBucket(request);
+  if (result.ok()) {
+    EnrichSpan(*span, *result);
+  } else {
+    MaybeInvalidate(result, request.metadata().name());
+  }
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::BucketMetadata> TracingConnection::PatchBucket(
     storage::internal::PatchBucketRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::PatchBucket");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->PatchBucket(request));
+  auto result = impl_->PatchBucket(request);
+  if (result.ok()) {
+    EnrichSpan(*span, *result);
+  } else {
+    MaybeInvalidate(result, request.bucket());
+  }
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::NativeIamPolicy> TracingConnection::GetNativeBucketIamPolicy(
     storage::internal::GetBucketIamPolicyRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::GetNativeBucketIamPolicy");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->GetNativeBucketIamPolicy(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->GetNativeBucketIamPolicy(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::NativeIamPolicy> TracingConnection::SetNativeBucketIamPolicy(
     storage::internal::SetNativeBucketIamPolicyRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::SetNativeBucketIamPolicy");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->SetNativeBucketIamPolicy(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->SetNativeBucketIamPolicy(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::TestBucketIamPermissionsResponse>
@@ -93,35 +208,46 @@ TracingConnection::TestBucketIamPermissions(
     storage::internal::TestBucketIamPermissionsRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::TestBucketIamPermissions");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->TestBucketIamPermissions(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->TestBucketIamPermissions(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::BucketMetadata> TracingConnection::LockBucketRetentionPolicy(
     storage::internal::LockBucketRetentionPolicyRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::LockBucketRetentionPolicy");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->LockBucketRetentionPolicy(request));
+  auto result = impl_->LockBucketRetentionPolicy(request);
+  if (result.ok()) EnrichSpan(*span, *result);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectMetadata> TracingConnection::InsertObjectMedia(
     storage::internal::InsertObjectMediaRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::InsertObjectMedia");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->InsertObjectMedia(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->InsertObjectMedia(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectMetadata> TracingConnection::CopyObject(
     storage::internal::CopyObjectRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::CopyObject");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->CopyObject(request));
+  EnrichSpan(*span, request.destination_bucket());
+  auto result = impl_->CopyObject(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectMetadata> TracingConnection::GetObjectMetadata(
     storage::internal::GetObjectMetadataRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::GetObjectMetadata");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->GetObjectMetadata(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->GetObjectMetadata(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<std::unique_ptr<storage::internal::ObjectReadSource>>
@@ -129,8 +255,11 @@ TracingConnection::ReadObject(
     storage::internal::ReadObjectRangeRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::ReadObject");
   auto scope = opentelemetry::trace::Scope(span);
+  EnrichSpan(*span, request.bucket_name());
   auto reader = impl_->ReadObject(request);
-  if (!reader) return internal::EndSpan(*span, std::move(reader));
+  if (!reader) {
+    return internal::EndSpan(*span, std::move(reader));
+  }
   return std::unique_ptr<storage::internal::ObjectReadSource>(
       std::make_unique<TracingObjectReadSource>(std::move(span),
                                                 *std::move(reader)));
@@ -141,42 +270,55 @@ StatusOr<storage::internal::ListObjectsResponse> TracingConnection::ListObjects(
   // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client
   auto span = internal::MakeSpan("storage::Client::ListObjects");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->ListObjects(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->ListObjects(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::EmptyResponse> TracingConnection::DeleteObject(
     storage::internal::DeleteObjectRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::DeleteObject");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->DeleteObject(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->DeleteObject(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectMetadata> TracingConnection::UpdateObject(
     storage::internal::UpdateObjectRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::UpdateObject");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->UpdateObject(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->UpdateObject(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectMetadata> TracingConnection::MoveObject(
     storage::internal::MoveObjectRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::MoveObject");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->MoveObject(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->MoveObject(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectMetadata> TracingConnection::PatchObject(
     storage::internal::PatchObjectRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::PatchObject");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->PatchObject(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->PatchObject(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectMetadata> TracingConnection::ComposeObject(
     storage::internal::ComposeObjectRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::ComposeObject");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->ComposeObject(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->ComposeObject(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::RewriteObjectResponse>
@@ -184,14 +326,18 @@ TracingConnection::RewriteObject(
     storage::internal::RewriteObjectRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::RewriteObject");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->RewriteObject(request));
+  EnrichSpan(*span, request.destination_bucket());
+  auto result = impl_->RewriteObject(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectMetadata> TracingConnection::RestoreObject(
     storage::internal::RestoreObjectRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::RestoreObject");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->RestoreObject(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->RestoreObject(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::CreateResumableUploadResponse>
@@ -201,7 +347,9 @@ TracingConnection::CreateResumableUpload(
   auto span =
       internal::MakeSpan("storage::Client::WriteObject/CreateResumableUpload");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->CreateResumableUpload(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->CreateResumableUpload(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::QueryResumableUploadResponse>
@@ -237,8 +385,9 @@ StatusOr<std::unique_ptr<std::string>> TracingConnection::UploadFileSimple(
   auto span =
       internal::MakeSpan("storage::Client::UploadFile/UploadFileSimple");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(
-      *span, impl_->UploadFileSimple(file_name, file_size, request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->UploadFileSimple(file_name, file_size, request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<std::unique_ptr<std::istream>> TracingConnection::UploadFileResumable(
@@ -247,8 +396,9 @@ StatusOr<std::unique_ptr<std::istream>> TracingConnection::UploadFileResumable(
   auto span =
       internal::MakeSpan("storage::Client::UploadFile/UploadFileResumable");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span,
-                           impl_->UploadFileResumable(file_name, request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->UploadFileResumable(file_name, request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 Status TracingConnection::DownloadStreamToFile(
@@ -257,8 +407,10 @@ Status TracingConnection::DownloadStreamToFile(
   auto span = internal::MakeSpan(
       "storage::Client::DownloadToFile/DownloadStreamToFile");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->DownloadStreamToFile(
-                                      std::move(stream), file_name, request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result =
+      impl_->DownloadStreamToFile(std::move(stream), file_name, request);
+  return internal::EndSpan(*span, result);
 }
 
 StatusOr<storage::ObjectMetadata> TracingConnection::ExecuteParallelUploadFile(
@@ -279,42 +431,60 @@ TracingConnection::ListBucketAcl(
   // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client
   auto span = internal::MakeSpan("storage::Client::ListBucketAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->ListBucketAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->ListBucketAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::BucketAccessControl> TracingConnection::CreateBucketAcl(
     storage::internal::CreateBucketAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::CreateBucketAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->CreateBucketAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->CreateBucketAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::EmptyResponse> TracingConnection::DeleteBucketAcl(
     storage::internal::DeleteBucketAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::DeleteBucketAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->DeleteBucketAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->DeleteBucketAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::BucketAccessControl> TracingConnection::GetBucketAcl(
     storage::internal::GetBucketAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::GetBucketAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->GetBucketAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->GetBucketAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::BucketAccessControl> TracingConnection::UpdateBucketAcl(
     storage::internal::UpdateBucketAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::UpdateBucketAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->UpdateBucketAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->UpdateBucketAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::BucketAccessControl> TracingConnection::PatchBucketAcl(
     storage::internal::PatchBucketAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::PatchBucketAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->PatchBucketAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->PatchBucketAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::ListObjectAclResponse>
@@ -323,42 +493,54 @@ TracingConnection::ListObjectAcl(
   // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client
   auto span = internal::MakeSpan("storage::Client::ListObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->ListObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->ListObjectAcl(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectAccessControl> TracingConnection::CreateObjectAcl(
     storage::internal::CreateObjectAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::CreateObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->CreateObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->CreateObjectAcl(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::EmptyResponse> TracingConnection::DeleteObjectAcl(
     storage::internal::DeleteObjectAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::DeleteObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->DeleteObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->DeleteObjectAcl(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectAccessControl> TracingConnection::GetObjectAcl(
     storage::internal::GetObjectAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::GetObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->GetObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->GetObjectAcl(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectAccessControl> TracingConnection::UpdateObjectAcl(
     storage::internal::UpdateObjectAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::UpdateObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->UpdateObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->UpdateObjectAcl(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectAccessControl> TracingConnection::PatchObjectAcl(
     storage::internal::PatchObjectAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::PatchObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->PatchObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->PatchObjectAcl(request);
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::ListDefaultObjectAclResponse>
@@ -367,7 +549,10 @@ TracingConnection::ListDefaultObjectAcl(
   // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client
   auto span = internal::MakeSpan("storage::Client::ListDefaultObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->ListDefaultObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->ListDefaultObjectAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectAccessControl>
@@ -375,7 +560,10 @@ TracingConnection::CreateDefaultObjectAcl(
     storage::internal::CreateDefaultObjectAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::CreateDefaultObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->CreateDefaultObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->CreateDefaultObjectAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::EmptyResponse>
@@ -383,14 +571,20 @@ TracingConnection::DeleteDefaultObjectAcl(
     storage::internal::DeleteDefaultObjectAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::DeleteDefaultObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->DeleteDefaultObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->DeleteDefaultObjectAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectAccessControl> TracingConnection::GetDefaultObjectAcl(
     storage::internal::GetDefaultObjectAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::GetDefaultObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->GetDefaultObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->GetDefaultObjectAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectAccessControl>
@@ -398,14 +592,20 @@ TracingConnection::UpdateDefaultObjectAcl(
     storage::internal::UpdateDefaultObjectAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::UpdateDefaultObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->UpdateDefaultObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->UpdateDefaultObjectAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ObjectAccessControl> TracingConnection::PatchDefaultObjectAcl(
     storage::internal::PatchDefaultObjectAclRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::PatchDefaultObjectAcl");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->PatchDefaultObjectAcl(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->PatchDefaultObjectAcl(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::ServiceAccount> TracingConnection::GetServiceAccount(
@@ -466,21 +666,30 @@ TracingConnection::ListNotifications(
   // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client
   auto span = internal::MakeSpan("storage::Client::ListNotifications");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->ListNotifications(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->ListNotifications(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::NotificationMetadata> TracingConnection::CreateNotification(
     storage::internal::CreateNotificationRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::CreateNotification");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->CreateNotification(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->CreateNotification(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::NotificationMetadata> TracingConnection::GetNotification(
     storage::internal::GetNotificationRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::GetNotification");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->GetNotification(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->GetNotification(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 StatusOr<storage::internal::EmptyResponse>
@@ -488,7 +697,10 @@ TracingConnection::DeleteNotification(
     storage::internal::DeleteNotificationRequest const& request) {
   auto span = internal::MakeSpan("storage::Client::DeleteNotification");
   auto scope = opentelemetry::trace::Scope(span);
-  return internal::EndSpan(*span, impl_->DeleteNotification(request));
+  EnrichSpan(*span, request.bucket_name());
+  auto result = impl_->DeleteNotification(request);
+  MaybeInvalidate(result, request.bucket_name());
+  return internal::EndSpan(*span, std::move(result));
 }
 
 std::vector<std::string> TracingConnection::InspectStackStructure() const {
diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h
index 45df6239f956d..0ca9643905354 100644
--- a/google/cloud/storage/internal/tracing_connection.h
+++ b/google/cloud/storage/internal/tracing_connection.h
@@ -15,12 +15,19 @@
 #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_TRACING_CONNECTION_H
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_TRACING_CONNECTION_H
 
+#include "google/cloud/storage/internal/bucket_metadata_cache.h"
 #include "google/cloud/storage/internal/storage_connection.h"
 #include "google/cloud/storage/parallel_upload.h"
 #include "google/cloud/storage/version.h"
+#include "google/cloud/internal/generic_background_threads_impl.h"
+#if GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC
+#include "google/cloud/background_threads.h"
+#include "google/cloud/completion_queue.h"
+#else
+#include "google/cloud/internal/rest_pure_background_threads_impl.h"
+#endif
 #include <memory>
 #include <string>
-#include <vector>
 
 namespace google {
 namespace cloud {
@@ -30,7 +37,9 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 class TracingConnection : public storage::internal::StorageConnection {
  public:
   explicit TracingConnection(std::shared_ptr<StorageConnection> impl);
-  ~TracingConnection() override = default;
+  ~TracingConnection() override;
+
+  static void ResetCacheForTesting();
 
   Options options() const override;
 
@@ -178,7 +187,46 @@ class TracingConnection : public storage::internal::StorageConnection {
   std::vector<std::string> InspectStackStructure() const override;
 
  private:
+  void EnrichSpan(opentelemetry::trace::Span& span,
+                  std::string const& bucket_name);
+  void EnrichSpan(opentelemetry::trace::Span& span,
+                  storage::BucketMetadata const& metadata) const;
+  static void EnrichSpan(opentelemetry::trace::Span& span,
+                         BucketCacheEntry const& entry);
+  void MaybeTriggerBackgroundFetch(std::string const& bucket_name);
+
+  static void MaybeInvalidate(Status const& status,
+                              std::string const& bucket_name) {
+    if (!status.ok() && status.code() == StatusCode::kNotFound) {
+      cache().Invalidate(bucket_name);
+    }
+  }
+
+  template <typename T>
+  static void MaybeInvalidate(StatusOr<T> const& result,
+                              std::string const& bucket_name) {
+    MaybeInvalidate(result.status(), bucket_name);
+  }
+
+  static BucketMetadataCache& cache();
+
+#if GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC
+  using StorageBackgroundThreads = google::cloud::BackgroundThreads;
+  using AutomaticallyCreatedStorageBackgroundThreads =
+      google::cloud::internal::AutomaticallyCreatedBackgroundThreadsImpl<
+          google::cloud::CompletionQueue, google::cloud::BackgroundThreads>;
+#else
+  using StorageBackgroundThreads =
+      google::cloud::rest_internal::RestPureBackgroundThreads;
+  using AutomaticallyCreatedStorageBackgroundThreads =
+      google::cloud::internal::AutomaticallyCreatedBackgroundThreadsImpl<
+          rest_internal::RestPureCompletionQueue,
+          rest_internal::RestPureBackgroundThreads,
+          rest_internal::RestPureQueueTraits>;
+#endif
+
   std::shared_ptr<StorageConnection> impl_;
+  std::unique_ptr<StorageBackgroundThreads> background_threads_;
 };
 
 std::shared_ptr<storage::internal::StorageConnection> MakeTracingClient(
diff --git a/google/cloud/storage/internal/tracing_connection_test.cc b/google/cloud/storage/internal/tracing_connection_test.cc
index bda9bc46d0a2d..79d130996005b 100644
--- a/google/cloud/storage/internal/tracing_connection_test.cc
+++ b/google/cloud/storage/internal/tracing_connection_test.cc
@@ -22,6 +22,7 @@
 #include <gmock/gmock.h>
 #include <memory>
 #include <string>
+#include <thread>
 #include <utility>
 
 namespace google {
@@ -33,7 +34,9 @@ namespace {
 using ::google::cloud::storage::testing::MockClient;
 using ::google::cloud::storage::testing::MockObjectReadSource;
 using ::google::cloud::storage::testing::canonical_errors::PermanentError;
+using ::google::cloud::storage::testing::canonical_errors::TransientError;
 using ::google::cloud::testing_util::InstallSpanCatcher;
+using ::google::cloud::testing_util::IsOk;
 using ::google::cloud::testing_util::OTelAttribute;
 using ::google::cloud::testing_util::SpanHasAttributes;
 using ::google::cloud::testing_util::SpanHasInstrumentationScope;
@@ -54,7 +57,8 @@ TEST(TracingClientTest, Options) {
   };
 
   auto mock = std::make_shared<MockClient>();
-  EXPECT_CALL(*mock, options).WillOnce(Return(Options{}.set<TestOption>(42)));
+  EXPECT_CALL(*mock, options)
+      .WillRepeatedly(Return(Options{}.set<TestOption>(42)));
   auto under_test = TracingConnection(mock);
   auto const options = under_test.options();
   EXPECT_EQ(42, options.get<TestOption>());
@@ -105,6 +109,35 @@ TEST(TracingClientTest, CreateBucket) {
                       "gl-cpp.status_code", code_str)))));
 }
 
+TEST(TracingClientTest, CreateBucketSuccess) {
+  auto span_catcher = InstallSpanCatcher();
+  auto mock = std::make_shared<MockClient>();
+  EXPECT_CALL(*mock, CreateBucket).WillOnce([](auto const&) {
+    EXPECT_TRUE(ThereIsAnActiveSpan());
+    storage::BucketMetadata metadata;
+    metadata.set_name("test-bucket");
+    metadata.set_project_number(123456);
+    metadata.set_location("us-east1");
+    metadata.set_location_type("regional");
+    return metadata;
+  });
+  auto under_test = TracingConnection(mock);
+  auto actual =
+      under_test.CreateBucket(storage::internal::CreateBucketRequest());
+  EXPECT_THAT(actual, IsOk());
+  EXPECT_THAT(
+      span_catcher->GetSpans(),
+      ElementsAre(AllOf(
+          SpanHasInstrumentationScope(), SpanKindIsClient(),
+          SpanNamed("storage::Client::CreateBucket"),
+          SpanWithStatus(opentelemetry::trace::StatusCode::kOk),
+          SpanHasAttributes(
+              OTelAttribute<std::string>("gcp.resource.destination.id",
+                                         "projects/123456/buckets/test-bucket"),
+              OTelAttribute<std::string>("gcp.resource.destination.location",
+                                         "us-east1")))));
+}
+
 TEST(TracingClientTest, GetBucketMetadata) {
   auto span_catcher = InstallSpanCatcher();
   auto mock = std::make_shared<MockClient>();
@@ -128,6 +161,89 @@ TEST(TracingClientTest, GetBucketMetadata) {
                       "gl-cpp.status_code", code_str)))));
 }
 
+TEST(TracingClientTest, GetBucketMetadataSuccess) {
+  auto span_catcher = InstallSpanCatcher();
+  auto mock = std::make_shared<MockClient>();
+  EXPECT_CALL(*mock, GetBucketMetadata).WillOnce([](auto const&) {
+    EXPECT_TRUE(ThereIsAnActiveSpan());
+    storage::BucketMetadata metadata;
+    metadata.set_name("test-bucket");
+    metadata.set_project_number(123456);
+    metadata.set_location("us-east1");
+    metadata.set_location_type("regional");
+    return metadata;
+  });
+  auto under_test = TracingConnection(mock);
+  auto actual = under_test.GetBucketMetadata(
+      storage::internal::GetBucketMetadataRequest("test-bucket"));
+  EXPECT_THAT(actual, IsOk());
+  EXPECT_THAT(
+      span_catcher->GetSpans(),
+      ElementsAre(AllOf(
+          SpanHasInstrumentationScope(), SpanKindIsClient(),
+          SpanNamed("storage::Client::GetBucketMetadata"),
+          SpanWithStatus(opentelemetry::trace::StatusCode::kOk),
+          SpanHasAttributes(
+              OTelAttribute<std::string>("gcp.resource.destination.id",
+                                         "projects/123456/buckets/test-bucket"),
+              OTelAttribute<std::string>("gcp.resource.destination.location",
+                                         "us-east1")))));
+}
+
+TEST(TracingClientTest, BucketMetadataCacheSuccess) {
+  TracingConnection::ResetCacheForTesting();
+  auto span_catcher = InstallSpanCatcher();
+  auto mock = std::make_shared<MockClient>();
+
+  std::promise<void> bg_fetch_done;
+  auto bg_fetch_future = bg_fetch_done.get_future();
+
+  EXPECT_CALL(*mock, GetObjectMetadata).WillOnce([](auto const&) {
+    return TransientError();
+  });
+
+  EXPECT_CALL(*mock, GetBucketMetadata)
+      .WillOnce([&bg_fetch_done](auto const& request) {
+        EXPECT_EQ("test-bucket", request.bucket_name());
+        storage::BucketMetadata metadata;
+        metadata.set_name("test-bucket");
+        metadata.set_project_number(123456);
+        metadata.set_location("us-east1");
+        metadata.set_location_type("regional");
+        bg_fetch_done.set_value();
+        return metadata;
+      });
+
+  auto under_test = TracingConnection(mock);
+
+  (void)under_test.GetObjectMetadata(
+      storage::internal::GetObjectMetadataRequest("test-bucket",
+                                                  "test-object"));
+
+  bg_fetch_future.wait_for(std::chrono::seconds(5));
+  std::this_thread::sleep_for(std::chrono::milliseconds(50));
+
+  EXPECT_CALL(*mock, DeleteObject).WillOnce([](auto const&) {
+    return PermanentError();
+  });
+
+  // Clear spans from GetObjectMetadata
+  (void)span_catcher->GetSpans();
+
+  (void)under_test.DeleteObject(
+      storage::internal::DeleteObjectRequest("test-bucket", "test-object"));
+
+  EXPECT_THAT(
+      span_catcher->GetSpans(),
+      ElementsAre(AllOf(
+          SpanNamed("storage::Client::DeleteObject"),
+          SpanHasAttributes(
+              OTelAttribute<std::string>("gcp.resource.destination.id",
+                                         "projects/123456/buckets/test-bucket"),
+              OTelAttribute<std::string>("gcp.resource.destination.location",
+                                         "us-east1")))));
+}
+
 TEST(TracingClientTest, DeleteBucket) {
   auto span_catcher = InstallSpanCatcher();
   auto mock = std::make_shared<MockClient>();
@@ -174,6 +290,35 @@ TEST(TracingClientTest, UpdateBucket) {
                       "gl-cpp.status_code", code_str)))));
 }
 
+TEST(TracingClientTest, UpdateBucketSuccess) {
+  auto span_catcher = InstallSpanCatcher();
+  auto mock = std::make_shared<MockClient>();
+  EXPECT_CALL(*mock, UpdateBucket).WillOnce([](auto const&) {
+    EXPECT_TRUE(ThereIsAnActiveSpan());
+    storage::BucketMetadata metadata;
+    metadata.set_name("test-bucket");
+    metadata.set_project_number(123456);
+    metadata.set_location("us-east1");
+    metadata.set_location_type("regional");
+    return metadata;
+  });
+  auto under_test = TracingConnection(mock);
+  auto actual =
+      under_test.UpdateBucket(storage::internal::UpdateBucketRequest());
+  EXPECT_THAT(actual, IsOk());
+  EXPECT_THAT(
+      span_catcher->GetSpans(),
+      ElementsAre(AllOf(
+          SpanHasInstrumentationScope(), SpanKindIsClient(),
+          SpanNamed("storage::Client::UpdateBucket"),
+          SpanWithStatus(opentelemetry::trace::StatusCode::kOk),
+          SpanHasAttributes(
+              OTelAttribute<std::string>("gcp.resource.destination.id",
+                                         "projects/123456/buckets/test-bucket"),
+              OTelAttribute<std::string>("gcp.resource.destination.location",
+                                         "us-east1")))));
+}
+
 TEST(TracingClientTest, PatchBucket) {
   auto span_catcher = InstallSpanCatcher();
   auto mock = std::make_shared<MockClient>();
@@ -196,6 +341,34 @@ TEST(TracingClientTest, PatchBucket) {
                       "gl-cpp.status_code", code_str)))));
 }
 
+TEST(TracingClientTest, PatchBucketSuccess) {
+  auto span_catcher = InstallSpanCatcher();
+  auto mock = std::make_shared<MockClient>();
+  EXPECT_CALL(*mock, PatchBucket).WillOnce([](auto const&) {
+    EXPECT_TRUE(ThereIsAnActiveSpan());
+    storage::BucketMetadata metadata;
+    metadata.set_name("test-bucket");
+    metadata.set_project_number(123456);
+    metadata.set_location("us-east1");
+    metadata.set_location_type("regional");
+    return metadata;
+  });
+  auto under_test = TracingConnection(mock);
+  auto actual = under_test.PatchBucket(storage::internal::PatchBucketRequest());
+  EXPECT_THAT(actual, IsOk());
+  EXPECT_THAT(
+      span_catcher->GetSpans(),
+      ElementsAre(AllOf(
+          SpanHasInstrumentationScope(), SpanKindIsClient(),
+          SpanNamed("storage::Client::PatchBucket"),
+          SpanWithStatus(opentelemetry::trace::StatusCode::kOk),
+          SpanHasAttributes(
+              OTelAttribute<std::string>("gcp.resource.destination.id",
+                                         "projects/123456/buckets/test-bucket"),
+              OTelAttribute<std::string>("gcp.resource.destination.location",
+                                         "us-east1")))));
+}
+
 TEST(TracingClientTest, GetNativeBucketIamPolicy) {
   auto span_catcher = InstallSpanCatcher();
   auto mock = std::make_shared<MockClient>();
@@ -290,6 +463,35 @@ TEST(TracingClientTest, LockBucketRetentionPolicy) {
                       "gl-cpp.status_code", code_str)))));
 }
 
+TEST(TracingClientTest, LockBucketRetentionPolicySuccess) {
+  auto span_catcher = InstallSpanCatcher();
+  auto mock = std::make_shared<MockClient>();
+  EXPECT_CALL(*mock, LockBucketRetentionPolicy).WillOnce([](auto const&) {
+    EXPECT_TRUE(ThereIsAnActiveSpan());
+    storage::BucketMetadata metadata;
+    metadata.set_name("test-bucket");
+    metadata.set_project_number(123456);
+    metadata.set_location("us-east1");
+    metadata.set_location_type("regional");
+    return metadata;
+  });
+  auto under_test = TracingConnection(mock);
+  auto actual = under_test.LockBucketRetentionPolicy(
+      storage::internal::LockBucketRetentionPolicyRequest());
+  EXPECT_THAT(actual, IsOk());
+  EXPECT_THAT(
+      span_catcher->GetSpans(),
+      ElementsAre(AllOf(
+          SpanHasInstrumentationScope(), SpanKindIsClient(),
+          SpanNamed("storage::Client::LockBucketRetentionPolicy"),
+          SpanWithStatus(opentelemetry::trace::StatusCode::kOk),
+          SpanHasAttributes(
+              OTelAttribute<std::string>("gcp.resource.destination.id",
+                                         "projects/123456/buckets/test-bucket"),
+              OTelAttribute<std::string>("gcp.resource.destination.location",
+                                         "us-east1")))));
+}
+
 TEST(TracingClientTest, InsertObjectMedia) {
   auto span_catcher = InstallSpanCatcher();
   auto mock = std::make_shared<MockClient>();
@@ -1484,6 +1686,99 @@ TEST(TracingClientTest, DeleteNotification) {
                       "gl-cpp.status_code", code_str)))));
 }
 
+TEST(TracingClientTest, BucketMetadataMaybeInvalidateBucketLevelEvict) {
+  TracingConnection::ResetCacheForTesting();
+  auto mock = std::make_shared<MockClient>();
+
+  EXPECT_CALL(*mock, options)
+      .WillRepeatedly(testing::Return(
+          Options{}.set<storage_experimental::OTelSpanEnrichmentOption>(true)));
+
+  // Seed cache
+  EXPECT_CALL(*mock, GetBucketMetadata).WillOnce([](auto const&) {
+    storage::BucketMetadata metadata;
+    metadata.set_name("test-bucket");
+    metadata.set_project_number(123456);
+    metadata.set_location("us-east1");
+    metadata.set_location_type("regional");
+    return metadata;
+  });
+
+  auto under_test = TracingConnection(mock);
+  (void)under_test.GetBucketMetadata(
+      storage::internal::GetBucketMetadataRequest("test-bucket"));
+
+  // Fail a bucket-level operation with 404 (DeleteBucket)
+  EXPECT_CALL(*mock, DeleteBucket).WillOnce([](auto const&) {
+    return Status(StatusCode::kNotFound, "Bucket not found");
+  });
+  (void)under_test.DeleteBucket(
+      storage::internal::DeleteBucketRequest("test-bucket"));
+
+  // Verify that the cache entry was evicted.
+  testing::Mock::VerifyAndClearExpectations(mock.get());
+  EXPECT_CALL(*mock, options)
+      .WillRepeatedly(testing::Return(
+          Options{}.set<storage_experimental::OTelSpanEnrichmentOption>(true)));
+
+  EXPECT_CALL(*mock, GetObjectMetadata).WillOnce([](auto const&) {
+    return storage::ObjectMetadata();
+  });
+  EXPECT_CALL(*mock, GetBucketMetadata).WillOnce([](auto const&) {
+    return Status(StatusCode::kNotFound, "Bucket not found");
+  });
+
+  (void)under_test.GetObjectMetadata(
+      storage::internal::GetObjectMetadataRequest("test-bucket",
+                                                  "test-object"));
+}
+
+TEST(TracingClientTest, BucketMetadataMaybeInvalidateObjectLevelNoEvict) {
+  TracingConnection::ResetCacheForTesting();
+  auto mock = std::make_shared<MockClient>();
+
+  EXPECT_CALL(*mock, options)
+      .WillRepeatedly(testing::Return(
+          Options{}.set<storage_experimental::OTelSpanEnrichmentOption>(true)));
+
+  // Seed cache
+  EXPECT_CALL(*mock, GetBucketMetadata).WillOnce([](auto const&) {
+    storage::BucketMetadata metadata;
+    metadata.set_name("test-bucket");
+    metadata.set_project_number(123456);
+    metadata.set_location("us-east1");
+    metadata.set_location_type("regional");
+    return metadata;
+  });
+
+  auto under_test = TracingConnection(mock);
+  (void)under_test.GetBucketMetadata(
+      storage::internal::GetBucketMetadataRequest("test-bucket"));
+
+  // Fail an object-level operation with 404 (GetObjectMetadata)
+  EXPECT_CALL(*mock, GetObjectMetadata).WillOnce([](auto const&) {
+    return Status(StatusCode::kNotFound, "Object not found");
+  });
+  (void)under_test.GetObjectMetadata(
+      storage::internal::GetObjectMetadataRequest("test-bucket",
+                                                  "test-object"));
+
+  // Verify that the cache entry was NOT evicted.
+  testing::Mock::VerifyAndClearExpectations(mock.get());
+  EXPECT_CALL(*mock, options)
+      .WillRepeatedly(testing::Return(
+          Options{}.set<storage_experimental::OTelSpanEnrichmentOption>(true)));
+
+  EXPECT_CALL(*mock, GetObjectMetadata).WillOnce([](auto const&) {
+    return storage::ObjectMetadata();
+  });
+  EXPECT_CALL(*mock, GetBucketMetadata).Times(0);
+
+  (void)under_test.GetObjectMetadata(
+      storage::internal::GetObjectMetadataRequest("test-bucket",
+                                                  "test-object"));
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal
diff --git a/google/cloud/storage/options.h b/google/cloud/storage/options.h
index f64f5d1f28e51..5c3fddeae543b 100644
--- a/google/cloud/storage/options.h
+++ b/google/cloud/storage/options.h
@@ -52,6 +52,19 @@ struct HttpVersionOption {
   using Type = std::string;
 };
 
+/**
+ * Enable/disable OpenTelemetry trace span enrichment with GCS bucket resource
+ * metadata.
+ *
+ * When enabled, the GCS client decorates spans with gcp.resource.destination.id
+ * and location attributes by fetching metadata in the background.
+ *
+ * @ingroup storage-options
+ */
+struct OTelSpanEnrichmentOption {
+  using Type = bool;
+};
+
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_experimental
 
@@ -325,7 +338,8 @@ using ClientOptionList = ::google::cloud::OptionList<
     MaximumCurlSocketRecvSizeOption, MaximumCurlSocketSendSizeOption,
     TransferStallTimeoutOption, RetryPolicyOption, BackoffPolicyOption,
     IdempotencyPolicyOption, CARootsFilePathOption,
-    storage_experimental::HttpVersionOption>;
+    storage_experimental::HttpVersionOption,
+    storage_experimental::OTelSpanEnrichmentOption>;
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage
diff --git a/google/cloud/storage/storage_client_unit_tests.bzl b/google/cloud/storage/storage_client_unit_tests.bzl
index 54c1c64a555b6..c08dc7b2119c7 100644
--- a/google/cloud/storage/storage_client_unit_tests.bzl
+++ b/google/cloud/storage/storage_client_unit_tests.bzl
@@ -43,6 +43,7 @@ storage_client_unit_tests = [
     "idempotency_policy_test.cc",
     "internal/base64_test.cc",
     "internal/bucket_acl_requests_test.cc",
+    "internal/bucket_metadata_cache_test.cc",
     "internal/bucket_requests_test.cc",
     "internal/complex_option_test.cc",
     "internal/compute_engine_util_test.cc",
diff --git a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc
index f947d62118c76..508010073c09c 100644
--- a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc
+++ b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc
@@ -16,6 +16,7 @@
 #include "google/cloud/storage/testing/object_integration_test.h"
 #include "google/cloud/storage/testing/storage_integration_test.h"
 #include "google/cloud/log.h"
+#include "google/cloud/opentelemetry_options.h"
 #include "google/cloud/status_or.h"
 #include "google/cloud/testing_util/expect_exception.h"
 #include "google/cloud/testing_util/status_matchers.h"
@@ -42,15 +43,18 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) {
   // own tests.
   if (UsingGrpc()) GTEST_SKIP();
 
-  auto client = MakeIntegrationTestClient();
+  auto options =
+      Options{}.set<google::cloud::OpenTelemetryTracingOption>(false);
   auto object_name = MakeRandomObjectName();
-
   std::string expected = LoremIpsum();
 
-  StatusOr<ObjectMetadata> meta = client.InsertObject(
-      bucket_name_, object_name, expected, IfGenerationMatch(0));
-  ASSERT_STATUS_OK(meta);
-  ScheduleForDelete(*meta);
+  {
+    auto client = MakeIntegrationTestClient(options);
+    StatusOr<ObjectMetadata> meta = client.InsertObject(
+        bucket_name_, object_name, expected, IfGenerationMatch(0));
+    ASSERT_STATUS_OK(meta);
+    ScheduleForDelete(*meta);
+  }
 
   // Track the number of open files to ensure every client creates the same
   // number of file descriptors and none are leaked.
@@ -64,7 +68,7 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) {
   }
   std::size_t delta = 0;
   for (int i = 0; i != 100; ++i) {
-    auto read_client = MakeIntegrationTestClient();
+    auto read_client = MakeIntegrationTestClient(options);
     auto stream = read_client.ReadObject(bucket_name_, object_name);
     char c;
     stream.read(&c, 1);
diff --git a/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc
index 3e489b2e23db0..26204cc90c5fe 100644
--- a/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc
+++ b/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc
@@ -16,6 +16,7 @@
 #include "google/cloud/storage/testing/object_integration_test.h"
 #include "google/cloud/storage/testing/storage_integration_test.h"
 #include "google/cloud/log.h"
+#include "google/cloud/opentelemetry_options.h"
 #include "google/cloud/status_or.h"
 #include "google/cloud/testing_util/expect_exception.h"
 #include "google/cloud/testing_util/status_matchers.h"
@@ -45,22 +46,24 @@ TEST_F(ObjectPlentyClientsSimultaneouslyIntegrationTest,
   // own tests.
   if (UsingGrpc()) GTEST_SKIP();
 
-  auto client = MakeIntegrationTestClient();
+  auto options =
+      Options{}.set<google::cloud::OpenTelemetryTracingOption>(false);
   auto object_name = MakeRandomObjectName();
-
   std::string expected = LoremIpsum();
 
-  // Create the object, but only if it does not exist already.
-  StatusOr<ObjectMetadata> meta = client.InsertObject(
-      bucket_name_, object_name, expected, IfGenerationMatch(0));
-  ASSERT_STATUS_OK(meta);
-  ScheduleForDelete(*meta);
+  {
+    auto client = MakeIntegrationTestClient(options);
+    StatusOr<ObjectMetadata> meta = client.InsertObject(
+        bucket_name_, object_name, expected, IfGenerationMatch(0));
+    ASSERT_STATUS_OK(meta);
+    ScheduleForDelete(*meta);
+  }
 
   auto num_fds_before_test = GetNumOpenFiles();
   std::vector<Client> read_clients;
   std::vector<ObjectReadStream> read_streams;
   for (int i = 0; i != 100; ++i) {
-    auto read_client = MakeIntegrationTestClient();
+    auto read_client = MakeIntegrationTestClient(options);
     auto stream = read_client.ReadObject(bucket_name_, object_name);
     char c;
     stream.read(&c, 1);