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

[refactor](trx-rs): resolve all improvement areas (P0-P3)

Browse Source 
Addresses every item in docs/Improvement-Areas.md:

P0 - Plugin signing: new src/trx-app/src/plugins.rs with SHA-256 checksum
     manifest, filename allowlisting, API version compatibility checks,
     and cross-platform file permission validation.

P1 - Session store mutex poisoning: all .unwrap() calls on RwLock/Mutex in
     auth.rs replaced with .unwrap_or_else(|e| e.into_inner()) + warning logs.
   - TCP listener rate limiting: added ConnectionTracker with per-IP connection
     cap (10 concurrent connections per IP).
   - RigState refactoring: decoder fields grouped into DecoderConfig and
     DecoderResetSeqs sub-structs with #[serde(flatten)] for wire compat.
   - spawn_blocking timeout: satellite pass computation wrapped in 30s timeout.

P2 - Command handler macro: rig_command! macro generates 7 unit-struct command
     implementations, reducing ~200 lines of boilerplate.
   - Protocol versioning: added protocol_version field to ClientEnvelope and
     ClientResponse; improved unknown command error handling in parse_envelope.
   - Unsafe string: replaced from_utf8_unchecked with safe from_utf8().expect().
   - Dead code: removed 2 unnecessary annotations, documented remaining 4.

P3 - Tests: added 4 unit tests for history_store.rs (round-trip, expiry, etc).
   - FT-817 VFO: improved inference for ambiguous same-frequency case.
   - Configurator: implemented serial port detection via tokio_serial.
   - Plugin versioning: integrated into plugin manifest (api_version field).
   - Naming: documented as intentional semantic distinctions, not inconsistencies.

https://claude.ai/code/session_01Gj1vEkP6GKVcVaMqzFW885
Signed-off-by: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
2026-03-29 11:06:23 +00:00
committed by Stan Grams
parent 8e3162d7e6
commit a69c5143e6
23 changed files with 1129 additions and 603 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/trx-server/src/audio.rs
+66 -66
View File
@@ -1194,8 +1194,8 @@ pub async fn run_aprs_decoder(
if active {
pcm_rx = pcm_rx.resubscribe();
}
if state.aprs_decode_reset_seq != last_reset_seq {
last_reset_seq = state.aprs_decode_reset_seq;
if state.reset_seqs.aprs_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.aprs_decode_reset_seq;
decoder.reset();
info!("APRS decoder reset (seq={})", last_reset_seq);
}
@@ -1211,7 +1211,7 @@ pub async fn run_aprs_decoder(
Ok(frame) => {
let reset_seq = {
let state = state_rx.borrow();
state.aprs_decode_reset_seq
state.reset_seqs.aprs_decode_reset_seq
};
if reset_seq != last_reset_seq {
last_reset_seq = reset_seq;
@@ -1236,7 +1236,7 @@ pub async fn run_aprs_decoder(
was_active = true;
let packets = tokio::task::block_in_place(|| decoder.process_samples(&mono));
let latest_reset_seq = state_rx.borrow().aprs_decode_reset_seq;
let latest_reset_seq = state_rx.borrow().reset_seqs.aprs_decode_reset_seq;
if latest_reset_seq != reset_seq {
last_reset_seq = latest_reset_seq;
decoder.reset();
@@ -1269,8 +1269,8 @@ pub async fn run_aprs_decoder(
Ok(()) => {
let state = state_rx.borrow();
active = matches!(state.status.mode, RigMode::PKT);
if state.aprs_decode_reset_seq != last_reset_seq {
last_reset_seq = state.aprs_decode_reset_seq;
if state.reset_seqs.aprs_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.aprs_decode_reset_seq;
decoder.reset();
info!("APRS decoder reset (seq={})", last_reset_seq);
}
@@ -1316,8 +1316,8 @@ pub async fn run_hf_aprs_decoder(
if active {
pcm_rx = pcm_rx.resubscribe();
}
if state.hf_aprs_decode_reset_seq != last_reset_seq {
last_reset_seq = state.hf_aprs_decode_reset_seq;
if state.reset_seqs.hf_aprs_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.hf_aprs_decode_reset_seq;
decoder.reset();
info!("HF APRS decoder reset (seq={})", last_reset_seq);
}
@@ -1333,7 +1333,7 @@ pub async fn run_hf_aprs_decoder(
Ok(frame) => {
let reset_seq = {
let state = state_rx.borrow();
state.hf_aprs_decode_reset_seq
state.reset_seqs.hf_aprs_decode_reset_seq
};
if reset_seq != last_reset_seq {
last_reset_seq = reset_seq;
@@ -1348,7 +1348,7 @@ pub async fn run_hf_aprs_decoder(
was_active = true;
let packets = tokio::task::block_in_place(|| decoder.process_samples(&mono));
let latest_reset_seq = state_rx.borrow().hf_aprs_decode_reset_seq;
let latest_reset_seq = state_rx.borrow().reset_seqs.hf_aprs_decode_reset_seq;
if latest_reset_seq != reset_seq {
last_reset_seq = latest_reset_seq;
decoder.reset();
@@ -1381,8 +1381,8 @@ pub async fn run_hf_aprs_decoder(
Ok(()) => {
let state = state_rx.borrow();
active = matches!(state.status.mode, RigMode::DIG);
if state.hf_aprs_decode_reset_seq != last_reset_seq {
last_reset_seq = state.hf_aprs_decode_reset_seq;
if state.reset_seqs.hf_aprs_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.hf_aprs_decode_reset_seq;
decoder.reset();
info!("HF APRS decoder reset (seq={})", last_reset_seq);
}
@@ -1594,7 +1594,7 @@ pub async fn run_cw_decoder(
let mut decoder = CwDecoder::new(sample_rate);
let mut was_active = false;
let mut last_reset_seq: u64 = 0;
let mut active = state_rx.borrow().cw_decode_enabled
let mut active = state_rx.borrow().decoders.cw_decode_enabled
&& matches!(state_rx.borrow().status.mode, RigMode::CW | RigMode::CWR);
let mut last_auto = state_rx.borrow().cw_auto;
let mut last_wpm = state_rx.borrow().cw_wpm;
@@ -1608,7 +1608,7 @@ pub async fn run_cw_decoder(
match state_rx.changed().await {
Ok(()) => {
let state = state_rx.borrow();
active = state.cw_decode_enabled
active = state.decoders.cw_decode_enabled
&& matches!(state.status.mode, RigMode::CW | RigMode::CWR);
if active {
pcm_rx = pcm_rx.resubscribe();
@@ -1625,8 +1625,8 @@ pub async fn run_cw_decoder(
last_tone = state.cw_tone_hz;
decoder.set_tone_hz(last_tone);
}
if state.cw_decode_reset_seq != last_reset_seq {
last_reset_seq = state.cw_decode_reset_seq;
if state.reset_seqs.cw_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.cw_decode_reset_seq;
decoder.reset();
info!("CW decoder reset (seq={})", last_reset_seq);
}
@@ -1643,12 +1643,12 @@ pub async fn run_cw_decoder(
let (process_enabled, cw_auto, cw_wpm, cw_tone_hz, reset_seq) = {
let state = state_rx.borrow();
(
state.cw_decode_enabled
state.decoders.cw_decode_enabled
&& matches!(state.status.mode, RigMode::CW | RigMode::CWR),
state.cw_auto,
state.cw_wpm,
state.cw_tone_hz,
state.cw_decode_reset_seq,
state.reset_seqs.cw_decode_reset_seq,
)
};
if cw_auto != last_auto {
@@ -1692,7 +1692,7 @@ pub async fn run_cw_decoder(
};
was_active = true;
let events = tokio::task::block_in_place(|| decoder.process_samples(&mono));
let latest_reset_seq = state_rx.borrow().cw_decode_reset_seq;
let latest_reset_seq = state_rx.borrow().reset_seqs.cw_decode_reset_seq;
if latest_reset_seq != reset_seq {
last_reset_seq = latest_reset_seq;
decoder.reset();
@@ -1718,7 +1718,7 @@ pub async fn run_cw_decoder(
match changed {
Ok(()) => {
let state = state_rx.borrow();
active = state.cw_decode_enabled
active = state.decoders.cw_decode_enabled
&& matches!(state.status.mode, RigMode::CW | RigMode::CWR);
if state.cw_auto != last_auto {
last_auto = state.cw_auto;
@@ -1732,8 +1732,8 @@ pub async fn run_cw_decoder(
last_tone = state.cw_tone_hz;
decoder.set_tone_hz(last_tone);
}
if state.cw_decode_reset_seq != last_reset_seq {
last_reset_seq = state.cw_decode_reset_seq;
if state.reset_seqs.cw_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.cw_decode_reset_seq;
decoder.reset();
info!("CW decoder reset (seq={})", last_reset_seq);
}
@@ -1826,7 +1826,7 @@ pub async fn run_ft8_decoder(
}
};
let mut last_reset_seq: u64 = 0;
let mut active = state_rx.borrow().ft8_decode_enabled
let mut active = state_rx.borrow().decoders.ft8_decode_enabled
&& matches!(state_rx.borrow().status.mode, RigMode::DIG | RigMode::USB);
let mut ft8_buf: Vec<f32> = Vec::new();
let mut last_slot: i64 = -1;
@@ -1837,13 +1837,13 @@ pub async fn run_ft8_decoder(
match state_rx.changed().await {
Ok(()) => {
let state = state_rx.borrow();
active = state.ft8_decode_enabled
active = state.decoders.ft8_decode_enabled
&& matches!(state.status.mode, RigMode::DIG | RigMode::USB);
if active {
pcm_rx = pcm_rx.resubscribe();
}
if state.ft8_decode_reset_seq != last_reset_seq {
last_reset_seq = state.ft8_decode_reset_seq;
if state.reset_seqs.ft8_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.ft8_decode_reset_seq;
decoder.reset();
ft8_buf.clear();
}
@@ -1871,7 +1871,7 @@ pub async fn run_ft8_decoder(
let reset_seq = {
let state = state_rx.borrow();
state.ft8_decode_reset_seq
state.reset_seqs.ft8_decode_reset_seq
};
if reset_seq != last_reset_seq {
last_reset_seq = reset_seq;
@@ -1895,7 +1895,7 @@ pub async fn run_ft8_decoder(
decoder.process_block(&block);
decoder.decode_if_ready(100)
});
let latest_reset_seq = state_rx.borrow().ft8_decode_reset_seq;
let latest_reset_seq = state_rx.borrow().reset_seqs.ft8_decode_reset_seq;
if latest_reset_seq != reset_seq {
last_reset_seq = latest_reset_seq;
decoder.reset();
@@ -1942,10 +1942,10 @@ pub async fn run_ft8_decoder(
match changed {
Ok(()) => {
let state = state_rx.borrow();
active = state.ft8_decode_enabled
active = state.decoders.ft8_decode_enabled
&& matches!(state.status.mode, RigMode::DIG | RigMode::USB);
if state.ft8_decode_reset_seq != last_reset_seq {
last_reset_seq = state.ft8_decode_reset_seq;
if state.reset_seqs.ft8_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.ft8_decode_reset_seq;
decoder.reset();
ft8_buf.clear();
}
@@ -1982,7 +1982,7 @@ pub async fn run_ft4_decoder(
}
};
let mut last_reset_seq: u64 = 0;
let mut active = state_rx.borrow().ft4_decode_enabled
let mut active = state_rx.borrow().decoders.ft4_decode_enabled
&& matches!(state_rx.borrow().status.mode, RigMode::DIG | RigMode::USB);
let mut ft4_buf: Vec<f32> = Vec::new();
let mut last_slot: i64 = -1;
@@ -1992,13 +1992,13 @@ pub async fn run_ft4_decoder(
match state_rx.changed().await {
Ok(()) => {
let state = state_rx.borrow();
active = state.ft4_decode_enabled
active = state.decoders.ft4_decode_enabled
&& matches!(state.status.mode, RigMode::DIG | RigMode::USB);
if active {
pcm_rx = pcm_rx.resubscribe();
}
if state.ft4_decode_reset_seq != last_reset_seq {
last_reset_seq = state.ft4_decode_reset_seq;
if state.reset_seqs.ft4_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.ft4_decode_reset_seq;
decoder.reset();
ft4_buf.clear();
}
@@ -2027,7 +2027,7 @@ pub async fn run_ft4_decoder(
let reset_seq = {
let state = state_rx.borrow();
state.ft4_decode_reset_seq
state.reset_seqs.ft4_decode_reset_seq
};
if reset_seq != last_reset_seq {
last_reset_seq = reset_seq;
@@ -2051,7 +2051,7 @@ pub async fn run_ft4_decoder(
decoder.process_block(&block);
decoder.decode_if_ready(100)
});
let latest_reset_seq = state_rx.borrow().ft4_decode_reset_seq;
let latest_reset_seq = state_rx.borrow().reset_seqs.ft4_decode_reset_seq;
if latest_reset_seq != reset_seq {
last_reset_seq = latest_reset_seq;
decoder.reset();
@@ -2095,10 +2095,10 @@ pub async fn run_ft4_decoder(
match changed {
Ok(()) => {
let state = state_rx.borrow();
active = state.ft4_decode_enabled
active = state.decoders.ft4_decode_enabled
&& matches!(state.status.mode, RigMode::DIG | RigMode::USB);
if state.ft4_decode_reset_seq != last_reset_seq {
last_reset_seq = state.ft4_decode_reset_seq;
if state.reset_seqs.ft4_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.ft4_decode_reset_seq;
decoder.reset();
ft4_buf.clear();
}
@@ -2135,7 +2135,7 @@ pub async fn run_ft2_decoder(
}
};
let mut last_reset_seq: u64 = 0;
let mut active = state_rx.borrow().ft2_decode_enabled
let mut active = state_rx.borrow().decoders.ft2_decode_enabled
&& matches!(state_rx.borrow().status.mode, RigMode::DIG | RigMode::USB);
let mut ft2_buf: Vec<f32> = Vec::new();
let mut pending_decode_samples: usize = 0;
@@ -2146,13 +2146,13 @@ pub async fn run_ft2_decoder(
match state_rx.changed().await {
Ok(()) => {
let state = state_rx.borrow();
active = state.ft2_decode_enabled
active = state.decoders.ft2_decode_enabled
&& matches!(state.status.mode, RigMode::DIG | RigMode::USB);
if active {
pcm_rx = pcm_rx.resubscribe();
}
if state.ft2_decode_reset_seq != last_reset_seq {
last_reset_seq = state.ft2_decode_reset_seq;
if state.reset_seqs.ft2_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.ft2_decode_reset_seq;
decoder.reset();
ft2_buf.clear();
pending_decode_samples = 0;
@@ -2170,7 +2170,7 @@ pub async fn run_ft2_decoder(
Ok(frame) => {
let reset_seq = {
let state = state_rx.borrow();
state.ft2_decode_reset_seq
state.reset_seqs.ft2_decode_reset_seq
};
if reset_seq != last_reset_seq {
last_reset_seq = reset_seq;
@@ -2199,7 +2199,7 @@ pub async fn run_ft2_decoder(
let results = tokio::task::block_in_place(|| {
decode_ft2_window(&mut decoder, &ft2_buf, 100)
});
let latest_reset_seq = state_rx.borrow().ft2_decode_reset_seq;
let latest_reset_seq = state_rx.borrow().reset_seqs.ft2_decode_reset_seq;
if latest_reset_seq != reset_seq {
last_reset_seq = latest_reset_seq;
decoder.reset();
@@ -2252,10 +2252,10 @@ pub async fn run_ft2_decoder(
match changed {
Ok(()) => {
let state = state_rx.borrow();
active = state.ft2_decode_enabled
active = state.decoders.ft2_decode_enabled
&& matches!(state.status.mode, RigMode::DIG | RigMode::USB);
if state.ft2_decode_reset_seq != last_reset_seq {
last_reset_seq = state.ft2_decode_reset_seq;
if state.reset_seqs.ft2_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.ft2_decode_reset_seq;
decoder.reset();
ft2_buf.clear();
pending_decode_samples = 0;
@@ -2299,7 +2299,7 @@ pub async fn run_wspr_decoder(
}
};
let mut last_reset_seq: u64 = 0;
let mut active = state_rx.borrow().wspr_decode_enabled
let mut active = state_rx.borrow().decoders.wspr_decode_enabled
&& matches!(state_rx.borrow().status.mode, RigMode::DIG | RigMode::USB);
let mut slot_buf: Vec<f32> = Vec::new();
let mut last_slot: i64 = -1;
@@ -2310,13 +2310,13 @@ pub async fn run_wspr_decoder(
match state_rx.changed().await {
Ok(()) => {
let state = state_rx.borrow();
active = state.wspr_decode_enabled
active = state.decoders.wspr_decode_enabled
&& matches!(state.status.mode, RigMode::DIG | RigMode::USB);
if active {
pcm_rx = pcm_rx.resubscribe();
}
if state.wspr_decode_reset_seq != last_reset_seq {
last_reset_seq = state.wspr_decode_reset_seq;
if state.reset_seqs.wspr_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.wspr_decode_reset_seq;
}
slot_buf.clear();
last_slot = -1;
@@ -2337,7 +2337,7 @@ pub async fn run_wspr_decoder(
let slot = now / slot_len_s;
let reset_seq = {
let state = state_rx.borrow();
state.wspr_decode_reset_seq
state.reset_seqs.wspr_decode_reset_seq
};
if reset_seq != last_reset_seq {
last_reset_seq = reset_seq;
@@ -2353,7 +2353,7 @@ pub async fn run_wspr_decoder(
let decode_results = tokio::task::block_in_place(|| {
decoder.decode_slot(&slot_buf, Some(base_freq))
});
let latest_reset_seq = state_rx.borrow().wspr_decode_reset_seq;
let latest_reset_seq = state_rx.borrow().reset_seqs.wspr_decode_reset_seq;
if latest_reset_seq != reset_seq {
last_reset_seq = latest_reset_seq;
slot_buf.clear();
@@ -2388,7 +2388,7 @@ pub async fn run_wspr_decoder(
slot_buf.clear();
last_slot = slot;
}
let latest_reset_seq = state_rx.borrow().wspr_decode_reset_seq;
let latest_reset_seq = state_rx.borrow().reset_seqs.wspr_decode_reset_seq;
if latest_reset_seq != last_reset_seq {
last_reset_seq = latest_reset_seq;
slot_buf.clear();
@@ -2422,10 +2422,10 @@ pub async fn run_wspr_decoder(
match changed {
Ok(()) => {
let state = state_rx.borrow();
active = state.wspr_decode_enabled
active = state.decoders.wspr_decode_enabled
&& matches!(state.status.mode, RigMode::DIG | RigMode::USB);
if state.wspr_decode_reset_seq != last_reset_seq {
last_reset_seq = state.wspr_decode_reset_seq;
if state.reset_seqs.wspr_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.wspr_decode_reset_seq;
slot_buf.clear();
last_slot = -1;
}
@@ -2449,7 +2449,7 @@ pub async fn run_wspr_decoder(
/// Decode Meteor-M LRPT satellite images from QPSK-demodulated baseband.
///
/// The task is idle until `state.lrpt_decode_enabled` becomes `true`.
/// The task is idle until `state.decoders.lrpt_decode_enabled` becomes `true`.
/// When disabled (or 30 s of silence elapses with no new MCUs), the
/// accumulated image is saved and broadcast.
pub async fn run_lrpt_decoder(
@@ -2466,7 +2466,7 @@ pub async fn run_lrpt_decoder(
info!("LRPT decoder started ({}Hz, {} ch)", sample_rate, channels);
let mut decoder = LrptDecoder::new(sample_rate);
let mut last_reset_seq: u64 = 0;
let mut active = state_rx.borrow().lrpt_decode_enabled;
let mut active = state_rx.borrow().decoders.lrpt_decode_enabled;
let mut pass_start_ms: i64 = 0;
let mut last_mcu_at = tokio::time::Instant::now();
@@ -2475,15 +2475,15 @@ pub async fn run_lrpt_decoder(
match state_rx.changed().await {
Ok(()) => {
let state = state_rx.borrow();
active = state.lrpt_decode_enabled;
active = state.decoders.lrpt_decode_enabled;
if active {
decoder.reset();
pass_start_ms = current_timestamp_ms();
last_mcu_at = tokio::time::Instant::now();
pcm_rx = pcm_rx.resubscribe();
}
if state.lrpt_decode_reset_seq != last_reset_seq {
last_reset_seq = state.lrpt_decode_reset_seq;
if state.reset_seqs.lrpt_decode_reset_seq != last_reset_seq {
last_reset_seq = state.reset_seqs.lrpt_decode_reset_seq;
decoder.reset();
}
}
@@ -2498,7 +2498,7 @@ pub async fn run_lrpt_decoder(
recv = pcm_rx.recv() => {
match recv {
Ok(frame) => {
let reset_seq = state_rx.borrow().lrpt_decode_reset_seq;
let reset_seq = state_rx.borrow().reset_seqs.lrpt_decode_reset_seq;
if reset_seq != last_reset_seq {
last_reset_seq = reset_seq;
decoder.reset();
@@ -2523,7 +2523,7 @@ pub async fn run_lrpt_decoder(
if changed.is_ok() {
let (new_active, new_reset_seq) = {
let state = state_rx.borrow();
(state.lrpt_decode_enabled, state.lrpt_decode_reset_seq)
(state.decoders.lrpt_decode_enabled, state.reset_seqs.lrpt_decode_reset_seq)
};
let was_active = active;
active = new_active;
src/trx-server/src/history_store.rs
+100
View File
@@ -191,3 +191,103 @@ pub fn spawn_flush_task(
}
});
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn now_unix_ms_returns_positive() {
let ms = now_unix_ms();
// Should be well past epoch (year 2020+).
assert!(ms > 1_577_836_800_000);
}
#[test]
fn stored_entry_roundtrip_serde() {
let entry = StoredEntry {
ts_ms: 1_700_000_000_000i64,
data: "test message".to_string(),
};
let json = serde_json::to_string(&entry).unwrap();
let decoded: StoredEntry<String> = serde_json::from_str(&json).unwrap();
assert_eq!(decoded.ts_ms, 1_700_000_000_000);
assert_eq!(decoded.data, "test message");
}
#[test]
fn save_and_load_key_roundtrip() {
let dir = std::env::temp_dir().join("trx_history_test");
let _ = std::fs::create_dir_all(&dir);
let db_file = dir.join("test.db");
let mut db = PickleDb::new(
&db_file,
PickleDbDumpPolicy::DumpUponRequest,
SerializationMethod::Json,
);
let mut deque = VecDeque::new();
deque.push_back((Instant::now(), "entry_a".to_string()));
deque.push_back((Instant::now(), "entry_b".to_string()));
save_key(&mut db, "test_key", &deque);
let loaded: Vec<(Instant, String)> = load_key(&db, "test_key");
assert_eq!(loaded.len(), 2);
assert_eq!(loaded[0].1, "entry_a");
assert_eq!(loaded[1].1, "entry_b");
let _ = std::fs::remove_file(&db_file);
let _ = std::fs::remove_dir(&dir);
}
#[test]
fn load_key_filters_expired_entries() {
let dir = std::env::temp_dir().join("trx_history_test_expired");
let _ = std::fs::create_dir_all(&dir);
let db_file = dir.join("test.db");
let mut db = PickleDb::new(
&db_file,
PickleDbDumpPolicy::DumpUponRequest,
SerializationMethod::Json,
);
// Manually insert an entry with an old timestamp.
let entries = vec![
StoredEntry {
ts_ms: 1_000, // Way in the past
data: "old".to_string(),
},
StoredEntry {
ts_ms: now_unix_ms(), // Current
data: "fresh".to_string(),
},
];
let _ = db.set("expiry_test", &entries);
let loaded: Vec<(Instant, String)> = load_key(&db, "expiry_test");
assert_eq!(loaded.len(), 1);
assert_eq!(loaded[0].1, "fresh");
let _ = std::fs::remove_file(&db_file);
let _ = std::fs::remove_dir(&dir);
}
#[test]
fn load_key_missing_returns_empty() {
let dir = std::env::temp_dir().join("trx_history_test_missing");
let _ = std::fs::create_dir_all(&dir);
let db_file = dir.join("test.db");
let db = PickleDb::new(
&db_file,
PickleDbDumpPolicy::DumpUponRequest,
SerializationMethod::Json,
);
let loaded: Vec<(Instant, String)> = load_key(&db, "nonexistent");
assert!(loaded.is_empty());
let _ = std::fs::remove_file(&db_file);
let _ = std::fs::remove_dir(&dir);
}
}
src/trx-server/src/listener.rs
+89 -8
View File
@@ -37,6 +37,8 @@ const DEFAULT_IO_TIMEOUT: Duration = Duration::from_secs(10);
/// Fallback request timeout used when no config value is provided.
const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(12);
const MAX_JSON_LINE_BYTES: usize = 256 * 1024;
/// Maximum concurrent connections allowed from a single IP address.
const MAX_CONNECTIONS_PER_IP: usize = 10;
/// Configurable timeout values for the listener, threaded from `[timeouts]`.
#[derive(Debug, Clone, Copy)]
@@ -65,6 +67,38 @@ struct SatPassCache {
result: trx_core::geo::PassPredictionResult,
computed_at: Instant,
}
/// Per-IP connection tracker for rate limiting.
struct ConnectionTracker {
counts: HashMap<std::net::IpAddr, usize>,
}
impl ConnectionTracker {
fn new() -> Self {
Self {
counts: HashMap::new(),
}
}
fn try_acquire(&mut self, ip: std::net::IpAddr) -> bool {
let count = self.counts.entry(ip).or_insert(0);
if *count >= MAX_CONNECTIONS_PER_IP {
false
} else {
*count += 1;
true
}
}
fn release(&mut self, ip: std::net::IpAddr) {
if let Some(count) = self.counts.get_mut(&ip) {
*count = count.saturating_sub(1);
if *count == 0 {
self.counts.remove(&ip);
}
}
}
}
/// Shared state passed to each client handler.
struct ClientContext {
rigs: Arc<HashMap<String, RigHandle>>,
@@ -93,11 +127,24 @@ pub async fn run_listener(
info!("Listening on {}", addr);
let validator = Arc::new(SimpleTokenValidator::new(auth_tokens));
let sat_pass_cache: Arc<Mutex<Option<SatPassCache>>> = Arc::new(Mutex::new(None));
let conn_tracker = Arc::new(Mutex::new(ConnectionTracker::new()));
loop {
tokio::select! {
accept = listener.accept() => {
let (socket, peer) = accept?;
// Per-IP connection rate limiting.
let peer_ip = peer.ip();
{
let mut tracker = conn_tracker.lock().unwrap_or_else(|e| e.into_inner());
if !tracker.try_acquire(peer_ip) {
warn!("Rejecting connection from {} (per-IP limit reached)", peer);
drop(socket);
continue;
}
}
info!("Client connected: {}", peer);
let ctx = ClientContext {
@@ -109,10 +156,15 @@ pub async fn run_listener(
timeouts,
};
let client_shutdown_rx = shutdown_rx.clone();
let tracker_clone = Arc::clone(&conn_tracker);
tokio::spawn(async move {
if let Err(e) = handle_client(socket, peer, ctx, client_shutdown_rx).await {
error!("Client {} error: {:?}", peer, e);
}
// Release connection slot when client disconnects.
if let Ok(mut tracker) = tracker_clone.lock() {
tracker.release(peer_ip);
}
});
}
changed = shutdown_rx.changed() => {
@@ -266,6 +318,7 @@ async fn handle_client(
let resp = ClientResponse {
success: false,
rig_id: None,
protocol_version: None,
state: None,
rigs: None,
sat_passes: None,
@@ -280,6 +333,7 @@ async fn handle_client(
let resp = ClientResponse {
success: false,
rig_id: None,
protocol_version: None,
state: None,
rigs: None,
sat_passes: None,
@@ -313,6 +367,7 @@ async fn handle_client(
let resp = ClientResponse {
success: true,
rig_id: Some("server".to_string()),
protocol_version: None,
state: None,
rigs: Some(entries),
sat_passes: None,
@@ -348,15 +403,32 @@ async fn handle_client(
.unwrap_or_default()
.as_millis() as i64;
let window_ms = 24 * 3600 * 1000; // 24 hours
let fresh = tokio::task::spawn_blocking(move || {
trx_core::geo::compute_upcoming_passes(lat, lon, now_ms, window_ms)
})
let fresh = match time::timeout(
Duration::from_secs(30),
tokio::task::spawn_blocking(move || {
trx_core::geo::compute_upcoming_passes(lat, lon, now_ms, window_ms)
}),
)
.await
.unwrap_or_else(|_| trx_core::geo::PassPredictionResult {
passes: vec![],
satellite_count: 0,
tle_source: trx_core::geo::TleSource::Unavailable,
});
{
Ok(Ok(result)) => result,
Ok(Err(e)) => {
warn!("Satellite pass computation panicked: {:?}", e);
trx_core::geo::PassPredictionResult {
passes: vec![],
satellite_count: 0,
tle_source: trx_core::geo::TleSource::Unavailable,
}
}
Err(_) => {
warn!("Satellite pass computation timed out after 30s");
trx_core::geo::PassPredictionResult {
passes: vec![],
satellite_count: 0,
tle_source: trx_core::geo::TleSource::Unavailable,
}
}
};
// Update cache.
if let Ok(mut guard) = sat_pass_cache.lock() {
*guard = Some(SatPassCache {
@@ -375,6 +447,7 @@ async fn handle_client(
let resp = ClientResponse {
success: true,
rig_id: Some("server".to_string()),
protocol_version: None,
state: None,
rigs: None,
sat_passes: Some(result),
@@ -392,6 +465,7 @@ async fn handle_client(
let resp = ClientResponse {
success: false,
rig_id: Some(target_rig_id.clone()),
protocol_version: None,
state: None,
rigs: None,
sat_passes: None,
@@ -411,6 +485,7 @@ async fn handle_client(
let resp = ClientResponse {
success: true,
rig_id: Some(target_rig_id.clone()),
protocol_version: None,
state: Some(snapshot),
rigs: None,
sat_passes: None,
@@ -438,6 +513,7 @@ async fn handle_client(
let resp = ClientResponse {
success: false,
rig_id: Some(target_rig_id.clone()),
protocol_version: None,
state: None,
rigs: None,
sat_passes: None,
@@ -450,6 +526,7 @@ async fn handle_client(
let resp = ClientResponse {
success: false,
rig_id: Some(target_rig_id.clone()),
protocol_version: None,
state: None,
rigs: None,
sat_passes: None,
@@ -468,6 +545,7 @@ async fn handle_client(
let resp = ClientResponse {
success: false,
rig_id: Some(target_rig_id.clone()),
protocol_version: None,
state: None,
rigs: None,
sat_passes: None,
@@ -493,6 +571,7 @@ async fn handle_client(
let resp = ClientResponse {
success: true,
rig_id: Some(target_rig_id.clone()),
protocol_version: None,
state: Some(snapshot),
rigs: None,
sat_passes: None,
@@ -504,6 +583,7 @@ async fn handle_client(
let resp = ClientResponse {
success: false,
rig_id: Some(target_rig_id.clone()),
protocol_version: None,
state: None,
rigs: None,
sat_passes: None,
@@ -516,6 +596,7 @@ async fn handle_client(
let resp = ClientResponse {
success: false,
rig_id: Some(target_rig_id.clone()),
protocol_version: None,
state: None,
rigs: None,
sat_passes: None,
src/trx-server/src/rig_task.rs
+69 -69
View File
@@ -463,12 +463,12 @@ async fn process_command(
// Handle decoder commands early — they don't touch the rig CAT.
match cmd {
RigCommand::SetAprsDecodeEnabled(en) => {
ctx.state.aprs_decode_enabled = en;
ctx.state.decoders.aprs_decode_enabled = en;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::SetCwDecodeEnabled(en) => {
ctx.state.cw_decode_enabled = en;
ctx.state.decoders.cw_decode_enabled = en;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
@@ -488,85 +488,85 @@ async fn process_command(
return snapshot_from(ctx.state);
}
RigCommand::SetFt8DecodeEnabled(en) => {
ctx.state.ft8_decode_enabled = en;
ctx.state.decoders.ft8_decode_enabled = en;
info!("FT8 decode {}", if en { "enabled" } else { "disabled" });
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::SetFt4DecodeEnabled(en) => {
ctx.state.ft4_decode_enabled = en;
ctx.state.decoders.ft4_decode_enabled = en;
info!("FT4 decode {}", if en { "enabled" } else { "disabled" });
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::SetFt2DecodeEnabled(en) => {
ctx.state.ft2_decode_enabled = en;
ctx.state.decoders.ft2_decode_enabled = en;
info!("FT2 decode {}", if en { "enabled" } else { "disabled" });
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::SetWsprDecodeEnabled(en) => {
ctx.state.wspr_decode_enabled = en;
ctx.state.decoders.wspr_decode_enabled = en;
info!("WSPR decode {}", if en { "enabled" } else { "disabled" });
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::ResetAprsDecoder => {
ctx.histories.clear_aprs_history();
ctx.state.aprs_decode_reset_seq += 1;
ctx.state.reset_seqs.aprs_decode_reset_seq += 1;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::SetHfAprsDecodeEnabled(en) => {
ctx.state.hf_aprs_decode_enabled = en;
ctx.state.decoders.hf_aprs_decode_enabled = en;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::ResetHfAprsDecoder => {
ctx.histories.clear_hf_aprs_history();
ctx.state.hf_aprs_decode_reset_seq += 1;
ctx.state.reset_seqs.hf_aprs_decode_reset_seq += 1;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::ResetCwDecoder => {
ctx.histories.clear_cw_history();
ctx.state.cw_decode_reset_seq += 1;
ctx.state.reset_seqs.cw_decode_reset_seq += 1;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::ResetFt8Decoder => {
ctx.histories.clear_ft8_history();
ctx.state.ft8_decode_reset_seq += 1;
ctx.state.reset_seqs.ft8_decode_reset_seq += 1;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::ResetFt4Decoder => {
ctx.histories.clear_ft4_history();
ctx.state.ft4_decode_reset_seq += 1;
ctx.state.reset_seqs.ft4_decode_reset_seq += 1;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::ResetFt2Decoder => {
ctx.histories.clear_ft2_history();
ctx.state.ft2_decode_reset_seq += 1;
ctx.state.reset_seqs.ft2_decode_reset_seq += 1;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::ResetWsprDecoder => {
ctx.histories.clear_wspr_history();
ctx.state.wspr_decode_reset_seq += 1;
ctx.state.reset_seqs.wspr_decode_reset_seq += 1;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::SetLrptDecodeEnabled(en) => {
ctx.state.lrpt_decode_enabled = en;
ctx.state.decoders.lrpt_decode_enabled = en;
info!("LRPT decode {}", if en { "enabled" } else { "disabled" });
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
RigCommand::ResetLrptDecoder => {
ctx.histories.clear_lrpt_history();
ctx.state.lrpt_decode_reset_seq += 1;
ctx.state.reset_seqs.lrpt_decode_reset_seq += 1;
let _ = ctx.state_tx.send(ctx.state.clone());
return snapshot_from(ctx.state);
}
@@ -1065,23 +1065,23 @@ fn invalidate_main_decoder_windows_on_freq_change(state: &mut RigState, prev_fre
match state.status.mode {
RigMode::PKT => {
state.aprs_decode_reset_seq += 1;
state.reset_seqs.aprs_decode_reset_seq += 1;
}
RigMode::DIG => {
state.hf_aprs_decode_reset_seq += 1;
state.ft8_decode_reset_seq += 1;
state.ft4_decode_reset_seq += 1;
state.ft2_decode_reset_seq += 1;
state.wspr_decode_reset_seq += 1;
state.reset_seqs.hf_aprs_decode_reset_seq += 1;
state.reset_seqs.ft8_decode_reset_seq += 1;
state.reset_seqs.ft4_decode_reset_seq += 1;
state.reset_seqs.ft2_decode_reset_seq += 1;
state.reset_seqs.wspr_decode_reset_seq += 1;
}
RigMode::USB => {
state.ft8_decode_reset_seq += 1;
state.ft4_decode_reset_seq += 1;
state.ft2_decode_reset_seq += 1;
state.wspr_decode_reset_seq += 1;
state.reset_seqs.ft8_decode_reset_seq += 1;
state.reset_seqs.ft4_decode_reset_seq += 1;
state.reset_seqs.ft2_decode_reset_seq += 1;
state.reset_seqs.wspr_decode_reset_seq += 1;
}
RigMode::CW | RigMode::CWR => {
state.cw_decode_reset_seq += 1;
state.reset_seqs.cw_decode_reset_seq += 1;
}
_ => {}
}
@@ -1235,13 +1235,13 @@ mod tests {
invalidate_main_decoder_windows_on_freq_change(&mut state, prev_freq_hz);
assert_eq!(state.aprs_decode_reset_seq, 1);
assert_eq!(state.hf_aprs_decode_reset_seq, 0);
assert_eq!(state.cw_decode_reset_seq, 0);
assert_eq!(state.ft8_decode_reset_seq, 0);
assert_eq!(state.ft4_decode_reset_seq, 0);
assert_eq!(state.ft2_decode_reset_seq, 0);
assert_eq!(state.wspr_decode_reset_seq, 0);
assert_eq!(state.reset_seqs.aprs_decode_reset_seq, 1);
assert_eq!(state.reset_seqs.hf_aprs_decode_reset_seq, 0);
assert_eq!(state.reset_seqs.cw_decode_reset_seq, 0);
assert_eq!(state.reset_seqs.ft8_decode_reset_seq, 0);
assert_eq!(state.reset_seqs.ft4_decode_reset_seq, 0);
assert_eq!(state.reset_seqs.ft2_decode_reset_seq, 0);
assert_eq!(state.reset_seqs.wspr_decode_reset_seq, 0);
}
#[test]
@@ -1255,26 +1255,26 @@ mod tests {
invalidate_main_decoder_windows_on_freq_change(&mut state, prev_freq_hz);
assert_eq!(state.aprs_decode_reset_seq, 0);
assert_eq!(state.hf_aprs_decode_reset_seq, 1);
assert_eq!(state.cw_decode_reset_seq, 0);
assert_eq!(state.ft8_decode_reset_seq, 1);
assert_eq!(state.ft4_decode_reset_seq, 1);
assert_eq!(state.ft2_decode_reset_seq, 1);
assert_eq!(state.wspr_decode_reset_seq, 1);
assert_eq!(state.reset_seqs.aprs_decode_reset_seq, 0);
assert_eq!(state.reset_seqs.hf_aprs_decode_reset_seq, 1);
assert_eq!(state.reset_seqs.cw_decode_reset_seq, 0);
assert_eq!(state.reset_seqs.ft8_decode_reset_seq, 1);
assert_eq!(state.reset_seqs.ft4_decode_reset_seq, 1);
assert_eq!(state.reset_seqs.ft2_decode_reset_seq, 1);
assert_eq!(state.reset_seqs.wspr_decode_reset_seq, 1);
}
#[test]
fn wfm_freq_change_does_not_touch_main_decoders() {
let mut state = RigState::new_uninitialized();
state.apply_mode(RigMode::WFM);
state.aprs_decode_reset_seq = 2;
state.hf_aprs_decode_reset_seq = 3;
state.cw_decode_reset_seq = 4;
state.ft8_decode_reset_seq = 5;
state.ft4_decode_reset_seq = 6;
state.ft2_decode_reset_seq = 7;
state.wspr_decode_reset_seq = 8;
state.reset_seqs.aprs_decode_reset_seq = 2;
state.reset_seqs.hf_aprs_decode_reset_seq = 3;
state.reset_seqs.cw_decode_reset_seq = 4;
state.reset_seqs.ft8_decode_reset_seq = 5;
state.reset_seqs.ft4_decode_reset_seq = 6;
state.reset_seqs.ft2_decode_reset_seq = 7;
state.reset_seqs.wspr_decode_reset_seq = 8;
let prev_freq_hz = state.status.freq.hz;
state.apply_freq(Freq {
hz: prev_freq_hz + 200_000,
@@ -1282,35 +1282,35 @@ mod tests {
invalidate_main_decoder_windows_on_freq_change(&mut state, prev_freq_hz);
assert_eq!(state.aprs_decode_reset_seq, 2);
assert_eq!(state.hf_aprs_decode_reset_seq, 3);
assert_eq!(state.cw_decode_reset_seq, 4);
assert_eq!(state.ft8_decode_reset_seq, 5);
assert_eq!(state.ft4_decode_reset_seq, 6);
assert_eq!(state.ft2_decode_reset_seq, 7);
assert_eq!(state.wspr_decode_reset_seq, 8);
assert_eq!(state.reset_seqs.aprs_decode_reset_seq, 2);
assert_eq!(state.reset_seqs.hf_aprs_decode_reset_seq, 3);
assert_eq!(state.reset_seqs.cw_decode_reset_seq, 4);
assert_eq!(state.reset_seqs.ft8_decode_reset_seq, 5);
assert_eq!(state.reset_seqs.ft4_decode_reset_seq, 6);
assert_eq!(state.reset_seqs.ft2_decode_reset_seq, 7);
assert_eq!(state.reset_seqs.wspr_decode_reset_seq, 8);
}
#[test]
fn unchanged_freq_keeps_decoder_windows_intact() {
let mut state = RigState::new_uninitialized();
state.aprs_decode_reset_seq = 2;
state.hf_aprs_decode_reset_seq = 3;
state.cw_decode_reset_seq = 4;
state.ft8_decode_reset_seq = 5;
state.ft4_decode_reset_seq = 6;
state.ft2_decode_reset_seq = 7;
state.wspr_decode_reset_seq = 8;
state.reset_seqs.aprs_decode_reset_seq = 2;
state.reset_seqs.hf_aprs_decode_reset_seq = 3;
state.reset_seqs.cw_decode_reset_seq = 4;
state.reset_seqs.ft8_decode_reset_seq = 5;
state.reset_seqs.ft4_decode_reset_seq = 6;
state.reset_seqs.ft2_decode_reset_seq = 7;
state.reset_seqs.wspr_decode_reset_seq = 8;
let prev_freq_hz = state.status.freq.hz;
invalidate_main_decoder_windows_on_freq_change(&mut state, prev_freq_hz);
assert_eq!(state.aprs_decode_reset_seq, 2);
assert_eq!(state.hf_aprs_decode_reset_seq, 3);
assert_eq!(state.cw_decode_reset_seq, 4);
assert_eq!(state.ft8_decode_reset_seq, 5);
assert_eq!(state.ft4_decode_reset_seq, 6);
assert_eq!(state.ft2_decode_reset_seq, 7);
assert_eq!(state.wspr_decode_reset_seq, 8);
assert_eq!(state.reset_seqs.aprs_decode_reset_seq, 2);
assert_eq!(state.reset_seqs.hf_aprs_decode_reset_seq, 3);
assert_eq!(state.reset_seqs.cw_decode_reset_seq, 4);
assert_eq!(state.reset_seqs.ft8_decode_reset_seq, 5);
assert_eq!(state.reset_seqs.ft4_decode_reset_seq, 6);
assert_eq!(state.reset_seqs.ft2_decode_reset_seq, 7);
assert_eq!(state.reset_seqs.wspr_decode_reset_seq, 8);
}
}
src/trx-server/trx-backend/trx-backend-ft817/src/lib.rs
+21 -6
View File
@@ -448,13 +448,27 @@ impl Ft817 {
Ft817VfoSide::A => self.vfo_a_freq = Some(freq),
Ft817VfoSide::B => self.vfo_b_freq = Some(freq),
Ft817VfoSide::Unknown => {
// Try to infer which VFO we are on using cached values; default to A only.
if self.vfo_b_freq.map(|f| f.hz == freq.hz).unwrap_or(false)
&& self.vfo_a_freq.is_none()
{
self.vfo_side = Ft817VfoSide::B;
self.vfo_b_freq = Some(freq);
// Infer which VFO we are on using cached values.
//
// When VFO B has a known frequency that differs from the current
// reading and VFO A is unset, we can infer VFO A. When frequencies
// match (ambiguous case), default to VFO A — the ambiguity is
// resolved after the first VFO toggle (see toggle_vfo_side).
if let Some(cached_b) = self.vfo_b_freq {
if cached_b.hz == freq.hz && self.vfo_a_freq.is_none() {
// Could be either VFO; default to A (will be corrected
// after toggle_vfo primes both sides).
self.vfo_side = Ft817VfoSide::A;
self.vfo_a_freq = Some(freq);
} else if cached_b.hz != freq.hz {
// Different frequency from cached B → must be A.
self.vfo_side = Ft817VfoSide::A;
self.vfo_a_freq = Some(freq);
} else {
self.vfo_b_freq = Some(freq);
}
} else {
// No cached B at all; assume A.
self.vfo_side = Ft817VfoSide::A;
self.vfo_a_freq = Some(freq);
}
@@ -472,6 +486,7 @@ impl Ft817 {
}
}
}
}
impl Rig for Ft817 {