1use {
2 async_trait::async_trait,
3 carbon_core::{
4 datasource::{
5 AccountUpdate, Datasource, DatasourceId, TransactionUpdate, Update, UpdateType,
6 },
7 error::CarbonResult,
8 metrics::{Counter, Histogram, MetricsRegistry},
9 transformers::yellowstone::{create_tx_meta, create_tx_versioned},
10 },
11 futures::{sink::SinkExt, StreamExt},
12 solana_account::Account,
13 solana_pubkey::Pubkey,
14 solana_signature::Signature,
15 std::{collections::HashMap, convert::TryFrom, sync::LazyLock, time::Duration},
16 tokio::{sync::mpsc::Sender, time::sleep},
17 tokio_util::sync::CancellationToken,
18 uuid::Uuid,
19 yellowstone_grpc_client::{GeyserGrpcBuilder, GeyserGrpcBuilderResult, GeyserGrpcClient},
20 yellowstone_grpc_proto::{
21 geyser::{
22 subscribe_update::UpdateOneof, CommitmentLevel, SubscribeRequest,
23 SubscribeRequestFilterAccounts, SubscribeRequestFilterBlocks,
24 SubscribeRequestFilterSlots, SubscribeRequestFilterTransactions, SubscribeRequestPing,
25 SubscribeUpdateAccountInfo, SubscribeUpdateTransactionInfo,
26 },
27 tonic::{codec::CompressionEncoding, transport::ClientTlsConfig},
28 },
29};
30
31const MAX_RECONNECTION_ATTEMPTS: u32 = 10;
32const RECONNECTION_DELAY_MS: u64 = 3000;
33
34static ACCOUNT_PROCESS_TIME_NANOS: LazyLock<Histogram> = LazyLock::new(|| {
35 Histogram::new(
36 "laserstream_account_process_time_nanoseconds",
37 "Time to process account update in nanoseconds",
38 vec![
39 1_000.0,
40 10_000.0,
41 100_000.0,
42 1_000_000.0,
43 10_000_000.0,
44 100_000_000.0,
45 1_000_000_000.0,
46 ],
47 )
48});
49static ACCOUNT_UPDATES_RECEIVED: Counter = Counter::new(
50 "laserstream_account_updates_received_total",
51 "Account updates received from Laserstream",
52);
53static ACCOUNT_DELETION_PROCESS_TIME_NANOS: LazyLock<Histogram> = LazyLock::new(|| {
54 Histogram::new(
55 "laserstream_account_deletion_process_time_nanoseconds",
56 "Time to process account deletions in nanoseconds",
57 vec![
58 1_000.0,
59 10_000.0,
60 100_000.0,
61 1_000_000.0,
62 10_000_000.0,
63 100_000_000.0,
64 1_000_000_000.0,
65 ],
66 )
67});
68static ACCOUNT_DELETIONS_RECEIVED: Counter = Counter::new(
69 "laserstream_account_deletions_received_total",
70 "Account deletions received from Laserstream",
71);
72static TRANSACTION_PROCESS_TIME_NANOS: LazyLock<Histogram> = LazyLock::new(|| {
73 Histogram::new(
74 "laserstream_transaction_process_time_nanoseconds",
75 "Time to process transaction update in nanoseconds",
76 vec![
77 1_000.0,
78 10_000.0,
79 100_000.0,
80 1_000_000.0,
81 10_000_000.0,
82 100_000_000.0,
83 1_000_000_000.0,
84 ],
85 )
86});
87static TRANSACTION_UPDATES_RECEIVED: Counter = Counter::new(
88 "laserstream_transaction_updates_received_total",
89 "Transaction updates received from Laserstream",
90);
91
92fn register_laserstream_metrics() {
93 let registry = MetricsRegistry::global();
94 registry.register_counter(&ACCOUNT_UPDATES_RECEIVED);
95 registry.register_counter(&ACCOUNT_DELETIONS_RECEIVED);
96 registry.register_counter(&TRANSACTION_UPDATES_RECEIVED);
97 registry.register_histogram(&ACCOUNT_PROCESS_TIME_NANOS);
98 registry.register_histogram(&ACCOUNT_DELETION_PROCESS_TIME_NANOS);
99 registry.register_histogram(&TRANSACTION_PROCESS_TIME_NANOS);
100}
101
102#[derive(Debug)]
103pub struct LaserStreamGeyserClient {
104 pub endpoint: String,
105 pub x_token: Option<String>,
106 pub commitment: Option<CommitmentLevel>,
107 pub account_filters: HashMap<String, SubscribeRequestFilterAccounts>,
108 pub transaction_filters: HashMap<String, SubscribeRequestFilterTransactions>,
109 pub block_filters: BlockFilters,
110 pub geyser_config: LaserStreamClientConfig,
111}
112
113#[derive(Debug, Clone)]
114pub struct LaserStreamClientConfig {
115 pub compression: Option<CompressionEncoding>,
116 pub connect_timeout: Option<Duration>,
117 pub timeout: Option<Duration>,
118 pub max_decoding_message_size: Option<usize>,
119 pub tls_config: Option<ClientTlsConfig>,
120 pub tcp_nodelay: Option<bool>,
121 pub replay_enabled: bool,
122 pub from_slot: Option<u64>,
123}
124
125impl Default for LaserStreamClientConfig {
126 fn default() -> Self {
127 Self {
128 compression: None,
129 connect_timeout: Some(Duration::from_secs(15)),
130 timeout: Some(Duration::from_secs(15)),
131 max_decoding_message_size: None,
132 tls_config: None,
133 tcp_nodelay: None,
134 replay_enabled: true,
135 from_slot: None,
136 }
137 }
138}
139
140#[derive(Default, Debug, Clone)]
141pub struct BlockFilters {
142 pub filters: HashMap<String, SubscribeRequestFilterBlocks>,
143 pub failed_transactions: Option<bool>,
144}
145
146impl LaserStreamGeyserClient {
147 #[allow(clippy::too_many_arguments)]
148 pub const fn new(
149 endpoint: String,
150 x_token: Option<String>,
151 commitment: Option<CommitmentLevel>,
152 account_filters: HashMap<String, SubscribeRequestFilterAccounts>,
153 transaction_filters: HashMap<String, SubscribeRequestFilterTransactions>,
154 block_filters: BlockFilters,
155 geyser_config: LaserStreamClientConfig,
156 ) -> Self {
157 LaserStreamGeyserClient {
158 endpoint,
159 x_token,
160 commitment,
161 account_filters,
162 transaction_filters,
163 block_filters,
164 geyser_config,
165 }
166 }
167}
168
169impl LaserStreamClientConfig {
170 #[allow(clippy::too_many_arguments)]
171 pub const fn new(
172 compression: Option<CompressionEncoding>,
173 connect_timeout: Option<Duration>,
174 timeout: Option<Duration>,
175 max_decoding_message_size: Option<usize>,
176 tls_config: Option<ClientTlsConfig>,
177 tcp_nodelay: Option<bool>,
178 replay_enabled: bool,
179 from_slot: Option<u64>,
180 ) -> Self {
181 LaserStreamClientConfig {
182 compression,
183 connect_timeout,
184 timeout,
185 max_decoding_message_size,
186 tls_config,
187 tcp_nodelay,
188 replay_enabled,
189 from_slot,
190 }
191 }
192
193 pub fn geyser_config_builder(
194 &self,
195 mut builder: GeyserGrpcBuilder,
196 ) -> GeyserGrpcBuilderResult<GeyserGrpcBuilder> {
197 builder = builder.connect_timeout(self.connect_timeout.unwrap_or(Duration::from_secs(15)));
198
199 builder = builder.timeout(self.timeout.unwrap_or(Duration::from_secs(15)));
200 let tls = self
201 .tls_config
202 .clone()
203 .unwrap_or_else(|| ClientTlsConfig::new().with_enabled_roots());
204 builder = builder.tls_config(tls)?;
205
206 if let Some(compression) = self.compression {
207 builder = builder
208 .send_compressed(compression)
209 .accept_compressed(compression);
210 }
211 if let Some(val) = self.max_decoding_message_size {
212 builder = builder.max_decoding_message_size(val);
213 }
214
215 if let Some(val) = self.tcp_nodelay {
216 builder = builder.tcp_nodelay(val);
217 }
218 Ok(builder)
219 }
220}
221
222#[async_trait]
223impl Datasource for LaserStreamGeyserClient {
224 async fn consume(
225 &self,
226 id: DatasourceId,
227 sender: Sender<(Update, DatasourceId)>,
228 cancellation_token: CancellationToken,
229 ) -> CarbonResult<()> {
230 register_laserstream_metrics();
231 let endpoint = self.endpoint.clone();
232 let x_token = self.x_token.clone();
233 let commitment = self.commitment;
234 let account_filters = self.account_filters.clone();
235 let transaction_filters = self.transaction_filters.clone();
236 let BlockFilters {
237 filters,
238 failed_transactions: block_failed_transactions,
239 } = self.block_filters.clone();
240 let retain_block_failed_transactions = block_failed_transactions.unwrap_or(true);
241 let geyser_config = self.geyser_config.clone();
242 let replay_enabled = geyser_config.replay_enabled;
243
244 let builder = GeyserGrpcClient::build_from_shared(endpoint.clone())
245 .map_err(|err| carbon_core::error::Error::FailedToConsumeDatasource(err.to_string()))?
246 .x_token(x_token.clone())
247 .map_err(|err| carbon_core::error::Error::FailedToConsumeDatasource(err.to_string()))?;
248
249 let mut geyser_client = geyser_config
250 .geyser_config_builder(builder)
251 .map_err(|err| carbon_core::error::Error::FailedToConsumeDatasource(err.to_string()))?
252 .connect()
253 .await
254 .map_err(|err| carbon_core::error::Error::FailedToConsumeDatasource(err.to_string()))?;
255
256 tokio::spawn(async move {
257 let mut reconnect_attempts = 0;
258 let mut tracked_slot: u64 = 0;
259
260 let mut subscribe_request = SubscribeRequest {
261 slots: HashMap::new(),
262 accounts: account_filters,
263 transactions: transaction_filters,
264 transactions_status: HashMap::new(),
265 entry: HashMap::new(),
266 blocks: filters,
267 blocks_meta: HashMap::new(),
268 commitment: commitment.map(|x| x as i32),
269 accounts_data_slice: vec![],
270 ping: None,
271 from_slot: geyser_config.from_slot,
272 };
273
274 let internal_slot_sub_id = if replay_enabled {
275 let slot_id = format!(
276 "internal-{}",
277 Uuid::new_v4().to_string().split('-').next().unwrap()
278 );
279 subscribe_request.slots.insert(
280 slot_id.clone(),
281 SubscribeRequestFilterSlots {
282 filter_by_commitment: Some(true),
283 ..Default::default()
284 },
285 );
286 Some(slot_id)
287 } else {
288 None
289 };
290
291 let id_for_loop = id.clone();
292
293 loop {
294 tokio::select! {
295 _ = cancellation_token.cancelled() => {
296 log::info!("Cancelling Laserstream subscription.");
297 break;
298 }
299 _ = async {
300 if reconnect_attempts > 0 && tracked_slot > 0 && replay_enabled {
302 let commitment_level = subscribe_request.commitment.unwrap_or(0);
303 let from_slot = match commitment_level {
304 0 => tracked_slot.saturating_sub(31), 1 | 2 => tracked_slot, _ => tracked_slot.saturating_sub(31), };
308 subscribe_request.from_slot = Some(from_slot);
309 } else if !replay_enabled {
310 subscribe_request.from_slot = None;
311 }
312
313
314
315 match geyser_client.subscribe_with_request(Some(subscribe_request.clone())).await {
316 Ok((mut subscribe_tx, mut stream)) => {
317 reconnect_attempts = 0; while let Some(message) = stream.next().await {
320 if cancellation_token.is_cancelled() {
321 return Ok(());
322 }
323
324 match message {
325 Ok(msg) => match msg.update_oneof {
326 Some(UpdateOneof::Account(account_update)) => {
327 send_subscribe_account_update_info(
328 account_update.account,
329 &sender,
330 id_for_loop.clone(),
331 account_update.slot,
332 )
333 .await
334 }
335 Some(UpdateOneof::Transaction(transaction_update)) => {
336 send_subscribe_update_transaction_info(
337 transaction_update.transaction,
338 &sender,
339 id_for_loop.clone(),
340 transaction_update.slot,
341 None,
342 ).await
343 }
344 Some(UpdateOneof::Block(block_update)) => {
345 let block_time = block_update.block_time.map(|ts| ts.timestamp);
346
347 for transaction_update in block_update.transactions {
348 if retain_block_failed_transactions || transaction_update.meta.as_ref().map(|meta| meta.err.is_none()).unwrap_or(false) {
349 send_subscribe_update_transaction_info(Some(transaction_update), &sender, id_for_loop.clone(), block_update.slot, block_time).await
350 }
351 }
352
353 for account_info in block_update.accounts {
354 send_subscribe_account_update_info(
355 Some(account_info),
356 &sender,
357 id_for_loop.clone(),
358 block_update.slot,
359 )
360 .await;
361 }
362 }
363 Some(UpdateOneof::Slot(slot_update)) => {
364 if replay_enabled {
365 tracked_slot = slot_update.slot;
366 }
367
368 if let Some(ref internal_id) = internal_slot_sub_id {
370 if msg.filters.len() == 1 && msg.filters.contains(internal_id) {
371 continue;
372 }
373 }
374 }
375 Some(UpdateOneof::Ping(_)) => {
376 match subscribe_tx
377 .send(SubscribeRequest {
378 ping: Some(SubscribeRequestPing { id: 1 }),
379 ..Default::default()
380 })
381 .await {
382 Ok(()) => (),
383 Err(error) => {
384 log::error!("Failed to send ping error: {error:?}");
385 break;
386 },
387 }
388 }
389
390 _ => {}
391 },
392 Err(error) => {
393 log::error!("Geyser stream error, will reconnect: {error:?}");
394 break;
395 }
396 }
397 }
398 Ok(())
399 }
400 Err(e) => {
401 log::error!("Failed to subscribe, will retry after 5s delay: {e:?}");
402 Err(carbon_core::error::Error::FailedToConsumeDatasource(e.to_string()))
403 }
404 }
405 } => {
406 reconnect_attempts += 1;
407 if reconnect_attempts >= MAX_RECONNECTION_ATTEMPTS {
408 log::error!("Max reconnection attempts reached: {MAX_RECONNECTION_ATTEMPTS}");
409 break;
410 }
411
412 let delay = Duration::from_millis(RECONNECTION_DELAY_MS);
413 sleep(delay).await;
414 }
415 }
416 }
417 });
418
419 Ok(())
420 }
421
422 fn update_types(&self) -> Vec<UpdateType> {
423 vec![
424 UpdateType::AccountUpdate,
425 UpdateType::Transaction,
426 UpdateType::AccountDeletion,
427 ]
428 }
429}
430
431async fn send_subscribe_account_update_info(
432 account_update_info: Option<SubscribeUpdateAccountInfo>,
433 sender: &Sender<(Update, DatasourceId)>,
434 id: DatasourceId,
435 slot: u64,
436) {
437 let start_time = std::time::Instant::now();
438
439 if let Some(account_info) = account_update_info {
440 let Ok(account_pubkey) = Pubkey::try_from(account_info.pubkey) else {
441 return;
442 };
443
444 let Ok(account_owner_pubkey) = Pubkey::try_from(account_info.owner) else {
445 return;
446 };
447
448 let account = Account {
449 lamports: account_info.lamports,
450 data: account_info.data,
451 owner: account_owner_pubkey,
452 executable: account_info.executable,
453 rent_epoch: account_info.rent_epoch,
454 };
455
456 let update = AccountUpdate {
457 pubkey: account_pubkey,
458 account,
459 slot,
460 transaction_signature: account_info
461 .txn_signature
462 .and_then(|sig| Signature::try_from(sig).ok()),
463 }
464 .into_update();
465 let is_deletion = matches!(&update, Update::AccountDeletion(_));
466
467 if let Err(e) = sender.try_send((update, id)) {
468 log::error!(
469 "Failed to send account event for pubkey {account_pubkey:?} at slot {slot}: {e:?}"
470 );
471 }
472
473 if is_deletion {
474 ACCOUNT_DELETION_PROCESS_TIME_NANOS.record(start_time.elapsed().as_nanos() as f64);
475 ACCOUNT_DELETIONS_RECEIVED.inc();
476 } else {
477 ACCOUNT_PROCESS_TIME_NANOS.record(start_time.elapsed().as_nanos() as f64);
478 ACCOUNT_UPDATES_RECEIVED.inc();
479 }
480 } else {
481 log::error!("No account info in UpdateOneof::Account at slot {slot}");
482 }
483}
484
485async fn send_subscribe_update_transaction_info(
486 transaction_info: Option<SubscribeUpdateTransactionInfo>,
487 sender: &Sender<(Update, DatasourceId)>,
488 id: DatasourceId,
489 slot: u64,
490 block_time: Option<i64>,
491) {
492 let start_time = std::time::Instant::now();
493
494 if let Some(transaction_info) = transaction_info {
495 let Ok(signature) = Signature::try_from(transaction_info.signature) else {
496 return;
497 };
498 let Some(yellowstone_transaction) = transaction_info.transaction else {
499 return;
500 };
501 let Some(yellowstone_tx_meta) = transaction_info.meta else {
502 return;
503 };
504 let Ok(versioned_transaction) = create_tx_versioned(yellowstone_transaction) else {
505 return;
506 };
507 let meta_original = match create_tx_meta(yellowstone_tx_meta) {
508 Ok(meta) => meta,
509 Err(err) => {
510 log::error!("Failed to create transaction meta: {err:?}");
511 return;
512 }
513 };
514 let update = Update::Transaction(Box::new(TransactionUpdate {
515 signature,
516 transaction: versioned_transaction,
517 meta: meta_original,
518 is_vote: transaction_info.is_vote,
519 slot,
520 index: Some(transaction_info.index),
521 block_time,
522 block_hash: None,
523 }));
524 if let Err(e) = sender.try_send((update, id)) {
525 log::error!(
526 "Failed to send transaction update with signature {signature:?} at slot {slot}: {e:?}"
527 );
528 return;
529 }
530
531 TRANSACTION_PROCESS_TIME_NANOS.record(start_time.elapsed().as_nanos() as f64);
532 TRANSACTION_UPDATES_RECEIVED.inc();
533 } else {
534 log::error!("No transaction info in `UpdateOneof::Transaction` at slot {slot}");
535 }
536}