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

nautilus_binance/common/
execution.rs
