pyth-stream mvp #1

ayazabbas · 2024-10-18T14:28:46Z

No description provided.

thmzlt · 2024-10-20T17:45:43Z

A few overall points:

From reading the code, I see that it is possible to configure the services using env. variables. We should specify in the README what the config options are. Also, we've had some discussions in the past about defining configuration for services. I tend to favor env. variables and CLI args rather than config files as it makes it easier to pass config in Kubernetes, but it is mostly personal.
When you connect to the websocket server and send an invalid message, you get no feedback from the server unless you enable a more verbose log level and read the server logs. We should send back something to the client, even if it is "I couldn't understand your message".
Better yet - would it be possible to pass the list of price ids in the websocket URL so the clients are subscribed automatically when they connect? That would make the websocket connection stateless and simplify the implementation (i.e. it is now one-way).

thmzlt · 2024-10-20T14:17:34Z

src/pyth_reader.rs

+        };
+
+        if price_account.agg.status == PriceStatus::Trading
+            && (update.context.slot - price_account.agg.pub_slot)


Maybe add a comment for context here? Looks like you want to discard account updates when the update slot is more than X slots ahead of the price aggregate slot. What is a invalid price update in this context?

I took this from benchmarks/rust-streaming https://github.com/pyth-network/benchmarks/blob/main/rust-streaming/src/pythnet.rs#L56 - I don't really have full context on this myself.

cc @ali-bahjati

Ideally we want to send price updates whenever the aggregate changes but sometimes the accounts can change without the aggregate changing; for example it happens when there is an update but min_pub is not reached and it can happen for feeds in closed market hours.

thmzlt · 2024-10-20T14:24:43Z

src/pyth_reader.rs

+                price_feed: PriceFeed {
+                    id: update.value.pubkey.to_string(),
+                    price: PriceInfo {
+                        price: price_account.agg.price.to_string(),


I'd check with Ali if it makes sense to send the price and expo components separately or formatted as a regular decimal.

I think @ayazabbas made it compatible with Hermes. I'd say let's keep it as our partners are trying it out with Hermes.

thmzlt · 2024-10-20T17:43:23Z

src/websocket_server.rs

+    Ok(jetstream)
+}
+
+async fn handle_connection(stream: TcpStream, clients: Clients) {


This function is difficult to parse. In addition to the websocket sender/receiver this is defining two additional channels (tx/rx and outgoing_tx/outgoing_rx), I am assuming it is so you can send both price updates and responses from the client through the ws_sender. Is that correct? Is it not possible to clone the ws_sender and use it in more than one place?

The idea behind multiple channels is to separate message generation from message sending. The broadcast channel is used for distributing price updates to all interested clients, and the mspc channel is used to queue outgoing messages for a specific client.

I agree that this can be simplified but I'm not sure if cloning the ws_sender is safe/recommended. I think an easy simplification might be to eliminate the broadcast channel and just have the nats message handler forward price updates directly to the outgoing_tx channel for each client. Gonna try this.

P.S. Using the broadcast channel approach would be more efficient when you have many clients subscribed to the same price feeds but in this case the difference is probably negligible.

I think if ws_sender is clone able, then we should do it to make things easier.

ali-behjati

LGTM. I left some comments that are mostly minor and language specific.

ali-behjati · 2024-10-21T11:52:49Z

.github/workflows/build-and-push-image.yaml

I think we can separate it to two workflows: one for ci, and one for image build&push (and add manual dispatch there)

ali-behjati · 2024-10-21T11:53:12Z

Cargo.toml

+tokio-tungstenite = "0.24.0"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+zeroize = { version = "1.7", default-features = false }


do we need this after the patch below?

ali-behjati · 2024-10-21T11:58:16Z

src/pyth_reader.rs

+
+#[derive(Debug, Deserialize)]
+struct NatsConfig {
+    url: String,


nit: you can use url::Url or nats ServerAddr here to make sure right argument is passed upon parsing the config.

ali-behjati · 2024-10-21T11:58:26Z

src/pyth_reader.rs

+    http_addr: String,
+    websocket_addr: String,


Same as above

ali-behjati · 2024-10-21T11:58:33Z

src/pyth_reader.rs

+    program_key: String,
+    mapping_key: String,


Please use Pubkey here.

ali-behjati · 2024-10-21T12:26:42Z

src/websocket_server.rs

+
+    info!(client_addr = %addr, "New WebSocket connection established");
+
+    let (tx, mut rx) = broadcast::channel(100);


nit: let's extract 100 and make it a constant CHANNEL_SIZE

ali-behjati · 2024-10-21T12:27:42Z

src/websocket_server.rs

+#[derive(Debug, Serialize, Deserialize)]
+struct ClientMessage {
+    #[serde(rename = "type")]
+    message_type: String,


we can have this as enum and act based on that. right? (like this)

ali-behjati · 2024-10-21T12:29:57Z

src/websocket_server.rs

+    let jetstream_clone = jetstream.clone();
+    let clients_clone = clients.clone();
+    tokio::spawn(async move {
+        handle_nats_messages(jetstream_clone, clients_clone).await;


one nit thing here is that if these threads fail the app will enter a dead state and we might not want it. So either reconnection is good here or panic there and set the panic behaviour to abort the process (see this)

ali-behjati · 2024-10-21T12:30:50Z

src/websocket_server.rs

+    Ok(jetstream)
+}
+
+async fn handle_connection(stream: TcpStream, clients: Clients) {


I think if ws_sender is clone able, then we should do it to make things easier.

ali-behjati · 2024-10-21T12:31:49Z

src/websocket_server.rs

+    let outgoing_tx_clone = outgoing_tx.clone();
+    let broadcast_task = tokio::spawn(async move {
+        while let Ok(msg) = rx.recv().await {
+            if let Err(e) = outgoing_tx_clone.send(Message::Text(msg)).await {


for future: as we have dozens of prices coming at each slot, it might be better to feed them first, and then flush once we've sent all of them.

add configuration options to readme use specific types in structs use expect or context instead of unwrap remove use of mapping account spawn separate threads for blocking operations move duplicate nats code to utils move binary entrypoints under src/bin add reconnection logic and abort process on panic

pyth stream mvp

2c8f129

ayazabbas requested review from ali-behjati and thmzlt October 18, 2024 14:28

add github actions workflow

432476e

thmzlt approved these changes Oct 20, 2024

View reviewed changes

ali-behjati approved these changes Oct 21, 2024

View reviewed changes

ayazabbas added 2 commits October 22, 2024 11:21

remove redundant import

a2b83fb

ayazabbas merged commit e850536 into main Oct 22, 2024
2 checks passed

ayazabbas deleted the pyth-stream branch October 22, 2024 10:56


		info!(client_addr = %addr, "New WebSocket connection established");

		let (tx, mut rx) = broadcast::channel(100);

pyth-stream mvp #1

pyth-stream mvp #1

Uh oh!

Conversation

ayazabbas commented Oct 18, 2024

Uh oh!

thmzlt commented Oct 20, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ayazabbas Oct 20, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ali-behjati left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

ayazabbas Oct 20, 2024 •

edited

Loading