-
Notifications
You must be signed in to change notification settings - Fork 20
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
node-support: event-export-sidecar: create event export sidecar
- Loading branch information
Showing
13 changed files
with
366 additions
and
35 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| [package] | ||
| name = "event-export-sidecar" | ||
| version = "0.1.0" | ||
| edition = "2021" | ||
|
|
||
| [dependencies] | ||
| # === Async + Runtime === # | ||
| tokio = { workspace = true, features = ["full"] } | ||
|
|
||
| # === Networking Dependencies === # | ||
| reqwest = { version = "0.11", features = ["json"] } | ||
|
|
||
| # === Workspace Dependencies === # | ||
| common = { path = "../../common" } | ||
| config = { path = "../../config" } | ||
| external-api = { path = "../../external-api", features = ["auth"] } | ||
| job-types = { path = "../../workers/job-types" } | ||
| event-manager = { path = "../../workers/event-manager" } | ||
|
|
||
| # === Misc Dependencies === # | ||
| url = "2.4" | ||
| clap = { version = "4", features = ["derive"] } | ||
| tracing = { workspace = true } | ||
| serde = { workspace = true } | ||
| serde_json = { workspace = true } | ||
| eyre = { workspace = true } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| //! A managed Unix listener that removes the socket file when dropped | ||
| use std::{fs, io, path::Path}; | ||
|
|
||
| use event_manager::manager::extract_unix_socket_path; | ||
| use eyre::{eyre, Error}; | ||
| use job_types::event_manager::RelayerEvent; | ||
| use tokio::net::{UnixListener, UnixStream}; | ||
| use tracing::{error, info, warn}; | ||
| use url::Url; | ||
|
|
||
| use crate::hse_client::HseClient; | ||
|
|
||
| // ------------- | ||
| // | Constants | | ||
| // ------------- | ||
|
|
||
| /// The maximum message size to read from the event export socket | ||
| const MAX_MESSAGE_SIZE: usize = 1024 * 1024; // 1MB | ||
|
|
||
| // ---------------- | ||
| // | Event Socket | | ||
| // ---------------- | ||
|
|
||
| /// A managed Unix socket that listens for events on a given path | ||
| /// and submits them to the historical state engine. | ||
| /// | ||
| /// The socket file is removed when the socket is dropped. | ||
| pub struct EventSocket { | ||
| /// The underlying Unix socket | ||
| socket: UnixStream, | ||
|
|
||
| /// The path to the Unix socket | ||
| path: String, | ||
|
|
||
| /// The historical state engine client | ||
| hse_client: HseClient, | ||
| } | ||
|
|
||
| impl EventSocket { | ||
| /// Creates a new event socket from the given URL | ||
| pub async fn new(url: &Url, hse_client: HseClient) -> Result<Self, Error> { | ||
| let path = extract_unix_socket_path(url)?; | ||
| let socket = Self::establish_socket_connection(&path).await?; | ||
| Ok(Self { socket, path, hse_client }) | ||
| } | ||
|
|
||
| /// Sets up a Unix socket listening on the given path | ||
| /// and awaits a single connection on it | ||
| async fn establish_socket_connection(path: &str) -> Result<UnixStream, Error> { | ||
| let listener = UnixListener::bind(Path::new(path))?; | ||
|
|
||
| // We only expect one connection, so we can just block on it | ||
| info!("Waiting for event export socket connection..."); | ||
| match listener.accept().await { | ||
| Ok((socket, _)) => Ok(socket), | ||
| Err(e) => Err(eyre!("error accepting Unix socket connection: {e}")), | ||
| } | ||
| } | ||
|
|
||
| /// Listens for events on the socket and submits them to the historical | ||
| /// state engine | ||
| pub async fn listen_for_events(&self) -> Result<(), Error> { | ||
| loop { | ||
| // Wait for the socket to be readable | ||
| self.socket.readable().await?; | ||
|
|
||
| let mut buf = [0; MAX_MESSAGE_SIZE]; | ||
|
|
||
| // Try to read data, this may still fail with `WouldBlock` | ||
| // if the readiness event is a false positive. | ||
| match self.socket.try_read(&mut buf) { | ||
| Ok(0) => { | ||
| warn!("Event export socket closed"); | ||
| return Ok(()); | ||
| }, | ||
| Ok(n) => { | ||
| let msg = &buf[..n]; | ||
| if let Err(e) = self.handle_relayer_event(msg).await { | ||
| // Events that fail to be submitted are effectively dropped here. | ||
| // We can consider retry logic or a local dead-letter queue, but | ||
| // for now we keep things simple. | ||
| error!("Error handling relayer event: {e}"); | ||
| } | ||
| }, | ||
| Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { | ||
| continue; | ||
| }, | ||
| Err(e) => { | ||
| return Err(e.into()); | ||
| }, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /// Handles an event received from the event export socket | ||
| async fn handle_relayer_event(&self, msg: &[u8]) -> Result<(), Error> { | ||
| let event = serde_json::from_slice::<RelayerEvent>(msg)?; | ||
| self.hse_client.submit_event(&event).await?; | ||
|
|
||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| impl Drop for EventSocket { | ||
| fn drop(&mut self) { | ||
| if let Err(e) = fs::remove_file(&self.path) { | ||
| warn!("Failed to remove Unix socket file: {}", e); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| //! A client for the historical state engine | ||
| use std::time::Duration; | ||
|
|
||
| use common::types::wallet::keychain::HmacKey; | ||
| use external_api::auth::add_expiring_auth_to_headers; | ||
| use eyre::{eyre, Error}; | ||
| use job_types::event_manager::RelayerEvent; | ||
| use reqwest::{header::HeaderMap, Client, Method, Response}; | ||
| use serde::Serialize; | ||
|
|
||
| // ------------- | ||
| // | Constants | | ||
| // ------------- | ||
|
|
||
| /// The buffer to add to the expiration timestamp for the signature | ||
| const SIG_EXPIRATION_BUFFER_MS: u64 = 5_000; // 5 seconds | ||
|
|
||
| /// The path to submit events to | ||
| const EVENT_SUBMISSION_PATH: &str = "/event"; | ||
|
|
||
| // ---------- | ||
| // | Client | | ||
| // ---------- | ||
|
|
||
| /// A client for the historical state engine | ||
| pub struct HseClient { | ||
| /// The base URL of the historical state engine | ||
| base_url: String, | ||
| /// The auth key for the historical state engine | ||
| auth_key: HmacKey, | ||
| } | ||
|
|
||
| impl HseClient { | ||
| /// Create a new historical state engine client | ||
| pub fn new(base_url: String, auth_key: HmacKey) -> Self { | ||
| Self { base_url, auth_key } | ||
| } | ||
|
|
||
| /// Submit an event to the historical state engine | ||
| pub async fn submit_event(&self, event: &RelayerEvent) -> Result<(), Error> { | ||
| send_authenticated_request( | ||
| &format!("{}{}", self.base_url, EVENT_SUBMISSION_PATH), | ||
| EVENT_SUBMISSION_PATH, | ||
| Method::POST, | ||
| event, | ||
| &self.auth_key, | ||
| ) | ||
| .await | ||
| .map(|_| ()) | ||
| } | ||
| } | ||
|
|
||
| // ----------- | ||
| // | Helpers | | ||
| // ----------- | ||
|
|
||
| /// Send a request w/ an expiring auth header | ||
| async fn send_authenticated_request<Req: Serialize>( | ||
| url: &str, | ||
| path: &str, | ||
| method: Method, | ||
| body: &Req, | ||
| key: &HmacKey, | ||
| ) -> Result<Response, Error> { | ||
| let expiration = Duration::from_millis(SIG_EXPIRATION_BUFFER_MS); | ||
|
|
||
| let body_bytes = serde_json::to_vec(body).expect("failed to serialize request body"); | ||
|
|
||
| let mut headers = HeaderMap::new(); | ||
| add_expiring_auth_to_headers(path, &mut headers, &body_bytes, key, expiration); | ||
|
|
||
| let route = format!("{}{}", url, path); | ||
| let response = send_request(&route, method, body, headers).await?; | ||
| Ok(response) | ||
| } | ||
|
|
||
| /// Send a basic HTTP request | ||
| async fn send_request<Req: Serialize>( | ||
| route: &str, | ||
| method: Method, | ||
| body: &Req, | ||
| headers: HeaderMap, | ||
| ) -> Result<Response, Error> { | ||
| let response = Client::new() | ||
| .request(method, route) | ||
| .headers(headers) | ||
| .json(body) | ||
| .send() | ||
| .await | ||
| .map_err(|e| eyre!("Failed to send request: {e}"))?; | ||
|
|
||
| // Check if the request was successful | ||
| if !response.status().is_success() { | ||
| return Err(eyre!("Request failed with status: {}", response.status())); | ||
| } | ||
|
|
||
| Ok(response) | ||
| } |
Oops, something went wrong.