fix: schedule reconnect after parse error during streaming (#134)

## Summary

Every error path in `ReconnectingRequest::poll_next` during the
`Connected` state sets state to `WaitingToReconnect` before returning
the error — except for parser errors propagated through
`process_bytes(...)?`. After a parse error, the state stayed at
`Connected` and the next poll continued draining the (already-broken)
response body before finally hitting end-of-body and emitting a spurious
`UnexpectedEof` before the reconnect actually kicked in.

This PR schedules a reconnect on parse errors when reconnect is enabled,
matching the pattern used by every other error path. The error is still
returned to the caller; the only behavioral change is that the next poll
cleanly transitions to reconnect rather than draining the broken body.

**Before:** `Connected`, `Err(InvalidLine)`, `Err(UnexpectedEof)`,
`Connected`, …
**After:**  `Connected`, `Err(InvalidLine)`, `Connected`, …

## Context

Surfaced while investigating
[launchdarkly/rust-server-sdk#116](launchdarkly/rust-server-sdk#116)
(`StreamingDataSource` silently shuts down on stream errors). That
report has two contributing bugs; this PR addresses the
rust-eventsource-client side. The rust-server-sdk side will be fixed in
a follow-up PR — together they restore the reconnection contract: the
eventsource-client owns reconnection, the SDK keeps polling and trusts
it.

Tracked in
[SDK-2345](https://launchdarkly.atlassian.net/browse/SDK-2345).

## Test plan

- [x] New test `parser_error_schedules_reconnect_immediately`
(`eventsource-client/src/client.rs`) — verifies the next stream item
after a parse error is `Connected` from the reconnect, not a spurious
`UnexpectedEof`.
- [x] `cargo test` — 60 lib tests + 2 doc tests pass; no regressions.

[SDK-2345]:
https://launchdarkly.atlassian.net/browse/SDK-2345?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes streaming error-handling semantics by scheduling reconnects on
parser failures, which can affect how downstream consumers observe
errors and reconnect timing. Scope is small and covered by a new
integration-style test, but touches core stream state transitions.
> 
> **Overview**
> **Fixes reconnection behavior after SSE parse errors during
streaming.** When `EventParser::process_bytes` fails in the `Connected`
state, the client now (when `ReconnectOptions::reconnect` is enabled)
transitions to `WaitingToReconnect` before yielding the parse error,
avoiding continued draining of a broken response body and the resulting
follow-on errors.
> 
> Updates `Client::stream` docs to clarify that stream errors are
non-terminal and that the stream only ends on `Poll::Ready(None)`, and
adds a new test (`parser_error_schedules_reconnect_immediately`)
asserting the sequence `Connected -> Err(InvalidLine) -> Connected`.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
ddea87d. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: kinyoklion

eventsource-client/src/client.rs

@@ -199,10 +199,20 @@ impl<T: HttpTransport> Client for ClientImpl<T> {
     /// Connect to the server and begin consuming the stream. Produces a
     /// [`Stream`] of [`Event`](crate::Event)s wrapped in [`Result`].
     ///
-    /// Do not use the stream after it returned an error!
+    /// Errors yielded by the stream are not terminal: keep polling.
+    /// When [`ReconnectOptions::reconnect`] is enabled (the default),
+    /// the stream schedules a reconnect on retryable errors and the
+    /// next poll resumes from a fresh connection.
     ///
-    /// After the first successful connection, the stream will
-    /// reconnect for retryable errors.
+    /// The stream is exhausted only when [`Stream::poll_next`] returns
+    /// [`Poll::Ready(None)`]. That happens when the underlying state
+    /// machine reaches `StreamClosed` (e.g. a redirect-limit overrun,
+    /// a malformed `Location` header, or an error during initial
+    /// connection while [`ReconnectOptions::retry_initial`] is
+    /// disabled), or after any error when reconnect is disabled.
+    ///
+    /// [`Poll::Ready(None)`]: std::task::Poll::Ready
+    /// [`Stream::poll_next`]: futures::Stream::poll_next
     fn stream(&self) -> BoxStream<Result<SSE>> {
         Box::pin(ReconnectingRequest::new(
             Arc::clone(&self.transport),
@@ -495,7 +505,21 @@ impl<T: HttpTransport> Stream for ReconnectingRequest<T> {
                 },
                 StateProj::Connected(mut body) => match ready!(body.as_mut().poll_next(cx)) {
                     Some(Ok(result)) => {
-                        this.event_parser.process_bytes(result)?;
+                        if let Err(e) = this.event_parser.process_bytes(result) {
+                            // A parse error means the current response body is
+                            // unusable; schedule a reconnect to abandon it.
+                            if self.props.reconnect_opts.reconnect {
+                                let duration = self
+                                    .as_mut()
+                                    .project()
+                                    .retry_strategy
+                                    .next_delay(Instant::now());
+                                self.as_mut().project().state.set(State::WaitingToReconnect(
+                                    delay(duration, "reconnecting"),
+                                ));
+                            }
+                            return Poll::Ready(Some(Err(e)));
+                        }
                         continue;
                     }
                     Some(Err(e)) => {
@@ -712,4 +736,67 @@ mod tests {
 
         initial_connection(&mock_server.url(), retry_initial, expected).await;
     }
+
+    // When a parse error happens during streaming and reconnect is
+    // enabled, the next stream item should be a fresh `Connected` from
+    // the reconnect, not another error from continuing to drain the
+    // broken response body.
+    #[cfg(feature = "hyper")]
+    #[tokio::test(flavor = "multi_thread")]
+    async fn parser_error_schedules_reconnect_immediately() {
+        use crate::{Client, ClientBuilder, Error, ReconnectOptionsBuilder, SSE};
+        use futures::StreamExt;
+        use launchdarkly_sdk_transport::HyperTransport;
+
+        let mut server = mockito::Server::new_async().await;
+        let _mock = server
+            .mock("GET", "/")
+            .with_status(200)
+            .with_body(b"\xff\xfe:bad\n\n".as_ref())
+            .create_async()
+            .await;
+
+        let transport = HyperTransport::new().expect("failed to build transport");
+        let client = ClientBuilder::for_url(&server.url())
+            .unwrap()
+            .reconnect(
+                ReconnectOptionsBuilder::new(true)
+                    .delay(Duration::from_millis(10))
+                    .delay_max(Duration::from_millis(10))
+                    .retry_initial(true)
+                    .build(),
+            )
+            .build_with_transport(transport);
+
+        let mut stream = client.stream();
+
+        // Expected order: Connected, parse error, Connected (reconnect).
+        let mut items = Vec::new();
+        tokio::time::timeout(Duration::from_secs(2), async {
+            while items.len() < 3 {
+                match stream.next().await {
+                    Some(item) => items.push(item),
+                    None => break,
+                }
+            }
+        })
+        .await
+        .expect("timed out waiting for parse error and reconnect");
+
+        assert!(
+            matches!(items.first(), Some(Ok(SSE::Connected(_)))),
+            "expected initial Connected, got {:?}",
+            items.first()
+        );
+        assert!(
+            matches!(items.get(1), Some(Err(Error::InvalidLine(_)))),
+            "expected InvalidLine error after first connection, got {:?}",
+            items.get(1)
+        );
+        assert!(
+            matches!(items.get(2), Some(Ok(SSE::Connected(_)))),
+            "expected reconnect (Connected) immediately after parse error, got {:?}",
+            items.get(2)
+        );
+    }
 }