nautilus_binance/common/
execution.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
16//! Shared execution client utilities for Binance Spot and Futures adapters.
17
18use std::{
19    future::Future,
20    time::{Duration, Instant},
21};
22
23use nautilus_live::{ExecutionClientCore, task::TaskGroup};
24use nautilus_model::identifiers::AccountId;
25
26/// Spawns an async task and tracks its handle in `pending_tasks`.
27pub fn spawn_task<F>(pending_tasks: &TaskGroup, description: &'static str, fut: F)
28where
29    F: Future<Output = anyhow::Result<()>> + Send + 'static,
30{
31    let future = async move {
32        if let Err(e) = fut.await {
33            log::warn!("{description} failed: {e}");
34        }
35    };
36
37    if let Err(e) = pending_tasks.spawn(future) {
38        log::warn!("Skipping Binance {description} after shutdown began: {e}");
39    }
40}
41
42/// Aborts all pending tasks stored in `pending_tasks`.
43pub fn abort_pending_tasks(pending_tasks: &TaskGroup) {
44    pending_tasks.begin_shutdown();
45}
46
47/// Completes bounded shutdown for Binance command tasks.
48///
49/// # Errors
50///
51/// Returns an error when bounded task shutdown fails.
52pub async fn await_pending_tasks(pending_tasks: &TaskGroup) -> anyhow::Result<()> {
53    pending_tasks.begin_shutdown();
54    pending_tasks
55        .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
56        .await
57        .map_err(|e| anyhow::anyhow!("Failed to terminate Binance execution tasks: {e}"))?;
58    Ok(())
59}
60
61/// Polls the cache until the account is registered or timeout is reached.
62///
63/// Each iteration borrows and drops the cache Ref to avoid holding the
64/// RefCell borrow across await points, which would block mutable access
65/// when the account state is registered by another task.
66///
67/// # Errors
68///
69/// Returns an error if the timeout is reached before the account is registered.
70pub async fn await_account_registered(
71    core: &ExecutionClientCore,
72    account_id: AccountId,
73    timeout_secs: f64,
74) -> anyhow::Result<()> {
75    if core.cache().account(&account_id).is_some() {
76        log::info!("Account {account_id} registered");
77        return Ok(());
78    }
79
80    let start = Instant::now();
81    let timeout = Duration::from_secs_f64(timeout_secs);
82    let interval = Duration::from_millis(10);
83
84    loop {
85        tokio::time::sleep(interval).await;
86
87        if core.cache().account(&account_id).is_some() {
88            log::info!("Account {account_id} registered");
89            return Ok(());
90        }
91
92        if start.elapsed() >= timeout {
93            anyhow::bail!(
94                "Timeout waiting for account {account_id} to be registered after {timeout_secs}s"
95            );
96        }
97    }
98}
99
100#[cfg(test)]
101mod tests {
102    use std::{cell::RefCell, rc::Rc};
103
104    use nautilus_common::cache::Cache;
105    use nautilus_live::ExecutionClientCore;
106    use nautilus_model::{
107        accounts::{AccountAny, CashAccount},
108        enums::{AccountType, OmsType},
109        events::AccountState,
110        identifiers::{AccountId, TraderId},
111        types::{AccountBalance, Money},
112    };
113    use rstest::rstest;
114
115    use super::*;
116    use crate::common::consts::{BINANCE_CLIENT_ID, BINANCE_VENUE};
117
118    #[rstest]
119    #[tokio::test]
120    async fn test_spawn_task_unregisters_finished_task_before_shutdown() {
121        let pending_tasks = TaskGroup::new();
122
123        spawn_task(&pending_tasks, "test task", async { Ok(()) });
124        tokio::time::timeout(Duration::from_secs(1), async {
125            while !pending_tasks.all_finished() {
126                tokio::task::yield_now().await;
127            }
128        })
129        .await
130        .expect("task should finish");
131
132        assert!(pending_tasks.is_empty());
133        abort_pending_tasks(&pending_tasks);
134        await_pending_tasks(&pending_tasks)
135            .await
136            .expect("task shutdown");
137        assert!(pending_tasks.is_empty());
138    }
139
140    #[rstest]
141    #[tokio::test]
142    async fn test_abort_pending_tasks_aborts_running_tasks() {
143        let (drop_tx, drop_rx) = tokio::sync::oneshot::channel();
144        let guard = AbortDropSignal { tx: Some(drop_tx) };
145
146        let pending_tasks = TaskGroup::new();
147        pending_tasks
148            .spawn(async move {
149                let _guard = guard;
150                tokio::time::sleep(Duration::from_secs(60)).await;
151            })
152            .expect("task spawn");
153
154        abort_pending_tasks(&pending_tasks);
155        await_pending_tasks(&pending_tasks)
156            .await
157            .expect("task shutdown");
158
159        assert!(pending_tasks.is_empty());
160        tokio::time::timeout(Duration::from_secs(1), drop_rx)
161            .await
162            .expect("Aborted task should drop its future")
163            .expect("Drop signal should be sent");
164    }
165
166    #[rstest]
167    #[tokio::test]
168    async fn test_await_account_registered_returns_when_account_is_added() {
169        let account_id = AccountId::from("BINANCE-001");
170        let cache = Rc::new(RefCell::new(Cache::default()));
171        let core = create_test_core(cache.clone(), account_id);
172
173        let wait_fut = await_account_registered(&core, account_id, 0.5);
174        let register_fut = async move {
175            tokio::time::sleep(Duration::from_millis(20)).await;
176            add_test_account_to_cache(&cache, account_id);
177        };
178
179        let (result, ()) = tokio::join!(wait_fut, register_fut);
180        result.unwrap();
181    }
182
183    #[rstest]
184    #[tokio::test]
185    async fn test_await_account_registered_times_out() {
186        let account_id = AccountId::from("BINANCE-001");
187        let cache = Rc::new(RefCell::new(Cache::default()));
188        let core = create_test_core(cache, account_id);
189
190        let error = await_account_registered(&core, account_id, 0.02)
191            .await
192            .expect_err("Missing account should time out");
193
194        assert!(error.to_string().contains("BINANCE-001"));
195    }
196
197    struct AbortDropSignal {
198        tx: Option<tokio::sync::oneshot::Sender<()>>,
199    }
200
201    impl Drop for AbortDropSignal {
202        fn drop(&mut self) {
203            if let Some(tx) = self.tx.take() {
204                let _ = tx.send(());
205            }
206        }
207    }
208
209    fn create_test_core(cache: Rc<RefCell<Cache>>, account_id: AccountId) -> ExecutionClientCore {
210        ExecutionClientCore::new(
211            TraderId::from("TESTER-001"),
212            *BINANCE_CLIENT_ID,
213            *BINANCE_VENUE,
214            OmsType::Hedging,
215            account_id,
216            AccountType::Cash,
217            None,
218            cache,
219        )
220    }
221
222    fn add_test_account_to_cache(cache: &Rc<RefCell<Cache>>, account_id: AccountId) {
223        let state = AccountState::new(
224            account_id,
225            AccountType::Cash,
226            vec![AccountBalance::new(
227                Money::from("1.0 BTC"),
228                Money::from("0 BTC"),
229                Money::from("1.0 BTC"),
230            )],
231            vec![],
232            true,
233            nautilus_core::UUID4::new(),
234            nautilus_core::UnixNanos::default(),
235            nautilus_core::UnixNanos::default(),
236            None,
237        );
238
239        let account = AccountAny::Cash(CashAccount::new(state, true, false));
240        cache.borrow_mut().add_account(account).unwrap();
241    }
242}
nautilus_binance/common/execution.rs

nautilus_binance/common/
execution.rs
