Skip to content

fix(storage): enforce mutual exclusion between Close() and Finalize() in async writers #16211

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
kalragauri wants to merge 1 commit into
base: main
Choose a base branch
from
+162 −2
Draft

fix(storage): enforce mutual exclusion between Close() and Finalize() in async writers #16211

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion google/cloud/storage/internal/async/writer_connection_buffered.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,16 @@ class AsyncWriterConnectionBufferedState
future<StatusOr<google::storage::v2::Object>> Finalize(
storage::WritePayload const& p) {
std::unique_lock<std::mutex> lk(mu_);
if (!finalized_future_.valid()) {
return make_ready_future(StatusOr<google::storage::v2::Object>(
internal::FailedPreconditionError("Finalize() already called",
GCP_ERROR_INFO())));
}
if (!closed_future_.valid()) {
return make_ready_future(StatusOr<google::storage::v2::Object>(
internal::FailedPreconditionError(
"Finalize() cannot be called after Close()", GCP_ERROR_INFO())));
}
Comment on lines +118 to +127

@gemini-code-assist gemini-code-assist Bot Jun 29, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

There is a race condition here. finalized_future_ is only moved out at the very end of Finalize(). However, HandleNewData() releases the lock mu_ before the function returns. If another thread calls Finalize() or Close() concurrently after the lock is released but before the function returns, finalized_future_.valid() will still be true, allowing the check to be bypassed. This can lead to duplicate execution and returning an invalid/moved-from future.

Using the existing finalize_ and close_ boolean flags (which are checked and set under the lock) completely avoids this race condition and is much safer.

    if (finalize_) {
      return make_ready_future(StatusOr<google::storage::v2::Object>(
          internal::FailedPreconditionError("Finalize() already called",
                                            GCP_ERROR_INFO())));
    }
    if (close_) {
      return make_ready_future(StatusOr<google::storage::v2::Object>(
          internal::FailedPreconditionError(
              "Finalize() cannot be called after Close()", GCP_ERROR_INFO())));
    }

resend_buffer_.Append(WritePayloadImpl::GetImpl(p));
finalize_ = true;
HandleNewData(std::move(lk), true);
Expand All @@ -124,10 +134,14 @@ class AsyncWriterConnectionBufferedState

future<Status> Close(storage::WritePayload const& p) {
std::unique_lock<std::mutex> lk(mu_);
if (close_ || closed_promise_completed_) {
if (!closed_future_.valid()) {
return make_ready_future(internal::FailedPreconditionError(
"Close() already called", GCP_ERROR_INFO()));
}
if (!finalized_future_.valid()) {
return make_ready_future(internal::FailedPreconditionError(
"Close() cannot be called after Finalize()", GCP_ERROR_INFO()));
}
Comment on lines +137 to +144

@gemini-code-assist gemini-code-assist Bot Jun 29, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

There is a race condition here. closed_future_ is only moved out at the very end of Close(). However, the lock mu_ is released before the function returns. If another thread calls Close() or Finalize() concurrently after the lock is released but before the function returns, closed_future_.valid() will still be true, allowing the check to be bypassed.

Using the existing close_ and finalize_ boolean flags (which are checked and set under the lock) completely avoids this race condition and is much safer.

    if (close_) {
      return make_ready_future(internal::FailedPreconditionError(
          "Close() already called", GCP_ERROR_INFO()));
    }
    if (finalize_) {
      return make_ready_future(internal::FailedPreconditionError(
          "Close() cannot be called after Finalize()", GCP_ERROR_INFO()));
    }

resend_buffer_.Append(WritePayloadImpl::GetImpl(p));
close_ = true;
// Force flush to drain the buffer first.
Expand Down
60 changes: 60 additions & 0 deletions google/cloud/storage/internal/async/writer_connection_buffered_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1665,6 +1665,66 @@ TEST(WriteConnectionBuffered, ResetWriteOffsetOnResume) {
EXPECT_STATUS_OK(write.get());
}

TEST(WriteConnectionBuffered, CloseAfterFinalizeFails) {
auto mock = std::make_unique<MockAsyncWriterConnection>();
EXPECT_CALL(*mock, UploadId).WillRepeatedly(Return("test-upload-id"));
EXPECT_CALL(*mock, PersistedState)
.WillRepeatedly(Return(MakePersistedState(0)));
EXPECT_CALL(*mock, Finalize).WillOnce([](auto) {
return make_ready_future(make_status_or(google::storage::v2::Object{}));
});

MockFactory mock_factory;
auto connection = MakeWriterConnectionBuffered(
mock_factory.AsStdFunction(), std::move(mock), TestOptions());

auto finalize = connection->Finalize({});
auto close = connection->Close({});

EXPECT_STATUS_OK(finalize.get());
EXPECT_THAT(close.get(), StatusIs(StatusCode::kFailedPrecondition));
}

TEST(WriteConnectionBuffered, FinalizeAfterCloseFails) {
auto mock = std::make_unique<MockAsyncWriterConnection>();
EXPECT_CALL(*mock, UploadId).WillRepeatedly(Return("test-upload-id"));
EXPECT_CALL(*mock, PersistedState)
.WillRepeatedly(Return(MakePersistedState(0)));
EXPECT_CALL(*mock, Close).WillOnce([](auto) {
return make_ready_future(Status{});
});

MockFactory mock_factory;
auto connection = MakeWriterConnectionBuffered(
mock_factory.AsStdFunction(), std::move(mock), TestOptions());

auto close = connection->Close({});
auto finalize = connection->Finalize({});

EXPECT_STATUS_OK(close.get());
EXPECT_THAT(finalize.get(), StatusIs(StatusCode::kFailedPrecondition));
}

TEST(WriteConnectionBuffered, DuplicateFinalizeFails) {
auto mock = std::make_unique<MockAsyncWriterConnection>();
EXPECT_CALL(*mock, UploadId).WillRepeatedly(Return("test-upload-id"));
EXPECT_CALL(*mock, PersistedState)
.WillRepeatedly(Return(MakePersistedState(0)));
EXPECT_CALL(*mock, Finalize).WillOnce([](auto) {
return make_ready_future(make_status_or(google::storage::v2::Object{}));
});

MockFactory mock_factory;
auto connection = MakeWriterConnectionBuffered(
mock_factory.AsStdFunction(), std::move(mock), TestOptions());

auto finalize1 = connection->Finalize({});
auto finalize2 = connection->Finalize({});

EXPECT_STATUS_OK(finalize1.get());
EXPECT_THAT(finalize2.get(), StatusIs(StatusCode::kFailedPrecondition));
}

} // namespace
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
} // namespace storage_internal
Expand Down
16 changes: 15 additions & 1 deletion google/cloud/storage/internal/async/writer_connection_resumed.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,16 @@ class AsyncWriterConnectionResumedState
future<StatusOr<google::storage::v2::Object>> Finalize(
storage::WritePayload const& p) {
std::unique_lock<std::mutex> lk(mu_);
if (!finalized_future_.valid()) {
return make_ready_future(StatusOr<google::storage::v2::Object>(
internal::FailedPreconditionError("Finalize() already called",
GCP_ERROR_INFO())));
}
if (!closed_future_.valid()) {
return make_ready_future(StatusOr<google::storage::v2::Object>(
internal::FailedPreconditionError(
"Finalize() cannot be called after Close()", GCP_ERROR_INFO())));
}
Comment on lines +128 to +137

@gemini-code-assist gemini-code-assist Bot Jun 29, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

There is a race condition here. finalized_future_ is only moved out at the very end of Finalize(). However, HandleNewData() releases the lock mu_ before the function returns. If another thread calls Finalize() or Close() concurrently after the lock is released but before the function returns, finalized_future_.valid() will still be true, allowing the check to be bypassed. This can lead to duplicate execution and returning an invalid/moved-from future.

Using the existing finalize_ and close_ boolean flags (which are checked and set under the lock) completely avoids this race condition and is much safer.

    if (finalize_) {
      return make_ready_future(StatusOr<google::storage::v2::Object>(
          internal::FailedPreconditionError("Finalize() already called",
                                            GCP_ERROR_INFO())));
    }
    if (close_) {
      return make_ready_future(StatusOr<google::storage::v2::Object>(
          internal::FailedPreconditionError(
              "Finalize() cannot be called after Close()", GCP_ERROR_INFO())));
    }

resend_buffer_.Append(WritePayloadImpl::GetImpl(p));
finalize_ = true;
HandleNewData(std::move(lk));
Expand All @@ -148,10 +158,14 @@ class AsyncWriterConnectionResumedState

future<Status> Close(storage::WritePayload const& p) {
std::unique_lock<std::mutex> lk(mu_);
if (close_ || closed_promise_completed_) {
if (!closed_future_.valid()) {
return make_ready_future(internal::FailedPreconditionError(
"Close() already called", GCP_ERROR_INFO()));
}
if (!finalized_future_.valid()) {
return make_ready_future(internal::FailedPreconditionError(
"Close() cannot be called after Finalize()", GCP_ERROR_INFO()));
}
Comment on lines +161 to +168

@gemini-code-assist gemini-code-assist Bot Jun 29, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

There is a race condition here. closed_future_ is only moved out at the very end of Close(). However, the lock mu_ is released before the function returns. If another thread calls Close() or Finalize() concurrently after the lock is released but before the function returns, closed_future_.valid() will still be true, allowing the check to be bypassed.

Using the existing close_ and finalize_ boolean flags (which are checked and set under the lock) completely avoids this race condition and is much safer.

    if (close_) {
      return make_ready_future(internal::FailedPreconditionError(
          "Close() already called", GCP_ERROR_INFO()));
    }
    if (finalize_) {
      return make_ready_future(internal::FailedPreconditionError(
          "Close() cannot be called after Finalize()", GCP_ERROR_INFO()));
    }

resend_buffer_.Append(WritePayloadImpl::GetImpl(p));
close_ = true;
// Force flush to drain the buffer first.
Expand Down
72 changes: 72 additions & 0 deletions google/cloud/storage/internal/async/writer_connection_resumed_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1198,6 +1198,78 @@ TEST(WriterConnectionResumed, CloseFailsAndResumeSucceedsAndFinalized) {
EXPECT_STATUS_OK(close.get());
}

TEST(WriterConnectionResumed, CloseAfterFinalizeFails) {
auto mock = std::make_unique<MockAsyncWriterConnection>();
auto initial_request = google::storage::v2::BidiWriteObjectRequest{};
auto first_response = google::storage::v2::BidiWriteObjectResponse{};

EXPECT_CALL(*mock, UploadId).WillRepeatedly(Return("test-upload-id"));
EXPECT_CALL(*mock, PersistedState)
.WillRepeatedly(Return(MakePersistedState(0)));
EXPECT_CALL(*mock, Finalize).WillOnce([](auto) {
return make_ready_future(make_status_or(google::storage::v2::Object{}));
});

MockFactory mock_factory;
auto connection = MakeWriterConnectionResumed(
mock_factory.AsStdFunction(), std::move(mock), initial_request, nullptr,
first_response, Options{});

auto finalize = connection->Finalize({});
auto close = connection->Close({});

EXPECT_STATUS_OK(finalize.get());
EXPECT_THAT(close.get(), StatusIs(StatusCode::kFailedPrecondition));
}

TEST(WriterConnectionResumed, FinalizeAfterCloseFails) {
auto mock = std::make_unique<MockAsyncWriterConnection>();
auto initial_request = google::storage::v2::BidiWriteObjectRequest{};
auto first_response = google::storage::v2::BidiWriteObjectResponse{};

EXPECT_CALL(*mock, UploadId).WillRepeatedly(Return("test-upload-id"));
EXPECT_CALL(*mock, PersistedState)
.WillRepeatedly(Return(MakePersistedState(0)));
EXPECT_CALL(*mock, Close).WillOnce([](auto) {
return make_ready_future(Status{});
});

MockFactory mock_factory;
auto connection = MakeWriterConnectionResumed(
mock_factory.AsStdFunction(), std::move(mock), initial_request, nullptr,
first_response, Options{});

auto close = connection->Close({});
auto finalize = connection->Finalize({});

EXPECT_STATUS_OK(close.get());
EXPECT_THAT(finalize.get(), StatusIs(StatusCode::kFailedPrecondition));
}

TEST(WriterConnectionResumed, DuplicateFinalizeFails) {
auto mock = std::make_unique<MockAsyncWriterConnection>();
auto initial_request = google::storage::v2::BidiWriteObjectRequest{};
auto first_response = google::storage::v2::BidiWriteObjectResponse{};

EXPECT_CALL(*mock, UploadId).WillRepeatedly(Return("test-upload-id"));
EXPECT_CALL(*mock, PersistedState)
.WillRepeatedly(Return(MakePersistedState(0)));
EXPECT_CALL(*mock, Finalize).WillOnce([](auto) {
return make_ready_future(make_status_or(google::storage::v2::Object{}));
});

MockFactory mock_factory;
auto connection = MakeWriterConnectionResumed(
mock_factory.AsStdFunction(), std::move(mock), initial_request, nullptr,
first_response, Options{});

auto finalize1 = connection->Finalize({});
auto finalize2 = connection->Finalize({});

EXPECT_STATUS_OK(finalize1.get());
EXPECT_THAT(finalize2.get(), StatusIs(StatusCode::kFailedPrecondition));
}

} // namespace
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
} // namespace storage_internal
Expand Down
Loading