feat: Lazer publisher agent

Author: merolish

lazer_agent/Cargo.lock

@@ -0,0 +1,34 @@
+[package]
+name = "pyth-lazer-agent"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+pyth-lazer-publisher-sdk = "0.1.3"
+pyth-lazer-protocol = "0.7.2"
+
+anyhow = "1.0.98"
+bincode = { version = "2.0.1", features = ["serde"] }
+clap = { version = "4.5.32", features = ["derive"] }
+config = "0.15.11"
+derivative = "2.2.0"
+ed25519-dalek = { version = "2.1.1", features = ["rand_core"] }
+futures = "0.3.31"
+futures-util = "0.3.31"
+http = "1.3.1"
+http-body-util = "0.1.3"
+humantime-serde = "1.1.1"
+hyper = { version = "1.6.0", features = ["http1", "server"] }
+hyper-util = { version = "0.1.10", features = ["tokio"] }
+protobuf = "3.7.2"
+serde = { version = "1.0.219", features = ["derive"] }
+serde_json = "1.0.140"
+soketto = { version = "0.8.1", features = ["http"] }
+solana-keypair = "2.2.1"
+tokio = { version = "1.44.1", features = ["full"] }
+tokio-stream = "0.1.17"
+tokio-tungstenite = { version = "0.26.2", features = ["native-tls", "url"] }
+tokio-util = { version = "0.7.14", features = ["compat"] }
+tracing = "0.1.41"
+tracing-subscriber = { version = "0.3.19", features = ["env-filter", "json"] }
+url = { version = "2.5.4", features = ["serde"] }

lazer_agent/Cargo.toml

@@ -0,0 +1,5 @@
+relayer_urls = ["ws://localhost:1235/v1/transaction", "ws://localhost:1335/v1/transaction"]
+authorization_token = "token1"
+publish_keypair_path = "/path/to/solana/id.json"
+listen_address = "0.0.0.0:1234"
+publish_interval_duration = "50ms"

lazer_agent/config.toml

@@ -0,0 +1,37 @@
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::time::Duration;
+
+use config::{Environment, File};
+use derivative::Derivative;
+use pyth_lazer_protocol::router::PublisherId;
+use serde::Deserialize;
+use url::Url;
+
+#[derive(Deserialize, Derivative, Clone, PartialEq)]
+#[derivative(Debug)]
+pub struct Config {
+    pub listen_address: SocketAddr,
+    pub publisher_id: PublisherId,
+    pub relayer_urls: Vec<Url>,
+    pub authorization_token: String,
+    pub publish_keypair_path: PathBuf,
+    #[serde(with = "humantime_serde", default = "default_publish_interval")]
+    pub publish_interval_duration: Duration,
+}
+
+fn default_publish_interval() -> Duration {
+    Duration::from_millis(50)
+}
+
+pub fn load_config(config_path: String) -> anyhow::Result<Config> {
+    let config = config::Config::builder()
+        .add_source(File::with_name(&config_path))
+        .add_source(Environment::with_prefix("LAZER_AGENT").separator("__"))
+        .build()?
+        .try_deserialize()?;
+    Ok(config)
+}
+
+// Default capacity for all tokio mpsc channels that communicate between tasks.
+pub const CHANNEL_CAPACITY: usize = 1000;

lazer_agent/src/config.rs

@@ -0,0 +1,131 @@
+use anyhow::{Context, Result};
+use hyper::{Response, StatusCode, body::Bytes, server::conn::http1, service::service_fn};
+use hyper_util::rt::TokioIo;
+use soketto::{
+    BoxedError,
+    handshake::http::{Server, is_upgrade_request},
+};
+use std::{io, net::SocketAddr};
+use tokio::net::{TcpListener, TcpStream};
+use tracing::{debug, info, instrument, warn};
+
+use crate::{
+    config::Config,
+    lazer_publisher::LazerPublisher,
+    publisher_handle::{PublisherConnectionContext, handle_publisher},
+};
+
+type FullBody = http_body_util::Full<Bytes>;
+
+#[derive(Debug)]
+pub enum Request {
+    PublisherV1,
+    PublisherV2,
+}
+
+pub struct RelayerRequest(pub http::Request<hyper::body::Incoming>);
+
+const PUBLISHER_WS_URI: &str = "/v1/publisher";
+const PUBLISHER_WS_URI_V2: &str = "/v2/publisher";
+
+pub async fn run(config: Config, lazer_publisher: LazerPublisher) -> Result<()> {
+    let listener = TcpListener::bind(&config.listen_address).await?;
+    info!("listening on {:?}", &config.listen_address);
+
+    loop {
+        let stream_addr = listener.accept().await;
+        let config_clone = config.clone();
+        let lazer_publisher_clone = lazer_publisher.clone();
+        tokio::spawn(async {
+            if let Err(err) =
+                try_handle_connection(stream_addr, config_clone, lazer_publisher_clone).await
+            {
+                warn!("error while handling connection: {err:?}");
+            }
+        });
+    }
+}
+
+async fn try_handle_connection(
+    stream_addr: io::Result<(TcpStream, SocketAddr)>,
+    config: Config,
+    lazer_publisher: LazerPublisher,
+) -> Result<()> {
+    let (stream, remote_addr) = stream_addr?;
+    debug!("accepted connection from {}", remote_addr);
+    stream.set_nodelay(true)?;
+    http1::Builder::new()
+        .serve_connection(
+            TokioIo::new(stream),
+            service_fn(move |r| {
+                let request = RelayerRequest(r);
+                request_handler(
+                    request,
+                    remote_addr,
+                    config.clone(),
+                    lazer_publisher.clone(),
+                )
+            }),
+        )
+        .with_upgrades()
+        .await?;
+    Ok(())
+}
+
+#[instrument(skip_all, fields(component = "http_server", remote_addr = remote_addr.to_string()))]
+async fn request_handler(
+    request: RelayerRequest,
+    remote_addr: SocketAddr,
+    config: Config,
+    lazer_publisher: LazerPublisher,
+) -> Result<Response<FullBody>, BoxedError> {
+    let path = request.0.uri().path();
+
+    let request_type = match path {
+        PUBLISHER_WS_URI => Request::PublisherV1,
+        PUBLISHER_WS_URI_V2 => Request::PublisherV2,
+        _ => {
+            return Ok(Response::builder()
+                .status(StatusCode::NOT_FOUND)
+                .body(FullBody::from("not found"))
+                .context("builder failed")?);
+        }
+    };
+
+    if !is_upgrade_request(&request.0) {
+        return Ok(Response::builder()
+            .status(StatusCode::BAD_REQUEST)
+            .body(FullBody::from("bad request"))
+            .context("builder failed")?);
+    }
+
+    let mut server = Server::new();
+    match server.receive_request(&request.0) {
+        Ok(response) => {
+            info!("accepted connection from publisher");
+            match request_type {
+                Request::PublisherV1 | Request::PublisherV2 => {
+                    let publisher_connection_context = PublisherConnectionContext {
+                        request_type,
+                        publisher_id: config.publisher_id,
+                        _remote_addr: remote_addr,
+                    };
+                    tokio::spawn(handle_publisher(
+                        server,
+                        request.0,
+                        publisher_connection_context,
+                        lazer_publisher,
+                    ));
+                    Ok(response.map(|()| FullBody::default()))
+                }
+            }
+        }
+        Err(e) => {
+            warn!("Could not upgrade connection: {}", e);
+            Ok(Response::builder()
+                .status(StatusCode::INTERNAL_SERVER_ERROR)
+                .body(FullBody::from("internal server error"))
+                .context("builder failed")?)
+        }
+    }
+}