oanda_client/
admission.rs

1// SPDX-FileCopyrightText: 2026 Kevin Monaghan
2// SPDX-License-Identifier: MIT-0
3
4//! Credential-free admission shared by clients using one caller-selected egress scope.
5
6use crate::{Error, Result};
7use std::{
8    collections::VecDeque,
9    fmt,
10    future::Future,
11    pin::Pin,
12    sync::{Arc, Mutex},
13    task::{Context, Poll},
14    time::Duration,
15};
16use tokio::{sync::Notify, time::Instant};
17use tower_layer::Layer;
18use tower_service::Service;
19
20pub(crate) const REQUESTS_PER_SECOND: u32 = 120;
21const STREAM_LIMIT: usize = 20;
22const ADMISSION_WINDOW: Duration = Duration::from_secs(1);
23
24/// Local admission for clients whose traffic shares one caller-selected egress scope.
25///
26/// Clones share REST request capacity, provider cooldown, stream leases and
27/// connector-start pacing. The owner contains no credentials or account mutation state.
28#[derive(Clone)]
29pub struct Admission {
30    pub(crate) inner: Arc<AdmissionInner>,
31}
32
33impl fmt::Debug for Admission {
34    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
35        f.debug_struct("Admission").finish_non_exhaustive()
36    }
37}
38
39pub(crate) struct AdmissionInner {
40    pub(crate) rate: Mutex<Rate>,
41    policy_changed: Notify,
42    pub(crate) stream_slots: Mutex<StreamSlots>,
43    connection_gate: Arc<ConnectionGate>,
44}
45
46pub(crate) struct Rate {
47    pub(crate) exit_reserve: u32,
48    pub(crate) admitted: VecDeque<Instant>,
49    pub(crate) cooldown_until: Option<Instant>,
50}
51
52pub(crate) struct StreamSlots {
53    pub(crate) active: usize,
54}
55
56impl Default for Admission {
57    fn default() -> Self {
58        Self::new()
59    }
60}
61
62impl Admission {
63    /// Create an independent owner with no exit reserve.
64    #[must_use]
65    pub fn new() -> Self {
66        Self::create(0)
67    }
68
69    /// Create an independent owner with an initial exit reserve.
70    ///
71    /// # Errors
72    ///
73    /// Returns [`Error::InvalidInput`] when `requests` is at least 120.
74    pub fn with_exit_reserve(requests: u32) -> Result<Self> {
75        if requests >= REQUESTS_PER_SECOND {
76            return Err(Error::InvalidInput(
77                "exit reserve must be less than 120 requests".into(),
78            ));
79        }
80        Ok(Self::create(requests))
81    }
82
83    fn create(requests: u32) -> Self {
84        Self {
85            inner: Arc::new(AdmissionInner {
86                rate: Mutex::new(Rate {
87                    exit_reserve: requests,
88                    admitted: VecDeque::new(),
89                    cooldown_until: None,
90                }),
91                policy_changed: Notify::new(),
92                stream_slots: Mutex::new(StreamSlots { active: 0 }),
93                connection_gate: Arc::new(ConnectionGate {
94                    starts: Mutex::new(ConnectionStarts::Empty),
95                    #[cfg(test)]
96                    wait_observer: Mutex::new(None),
97                }),
98            }),
99        }
100    }
101
102    /// Replace the reserve without refunding spent capacity or shortening cooldown.
103    ///
104    /// # Errors
105    ///
106    /// Returns [`Error::InvalidInput`] for a reserve of at least 120, or
107    /// [`Error::Invariant`] for a poisoned lock.
108    pub fn set_exit_reserve(&self, requests: u32) -> Result<()> {
109        if requests >= REQUESTS_PER_SECOND {
110            return Err(Error::InvalidInput(
111                "exit reserve must be less than 120 requests".into(),
112            ));
113        }
114        let mut rate = self
115            .inner
116            .rate
117            .lock()
118            .map_err(|_| Error::Invariant("rate lock poisoned"))?;
119        let changed = rate.exit_reserve != requests;
120        rate.exit_reserve = requests;
121        drop(rate);
122        if changed {
123            self.inner.policy_changed.notify_waiters();
124        }
125        Ok(())
126    }
127
128    pub(crate) fn exit_reserve(&self) -> Result<u32> {
129        self.inner
130            .rate
131            .lock()
132            .map(|rate| rate.exit_reserve)
133            .map_err(|_| Error::Invariant("rate lock poisoned"))
134    }
135
136    pub(crate) async fn admit(&self, mutation: bool, exit: bool) -> Result<()> {
137        loop {
138            let changed = self.inner.policy_changed.notified();
139            tokio::pin!(changed);
140            changed.as_mut().enable();
141            let wait = {
142                let mut rate = self
143                    .inner
144                    .rate
145                    .lock()
146                    .map_err(|_| Error::Invariant("rate lock poisoned"))?;
147                let capacity = if exit {
148                    REQUESTS_PER_SECOND
149                } else {
150                    REQUESTS_PER_SECOND - rate.exit_reserve
151                };
152                let now = Instant::now();
153                while rate
154                    .admitted
155                    .front()
156                    .is_some_and(|at| now.duration_since(*at) >= ADMISSION_WINDOW)
157                {
158                    rate.admitted.pop_front();
159                }
160                if let Some(until) = rate.cooldown_until.filter(|until| *until > now) {
161                    until.duration_since(now)
162                } else if u32::try_from(rate.admitted.len())
163                    .map_err(|_| Error::Invariant("rate admission count exceeds u32"))?
164                    < capacity
165                {
166                    rate.admitted.push_back(now);
167                    return Ok(());
168                } else {
169                    let oldest = rate
170                        .admitted
171                        .front()
172                        .ok_or(Error::Invariant("full rate window has no admission"))?;
173                    ADMISSION_WINDOW.saturating_sub(now.duration_since(*oldest))
174                }
175            };
176            if mutation {
177                return Err(Error::RateLimited);
178            }
179            tokio::select! {
180                () = &mut changed => {},
181                () = tokio::time::sleep(wait) => {},
182            }
183        }
184    }
185
186    pub(crate) fn install_cooldown(&self, cooldown: Duration) -> Result<()> {
187        let mut rate = self
188            .inner
189            .rate
190            .lock()
191            .map_err(|_| Error::Invariant("rate lock poisoned while installing cooldown"))?;
192        let until = Instant::now() + cooldown;
193        rate.cooldown_until = Some(
194            rate.cooldown_until
195                .map_or(until, |existing| existing.max(until)),
196        );
197        Ok(())
198    }
199
200    pub(crate) fn lease_stream(&self) -> Result<StreamLease> {
201        let mut slots = self
202            .inner
203            .stream_slots
204            .lock()
205            .map_err(|_| Error::Invariant("stream slots lock poisoned"))?;
206        if slots.active >= STREAM_LIMIT {
207            return Err(Error::RateLimited);
208        }
209        slots.active += 1;
210        Ok(StreamLease {
211            owner: self.clone(),
212        })
213    }
214
215    pub(crate) fn connection_layer(&self) -> ConnectionLayer {
216        ConnectionLayer {
217            gate: Arc::clone(&self.inner.connection_gate),
218        }
219    }
220}
221
222pub(crate) struct StreamLease {
223    pub(crate) owner: Admission,
224}
225
226impl Drop for StreamLease {
227    fn drop(&mut self) {
228        let mut slots = self
229            .owner
230            .inner
231            .stream_slots
232            .lock()
233            .unwrap_or_else(std::sync::PoisonError::into_inner);
234        slots.active = slots.active.saturating_sub(1);
235    }
236}
237
238#[derive(Clone)]
239pub(crate) struct ConnectionLayer {
240    gate: Arc<ConnectionGate>,
241}
242
243impl<S> Layer<S> for ConnectionLayer {
244    type Service = ConnectionService<S>;
245
246    fn layer(&self, inner: S) -> Self::Service {
247        ConnectionService {
248            inner,
249            gate: Arc::clone(&self.gate),
250        }
251    }
252}
253
254#[derive(Clone)]
255pub(crate) struct ConnectionService<S> {
256    inner: S,
257    gate: Arc<ConnectionGate>,
258}
259
260struct ConnectionGate {
261    starts: Mutex<ConnectionStarts>,
262    #[cfg(test)]
263    wait_observer: Mutex<Option<tokio::sync::mpsc::UnboundedSender<Instant>>>,
264}
265
266enum ConnectionStarts {
267    Empty,
268    One(Instant),
269    Two { oldest: Instant, newest: Instant },
270}
271
272impl ConnectionStarts {
273    fn admit_or_wait(&mut self, now: Instant) -> Option<Instant> {
274        match self {
275            Self::Empty => {
276                *self = Self::One(now);
277                None
278            }
279            Self::One(oldest) if now.duration_since(*oldest) >= ADMISSION_WINDOW => {
280                *oldest = now;
281                None
282            }
283            Self::One(oldest) => {
284                *self = Self::Two {
285                    oldest: *oldest,
286                    newest: now,
287                };
288                None
289            }
290            Self::Two { newest, .. } if now.duration_since(*newest) >= ADMISSION_WINDOW => {
291                *self = Self::One(now);
292                None
293            }
294            Self::Two { oldest, newest } if now.duration_since(*oldest) >= ADMISSION_WINDOW => {
295                *oldest = *newest;
296                *newest = now;
297                None
298            }
299            Self::Two { oldest, .. } => Some(*oldest + ADMISSION_WINDOW),
300        }
301    }
302}
303
304impl ConnectionGate {
305    async fn admit_start(&self) {
306        loop {
307            let wait_until = {
308                // Only this exhaustive, panic-free Instant state transition owns the lock.
309                // Poison recovery retains every previously admitted connector start.
310                let mut starts = self
311                    .starts
312                    .lock()
313                    .unwrap_or_else(std::sync::PoisonError::into_inner);
314                starts.admit_or_wait(Instant::now())
315            };
316            if let Some(until) = wait_until {
317                #[cfg(test)]
318                if until > Instant::now() {
319                    let observer = self
320                        .wait_observer
321                        .lock()
322                        .unwrap_or_else(std::sync::PoisonError::into_inner);
323                    if let Some(observer) = observer.as_ref() {
324                        let _ = observer.send(until);
325                    }
326                }
327                tokio::time::sleep_until(until).await;
328            } else {
329                return;
330            }
331        }
332    }
333}
334
335impl<S, Request> Service<Request> for ConnectionService<S>
336where
337    S: Service<Request>,
338    S::Future: Send + 'static,
339{
340    type Response = S::Response;
341    type Error = S::Error;
342    type Future =
343        Pin<Box<dyn Future<Output = std::result::Result<Self::Response, Self::Error>> + Send>>;
344
345    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
346        self.inner.poll_ready(cx)
347    }
348
349    fn call(&mut self, request: Request) -> Self::Future {
350        let connection = self.inner.call(request);
351        let gate = Arc::clone(&self.gate);
352        Box::pin(async move {
353            gate.admit_start().await;
354            connection.await
355        })
356    }
357}
358
359#[cfg(test)]
360// The fake connector supplies only the transport boundary and its completion signal.
361#[allow(clippy::expect_used, clippy::unwrap_used)]
362mod tests {
363    use super::*;
364    use crate::{AccountID, Client, Environment};
365    use std::convert::Infallible;
366    use tokio::{
367        io::{AsyncReadExt, AsyncWriteExt},
368        net::{TcpListener, TcpStream},
369        sync::{mpsc, oneshot},
370    };
371    use url::Url;
372
373    struct HeldConnector {
374        id: u8,
375        begun: mpsc::UnboundedSender<u8>,
376        release: Option<oneshot::Receiver<()>>,
377    }
378
379    impl Service<()> for HeldConnector {
380        type Response = ();
381        type Error = Infallible;
382        type Future = Pin<Box<dyn Future<Output = std::result::Result<(), Infallible>> + Send>>;
383
384        fn poll_ready(
385            &mut self,
386            _cx: &mut Context<'_>,
387        ) -> Poll<std::result::Result<(), Self::Error>> {
388            Poll::Ready(Ok(()))
389        }
390
391        fn call(&mut self, _request: ()) -> Self::Future {
392            let begun = self.begun.clone();
393            let release = self.release.take().expect("one call");
394            let id = self.id;
395            Box::pin(async move {
396                begun.send(id).expect("observer");
397                let _ = release.await;
398                Ok(())
399            })
400        }
401    }
402
403    async fn loopback() -> (
404        Url,
405        mpsc::UnboundedReceiver<Instant>,
406        tokio::task::JoinHandle<()>,
407    ) {
408        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
409        let url = Url::parse(&format!(
410            "http://{}/",
411            listener.local_addr().expect("address")
412        ))
413        .expect("URL");
414        let (accepted, observations) = mpsc::unbounded_channel();
415        let task = tokio::spawn(async move {
416            while let Ok((socket, _)) = listener.accept().await {
417                accepted.send(Instant::now()).expect("accept receiver");
418                tokio::spawn(serve_socket(socket));
419            }
420        });
421        (url, observations, task)
422    }
423
424    async fn serve_socket(mut socket: TcpStream) {
425        let mut pending = Vec::new();
426        loop {
427            let mut buffer = [0_u8; 4096];
428            let read = socket.read(&mut buffer).await.expect("read");
429            if read == 0 {
430                return;
431            }
432            pending.extend_from_slice(&buffer[..read]);
433            let Some(end) = pending.windows(4).position(|bytes| bytes == b"\r\n\r\n") else {
434                continue;
435            };
436            let stream = pending[..end]
437                .windows(b"/transactions/stream".len())
438                .any(|part| part == b"/transactions/stream");
439            pending.drain(..end + 4);
440            if stream {
441                socket
442                    .write_all(
443                        b"HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\nTransfer-Encoding: chunked\r\n\r\n",
444                    )
445                    .await
446                    .expect("stream header");
447                std::future::pending::<()>().await;
448            }
449            let body = br#"{"accounts":[]}"#;
450            let header = format!(
451                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: keep-alive\r\n\r\n",
452                body.len()
453            );
454            socket.write_all(header.as_bytes()).await.expect("header");
455            socket.write_all(body).await.expect("body");
456        }
457    }
458
459    fn keep_clock_frozen() -> tokio::task::JoinHandle<()> {
460        tokio::spawn(async {
461            loop {
462                tokio::task::yield_now().await;
463            }
464        })
465    }
466
467    #[tokio::test(start_paused = true)]
468    async fn built_rest_and_stream_clients_share_two_connector_starts_per_second() {
469        let clock_guard = keep_clock_frozen();
470        let (url, mut accepts, server) = loopback().await;
471        let owner = Admission::new();
472        let (wait_sender, mut waits) = mpsc::unbounded_channel();
473        *owner
474            .inner
475            .connection_gate
476            .wait_observer
477            .lock()
478            .expect("wait observer") = Some(wait_sender);
479        let first_client = Client::builder(Environment::Practice, "token-a")
480            .admission(owner.clone())
481            .endpoints(url.clone(), url.clone())
482            .build()
483            .expect("first client");
484        let second_client = Client::builder(Environment::Practice, "token-b")
485            .admission(owner)
486            .endpoints(url.clone(), url)
487            .build()
488            .expect("second client");
489        let (stop_watchdog, watchdog_stopped) = std::sync::mpsc::channel();
490        let (expired, mut watchdog_expired) = oneshot::channel();
491        let watchdog = std::thread::spawn(move || {
492            if matches!(
493                watchdog_stopped.recv_timeout(Duration::from_secs(5)),
494                Err(std::sync::mpsc::RecvTimeoutError::Timeout)
495            ) {
496                let _ = expired.send(());
497            }
498        });
499
500        first_client.list_accounts().await.expect("first REST");
501        let first = accepts.recv().await.expect("first accept");
502        first_client
503            .list_accounts()
504            .await
505            .expect("reused REST socket");
506        assert!(accepts.try_recv().is_err());
507        assert!(waits.try_recv().is_err());
508
509        second_client
510            .list_accounts()
511            .await
512            .expect("second REST response");
513        let second = accepts.recv().await.expect("second accept");
514        assert_eq!(second, first);
515        assert!(waits.try_recv().is_err());
516
517        let account = AccountID::new("101-001-1-001").expect("account");
518        let mut stream_task =
519            tokio::spawn(async move { first_client.stream_transactions(&account).await });
520        let stream_deadline = tokio::select! {
521            observed = waits.recv() => observed,
522            _ = &mut stream_task => None,
523            _ = &mut watchdog_expired => None,
524        }
525        .expect("stream reached the connector start budget");
526        assert_eq!(
527            stream_deadline.duration_since(first),
528            Duration::from_secs(1)
529        );
530        tokio::time::advance(Duration::from_millis(999)).await;
531        assert!(accepts.try_recv().is_err());
532        tokio::time::advance(Duration::from_millis(1)).await;
533        let stream = stream_task
534            .await
535            .expect("stream task")
536            .expect("stream response");
537        let third = accepts.recv().await.expect("stream accept");
538        assert_eq!(third, stream_deadline);
539
540        drop(stream);
541        stop_watchdog.send(()).expect("stop watchdog");
542        watchdog.join().expect("watchdog");
543        clock_guard.abort();
544        server.abort();
545    }
546
547    #[tokio::test(start_paused = true)]
548    async fn two_connectors_start_while_held_and_third_waits_for_a_rolling_second() {
549        let owner = Admission::new();
550        let clock_guard = keep_clock_frozen();
551        let (begun, mut observed) = mpsc::unbounded_channel();
552        let (release_first, first_done) = oneshot::channel();
553        let (release_second, second_done) = oneshot::channel();
554        let (release_third, third_done) = oneshot::channel();
555        let mut first = owner.connection_layer().layer(HeldConnector {
556            id: 1,
557            begun: begun.clone(),
558            release: Some(first_done),
559        });
560        let mut second = owner.connection_layer().layer(HeldConnector {
561            id: 2,
562            begun: begun.clone(),
563            release: Some(second_done),
564        });
565        let mut third = owner.connection_layer().layer(HeldConnector {
566            id: 3,
567            begun,
568            release: Some(third_done),
569        });
570        let started = Instant::now();
571        let first_task = tokio::spawn(async move { first.call(()).await });
572        assert_eq!(observed.recv().await, Some(1));
573        let second_task = tokio::spawn(async move { second.call(()).await });
574        assert_eq!(observed.recv().await, Some(2));
575        assert_eq!(Instant::now(), started);
576        let third_task = tokio::spawn(async move { third.call(()).await });
577        tokio::time::advance(Duration::from_millis(999)).await;
578        assert!(observed.try_recv().is_err());
579        tokio::time::advance(Duration::from_millis(1)).await;
580        assert_eq!(observed.recv().await, Some(3));
581        assert_eq!(Instant::now(), started + Duration::from_secs(1));
582        release_first.send(()).expect("release first");
583        release_second.send(()).expect("release second");
584        release_third.send(()).expect("release third");
585        first_task
586            .await
587            .expect("first task")
588            .expect("first connect");
589        second_task
590            .await
591            .expect("second task")
592            .expect("second connect");
593        third_task
594            .await
595            .expect("third task")
596            .expect("third connect");
597        clock_guard.abort();
598    }
599
600    #[tokio::test(start_paused = true)]
601    async fn cancelled_started_connector_keeps_budget_but_queued_cancellation_spends_none() {
602        let owner = Admission::new();
603        let clock_guard = keep_clock_frozen();
604        let (begun, mut observed) = mpsc::unbounded_channel();
605        let (_release_first, first_done) = oneshot::channel();
606        let (_release_second, second_done) = oneshot::channel();
607        let (_release_third, third_done) = oneshot::channel();
608        let (release_fourth, fourth_done) = oneshot::channel();
609        let mut first = owner.connection_layer().layer(HeldConnector {
610            id: 1,
611            begun: begun.clone(),
612            release: Some(first_done),
613        });
614        let mut second = owner.connection_layer().layer(HeldConnector {
615            id: 2,
616            begun: begun.clone(),
617            release: Some(second_done),
618        });
619        let mut third = owner.connection_layer().layer(HeldConnector {
620            id: 3,
621            begun: begun.clone(),
622            release: Some(third_done),
623        });
624        let mut fourth = owner.connection_layer().layer(HeldConnector {
625            id: 4,
626            begun,
627            release: Some(fourth_done),
628        });
629        let first_task = tokio::spawn(async move { first.call(()).await });
630        assert_eq!(observed.recv().await, Some(1));
631        first_task.abort();
632        let _ = first_task.await;
633        let second_task = tokio::spawn(async move { second.call(()).await });
634        assert_eq!(observed.recv().await, Some(2));
635        let third_task = tokio::spawn(async move { third.call(()).await });
636        tokio::time::advance(Duration::from_millis(999)).await;
637        assert!(observed.try_recv().is_err());
638        third_task.abort();
639        let _ = third_task.await;
640        let fourth_task = tokio::spawn(async move { fourth.call(()).await });
641        assert!(observed.try_recv().is_err());
642        tokio::time::advance(Duration::from_millis(1)).await;
643        assert_eq!(observed.recv().await, Some(4));
644        release_fourth.send(()).expect("release fourth");
645        fourth_task
646            .await
647            .expect("fourth task")
648            .expect("fourth connect");
649        second_task.abort();
650        let _ = second_task.await;
651        clock_guard.abort();
652    }
653
654    #[tokio::test(start_paused = true)]
655    async fn shared_rate_and_cooldown_keep_exit_reserve_spent_across_clones() {
656        let owner = Admission::with_exit_reserve(119).expect("reserve");
657        owner.admit(true, false).await.expect("ordinary capacity");
658        assert!(matches!(
659            owner.clone().admit(true, false).await,
660            Err(Error::RateLimited)
661        ));
662        owner
663            .admit(true, true)
664            .await
665            .expect("reserved exit capacity");
666        for _ in 0..118 {
667            owner
668                .admit(true, true)
669                .await
670                .expect("remaining exit capacity");
671        }
672        owner.set_exit_reserve(0).expect("release reserve");
673        assert!(matches!(
674            owner.admit(true, false).await,
675            Err(Error::RateLimited)
676        ));
677        owner
678            .install_cooldown(Duration::from_secs(2))
679            .expect("install cooldown");
680        assert!(matches!(
681            owner.clone().admit(true, true).await,
682            Err(Error::RateLimited)
683        ));
684        assert!(matches!(
685            owner.admit(true, false).await,
686            Err(Error::RateLimited)
687        ));
688        tokio::time::advance(Duration::from_secs(2)).await;
689        owner.admit(true, false).await.expect("cooldown expired");
690    }
691
692    #[tokio::test(start_paused = true)]
693    async fn shared_owner_admits_one_hundred_twenty_rest_requests_and_refuses_the_next_mutation() {
694        let owner = Admission::new();
695        for _ in 0..120 {
696            owner
697                .clone()
698                .admit(true, false)
699                .await
700                .expect("documented REST capacity");
701        }
702        assert!(matches!(
703            owner.admit(true, false).await,
704            Err(Error::RateLimited)
705        ));
706    }
707
708    #[tokio::test(start_paused = true)]
709    async fn rolling_window_does_not_refill_a_late_burst_at_the_first_requests_expiry() {
710        let owner = Admission::new();
711        owner.admit(true, false).await.expect("first request");
712        tokio::time::advance(Duration::from_millis(999)).await;
713        for _ in 0..119 {
714            owner.admit(true, false).await.expect("late burst");
715        }
716        tokio::time::advance(Duration::from_millis(1)).await;
717        owner.admit(true, false).await.expect("one expired request");
718        assert!(matches!(
719            owner.admit(true, false).await,
720            Err(Error::RateLimited)
721        ));
722        tokio::time::advance(Duration::from_millis(998)).await;
723        assert!(matches!(
724            owner.admit(true, false).await,
725            Err(Error::RateLimited)
726        ));
727        tokio::time::advance(Duration::from_millis(1)).await;
728        owner.admit(true, false).await.expect("late burst expired");
729    }
730
731    #[tokio::test(start_paused = true)]
732    async fn reserve_of_one_hundred_nineteen_leaves_one_ordinary_request_and_all_exits() {
733        let owner = Admission::with_exit_reserve(119).expect("largest valid reserve");
734        owner.admit(true, false).await.expect("ordinary request");
735        assert!(matches!(
736            owner.admit(true, false).await,
737            Err(Error::RateLimited)
738        ));
739        for _ in 0..119 {
740            owner.admit(true, true).await.expect("reserved exit");
741        }
742        assert!(matches!(
743            owner.admit(true, true).await,
744            Err(Error::RateLimited)
745        ));
746        assert!(matches!(
747            Admission::with_exit_reserve(120),
748            Err(Error::InvalidInput(_))
749        ));
750    }
751
752    #[tokio::test(start_paused = true)]
753    async fn twenty_stream_leases_start_together_and_the_next_is_unsent() {
754        let owner = Admission::new();
755        let mut leases = Vec::new();
756        for _ in 0..20 {
757            leases.push(owner.lease_stream().expect("active stream lease"));
758        }
759        assert!(matches!(owner.lease_stream(), Err(Error::RateLimited)));
760        drop(leases.pop());
761        leases.push(owner.lease_stream().expect("released lease"));
762    }
763}
oanda_client/admission.rs

oanda_client/
admission.rs
