sjg/trx-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[refactor](trx-server): extract history policy

Browse Source
This commit is contained in:
sjg
2026-08-01 01:44:26 +02:00
parent 7db7fad9b0
commit 4728b578ae
3 changed files with 64 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/trx-server/src/audio.rs
+14 -59
View File
@@ -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<T>, 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<T>(deque: &mut VecDeque<T>, 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<T>(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<Self> {
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) {
src/trx-server/src/history_policy.rs
+49
View File
@@ -0,0 +1,49 @@
// SPDX-FileCopyrightText: 2026 Stan Grams <sjg@haxx.space>
//
// 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<T>, 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<T>(deque: &mut VecDeque<T>, 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<T>(
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;
}
}
}
src/trx-server/src/main.rs
+1
View File
@@ -5,6 +5,7 @@
mod audio;
mod config;
mod error;
mod history_policy;
mod history_store;
mod listener;
mod rig_handle;