bhtune_runtime/
history.rs

1//! Shared history operations that can affect a live controller.
2
3use bhtune_core::ResponseLevel;
4use bhtune_db::SqlitePool;
5use bhtune_db::models::{
6    LiveOperationKind, TuneDriver, TuneRunRow, TuneWriteRow, WriteKind, WriteReadback,
7};
8use bhtune_driver::OpcDaDriver;
9use thiserror::Error;
10
11use crate::{
12    live_ownership::{AuditedDriver, LiveOperationGuard},
13    tune::{PidWriteOutcome, write_pid_values_with_owner},
14};
15
16#[derive(Debug, Error)]
17pub enum RevertRunError {
18    #[error("{0}")]
19    Invalid(anyhow::Error),
20    #[error("{0}")]
21    Connection(anyhow::Error),
22    #[error("{0}")]
23    Gateway(anyhow::Error),
24    #[error(transparent)]
25    Persistence(anyhow::Error),
26}
27
28pub struct RevertRequest {
29    pub run_id: i64,
30    pub bridge_host: Option<String>,
31    pub server: Option<String>,
32    pub confirmed: bool,
33}
34
35pub struct RevertProgress<'a> {
36    pub run_id: i64,
37    pub loop_name: &'a str,
38    pub response_level: ResponseLevel,
39    pub target: WriteReadback,
40}
41
42pub struct RevertRunResult {
43    pub response_level: ResponseLevel,
44    pub target: WriteReadback,
45    pub outcome: PidWriteOutcome,
46}
47
48/// Resolves a revert connection only from the original run's recorded connection. Explicit
49/// flags are cross-checks and cannot redirect a revert to another controller or gateway.
50pub fn resolve_revert_connection(
51    run: &TuneRunRow,
52    bridge_host_flag: Option<&str>,
53    server_flag: Option<&str>,
54) -> anyhow::Result<(String, String)> {
55    let stored_server = run.opc_server.as_deref().ok_or_else(|| {
56        anyhow::anyhow!(
57            "run {}'s recorded OPC server is missing; refusing to guess which server to \
58             revert against",
59            run.id
60        )
61    })?;
62    let stored_bridge_host = run.bridge_host.as_deref().ok_or_else(|| {
63        anyhow::anyhow!(
64            "run {}'s recorded bridge host is missing; refusing to guess which gateway to \
65             revert against",
66            run.id
67        )
68    })?;
69
70    if let Some(server_flag) = server_flag
71        && server_flag != stored_server
72    {
73        anyhow::bail!(
74            "--server {server_flag:?} contradicts run {}'s recorded OPC server \
75             {stored_server:?}; refusing to revert against a different server than the run \
76             actually used -- omit --server to use the recorded value",
77            run.id
78        );
79    }
80    if let Some(bridge_host_flag) = bridge_host_flag
81        && bridge_host_flag != stored_bridge_host
82    {
83        anyhow::bail!(
84            "--bridge-host {bridge_host_flag:?} contradicts run {}'s recorded bridge host \
85             {stored_bridge_host:?}; refusing to revert through a different gateway than the \
86             run actually used -- omit --bridge-host to use the recorded value",
87            run.id
88        );
89    }
90
91    Ok((stored_bridge_host.to_string(), stored_server.to_string()))
92}
93
94/// Restores the pre-write-back PID values recorded for a run. Every eligibility, confirmation,
95/// and connection check completes before the live write sequence begins.
96pub async fn revert_run(
97    pool: &SqlitePool,
98    request: RevertRequest,
99    allow_uncertain_quality: bool,
100) -> Result<RevertRunResult, RevertRunError> {
101    revert_run_with_progress(pool, request, allow_uncertain_quality, |_| {}).await
102}
103
104/// Revert the recorded PID values and notify an adapter after all safety checks have passed,
105/// immediately before the first live write.
106pub async fn revert_run_with_progress(
107    pool: &SqlitePool,
108    request: RevertRequest,
109    allow_uncertain_quality: bool,
110    on_ready: impl FnOnce(RevertProgress<'_>),
111) -> Result<RevertRunResult, RevertRunError> {
112    let run_id = request.run_id;
113    let run = TuneRunRow::get(pool, run_id)
114        .await
115        .map_err(|error| RevertRunError::Persistence(error.into()))?
116        .ok_or_else(|| RevertRunError::Invalid(anyhow::anyhow!("no run with id {run_id}")))?;
117
118    if run.driver != TuneDriver::Opcda {
119        return Err(RevertRunError::Invalid(anyhow::anyhow!(
120            "run {run_id} used the {:?} driver, which has no live loop to revert a write \
121             against",
122            run.driver
123        )));
124    }
125
126    let writes = TuneWriteRow::list_for_run(pool, run_id)
127        .await
128        .map_err(|error| RevertRunError::Persistence(error.into()))?;
129    let last_write = writes
130        .iter()
131        .rev()
132        .find(|write| write.kind == WriteKind::Write)
133        .ok_or_else(|| {
134            RevertRunError::Invalid(anyhow::anyhow!(
135                "run {run_id} has no recorded PID write-back to revert"
136            ))
137        })?;
138    let response_level = last_write.response_level;
139    let target = last_write.previous.ok_or_else(|| {
140        RevertRunError::Invalid(anyhow::anyhow!(
141            "run {run_id}'s {response_level:?} PID write-back never recorded pre-write \
142             values (its pre-read failed at the time); nothing to revert to"
143        ))
144    })?;
145
146    if !request.confirmed {
147        return Err(RevertRunError::Invalid(anyhow::anyhow!(
148            "reverting writes PID constants back to a live loop; pass --yes to confirm"
149        )));
150    }
151
152    let (Some(p_tag), Some(i_tag), Some(d_tag)) = (
153        &run.tags.proportional_constant,
154        &run.tags.integral_constant,
155        &run.tags.derivative_constant,
156    ) else {
157        return Err(RevertRunError::Invalid(anyhow::anyhow!(
158            "run {run_id}'s snapshotted tags have no PID constant tags configured"
159        )));
160    };
161
162    let (bridge_host, server) = resolve_revert_connection(
163        &run,
164        request.bridge_host.as_deref(),
165        request.server.as_deref(),
166    )
167    .map_err(RevertRunError::Invalid)?;
168    let restore_intent = serde_json::to_string(&serde_json::json!({
169        "version": 1,
170        "kind": "pid_restore",
171        "run_id": run_id,
172        "write_kind": WriteKind::Revert,
173        "state": "awaiting_prewrite_readings",
174    }))
175    .map_err(|error| RevertRunError::Invalid(anyhow::Error::new(error)))?;
176    let ownership = LiveOperationGuard::acquire_for_recorded_run(
177        pool,
178        &run,
179        LiveOperationKind::PidRevert,
180        restore_intent,
181    )
182    .await
183    .map_err(|error| RevertRunError::Persistence(error.into()))?;
184    let operation_result = async {
185        ownership
186            .ensure_healthy()
187            .map_err(RevertRunError::Persistence)?;
188        let driver = OpcDaDriver::connect(&bridge_host, &server)
189            .await
190            .map_err(|error| RevertRunError::Connection(anyhow::Error::new(error)))?;
191        let compatibility =
192            crate::gateway::require_live_gateway_compatible(&bridge_host, Some(&server))
193                .await
194                .map_err(RevertRunError::Gateway)?;
195        record_gateway_compatibility_if_missing(pool, &run, &compatibility).await?;
196
197        on_ready(RevertProgress {
198            run_id,
199            loop_name: &run.loop_name,
200            response_level,
201            target,
202        });
203
204        let audited = AuditedDriver::new(&driver, pool, &ownership, Some(run_id));
205        let outcome = write_pid_values_with_owner(
206            pool,
207            run_id,
208            &audited,
209            p_tag,
210            i_tag,
211            d_tag,
212            response_level,
213            target,
214            WriteKind::Revert,
215            allow_uncertain_quality,
216        )
217        .await
218        .map_err(RevertRunError::Persistence)?;
219        Ok::<_, RevertRunError>(RevertRunResult {
220            response_level,
221            target,
222            outcome,
223        })
224    }
225    .await;
226    let release_result = ownership.release().await;
227    combine_revert_and_release(operation_result, release_result)
228}
229
230fn combine_revert_and_release<T>(
231    operation_result: Result<T, RevertRunError>,
232    release_result: anyhow::Result<()>,
233) -> Result<T, RevertRunError> {
234    match (operation_result, release_result) {
235        (Ok(result), Ok(())) => Ok(result),
236        (Err(error), Ok(())) => Err(error),
237        (Ok(_), Err(error)) => Err(RevertRunError::Persistence(error)),
238        (Err(error), Err(release_error)) => Err(RevertRunError::Persistence(anyhow::anyhow!(
239            "{error}; additionally failed to persist live ownership release: {release_error}"
240        ))),
241    }
242}
243
244async fn record_gateway_compatibility_if_missing(
245    pool: &SqlitePool,
246    run: &TuneRunRow,
247    compatibility: &bhtune_driver::OpcDaGatewayCompatibility,
248) -> Result<(), RevertRunError> {
249    if run.gateway_compatibility_json.is_none() {
250        let snapshot = crate::gateway::compatibility_json(compatibility);
251        TuneRunRow::record_gateway_compatibility(pool, run.id, &snapshot)
252            .await
253            .map_err(|error| RevertRunError::Persistence(error.into()))?;
254    }
255    Ok(())
256}
257
258#[cfg(test)]
259mod tests {
260    use super::*;
261    use bhtune_core::{ControllerType, LoopConfig, ProcessType};
262    use bhtune_db::models::{TemplateOrigin, TuneRunRow};
263
264    fn recorded_run() -> TuneRunRow {
265        let template = bhtune_core::built_in_templates().remove(0);
266        let tags = bhtune_core::LoopTags::derive_from_pv_tag("Unit1.LIC101.PV", &template);
267        TuneRunRow {
268            id: 7,
269            loop_id: None,
270            demo_session_id: None,
271            loop_name: "Unit1.LIC101.PV".to_string(),
272            driver: TuneDriver::Opcda,
273            opc_server: Some("Kepware.KEPServerEX.V6".to_string()),
274            bridge_host: Some("127.0.0.1:7600".to_string()),
275            started_at: chrono::Utc::now(),
276            completed_at: None,
277            outcome: bhtune_db::models::TuneOutcome::Running,
278            failure_reason: None,
279            config: bhtune_core::LoopConfig {
280                process_type: bhtune_core::ProcessType::Flow,
281                controller_type: bhtune_core::ControllerType::Pi,
282                relay_amp_percent: 10.0,
283                num_cycles_skip: 1,
284                num_cycles_count: 2,
285                noise_protection_secs: 3,
286                mrft_delay_secs: 0,
287            },
288            template_origin: bhtune_db::models::TemplateOrigin::Builtin,
289            tags,
290            template,
291            request_json: "{}".to_string(),
292            notes: None,
293            initial_readings: None,
294            allow_uncertain_quality: false,
295            timing_metrics: None,
296            effective_tuning: None,
297            gateway_compatibility_json: None,
298            restore_status: None,
299            restore_detail: None,
300            recovery_state: None,
301            recovery_evidence_json: None,
302            created_at: chrono::Utc::now(),
303        }
304    }
305
306    #[test]
307    fn revert_and_release_results_preserve_both_failures() {
308        assert!(combine_revert_and_release::<()>(Ok(()), Ok(())).is_ok());
309
310        let operation_only = combine_revert_and_release::<()>(
311            Err(RevertRunError::Invalid(anyhow::anyhow!("operation failed"))),
312            Ok(()),
313        )
314        .unwrap_err();
315        assert!(operation_only.to_string().contains("operation failed"));
316
317        let release_only =
318            combine_revert_and_release::<()>(Ok(()), Err(anyhow::anyhow!("release failed")))
319                .unwrap_err();
320        assert!(release_only.to_string().contains("release failed"));
321
322        let both = combine_revert_and_release::<()>(
323            Err(RevertRunError::Invalid(anyhow::anyhow!("operation failed"))),
324            Err(anyhow::anyhow!("release failed")),
325        )
326        .unwrap_err();
327        assert!(both.to_string().contains("operation failed"));
328        assert!(both.to_string().contains("release failed"));
329    }
330
331    #[test]
332    fn revert_connection_uses_the_recorded_values() {
333        let run = recorded_run();
334
335        assert_eq!(
336            resolve_revert_connection(&run, None, None).unwrap(),
337            (
338                "127.0.0.1:7600".to_string(),
339                "Kepware.KEPServerEX.V6".to_string()
340            )
341        );
342        assert!(
343            resolve_revert_connection(&run, Some("127.0.0.1:7600"), Some("Kepware.KEPServerEX.V6"))
344                .is_ok()
345        );
346    }
347
348    #[test]
349    fn revert_connection_rejects_missing_or_contradictory_values() {
350        let mut run = recorded_run();
351        run.bridge_host = None;
352        assert!(
353            resolve_revert_connection(&run, None, None)
354                .unwrap_err()
355                .to_string()
356                .contains("recorded bridge host is missing")
357        );
358
359        let run = recorded_run();
360        assert!(
361            resolve_revert_connection(&run, None, Some("Different.Server"))
362                .unwrap_err()
363                .to_string()
364                .contains("contradicts")
365        );
366        assert!(
367            resolve_revert_connection(&run, Some("different:7600"), None)
368                .unwrap_err()
369                .to_string()
370                .contains("contradicts")
371        );
372    }
373
374    #[tokio::test]
375    async fn gateway_compatibility_snapshot_is_recorded_only_when_missing() {
376        let pool = bhtune_db::connect_in_memory().await.unwrap();
377        bhtune_db::seed_builtin_templates(&pool, chrono::Utc::now())
378            .await
379            .unwrap();
380        let template = bhtune_core::built_in_templates().remove(0);
381        let tags = bhtune_core::LoopTags::derive_from_pv_tag("Unit1.LIC101.PV", &template);
382        let config = LoopConfig {
383            process_type: ProcessType::Flow,
384            controller_type: ControllerType::Pi,
385            relay_amp_percent: 10.0,
386            num_cycles_skip: 1,
387            num_cycles_count: 2,
388            noise_protection_secs: 3,
389            mrft_delay_secs: 0,
390        };
391        let run = TuneRunRow::start(
392            &pool,
393            None,
394            "Unit1.LIC101.PV",
395            TuneDriver::Opcda,
396            config,
397            TemplateOrigin::Builtin,
398            &template,
399            &tags,
400            chrono::Utc::now(),
401        )
402        .await
403        .unwrap();
404        let compatibility = bhtune_driver::OpcDaGatewayCompatibility::unverified();
405        record_gateway_compatibility_if_missing(&pool, &run, &compatibility)
406            .await
407            .unwrap();
408
409        let expected = crate::gateway::compatibility_json(&compatibility);
410        let saved = TuneRunRow::get(&pool, run.id).await.unwrap().unwrap();
411        assert_eq!(
412            saved.gateway_compatibility_json.as_deref(),
413            Some(expected.as_str())
414        );
415
416        let existing = r#"{"status":"full","source":"original"}"#;
417        TuneRunRow::record_gateway_compatibility(&pool, run.id, existing)
418            .await
419            .unwrap();
420        let saved = TuneRunRow::get(&pool, run.id).await.unwrap().unwrap();
421        record_gateway_compatibility_if_missing(&pool, &saved, &compatibility)
422            .await
423            .unwrap();
424        let saved = TuneRunRow::get(&pool, run.id).await.unwrap().unwrap();
425        assert_eq!(saved.gateway_compatibility_json.as_deref(), Some(existing));
426    }
427}
bhtune_runtime/history.rs

bhtune_runtime/
history.rs
