1use std::{collections::VecDeque, time::Duration};
17
18use anyhow::Context;
19use nautilus_common::{
20 clients::DataClient,
21 defi::RequestPoolSnapshot,
22 messages::{
23 DataEvent,
24 defi::{
25 DefiDataCommand, DefiRequestCommand, DefiSubscribeCommand, DefiUnsubscribeCommand,
26 SubscribeBlocks, SubscribePool, SubscribePoolFeeCollects, SubscribePoolFlashEvents,
27 SubscribePoolLiquidityUpdates, SubscribePoolSwaps, UnsubscribeBlocks, UnsubscribePool,
28 UnsubscribePoolFeeCollects, UnsubscribePoolFlashEvents,
29 UnsubscribePoolLiquidityUpdates, UnsubscribePoolSwaps,
30 },
31 },
32};
33use nautilus_live::{
34 SocketControlFactory,
35 task::{TaskGroup, TaskGroupGuard},
36};
37use nautilus_model::{
38 defi::{DefiData, DexType, PoolIdentifier, SharedChain, validation::validate_address},
39 identifiers::{ClientId, Venue},
40};
41use ustr::Ustr;
42
43use crate::{
44 cache::BlockchainCache,
45 config::BlockchainDataClientConfig,
46 data::{
47 core::BlockchainDataClientCore,
48 subscription::{BlockFeedBackend, BlockFeedOwner},
49 },
50 exchanges::get_dex_extended,
51 rpc::{
52 BlockchainRpcClient,
53 types::{BlockchainMessage, RpcEventType},
54 },
55};
56
57const MAX_PENDING_POOL_MESSAGES: usize = 10_000;
58
59#[derive(Debug)]
69pub struct BlockchainDataClient {
70 pub client_id: ClientId,
72 pub chain: SharedChain,
74 pub config: BlockchainDataClientConfig,
76 pub core_client: Option<BlockchainDataClientCore>,
79 socket_factory: SocketControlFactory,
80 hypersync_rx: Option<tokio::sync::mpsc::UnboundedReceiver<BlockchainMessage>>,
81 hypersync_tx: Option<tokio::sync::mpsc::UnboundedSender<BlockchainMessage>>,
82 command_tx: tokio::sync::mpsc::UnboundedSender<DefiDataCommand>,
83 command_rx: Option<tokio::sync::mpsc::UnboundedReceiver<DefiDataCommand>>,
84 session_tasks: TaskGroup,
85 cancellation_token: tokio_util::sync::CancellationToken,
86}
87
88impl BlockchainDataClient {
89 #[must_use]
91 pub fn new(client_id: ClientId, config: BlockchainDataClientConfig) -> Self {
92 let chain = config.chain.clone();
93 let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
94 let (hypersync_tx, hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
95 let socket_factory = SocketControlFactory::new(client_id, None);
96 let session_tasks = TaskGroup::new();
97 Self {
98 client_id,
99 chain,
100 core_client: None,
101 socket_factory,
102 config,
103 hypersync_rx: Some(hypersync_rx),
104 hypersync_tx: Some(hypersync_tx),
105 command_tx,
106 command_rx: Some(command_rx),
107 cancellation_token: session_tasks.cancellation_token(),
108 session_tasks,
109 }
110 }
111
112 fn spawn_process_task(
120 &mut self,
121 ) -> anyhow::Result<tokio::sync::oneshot::Receiver<anyhow::Result<()>>> {
122 let command_rx = if let Some(r) = self.command_rx.take() {
123 r
124 } else {
125 log::error!("Command receiver already taken, not spawning handler");
126 anyhow::bail!("Command receiver already taken");
127 };
128
129 let cancellation_token = self.cancellation_token.clone();
130
131 let data_tx = nautilus_common::live::runner::get_data_event_sender();
132
133 let mut hypersync_rx = self.hypersync_rx.take().unwrap();
134 let hypersync_tx = self.hypersync_tx.take();
135
136 let mut core_client = BlockchainDataClientCore::new(
137 self.config.clone(),
138 hypersync_tx,
139 Some(data_tx),
140 cancellation_token.clone(),
141 );
142 core_client.set_socket_control(self.socket_factory.control("blockchain-rpc"));
143
144 let (startup_tx, startup_rx) = tokio::sync::oneshot::channel();
145 let future = async move {
146 log::debug!("Started task 'process'");
147
148 if let Err(e) = core_client.connect().await {
149 if e.to_string().contains("cancelled") || e.to_string().contains("Sync cancelled") {
152 log::warn!("Blockchain core client connection interrupted: {e}");
153 } else {
154 log::error!("Failed to connect blockchain core client: {e}");
155 }
156 let _ = startup_tx.send(Err(e));
157 return;
158 }
159 let _ = startup_tx.send(Ok(()));
160
161 let mut command_rx = command_rx;
162 let mut pending_pool_messages = VecDeque::new();
163
164 loop {
165 tokio::select! {
166 () = cancellation_token.cancelled() => {
167 log::debug!("Received cancellation signal in Blockchain data client process task");
168 core_client.disconnect().await;
169 break;
170 }
171 command = command_rx.recv() => {
172 if let Some(cmd) = command {
173 match cmd {
174 DefiDataCommand::Subscribe(cmd) => {
175 let chain = cmd.blockchain();
176 if chain != core_client.chain.name {
177 log::error!("Incorrect blockchain for subscribe command: {chain}");
178 continue;
179 }
180
181 if let Err(e) = Self::handle_subscribe_command(cmd, &mut core_client).await{
182 log::error!("Error processing subscribe command: {e}");
183 }
184 }
185 DefiDataCommand::Unsubscribe(cmd) => {
186 let chain = cmd.blockchain();
187 if chain != core_client.chain.name {
188 log::error!("Incorrect blockchain for subscribe command: {chain}");
189 continue;
190 }
191
192 if let Err(e) = Self::handle_unsubscribe_command(cmd, &mut core_client).await{
193 log::error!("Error processing subscribe command: {e}");
194 }
195 }
196 DefiDataCommand::Request(cmd) => {
197 if let Err(e) = Self::handle_request_command(cmd, &mut core_client).await {
198 log::error!("Error processing request command: {e}");
199 }
200 }
201 }
202 } else {
203 log::debug!("Command channel closed");
204 break;
205 }
206 }
207 data = hypersync_rx.recv() => {
208 if let Some(msg) = data {
209 Self::process_live_blockchain_message(
210 msg,
211 &mut core_client,
212 &mut pending_pool_messages,
213 )
214 .await;
215 } else {
216 log::debug!("HyperSync data channel closed");
217 break;
218 }
219 }
220 msg = async {
221 match core_client.rpc_client {
222 Some(ref mut rpc_client) => rpc_client.next_rpc_message().await,
223 None => std::future::pending().await, }
225 } => {
226 match msg {
228 Ok(msg) => {
229 Self::process_live_blockchain_message(
230 msg,
231 &mut core_client,
232 &mut pending_pool_messages,
233 )
234 .await;
235 }
236 Err(e) => {
237 log::error!("Error processing RPC message: {e}");
238 }
239 }
240 }
241 }
242 }
243
244 log::debug!("Stopped task 'process'");
245 };
246
247 self.session_tasks
248 .spawn(future)
249 .map_err(|e| anyhow::anyhow!("Failed to register blockchain process task: {e}"))?;
250 Ok(startup_rx)
251 }
252
253 async fn process_live_blockchain_message(
254 msg: BlockchainMessage,
255 core_client: &mut BlockchainDataClientCore,
256 pending_pool_messages: &mut VecDeque<BlockchainMessage>,
257 ) {
258 let is_block = matches!(&msg, BlockchainMessage::Block(_));
259 let Some(msg) =
260 Self::ready_live_blockchain_message(msg, &core_client.cache, pending_pool_messages)
261 else {
262 return;
263 };
264
265 if let Some(data) = Self::data_event_from_blockchain_message(msg, core_client).await {
266 core_client.send_data(data);
267 }
268
269 if is_block {
270 for data in Self::drain_pending_pool_messages(core_client, pending_pool_messages).await
271 {
272 core_client.send_data(data);
273 }
274 }
275 }
276
277 async fn drain_pending_pool_messages(
278 core_client: &mut BlockchainDataClientCore,
279 pending_pool_messages: &mut VecDeque<BlockchainMessage>,
280 ) -> Vec<DataEvent> {
281 let ready_messages = Self::drain_pending_pool_messages_with_cached_metadata(
282 pending_pool_messages,
283 &core_client.cache,
284 );
285 let mut data_events = Vec::with_capacity(ready_messages.len());
286
287 for msg in ready_messages {
288 if let Some(data) = Self::data_event_from_blockchain_message(msg, core_client).await {
289 data_events.push(data);
290 }
291 }
292
293 data_events
294 }
295
296 fn drain_pending_pool_messages_with_cached_metadata(
297 pending_pool_messages: &mut VecDeque<BlockchainMessage>,
298 cache: &BlockchainCache,
299 ) -> Vec<BlockchainMessage> {
300 let pending_count = pending_pool_messages.len();
301 let mut ready_messages = Vec::new();
302
303 for _ in 0..pending_count {
304 let Some(msg) = pending_pool_messages.pop_front() else {
305 break;
306 };
307
308 if Self::pool_event_missing_block_metadata(&msg, cache).is_some() {
309 pending_pool_messages.push_back(msg);
310 continue;
311 }
312
313 ready_messages.push(msg);
314 }
315
316 ready_messages
317 }
318
319 fn queue_pending_pool_message(
320 pending_pool_messages: &mut VecDeque<BlockchainMessage>,
321 msg: BlockchainMessage,
322 ) {
323 if pending_pool_messages.len() >= MAX_PENDING_POOL_MESSAGES
324 && let Some(dropped_msg) = pending_pool_messages.pop_front()
325 && let Some(block_number) = Self::pool_event_block_number(&dropped_msg)
326 {
327 log::warn!(
328 "Dropping oldest live pool event waiting for uncached block {block_number}; pending buffer reached {MAX_PENDING_POOL_MESSAGES} messages"
329 );
330 }
331
332 pending_pool_messages.push_back(msg);
333 }
334
335 fn ready_live_blockchain_message(
336 msg: BlockchainMessage,
337 cache: &BlockchainCache,
338 pending_pool_messages: &mut VecDeque<BlockchainMessage>,
339 ) -> Option<BlockchainMessage> {
340 if let Some(block_number) = Self::pool_event_missing_block_metadata(&msg, cache) {
341 log::debug!("Deferring live pool event until block {block_number} metadata is cached");
342 Self::queue_pending_pool_message(pending_pool_messages, msg);
343 None
344 } else {
345 Some(msg)
346 }
347 }
348
349 fn pool_event_missing_block_metadata(
350 msg: &BlockchainMessage,
351 cache: &BlockchainCache,
352 ) -> Option<u64> {
353 let block_number = Self::pool_event_block_number(msg)?;
354 (cache.get_block_timestamp(block_number).is_none()
355 || cache.get_block_hash(block_number).is_none())
356 .then_some(block_number)
357 }
358
359 fn pool_event_block_number(msg: &BlockchainMessage) -> Option<u64> {
360 match msg {
361 BlockchainMessage::SwapEvent(event) => Some(event.block_number),
362 BlockchainMessage::MintEvent(event) => Some(event.block_number),
363 BlockchainMessage::BurnEvent(event) => Some(event.block_number),
364 BlockchainMessage::CollectEvent(event) => Some(event.block_number),
365 BlockchainMessage::FlashEvent(event) => Some(event.block_number),
366 BlockchainMessage::FeeProtocolUpdateEvent(event) => Some(event.block_number),
367 BlockchainMessage::FeeProtocolCollectEvent(event) => Some(event.block_number),
368 BlockchainMessage::Block(_) => None,
369 }
370 }
371
372 async fn data_event_from_blockchain_message(
373 msg: BlockchainMessage,
374 core_client: &mut BlockchainDataClientCore,
375 ) -> Option<DataEvent> {
376 match msg {
377 BlockchainMessage::Block(block) => {
378 if let Err(e) = core_client.cache.add_block(block.clone()).await {
379 log::error!("Failed to cache block {}: {e}", block.number);
380 }
381
382 Some(DataEvent::DeFi(DefiData::Block(block)))
383 }
384 BlockchainMessage::SwapEvent(swap_event) => {
385 match core_client.get_pool(&swap_event.pool_identifier) {
386 Ok(pool) => match core_client.process_pool_swap_event(&swap_event, pool) {
387 Ok(swap) => Some(DataEvent::DeFi(DefiData::PoolSwap(swap))),
388 Err(e) => {
389 log::error!("Error processing pool swap event: {e}");
390 None
391 }
392 },
393 Err(e) => {
394 log::error!(
395 "Failed to get pool {} with error {:?}",
396 swap_event.pool_identifier,
397 e
398 );
399 None
400 }
401 }
402 }
403 BlockchainMessage::BurnEvent(burn_event) => {
404 match core_client.get_pool(&burn_event.pool_identifier) {
405 Ok(pool) => {
406 let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name)
407 .expect("Failed to get dex extended");
408
409 match core_client.process_pool_burn_event(&burn_event, pool, dex_extended) {
410 Ok(update) => {
411 Some(DataEvent::DeFi(DefiData::PoolLiquidityUpdate(update)))
412 }
413 Err(e) => {
414 log::error!("Error processing pool burn event: {e}");
415 None
416 }
417 }
418 }
419 Err(e) => {
420 log::error!(
421 "Failed to get pool {} with error {:?}",
422 burn_event.pool_identifier,
423 e
424 );
425 None
426 }
427 }
428 }
429 BlockchainMessage::MintEvent(mint_event) => {
430 match core_client.get_pool(&mint_event.pool_identifier) {
431 Ok(pool) => {
432 let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name)
433 .expect("Failed to get dex extended");
434
435 match core_client.process_pool_mint_event(&mint_event, pool, dex_extended) {
436 Ok(update) => {
437 Some(DataEvent::DeFi(DefiData::PoolLiquidityUpdate(update)))
438 }
439 Err(e) => {
440 log::error!("Error processing pool mint event: {e}");
441 None
442 }
443 }
444 }
445 Err(e) => {
446 log::error!(
447 "Failed to get pool {} with error {:?}",
448 mint_event.pool_identifier,
449 e
450 );
451 None
452 }
453 }
454 }
455 BlockchainMessage::CollectEvent(collect_event) => {
456 match core_client.get_pool(&collect_event.pool_identifier) {
457 Ok(pool) => {
458 let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name)
459 .expect("Failed to get dex extended");
460
461 match core_client.process_pool_collect_event(
462 &collect_event,
463 pool,
464 dex_extended,
465 ) {
466 Ok(update) => Some(DataEvent::DeFi(DefiData::PoolFeeCollect(update))),
467 Err(e) => {
468 log::error!("Error processing pool collect event: {e}");
469 None
470 }
471 }
472 }
473 Err(e) => {
474 log::error!(
475 "Failed to get pool {} with error {:?}",
476 collect_event.pool_identifier,
477 e
478 );
479 None
480 }
481 }
482 }
483 BlockchainMessage::FlashEvent(flash_event) => {
484 match core_client.get_pool(&flash_event.pool_identifier) {
485 Ok(pool) => match core_client.process_pool_flash_event(&flash_event, pool) {
486 Ok(flash) => Some(DataEvent::DeFi(DefiData::PoolFlash(flash))),
487 Err(e) => {
488 log::error!("Error processing pool flash event: {e}");
489 None
490 }
491 },
492 Err(e) => {
493 log::error!(
494 "Failed to get pool {} with error {:?}",
495 flash_event.pool_identifier,
496 e
497 );
498 None
499 }
500 }
501 }
502 BlockchainMessage::FeeProtocolUpdateEvent(update_event) => {
503 match core_client.get_pool(&update_event.pool_identifier) {
504 Ok(pool) => match core_client
505 .process_pool_fee_protocol_update_event(&update_event, pool)
506 {
507 Ok(update) => {
508 Some(DataEvent::DeFi(DefiData::PoolFeeProtocolUpdate(update)))
509 }
510 Err(e) => {
511 log::error!("Error processing pool fee-protocol update event: {e}");
512 None
513 }
514 },
515 Err(e) => {
516 log::error!(
517 "Failed to get pool {} with error {:?}",
518 update_event.pool_identifier,
519 e
520 );
521 None
522 }
523 }
524 }
525 BlockchainMessage::FeeProtocolCollectEvent(collect_event) => {
526 match core_client.get_pool(&collect_event.pool_identifier) {
527 Ok(pool) => match core_client
528 .process_pool_fee_protocol_collect_event(&collect_event, pool)
529 {
530 Ok(collect) => {
531 Some(DataEvent::DeFi(DefiData::PoolFeeProtocolCollect(collect)))
532 }
533 Err(e) => {
534 log::error!("Error processing pool fee-protocol collect event: {e}");
535 None
536 }
537 },
538 Err(e) => {
539 log::error!(
540 "Failed to get pool {} with error {:?}",
541 collect_event.pool_identifier,
542 e
543 );
544 None
545 }
546 }
547 }
548 }
549 }
550
551 async fn handle_subscribe_command(
553 command: DefiSubscribeCommand,
554 core_client: &mut BlockchainDataClientCore,
555 ) -> anyhow::Result<()> {
556 match command {
557 DefiSubscribeCommand::Blocks(_cmd) => {
558 log::debug!("Processing subscribe blocks command");
559
560 Self::subscribe_block_feed(core_client, BlockFeedOwner::Explicit).await?;
561
562 Ok(())
563 }
564 DefiSubscribeCommand::Pool(cmd) => {
565 log::debug!(
566 "Processing subscribe pool command for {}",
567 cmd.instrument_id
568 );
569
570 if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
571 let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
572 .map_err(|e| {
573 anyhow::anyhow!(
574 "Invalid pool address '{}' failed with error: {:?}",
575 cmd.instrument_id,
576 e
577 )
578 })?;
579
580 core_client
582 .subscription_manager
583 .subscribe_swaps(dex, pool_address);
584 core_client
585 .subscription_manager
586 .subscribe_burns(dex, pool_address);
587 core_client
588 .subscription_manager
589 .subscribe_mints(dex, pool_address);
590 core_client
591 .subscription_manager
592 .subscribe_collects(dex, pool_address);
593 core_client
594 .subscription_manager
595 .subscribe_flashes(dex, pool_address);
596 core_client
597 .subscription_manager
598 .subscribe_fee_protocol_updates(dex, pool_address);
599 core_client
600 .subscription_manager
601 .subscribe_fee_protocol_collects(dex, pool_address);
602 Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
603 Self::update_hypersync_pool_event_stream(core_client, dex).await?;
604
605 log::debug!(
606 "Subscribed to all pool events for {} at address {}",
607 cmd.instrument_id,
608 pool_address
609 );
610 } else {
611 anyhow::bail!(
612 "Invalid venue {}, expected Blockchain DEX format",
613 cmd.instrument_id.venue
614 )
615 }
616
617 Ok(())
618 }
619 DefiSubscribeCommand::PoolSwaps(cmd) => {
620 log::debug!(
621 "Processing subscribe pool swaps command for {}",
622 cmd.instrument_id
623 );
624
625 if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
626 let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
627 .map_err(|e| {
628 anyhow::anyhow!(
629 "Invalid pool swap address '{}' failed with error: {:?}",
630 cmd.instrument_id,
631 e
632 )
633 })?;
634 core_client
635 .subscription_manager
636 .subscribe_swaps(dex, pool_address);
637 Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
638 Self::update_hypersync_pool_event_stream(core_client, dex).await?;
639 } else {
640 anyhow::bail!(
641 "Invalid venue {}, expected Blockchain DEX format",
642 cmd.instrument_id.venue
643 )
644 }
645
646 Ok(())
647 }
648 DefiSubscribeCommand::PoolLiquidityUpdates(cmd) => {
649 log::debug!(
650 "Processing subscribe pool liquidity updates command for address: {}",
651 cmd.instrument_id
652 );
653
654 if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
655 let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
656 .map_err(|_| {
657 anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
658 })?;
659 core_client
660 .subscription_manager
661 .subscribe_burns(dex, pool_address);
662 core_client
663 .subscription_manager
664 .subscribe_mints(dex, pool_address);
665 Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
666 Self::update_hypersync_pool_event_stream(core_client, dex).await?;
667 } else {
668 anyhow::bail!(
669 "Invalid venue {}, expected Blockchain DEX format",
670 cmd.instrument_id.venue
671 )
672 }
673
674 Ok(())
675 }
676 DefiSubscribeCommand::PoolFeeCollects(cmd) => {
677 log::debug!(
678 "Processing subscribe pool fee collects command for address: {}",
679 cmd.instrument_id
680 );
681
682 if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
683 let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
684 .map_err(|_| {
685 anyhow::anyhow!(
686 "Invalid pool fee collect address: {}",
687 cmd.instrument_id
688 )
689 })?;
690 core_client
691 .subscription_manager
692 .subscribe_collects(dex, pool_address);
693 Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
694 Self::update_hypersync_pool_event_stream(core_client, dex).await?;
695 } else {
696 anyhow::bail!(
697 "Invalid venue {}, expected Blockchain DEX format",
698 cmd.instrument_id.venue
699 )
700 }
701
702 Ok(())
703 }
704 DefiSubscribeCommand::PoolFlashEvents(cmd) => {
705 log::debug!(
706 "Processing subscribe pool flash command for address: {}",
707 cmd.instrument_id
708 );
709
710 if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
711 let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
712 .map_err(|_| {
713 anyhow::anyhow!(
714 "Invalid pool flash subscribe address: {}",
715 cmd.instrument_id
716 )
717 })?;
718 core_client
719 .subscription_manager
720 .subscribe_flashes(dex, pool_address);
721 Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
722 Self::update_hypersync_pool_event_stream(core_client, dex).await?;
723 } else {
724 anyhow::bail!(
725 "Invalid venue {}, expected Blockchain DEX format",
726 cmd.instrument_id.venue
727 )
728 }
729
730 Ok(())
731 }
732 }
733 }
734
735 async fn handle_unsubscribe_command(
737 command: DefiUnsubscribeCommand,
738 core_client: &mut BlockchainDataClientCore,
739 ) -> anyhow::Result<()> {
740 match command {
741 DefiUnsubscribeCommand::Blocks(_cmd) => {
742 log::debug!("Processing unsubscribe blocks command");
743
744 Self::unsubscribe_block_feed(core_client, BlockFeedOwner::Explicit).await?;
745
746 Ok(())
747 }
748 DefiUnsubscribeCommand::Pool(cmd) => {
749 log::debug!(
750 "Processing unsubscribe pool command for {}",
751 cmd.instrument_id
752 );
753
754 if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
755 let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
756 .map_err(|_| {
757 anyhow::anyhow!("Invalid pool address: {}", cmd.instrument_id)
758 })?;
759
760 core_client
762 .subscription_manager
763 .unsubscribe_swaps(dex, pool_address);
764 core_client
765 .subscription_manager
766 .unsubscribe_burns(dex, pool_address);
767 core_client
768 .subscription_manager
769 .unsubscribe_mints(dex, pool_address);
770 core_client
771 .subscription_manager
772 .unsubscribe_collects(dex, pool_address);
773 core_client
774 .subscription_manager
775 .unsubscribe_flashes(dex, pool_address);
776 core_client
777 .subscription_manager
778 .unsubscribe_fee_protocol_updates(dex, pool_address);
779 core_client
780 .subscription_manager
781 .unsubscribe_fee_protocol_collects(dex, pool_address);
782 Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
783 Self::update_hypersync_pool_event_stream(core_client, dex).await?;
784
785 log::debug!(
786 "Unsubscribed from all pool events for {} at address {}",
787 cmd.instrument_id,
788 pool_address
789 );
790 } else {
791 anyhow::bail!(
792 "Invalid venue {}, expected Blockchain DEX format",
793 cmd.instrument_id.venue
794 )
795 }
796
797 Ok(())
798 }
799 DefiUnsubscribeCommand::PoolSwaps(cmd) => {
800 log::debug!("Processing unsubscribe pool swaps command");
801
802 if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
803 let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
804 .map_err(|_| {
805 anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
806 })?;
807 core_client
808 .subscription_manager
809 .unsubscribe_swaps(dex, pool_address);
810 Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
811 Self::update_hypersync_pool_event_stream(core_client, dex).await?;
812 } else {
813 anyhow::bail!(
814 "Invalid venue {}, expected Blockchain DEX format",
815 cmd.instrument_id.venue
816 )
817 }
818
819 Ok(())
820 }
821 DefiUnsubscribeCommand::PoolLiquidityUpdates(cmd) => {
822 log::debug!(
823 "Processing unsubscribe pool liquidity updates command for {}",
824 cmd.instrument_id
825 );
826
827 if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
828 let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
829 .map_err(|_| {
830 anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
831 })?;
832 core_client
833 .subscription_manager
834 .unsubscribe_burns(dex, pool_address);
835 core_client
836 .subscription_manager
837 .unsubscribe_mints(dex, pool_address);
838 Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
839 Self::update_hypersync_pool_event_stream(core_client, dex).await?;
840 } else {
841 anyhow::bail!(
842 "Invalid venue {}, expected Blockchain DEX format",
843 cmd.instrument_id.venue
844 )
845 }
846
847 Ok(())
848 }
849 DefiUnsubscribeCommand::PoolFeeCollects(cmd) => {
850 log::debug!(
851 "Processing unsubscribe pool fee collects command for {}",
852 cmd.instrument_id
853 );
854
855 if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
856 let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
857 .map_err(|_| {
858 anyhow::anyhow!(
859 "Invalid pool fee collect address: {}",
860 cmd.instrument_id
861 )
862 })?;
863 core_client
864 .subscription_manager
865 .unsubscribe_collects(dex, pool_address);
866 Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
867 Self::update_hypersync_pool_event_stream(core_client, dex).await?;
868 } else {
869 anyhow::bail!(
870 "Invalid venue {}, expected Blockchain DEX format",
871 cmd.instrument_id.venue
872 )
873 }
874
875 Ok(())
876 }
877 DefiUnsubscribeCommand::PoolFlashEvents(cmd) => {
878 log::debug!(
879 "Processing unsubscribe pool flash command for {}",
880 cmd.instrument_id
881 );
882
883 if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
884 let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
885 .map_err(|_| {
886 anyhow::anyhow!("Invalid pool flash address: {}", cmd.instrument_id)
887 })?;
888 core_client
889 .subscription_manager
890 .unsubscribe_flashes(dex, pool_address);
891 Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
892 Self::update_hypersync_pool_event_stream(core_client, dex).await?;
893 } else {
894 anyhow::bail!(
895 "Invalid venue {}, expected Blockchain DEX format",
896 cmd.instrument_id.venue
897 )
898 }
899
900 Ok(())
901 }
902 }
903 }
904
905 async fn update_rpc_pool_event_subscriptions(
906 core_client: &mut BlockchainDataClientCore,
907 dex: DexType,
908 ) -> anyhow::Result<()> {
909 let updates = vec![
910 (
911 RpcEventType::PoolSwap(dex),
912 core_client
913 .subscription_manager
914 .get_subscribed_pool_swap_addresses(&dex),
915 core_client
916 .subscription_manager
917 .get_dex_pool_swap_event_signature(&dex),
918 ),
919 (
920 RpcEventType::PoolMint(dex),
921 core_client
922 .subscription_manager
923 .get_subscribed_pool_mint_addresses(&dex),
924 core_client
925 .subscription_manager
926 .get_dex_pool_mint_event_signature(&dex),
927 ),
928 (
929 RpcEventType::PoolBurn(dex),
930 core_client
931 .subscription_manager
932 .get_subscribed_pool_burn_addresses(&dex),
933 core_client
934 .subscription_manager
935 .get_dex_pool_burn_event_signature(&dex),
936 ),
937 (
938 RpcEventType::PoolCollect(dex),
939 core_client
940 .subscription_manager
941 .get_subscribed_pool_collect_addresses(&dex),
942 core_client
943 .subscription_manager
944 .get_dex_pool_collect_event_signature(&dex),
945 ),
946 (
947 RpcEventType::PoolFlash(dex),
948 core_client
949 .subscription_manager
950 .get_subscribed_pool_flash_addresses(&dex),
951 core_client
952 .subscription_manager
953 .get_dex_pool_flash_event_signature(&dex),
954 ),
955 (
956 RpcEventType::PoolFeeProtocolUpdate(dex),
957 core_client
958 .subscription_manager
959 .get_subscribed_pool_fee_protocol_update_addresses(&dex),
960 core_client
961 .subscription_manager
962 .get_dex_pool_fee_protocol_update_event_signature(&dex),
963 ),
964 (
965 RpcEventType::PoolFeeProtocolCollect(dex),
966 core_client
967 .subscription_manager
968 .get_subscribed_pool_fee_protocol_collect_addresses(&dex),
969 core_client
970 .subscription_manager
971 .get_dex_pool_fee_protocol_collect_event_signature(&dex),
972 ),
973 ];
974
975 let has_pool_event_subscriptions = Self::has_active_pool_event_subscriptions(core_client);
976
977 if core_client.rpc_client.is_none() {
978 return Ok(());
979 }
980
981 if has_pool_event_subscriptions {
982 Self::subscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
983 }
984
985 if let Some(ref mut rpc) = core_client.rpc_client {
986 for (event_type, addresses, event_signature) in updates {
987 if let Some(event_signature) = event_signature {
988 rpc.subscribe_pool_events(event_type, &addresses, event_signature)
989 .await?;
990 }
991 }
992 }
993
994 if !has_pool_event_subscriptions {
995 Self::unsubscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
996 }
997
998 Ok(())
999 }
1000
1001 async fn update_hypersync_pool_event_stream(
1002 core_client: &mut BlockchainDataClientCore,
1003 dex: DexType,
1004 ) -> anyhow::Result<()> {
1005 if core_client.rpc_client.is_some() {
1006 return Ok(());
1007 }
1008
1009 let addresses = core_client
1010 .subscription_manager
1011 .get_subscribed_dex_contract_addresses(&dex);
1012 let event_signatures = core_client
1013 .subscription_manager
1014 .get_active_subscribed_dex_event_signatures(&dex);
1015 let has_pool_event_subscriptions = Self::has_active_pool_event_subscriptions(core_client);
1016
1017 if has_pool_event_subscriptions {
1018 Self::subscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
1019 }
1020
1021 core_client
1022 .hypersync_client
1023 .update_dex_event_stream(dex, addresses, event_signatures)
1024 .await;
1025
1026 if !has_pool_event_subscriptions {
1027 Self::unsubscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
1028 }
1029
1030 Ok(())
1031 }
1032
1033 fn has_active_pool_event_subscriptions(core_client: &BlockchainDataClientCore) -> bool {
1034 core_client
1035 .subscription_manager
1036 .has_pool_event_subscriptions()
1037 }
1038
1039 async fn subscribe_block_feed(
1040 core_client: &mut BlockchainDataClientCore,
1041 owner: BlockFeedOwner,
1042 ) -> anyhow::Result<()> {
1043 let preferred_backend = if core_client.rpc_client.is_some() {
1044 BlockFeedBackend::Rpc
1045 } else {
1046 BlockFeedBackend::HyperSync
1047 };
1048 let Some(backend) = core_client
1049 .subscription_manager
1050 .add_block_demand(owner, preferred_backend)
1051 else {
1052 return Ok(());
1053 };
1054
1055 let started_backend = match backend {
1056 BlockFeedBackend::Rpc => {
1057 let Some(rpc) = core_client.rpc_client.as_mut() else {
1058 anyhow::bail!("RPC block feed selected without an RPC client")
1059 };
1060
1061 match rpc.subscribe_blocks().await {
1062 Ok(()) => {
1063 log::debug!("Successfully subscribed to blocks via RPC");
1064 BlockFeedBackend::Rpc
1065 }
1066 Err(e) if owner == BlockFeedOwner::Explicit => {
1067 log::warn!(
1068 "RPC blocks subscription failed: {e}, falling back to HyperSync"
1069 );
1070 core_client.hypersync_client.subscribe_blocks();
1071 tokio::task::yield_now().await;
1072 BlockFeedBackend::HyperSync
1073 }
1074 Err(e) => return Err(e.into()),
1075 }
1076 }
1077 BlockFeedBackend::HyperSync => {
1078 log::debug!("Subscribing to blocks via HyperSync");
1079 core_client.hypersync_client.subscribe_blocks();
1080 tokio::task::yield_now().await;
1081 BlockFeedBackend::HyperSync
1082 }
1083 };
1084
1085 core_client
1086 .subscription_manager
1087 .block_feed_started(started_backend);
1088 Ok(())
1089 }
1090
1091 async fn unsubscribe_block_feed(
1092 core_client: &mut BlockchainDataClientCore,
1093 owner: BlockFeedOwner,
1094 ) -> anyhow::Result<()> {
1095 let Some(backend) = core_client.subscription_manager.remove_block_demand(owner) else {
1096 log::debug!("Keeping block subscription active while another owner remains");
1097 return Ok(());
1098 };
1099
1100 match backend {
1101 BlockFeedBackend::Rpc => {
1102 let Some(rpc) = core_client.rpc_client.as_mut() else {
1103 anyhow::bail!("RPC block feed active without an RPC client")
1104 };
1105 rpc.unsubscribe_blocks().await?;
1106 log::debug!("Unsubscribed from blocks via RPC");
1107 }
1108 BlockFeedBackend::HyperSync => {
1109 core_client.hypersync_client.unsubscribe_blocks().await;
1110 log::debug!("Unsubscribed from blocks via HyperSync");
1111 }
1112 }
1113
1114 core_client.subscription_manager.block_feed_stopped(backend);
1115 Ok(())
1116 }
1117
1118 async fn handle_request_command(
1120 command: DefiRequestCommand,
1121 core_client: &mut BlockchainDataClientCore,
1122 ) -> anyhow::Result<()> {
1123 match command {
1124 DefiRequestCommand::PoolSnapshot(cmd) => {
1125 log::debug!("Processing pool snapshot request for {}", cmd.instrument_id);
1126
1127 let pool_address =
1128 validate_address(cmd.instrument_id.symbol.as_str()).map_err(|e| {
1129 anyhow::anyhow!(
1130 "Invalid pool address '{}' failed with error: {:?}",
1131 cmd.instrument_id,
1132 e
1133 )
1134 })?;
1135
1136 let pool_identifier =
1137 PoolIdentifier::Address(Ustr::from(&pool_address.to_string()));
1138
1139 match core_client.get_pool(&pool_identifier) {
1140 Ok(pool) => {
1141 let pool = pool.clone();
1142 log::debug!("Found pool for snapshot request: {}", cmd.instrument_id);
1143
1144 let pool_data = DataEvent::DeFi(DefiData::Pool(pool.as_ref().clone()));
1146 core_client.send_data(pool_data);
1147
1148 match core_client
1149 .bootstrap_latest_pool_profiler(&pool, None)
1150 .await
1151 {
1152 Ok((profiler, already_valid)) => match profiler.extract_snapshot() {
1153 Ok(snapshot) => {
1154 log::debug!(
1155 "Saving pool snapshot with {} positions and {} ticks to database...",
1156 snapshot.positions.len(),
1157 snapshot.ticks.len()
1158 );
1159 core_client
1160 .cache
1161 .add_pool_snapshot(
1162 &pool.dex.name,
1163 &pool.pool_identifier,
1164 &snapshot,
1165 )
1166 .await?;
1167
1168 if core_client
1170 .check_snapshot_validity(&profiler, already_valid)
1171 .await?
1172 .is_usable()
1173 {
1174 let snapshot_data =
1175 DataEvent::DeFi(DefiData::PoolSnapshot(snapshot));
1176 core_client.send_data(snapshot_data);
1177 }
1178 }
1179 Err(e) => log::error!(
1180 "Failed to extract snapshot for {}: {e}",
1181 cmd.instrument_id
1182 ),
1183 },
1184 Err(e) => log::error!(
1185 "Failed to bootstrap pool profiler for {} and extract snapshot with error {e}",
1186 cmd.instrument_id
1187 ),
1188 }
1189 }
1190 Err(e) => {
1191 log::warn!("Pool {} not found in cache: {e}", cmd.instrument_id);
1192 }
1193 }
1194
1195 Ok(())
1196 }
1197 }
1198 }
1199
1200 pub async fn await_process_task_close(&self) -> anyhow::Result<()> {
1209 self.session_tasks
1210 .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2))
1211 .await
1212 .map_err(|e| anyhow::anyhow!("Blockchain process task shutdown failed: {e}"))?;
1213 Ok(())
1214 }
1215
1216 async fn prepare_task_group(&mut self) -> anyhow::Result<()> {
1217 if !self.session_tasks.is_open() {
1218 self.session_tasks.begin_shutdown();
1219 self.await_process_task_close().await?;
1220 self.reset_channels();
1221 self.session_tasks
1222 .start_generation()
1223 .map_err(|e| anyhow::anyhow!("Failed to start blockchain task generation: {e}"))?;
1224 self.cancellation_token = self.session_tasks.cancellation_token();
1225 }
1226 Ok(())
1227 }
1228
1229 fn reset_channels(&mut self) {
1230 let (hypersync_tx, hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
1231 self.hypersync_tx = Some(hypersync_tx);
1232 self.hypersync_rx = Some(hypersync_rx);
1233 let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
1234 self.command_tx = command_tx;
1235 self.command_rx = Some(command_rx);
1236 }
1237}
1238
1239#[async_trait::async_trait(?Send)]
1240impl DataClient for BlockchainDataClient {
1241 fn client_id(&self) -> ClientId {
1242 self.client_id
1243 }
1244
1245 fn venue(&self) -> Option<Venue> {
1246 None
1249 }
1250
1251 fn start(&mut self) -> anyhow::Result<()> {
1252 log::info!(
1253 "Starting blockchain data client: chain_name={}, dex_ids={:?}, use_hypersync_for_live_data={}, proxy_url={:?}",
1254 self.chain.name,
1255 self.config.dex_ids,
1256 self.config.use_hypersync_for_live_data,
1257 self.config.proxy_url
1258 );
1259 Ok(())
1260 }
1261
1262 fn stop(&mut self) -> anyhow::Result<()> {
1263 log::info!(
1264 "Stopping blockchain data client for '{chain_name}'",
1265 chain_name = self.chain.name
1266 );
1267 self.session_tasks.begin_shutdown();
1268 Ok(())
1269 }
1270
1271 fn reset(&mut self) -> anyhow::Result<()> {
1272 log::info!(
1273 "Resetting blockchain data client for '{chain_name}'",
1274 chain_name = self.chain.name
1275 );
1276 self.session_tasks.begin_shutdown();
1277 Ok(())
1278 }
1279
1280 fn dispose(&mut self) -> anyhow::Result<()> {
1281 log::info!(
1282 "Disposing blockchain data client for '{chain_name}'",
1283 chain_name = self.chain.name
1284 );
1285 self.session_tasks.begin_shutdown();
1286 Ok(())
1287 }
1288
1289 async fn connect(&mut self) -> anyhow::Result<()> {
1290 if self.session_tasks.is_open() && !self.session_tasks.is_empty() {
1291 return Ok(());
1292 }
1293
1294 log::info!(
1295 "Connecting blockchain data client for '{}'",
1296 self.chain.name
1297 );
1298
1299 self.prepare_task_group().await?;
1300 let setup_guard = TaskGroupGuard::new(&[&self.session_tasks], || {});
1301 let startup = self.spawn_process_task()?;
1302 if let Err(e) = startup
1303 .await
1304 .context("Blockchain process task stopped before startup")?
1305 {
1306 self.session_tasks.begin_shutdown();
1307 let teardown_result = self.await_process_task_close().await;
1308 self.reset_channels();
1309
1310 if let Err(teardown_error) = teardown_result {
1311 return Err(e.context(format!(
1312 "Blockchain data startup teardown failed: {teardown_error}"
1313 )));
1314 }
1315 return Err(e);
1316 }
1317
1318 setup_guard.disarm();
1319 Ok(())
1320 }
1321
1322 async fn disconnect(&mut self) -> anyhow::Result<()> {
1323 log::info!(
1324 "Disconnecting blockchain data client for '{}'",
1325 self.chain.name
1326 );
1327
1328 self.session_tasks.begin_shutdown();
1329 let tasks_result = self.await_process_task_close().await;
1330 self.reset_channels();
1331
1332 tasks_result
1333 }
1334
1335 fn is_connected(&self) -> bool {
1336 self.session_tasks.is_open()
1337 && !self.session_tasks.is_empty()
1338 && !self.session_tasks.all_finished()
1339 }
1340
1341 fn is_disconnected(&self) -> bool {
1342 !self.is_connected()
1343 }
1344
1345 fn subscribe_blocks(&mut self, cmd: SubscribeBlocks) -> anyhow::Result<()> {
1346 let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::Blocks(cmd));
1347 self.command_tx.send(command)?;
1348 Ok(())
1349 }
1350
1351 fn subscribe_pool(&mut self, cmd: SubscribePool) -> anyhow::Result<()> {
1352 let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::Pool(cmd));
1353 self.command_tx.send(command)?;
1354 Ok(())
1355 }
1356
1357 fn subscribe_pool_swaps(&mut self, cmd: SubscribePoolSwaps) -> anyhow::Result<()> {
1358 let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolSwaps(cmd));
1359 self.command_tx.send(command)?;
1360 Ok(())
1361 }
1362
1363 fn subscribe_pool_liquidity_updates(
1364 &mut self,
1365 cmd: SubscribePoolLiquidityUpdates,
1366 ) -> anyhow::Result<()> {
1367 let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolLiquidityUpdates(cmd));
1368 self.command_tx.send(command)?;
1369 Ok(())
1370 }
1371
1372 fn subscribe_pool_fee_collects(&mut self, cmd: SubscribePoolFeeCollects) -> anyhow::Result<()> {
1373 let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolFeeCollects(cmd));
1374 self.command_tx.send(command)?;
1375 Ok(())
1376 }
1377
1378 fn subscribe_pool_flash_events(&mut self, cmd: SubscribePoolFlashEvents) -> anyhow::Result<()> {
1379 let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolFlashEvents(cmd));
1380 self.command_tx.send(command)?;
1381 Ok(())
1382 }
1383
1384 fn unsubscribe_blocks(&mut self, cmd: &UnsubscribeBlocks) -> anyhow::Result<()> {
1385 let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::Blocks(cmd.clone()));
1386 self.command_tx.send(command)?;
1387 Ok(())
1388 }
1389
1390 fn unsubscribe_pool(&mut self, cmd: &UnsubscribePool) -> anyhow::Result<()> {
1391 let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::Pool(cmd.clone()));
1392 self.command_tx.send(command)?;
1393 Ok(())
1394 }
1395
1396 fn unsubscribe_pool_swaps(&mut self, cmd: &UnsubscribePoolSwaps) -> anyhow::Result<()> {
1397 let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolSwaps(cmd.clone()));
1398 self.command_tx.send(command)?;
1399 Ok(())
1400 }
1401
1402 fn unsubscribe_pool_liquidity_updates(
1403 &mut self,
1404 cmd: &UnsubscribePoolLiquidityUpdates,
1405 ) -> anyhow::Result<()> {
1406 let command =
1407 DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolLiquidityUpdates(cmd.clone()));
1408 self.command_tx.send(command)?;
1409 Ok(())
1410 }
1411
1412 fn unsubscribe_pool_fee_collects(
1413 &mut self,
1414 cmd: &UnsubscribePoolFeeCollects,
1415 ) -> anyhow::Result<()> {
1416 let command =
1417 DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolFeeCollects(cmd.clone()));
1418 self.command_tx.send(command)?;
1419 Ok(())
1420 }
1421
1422 fn unsubscribe_pool_flash_events(
1423 &mut self,
1424 cmd: &UnsubscribePoolFlashEvents,
1425 ) -> anyhow::Result<()> {
1426 let command =
1427 DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolFlashEvents(cmd.clone()));
1428 self.command_tx.send(command)?;
1429 Ok(())
1430 }
1431
1432 fn request_pool_snapshot(&self, cmd: RequestPoolSnapshot) -> anyhow::Result<()> {
1433 let command = DefiDataCommand::Request(DefiRequestCommand::PoolSnapshot(cmd));
1434 self.command_tx.send(command)?;
1435 Ok(())
1436 }
1437}
1438
1439#[cfg(test)]
1440mod tests {
1441 use std::{sync::Arc, time::Duration};
1442
1443 use alloy::primitives::{I256, U160, U256, address};
1444 use nautilus_common::defi::RequestPoolSnapshot;
1445 use nautilus_core::{UUID4, UnixNanos};
1446 use nautilus_model::{
1447 defi::{Block, Blockchain, Chain, DexType, Pool, PoolIdentifier, Token},
1448 identifiers::{ClientId, InstrumentId},
1449 };
1450 use rstest::rstest;
1451 use tokio_util::sync::CancellationToken;
1452
1453 use super::*;
1454 use crate::events::{flash::FlashEvent, swap::SwapEvent};
1455
1456 const WETH_USDT_CREATION_BLOCK: u64 = 12_375_326;
1457
1458 #[tokio::test(flavor = "multi_thread")]
1459 #[ignore = "requires ENVIO_API_TOKEN and live HyperSync access"]
1460 async fn pool_snapshot_request_does_not_emit_snapshot_when_bootstrap_fails() {
1461 std::env::var("ENVIO_API_TOKEN").expect("ENVIO_API_TOKEN must be set");
1462
1463 let pool = weth_usdt_pool();
1464 let instrument_id = pool.instrument_id;
1465 let (hypersync_tx, _hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
1466 let (data_tx, mut data_rx) = tokio::sync::mpsc::unbounded_channel();
1467 let config = BlockchainDataClientConfig::builder()
1468 .chain(pool.chain.clone())
1469 .dex_ids(vec![DexType::UniswapV3])
1470 .http_rpc_url("http://127.0.0.1:9".to_string())
1471 .use_hypersync_for_live_data(true)
1472 .maybe_from_block(Some(WETH_USDT_CREATION_BLOCK))
1473 .build();
1474 let mut core = BlockchainDataClientCore::new(
1475 config,
1476 Some(hypersync_tx),
1477 Some(data_tx),
1478 CancellationToken::new(),
1479 );
1480 core.cache
1481 .add_pool(pool.as_ref().clone())
1482 .await
1483 .expect("Pool should be added to in-memory cache");
1484
1485 let request = RequestPoolSnapshot::new(
1486 instrument_id,
1487 Some(ClientId::new("BLOCKCHAIN")),
1488 UUID4::new(),
1489 UnixNanos::default(),
1490 None,
1491 );
1492
1493 BlockchainDataClient::handle_request_command(
1494 DefiRequestCommand::PoolSnapshot(request),
1495 &mut core,
1496 )
1497 .await
1498 .expect("Bootstrap failure should not fail the request handler");
1499
1500 let mut events = Vec::new();
1501 while let Ok(event) = data_rx.try_recv() {
1502 events.push(event);
1503 }
1504
1505 assert_eq!(events.len(), 1);
1506 match &events[0] {
1507 DataEvent::DeFi(DefiData::Pool(pool)) => {
1508 assert_eq!(pool.instrument_id, instrument_id);
1509 }
1510 _ => panic!("expected only the pool definition event"),
1511 }
1512 assert!(
1513 events
1514 .iter()
1515 .all(|event| !matches!(event, DataEvent::DeFi(DefiData::PoolSnapshot(_))))
1516 );
1517 }
1518
1519 #[rstest]
1520 fn pool_event_missing_block_metadata_clears_after_block_cache_update() {
1521 let chain = Arc::new(
1522 Chain::from_chain_id(1)
1523 .expect("Ethereum chain should exist")
1524 .clone(),
1525 );
1526 let mut cache = BlockchainCache::new(chain);
1527 let msg = flash_message(42);
1528
1529 assert_eq!(
1530 BlockchainDataClient::pool_event_block_number(&msg),
1531 Some(42)
1532 );
1533 assert_eq!(
1534 BlockchainDataClient::pool_event_missing_block_metadata(&msg, &cache),
1535 Some(42)
1536 );
1537
1538 let block = test_block(42, "0x1", 1_700_000_000_000_000_000);
1539 cache.cache_block_metadata(&block);
1540
1541 assert_eq!(
1542 BlockchainDataClient::pool_event_missing_block_metadata(&msg, &cache),
1543 None
1544 );
1545 }
1546
1547 #[rstest]
1548 fn queue_pending_pool_message_drops_oldest_entry_at_cap() {
1549 let mut pending = VecDeque::new();
1550
1551 for block_number in 0..MAX_PENDING_POOL_MESSAGES {
1552 BlockchainDataClient::queue_pending_pool_message(
1553 &mut pending,
1554 swap_message(block_number as u64),
1555 );
1556 }
1557
1558 BlockchainDataClient::queue_pending_pool_message(
1559 &mut pending,
1560 swap_message(MAX_PENDING_POOL_MESSAGES as u64),
1561 );
1562
1563 assert_eq!(pending.len(), MAX_PENDING_POOL_MESSAGES);
1564 assert_eq!(
1565 BlockchainDataClient::pool_event_block_number(pending.front().unwrap()),
1566 Some(1)
1567 );
1568 assert_eq!(
1569 BlockchainDataClient::pool_event_block_number(pending.back().unwrap()),
1570 Some(MAX_PENDING_POOL_MESSAGES as u64)
1571 );
1572 }
1573
1574 #[rstest]
1575 fn drain_pending_pool_messages_releases_events_after_block_metadata_is_cached() {
1576 let chain = Arc::new(
1577 Chain::from_chain_id(1)
1578 .expect("Ethereum chain should exist")
1579 .clone(),
1580 );
1581 let mut cache = BlockchainCache::new(chain);
1582 let block_42 = test_block(42, "0x42", 1_700_000_000_000_000_000);
1583 cache.cache_block_metadata(&block_42);
1584 let mut pending = VecDeque::from([flash_message(41), flash_message(42)]);
1585
1586 let ready_messages = BlockchainDataClient::drain_pending_pool_messages_with_cached_metadata(
1587 &mut pending,
1588 &cache,
1589 );
1590
1591 assert_eq!(ready_messages.len(), 1);
1592 assert_eq!(
1593 BlockchainDataClient::pool_event_block_number(&ready_messages[0]),
1594 Some(42)
1595 );
1596 assert_eq!(pending.len(), 1);
1597 assert_eq!(
1598 BlockchainDataClient::pool_event_block_number(pending.front().unwrap()),
1599 Some(41)
1600 );
1601
1602 let block_41 = test_block(41, "0x41", 1_700_000_000_000_000_001);
1603 cache.cache_block_metadata(&block_41);
1604
1605 let ready_messages = BlockchainDataClient::drain_pending_pool_messages_with_cached_metadata(
1606 &mut pending,
1607 &cache,
1608 );
1609
1610 assert_eq!(ready_messages.len(), 1);
1611 assert_eq!(
1612 BlockchainDataClient::pool_event_block_number(&ready_messages[0]),
1613 Some(41)
1614 );
1615 assert!(pending.is_empty());
1616 }
1617
1618 #[rstest]
1619 fn ready_live_blockchain_message_queues_until_block_metadata_is_cached() {
1620 let chain = Arc::new(
1621 Chain::from_chain_id(1)
1622 .expect("Ethereum chain should exist")
1623 .clone(),
1624 );
1625 let mut cache = BlockchainCache::new(chain);
1626 let mut pending = VecDeque::new();
1627
1628 let message = BlockchainDataClient::ready_live_blockchain_message(
1629 flash_message(42),
1630 &cache,
1631 &mut pending,
1632 );
1633
1634 assert!(message.is_none());
1635 assert_eq!(pending.len(), 1);
1636 assert_eq!(
1637 BlockchainDataClient::pool_event_block_number(pending.front().unwrap()),
1638 Some(42)
1639 );
1640
1641 let block = test_block(42, "0x1", 1_700_000_000_000_000_000);
1642 cache.cache_block_metadata(&block);
1643
1644 let message = BlockchainDataClient::ready_live_blockchain_message(
1645 BlockchainMessage::Block(block),
1646 &cache,
1647 &mut pending,
1648 );
1649 let ready_messages = BlockchainDataClient::drain_pending_pool_messages_with_cached_metadata(
1650 &mut pending,
1651 &cache,
1652 );
1653
1654 assert!(matches!(message, Some(BlockchainMessage::Block(_))));
1655 assert_eq!(ready_messages.len(), 1);
1656 assert_eq!(
1657 BlockchainDataClient::pool_event_block_number(&ready_messages[0]),
1658 Some(42)
1659 );
1660 assert!(pending.is_empty());
1661 }
1662
1663 #[tokio::test(flavor = "multi_thread")]
1664 #[ignore = "requires ENVIO_API_TOKEN and live HyperSync access"]
1665 async fn live_hypersync_pool_swap_subscription_receives_tip_event_and_unsubscribes() {
1666 std::env::var("ENVIO_API_TOKEN").expect("ENVIO_API_TOKEN must be set");
1667
1668 let chain = Arc::new(
1669 Chain::from_chain_id(42161)
1670 .expect("Arbitrum chain should exist")
1671 .clone(),
1672 );
1673 let dex_extended = get_dex_extended(chain.name, &DexType::UniswapV3)
1674 .expect("Arbitrum UniswapV3 should be registered");
1675 let pool_address = address!("C31E54c7A869B9FcBEcc14363CF510d1c41fa443");
1676 let instrument_id_value = format!("{}.Arbitrum:UniswapV3", pool_address.to_checksum(None));
1677 let instrument_id = InstrumentId::from(instrument_id_value.as_str());
1678 let expected_pool_id = PoolIdentifier::from_address(pool_address);
1679 let (hypersync_tx, mut hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
1680 let config = BlockchainDataClientConfig::builder()
1681 .chain(chain)
1682 .dex_ids(vec![DexType::UniswapV3])
1683 .http_rpc_url("http://127.0.0.1:9".to_string())
1684 .use_hypersync_for_live_data(true)
1685 .build();
1686 let mut core = BlockchainDataClientCore::new(
1687 config,
1688 Some(hypersync_tx),
1689 None,
1690 CancellationToken::new(),
1691 );
1692 core.cache
1693 .add_dex(dex_extended.dex.clone())
1694 .await
1695 .expect("DEX should be added to in-memory cache");
1696 core.subscription_manager.register_dex_for_subscriptions(
1697 DexType::UniswapV3,
1698 dex_extended.swap_created_event.as_ref(),
1699 dex_extended.mint_created_event.as_ref(),
1700 dex_extended.burn_created_event.as_ref(),
1701 dex_extended.collect_created_event.as_ref(),
1702 dex_extended.flash_created_event.as_deref(),
1703 );
1704 core.subscription_manager.register_dex_fee_protocol_events(
1705 DexType::UniswapV3,
1706 dex_extended.fee_protocol_update_event.as_deref(),
1707 dex_extended.fee_protocol_collect_event.as_deref(),
1708 );
1709
1710 BlockchainDataClient::handle_subscribe_command(
1711 DefiSubscribeCommand::PoolSwaps(SubscribePoolSwaps::new(
1712 instrument_id,
1713 Some(ClientId::new("BLOCKCHAIN")),
1714 UUID4::new(),
1715 UnixNanos::default(),
1716 None,
1717 )),
1718 &mut core,
1719 )
1720 .await
1721 .expect("live HyperSync pool swap subscribe should succeed");
1722
1723 let event = tokio::time::timeout(Duration::from_secs(240), async {
1724 loop {
1725 let msg = hypersync_rx
1726 .recv()
1727 .await
1728 .expect("HyperSync live stream channel should stay open");
1729
1730 if let BlockchainMessage::SwapEvent(event) = msg
1731 && event.pool_identifier == expected_pool_id
1732 {
1733 break event;
1734 }
1735 }
1736 })
1737 .await
1738 .expect("expected a live Arbitrum UniswapV3 swap within 240s");
1739
1740 BlockchainDataClient::handle_unsubscribe_command(
1741 DefiUnsubscribeCommand::PoolSwaps(UnsubscribePoolSwaps::new(
1742 instrument_id,
1743 Some(ClientId::new("BLOCKCHAIN")),
1744 UUID4::new(),
1745 UnixNanos::default(),
1746 None,
1747 )),
1748 &mut core,
1749 )
1750 .await
1751 .expect("live HyperSync pool swap unsubscribe should succeed");
1752 core.disconnect().await;
1753
1754 assert_eq!(event.pool_identifier, expected_pool_id);
1755 assert!(event.block_number > 0);
1756 }
1757
1758 fn swap_message(block_number: u64) -> BlockchainMessage {
1759 let pool = weth_usdt_pool();
1760 let address = address!("1111111111111111111111111111111111111111");
1761
1762 BlockchainMessage::SwapEvent(SwapEvent::new(
1763 pool.dex.clone(),
1764 pool.pool_identifier,
1765 block_number,
1766 "0x1".to_string(),
1767 0,
1768 0,
1769 address,
1770 address,
1771 I256::ZERO,
1772 I256::ZERO,
1773 U160::ZERO,
1774 0,
1775 0,
1776 ))
1777 }
1778
1779 fn test_block(number: u64, hash: &str, timestamp: u64) -> Block {
1780 Block::new(
1781 hash.to_string(),
1782 "0x0".to_string(),
1783 number,
1784 Ustr::from("0x0000000000000000000000000000000000000000"),
1785 30_000_000,
1786 21_000,
1787 UnixNanos::from(timestamp),
1788 Some(Blockchain::Ethereum),
1789 )
1790 }
1791
1792 fn flash_message(block_number: u64) -> BlockchainMessage {
1793 let pool = weth_usdt_pool();
1794 let address = address!("1111111111111111111111111111111111111111");
1795
1796 BlockchainMessage::FlashEvent(FlashEvent::new(
1797 pool.dex.clone(),
1798 pool.pool_identifier,
1799 block_number,
1800 "0x1".to_string(),
1801 0,
1802 0,
1803 address,
1804 address,
1805 U256::ZERO,
1806 U256::ZERO,
1807 U256::ZERO,
1808 U256::ZERO,
1809 ))
1810 }
1811
1812 fn weth_usdt_pool() -> Arc<Pool> {
1813 let chain = Arc::new(
1814 Chain::from_chain_id(1)
1815 .expect("Ethereum chain should exist")
1816 .clone(),
1817 );
1818 let dex = get_dex_extended(chain.name, &DexType::UniswapV3)
1819 .expect("Ethereum UniswapV3 should be registered")
1820 .dex
1821 .clone();
1822 let pool_address = address!("4e68ccd3e89f51c3074ca5072bbac773960dfa36");
1823 let token0 = Token::new(
1824 chain.clone(),
1825 address!("C02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
1826 "Wrapped Ether".to_string(),
1827 "WETH".to_string(),
1828 18,
1829 );
1830 let token1 = Token::new(
1831 chain.clone(),
1832 address!("dAC17F958D2ee523a2206206994597C13D831ec7"),
1833 "Tether USD".to_string(),
1834 "USDT".to_string(),
1835 6,
1836 );
1837
1838 Arc::new(Pool::new(
1839 chain,
1840 dex,
1841 pool_address,
1842 PoolIdentifier::from_address(pool_address),
1843 WETH_USDT_CREATION_BLOCK,
1844 token0,
1845 token1,
1846 Some(3_000),
1847 Some(60),
1848 UnixNanos::default(),
1849 ))
1850 }
1851}