nym_bandwidth_controller/requests/
sender.rs

1// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
2// SPDX-License-Identifier: Apache-2.0
3
4use async_trait::async_trait;
5use nym_credentials_interface::TicketType;
6use nym_crypto::asymmetric::ed25519;
7use nym_ecash_time::OffsetDateTime;
8use tokio::sync::mpsc::UnboundedSender;
9use tracing::instrument;
10
11use crate::{
12    error::BandwidthControllerError,
13    requests::{BandwidthControllerRequest, ReturnSender},
14    ticketbooks::AvailableTicketbooks,
15    traits::CredentialFetcher,
16    BandwidthTicketProvider, CredentialPublicDataFetcher, EcashTicketRequest, PreparedCredential,
17    PreparedCredentialMetadata,
18};
19use std::sync::Arc;
20
21#[derive(Clone)]
22pub struct BandwidthControllerRequestSender {
23    command_tx: UnboundedSender<BandwidthControllerRequest>,
24}
25
26// Basic set of commands that can be sent to the bandwidth controller
27
28impl BandwidthControllerRequestSender {
29    pub fn new(command_tx: UnboundedSender<BandwidthControllerRequest>) -> Self {
30        Self { command_tx }
31    }
32
33    #[instrument(skip(self), level = "debug")]
34    pub async fn get_ecash_ticket(
35        &self,
36        ticket_type: TicketType,
37        gateway_id: ed25519::PublicKey,
38        tickets_to_spend: u32,
39        spend_time: OffsetDateTime,
40    ) -> Result<Option<PreparedCredential>, BandwidthControllerError> {
41        let (tx, rx) = ReturnSender::new();
42        self.command_tx
43            .send(BandwidthControllerRequest::EcashTicket(
44                tx,
45                EcashTicketRequest {
46                    ticket_type,
47                    gateway_id,
48                    tickets_to_spend,
49                    spend_time,
50                },
51            ))
52            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
53        rx.await
54            .map_err(|_| BandwidthControllerError::ChannelClosed)?
55    }
56
57    #[instrument(skip(self), level = "debug")]
58    pub async fn get_upgrade_mode_token(&self) -> Result<Option<String>, BandwidthControllerError> {
59        let (tx, rx) = ReturnSender::new();
60        self.command_tx
61            .send(BandwidthControllerRequest::UpgradeModeToken(tx))
62            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
63        rx.await
64            .map_err(|_| BandwidthControllerError::ChannelClosed)?
65    }
66
67    #[instrument(skip(self), level = "debug")]
68    pub async fn attempt_revert_spending(
69        &self,
70        metadata: PreparedCredentialMetadata,
71    ) -> Result<bool, BandwidthControllerError> {
72        let (tx, rx) = ReturnSender::new();
73        self.command_tx
74            .send(BandwidthControllerRequest::AttemptRevertSpending(
75                tx, metadata,
76            ))
77            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
78        rx.await
79            .map_err(|_| BandwidthControllerError::ChannelClosed)?
80    }
81
82    /// Installs the credential fetcher; the controller immediately restocks any low types.
83    #[instrument(skip(self, credential_fetcher))]
84    pub async fn set_credential_fetcher(
85        &self,
86        credential_fetcher: Arc<impl CredentialFetcher + 'static>,
87    ) -> Result<(), BandwidthControllerError> {
88        let (tx, rx) = ReturnSender::new();
89        self.command_tx
90            .send(BandwidthControllerRequest::SetCredentialFetcher(
91                tx,
92                Some(credential_fetcher),
93            ))
94            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
95        rx.await
96            .map_err(|_| BandwidthControllerError::ChannelClosed)?
97    }
98
99    /// Removes the credential fetcher; no further automatic restocking happens until one is set.
100    #[instrument(skip(self))]
101    pub async fn unset_credential_fetcher(&self) -> Result<(), BandwidthControllerError> {
102        let (tx, rx) = ReturnSender::new();
103        self.command_tx
104            .send(BandwidthControllerRequest::SetCredentialFetcher(tx, None))
105            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
106        rx.await
107            .map_err(|_| BandwidthControllerError::ChannelClosed)?
108    }
109
110    /// Installs the global-data fetcher used to retrieve missing ecash signing materials.
111    #[instrument(skip(self, public_data_fetcher))]
112    pub async fn set_public_data_fetcher(
113        &self,
114        public_data_fetcher: Arc<impl CredentialPublicDataFetcher + 'static>,
115    ) -> Result<(), BandwidthControllerError> {
116        let (tx, rx) = ReturnSender::new();
117        self.command_tx
118            .send(BandwidthControllerRequest::SetPublicDataFetcher(
119                tx,
120                Some(public_data_fetcher),
121            ))
122            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
123        rx.await
124            .map_err(|_| BandwidthControllerError::ChannelClosed)?
125    }
126
127    /// Removes the global-data fetcher.
128    #[instrument(skip(self))]
129    pub async fn unset_public_data_fetcher(&self) -> Result<(), BandwidthControllerError> {
130        let (tx, rx) = ReturnSender::new();
131        self.command_tx
132            .send(BandwidthControllerRequest::SetPublicDataFetcher(tx, None))
133            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
134        rx.await
135            .map_err(|_| BandwidthControllerError::ChannelClosed)?
136    }
137
138    /// Cancels in-flight fetches, drops the fetcher, clears stored credentials, and fails any parked
139    /// readiness waiters. Used to fully de-provision the controller.
140    #[instrument(skip(self))]
141    pub async fn reset(&self) -> Result<(), BandwidthControllerError> {
142        let (tx, rx) = ReturnSender::new();
143        self.command_tx
144            .send(BandwidthControllerRequest::Reset(tx))
145            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
146        rx.await
147            .map_err(|_| BandwidthControllerError::ChannelClosed)?
148    }
149
150    /// Removes stored emergency (upgrade-mode) credentials, leaving ticketbooks intact.
151    #[instrument(skip(self))]
152    pub async fn clear_emergency_credentials(&self) -> Result<(), BandwidthControllerError> {
153        let (tx, rx) = ReturnSender::new();
154        self.command_tx
155            .send(BandwidthControllerRequest::ClearEmergencyCredentials(tx))
156            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
157        rx.await
158            .map_err(|_| BandwidthControllerError::ChannelClosed)?
159    }
160
161    /// Returns the currently stored ticketbooks.
162    #[instrument(skip(self))]
163    pub async fn get_available_ticketbooks(
164        &self,
165    ) -> Result<AvailableTicketbooks, BandwidthControllerError> {
166        let (tx, rx) = ReturnSender::new();
167        self.command_tx
168            .send(BandwidthControllerRequest::GetAvailableTicketbooks(tx))
169            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
170        rx.await
171            .map_err(|_| BandwidthControllerError::ChannelClosed)?
172    }
173
174    /// Kicks off a background restock for the given ticket types
175    /// Returns once the restock is scheduled - it does not wait for the fetches to finish
176    /// (use [`Self::wait_for_ticketbooks`] for that).
177    ///
178    /// Not to be used lightly: the automatic triggers (ticket handout, timer, fetcher install)
179    /// already keep every type stocked. This is a manual safety valve, not a routine call.
180    #[instrument(skip(self))]
181    pub async fn restock_ticketbooks(
182        &self,
183        types: Vec<TicketType>,
184    ) -> Result<(), BandwidthControllerError> {
185        let (tx, rx) = ReturnSender::new();
186        self.command_tx
187            .send(BandwidthControllerRequest::RestockTicketbooks(tx, types))
188            .map_err(BandwidthControllerError::internal)?;
189        rx.await.map_err(BandwidthControllerError::internal)?
190    }
191
192    /// Kicks off a background restock for every ticket type running low or about to expire.
193    /// Returns once the restock is scheduled, not once the fetches finish.
194    #[instrument(skip(self))]
195    pub async fn restock_all_ticketbooks(&self) -> Result<(), BandwidthControllerError> {
196        self.restock_ticketbooks(AvailableTicketbooks::ticketbook_types())
197            .await
198    }
199
200    /// Resolves once every listed type is usable (stocked or covered by upgrade mode). Errors if a
201    /// required type is neither stocked nor being fetched; otherwise waits while the unsatisfied
202    /// ones are still in flight.
203    #[instrument(skip(self))]
204    pub async fn wait_for_ticketbooks(
205        &self,
206        types: Vec<TicketType>,
207    ) -> Result<(), BandwidthControllerError> {
208        let (tx, rx) = ReturnSender::new();
209        self.command_tx
210            .send(BandwidthControllerRequest::WaitForTicketbooks(tx, types))
211            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
212        rx.await
213            .map_err(|_| BandwidthControllerError::ChannelClosed)?
214    }
215}
216
217#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
218#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
219impl BandwidthTicketProvider for BandwidthControllerRequestSender {
220    async fn get_ecash_ticket(
221        &self,
222        ticket_type: TicketType,
223        gateway_id: ed25519::PublicKey,
224        tickets_to_spend: u32,
225        spend_time: OffsetDateTime,
226    ) -> Result<Option<PreparedCredential>, BandwidthControllerError> {
227        self.get_ecash_ticket(ticket_type, gateway_id, tickets_to_spend, spend_time)
228            .await
229    }
230
231    async fn get_upgrade_mode_token(&self) -> Result<Option<String>, BandwidthControllerError> {
232        self.get_upgrade_mode_token().await
233    }
234
235    async fn attempt_revert_spending(
236        &self,
237        metadata: PreparedCredentialMetadata,
238    ) -> Result<bool, BandwidthControllerError> {
239        self.attempt_revert_spending(metadata).await
240    }
241
242    // No-op, the controller will close when stopped
243    async fn close(&self) {}
244}
nym_bandwidth_controller/requests/sender.rs

nym_bandwidth_controller/requests/
sender.rs
