fix(storage): preserve flush state across concurrent flushes in async writers

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

@@ -540,21 +540,23 @@ class AsyncWriterConnectionBufferedState
 
   void SetFlushed(std::unique_lock<std::mutex> lk, Status const& result) {
     if (!result.ok()) return SetError(std::move(lk), std::move(result));
-    flush_ = false;  // Reset flush flag; WriteLoop may set it again.
     // Do NOT reset finalize_ or finalizing_ here.
     auto handlers = ClearHandlers(lk);
     // Dequeue the promise corresponding to an explicit Flush() call, if any.
     if (pending_flush_promises_.empty()) {
       // This can happen if SetError cleared the queue first, or if this
       // flush was triggered internally by buffer size (not by an explicit
       // Flush() call) and thus has no promise in the queue.
+      flush_ = false;
       lk.unlock();
       for (auto& h : handlers) h->Execute(Status{});
       return;
     }
     auto flushed = std::move(pending_flush_promises_.front());
     pending_flush_promises_.pop_front();
-
+    if (pending_flush_promises_.empty()) {
+      flush_ = false;
+    }
     lk.unlock();  // Unlock only once before notifying
     // Notify handlers and the specific flush promise *after* releasing the
     // lock.

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

@@ -1376,12 +1376,17 @@ TEST(WriteConnectionBuffered, MultipleConcurrentFlushesAreQueued) {
       return sequencer.PushBack("Query1").then(
           [](auto) { return make_status_or(static_cast<std::int64_t>(4096)); });
     });
-    EXPECT_CALL(*mock, Write(expected_write_size(8192)))
+    EXPECT_CALL(*mock, Flush(expected_write_size(8192)))
         .Times(1)
         .WillOnce([&](auto) {
-          return sequencer.PushBack("Write").then(
+          return sequencer.PushBack("Flush2").then(
               [](auto) { return Status{}; });
         });
+    EXPECT_CALL(*mock, Query).WillOnce([&]() {
+      return sequencer.PushBack("Query2").then([](auto) {
+        return make_status_or(static_cast<std::int64_t>(4096 + 8192));
+      });
+    });
   }
 
   MockFactory mock_factory;
@@ -1409,10 +1414,17 @@ TEST(WriteConnectionBuffered, MultipleConcurrentFlushesAreQueued) {
   // After the Query, the first Flush future should be completed.
   EXPECT_STATUS_OK(f1.get());
 
-  // Satisfy the Write call for the remaining data.
+  // Satisfy the second Flush call.
   next = sequencer.PopFrontWithName();
-  EXPECT_EQ(next.second, "Write");
+  EXPECT_EQ(next.second, "Flush2");
+  next.first.set_value(true);
+
+  // The second Flush is also followed by a Query.
+  next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Query2");
   next.first.set_value(true);
+
+  EXPECT_STATUS_OK(f2.get());
 }
 
 TEST(WriteConnectionBuffered, FinalizeWithEmptyPayloadIsImmediate) {

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

@@ -578,21 +578,23 @@ class AsyncWriterConnectionResumedState
 
   void SetFlushed(std::unique_lock<std::mutex> lk, Status const& result) {
     if (!result.ok()) return SetError(std::move(lk), std::move(result));
-    flush_ = false;  // Reset flush flag; WriteLoop may set it again.
     // Do NOT reset finalize_ or finalizing_ here.
     auto handlers = ClearHandlers(lk);
     // Dequeue the promise corresponding to an explicit Flush() call, if any.
     if (pending_flush_promises_.empty()) {
       // This can happen if SetError cleared the queue first, or if this
       // flush was triggered internally by buffer size (not by an explicit
       // Flush() call) and thus has no promise in the queue.
+      flush_ = false;
       lk.unlock();
       for (auto& h : handlers) h->Execute(Status{});
       return;
     }
     auto flushed = std::move(pending_flush_promises_.front());
     pending_flush_promises_.pop_front();
-
+    if (pending_flush_promises_.empty()) {
+      flush_ = false;
+    }
     lk.unlock();  // Unlock only once before notifying
     // Notify handlers and the specific flush promise *after* releasing the
     // lock.