diff --git a/CHANGELOG.md b/CHANGELOG.md
index 59296ca090a33..b40aeb51e06a0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -90,6 +90,50 @@
 
 ## v1.36.0 - TBD
 
+### [Storage](https://github.com/googleapis/google-cloud-cpp/blob/main/google/cloud/storage/README.md)
+
+**BREAKING CHANGE:** with this release any use of the
+`storage::internal::ResumableUploadResponse` type require changes. Applications
+should have little need for this type, outside mocks, so the changes should not
+affect production code.
+
+Nevertheless, we apologize for the inconvenience, and while we would have
+preferred to avoid breaking changes, it was inevitable to introduce some
+breaking changes to fix a data loss bug.
+
+If you are affected by this change, you will need to change your tests following
+this guide. Any place where you return a `ResumableUploadResponse` needs to
+change from:
+
+```cc
+storage::internal::ResumableUploadResource{
+  /*upload_session_url=*/std::string{"typically-unused"},
+  /*last_committed_byte=*/std::uint64_t{value},
+  /*payload=*/absl::nullopt, // or some gcs::ObjectMetadata value
+  /*upload_state=*/ResumableUploadResponse::kInProgress, // or kDone
+  /*annotations=*/std::string{"typically-unused"}
+}
+```
+
+to:
+
+```cc
+storage::internal::ResumableUploadResource{
+  /*upload_session_url=*/std::string{"typically-unused"},
+  /*upload_state=*/ResumableUploadResponse::kInProgress, // or kDone
+  /*committed_size=*/value + 1, // or absl::nullopt
+  /*payload=*/absl::nullopt, // or some gcs::ObjectMetadata value
+  /*annotations=*/std::string{"typically-unused"}
+}
+```
+
+That is, you need to re-order the fields **and** increase the `value` to reflect
+the number of committed bytes vs. the index in the last committed byte.
+
+Changing the order of the fields was intentional. It will create a build
+failure, which is easier to detect and repair than a run-time error in
+your tests.
+
 ## v1.35.0 - 2022-01
 
 ### New GA Libraries
diff --git a/google/cloud/storage/client_write_object_test.cc b/google/cloud/storage/client_write_object_test.cc
index 0883417010f98..9b5203fdc0a62 100644
--- a/google/cloud/storage/client_write_object_test.cc
+++ b/google/cloud/storage/client_write_object_test.cc
@@ -62,15 +62,15 @@ TEST_F(WriteObjectTest, WriteObject) {
         EXPECT_CALL(*mock, next_expected_byte()).WillRepeatedly(Return(0));
         EXPECT_CALL(*mock, UploadChunk)
             .WillRepeatedly(Return(make_status_or(ResumableUploadResponse{
-                "fake-url", 0, {}, ResumableUploadResponse::kInProgress, {}})));
+                "fake-url", ResumableUploadResponse::kInProgress, 0, {}, {}})));
         EXPECT_CALL(*mock, ResetSession())
             .WillOnce(Return(make_status_or(ResumableUploadResponse{
-                "fake-url", 0, {}, ResumableUploadResponse::kInProgress, {}})));
+                "fake-url", ResumableUploadResponse::kInProgress, 0, {}, {}})));
         EXPECT_CALL(*mock, UploadFinalChunk)
             .WillOnce(
                 Return(StatusOr<ResumableUploadResponse>(TransientError())))
             .WillOnce(Return(make_status_or(ResumableUploadResponse{
-                "fake-url", 0, expected, ResumableUploadResponse::kDone, {}})));
+                "fake-url", ResumableUploadResponse::kDone, 0, expected, {}})));
 
         return make_status_or(
             std::unique_ptr<internal::ResumableUploadSession>(std::move(mock)));
@@ -227,7 +227,7 @@ TEST_F(WriteObjectTest, UploadStreamResumable) {
               bytes_written += internal::TotalBytes(data);
               EXPECT_EQ(bytes_written, size);
               return make_status_or(ResumableUploadResponse{
-                  "fake-url", 0, expected, ResumableUploadResponse::kDone, {}});
+                  "fake-url", ResumableUploadResponse::kDone, 0, expected, {}});
             });
 
         return make_status_or(
@@ -285,7 +285,7 @@ TEST_F(WriteObjectTest, UploadFile) {
               bytes_written += internal::TotalBytes(data);
               EXPECT_EQ(bytes_written, size);
               return make_status_or(ResumableUploadResponse{
-                  "fake-url", 0, expected, ResumableUploadResponse::kDone, {}});
+                  "fake-url", ResumableUploadResponse::kDone, 0, expected, {}});
             });
 
         return make_status_or(
diff --git a/google/cloud/storage/examples/storage_client_mock_samples.cc b/google/cloud/storage/examples/storage_client_mock_samples.cc
index a475fc1eb75ec..6ae851221040a 100644
--- a/google/cloud/storage/examples/storage_client_mock_samples.cc
+++ b/google/cloud/storage/examples/storage_client_mock_samples.cc
@@ -89,16 +89,16 @@ TEST(StorageMockingSamples, MockWriteObject) {
         EXPECT_CALL(*mock_result, UploadChunk)
             .WillRepeatedly(Return(google::cloud::make_status_or(
                 ResumableUploadResponse{"fake-url",
-                                        0,
-                                        {},
                                         ResumableUploadResponse::kInProgress,
+                                        /*committed_size=*/absl::nullopt,
+                                        /*object_metadata=*/absl::nullopt,
                                         {}})));
         EXPECT_CALL(*mock_result, UploadFinalChunk)
             .WillRepeatedly(Return(google::cloud::make_status_or(
                 ResumableUploadResponse{"fake-url",
-                                        0,
-                                        expected_metadata,
                                         ResumableUploadResponse::kDone,
+                                        /*committed_size=*/absl::nullopt,
+                                        /*object_metadata=*/expected_metadata,
                                         {}})));
 
         std::unique_ptr<gcs::internal::ResumableUploadSession> result(
diff --git a/google/cloud/storage/internal/curl_resumable_upload_session.cc b/google/cloud/storage/internal/curl_resumable_upload_session.cc
index 68dd3617bed9e..d766e9ddf0651 100644
--- a/google/cloud/storage/internal/curl_resumable_upload_session.cc
+++ b/google/cloud/storage/internal/curl_resumable_upload_session.cc
@@ -73,11 +73,9 @@ void CurlResumableUploadSession::Update(
     // value. In this case we update the next expected byte using the chunk
     // size, as we know the upload was successful.
     next_expected_ += chunk_size;
-  } else if (result->last_committed_byte != 0) {
-    next_expected_ = result->last_committed_byte + 1;
   } else {
     // Nothing has been committed on the server side yet, keep resending.
-    next_expected_ = 0;
+    next_expected_ = result->committed_size.value_or(0);
   }
   if (session_id_.empty() && !result->upload_session_url.empty()) {
     session_id_ = result->upload_session_url;
diff --git a/google/cloud/storage/internal/curl_resumable_upload_session_test.cc b/google/cloud/storage/internal/curl_resumable_upload_session_test.cc
index 1442bf79c441b..01d3b5e8cd976 100644
--- a/google/cloud/storage/internal/curl_resumable_upload_session_test.cc
+++ b/google/cloud/storage/internal/curl_resumable_upload_session_test.cc
@@ -55,7 +55,7 @@ MATCHER_P(MatchesPayload, value, "Checks whether payload matches a value") {
 
 TEST(CurlResumableUploadSessionTest, Simple) {
   auto mock = MockCurlClient::Create();
-  std::string test_url = "http://invalid.example.com/not-used-in-mock";
+  std::string test_url = "https://invalid.example.com/not-used-in-mock";
   ResumableUploadRequest request("test-bucket", "test-object");
   request.set_multiple_options(CustomHeader("x-test-custom", "custom-value"),
                                Fields("fields"), QuotaUser("quota-user"),
@@ -83,7 +83,7 @@ TEST(CurlResumableUploadSessionTest, Simple) {
         EXPECT_EQ(0, request.source_size());
         EXPECT_EQ(0, request.range_begin());
         return make_status_or(ResumableUploadResponse{
-            "", size - 1, {}, ResumableUploadResponse::kInProgress, {}});
+            "", ResumableUploadResponse::kInProgress, size, {}, {}});
       })
       .WillOnce([&](UploadChunkRequest const& request) {
         EXPECT_EQ(test_url, request.upload_session_url());
@@ -91,26 +91,26 @@ TEST(CurlResumableUploadSessionTest, Simple) {
         EXPECT_EQ(2 * size, request.source_size());
         EXPECT_EQ(size, request.range_begin());
         return make_status_or(ResumableUploadResponse{
-            "", 2 * size - 1, {}, ResumableUploadResponse::kDone, {}});
+            "", ResumableUploadResponse::kDone, 2 * size, {}, {}});
       });
 
   auto upload = session.UploadChunk({{payload}});
   EXPECT_STATUS_OK(upload);
-  EXPECT_EQ(size - 1, upload->last_committed_byte);
+  EXPECT_EQ(size, upload->committed_size.value_or(0));
   EXPECT_EQ(size, session.next_expected_byte());
   EXPECT_FALSE(session.done());
 
   upload = session.UploadFinalChunk({{payload}}, 2 * size, {});
   EXPECT_STATUS_OK(upload);
-  EXPECT_EQ(2 * size - 1, upload->last_committed_byte);
+  EXPECT_EQ(2 * size, upload->committed_size.value_or(0));
   EXPECT_EQ(2 * size, session.next_expected_byte());
   EXPECT_TRUE(session.done());
 }
 
 TEST(CurlResumableUploadSessionTest, Reset) {
   auto mock = MockCurlClient::Create();
-  std::string url1 = "http://invalid.example.com/not-used-in-mock-1";
-  std::string url2 = "http://invalid.example.com/not-used-in-mock-2";
+  std::string url1 = "https://invalid.example.com/not-used-in-mock-1";
+  std::string url2 = "https://invalid.example.com/not-used-in-mock-2";
   ResumableUploadRequest request("test-bucket", "test-object");
   request.set_multiple_options(CustomHeader("x-test-custom", "custom-value"),
                                Fields("fields"), QuotaUser("quota-user"),
@@ -125,14 +125,14 @@ TEST(CurlResumableUploadSessionTest, Reset) {
   EXPECT_CALL(*mock, UploadChunk)
       .WillOnce([&](UploadChunkRequest const&) {
         return make_status_or(ResumableUploadResponse{
-            "", size - 1, {}, ResumableUploadResponse::kInProgress, {}});
+            "", ResumableUploadResponse::kInProgress, size, {}, {}});
       })
       .WillOnce([&](UploadChunkRequest const&) {
         return StatusOr<ResumableUploadResponse>(
             AsStatus(HttpResponse{308, "uh oh", {}}));
       });
   const ResumableUploadResponse resume_response{
-      url2, 2 * size - 1, {}, ResumableUploadResponse::kInProgress, {}};
+      url2, ResumableUploadResponse::kInProgress, 2 * size, {}, {}};
   EXPECT_CALL(*mock, QueryResumableUpload)
       .WillOnce([&](QueryResumableUploadRequest const& request) {
         EXPECT_EQ(request.GetOption<CustomHeader>().value_or(""),
@@ -167,8 +167,8 @@ TEST(CurlResumableUploadSessionTest, Reset) {
 
 TEST(CurlResumableUploadSessionTest, SessionUpdatedInChunkUpload) {
   auto mock = MockCurlClient::Create();
-  std::string url1 = "http://invalid.example.com/not-used-in-mock-1";
-  std::string url2 = "http://invalid.example.com/not-used-in-mock-2";
+  std::string url1 = "https://invalid.example.com/not-used-in-mock-1";
+  std::string url2 = "https://invalid.example.com/not-used-in-mock-2";
   ResumableUploadRequest request("test-bucket", "test-object");
   CurlResumableUploadSession session(mock, request, url1);
 
@@ -179,11 +179,11 @@ TEST(CurlResumableUploadSessionTest, SessionUpdatedInChunkUpload) {
   EXPECT_CALL(*mock, UploadChunk)
       .WillOnce([&](UploadChunkRequest const&) {
         return make_status_or(ResumableUploadResponse{
-            "", size - 1, {}, ResumableUploadResponse::kInProgress, {}});
+            "", ResumableUploadResponse::kInProgress, size, {}, {}});
       })
       .WillOnce([&](UploadChunkRequest const&) {
         return make_status_or(ResumableUploadResponse{
-            url2, 2 * size - 1, {}, ResumableUploadResponse::kInProgress, {}});
+            url2, ResumableUploadResponse::kInProgress, 2 * size, {}, {}});
       });
 
   auto upload = session.UploadChunk({{payload}});
@@ -198,7 +198,7 @@ TEST(CurlResumableUploadSessionTest, SessionUpdatedInChunkUpload) {
 
 TEST(CurlResumableUploadSessionTest, Empty) {
   auto mock = MockCurlClient::Create();
-  std::string test_url = "http://invalid.example.com/not-used-in-mock";
+  std::string test_url = "https://invalid.example.com/not-used-in-mock";
   ResumableUploadRequest request("test-bucket", "test-object");
   CurlResumableUploadSession session(mock, request, test_url);
 
@@ -214,12 +214,12 @@ TEST(CurlResumableUploadSessionTest, Empty) {
         EXPECT_EQ(0, request.source_size());
         EXPECT_EQ(0, request.range_begin());
         return make_status_or(ResumableUploadResponse{
-            "", size, {}, ResumableUploadResponse::kDone, {}});
+            "", ResumableUploadResponse::kDone, size, {}, {}});
       });
 
   auto upload = session.UploadFinalChunk({{payload}}, size, {});
   EXPECT_STATUS_OK(upload);
-  EXPECT_EQ(size, upload->last_committed_byte);
+  EXPECT_EQ(size, upload->committed_size.value_or(0));
   EXPECT_EQ(size, session.next_expected_byte());
   EXPECT_TRUE(session.done());
 }
diff --git a/google/cloud/storage/internal/grpc_client.cc b/google/cloud/storage/internal/grpc_client.cc
index 34f91598b4fb7..6a45c2efd107f 100644
--- a/google/cloud/storage/internal/grpc_client.cc
+++ b/google/cloud/storage/internal/grpc_client.cc
@@ -188,12 +188,9 @@ StatusOr<ResumableUploadResponse> GrpcClient::QueryResumableUpload(
 
   ResumableUploadResponse response;
   response.upload_state = ResumableUploadResponse::kInProgress;
-  // TODO(#6880) - cleanup the committed_byte vs. size thing
-  if (status->has_persisted_size() && status->persisted_size()) {
-    response.last_committed_byte =
+  if (status->has_persisted_size()) {
+    response.committed_size =
         static_cast<std::uint64_t>(status->persisted_size());
-  } else {
-    response.last_committed_byte = 0;
   }
   if (status->has_resource()) {
     response.payload =
diff --git a/google/cloud/storage/internal/grpc_object_request_parser.cc b/google/cloud/storage/internal/grpc_object_request_parser.cc
index 84862bd6f7474..c9cc25d69d596 100644
--- a/google/cloud/storage/internal/grpc_object_request_parser.cc
+++ b/google/cloud/storage/internal/grpc_object_request_parser.cc
@@ -256,12 +256,8 @@ ResumableUploadResponse GrpcObjectRequestParser::FromProto(
     google::storage::v2::WriteObjectResponse const& p, Options const& options) {
   ResumableUploadResponse response;
   response.upload_state = ResumableUploadResponse::kInProgress;
-  if (p.has_persisted_size() && p.persisted_size() > 0) {
-    // TODO(#6880) - cleanup the committed_byte vs. size thing
-    response.last_committed_byte =
-        static_cast<std::uint64_t>(p.persisted_size()) - 1;
-  } else {
-    response.last_committed_byte = 0;
+  if (p.has_persisted_size()) {
+    response.committed_size = static_cast<std::uint64_t>(p.persisted_size());
   }
   if (p.has_resource()) {
     response.payload =
diff --git a/google/cloud/storage/internal/grpc_resumable_upload_session.cc b/google/cloud/storage/internal/grpc_resumable_upload_session.cc
index 66f8e2f9893f4..21d000d6db86e 100644
--- a/google/cloud/storage/internal/grpc_resumable_upload_session.cc
+++ b/google/cloud/storage/internal/grpc_resumable_upload_session.cc
@@ -61,7 +61,7 @@ StatusOr<ResumableUploadResponse> GrpcResumableUploadSession::ResetSession() {
   if (!last_response_) return last_response_;
 
   done_ = (last_response_->upload_state == ResumableUploadResponse::kDone);
-  next_expected_ = last_response_->last_committed_byte;
+  next_expected_ = last_response_->committed_size.value_or(0);
   return last_response_;
 }
 
diff --git a/google/cloud/storage/internal/grpc_resumable_upload_session_test.cc b/google/cloud/storage/internal/grpc_resumable_upload_session_test.cc
index 9cd8e9ff63ec2..bfe19d156ebd0 100644
--- a/google/cloud/storage/internal/grpc_resumable_upload_session_test.cc
+++ b/google/cloud/storage/internal/grpc_resumable_upload_session_test.cc
@@ -126,14 +126,14 @@ TEST(GrpcResumableUploadSessionTest, Simple) {
       });
 
   auto upload = session.UploadChunk({{payload}});
-  EXPECT_STATUS_OK(upload);
-  EXPECT_EQ(size - 1, upload->last_committed_byte);
+  ASSERT_STATUS_OK(upload);
+  EXPECT_EQ(size, upload->committed_size.value_or(0));
   EXPECT_EQ(size, session.next_expected_byte());
   EXPECT_FALSE(session.done());
 
   upload = session.UploadFinalChunk({{payload}}, 2 * size, hashes);
   EXPECT_STATUS_OK(upload);
-  EXPECT_EQ(2 * size - 1, upload->last_committed_byte);
+  EXPECT_EQ(2 * size, upload->committed_size.value_or(0));
   EXPECT_EQ(2 * size, session.next_expected_byte());
   EXPECT_TRUE(session.done());
 }
@@ -179,13 +179,13 @@ TEST(GrpcResumableUploadSessionTest, SingleStreamForLargeChunks) {
 
   auto upload = session.UploadChunk({{payload}});
   EXPECT_STATUS_OK(upload);
-  EXPECT_EQ(size - 1, upload->last_committed_byte);
+  EXPECT_EQ(size, upload->committed_size.value_or(0));
   EXPECT_EQ(size, session.next_expected_byte());
   EXPECT_FALSE(session.done());
 
   upload = session.UploadFinalChunk({{payload}}, 2 * size, HashValues{});
   EXPECT_STATUS_OK(upload);
-  EXPECT_EQ(2 * size - 1, upload->last_committed_byte);
+  EXPECT_EQ(2 * size, upload->committed_size.value_or(0));
   EXPECT_EQ(2 * size, session.next_expected_byte());
   EXPECT_TRUE(session.done());
 }
@@ -235,7 +235,7 @@ TEST(GrpcResumableUploadSessionTest, Reset) {
       .WillOnce([&](QueryResumableUploadRequest const& request) {
         EXPECT_EQ("test-upload-id", request.upload_session_url());
         return make_status_or(ResumableUploadResponse{
-            {}, size, {}, ResumableUploadResponse::kInProgress, {}});
+            {}, ResumableUploadResponse::kInProgress, size, {}, {}});
       });
 
   auto upload = session.UploadChunk({{payload}});
@@ -303,7 +303,7 @@ TEST(GrpcResumableUploadSessionTest, ResumeFromEmpty) {
       });
 
   ResumableUploadResponse const resume_response{
-      {}, 0, {}, ResumableUploadResponse::kInProgress, {}};
+      {}, ResumableUploadResponse::kInProgress, 0, {}, {}};
   EXPECT_CALL(*mock, QueryResumableUpload)
       .WillOnce([&](QueryResumableUploadRequest const& request) {
         EXPECT_EQ("test-upload-id", request.upload_session_url());
diff --git a/google/cloud/storage/internal/logging_resumable_upload_session_test.cc b/google/cloud/storage/internal/logging_resumable_upload_session_test.cc
index 5b38a588f9542..03a5eca24a0ef 100644
--- a/google/cloud/storage/internal/logging_resumable_upload_session_test.cc
+++ b/google/cloud/storage/internal/logging_resumable_upload_session_test.cc
@@ -119,7 +119,7 @@ TEST_F(LoggingResumableUploadSessionTest, LastResponseOk) {
   auto mock = absl::make_unique<testing::MockResumableUploadSession>();
 
   const StatusOr<ResumableUploadResponse> last_response(ResumableUploadResponse{
-      "upload url", 1, {}, ResumableUploadResponse::kInProgress, {}});
+      "upload url", ResumableUploadResponse::kInProgress, 1, {}, {}});
   EXPECT_CALL(*mock, last_response()).WillOnce(ReturnRef(last_response));
 
   LoggingResumableUploadSession session(std::move(mock));
diff --git a/google/cloud/storage/internal/object_write_streambuf.cc b/google/cloud/storage/internal/object_write_streambuf.cc
index c7f4bd1a919ce..32092c49b2018 100644
--- a/google/cloud/storage/internal/object_write_streambuf.cc
+++ b/google/cloud/storage/internal/object_write_streambuf.cc
@@ -36,7 +36,7 @@ ObjectWriteStreambuf::ObjectWriteStreambuf(
       hash_validator_(std::move(hash_validator)),
       auto_finalize_(auto_finalize),
       last_response_(ResumableUploadResponse{
-          {}, 0, {}, ResumableUploadResponse::kInProgress, {}}) {
+          {}, ResumableUploadResponse::kInProgress, 0, absl::nullopt, {}}) {
   current_ios_buffer_.resize(max_buffer_size_);
   auto* pbeg = current_ios_buffer_.data();
   auto* pend = pbeg + current_ios_buffer_.size();
diff --git a/google/cloud/storage/internal/object_write_streambuf_test.cc b/google/cloud/storage/internal/object_write_streambuf_test.cc
index 66ac61e1fdb29..c6099a4f4424a 100644
--- a/google/cloud/storage/internal/object_write_streambuf_test.cc
+++ b/google/cloud/storage/internal/object_write_streambuf_test.cc
@@ -51,7 +51,7 @@ TEST(ObjectWriteStreambufTest, EmptyStream) {
         EXPECT_EQ(0, TotalBytes(p));
         EXPECT_EQ(0, s);
         return make_status_or(ResumableUploadResponse{
-            "{}", 0, {}, ResumableUploadResponse::kInProgress, {}});
+            "{}", ResumableUploadResponse::kInProgress, 0, {}, {}});
       });
   EXPECT_CALL(*mock, next_expected_byte()).WillOnce(Return(0));
 
@@ -78,7 +78,7 @@ TEST(ObjectWriteStreambufTest, AutoFinalizeEnabled) {
         EXPECT_EQ(0, TotalBytes(p));
         EXPECT_EQ(0, s);
         return make_status_or(ResumableUploadResponse{
-            {}, 0, {}, ResumableUploadResponse::kDone, {}});
+            {}, ResumableUploadResponse::kDone, 0, {}, {}});
       });
   EXPECT_CALL(*mock, next_expected_byte()).WillOnce(Return(0));
 
@@ -122,12 +122,11 @@ TEST(ObjectWriteStreambufTest, SmallStream) {
         EXPECT_EQ(1, count);
         EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
         EXPECT_EQ(payload.size(), s);
-        auto last_committed_byte = payload.size() - 1;
         return make_status_or(
             ResumableUploadResponse{"{}",
-                                    last_committed_byte,
-                                    {},
                                     ResumableUploadResponse::kInProgress,
+                                    payload.size(),
+                                    {},
                                     {}});
       });
   EXPECT_CALL(*mock, next_expected_byte()).WillOnce(Return(0));
@@ -156,10 +155,9 @@ TEST(ObjectWriteStreambufTest, EmptyTrailer) {
     ++count;
     EXPECT_EQ(1, count);
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
-    auto last_committed_byte = payload.size() - 1;
-    next_byte = last_committed_byte + 1;
+    next_byte = payload.size();
     return make_status_or(ResumableUploadResponse{
-        "", last_committed_byte, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, payload.size(), {}, {}});
   });
   EXPECT_CALL(*mock, UploadFinalChunk)
       .WillOnce([&](ConstBufferSequence const& p, std::uint64_t s,
@@ -168,13 +166,8 @@ TEST(ObjectWriteStreambufTest, EmptyTrailer) {
         EXPECT_EQ(2, count);
         EXPECT_EQ(0, TotalBytes(p));
         EXPECT_EQ(quantum, s);
-        auto last_committed_byte = quantum - 1;
-        return make_status_or(
-            ResumableUploadResponse{"{}",
-                                    last_committed_byte,
-                                    {},
-                                    ResumableUploadResponse::kInProgress,
-                                    {}});
+        return make_status_or(ResumableUploadResponse{
+            "{}", ResumableUploadResponse::kInProgress, quantum, {}, {}});
       });
   EXPECT_CALL(*mock, next_expected_byte()).WillRepeatedly([&]() {
     return next_byte;
@@ -202,20 +195,16 @@ TEST(ObjectWriteStreambufTest, FlushAfterLargePayload) {
     InSequence seq;
     EXPECT_CALL(*mock, UploadChunk).WillOnce(InvokeWithoutArgs([&]() {
       next_byte = payload_1.size();
-      return make_status_or(
-          ResumableUploadResponse{"",
-                                  payload_1.size() - 1,
-                                  {},
-                                  ResumableUploadResponse::kInProgress,
-                                  {}});
+      return make_status_or(ResumableUploadResponse{
+          "", ResumableUploadResponse::kInProgress, payload_1.size(), {}, {}});
     }));
     EXPECT_CALL(*mock, UploadFinalChunk({{payload_2}},
                                         payload_1.size() + payload_2.size(), _))
         .WillOnce(Return(make_status_or(
             ResumableUploadResponse{"{}",
-                                    payload_1.size() + payload_2.size() - 1,
-                                    {},
                                     ResumableUploadResponse::kInProgress,
+                                    payload_1.size() + payload_2.size(),
+                                    {},
                                     {}})));
   }
   EXPECT_CALL(*mock, next_expected_byte()).WillRepeatedly([&]() {
@@ -250,7 +239,7 @@ TEST(ObjectWriteStreambufTest, FlushAfterFullQuantum) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload_1}, ConstBuffer{trailer}));
     next_byte += TotalBytes(p);
     return make_status_or(ResumableUploadResponse{
-        "", quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, quantum, {}, {}});
   });
   EXPECT_CALL(*mock, UploadFinalChunk)
       .WillOnce([&](ConstBufferSequence const& p, std::uint64_t s,
@@ -263,9 +252,9 @@ TEST(ObjectWriteStreambufTest, FlushAfterFullQuantum) {
         auto last_committed_byte = payload_1.size() + payload_2.size() - 1;
         return make_status_or(
             ResumableUploadResponse{"{}",
+                                    ResumableUploadResponse::kInProgress,
                                     last_committed_byte,
                                     {},
-                                    ResumableUploadResponse::kInProgress,
                                     {}});
       });
   EXPECT_CALL(*mock, next_expected_byte()).WillRepeatedly([&]() {
@@ -302,7 +291,7 @@ TEST(ObjectWriteStreambufTest, OverflowFlushAtFullQuantum) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
     next_byte += TotalBytes(p);
     return make_status_or(ResumableUploadResponse{
-        "", next_byte - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, next_byte, {}, {}});
   });
   EXPECT_CALL(*mock, UploadFinalChunk)
       .WillOnce([&](ConstBufferSequence const& p, std::uint64_t s,
@@ -315,7 +304,7 @@ TEST(ObjectWriteStreambufTest, OverflowFlushAtFullQuantum) {
         EXPECT_EQ(next_byte, s);
         mock_done = true;
         return make_status_or(ResumableUploadResponse{
-            "{}", next_byte - 1, {}, ResumableUploadResponse::kDone, {}});
+            "{}", ResumableUploadResponse::kDone, next_byte, {}, {}});
       });
 
   ObjectWriteStreambuf streambuf(
@@ -344,7 +333,7 @@ TEST(ObjectWriteStreambufTest, SomeBytesNotAccepted) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{expected}));
     next_byte += quantum;
     return make_status_or(ResumableUploadResponse{
-        "", next_byte - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, next_byte, {}, {}});
   });
   EXPECT_CALL(*mock, UploadFinalChunk)
       .WillOnce([&](ConstBufferSequence const& p, std::uint64_t s,
@@ -354,7 +343,7 @@ TEST(ObjectWriteStreambufTest, SomeBytesNotAccepted) {
         EXPECT_EQ(payload.size(), s);
         next_byte += content.size();
         return make_status_or(ResumableUploadResponse{
-            "{}", next_byte - 1, {}, ResumableUploadResponse::kInProgress, {}});
+            "{}", ResumableUploadResponse::kInProgress, next_byte, {}, {}});
       });
   EXPECT_CALL(*mock, next_expected_byte()).WillRepeatedly([&]() {
     return next_byte;
@@ -387,7 +376,7 @@ TEST(ObjectWriteStreambufTest, NextExpectedByteJumpsAhead) {
     // than expected
     next_byte += quantum * 3;
     return make_status_or(ResumableUploadResponse{
-        "", next_byte - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, next_byte, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte()).WillRepeatedly([&]() {
     return next_byte;
@@ -417,7 +406,7 @@ TEST(ObjectWriteStreambufTest, NextExpectedByteDecreases) {
   EXPECT_CALL(*mock, UploadChunk).WillOnce(InvokeWithoutArgs([&]() {
     next_byte--;
     return make_status_or(ResumableUploadResponse{
-        "", next_byte - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, next_byte, {}, {}});
   }));
   EXPECT_CALL(*mock, next_expected_byte()).WillRepeatedly([&]() {
     return next_byte;
@@ -454,7 +443,7 @@ TEST(ObjectWriteStreambufTest, MixPutcPutn) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload_1}, ConstBuffer{expected}));
     next_byte += TotalBytes(p);
     return make_status_or(ResumableUploadResponse{
-        "", quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, quantum, {}, {}});
   });
   EXPECT_CALL(*mock, UploadFinalChunk)
       .WillOnce([&](ConstBufferSequence const& p, std::uint64_t s,
@@ -464,12 +453,11 @@ TEST(ObjectWriteStreambufTest, MixPutcPutn) {
         auto expected = payload_2.substr(payload_2.size() - payload_1.size());
         EXPECT_THAT(p, ElementsAre(ConstBuffer{expected}));
         EXPECT_EQ(payload_1.size() + payload_2.size(), s);
-        auto last_committed_byte = payload_1.size() + payload_2.size() - 1;
         return make_status_or(
             ResumableUploadResponse{"{}",
-                                    last_committed_byte,
-                                    {},
                                     ResumableUploadResponse::kInProgress,
+                                    payload_1.size() + payload_2.size(),
+                                    {},
                                     {}});
       });
   EXPECT_CALL(*mock, next_expected_byte()).WillRepeatedly([&]() {
@@ -494,7 +482,7 @@ TEST(ObjectWriteStreambufTest, CreatedForFinalizedUpload) {
   auto mock = absl::make_unique<testing::MockResumableUploadSession>();
   EXPECT_CALL(*mock, done).WillRepeatedly(Return(true));
   auto last_upload_response = make_status_or(ResumableUploadResponse{
-      "url-for-test", 0, {}, ResumableUploadResponse::kDone, {}});
+      "url-for-test", ResumableUploadResponse::kDone, 0, {}, {}});
   EXPECT_CALL(*mock, last_response).WillOnce(ReturnRef(last_upload_response));
 
   ObjectWriteStreambuf streambuf(
@@ -589,13 +577,13 @@ TEST(ObjectWriteStreambufTest, KnownSizeUpload) {
         EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
         mock_next_byte += TotalBytes(p);
         return make_status_or(ResumableUploadResponse{
-            "", 2 * quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+            "", ResumableUploadResponse::kInProgress, 2 * quantum, {}, {}});
       })
       .WillOnce([&](ConstBufferSequence const& p) {
         EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
         mock_next_byte += TotalBytes(p);
         return make_status_or(ResumableUploadResponse{
-            "", 4 * quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+            "", ResumableUploadResponse::kInProgress, 4 * quantum, {}, {}});
       })
       .WillOnce([&](ConstBufferSequence const& p) {
         EXPECT_THAT(p, ElementsAre(ConstBuffer{payload.data(), quantum}));
@@ -606,7 +594,7 @@ TEST(ObjectWriteStreambufTest, KnownSizeUpload) {
         // byte of "0".
         mock_is_done = true;
         return make_status_or(ResumableUploadResponse{
-            "", 0, {}, ResumableUploadResponse::kDone, {}});
+            "", ResumableUploadResponse::kDone, 0, {}, {}});
       });
 
   ObjectWriteStreambuf streambuf(
@@ -642,17 +630,17 @@ TEST(ObjectWriteStreambufTest, Pubsync) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
     mock_next_byte += TotalBytes(p);
     return make_status_or(ResumableUploadResponse{
-        "", mock_next_byte - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, mock_next_byte, {}, {}});
   });
   EXPECT_CALL(*mock, UploadFinalChunk)
-      .WillOnce([&](ConstBufferSequence const& p, std::uint64_t,
-                    HashValues const&) {
-        EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
-        mock_next_byte += TotalBytes(p);
-        mock_is_done = true;
-        return make_status_or(ResumableUploadResponse{
-            "", mock_next_byte - 1, {}, ResumableUploadResponse::kDone, {}});
-      });
+      .WillOnce(
+          [&](ConstBufferSequence const& p, std::uint64_t, HashValues const&) {
+            EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
+            mock_next_byte += TotalBytes(p);
+            mock_is_done = true;
+            return make_status_or(ResumableUploadResponse{
+                "", ResumableUploadResponse::kDone, mock_next_byte, {}, {}});
+          });
 
   ObjectWriteStreambuf streambuf(
       std::move(mock), 2 * quantum, CreateNullHashFunction(), HashValues{},
diff --git a/google/cloud/storage/internal/resumable_upload_session.cc b/google/cloud/storage/internal/resumable_upload_session.cc
index 7f1fd8dd5db14..8a0c55344c02b 100644
--- a/google/cloud/storage/internal/resumable_upload_session.cc
+++ b/google/cloud/storage/internal/resumable_upload_session.cc
@@ -24,6 +24,32 @@ namespace cloud {
 namespace storage {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
+namespace {
+StatusOr<std::uint64_t> ParseRangeHeader(std::string const& range) {
+  // We expect a `Range:` header in the format described here:
+  //    https://cloud.google.com/storage/docs/json_api/v1/how-tos/resumable-upload
+  // that is the value should match `bytes=0-[0-9]+`:
+  char const prefix[] = "bytes=0-";
+  auto constexpr kPrefixLen = sizeof(prefix) - 1;
+  if (range.rfind(prefix, 0) != 0) {
+    return Status(
+        StatusCode::kInternal,
+        "cannot parse Range header in resumable upload response, value=" +
+            range);
+  }
+  char const* buffer = range.data() + kPrefixLen;
+  char* endptr;
+  auto constexpr kBytesBase = 10;
+  auto last = std::strtoll(buffer, &endptr, kBytesBase);
+  if (buffer != endptr && *endptr == '\0' && 0 <= last) {
+    return last;
+  }
+  return Status(
+      StatusCode::kInternal,
+      "cannot parse Range header in resumable upload response, value=" + range);
+}
+}  // namespace
+
 StatusOr<ResumableUploadResponse> ResumableUploadResponse::FromHttpResponse(
     HttpResponse response) {
   ResumableUploadResponse result;
@@ -33,55 +59,24 @@ StatusOr<ResumableUploadResponse> ResumableUploadResponse::FromHttpResponse(
   } else {
     result.upload_state = kInProgress;
   }
-  result.last_committed_byte = 0;
   result.annotations += "code=" + std::to_string(response.status_code);
   // For the JSON API, the payload contains the object resource when the upload
   // is finished. In that case, we try to parse it.
   if (result.upload_state == kDone && !response.payload.empty()) {
     auto contents = ObjectMetadataParser::FromString(response.payload);
-    if (!contents) {
-      return std::move(contents).status();
-    }
+    if (!contents) return std::move(contents).status();
     result.payload = *std::move(contents);
   }
   if (response.headers.find("location") != response.headers.end()) {
     result.upload_session_url = response.headers.find("location")->second;
   }
   auto r = response.headers.find("range");
-  if (r == response.headers.end()) {
-    std::ostringstream os;
-    os << __func__ << "() missing range header in resumable upload response"
-       << ", response=" << response;
-    result.annotations += " " + std::move(os).str();
-    return result;
-  }
-  // We expect a `Range:` header in the format described here:
-  //    https://cloud.google.com/storage/docs/json_api/v1/how-tos/resumable-upload
-  // that is the value should match `bytes=0-[0-9]+`:
-  std::string const& range = r->second;
-  result.annotations += " range=" + range;
+  if (r == response.headers.end()) return result;
 
-  char const prefix[] = "bytes=0-";
-  auto constexpr kPrefixLen = sizeof(prefix) - 1;
-  if (range.rfind(prefix, 0) != 0) {
-    std::ostringstream os;
-    os << __func__ << "() cannot parse range: header in resumable upload"
-       << " response, header=" << range << ", response=" << response;
-    result.annotations += " " + std::move(os).str();
-    return result;
-  }
-  char const* buffer = range.data() + kPrefixLen;
-  char* endptr;
-  auto constexpr kBytesBase = 10;
-  auto last = std::strtoll(buffer, &endptr, kBytesBase);
-  if (buffer != endptr && *endptr == '\0' && 0 <= last) {
-    result.last_committed_byte = static_cast<std::uint64_t>(last);
-  } else {
-    std::ostringstream os;
-    os << __func__ << "() cannot parse range: header in resumable upload"
-       << " response, header=" << range << ", response=" << response;
-    result.annotations += " " + std::move(os).str();
-  }
+  result.annotations += " range=" + r->second;
+  auto last_committed_byte = ParseRangeHeader(r->second);
+  if (!last_committed_byte) return std::move(last_committed_byte).status();
+  result.committed_size = *last_committed_byte + 1;
 
   return result;
 }
@@ -89,7 +84,7 @@ StatusOr<ResumableUploadResponse> ResumableUploadResponse::FromHttpResponse(
 bool operator==(ResumableUploadResponse const& lhs,
                 ResumableUploadResponse const& rhs) {
   return lhs.upload_session_url == rhs.upload_session_url &&
-         lhs.last_committed_byte == rhs.last_committed_byte &&
+         lhs.committed_size == rhs.committed_size &&
          lhs.payload == rhs.payload && lhs.upload_state == rhs.upload_state;
 }
 
@@ -99,18 +94,23 @@ bool operator!=(ResumableUploadResponse const& lhs,
 }
 
 std::ostream& operator<<(std::ostream& os, ResumableUploadResponse const& r) {
+  char const* state = r.upload_state == ResumableUploadResponse::kDone
+                          ? "kDone"
+                          : "kInProgress";
   os << "ResumableUploadResponse={upload_session_url=" << r.upload_session_url
-     << ", last_committed_byte=" << r.last_committed_byte << ", payload=";
+     << ", upload_state=" << state << ", committed_size=";
+  if (r.committed_size.has_value()) {
+    os << *r.committed_size;
+  } else {
+    os << "{}";
+  }
+  os << ", payload=";
   if (r.payload.has_value()) {
     os << *r.payload;
   } else {
     os << "{}";
   }
-  return os << ", upload_state="
-            << (r.upload_state == ResumableUploadResponse::kDone
-                    ? "kDone"
-                    : "kInProgress")
-            << ", annotations=" << r.annotations << "}";
+  return os << ", annotations=" << r.annotations << "}";
 }
 
 }  // namespace internal
diff --git a/google/cloud/storage/internal/resumable_upload_session.h b/google/cloud/storage/internal/resumable_upload_session.h
index 3823f7b6da797..ed7db987af95e 100644
--- a/google/cloud/storage/internal/resumable_upload_session.h
+++ b/google/cloud/storage/internal/resumable_upload_session.h
@@ -89,13 +89,14 @@ class ResumableUploadSession {
 
 struct ResumableUploadResponse {
   enum UploadState { kInProgress, kDone };
+
   static StatusOr<ResumableUploadResponse> FromHttpResponse(
       HttpResponse response);
 
   std::string upload_session_url;
-  std::uint64_t last_committed_byte;
-  absl::optional<google::cloud::storage::ObjectMetadata> payload;
   UploadState upload_state;
+  absl::optional<std::uint64_t> committed_size;
+  absl::optional<google::cloud::storage::ObjectMetadata> payload;
   std::string annotations;
 };
 
diff --git a/google/cloud/storage/internal/resumable_upload_session_test.cc b/google/cloud/storage/internal/resumable_upload_session_test.cc
index 9d9096850e73e..41f004b758f23 100644
--- a/google/cloud/storage/internal/resumable_upload_session_test.cc
+++ b/google/cloud/storage/internal/resumable_upload_session_test.cc
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/resumable_upload_session.h"
+#include "google/cloud/testing_util/status_matchers.h"
 #include <gmock/gmock.h>
 
 namespace google {
@@ -22,6 +23,7 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
 namespace {
 
+using ::google::cloud::testing_util::StatusIs;
 using ::testing::HasSubstr;
 
 TEST(ResumableUploadResponseTest, Base) {
@@ -30,29 +32,29 @@ TEST(ResumableUploadResponseTest, Base) {
                                  R"""({"name": "test-object-name"})""",
                                  {{"ignored-header", "value"},
                                   {"location", "location-value"},
-                                  {"range", "bytes=0-2000"}}})
+                                  {"range", "bytes=0-1999"}}})
                     .value();
   ASSERT_TRUE(actual.payload.has_value());
   EXPECT_EQ("test-object-name", actual.payload->name());
   EXPECT_EQ("location-value", actual.upload_session_url);
-  EXPECT_EQ(2000, actual.last_committed_byte);
+  EXPECT_EQ(2000, actual.committed_size);
   EXPECT_EQ(ResumableUploadResponse::kDone, actual.upload_state);
 
   std::ostringstream os;
   os << actual;
   auto actual_str = os.str();
   EXPECT_THAT(actual_str, HasSubstr("upload_session_url=location-value"));
-  EXPECT_THAT(actual_str, HasSubstr("last_committed_byte=2000"));
+  EXPECT_THAT(actual_str, HasSubstr("committed_size=2000"));
   EXPECT_THAT(actual_str, HasSubstr("annotations="));
 }
 
 TEST(ResumableUploadResponseTest, NoLocation) {
   auto actual = ResumableUploadResponse::FromHttpResponse(
-                    HttpResponse{308, {}, {{"range", "bytes=0-2000"}}})
+                    HttpResponse{308, {}, {{"range", "bytes=0-1999"}}})
                     .value();
   EXPECT_FALSE(actual.payload.has_value());
   EXPECT_EQ("", actual.upload_session_url);
-  EXPECT_EQ(2000, actual.last_committed_byte);
+  EXPECT_EQ(2000, actual.committed_size);
   EXPECT_EQ(ResumableUploadResponse::kInProgress, actual.upload_state);
 }
 
@@ -65,71 +67,48 @@ TEST(ResumableUploadResponseTest, NoRange) {
   ASSERT_TRUE(actual.payload.has_value());
   EXPECT_EQ("test-object-name", actual.payload->name());
   EXPECT_EQ("location-value", actual.upload_session_url);
-  EXPECT_EQ(0, actual.last_committed_byte);
+  EXPECT_FALSE(actual.committed_size.has_value());
   EXPECT_EQ(ResumableUploadResponse::kDone, actual.upload_state);
 }
 
 TEST(ResumableUploadResponseTest, MissingBytesInRange) {
-  auto actual = ResumableUploadResponse::FromHttpResponse(
-                    HttpResponse{308,
-                                 {},
-                                 {{"location", "location-value"},
-                                  {"range", "units=0-2000"}}})
-                    .value();
-  EXPECT_FALSE(actual.payload.has_value());
-  EXPECT_EQ("location-value", actual.upload_session_url);
-  EXPECT_EQ(0, actual.last_committed_byte);
-  EXPECT_EQ(ResumableUploadResponse::kInProgress, actual.upload_state);
+  auto actual = ResumableUploadResponse::FromHttpResponse(HttpResponse{
+      308, {}, {{"location", "location-value"}, {"range", "units=0-2000"}}});
+  EXPECT_THAT(actual,
+              StatusIs(StatusCode::kInternal, HasSubstr("units=0-2000")));
 }
 
 TEST(ResumableUploadResponseTest, MissingRangeEnd) {
   auto actual = ResumableUploadResponse::FromHttpResponse(
-                    HttpResponse{308, {}, {{"range", "bytes=0-"}}})
-                    .value();
-  EXPECT_FALSE(actual.payload.has_value());
-  EXPECT_EQ("", actual.upload_session_url);
-  EXPECT_EQ(0, actual.last_committed_byte);
-  EXPECT_EQ(ResumableUploadResponse::kInProgress, actual.upload_state);
+      HttpResponse{308, {}, {{"range", "bytes=0-"}}});
+  EXPECT_THAT(actual, StatusIs(StatusCode::kInternal, HasSubstr("bytes=0-")));
 }
 
 TEST(ResumableUploadResponseTest, InvalidRangeEnd) {
   auto actual = ResumableUploadResponse::FromHttpResponse(
-                    HttpResponse{308, {}, {{"range", "bytes=0-abcd"}}})
-                    .value();
-  EXPECT_FALSE(actual.payload.has_value());
-  EXPECT_EQ("", actual.upload_session_url);
-  EXPECT_EQ(0, actual.last_committed_byte);
-  EXPECT_EQ(ResumableUploadResponse::kInProgress, actual.upload_state);
+      HttpResponse{308, {}, {{"range", "bytes=0-abcd"}}});
+  EXPECT_THAT(actual,
+              StatusIs(StatusCode::kInternal, HasSubstr("bytes=0-abcd")));
 }
 
 TEST(ResumableUploadResponseTest, InvalidRangeBegin) {
   auto actual = ResumableUploadResponse::FromHttpResponse(
-                    HttpResponse{308, {}, {{"range", "bytes=abcd-2000"}}})
-                    .value();
-  EXPECT_FALSE(actual.payload.has_value());
-  EXPECT_EQ("", actual.upload_session_url);
-  EXPECT_EQ(0, actual.last_committed_byte);
-  EXPECT_EQ(ResumableUploadResponse::kInProgress, actual.upload_state);
+      HttpResponse{308, {}, {{"range", "bytes=abcd-2000"}}});
+  EXPECT_THAT(actual,
+              StatusIs(StatusCode::kInternal, HasSubstr("bytes=abcd-2000")));
 }
 
 TEST(ResumableUploadResponseTest, UnexpectedRangeBegin) {
   auto actual = ResumableUploadResponse::FromHttpResponse(
-                    HttpResponse{308, {}, {{"range", "bytes=3000-2000"}}})
-                    .value();
-  EXPECT_FALSE(actual.payload.has_value());
-  EXPECT_EQ("", actual.upload_session_url);
-  EXPECT_EQ(0, actual.last_committed_byte);
-  EXPECT_EQ(ResumableUploadResponse::kInProgress, actual.upload_state);
+      HttpResponse{308, {}, {{"range", "bytes=3000-2000"}}});
+  EXPECT_THAT(actual,
+              StatusIs(StatusCode::kInternal, HasSubstr("bytes=3000-2000")));
 }
 
 TEST(ResumableUploadResponseTest, NegativeEnd) {
   auto actual = ResumableUploadResponse::FromHttpResponse(
-                    HttpResponse{308, {}, {{"range", "bytes=0--7"}}})
-                    .value();
-  EXPECT_FALSE(actual.payload.has_value());
-  EXPECT_EQ("", actual.upload_session_url);
-  EXPECT_EQ(0, actual.last_committed_byte);
-  EXPECT_EQ(ResumableUploadResponse::kInProgress, actual.upload_state);
+      HttpResponse{308, {}, {{"range", "bytes=0--7"}}});
+  EXPECT_THAT(actual, StatusIs(StatusCode::kInternal, HasSubstr("bytes=0--7")));
 }
 
 }  // namespace
diff --git a/google/cloud/storage/internal/retry_resumable_upload_session_test.cc b/google/cloud/storage/internal/retry_resumable_upload_session_test.cc
index d15560bde7890..68bfdacafa4e0 100644
--- a/google/cloud/storage/internal/retry_resumable_upload_session_test.cc
+++ b/google/cloud/storage/internal/retry_resumable_upload_session_test.cc
@@ -68,14 +68,14 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransient) {
           [&]() { return StatusOr<ResumableUploadResponse>(TransientError()); })
       .WillOnce([&]() {
         return make_status_or(ResumableUploadResponse{
-            "", 0, {}, ResumableUploadResponse::kInProgress, {}});
+            "", ResumableUploadResponse::kInProgress, 0, {}, {}});
       });
   EXPECT_CALL(*mock, next_expected_byte).Times(1).WillRepeatedly(Return(0));
 
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const& p) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
     return make_status_or(ResumableUploadResponse{
-        "", quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, quantum, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte)
       .Times(1)
@@ -93,7 +93,7 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransient) {
   });
   EXPECT_CALL(*mock, ResetSession).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, quantum, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte)
       .Times(1)
@@ -101,7 +101,7 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransient) {
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const& p) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
     return make_status_or(ResumableUploadResponse{
-        "", 2 * quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 2 * quantum, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte)
       .Times(1)
@@ -115,7 +115,7 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransient) {
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const& p) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
     return make_status_or(ResumableUploadResponse{
-        "", 3 * quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 3 * quantum, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte)
       .Times(1)
@@ -128,15 +128,15 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransient) {
   StatusOr<ResumableUploadResponse> response;
   response = session.UploadChunk({{payload}});
   EXPECT_STATUS_OK(response);
-  EXPECT_EQ(quantum - 1, response->last_committed_byte);
+  EXPECT_EQ(quantum, response->committed_size.value_or(0));
 
   response = session.UploadChunk({{payload}});
   EXPECT_STATUS_OK(response);
-  EXPECT_EQ(2 * quantum - 1, response->last_committed_byte);
+  EXPECT_EQ(2 * quantum, response->committed_size.value_or(0));
 
   response = session.UploadChunk({{payload}});
   EXPECT_STATUS_OK(response);
-  EXPECT_EQ(3 * quantum - 1, response->last_committed_byte);
+  EXPECT_EQ(3 * quantum, response->committed_size.value_or(0));
 }
 
 /// @test Verify that a permanent error on UploadChunk results in a failure.
@@ -207,7 +207,7 @@ TEST_F(RetryResumableUploadSessionTest, TooManyTransientOnUploadChunk) {
   });
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", 0, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 0, {}, {}});
   });
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const& p) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
@@ -215,7 +215,7 @@ TEST_F(RetryResumableUploadSessionTest, TooManyTransientOnUploadChunk) {
   });
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", 0, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 0, {}, {}});
   });
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const& p) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
@@ -292,64 +292,44 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransiensOnSeparateChunks) {
   });
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", 0, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 0, {}, {}});
   });
 
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const& p) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
     next_expected_byte = quantum;
-    return make_status_or(
-        ResumableUploadResponse{"",
-                                next_expected_byte - 1,
-                                {},
-                                ResumableUploadResponse::kInProgress,
-                                {}});
+    return make_status_or(ResumableUploadResponse{
+        "", ResumableUploadResponse::kInProgress, next_expected_byte, {}, {}});
   });
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const& p) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
     return StatusOr<ResumableUploadResponse>(TransientError());
   });
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
-    return make_status_or(
-        ResumableUploadResponse{"",
-                                next_expected_byte - 1,
-                                {},
-                                ResumableUploadResponse::kInProgress,
-                                {}});
+    return make_status_or(ResumableUploadResponse{
+        "", ResumableUploadResponse::kInProgress, next_expected_byte, {}, {}});
   });
 
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const& p) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
     next_expected_byte = 2 * quantum;
-    return make_status_or(
-        ResumableUploadResponse{"",
-                                next_expected_byte - 1,
-                                {},
-                                ResumableUploadResponse::kInProgress,
-                                {}});
+    return make_status_or(ResumableUploadResponse{
+        "", ResumableUploadResponse::kInProgress, next_expected_byte, {}, {}});
   });
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const& p) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
     return StatusOr<ResumableUploadResponse>(TransientError());
   });
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
-    return make_status_or(
-        ResumableUploadResponse{"",
-                                next_expected_byte - 1,
-                                {},
-                                ResumableUploadResponse::kInProgress,
-                                {}});
+    return make_status_or(ResumableUploadResponse{
+        "", ResumableUploadResponse::kInProgress, next_expected_byte, {}, {}});
   });
 
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const& p) {
     EXPECT_THAT(p, ElementsAre(ConstBuffer{payload}));
     next_expected_byte = 3 * quantum;
-    return make_status_or(
-        ResumableUploadResponse{"",
-                                next_expected_byte - 1,
-                                {},
-                                ResumableUploadResponse::kInProgress,
-                                {}});
+    return make_status_or(ResumableUploadResponse{
+        "", ResumableUploadResponse::kInProgress, next_expected_byte, {}, {}});
   });
 
   // Configure a session that tolerates 2 transient errors per call. None of
@@ -360,15 +340,15 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransiensOnSeparateChunks) {
 
   StatusOr<ResumableUploadResponse> response = session.UploadChunk({{payload}});
   EXPECT_STATUS_OK(response);
-  EXPECT_EQ(response->last_committed_byte, quantum - 1);
+  EXPECT_EQ(response->committed_size.value_or(0), quantum);
 
   response = session.UploadChunk({{payload}});
   EXPECT_STATUS_OK(response);
-  EXPECT_EQ(response->last_committed_byte, 2 * quantum - 1);
+  EXPECT_EQ(response->committed_size.value_or(0), 2 * quantum);
 
   response = session.UploadChunk({{payload}});
   EXPECT_STATUS_OK(response);
-  EXPECT_EQ(response->last_committed_byte, 3 * quantum - 1);
+  EXPECT_EQ(response->committed_size.value_or(0), 3 * quantum);
 }
 
 /// @test Verify that a permanent error on UploadFinalChunk results in a
@@ -420,7 +400,7 @@ TEST_F(RetryResumableUploadSessionTest, TooManyTransientOnUploadFinalChunk) {
       });
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", 0, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 0, {}, {}});
   });
 
   EXPECT_CALL(*mock, UploadFinalChunk)
@@ -435,7 +415,7 @@ TEST_F(RetryResumableUploadSessionTest, TooManyTransientOnUploadFinalChunk) {
 
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", 0, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 0, {}, {}});
   });
   EXPECT_CALL(*mock, UploadFinalChunk)
       .WillOnce([&](ConstBufferSequence const& p, std::uint64_t s,
@@ -471,7 +451,7 @@ TEST(RetryResumableUploadSession, Done) {
 TEST(RetryResumableUploadSession, LastResponse) {
   auto mock = absl::make_unique<testing::MockResumableUploadSession>();
   StatusOr<ResumableUploadResponse> const last_response(ResumableUploadResponse{
-      "url", 1, {}, ResumableUploadResponse::kDone, {}});
+      "url", ResumableUploadResponse::kDone, 1, {}, {}});
   EXPECT_CALL(*mock, last_response()).WillOnce(ReturnRef(last_response));
 
   RetryResumableUploadSession session(
@@ -546,7 +526,7 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransientPartialFailures) {
   });
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, quantum, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte)
       .Times(1)
@@ -558,7 +538,7 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransientPartialFailures) {
   });
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", 2 * quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 2 * quantum, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte)
       .Times(1)
@@ -567,7 +547,7 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransientPartialFailures) {
     EXPECT_EQ(quantum, TotalBytes(p));
     EXPECT_EQ('Z', p[0][0]);
     return make_status_or(ResumableUploadResponse{
-        "", 3 * quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 3 * quantum, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte)
       .Times(3)
@@ -583,7 +563,7 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransientPartialFailures) {
           });
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", 4 * quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 4 * quantum, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte)
       .Times(1)
@@ -597,7 +577,7 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransientPartialFailures) {
           });
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", 5 * quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 5 * quantum, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte)
       .Times(1)
@@ -608,7 +588,7 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransientPartialFailures) {
             EXPECT_EQ(quantum, TotalBytes(p));
             EXPECT_EQ('C', p[0][0]);
             return make_status_or(ResumableUploadResponse{
-                "", 6 * quantum - 1, {}, ResumableUploadResponse::kDone, {}});
+                "", ResumableUploadResponse::kDone, 6 * quantum, {}, {}});
           });
 
   RetryResumableUploadSession session(std::move(mock),
@@ -618,12 +598,12 @@ TEST_F(RetryResumableUploadSessionTest, HandleTransientPartialFailures) {
   StatusOr<ResumableUploadResponse> response;
   response = session.UploadChunk({{payload}});
   EXPECT_STATUS_OK(response);
-  EXPECT_EQ(3 * quantum - 1, response->last_committed_byte);
+  EXPECT_EQ(3 * quantum, response->committed_size.value_or(0));
 
   response =
       session.UploadFinalChunk({{payload_final}}, 6 * quantum, HashValues{});
   EXPECT_STATUS_OK(response);
-  EXPECT_EQ(6 * quantum - 1, response->last_committed_byte);
+  EXPECT_EQ(6 * quantum, response->committed_size.value_or(0));
 }
 
 /// @test Verify that erroneous server behavior (uncommitting data) is handled.
@@ -640,7 +620,7 @@ TEST_F(RetryResumableUploadSessionTest, UploadFinalChunkUncommitted) {
   EXPECT_CALL(*mock, next_expected_byte()).Times(2).WillRepeatedly(Return(0));
   EXPECT_CALL(*mock, UploadChunk).WillOnce([&](ConstBufferSequence const&) {
     return make_status_or(ResumableUploadResponse{
-        "", quantum - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, quantum, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte())
       .Times(3)
@@ -654,7 +634,7 @@ TEST_F(RetryResumableUploadSessionTest, UploadFinalChunkUncommitted) {
   // This should not happen, the last committed byte should not go back
   EXPECT_CALL(*mock, ResetSession()).WillOnce([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", 0, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, 0, {}, {}});
   });
   EXPECT_CALL(*mock, next_expected_byte()).Times(1).WillRepeatedly(Return(0));
 
@@ -670,7 +650,7 @@ TEST_F(RetryResumableUploadSessionTest, UploadFinalChunkUncommitted) {
 
   response = session.UploadChunk({{payload}});
   EXPECT_STATUS_OK(response);
-  EXPECT_EQ(quantum - 1, response->last_committed_byte);
+  EXPECT_EQ(quantum, response->committed_size.value_or(0));
 
   response = session.UploadFinalChunk({{payload}}, 2 * quantum, HashValues{});
   ASSERT_FALSE(response);
@@ -693,7 +673,7 @@ TEST_F(RetryResumableUploadSessionTest, ShortWriteRetryExhausted) {
   EXPECT_CALL(*mock, UploadChunk)
       .WillOnce([&](ConstBufferSequence const&) {
         return make_status_or(ResumableUploadResponse{
-            "", neb - 1, {}, ResumableUploadResponse::kInProgress, {}});
+            "", ResumableUploadResponse::kInProgress, neb, {}, {}});
       })
       .WillOnce([&](ConstBufferSequence const& p) {
         EXPECT_EQ(TotalBytes(p), payload.size() - neb);
@@ -710,7 +690,7 @@ TEST_F(RetryResumableUploadSessionTest, ShortWriteRetryExhausted) {
 
   EXPECT_CALL(*mock, ResetSession()).WillRepeatedly([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", neb - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, neb, {}, {}});
   });
 
   RetryResumableUploadSession session(std::move(mock),
@@ -736,17 +716,17 @@ TEST_F(RetryResumableUploadSessionTest, ShortWriteRetrySucceeds) {
       .WillOnce([&](ConstBufferSequence const&) {
         neb = quantum;
         return make_status_or(ResumableUploadResponse{
-            "", neb - 1, {}, ResumableUploadResponse::kInProgress, {}});
+            "", ResumableUploadResponse::kInProgress, neb, {}, {}});
       })
       .WillOnce([&](ConstBufferSequence const&) {
         neb = 2 * quantum;
         return make_status_or(ResumableUploadResponse{
-            "", neb - 1, {}, ResumableUploadResponse::kInProgress, {}});
+            "", ResumableUploadResponse::kInProgress, neb, {}, {}});
       });
 
   EXPECT_CALL(*mock, ResetSession()).WillRepeatedly([&]() {
     return make_status_or(ResumableUploadResponse{
-        "", neb - 1, {}, ResumableUploadResponse::kInProgress, {}});
+        "", ResumableUploadResponse::kInProgress, neb, {}, {}});
   });
 
   RetryResumableUploadSession session(std::move(mock),
@@ -756,7 +736,7 @@ TEST_F(RetryResumableUploadSessionTest, ShortWriteRetrySucceeds) {
   StatusOr<ResumableUploadResponse> response;
   response = session.UploadChunk({{payload}});
   ASSERT_STATUS_OK(response);
-  EXPECT_EQ(2 * quantum - 1, response->last_committed_byte);
+  EXPECT_EQ(2 * quantum, response->committed_size.value_or(0));
 }
 
 }  // namespace
diff --git a/google/cloud/storage/parallel_uploads_test.cc b/google/cloud/storage/parallel_uploads_test.cc
index b514b8afa72d9..76007920cf8db 100644
--- a/google/cloud/storage/parallel_uploads_test.cc
+++ b/google/cloud/storage/parallel_uploads_test.cc
@@ -172,7 +172,7 @@ class ParallelUploadTest
     EXPECT_CALL(res, next_expected_byte()).WillRepeatedly(Return(0));
     EXPECT_CALL(res, UploadChunk)
         .WillRepeatedly(Return(make_status_or(ResumableUploadResponse{
-            "fake-url", 0, {}, ResumableUploadResponse::kInProgress, {}})));
+            "fake-url", ResumableUploadResponse::kInProgress, 0, {}, {}})));
     EXPECT_CALL(res, UploadFinalChunk)
         .WillRepeatedly(Return(std::move(status)));
     AddNewExpectation(object_name, resumable_session_id);
@@ -204,18 +204,18 @@ class ParallelUploadTest
             EXPECT_THAT(content, ElementsAre(ConstBuffer(*expected_content)));
             return make_status_or(
                 ResumableUploadResponse{"fake-url",
+                                        ResumableUploadResponse::kDone,
                                         0,
                                         MockObject(object_name, generation),
-                                        ResumableUploadResponse::kDone,
                                         {}});
           });
     } else {
       EXPECT_CALL(res, UploadFinalChunk)
           .WillOnce(Return(make_status_or(
               ResumableUploadResponse{"fake-url",
+                                      ResumableUploadResponse::kDone,
                                       0,
                                       MockObject(object_name, generation),
-                                      ResumableUploadResponse::kDone,
                                       {}})));
     }
     AddNewExpectation(object_name, resumable_session_id);