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.25328% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 92.39%. Comparing base (68190fe) to head (8112944).

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.39%            
========================================
  Files        2265     2265            
  Lines      219107   219328   +221     
========================================
+ Hits       202416   202641   +225     
+ Misses      16691    16687     -4     

☔ 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.

[v-pratap]

LGTM — the lifetime fixes in AsyncWriter and the buffered/resumed writer connections cleanly prevent member access after synchronous callback destruction, and the new unit tests pass under ASan. (Note: clang-tidy-storage-pr failed after 30m00s in CI and may need a re-run or log check before merging.)