diff --git a/google/cloud/storage/internal/async/connection_impl.cc b/google/cloud/storage/internal/async/connection_impl.cc
index 5516c23cfcd41..6a6ad66d49a0b 100644
--- a/google/cloud/storage/internal/async/connection_impl.cc
+++ b/google/cloud/storage/internal/async/connection_impl.cc
@@ -26,6 +26,7 @@
 #include "google/cloud/storage/internal/async/object_descriptor_impl.h"
 #include "google/cloud/storage/internal/async/open_object.h"
 #include "google/cloud/storage/internal/async/open_stream.h"
+#include "google/cloud/storage/internal/async/options.h"
 #include "google/cloud/storage/internal/async/read_payload_impl.h"
 #include "google/cloud/storage/internal/async/reader_connection_impl.h"
 #include "google/cloud/storage/internal/async/reader_connection_resume.h"
@@ -58,6 +59,7 @@
 #include "google/cloud/internal/make_status.h"
 #include <grpcpp/grpcpp.h>
 #include <memory>
+#include <set>
 #include <utility>
 
 namespace google {
@@ -220,6 +222,24 @@ AsyncConnectionImpl::Open(OpenParams p) {
   auto initial_request = google::storage::v2::BidiReadObjectRequest{};
   *initial_request.mutable_read_object_spec() = p.read_spec;
   auto current = internal::MakeImmutableOptions(p.options);
+  // If pre-warmed ranges are configured, populate the initial request
+  // with these ranges to start downloading them as soon as the stream opens.
+  if (current->has<ReadRangesOption>()) {
+    auto const& ranges = current->get<ReadRangesOption>();
+    std::int64_t id = 0;
+    std::set<std::pair<std::int64_t, std::int64_t>> seen_ranges;
+    for (auto const& r : ranges) {
+      auto range_key = std::make_pair(r.offset, r.length);
+      if (!seen_ranges.insert(range_key).second) {
+        continue;  // Skip duplicate range.
+      }
+      auto* proto_range = initial_request.add_read_ranges();
+      proto_range->set_read_offset(r.offset);
+      proto_range->set_read_length(r.length);
+      // Generate sequential IDs starting at 1. The receiver must match these.
+      proto_range->set_read_id(++id);
+    }
+  }
   // Get the policy factory and immediately create a policy.
   auto resume_policy = current->get<storage::ResumePolicyOption>()();
 
diff --git a/google/cloud/storage/internal/async/connection_impl_open_test.cc b/google/cloud/storage/internal/async/connection_impl_open_test.cc
index e619490e6c7c0..32609f98aec4e 100644
--- a/google/cloud/storage/internal/async/connection_impl_open_test.cc
+++ b/google/cloud/storage/internal/async/connection_impl_open_test.cc
@@ -16,6 +16,7 @@
 #include "google/cloud/storage/async/retry_policy.h"
 #include "google/cloud/storage/internal/async/connection_impl.h"
 #include "google/cloud/storage/internal/async/default_options.h"
+#include "google/cloud/storage/internal/async/options.h"
 #include "google/cloud/storage/internal/grpc/ctype_cord_workaround.h"
 #include "google/cloud/storage/testing/canonical_errors.h"
 #include "google/cloud/storage/testing/mock_resume_policy.h"
@@ -405,6 +406,150 @@ TEST(AsyncConnectionImplTest, TooManyTransienErrors) {
   ASSERT_THAT(pending.get(), StatusIs(TransientError().code()));
 }
 
+TEST(AsyncConnectionImplTest, OpenWithReadRanges) {
+  auto constexpr kExpectedRequestSpec = R"pb(
+    bucket: "test-only-invalid"
+    object: "test-object"
+    generation: 42
+    if_metageneration_match: 7
+  )pb";
+  auto constexpr kExpectedRequest = R"pb(
+    read_object_spec {
+      bucket: "test-only-invalid"
+      object: "test-object"
+      generation: 42
+      if_metageneration_match: 7
+    }
+    read_ranges { read_offset: 0 read_length: 1024 read_id: 1 }
+    read_ranges { read_offset: 2048 read_length: 4096 read_id: 2 }
+  )pb";
+  auto constexpr kMetadataText = R"pb(
+    bucket: "projects/_/buckets/test-bucket"
+    name: "test-object"
+    generation: 42
+  )pb";
+
+  AsyncSequencer<bool> sequencer;
+  auto mock = std::make_shared<storage::testing::MockStorageStub>();
+  EXPECT_CALL(*mock, AsyncBidiReadObject)
+      .WillOnce([&](CompletionQueue const&,
+                    std::shared_ptr<grpc::ClientContext> const&,
+                    google::cloud::internal::ImmutableOptions const& options) {
+        EXPECT_EQ(options->get<AuthorityOption>(), kAuthority);
+
+        auto stream = std::make_unique<MockStream>();
+        EXPECT_CALL(*stream, Start).WillOnce([&sequencer]() {
+          return sequencer.PushBack("Start").then(
+              [](auto f) { return f.get(); });
+        });
+        EXPECT_CALL(*stream, Write)
+            .WillOnce(
+                [=, &sequencer](
+                    google::storage::v2::BidiReadObjectRequest const& request,
+                    grpc::WriteOptions) {
+                  auto expected = google::storage::v2::BidiReadObjectRequest{};
+                  EXPECT_TRUE(
+                      TextFormat::ParseFromString(kExpectedRequest, &expected));
+                  EXPECT_THAT(request, IsProtoEqual(expected));
+                  return sequencer.PushBack("Write").then(
+                      [](auto f) { return f.get(); });
+                });
+        EXPECT_CALL(*stream, Read)
+            .WillOnce([&]() {
+              return sequencer.PushBack("Read").then(
+                  [=](auto f) -> absl::optional<
+                                  google::storage::v2::BidiReadObjectResponse> {
+                    if (!f.get()) return absl::nullopt;
+                    auto constexpr kHandleText = R"pb(
+                      handle: "handle-12345"
+                    )pb";
+                    auto response =
+                        google::storage::v2::BidiReadObjectResponse{};
+                    EXPECT_TRUE(TextFormat::ParseFromString(
+                        kMetadataText, response.mutable_metadata()));
+                    EXPECT_TRUE(TextFormat::ParseFromString(
+                        kHandleText, response.mutable_read_handle()));
+                    return response;
+                  });
+            })
+            .WillOnce([&sequencer]() {
+              return sequencer.PushBack("Read[N]").then(
+                  [](auto f) -> absl::optional<
+                                 google::storage::v2::BidiReadObjectResponse> {
+                    if (!f.get()) return absl::nullopt;
+                    return google::storage::v2::BidiReadObjectResponse{};
+                  });
+            });
+        EXPECT_CALL(*stream, Cancel).WillOnce([&sequencer]() {
+          (void)sequencer.PushBack("Cancel");
+        });
+        EXPECT_CALL(*stream, Finish).WillOnce([&sequencer]() {
+          return sequencer.PushBack("Finish").then(
+              [](auto) { return Status{}; });
+        });
+
+        return std::unique_ptr<BidiReadStream>(std::move(stream));
+      })
+      .WillRepeatedly([](CompletionQueue const&,
+                         std::shared_ptr<grpc::ClientContext> const&,
+                         google::cloud::internal::ImmutableOptions const&) {
+        auto stream = std::make_unique<NiceMock<MockStream>>();
+        ON_CALL(*stream, Start).WillByDefault(InvokeWithoutArgs([] {
+          return make_ready_future(false);
+        }));
+        ON_CALL(*stream, Finish).WillByDefault(InvokeWithoutArgs([] {
+          return make_ready_future(Status{});
+        }));
+        ON_CALL(*stream, Cancel).WillByDefault([] {});
+        return std::unique_ptr<BidiReadStream>(std::move(stream));
+      });
+
+  auto mock_cq = std::make_shared<MockCompletionQueueImpl>();
+  EXPECT_CALL(*mock_cq, MakeRelativeTimer)
+      .WillRepeatedly([](std::chrono::nanoseconds) {
+        return make_ready_future(
+            StatusOr<std::chrono::system_clock::time_point>(
+                std::chrono::system_clock::now()));
+      });
+  auto connection = std::make_shared<AsyncConnectionImpl>(
+      CompletionQueue(mock_cq), std::shared_ptr<GrpcChannelRefresh>(), mock,
+      TestOptions());
+
+  auto request = google::storage::v2::BidiReadObjectSpec{};
+  ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequestSpec, &request));
+  auto options =
+      connection->options().set<ReadRangesOption>({{0, 1024}, {2048, 4096}});
+  auto pending = connection->Open({std::move(request), std::move(options)});
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Start");
+  next.first.set_value(true);
+  next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Write");
+  next.first.set_value(true);
+
+  next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Read");
+  next.first.set_value(true);
+
+  auto p = pending.get();
+  ASSERT_THAT(p, IsOkAndHolds(NotNull()));
+  auto descriptor = *std::move(p);
+
+  descriptor.reset();
+
+  auto last_read = sequencer.PopFrontWithName();
+  EXPECT_EQ(last_read.second, "Read[N]");
+  next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Cancel");
+  next.first.set_value(true);
+  last_read.first.set_value(false);
+
+  next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Finish");
+  next.first.set_value(true);
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal
diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.cc b/google/cloud/storage/internal/async/object_descriptor_impl.cc
index 9430880b8d966..a6a23ab90bcc4 100644
--- a/google/cloud/storage/internal/async/object_descriptor_impl.cc
+++ b/google/cloud/storage/internal/async/object_descriptor_impl.cc
@@ -18,6 +18,7 @@
 #include "google/cloud/storage/internal/async/handle_redirect_error.h"
 #include "google/cloud/storage/internal/async/multi_stream_manager.h"
 #include "google/cloud/storage/internal/async/object_descriptor_reader_tracing.h"
+#include "google/cloud/storage/internal/async/options.h"
 #include "google/cloud/storage/internal/grpc/object_metadata_parser.h"
 #include "google/cloud/storage/internal/hash_function.h"
 #include "google/cloud/storage/internal/hash_function_impl.h"
@@ -28,6 +29,7 @@
 #include "google/cloud/grpc_error_delegate.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/rpc/status.pb.h"
+#include <algorithm>
 #include <limits>
 #include <memory>
 #include <utility>
@@ -52,6 +54,34 @@ ObjectDescriptorImpl::ObjectDescriptorImpl(
       []() -> std::shared_ptr<ReadStream> { return nullptr; },  // NOLINT
       std::make_shared<ReadStream>(std::move(stream),
                                    resume_policy_prototype_->clone()));
+  // If pre-warmed ranges are specified, initialize their `ReadRange` objects,
+  // register them as active on the initial stream, and cache them.
+  if (options_.has<ReadRangesOption>()) {
+    auto const& ranges = options_.get<ReadRangesOption>();
+    auto it = stream_manager_->GetFirstStream();
+    if (it != stream_manager_->End()) {
+      std::int64_t id = 0;
+      std::set<std::pair<std::int64_t, std::int64_t>> seen_ranges;
+      for (auto const& r : ranges) {
+        auto range_key = std::make_pair(r.offset, r.length);
+        if (!seen_ranges.insert(range_key).second) {
+          continue;  // Skip duplicate range.
+        }
+        ++id;
+        auto range = std::make_shared<ReadRange>(r.offset, r.length,
+                                                 read_object_spec_.bucket(),
+                                                 read_object_spec_.object());
+        // Registering on the stream allows `OnRead` to route incoming data to
+        // these ranges.
+        it->active_ranges.emplace(id, range);
+        // Cache them so subsequent `Read()` calls can claim them.
+        prewarmed_ranges_.emplace(range_key, PrewarmedRange{range, id});
+      }
+      // Ensure new dynamically requested ranges use IDs that don't conflict
+      // with pre-warmed ones.
+      read_id_generator_ = id;
+    }
+  }
 }
 
 ObjectDescriptorImpl::~ObjectDescriptorImpl() { Cancel(); }
@@ -193,6 +223,21 @@ std::unique_ptr<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
       read_object_spec_.bucket(), read_object_spec_.object());
 
   std::unique_lock<std::mutex> lk(mu_);
+  // Check if this range matches a pre-warmed range.
+  auto cache_key = std::make_pair(p.start, p.length);
+  auto cache_it = prewarmed_ranges_.find(cache_key);
+  if (cache_it != prewarmed_ranges_.end()) {
+    // Cache hit. Claim the pre-warmed range and return it to the user.
+    auto prewarmed = std::move(cache_it->second);
+    prewarmed_ranges_.erase(cache_it);
+    lk.unlock();
+    if (!internal::TracingEnabled(options_)) {
+      return std::unique_ptr<storage::AsyncReaderConnection>(
+          std::make_unique<ObjectDescriptorReader>(std::move(prewarmed.range)));
+    }
+    return MakeTracingObjectDescriptorReader(std::move(prewarmed.range));
+  }
+
   if (stream_manager_->Empty()) {
     lk.unlock();
     range->OnFinish(Status(StatusCode::kFailedPrecondition,
@@ -366,9 +411,17 @@ void ObjectDescriptorImpl::OnRead(
     auto id = range_data.read_range().read_id();
     auto const l = copy.find(id);
     if (l == copy.end()) continue;
-    // TODO(#15104) - Consider returning if the range is done, and then
-    // skipping CleanupDoneRanges().
-    l->second->OnRead(std::move(range_data), is_transcoded, object_size);
+
+    auto range = l->second;
+    bool active = false;
+    lk.lock();
+    active = it->active_ranges.count(id) != 0;
+    lk.unlock();
+    if (active) {
+      // TODO(#15104) - Consider returning if the range is done, and then
+      // skipping CleanupDoneRanges().
+      range->OnRead(std::move(range_data), is_transcoded, object_size);
+    }
   }
   lk.lock();
   stream_manager_->CleanupDoneRanges(it);
diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.h b/google/cloud/storage/internal/async/object_descriptor_impl.h
index c7b2e2abffc0b..023a0084889b2 100644
--- a/google/cloud/storage/internal/async/object_descriptor_impl.h
+++ b/google/cloud/storage/internal/async/object_descriptor_impl.h
@@ -27,9 +27,11 @@
 #include "google/storage/v2/storage.pb.h"
 #include <cstdint>
 #include <functional>
+#include <map>
 #include <memory>
 #include <mutex>
 #include <optional>
+#include <set>
 #include <unordered_map>
 
 namespace google {
@@ -124,6 +126,15 @@ class ObjectDescriptorImpl
   std::optional<google::storage::v2::Object> metadata_;
   std::int64_t read_id_generator_ = 0;
 
+  // Information about a pre-warmed range that has not been claimed yet.
+  struct PrewarmedRange {
+    std::shared_ptr<ReadRange> range;
+    std::int64_t read_id;
+  };
+  // Cache of pre-warmed ranges, keyed by (offset, length).
+  std::map<std::pair<std::int64_t, std::int64_t>, PrewarmedRange>
+      prewarmed_ranges_;
+
   Options options_;
   std::unique_ptr<StreamManager> stream_manager_;
   // The future for the proactive background stream.
diff --git a/google/cloud/storage/internal/async/object_descriptor_impl_test.cc b/google/cloud/storage/internal/async/object_descriptor_impl_test.cc
index 07830a3e63fdb..e29f15e53bedc 100644
--- a/google/cloud/storage/internal/async/object_descriptor_impl_test.cc
+++ b/google/cloud/storage/internal/async/object_descriptor_impl_test.cc
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/async/object_descriptor_impl.h"
+#include "google/cloud/storage/internal/async/options.h"
 
 // TODO(v-pratap): Remove this when EnableMD5ValidationOption and
 // EnableCrc32cValidationOption are removed.
@@ -2298,6 +2299,197 @@ TEST(ObjectDescriptorImpl, PartialReadChecksumValidationBypassed) {
   next.first.set_value(true);
 }
 
+TEST(ObjectDescriptorImpl, PrewarmedCacheHit) {
+  auto constexpr kResponse0 = R"pb(
+    metadata {
+      bucket: "projects/_/buckets/test-bucket"
+      name: "test-object"
+      generation: 42
+    }
+    read_handle { handle: "handle-12345" }
+  )pb";
+
+  auto constexpr kResponse1 = R"pb(
+    read_handle { handle: "handle-23456" }
+    object_data_ranges {
+      range_end: true
+      read_range { read_id: 1 read_offset: 0 }
+      checksummed_data { content: "Pre-warmed data" }
+    }
+  )pb";
+
+  AsyncSequencer<bool> sequencer;
+  auto stream = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream, Write).Times(0);
+
+  EXPECT_CALL(*stream, Read)
+      .WillOnce([=, &sequencer]() {
+        return sequencer.PushBack("Read[1]").then([&](auto) {
+          auto response = Response{};
+          EXPECT_TRUE(TextFormat::ParseFromString(kResponse1, &response));
+          return absl::make_optional(response);
+        });
+      })
+      .WillOnce([&sequencer]() {
+        return sequencer.PushBack("Read[2]").then(
+            [](auto) { return absl::optional<Response>{}; });
+      });
+  EXPECT_CALL(*stream, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish").then(
+        [](auto) { return PermanentError(); });
+  });
+  EXPECT_CALL(*stream, Cancel).Times(AtMost(1));
+
+  MockFactory factory;
+  EXPECT_CALL(factory, Call).WillOnce([](Request const&) {
+    return make_ready_future(StatusOr<OpenStreamResult>(PermanentError()));
+  });
+
+  Options options;
+  options.set<storage::EnableMultiStreamOptimizationOption>(true);
+  options.set<ReadRangesOption>({{0, 15}});
+
+  auto tested = std::make_shared<ObjectDescriptorImpl>(
+      NoResume(), factory.AsStdFunction(),
+      google::storage::v2::BidiReadObjectSpec{},
+      std::make_shared<OpenStream>(std::move(stream)), options);
+
+  auto response = Response{};
+  EXPECT_TRUE(TextFormat::ParseFromString(kResponse0, &response));
+  tested->Start(std::move(response));
+
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read1.second, "Read[1]");
+
+  auto s1 = tested->Read({0, 15});
+  ASSERT_THAT(s1, NotNull());
+
+  auto s1r1 = s1->Read();
+  EXPECT_FALSE(s1r1.is_ready());
+
+  read1.first.set_value(true);
+
+  EXPECT_TRUE(s1r1.is_ready());
+  EXPECT_THAT(s1r1.get(),
+              VariantWith<storage::ReadPayload>(ResultOf(
+                  "contents are",
+                  [](storage::ReadPayload const& p) { return p.contents(); },
+                  ElementsAre(absl::string_view{"Pre-warmed data"}))));
+
+  EXPECT_THAT(s1->Read().get(), VariantWith<Status>(IsOk()));
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Read[2]");
+  next.first.set_value(true);
+
+  next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Finish");
+  next.first.set_value(true);
+}
+
+TEST(ObjectDescriptorImpl, PrewarmedCacheMiss) {
+  auto constexpr kResponse0 = R"pb(
+    metadata {
+      bucket: "projects/_/buckets/test-bucket"
+      name: "test-object"
+      generation: 42
+    }
+    read_handle { handle: "handle-12345" }
+  )pb";
+
+  auto constexpr kExpectedRequest = R"pb(
+    read_ranges { read_id: 2 read_offset: 100 read_length: 15 }
+  )pb";
+
+  auto constexpr kResponse1 = R"pb(
+    object_data_ranges {
+      range_end: true
+      read_range { read_id: 2 read_offset: 100 }
+      checksummed_data { content: "Missed range data" }
+    }
+  )pb";
+
+  AsyncSequencer<bool> sequencer;
+  auto stream = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream, Write)
+      .WillOnce([&](Request const& request, grpc::WriteOptions) {
+        auto expected = Request{};
+        EXPECT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &expected));
+        EXPECT_THAT(request, IsProtoEqual(expected));
+        return sequencer.PushBack("Write[1]").then([](auto f) {
+          return f.get();
+        });
+      });
+
+  EXPECT_CALL(*stream, Read)
+      .WillOnce([=, &sequencer]() {
+        return sequencer.PushBack("Read[1]").then([&](auto) {
+          auto response = Response{};
+          EXPECT_TRUE(TextFormat::ParseFromString(kResponse1, &response));
+          return absl::make_optional(response);
+        });
+      })
+      .WillOnce([&sequencer]() {
+        return sequencer.PushBack("Read[2]").then(
+            [](auto) { return absl::optional<Response>{}; });
+      });
+  EXPECT_CALL(*stream, Finish).WillOnce([&sequencer]() {
+    return sequencer.PushBack("Finish").then(
+        [](auto) { return PermanentError(); });
+  });
+  EXPECT_CALL(*stream, Cancel).Times(AtMost(1));
+
+  MockFactory factory;
+  EXPECT_CALL(factory, Call).WillOnce([](Request const&) {
+    return make_ready_future(StatusOr<OpenStreamResult>(PermanentError()));
+  });
+
+  Options options;
+  options.set<storage::EnableMultiStreamOptimizationOption>(true);
+  options.set<ReadRangesOption>({{0, 15}});
+
+  auto tested = std::make_shared<ObjectDescriptorImpl>(
+      NoResume(), factory.AsStdFunction(),
+      google::storage::v2::BidiReadObjectSpec{},
+      std::make_shared<OpenStream>(std::move(stream)), options);
+
+  auto response = Response{};
+  EXPECT_TRUE(TextFormat::ParseFromString(kResponse0, &response));
+  tested->Start(std::move(response));
+
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read1.second, "Read[1]");
+
+  auto s1 = tested->Read({100, 15});
+  ASSERT_THAT(s1, NotNull());
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Write[1]");
+  next.first.set_value(true);
+
+  auto s1r1 = s1->Read();
+  EXPECT_FALSE(s1r1.is_ready());
+
+  read1.first.set_value(true);
+
+  EXPECT_TRUE(s1r1.is_ready());
+  EXPECT_THAT(s1r1.get(),
+              VariantWith<storage::ReadPayload>(ResultOf(
+                  "contents are",
+                  [](storage::ReadPayload const& p) { return p.contents(); },
+                  ElementsAre(absl::string_view{"Missed range data"}))));
+
+  EXPECT_THAT(s1->Read().get(), VariantWith<Status>(IsOk()));
+
+  next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Read[2]");
+  next.first.set_value(true);
+
+  next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Finish");
+  next.first.set_value(true);
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal
diff --git a/google/cloud/storage/internal/async/options.h b/google/cloud/storage/internal/async/options.h
new file mode 100644
index 0000000000000..f6fe7fa70593c
--- /dev/null
+++ b/google/cloud/storage/internal/async/options.h
@@ -0,0 +1,48 @@
+// 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_ASYNC_OPTIONS_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OPTIONS_H
+
+#include "google/cloud/options.h"
+#include "google/cloud/version.h"
+#include <cstdint>
+#include <vector>
+
+namespace google {
+namespace cloud {
+namespace storage_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+// Configuration for a single read range to be pre-warmed.
+struct ReadRangeConfig {
+  std::int64_t offset;
+  std::int64_t length;
+};
+
+// Internal option to pass the list of ranges to pre-warm from `AsyncClient` to
+// `Connection`.
+struct ReadRangesOption {
+  using Type = std::vector<ReadRangeConfig>;
+  static char const* name() {
+    return "google::cloud::storage::ReadRangesOption";
+  }
+};
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace storage_internal
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OPTIONS_H