fix: deplete read session wait budget on retry

[infiniteregrets]

No description provided.

[greptile-apps]

Greptile Overview

Greptile Summary

• Tracks a baseline ReadEnd.wait and an Instant to treat wait as a total budget across retries rather than per-attempt.
• Before each new read_session attempt (initial and subsequent retries), rewrites end.wait to the remaining seconds (saturating at 0).
• Leaves the per-batch heartbeat timeout behavior unchanged (timeout(20s, batches.next())).

Confidence Score: 4/5

• This PR is likely safe to merge with low risk; the change is small and localized.
• The change cleanly implements a cumulative wait budget across retries via Instant + saturating subtraction. No other code paths are affected. I could not run tests/compile in this environment (no Rust toolchain available), so the score is not a 5/5.
• src/session/read.rs (verify intended wait semantics and units; consider adding a regression test if the project has coverage here)

Important Files Changed

Filename	Overview
src/session/read.rs	Adjusts ReadEnd.wait on each retry by tracking elapsed time since session start and subtracting it from the initial wait budget via remaining_wait().

Sequence Diagram

sequenceDiagram
    participant Caller
    participant ReadSession as read_session()
    participant Retry as can_retry()/RetryBackoff
    participant Inner as session_inner()
    participant Client as BasinClient.read_session()
    participant Net as Stream (batches.next())

    Caller->>ReadSession: read_session(client, name, start, end)
    Note over ReadSession: baseline_wait = end.wait
    Note over ReadSession: start_time = Instant::now()

    loop Create initial session (with retries)
        ReadSession->>ReadSession: end.wait = remaining_wait(baseline_wait, start_time)
        ReadSession->>Inner: session_inner(client, name, start, end)
        Inner->>Client: read_session(name, start, end)
        Client-->>Inner: batches stream
        Inner->>Net: timeout(20s, batches.next())
        alt heartbeat timeout / retryable error
            Inner-->>ReadSession: Err(ReadSessionError)
            ReadSession->>Retry: can_retry(err)
            alt retry
                Retry-->>ReadSession: sleep(backoff)
            else no retry
                Retry-->>ReadSession: false
                ReadSession-->>Caller: return Err
            end
        else ok
            Inner-->>ReadSession: Ok(stream)
        end
    end

    ReadSession-->>Caller: Streaming<ReadBatch>

    loop Consume stream
        Caller->>ReadSession: next()
        ReadSession->>Inner: (underlying stream) timeout(20s, batches.next())
        alt stream item ok
            Inner-->>ReadSession: Ok(batch)
            ReadSession-->>Caller: Ok(batch)
        else stream item err
            Inner-->>ReadSession: Err(err)
            Note over ReadSession: batches = None
            ReadSession->>Retry: can_retry(err)
            alt retry
                Retry-->>ReadSession: sleep(backoff)
                ReadSession->>ReadSession: end.wait = remaining_wait(baseline_wait, start_time)
                ReadSession->>Inner: session_inner(...)
            else no retry
                ReadSession-->>Caller: Err(err)
            end
        else stream ends
            Inner-->>ReadSession: None
            ReadSession-->>Caller: None
        end
    end

Loading

[quettabit]

LGTM!