diff --git a/src/trx-server/src/audio.rs b/src/trx-server/src/audio.rs index 52cdfbf2..bf9745c8 100644 --- a/src/trx-server/src/audio.rs +++ b/src/trx-server/src/audio.rs @@ -47,21 +47,11 @@ use trx_wspr::WsprDecoder; use uuid::Uuid; use crate::config::AudioConfig; +use crate::history_policy::{ + enforce_capacity, lock_or_recover, prune_by_age, HISTORY_RETENTION, MAX_HISTORY_ENTRIES, +}; use trx_decode_log::DecoderLoggers; -const APRS_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); -const HF_APRS_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); -const AIS_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); -const VDES_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); -const CW_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); -const FT8_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); -const WSPR_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); -const LRPT_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); -const WEFAX_HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); -/// Maximum entries per decoder history queue. Prevents unbounded memory growth -/// on busy channels (e.g. AIS near a port). Oldest entries are evicted when -/// the limit is reached, independent of the time-based pruning. -const MAX_HISTORY_ENTRIES: usize = 10_000; /// Silence timeout before auto-finalising an LRPT pass (30 s without new MCUs). const LRPT_PASS_SILENCE_TIMEOUT: Duration = Duration::from_secs(30); const FT8_SAMPLE_RATE: u32 = 12_000; @@ -382,41 +372,6 @@ pub struct DecoderHistories { total_count: AtomicUsize, } -/// Acquire a mutex, recovering from poisoning with a warning log. -fn lock_or_recover<'a, T>(mutex: &'a Mutex, label: &str) -> std::sync::MutexGuard<'a, T> { - mutex.lock().unwrap_or_else(|e| { - tracing::warn!( - "Mutex for {} was poisoned (prior panic); recovering with potentially inconsistent data", - label - ); - e.into_inner() - }) -} - -/// Enforce capacity limit on a history deque by evicting oldest entries. -fn enforce_capacity(deque: &mut VecDeque, max: usize) { - while deque.len() > max { - deque.pop_front(); - } -} - -/// Drop entries older than `retention` from the front of a time-tagged deque. -/// Uses `checked_sub` so an early `now` (before `retention` elapses since the -/// monotonic clock origin) is treated as "no entries are old enough to prune" -/// rather than panicking. -fn prune_by_age(deque: &mut VecDeque<(Instant, T)>, retention: Duration, now: Instant) { - let Some(cutoff) = now.checked_sub(retention) else { - return; - }; - while let Some((ts, _)) = deque.front() { - if *ts < cutoff { - deque.pop_front(); - } else { - break; - } - } -} - impl DecoderHistories { pub fn new() -> Arc { Arc::new(Self { @@ -465,7 +420,7 @@ impl DecoderHistories { // --- AIS --- fn prune_ais(history: &mut VecDeque<(Instant, AisMessage)>) { - prune_by_age(history, AIS_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } pub fn record_ais_message(&self, mut msg: AisMessage) { @@ -491,7 +446,7 @@ impl DecoderHistories { // --- VDES --- fn prune_vdes(history: &mut VecDeque<(Instant, VdesMessage)>) { - prune_by_age(history, VDES_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } pub fn record_vdes_message(&self, mut msg: VdesMessage) { @@ -517,7 +472,7 @@ impl DecoderHistories { // --- APRS --- fn prune_aprs(history: &mut VecDeque<(Instant, AprsPacket)>) { - prune_by_age(history, APRS_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } pub fn record_aprs_packet(&self, mut pkt: AprsPacket) { @@ -555,7 +510,7 @@ impl DecoderHistories { // --- HF APRS --- fn prune_hf_aprs(history: &mut VecDeque<(Instant, AprsPacket)>) { - prune_by_age(history, HF_APRS_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } pub fn record_hf_aprs_packet(&self, mut pkt: AprsPacket) { @@ -593,7 +548,7 @@ impl DecoderHistories { // --- CW --- fn prune_cw(history: &mut VecDeque<(Instant, CwEvent)>) { - prune_by_age(history, CW_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } pub fn record_cw_event(&self, evt: CwEvent) { @@ -625,7 +580,7 @@ impl DecoderHistories { // --- FT8 --- fn prune_ft8(history: &mut VecDeque<(Instant, Ft8Message)>) { - prune_by_age(history, FT8_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } pub fn record_ft8_message(&self, msg: Ft8Message) { @@ -657,7 +612,7 @@ impl DecoderHistories { // --- FT4 --- fn prune_ft4(history: &mut VecDeque<(Instant, Ft8Message)>) { - prune_by_age(history, FT8_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } pub fn record_ft4_message(&self, msg: Ft8Message) { @@ -690,7 +645,7 @@ impl DecoderHistories { #[cfg_attr(not(feature = "ft2"), allow(dead_code))] fn prune_ft2(history: &mut VecDeque<(Instant, Ft8Message)>) { - prune_by_age(history, FT8_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } #[cfg_attr(not(feature = "ft2"), allow(dead_code))] @@ -724,7 +679,7 @@ impl DecoderHistories { // --- WSPR --- fn prune_wspr(history: &mut VecDeque<(Instant, WsprMessage)>) { - prune_by_age(history, WSPR_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } pub fn record_wspr_message(&self, msg: WsprMessage) { @@ -756,7 +711,7 @@ impl DecoderHistories { // --- LRPT --- fn prune_lrpt(history: &mut VecDeque<(Instant, LrptImage)>) { - prune_by_age(history, LRPT_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } pub fn record_lrpt_image(&self, mut img: LrptImage) { @@ -791,7 +746,7 @@ impl DecoderHistories { // --- WEFAX --- fn prune_wefax(history: &mut VecDeque<(Instant, WefaxMessage)>) { - prune_by_age(history, WEFAX_HISTORY_RETENTION, Instant::now()); + prune_by_age(history, HISTORY_RETENTION, Instant::now()); } pub fn record_wefax_message(&self, mut msg: WefaxMessage) { diff --git a/src/trx-server/src/history_policy.rs b/src/trx-server/src/history_policy.rs new file mode 100644 index 00000000..2ee54e3a --- /dev/null +++ b/src/trx-server/src/history_policy.rs @@ -0,0 +1,49 @@ +// SPDX-FileCopyrightText: 2026 Stan Grams +// +// SPDX-License-Identifier: GPL-2.0-or-later + +//! Shared retention and synchronization policy for decoder histories. + +use std::collections::VecDeque; +use std::sync::{Mutex, MutexGuard}; +use std::time::{Duration, Instant}; + +pub(crate) const HISTORY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); + +/// Maximum entries per decoder history queue. Oldest entries are evicted on +/// busy channels independently of time-based pruning. +pub(crate) const MAX_HISTORY_ENTRIES: usize = 10_000; + +pub(crate) fn lock_or_recover<'a, T>(mutex: &'a Mutex, label: &str) -> MutexGuard<'a, T> { + mutex.lock().unwrap_or_else(|error| { + tracing::warn!( + "Mutex for {} was poisoned (prior panic); recovering with potentially inconsistent data", + label + ); + error.into_inner() + }) +} + +pub(crate) fn enforce_capacity(deque: &mut VecDeque, max: usize) { + while deque.len() > max { + deque.pop_front(); + } +} + +/// Drop entries older than `retention` from an ordered history queue. +pub(crate) fn prune_by_age( + deque: &mut VecDeque<(Instant, T)>, + retention: Duration, + now: Instant, +) { + let Some(cutoff) = now.checked_sub(retention) else { + return; + }; + while let Some((timestamp, _)) = deque.front() { + if *timestamp < cutoff { + deque.pop_front(); + } else { + break; + } + } +} diff --git a/src/trx-server/src/main.rs b/src/trx-server/src/main.rs index d3430b77..3110095f 100644 --- a/src/trx-server/src/main.rs +++ b/src/trx-server/src/main.rs @@ -5,6 +5,7 @@ mod audio; mod config; mod error; +mod history_policy; mod history_store; mod listener; mod rig_handle;