fix(machine-validation): address machine validation stale review feedback (#2911)

Addresses the review feedback on the machine validation stale-run
changes (#2838).

 Summary

  - Simplified stale timeout clamping logic.
- Kept the minimum stale timeout enforcement so healthy runs do not go
stale between heartbeats.
  - Added the stale reconciliation error message to logs.
- Added focused DB/API test coverage for stale-run detection and
reconciliation.
  

## Related issues
Fixes #454

## Type of Change
<!-- Check one that best describes this PR -->
- [ ] **Add** - New feature or capability
- [ ] **Change** - Changes in existing functionality
- [x] **Fix** - Bug fixes
- [ ] **Remove** - Removed features or deprecated functionality
- [ ] **Internal** - Internal changes (refactoring, tests, docs, etc.)

## Breaking Changes
<!-- If checked, describe the breaking changes and migration steps -->
<!-- Breaking changes are not generally permitted, please discuss on a
GitHub discussion or with the development team if you believe you need
to break a backward compatibility guarantee -->
- [ ] **This PR contains breaking changes**

## Testing
<!-- How was this tested? Check all that apply -->
- [x] Unit tests added/updated
- [x] Integration tests added/updated
- [ ] Manual testing performed
- [ ] No testing required (docs, internal refactor, etc.)

## Additional Notes

Author: sunilkumar-nvidia

crates/api-core/src/machine_validation/mod.rs

@@ -45,18 +45,17 @@ pub struct MachineValidationManager {
 impl MachineValidationManager {
     pub fn new(
         database_connection: sqlx::PgPool,
-        config: MachineValidationConfig,
+        mut config: MachineValidationConfig,
         meter: opentelemetry::metrics::Meter,
     ) -> Self {
-        let configured_stale_run_timeout = config.stale_run_timeout;
-        let config = config.with_minimum_stale_run_timeout();
-        if config.stale_run_timeout != configured_stale_run_timeout {
+        if config.stale_run_timeout < MachineValidationConfig::MIN_STALE_RUN_TIMEOUT {
             tracing::warn!(
-                configured_stale_run_timeout_seconds = configured_stale_run_timeout.as_secs(),
+                configured_stale_run_timeout_seconds = config.stale_run_timeout.as_secs(),
                 minimum_stale_run_timeout_seconds =
                     MachineValidationConfig::MIN_STALE_RUN_TIMEOUT.as_secs(),
                 "machine validation stale_run_timeout is below minimum; using minimum"
             );
+            config.stale_run_timeout = MachineValidationConfig::MIN_STALE_RUN_TIMEOUT;
         }
 
         let hold_period = config
@@ -399,14 +398,15 @@ async fn reconcile_stale_validation(
     record_failed_validation_side_effects(
         txn,
         &validation,
-        error_message,
+        error_message.clone(),
         "StaleMachineValidationRun",
     )
     .await?;
 
     tracing::warn!(
         validation_id = %validation.id,
         machine_id = %validation.machine_id,
+        error = %error_message,
         "reconciled stale machine validation run"
     );
 

crates/api-db/src/machine_validation.rs

@@ -368,3 +368,122 @@ pub async fn mark_machine_validation_complete(
 
     Ok(true)
 }
+
+#[cfg(test)]
+mod tests {
+    use std::str::FromStr;
+
+    use super::*;
+
+    fn test_machine_id() -> MachineId {
+        MachineId::from_str("fm100htes3rn1npvbtm5qd57dkilaag7ljugl1llmm7rfuq1ov50i0rpl30").unwrap()
+    }
+
+    async fn insert_active_validation(
+        txn: &mut PgConnection,
+        start_time: chrono::DateTime<chrono::Utc>,
+        duration_to_complete: i64,
+        last_heartbeat_at: Option<chrono::DateTime<chrono::Utc>>,
+    ) -> DatabaseResult<MachineValidationId> {
+        let id = MachineValidationId::new();
+        const QUERY: &str = "
+            INSERT INTO machine_validation (
+                id,
+                machine_id,
+                start_time,
+                name,
+                end_time,
+                context,
+                total,
+                completed,
+                state,
+                duration_to_complete,
+                last_heartbeat_at
+            )
+            VALUES ($1, $2, $3, $4, NULL, $5, 1, 0, $6, $7, $8)";
+
+        sqlx::query(QUERY)
+            .bind(id)
+            .bind(test_machine_id())
+            .bind(start_time)
+            .bind(format!("Test_{id}"))
+            .bind("OnDemand")
+            .bind(MachineValidationState::InProgress.to_string())
+            .bind(duration_to_complete)
+            .bind(last_heartbeat_at)
+            .execute(txn)
+            .await
+            .map_err(|e| DatabaseError::query(QUERY, e))?;
+
+        Ok(id)
+    }
+
+    #[crate::sqlx_test]
+    async fn mark_stale_if_active_uses_heartbeat_when_present(
+        pool: sqlx::PgPool,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let mut txn = pool.begin().await?;
+        let now = chrono::Utc::now();
+        let stale_run_timeout = std::time::Duration::from_secs(60);
+        let status = MachineValidationStatus {
+            state: MachineValidationState::Failed,
+            ..MachineValidationStatus::default()
+        };
+
+        let fresh_heartbeat = insert_active_validation(
+            txn.as_mut(),
+            now - chrono::Duration::minutes(10),
+            1,
+            Some(now - chrono::Duration::seconds(30)),
+        )
+        .await?;
+        let stale_heartbeat = insert_active_validation(
+            txn.as_mut(),
+            now - chrono::Duration::seconds(30),
+            1,
+            Some(now - chrono::Duration::seconds(61)),
+        )
+        .await?;
+        let stale_without_heartbeat =
+            insert_active_validation(txn.as_mut(), now - chrono::Duration::seconds(120), 1, None)
+                .await?;
+
+        assert!(
+            mark_stale_if_active(
+                txn.as_mut(),
+                &fresh_heartbeat,
+                stale_run_timeout,
+                now,
+                &status,
+            )
+            .await?
+            .is_none()
+        );
+        assert_eq!(
+            mark_stale_if_active(
+                txn.as_mut(),
+                &stale_heartbeat,
+                stale_run_timeout,
+                now,
+                &status,
+            )
+            .await?
+            .map(|validation| validation.id),
+            Some(stale_heartbeat)
+        );
+        assert_eq!(
+            mark_stale_if_active(
+                txn.as_mut(),
+                &stale_without_heartbeat,
+                stale_run_timeout,
+                now,
+                &status,
+            )
+            .await?
+            .map(|validation| validation.id),
+            Some(stale_without_heartbeat)
+        );
+
+        Ok(())
+    }
+}

crates/api-db/src/machine_validation_execution.rs

@@ -845,3 +845,187 @@ fn truncate_summary(value: &str) -> Option<String> {
         Some(value.chars().take(SUMMARY_LIMIT).collect())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeSet;
+    use std::str::FromStr;
+
+    use carbide_uuid::machine::MachineId;
+
+    use super::*;
+
+    fn test_machine_id() -> MachineId {
+        MachineId::from_str("fm100htes3rn1npvbtm5qd57dkilaag7ljugl1llmm7rfuq1ov50i0rpl30").unwrap()
+    }
+
+    async fn insert_active_validation(
+        txn: &mut PgConnection,
+        now: chrono::DateTime<chrono::Utc>,
+    ) -> DatabaseResult<MachineValidationId> {
+        let id = MachineValidationId::new();
+        const QUERY: &str = "
+            INSERT INTO machine_validation (
+                id,
+                machine_id,
+                start_time,
+                name,
+                end_time,
+                context,
+                total,
+                completed,
+                state,
+                duration_to_complete,
+                last_heartbeat_at
+            )
+            VALUES ($1, $2, $3, $4, NULL, $5, 1, 0, $6, 0, NULL)";
+
+        sqlx::query(QUERY)
+            .bind(id)
+            .bind(test_machine_id())
+            .bind(now)
+            .bind(format!("Test_{id}"))
+            .bind("OnDemand")
+            .bind("InProgress")
+            .execute(txn)
+            .await
+            .map_err(|e| DatabaseError::query(QUERY, e))?;
+
+        Ok(id)
+    }
+
+    async fn insert_attempt(
+        txn: &mut PgConnection,
+        validation_id: &MachineValidationId,
+        test_id: &str,
+        order_index: i32,
+        state: MachineValidationAttemptState,
+        started_at: Option<chrono::DateTime<chrono::Utc>>,
+        last_heartbeat_at: Option<chrono::DateTime<chrono::Utc>>,
+    ) -> DatabaseResult<MachineValidationAttemptId> {
+        let run_item_id = MachineValidationRunItemId::new();
+        let attempt_id = MachineValidationAttemptId::new();
+        const RUN_ITEM_QUERY: &str = "
+            INSERT INTO machine_validation_run_items (
+                id,
+                run_id,
+                test_id,
+                display_name,
+                context,
+                state,
+                order_index,
+                attempt,
+                max_attempts,
+                timeout_seconds,
+                started_at,
+                last_heartbeat_at
+            )
+            VALUES ($1, $2, $3, $4, $5, $6, $7, 1, 1, 1, $8, $9)";
+
+        sqlx::query(RUN_ITEM_QUERY)
+            .bind(run_item_id)
+            .bind(validation_id)
+            .bind(test_id)
+            .bind(test_id)
+            .bind("OnDemand")
+            .bind(MachineValidationRunItemState::Running.to_string())
+            .bind(order_index)
+            .bind(started_at)
+            .bind(last_heartbeat_at)
+            .execute(&mut *txn)
+            .await
+            .map_err(|e| DatabaseError::query(RUN_ITEM_QUERY, e))?;
+
+        const ATTEMPT_QUERY: &str = "
+            INSERT INTO machine_validation_attempts (
+                id,
+                run_item_id,
+                attempt_number,
+                state,
+                command,
+                args,
+                started_at,
+                last_heartbeat_at
+            )
+            VALUES ($1, $2, 1, $3, $4, $5, $6, $7)";
+
+        sqlx::query(ATTEMPT_QUERY)
+            .bind(attempt_id)
+            .bind(run_item_id)
+            .bind(state.to_string())
+            .bind("echo")
+            .bind("ok")
+            .bind(started_at)
+            .bind(last_heartbeat_at)
+            .execute(txn)
+            .await
+            .map_err(|e| DatabaseError::query(ATTEMPT_QUERY, e))?;
+
+        Ok(attempt_id)
+    }
+
+    #[crate::sqlx_test]
+    async fn find_stale_active_attempts_respects_heartbeat_and_duration_fallback(
+        pool: sqlx::PgPool,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let mut txn = pool.begin().await?;
+        let now = chrono::Utc::now();
+        let validation_id = insert_active_validation(txn.as_mut(), now).await?;
+
+        insert_attempt(
+            txn.as_mut(),
+            &validation_id,
+            "stale-heartbeat",
+            0,
+            MachineValidationAttemptState::Running,
+            Some(now - chrono::Duration::seconds(10)),
+            Some(now - chrono::Duration::seconds(61)),
+        )
+        .await?;
+        let _fresh_heartbeat_attempt = insert_attempt(
+            txn.as_mut(),
+            &validation_id,
+            "fresh-heartbeat",
+            1,
+            MachineValidationAttemptState::Running,
+            Some(now - chrono::Duration::seconds(10)),
+            Some(now - chrono::Duration::seconds(30)),
+        )
+        .await?;
+        insert_attempt(
+            txn.as_mut(),
+            &validation_id,
+            "legacy-stale",
+            2,
+            MachineValidationAttemptState::Running,
+            Some(now - chrono::Duration::seconds(120)),
+            None,
+        )
+        .await?;
+        let _terminal_attempt = insert_attempt(
+            txn.as_mut(),
+            &validation_id,
+            "terminal",
+            3,
+            MachineValidationAttemptState::Failed,
+            Some(now - chrono::Duration::seconds(120)),
+            Some(now - chrono::Duration::seconds(120)),
+        )
+        .await?;
+
+        let stale_attempts =
+            find_stale_active_attempts(txn.as_mut(), std::time::Duration::from_secs(60), now)
+                .await?;
+        let stale_test_ids = stale_attempts
+            .iter()
+            .map(|attempt| attempt.test_id.as_str())
+            .collect::<BTreeSet<_>>();
+
+        assert_eq!(
+            stale_test_ids,
+            BTreeSet::from(["legacy-stale", "stale-heartbeat"])
+        );
+
+        Ok(())
+    }
+}