feat(sdk): opt-in indefinite retries for read sessions

[devin-ai-integration]

Summary

read_session already resumes gap-free across retryable errors, but its retry budget (retry.max_attempts) is finite (reset only on a successful batch), and a clean stream end terminates the session. Long-lived consumers — e.g. the connector in feldera/feldera#5728 — therefore wrap their own infinite reconnect loop around the SDK. This adds an opt-in mode that makes the SDK own that loop:

let session = stream
    .read_session_with_config(input, ReadSessionConfig::new().with_retry_indefinitely(true))
    .await?;

Semantics when retry_indefinitely is enabled:

• Retryable errors retry indefinitely: when the backoff iterator is exhausted, the delay pins at RetryBackoff::max_base_delay() (rather than resetting, which would hammer at the min delay). Non-retryable errors still terminate.
• Clean stream ends reconnect too, but only for fully open-ended reads (no count/bytes/until/wait limit — wait counts as a limit since a clean end then means the wait budget expired). Reads with any limit terminate normally. A capped backoff step is applied before reconnecting so this can't hot-loop.
• Behavior with retry_indefinitely == false (and via the unchanged read_session) is exactly today's.

New API: types::ReadSessionConfig (#[non_exhaustive], builder-style), S2Stream::read_session_with_config, and a RetryBackoff::max_base_delay() getter.

Testing

just fmt, just clippy (-D warnings), just test (679 passed). Unit tests cover the config builder/default, open-endedness classification (incl. wait), and retry-past-budget pinning at the cap.

Part of a set of SDK improvements from reviewing feldera/feldera#5728 (see #653, #654, #655, #659).

Link to Devin session: https://app.devin.ai/sessions/6a1c8bd8dcd144128571f1cab5af0dcb
Requested by: @shikhar

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring

[greptile-apps]

Greptile Summary

This PR adds opt-in auto_reconnect support to ReadSession via a new ReadSessionConfig type and S2Stream::read_session_with_config. When enabled, retryable errors pin to max_base_delay after the normal backoff budget is exhausted (rather than giving up), and clean stream ends trigger a capped-backoff reconnect for fully open-ended reads.

• ReadSessionConfig is #[non_exhaustive] with a builder API; the existing read_session delegates to read_session_with_config(input, Default::default()), so backward compatibility is preserved.
• RetryBackoff gains a max_base_delay() getter used to pin the delay when the budget is exhausted; is_open_ended guards the clean-end reconnect so reads with any count/byte/timestamp/wait limit still terminate normally.

Confidence Score: 4/5

Safe to merge. The auto_reconnect path is well-isolated behind a false-by-default flag, existing behavior is unchanged, and the clean-end reconnect is correctly guarded by is_open_ended.

The implementation is correct and well-tested. The only gaps are a missing bytes assertion in the is_open_ended unit test and a log field that could be improved for debuggability in pinned-delay mode.

sdk/src/session/read.rs — the open_ended_read_has_no_limits test and the retry_delay log statement.

Important Files Changed

Filename	Overview
sdk/src/session/read.rs	Core auto_reconnect logic added to the read session stream loop: handles clean-end reconnect for open-ended reads and pins retries at max_base_delay when the backoff budget is exhausted. Logic is correct; missing test case for bytes limit in is_open_ended and a potentially misleading log field when the delay is pinned.
sdk/src/retry.rs	Adds max_base_delay() getter to RetryBackoff with a covering unit test. Straightforward field accessor, no concerns.
sdk/src/types.rs	Introduces ReadSessionConfig with #[non_exhaustive] and builder-style API. Well-structured, default is safe (auto_reconnect=false), and tested.
sdk/src/ops.rs	Adds read_session_with_config and delegates the existing read_session to it with a default config. Backward-compatible change, no concerns.

Sequence Diagram

sequenceDiagram
    participant C as Consumer
    participant RS as ReadSession
    participant SI as session_inner
    participant S as S2 Server

    C->>RS: "read_session_with_config(input, auto_reconnect=true)"
    RS->>SI: session_inner(start, end)
    SI->>S: open read stream
    S-->>SI: stream batches
    SI-->>RS: Ok(batches)
    RS->>RS: retry_backoff.reset()

    loop Stream batch loop
        RS->>SI: batches.next()
        S-->>SI: batch
        SI-->>RS: Some(Ok(batch))
        RS->>RS: update start, retry_backoff.reset()
        RS-->>C: yield ReadBatch

        alt Retryable error
            S-->>SI: gRPC error
            SI-->>RS: Some(Err(retryable))
            RS->>RS: "retry_delay() -> backoff or max_base_delay"
            RS-->>C: yield ReadUpdate::behind()
            RS->>RS: sleep(backoff), reconnect
        else Clean stream end (open-ended only)
            S-->>SI: stream closed
            SI-->>RS: None
            RS->>RS: "is_open_ended? -> next_retry_delay"
            RS->>RS: sleep(backoff), reconnect
        else Clean stream end (with limits)
            RS-->>C: stream ends normally
        end
    end

Loading

Comments Outside Diff (1)

1. sdk/src/session/read.rs, line 465-472 (link)

   When the backoff is exhausted and auto_reconnect pins to max_base_delay, the log still emits num_retries_remaining = 0. In production traces this looks identical to "retries run out and gave up", making it hard to tell the session is still alive and operating in pinned-delay mode.

   Prompt To Fix With AI

   This is a comment left during a code review.
   Path: sdk/src/session/read.rs
   Line: 465-472
   
   Comment:
   When the backoff is exhausted and auto_reconnect pins to `max_base_delay`, the log still emits `num_retries_remaining = 0`. In production traces this looks identical to "retries run out and gave up", making it hard to tell the session is still alive and operating in pinned-delay mode.
   
   
   
   How can I resolve this? If you propose a fix, please make it concise.

   Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

Prompt To Fix All With AI

Fix the following 2 code review issues. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 2
sdk/src/session/read.rs:759-765
The `open_ended_read_has_no_limits` test verifies `count`, `until`, and `wait` fields but skips `bytes`. If `is_open_ended` ever dropped the `bytes` check by mistake, auto_reconnect would silently fire for byte-limited reads and loop forever — no test would catch it.

```suggestion
        assert!(!is_open_ended(&ReadEnd {
            count: None,
            bytes: None,
            until: None,
            wait: Some(1),
        }));
        assert!(!is_open_ended(&ReadEnd {
            count: None,
            bytes: Some(1),
            until: None,
            wait: None,
        }));
    }
```

### Issue 2 of 2
sdk/src/session/read.rs:465-472
When the backoff is exhausted and auto_reconnect pins to `max_base_delay`, the log still emits `num_retries_remaining = 0`. In production traces this looks identical to "retries run out and gave up", making it hard to tell the session is still alive and operating in pinned-delay mode.

```suggestion
    if let Some(backoff) = next_retry_delay(backoffs, auto_reconnect) {
        debug!(
            %err,
            ?backoff,
            num_retries_remaining = backoffs.remaining(),
            pinned_to_max_delay = backoffs.is_exhausted() && auto_reconnect,
            "retrying read session"
        );
        Some(backoff)
```

_{Reviews (1): Last reviewed commit: "fix(sdk): honor wait limits for read rec..." | Re-trigger Greptile}

[greptile-apps]

The open_ended_read_has_no_limits test verifies count, until, and wait fields but skips bytes. If is_open_ended ever dropped the bytes check by mistake, auto_reconnect would silently fire for byte-limited reads and loop forever — no test would catch it.

Suggested change

	assert!(!is_open_ended(&ReadEnd {
	count: None,
	bytes: None,
	until: None,
	wait: Some(1),
	}));
	}
	assert!(!is_open_ended(&ReadEnd {
	count: None,
	bytes: None,
	until: None,
	wait: Some(1),
	}));
	assert!(!is_open_ended(&ReadEnd {
	count: None,
	bytes: Some(1),
	until: None,
	wait: None,
	}));
	}

Prompt To Fix With AI

This is a comment left during a code review.
Path: sdk/src/session/read.rs
Line: 759-765

Comment:
The `open_ended_read_has_no_limits` test verifies `count`, `until`, and `wait` fields but skips `bytes`. If `is_open_ended` ever dropped the `bytes` check by mistake, auto_reconnect would silently fire for byte-limited reads and loop forever — no test would catch it.

```suggestion
        assert!(!is_open_ended(&ReadEnd {
            count: None,
            bytes: None,
            until: None,
            wait: Some(1),
        }));
        assert!(!is_open_ended(&ReadEnd {
            count: None,
            bytes: Some(1),
            until: None,
            wait: None,
        }));
    }
```

How can I resolve this? If you propose a fix, please make it concise.

[shikhar]

auto_reconnect happens already. need a better name to signal this is indefinite retries

[devin-ai-integration]

Renaming to retry_indefinitely (with_retry_indefinitely(bool)) — will push shortly.