fix(storage): ensure thread-safe stream resumption in ObjectDescriptorImpl

[kalragauri]

This PR addresses the following lifecycle and resumption bugs in ObjectDescriptorImpl when handling bidirectional asynchronous read streams:

1. Concurrent Read() During Resumption:
   • Issue: Read() calls arriving while reconnection was in flight attempted to write to the dying stream, and OnResume() previously overwrote it->stream, dropping ranges accumulated in next_request.
   • Fix: Added resuming = true to ReadStream to block Flush() during reconnection. OnResume() preserves all accumulated next_request ranges across transient reconnection retries and flushes them to the new stream.
2. Stale Callback Hygiene & Safe Async Continuations:
   • Issue: Callbacks capturing std::list iterators could access invalidated iterators upon stream removal (triggering undefined behavior) or mistakenly trigger Finish on replacement streams.
   • Fix: Async callbacks now capture std::shared_ptr<ReadStream> and std::shared_ptr<OpenStream>. Under lock, callbacks resolve membership via MultiStreamManager::Find(read_stream) and validate stream identity (it->stream->stream == stream) before processing. Orphaned streams returned from delayed resume attempts are cancelled explicitly upon arrival.
3. Watchdog Stall Timeout vs. User Cancellation:
   • Issue: Stall watchdog cancellations (StatusCode::kCancelled) were treated as permanent failures by the default resume policy.
   • Fix: Translated watchdog cancellations to StatusCode::kUnavailable in IsResumable() to enable automatic reconnection. Guarded with cancelled_ to ensure explicit user cancellations terminate immediately without attempting resumption.
4. Resume Policy Notification on Successful Start:
   • Initialized resume_policy->OnStartSuccess() when streams are established in the constructor, MakeSubsequentStream(), and OnResume().

---

Sequence Diagrams

1. Concurrent Read() During Resumption

Before (Race Condition & Dropped Ranges)

sequenceDiagram
    autonumber
    participant App as User Thread
    participant Obj as ObjectDescriptorImpl
    participant S1 as Stream 1 (Failing)
    participant S2 as Stream 2 (New)

    S1-->>Obj: Stream failure (read/write error)
    Obj->>Obj: OnFinish() -> Resume() (async make_stream_)
    Note over Obj: Reconnection in flight...
    App->>Obj: Read(range_B)
    Obj->>S1: Flush() writes range_B to dying Stream 1 (Fails)
    S2-->>Obj: OnResume(Stream 2)
    Note over Obj,S2: it->stream replaced with new ReadStream<br/>next_request overwritten & range_B dropped

Loading

After (Safe Queueing & Automatic Flushing)

sequenceDiagram
    autonumber
    participant App as User Thread
    participant Obj as ObjectDescriptorImpl
    participant S1 as Stream 1 (Failing)
    participant S2 as Stream 2 (New)

    S1-->>Obj: Stream failure (read/write error)
    Obj->>Obj: Resume() sets resuming = true
    Note over Obj: Reconnection in flight...
    App->>Obj: Read(range_B)
    Obj->>Obj: Flush() skipped (resuming == true)<br/>range_B added to active_ranges & next_request
    S2-->>Obj: OnResume(Stream 2)
    Note over Obj,S2: Preserves next_request<br/>Replaces stream, sets resuming = false
    Obj->>S2: Flush() queued next_request (range_B)
    Obj->>S2: DoRead() starts reading

Loading

---

2. Stale Callback Hygiene

Before

sequenceDiagram
    autonumber
    participant S1 as Stream 1 (Old)
    participant Obj as ObjectDescriptorImpl
    participant S2 as Stream 2 (Active)

    S1-->>Obj: Read error triggers Resume()
    Obj->>S2: OnResume() establishes Stream 2 as active
    Note over S1,Obj: Delayed Write callback from Stream 1 arrives late (ok = false)
    S1-->>Obj: OnWrite(ok = false)
    Obj->>Obj: DoFinish() without checking stream identity
    Obj->>S2: stream_manager_->RemoveStream(it) -> Kills new Stream 2

Loading

After

sequenceDiagram
    autonumber
    participant S1 as Stream 1 (Old)
    participant Obj as ObjectDescriptorImpl
    participant S2 as Stream 2 (Active)

    S1-->>Obj: Read error triggers Resume()
    Obj->>S2: OnResume() establishes Stream 2 as active
    Note over S1,Obj: Delayed Write callback from Stream 1 arrives late (ok = false)
    S1-->>Obj: OnWrite(Stream 1, ok = false)
    Obj->>Obj: Check: stream_manager_->Find(read_stream) != End() && it->stream->stream == Stream 1?
    Note over Obj: Stream identity mismatch (active stream is Stream 2)<br/>Callback safely discarded
    Note over S2: Stream 2 continues healthy operation uninterrupted

Loading

[gemini-code-assist]

Code Review

This pull request introduces stream membership tracking via a new Contains method in MultiStreamManager and updates ObjectDescriptorImpl to pass stream pointers to asynchronous callbacks, allowing it to safely discard callbacks from stale or removed streams. However, the review feedback correctly identifies a critical issue with undefined behavior: passing and comparing potentially invalidated StreamIterator objects (such as in Contains and the asynchronous callbacks) violates C++ iterator lifetime rules. It is highly recommended to refactor the design to identify streams using std::shared_ptr and implement a safe lookup method like Find to resolve this safety concern.

[kalragauri]

/gemini review

[gemini-code-assist]

Code Review

This pull request refactors stream handling in ObjectDescriptorImpl to use std::shared_ptr<ReadStream> instead of StreamIterator in asynchronous callbacks, allowing the system to safely discard callbacks from stale or removed streams. It also maps StatusCode::kCancelled to StatusCode::kUnavailable during resumption and adds comprehensive unit tests. The review feedback correctly identifies several violations of the repository's style guide regarding the over-use of auto where explicit types are required for domain objects, primitives, iterators, and protobuf messages.

[codecov]

Codecov Report

❌ Patch coverage is 95.12195% with 24 lines in your changes missing coverage. Please review.
✅ Project coverage is 92.29%. Comparing base (6e0a08e) to head (08e9fd2).

Files with missing lines	Patch %	Lines
...rage/internal/async/object_descriptor_impl_test.cc	95.79%	14 Missing ⚠️
...d/storage/internal/async/object_descriptor_impl.cc	92.95%	10 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #16440      +/-   ##
==========================================
+ Coverage   92.28%   92.29%   +0.01%     
==========================================
  Files        2246     2246              
  Lines      212894   213315     +421     
==========================================
+ Hits       196474   196888     +414     
- Misses      16420    16427       +7     

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