diff --git a/google/cloud/storage/internal/connection_impl.cc b/google/cloud/storage/internal/connection_impl.cc
index bac46cfd22f53..503636d4c0bd4 100644
--- a/google/cloud/storage/internal/connection_impl.cc
+++ b/google/cloud/storage/internal/connection_impl.cc
@@ -420,16 +420,46 @@ StatusOr<std::unique_ptr<ObjectReadSource>> StorageConnectionImpl::ReadObject(
         *current, request, where);
   };
 
-  auto retry_source_factory =
-      [factory, current,
-       request]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
-    auto retry_policy = current->get<RetryPolicyOption>()->clone();
-    auto backoff_policy = current->get<BackoffPolicyOption>()->clone();
-    auto child = factory(request, *retry_policy, *backoff_policy);
+  HedgedObjectReadSource::Position position;
+  position.direction =
+      request.HasOption<ReadLast>() ? kFromEnd : kFromBeginning;
+  position.offset = position.direction == kFromEnd
+                        ? request.GetOption<ReadLast>().value()
+                        : request.StartingByte();
+  if (request.HasOption<ReadRange>()) {
+    position.end_offset = request.GetOption<ReadRange>().value().end;
+  }
+  if (request.HasOption<Generation>()) {
+    position.generation = request.GetOption<Generation>().value();
+  }
+
+  // Creates a `RetryObjectReadSource` positioned at `current_offset`, this is
+  // the same request rewrite `RetryObjectReadSource` applies when it resumes
+  // after a failure.
+  auto child_factory = [factory, current, request](
+                           std::int64_t current_offset,
+                           std::optional<std::int64_t> generation)
+      -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    ReadObjectRangeRequest req = request;
+    if (req.HasOption<ReadLast>()) {
+      req.set_option(ReadLast(current_offset));
+    } else if (current_offset != 0 || req.HasOption<ReadRange>() ||
+               req.HasOption<ReadFromOffset>()) {
+      req.set_option(ReadFromOffset(current_offset));
+    }
+    if (generation) {
+      req.set_option(Generation(*generation));
+    }
+    std::unique_ptr<RetryPolicy> retry_policy =
+        current->get<RetryPolicyOption>()->clone();
+    std::unique_ptr<BackoffPolicy> backoff_policy =
+        current->get<BackoffPolicyOption>()->clone();
+    StatusOr<std::unique_ptr<ObjectReadSource>> child =
+        factory(req, *retry_policy, *backoff_policy);
     if (!child) return child;
     return std::unique_ptr<ObjectReadSource>(
         std::make_unique<RetryObjectReadSource>(
-            factory, current, request, *std::move(child),
+            factory, current, std::move(req), *std::move(child),
             std::move(retry_policy), std::move(backoff_policy)));
   };
 
@@ -442,15 +472,15 @@ StatusOr<std::unique_ptr<ObjectReadSource>> StorageConnectionImpl::ReadObject(
       current->get<storage_experimental::MaximumHedgeBufferOption>();
 
   if (!enable_hedging || max_hedges <= 0 || !hedge_pool_ || !read_pool_) {
-    return retry_source_factory();
+    return child_factory(position.offset, position.generation);
   }
 
   // `max_buffer` bounds the size of an individual read, which is only known
   // when the application calls `Read()`; the source applies it there.
   return std::unique_ptr<ObjectReadSource>(
-      std::make_unique<HedgedObjectReadSource>(read_pool_, hedge_pool_,
-                                               std::move(retry_source_factory),
-                                               delay, max_hedges, max_buffer));
+      std::make_unique<HedgedObjectReadSource>(
+          read_pool_, hedge_pool_, std::move(child_factory), delay, max_hedges,
+          max_buffer, position));
 }
 
 StatusOr<ListObjectsResponse> StorageConnectionImpl::ListObjects(
diff --git a/google/cloud/storage/internal/connection_impl_test.cc b/google/cloud/storage/internal/connection_impl_test.cc
index f18ae8e209386..1e8a4dbecaa88 100644
--- a/google/cloud/storage/internal/connection_impl_test.cc
+++ b/google/cloud/storage/internal/connection_impl_test.cc
@@ -16,11 +16,13 @@
 #include "google/cloud/storage/internal/tracing_connection.h"
 #include "google/cloud/storage/options.h"
 #include "google/cloud/storage/testing/canonical_errors.h"
+#include "google/cloud/storage/testing/mock_client.h"
 #include "google/cloud/storage/testing/mock_generic_stub.h"
 #include "google/cloud/testing_util/chrono_literals.h"
 #include "google/cloud/testing_util/opentelemetry_matchers.h"
 #include "google/cloud/testing_util/status_matchers.h"
 #include <gmock/gmock.h>
+#include <functional>
 #include <memory>
 #include <string>
 #include <utility>
@@ -695,6 +697,95 @@ TEST(RetryClientTest, BackoffSpansUploadChunk) {
                   SpanNamed("storage::Client::WriteObject/UploadChunk")));
 }
 
+// `ReadObject()` positions the child stream it creates, using the same request
+// rewrite a resumed or hedged read applies. Verify the request that reaches
+// the stub for each way a caller can express a starting position.
+StatusOr<std::unique_ptr<ObjectReadSource>> ReadObjectWithRequest(
+    ReadObjectRangeRequest const& request,
+    std::function<void(ReadObjectRangeRequest const&)> check) {
+  auto mock = std::make_unique<MockGenericStub>();
+  EXPECT_CALL(*mock, options).Times(AtLeast(0));
+  EXPECT_CALL(*mock, ReadObject)
+      .WillOnce([check = std::move(check)](auto&, auto const&,
+                                           ReadObjectRangeRequest const& req) {
+        check(req);
+        return std::unique_ptr<ObjectReadSource>(
+            std::make_unique<testing::MockObjectReadSource>());
+      });
+  auto client = StorageConnectionImpl::Create(std::move(mock));
+  google::cloud::internal::OptionsSpan const span(BasicTestPolicies());
+  return client->ReadObject(request);
+}
+
+TEST(RetryClientTest, ReadObjectPositionsPlainRequest) {
+  ReadObjectRangeRequest request("test-bucket", "test-object");
+  EXPECT_THAT(
+      ReadObjectWithRequest(request,
+                            [](ReadObjectRangeRequest const& req) {
+                              // Offset 0 with no range: the rewrite must not
+                              // introduce a `ReadFromOffset(0)`, which would
+                              // change the request.
+                              EXPECT_FALSE(req.HasOption<ReadFromOffset>());
+                              EXPECT_FALSE(req.HasOption<ReadLast>());
+                              EXPECT_FALSE(req.HasOption<Generation>());
+                            }),
+      IsOk());
+}
+
+TEST(RetryClientTest, ReadObjectPositionsReadFromOffset) {
+  ReadObjectRangeRequest request("test-bucket", "test-object");
+  request.set_option(ReadFromOffset(1024));
+  EXPECT_THAT(ReadObjectWithRequest(
+                  request,
+                  [](ReadObjectRangeRequest const& req) {
+                    EXPECT_EQ(1024,
+                              req.GetOption<ReadFromOffset>().value_or(0));
+                  }),
+              IsOk());
+}
+
+TEST(RetryClientTest, ReadObjectPositionsReadRange) {
+  ReadObjectRangeRequest request("test-bucket", "test-object");
+  request.set_option(ReadRange(100, 200));
+  EXPECT_THAT(ReadObjectWithRequest(
+                  request,
+                  [](ReadObjectRangeRequest const& req) {
+                    // The range is preserved, and the offset is pinned to its
+                    // start so a resumed read picks up where this one left off.
+                    EXPECT_EQ(100, req.GetOption<ReadFromOffset>().value_or(0));
+                    ASSERT_TRUE(req.HasOption<ReadRange>());
+                    EXPECT_EQ(200, req.GetOption<ReadRange>().value().end);
+                  }),
+              IsOk());
+}
+
+TEST(RetryClientTest, ReadObjectPositionsReadLast) {
+  ReadObjectRangeRequest request("test-bucket", "test-object");
+  request.set_option(ReadLast(512));
+  EXPECT_THAT(ReadObjectWithRequest(
+                  request,
+                  [](ReadObjectRangeRequest const& req) {
+                    // `ReadLast` counts from the end, so it is rewritten in
+                    // place and must not become a `ReadFromOffset`.
+                    EXPECT_EQ(512, req.GetOption<ReadLast>().value_or(0));
+                    EXPECT_FALSE(req.HasOption<ReadFromOffset>());
+                  }),
+              IsOk());
+}
+
+TEST(RetryClientTest, ReadObjectPinsGeneration) {
+  ReadObjectRangeRequest request("test-bucket", "test-object");
+  request.set_option(Generation(12345));
+  request.set_option(ReadFromOffset(64));
+  EXPECT_THAT(ReadObjectWithRequest(
+                  request,
+                  [](ReadObjectRangeRequest const& req) {
+                    EXPECT_EQ(12345, req.GetOption<Generation>().value_or(0));
+                    EXPECT_EQ(64, req.GetOption<ReadFromOffset>().value_or(0));
+                  }),
+              IsOk());
+}
+
 }  // namespace
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
diff --git a/google/cloud/storage/internal/hedged_object_read_source.cc b/google/cloud/storage/internal/hedged_object_read_source.cc
index 8378ba8f6ad29..6c6f1aa0a56b7 100644
--- a/google/cloud/storage/internal/hedged_object_read_source.cc
+++ b/google/cloud/storage/internal/hedged_object_read_source.cc
@@ -13,10 +13,13 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/hedged_object_read_source.h"
+#include "google/cloud/storage/retry_policy.h"
 #include "google/cloud/internal/make_status.h"
 #include <atomic>
+#include <chrono>
 #include <cstring>
 #include <future>
+#include <mutex>
 #include <utility>
 
 namespace google {
@@ -26,28 +29,98 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
 namespace {
 
+// The number of races a single stream may hedge, as a multiple of
+// `max_hedges`. `max_hedges` bounds one race; without this budget a stream
+// that is merely slow (every read takes longer than the hedge delay) would
+// re-race on every read and issue an unbounded number of duplicate requests,
+// because the rate limit and concurrency backstops are both disabled by
+// default.
+int constexpr kMaxHedgeRoundsPerStream = 8;
+
 struct RaceResult {
   StatusOr<ReadSourceResult> result;
   std::unique_ptr<ObjectReadSource> source;
   std::unique_ptr<char[]> buffer;
+  std::size_t buffer_capacity = 0;
 };
 
+// Shared between the caller, which schedules the attempts and waits for the
+// winner, and the attempts themselves, which may outlive the caller's wait.
 struct RaceState {
   std::promise<RaceResult> promise;
   std::atomic<bool> resolved{false};
+  std::atomic<int> active_attempts{0};
+
+  // The primary attempt reads from the active child (if any) into the staging
+  // buffer kept from the previous race (if large enough). Both are consumed
+  // by the primary attempt when it starts.
+  std::unique_ptr<ObjectReadSource> primary_child;
+  std::unique_ptr<char[]> primary_buffer;
+  std::size_t primary_buffer_capacity = 0;
+
+  std::mutex mu;
+  Status primary_error;  // GUARDED_BY(mu)
+  Status last_error;     // GUARDED_BY(mu)
+
+  // Returns true for exactly one caller: the one that gets to set the result.
+  bool TryClaim() {
+    bool expected = false;
+    return resolved.compare_exchange_strong(expected, true);
+  }
+
+  // The error reported when every attempt fails. The primary describes the
+  // stream the caller is actually reading, so its error takes precedence over
+  // whatever a hedge happened to fail with last.
+  Status FinalError() {
+    std::lock_guard<std::mutex> lock(mu);
+    if (!primary_error.ok()) return primary_error;
+    return last_error;
+  }
+
+  // Called once for every attempt that ends without a result: an open error,
+  // a read error, or a hedge that could not be dispatched. The last attempt
+  // to retire resolves the race with the collected error.
+  void RetireAttempt() {
+    if (active_attempts.fetch_sub(1) != 1) return;
+    if (!TryClaim()) return;
+    promise.set_value(RaceResult{FinalError(), nullptr, nullptr});
+  }
+
+  void Fail(Status status, bool is_primary) {
+    bool const permanent =
+        is_primary && StatusTraits::IsPermanentFailure(status);
+    {
+      std::lock_guard<std::mutex> lock(mu);
+      if (is_primary) {
+        primary_error = std::move(status);
+      } else {
+        last_error = std::move(status);
+      }
+    }
+    // A hedge cannot fix a permanent error on the primary (the object is gone,
+    // access is denied, ...). Report it now instead of holding the caller
+    // until every in-flight hedge has exhausted its own retry budget.
+    if (permanent && TryClaim()) {
+      promise.set_value(RaceResult{FinalError(), nullptr, nullptr});
+    }
+    RetireAttempt();
+  }
 };
 
-// Opens a new child and performs its initial read, resolving the race if this
-// attempt finishes first. Losing attempts close their child. Only the primary
-// attempt resolves the race on an open error: a hedge that fails to open must
-// not mask a slower, but successful, primary.
+// Runs a single read attempt. The primary attempt reads from the active child
+// when the stream has one, any other attempt opens a new child at @p offset
+// and @p generation. A successful read resolves the race immediately; the
+// loser closes its own child. A failed attempt only resolves the race if it
+// is the last one standing, or if it is the primary failing permanently.
 void RunAttempt(std::shared_ptr<RaceState> const& state,
                 HedgedObjectReadSource::ChildFactory const& factory,
-                std::size_t n, bool resolve_on_open_error,
+                std::unique_ptr<ObjectReadSource> child,
+                std::unique_ptr<char[]> buffer, std::size_t buffer_capacity,
+                std::int64_t offset, std::optional<std::int64_t> generation,
+                std::size_t n, bool is_primary,
                 std::shared_ptr<HedgingThreadPool> release_slot) {
   // Releases the acquired hedge concurrency slot upon function exit across
-  // all code paths (early return on open/allocation error, race winner, or
-  // race loser). For primary attempts, release_slot is nullptr.
+  // all code paths. For the primary attempt, release_slot is nullptr.
   struct SlotGuard {
     std::shared_ptr<HedgingThreadPool> pool;
     ~SlotGuard() {
@@ -55,37 +128,40 @@ void RunAttempt(std::shared_ptr<RaceState> const& state,
     }
   } guard{std::move(release_slot)};
 
-  auto source = factory();
-  if (!source) {
-    if (!resolve_on_open_error) return;
-    bool expected = false;
-    if (state->resolved.compare_exchange_strong(expected, true)) {
-      state->promise.set_value(
-          RaceResult{std::move(source).status(), nullptr, {}});
-    }
-    return;
+  if (!child) {
+    StatusOr<std::unique_ptr<ObjectReadSource>> source =
+        factory(offset, generation);
+    if (!source) return state->Fail(std::move(source).status(), is_primary);
+    child = *std::move(source);
   }
-  std::unique_ptr<char[]> buffer(new (std::nothrow) char[n]);
+
   if (!buffer) {
-    if (!resolve_on_open_error) return;
-    bool expected = false;
-    if (state->resolved.compare_exchange_strong(expected, true)) {
-      state->promise.set_value(RaceResult{
+    buffer.reset(new (std::nothrow) char[n]);
+    if (!buffer) {
+      return state->Fail(
           google::cloud::internal::ResourceExhaustedError(
               "Out of memory allocating hedge buffer", GCP_ERROR_INFO()),
-          nullptr,
-          {}});
+          is_primary);
     }
-    return;
+    buffer_capacity = n;
   }
-  auto result = (*source)->Read(buffer.get(), n);
-  bool expected = false;
-  if (state->resolved.compare_exchange_strong(expected, true)) {
-    state->promise.set_value(
-        RaceResult{std::move(result), *std::move(source), std::move(buffer)});
-  } else {
-    (*source)->Close();
+
+  StatusOr<ReadSourceResult> result = child->Read(buffer.get(), n);
+  if (!result) {
+    // A child that failed may have already torn down its connection, e.g. a
+    // `RetryObjectReadSource` that exhausted its retry policy has no child of
+    // its own to close.
+    if (child->IsOpen()) child->Close();
+    return state->Fail(std::move(result).status(), is_primary);
+  }
+
+  if (!state->TryClaim()) {
+    // Lost the race, the winner's data was already returned to the caller.
+    child->Close();
+    return;
   }
+  state->promise.set_value(RaceResult{std::move(result), std::move(child),
+                                      std::move(buffer), buffer_capacity});
 }
 
 }  // namespace
@@ -93,13 +169,27 @@ void RunAttempt(std::shared_ptr<RaceState> const& state,
 HedgedObjectReadSource::HedgedObjectReadSource(
     std::shared_ptr<ThreadPool> read_pool,
     std::shared_ptr<HedgingThreadPool> hedge_pool, ChildFactory child_factory,
-    std::chrono::milliseconds delay, int max_hedges, std::size_t max_buffer)
+    std::chrono::milliseconds delay, int max_hedges, std::size_t max_buffer,
+    Position position)
     : read_pool_(std::move(read_pool)),
       hedge_pool_(std::move(hedge_pool)),
-      child_factory_(std::move(child_factory)),
+      child_factory_(
+          std::make_shared<ChildFactory const>(std::move(child_factory))),
       delay_(delay),
       max_hedges_(max_hedges),
-      max_buffer_(max_buffer) {}
+      max_buffer_(max_buffer),
+      current_offset_(position.offset),
+      offset_direction_(position.direction),
+      end_offset_(position.end_offset),
+      generation_(position.generation) {}
+
+HedgedObjectReadSource::HedgedObjectReadSource(
+    std::shared_ptr<ThreadPool> read_pool,
+    std::shared_ptr<HedgingThreadPool> hedge_pool, ChildFactory child_factory,
+    std::chrono::milliseconds delay, int max_hedges, std::size_t max_buffer)
+    : HedgedObjectReadSource(std::move(read_pool), std::move(hedge_pool),
+                             std::move(child_factory), delay, max_hedges,
+                             max_buffer, Position{}) {}
 
 bool HedgedObjectReadSource::IsOpen() const {
   if (active_child_) return active_child_->IsOpen();
@@ -119,28 +209,93 @@ StatusOr<ReadSourceResult> HedgedObjectReadSource::Read(char* buf,
   if (is_closed_) {
     return ReadSourceResult{0, HttpResponse{HttpStatusCode::kOk, {}, {}}};
   }
+  std::chrono::steady_clock::time_point const start =
+      std::chrono::steady_clock::now();
+  StatusOr<ReadSourceResult> result =
+      ShouldRace(n) ? ReadRaced(buf, n) : ReadDirect(buf, n);
+  last_read_stalled_ = std::chrono::steady_clock::now() - start > delay_;
+  UpdateState(result);
+  return result;
+}
 
-  // Only the stream open is hedged. Once a child has won the race all
-  // subsequent reads continue on it, at its current offset, without any
-  // thread hops or extra copies.
-  if (active_child_) return active_child_->Read(buf, n);
+bool HedgedObjectReadSource::ShouldRace(std::size_t n) const {
+  if (max_hedges_ <= 0 || !read_pool_ || !hedge_pool_) return false;
+  // Racing stages one copy of `n` bytes per attempt on top of the caller's
+  // buffer. For a large read that multiplication is worse than the tail
+  // latency it avoids.
+  if (n > max_buffer_) return false;
+  // The stream open is always raced, that is where most tail latency lives.
+  if (!active_child_) return true;
+  // Decompressive transcoding does not respect byte ranges (HTTP 206). A
+  // mid-stream hedge would have to re-read and discard from offset 0, which
+  // is worse than reading directly on the active child.
+  if (is_gunzipped_) return false;
+  // The caller drains a stream with one more read at the end of the requested
+  // data. A hedge there would request an empty or inverted range, and could
+  // even win the race with bytes from the wrong offset.
+  if (AtEnd()) return false;
+  // A stream that is uniformly slow, rather than intermittently stalled, would
+  // otherwise re-race every read for the life of the stream.
+  if (total_hedges_ >= max_hedges_ * kMaxHedgeRoundsPerStream) return false;
+  // Otherwise only re-race a stream that has shown signs of stalling, so a
+  // healthy stream keeps the zero-cost direct path.
+  return last_read_stalled_;
+}
+
+bool HedgedObjectReadSource::AtEnd() const {
+  if (offset_direction_ == kFromEnd) return current_offset_ <= 0;
+  // An explicit range end is authoritative. `size_` comes from the response,
+  // and on the REST path it falls back to `content-length`, which is the
+  // length of that response rather than the size of the object. For a ranged
+  // read that is smaller than the stream's offset, so consulting it here would
+  // end the stream on the first check.
+  if (end_offset_) return current_offset_ >= *end_offset_;
+  return size_ && current_offset_ >= static_cast<std::int64_t>(*size_);
+}
 
-  // Racing requires one staging buffer of `n` bytes per attempt, on top of the
-  // caller's own buffer. For a large read that multiplication is worse than
-  // the tail latency it avoids, so open the stream without hedging and read
-  // straight into the caller's buffer.
-  if (n > max_buffer_) {
-    auto child = child_factory_();
-    if (!child) return std::move(child).status();
+StatusOr<ReadSourceResult> HedgedObjectReadSource::ReadDirect(char* buf,
+                                                              std::size_t n) {
+  if (!active_child_) {
+    StatusOr<std::unique_ptr<ObjectReadSource>> child =
+        (*child_factory_)(current_offset_, generation_);
+    if (!child) {
+      // The stream never opened, there is nothing to read from or to close.
+      is_closed_ = true;
+      return std::move(child).status();
+    }
     active_child_ = *std::move(child);
-    return active_child_->Read(buf, n);
   }
+  StatusOr<ReadSourceResult> result = active_child_->Read(buf, n);
+  if (!result) {
+    // Match `ReadRaced()`: a child whose read failed has exhausted its retry
+    // policy, there is nothing left to read from. A child that is still open
+    // holds a connection that must be released.
+    if (active_child_->IsOpen()) active_child_->Close();
+    active_child_.reset();
+    is_closed_ = true;
+  }
+  return result;
+}
 
+StatusOr<ReadSourceResult> HedgedObjectReadSource::ReadRaced(char* buf,
+                                                             std::size_t n) {
   auto state = std::make_shared<RaceState>();
-  auto future = state->promise.get_future();
+  std::future<RaceResult> future = state->promise.get_future();
+  state->active_attempts.store(1);
+  state->primary_child = std::move(active_child_);
+  if (staging_buffer_capacity_ >= n) {
+    state->primary_buffer = std::move(staging_buffer_);
+    state->primary_buffer_capacity = staging_buffer_capacity_;
+  }
+  staging_buffer_.reset();
+  staging_buffer_capacity_ = 0;
 
-  auto primary = [state, factory = child_factory_, n] {
-    RunAttempt(state, factory, n, /*resolve_on_open_error=*/true, nullptr);
+  auto primary = [state, factory = child_factory_, offset = current_offset_,
+                  gen = generation_, n] {
+    RunAttempt(state, *factory, std::move(state->primary_child),
+               std::move(state->primary_buffer), state->primary_buffer_capacity,
+               offset, gen, n,
+               /*is_primary=*/true, nullptr);
   };
   // The primary attempt is scheduled on the dedicated read pool.
   // If the pool is shutting down run the attempt inline, the read must
@@ -161,24 +316,65 @@ StatusOr<ReadSourceResult> HedgedObjectReadSource::Read(char* buf,
       }
       continue;
     }
-    auto hedge = [state, factory = child_factory_, n, pool = hedge_pool_] {
-      RunAttempt(state, factory, n, /*resolve_on_open_error=*/false, pool);
+    state->active_attempts.fetch_add(1);
+    auto hedge = [state, factory = child_factory_, offset = current_offset_,
+                  gen = generation_, n, pool = hedge_pool_] {
+      RunAttempt(state, *factory, /*child=*/nullptr, /*buffer=*/nullptr,
+                 /*buffer_capacity=*/0, offset, gen, n, /*is_primary=*/false,
+                 pool);
     };
     if (!hedge_pool_->Enqueue(hedge)) {
       hedge_pool_->ReleaseHedgeSlot();
+      state->RetireAttempt();
       break;
     }
     ++hedges_dispatched;
+    ++total_hedges_;
   }
 
-  auto race = future.get();
+  RaceResult race = future.get();
   active_child_ = std::move(race.source);
-  if (race.result.ok() && race.result->bytes_received > 0) {
+  if (!race.result) {
+    // Every attempt failed and closed its own child, there is nothing left to
+    // read from or to close.
+    is_closed_ = true;
+    return std::move(race.result).status();
+  }
+  if (race.result->bytes_received > 0) {
     std::memcpy(buf, race.buffer.get(), race.result->bytes_received);
   }
+  staging_buffer_ = std::move(race.buffer);
+  staging_buffer_capacity_ = race.buffer_capacity;
   return race.result;
 }
 
+void HedgedObjectReadSource::UpdateState(
+    StatusOr<ReadSourceResult> const& result) {
+  if (!result) return;
+  if (result->generation) generation_ = result->generation;
+  if (result->size && !size_) size_ = result->size;
+  if (result->transformation.value_or("") == "gunzipped") {
+    // Decompressive transcoding does not respect byte ranges, so `ShouldRace()`
+    // disengages for the rest of the stream once this is set. No hedge will
+    // reopen the object, so there is no resume position left to track.
+    is_gunzipped_ = true;
+    return;
+  }
+  auto const received = static_cast<std::int64_t>(result->bytes_received);
+  if (offset_direction_ == kFromEnd) {
+    // `ReadLast(N)` with `N` larger than the object returns the whole object.
+    // The bytes still to read are then bounded by the object size, not by
+    // `N`, otherwise a hedge opened with the remaining count would cover the
+    // whole object again and return data from the first byte.
+    if (size_ && current_offset_ > static_cast<std::int64_t>(*size_)) {
+      current_offset_ = static_cast<std::int64_t>(*size_);
+    }
+    current_offset_ -= received;
+  } else {
+    current_offset_ += received;
+  }
+}
+
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage
diff --git a/google/cloud/storage/internal/hedged_object_read_source.h b/google/cloud/storage/internal/hedged_object_read_source.h
index c224b3ecedafb..ac587e50daf67 100644
--- a/google/cloud/storage/internal/hedged_object_read_source.h
+++ b/google/cloud/storage/internal/hedged_object_read_source.h
@@ -17,10 +17,13 @@
 
 #include "google/cloud/storage/internal/hedging_thread_pool.h"
 #include "google/cloud/storage/internal/object_read_source.h"
+#include "google/cloud/storage/internal/retry_object_read_source.h"
 #include "google/cloud/storage/version.h"
 #include <chrono>
+#include <cstdint>
 #include <functional>
 #include <memory>
+#include <optional>
 
 namespace google {
 namespace cloud {
@@ -29,31 +32,64 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
 
 /**
- * Hedge the *open* of an `ObjectReadSource` to reduce tail latency.
+ * Hedge reads of an `ObjectReadSource` to reduce tail latency.
  *
- * The first `Read()` races one or more children created by `child_factory`:
- * a primary attempt starts immediately, and up to @p max_hedges additional
- * attempts start, staggered by @p delay, while no attempt has completed. The
- * first attempt to complete its initial read wins; losing attempts are closed
- * when they eventually complete.
+ * The first `Read()` (the stream open) races a primary attempt against up to
+ * @p max_hedges additional attempts created by @p child_factory, each started
+ * after @p delay elapses without a winner. The first attempt to complete its
+ * read wins and becomes the active child; losing attempts are closed when they
+ * eventually complete.
  *
- * Only the initial open is hedged. `ObjectReadSource` is a stream, so a hedge
- * started mid-stream would restart from the request's initial offset and
- * could return the wrong bytes. After the race, all subsequent reads simply
- * continue on the winning child at its current offset, with no extra threads
- * or copies.
+ * Later reads normally continue on the active child, on the caller's thread,
+ * with no thread hops or copies. A read that takes longer than @p delay marks
+ * the stream as stalled, and the next read is raced again: the active child is
+ * the primary attempt, and hedges are opened by @p child_factory at the
+ * stream's current offset, pinned to the generation observed so far. A hedge
+ * that wins replaces the active child. Once a read completes within @p delay
+ * the stream goes back to direct reads. The number of hedges a single stream
+ * may issue this way is capped at a small multiple of @p max_hedges, so a
+ * stream that is uniformly slow rather than intermittently stalled stops
+ * racing instead of duplicating every read.
  *
- * Each racing attempt reads into its own buffer, because a losing attempt
- * keeps writing until it completes and must not touch the caller's buffer.
- * Peak memory for the race is therefore proportional to the size of the first
- * read. Reads larger than @p max_buffer are served without hedging, directly
- * into the caller's buffer, so a large read cannot multiply memory use.
+ * Racing is skipped where it cannot produce correct data or cannot help: under
+ * decompressive transcoding (byte ranges are not honored, a hedge would restart
+ * from the first byte), for reads larger than @p max_buffer (each attempt
+ * stages its own copy of the data, so a large read would multiply memory use),
+ * and once the stream has reached the end of the requested data (a hedge would
+ * request an empty or invalid range).
  */
 class HedgedObjectReadSource : public ObjectReadSource {
  public:
+  /**
+   * Creates a child stream positioned at @p current_offset.
+   *
+   * With `kFromBeginning` the offset counts bytes from the start of the object,
+   * with `kFromEnd` it is the number of bytes still to read from the end of the
+   * object (`ReadLast`). The child must read the given @p generation when one
+   * is known.
+   */
   using ChildFactory =
-      std::function<StatusOr<std::unique_ptr<ObjectReadSource>>()>;
+      std::function<StatusOr<std::unique_ptr<ObjectReadSource>>(
+          std::int64_t current_offset, std::optional<std::int64_t> generation)>;
+
+  /// Where the stream starts, as derived from the original request.
+  struct Position {
+    /// Bytes from the start of the object, or for `ReadLast` the bytes
+    /// remaining to read from the end of the object.
+    std::int64_t offset = 0;
+    OffsetDirection direction = kFromBeginning;
+    /// Exclusive end of the requested range, if the request has one.
+    std::optional<std::int64_t> end_offset;
+    std::optional<std::int64_t> generation;
+  };
 
+  HedgedObjectReadSource(std::shared_ptr<ThreadPool> read_pool,
+                         std::shared_ptr<HedgingThreadPool> hedge_pool,
+                         ChildFactory child_factory,
+                         std::chrono::milliseconds delay, int max_hedges,
+                         std::size_t max_buffer, Position position);
+
+  /// A stream that starts at the beginning of the object.
   HedgedObjectReadSource(std::shared_ptr<ThreadPool> read_pool,
                          std::shared_ptr<HedgingThreadPool> hedge_pool,
                          ChildFactory child_factory,
@@ -67,13 +103,35 @@ class HedgedObjectReadSource : public ObjectReadSource {
   StatusOr<ReadSourceResult> Read(char* buf, std::size_t n) override;
 
  private:
+  bool ShouldRace(std::size_t n) const;
+  bool AtEnd() const;
+  StatusOr<ReadSourceResult> ReadDirect(char* buf, std::size_t n);
+  StatusOr<ReadSourceResult> ReadRaced(char* buf, std::size_t n);
+  void UpdateState(StatusOr<ReadSourceResult> const& result);
+
   std::shared_ptr<ThreadPool> read_pool_;
   std::shared_ptr<HedgingThreadPool> hedge_pool_;
-  ChildFactory child_factory_;
+  // Shared with the racing attempts, which may outlive this object.
+  std::shared_ptr<ChildFactory const> child_factory_;
   std::chrono::milliseconds delay_;
   int max_hedges_;
   std::size_t max_buffer_;
 
+  std::int64_t current_offset_;
+  OffsetDirection offset_direction_;
+  std::optional<std::int64_t> end_offset_;
+  std::optional<std::int64_t> generation_;
+  std::optional<std::uint64_t> size_;
+  bool is_gunzipped_ = false;
+  bool last_read_stalled_ = false;
+  // Hedges dispatched over the life of this stream, bounded so a uniformly
+  // slow stream cannot race forever.
+  int total_hedges_ = 0;
+
+  // The staging buffer of the last winning attempt, reused by the next race.
+  std::unique_ptr<char[]> staging_buffer_;
+  std::size_t staging_buffer_capacity_ = 0;
+
   std::unique_ptr<ObjectReadSource> active_child_;
   bool is_closed_ = false;
 };
diff --git a/google/cloud/storage/internal/hedged_object_read_source_test.cc b/google/cloud/storage/internal/hedged_object_read_source_test.cc
index 0b934d3a31216..0234516a6517d 100644
--- a/google/cloud/storage/internal/hedged_object_read_source_test.cc
+++ b/google/cloud/storage/internal/hedged_object_read_source_test.cc
@@ -23,6 +23,7 @@
 #include <future>
 #include <memory>
 #include <string>
+#include <thread>
 #include <vector>
 
 namespace google {
@@ -36,10 +37,20 @@ using ::google::cloud::storage::testing::MockObjectReadSource;
 using ::google::cloud::testing_util::IsOk;
 using ::google::cloud::testing_util::StatusIs;
 using ::testing::Eq;
+using ::testing::Return;
 
 // Large enough that no test read is treated as oversized.
 std::size_t constexpr kUnlimitedBuffer = std::size_t{1} << 30;
 
+// A hedge delay no test read exceeds, so a stream that answers immediately is
+// never considered stalled and never hedged after the open.
+auto constexpr kLongDelay = std::chrono::seconds(30);
+
+// The hedge delay for tests that drive a stall: `kStall` is comfortably above
+// it, while an immediate answer is comfortably below it.
+auto constexpr kDelay = std::chrono::milliseconds(100);
+auto constexpr kStall = std::chrono::milliseconds(200);
+
 std::shared_ptr<ThreadPool> MakeUnlimitedReadPool() {
   return std::make_shared<ThreadPool>(/*max_threads=*/4);
 }
@@ -50,11 +61,84 @@ std::shared_ptr<HedgingThreadPool> MakeUnlimitedHedgePool() {
       /*max_concurrent=*/0);
 }
 
+// Most tests do not care about the offset or generation a child is opened at.
+template <typename F>
+HedgedObjectReadSource::ChildFactory Adapt(F f) {
+  return [f = std::move(f)](std::int64_t, std::optional<std::int64_t>) {
+    return f();
+  };
+}
+
 ReadSourceResult MakeReadResult(std::string const& payload) {
   return ReadSourceResult{payload.size(),
                           HttpResponse{HttpStatusCode::kOk, {}, {}}};
 }
 
+// A `Read()` action that returns @p payload immediately.
+auto ImmediateRead(std::string payload) {
+  return [payload = std::move(payload)](char* buf, std::size_t n) {
+    EXPECT_LE(payload.size(), n);
+    std::copy(payload.begin(), payload.end(), buf);
+    return MakeReadResult(payload);
+  };
+}
+
+// A `Read()` action that returns @p payload after @p delay. With a delay above
+// the source's hedge delay this marks the stream as stalled.
+auto DelayedRead(std::string payload, std::chrono::milliseconds delay) {
+  return [payload = std::move(payload), delay](char* buf, std::size_t n) {
+    EXPECT_LE(payload.size(), n);
+    std::this_thread::sleep_for(delay);
+    std::copy(payload.begin(), payload.end(), buf);
+    return MakeReadResult(payload);
+  };
+}
+
+// A `Read()` action that returns @p payload once @p unblock is set.
+auto BlockedRead(std::shared_ptr<std::promise<void>> unblock,
+                 std::string payload) {
+  return [unblock = std::move(unblock), payload = std::move(payload)](
+             char* buf, std::size_t n) {
+    EXPECT_LE(payload.size(), n);
+    unblock->get_future().get();
+    std::copy(payload.begin(), payload.end(), buf);
+    return MakeReadResult(payload);
+  };
+}
+
+// Blocks until @p signal is set, or until a generous timeout expires.
+//
+// Tests that need an attempt to outlive the hedge dispatch must wait for the
+// hedge itself rather than sleep: the two run on different threads, and a
+// sleep long enough on an idle machine can still be too short on a loaded one.
+// The timeout keeps a regression a test failure instead of a hang.
+void WaitForSignal(std::shared_ptr<std::promise<void>> const& signal) {
+  EXPECT_EQ(signal->get_future().wait_for(std::chrono::seconds(10)),
+            std::future_status::ready);
+}
+
+// Records that a hedge reached the factory. `Signal()` is safe to call from
+// several attempts, only the first one sets the promise.
+struct HedgeSignal {
+  std::shared_ptr<std::promise<void>> reached =
+      std::make_shared<std::promise<void>>();
+  std::shared_ptr<std::atomic<bool>> signalled =
+      std::make_shared<std::atomic<bool>>(false);
+
+  void Signal() const {
+    if (!signalled->exchange(true)) reached->set_value();
+  }
+  void Wait() const { WaitForSignal(reached); }
+};
+
+// A `Close()` action that sets @p closed.
+auto NotifyClose(std::shared_ptr<std::promise<void>> closed) {
+  return [closed = std::move(closed)]() {
+    closed->set_value();
+    return make_status_or(HttpResponse{HttpStatusCode::kOk, {}, {}});
+  };
+}
+
 auto MakeStallingPrimaryFactory(
     std::shared_ptr<std::promise<void>> const& unblock_primary,
     std::shared_ptr<std::promise<void>> const& primary_closed,
@@ -63,23 +147,10 @@ auto MakeStallingPrimaryFactory(
           calls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
     auto mock = std::make_unique<MockObjectReadSource>();
     if (++*calls == 1) {
-      EXPECT_CALL(*mock, Read)
-          .WillOnce([unblock_primary](char* buf, std::size_t) {
-            unblock_primary->get_future().get();
-            std::string const payload = "slow";
-            std::copy(payload.begin(), payload.end(), buf);
-            return MakeReadResult(payload);
-          });
-      EXPECT_CALL(*mock, Close).WillOnce([primary_closed]() {
-        primary_closed->set_value();
-        return make_status_or(HttpResponse{HttpStatusCode::kOk, {}, {}});
-      });
+      EXPECT_CALL(*mock, Read).WillOnce(BlockedRead(unblock_primary, "slow"));
+      EXPECT_CALL(*mock, Close).WillOnce(NotifyClose(primary_closed));
     } else {
-      EXPECT_CALL(*mock, Read).WillOnce([](char* buf, std::size_t) {
-        std::string const payload = "hedge";
-        std::copy(payload.begin(), payload.end(), buf);
-        return MakeReadResult(payload);
-      });
+      EXPECT_CALL(*mock, Read).WillOnce(ImmediateRead("hedge"));
     }
     return std::unique_ptr<ObjectReadSource>(std::move(mock));
   };
@@ -88,18 +159,13 @@ auto MakeStallingPrimaryFactory(
 TEST(HedgedObjectReadSourceTest, PrimaryWins) {
   auto factory = []() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
     auto mock = std::make_unique<MockObjectReadSource>();
-    EXPECT_CALL(*mock, Read).WillOnce([](char* buf, std::size_t) {
-      std::string const payload = "payload";
-      std::copy(payload.begin(), payload.end(), buf);
-      return MakeReadResult(payload);
-    });
+    EXPECT_CALL(*mock, Read).WillOnce(ImmediateRead("payload"));
     return std::unique_ptr<ObjectReadSource>(std::move(mock));
   };
 
   HedgedObjectReadSource source(MakeUnlimitedReadPool(),
-                                MakeUnlimitedHedgePool(), factory,
-                                std::chrono::milliseconds(500),
-                                /*max_hedges=*/2, kUnlimitedBuffer);
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kLongDelay, /*max_hedges=*/2, kUnlimitedBuffer);
 
   std::vector<char> buffer(100);
   auto result = source.Read(buffer.data(), buffer.size());
@@ -110,36 +176,32 @@ TEST(HedgedObjectReadSourceTest, PrimaryWins) {
 }
 
 TEST(HedgedObjectReadSourceTest, SubsequentReadsContinueOnWinner) {
-  // The factory must be called exactly once: after the open race is decided,
-  // reads must continue on the winning child without creating new children,
-  // otherwise the stream would restart at the wrong offset.
+  // Once the open race is decided, reads on a healthy stream continue on the
+  // winning child without opening new children: the factory is called exactly
+  // once.
   auto factory_calls = std::make_shared<std::atomic<int>>(0);
   auto factory =
       [factory_calls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
     ++*factory_calls;
     auto mock = std::make_unique<MockObjectReadSource>();
     EXPECT_CALL(*mock, Read)
-        .WillOnce([](char* buf, std::size_t) {
-          std::string const payload = "chunk-1";
-          std::copy(payload.begin(), payload.end(), buf);
-          return MakeReadResult(payload);
-        })
-        .WillOnce([](char* buf, std::size_t) {
-          std::string const payload = "chunk-2";
-          std::copy(payload.begin(), payload.end(), buf);
-          return MakeReadResult(payload);
-        });
+        .WillOnce(ImmediateRead("chunk-1"))
+        .WillOnce(ImmediateRead("chunk-2"))
+        .WillOnce(ImmediateRead("chunk-3"));
     return std::unique_ptr<ObjectReadSource>(std::move(mock));
   };
 
   HedgedObjectReadSource source(MakeUnlimitedReadPool(),
-                                MakeUnlimitedHedgePool(), factory,
-                                std::chrono::milliseconds(500),
-                                /*max_hedges=*/2, kUnlimitedBuffer);
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kLongDelay, /*max_hedges=*/2, kUnlimitedBuffer);
 
   std::vector<char> buffer(100);
-  EXPECT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
-  EXPECT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  for (auto const* expected : {"chunk-1", "chunk-2", "chunk-3"}) {
+    auto result = source.Read(buffer.data(), buffer.size());
+    ASSERT_THAT(result, IsOk());
+    EXPECT_THAT(std::string(buffer.data(), result->bytes_received),
+                Eq(expected));
+  }
   EXPECT_THAT(factory_calls->load(), Eq(1));
 }
 
@@ -154,7 +216,7 @@ TEST(HedgedObjectReadSourceTest, HedgeWinsWhenPrimaryStalls) {
       MakeStallingPrimaryFactory(unblock_primary, primary_closed, calls);
 
   auto source = std::make_unique<HedgedObjectReadSource>(
-      MakeUnlimitedReadPool(), MakeUnlimitedHedgePool(), factory,
+      MakeUnlimitedReadPool(), MakeUnlimitedHedgePool(), Adapt(factory),
       std::chrono::milliseconds(1),
       /*max_hedges=*/2, kUnlimitedBuffer);
 
@@ -178,7 +240,7 @@ TEST(HedgedObjectReadSourceTest, ReadPoolSaturationDoesNotBlockHedges) {
       MakeStallingPrimaryFactory(unblock_primary, primary_closed, calls);
 
   HedgedObjectReadSource source(std::make_shared<ThreadPool>(/*max_threads=*/1),
-                                MakeUnlimitedHedgePool(), factory,
+                                MakeUnlimitedHedgePool(), Adapt(factory),
                                 std::chrono::milliseconds(1),
                                 /*max_hedges=*/2, kUnlimitedBuffer);
 
@@ -206,15 +268,11 @@ TEST(HedgedObjectReadSourceTest, HedgePoolExhaustionDoesNotBlockPrimary) {
   auto factory = [calls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
     ++*calls;
     auto mock = std::make_unique<MockObjectReadSource>();
-    EXPECT_CALL(*mock, Read).WillOnce([](char* buf, std::size_t) {
-      std::string const payload = "primary_only";
-      std::copy(payload.begin(), payload.end(), buf);
-      return MakeReadResult(payload);
-    });
+    EXPECT_CALL(*mock, Read).WillOnce(ImmediateRead("primary_only"));
     return std::unique_ptr<ObjectReadSource>(std::move(mock));
   };
 
-  HedgedObjectReadSource source(read_pool, hedge_pool, factory,
+  HedgedObjectReadSource source(read_pool, hedge_pool, Adapt(factory),
                                 std::chrono::milliseconds(10),
                                 /*max_hedges=*/2, kUnlimitedBuffer);
 
@@ -254,8 +312,8 @@ TEST(HedgedObjectReadSourceTest,
     hedge_pool->ReleaseHedgeSlot();
   });
 
-  HedgedObjectReadSource source(MakeUnlimitedReadPool(), hedge_pool, factory,
-                                std::chrono::milliseconds(10),
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(), hedge_pool,
+                                Adapt(factory), std::chrono::milliseconds(10),
                                 /*max_hedges=*/1, kUnlimitedBuffer);
 
   std::vector<char> buffer(100);
@@ -287,12 +345,7 @@ TEST(HedgedObjectReadSourceTest, HedgeOpenFailureReleasesSlot) {
       // Primary attempt: stalls until unblocked.
       auto mock = std::make_unique<MockObjectReadSource>();
       EXPECT_CALL(*mock, Read)
-          .WillOnce([unblock_primary](char* buf, std::size_t) {
-            unblock_primary->get_future().get();
-            std::string const payload = "primary";
-            std::copy(payload.begin(), payload.end(), buf);
-            return MakeReadResult(payload);
-          });
+          .WillOnce(BlockedRead(unblock_primary, "primary"));
       return std::unique_ptr<ObjectReadSource>(std::move(mock));
     }
     // Hedge attempt: fails to open.
@@ -303,8 +356,8 @@ TEST(HedgedObjectReadSourceTest, HedgeOpenFailureReleasesSlot) {
       /*max_threads=*/2, /*rate_limit=*/0.0, /*capacity=*/0.0,
       /*max_concurrent=*/1);
 
-  HedgedObjectReadSource source(MakeUnlimitedReadPool(), hedge_pool, factory,
-                                std::chrono::milliseconds(1),
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(), hedge_pool,
+                                Adapt(factory), std::chrono::milliseconds(1),
                                 /*max_hedges=*/1, kUnlimitedBuffer);
 
   std::vector<char> buffer(100);
@@ -338,12 +391,7 @@ TEST(HedgedObjectReadSourceTest, ZeroDelayBacksOffOnHedgeTokenExhaustion) {
     ++*calls;
     auto mock = std::make_unique<MockObjectReadSource>();
     EXPECT_CALL(*mock, Read)
-        .WillOnce([unblock_primary](char* buf, std::size_t) {
-          unblock_primary->get_future().get();
-          std::string const payload = "primary_data";
-          std::copy(payload.begin(), payload.end(), buf);
-          return MakeReadResult(payload);
-        });
+        .WillOnce(BlockedRead(unblock_primary, "primary_data"));
     return std::unique_ptr<ObjectReadSource>(std::move(mock));
   };
 
@@ -353,8 +401,8 @@ TEST(HedgedObjectReadSourceTest, ZeroDelayBacksOffOnHedgeTokenExhaustion) {
   // Exhaust all hedge slots so TryAcquireHedgeToken fails.
   ASSERT_TRUE(hedge_pool->TryAcquireHedgeToken());
 
-  HedgedObjectReadSource source(MakeUnlimitedReadPool(), hedge_pool, factory,
-                                std::chrono::milliseconds(0),
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(), hedge_pool,
+                                Adapt(factory), std::chrono::milliseconds(0),
                                 /*max_hedges=*/2, kUnlimitedBuffer);
 
   std::vector<char> buffer(100);
@@ -381,16 +429,12 @@ TEST(HedgedObjectReadSourceTest, NonPositiveMaxHedgesDoesNotHedge) {
   auto factory = [calls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
     ++*calls;
     auto mock = std::make_unique<MockObjectReadSource>();
-    EXPECT_CALL(*mock, Read).WillOnce([](char* buf, std::size_t) {
-      std::string const payload = "primary_only";
-      std::copy(payload.begin(), payload.end(), buf);
-      return MakeReadResult(payload);
-    });
+    EXPECT_CALL(*mock, Read).WillOnce(ImmediateRead("primary_only"));
     return std::unique_ptr<ObjectReadSource>(std::move(mock));
   };
 
   HedgedObjectReadSource source(MakeUnlimitedReadPool(),
-                                MakeUnlimitedHedgePool(), factory,
+                                MakeUnlimitedHedgePool(), Adapt(factory),
                                 std::chrono::milliseconds(0),
                                 /*max_hedges=*/-1, kUnlimitedBuffer);
 
@@ -408,13 +452,102 @@ TEST(HedgedObjectReadSourceTest, PrimaryOpenErrorPropagates) {
   };
 
   HedgedObjectReadSource source(MakeUnlimitedReadPool(),
-                                MakeUnlimitedHedgePool(), factory,
-                                std::chrono::milliseconds(500),
-                                /*max_hedges=*/2, kUnlimitedBuffer);
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kLongDelay, /*max_hedges=*/2, kUnlimitedBuffer);
 
   std::vector<char> buffer(100);
   EXPECT_THAT(source.Read(buffer.data(), buffer.size()),
               StatusIs(StatusCode::kPermissionDenied));
+  // Nothing was opened, the stream must not report itself as open.
+  EXPECT_FALSE(source.IsOpen());
+}
+
+TEST(HedgedObjectReadSourceTest, PermanentPrimaryErrorResolvesImmediately) {
+  // The primary fails with a permanent error while a hedge is in flight and
+  // stalled. Waiting for the hedge cannot change the outcome, so the error
+  // must be reported at once, and the hedge must be closed when it completes.
+  auto unblock_hedge = std::make_shared<std::promise<void>>();
+  auto hedge_closed = std::make_shared<std::promise<void>>();
+  auto calls = std::make_shared<std::atomic<int>>(0);
+  HedgeSignal hedge_started;
+  auto factory =
+      [unblock_hedge, hedge_closed, calls,
+       hedge_started]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    auto mock = std::make_unique<MockObjectReadSource>();
+    if (++*calls == 1) {
+      // Fail only once the hedge has been dispatched, otherwise the race is
+      // over before there is anything to hedge. The failed child reports
+      // itself as already closed, so it must not be closed again.
+      EXPECT_CALL(*mock, Read).WillOnce([hedge_started](char*, std::size_t) {
+        hedge_started.Wait();
+        return StatusOr<ReadSourceResult>(
+            Status(StatusCode::kNotFound, "object deleted"));
+      });
+      EXPECT_CALL(*mock, IsOpen).WillRepeatedly(Return(false));
+      EXPECT_CALL(*mock, Close).Times(0);
+    } else {
+      hedge_started.Signal();
+      EXPECT_CALL(*mock, Read).WillOnce(BlockedRead(unblock_hedge, "hedge"));
+      EXPECT_CALL(*mock, Close).WillOnce(NotifyClose(hedge_closed));
+    }
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kDelay, /*max_hedges=*/1, kUnlimitedBuffer);
+
+  std::vector<char> buffer(100);
+  auto result = source.Read(buffer.data(), buffer.size());
+  EXPECT_THAT(result, StatusIs(StatusCode::kNotFound));
+  EXPECT_FALSE(source.IsOpen());
+
+  unblock_hedge->set_value();
+  WaitForSignal(hedge_closed);
+  // Asserted only once the hedge has run to completion. A permanent primary
+  // error resolves the race without waiting for the hedge to retire, so
+  // `Read()` returning says nothing about how far the hedge has progressed.
+  EXPECT_THAT(calls->load(), Eq(2));
+}
+
+TEST(HedgedObjectReadSourceTest, AllAttemptsFailReportsPrimaryError) {
+  // The hedge fails first with one error, the primary later with another.
+  // The stream the caller is reading is the primary, so its error is the one
+  // reported.
+  auto calls = std::make_shared<std::atomic<int>>(0);
+  HedgeSignal hedge_started;
+  auto factory =
+      [calls, hedge_started]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    if (++*calls != 1) {
+      hedge_started.Signal();
+      return Status(StatusCode::kNotFound, "hedge error");
+    }
+    auto mock = std::make_unique<MockObjectReadSource>();
+    EXPECT_CALL(*mock, Read).WillOnce([hedge_started](char*, std::size_t) {
+      // Outlive the hedge dispatch, otherwise the primary error is reported
+      // before there is a hedge error to lose to it.
+      hedge_started.Wait();
+      return StatusOr<ReadSourceResult>(
+          Status(StatusCode::kUnavailable, "retry policy exhausted"));
+    });
+    // The failed child is still open, e.g. a permanent HTTP error, so the
+    // race must close it.
+    EXPECT_CALL(*mock, IsOpen).WillRepeatedly(Return(true));
+    EXPECT_CALL(*mock, Close)
+        .WillOnce(
+            Return(make_status_or(HttpResponse{HttpStatusCode::kOk, {}, {}})));
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kDelay, /*max_hedges=*/1, kUnlimitedBuffer);
+
+  std::vector<char> buffer(100);
+  auto result = source.Read(buffer.data(), buffer.size());
+  EXPECT_THAT(result, StatusIs(StatusCode::kUnavailable));
+  EXPECT_THAT(calls->load(), Eq(2));
+  EXPECT_FALSE(source.IsOpen());
 }
 
 TEST(HedgedObjectReadSourceTest, CloseWithoutReadSucceeds) {
@@ -422,9 +555,8 @@ TEST(HedgedObjectReadSourceTest, CloseWithoutReadSucceeds) {
     return Status(StatusCode::kUnimplemented, "never called");
   };
   HedgedObjectReadSource source(MakeUnlimitedReadPool(),
-                                MakeUnlimitedHedgePool(), factory,
-                                std::chrono::milliseconds(500),
-                                /*max_hedges=*/2, kUnlimitedBuffer);
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kLongDelay, /*max_hedges=*/2, kUnlimitedBuffer);
   EXPECT_TRUE(source.IsOpen());
   EXPECT_THAT(source.Close(), IsOk());
 }
@@ -436,7 +568,7 @@ TEST(HedgedObjectReadSourceTest, CloseBeforeRead) {
     return std::unique_ptr<ObjectReadSource>(
         std::make_unique<MockObjectReadSource>());
   };
-  HedgedObjectReadSource source(read_pool, hedge_pool, factory,
+  HedgedObjectReadSource source(read_pool, hedge_pool, Adapt(factory),
                                 std::chrono::milliseconds(10), 2,
                                 kUnlimitedBuffer);
   EXPECT_TRUE(source.IsOpen());
@@ -454,18 +586,14 @@ TEST(HedgedObjectReadSourceTest, OversizedReadIsNotHedged) {
   auto factory = [calls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
     ++*calls;
     auto mock = std::make_unique<MockObjectReadSource>();
-    EXPECT_CALL(*mock, Read).WillOnce([](char* buf, std::size_t) {
-      std::string const payload = "direct";
-      std::copy(payload.begin(), payload.end(), buf);
-      return MakeReadResult(payload);
-    });
+    EXPECT_CALL(*mock, Read).WillOnce(ImmediateRead("direct"));
     return std::unique_ptr<ObjectReadSource>(std::move(mock));
   };
 
   // A zero delay would let a hedge start immediately if the limit were not
   // honored, so any race would be observable as extra factory calls.
   HedgedObjectReadSource source(MakeUnlimitedReadPool(),
-                                MakeUnlimitedHedgePool(), factory,
+                                MakeUnlimitedHedgePool(), Adapt(factory),
                                 std::chrono::milliseconds(0),
                                 /*max_hedges=*/2, /*max_buffer=*/8);
 
@@ -483,49 +611,578 @@ TEST(HedgedObjectReadSourceTest, OversizedReadPropagatesOpenError) {
   };
 
   HedgedObjectReadSource source(MakeUnlimitedReadPool(),
-                                MakeUnlimitedHedgePool(), factory,
+                                MakeUnlimitedHedgePool(), Adapt(factory),
                                 std::chrono::milliseconds(0),
                                 /*max_hedges=*/2, /*max_buffer=*/8);
 
   std::vector<char> buffer(64);
   EXPECT_THAT(source.Read(buffer.data(), buffer.size()),
               StatusIs(StatusCode::kPermissionDenied));
+  EXPECT_FALSE(source.IsOpen());
 }
 
-TEST(HedgedObjectReadSourceTest, SubsequentReadsIgnoreBufferLimit) {
-  // The limit only decides whether the *open* is hedged. Once a child exists,
-  // reads of any size continue on it without staging buffers.
+TEST(HedgedObjectReadSourceTest, OversizedReadOnStalledStreamIsNotHedged) {
+  // The buffer limit applies to every read, not only to the open: a stalled
+  // stream is not raced for a read larger than the limit either.
   auto calls = std::make_shared<std::atomic<int>>(0);
   auto factory = [calls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
     ++*calls;
     auto mock = std::make_unique<MockObjectReadSource>();
     EXPECT_CALL(*mock, Read)
-        .WillOnce([](char* buf, std::size_t) {
-          std::string const payload = "small";
-          std::copy(payload.begin(), payload.end(), buf);
-          return MakeReadResult(payload);
-        })
-        .WillOnce([](char* buf, std::size_t) {
-          std::string const payload = "large";
-          std::copy(payload.begin(), payload.end(), buf);
-          return MakeReadResult(payload);
-        });
+        .WillOnce(ImmediateRead("open"))
+        .WillOnce(DelayedRead("small", kStall))
+        .WillOnce(ImmediateRead("large"));
     return std::unique_ptr<ObjectReadSource>(std::move(mock));
   };
 
   HedgedObjectReadSource source(MakeUnlimitedReadPool(),
-                                MakeUnlimitedHedgePool(), factory,
-                                std::chrono::milliseconds(500),
-                                /*max_hedges=*/2, /*max_buffer=*/64);
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kDelay, /*max_hedges=*/2, /*max_buffer=*/64);
 
   std::vector<char> small(8);
   EXPECT_THAT(source.Read(small.data(), small.size()), IsOk());
-  // Well past the limit, but the winner is already open.
+  EXPECT_THAT(source.Read(small.data(), small.size()), IsOk());
+  // The previous read stalled, but this one is well past the limit.
   std::vector<char> large(4096);
   EXPECT_THAT(source.Read(large.data(), large.size()), IsOk());
   EXPECT_THAT(calls->load(), Eq(1));
 }
 
+TEST(HedgedObjectReadSourceTest, SubsequentReadHedgeWinsWhenPrimaryStalls) {
+  // Read 1 opens the stream, read 2 is slow and marks the stream as stalled,
+  // so read 3 is raced. The primary blocks on read 3 and the hedge, opened at
+  // the current offset, wins and serves the rest of the stream.
+  auto unblock_primary = std::make_shared<std::promise<void>>();
+  auto primary_closed = std::make_shared<std::promise<void>>();
+  auto recorded_offset = std::make_shared<std::atomic<std::int64_t>>(-1);
+  auto factory_calls = std::make_shared<std::atomic<int>>(0);
+
+  auto factory = [unblock_primary, primary_closed, recorded_offset,
+                  factory_calls](std::int64_t offset,
+                                 std::optional<std::int64_t>)
+      -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    auto mock = std::make_unique<MockObjectReadSource>();
+    if (++*factory_calls == 1) {
+      EXPECT_CALL(*mock, Read)
+          .WillOnce(ImmediateRead("chunk-1"))
+          .WillOnce(DelayedRead("chunk-2", kStall))
+          .WillOnce(BlockedRead(unblock_primary, "chunk-3-slow"));
+      EXPECT_CALL(*mock, Close).WillOnce(NotifyClose(primary_closed));
+    } else {
+      recorded_offset->store(offset);
+      EXPECT_CALL(*mock, Read)
+          .WillOnce(ImmediateRead("chunk-3-hedge"))
+          .WillOnce(ImmediateRead("chunk-4"));
+    }
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), factory, kDelay,
+                                /*max_hedges=*/1, kUnlimitedBuffer);
+
+  std::vector<char> buffer(100);
+  for (auto const* expected : {"chunk-1", "chunk-2"}) {
+    auto result = source.Read(buffer.data(), buffer.size());
+    ASSERT_THAT(result, IsOk());
+    EXPECT_THAT(std::string(buffer.data(), result->bytes_received),
+                Eq(expected));
+  }
+
+  auto r3 = source.Read(buffer.data(), buffer.size());
+  ASSERT_THAT(r3, IsOk());
+  EXPECT_THAT(std::string(buffer.data(), r3->bytes_received),
+              Eq("chunk-3-hedge"));
+  EXPECT_THAT(recorded_offset->load(), Eq(14));
+
+  unblock_primary->set_value();
+  primary_closed->get_future().get();
+
+  auto r4 = source.Read(buffer.data(), buffer.size());
+  ASSERT_THAT(r4, IsOk());
+  EXPECT_THAT(std::string(buffer.data(), r4->bytes_received), Eq("chunk-4"));
+  EXPECT_THAT(factory_calls->load(), Eq(2));
+}
+
+TEST(HedgedObjectReadSourceTest, StalledStreamReturnsToDirectReads) {
+  // Read 1 opens the stream, read 2 stalls, read 3 is therefore raced and
+  // the hedge wins. Read 4 (on the hedge) completes quickly, so read 5 is a
+  // direct read again: no further children are opened.
+  auto unblock_primary = std::make_shared<std::promise<void>>();
+  auto primary_closed = std::make_shared<std::promise<void>>();
+  auto factory_calls = std::make_shared<std::atomic<int>>(0);
+
+  auto factory =
+      [unblock_primary, primary_closed,
+       factory_calls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    auto mock = std::make_unique<MockObjectReadSource>();
+    if (++*factory_calls == 1) {
+      EXPECT_CALL(*mock, Read)
+          .WillOnce(ImmediateRead("chunk-1"))
+          .WillOnce(DelayedRead("chunk-2", kStall))
+          .WillOnce(BlockedRead(unblock_primary, "chunk-3-slow"));
+      EXPECT_CALL(*mock, Close).WillOnce(NotifyClose(primary_closed));
+    } else {
+      EXPECT_CALL(*mock, Read)
+          .WillOnce(ImmediateRead("chunk-3-hedge"))
+          .WillOnce(ImmediateRead("chunk-4"))
+          .WillOnce(ImmediateRead("chunk-5"));
+    }
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kDelay, /*max_hedges=*/1, kUnlimitedBuffer);
+
+  std::vector<char> buffer(100);
+  for (auto const* expected :
+       {"chunk-1", "chunk-2", "chunk-3-hedge", "chunk-4", "chunk-5"}) {
+    auto result = source.Read(buffer.data(), buffer.size());
+    ASSERT_THAT(result, IsOk());
+    EXPECT_THAT(std::string(buffer.data(), result->bytes_received),
+                Eq(expected));
+  }
+  EXPECT_THAT(factory_calls->load(), Eq(2));
+
+  unblock_primary->set_value();
+  primary_closed->get_future().get();
+}
+
+TEST(HedgedObjectReadSourceTest, SubsequentReadPinsGeneration) {
+  auto unblock_primary = std::make_shared<std::promise<void>>();
+  auto primary_closed = std::make_shared<std::promise<void>>();
+  auto recorded_gen = std::make_shared<std::atomic<std::int64_t>>(-1);
+  auto factory_calls = std::make_shared<std::atomic<int>>(0);
+
+  auto factory = [unblock_primary, primary_closed, recorded_gen, factory_calls](
+                     std::int64_t, std::optional<std::int64_t> generation)
+      -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    auto mock = std::make_unique<MockObjectReadSource>();
+    if (++*factory_calls == 1) {
+      EXPECT_CALL(*mock, Read)
+          .WillOnce([](char* buf, std::size_t) {
+            std::string const payload = "chunk-1";
+            std::copy(payload.begin(), payload.end(), buf);
+            auto r = MakeReadResult(payload);
+            r.generation = 987654321;
+            return r;
+          })
+          .WillOnce(DelayedRead("chunk-2", kStall))
+          .WillOnce(BlockedRead(unblock_primary, "chunk-3-slow"));
+      EXPECT_CALL(*mock, Close).WillOnce(NotifyClose(primary_closed));
+    } else {
+      if (generation) recorded_gen->store(*generation);
+      EXPECT_CALL(*mock, Read).WillOnce(ImmediateRead("chunk-3-hedge"));
+    }
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), factory, kDelay,
+                                /*max_hedges=*/1, kUnlimitedBuffer);
+
+  std::vector<char> buffer(100);
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  EXPECT_THAT(recorded_gen->load(), Eq(987654321));
+
+  unblock_primary->set_value();
+  primary_closed->get_future().get();
+}
+
+TEST(HedgedObjectReadSourceTest, SubsequentReadGunzippedBypassesHedging) {
+  // Read 1 discovers decompressive transcoding, read 2 stalls. Read 3 would
+  // be raced, but under transcoding a hedge cannot resume at an offset, so it
+  // must continue directly on the active child.
+  auto factory_calls = std::make_shared<std::atomic<int>>(0);
+  auto factory =
+      [factory_calls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    ++*factory_calls;
+    auto mock = std::make_unique<MockObjectReadSource>();
+    EXPECT_CALL(*mock, Read)
+        .WillOnce([](char* buf, std::size_t) {
+          std::string const payload = "chunk-1";
+          std::copy(payload.begin(), payload.end(), buf);
+          auto r = MakeReadResult(payload);
+          r.transformation = "gunzipped";
+          return r;
+        })
+        .WillOnce(DelayedRead("chunk-2", kStall))
+        .WillOnce(ImmediateRead("chunk-3"));
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kDelay, /*max_hedges=*/2, kUnlimitedBuffer);
+
+  std::vector<char> buffer(100);
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  auto r3 = source.Read(buffer.data(), buffer.size());
+  ASSERT_THAT(r3, IsOk());
+  EXPECT_THAT(std::string(buffer.data(), r3->bytes_received), Eq("chunk-3"));
+  EXPECT_THAT(factory_calls->load(), Eq(1));
+}
+
+TEST(HedgedObjectReadSourceTest,
+     SubsequentReadHedgeFailureDoesNotAbortPrimary) {
+  auto unblock_primary = std::make_shared<std::promise<void>>();
+  auto hedge_attempted = std::make_shared<std::promise<void>>();
+  auto hedge_signalled = std::make_shared<std::atomic<bool>>(false);
+  auto factory_calls = std::make_shared<std::atomic<int>>(0);
+
+  auto factory =
+      [unblock_primary, hedge_attempted, hedge_signalled,
+       factory_calls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    if (++*factory_calls != 1) {
+      // Tell the test the hedge has been dispatched. Guarded because setting a
+      // promise twice throws, and only the first hedge needs to be observed.
+      if (!hedge_signalled->exchange(true)) hedge_attempted->set_value();
+      return Status(StatusCode::kUnavailable, "hedge open error");
+    }
+    auto mock = std::make_unique<MockObjectReadSource>();
+    EXPECT_CALL(*mock, Read)
+        .WillOnce(ImmediateRead("chunk-1"))
+        .WillOnce(DelayedRead("chunk-2", kStall))
+        .WillOnce(BlockedRead(unblock_primary, "chunk-3-primary"));
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kDelay, /*max_hedges=*/1, kUnlimitedBuffer);
+
+  std::vector<char> buffer(100);
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+
+  // Release the primary only once the hedge has actually been attempted.
+  // Sleeping instead would race the hedge dispatch against an unrelated clock,
+  // and on a slow machine the primary could finish before the hedge is ever
+  // dispatched. The bounded wait makes a regression fail the assertions below
+  // rather than hang the test.
+  std::thread unblocker([unblock_primary, hedge_attempted] {
+    hedge_attempted->get_future().wait_for(std::chrono::seconds(10));
+    unblock_primary->set_value();
+  });
+
+  auto r3 = source.Read(buffer.data(), buffer.size());
+  unblocker.join();
+
+  ASSERT_THAT(r3, IsOk());
+  EXPECT_THAT(std::string(buffer.data(), r3->bytes_received),
+              Eq("chunk-3-primary"));
+  EXPECT_THAT(factory_calls->load(), Eq(2));
+}
+
+// Returns a factory whose first child answers @p result twice (the second time
+// after `kStall`, so the next read is raced) and then blocks, and whose second
+// child records the offset it was opened at and answers "hedge".
+auto MakeOffsetRecordingFactory(
+    ReadSourceResult result,
+    std::shared_ptr<std::promise<void>> const& unblock_primary,
+    std::shared_ptr<std::promise<void>> const& primary_closed,
+    std::shared_ptr<std::atomic<std::int64_t>> const& recorded_offset,
+    std::shared_ptr<std::atomic<int>> const& factory_calls) {
+  return [result = std::move(result), unblock_primary, primary_closed,
+          recorded_offset,
+          factory_calls](std::int64_t offset, std::optional<std::int64_t>)
+             -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    auto mock = std::make_unique<MockObjectReadSource>();
+    if (++*factory_calls == 1) {
+      EXPECT_CALL(*mock, Read)
+          .WillOnce([result](char* buf, std::size_t) {
+            std::fill(buf, buf + result.bytes_received, 'x');
+            return result;
+          })
+          .WillOnce([result](char* buf, std::size_t) {
+            std::this_thread::sleep_for(kStall);
+            std::fill(buf, buf + result.bytes_received, 'x');
+            return result;
+          })
+          .WillOnce(BlockedRead(unblock_primary, "slow"));
+      EXPECT_CALL(*mock, Close).WillOnce(NotifyClose(primary_closed));
+    } else {
+      recorded_offset->store(offset);
+      EXPECT_CALL(*mock, Read).WillOnce(ImmediateRead("hedge"));
+    }
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+}
+
+TEST(HedgedObjectReadSourceTest, SubsequentReadFromEndTracksOffset) {
+  auto unblock_primary = std::make_shared<std::promise<void>>();
+  auto primary_closed = std::make_shared<std::promise<void>>();
+  auto recorded_offset = std::make_shared<std::atomic<std::int64_t>>(-1);
+  auto factory_calls = std::make_shared<std::atomic<int>>(0);
+  auto factory = MakeOffsetRecordingFactory(MakeReadResult("1234567890"),
+                                            unblock_primary, primary_closed,
+                                            recorded_offset, factory_calls);
+
+  HedgedObjectReadSource::Position position;
+  position.offset = 100;
+  position.direction = kFromEnd;
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), factory, kDelay,
+                                /*max_hedges=*/1, kUnlimitedBuffer, position);
+
+  std::vector<char> buffer(100);
+  auto r1 = source.Read(buffer.data(), buffer.size());
+  ASSERT_THAT(r1, IsOk());
+  EXPECT_THAT(r1->bytes_received, Eq(10));
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+
+  auto r3 = source.Read(buffer.data(), buffer.size());
+  ASSERT_THAT(r3, IsOk());
+  EXPECT_THAT(recorded_offset->load(), Eq(80));
+
+  unblock_primary->set_value();
+  primary_closed->get_future().get();
+}
+
+TEST(HedgedObjectReadSourceTest, ReadLastLargerThanObjectClampsOffset) {
+  // `ReadLast(100)` on a 40 byte object returns the whole object. After 20
+  // bytes, 20 remain: a hedge asking for the last 80 bytes would receive the
+  // whole object again, from the first byte.
+  auto unblock_primary = std::make_shared<std::promise<void>>();
+  auto primary_closed = std::make_shared<std::promise<void>>();
+  auto recorded_offset = std::make_shared<std::atomic<std::int64_t>>(-1);
+  auto factory_calls = std::make_shared<std::atomic<int>>(0);
+  auto first = MakeReadResult("1234567890");
+  first.size = 40;
+  auto factory = MakeOffsetRecordingFactory(
+      first, unblock_primary, primary_closed, recorded_offset, factory_calls);
+
+  HedgedObjectReadSource::Position position;
+  position.offset = 100;
+  position.direction = kFromEnd;
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), factory, kDelay,
+                                /*max_hedges=*/1, kUnlimitedBuffer, position);
+
+  std::vector<char> buffer(100);
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  EXPECT_THAT(recorded_offset->load(), Eq(20));
+
+  unblock_primary->set_value();
+  primary_closed->get_future().get();
+}
+
+// Verifies that a stalled stream is *not* raced once it has reached the end
+// of the requested data: the drain read at the end must go to the active
+// child, a hedge would request an empty or inverted range. The child answers
+// @p chunk twice, reaching the end of the data with a stalled read, then
+// answers the (equally slow) drain read with no data.
+void ExpectNoRaceAtEnd(HedgedObjectReadSource::Position position,
+                       ReadSourceResult chunk) {
+  auto factory_calls = std::make_shared<std::atomic<int>>(0);
+  auto factory = [factory_calls, chunk = std::move(chunk)]()
+      -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    ++*factory_calls;
+    auto mock = std::make_unique<MockObjectReadSource>();
+    EXPECT_CALL(*mock, Read)
+        .WillOnce([chunk](char* buf, std::size_t) {
+          std::fill(buf, buf + chunk.bytes_received, 'x');
+          return chunk;
+        })
+        .WillOnce([chunk](char* buf, std::size_t) {
+          std::this_thread::sleep_for(kStall);
+          std::fill(buf, buf + chunk.bytes_received, 'x');
+          return chunk;
+        })
+        .WillOnce(DelayedRead("", kStall));
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  HedgedObjectReadSource source(
+      MakeUnlimitedReadPool(), MakeUnlimitedHedgePool(), Adapt(factory), kDelay,
+      /*max_hedges=*/2, kUnlimitedBuffer, position);
+
+  std::vector<char> buffer(100);
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  auto drain = source.Read(buffer.data(), buffer.size());
+  ASSERT_THAT(drain, IsOk());
+  EXPECT_THAT(drain->bytes_received, Eq(0));
+  EXPECT_THAT(factory_calls->load(), Eq(1));
+}
+
+TEST(HedgedObjectReadSourceTest, NoRaceAtObjectSize) {
+  auto chunk = MakeReadResult("12345");
+  chunk.size = 10;
+  ExpectNoRaceAtEnd(HedgedObjectReadSource::Position{}, chunk);
+}
+
+TEST(HedgedObjectReadSourceTest, NoRaceAtRangeEnd) {
+  HedgedObjectReadSource::Position position;
+  position.offset = 5;
+  position.end_offset = 15;
+  auto chunk = MakeReadResult("12345");
+  chunk.size = 1000;
+  ExpectNoRaceAtEnd(position, chunk);
+}
+
+TEST(HedgedObjectReadSourceTest, NoRaceAtReadLastEnd) {
+  HedgedObjectReadSource::Position position;
+  position.offset = 10;
+  position.direction = kFromEnd;
+  ExpectNoRaceAtEnd(position, MakeReadResult("12345"));
+}
+
+TEST(HedgedObjectReadSourceTest, RangeEndTakesPrecedenceOverResponseSize) {
+  // On the REST path `ReadSourceResult::size` falls back to `content-length`,
+  // which is the length of that response rather than the size of the object.
+  // For a ranged read that value is far below the stream's offset, and must
+  // not be mistaken for the end of the requested data: the range end wins.
+  HedgedObjectReadSource::Position position;
+  position.offset = 1000;
+  position.end_offset = 1100;
+
+  auto unblock_primary = std::make_shared<std::promise<void>>();
+  auto primary_closed = std::make_shared<std::promise<void>>();
+  auto factory_calls = std::make_shared<std::atomic<int>>(0);
+  auto factory =
+      [unblock_primary, primary_closed,
+       factory_calls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    auto mock = std::make_unique<MockObjectReadSource>();
+    if (++*factory_calls != 1) {
+      EXPECT_CALL(*mock, Read).WillOnce(ImmediateRead("chunk-3-hedge"));
+      return std::unique_ptr<ObjectReadSource>(std::move(mock));
+    }
+    // `size` is this response's content-length, well below `position.offset`.
+    ReadSourceResult chunk = MakeReadResult("chunk-1");
+    chunk.size = 7;
+    EXPECT_CALL(*mock, Read)
+        .WillOnce([chunk](char* buf, std::size_t n) {
+          EXPECT_LE(chunk.bytes_received, n);
+          std::fill(buf, buf + chunk.bytes_received, 'x');
+          return chunk;
+        })
+        .WillOnce(DelayedRead("chunk-2", kStall))
+        .WillOnce(BlockedRead(unblock_primary, "chunk-3-slow"));
+    EXPECT_CALL(*mock, Close).WillOnce(NotifyClose(primary_closed));
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  HedgedObjectReadSource source(
+      MakeUnlimitedReadPool(), MakeUnlimitedHedgePool(), Adapt(factory), kDelay,
+      /*max_hedges=*/1, kUnlimitedBuffer, position);
+
+  std::vector<char> buffer(100);
+  // Read 1 opens the stream, read 2 is slow and marks it stalled.
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+
+  // Read 3 must still be raced. The stream is 1014 bytes in and the range ends
+  // at 1100, so it has not reached the end of the requested data, even though
+  // the reported `size` of 7 is long behind it.
+  auto r3 = source.Read(buffer.data(), buffer.size());
+  ASSERT_THAT(r3, IsOk());
+  EXPECT_THAT(std::string(buffer.data(), r3->bytes_received),
+              Eq("chunk-3-hedge"));
+  EXPECT_THAT(factory_calls->load(), Eq(2));
+
+  unblock_primary->set_value();
+  WaitForSignal(primary_closed);
+}
+
+TEST(HedgedObjectReadSourceTest, HedgesAreBoundedPerStream) {
+  // Every read on this stream is slow enough to look stalled, so without a
+  // per-stream budget the source would race, and hedge, forever. The hedges
+  // all fail so the primary always wins and stays the active child.
+  auto constexpr kShortDelay = std::chrono::milliseconds(5);
+  auto constexpr kSlowRead = std::chrono::milliseconds(20);
+  int constexpr kMaxHedges = 1;
+  int constexpr kReads = 20;
+  // One call opens the primary, the rest are hedges. `ShouldRace()` stops
+  // racing once the stream has spent `max_hedges * kMaxHedgeRoundsPerStream`.
+  int constexpr kExpectedCalls = 1 + kMaxHedges * 8;
+
+  auto mu = std::make_shared<std::mutex>();
+  auto cv = std::make_shared<std::condition_variable>();
+  auto factory_calls = std::make_shared<int>(0);
+  auto raced_rounds = std::make_shared<int>(0);
+  auto factory =
+      [mu, cv, factory_calls, raced_rounds, kSlowRead,
+       expected_calls =
+           kExpectedCalls]() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    int calls = 0;
+    {
+      std::lock_guard<std::mutex> lock(*mu);
+      calls = ++*factory_calls;
+    }
+    cv->notify_all();
+    if (calls != 1) return Status(StatusCode::kUnavailable, "hedge");
+    auto mock = std::make_unique<MockObjectReadSource>();
+    EXPECT_CALL(*mock, Read)
+        .WillRepeatedly([mu, cv, factory_calls, raced_rounds, kSlowRead,
+                         expected_calls](char* buf, std::size_t n) {
+          // While the per-stream hedge budget remains, each raced primary read
+          // waits until its hedge has actually entered `factory` before
+          // completing, so a slow CI scheduler cannot let the primary finish
+          // ahead of `future.wait_for(kShortDelay)`.
+          int const target_calls = 1 + ++*raced_rounds;
+          if (target_calls <= expected_calls) {
+            std::unique_lock<std::mutex> lock(*mu);
+            EXPECT_TRUE(cv->wait_for(lock, std::chrono::seconds(10), [&] {
+              return *factory_calls >= target_calls;
+            }));
+          }
+          return DelayedRead("chunk", kSlowRead)(buf, n);
+        });
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kShortDelay, kMaxHedges, kUnlimitedBuffer);
+
+  std::vector<char> buffer(100);
+  for (int i = 0; i != kReads; ++i) {
+    ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk()) << "i=" << i;
+  }
+
+  std::lock_guard<std::mutex> lock(*mu);
+  EXPECT_THAT(*factory_calls, Eq(kExpectedCalls));
+}
+
+TEST(HedgedObjectReadSourceTest, DirectReadFailureClosesStream) {
+  // A direct read that fails must tear the stream down the same way a raced
+  // read does, otherwise `Close()` would reach a child that has already
+  // released its connection.
+  auto factory = []() -> StatusOr<std::unique_ptr<ObjectReadSource>> {
+    auto mock = std::make_unique<MockObjectReadSource>();
+    EXPECT_CALL(*mock, Read)
+        .WillOnce(ImmediateRead("payload"))
+        .WillOnce(Return(StatusOr<ReadSourceResult>(
+            Status(StatusCode::kUnavailable, "retry policy exhausted"))));
+    EXPECT_CALL(*mock, IsOpen).WillRepeatedly(Return(true));
+    EXPECT_CALL(*mock, Close)
+        .WillOnce(
+            Return(make_status_or(HttpResponse{HttpStatusCode::kOk, {}, {}})));
+    return std::unique_ptr<ObjectReadSource>(std::move(mock));
+  };
+
+  // `kLongDelay` keeps the open race from dispatching a hedge, and the fast
+  // first read leaves the stream looking healthy, so the second read is
+  // direct.
+  HedgedObjectReadSource source(MakeUnlimitedReadPool(),
+                                MakeUnlimitedHedgePool(), Adapt(factory),
+                                kLongDelay, /*max_hedges=*/1, kUnlimitedBuffer);
+
+  std::vector<char> buffer(100);
+  ASSERT_THAT(source.Read(buffer.data(), buffer.size()), IsOk());
+  EXPECT_THAT(source.Read(buffer.data(), buffer.size()),
+              StatusIs(StatusCode::kUnavailable));
+  EXPECT_FALSE(source.IsOpen());
+  EXPECT_THAT(source.Close(), IsOk());
+}
+
 }  // namespace
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
diff --git a/google/cloud/storage/internal/retry_object_read_source.cc b/google/cloud/storage/internal/retry_object_read_source.cc
index 0df66aa82e0c4..ba9af0db93f06 100644
--- a/google/cloud/storage/internal/retry_object_read_source.cc
+++ b/google/cloud/storage/internal/retry_object_read_source.cc
@@ -143,8 +143,14 @@ bool RetryObjectReadSource::HandleResult(StatusOr<ReadSourceResult> const& r) {
   if (r->generation) generation_ = r->generation;
   if (r->transformation.value_or("") == "gunzipped") is_gunzipped_ = true;
   // Since decompressive transcoding does not respect `ReadLast()` we need
-  // to ensure the offset is incremented, so the discard loop works.
-  if (is_gunzipped_) offset_direction_ = kFromBeginning;
+  // to ensure the offset is incremented, so the discard loop works. The
+  // response is the whole object from the first byte, so the bytes consumed
+  // so far are counted from the start of the object and the offset restarts
+  // at zero. Resuming from the un-reset `ReadLast()` count would skip data.
+  if (is_gunzipped_ && offset_direction_ == kFromEnd) {
+    offset_direction_ = kFromBeginning;
+    current_offset_ = 0;
+  }
   if (offset_direction_ == kFromEnd) {
     current_offset_ -= r->bytes_received;
   } else {
diff --git a/google/cloud/storage/internal/retry_object_read_source.h b/google/cloud/storage/internal/retry_object_read_source.h
index 086c4c9eaf533..db7e0460ba20d 100644
--- a/google/cloud/storage/internal/retry_object_read_source.h
+++ b/google/cloud/storage/internal/retry_object_read_source.h
@@ -62,7 +62,14 @@ class RetryObjectReadSource : public ObjectReadSource {
                         std::unique_ptr<BackoffPolicy> backoff_policy);
 
   bool IsOpen() const override { return child_ && child_->IsOpen(); }
-  StatusOr<HttpResponse> Close() override { return child_->Close(); }
+  StatusOr<HttpResponse> Close() override {
+    // `Read()` releases the child when it retries, and leaves it released if
+    // the retry policy is then exhausted. There is no connection left to
+    // close, and callers (e.g. `ObjectReadStreambuf::Close()`) do not check
+    // `IsOpen()` first.
+    if (!child_) return HttpResponse{HttpStatusCode::kOk, {}, {}};
+    return child_->Close();
+  }
   StatusOr<ReadSourceResult> Read(char* buf, std::size_t n) override;
 
  private:
diff --git a/google/cloud/storage/internal/retry_object_read_source_test.cc b/google/cloud/storage/internal/retry_object_read_source_test.cc
index f264e3e743539..f27f951109f2b 100644
--- a/google/cloud/storage/internal/retry_object_read_source_test.cc
+++ b/google/cloud/storage/internal/retry_object_read_source_test.cc
@@ -489,6 +489,84 @@ TEST(RetryObjectReadSourceTest, DiscardDataForDecompressiveTranscoding) {
               Contains(Pair("x-test-only", "download 3 r1")));
 }
 
+/// @test `ReadLast` downloads subject to decompressive transcoding restart
+///     counting from the first byte, not from the tail.
+TEST(RetryObjectReadSourceTest, ReadLastWithDecompressiveTranscoding) {
+  auto mock = std::make_unique<MockGenericStub>();
+  EXPECT_CALL(*mock, options);  // Required in RetryClient::Create()
+  EXPECT_CALL(*mock, ReadObject)
+      .WillOnce([](auto&, auto const&, ReadObjectRangeRequest const& req) {
+        EXPECT_EQ(1029, req.GetOption<ReadLast>().value());
+        // The response reveals the object is served with decompressive
+        // transcoding, which ignores the range and returns the whole object
+        // from the first byte.
+        auto r0 = ReadSourceResult{static_cast<std::size_t>(1024),
+                                   HttpResponse{100, "", {}}};
+        r0.transformation = "gunzipped";
+        auto source = std::make_unique<MockObjectReadSource>();
+        EXPECT_CALL(*source, Read)
+            .WillOnce(Return(r0))
+            .WillOnce(Return(TransientError()));
+        return std::unique_ptr<ObjectReadSource>(std::move(source));
+      })
+      .WillOnce([](auto&, auto const&, ReadObjectRangeRequest const& req) {
+        // 1024 bytes were consumed, counted from the start of the object. The
+        // resumed download must discard exactly those, not the un-reset
+        // `ReadLast()` count plus them (1029 + 1024), which would skip data.
+        EXPECT_EQ(1024, req.GetOption<ReadFromOffset>().value_or(0));
+        auto discard = ReadSourceResult{static_cast<std::size_t>(1024),
+                                        HttpResponse{100, "", {}}};
+        discard.transformation = "gunzipped";
+        auto payload = ReadSourceResult{static_cast<std::size_t>(64),
+                                        HttpResponse{200, "", {}}};
+        auto source = std::make_unique<MockObjectReadSource>();
+        ::testing::InSequence sequence;
+        EXPECT_CALL(*source, Read(_, 1024L)).WillOnce(Return(discard));
+        EXPECT_CALL(*source, Read(_, 2048L)).WillOnce(Return(payload));
+        return std::unique_ptr<ObjectReadSource>(std::move(source));
+      });
+
+  auto client = StorageConnectionImpl::Create(std::move(mock));
+  google::cloud::internal::OptionsSpan const span(BasicTestPolicies());
+
+  ReadObjectRangeRequest req("test_bucket", "test_object");
+  req.set_option(ReadLast(1029));
+  auto source = client->ReadObject(req);
+  ASSERT_STATUS_OK(source);
+  ASSERT_STATUS_OK((*source)->Read(nullptr, 1024));
+  auto response = (*source)->Read(nullptr, 2048);
+  ASSERT_STATUS_OK(response);
+  EXPECT_EQ(response->bytes_received, 64);
+}
+
+/// @test Closing a source whose retry policy was exhausted is safe.
+TEST(RetryObjectReadSourceTest, CloseAfterRetryPolicyExhausted) {
+  auto mock = std::make_unique<MockGenericStub>();
+  EXPECT_CALL(*mock, options);  // Required in RetryClient::Create()
+  EXPECT_CALL(*mock, ReadObject)
+      .WillOnce([] {
+        auto source = std::make_unique<MockObjectReadSource>();
+        EXPECT_CALL(*source, Read).WillOnce(Return(TransientError()));
+        return std::unique_ptr<ObjectReadSource>(std::move(source));
+      })
+      // Resuming the download fails until the retry policy is exhausted, so
+      // the source is left without a child.
+      .WillRepeatedly([] { return TransientError(); });
+
+  auto client = StorageConnectionImpl::Create(std::move(mock));
+  google::cloud::internal::OptionsSpan const span(BasicTestPolicies());
+
+  auto source = client->ReadObject(ReadObjectRangeRequest{});
+  ASSERT_STATUS_OK(source);
+  EXPECT_THAT((*source)->Read(nullptr, 1024),
+              StatusIs(TransientError().code()));
+  EXPECT_FALSE((*source)->IsOpen());
+  // Callers such as `ObjectReadStreambuf::Close()` close the stream without
+  // checking `IsOpen()` first, so this must not dereference the released
+  // child.
+  EXPECT_STATUS_OK((*source)->Close());
+}
+
 using ::google::cloud::testing_util::DisableTracing;
 using ::google::cloud::testing_util::EnableTracing;
 using ::google::cloud::testing_util::SpanNamed;
diff --git a/google/cloud/storage/options.h b/google/cloud/storage/options.h
index 696f1abad3149..6d7bbc81a9afb 100644
--- a/google/cloud/storage/options.h
+++ b/google/cloud/storage/options.h
@@ -39,7 +39,10 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
  *
  * When enabled, opening a download races the initial request against one or
  * more delayed, duplicate ("hedged") requests, and the first to respond wins.
- * This reduces tail latency at the cost of additional requests.
+ * A read that stalls mid-stream, that is, one that takes longer than
+ * `ReadHedgeDelayOption`, is raced the same way against duplicate requests
+ * resuming from the stream's current offset. This reduces tail latency at the
+ * cost of additional requests.
  *
  * @ingroup storage-options
  */
@@ -74,11 +77,15 @@ struct MaxConcurrentHedgesOption {
  * The largest read, in bytes, that is eligible for hedging.
  *
  * Racing requests each buffer their own copy of the data, so the memory used
- * while opening a stream grows with the size of the first read. A read larger
- * than this value is served without hedging, reading directly into the
- * application's buffer, which bounds that growth. Note this is the size the
- * application asks for in a single read (e.g. `stream.read(buf, n)`), not the
- * size of the object or of a requested range.
+ * by a raced read grows with the size of that read. A read larger than this
+ * value is served without hedging, reading directly into the application's
+ * buffer, which bounds that growth. Note this is the size the application asks
+ * for in a single read (e.g. `stream.read(buf, n)`), not the size of the
+ * object or of a requested range.
+ *
+ * This bound applies to every raced read, both the stream open and any
+ * stalled read that is raced later, so a stream can hold up to
+ * `MaxReadHedgesOption` + 1 buffers of this size at once.
  *
  * The default is 64 MiB (64 * 1024 * 1024).
  *
@@ -91,6 +98,9 @@ struct MaximumHedgeBufferOption {
 /**
  * The delay before starting a hedged request.
  *
+ * This is also the mid-stream stall threshold: a read that takes longer than
+ * this marks the stream as stalled, and the next read is raced.
+ *
  * The default is 500 milliseconds.
  *
  * @ingroup storage-options
@@ -100,7 +110,12 @@ struct ReadHedgeDelayOption {
 };
 
 /**
- * The maximum number of hedged requests per stream open.
+ * The maximum number of hedged requests per raced read.
+ *
+ * A read is raced when the stream is opened, and again whenever a read stalls
+ * mid-stream, so this bounds a single race rather than the whole stream. The
+ * total number of hedges a stream may issue is bounded at a small multiple of
+ * this value, so a uniformly slow stream cannot keep racing indefinitely.
  *
  * The default is 2. Set to 0 to disable hedging for reads even when
  * `EnableReadHedgingOption` is set.