broker: hourly vacuum of delivered messages older than 30 days
undelivered rows are always kept regardless of age (still in flight). sweep runs immediately on serve start then every hour. logs row count when non-zero. keep_secs is hard-coded for now (30 days); can be config-driven later if a host wants to retain more / less for audit.
This commit is contained in:
parent
a9ed33d94f
commit
6d52f67292
3 changed files with 31 additions and 2 deletions
|
|
@ -110,6 +110,22 @@ async fn main() -> Result<()> {
|
|||
tracing::warn!(error = ?e, "auto-update task failed");
|
||||
}
|
||||
});
|
||||
// Periodic broker vacuum: drop delivered messages older than
|
||||
// 30 days. Undelivered messages are always kept (still in
|
||||
// flight). Runs hourly; first sweep happens immediately.
|
||||
let vacuum_coord = coord.clone();
|
||||
tokio::spawn(async move {
|
||||
let interval_secs = 3600u64;
|
||||
let keep_secs: i64 = 30 * 24 * 3600;
|
||||
loop {
|
||||
match vacuum_coord.broker.vacuum_delivered(keep_secs) {
|
||||
Ok(0) => {}
|
||||
Ok(n) => tracing::info!(removed = n, "broker vacuum"),
|
||||
Err(e) => tracing::warn!(error = ?e, "broker vacuum failed"),
|
||||
}
|
||||
tokio::time::sleep(std::time::Duration::from_secs(interval_secs)).await;
|
||||
}
|
||||
});
|
||||
let dash_coord = coord.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue