Jędrzej Stuczyński
0ee387d983
* squashing work on using cancellation in nym crates making nym-task wasm compilable removed sending of status messages replaced TaskManager with ShutdownManager in the validator rewarder additional helpers for ShutdownManager simplified ShutdownToken by removing the name field TaskClient => ShutdownToken within all client tasks wip: remove TaskHandle * track all long-living client tasks * add task tracking for most top level tasks within nym-node * improved default builder * split up cancellation module * module documentation and unit tests * nym node fixes and naming consistency * wasm fixes * assert_eq => assert * wasm fixes and made 'run_until_shutdown' take reference instead of ownership * linux-specific fixes to IpPacketRouter * post rebasing fixes for signing monitor * add ShutdownManager constructor to build it from an external token * applying PR review suggestions
54 lines
1.6 KiB
Rust
54 lines
1.6 KiB
Rust
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
use nym_gateway_storage::{GatewayStorage, InboxManager};
|
|
use nym_task::ShutdownToken;
|
|
use std::error::Error;
|
|
use std::time::Duration;
|
|
use time::OffsetDateTime;
|
|
use tracing::{debug, trace, warn};
|
|
|
|
pub struct StaleMessagesCleaner {
|
|
inbox_manager: InboxManager,
|
|
max_message_age: Duration,
|
|
run_interval: Duration,
|
|
}
|
|
|
|
impl StaleMessagesCleaner {
|
|
pub(crate) fn new(
|
|
storage: &GatewayStorage,
|
|
max_message_age: Duration,
|
|
run_interval: Duration,
|
|
) -> Self {
|
|
StaleMessagesCleaner {
|
|
inbox_manager: storage.inbox_manager().clone(),
|
|
max_message_age,
|
|
run_interval,
|
|
}
|
|
}
|
|
|
|
async fn clean_up_stale_messages(&mut self) -> Result<(), impl Error> {
|
|
let cutoff = OffsetDateTime::now_utc() - self.max_message_age;
|
|
self.inbox_manager.remove_stale(cutoff).await
|
|
}
|
|
|
|
pub async fn run(&mut self, shutdown_token: ShutdownToken) {
|
|
let mut interval = tokio::time::interval(self.run_interval);
|
|
loop {
|
|
tokio::select! {
|
|
biased;
|
|
_ = shutdown_token.cancelled() => {
|
|
trace!("StaleMessagesCleaner: received shutdown");
|
|
break;
|
|
}
|
|
_ = interval.tick() => {
|
|
if let Err(err) = self.clean_up_stale_messages().await {
|
|
warn!("failed to clean up stale messages: {err}");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
debug!("StaleMessagesCleaner: Exiting");
|
|
}
|
|
}
|