feat(lazer): add resilient client in rust #2859
Conversation
|
The latest updates on your projects. Learn more about Vercel for Git ↗︎
|
lazer/sdk/rust/client/src/client.rs
Outdated
| /// * `access_token` - The access token for authentication | ||
| /// * `num_connections` - The number of WebSocket connections to maintain | ||
| pub fn new( | ||
| endpoints: Vec<String>, |
| let (sender, receiver) = mpsc::channel::<AnyResponse>(CHANNEL_CAPACITY); | ||
|
|
||
| for i in 0..self.num_connections { | ||
| let endpoint = self.endpoints[i % self.endpoints.len()].clone(); |
There was a problem hiding this comment.
It will panic if endpoints is empty. We should probably check for that in the constructor.
lazer/sdk/rust/client/src/client.rs
Outdated
| } | ||
|
|
||
| pub async fn start(&mut self) -> Result<mpsc::Receiver<AnyResponse>> { | ||
| let (sender, receiver) = mpsc::channel::<AnyResponse>(CHANNEL_CAPACITY); |
There was a problem hiding this comment.
Channel capacity should be configurable by the caller.
There was a problem hiding this comment.
Done. Internal channel don't need to be configurable right?
lazer/sdk/rust/client/src/client.rs
Outdated
| } | ||
|
|
||
| let streams: Vec<_> = self.receivers.drain(..).map(ReceiverStream::new).collect(); | ||
| let mut merged_stream = stream::select_all(streams); |
There was a problem hiding this comment.
I think it's easier and more efficient to use a single channel here.
There was a problem hiding this comment.
right. fixed.
| if last_failure_time.elapsed() > BACKOFF_RESET_DURATION { | ||
| self.backoff.reset(); | ||
| } |
There was a problem hiding this comment.
This feels like a hack. I think we should reset the backoff object in start() when we successfully connect and subscribe.
There was a problem hiding this comment.
I want the behavior to be a little different than that. For example, if client can connect and subscribe, but get disconnected immediately after we should not reset.
There was a problem hiding this comment.
This code means that when the endpoint is down, rather than doing a proper exponential backoff, it will start spamming the endpoint again every 10 seconds. I think what you want can be better achieved with this:
pub async fn start(...) {
//...
let stream = ws_connection.start().await?;
let connected_at = Instant::now();
let mut num_responses = 0;
let mut confirmed_healthy = false;
//...
Ok(response) => {
if !confirmed_healthy {
num_responses += 1;
if num_responses >= CONFIRM_HEALTHY_NUM_RESPONSES &&
connected_at.elapsed() > CONFIRM_HEALTHY_DELAY {
confirmed_healthy = true;
self.backoff.reset();
}
}
sender
.send(response)
.await
.context("Failed to send response")?;
}
//...
}
There was a problem hiding this comment.
It only happens when max_interval is greater than 10 seconds. Otherwise it will try connecting every max_interval.
I think the best solution is to set BACKOFF_RESET_DURATION to be 2*max_interval.
There was a problem hiding this comment.
I fixed it another way. Now we check the duration of the connection and if was connected for more than BACKOFF_RESET_DURATION and timout + second it was connected successfully and we reset the backoff.
| request: SubscribeRequest, | ||
| ) -> Result<()> { | ||
| self.subscriptions.push(request.clone()); | ||
| return ws_connection.subscribe(request).await; |
| request.subscription_id | ||
| ); | ||
| } | ||
| return ws_connection.unsubscribe(request).await; |
| endpoints: Vec<String>, | ||
| access_token: String, | ||
| num_connections: usize, | ||
| backoff: ExponentialBackoff, |
There was a problem hiding this comment.
this seems to be part of your public interface. you need to reexport it to downstream consumers. Other thing is to have a wrapper that has what you need (because i see you are not supporting everything there)
There was a problem hiding this comment.
added a builder wrapper.
There was a problem hiding this comment.
A builder would also be nice for the PythLazerClient itself. You can move channel_capacity from start() argument to the builder. The builder can also provide default values for everything except access_token.
| }; | ||
|
|
||
| #[derive(Debug)] | ||
| pub struct PythLazerExponentialBackoffBuilder { |
There was a problem hiding this comment.
document here as well? maybe default as well?
There was a problem hiding this comment.
The reason I didn't implement default is that we need to have access token. Default doesn't make sense since it will be in an invalid state.
| .with_max_interval(self.max_interval) | ||
| .with_max_elapsed_time(None) | ||
| .build() | ||
| } |
There was a problem hiding this comment.
i'm fine with this but generally my take was reexposing what we use in the API and we still use ExponentialBackoff from the backoff crate. now that you have a builder it makes sense that you have your own wrapper type for the api of the lazer client. later when you convert it you can remove the following error as well: bail!("max_elapsed_time is not supported in Pyth Lazer client");
There was a problem hiding this comment.
yeah it's a good improvement. I'll add this + docs in another PR to unblock the progress of monitor for now.
Summary
This PR updates the rust client to be a resilient client with multiple connections.
Rationale
We need this both for consumers and for monitor service.
How has this been tested?
Manually ran and tested the connections.