feat: solana logs get parsed after the relayers initial startup (#18)

* refactor: using ws endpoint for fetching logs* refactor: improve "catchup" mechanism fetching logic
* refactor: clean up log fetching logic
* refactor: make linter happy
* docs: adr for Solana event ingestion
* refactor: added websocket recovery when waiting for initial msg

config.example.toml

@@ -16,4 +16,10 @@ chain = "solana-devnet"
 bind_addr = "0.0.0.0:3000"
 
 [solana_listener_component]
-solana_rpc = "https://api.devnet.solana.com"
+solana_http_rpc = "https://api.devnet.solana.com"
+
+# the officia ws endpoints hosted by solana sometimes refuse to accept new connections / drop the connection frequently.
+# using helius hosted nodes (even the free tier) often at times yields better results.
+solana_ws = "wss://api.devnet.solana.com"
+# solana_ws = "wss://devnet.helius-rpc.com/?{your api key}"
+missed_signature_catchup_strategy = "until_beginning"

crates/solana-axelar-relayer/src/main.rs

@@ -78,6 +78,8 @@ mod tests {
     use amplifier_api::identity::Identity;
     use pretty_assertions::assert_eq;
     use solana_listener::solana_sdk::pubkey::Pubkey;
+    use solana_listener::solana_sdk::signature::Signature;
+    use solana_listener::MissedSignatureCatchupStrategy;
 
     use crate::Config;
 
@@ -88,11 +90,14 @@ mod tests {
         let chain = "solana-devnet";
         let gateway_program_address = Pubkey::new_unique();
         let gateway_program_address_as_str = gateway_program_address.to_string();
-        let solana_rpc = "https://solana-devnet.com".parse()?;
+        let solana_rpc = "https://api.solana-devnet.com".parse()?;
+        let solana_ws = "wss://api.solana-devnet.com".parse()?;
         let solana_tx_scan_poll_period = Duration::from_millis(42);
         let solana_tx_scan_poll_period_ms = solana_tx_scan_poll_period.as_millis();
         let max_concurrent_rpc_requests = 100;
+        let latest_processed_signature = Signature::new_unique().to_string();
         let identity = identity_fixture();
+        let missed_signature_catchup_strategy = "until_beginning";
         let input = indoc::formatdoc! {r#"
             [amplifier_component]
             identity = '''
@@ -107,9 +112,12 @@ mod tests {
 
             [solana_listener_component]
             gateway_program_address = "{gateway_program_address_as_str}"
-            solana_rpc = "{solana_rpc}"
+            solana_http_rpc = "{solana_rpc}"
+            solana_ws = "{solana_ws}"
             tx_scan_poll_period_in_milliseconds = {solana_tx_scan_poll_period_ms}
             max_concurrent_rpc_requests = {max_concurrent_rpc_requests}
+            missed_signature_catchup_strategy = "{missed_signature_catchup_strategy}"
+            latest_processed_signature = "{latest_processed_signature}"
         "#};
 
         let parsed: Config = toml::from_str(&input)?;
@@ -126,9 +134,12 @@ mod tests {
             },
             solana_listener_component: solana_listener::Config {
                 gateway_program_address,
-                solana_rpc,
+                solana_http_rpc: solana_rpc,
                 tx_scan_poll_period: solana_tx_scan_poll_period,
                 max_concurrent_rpc_requests,
+                solana_ws,
+                missed_signature_catchup_strategy: MissedSignatureCatchupStrategy::UntilBeginning,
+                latest_processed_signature: Some(Signature::from_str(&latest_processed_signature)?),
             },
         };
         assert_eq!(parsed, expected);

crates/solana-event-forwarder/src/component.rs

@@ -3,7 +3,6 @@ use core::pin::Pin;
 
 use futures::{SinkExt, StreamExt};
 use gmp_gateway::events::{EventContainer, GatewayEvent};
-use relayer_amplifier_api_integration::amplifier_api::chrono::DateTime;
 use relayer_amplifier_api_integration::amplifier_api::types::{
     CallEvent, Event, EventBase, EventId, EventMetadata, GatewayV2Message, MessageId,
     PublishEventsRequest, TxId,
@@ -200,9 +199,7 @@ fn map_gateway_event_to_amplifier_event(
                     event_id,
                     meta: Some(EventMetadata {
                         tx_id: Some(tx_id),
-                        timestamp: message
-                            .block_time
-                            .and_then(|date_time| DateTime::from_timestamp(date_time, 0)),
+                        timestamp: message.timestamp,
                         from_address: Some(source_address.clone()),
                         finalized: Some(true),
                     }),

crates/solana-listener/Cargo.toml

@@ -11,16 +11,15 @@ edition.workspace = true
 typed-builder.workspace = true
 futures.workspace = true
 tracing.workspace = true
-futures-concurrency.workspace = true
 tokio.workspace = true
 eyre.workspace = true
 url.workspace = true
 serde.workspace = true
 relayer-engine.workspace = true
-thiserror.workspace = true
 backoff.workspace = true
 async-trait.workspace = true
 serde_json.workspace = true
+chrono.workspace = true
 
 gmp-gateway.workspace = true
 common-serde-utils.workspace = true