nautilus_blockchain/hypersync/
client.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use std::{
17    env::VarError,
18    sync::{
19        Arc,
20        atomic::{AtomicU64, Ordering},
21    },
22    time::Duration,
23};
24
25use ahash::AHashMap;
26use alloy::primitives::Address;
27use anyhow::Context;
28use futures_util::Stream;
29use hypersync_client::{
30    StreamConfig,
31    net_types::{BlockField, BlockSelection, FieldSelection, Query},
32    simple_types::Log,
33};
34use nautilus_common::config::{ConfigError, check_non_empty_field};
35use nautilus_core::{consts::NAUTILUS_USER_AGENT, hex};
36use nautilus_live::task::{TaskJoinOutcome, TaskSlot, finish_task};
37use nautilus_model::{
38    defi::{Block, Blockchain, DexType, SharedChain},
39    identifiers::InstrumentId,
40};
41
42use crate::{
43    exchanges::{extended::DexExtended, get_dex_extended},
44    hypersync::transform::transform_hypersync_block,
45    rpc::{http::validate_execution_endpoint, types::BlockchainMessage},
46};
47
48const API_TOKEN_ENV: &str = "ENVIO_API_TOKEN";
49const API_TOKEN_FORMAT: &str = "a UUID";
50
51/// An item yielded by the contract-events stream.
52///
53/// Blocks are surfaced ahead of the logs from the same response so callers can populate their
54/// block-timestamp cache before converting events from those blocks.
55#[derive(Debug)]
56pub enum PoolEventStreamItem {
57    /// A block referenced by subsequent logs.
58    Block(Block),
59    /// A contract event log.
60    Log(Log),
61}
62
63/// The interval in milliseconds at which to check for new blocks when waiting
64/// for the hypersync to index the block.
65const BLOCK_POLLING_INTERVAL_MS: u64 = 50;
66
67/// Timeout in seconds for HyperSync HTTP requests.
68const HYPERSYNC_REQUEST_TIMEOUT_SECS: u64 = 30;
69
70/// Timeout in seconds for graceful task shutdown during disconnect.
71/// If the task doesn't finish within this time, it will be forcefully aborted.
72const DISCONNECT_TIMEOUT_SECS: u64 = 5;
73
74/// Delay before restarting a DEX event stream after it reaches the current indexed tip.
75const DEX_EVENT_STREAM_RETRY_DELAY_MS: u64 = 1_000;
76
77/// A client for interacting with a HyperSync API to retrieve blockchain data.
78#[derive(Debug)]
79pub struct HyperSyncClient {
80    /// The target blockchain identifier (e.g. Ethereum, Arbitrum).
81    chain: SharedChain,
82    /// The underlying HyperSync Rust client for making API requests.
83    client: Arc<hypersync_client::Client>,
84    /// Background task handle for the block subscription task.
85    blocks_task: TaskSlot<()>,
86    /// Cancellation token for the blocks subscription task.
87    blocks_cancellation_token: Option<tokio_util::sync::CancellationToken>,
88    /// Background DEX event stream tasks keyed by DEX type.
89    dex_event_tasks: AHashMap<DexType, DexEventStreamTask>,
90    /// Channel for sending blockchain messages to the adapter data client.
91    tx: Option<tokio::sync::mpsc::UnboundedSender<BlockchainMessage>>,
92    /// Index of pool addressed keyed by instrument ID.
93    pool_addresses: AHashMap<InstrumentId, Address>,
94    /// Cancellation token for graceful shutdown of background tasks.
95    cancellation_token: tokio_util::sync::CancellationToken,
96}
97
98impl HyperSyncClient {
99    /// Creates a new [`HyperSyncClient`] instance for the given chain and message sender.
100    ///
101    /// # Errors
102    ///
103    /// Returns an error if the HyperSync endpoint or `ENVIO_API_TOKEN` is invalid, the token
104    /// is missing, or the underlying client cannot be initialized.
105    pub fn new(
106        chain: SharedChain,
107        tx: Option<tokio::sync::mpsc::UnboundedSender<BlockchainMessage>>,
108        cancellation_token: tokio_util::sync::CancellationToken,
109    ) -> anyhow::Result<Self> {
110        let mut config = hypersync_client::ClientConfig::default();
111
112        let hypersync_url = validate_execution_endpoint(chain.hypersync_url.as_str(), "HyperSync")
113            .map_err(|e| ConfigError::invalid_value("chain.hypersync_url", e.to_string()))?;
114
115        config.url = hypersync_url.to_string();
116        config.api_token = std::env::var(API_TOKEN_ENV).map_err(|e| match e {
117            VarError::NotPresent => ConfigError::missing_field(API_TOKEN_ENV),
118            VarError::NotUnicode(_) => ConfigError::invalid_format(API_TOKEN_ENV, API_TOKEN_FORMAT),
119        })?;
120
121        check_non_empty_field(API_TOKEN_ENV, &config.api_token)?;
122
123        config
124            .validate()
125            .map_err(|_| ConfigError::invalid_format(API_TOKEN_ENV, API_TOKEN_FORMAT))?;
126        let client = hypersync_client::Client::new_with_agent(config, NAUTILUS_USER_AGENT)
127            .context("failed to create HyperSync client")?;
128
129        Ok(Self {
130            chain,
131            client: Arc::new(client),
132            blocks_task: TaskSlot::new(),
133            blocks_cancellation_token: None,
134            dex_event_tasks: AHashMap::new(),
135            tx,
136            pool_addresses: AHashMap::new(),
137            cancellation_token,
138        })
139    }
140
141    #[must_use]
142    pub fn get_pool_address(&self, instrument_id: InstrumentId) -> Option<&Address> {
143        self.pool_addresses.get(&instrument_id)
144    }
145
146    /// Starts, refreshes, or stops the live DEX event stream for one DEX.
147    ///
148    /// The stream query is open-ended (`to_block = None`) and the background task resumes from the
149    /// last HyperSync `next_block` whenever the SDK stream reaches the current indexed tip.
150    pub async fn update_dex_event_stream(
151        &mut self,
152        dex: DexType,
153        contract_addresses: Vec<Address>,
154        event_signatures: Vec<String>,
155    ) {
156        let filter = DexEventStreamFilter::new(contract_addresses, event_signatures);
157
158        if filter.is_empty() {
159            self.stop_dex_event_stream(dex).await;
160            return;
161        }
162
163        if self.dex_event_tasks.get(&dex).is_some_and(|task| {
164            task.filter == filter
165                && task
166                    .task
167                    .as_ref()
168                    .is_some_and(|handle| !handle.is_finished())
169        }) {
170            return;
171        }
172
173        let next_from_block = self.stop_dex_event_stream(dex).await;
174        if self.dex_event_tasks.contains_key(&dex) {
175            log::error!("Previous HyperSync DEX event stream for {dex} is still stopping");
176            return;
177        }
178
179        let from_block = match next_from_block {
180            Some(block) => block,
181            None => match self.client.get_height().await {
182                Ok(block) => block,
183                Err(e) => {
184                    log::error!("Failed to get HyperSync height for DEX event stream: {e}");
185                    return;
186                }
187            },
188        };
189
190        let tx = if let Some(tx) = &self.tx {
191            tx.clone()
192        } else {
193            log::error!("Hypersync client channel should have been initialized");
194            return;
195        };
196
197        let client = Arc::clone(&self.client);
198        let chain = self.chain.name;
199        let Some(dex_extended) = get_dex_extended(chain, &dex) else {
200            log::error!("Failed to get DEX registration for {dex} on {chain}");
201            return;
202        };
203        let stream_token = self.cancellation_token.child_token();
204        let task_token = stream_token.clone();
205        let next_from_block = Arc::new(AtomicU64::new(from_block));
206        let task_next_from_block = Arc::clone(&next_from_block);
207        let task_filter = filter.clone();
208
209        let mut task = TaskSlot::new();
210        if let Err(e) = task.spawn(async move {
211            Self::run_dex_event_stream(
212                dex,
213                client,
214                tx,
215                task_filter,
216                dex_extended,
217                task_next_from_block,
218                task_token,
219            )
220            .await;
221        }) {
222            stream_token.cancel();
223            log::error!("Failed to start HyperSync DEX event stream for {dex}: {e}");
224        }
225
226        self.dex_event_tasks.insert(
227            dex,
228            DexEventStreamTask {
229                filter,
230                next_from_block,
231                cancellation_token: stream_token,
232                task,
233            },
234        );
235    }
236
237    /// Creates a stream of contract event logs matching the specified criteria.
238    ///
239    /// # Panics
240    ///
241    /// Panics if the contract address cannot be parsed as a valid Ethereum address.
242    pub async fn request_contract_events_stream(
243        &self,
244        from_block: u64,
245        to_block: Option<u64>,
246        contract_address: &Address,
247        topics: Vec<&str>,
248    ) -> impl Stream<Item = PoolEventStreamItem> + use<> {
249        let query = Self::construct_contract_events_query(
250            from_block,
251            to_block,
252            &[*contract_address],
253            &topics,
254        );
255
256        let chain = self.chain.name;
257        let mut rx = self
258            .client
259            .stream(query, StreamConfig::default())
260            .await
261            .expect("Failed to create stream");
262
263        async_stream::stream! {
264              while let Some(response) = rx.recv().await {
265                let response = response.unwrap();
266                for item in pool_events_from_response(chain, response.data.blocks, response.data.logs) {
267                    yield item;
268                }
269            }
270        }
271    }
272
273    /// Disconnects from the HyperSync service and stops all background tasks.
274    pub async fn disconnect(&mut self) {
275        log::debug!("Disconnecting HyperSync client");
276        self.cancellation_token.cancel();
277
278        if let Some(outcome) = finish_task(
279            &mut self.blocks_task,
280            Duration::from_secs(DISCONNECT_TIMEOUT_SECS),
281            Duration::from_secs(DISCONNECT_TIMEOUT_SECS),
282        )
283        .await
284        {
285            match outcome {
286                TaskJoinOutcome::Completed(()) | TaskJoinOutcome::Aborted => {}
287                TaskJoinOutcome::Failed(error) => {
288                    log::error!("HyperSync blocks task failed: {error}");
289                }
290                TaskJoinOutcome::Incomplete => {
291                    log::error!("HyperSync blocks task did not stop after abort");
292                }
293            }
294        }
295
296        let dexes = self.dex_event_tasks.keys().copied().collect::<Vec<_>>();
297        for dex in dexes {
298            self.stop_dex_event_stream(dex).await;
299        }
300
301        log::debug!("HyperSync client disconnected");
302    }
303
304    /// Returns the current block
305    ///
306    /// # Panics
307    ///
308    /// Panics if the client height request fails.
309    pub async fn current_block(&self) -> u64 {
310        self.client.get_height().await.unwrap()
311    }
312
313    /// Creates a stream that yields blockchain blocks within the specified range.
314    ///
315    /// # Panics
316    ///
317    /// Panics if the stream creation or block transformation fails.
318    pub async fn request_blocks_stream(
319        &self,
320        from_block: u64,
321        to_block: Option<u64>,
322    ) -> impl Stream<Item = Block> {
323        let query = Self::construct_block_query(from_block, to_block);
324        let mut rx = self
325            .client
326            .stream(query, StreamConfig::default())
327            .await
328            .unwrap();
329
330        let chain = self.chain.name;
331
332        async_stream::stream! {
333            while let Some(response) = rx.recv().await {
334                let response = response.unwrap();
335                for received_block in response.data.blocks.into_iter().flatten() {
336                    let block = transform_hypersync_block(chain, received_block).unwrap();
337                    yield block
338                }
339            }
340        }
341    }
342
343    /// Starts a background task that continuously polls for new blockchain blocks.
344    ///
345    /// # Panics
346    ///
347    /// Panics if client height requests or block transformations fail.
348    pub fn subscribe_blocks(&mut self) {
349        if self.blocks_task.is_some() {
350            return;
351        }
352
353        let chain = self.chain.name;
354        let client = Arc::clone(&self.client);
355        let tx = if let Some(tx) = &self.tx {
356            tx.clone()
357        } else {
358            log::error!("Hypersync client channel should have been initialized");
359            return;
360        };
361
362        // Create a child token that can be cancelled independently
363        let blocks_token = self.cancellation_token.child_token();
364        let cancellation_token = blocks_token.clone();
365        self.blocks_cancellation_token = Some(blocks_token);
366
367        if let Err(e) = self.blocks_task.spawn(async move {
368            log::debug!("Starting task 'blocks_feed");
369
370            let current_block_height = client.get_height().await.unwrap();
371            let mut query = Self::construct_block_query(current_block_height, None);
372
373            loop {
374                tokio::select! {
375                    () = cancellation_token.cancelled() => {
376                        log::debug!("Blocks subscription task received cancellation signal");
377                        break;
378                    }
379                    result = tokio::time::timeout(
380                        Duration::from_secs(HYPERSYNC_REQUEST_TIMEOUT_SECS),
381                        client.get(&query)
382                    ) => {
383                        let response = match result {
384                            Ok(Ok(resp)) => resp,
385                            Ok(Err(e)) => {
386                                log::error!("Hypersync request failed: {e}");
387                                break;
388                            }
389                            Err(_) => {
390                                log::warn!("Hypersync request timed out after {HYPERSYNC_REQUEST_TIMEOUT_SECS}s, retrying...");
391                                continue;
392                            }
393                        };
394
395                        for received_block in response.data.blocks.into_iter().flatten() {
396                            let block = transform_hypersync_block(chain, received_block).unwrap();
397                            let msg = BlockchainMessage::Block(block);
398                            if let Err(e) = tx.send(msg) {
399                                log::error!("Error sending message: {e}");
400                            }
401                        }
402
403                        if let Some(archive_block_height) = response.archive_height
404                            && archive_block_height < response.next_block
405                        {
406                            while client.get_height().await.unwrap() < response.next_block {
407                                tokio::select! {
408                                    () = cancellation_token.cancelled() => {
409                                        log::debug!("Blocks subscription task received cancellation signal during polling");
410                                        return;
411                                    }
412                                    () = tokio::time::sleep(Duration::from_millis(
413                                        BLOCK_POLLING_INTERVAL_MS,
414                                    )) => {}
415                                }
416                            }
417                        }
418
419                        query.from_block = response.next_block;
420                    }
421                }
422            }
423        }) {
424            if let Some(token) = self.blocks_cancellation_token.take() {
425                token.cancel();
426            }
427            log::error!("Failed to start HyperSync blocks subscription task: {e}");
428        }
429    }
430
431    /// Unsubscribes from new blocks by stopping the background watch task.
432    pub async fn unsubscribe_blocks(&mut self) {
433        if self.blocks_task.is_none() {
434            return;
435        }
436
437        // Cancel only the blocks child token, not the main cancellation token
438        if let Some(token) = self.blocks_cancellation_token.take() {
439            token.cancel();
440        }
441
442        if let Some(outcome) = finish_task(
443            &mut self.blocks_task,
444            Duration::from_secs(DISCONNECT_TIMEOUT_SECS),
445            Duration::from_secs(DISCONNECT_TIMEOUT_SECS),
446        )
447        .await
448        {
449            match outcome {
450                TaskJoinOutcome::Completed(()) | TaskJoinOutcome::Aborted => {}
451                TaskJoinOutcome::Failed(error) => {
452                    log::error!("HyperSync blocks task failed during unsubscribe: {error}");
453                }
454                TaskJoinOutcome::Incomplete => {
455                    log::error!("HyperSync blocks task did not stop after abort");
456                }
457            }
458        }
459        log::debug!("Unsubscribed from blocks");
460    }
461
462    /// Constructs a HyperSync query for fetching blocks with all available fields within the specified range.
463    fn construct_block_query(from_block: u64, to_block: Option<u64>) -> Query {
464        Query {
465            from_block,
466            to_block: Self::to_hypersync_exclusive_bound(to_block),
467            blocks: vec![BlockSelection::default()],
468            field_selection: FieldSelection {
469                block: BlockField::all(),
470                ..Default::default()
471            },
472            ..Default::default()
473        }
474    }
475
476    fn construct_contract_events_query(
477        from_block: u64,
478        to_block: Option<u64>,
479        contract_addresses: &[Address],
480        topics: &[&str],
481    ) -> Query {
482        let mut query_value = serde_json::json!({
483            "from_block": from_block,
484            "logs": [{
485                "topics": [topics],
486                "address": contract_addresses
487            }],
488            "field_selection": {
489                "log": [
490                    "block_number",
491                    "transaction_hash",
492                    "transaction_index",
493                    "log_index",
494                    "address",
495                    "data",
496                    "topic0",
497                    "topic1",
498                    "topic2",
499                    "topic3",
500                ],
501                // Join block fields so callers can resolve each event's ts_event
502                "block": [
503                    "number",
504                    "hash",
505                    "parent_hash",
506                    "miner",
507                    "gas_limit",
508                    "gas_used",
509                    "timestamp",
510                ]
511            }
512        });
513
514        if let Some(to_block) = Self::to_hypersync_exclusive_bound(to_block)
515            && let Some(obj) = query_value.as_object_mut()
516        {
517            obj.insert("to_block".to_string(), serde_json::json!(to_block));
518        }
519
520        serde_json::from_value(query_value).unwrap()
521    }
522
523    fn to_hypersync_exclusive_bound(to_block: Option<u64>) -> Option<u64> {
524        to_block.map(|block| block.saturating_add(1))
525    }
526
527    async fn run_dex_event_stream(
528        dex: DexType,
529        client: Arc<hypersync_client::Client>,
530        tx: tokio::sync::mpsc::UnboundedSender<BlockchainMessage>,
531        filter: DexEventStreamFilter,
532        dex_extended: &'static DexExtended,
533        next_from_block: Arc<AtomicU64>,
534        cancellation_token: tokio_util::sync::CancellationToken,
535    ) {
536        log::debug!("Starting task 'dex_event_stream' for {dex}");
537
538        loop {
539            let from_block = next_from_block.load(Ordering::Relaxed);
540            Self::wait_for_stream_start_block(&client, from_block, &cancellation_token).await;
541            if cancellation_token.is_cancelled() {
542                break;
543            }
544
545            let topics = filter
546                .event_signatures
547                .iter()
548                .map(String::as_str)
549                .collect::<Vec<_>>();
550            let query = Self::construct_contract_events_query(
551                from_block,
552                None,
553                &filter.contract_addresses,
554                &topics,
555            );
556            let mut rx = match client.stream(query, StreamConfig::default()).await {
557                Ok(rx) => rx,
558                Err(e) => {
559                    log::warn!("Failed to create DEX event stream for {dex}: {e}");
560
561                    if !Self::sleep_or_cancel(
562                        Duration::from_millis(DEX_EVENT_STREAM_RETRY_DELAY_MS),
563                        &cancellation_token,
564                    )
565                    .await
566                    {
567                        break;
568                    }
569                    continue;
570                }
571            };
572
573            let mut received_response = false;
574
575            loop {
576                tokio::select! {
577                    () = cancellation_token.cancelled() => {
578                        log::debug!("DEX event stream task for {dex} received cancellation signal");
579                        break;
580                    }
581                    response = rx.recv() => {
582                        let Some(response) = response else {
583                            break;
584                        };
585
586                        let response = match response {
587                            Ok(resp) => resp,
588                            Err(e) => {
589                                if received_response {
590                                    log::debug!("DEX event stream drained for {dex}: {e}");
591                                } else {
592                                    log::warn!("Failed to receive DEX event stream response for {dex}: {e}");
593                                }
594                                break;
595                            }
596                        };
597
598                        received_response = true;
599                        next_from_block.fetch_max(response.next_block, Ordering::Relaxed);
600
601                        for log in response.data.logs.into_iter().flatten() {
602                            Self::send_dex_event_log(&tx, dex_extended, &log);
603                        }
604                    }
605                }
606            }
607
608            if !Self::sleep_or_cancel(
609                Duration::from_millis(DEX_EVENT_STREAM_RETRY_DELAY_MS),
610                &cancellation_token,
611            )
612            .await
613            {
614                break;
615            }
616        }
617
618        log::debug!("Stopped task 'dex_event_stream' for {dex}");
619    }
620
621    async fn wait_for_stream_start_block(
622        client: &hypersync_client::Client,
623        from_block: u64,
624        cancellation_token: &tokio_util::sync::CancellationToken,
625    ) {
626        loop {
627            match client.get_height().await {
628                Ok(height) if height >= from_block => return,
629                Ok(_) => {}
630                Err(e) => log::warn!("Failed to get HyperSync height for DEX event stream: {e}"),
631            }
632
633            if !Self::sleep_or_cancel(
634                Duration::from_millis(BLOCK_POLLING_INTERVAL_MS),
635                cancellation_token,
636            )
637            .await
638            {
639                return;
640            }
641        }
642    }
643
644    async fn sleep_or_cancel(
645        duration: Duration,
646        cancellation_token: &tokio_util::sync::CancellationToken,
647    ) -> bool {
648        tokio::select! {
649            () = cancellation_token.cancelled() => false,
650            () = tokio::time::sleep(duration) => true,
651        }
652    }
653
654    fn send_dex_event_log(
655        tx: &tokio::sync::mpsc::UnboundedSender<BlockchainMessage>,
656        dex_extended: &DexExtended,
657        log: &Log,
658    ) {
659        let event_signature = match log.topics.first().and_then(|t| t.as_ref()) {
660            Some(log_argument) => hex::encode_prefixed(log_argument.as_ref()),
661            None => return,
662        };
663
664        let (event_name, event) = if event_signature == dex_extended.swap_created_event.as_ref() {
665            (
666                "swap",
667                dex_extended
668                    .parse_swap_event_hypersync(log)
669                    .map(BlockchainMessage::SwapEvent),
670            )
671        } else if event_signature == dex_extended.mint_created_event.as_ref() {
672            (
673                "mint",
674                dex_extended
675                    .parse_mint_event_hypersync(log)
676                    .map(BlockchainMessage::MintEvent),
677            )
678        } else if event_signature == dex_extended.burn_created_event.as_ref() {
679            (
680                "burn",
681                dex_extended
682                    .parse_burn_event_hypersync(log)
683                    .map(BlockchainMessage::BurnEvent),
684            )
685        } else if event_signature == dex_extended.collect_created_event.as_ref() {
686            (
687                "collect",
688                dex_extended
689                    .parse_collect_event_hypersync(log)
690                    .map(BlockchainMessage::CollectEvent),
691            )
692        } else if dex_extended.flash_created_event.as_deref() == Some(event_signature.as_str()) {
693            (
694                "flash",
695                dex_extended
696                    .parse_flash_event_hypersync(log)
697                    .map(BlockchainMessage::FlashEvent),
698            )
699        } else if dex_extended.fee_protocol_update_event.as_deref()
700            == Some(event_signature.as_str())
701        {
702            (
703                "fee-protocol update",
704                dex_extended
705                    .parse_fee_protocol_update_event_hypersync(log)
706                    .map(BlockchainMessage::FeeProtocolUpdateEvent),
707            )
708        } else if dex_extended.fee_protocol_collect_event.as_deref()
709            == Some(event_signature.as_str())
710        {
711            (
712                "fee-protocol collect",
713                dex_extended
714                    .parse_fee_protocol_collect_event_hypersync(log)
715                    .map(BlockchainMessage::FeeProtocolCollectEvent),
716            )
717        } else {
718            log::error!("Unknown event signature: {event_signature}");
719            return;
720        };
721
722        match event {
723            Ok(event) => {
724                if let Err(e) = tx.send(event) {
725                    log::error!("Failed to send {event_name} event: {e}");
726                }
727            }
728            Err(e) => {
729                log::error!("Failed to parse {event_name} with error '{e:?}' for event: {log:?}",);
730            }
731        }
732    }
733
734    async fn stop_dex_event_stream(&mut self, dex: DexType) -> Option<u64> {
735        let task = self.dex_event_tasks.get_mut(&dex)?;
736        let terminated = Self::stop_dex_event_task(dex, task).await;
737        let next_from_block = task.next_from_block.load(Ordering::Relaxed);
738
739        if terminated {
740            self.dex_event_tasks.remove(&dex);
741        }
742
743        Some(next_from_block)
744    }
745
746    async fn stop_dex_event_task(dex: DexType, task: &mut DexEventStreamTask) -> bool {
747        task.cancellation_token.cancel();
748
749        let outcome = finish_task(
750            &mut task.task,
751            Duration::from_secs(DISCONNECT_TIMEOUT_SECS),
752            Duration::from_secs(DISCONNECT_TIMEOUT_SECS),
753        )
754        .await;
755
756        match outcome {
757            None | Some(TaskJoinOutcome::Completed(()) | TaskJoinOutcome::Aborted) => true,
758            Some(TaskJoinOutcome::Failed(error)) => {
759                log::error!("HyperSync DEX event stream task for {dex} failed: {error}");
760                true
761            }
762            Some(TaskJoinOutcome::Incomplete) => {
763                log::error!("HyperSync DEX event stream task for {dex} did not stop after abort");
764                false
765            }
766        }
767    }
768}
769
770impl Drop for HyperSyncClient {
771    fn drop(&mut self) {
772        self.cancellation_token.cancel();
773        if let Some(token) = self.blocks_cancellation_token.as_ref() {
774            token.cancel();
775        }
776        self.blocks_task.abort();
777
778        for task in self.dex_event_tasks.values_mut() {
779            task.cancellation_token.cancel();
780            task.task.abort();
781        }
782    }
783}
784
785#[derive(Debug, Clone, PartialEq, Eq)]
786struct DexEventStreamFilter {
787    contract_addresses: Vec<Address>,
788    event_signatures: Vec<String>,
789}
790
791impl DexEventStreamFilter {
792    fn new(mut contract_addresses: Vec<Address>, mut event_signatures: Vec<String>) -> Self {
793        contract_addresses.sort_unstable();
794        contract_addresses.dedup();
795        event_signatures.sort_unstable();
796        event_signatures.dedup();
797        Self {
798            contract_addresses,
799            event_signatures,
800        }
801    }
802
803    fn is_empty(&self) -> bool {
804        self.contract_addresses.is_empty() || self.event_signatures.is_empty()
805    }
806}
807
808#[derive(Debug)]
809struct DexEventStreamTask {
810    filter: DexEventStreamFilter,
811    next_from_block: Arc<AtomicU64>,
812    cancellation_token: tokio_util::sync::CancellationToken,
813    task: TaskSlot<()>,
814}
815
816/// Maps one HyperSync response into stream items, surfacing blocks ahead of the logs from the
817/// same response so callers can cache them before converting events from those blocks.
818///
819/// Blocks that fail to transform are logged and skipped without dropping the response's logs.
820fn pool_events_from_response(
821    chain: Blockchain,
822    blocks: Vec<Vec<hypersync_client::simple_types::Block>>,
823    logs: Vec<Vec<Log>>,
824) -> Vec<PoolEventStreamItem> {
825    let mut items = Vec::new();
826
827    for block in blocks.into_iter().flatten() {
828        match transform_hypersync_block(chain, block) {
829            Ok(block) => items.push(PoolEventStreamItem::Block(block)),
830            Err(e) => log::error!("Failed to transform block for timestamp: {e}"),
831        }
832    }
833
834    items.extend(logs.into_iter().flatten().map(PoolEventStreamItem::Log));
835    items
836}
837
838#[cfg(test)]
839mod tests {
840    use std::{
841        str::FromStr,
842        sync::{
843            Arc,
844            atomic::{AtomicU64, Ordering},
845        },
846    };
847
848    use hypersync_client::{
849        format::{Address as HypersyncAddress, Hash, Quantity},
850        simple_types::{Block as HypersyncBlock, Log},
851    };
852    #[cfg(not(madsim))]
853    use nautilus_common::{clients::DataClient, live::runner::replace_data_event_sender};
854    use nautilus_core::{UnixNanos, datetime::NANOSECONDS_IN_SECOND};
855    use nautilus_model::defi::Chain;
856    #[cfg(not(madsim))]
857    use nautilus_model::identifiers::ClientId;
858    use rstest::rstest;
859
860    use super::*;
861    #[cfg(not(madsim))]
862    use crate::{config::BlockchainDataClientConfig, data::client::BlockchainDataClient};
863
864    #[cfg(not(madsim))]
865    const TOKEN_TEST_ENV: &str = "BLOCKCHAIN_HYPERSYNC_TOKEN_TEST";
866
867    #[cfg(not(madsim))]
868    #[rstest]
869    #[case::missing(None, "missing")]
870    #[case::empty(Some(""), "empty")]
871    #[case::malformed(Some("not-a-uuid"), "malformed")]
872    fn test_constructor_rejects_invalid_token(#[case] token: Option<&str>, #[case] expected: &str) {
873        let module = module_path!().split_once("::").unwrap().1;
874        let child_test = format!("{module}::hypersync_constructor_token_child");
875        let executable = std::env::current_exe().unwrap();
876        let mut command = std::process::Command::new(executable);
877        command
878            .args(["--exact", &child_test, "--ignored"])
879            .env(TOKEN_TEST_ENV, expected)
880            .env_remove(API_TOKEN_ENV);
881
882        if let Some(token) = token {
883            command.env(API_TOKEN_ENV, token);
884        }
885
886        let output = command.output().unwrap();
887        let stdout = String::from_utf8(output.stdout).unwrap();
888
889        assert_eq!(output.status.code(), Some(0), "{stdout}");
890        assert!(
891            stdout.contains(&format!("test {child_test} ... ok")),
892            "{stdout}"
893        );
894    }
895
896    #[cfg(not(madsim))]
897    #[ignore = "runs only in the isolated HyperSync token test process"]
898    #[tokio::test]
899    async fn hypersync_constructor_token_child() {
900        let Ok(case) = std::env::var(TOKEN_TEST_ENV) else {
901            return;
902        };
903
904        let expected = match case.as_str() {
905            "missing" => ConfigError::missing_field(API_TOKEN_ENV),
906            "empty" => ConfigError::empty_field(API_TOKEN_ENV),
907            "malformed" => ConfigError::invalid_format(API_TOKEN_ENV, "a UUID"),
908            _ => panic!("unexpected token test case: {case}"),
909        };
910
911        let mut chain = Chain::new(Blockchain::Arbitrum, 42_161);
912        chain.hypersync_url = "https://42161.hypersync.xyz".to_string();
913        let chain = Arc::new(chain);
914        let error = HyperSyncClient::new(
915            Arc::clone(&chain),
916            None,
917            tokio_util::sync::CancellationToken::new(),
918        )
919        .unwrap_err();
920        let config = BlockchainDataClientConfig::builder()
921            .chain(chain)
922            .http_rpc_url("http://127.0.0.1:1".into())
923            .build();
924        let mut client = BlockchainDataClient::new(ClientId::from("BLOCKCHAIN"), config).unwrap();
925        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
926        replace_data_event_sender(sender);
927
928        assert_eq!(error.downcast_ref::<ConfigError>(), Some(&expected));
929
930        for _ in 0..2 {
931            let error = client.connect().await.unwrap_err();
932            assert_eq!(error.downcast_ref::<ConfigError>(), Some(&expected));
933        }
934    }
935
936    fn synthetic_block(number: u64, timestamp_secs: u64) -> HypersyncBlock {
937        HypersyncBlock {
938            number: Some(number),
939            hash: Some(
940                Hash::from_str(
941                    "0x0000000000000000000000000000000000000000000000000000000000000001",
942                )
943                .unwrap(),
944            ),
945            parent_hash: Some(
946                Hash::from_str(
947                    "0x0000000000000000000000000000000000000000000000000000000000000000",
948                )
949                .unwrap(),
950            ),
951            miner: Some(
952                HypersyncAddress::from_str("0x0000000000000000000000000000000000000001").unwrap(),
953            ),
954            gas_limit: Some(Quantity::from(21_000u64)),
955            gas_used: Some(Quantity::from(21_000u64)),
956            timestamp: Some(Quantity::from(timestamp_secs)),
957            ..Default::default()
958        }
959    }
960
961    #[rstest]
962    fn test_constructor_rejects_remote_cleartext_before_loading_token() {
963        let mut chain = Chain::new(Blockchain::Arbitrum, 42_161);
964        chain.hypersync_url = "http://localhost:8080".to_string();
965
966        let error = HyperSyncClient::new(
967            Arc::new(chain),
968            None,
969            tokio_util::sync::CancellationToken::new(),
970        )
971        .unwrap_err();
972
973        assert_eq!(
974            error.downcast_ref::<ConfigError>(),
975            Some(&ConfigError::invalid_value(
976                "chain.hypersync_url",
977                "HyperSync endpoint must use HTTPS unless its host is a canonical loopback IP literal",
978            )),
979        );
980    }
981
982    #[rstest]
983    fn pool_events_yields_blocks_before_logs() {
984        let items = pool_events_from_response(
985            Blockchain::Ethereum,
986            vec![vec![synthetic_block(12, 100)]],
987            vec![vec![Log::default(), Log::default()]],
988        );
989
990        assert_eq!(items.len(), 3);
991        match &items[0] {
992            PoolEventStreamItem::Block(block) => {
993                assert_eq!(block.number, 12);
994                assert_eq!(block.timestamp, UnixNanos::new(100 * NANOSECONDS_IN_SECOND));
995            }
996            other => panic!("expected Block first, was {other:?}"),
997        }
998        assert!(matches!(items[1], PoolEventStreamItem::Log(_)));
999        assert!(matches!(items[2], PoolEventStreamItem::Log(_)));
1000    }
1001
1002    #[rstest]
1003    fn pool_events_skips_unparsable_block_but_keeps_logs() {
1004        // A block missing required fields (gas, hash, ...) fails transform and is skipped, but
1005        // the response's logs must still be yielded.
1006        let bad_block = HypersyncBlock {
1007            number: Some(7),
1008            ..Default::default()
1009        };
1010        let items = pool_events_from_response(
1011            Blockchain::Ethereum,
1012            vec![vec![bad_block]],
1013            vec![vec![Log::default()]],
1014        );
1015
1016        assert_eq!(items.len(), 1);
1017        assert!(matches!(items[0], PoolEventStreamItem::Log(_)));
1018    }
1019
1020    #[rstest]
1021    fn construct_block_query_converts_to_block_to_hypersync_exclusive_bound() {
1022        let query = HyperSyncClient::construct_block_query(10, Some(12));
1023
1024        assert_eq!(query.from_block, 10);
1025        assert_eq!(query.to_block, Some(13));
1026    }
1027
1028    #[rstest]
1029    fn construct_contract_events_query_converts_to_block_to_hypersync_exclusive_bound() {
1030        let address = Address::from_str("0x0000000000000000000000000000000000000001").unwrap();
1031        let query = HyperSyncClient::construct_contract_events_query(
1032            10,
1033            Some(12),
1034            &[address],
1035            &["0xc42079f94a6350d7e6235f29174924f928cc2ac818eb64fed8004e115fbcca67"],
1036        );
1037
1038        assert_eq!(query.from_block, 10);
1039        assert_eq!(query.to_block, Some(13));
1040    }
1041
1042    #[rstest]
1043    fn construct_contract_events_query_single_block_uses_next_block_as_exclusive_bound() {
1044        let address = Address::from_str("0x0000000000000000000000000000000000000001").unwrap();
1045        let query = HyperSyncClient::construct_contract_events_query(
1046            10,
1047            Some(10),
1048            &[address],
1049            &["0xc42079f94a6350d7e6235f29174924f928cc2ac818eb64fed8004e115fbcca67"],
1050        );
1051
1052        assert_eq!(query.from_block, 10);
1053        assert_eq!(query.to_block, Some(11));
1054    }
1055
1056    #[rstest]
1057    fn construct_contract_events_query_open_upper_bound_stays_open() {
1058        let address = Address::from_str("0x0000000000000000000000000000000000000001").unwrap();
1059        let query = HyperSyncClient::construct_contract_events_query(
1060            10,
1061            None,
1062            &[address],
1063            &["0xc42079f94a6350d7e6235f29174924f928cc2ac818eb64fed8004e115fbcca67"],
1064        );
1065
1066        assert_eq!(query.from_block, 10);
1067        assert_eq!(query.to_block, None);
1068    }
1069
1070    #[rstest]
1071    fn dex_event_stream_filter_sorts_and_deduplicates_inputs() {
1072        let address1 = Address::from_str("0x0000000000000000000000000000000000000001").unwrap();
1073        let address2 = Address::from_str("0x0000000000000000000000000000000000000002").unwrap();
1074        let filter = DexEventStreamFilter::new(
1075            vec![address2, address1, address2],
1076            vec![
1077                "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".to_string(),
1078                "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string(),
1079                "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".to_string(),
1080            ],
1081        );
1082
1083        assert_eq!(filter.contract_addresses, vec![address1, address2]);
1084        assert_eq!(
1085            filter.event_signatures,
1086            vec![
1087                "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string(),
1088                "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".to_string(),
1089            ]
1090        );
1091    }
1092
1093    #[rstest]
1094    fn dex_event_stream_filter_requires_addresses_and_signatures() {
1095        let address = Address::from_str("0x0000000000000000000000000000000000000001").unwrap();
1096
1097        assert!(DexEventStreamFilter::new(vec![], vec!["0x01".to_string()]).is_empty());
1098        assert!(DexEventStreamFilter::new(vec![address], vec![]).is_empty());
1099        assert!(!DexEventStreamFilter::new(vec![address], vec!["0x01".to_string()]).is_empty());
1100    }
1101
1102    #[tokio::test]
1103    async fn stop_dex_event_task_cancels_and_awaits_task() {
1104        let cancellation_token = tokio_util::sync::CancellationToken::new();
1105        let task_token = cancellation_token.clone();
1106        let next_from_block = Arc::new(AtomicU64::new(42));
1107        let task_next_from_block = Arc::clone(&next_from_block);
1108
1109        let task = tokio::spawn(async move {
1110            task_token.cancelled().await;
1111            task_next_from_block.store(99, Ordering::Relaxed);
1112        });
1113        let stream_task = DexEventStreamTask {
1114            filter: DexEventStreamFilter::new(vec![], vec![]),
1115            next_from_block,
1116            cancellation_token,
1117            task: TaskSlot::from_handle(task),
1118        };
1119        let mut stream_task = stream_task;
1120
1121        let terminated =
1122            HyperSyncClient::stop_dex_event_task(DexType::UniswapV3, &mut stream_task).await;
1123
1124        assert!(terminated);
1125        assert_eq!(stream_task.next_from_block.load(Ordering::Relaxed), 99);
1126        assert!(stream_task.task.is_none());
1127    }
1128}
nautilus_blockchain/hypersync/client.rs

nautilus_blockchain/hypersync/
client.rs
