fix(storage): enable fast resume for rapid object write streams by including write handle (#15762)

Author: shubham-up-47

google/cloud/storage/internal/async/writer_connection_resumed.cc

@@ -334,6 +334,12 @@ class AsyncWriterConnectionResumedState
       append_object_spec.set_bucket(spec.bucket());
       append_object_spec.set_object(spec.object());
     }
+    // Include write_handle to enable fast resume instead of slow
+    // takeover. Without handle, server performs full state validation.
+    if (first_response_.has_write_handle()) {
+      *append_object_spec.mutable_write_handle() =
+          first_response_.write_handle();
+    }
     append_object_spec.set_generation(first_response_.resource().generation());
     ApplyWriteRedirectErrors(append_object_spec, std::move(proto_status));
 

google/cloud/storage/internal/async/writer_connection_resumed_test.cc

@@ -268,6 +268,7 @@ TEST(WriteConnectionResumed, ResumeUsesWriteObjectSpecFromInitialRequest) {
       "test-object");
 
   google::storage::v2::BidiWriteObjectResponse first_response;
+  first_response.mutable_write_handle();
   first_response.mutable_resource()->set_generation(12345);
 
   EXPECT_CALL(*mock, PersistedState)
@@ -318,6 +319,7 @@ TEST(WriteConnectionResumed, ResumeUsesWriteObjectSpecFromInitialRequest) {
 
   EXPECT_FALSE(captured_request.has_write_object_spec());
   EXPECT_TRUE(captured_request.has_append_object_spec());
+  EXPECT_TRUE(captured_request.append_object_spec().has_write_handle());
   EXPECT_EQ(captured_request.append_object_spec().generation(), 12345);
   EXPECT_EQ(captured_request.append_object_spec().object(), "test-object");
   EXPECT_EQ(captured_request.append_object_spec().bucket(),
@@ -383,6 +385,7 @@ TEST(WriteConnectionResumed, ResumeUsesAppendObjectSpecFromInitialRequest) {
 
   EXPECT_FALSE(captured_request.has_write_object_spec());
   EXPECT_TRUE(captured_request.has_append_object_spec());
+  EXPECT_FALSE(captured_request.append_object_spec().has_write_handle());
   EXPECT_EQ(captured_request.append_object_spec().generation(), 12345);
   EXPECT_EQ(captured_request.append_object_spec().object(), "test-object");
   EXPECT_EQ(captured_request.append_object_spec().bucket(),