fn reset_temporary_condition_observations_after_restart( state: &AppState, ) -> Result { 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 cloud_reconnect_provider = state.providers.cloud().clone(); let cloud_reconnect_settings = state.settings.clone(); let cloud_reconnect_db = state.db.clone(); tokio::spawn(async move { crate::provider::cloud_reconnect_loop( cloud_reconnect_provider, cloud_reconnect_settings, cloud_reconnect_db, ) .await; }); let cloud_push_state = state.clone(); tokio::spawn(async move { cloud_push_loop(cloud_push_state).await; }); let ha_energy_state = state.clone(); tokio::spawn(async move { home_assistant_energy_loop(ha_energy_state).await; }); let local_connectivity_state = state.clone(); tokio::spawn(async move { local_connectivity_loop(local_connectivity_state).await; }); let cloud_connectivity_state = state.clone(); tokio::spawn(async move { cloud_connectivity_loop(cloud_connectivity_state).await; }); let control_plan_state = state.clone(); tokio::spawn(async move { control_plan_cache_loop(control_plan_state).await; }); 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"), } match maintenance_state .db .compact_network_history(compaction_days) { Ok(count) if count > 0 => { tracing::info!(count, "connectivity history samples compacted") } Ok(_) => {} Err(err) => tracing::warn!(error=?err, "cannot compact connectivity 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"), } match maintenance_state.db.prune_energy_readings(retention_days) { Ok(count) if count > 0 => { tracing::info!(count, retention_days, "old energy readings pruned") } Ok(_) => {} Err(err) => tracing::warn!(error=?err, "cannot prune energy readings"), } match maintenance_state.db.prune_network_readings(retention_days) { Ok(count) if count > 0 => { tracing::info!(count, retention_days, "old connectivity readings pruned") } Ok(_) => {} Err(err) => tracing::warn!(error=?err, "cannot prune connectivity 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 { 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; } } for _ in 0..50 { let energy = state.db.energy_before(cutoff, 1_000)?; if energy.is_empty() { break; } influxdb::write_energy_batch(&state.http, &settings, &energy).await?; let deleted = state.db.delete_energy_batch(&energy)?; moved += deleted; if deleted == 0 { break; } } for _ in 0..50 { let network = state.db.network_before(cutoff, 1_000)?; if network.is_empty() { break; } influxdb::write_network_batch(&state.http, &settings, &network).await?; let deleted = state.db.delete_network_batch(&network)?; moved += deleted; if deleted == 0 { break; } } Ok(moved) }