bhtune_server/routes/
stream.rs

1//! `GET /api/runs/{id}/stream` -- pushes per-tick engine state to the browser over
2//! Server-Sent Events as a run progresses, so the frontend's live trend chart doesn't have
3//! to poll `GET /api/runs/{id}` and re-fetch the whole (ever-growing) `samples` array once a
4//! second, the way `frontend/src/api/runs.ts`'s `useRun` interim-substitute polling did
5//! before this endpoint existed (`frontend-live-stream`, the last piece `frontend-screens`
6//! was blocked on).
7//!
8//! SSE, not WebSocket: the flow is strictly server -> client, and SSE gives every browser
9//! automatic reconnection on a dropped connection, survives ordinary HTTP proxies, and is
10//! trivially inspectable with `curl -N` -- see AGENTS.md's "Web app architecture" section for
11//! the full rationale.
12//!
13//! Implemented as an internal poll of `tune_samples`/`tune_runs`, **not** a broadcast channel
14//! threaded through `bhtune-cli::commands::tune`'s already-heavily-tested tick loop --
15//! deliberately, so this endpoint adds zero risk to the shared CLI/server tune-execution code
16//! path (`run_polling_loop` keeps its existing, already-proven signature and test suite
17//! untouched). At this project's documented data volumes (a pathological 2-hour run is
18//! ~9,000 samples -- see AGENTS.md's "History explorer" notes) polling the database every
19//! [`POLL_INTERVAL`] is negligible cost, not a premature optimization to avoid.
20
21use std::convert::Infallible;
22use std::time::Duration;
23use std::time::Instant;
24
25use async_stream::stream;
26use axum::Router;
27use axum::extract::{Path, State};
28use axum::response::IntoResponse;
29use axum::response::sse::{Event, KeepAlive, Sse};
30use axum::routing::get;
31use bhtune_db::models::{TuneOutcome, TuneRunRow, TuneSampleRow};
32use serde::Serialize;
33use utoipa::ToSchema;
34
35use crate::error::ApiError;
36use crate::routes::history::{InitialReadingsResponse, SampleResponse};
37use crate::state::AppState;
38
39/// How often the stream re-polls the database for newly recorded samples or a changed run
40/// outcome. Well under the CLI's own default 800ms tick interval, so a connected client never
41/// perceives added latency beyond this poll interval itself; see the module doc for why
42/// polling the database is an acceptable (not merely expedient) choice at this project's data
43/// volumes.
44const POLL_INTERVAL: Duration = Duration::from_millis(300);
45
46/// The final event emitted on every `GET /api/runs/{id}/stream` connection, named `done`,
47/// immediately before the stream closes. Deliberately carries only the outcome rather than
48/// duplicating the full run detail: the frontend already has `GET /api/runs/{id}` (via
49/// `useRun`) for the config/results/write-back-audit shape, so `done` just signals "stop
50/// listening and go refetch that" rather than growing this endpoint a second response shape.
51#[derive(Debug, Serialize, ToSchema)]
52pub(crate) struct RunStreamDone {
53    outcome: TuneOutcome,
54}
55
56/// The stream's item type is `Result<Event, Infallible>` -- see the module doc for why a
57/// real transport-level error is never produced (a DB failure mid-stream is represented as a
58/// named `error` SSE event instead, followed by ending the stream). A tiny helper rather
59/// than a turbofish (`Ok::<Event, Infallible>(event)`) at every `yield` site, since the
60/// `stream!` macro can't infer `Infallible` from context on its own (every `yield` only ever
61/// produces `Ok`, so there's no `Err(...)` arm anywhere for type inference to anchor on).
62fn ok_event(event: Event) -> Result<Event, Infallible> {
63    Ok(event)
64}
65
66/// Stream per-tick engine state for one run over Server-Sent Events.
67///
68/// `GET /api/runs/{id}/stream` -- 404 if no run has that id. Emits one `initial` event (JSON
69/// body: [`InitialReadingsResponse`]) as soon as the driver's initial snapshot is persisted,
70/// then a `sample` event (JSON body: [`SampleResponse`], the same shape
71/// `GET /api/runs/{id}`'s `samples` array already uses) for every tick recorded so far and
72/// every new tick recorded while connected, followed by exactly one final `done` event (JSON
73/// body: [`RunStreamDone`]) once the run reaches a terminal outcome -- after which the
74/// connection closes. Safe to open at any point in a run's lifecycle, including after it has
75/// already finished: the initial snapshot (when available) and every sample are replayed once,
76/// immediately followed by `done`.
77#[utoipa::path(
78    get,
79    path = "/api/runs/{id}/stream",
80    tag = "runs",
81    params(
82        ("id" = i64, Path, description = "Run id"),
83    ),
84    responses(
85        (
86            status = 200,
87            description = "A `text/event-stream` with an optional `initial` event \
88                (data: InitialReadingsResponse), `sample` events (data: SampleResponse), \
89                and one final `done` event (data: RunStreamDone).",
90            content_type = "text/event-stream",
91            body = SampleResponse,
92        ),
93        (status = 404, description = "No run with that id.", body = crate::error::ErrorBody),
94    ),
95)]
96pub(crate) async fn stream_run(
97    State(state): State<AppState>,
98    Path(run_id): Path<i64>,
99) -> Result<impl IntoResponse, ApiError> {
100    stream_run_with_permit(state, run_id, None, None).await
101}
102
103pub(crate) async fn stream_run_with_permit(
104    state: AppState,
105    run_id: i64,
106    permit: Option<Vec<tokio::sync::OwnedSemaphorePermit>>,
107    timeout: Option<Duration>,
108) -> Result<impl IntoResponse, ApiError> {
109    if TuneRunRow::get(&state.pool, run_id).await?.is_none() {
110        return Err(ApiError::NotFound(format!("no run with id {run_id}")));
111    }
112
113    let pool = state.pool.clone();
114    let events = stream! {
115        let _permit = permit;
116        let started = Instant::now();
117        // `-1`: nothing sent yet, so the first poll below fetches every tick recorded so
118        // far -- see `TuneSampleRow::list_for_run_since`'s own doc comment for why `-1` is a
119        // safe "everything" sentinel (`tick` is always `>= 0`).
120        let mut last_tick: i64 = -1;
121        let mut sent_initial = false;
122        loop {
123            if timeout.is_some_and(|limit| started.elapsed() >= limit) {
124                yield ok_event(Event::default().event("error").data("SSE demo time limit exceeded"));
125                break;
126            }
127            let run_outcome = match TuneRunRow::get(&pool, run_id).await {
128                Ok(Some(run)) => {
129                    if !sent_initial && let Some(initial) = run.initial_readings.as_ref() {
130                        let response = InitialReadingsResponse::from(initial.clone());
131                        match Event::default().event("initial").json_data(response) {
132                            Ok(event) => {
133                                sent_initial = true;
134                                yield ok_event(event);
135                            }
136                            Err(err) => tracing::error!(
137                                run_id,
138                                error = %err,
139                                "failed to encode initial readings as an SSE event"
140                            ),
141                        }
142                    }
143                    Some(run.outcome)
144                }
145                // Still running (or -- defensively -- the run vanished from under us mid-
146                // stream, which cannot happen in practice since nothing ever deletes a
147                // `tune_runs` row): continue to the sample query.
148                Ok(_) => None,
149                Err(err) => {
150                    tracing::error!(
151                        run_id,
152                        error = %err,
153                        "failed to poll tune_runs for the run stream; ending the stream"
154                    );
155                    yield ok_event(Event::default().event("error").data(err.to_string()));
156                    break;
157                }
158            };
159
160            match TuneSampleRow::list_for_run_since(&pool, run_id, last_tick).await {
161                Ok(samples) => {
162                    for sample in &samples {
163                        last_tick = sample.tick_index;
164                        let response = SampleResponse::from(sample);
165                        match Event::default().event("sample").json_data(response) {
166                            Ok(event) => yield ok_event(event),
167                            Err(err) => tracing::error!(
168                                run_id,
169                                error = %err,
170                                "failed to encode a tune sample as an SSE event"
171                            ),
172                        }
173                    }
174                }
175                Err(err) => {
176                    tracing::error!(
177                        run_id,
178                        error = %err,
179                        "failed to poll tune_samples for the run stream; ending the stream"
180                    );
181                    yield ok_event(Event::default().event("error").data(err.to_string()));
182                    break;
183                }
184            }
185
186            if let Some(outcome) = run_outcome
187                && outcome != TuneOutcome::Running
188            {
189                let done = RunStreamDone { outcome };
190                if let Ok(event) = Event::default().event("done").json_data(done) {
191                    yield ok_event(event);
192                }
193                break;
194            }
195
196            tokio::time::sleep(POLL_INTERVAL).await;
197        }
198    };
199
200    Ok(Sse::new(events).keep_alive(KeepAlive::default()))
201}
202
203pub fn router() -> Router<AppState> {
204    Router::new().route("/api/runs/{id}/stream", get(stream_run))
205}
206
207#[cfg(test)]
208mod tests {
209    use super::*;
210    use axum::body::{Body, to_bytes};
211    use axum::http::{Request, StatusCode};
212    use bhtune_core::{ControllerType, LoopConfig, LoopTags, MrftState, ProcessType, Tick};
213    use bhtune_db::models::{SampleQuality, TuneDriver, TuneRunInitialReadings};
214    use chrono::Utc;
215    use tower::ServiceExt;
216
217    async fn seed_running_run(state: &AppState, name: &str) -> i64 {
218        let template_row =
219            bhtune_db::models::DcsTemplateRow::get_by_name(&state.pool, "Yokogawa CentumVP")
220                .await
221                .unwrap()
222                .unwrap();
223        let template = template_row.template;
224        let config = LoopConfig {
225            process_type: ProcessType::Flow,
226            controller_type: ControllerType::Pi,
227            relay_amp_percent: 5.0,
228            num_cycles_skip: 1,
229            num_cycles_count: 3,
230            noise_protection_secs: 0,
231            mrft_delay_secs: 0,
232        };
233        let tags = LoopTags::derive_from_pv_tag(&format!("{name}.PV"), &template);
234        let run = TuneRunRow::start(
235            &state.pool,
236            None,
237            name,
238            TuneDriver::Simulator,
239            config,
240            template_row.origin,
241            &template,
242            &tags,
243            Utc::now(),
244        )
245        .await
246        .unwrap();
247        run.id
248    }
249
250    async fn insert_sample(state: &AppState, run_id: i64, tick: i64) {
251        let sample = Tick {
252            time: Utc::now(),
253            pv: 50.0 + tick as f32,
254        };
255        let mrft_state = MrftState {
256            hysteresis: 1.0,
257            mv_value_current: 55.0,
258            mv_sign_next_step: 1,
259            counter_all_switches: tick as u32,
260            cycles_completed: 0,
261            cycles_remaining: 2,
262        };
263        TuneSampleRow::insert(
264            &state.pool,
265            run_id,
266            tick,
267            sample,
268            mrft_state,
269            SampleQuality::Good,
270        )
271        .await
272        .unwrap();
273    }
274
275    async fn record_initial_readings(state: &AppState, run_id: i64) {
276        TuneRunRow::record_initial_readings(
277            &state.pool,
278            run_id,
279            TuneRunInitialReadings {
280                pv_ini: 48.0,
281                mv_ini: 42.0,
282                mv_range_low: 0.0,
283                mv_range_high: 100.0,
284                pv_range_high: 100.0,
285                pv_range_low: 0.0,
286                controller_direction: bhtune_core::ControllerDirection::Reverse,
287                mode_raw: Some("AUTO".to_string()),
288                mode_attribute_raw: None,
289                setpoint_ini: Some(50.0),
290            },
291        )
292        .await
293        .unwrap();
294    }
295
296    /// Parses a `text/event-stream` body into `(event, data)` pairs, in order. A minimal,
297    /// deliberately non-general parser matching exactly the shape this module's own handler
298    /// produces (one `event:`/`data:` line pair per event, `\n\n`-terminated) -- good enough
299    /// for asserting on our own output without pulling in a full SSE-client crate just for
300    /// tests.
301    fn parse_sse(body: &str) -> Vec<(String, String)> {
302        body.split("\n\n")
303            .filter(|chunk| !chunk.trim().is_empty())
304            .map(|chunk| {
305                let mut event = String::new();
306                let mut data = String::new();
307                for line in chunk.lines() {
308                    if let Some(rest) = line.strip_prefix("event:") {
309                        event = rest.trim().to_string();
310                    } else if let Some(rest) = line.strip_prefix("data:") {
311                        data = rest.trim().to_string();
312                    }
313                }
314                (event, data)
315            })
316            .collect()
317    }
318
319    #[test]
320    fn sse_parser_ignores_unrecognized_lines_between_event_and_data() {
321        assert_eq!(
322            parse_sse("event: ping\nid: 42\ndata: payload\n\n"),
323            vec![("ping".to_string(), "payload".to_string())]
324        );
325    }
326
327    #[tokio::test]
328    async fn streaming_an_unknown_run_returns_404() {
329        let state = crate::test_support::in_memory_state().await;
330        let app = crate::build_router(state);
331        let response = app
332            .oneshot(
333                Request::get("/api/runs/999/stream")
334                    .body(Body::empty())
335                    .unwrap(),
336            )
337            .await
338            .unwrap();
339        assert_eq!(response.status(), StatusCode::NOT_FOUND);
340    }
341
342    #[tokio::test]
343    async fn streaming_after_its_deadline_emits_an_error_event_and_closes() {
344        let state = crate::test_support::in_memory_state().await;
345        let run_id = seed_running_run(&state, "LIC-STREAM-TIMEOUT").await;
346        let response = stream_run_with_permit(state, run_id, None, Some(Duration::ZERO))
347            .await
348            .unwrap()
349            .into_response();
350
351        let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
352        let events = parse_sse(core::str::from_utf8(&bytes).unwrap());
353
354        assert_eq!(
355            events,
356            vec![(
357                "error".to_string(),
358                "SSE demo time limit exceeded".to_string()
359            )]
360        );
361    }
362
363    #[tokio::test]
364    async fn streaming_a_completed_run_replays_every_sample_then_emits_done() {
365        let state = crate::test_support::in_memory_state().await;
366        let run_id = seed_running_run(&state, "LIC-STREAM-1").await;
367        insert_sample(&state, run_id, 0).await;
368        insert_sample(&state, run_id, 1).await;
369        TuneRunRow::complete(&state.pool, run_id, Utc::now())
370            .await
371            .unwrap();
372
373        let app = crate::build_router(state);
374        let response = app
375            .oneshot(
376                Request::get(format!("/api/runs/{run_id}/stream"))
377                    .body(Body::empty())
378                    .unwrap(),
379            )
380            .await
381            .unwrap();
382        assert_eq!(response.status(), StatusCode::OK);
383        assert_eq!(
384            response
385                .headers()
386                .get(axum::http::header::CONTENT_TYPE)
387                .unwrap(),
388            "text/event-stream"
389        );
390
391        let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
392        let body = String::from_utf8(bytes.to_vec()).unwrap();
393        let events = parse_sse(&body);
394
395        let sample_events: Vec<_> = events.iter().filter(|(e, _)| e == "sample").collect();
396        assert_eq!(
397            sample_events.len(),
398            2,
399            "both pre-recorded samples must be replayed"
400        );
401        let first: serde_json::Value = serde_json::from_str(&sample_events[0].1).unwrap();
402        assert_eq!(first["tick_index"], 0);
403        let second: serde_json::Value = serde_json::from_str(&sample_events[1].1).unwrap();
404        assert_eq!(second["tick_index"], 1);
405
406        let (last_event, last_data) = events.last().expect("stream must emit at least `done`");
407        assert_eq!(last_event, "done");
408        let done: serde_json::Value = serde_json::from_str(last_data).unwrap();
409        assert_eq!(done["outcome"], "completed");
410    }
411
412    #[tokio::test]
413    async fn streaming_emits_initial_readings_before_replayed_samples() {
414        let state = crate::test_support::in_memory_state().await;
415        let run_id = seed_running_run(&state, "LIC-STREAM-INITIAL").await;
416        record_initial_readings(&state, run_id).await;
417        insert_sample(&state, run_id, 0).await;
418        TuneRunRow::complete(&state.pool, run_id, Utc::now())
419            .await
420            .unwrap();
421
422        let app = crate::build_router(state);
423        let response = app
424            .oneshot(
425                Request::get(format!("/api/runs/{run_id}/stream"))
426                    .body(Body::empty())
427                    .unwrap(),
428            )
429            .await
430            .unwrap();
431        let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
432        let events = parse_sse(core::str::from_utf8(&bytes).unwrap());
433
434        assert_eq!(events[0].0, "initial");
435        let initial: serde_json::Value = serde_json::from_str(&events[0].1).unwrap();
436        assert_eq!(initial["pv_ini"], 48.0);
437        assert_eq!(initial["mv_ini"], 42.0);
438        assert_eq!(events[1].0, "sample");
439        assert_eq!(events[2].0, "done");
440    }
441
442    #[tokio::test]
443    async fn streaming_a_run_with_no_samples_still_terminates_with_done() {
444        let state = crate::test_support::in_memory_state().await;
445        let run_id = seed_running_run(&state, "LIC-STREAM-2").await;
446        TuneRunRow::abort(&state.pool, run_id, Utc::now())
447            .await
448            .unwrap();
449
450        let app = crate::build_router(state);
451        let response = app
452            .oneshot(
453                Request::get(format!("/api/runs/{run_id}/stream"))
454                    .body(Body::empty())
455                    .unwrap(),
456            )
457            .await
458            .unwrap();
459        let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
460        let events = parse_sse(core::str::from_utf8(&bytes).unwrap());
461        assert_eq!(
462            events.len(),
463            1,
464            "no samples were recorded, so only `done` is emitted"
465        );
466        assert_eq!(events[0].0, "done");
467        let done: serde_json::Value = serde_json::from_str(&events[0].1).unwrap();
468        assert_eq!(done["outcome"], "aborted");
469    }
470
471    #[tokio::test]
472    async fn streaming_a_still_running_run_waits_for_it_to_finish_before_closing() {
473        let state = crate::test_support::in_memory_state().await;
474        let run_id = seed_running_run(&state, "LIC-STREAM-3").await;
475        insert_sample(&state, run_id, 0).await;
476        // Left `running` -- the stream must poll past at least one `POLL_INTERVAL` sleep
477        // before observing a terminal outcome. A background task flips it to `completed`
478        // shortly after the stream has almost certainly already seen it as `running` at
479        // least once (the handler's own pre-flight `TuneRunRow::get` existence check runs
480        // synchronously before this task is even spawned, so there is no race on "does the
481        // run exist").
482        let pool = state.pool.clone();
483        tokio::spawn(async move {
484            tokio::time::sleep(Duration::from_millis(400)).await;
485            TuneRunRow::complete(&pool, run_id, Utc::now())
486                .await
487                .unwrap();
488        });
489
490        let app = crate::build_router(state);
491        let response = tokio::time::timeout(
492            Duration::from_secs(5),
493            app.oneshot(
494                Request::get(format!("/api/runs/{run_id}/stream"))
495                    .body(Body::empty())
496                    .unwrap(),
497            ),
498        )
499        .await
500        .expect("the stream must eventually close on its own")
501        .unwrap();
502
503        let bytes = tokio::time::timeout(
504            Duration::from_secs(5),
505            to_bytes(response.into_body(), usize::MAX),
506        )
507        .await
508        .expect("reading the full (finite) SSE body must not hang")
509        .unwrap();
510        let events = parse_sse(core::str::from_utf8(&bytes).unwrap());
511
512        assert!(
513            events
514                .iter()
515                .any(|(e, d)| e == "sample" && d.contains("\"tick_index\":0")),
516            "the pre-recorded sample must have been replayed: {events:?}"
517        );
518        let (last_event, last_data) = events.last().unwrap();
519        assert_eq!(last_event, "done");
520        let done: serde_json::Value = serde_json::from_str(last_data).unwrap();
521        assert_eq!(done["outcome"], "completed");
522    }
523}
bhtune_server/routes/stream.rs

bhtune_server/routes/
stream.rs
