diff --git a/google/cloud/storage/internal/async/connection_tracing.cc b/google/cloud/storage/internal/async/connection_tracing.cc
index 2e29660276078..af4fa2e940536 100644
--- a/google/cloud/storage/internal/async/connection_tracing.cc
+++ b/google/cloud/storage/internal/async/connection_tracing.cc
@@ -18,12 +18,21 @@
 #include "google/cloud/storage/internal/async/reader_connection_tracing.h"
 #include "google/cloud/storage/internal/async/rewriter_connection_tracing.h"
 #include "google/cloud/storage/internal/async/writer_connection_tracing.h"
+#include "google/cloud/storage/internal/bucket_metadata_cache.h"
+#include "google/cloud/storage/options.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/cloud/version.h"
+#include "google/storage/v2/storage.pb.h"
+#include <algorithm>
+#include <chrono>
+#include <future>
 #include <memory>
+#include <mutex>
+#include <vector>
 
 namespace google {
 namespace cloud {
+
 namespace storage_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
@@ -33,28 +42,32 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
  public:
   explicit AsyncConnectionTracing(
       std::shared_ptr<storage::AsyncConnection> impl)
-      : impl_(std::move(impl)) {}
+      : impl_(std::move(impl)),
+        cache_(std::make_shared<BucketMetadataCache>()) {}
 
-  Options options() const override { return impl_->options(); }
+  ~AsyncConnectionTracing() override = default;
 
-  future<StatusOr<google::storage::v2::Bucket>> GetBucket(
-      GetBucketParams p) override {
-    auto span = internal::MakeSpan("storage::AsyncConnection::GetBucket");
-    internal::OTelScope scope(span);
-    return internal::EndSpan(std::move(span), impl_->GetBucket(std::move(p)));
-  }
+  Options options() const override { return impl_->options(); }
 
   future<StatusOr<google::storage::v2::Object>> InsertObject(
       InsertObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::InsertObject");
+    EnrichSpan(*span, p.options,
+               p.request.write_object_spec().resource().bucket());
     internal::OTelScope scope(span);
-    return internal::EndSpan(std::move(span),
-                             impl_->InsertObject(std::move(p)));
+    return impl_->InsertObject(std::move(p))
+        .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
+               span = std::move(span)](auto f) {
+          auto result = f.get();
+          internal::DetachOTelContext(oc);
+          return internal::EndSpan(*span, std::move(result));
+        });
   }
 
   future<StatusOr<std::shared_ptr<storage::ObjectDescriptorConnection>>> Open(
       OpenParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::Open");
+    EnrichSpan(*span, p.options, p.read_spec.bucket());
     internal::OTelScope scope(span);
     return impl_->Open(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -74,13 +87,16 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   future<StatusOr<std::unique_ptr<storage::AsyncReaderConnection>>> ReadObject(
       ReadObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::ReadObject");
+    EnrichSpan(*span, p.options, p.request.bucket());
     internal::OTelScope scope(span);
     auto wrap = [oc = opentelemetry::context::RuntimeContext::GetCurrent(),
                  span = std::move(span)](auto f)
         -> StatusOr<std::unique_ptr<storage::AsyncReaderConnection>> {
       auto reader = f.get();
       internal::DetachOTelContext(oc);
-      if (!reader) return internal::EndSpan(*span, std::move(reader).status());
+      if (!reader) {
+        return internal::EndSpan(*span, std::move(reader).status());
+      }
       return MakeTracingReaderConnection(std::move(span), *std::move(reader));
     };
     return impl_->ReadObject(std::move(p)).then(std::move(wrap));
@@ -89,6 +105,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   future<StatusOr<storage::ReadPayload>> ReadObjectRange(
       ReadObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::ReadObjectRange");
+    EnrichSpan(*span, p.options, p.request.bucket());
     internal::OTelScope scope(span);
     return impl_->ReadObjectRange(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -103,6 +120,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   StartAppendableObjectUpload(AppendableUploadParams p) override {
     auto span = internal::MakeSpan(
         "storage::AsyncConnection::StartAppendableObjectUpload");
+    EnrichSpan(*span, p.options,
+               p.request.write_object_spec().resource().bucket());
     internal::OTelScope scope(span);
     return impl_->StartAppendableObjectUpload(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -110,7 +129,9 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
                   -> StatusOr<std::unique_ptr<storage::AsyncWriterConnection>> {
           auto w = f.get();
           internal::DetachOTelContext(oc);
-          if (!w) return internal::EndSpan(*span, std::move(w).status());
+          if (!w) {
+            return internal::EndSpan(*span, std::move(w).status());
+          }
           return MakeTracingWriterConnection(span, *std::move(w));
         });
   }
@@ -119,6 +140,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   ResumeAppendableObjectUpload(AppendableUploadParams p) override {
     auto span = internal::MakeSpan(
         "storage::AsyncConnection::ResumeAppendableObjectUpload");
+    EnrichSpan(*span, p.options,
+               p.request.write_object_spec().resource().bucket());
     internal::OTelScope scope(span);
     return impl_->ResumeAppendableObjectUpload(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -126,7 +149,9 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
                   -> StatusOr<std::unique_ptr<storage::AsyncWriterConnection>> {
           auto w = f.get();
           internal::DetachOTelContext(oc);
-          if (!w) return internal::EndSpan(*span, std::move(w).status());
+          if (!w) {
+            return internal::EndSpan(*span, std::move(w).status());
+          }
           return MakeTracingWriterConnection(span, *std::move(w));
         });
   }
@@ -135,6 +160,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   StartUnbufferedUpload(UploadParams p) override {
     auto span =
         internal::MakeSpan("storage::AsyncConnection::StartUnbufferedUpload");
+    EnrichSpan(*span, p.options,
+               p.request.write_object_spec().resource().bucket());
     internal::OTelScope scope(span);
     return impl_->StartUnbufferedUpload(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -142,7 +169,9 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
                   -> StatusOr<std::unique_ptr<storage::AsyncWriterConnection>> {
           auto w = f.get();
           internal::DetachOTelContext(oc);
-          if (!w) return internal::EndSpan(*span, std::move(w).status());
+          if (!w) {
+            return internal::EndSpan(*span, std::move(w).status());
+          }
           return MakeTracingWriterConnection(span, *std::move(w));
         });
   }
@@ -151,6 +180,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   StartBufferedUpload(UploadParams p) override {
     auto span =
         internal::MakeSpan("storage::AsyncConnection::StartBufferedUpload");
+    EnrichSpan(*span, p.options,
+               p.request.write_object_spec().resource().bucket());
     internal::OTelScope scope(span);
     return impl_->StartBufferedUpload(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -158,7 +189,9 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
                   -> StatusOr<std::unique_ptr<storage::AsyncWriterConnection>> {
           auto w = f.get();
           internal::DetachOTelContext(oc);
-          if (!w) return internal::EndSpan(*span, std::move(w).status());
+          if (!w) {
+            return internal::EndSpan(*span, std::move(w).status());
+          }
           return MakeTracingWriterConnection(span, *std::move(w));
         });
   }
@@ -198,16 +231,28 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   future<StatusOr<google::storage::v2::Object>> ComposeObject(
       ComposeObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::ComposeObject");
+    EnrichSpan(*span, p.options, p.request.destination().bucket());
     internal::OTelScope scope(span);
-    return internal::EndSpan(std::move(span),
-                             impl_->ComposeObject(std::move(p)));
+    return impl_->ComposeObject(std::move(p))
+        .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
+               span = std::move(span)](auto f) {
+          auto result = f.get();
+          internal::DetachOTelContext(oc);
+          return internal::EndSpan(*span, std::move(result));
+        });
   }
 
   future<Status> DeleteObject(DeleteObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::DeleteObject");
+    EnrichSpan(*span, p.options, p.request.bucket());
     internal::OTelScope scope(span);
-    return internal::EndSpan(std::move(span),
-                             impl_->DeleteObject(std::move(p)));
+    return impl_->DeleteObject(std::move(p))
+        .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
+               span = std::move(span)](auto f) {
+          auto result = f.get();
+          internal::DetachOTelContext(oc);
+          return internal::EndSpan(*span, std::move(result));
+        });
   }
 
   std::shared_ptr<storage::AsyncRewriterConnection> RewriteObject(
@@ -217,8 +262,101 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
         impl_->RewriteObject(std::move(p)), enabled);
   }
 
+  future<StatusOr<google::storage::v2::Bucket>> GetBucket(
+      GetBucketParams p) override {
+    auto span = internal::MakeSpan("storage::AsyncConnection::GetBucket");
+    internal::OTelScope scope(span);
+    auto const bucket_name = p.request.name();
+    auto const options = p.options;
+    return impl_->GetBucket(std::move(p))
+        .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
+               span = std::move(span), this, bucket_name,
+               options](future<StatusOr<google::storage::v2::Bucket>> f)
+                  -> StatusOr<google::storage::v2::Bucket> {
+          auto result = f.get();
+          internal::DetachOTelContext(oc);
+          if (result.ok()) {
+            EnrichSpan(*span, options, *result, bucket_name);
+          } else {
+            cache().MaybeInvalidate(result, bucket_name);
+          }
+          return internal::EndSpan(*span, std::move(result));
+        });
+  }
+
  private:
+  BucketMetadataCache& cache() const { return *cache_; }
+
+  void MaybeTriggerBackgroundFetch(Options const& options,
+                                   std::string const& bucket_name) {
+    if (!cache().StartFetch(bucket_name)) {
+      return;
+    }
+
+    auto guard = ScopedFetch(cache_, bucket_name);
+    google::storage::v2::GetBucketRequest request;
+    auto const normalized_bucket_name =
+        BucketMetadataCache::NormalizeBucketName(bucket_name);
+    request.set_name("projects/_/buckets/" + normalized_bucket_name);
+    GetBucketParams params{std::move(request), options};
+
+    impl_->GetBucket(std::move(params))
+        .then([cache = cache_, bucket_name, guard = std::move(guard)](
+                  future<StatusOr<google::storage::v2::Bucket>> f) {
+          auto metadata = f.get();
+          if (metadata.ok()) {
+            BucketCacheEntry entry = BucketCacheEntry::FromLocation(
+                metadata->project() + "/buckets/" +
+                    BucketMetadataCache::NormalizeBucketName(bucket_name),
+                metadata->location(), metadata->location_type());
+            cache->Put(bucket_name, std::move(entry));
+          } else if (metadata.status().code() ==
+                     StatusCode::kPermissionDenied) {
+            BucketCacheEntry entry{
+                "projects/_/buckets/" +
+                    BucketMetadataCache::NormalizeBucketName(bucket_name),
+                "global"};
+            cache->Put(bucket_name, std::move(entry));
+          }
+        });
+  }
+
+  static void 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 EnrichSpan(opentelemetry::trace::Span& span, Options const& options,
+                  google::storage::v2::Bucket const& bucket,
+                  std::string const& bucket_name) {
+    auto const enabled = options.get<
+        google::cloud::storage_experimental::OTelSpanEnrichmentOption>();
+    if (!enabled) return;
+    auto entry = BucketCacheEntry::FromLocation(
+        bucket.project() + "/buckets/" +
+            BucketMetadataCache::NormalizeBucketName(bucket_name),
+        bucket.location(), bucket.location_type());
+    EnrichSpan(span, entry);
+    cache().Put(bucket_name, std::move(entry));
+  }
+
+  void EnrichSpan(opentelemetry::trace::Span& span, Options const& options,
+                  std::string const& bucket_name) {
+    if (bucket_name.empty()) return;
+    auto const enabled = options.get<
+        google::cloud::storage_experimental::OTelSpanEnrichmentOption>();
+    if (!enabled) return;
+    auto entry = cache().Get(bucket_name);
+    if (entry.has_value()) {
+      EnrichSpan(span, *entry);
+    } else {
+      MaybeTriggerBackgroundFetch(options, bucket_name);
+    }
+  }
+
   std::shared_ptr<storage::AsyncConnection> impl_;
+  std::shared_ptr<BucketMetadataCache> cache_;
 };
 
 }  // namespace
diff --git a/google/cloud/storage/internal/async/connection_tracing_test.cc b/google/cloud/storage/internal/async/connection_tracing_test.cc
index bdc562baef97c..f793243f42c05 100644
--- a/google/cloud/storage/internal/async/connection_tracing_test.cc
+++ b/google/cloud/storage/internal/async/connection_tracing_test.cc
@@ -20,6 +20,7 @@
 #include "google/cloud/storage/mocks/mock_async_reader_connection.h"
 #include "google/cloud/storage/mocks/mock_async_rewriter_connection.h"
 #include "google/cloud/storage/mocks/mock_async_writer_connection.h"
+#include "google/cloud/storage/options.h"
 #include "google/cloud/storage/testing/canonical_errors.h"
 #include "google/cloud/opentelemetry_options.h"
 #include "google/cloud/testing_util/opentelemetry_matchers.h"
@@ -44,8 +45,10 @@ using ::google::cloud::testing_util::EventNamed;
 using ::google::cloud::testing_util::InstallSpanCatcher;
 using ::google::cloud::testing_util::IsOk;
 using ::google::cloud::testing_util::IsOkAndHolds;
+using ::google::cloud::testing_util::OTelAttribute;
 using ::google::cloud::testing_util::OTelContextCaptured;
 using ::google::cloud::testing_util::PromiseWithOTelContext;
+using ::google::cloud::testing_util::SpanHasAttributes;
 using ::google::cloud::testing_util::SpanHasEvents;
 using ::google::cloud::testing_util::SpanHasInstrumentationScope;
 using ::google::cloud::testing_util::SpanKindIsClient;
@@ -762,6 +765,209 @@ TEST(ConnectionTracing, GetBucketError) {
                        SpanHasInstrumentationScope(), SpanKindIsClient())));
 }
 
+TEST(ConnectionTracing, GetBucketSpanEnrichment) {
+  auto span_catcher = InstallSpanCatcher();
+  PromiseWithOTelContext<StatusOr<google::storage::v2::Bucket>> p;
+
+  auto options =
+      TracingEnabled()
+          .set<google::cloud::storage_experimental::OTelSpanEnrichmentOption>(
+              true);
+  auto mock = std::make_unique<MockAsyncConnection>();
+  EXPECT_CALL(*mock, options).WillRepeatedly(Return(options));
+  EXPECT_CALL(*mock, GetBucket).WillOnce(expect_context(p));
+  auto actual = MakeTracingAsyncConnection(std::move(mock));
+
+  google::storage::v2::GetBucketRequest req;
+  req.set_name("projects/_/buckets/test-bucket");
+  auto result =
+      actual->GetBucket({std::move(req), options}).then(expect_no_context);
+
+  google::storage::v2::Bucket bucket_meta;
+  bucket_meta.set_project("projects/123456");
+  bucket_meta.set_location("us-east1");
+  bucket_meta.set_location_type("regional");
+  p.set_value(make_status_or(std::move(bucket_meta)));
+  ASSERT_STATUS_OK(result.get());
+
+  auto spans = span_catcher->GetSpans();
+  EXPECT_THAT(
+      spans,
+      ElementsAre(AllOf(
+          SpanNamed("storage::AsyncConnection::GetBucket"),
+          SpanWithStatus(opentelemetry::trace::StatusCode::kOk),
+          SpanHasInstrumentationScope(), SpanKindIsClient(),
+          SpanHasAttributes(
+              OTelAttribute<std::string>("gcp.resource.destination.id",
+                                         "projects/123456/buckets/test-bucket"),
+              OTelAttribute<std::string>("gcp.resource.destination.location",
+                                         "us-east1")))));
+}
+
+TEST(ConnectionTracing, BucketMetadataCacheSuccess) {
+  auto span_catcher = InstallSpanCatcher();
+  PromiseWithOTelContext<Status> delete_promise_1;
+  PromiseWithOTelContext<StatusOr<google::storage::v2::Bucket>> bucket_promise;
+  PromiseWithOTelContext<Status> delete_promise_2;
+
+  auto options =
+      TracingEnabled()
+          .set<google::cloud::storage_experimental::OTelSpanEnrichmentOption>(
+              true);
+
+  auto mock = std::make_unique<MockAsyncConnection>();
+  EXPECT_CALL(*mock, options).WillRepeatedly(Return(options));
+
+  // 1st call: DeleteObject on a cache miss triggers background GetBucket
+  EXPECT_CALL(*mock, DeleteObject)
+      .WillOnce(expect_context(delete_promise_1))
+      .WillOnce(expect_context(delete_promise_2));
+
+  EXPECT_CALL(*mock, GetBucket).WillOnce([&](auto const&) {
+    return bucket_promise.get_future();
+  });
+
+  auto actual = MakeTracingAsyncConnection(std::move(mock));
+
+  google::storage::v2::DeleteObjectRequest req;
+  req.set_bucket("projects/_/buckets/test-bucket");
+  req.set_object("test-object-1");
+
+  auto res1 = actual->DeleteObject({req, options}).then(expect_no_context);
+  delete_promise_1.set_value(Status{});
+  ASSERT_STATUS_OK(res1.get());
+
+  // Complete the background GetBucket fetch to populate the cache
+  google::storage::v2::Bucket bucket_meta;
+  bucket_meta.set_project("projects/123456");
+  bucket_meta.set_location("us-east1");
+  bucket_meta.set_location_type("regional");
+  bucket_promise.set_value(make_status_or(std::move(bucket_meta)));
+
+  // Clear spans from the first operation and background GetBucket
+  (void)span_catcher->GetSpans();
+
+  // 2nd call: DeleteObject hits the cache and span is enriched
+  req.set_object("test-object-2");
+  auto res2 = actual->DeleteObject({req, options}).then(expect_no_context);
+  delete_promise_2.set_value(Status{});
+  ASSERT_STATUS_OK(res2.get());
+
+  auto spans = span_catcher->GetSpans();
+  EXPECT_THAT(
+      spans,
+      ElementsAre(AllOf(
+          SpanNamed("storage::AsyncConnection::DeleteObject"),
+          SpanWithStatus(opentelemetry::trace::StatusCode::kOk),
+          SpanHasInstrumentationScope(), SpanKindIsClient(),
+          SpanHasAttributes(
+              OTelAttribute<std::string>("gcp.resource.destination.id",
+                                         "projects/123456/buckets/test-bucket"),
+              OTelAttribute<std::string>("gcp.resource.destination.location",
+                                         "us-east1")))));
+}
+
+TEST(ConnectionTracing, GetBucketMaybeInvalidateEvicts) {
+  auto span_catcher = InstallSpanCatcher();
+  PromiseWithOTelContext<StatusOr<google::storage::v2::Bucket>> p1;
+  PromiseWithOTelContext<StatusOr<google::storage::v2::Bucket>> p2;
+
+  auto options =
+      TracingEnabled()
+          .set<google::cloud::storage_experimental::OTelSpanEnrichmentOption>(
+              true);
+  auto mock = std::make_unique<MockAsyncConnection>();
+  EXPECT_CALL(*mock, options).WillRepeatedly(Return(options));
+
+  EXPECT_CALL(*mock, GetBucket)
+      .WillOnce(expect_context(p1))
+      .WillOnce(expect_context(p2));
+
+  auto actual = MakeTracingAsyncConnection(std::move(mock));
+
+  // 1st call populates the cache
+  google::storage::v2::GetBucketRequest req;
+  req.set_name("projects/_/buckets/test-bucket");
+  auto res1 = actual->GetBucket({req, options}).then(expect_no_context);
+  google::storage::v2::Bucket bucket_meta;
+  bucket_meta.set_project("projects/123456");
+  bucket_meta.set_location("us-east1");
+  bucket_meta.set_location_type("regional");
+  p1.set_value(make_status_or(std::move(bucket_meta)));
+  ASSERT_STATUS_OK(res1.get());
+
+  // 2nd call returns kNotFound and evicts the cached metadata
+  auto res2 = actual->GetBucket({req, options}).then(expect_no_context);
+  p2.set_value(StatusOr<google::storage::v2::Bucket>(
+      Status(StatusCode::kNotFound, "not found")));
+  EXPECT_THAT(res2.get(), StatusIs(StatusCode::kNotFound));
+}
+
+TEST(ConnectionTracing, DeleteObjectNoEvictOnError) {
+  auto span_catcher = InstallSpanCatcher();
+  PromiseWithOTelContext<StatusOr<google::storage::v2::Bucket>> bucket_p;
+  PromiseWithOTelContext<Status> delete_p1;
+  PromiseWithOTelContext<Status> delete_p2;
+
+  auto options =
+      TracingEnabled()
+          .set<google::cloud::storage_experimental::OTelSpanEnrichmentOption>(
+              true);
+  auto mock = std::make_unique<MockAsyncConnection>();
+  EXPECT_CALL(*mock, options).WillRepeatedly(Return(options));
+
+  EXPECT_CALL(*mock, GetBucket).WillOnce(expect_context(bucket_p));
+  EXPECT_CALL(*mock, DeleteObject)
+      .WillOnce(expect_context(delete_p1))
+      .WillOnce(expect_context(delete_p2));
+
+  auto actual = MakeTracingAsyncConnection(std::move(mock));
+
+  // 1st call: GetBucket populates the cache
+  google::storage::v2::GetBucketRequest bucket_req;
+  bucket_req.set_name("projects/_/buckets/test-bucket");
+  auto res_bucket =
+      actual->GetBucket({bucket_req, options}).then(expect_no_context);
+  google::storage::v2::Bucket bucket_meta;
+  bucket_meta.set_project("projects/123456");
+  bucket_meta.set_location("us-east1");
+  bucket_meta.set_location_type("regional");
+  bucket_p.set_value(make_status_or(std::move(bucket_meta)));
+  ASSERT_STATUS_OK(res_bucket.get());
+
+  // 2nd call: DeleteObject fails with kNotFound (object missing)
+  google::storage::v2::DeleteObjectRequest del_req;
+  del_req.set_bucket("projects/_/buckets/test-bucket");
+  del_req.set_object("missing-object");
+  auto res_del1 =
+      actual->DeleteObject({del_req, options}).then(expect_no_context);
+  delete_p1.set_value(Status(StatusCode::kNotFound, "not found"));
+  EXPECT_THAT(res_del1.get(), StatusIs(StatusCode::kNotFound));
+
+  // Clear spans to verify next span enrichment
+  (void)span_catcher->GetSpans();
+
+  // 3rd call: DeleteObject proves bucket metadata was NOT evicted by object
+  // error
+  del_req.set_object("another-object");
+  auto res_del2 =
+      actual->DeleteObject({del_req, options}).then(expect_no_context);
+  delete_p2.set_value(Status{});
+  ASSERT_STATUS_OK(res_del2.get());
+
+  auto spans = span_catcher->GetSpans();
+  EXPECT_THAT(
+      spans,
+      ElementsAre(AllOf(
+          SpanNamed("storage::AsyncConnection::DeleteObject"),
+          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")))));
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal
diff --git a/google/cloud/storage/internal/async/default_options.cc b/google/cloud/storage/internal/async/default_options.cc
index bee57a17cfe06..c367c6d0a6a02 100644
--- a/google/cloud/storage/internal/async/default_options.cc
+++ b/google/cloud/storage/internal/async/default_options.cc
@@ -19,6 +19,7 @@
 #include "google/cloud/storage/async/retry_policy.h"
 #include "google/cloud/storage/async/writer_connection.h"
 #include "google/cloud/storage/internal/grpc/default_options.h"
+#include "google/cloud/storage/options.h"
 #include <limits>
 
 namespace google {
@@ -76,7 +77,8 @@ Options DefaultOptionsAsync(Options opts) {
               storage::MakeStrictAsyncIdempotencyPolicy)
           .set<storage::EnableCrc32cValidationOption>(true)
           .set<storage::MaximumRangeSizeOption>(128 * 1024 * 1024L)
-          .set<storage::EnableMultiStreamOptimizationOption>(true));
+          .set<storage::EnableMultiStreamOptimizationOption>(true)
+          .set<storage_experimental::OTelSpanEnrichmentOption>(true));
   return Adjust(DefaultOptionsGrpc(std::move(opts)));
 }
 
diff --git a/google/cloud/storage/internal/async/default_options_test.cc b/google/cloud/storage/internal/async/default_options_test.cc
index 7bdc79553a84b..923bc39d1a33f 100644
--- a/google/cloud/storage/internal/async/default_options_test.cc
+++ b/google/cloud/storage/internal/async/default_options_test.cc
@@ -21,6 +21,7 @@
 #include "google/cloud/storage/async/options.h"
 #include "google/cloud/storage/async/resume_policy.h"
 #include "google/cloud/storage/async/writer_connection.h"
+#include "google/cloud/storage/options.h"
 #include "google/cloud/common_options.h"
 #include <gmock/gmock.h>
 
@@ -109,6 +110,20 @@ TEST(DefaultOptionsAsync, EnableMultiStreamOptimizationOption) {
       updated_options.get<storage::EnableMultiStreamOptimizationOption>());
 }
 
+TEST(DefaultOptionsAsync, OTelSpanEnrichmentOption) {
+  auto const options = DefaultOptionsAsync({});
+  EXPECT_TRUE(options.get<
+              google::cloud::storage_experimental::OTelSpanEnrichmentOption>());
+
+  auto const updated_options = DefaultOptionsAsync(
+      Options{}
+          .set<google::cloud::storage_experimental::OTelSpanEnrichmentOption>(
+              false));
+  EXPECT_FALSE(
+      updated_options.get<
+          google::cloud::storage_experimental::OTelSpanEnrichmentOption>());
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal
diff --git a/google/cloud/storage/internal/bucket_metadata_cache.cc b/google/cloud/storage/internal/bucket_metadata_cache.cc
index 09e74119e8041..583033393967f 100644
--- a/google/cloud/storage/internal/bucket_metadata_cache.cc
+++ b/google/cloud/storage/internal/bucket_metadata_cache.cc
@@ -14,6 +14,7 @@
 
 #include "google/cloud/storage/internal/bucket_metadata_cache.h"
 #include "google/cloud/storage/bucket_metadata.h"
+#include "absl/strings/match.h"
 #include <mutex>
 #include <utility>
 
@@ -22,16 +23,28 @@ namespace cloud {
 namespace storage_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
+BucketCacheEntry BucketCacheEntry::FromLocation(
+    std::string id, std::string location, std::string const& location_type) {
+  if (location_type == "multi-region" || location_type == "dual-region") {
+    location = "global";
+  }
+  return {std::move(id), std::move(location)};
+}
+
 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 {
+  return FromLocation(
       "projects/" + std::to_string(m.project_number()) + "/buckets/" + m.name(),
-      std::move(loc)};
+      m.location(), m.location_type());
+}
+
+std::string BucketMetadataCache::NormalizeBucketName(
+    std::string const& bucket) {
+  auto const prefix = std::string("projects/_/buckets/");
+  if (absl::StartsWith(bucket, prefix)) {
+    return bucket.substr(prefix.size());
+  }
+  return bucket;
 }
 
 void BucketMetadataCache::MoveToFront(std::list<std::string>::iterator it) {
diff --git a/google/cloud/storage/internal/bucket_metadata_cache.h b/google/cloud/storage/internal/bucket_metadata_cache.h
index f3023126b2d28..4292140469ba6 100644
--- a/google/cloud/storage/internal/bucket_metadata_cache.h
+++ b/google/cloud/storage/internal/bucket_metadata_cache.h
@@ -16,6 +16,8 @@
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H
 
 #include "google/cloud/storage/version.h"
+#include "google/cloud/status.h"
+#include "google/cloud/status_or.h"
 #include "absl/types/optional.h"
 #include <cstddef>
 #include <list>
@@ -40,6 +42,8 @@ struct BucketCacheEntry {
   std::string id;
   std::string location;
 
+  static BucketCacheEntry FromLocation(std::string id, std::string location,
+                                       std::string const& location_type);
   static BucketCacheEntry FromMetadata(storage::BucketMetadata const& m);
 };
 
@@ -48,6 +52,8 @@ class BucketMetadataCache {
   explicit BucketMetadataCache(std::size_t max_size = 10000)
       : max_size_(max_size) {}
 
+  static std::string NormalizeBucketName(std::string const& bucket);
+
   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);
@@ -55,6 +61,18 @@ class BucketMetadataCache {
   bool StartFetch(std::string const& bucket_name);
   void EndFetch(std::string const& bucket_name);
 
+  void MaybeInvalidate(Status const& status, std::string const& bucket_name) {
+    if (!status.ok() && status.code() == StatusCode::kNotFound) {
+      Invalidate(bucket_name);
+    }
+  }
+
+  template <typename T>
+  void MaybeInvalidate(StatusOr<T> const& result,
+                       std::string const& bucket_name) {
+    MaybeInvalidate(result.status(), bucket_name);
+  }
+
  private:
   void MoveToFront(std::list<std::string>::iterator it);