fix(storage): avoid use-after-free when callback destroys async writer

[kalragauri]

In AsyncWriterConnectionBufferedState and AsyncWriterConnectionResumedState, Finalize() and Close() read a member variable after HandleNewData() returns:

HandleNewData(std::move(lk));
return std::move(finalized_future_);  // `closed_future_` in `Close()`

Problem

• When the write loop is idle, HandleNewData() starts it synchronously (HandleNewData() -> StartWriting() -> WriteLoop() -> FlushStep()).
• FlushStep() calls impl_->Flush(...) and attaches a .then() continuation that locks WeakFromThis() into a local shared_ptr (self) and calls self->OnFlush(...).
• If impl_->Flush(...) completes inline with a permanent error, OnFlush() calls SetError(), which unlocks mu_ and satisfies all pending Write() and Flush() promises (flush_handlers_ and pending_flush_promises_).
• Because promise::set_value() runs attached .then() callbacks synchronously on the calling thread, a user callback on one of those pending futures can destroy the connection (or AsyncWriter) before SetError() returns.
• When OnFlush() then returns and the FlushStep() continuation finishes (still inside HandleNewData()) its local self shared_ptr is destroyed. If no other reference to the state remains, *this is freed before HandleNewData() returns, and reading finalized_future_ or closed_future_ is a heap-use-after-free.

AsyncWriter has the same issue one layer up in Write(), Finalize(), Flush(), and Close():

return impl_->Finalize(std::move(payload)).then([impl = impl_](auto f)

In C++17, the call impl_->Finalize(...) is sequenced before the lambda capture [impl = impl_]. If a synchronous callback invoked during that call destroys the AsyncWriter, capturing impl_ reads a member of the destroyed AsyncWriter.

Fix

This PR updates both layers by moving or copying the required member into a local variable before calling HandleNewData() or impl_->..., so no member of *this is accessed after the call. In AsyncWriter::Close(), moving impl_ before calling impl->Close(...) also means any callback that re-enters the writer during Close() observes a closed stream, matching the documented contract that the writer is not usable after Close().

[gemini-code-assist]

Code Review

This pull request prevents potential use-after-free issues in AsyncWriter and its connection implementations by ensuring member variables and futures are copied or moved before invoking operations that might synchronously destroy the object. It also adds comprehensive unit tests to verify lifetime safety. The review feedback highlights several violations of the repository style guide's type deduction rules, requesting that auto be replaced with explicit types for connection pointers and complex future types to avoid obscuring domain and return types.

[codecov]

Codecov Report

❌ Patch coverage is 98.27586% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 92.37%. Comparing base (78b7a16) to head (7083f58).

Files with missing lines	Patch %	Lines
.../internal/async/writer_connection_buffered_test.cc	97.56%	2 Missing ⚠️
...e/internal/async/writer_connection_resumed_test.cc	97.64%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #16555      +/-   ##
==========================================
- Coverage   92.38%   92.37%   -0.01%     
==========================================
  Files        2265     2265              
  Lines      219046   219266     +220     
==========================================
+ Hits       202355   202546     +191     
- Misses      16691    16720      +29     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.