Add prune command to BandwidthController

common/bandwidth-controller/src/controller.rs

@@ -228,6 +228,9 @@ impl<St: Storage> BandwidthController<St> {
                         .await
                         .map_err(BandwidthControllerError::credential_storage_error),
                 ),
+            BandwidthControllerRequest::Prune(return_sender) => {
+                return_sender.send(self.handle_prune().await)
+            }
             BandwidthControllerRequest::GetAvailableTicketbooks(return_sender) => {
                 return_sender.send(self.handle_get_available_ticketbooks().await)
             }
@@ -296,6 +299,37 @@ impl<St: Storage> BandwidthController<St> {
             .map_err(BandwidthControllerError::credential_storage_error)
     }
 
+    // Removes expired ticketbooks from storage and expired ticketbooks that are still in pending.
+    async fn handle_prune(&mut self) -> Result<(), BandwidthControllerError> {
+        let expired_stored_ret = self
+            .storage
+            .cleanup_expired()
+            .await
+            .map_err(BandwidthControllerError::credential_storage_error);
+
+        let expired_pending_ret = if let Some(fetcher) = &self.credential_fetcher {
+            Some(
+                fetcher
+                    .prune()
+                    .await
+                    .map_err(BandwidthControllerError::fetcher_error),
+            )
+        } else {
+            tracing::debug!("No credential fetcher set. No pruning possible there");
+            None
+        };
+
+        // We propagate the errors after trying to prune as much expired data as possible
+        // Storage pruning failure is more important, as pending data eventually converts
+        // into stored data anyway, so we propagate potential storage pruning failure first.
+        expired_stored_ret?;
+        if let Some(expired_pending_ret) = expired_pending_ret {
+            expired_pending_ret?;
+        }
+
+        Ok(())
+    }
+
     async fn handle_get_available_ticketbooks(
         &self,
     ) -> Result<AvailableTicketbooks, BandwidthControllerError> {

common/bandwidth-controller/src/in_flight/mod.rs

@@ -278,6 +278,10 @@ mod tests {
             }
         }
 
+        async fn prune(&self) -> Result<(), CredentialFetcherError> {
+            Ok(())
+        }
+
         async fn cleanup(&self) {}
 
         async fn reset(self) -> Result<(), CredentialFetcherError> {

common/bandwidth-controller/src/requests/mod.rs

@@ -34,6 +34,9 @@ pub enum BandwidthControllerRequest {
     Reset(ReturnSender<()>),
     /// Removes the stored emergency (upgrade-mode) credentials only, leaving ticketbooks intact.
     ClearEmergencyCredentials(ReturnSender<()>),
+    /// Prunes expired ticketbooks from storage and stops in-flight retrieval of blinded ticketbook
+    /// shares whose validity has passed.
+    Prune(ReturnSender<()>),
     /// Returns the currently stored ticketbooks (also logs a stock summary).
     GetAvailableTicketbooks(ReturnSender<AvailableTicketbooks>),
 

common/bandwidth-controller/src/requests/sender.rs

@@ -158,6 +158,18 @@ impl BandwidthControllerRequestSender {
             .map_err(|_| BandwidthControllerError::ChannelClosed)?
     }
 
+    /// Prunes expired ticketbooks from storage and stops in-flight retrieval of blinded ticketbook
+    /// shares whose validity has passed.
+    #[instrument(skip(self))]
+    pub async fn prune(&self) -> Result<(), BandwidthControllerError> {
+        let (tx, rx) = ReturnSender::new();
+        self.command_tx
+            .send(BandwidthControllerRequest::Prune(tx))
+            .map_err(|_| BandwidthControllerError::ChannelClosed)?;
+        rx.await
+            .map_err(|_| BandwidthControllerError::ChannelClosed)?
+    }
+
     /// Returns the currently stored ticketbooks.
     #[instrument(skip(self))]
     pub async fn get_available_ticketbooks(

common/bandwidth-controller/src/traits.rs

@@ -98,6 +98,9 @@ pub trait CredentialFetcher: CredentialPublicDataFetcher + Send + Sync {
         ticketbook_type: TicketType,
     ) -> Result<Vec<NymCredential>, CredentialFetcherError>;
 
+    /// Stops in-flight retrieval of blinded ticketbook shares whose validity has passed.
+    async fn prune(&self) -> Result<(), CredentialFetcherError>;
+
     /// Persists any in-progress state (e.g. closing storage) before the fetcher is dropped, such
     /// that it can be resumed later.
     async fn cleanup(&self);

common/bandwidth-controller/tests/managed_ticket_types.rs

@@ -68,6 +68,10 @@ impl CredentialFetcher for RecordingFetcher {
         Ok(Vec::new())
     }
 
+    async fn prune(&self) -> Result<(), CredentialFetcherError> {
+        Ok(())
+    }
+
     async fn cleanup(&self) {}
 
     async fn reset(self) -> Result<(), CredentialFetcherError> {

common/bandwidth-fetcher/Cargo.toml

@@ -34,6 +34,9 @@ nym-validator-client = { workspace = true }
 sqlx = { workspace = true }
 nym-sqlx-pool-guard = { workspace = true }
 
+[dev-dependencies]
+serde = { workspace = true }
+
 [build-dependencies]
 anyhow = { workspace = true }
 sqlx = { workspace = true, features = [

common/bandwidth-fetcher/src/credentials.rs

@@ -185,6 +185,41 @@ where
 
         Ok(issuance_data.to_issued_ticketbook(wallet, epoch_id))
     }
+
+    async fn cancel_expired_ticketbooks(&self) -> Result<(), NyxdFetcherError> {
+        let mut pruned_pending_ticketbooks = 0;
+
+        for expired_pending_ticketbook_id in self
+            .pending_storage
+            .get_pending_ticketbooks()
+            .await?
+            .iter()
+            .filter_map(|ticket_book| {
+                if ticket_book.pending_ticketbook.expired() {
+                    Some(ticket_book.pending_id)
+                } else {
+                    None
+                }
+            })
+        {
+            if let Err(err) = self
+                .pending_storage
+                .remove_pending_ticketbook(expired_pending_ticketbook_id)
+                .await
+            {
+                tracing::warn!(
+                    "Failed to remove expired ticketbook id {expired_pending_ticketbook_id} from pending storage: {err}"
+                );
+            } else {
+                pruned_pending_ticketbooks += 1;
+            }
+        }
+        tracing::debug!(
+            "Cancelled {pruned_pending_ticketbooks} expired ticketbooks that were pending"
+        );
+
+        Ok(())
+    }
 }
 
 impl<C> NyxdCredentialFetcher<C>
@@ -359,6 +394,11 @@ where
         }
     }
 
+    async fn prune(&self) -> Result<(), CredentialFetcherError> {
+        self.cancel_expired_ticketbooks().await?;
+        Ok(())
+    }
+
     async fn cleanup(&self) {
         self.pending_storage.close().await;
     }
@@ -446,6 +486,11 @@ pub(crate) mod recovery {
             Ok(recovered_ticketbooks)
         }
 
+        async fn prune(&self) -> Result<(), CredentialFetcherError> {
+            self.0.cancel_expired_ticketbooks().await?;
+            Ok(())
+        }
+
         async fn cleanup(&self) {
             self.0.pending_storage.close().await;
         }
@@ -460,3 +505,131 @@ pub(crate) mod recovery {
         }
     }
 }
+
+#[cfg(test)]
+#[allow(clippy::unreachable)]
+mod tests {
+    use std::env::temp_dir;
+
+    use serde::Deserialize;
+
+    use nym_validator_client::nyxd::{
+        Fee,
+        contract_traits::dkg_query_client::DkgQueryMsg,
+        cosmwasm_client::types::ExecuteResult,
+        error::NyxdError,
+        nym_ecash_contract_common::msg::{ExecuteMsg, QueryMsg},
+    };
+    use tokio::fs::remove_file;
+
+    use super::*;
+
+    struct MockPruneClient {}
+
+    #[async_trait]
+    impl DkgQueryClient for MockPruneClient {
+        async fn query_dkg_contract<T>(
+            &self,
+            _query: DkgQueryMsg,
+        ) -> std::result::Result<T, NyxdError>
+        where
+            for<'a> T: Deserialize<'a>,
+        {
+            unreachable!("client not used in prune unit tests");
+        }
+    }
+
+    #[async_trait]
+    impl EcashSigningClient for MockPruneClient {
+        async fn execute_ecash_contract(
+            &self,
+            _fee: Option<Fee>,
+            _msg: ExecuteMsg,
+            _memo: String,
+            _funds: Vec<Coin>,
+        ) -> Result<ExecuteResult, NyxdError> {
+            unreachable!("client not used in prune unit tests");
+        }
+    }
+
+    #[async_trait]
+    impl EcashQueryClient for MockPruneClient {
+        async fn query_ecash_contract<T>(&self, _query: QueryMsg) -> Result<T, NyxdError>
+        where
+            for<'a> T: Deserialize<'a>,
+        {
+            unreachable!("client not used in prune unit tests");
+        }
+    }
+
+    #[tokio::test]
+    async fn prune_expired() {
+        let mut db_path = temp_dir();
+        db_path.push("prune_expired_unittest.db");
+        let fetcher = NyxdCredentialFetcher::new(
+            Arc::new(MockPruneClient {}),
+            &db_path,
+            Zeroizing::new(Vec::new()),
+        )
+        .await
+        .unwrap();
+
+        // pruning empty database doesn't fail
+        fetcher.prune().await.unwrap();
+
+        // insert late expiration ticketbook
+        let expired_ticketbook = IssuanceTicketBook::new_with_expiration(
+            0,
+            [],
+            ed25519::PrivateKey::new(&mut OsRng),
+            TicketType::V1WireguardEntry,
+            Date::MIN,
+        );
+        fetcher
+            .pending_storage
+            .insert_pending_ticketbook(&expired_ticketbook)
+            .await
+            .unwrap();
+
+        // check pruning emptied it
+        fetcher.prune().await.unwrap();
+        assert_eq!(
+            fetcher
+                .pending_storage
+                .get_pending_ticketbooks()
+                .await
+                .unwrap()
+                .len(),
+            0
+        );
+
+        // insert late expiration ticketbook
+        let unexpired_ticketbook = IssuanceTicketBook::new_with_expiration(
+            0,
+            [],
+            ed25519::PrivateKey::new(&mut OsRng),
+            TicketType::V1WireguardEntry,
+            Date::MAX,
+        );
+        fetcher
+            .pending_storage
+            .insert_pending_ticketbook(&unexpired_ticketbook)
+            .await
+            .unwrap();
+
+        // check pruning doesn't affect it
+        fetcher.prune().await.unwrap();
+        assert_ne!(
+            fetcher
+                .pending_storage
+                .get_pending_ticketbooks()
+                .await
+                .unwrap()
+                .len(),
+            0
+        );
+
+        fetcher.pending_storage.close().await;
+        remove_file(db_path).await.unwrap();
+    }
+}

sdk/rust/nym-sdk-session/src/fetcher.rs

@@ -141,6 +141,10 @@ impl<F: CredentialFetcher> CredentialFetcher for TimeoutFetcher<F> {
         self.inner.fetch_ticketbooks(ticketbook_type).await
     }
 
+    async fn prune(&self) -> Result<(), CredentialFetcherError> {
+        self.inner.prune().await
+    }
+
     async fn cleanup(&self) {
         self.inner.cleanup().await
     }
@@ -229,6 +233,10 @@ mod tests {
             self.act().await
         }
 
+        async fn prune(&self) -> Result<(), CredentialFetcherError> {
+            Ok(())
+        }
+
         async fn cleanup(&self) {}
 
         async fn reset(self) -> Result<(), CredentialFetcherError> {

sdk/rust/nym-sdk-session/tests/support/mod.rs

@@ -13,6 +13,7 @@
 
 pub mod fake_dkg;
 pub mod http_harness;
+pub mod prune;
 
 use nym_compact_ecash::scheme::keygen::{KeyPairAuth, SecretKeyAuth};
 use nym_compact_ecash::tests::helpers::{
@@ -77,11 +78,13 @@ impl TestEcash {
         }
     }
 
-    /// Fabricate a real threshold-signed ticketbook for `typ`. Distinct `seed`s
-    /// yield distinct books (fresh user keypair per book).
-    pub fn ticketbook(&self, typ: TicketType, seed: u64) -> IssuedTicketBook {
+    pub fn ticketbook_with_expiration(
+        &self,
+        typ: TicketType,
+        seed: u64,
+        expiration_date: Date,
+    ) -> IssuedTicketBook {
         let user = generate_keypair_user_from_seed(seed.to_be_bytes());
-        let expiration_date = ecash_default_expiration_date();
         let expiration_ts = expiration_date.ecash_unix_timestamp();
 
         let (req, req_info) = withdrawal_request(user.secret_key(), expiration_ts, typ.encode())
@@ -118,6 +121,13 @@ impl TestEcash {
         )
     }
 
+    /// Fabricate a real threshold-signed ticketbook for `typ`. Distinct `seed`s
+    /// yield distinct books (fresh user keypair per book).
+    pub fn ticketbook(&self, typ: TicketType, seed: u64) -> IssuedTicketBook {
+        let expiration_date = ecash_default_expiration_date();
+        self.ticketbook_with_expiration(typ, seed, expiration_date)
+    }
+
     /// The master verification key as the fetcher would serve it.
     pub fn epoch_verification_key(&self, epoch_id: u64) -> EpochVerificationKey {
         EpochVerificationKey {