jan/rusty-telemetry
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add run deletion, session reopen, and feed merging (v0.6.0)

Browse source 
- DELETE /api/sessions/{id}/runs/{run_id} removes stopped runs
- POST /api/sessions/{id}/reopen transitions Completed → Active
- Optional merge_feeds field on POST /api/sessions/{id}/runs combines
  frames from multiple feeds with priority-based deduplication
- In-place merge_frames() avoids 2× peak memory
- All new fields use #[serde(default)] for backward compatibility
- Updated README.md with new endpoints and feed merging docs
This commit is contained in:
Jan-Peter von Hunnius 2026-06-06 09:38:42 +02:00
parent e3aa511ac7
commit 9fae8ba9d8
5 changed files with 208 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

30
src/api.rs
View file

@ -7,7 +7,7 @@ use axum::{
extract::{Path, State},
http::StatusCode,
response::{IntoResponse, Response},
routing::{get, post},
routing::{delete, get, post},
};
use serde::{Deserialize, Serialize};
use std::sync::{Arc, Mutex, MutexGuard};
@ -85,6 +85,8 @@ struct CreateSessionRequest {
struct StartRunRequest {
feed_name: String,
label: Option<String>,
#[serde(default)]
merge_feeds: Vec<String>,
}
async fn create_session(
@ -132,7 +134,7 @@ async fn start_run(
Json(req): Json<StartRunRequest>,
) -> Response {
let mut mgr = lock_sessions(&state);
match mgr.start_run(&id, &req.feed_name, req.label) {
match mgr.start_run(&id, &req.feed_name, req.label, req.merge_feeds) {
Ok(run) => (StatusCode::CREATED, Json(run)).into_response(),
Err(msg) => err(StatusCode::BAD_REQUEST, &msg),
}
@ -149,6 +151,28 @@ async fn stop_run(
}
}
async fn delete_run(
State(state): State<Arc<AppState>>,
Path((id, run_id)): Path<(String, String)>,
) -> Response {
let mut mgr = lock_sessions(&state);
match mgr.delete_run(&id, &run_id) {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(msg) => err(StatusCode::BAD_REQUEST, &msg),
}
}
async fn reopen_session(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let mut mgr = lock_sessions(&state);
match mgr.reopen_session(&id) {
Ok(session) => ok_json(session),
Err(msg) => err(StatusCode::BAD_REQUEST, &msg),
}
}
async fn enrich_with_recording_meta(
state: &Arc<AppState>,
result: &mut serde_json::Value,
@ -521,6 +545,8 @@ pub fn build_router(state: Arc<AppState>) -> Router {
.route("/api/sessions/{id}", get(get_session).delete(delete_session))
.route("/api/sessions/{id}/runs", post(start_run))
.route("/api/sessions/{id}/runs/{run_id}/stop", post(stop_run))
.route("/api/sessions/{id}/runs/{run_id}", delete(delete_run))
.route("/api/sessions/{id}/reopen", post(reopen_session))
.route("/api/sessions/{id}/complete", post(complete_session))
.route("/api/sessions/{id}/analysis", get(get_analysis))
.route("/api/sessions/{id}/analyze", post(run_analysis))

137
src/session.rs
View file

@ -31,9 +31,15 @@ pub struct SessionRun {
pub id: String,
pub label: String,
pub feed_name: String,
#[serde(default)]
pub merge_feeds: Vec<String>,
pub status: RunStatus,
pub start_marker: u64,
#[serde(default)]
pub merge_start_markers: HashMap<String, u64>,
pub end_marker: Option<u64>,
#[serde(default)]
pub merge_end_markers: HashMap<String, Option<u64>>,
pub frame_count: usize,
pub data_truncated: bool,
#[serde(skip)]
@ -46,9 +52,12 @@ impl SessionRun {
id: self.id.clone(),
label: self.label.clone(),
feed_name: self.feed_name.clone(),
merge_feeds: self.merge_feeds.clone(),
status: self.status.clone(),
start_marker: self.start_marker,
merge_start_markers: self.merge_start_markers.clone(),
end_marker: self.end_marker,
merge_end_markers: self.merge_end_markers.clone(),
frame_count: self.frame_count,
data_truncated: self.data_truncated,
frames: Vec::new(),
@ -145,6 +154,7 @@ impl SessionManager {
session_id: &str,
feed_name: &str,
label: Option<String>,
merge_feeds: Vec<String>,
) -> Result<SessionRun, String> {
let session = self
.sessions
@ -156,14 +166,22 @@ impl SessionManager {
}
let marker = self.recording_manager.ring_marker(feed_name);
let mut merge_start_markers = HashMap::new();
for mf in &merge_feeds {
merge_start_markers.insert(mf.clone(), self.recording_manager.ring_marker(mf));
}
let run_id = uuid::Uuid::new_v4().to_string();
let run = SessionRun {
id: run_id,
label: label.unwrap_or_else(|| format!("Run {}", session.runs.len() + 1)),
feed_name: feed_name.to_string(),
merge_feeds,
status: RunStatus::Recording,
start_marker: marker,
merge_start_markers,
end_marker: None,
merge_end_markers: HashMap::new(),
frame_count: 0,
data_truncated: false,
frames: Vec::new(),
@ -200,17 +218,37 @@ impl SessionManager {
fn finalize_run(recording_manager: &Arc<RecordingManager>, run: &mut SessionRun) {
let end_marker = recording_manager.ring_marker(&run.feed_name);
let frames = recording_manager
let primary_frames = recording_manager
.extract_since(&run.feed_name, run.start_marker);
let expected = end_marker.saturating_sub(run.start_marker);
let data_truncated = (frames.len() as u64) < expected && expected > 0;
let primary_truncated = (primary_frames.len() as u64) < expected && expected > 0;
let mut all_frames = primary_frames;
let mut any_truncated = primary_truncated;
for feed in &run.merge_feeds {
if let Some(&start_m) = run.merge_start_markers.get(feed) {
let end_m = recording_manager.ring_marker(feed);
let frames = recording_manager.extract_since(feed, start_m);
let exp = end_m.saturating_sub(start_m);
if (frames.len() as u64) < exp && exp > 0 {
any_truncated = true;
}
run.merge_end_markers.insert(feed.clone(), Some(end_m));
all_frames.extend(frames);
}
}
if !run.merge_feeds.is_empty() {
all_frames = merge_frames(all_frames);
}
run.end_marker = Some(end_marker);
run.frame_count = frames.len();
run.frames = frames;
run.frame_count = all_frames.len();
run.frames = all_frames;
run.status = RunStatus::Stopped;
run.data_truncated = data_truncated;
run.data_truncated = any_truncated;
}
pub fn complete_session(&mut self, id: &str) -> Result<UseCaseSession, String> {
@ -235,6 +273,48 @@ impl SessionManager {
Ok(session.summary_clone())
}
pub fn delete_run(&mut self, session_id: &str, run_id: &str) -> Result<(), String> {
let session = self
.sessions
.get_mut(session_id)
.ok_or_else(|| format!("Session '{}' not found", session_id))?;
let run_idx = session
.runs
.iter()
.position(|r| r.id == run_id)
.ok_or_else(|| format!("Run '{}' not found in session '{}'", run_id, session_id))?;
if session.runs[run_idx].status == RunStatus::Recording {
return Err("Cannot delete a recording run. Stop it first.".to_string());
}
session.runs.remove(run_idx);
if session.status == SessionStatus::Completed && session.analysis.is_some() {
session.analysis = None;
}
Ok(())
}
pub fn reopen_session(&mut self, session_id: &str) -> Result<UseCaseSession, String> {
let session = self
.sessions
.get_mut(session_id)
.ok_or_else(|| format!("Session '{}' not found", session_id))?;
if session.status != SessionStatus::Completed {
return Err(format!("Session '{}' is not completed", session_id));
}
session.status = SessionStatus::Active;
session.completed_at = None;
session.analysis = None;
Ok(session.summary_clone())
}
pub fn set_analysis(
&mut self,
session_id: &str,
@ -271,3 +351,50 @@ impl SessionManager {
Ok(())
}
}
/// Merge frames from multiple feeds into a single sorted, deduplicated sequence.
///
/// Frames are sorted by `(lap_count, lap_time_ms)`. Near-duplicate frames
/// (within 15ms on the same lap) are resolved by keeping the frame from the
/// higher-priority feed: Telemetry Tool > KSUDP > PCARS.
fn merge_frames(mut frames: Vec<TelemetryFrame>) -> Vec<TelemetryFrame> {
if frames.len() <= 1 {
return frames;
}
frames.sort_by(|a, b| {
a.lap_count
.cmp(&b.lap_count)
.then_with(|| a.lap_time_ms.cmp(&b.lap_time_ms))
});
const TOLERANCE_MS: i32 = 15;
let mut write = 0;
for read in 1..frames.len() {
if frames[write].lap_count == frames[read].lap_count
&& (frames[read].lap_time_ms - frames[write].lap_time_ms).abs() <= TOLERANCE_MS
{
// Near-duplicate: keep higher-priority frame at write position
if feed_priority(frames[read].source) > feed_priority(frames[write].source) {
frames[write] = std::mem::replace(&mut frames[read], TelemetryFrame::default());
}
} else {
write += 1;
frames.swap(write, read);
}
}
frames.truncate(write + 1);
frames
}
/// Feed priority for merge deduplication: TelemetryTool > KSUDP > PCARS.
fn feed_priority(source: crate::telemetry_frame::FeedSource) -> u8 {
use crate::telemetry_frame::FeedSource;
match source {
FeedSource::TelemetryTool => 3,
FeedSource::Ksudp => 2,
FeedSource::Pcars => 1,
}
}