v0.8.14

src/engine/runtime.rs

@@ -0,0 +1,128 @@
+fn reset_temporary_condition_observations_after_restart(state: &AppState) -> Result<usize, AppError> {
+    let mut changed = 0usize;
+    for mut zone in state.db.list_zones()? {
+        let Some(session) = zone.temporary_quick_thermostat.as_mut() else { continue; };
+        if session.condition_started_at.is_none() && session.condition_last_observed_at.is_none() { continue; }
+        session.condition_started_at = None;
+        session.condition_last_observed_at = None;
+        zone.updated_at = Utc::now();
+        state.db.save_zone(&zone)?;
+        changed += 1;
+    }
+    Ok(changed)
+}
+
+pub fn start(state: AppState) {
+    // A continuous temperature hold cannot span controller downtime. Preserve the session
+    // itself, but require fresh observations after every process restart (H11).
+    if let Err(err) = reset_temporary_condition_observations_after_restart(&state) {
+        tracing::warn!(error=?err, "cannot reset temporary thermostat observation continuity after restart");
+    }
+    let poll_state = state.clone();
+    tokio::spawn(async move {
+        sleep(Duration::from_millis(500)).await;
+        loop {
+            match poll_all(&poll_state).await {
+                Ok(()) => {
+                    if !poll_state.initial_device_sync_complete.swap(true, Ordering::AcqRel) {
+                        tracing::info!("initial device state synchronized; thermostat control enabled");
+                    }
+                }
+                Err(err) => tracing::error!(error=?err, "device poll cycle failed"),
+            }
+            let seconds = poll_state.settings.read().await.poll_interval_seconds.max(2);
+            sleep(Duration::from_secs(seconds)).await;
+        }
+    });
+
+    let control_state = state.clone();
+    tokio::spawn(async move {
+        sleep(Duration::from_secs(2)).await;
+        loop {
+            // A restart must never make decisions from the persisted, potentially stale
+            // device snapshot. Wait for one full live poll before thermostat/schedule/automation
+            // ownership can emit commands. Manual API/remote control remains available.
+            if !control_state.initial_device_sync_complete.load(Ordering::Acquire) {
+                sleep(Duration::from_millis(250)).await;
+                continue;
+            }
+            if let Err(err) = control_zones(&control_state).await {
+                tracing::error!(error=?err, "zone cycle failed");
+            }
+            if let Err(err) = run_automations(&control_state).await {
+                tracing::error!(error=?err, "automation cycle failed");
+            }
+            let seconds = control_state.settings.read().await.zone_interval_seconds.max(2);
+            let normal_delay = Duration::from_secs(seconds);
+            let resume_delay = match next_zone_control_deadline_delay(&control_state) {
+                Ok(value) => value,
+                Err(err) => {
+                    tracing::warn!(error=?err, "cannot calculate thermostat control deadline");
+                    None
+                }
+            };
+            let sleep_for = resume_delay.map(|delay| delay.min(normal_delay)).unwrap_or(normal_delay);
+            tokio::select! {
+                _ = sleep(sleep_for) => {},
+                _ = control_state.zone_control_wakeup.notified() => {},
+            }
+        }
+    });
+
+    let maintenance_state = state;
+    tokio::spawn(async move {
+        sleep(Duration::from_secs(60)).await;
+        loop {
+            let settings = maintenance_state.settings.read().await.clone();
+            // When InfluxDB is enabled, compact all locally retained legacy history before
+            // transferring old buckets. Without Influx, compact only the configured retention window.
+            let compaction_days = if settings.influxdb.enabled { 3650 } else { settings.history_retention_days.max(1) } as i64;
+            if settings.history_compaction_enabled {
+                match maintenance_state.db.compact_history(compaction_days) {
+                    Ok(count) if count > 0 => tracing::info!(count, "history samples compacted"),
+                    Ok(_) => {}
+                    Err(err) => tracing::warn!(error=?err, "cannot compact history"),
+                }
+            }
+            if settings.influxdb.enabled {
+                match archive_old_history(&maintenance_state, settings.influxdb.history_threshold_days.max(1)).await {
+                    Ok(count) if count > 0 => tracing::info!(count, "old local readings archived to InfluxDB and removed from SQLite"),
+                    Ok(_) => {}
+                    Err(err) => tracing::warn!(error=?err, "cannot archive old history to InfluxDB; SQLite copies were kept"),
+                }
+            } else {
+                let retention_days = settings.history_retention_days.max(1) as i64;
+                match maintenance_state.db.prune_readings(retention_days) {
+                    Ok(count) if count > 0 => tracing::info!(count, retention_days, "old local readings pruned"),
+                    Ok(_) => {}
+                    Err(err) => tracing::warn!(error=?err, "cannot prune readings"),
+                }
+            }
+            let event_retention_days = settings.event_log_retention_days.max(1) as i64;
+            match maintenance_state.db.prune_events(event_retention_days) {
+                Ok(count) if count > 0 => tracing::info!(count, event_retention_days, "old event log rows pruned"),
+                Ok(_) => {}
+                Err(err) => tracing::warn!(error=?err, "cannot prune event log"),
+            }
+            sleep(Duration::from_secs(6 * 60 * 60)).await;
+        }
+    });
+}
+
+async fn archive_old_history(state: &AppState, threshold_days: u32) -> Result<u64> {
+    let cutoff = Utc::now() - chrono::Duration::days(threshold_days.max(1) as i64);
+    let settings = state.settings.read().await.influxdb.clone();
+    let mut moved = 0_u64;
+    // Bound one maintenance pass so a very large legacy database never monopolizes the runtime.
+    // Successful batches are deleted from SQLite, so the next pass naturally continues forward.
+    for _ in 0..50 {
+        let (devices, zones, ha) = state.db.history_before(cutoff, 1_000)?;
+        if devices.is_empty() && zones.is_empty() && ha.is_empty() { break; }
+        influxdb::write_batch(&state.http, &settings, &devices, &zones, &ha).await?;
+        let deleted = state.db.delete_history_batch(&devices, &zones, &ha)?;
+        moved += deleted;
+        if deleted == 0 { break; }
+    }
+    Ok(moved)
+}
+