bhtune_runtime/tune/
actuation.rs

1#![allow(rustdoc::broken_intra_doc_links)]
2
3use std::collections::HashMap;
4use std::time::Duration;
5
6use bhtune_core::LoopConfig;
7use bhtune_core::mrft::clamp_relay_amplitude;
8use bhtune_db::SqlitePool;
9use bhtune_db::models::{
10    MvActuationKind, MvActuationStatus, NewTuneMvActuation, SampleQuality, TuneMvActuationRow,
11};
12use bhtune_driver::{Driver, TagValue};
13use chrono::{DateTime, Utc};
14use tokio::time::Instant;
15
16use super::request::{DriverKind, TuneRequest};
17use crate::cancel::CtrlC;
18use crate::timing::PollTimingAccumulator;
19
20use super::RequireInvariant;
21use super::config::EffectiveTiming;
22#[cfg(test)]
23use super::config::test_effective_timing;
24use super::outcome::AbortReason;
25use super::poll::{TickOperation, bounded_driver_call};
26use super::prepare::InitialState;
27use super::quality::{
28    check_quality, read_numeric_from_batch, read_numeric_sample, sample_quality_from_driver,
29};
30
31/// Maximum interval from an accepted OPC DA MV write to its mandatory confirmation check.
32///
33/// Public so HTTP and other non-clap adapters can expose the same validation policy without
34/// copying the safety-critical literal.
35pub const MV_ACTUATION_CONFIRMATION_SECS: u64 = 4;
36pub(super) const MV_ACTUATION_RETRY_INTERVAL: Duration = Duration::from_millis(100);
37pub(super) const MV_ACTUATION_DEADLINE_READ_MAX: Duration = Duration::from_secs(1);
38pub(super) const MV_ACTUATION_FALLBACK_HEADROOM: Duration = Duration::from_secs(1);
39pub(super) const MV_RESTORE_HANDOFF_READ_MAX: Duration = Duration::from_secs(1);
40pub(super) const MV_SPAN_TOLERANCE_FRACTION: f32 = 0.001;
41pub(super) const RELAY_STEP_TOLERANCE_FRACTION: f32 = 0.25;
42pub(super) const MIN_RELAY_STEP: f32 = 0.01;
43#[derive(Debug)]
44pub(super) struct PendingMvActuation {
45    pub(super) id: Option<i64>,
46    pub(super) kind: MvActuationKind,
47    pub(super) target: f32,
48    pub(super) tolerance: f32,
49    pub(super) switch_tick: DateTime<Utc>,
50    pub(super) switch_instant: Instant,
51    pub(super) accepted_instant: Instant,
52    pub(super) first_check_at: Instant,
53    pub(super) deadline: Instant,
54    pub(super) last_readback: Option<f32>,
55}
56#[derive(Debug)]
57pub(super) struct MvActuationTracker {
58    pub(super) next_sequence: i64,
59    pub(super) previous_commanded_mv: f32,
60    pub(super) confirmed_mv: Option<f32>,
61    pub(super) pending: Option<PendingMvActuation>,
62    pub(super) mv_span: f32,
63}
64impl MvActuationTracker {
65    pub(super) fn for_run(args: &TuneRequest, initial: &InitialState) -> Option<Self> {
66        (args.driver == DriverKind::Opcda).then_some(Self {
67            next_sequence: 0,
68            previous_commanded_mv: initial.mv_ini,
69            confirmed_mv: None,
70            pending: None,
71            mv_span: initial.mv_range_high - initial.mv_range_low,
72        })
73    }
74
75    #[cfg(test)]
76    #[allow(clippy::too_many_arguments)]
77    pub(super) async fn record_accepted(
78        &mut self,
79        pool: &SqlitePool,
80        run_id: i64,
81        kind: MvActuationKind,
82        target: f32,
83        first_check_at: Instant,
84        accepted_at: DateTime<Utc>,
85        accepted_instant: Instant,
86        tolerance: f32,
87    ) -> anyhow::Result<()> {
88        self.record_accepted_at_switch(
89            pool,
90            run_id,
91            kind,
92            target,
93            accepted_at,
94            accepted_instant,
95            first_check_at,
96            accepted_at,
97            accepted_instant,
98            tolerance,
99        )
100        .await
101    }
102
103    #[allow(clippy::too_many_arguments)]
104    pub(super) async fn record_accepted_at_switch(
105        &mut self,
106        pool: &SqlitePool,
107        run_id: i64,
108        kind: MvActuationKind,
109        target: f32,
110        switch_tick: DateTime<Utc>,
111        switch_instant: Instant,
112        first_check_at: Instant,
113        accepted_at: DateTime<Utc>,
114        accepted_instant: Instant,
115        tolerance: f32,
116    ) -> anyhow::Result<()> {
117        let deadline = accepted_instant + Duration::from_secs(MV_ACTUATION_CONFIRMATION_SECS);
118        let confirmation_due_at =
119            accepted_at + chrono::Duration::seconds(MV_ACTUATION_CONFIRMATION_SECS as i64);
120        let previous_commanded_mv = Some(self.previous_commanded_mv);
121        let row = TuneMvActuationRow::insert_pending(
122            pool,
123            run_id,
124            NewTuneMvActuation {
125                sequence: self.next_sequence,
126                kind,
127                commanded_at: accepted_at,
128                target_mv: target,
129                previous_commanded_mv,
130                tolerance,
131                confirmation_due_at,
132            },
133        )
134        .await?;
135        self.accept_pending(PendingMvActuation {
136            id: Some(row.id),
137            kind,
138            target,
139            tolerance,
140            switch_tick,
141            switch_instant,
142            accepted_instant,
143            first_check_at: first_check_at.min(deadline),
144            deadline,
145            last_readback: None,
146        });
147        Ok(())
148    }
149
150    #[allow(clippy::too_many_arguments)]
151    pub(super) async fn record_restore_accepted_best_effort(
152        &mut self,
153        pool: &SqlitePool,
154        run_id: i64,
155        target: f32,
156        accepted_at: DateTime<Utc>,
157        accepted_instant: Instant,
158        tolerance: f32,
159    ) {
160        let deadline = accepted_instant + Duration::from_secs(MV_ACTUATION_CONFIRMATION_SECS);
161        let confirmation_due_at =
162            accepted_at + chrono::Duration::seconds(MV_ACTUATION_CONFIRMATION_SECS as i64);
163        let pending = PendingMvActuation {
164            id: None,
165            kind: MvActuationKind::Restore,
166            target,
167            tolerance,
168            switch_tick: accepted_at,
169            switch_instant: accepted_instant,
170            accepted_instant,
171            first_check_at: accepted_instant,
172            deadline,
173            last_readback: None,
174        };
175        let row = TuneMvActuationRow::insert_pending(
176            pool,
177            run_id,
178            NewTuneMvActuation {
179                sequence: self.next_sequence,
180                kind: MvActuationKind::Restore,
181                commanded_at: accepted_at,
182                target_mv: target,
183                previous_commanded_mv: Some(self.previous_commanded_mv),
184                tolerance,
185                confirmation_due_at,
186            },
187        )
188        .await;
189        let mut pending = pending;
190        match row {
191            Ok(row) => pending.id = Some(row.id),
192            Err(error) => {
193                tracing::error!(
194                    run_id,
195                    error = %error,
196                    "failed to record accepted restore MV command; continuing physical restore"
197                );
198            }
199        }
200        self.accept_pending(pending);
201    }
202
203    pub(super) async fn record_restore_accepted_required(
204        &mut self,
205        pool: &SqlitePool,
206        run_id: i64,
207        target: f32,
208        accepted_at: DateTime<Utc>,
209        accepted_instant: Instant,
210        tolerance: f32,
211    ) -> anyhow::Result<()> {
212        let confirmation_due_at =
213            accepted_at + chrono::Duration::seconds(MV_ACTUATION_CONFIRMATION_SECS as i64);
214        let row = TuneMvActuationRow::insert_pending(
215            pool,
216            run_id,
217            NewTuneMvActuation {
218                sequence: self.next_sequence,
219                kind: MvActuationKind::Restore,
220                commanded_at: accepted_at,
221                target_mv: target,
222                previous_commanded_mv: Some(self.previous_commanded_mv),
223                tolerance,
224                confirmation_due_at,
225            },
226        )
227        .await?;
228        let deadline = accepted_instant + Duration::from_secs(MV_ACTUATION_CONFIRMATION_SECS);
229        self.accept_pending(PendingMvActuation {
230            id: Some(row.id),
231            kind: MvActuationKind::Restore,
232            target,
233            tolerance,
234            switch_tick: accepted_at,
235            switch_instant: accepted_instant,
236            accepted_instant,
237            first_check_at: accepted_instant,
238            deadline,
239            last_readback: None,
240        });
241        Ok(())
242    }
243
244    pub(super) fn accept_pending(&mut self, pending: PendingMvActuation) {
245        self.next_sequence += 1;
246        self.previous_commanded_mv = pending.target;
247        self.confirmed_mv = None;
248        self.pending = Some(pending);
249    }
250
251    pub(super) fn next_verification_wakeup(&self) -> Option<Instant> {
252        self.pending.as_ref().map(|pending| {
253            if pending.last_readback.is_some() {
254                pending.deadline
255            } else {
256                pending
257                    .first_check_at
258                    .max(pending.deadline - MV_ACTUATION_FALLBACK_HEADROOM)
259            }
260        })
261    }
262}
263#[allow(clippy::too_many_arguments)]
264pub(super) async fn record_relay_actuation(
265    tracker: &mut MvActuationTracker,
266    pool: &SqlitePool,
267    run_id: i64,
268    target: f32,
269    switch_tick: DateTime<Utc>,
270    switch_instant: Instant,
271    first_check_at: Instant,
272    elapsed_since_observation: Duration,
273    accepted_instant: Instant,
274    tolerance: f32,
275) -> anyhow::Result<()> {
276    let accepted_at = utc_after_elapsed(switch_tick, elapsed_since_observation)?;
277    tracker
278        .record_accepted_at_switch(
279            pool,
280            run_id,
281            MvActuationKind::Relay,
282            target,
283            switch_tick,
284            switch_instant,
285            first_check_at,
286            accepted_at,
287            accepted_instant,
288            tolerance,
289        )
290        .await
291}
292pub(super) fn f32_precision_floor(target: f32, previous: f32) -> f32 {
293    4.0 * f32::EPSILON * target.abs().max(previous.abs()).max(1.0)
294}
295pub(super) fn mv_actuation_tolerance(
296    kind: MvActuationKind,
297    target: f32,
298    previous: f32,
299    mv_span: f32,
300) -> anyhow::Result<f32> {
301    let uncapped = mv_actuation_uncapped_tolerance(target, previous, mv_span);
302    if kind == MvActuationKind::Restore {
303        return Ok(uncapped);
304    }
305
306    let step = (target - previous).abs();
307    let relay_cap = step * RELAY_STEP_TOLERANCE_FRACTION;
308    if !step.is_finite()
309        || step < MIN_RELAY_STEP
310        || relay_cap <= f32_precision_floor(target, previous)
311    {
312        let minimum_step = MIN_RELAY_STEP
313            .max(f32_precision_floor(target, previous) / RELAY_STEP_TOLERANCE_FRACTION);
314        anyhow::bail!(
315            "the effective relay step {step} is too small to verify safely (minimum {minimum_step})"
316        );
317    }
318    Ok(uncapped.min(relay_cap))
319}
320pub(super) fn mv_actuation_uncapped_tolerance(target: f32, previous: f32, mv_span: f32) -> f32 {
321    let precision_floor = f32_precision_floor(target, previous);
322    let span_tolerance = mv_span.abs() * MV_SPAN_TOLERANCE_FRACTION;
323    precision_floor + span_tolerance
324}
325pub(super) fn validate_relay_actuation_step(
326    args: &TuneRequest,
327    config: LoopConfig,
328    initial: &InitialState,
329) -> anyhow::Result<()> {
330    if args.driver != DriverKind::Opcda {
331        return Ok(());
332    }
333    let relay_step = clamp_relay_amplitude(
334        config.relay_amp_percent,
335        initial.mv_ini,
336        initial.mv_range_low,
337        initial.mv_range_high,
338    );
339    mv_actuation_tolerance(
340        MvActuationKind::Relay,
341        initial.mv_ini + relay_step,
342        initial.mv_ini,
343        initial.mv_range_high - initial.mv_range_low,
344    )
345    .map(|_| ())
346}
347pub(super) fn actuation_abort_reason(
348    tag: &str,
349    pending: &PendingMvActuation,
350    readback: Option<f32>,
351    now: Instant,
352) -> AbortReason {
353    let elapsed_ms = now
354        .saturating_duration_since(pending.accepted_instant)
355        .as_millis()
356        .min(u128::from(u64::MAX)) as u64;
357    AbortReason::MvActuationUnconfirmed {
358        tag: tag.to_string(),
359        target: pending.target,
360        readback,
361        tolerance: pending.tolerance,
362        elapsed_ms,
363        deadline_secs: MV_ACTUATION_CONFIRMATION_SECS,
364    }
365}
366pub(super) fn actuation_matches(target: f32, readback: f32, tolerance: f32) -> bool {
367    (target - readback).abs() <= tolerance
368}
369#[derive(Debug, Clone, Copy, PartialEq, Eq)]
370pub(super) enum MvVerificationTrigger {
371    Scheduled,
372    Deadline,
373}
374#[derive(Debug, Clone, Copy, PartialEq, Eq)]
375pub(super) enum ActuationAuditPolicy {
376    Required,
377    BestEffort,
378}
379#[derive(Debug, Clone, Copy)]
380pub(super) enum MvVerificationCallLimit {
381    None,
382    Restore(Instant),
383}
384#[derive(Debug, Clone, Copy, PartialEq, Eq)]
385pub(super) enum MvVerificationLimitKind {
386    Confirmation,
387    Deadline,
388    Restore,
389}
390pub(super) async fn record_actuation_observation(
391    pool: &SqlitePool,
392    pending: &PendingMvActuation,
393    checked_at: DateTime<Utc>,
394    readback: Option<f32>,
395    quality: Option<SampleQuality>,
396    policy: ActuationAuditPolicy,
397) -> anyhow::Result<Option<i64>> {
398    let Some(id) = pending.id else {
399        return Ok(None);
400    };
401    match TuneMvActuationRow::record_observation(pool, id, checked_at, readback, quality).await {
402        Ok(row) => Ok(Some(row.attempt_count)),
403        Err(error) if policy == ActuationAuditPolicy::BestEffort => {
404            tracing::error!(
405                actuation_id = id,
406                error = %error,
407                "failed to record MV verification observation; continuing physical restore"
408            );
409            Ok(None)
410        }
411        Err(error) => Err(error.into()),
412    }
413}
414#[allow(clippy::too_many_arguments)]
415pub(super) async fn record_final_actuation_observation(
416    pool: &SqlitePool,
417    pending: &PendingMvActuation,
418    checked_at: DateTime<Utc>,
419    readback: Option<f32>,
420    quality: Option<SampleQuality>,
421    status: MvActuationStatus,
422    detail: &str,
423) -> Option<i64> {
424    let id = pending.id?;
425    match TuneMvActuationRow::record_final_observation(
426        pool,
427        id,
428        checked_at,
429        readback,
430        quality,
431        status,
432        (!detail.is_empty()).then_some(detail),
433    )
434    .await
435    {
436        Ok(row) => Some(row.attempt_count),
437        Err(error) => {
438            tracing::error!(
439                actuation_id = id,
440                error = %error,
441                "failed to record terminal MV verification observation"
442            );
443            None
444        }
445    }
446}
447pub(super) async fn finalize_actuation_best_effort(
448    pool: &SqlitePool,
449    pending: &PendingMvActuation,
450    status: MvActuationStatus,
451    detail: &str,
452) {
453    let Some(id) = pending.id else {
454        return;
455    };
456    if let Err(error) = TuneMvActuationRow::finalize(pool, id, status, Some(detail)).await {
457        tracing::error!(
458            actuation_id = id,
459            error = %error,
460            "failed to finalize MV actuation audit row"
461        );
462    }
463}
464pub(super) async fn reject_replacement_for_pending_actuation(
465    pool: &SqlitePool,
466    tag: &str,
467    tracker: &mut MvActuationTracker,
468) -> anyhow::Result<AbortReason> {
469    let pending = tracker
470        .pending
471        .take()
472        .require_invariant("called only when an MV actuation is pending")?;
473    let (status, detail) = if pending.last_readback.is_some() {
474        (
475            MvActuationStatus::Failed,
476            "a replacement relay command was requested while the prior MV readback remained outside tolerance",
477        )
478    } else {
479        (
480            MvActuationStatus::Unverified,
481            "a replacement relay command was requested before an acceptable prior MV readback was available",
482        )
483    };
484    finalize_actuation_best_effort(pool, &pending, status, detail).await;
485    Ok(actuation_abort_reason(
486        tag,
487        &pending,
488        pending.last_readback,
489        Instant::now(),
490    ))
491}
492pub(super) fn verification_trigger(
493    pending: &PendingMvActuation,
494    now: Instant,
495) -> Option<MvVerificationTrigger> {
496    if now >= pending.deadline {
497        Some(MvVerificationTrigger::Deadline)
498    } else if pending.last_readback.is_none() && now >= pending.first_check_at {
499        Some(MvVerificationTrigger::Scheduled)
500    } else {
501        None
502    }
503}
504pub(super) async fn wait_for_mv_verification(wakeup: Option<Instant>) {
505    match wakeup {
506        Some(wakeup) => tokio::time::sleep_until(wakeup).await,
507        None => std::future::pending::<()>().await,
508    }
509}
510pub(super) fn mv_verification_read_limit(
511    trigger: MvVerificationTrigger,
512    pending: &PendingMvActuation,
513    call_limit: MvVerificationCallLimit,
514) -> (Instant, MvVerificationLimitKind) {
515    let external = match call_limit {
516        MvVerificationCallLimit::None => None,
517        MvVerificationCallLimit::Restore(deadline) => {
518            Some((deadline, MvVerificationLimitKind::Restore))
519        }
520    };
521    if trigger == MvVerificationTrigger::Deadline {
522        let deadline_read_limit = (
523            Instant::now() + MV_ACTUATION_DEADLINE_READ_MAX,
524            MvVerificationLimitKind::Deadline,
525        );
526        return match external {
527            Some(external) if external.0 < deadline_read_limit.0 => external,
528            _ => deadline_read_limit,
529        };
530    }
531    match external {
532        Some(external @ (deadline, _)) if deadline < pending.deadline => external,
533        _ => (pending.deadline, MvVerificationLimitKind::Confirmation),
534    }
535}
536/// Checks the accepted OPC DA MV command without producing a tune sample. Transport failures
537/// remain ordinary failed-run errors, an individual operation timeout remains
538/// [`AbortReason::OperationTimedOut`], and rejected quality remains [`AbortReason::PoorQuality`].
539/// Only a finite, acceptable-quality mismatch can become
540/// [`AbortReason::MvActuationUnconfirmed`]. Successful verification-read duration is optionally
541/// recorded when this is called from the polling loop.
542#[allow(clippy::too_many_arguments)]
543pub(super) async fn verify_pending_mv_actuation_with_timing(
544    pool: &SqlitePool,
545    _args: &TuneRequest,
546    effective_timing: EffectiveTiming,
547    tag: &str,
548    driver: &dyn Driver,
549    ctrl_c: &mut CtrlC,
550    allow_uncertain_quality: bool,
551    tracker: &mut MvActuationTracker,
552    trigger: MvVerificationTrigger,
553    call_limit: MvVerificationCallLimit,
554    audit_policy: ActuationAuditPolicy,
555    mut timing: Option<&mut PollTimingAccumulator>,
556) -> anyhow::Result<Option<AbortReason>> {
557    let Some(pending) = tracker.pending.as_ref() else {
558        return Ok(None);
559    };
560    let now = Instant::now();
561    if trigger == MvVerificationTrigger::Scheduled && now < pending.first_check_at {
562        return Ok(None);
563    }
564
565    match read_pending_mv_verification_with_timing(
566        effective_timing,
567        tag,
568        driver,
569        ctrl_c,
570        tracker,
571        trigger,
572        call_limit,
573    )
574    .await?
575    {
576        PendingMvVerificationRead::DeadlineTimedOut {
577            pending,
578            checked_at,
579            checked_instant,
580        } => {
581            record_final_actuation_observation(
582                pool,
583                &pending,
584                checked_at,
585                pending.last_readback,
586                None,
587                MvActuationStatus::Unverified,
588                "the fresh MV read at the confirmation deadline did not finish within its bounded verification window",
589            )
590            .await;
591            Ok(Some(actuation_abort_reason(
592                tag,
593                &pending,
594                pending.last_readback,
595                checked_instant,
596            )))
597        }
598        PendingMvVerificationRead::RestoreTimedOut {
599            pending,
600            checked_at,
601            checked_instant,
602        } => {
603            record_final_actuation_observation(
604                pool,
605                &pending,
606                checked_at,
607                pending.last_readback,
608                None,
609                MvActuationStatus::Unverified,
610                "the restore timeout elapsed before MV confirmation completed",
611            )
612            .await;
613            Ok(Some(actuation_abort_reason(
614                tag,
615                &pending,
616                pending.last_readback,
617                checked_instant,
618            )))
619        }
620        PendingMvVerificationRead::Ready {
621            operation,
622            checked_at,
623            checked_instant,
624            read_duration,
625        } => match resolve_pending_mv_read(
626            pool,
627            effective_timing,
628            tag,
629            tracker,
630            operation,
631            checked_at,
632            checked_instant,
633            allow_uncertain_quality,
634        )
635        .await?
636        {
637            PendingMvVerificationResult::Abort(reason) => Ok(Some(reason)),
638            PendingMvVerificationResult::Value(value) => {
639                if let Some(timing) = timing.as_mut() {
640                    timing.observe_mv_verification(read_duration);
641                }
642                finalize_pending_mv_verification(pool, tag, tracker, value, trigger, audit_policy)
643                    .await
644            }
645        },
646    }
647}
648#[cfg(test)]
649#[allow(clippy::too_many_arguments)]
650pub(super) async fn verify_pending_mv_actuation_with(
651    pool: &SqlitePool,
652    args: &TuneRequest,
653    tag: &str,
654    driver: &dyn Driver,
655    ctrl_c: &mut CtrlC,
656    allow_uncertain_quality: bool,
657    tracker: &mut MvActuationTracker,
658    trigger: MvVerificationTrigger,
659    call_limit: MvVerificationCallLimit,
660    audit_policy: ActuationAuditPolicy,
661) -> anyhow::Result<Option<AbortReason>> {
662    verify_pending_mv_actuation_with_timing(
663        pool,
664        args,
665        test_effective_timing(args),
666        tag,
667        driver,
668        ctrl_c,
669        allow_uncertain_quality,
670        tracker,
671        trigger,
672        call_limit,
673        audit_policy,
674        None,
675    )
676    .await
677}
678pub(super) enum PendingMvVerificationRead {
679    Ready {
680        operation: anyhow::Result<TickOperation<(f32, bhtune_driver::Quality)>>,
681        checked_at: DateTime<Utc>,
682        checked_instant: Instant,
683        read_duration: Duration,
684    },
685    DeadlineTimedOut {
686        pending: PendingMvActuation,
687        checked_at: DateTime<Utc>,
688        checked_instant: Instant,
689    },
690    RestoreTimedOut {
691        pending: PendingMvActuation,
692        checked_at: DateTime<Utc>,
693        checked_instant: Instant,
694    },
695}
696pub(super) struct PendingMvVerificationValue {
697    pub(super) readback: f32,
698    pub(super) sample_quality: SampleQuality,
699    pub(super) checked_at: DateTime<Utc>,
700    pub(super) checked_instant: Instant,
701}
702pub(super) enum PendingMvVerificationResult {
703    Value(PendingMvVerificationValue),
704    Abort(AbortReason),
705}
706pub(super) fn pending_verification_ready(
707    tracker: &MvActuationTracker,
708    operation: anyhow::Result<TickOperation<(f32, bhtune_driver::Quality)>>,
709    read_duration: Duration,
710) -> anyhow::Result<PendingMvVerificationRead> {
711    let checked_instant = Instant::now();
712    let pending = tracker
713        .pending
714        .as_ref()
715        .require_invariant("pending actuation existed after the bounded read")?;
716    Ok(PendingMvVerificationRead::Ready {
717        operation,
718        checked_at: checked_at_for_pending(pending, checked_instant)?,
719        checked_instant,
720        read_duration,
721    })
722}
723#[allow(clippy::too_many_arguments)]
724pub(super) async fn read_pending_mv_verification_with_timing(
725    effective_timing: EffectiveTiming,
726    tag: &str,
727    driver: &dyn Driver,
728    ctrl_c: &mut CtrlC,
729    tracker: &mut MvActuationTracker,
730    mut trigger: MvVerificationTrigger,
731    call_limit: MvVerificationCallLimit,
732) -> anyhow::Result<PendingMvVerificationRead> {
733    loop {
734        let limit = {
735            let pending = tracker
736                .pending
737                .as_ref()
738                .require_invariant("pending actuation existed before the bounded read")?;
739            mv_verification_read_limit(trigger, pending, call_limit)
740        };
741        let read_started = Instant::now();
742        let read = bounded_driver_call(
743            effective_timing.op_timeout_secs,
744            ctrl_c,
745            read_numeric_sample(driver, tag),
746        );
747        let (deadline, limit_kind) = limit;
748        match tokio::time::timeout_at(deadline, read).await {
749            Ok(operation) => {
750                return pending_verification_ready(tracker, operation, read_started.elapsed());
751            }
752            Err(_) => match limit_kind {
753                MvVerificationLimitKind::Confirmation => {
754                    trigger = MvVerificationTrigger::Deadline;
755                }
756
757                MvVerificationLimitKind::Deadline => {
758                    const MSG: &str = "pending actuation existed before the deadline read";
759                    let pending = tracker.pending.take().require_invariant(MSG)?;
760                    let checked_instant = Instant::now();
761                    let checked_at = checked_at_for_pending(&pending, checked_instant)?;
762                    return Ok(PendingMvVerificationRead::DeadlineTimedOut {
763                        pending,
764                        checked_at,
765                        checked_instant,
766                    });
767                }
768                MvVerificationLimitKind::Restore => {
769                    const MSG: &str = "pending actuation existed before the bounded read";
770                    let pending = tracker.pending.take().require_invariant(MSG)?;
771                    let checked_instant = Instant::now();
772                    let checked_at = checked_at_for_pending(&pending, checked_instant)?;
773                    return Ok(PendingMvVerificationRead::RestoreTimedOut {
774                        pending,
775                        checked_at,
776                        checked_instant,
777                    });
778                }
779            },
780        }
781    }
782}
783#[allow(clippy::too_many_arguments)]
784pub(super) async fn resolve_pending_mv_read(
785    pool: &SqlitePool,
786    effective_timing: EffectiveTiming,
787    tag: &str,
788    tracker: &mut MvActuationTracker,
789    operation: anyhow::Result<TickOperation<(f32, bhtune_driver::Quality)>>,
790    checked_at: DateTime<Utc>,
791    checked_instant: Instant,
792    allow_uncertain_quality: bool,
793) -> anyhow::Result<PendingMvVerificationResult> {
794    let operation = match operation {
795        Ok(operation) => operation,
796        Err(error) => {
797            let pending = tracker
798                .pending
799                .take()
800                .require_invariant("pending actuation existed before the verification read")?;
801            let detail = format!("MV verification read failed: {error}");
802            record_final_actuation_observation(
803                pool,
804                &pending,
805                checked_at,
806                None,
807                None,
808                MvActuationStatus::Unverified,
809                &detail,
810            )
811            .await;
812            return Err(error);
813        }
814    };
815
816    let (readback, quality) = match operation {
817        TickOperation::Completed(value) => value,
818        TickOperation::Cancelled => {
819            let pending = tracker
820                .pending
821                .take()
822                .require_invariant("pending actuation existed before the verification read")?;
823            finalize_actuation_best_effort(
824                pool,
825                &pending,
826                MvActuationStatus::Unverified,
827                "MV verification was interrupted before confirmation completed",
828            )
829            .await;
830            return Ok(PendingMvVerificationResult::Abort(
831                AbortReason::UserInterrupt,
832            ));
833        }
834        TickOperation::TimedOut => {
835            let pending = tracker
836                .pending
837                .take()
838                .require_invariant("pending actuation existed before the verification read")?;
839            let detail = format!(
840                "MV verification read did not complete within {} seconds",
841                effective_timing.op_timeout_secs
842            );
843            record_final_actuation_observation(
844                pool,
845                &pending,
846                checked_at,
847                None,
848                None,
849                MvActuationStatus::Unverified,
850                &detail,
851            )
852            .await;
853            return Ok(PendingMvVerificationResult::Abort(
854                AbortReason::OperationTimedOut {
855                    tag: tag.to_string(),
856                    op_timeout_secs: effective_timing.op_timeout_secs,
857                },
858            ));
859        }
860    };
861    let sample_quality = sample_quality_from_driver(quality);
862    if check_quality(tag, quality, allow_uncertain_quality).is_err() {
863        let pending = tracker
864            .pending
865            .take()
866            .require_invariant("pending actuation existed before the verification read")?;
867        let detail = format!("MV verification read reported OPC quality {quality:?}");
868        record_final_actuation_observation(
869            pool,
870            &pending,
871            checked_at,
872            Some(readback),
873            Some(sample_quality),
874            MvActuationStatus::Unverified,
875            &detail,
876        )
877        .await;
878        return Ok(PendingMvVerificationResult::Abort(
879            AbortReason::PoorQuality {
880                tag: tag.to_string(),
881                quality,
882            },
883        ));
884    }
885
886    Ok(PendingMvVerificationResult::Value(
887        PendingMvVerificationValue {
888            readback,
889            sample_quality,
890            checked_at,
891            checked_instant,
892        },
893    ))
894}
895#[allow(clippy::too_many_arguments)]
896pub(super) async fn resolve_pending_mv_poll(
897    pool: &SqlitePool,
898    effective_timing: EffectiveTiming,
899    pv_mv_values: TickOperation<HashMap<String, TagValue>>,
900    mv_tag: &str,
901    checked_at: DateTime<Utc>,
902    checked_instant: Instant,
903    read_duration: Duration,
904    allow_uncertain_quality: bool,
905    tracker: &mut MvActuationTracker,
906    timing: &mut PollTimingAccumulator,
907) -> anyhow::Result<(Option<AbortReason>, bool)> {
908    let operation = match pv_mv_values {
909        TickOperation::Completed(values) => match read_numeric_from_batch(&values, mv_tag) {
910            Ok(value) => Ok(TickOperation::Completed(value)),
911            Err(error) => Err(error),
912        },
913        TickOperation::Cancelled => Ok(TickOperation::Cancelled),
914        TickOperation::TimedOut => Ok(TickOperation::TimedOut),
915    };
916
917    match resolve_pending_mv_read(
918        pool,
919        effective_timing,
920        mv_tag,
921        tracker,
922        operation,
923        checked_at,
924        checked_instant,
925        allow_uncertain_quality,
926    )
927    .await?
928    {
929        PendingMvVerificationResult::Abort(reason) => Ok((Some(reason), false)),
930        PendingMvVerificationResult::Value(value) => {
931            timing.observe_mv_verification(read_duration);
932            let outcome = finalize_pending_mv_verification(
933                pool,
934                mv_tag,
935                tracker,
936                value,
937                MvVerificationTrigger::Scheduled,
938                ActuationAuditPolicy::Required,
939            )
940            .await?;
941            Ok((outcome, true))
942        }
943    }
944}
945#[allow(clippy::too_many_arguments)]
946pub(super) async fn confirm_pending_mv_actuation(
947    pool: &SqlitePool,
948    tracker: &mut MvActuationTracker,
949    pending: PendingMvActuation,
950    checked_at: DateTime<Utc>,
951    readback: f32,
952    sample_quality: SampleQuality,
953    audit_policy: ActuationAuditPolicy,
954) -> anyhow::Result<()> {
955    let recorded_attempts = match pending.id {
956        Some(id) => {
957            let result = TuneMvActuationRow::record_final_observation(
958                pool,
959                id,
960                checked_at,
961                Some(readback),
962                Some(sample_quality),
963                MvActuationStatus::Confirmed,
964                None,
965            )
966            .await;
967            match result {
968                Ok(row) => Some(row.attempt_count),
969                Err(error) if audit_policy == ActuationAuditPolicy::BestEffort => {
970                    tracing::error!(
971                        actuation_id = id,
972                        error = %error,
973                        "failed to record confirmed MV restore observation"
974                    );
975                    None
976                }
977                Err(error) => {
978                    tracker.pending = Some(pending);
979                    return Err(error.into());
980                }
981            }
982        }
983        None => None,
984    };
985    if let Some(attempt_count) = recorded_attempts.filter(|attempt_count| *attempt_count > 1) {
986        tracing::warn!(
987            actuation_id = ?pending.id,
988            attempt_count,
989            target = pending.target,
990            readback,
991            tolerance = pending.tolerance,
992            "MV actuation confirmed after earlier unsuccessful observations"
993        );
994    }
995    tracker.confirmed_mv = Some(pending.target);
996    Ok(())
997}
998pub(super) async fn finalize_pending_mv_verification(
999    pool: &SqlitePool,
1000    tag: &str,
1001    tracker: &mut MvActuationTracker,
1002    value: PendingMvVerificationValue,
1003    trigger: MvVerificationTrigger,
1004    audit_policy: ActuationAuditPolicy,
1005) -> anyhow::Result<Option<AbortReason>> {
1006    let pending = tracker
1007        .pending
1008        .take()
1009        .require_invariant("pending actuation existed before the verification result")?;
1010    if value.checked_instant > pending.deadline {
1011        let detail = if actuation_matches(pending.target, value.readback, pending.tolerance) {
1012            "MV readback matched the target only after the confirmation deadline"
1013        } else {
1014            "MV readback remained outside tolerance after the confirmation deadline"
1015        };
1016        record_final_actuation_observation(
1017            pool,
1018            &pending,
1019            value.checked_at,
1020            Some(value.readback),
1021            Some(value.sample_quality),
1022            MvActuationStatus::Failed,
1023            detail,
1024        )
1025        .await;
1026        return Ok(Some(actuation_abort_reason(
1027            tag,
1028            &pending,
1029            Some(value.readback),
1030            value.checked_instant,
1031        )));
1032    }
1033    if actuation_matches(pending.target, value.readback, pending.tolerance) {
1034        confirm_pending_mv_actuation(
1035            pool,
1036            tracker,
1037            pending,
1038            value.checked_at,
1039            value.readback,
1040            value.sample_quality,
1041            audit_policy,
1042        )
1043        .await?;
1044        return Ok(None);
1045    }
1046
1047    let deadline_reached =
1048        trigger == MvVerificationTrigger::Deadline || value.checked_instant >= pending.deadline;
1049    if deadline_reached {
1050        record_final_actuation_observation(
1051            pool,
1052            &pending,
1053            value.checked_at,
1054            Some(value.readback),
1055            Some(value.sample_quality),
1056            MvActuationStatus::Failed,
1057            "MV readback remained outside tolerance at the confirmation deadline",
1058        )
1059        .await;
1060        return Ok(Some(actuation_abort_reason(
1061            tag,
1062            &pending,
1063            Some(value.readback),
1064            value.checked_instant,
1065        )));
1066    }
1067
1068    let attempt_count = record_actuation_observation(
1069        pool,
1070        &pending,
1071        value.checked_at,
1072        Some(value.readback),
1073        Some(value.sample_quality),
1074        audit_policy,
1075    )
1076    .await?;
1077    tracing::warn!(
1078        actuation_id = ?pending.id,
1079        attempt_count,
1080        target = pending.target,
1081        readback = value.readback,
1082        tolerance = pending.tolerance,
1083        "MV readback is outside tolerance; confirmation remains pending"
1084    );
1085    let mut pending = pending;
1086    pending.last_readback = Some(value.readback);
1087    tracker.pending = Some(pending);
1088    Ok(None)
1089}
1090#[cfg(test)]
1091#[allow(clippy::too_many_arguments)]
1092pub(super) async fn verify_pending_mv_actuation(
1093    pool: &SqlitePool,
1094    args: &TuneRequest,
1095    tag: &str,
1096    driver: &dyn Driver,
1097    ctrl_c: &mut CtrlC,
1098    allow_uncertain_quality: bool,
1099    tracker: &mut MvActuationTracker,
1100    call_deadline: Option<Instant>,
1101) -> anyhow::Result<Option<AbortReason>> {
1102    let trigger = tracker
1103        .pending
1104        .as_ref()
1105        .and_then(|pending| verification_trigger(pending, Instant::now()))
1106        .unwrap_or(MvVerificationTrigger::Scheduled);
1107    verify_pending_mv_actuation_with_timing(
1108        pool,
1109        args,
1110        test_effective_timing(args),
1111        tag,
1112        driver,
1113        ctrl_c,
1114        allow_uncertain_quality,
1115        tracker,
1116        trigger,
1117        call_deadline.map_or(
1118            MvVerificationCallLimit::None,
1119            MvVerificationCallLimit::Restore,
1120        ),
1121        ActuationAuditPolicy::Required,
1122        None,
1123    )
1124    .await
1125}
1126pub(super) async fn finalize_pending_for_run_best_effort(
1127    pool: &SqlitePool,
1128    run_id: i64,
1129    detail: &str,
1130) {
1131    if let Err(error) = TuneMvActuationRow::finalize_pending_for_run(
1132        pool,
1133        run_id,
1134        MvActuationStatus::Unverified,
1135        Some(detail),
1136    )
1137    .await
1138    {
1139        tracing::error!(
1140            run_id,
1141            error = %error,
1142            "failed to finalize pending MV actuation rows"
1143        );
1144    }
1145}
1146pub(super) async fn supersede_pending_actuation_best_effort(
1147    pool: &SqlitePool,
1148    tracker: &mut MvActuationTracker,
1149    detail: &str,
1150) {
1151    let Some(pending) = tracker.pending.take() else {
1152        return;
1153    };
1154    finalize_actuation_best_effort(pool, &pending, MvActuationStatus::Superseded, detail).await;
1155}
1156pub(super) async fn record_handoff_observation_best_effort(
1157    pool: &SqlitePool,
1158    pending: &PendingMvActuation,
1159    checked_at: DateTime<Utc>,
1160    readback: Option<f32>,
1161    quality: Option<SampleQuality>,
1162    detail: &str,
1163) {
1164    record_final_actuation_observation(
1165        pool,
1166        pending,
1167        checked_at,
1168        readback,
1169        quality,
1170        MvActuationStatus::Superseded,
1171        detail,
1172    )
1173    .await;
1174}
1175pub(super) fn checked_at_for_pending(
1176    pending: &PendingMvActuation,
1177    checked_instant: Instant,
1178) -> anyhow::Result<DateTime<Utc>> {
1179    Ok(pending.switch_tick
1180        + chrono::Duration::from_std(
1181            checked_instant.saturating_duration_since(pending.switch_instant),
1182        )
1183        .map_err(|_| anyhow::anyhow!("MV actuation observation time exceeded chrono's range"))?)
1184}
1185pub(super) fn utc_after_elapsed(
1186    now: DateTime<Utc>,
1187    elapsed: Duration,
1188) -> anyhow::Result<DateTime<Utc>> {
1189    Ok(now
1190        + chrono::Duration::from_std(elapsed)
1191            .map_err(|_| anyhow::anyhow!("MV command time exceeded chrono's range"))?)
1192}
bhtune_runtime/tune/actuation.rs

bhtune_runtime/tune/
actuation.rs
