feat(llc)!: bound and authenticate a connection attempt

[xsahil03x]

Submit a pull request

Linear: FLU-

Github Issue: #

CLA

• I have signed the Stream CLA (required).
• The code changes follow best practices
• Code changes are tested (add some information if not applicable)

Stacked on #159 — review that one first. This PR's base is feat/token-manager-user-switching, so its diff is the WebSocket layer plus the credential path that feeds it.

Description of the pull request

StreamWebSocketClient treated opening the socket as the end of connecting: it called onConnectionEstablished, discarded whatever that returned, and then waited indefinitely for a health check to arrive. Everything below follows from unpicking that, in the guest flow that motivated it.

options → optionsBuilder

typedef WebSocketOptionsBuilder = WebSocketOptions Function();

Called once per attempt. The options carry values that change over a client's lifetime — the stream-auth-type a connection needs depends on the token it will present, and a client that switches users presents a different one — so a single instance built at construction time describes only the first attempt.

onConnectionEstablished → onAuthenticate

typedef WsRequestSender = Result<void> Function(WsRequest request);
typedef WebSocketAuthenticator = Future<void> Function(WsRequestSender send, StreamApiError? previousError);

Renamed for what it is called for and when: the socket is open, the state is Authenticating, and the connection is not usable until credentials have been sent.

The signature change is the substantive part. As a void Function() it could not report a failure to send those credentials, and it silently accepted an async callback whose future was then discarded — so a token that failed to load left the connection sitting in Authenticating until something else closed it. It is now awaited, and throwing is how it says the credentials did not go out, whether because sending failed or because it chose not to send them. The connection is then closed with the new AuthenticationFailed source carrying the cause, and is not reconnected.

It is handed a WsRequestSender rather than the client because it runs while the connection is being established — the client cannot hand out an interface that implies the connection is usable. The sender belongs to the attempt it was given to and fails once that attempt is no longer the one in flight, so an authenticator still awaiting a token for an abandoned attempt cannot send it over the connection that replaced it, nor close that one as AuthenticationFailed.

It also fixes the doc example, which never compiled: onConnectionEstablished: () { client.send(...) } is referenced_before_declaration.

previousError: telling an authenticator why the last attempt was refused

The second parameter is the error the server closed the previous attempt with. Without it, an authenticator that caches its token has no way to know the token it is about to present is the one just refused, so it offers the same one for the life of the client.

Set only for the attempt directly after a refusal. Cleared once a connection is established, and once the caller disconnects — a caller that takes connecting back may sign a different user in, and a refusal recorded against the user before them says nothing about the credentials they will present. An attempt abandoned before its authenticator finished leaves the refusal behind for the attempt that replaces it, which has yet to answer it.

connectTimeout was dead API

Declared on WebSocketOptions and never read. It now bounds the whole attempt rather than just opening the socket, because the attempt that hangs is precisely the one that opens and never receives its first health check — and nothing else watches Authenticating. Abandoning it reports the new ConnectTimeout source.

The field is no longer nullable. Its doc claimed null meant "the platform default", which was never consulted, so null meant no timeout at all; it now defaults to WebSocketOptions.defaultConnectTimeout, 30 seconds.

What reconnects, and what does not

ConnectTimeout is eligible for automatic reconnection — a handshake that was slow once may not be next time. AuthenticationFailed is not: credentials that never went out will not go out on a retry either.

Eligibility is necessary but not sufficient, and this is the part worth reading twice. ConnectionRecoveryHandler recovers only a connection that was established. So a connection that times out on its way back is reconnected, while a first connection that times out is reported through connectionState and left there — making another attempt belongs to whoever called connect. DisconnectionSource.isReconnectable is new and is the whole of isAutomaticReconnectionEnabled; the established-connection gate sits on top of it.

Two reconnection rules were also simply broken: the deliberate-close check compared a Stream error code against 1000 when it needed the WebSocket close code, and isClientError compared the Stream error code against 400..499, a range it never falls in. Neither had ever matched. A rate limit is now reconnectable too, since it clears on its own.

A health check arriving while disconnecting

Pre-existing. A pong was handled the same whether the connection was live or already on its way down, so it set the state back to Connected — which replaced the Disconnecting source. A deliberate UserInitiated disconnect could therefore close as ServerInitiated and be automatically reconnected, the opposite of what the caller asked for. Pongs are now ignored once the state is Disconnecting or Disconnected.

A handshake that failed read as a deliberate close

Found while reviewing the above, and the sharpest bug here. connect closed the socket of a failed attempt using the engine's default close code — 1000 — and a closure with that code is never reconnected. So an established connection that dropped, then failed to upgrade on its retry, had its recovery cancelled and stayed down for good, indistinguishable from a clean server close. The original error was discarded entirely.

The attempt now reports the error that actually failed it before letting go of its socket, which leaves the closure reconnectable and carrying its cause. Pinned by a test that walks the whole loop: established → drop → retry that cannot upgrade → still retrying.

ConnectUserDetailsRequest.fromUser

Here because an authenticator builds its auth frame from the client's User, and every product was mapping the same four fields by hand. role and teams are deliberately left out — the server assigns both and ignores them from a client. name comes from originalName, so a user with no name does not have their id sent as one, which the hand-rolled mappings got wrong.

StreamWebSocketClient.dispose

The client is now Disposable: dispose closes the connection along with events and connectionState. connect throws a StateError afterwards, in release builds as well as debug — it previously asserted and then returned, so a release build opened a socket nothing could observe or close.

The credential path feeding all of this

• AuthInterceptor extends Interceptor rather than QueuedInterceptor. A queue slot is freed only once a handler completes, so the retry sent from onError waited behind the request still holding one and neither finished. TokenManager serialises the token loads, which is the part that needs it. The interceptor now retries a refused request at most once, expires only the token that request actually carried, clones a multipart body whose streams the refused attempt consumed, and refuses to retry a request signed for a user the manager has since been pointed away from — that retry would have performed one user's request as another.
• StreamApiError — isTokenExpiredError now means code 40 alone. The codes another token cannot fix (41–43) and a wrong API key (2) are the new isInvalidTokenError. This is the distinction that decides whether reconnecting is worth anything.
• DioException.apiError — reads the Stream error from a body Dio decoded or handed over as a string. A token-expired response sent without a JSON content type was previously never retried, while the same body was already read as a string when the error was surfaced to the caller.
• Result — getOrElse, getOrDefault, recover and recoverCatching return the result's own type and no longer take a type parameter. The old signatures used an unchecked data as R, which threw at runtime for any R that was not T.

Migration

// before
StreamWebSocketClient(
  options: WebSocketOptions(url: url),
  onConnectionEstablished: () => client.send(authRequest),
);

// after
StreamWebSocketClient(
  optionsBuilder: () => WebSocketOptions(url: url),
  onAuthenticate: (send, previousError) async => send(authRequest).getOrThrow(),
);

Breaking, and stream-video-flutter has two call sites that will need updating when it bumps — it pins stream_core: ^0.4.0, so nothing there breaks today:

• coordinator_ws.dart:37 — options: → optionsBuilder:
• coordinator_ws.dart:41 — onConnectionEstablished: _authenticateUser → onAuthenticate:, and _authenticateUser (:115) has to change shape from Future<void> Function() to Future<void> Function(WsRequestSender, StreamApiError?)
• coordinator_ws.dart:116 — reads _client.options.url for a log line; the options field is gone
• sfu_ws.dart:67 — options: → optionsBuilder:
• sfu_ws.dart:85 — String get url => _client.options.url; is a public getter on SfuWs, so this one surfaces in video's own API

stream-feeds-flutter pins core to a git ref and its branch already implements the two-parameter authenticator — it moves when the ref moves.

Also removed: WebSocketEngineException.stopErrorCode, replaced by CloseCode.normalClosure.

Behaviour change, not just API

connectTimeout was declared and never read, so every connection previously waited indefinitely for its first health check. It is now abandoned after 30s. Video's two clients pass no timeout today and so inherit that default. A backend slow to send the first health check goes from "connects eventually" to "drops after 30s" — reconnected if the connection had been established before, and reported to the caller if this was its first attempt.

Test plan

• dart test in packages/stream_core — 544 pass, up from 383 on the base branch
• dart analyze — clean
• dart format — clean

StreamWebSocketClient had no test file at all before this, so most of the +161 is new coverage rather than adjusted coverage; the AuthInterceptor suite is the one that was rewritten rather than added to. Most of it drives the real client through a fake socket — the engine, codec, authentication handler, health monitor and recovery handler are all the production ones, so a test drives the client the way an app does and the fake server answers what the client actually sent.

Highlights:

• optionsBuilder — called for every attempt, not once per client
• onAuthenticate — called once the socket is open while Authenticating, once per attempt, handed a sender that reaches the socket; a throw closes the connection as AuthenticationFailed and is not retried; a sender belonging to an abandoned attempt fails rather than reaching the connection that replaced it
• previousError — handed to the attempt after a refusal and no later one, survives a closure the server did not explain, forgotten once a connection is established and once the caller disconnects, and the guest case end to end: refused expired token → fresh token → connected
• connect timeout (9, on fake_async) — abandons an attempt that never becomes connected, one whose socket never opens, and one whose authenticator never returns; armed again for a later attempt; honours a timeout given in the options; does not fire once established; does not replace the source of a closure or disconnect that came first
• health check while disconnecting (2) — does not report the connection as established again, and leaves the disconnection source intact once the socket closes
• recovery — a first attempt that never connected is not retried; one that dropped after being established is; and a retry that fails its handshake keeps the recovery going rather than ending it
• reconnection rules — ConnectTimeout and UnHealthyConnection reconnect, AuthenticationFailed and UserInitiated do not, close code 1000 does not, an expired token and a rate limit do, an invalid signature does not
• plus closeReason uniqueness across all six sources, dispose, the AuthInterceptor suite rebuilt around one fake backend, and Result's new signatures

Adds fake_async as a dev dependency, used for the timeout tests so a 30s timer does not cost 30s of wall clock.

Also drops test/query/filter_test.dart (16 tests, 287 lines). Thirteen asserted that a constructor stored its arguments, which the serialisation test beside it already covers — a wrong field, operator or value shows up as wrong JSON. The other three re-ran assertions their parts already make. Line coverage of filter.dart is unchanged at 100%.

Screenshots / Videos

n/a — no UI changes.

🤖 Generated with Claude Code

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 01d85174-f0ee-4a92-a170-03a63027036c

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[codecov]

Codecov Report

❌ Patch coverage is 97.14286% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 64.44%. Comparing base (863c881) to head (9c666e0).

Files with missing lines	Patch %	Lines
...stream_core/lib/src/api/stream_core_dio_error.dart	66.66%	2 Missing ⚠️
...lib/src/ws/client/web_socket_connection_state.dart	88.88%	2 Missing ⚠️

Additional details and impacted files

@@                          Coverage Diff                          @@
##           feat/token-manager-user-switching     #160      +/-   ##
=====================================================================
+ Coverage                              60.32%   64.44%   +4.11%     
=====================================================================
  Files                                    192      193       +1     
  Lines                                   7834     7920      +86     
=====================================================================
+ Hits                                    4726     5104     +378     
+ Misses                                  3108     2816     -292     

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

[renefloor]

Reviewed the WS lifecycle changes with the branch checked out; suite is green (387 pass) and dart analyze --fatal-infos is clean. I probed the new lifecycle paths and four of them reproduced — details inline, ordered by how much I'd worry about them.

Worth fixing before merge

1. An authenticator that throws (rather than returning a failed Result) produces an unhandled async error and leaves the connection stuck in Authenticating. This is the shape almost everyone will write, because TokenManager.getToken() throws. See the comment on _authenticate.
2. The new connect timer can overwrite a ServerInitiated source and flip isAutomaticReconnectionEnabled from true to false — the same bug class this PR fixes for late pongs, but the guard only went into the pong path. See the comment on disconnect.

Worth a deliberate decision

3. Whether ConnectTimeout and AuthenticationFailed should really block reconnection, given UnHealthyConnection doesn't. See the comment on isAutomaticReconnectionEnabled.

Pre-existing, but this PR makes it sharper

connect() still doesn't guard Disconnecting, and onClose now cancels the connect timer — so a stale close from an old socket can disarm the new attempt's timeout. Comment on connect has the trace.

What I liked

optionsBuilder is the right call and correctly motivated — stream-auth-type depends on the token an attempt will present, so a single instance built at construction can only ever describe the first attempt. Handing the authenticator a WsSender instead of the client is the right boundary, and it fixes a doc example that genuinely never compiled. connectTimeout was dead API and bounding the whole handshake rather than just the socket open is the correct scope, since nothing watched Authenticating. The pong-while-disconnecting fix is a real bug with a regression test that pins the reconnect consequence rather than just the state. And 25 tests on a class that had none — with fake_async for the timers instead of real waits — is the right way to land this.

One small thing not worth its own comment: connect()'s doc still says it "completes when the connection attempt finishes". It resolves once the socket opens — before authentication, well before Connected. Given this PR is precisely about not treating an open socket as a finished attempt, that sentence should probably say so.

[renefloor]

An authenticator that throws instead of returning a failed Result isn't handled here, and since _authenticate() is unawaited the error escapes:

Bad state: token load failed
  stream_web_socket_client.dart 203  StreamWebSocketClient._authenticate
  stream_web_socket_client.dart 199  StreamWebSocketClient.onOpen
state: Authenticating()

Unhandled async error, and the connection sits in Authenticating until the 15s timeout — then reports ConnectTimeout, which carries no error, so the real cause is lost.

This isn't hypothetical. The natural authenticator for the flow this stack exists to serve is:

onAuthenticate: (send) async => send(ConnectRequest(token: await manager.getToken())),

and getToken() throws (ClientException) on an unconfigured/reset manager or a failing provider — that's #159's own contract. The typedef asks for a Result, but the one authenticator everybody will write can't honour it without an explicit try/catch.

Could we route a throw to the same place a failed Result goes?

final result = await Result.guard(() => onAuthenticate!.call(send));

so the cause lands in AuthenticationFailed(error: ...) instead of being lost to a timeout.

[renefloor]

Cancelling the timer here covers the case the test does not replace the source of a disconnect that came first pins — disconnect() ran first, so the timer never fires. But the reverse direction isn't covered, because disconnect() only early-returns on Disconnected, not Disconnecting.

onError (line 239) sets Disconnecting(ServerInitiated) and does not cancel the timer. If onClose doesn't follow promptly:

after onError:          Disconnecting(ServerInitiated(...))
after timeout elapsed:  Disconnecting(ConnectTimeout())
after onClose:          Disconnected(ConnectTimeout())    autoReconnect = false

Without the timer that last state is Disconnected(ServerInitiated) with autoReconnect = **true** (web_socket_connection_state.dart:109-114). So a recoverable socket error becomes a permanent disconnect — which is the same failure this PR fixes for late pongs, just via the timer instead of a pong.

Same shape with lower stakes: if the timeout fires and the authenticator then returns a failure, the source is overwritten (ConnectTimeout → AuthenticationFailed). The engine guards the second close, and both sources are non-reconnectable, so that one is only misreporting:

after timeout:           Disconnecting(ConnectTimeout())      engine closes = 1
after late auth failure: Disconnecting(AuthenticationFailed)  engine closes = 1

Both fall out of one fix: have disconnect() return early (or at least not replace source) when the state is already Disconnecting. That seems better than adding _cancelConnectTimeout() to each new call site as they appear.

[renefloor]

Pre-existing, but the new timer gives it a sharper edge: connect() guards Connecting/Authenticating/Connected but not Disconnecting, so it proceeds while an old socket is still closing.

after disconnect:                      Disconnecting
after connect() during disconnecting:  Authenticating      <- new socket opened
after the OLD socket's onClose:        Disconnected(ServerInitiated)

The stale close kills the new attempt — and because onClose now also calls _cancelConnectTimeout(), it disarms the new attempt's timer. If that socket then opens, we're back in Authenticating with nothing watching it, which is exactly the state the timeout was added for.

Adding Disconnecting to the early-return above would close it. Happy for it to be a follow-up since it predates this PR.

[renefloor]

I'd like to push back on both of these, and it's the PR description's own reasoning that makes me want to.

ConnectTimeout — UnHealthyConnection (no pong on an established connection) is retryable, but a missing first pong isn't. That's the same failure mode, usually a bad network, at a different moment. It also compounds connectTimeout going from "null = no timeout" to a mandatory 15s: a customer whose backend is slow to send the first health check now gets connections dropped where they previously worked, and not retried. A spurious timeout being permanent is a rough edge.

AuthenticationFailed — the description argues "credentials the server rejected fail the same way on a retry", but this source never means the server rejected anything. It fires when the client couldn't load or send credentials. send() failing because the socket died between onOpen and the send is exactly the transient case. A genuine "the server said no" arrives later, as an error frame.

One line either way, so mostly I'd like it decided deliberately rather than by analogy with UserInitiated.

[xsahil03x]

Decided as you asked, and I took your reading on one of the two.

ConnectTimeout is now reconnectable (web_socket_connection_state.dart:120). Your argument is the one that settles it: a first health check that never arrives is the same failure as one that stops arriving, and UnHealthyConnection already retries that.

Worth being explicit about what that does to the customer you raised, since it is a three-way change rather than a two-way one. For a backend slow to send the first health check: before this PR the connection hung indefinitely; with the timeout but non-reconnectable it dropped and stayed down; now it drops and reconnects with the recovery handler's backoff. So the flag turns "stays down" into "retries with backoff" rather than back into "connects eventually" — the 15s bound still applies. If that is the wrong trade for a slow backend, the lever is connectTimeout itself rather than the source, and it is per-attempt now.

AuthenticationFailed stays non-reconnectable, with your distinction written into the code as a comment: it is not the server refusing the credentials — that arrives as an error frame — but the client failing to load or send them, and it will fail the same way on a retry. The transient sub-case you named (the socket dying between onOpen and the send) is real, but it is also covered: that path closes the socket, and the resulting closure is reported by the engine rather than by this source. If it turns out to matter in practice, splitting the source is a smaller change than reversing this default.

Also fixed from your other comments: the throwing authenticator now goes through runSafely so the cause lands in AuthenticationFailed instead of escaping (:245), and disconnect early-returns when the connection is already Disconnecting, so the timer can no longer replace a ServerInitiated source — both with regression tests. The connectTimeout behaviour change is now in the changelog and the PR body, and connect's doc no longer claims its future completes when the attempt finishes.

[renefloor]

Agreed that the old null doc was a lie (nothing consulted a platform default, so null meant no timeout at all), and that a default is better than dead API.

Worth calling out in the changelog as a behaviour change though, not just an API one: every existing connection now gets abandoned after 15s if the first health check hasn't arrived, where before it waited indefinitely. Paired with ConnectTimeout not being reconnectable, a slow-first-pong backend goes from "connects eventually" to "drops and stays down".

[renefloor]

Nit: no doc comment on new public API. The class has none either so it's consistent as-is — but the two decisions worth writing down are the ones a caller can't infer: role/teams omitted because the server assigns them, and name coming from originalName so a user with no name doesn't get their id sent as one. That last part is a good catch; every product was getting it wrong by hand.