bhtune_db/models/
live_ownership.rs

1//! Durable ownership, mutation, and interrupted-run recovery evidence for live OPC DA work.
2
3use chrono::{DateTime, Utc};
4use sqlx::{Row, SqlitePool, sqlite::SqliteRow};
5
6use crate::{
7    convert::{enum_to_text, text_to_enum},
8    error::{DbError, DbResult},
9};
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
12#[serde(rename_all = "snake_case")]
13pub enum LiveOperationKind {
14    Tune,
15    PidWrite,
16    PidRevert,
17    Recovery,
18    OpcWrite,
19}
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
22#[serde(rename_all = "snake_case")]
23pub enum LiveOwnershipState {
24    Active,
25    Released,
26    Orphaned,
27    Recovered,
28}
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
31#[serde(rename_all = "snake_case")]
32pub enum MutationStepStatus {
33    Intent,
34    Confirmed,
35    Failed,
36    NotNeeded,
37}
38
39#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
40#[serde(rename_all = "snake_case")]
41pub enum RecoveryAttemptStatus {
42    Running,
43    Confirmed,
44    Incomplete,
45    Failed,
46}
47
48#[derive(Debug, Clone, PartialEq, serde::Serialize)]
49pub struct NewLiveOwnership {
50    pub run_id: Option<i64>,
51    pub operation_kind: LiveOperationKind,
52    pub database_key: String,
53    pub resource_key: String,
54    pub owner_pid: i64,
55    pub acquired_at: DateTime<Utc>,
56    pub restore_intent_json: Option<String>,
57}
58
59#[derive(Debug, Clone, PartialEq, serde::Serialize)]
60pub struct LiveOwnershipRow {
61    pub id: i64,
62    pub run_id: Option<i64>,
63    pub operation_kind: LiveOperationKind,
64    pub database_key: String,
65    pub resource_key: String,
66    pub owner_pid: i64,
67    pub acquired_at: DateTime<Utc>,
68    pub heartbeat_at: DateTime<Utc>,
69    pub released_at: Option<DateTime<Utc>>,
70    pub state: LiveOwnershipState,
71    pub restore_intent_json: Option<String>,
72    pub orphan_eligible: bool,
73    pub orphan_evidence_json: Option<String>,
74}
75
76#[derive(Debug, Clone, PartialEq, serde::Serialize)]
77pub struct LiveMutationStepRow {
78    pub id: i64,
79    pub owner_id: i64,
80    pub run_id: Option<i64>,
81    pub step: String,
82    pub target_json: String,
83    pub previous_json: Option<String>,
84    pub status: MutationStepStatus,
85    pub started_at: DateTime<Utc>,
86    pub completed_at: Option<DateTime<Utc>>,
87    pub readback_json: Option<String>,
88    pub detail: Option<String>,
89}
90
91#[derive(Debug, Clone, PartialEq, serde::Serialize)]
92pub struct NewLiveMutationStep {
93    pub owner_id: i64,
94    pub run_id: Option<i64>,
95    pub step: String,
96    pub target_json: String,
97    pub previous_json: Option<String>,
98    pub started_at: DateTime<Utc>,
99}
100
101#[derive(Debug, Clone, PartialEq, serde::Serialize)]
102pub struct TuneRecoveryAttemptRow {
103    pub id: i64,
104    pub run_id: i64,
105    pub source_owner_id: i64,
106    pub recovery_owner_id: i64,
107    pub export_path: String,
108    pub started_at: DateTime<Utc>,
109    pub completed_at: Option<DateTime<Utc>>,
110    pub status: RecoveryAttemptStatus,
111    pub detail: Option<String>,
112    pub evidence_json: String,
113}
114
115pub struct OrphanOwnerUpdate<'a> {
116    pub owner_id: i64,
117    pub expected_heartbeat: DateTime<Utc>,
118    pub expected_claim_key: &'a str,
119    pub now: DateTime<Utc>,
120    pub evidence_json: &'a str,
121    pub eligible: bool,
122    pub detail: &'a str,
123}
124
125pub struct UnrecoverableOwnerUpdate<'a> {
126    pub owner_id: i64,
127    pub expected_heartbeat: DateTime<Utc>,
128    pub expected_claim_key: &'a str,
129    pub now: DateTime<Utc>,
130    pub evidence_json: &'a str,
131    pub fail_running_run_id: Option<i64>,
132    pub detail: &'a str,
133}
134
135pub struct CompleteRecoveryUpdate<'a> {
136    pub attempt_id: i64,
137    pub recovery_owner_id: i64,
138    pub source_owner_id: i64,
139    pub run_id: i64,
140    pub claim_key: &'a str,
141    pub status: RecoveryAttemptStatus,
142    pub now: DateTime<Utc>,
143    pub detail: Option<&'a str>,
144    pub evidence_json: &'a str,
145}
146
147pub struct RestartRecoveryUpdate<'a> {
148    pub attempt_id: i64,
149    pub recovery_owner_id: i64,
150    pub expected_heartbeat: DateTime<Utc>,
151    pub source_owner_id: i64,
152    pub run_id: i64,
153    pub claim_key: &'a str,
154    pub now: DateTime<Utc>,
155    pub detail: &'a str,
156    pub evidence_json: &'a str,
157}
158
159impl LiveOwnershipRow {
160    /// Inserts an owner and conditionally claims its canonical database/resource key in one
161    /// transaction. Recovery claim transfer is handled atomically by
162    /// `TuneRecoveryAttemptRow::start_with_claim`.
163    pub async fn create_and_claim(
164        pool: &SqlitePool,
165        new: NewLiveOwnership,
166        claim_key: &str,
167    ) -> DbResult<Option<Self>> {
168        let mut transaction = pool.begin().await.map_err(DbError::Query)?;
169        let row = sqlx::query(
170            r"
171            INSERT INTO live_operation_owners (
172                run_id, operation_kind, database_key, resource_key, owner_pid,
173                acquired_at, heartbeat_at, state, restore_intent_json
174            ) VALUES (?, ?, ?, ?, ?, ?, ?, 'active', ?)
175            RETURNING *
176            ",
177        )
178        .bind(new.run_id)
179        .bind(enum_to_text(&new.operation_kind)?)
180        .bind(new.database_key)
181        .bind(new.resource_key)
182        .bind(new.owner_pid)
183        .bind(new.acquired_at)
184        .bind(new.acquired_at)
185        .bind(new.restore_intent_json)
186        .fetch_one(&mut *transaction)
187        .await
188        .map_err(DbError::Query)?;
189        let owner = row_to_live_ownership(row)?;
190
191        let claim_result = sqlx::query(
192            r"
193            INSERT INTO live_operation_claims (claim_key, owner_id, claimed_at)
194            VALUES (?, ?, ?)
195            ON CONFLICT(claim_key) DO NOTHING
196            ",
197        )
198        .bind(claim_key)
199        .bind(owner.id)
200        .bind(new.acquired_at)
201        .execute(&mut *transaction)
202        .await
203        .map_err(DbError::Query)?;
204
205        if claim_result.rows_affected() != 1 {
206            transaction.rollback().await.map_err(DbError::Query)?;
207            return Ok(None);
208        }
209        transaction.commit().await.map_err(DbError::Query)?;
210        Ok(Some(owner))
211    }
212
213    pub async fn heartbeat(pool: &SqlitePool, owner_id: i64, now: DateTime<Utc>) -> DbResult<()> {
214        let result = sqlx::query(
215            r"
216            UPDATE live_operation_owners
217            SET heartbeat_at = ?
218            WHERE id = ? AND state = 'active'
219                AND EXISTS (
220                    SELECT 1 FROM live_operation_claims WHERE owner_id = ?
221                )
222            ",
223        )
224        .bind(now)
225        .bind(owner_id)
226        .bind(owner_id)
227        .execute(pool)
228        .await
229        .map_err(DbError::Query)?;
230        if result.rows_affected() != 1 {
231            return Err(DbError::LiveOwnershipLost);
232        }
233        Ok(())
234    }
235
236    pub async fn release(pool: &SqlitePool, owner_id: i64, now: DateTime<Utc>) -> DbResult<()> {
237        let mut transaction = pool.begin().await.map_err(DbError::Query)?;
238        let result = sqlx::query(
239            r"
240            UPDATE live_operation_owners
241            SET state = 'released', released_at = ?
242            WHERE id = ? AND state = 'active'
243                AND EXISTS (
244                    SELECT 1 FROM live_operation_claims WHERE owner_id = ?
245                )
246            ",
247        )
248        .bind(now)
249        .bind(owner_id)
250        .bind(owner_id)
251        .execute(&mut *transaction)
252        .await
253        .map_err(DbError::Query)?;
254        if result.rows_affected() != 1 {
255            transaction.rollback().await.map_err(DbError::Query)?;
256            return Err(DbError::LiveOwnershipLost);
257        }
258        let claim_result = sqlx::query("DELETE FROM live_operation_claims WHERE owner_id = ?")
259            .bind(owner_id)
260            .execute(&mut *transaction)
261            .await
262            .map_err(DbError::Query)?;
263        if claim_result.rows_affected() != 1 {
264            transaction.rollback().await.map_err(DbError::Query)?;
265            return Err(DbError::LiveOwnershipLost);
266        }
267        transaction.commit().await.map_err(DbError::Query)
268    }
269
270    pub async fn get(pool: &SqlitePool, id: i64) -> DbResult<Option<Self>> {
271        let row = sqlx::query("SELECT * FROM live_operation_owners WHERE id = ?")
272            .bind(id)
273            .fetch_optional(pool)
274            .await
275            .map_err(DbError::Query)?;
276        row.map(row_to_live_ownership).transpose()
277    }
278
279    pub async fn has_claim(
280        pool: &SqlitePool,
281        owner_id: i64,
282        expected_claim_key: &str,
283    ) -> DbResult<bool> {
284        let claimed: i64 = sqlx::query_scalar(
285            "SELECT EXISTS(SELECT 1 FROM live_operation_claims WHERE owner_id = ? AND claim_key = ?)",
286        )
287        .bind(owner_id)
288        .bind(expected_claim_key)
289        .fetch_one(pool)
290        .await
291        .map_err(DbError::Query)?;
292        Ok(claimed == 1)
293    }
294
295    pub async fn list_for_run(pool: &SqlitePool, run_id: i64) -> DbResult<Vec<Self>> {
296        let rows = sqlx::query(
297            "SELECT * FROM live_operation_owners WHERE run_id = ? ORDER BY acquired_at, id",
298        )
299        .bind(run_id)
300        .fetch_all(pool)
301        .await
302        .map_err(DbError::Query)?;
303        rows.into_iter().map(row_to_live_ownership).collect()
304    }
305
306    pub async fn stale_active_owners(
307        pool: &SqlitePool,
308        stale_before: DateTime<Utc>,
309    ) -> DbResult<Vec<Self>> {
310        let rows = sqlx::query(
311            r"
312            SELECT owners.*
313            FROM live_operation_owners AS owners
314            WHERE owners.state = 'active'
315                AND owners.heartbeat_at < ?
316                AND EXISTS (
317                    SELECT 1 FROM live_operation_claims
318                    WHERE live_operation_claims.owner_id = owners.id
319                )
320            ORDER BY owners.heartbeat_at, owners.id
321            ",
322        )
323        .bind(stale_before)
324        .fetch_all(pool)
325        .await
326        .map_err(DbError::Query)?;
327        rows.into_iter().map(row_to_live_ownership).collect()
328    }
329
330    pub async fn set_restore_intent(
331        pool: &SqlitePool,
332        owner_id: i64,
333        intent_json: &str,
334    ) -> DbResult<()> {
335        let result = sqlx::query(
336            r"
337            UPDATE live_operation_owners SET restore_intent_json = ?
338            WHERE id = ? AND state = 'active'
339                AND EXISTS (SELECT 1 FROM live_operation_claims WHERE owner_id = ?)
340            ",
341        )
342        .bind(intent_json)
343        .bind(owner_id)
344        .bind(owner_id)
345        .execute(pool)
346        .await
347        .map_err(DbError::Query)?;
348        if result.rows_affected() != 1 {
349            return Err(DbError::LiveOwnershipLost);
350        }
351        Ok(())
352    }
353
354    pub async fn mark_orphaned(pool: &SqlitePool, update: OrphanOwnerUpdate<'_>) -> DbResult<bool> {
355        let mut transaction = pool.begin().await.map_err(DbError::Query)?;
356        let result = sqlx::query(
357            r"
358            UPDATE live_operation_owners
359            SET state = 'orphaned', released_at = ?, orphan_eligible = ?,
360                orphan_evidence_json = ?
361            WHERE id = ? AND state = 'active' AND heartbeat_at = ?
362                AND operation_kind = 'tune'
363                AND EXISTS (
364                    SELECT 1 FROM tune_runs
365                    WHERE tune_runs.id = live_operation_owners.run_id
366                        AND tune_runs.driver = 'opcda'
367                        AND tune_runs.outcome = 'running'
368                )
369                AND EXISTS (
370                    SELECT 1 FROM live_operation_claims
371                    WHERE live_operation_claims.owner_id = live_operation_owners.id
372                        AND live_operation_claims.claim_key = ?
373                )
374            ",
375        )
376        .bind(update.now)
377        .bind(update.eligible)
378        .bind(update.evidence_json)
379        .bind(update.owner_id)
380        .bind(update.expected_heartbeat)
381        .bind(update.expected_claim_key)
382        .execute(&mut *transaction)
383        .await
384        .map_err(DbError::Query)?;
385        if result.rows_affected() != 1 {
386            transaction.rollback().await.map_err(DbError::Query)?;
387            return Ok(false);
388        }
389
390        let state = if update.eligible {
391            "eligible"
392        } else {
393            "not_recoverable"
394        };
395        let run_result = sqlx::query(
396            r"
397            UPDATE tune_runs
398            SET outcome = 'failed', completed_at = ?, failure_reason = ?,
399                restore_status = 'incomplete', restore_detail = ?,
400                recovery_state = ?, recovery_evidence_json = ?
401            WHERE id = (
402                SELECT run_id FROM live_operation_owners WHERE id = ?
403            ) AND outcome = 'running'
404            ",
405        )
406        .bind(update.now)
407        .bind(update.detail)
408        .bind(update.detail)
409        .bind(state)
410        .bind(update.evidence_json)
411        .bind(update.owner_id)
412        .execute(&mut *transaction)
413        .await
414        .map_err(DbError::Query)?;
415        if run_result.rows_affected() != 1 {
416            transaction.rollback().await.map_err(DbError::Query)?;
417            return Ok(false);
418        }
419
420        if !update.eligible {
421            sqlx::query("DELETE FROM live_operation_claims WHERE owner_id = ?")
422                .bind(update.owner_id)
423                .execute(&mut *transaction)
424                .await
425                .map_err(DbError::Query)?;
426        }
427        transaction.commit().await.map_err(DbError::Query)?;
428        Ok(true)
429    }
430
431    /// Retires a stale owner only after its caller has exported the exact affected rows.
432    /// Any still-running row attached to the orphaned operation is failed closed rather than
433    /// made automatically recoverable.
434    pub async fn mark_orphaned_unrecoverable(
435        pool: &SqlitePool,
436        update: UnrecoverableOwnerUpdate<'_>,
437    ) -> DbResult<bool> {
438        let mut transaction = pool.begin().await.map_err(DbError::Query)?;
439        let owner_result = sqlx::query(
440            r"
441            UPDATE live_operation_owners
442            SET state = 'orphaned', released_at = ?, orphan_eligible = 0,
443                orphan_evidence_json = ?
444            WHERE id = ? AND state = 'active' AND heartbeat_at = ?
445                AND EXISTS (
446                    SELECT 1 FROM live_operation_claims
447                    WHERE owner_id = live_operation_owners.id AND claim_key = ?
448                )
449            ",
450        )
451        .bind(update.now)
452        .bind(update.evidence_json)
453        .bind(update.owner_id)
454        .bind(update.expected_heartbeat)
455        .bind(update.expected_claim_key)
456        .execute(&mut *transaction)
457        .await
458        .map_err(DbError::Query)?;
459        if owner_result.rows_affected() != 1 {
460            transaction.rollback().await.map_err(DbError::Query)?;
461            return Ok(false);
462        }
463
464        if let Some(run_id) = update.fail_running_run_id {
465            let run_result = sqlx::query(
466                r"
467                UPDATE tune_runs
468                SET outcome = 'failed', completed_at = ?, failure_reason = ?,
469                    restore_status = 'incomplete', restore_detail = ?,
470                    recovery_state = 'not_recoverable', recovery_evidence_json = ?
471                WHERE id = ? AND outcome = 'running'
472                    AND EXISTS (
473                        SELECT 1 FROM live_operation_owners
474                        WHERE id = ? AND run_id = ? AND state = 'orphaned'
475                    )
476                ",
477            )
478            .bind(update.now)
479            .bind(update.detail)
480            .bind(update.detail)
481            .bind(update.evidence_json)
482            .bind(run_id)
483            .bind(update.owner_id)
484            .bind(run_id)
485            .execute(&mut *transaction)
486            .await
487            .map_err(DbError::Query)?;
488            if run_result.rows_affected() != 1 {
489                transaction.rollback().await.map_err(DbError::Query)?;
490                return Ok(false);
491            }
492        }
493
494        let claim_result =
495            sqlx::query("DELETE FROM live_operation_claims WHERE owner_id = ? AND claim_key = ?")
496                .bind(update.owner_id)
497                .bind(update.expected_claim_key)
498                .execute(&mut *transaction)
499                .await
500                .map_err(DbError::Query)?;
501        if claim_result.rows_affected() != 1 {
502            transaction.rollback().await.map_err(DbError::Query)?;
503            return Ok(false);
504        }
505        transaction.commit().await.map_err(DbError::Query)?;
506        Ok(true)
507    }
508}
509
510impl LiveMutationStepRow {
511    pub async fn begin(pool: &SqlitePool, new: NewLiveMutationStep) -> DbResult<Self> {
512        let row = sqlx::query(
513            r"
514            INSERT INTO live_mutation_steps (
515                owner_id, run_id, step, target_json, previous_json, status, started_at
516            )
517            SELECT ?, ?, ?, ?, ?, 'intent', ?
518            WHERE EXISTS (
519                SELECT 1 FROM live_operation_owners
520                WHERE id = ? AND run_id IS ? AND state = 'active'
521            )
522                AND EXISTS (
523                    SELECT 1 FROM live_operation_claims WHERE owner_id = ?
524                )
525            RETURNING *
526            ",
527        )
528        .bind(new.owner_id)
529        .bind(new.run_id)
530        .bind(new.step)
531        .bind(new.target_json)
532        .bind(new.previous_json)
533        .bind(new.started_at)
534        .bind(new.owner_id)
535        .bind(new.run_id)
536        .bind(new.owner_id)
537        .fetch_optional(pool)
538        .await
539        .map_err(DbError::Query)?
540        .ok_or(DbError::LiveOwnershipLost)?;
541        row_to_live_mutation_step(row)
542    }
543
544    pub async fn finish(
545        pool: &SqlitePool,
546        step_id: i64,
547        status: MutationStepStatus,
548        now: DateTime<Utc>,
549        readback_json: Option<&str>,
550        detail: Option<&str>,
551    ) -> DbResult<Self> {
552        if status == MutationStepStatus::Intent {
553            return Err(DbError::LiveOwnershipLost);
554        }
555        let row = sqlx::query(
556            r"
557            UPDATE live_mutation_steps
558            SET status = ?, completed_at = ?, readback_json = ?, detail = ?
559            WHERE id = ? AND status = 'intent'
560            RETURNING *
561            ",
562        )
563        .bind(enum_to_text(&status)?)
564        .bind(now)
565        .bind(readback_json)
566        .bind(detail)
567        .bind(step_id)
568        .fetch_one(pool)
569        .await
570        .map_err(DbError::Query)?;
571        row_to_live_mutation_step(row)
572    }
573
574    pub async fn not_needed(
575        pool: &SqlitePool,
576        owner_id: i64,
577        run_id: i64,
578        step: &str,
579        now: DateTime<Utc>,
580        detail: &str,
581    ) -> DbResult<Self> {
582        let row = sqlx::query(
583            r"
584            INSERT INTO live_mutation_steps (
585                owner_id, run_id, step, target_json, status, started_at, completed_at, detail
586            )
587            SELECT ?, ?, ?, '{}', 'not_needed', ?, ?, ?
588            WHERE EXISTS (
589                SELECT 1 FROM live_operation_owners
590                WHERE id = ? AND run_id IS ? AND state = 'active'
591            )
592                AND EXISTS (
593                    SELECT 1 FROM live_operation_claims WHERE owner_id = ?
594                )
595            RETURNING *
596            ",
597        )
598        .bind(owner_id)
599        .bind(run_id)
600        .bind(step)
601        .bind(now)
602        .bind(now)
603        .bind(detail)
604        .bind(owner_id)
605        .bind(run_id)
606        .bind(owner_id)
607        .fetch_optional(pool)
608        .await
609        .map_err(DbError::Query)?
610        .ok_or(DbError::LiveOwnershipLost)?;
611        row_to_live_mutation_step(row)
612    }
613
614    pub async fn list_for_owner(pool: &SqlitePool, owner_id: i64) -> DbResult<Vec<Self>> {
615        let rows = sqlx::query(
616            "SELECT * FROM live_mutation_steps WHERE owner_id = ? ORDER BY started_at, id",
617        )
618        .bind(owner_id)
619        .fetch_all(pool)
620        .await
621        .map_err(DbError::Query)?;
622        rows.into_iter().map(row_to_live_mutation_step).collect()
623    }
624
625    pub async fn list_for_run(pool: &SqlitePool, run_id: i64) -> DbResult<Vec<Self>> {
626        let rows = sqlx::query(
627            "SELECT * FROM live_mutation_steps WHERE run_id = ? ORDER BY started_at, id",
628        )
629        .bind(run_id)
630        .fetch_all(pool)
631        .await
632        .map_err(DbError::Query)?;
633        rows.into_iter().map(row_to_live_mutation_step).collect()
634    }
635}
636
637impl TuneRecoveryAttemptRow {
638    pub async fn start_with_claim(
639        pool: &SqlitePool,
640        new_owner: NewLiveOwnership,
641        claim_key: &str,
642        source_owner_id: i64,
643        run_id: i64,
644        export_path: &str,
645        evidence_json: &str,
646    ) -> DbResult<Option<(LiveOwnershipRow, Self)>> {
647        let mut transaction = pool.begin().await.map_err(DbError::Query)?;
648        let owner_row = sqlx::query(
649            r"
650            INSERT INTO live_operation_owners (
651                run_id, operation_kind, database_key, resource_key, owner_pid,
652                acquired_at, heartbeat_at, state, restore_intent_json
653            ) VALUES (?, ?, ?, ?, ?, ?, ?, 'active', ?)
654            RETURNING *
655            ",
656        )
657        .bind(new_owner.run_id)
658        .bind(enum_to_text(&new_owner.operation_kind)?)
659        .bind(&new_owner.database_key)
660        .bind(&new_owner.resource_key)
661        .bind(new_owner.owner_pid)
662        .bind(new_owner.acquired_at)
663        .bind(new_owner.acquired_at)
664        .bind(&new_owner.restore_intent_json)
665        .fetch_one(&mut *transaction)
666        .await
667        .map_err(DbError::Query)?;
668        let recovery_owner = row_to_live_ownership(owner_row)?;
669
670        let claim_result = sqlx::query(
671            r"
672            UPDATE live_operation_claims
673            SET owner_id = ?, claimed_at = ?
674            WHERE claim_key = ? AND owner_id = ?
675                AND EXISTS (
676                    SELECT 1 FROM live_operation_owners
677                    WHERE id = ? AND run_id = ? AND operation_kind = 'tune'
678                        AND state = 'orphaned' AND orphan_eligible = 1
679                        AND database_key = ? AND resource_key = ?
680                )
681            ",
682        )
683        .bind(recovery_owner.id)
684        .bind(new_owner.acquired_at)
685        .bind(claim_key)
686        .bind(source_owner_id)
687        .bind(source_owner_id)
688        .bind(run_id)
689        .bind(&new_owner.database_key)
690        .bind(&new_owner.resource_key)
691        .execute(&mut *transaction)
692        .await
693        .map_err(DbError::Query)?;
694        if claim_result.rows_affected() != 1 {
695            transaction.rollback().await.map_err(DbError::Query)?;
696            return Ok(None);
697        }
698
699        let attempt_row = sqlx::query(
700            r"
701            INSERT INTO tune_recovery_attempts (
702                run_id, source_owner_id, recovery_owner_id, export_path, started_at,
703                status, evidence_json
704            )
705            SELECT ?, ?, ?, ?, ?, 'running', ?
706            WHERE EXISTS (
707                SELECT 1 FROM live_operation_owners
708                WHERE id = ? AND run_id = ? AND operation_kind = 'tune'
709                    AND state = 'orphaned' AND orphan_eligible = 1
710            )
711                AND EXISTS (
712                    SELECT 1 FROM live_operation_owners
713                    WHERE id = ? AND run_id = ? AND operation_kind = 'recovery'
714                        AND state = 'active'
715                )
716                AND EXISTS (
717                    SELECT 1 FROM live_operation_claims
718                    WHERE owner_id = ? AND claim_key = ?
719                )
720            RETURNING *
721            ",
722        )
723        .bind(run_id)
724        .bind(source_owner_id)
725        .bind(recovery_owner.id)
726        .bind(export_path)
727        .bind(new_owner.acquired_at)
728        .bind(evidence_json)
729        .bind(source_owner_id)
730        .bind(run_id)
731        .bind(recovery_owner.id)
732        .bind(run_id)
733        .bind(recovery_owner.id)
734        .bind(claim_key)
735        .fetch_optional(&mut *transaction)
736        .await
737        .map_err(DbError::Query)?
738        .ok_or(DbError::LiveOwnershipLost)?;
739        let attempt = row_to_recovery_attempt(attempt_row)?;
740
741        let run_result = sqlx::query(
742            r"
743            UPDATE tune_runs
744            SET recovery_state = 'running', recovery_evidence_json = ?
745            WHERE id = ? AND outcome = 'failed'
746                AND recovery_state IN ('eligible', 'incomplete')
747                AND EXISTS (
748                    SELECT 1 FROM live_operation_owners
749                    WHERE id = ? AND run_id = ? AND state = 'orphaned'
750                        AND orphan_eligible = 1
751                )
752            ",
753        )
754        .bind(evidence_json)
755        .bind(run_id)
756        .bind(source_owner_id)
757        .bind(run_id)
758        .execute(&mut *transaction)
759        .await
760        .map_err(DbError::Query)?;
761        if run_result.rows_affected() != 1 {
762            transaction.rollback().await.map_err(DbError::Query)?;
763            return Err(DbError::LiveOwnershipLost);
764        }
765
766        transaction.commit().await.map_err(DbError::Query)?;
767        Ok(Some((recovery_owner, attempt)))
768    }
769
770    pub async fn list_for_run(pool: &SqlitePool, run_id: i64) -> DbResult<Vec<Self>> {
771        let rows = sqlx::query(
772            "SELECT * FROM tune_recovery_attempts WHERE run_id = ? ORDER BY started_at, id",
773        )
774        .bind(run_id)
775        .fetch_all(pool)
776        .await
777        .map_err(DbError::Query)?;
778        rows.into_iter().map(row_to_recovery_attempt).collect()
779    }
780
781    pub async fn complete_with_ownership(
782        pool: &SqlitePool,
783        update: CompleteRecoveryUpdate<'_>,
784    ) -> DbResult<()> {
785        let (restore_status, recovery_state, owner_state, eligible, released_at) = match update
786            .status
787        {
788            RecoveryAttemptStatus::Confirmed => (
789                Some("confirmed"),
790                "confirmed",
791                "recovered",
792                false,
793                Some(update.now),
794            ),
795            RecoveryAttemptStatus::Incomplete => (
796                Some("incomplete"),
797                "incomplete",
798                "orphaned",
799                true,
800                Some(update.now),
801            ),
802            RecoveryAttemptStatus::Failed => (None, "eligible", "orphaned", true, Some(update.now)),
803            RecoveryAttemptStatus::Running => return Err(DbError::LiveOwnershipLost),
804        };
805        let mut transaction = pool.begin().await.map_err(DbError::Query)?;
806        let attempt_result = sqlx::query(
807            r"
808            UPDATE tune_recovery_attempts
809            SET status = ?, completed_at = ?, detail = ?, evidence_json = ?
810            WHERE id = ? AND run_id = ? AND recovery_owner_id = ? AND status = 'running'
811            ",
812        )
813        .bind(enum_to_text(&update.status)?)
814        .bind(update.now)
815        .bind(update.detail)
816        .bind(update.evidence_json)
817        .bind(update.attempt_id)
818        .bind(update.run_id)
819        .bind(update.recovery_owner_id)
820        .execute(&mut *transaction)
821        .await
822        .map_err(DbError::Query)?;
823        if attempt_result.rows_affected() != 1 {
824            transaction.rollback().await.map_err(DbError::Query)?;
825            return Err(DbError::LiveOwnershipLost);
826        }
827
828        let run_result = sqlx::query(
829            r"
830            UPDATE tune_runs
831            SET restore_status = COALESCE(?, restore_status),
832                restore_detail = CASE WHEN ? IS NULL THEN restore_detail ELSE ? END,
833                recovery_state = ?,
834                recovery_evidence_json = ?
835            WHERE id = ? AND outcome = 'failed'
836                AND EXISTS (
837                    SELECT 1 FROM live_operation_owners
838                    WHERE id = ? AND run_id = ? AND state = 'orphaned'
839                        AND orphan_eligible = 1
840                )
841            ",
842        )
843        .bind(restore_status)
844        .bind(restore_status)
845        .bind(update.detail)
846        .bind(recovery_state)
847        .bind(update.evidence_json)
848        .bind(update.run_id)
849        .bind(update.source_owner_id)
850        .bind(update.run_id)
851        .execute(&mut *transaction)
852        .await
853        .map_err(DbError::Query)?;
854        if run_result.rows_affected() != 1 {
855            transaction.rollback().await.map_err(DbError::Query)?;
856            return Err(DbError::LiveOwnershipLost);
857        }
858
859        let recovery_owner_result = sqlx::query(
860            "UPDATE live_operation_owners SET state = 'released', released_at = ? WHERE id = ? AND state = 'active'",
861        )
862        .bind(update.now)
863        .bind(update.recovery_owner_id)
864        .execute(&mut *transaction)
865        .await
866        .map_err(DbError::Query)?;
867        if recovery_owner_result.rows_affected() != 1 {
868            transaction.rollback().await.map_err(DbError::Query)?;
869            return Err(DbError::LiveOwnershipLost);
870        }
871
872        let source_owner_result = sqlx::query(
873            r"
874            UPDATE live_operation_owners
875            SET state = ?, released_at = ?, orphan_eligible = ?
876            WHERE id = ? AND run_id = ? AND state = 'orphaned' AND orphan_eligible = 1
877            ",
878        )
879        .bind(owner_state)
880        .bind(released_at)
881        .bind(eligible)
882        .bind(update.source_owner_id)
883        .bind(update.run_id)
884        .execute(&mut *transaction)
885        .await
886        .map_err(DbError::Query)?;
887        if source_owner_result.rows_affected() != 1 {
888            transaction.rollback().await.map_err(DbError::Query)?;
889            return Err(DbError::LiveOwnershipLost);
890        }
891
892        let claim_result =
893            sqlx::query("DELETE FROM live_operation_claims WHERE claim_key = ? AND owner_id = ?")
894                .bind(update.claim_key)
895                .bind(update.recovery_owner_id)
896                .execute(&mut *transaction)
897                .await
898                .map_err(DbError::Query)?;
899        if claim_result.rows_affected() != 1 {
900            transaction.rollback().await.map_err(DbError::Query)?;
901            return Err(DbError::LiveOwnershipLost);
902        }
903        if update.status != RecoveryAttemptStatus::Confirmed {
904            sqlx::query(
905                "INSERT INTO live_operation_claims (claim_key, owner_id, claimed_at) VALUES (?, ?, ?)",
906            )
907            .bind(update.claim_key)
908            .bind(update.source_owner_id)
909            .bind(update.now)
910            .execute(&mut *transaction)
911            .await
912            .map_err(DbError::Query)?;
913        }
914        transaction.commit().await.map_err(DbError::Query)
915    }
916
917    /// Rolls back only the database ownership handoff for a recovery process proven dead by
918    /// the released OS lock and stale heartbeat. Controller state is never inspected or
919    /// mutated here; the original eligible owner regains its claim for an explicit retry.
920    pub async fn abandon_after_restart(
921        pool: &SqlitePool,
922        update: RestartRecoveryUpdate<'_>,
923    ) -> DbResult<bool> {
924        let mut transaction = pool.begin().await.map_err(DbError::Query)?;
925        let attempt_result = sqlx::query(
926            r"
927            UPDATE tune_recovery_attempts
928            SET status = 'incomplete', completed_at = ?, detail = ?, evidence_json = ?
929            WHERE id = ? AND run_id = ? AND recovery_owner_id = ? AND status = 'running'
930            ",
931        )
932        .bind(update.now)
933        .bind(update.detail)
934        .bind(update.evidence_json)
935        .bind(update.attempt_id)
936        .bind(update.run_id)
937        .bind(update.recovery_owner_id)
938        .execute(&mut *transaction)
939        .await
940        .map_err(DbError::Query)?;
941        if attempt_result.rows_affected() != 1 {
942            transaction.rollback().await.map_err(DbError::Query)?;
943            return Ok(false);
944        }
945
946        let run_result = sqlx::query(
947            r"
948            UPDATE tune_runs
949            SET restore_status = 'incomplete', restore_detail = ?,
950                recovery_state = 'incomplete', recovery_evidence_json = ?
951            WHERE id = ? AND outcome = 'failed' AND recovery_state = 'running'
952                AND EXISTS (
953                    SELECT 1 FROM live_operation_owners AS source
954                    WHERE source.id = ? AND source.run_id = ? AND source.operation_kind = 'tune'
955                        AND source.state = 'orphaned' AND source.orphan_eligible = 1
956                )
957                AND EXISTS (
958                    SELECT 1 FROM live_operation_owners AS recovery
959                    WHERE recovery.id = ? AND recovery.run_id = ?
960                        AND recovery.operation_kind = 'recovery' AND recovery.state = 'active'
961                )
962            ",
963        )
964        .bind(update.detail)
965        .bind(update.evidence_json)
966        .bind(update.run_id)
967        .bind(update.source_owner_id)
968        .bind(update.run_id)
969        .bind(update.recovery_owner_id)
970        .bind(update.run_id)
971        .execute(&mut *transaction)
972        .await
973        .map_err(DbError::Query)?;
974        if run_result.rows_affected() != 1 {
975            transaction.rollback().await.map_err(DbError::Query)?;
976            return Ok(false);
977        }
978
979        let recovery_owner_result = sqlx::query(
980            r"
981            UPDATE live_operation_owners
982            SET state = 'released', released_at = ?
983            WHERE id = ? AND run_id = ? AND operation_kind = 'recovery'
984                AND state = 'active' AND heartbeat_at = ?
985                AND EXISTS (
986                    SELECT 1 FROM live_operation_claims
987                    WHERE owner_id = live_operation_owners.id AND claim_key = ?
988                )
989            ",
990        )
991        .bind(update.now)
992        .bind(update.recovery_owner_id)
993        .bind(update.run_id)
994        .bind(update.expected_heartbeat)
995        .bind(update.claim_key)
996        .execute(&mut *transaction)
997        .await
998        .map_err(DbError::Query)?;
999        if recovery_owner_result.rows_affected() != 1 {
1000            transaction.rollback().await.map_err(DbError::Query)?;
1001            return Ok(false);
1002        }
1003
1004        let claim_result = sqlx::query(
1005            r"
1006            UPDATE live_operation_claims
1007            SET owner_id = ?, claimed_at = ?
1008            WHERE claim_key = ? AND owner_id = ?
1009                AND EXISTS (
1010                    SELECT 1 FROM live_operation_owners AS source
1011                    WHERE source.id = ? AND source.run_id = ? AND source.operation_kind = 'tune'
1012                        AND source.state = 'orphaned' AND source.orphan_eligible = 1
1013                )
1014            ",
1015        )
1016        .bind(update.source_owner_id)
1017        .bind(update.now)
1018        .bind(update.claim_key)
1019        .bind(update.recovery_owner_id)
1020        .bind(update.source_owner_id)
1021        .bind(update.run_id)
1022        .execute(&mut *transaction)
1023        .await
1024        .map_err(DbError::Query)?;
1025        if claim_result.rows_affected() != 1 {
1026            transaction.rollback().await.map_err(DbError::Query)?;
1027            return Ok(false);
1028        }
1029
1030        transaction.commit().await.map_err(DbError::Query)?;
1031        Ok(true)
1032    }
1033}
1034
1035fn row_to_live_ownership(row: SqliteRow) -> DbResult<LiveOwnershipRow> {
1036    let operation_kind: String = row.try_get("operation_kind").map_err(DbError::Query)?;
1037    let state: String = row.try_get("state").map_err(DbError::Query)?;
1038    Ok(LiveOwnershipRow {
1039        id: row.try_get("id").map_err(DbError::Query)?,
1040        run_id: row.try_get("run_id").map_err(DbError::Query)?,
1041        operation_kind: text_to_enum("operation_kind", &operation_kind)?,
1042        database_key: row.try_get("database_key").map_err(DbError::Query)?,
1043        resource_key: row.try_get("resource_key").map_err(DbError::Query)?,
1044        owner_pid: row.try_get("owner_pid").map_err(DbError::Query)?,
1045        acquired_at: row.try_get("acquired_at").map_err(DbError::Query)?,
1046        heartbeat_at: row.try_get("heartbeat_at").map_err(DbError::Query)?,
1047        released_at: row.try_get("released_at").map_err(DbError::Query)?,
1048        state: text_to_enum("state", &state)?,
1049        restore_intent_json: row.try_get("restore_intent_json").map_err(DbError::Query)?,
1050        orphan_eligible: row.try_get("orphan_eligible").map_err(DbError::Query)?,
1051        orphan_evidence_json: row
1052            .try_get("orphan_evidence_json")
1053            .map_err(DbError::Query)?,
1054    })
1055}
1056
1057fn row_to_live_mutation_step(row: SqliteRow) -> DbResult<LiveMutationStepRow> {
1058    let status: String = row.try_get("status").map_err(DbError::Query)?;
1059    Ok(LiveMutationStepRow {
1060        id: row.try_get("id").map_err(DbError::Query)?,
1061        owner_id: row.try_get("owner_id").map_err(DbError::Query)?,
1062        run_id: row.try_get("run_id").map_err(DbError::Query)?,
1063        step: row.try_get("step").map_err(DbError::Query)?,
1064        target_json: row.try_get("target_json").map_err(DbError::Query)?,
1065        previous_json: row.try_get("previous_json").map_err(DbError::Query)?,
1066        status: text_to_enum("status", &status)?,
1067        started_at: row.try_get("started_at").map_err(DbError::Query)?,
1068        completed_at: row.try_get("completed_at").map_err(DbError::Query)?,
1069        readback_json: row.try_get("readback_json").map_err(DbError::Query)?,
1070        detail: row.try_get("detail").map_err(DbError::Query)?,
1071    })
1072}
1073
1074fn row_to_recovery_attempt(row: SqliteRow) -> DbResult<TuneRecoveryAttemptRow> {
1075    let status: String = row.try_get("status").map_err(DbError::Query)?;
1076    Ok(TuneRecoveryAttemptRow {
1077        id: row.try_get("id").map_err(DbError::Query)?,
1078        run_id: row.try_get("run_id").map_err(DbError::Query)?,
1079        source_owner_id: row.try_get("source_owner_id").map_err(DbError::Query)?,
1080        recovery_owner_id: row.try_get("recovery_owner_id").map_err(DbError::Query)?,
1081        export_path: row.try_get("export_path").map_err(DbError::Query)?,
1082        started_at: row.try_get("started_at").map_err(DbError::Query)?,
1083        completed_at: row.try_get("completed_at").map_err(DbError::Query)?,
1084        status: text_to_enum("status", &status)?,
1085        detail: row.try_get("detail").map_err(DbError::Query)?,
1086        evidence_json: row.try_get("evidence_json").map_err(DbError::Query)?,
1087    })
1088}
1089
1090#[cfg(test)]
1091mod tests;
bhtune_db/models/live_ownership.rs

bhtune_db/models/
live_ownership.rs
