bhtune_runtime/
live_ownership.rs

1//! Process-shared and database-backed ownership for live controller mutations.
2
3use std::{
4    collections::{HashMap, VecDeque},
5    fs::{File, OpenOptions},
6    io,
7    path::PathBuf,
8    sync::{
9        Arc,
10        atomic::{AtomicBool, Ordering},
11    },
12    time::Duration,
13};
14
15use bhtune_db::{
16    DbError, SqlitePool, database_path,
17    models::{
18        LiveMutationStepRow, LiveOperationKind, LiveOwnershipRow, MutationStepStatus,
19        NewLiveMutationStep, NewLiveOwnership, TuneDriver, TuneRunRow,
20    },
21};
22use bhtune_driver::{
23    BrowsePage, BrowsePageRequest, Driver, DriverError, DriverResult, TagId, TagValue, TagWrite,
24    WriteOutcome,
25};
26use chrono::Utc;
27use fs2::FileExt;
28use serde_json::{Value, json};
29use thiserror::Error;
30use tokio::sync::Mutex;
31use tokio::task::JoinHandle;
32
33const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
34
35#[derive(Debug, Error)]
36pub enum LiveOwnershipAcquireError {
37    #[error("another live controller operation holds this database's OS ownership lock")]
38    LockBusy,
39    #[error("the canonical live controller resource is already claimed")]
40    ResourceClaimed,
41    #[error(transparent)]
42    Persistence(#[from] anyhow::Error),
43}
44
45/// An OS lock for one canonical SQLite database. The lock file's existence is not evidence
46/// of ownership; only the held kernel lock prevents another process from acquiring it.
47#[derive(Debug)]
48pub struct DatabaseFileGuard {
49    _file: File,
50    database_key: String,
51    #[cfg(test)]
52    memory_lock_file: Option<Arc<tempfile::NamedTempFile>>,
53}
54
55impl DatabaseFileGuard {
56    /// Attempts to acquire the database's nonblocking, process-shared exclusive lock.
57    pub async fn try_acquire(pool: &SqlitePool) -> anyhow::Result<Option<Self>> {
58        let path = database_path(pool).await;
59        #[cfg(test)]
60        let pool_identity = std::ptr::from_ref(pool) as usize;
61        tokio::task::spawn_blocking(move || {
62            #[cfg(test)]
63            let mut memory_lock_file = None;
64            let (lock_path, database_key) = match path {
65                Ok(path) => {
66                    let canonical_path = canonicalize_database_path(&path)?;
67                    let database_key = database_key_from_canonical_path(&canonical_path)?;
68                    let mut lock_path = canonical_path.as_os_str().to_os_string();
69                    lock_path.push(".live-owner.lock");
70                    (PathBuf::from(lock_path), database_key)
71                }
72                #[cfg(test)]
73                Err(DbError::DatabasePathUnavailable) => {
74                    let file = tests::memory_database_lock_file(pool_identity)?;
75                    let path = file.path().to_path_buf();
76                    memory_lock_file = Some(file);
77                    (path, format!("test-memory:{pool_identity:x}"))
78                }
79                Err(error) => return Err(error.into()),
80            };
81            let file = OpenOptions::new()
82                .create(true)
83                .truncate(false)
84                .read(true)
85                .write(true)
86                .open(lock_path)
87                .map_err(|error| {
88                    anyhow::anyhow!("cannot open the live-operation lock file: {error}")
89                })?;
90            match classify_lock_result(FileExt::try_lock_exclusive(&file))? {
91                true => Ok(Some(Self {
92                    _file: file,
93                    database_key,
94                    #[cfg(test)]
95                    memory_lock_file,
96                })),
97                false => Ok(None),
98            }
99        })
100        .await?
101    }
102
103    pub fn database_key(&self) -> &str {
104        &self.database_key
105    }
106}
107
108/// A held OS lock paired with the conditional SQLite owner/claim and an independent
109/// five-second heartbeat. Dropping it releases the OS lock; successful completion must call
110/// [`Self::release`] so the database owner and claim are also durably retired.
111pub struct LiveOperationGuard {
112    file_guard: DatabaseFileGuard,
113    owner: LiveOwnershipRow,
114    claim_key: String,
115    pool: SqlitePool,
116    heartbeat_failed: Arc<AtomicBool>,
117    previous_write_values: Arc<Mutex<HashMap<String, VecDeque<String>>>>,
118    heartbeat_task: Option<JoinHandle<()>>,
119}
120
121/// A `Driver` decorator that commits a durable per-write intent before forwarding any live
122/// mutation and records whether the transport accepted or rejected it afterward.
123pub struct AuditedDriver<'a> {
124    inner: &'a dyn Driver,
125    pool: &'a SqlitePool,
126    guard: &'a LiveOperationGuard,
127    run_id: Option<i64>,
128}
129
130impl<'a> AuditedDriver<'a> {
131    pub fn new(
132        inner: &'a dyn Driver,
133        pool: &'a SqlitePool,
134        guard: &'a LiveOperationGuard,
135        run_id: Option<i64>,
136    ) -> Self {
137        Self {
138            inner,
139            pool,
140            guard,
141            run_id,
142        }
143    }
144
145    pub fn run_id(&self) -> Option<i64> {
146        self.run_id
147    }
148
149    pub fn owner(&self) -> &LiveOperationGuard {
150        self.guard
151    }
152}
153
154#[async_trait::async_trait]
155impl Driver for AuditedDriver<'_> {
156    async fn read(&self, tags: &[TagId]) -> DriverResult<Vec<TagValue>> {
157        self.inner.read(tags).await
158    }
159
160    async fn write(&self, tag: &TagId, value: TagWrite) -> DriverResult<WriteOutcome> {
161        self.guard.ensure_healthy().map_err(|error| {
162            DriverError::Operation(Box::new(io::Error::other(error.to_string())))
163        })?;
164        let target_json = write_target_json(tag, &value).map_err(|error| {
165            DriverError::Operation(Box::new(io::Error::other(error.to_string())))
166        })?;
167        let step = LiveMutationStepRow::begin(
168            self.pool,
169            NewLiveMutationStep {
170                owner_id: self.guard.owner.id,
171                run_id: self.run_id,
172                step: "controller_write".to_string(),
173                target_json,
174                previous_json: self.guard.take_previous_write_value(tag).await,
175                started_at: Utc::now(),
176            },
177        )
178        .await
179        .map_err(db_driver_error)?;
180
181        let result = self.inner.write(tag, value).await;
182        let (status, detail) = match &result {
183            Ok(outcome) if outcome.success => (
184                MutationStepStatus::Confirmed,
185                Some(
186                    "the driver accepted the write; value confirmation is recorded separately"
187                        .to_string(),
188                ),
189            ),
190            Ok(outcome) => (
191                MutationStepStatus::Failed,
192                Some(
193                    outcome
194                        .error_message
195                        .clone()
196                        .unwrap_or_else(|| "the driver rejected the write".to_string()),
197                ),
198            ),
199            Err(error) => (MutationStepStatus::Failed, Some(error.to_string())),
200        };
201        let audit_result = LiveMutationStepRow::finish(
202            self.pool,
203            step.id,
204            status,
205            Utc::now(),
206            None,
207            detail.as_deref(),
208        )
209        .await;
210        if let Err(error) = audit_result {
211            return Err(db_driver_error(error));
212        }
213        result
214    }
215
216    async fn browse(&self, request: BrowsePageRequest) -> DriverResult<BrowsePage> {
217        self.inner.browse(request).await
218    }
219}
220
221fn write_target_json(tag: &str, value: &TagWrite) -> anyhow::Result<String> {
222    let value = match value {
223        TagWrite::Float(value) => {
224            let number = serde_json::Number::from_f64(f64::from(*value))
225                .ok_or_else(|| anyhow::anyhow!("controller write target is not finite"))?;
226            Value::Number(number)
227        }
228        TagWrite::Raw(value) => Value::String(value.clone()),
229    };
230    serde_json::to_string(&json!({"tag": tag, "value": value}))
231        .map_err(|error| anyhow::anyhow!("failed to serialize controller write target: {error}"))
232}
233
234fn db_driver_error(error: DbError) -> DriverError {
235    DriverError::Operation(Box::new(error))
236}
237
238impl LiveOperationGuard {
239    /// Acquires the OS lock first, then creates the owner and atomically claims its canonical
240    /// database/controller/tag key. Recovery uses a separate transaction that also starts
241    /// the recovery attempt and records its evidence.
242    #[allow(clippy::too_many_arguments)]
243    pub async fn acquire(
244        pool: &SqlitePool,
245        run_id: Option<i64>,
246        operation_kind: LiveOperationKind,
247        bridge_host: &str,
248        server: &str,
249        resource_tag: &str,
250        restore_intent_json: Option<String>,
251    ) -> Result<Self, LiveOwnershipAcquireError> {
252        let file_guard = Self::try_acquire_database_file(pool)
253            .await
254            .map_err(LiveOwnershipAcquireError::Persistence)?
255            .ok_or(LiveOwnershipAcquireError::LockBusy)?;
256        let database_key = file_guard.database_key().to_owned();
257        let resource_key = resource_key(bridge_host, server, resource_tag)
258            .map_err(LiveOwnershipAcquireError::Persistence)?;
259        let claim_key = claim_key(&database_key, &resource_key)
260            .map_err(LiveOwnershipAcquireError::Persistence)?;
261        let now = Utc::now();
262        let owner = LiveOwnershipRow::create_and_claim(
263            pool,
264            NewLiveOwnership {
265                run_id,
266                operation_kind,
267                database_key,
268                resource_key,
269                owner_pid: i64::from(std::process::id()),
270                acquired_at: now,
271                restore_intent_json,
272            },
273            &claim_key,
274        )
275        .await
276        .map_err(|error| LiveOwnershipAcquireError::Persistence(error.into()))?
277        .ok_or(LiveOwnershipAcquireError::ResourceClaimed)?;
278        Ok(Self::start_heartbeat(pool, file_guard, owner, claim_key))
279    }
280
281    /// Acquires ownership for a post-run write or revert using only the connection and MV
282    /// recorded on `run`. Explicit caller values must be cross-checked before this function;
283    /// they are never used to select a different controller.
284    pub async fn acquire_for_recorded_run(
285        pool: &SqlitePool,
286        run: &TuneRunRow,
287        operation_kind: LiveOperationKind,
288        restore_intent_json: String,
289    ) -> Result<Self, LiveOwnershipAcquireError> {
290        if run.driver != TuneDriver::Opcda {
291            return Err(LiveOwnershipAcquireError::Persistence(anyhow::anyhow!(
292                "run {} used the {:?} driver and has no live OPC DA operation to own",
293                run.id,
294                run.driver
295            )));
296        }
297        let bridge_host = run.bridge_host.as_deref().ok_or_else(|| {
298            LiveOwnershipAcquireError::Persistence(anyhow::anyhow!(
299                "run {} has no recorded bridge host; refusing to guess which gateway to use",
300                run.id
301            ))
302        })?;
303        let server = run.opc_server.as_deref().ok_or_else(|| {
304            LiveOwnershipAcquireError::Persistence(anyhow::anyhow!(
305                "run {} has no recorded OPC server; refusing to guess which controller to use",
306                run.id
307            ))
308        })?;
309        let resource_tag = run.tags.manipulated_variable.as_str();
310        Self::acquire(
311            pool,
312            Some(run.id),
313            operation_kind,
314            bridge_host,
315            server,
316            resource_tag,
317            Some(restore_intent_json),
318        )
319        .await
320    }
321
322    /// Returns a database-file guard without creating a database owner. Full-mode startup
323    /// uses this to distinguish a stale heartbeat from a still-running, possibly paused owner.
324    pub async fn try_acquire_database_file(
325        pool: &SqlitePool,
326    ) -> anyhow::Result<Option<DatabaseFileGuard>> {
327        DatabaseFileGuard::try_acquire(pool).await
328    }
329
330    pub(crate) fn from_database_owner(
331        pool: &SqlitePool,
332        file_guard: DatabaseFileGuard,
333        owner: LiveOwnershipRow,
334        claim_key: String,
335    ) -> Self {
336        Self::start_heartbeat(pool, file_guard, owner, claim_key)
337    }
338
339    fn start_heartbeat(
340        pool: &SqlitePool,
341        file_guard: DatabaseFileGuard,
342        owner: LiveOwnershipRow,
343        claim_key: String,
344    ) -> Self {
345        let heartbeat_failed = Arc::new(AtomicBool::new(false));
346        let previous_write_values = Arc::new(Mutex::new(HashMap::new()));
347        let task_failed = Arc::clone(&heartbeat_failed);
348        let task_pool = pool.clone();
349        let owner_id = owner.id;
350        let heartbeat_task = tokio::spawn(async move {
351            let mut interval = tokio::time::interval(HEARTBEAT_INTERVAL);
352            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
353            interval.tick().await;
354            loop {
355                interval.tick().await;
356                if let Err(error) =
357                    LiveOwnershipRow::heartbeat(&task_pool, owner_id, Utc::now()).await
358                {
359                    task_failed.store(true, Ordering::Release);
360                    tracing::error!(
361                        owner_id,
362                        error = %error,
363                        "live controller owner heartbeat failed"
364                    );
365                    break;
366                }
367            }
368        });
369        Self {
370            file_guard,
371            owner,
372            claim_key,
373            pool: pool.clone(),
374            heartbeat_failed,
375            previous_write_values,
376            heartbeat_task: Some(heartbeat_task),
377        }
378    }
379
380    pub fn owner(&self) -> &LiveOwnershipRow {
381        &self.owner
382    }
383
384    pub fn claim_key(&self) -> &str {
385        &self.claim_key
386    }
387
388    pub fn database_key(&self) -> &str {
389        self.file_guard.database_key()
390    }
391
392    /// Prevents a caller from starting another controller mutation after an asynchronous
393    /// database-heartbeat failure. An already-issued driver call remains protected by the OS
394    /// lock until its owning future finishes or the process exits.
395    pub fn ensure_healthy(&self) -> anyhow::Result<()> {
396        if self.heartbeat_failed.load(Ordering::Acquire) {
397            anyhow::bail!(
398                "live controller ownership heartbeat failed; refusing further mutations and \
399                 retaining the pre-write audit intent"
400            );
401        }
402        Ok(())
403    }
404
405    /// Queues the value observed before the next write to `tag`. The audited driver consumes
406    /// it when it records the write intent, preserving correct pre-write data even for
407    /// write-and-rollback sequences on the same tag.
408    pub async fn queue_previous_write_value(&self, tag: &str, value_json: String) {
409        self.previous_write_values
410            .lock()
411            .await
412            .entry(tag.to_owned())
413            .or_default()
414            .push_back(value_json);
415    }
416
417    pub async fn clear_previous_write_values(&self) {
418        self.previous_write_values.lock().await.clear();
419    }
420
421    async fn take_previous_write_value(&self, tag: &str) -> Option<String> {
422        let mut values = self.previous_write_values.lock().await;
423        let by_tag = values.get_mut(tag)?;
424        let value = by_tag.pop_front();
425        if by_tag.is_empty() {
426            values.remove(tag);
427        }
428        value
429    }
430
431    /// Replaces the pre-mutation recovery intent with a more complete, durably persisted
432    /// snapshot. Failure prevents the caller from issuing its next live write.
433    pub async fn persist_restore_intent(&self, intent_json: &str) -> anyhow::Result<()> {
434        self.ensure_healthy()?;
435        LiveOwnershipRow::set_restore_intent(&self.pool, self.owner.id, intent_json).await?;
436        Ok(())
437    }
438
439    /// Retires the persisted owner/claim before releasing the OS lock. A database persistence
440    /// failure is returned to the caller and must never be reported as successful completion.
441    pub async fn release(mut self) -> anyhow::Result<()> {
442        self.stop_heartbeat().await;
443        self.ensure_healthy()?;
444        LiveOwnershipRow::release(&self.pool, self.owner.id, Utc::now()).await?;
445        Ok(())
446    }
447
448    /// Stops heartbeat updates while a recovery attempt atomically finalizes both owners and
449    /// the claim. The caller must keep this guard alive until that transaction succeeds or
450    /// fails.
451    pub async fn stop_heartbeat(&mut self) {
452        if let Some(task) = self.heartbeat_task.take() {
453            task.abort();
454            let _ = task.await;
455        }
456    }
457}
458
459impl Drop for LiveOperationGuard {
460    fn drop(&mut self) {
461        if let Some(task) = self.heartbeat_task.take() {
462            task.abort();
463        }
464    }
465}
466
467/// The canonical resource key is independent of the run ID. Host and ProgID comparisons are
468/// case-insensitive; the OPC ItemID remains exact because OPC namespaces may be case-sensitive.
469pub fn resource_key(bridge_host: &str, server: &str, resource_tag: &str) -> anyhow::Result<String> {
470    serde_json::to_string(&(
471        bridge_host.to_ascii_lowercase(),
472        server.to_ascii_lowercase(),
473        resource_tag,
474    ))
475    .map_err(|error| anyhow::anyhow!("failed to serialize the live resource identity: {error}"))
476}
477
478pub fn claim_key(database_key: &str, resource_key: &str) -> anyhow::Result<String> {
479    serde_json::to_string(&(database_key, resource_key))
480        .map_err(|error| anyhow::anyhow!("failed to serialize the live claim identity: {error}"))
481}
482
483fn lock_is_contended(error: &io::Error) -> bool {
484    error.kind() == io::ErrorKind::WouldBlock
485        || error
486            .raw_os_error()
487            .is_some_and(|code| fs2::lock_contended_error().raw_os_error() == Some(code))
488}
489
490fn classify_lock_result(result: io::Result<()>) -> anyhow::Result<bool> {
491    match result {
492        Ok(()) => Ok(true),
493        Err(error) if lock_is_contended(&error) => Ok(false),
494        Err(error) => Err(error.into()),
495    }
496}
497
498fn database_key_from_canonical_path(path: &std::path::Path) -> anyhow::Result<String> {
499    path.to_str()
500        .map(str::to_owned)
501        .ok_or_else(|| anyhow::anyhow!("database path is not valid UTF-8"))
502}
503
504fn canonicalize_database_path(path: &std::path::Path) -> anyhow::Result<PathBuf> {
505    std::fs::canonicalize(path).map_err(|error| {
506        anyhow::anyhow!(
507            "cannot canonicalize the live-operation database path {}: {error}",
508            path.display()
509        )
510    })
511}
512
513#[cfg(test)]
514mod tests;
bhtune_runtime/live_ownership.rs

bhtune_runtime/
live_ownership.rs
