make scheduling mandatory

Author: passcod

README.md

@@ -86,7 +86,7 @@ Defines a continuously-refreshed replica of a PostgreSQL database restored from
 |-------|------|----------|---------|-------------|
 | `kopiaSecretRef` | `SecretReference` | Yes | — | Reference to a Secret containing Kopia repository credentials (`bucket`, `region`, `repositoryPassword`, `accessKeyId`, `secretAccessKey`). |
 | `snapshotFilter` | `SnapshotFilter` | No | — | Filter criteria to select which Kopia snapshot to restore. |
-| `schedule` | `string` | No | — | Cron expression controlling how often new restores are triggered. |
+| `schedule` | `string` | Yes | — | Cron expression controlling how often new restores are triggered. |
 | `scheduleJitter` | `string` | No | `"10m"` | Random jitter added to scheduled restores (friendly duration, e.g. `"5m"`, `"1h"`). |
 | `minimumTtl` | `string` | No | — | Don't restore a new snapshot within this duration of the last restore completing. |
 | `switchoverGracePeriod` | `string` | No | `"5m"` | How long to wait before deleting the old restore after a switchover. |

src/controllers/replica.rs

@@ -20,6 +20,7 @@ use crate::{
 	kopia,
 	types::*,
 };
+use scheduling::ScheduleDecision;
 
 mod resources;
 mod scheduling;
@@ -239,13 +240,6 @@ pub async fn reconcile(replica: Arc<PostgresPhysicalReplica>, ctx: Arc<Context>)
 			.send_notifications(client, &ctx.http_client, switching, &ctx.metrics)
 			.await;
 
-		// Recompute next scheduled restore in case schedule changed while restore was in flight
-		if let Some(next) = replica.compute_next_scheduled_restore(now) {
-			replica
-				.update_status_field(client, "nextScheduledRestore", Time(next))
-				.await?;
-		}
-
 		return Ok(Action::requeue(Duration::from_secs(10)));
 	}
 
@@ -402,22 +396,42 @@ pub async fn reconcile(replica: Arc<PostgresPhysicalReplica>, ctx: Arc<Context>)
 			.and_then(|s| s.last_restore_completed_at.as_ref())
 			.is_none();
 
+	// Ensure nextScheduledRestore is populated
+	if replica
+		.status
+		.as_ref()
+		.and_then(|s| s.next_scheduled_restore.as_ref())
+		.is_none()
+		&& let Some(next) = replica.compute_next_scheduled_restore(now)
+	{
+		replica
+			.update_status_field(client, "nextScheduledRestore", Time(next))
+			.await?;
+	}
+
 	if never_restored {
 		info!(
 			replica = name,
 			"no successful restore yet, triggering immediately"
 		);
 	}
 
-	let should_restore = never_restored || replica.should_trigger_scheduled_restore();
+	let schedule_decision = replica.check_schedule();
 
-	if should_restore {
-		if let Some(next) = replica.compute_next_scheduled_restore(now) {
-			replica
-				.update_status_field(client, "nextScheduledRestore", Time(next))
-				.await?;
-		}
+	let should_restore = never_restored || matches!(schedule_decision, ScheduleDecision::Trigger);
+
+	// Recompute nextScheduledRestore when a trigger fires (whether it proceeds or is skipped by TTL)
+	if matches!(
+		schedule_decision,
+		ScheduleDecision::Trigger | ScheduleDecision::SkippedByTtl
+	) && let Some(next) = replica.compute_next_scheduled_restore(now)
+	{
+		replica
+			.update_status_field(client, "nextScheduledRestore", Time(next))
+			.await?;
+	}
 
+	if should_restore {
 		// Check concurrent restore limit
 		let mut queue = ctx.restore_queue.write().await;
 		if !queue.can_start(ctx.max_concurrent_restores()) {

src/controllers/replica/scheduling.rs

@@ -8,9 +8,16 @@ use tracing::{debug, info, warn};
 
 use crate::types::*;
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ScheduleDecision {
+	NotDue,
+	Trigger,
+	SkippedByTtl,
+}
+
 impl PostgresPhysicalReplica {
 	pub fn compute_next_scheduled_restore(&self, now: Timestamp) -> Option<Timestamp> {
-		let schedule = self.spec.schedule.as_deref()?;
+		let schedule = &self.spec.schedule;
 
 		let Ok(cron) = parse_crontab_with(schedule, {
 			let mut options = ParseOptions::default();
@@ -67,60 +74,250 @@ impl PostgresPhysicalReplica {
 		Some(next.into())
 	}
 
-	pub fn should_trigger_scheduled_restore(&self) -> bool {
+	pub fn check_schedule(&self) -> ScheduleDecision {
 		let name = self.name_any();
-		let Some(schedule) = &self.spec.schedule else {
-			debug!(replica = %name, "no schedule configured, skipping scheduled restore");
-			return false;
+		let schedule = &self.spec.schedule;
+		let status = self.status.as_ref();
+		let now = Timestamp::now().to_zoned(TimeZone::UTC);
+
+		let Some(next_scheduled) = status.and_then(|s| s.next_scheduled_restore.as_ref()) else {
+			info!(
+				replica = %name,
+				schedule = schedule,
+				"no nextScheduledRestore set, triggering"
+			);
+			return ScheduleDecision::Trigger;
 		};
 
-		let status = self.status.as_ref();
+		if now < next_scheduled.0.to_zoned(TimeZone::UTC) {
+			debug!(
+				replica = %name,
+				next_scheduled = %next_scheduled.0,
+				"scheduled restore not due yet, skipping"
+			);
+			return ScheduleDecision::NotDue;
+		}
 
-		let now = Timestamp::now().to_zoned(TimeZone::UTC);
+		info!(
+			replica = %name,
+			next_scheduled = %next_scheduled.0,
+			"scheduled restore time reached"
+		);
 
-		// Check minimumTTL (only if configured)
+		// Check minimumTTL: prevent restoring too soon after the last one
 		if let Some(ref minimum_ttl) = self.spec.minimum_ttl
 			&& let Some(last_completed) = status.and_then(|s| s.last_restore_completed_at.as_ref())
 		{
 			let not_before = last_completed
 				.0
 				.to_zoned(TimeZone::UTC)
 				.saturating_add(minimum_ttl.0);
-			if not_before < now {
-				debug!(
+			if now < not_before {
+				info!(
 					replica = %name,
 					last_completed = %last_completed.0,
 					%minimum_ttl,
 					%not_before,
 					"last restore completed within minimum TTL, skipping"
 				);
-				return false;
+				return ScheduleDecision::SkippedByTtl;
 			}
 		}
 
-		if let Some(next_scheduled) = status.and_then(|s| s.next_scheduled_restore.as_ref()) {
-			if now >= next_scheduled.0.to_zoned(TimeZone::UTC) {
-				info!(
-					replica = %name,
-					next_scheduled = %next_scheduled.0,
-					"scheduled restore time reached, triggering"
-				);
-			} else {
-				debug!(
-					replica = %name,
-					next_scheduled = %next_scheduled.0,
-					"scheduled restore not due yet, skipping"
-				);
-				return false;
-			}
-		} else {
-			info!(
-				replica = %name,
-				schedule = schedule,
-				"no nextScheduledRestore set, triggering"
-			);
+		ScheduleDecision::Trigger
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use std::collections::HashMap;
+
+	use jiff::Span;
+	use k8s_openapi::{api::core::v1::SecretReference, apimachinery::pkg::apis::meta::v1::Time};
+	use kube::api::ObjectMeta;
+
+	use crate::util::TimeSpan;
+
+	use super::*;
+
+	fn make_replica(
+		schedule: &str,
+		next_scheduled: Option<Timestamp>,
+		last_completed: Option<Timestamp>,
+		minimum_ttl: Option<TimeSpan>,
+	) -> PostgresPhysicalReplica {
+		PostgresPhysicalReplica {
+			metadata: ObjectMeta {
+				name: Some("test-replica".into()),
+				namespace: Some("default".into()),
+				uid: Some("test-uid-123".into()),
+				..Default::default()
+			},
+			spec: PostgresPhysicalReplicaSpec {
+				kopia_secret_ref: SecretReference {
+					name: Some("test-secret".into()),
+					namespace: None,
+				},
+				snapshot_filter: None,
+				schedule: schedule.into(),
+				schedule_jitter: TimeSpan(Span::new().seconds(0)),
+				minimum_ttl,
+				switchover_grace_period: TimeSpan(Span::new().minutes(5)),
+				analytics_username: "analytics".into(),
+				storage_class: None,
+				storage_size_override: None,
+				resources: None,
+				service_annotations: None,
+				pod_annotations: None,
+				affinity: None,
+				tolerations: Vec::new(),
+				read_only: true,
+				postgres_extra_config: None,
+				notifications: Vec::new(),
+				overlay_database: None,
+			},
+			status: Some(PostgresPhysicalReplicaStatus {
+				next_scheduled_restore: next_scheduled.map(Time),
+				last_restore_completed_at: last_completed.map(Time),
+				..Default::default()
+			}),
 		}
+	}
+
+	#[test]
+	fn check_schedule_no_next_scheduled_triggers() {
+		let replica = make_replica("0 */6 * * *", None, None, None);
+		assert_eq!(replica.check_schedule(), ScheduleDecision::Trigger);
+	}
+
+	#[test]
+	fn check_schedule_not_due_yet() {
+		let future = Timestamp::now()
+			.to_zoned(jiff::tz::TimeZone::UTC)
+			.saturating_add(Span::new().hours(3))
+			.timestamp();
+		let replica = make_replica("0 */6 * * *", Some(future), None, None);
+		assert_eq!(replica.check_schedule(), ScheduleDecision::NotDue);
+	}
+
+	#[test]
+	fn check_schedule_due_triggers() {
+		let past = Timestamp::now()
+			.to_zoned(jiff::tz::TimeZone::UTC)
+			.saturating_add(Span::new().hours(-1))
+			.timestamp();
+		let replica = make_replica("0 */6 * * *", Some(past), None, None);
+		assert_eq!(replica.check_schedule(), ScheduleDecision::Trigger);
+	}
+
+	#[test]
+	fn check_schedule_due_but_within_ttl_skips() {
+		let past = Timestamp::now()
+			.to_zoned(jiff::tz::TimeZone::UTC)
+			.saturating_add(Span::new().hours(-1))
+			.timestamp();
+		// Last restore completed 30 minutes ago, TTL is 2 hours
+		let last_completed = Timestamp::now()
+			.to_zoned(jiff::tz::TimeZone::UTC)
+			.saturating_add(Span::new().minutes(-30))
+			.timestamp();
+		let ttl = TimeSpan(Span::new().hours(2));
+		let replica = make_replica("0 */6 * * *", Some(past), Some(last_completed), Some(ttl));
+		assert_eq!(replica.check_schedule(), ScheduleDecision::SkippedByTtl);
+	}
+
+	#[test]
+	fn check_schedule_due_and_past_ttl_triggers() {
+		let past = Timestamp::now()
+			.to_zoned(jiff::tz::TimeZone::UTC)
+			.saturating_add(Span::new().hours(-1))
+			.timestamp();
+		// Last restore completed 3 hours ago, TTL is 2 hours
+		let last_completed = Timestamp::now()
+			.to_zoned(jiff::tz::TimeZone::UTC)
+			.saturating_add(Span::new().hours(-3))
+			.timestamp();
+		let ttl = TimeSpan(Span::new().hours(2));
+		let replica = make_replica("0 */6 * * *", Some(past), Some(last_completed), Some(ttl));
+		assert_eq!(replica.check_schedule(), ScheduleDecision::Trigger);
+	}
+
+	#[test]
+	fn check_schedule_no_ttl_configured_triggers() {
+		let past = Timestamp::now()
+			.to_zoned(jiff::tz::TimeZone::UTC)
+			.saturating_add(Span::new().hours(-1))
+			.timestamp();
+		// Last restore completed recently but no TTL configured
+		let last_completed = Timestamp::now()
+			.to_zoned(jiff::tz::TimeZone::UTC)
+			.saturating_add(Span::new().minutes(-10))
+			.timestamp();
+		let replica = make_replica("0 */6 * * *", Some(past), Some(last_completed), None);
+		assert_eq!(replica.check_schedule(), ScheduleDecision::Trigger);
+	}
+
+	#[test]
+	fn compute_next_scheduled_restore_valid_cron() {
+		let replica = make_replica("0 */6 * * *", None, None, None);
+		let now = Timestamp::now();
+		let next = replica.compute_next_scheduled_restore(now);
+		assert!(next.is_some());
+		assert!(next.unwrap() > now);
+	}
+
+	#[test]
+	fn compute_next_scheduled_restore_invalid_cron() {
+		let replica = make_replica("not a cron", None, None, None);
+		let now = Timestamp::now();
+		assert!(replica.compute_next_scheduled_restore(now).is_none());
+	}
+
+	#[test]
+	fn compute_next_scheduled_restore_with_jitter() {
+		let mut replica = make_replica("0 */6 * * *", None, None, None);
+		replica.spec.schedule_jitter = TimeSpan(Span::new().minutes(30));
+		let now = Timestamp::now();
+
+		let mut results: Vec<Timestamp> = (0..10)
+			.filter_map(|_| replica.compute_next_scheduled_restore(now))
+			.collect();
+		results.sort();
+		results.dedup();
+
+		// With 30 minutes of jitter, repeated calls should produce varying results
+		assert!(
+			results.len() > 1,
+			"jitter should produce different results across calls"
+		);
+	}
+
+	#[test]
+	fn check_schedule_ttl_with_no_previous_restore_triggers() {
+		let past = Timestamp::now()
+			.to_zoned(jiff::tz::TimeZone::UTC)
+			.saturating_add(Span::new().hours(-1))
+			.timestamp();
+		// TTL configured but no last_completed timestamp (first restore)
+		let ttl = TimeSpan(Span::new().hours(2));
+		let replica = make_replica("0 */6 * * *", Some(past), None, Some(ttl));
+		assert_eq!(replica.check_schedule(), ScheduleDecision::Trigger);
+	}
 
-		true
+	#[test]
+	fn check_schedule_uses_tags_for_filtering() {
+		// Verify that check_schedule works with snapshot filter tags
+		// (tags are used elsewhere but shouldn't affect scheduling)
+		let past = Timestamp::now()
+			.to_zoned(jiff::tz::TimeZone::UTC)
+			.saturating_add(Span::new().hours(-1))
+			.timestamp();
+		let mut replica = make_replica("0 */6 * * *", Some(past), None, None);
+		replica.spec.snapshot_filter = Some(SnapshotFilter {
+			tags: Some(HashMap::from([("env".into(), "prod".into())])),
+			host_pattern: None,
+			description_pattern: None,
+		});
+		assert_eq!(replica.check_schedule(), ScheduleDecision::Trigger);
 	}
 }

src/controllers/restore/tests.rs

@@ -29,7 +29,7 @@ fn make_replica_with_opts(
 				namespace: None,
 			},
 			snapshot_filter: None,
-			schedule: None,
+			schedule: "0 */6 * * *".into(),
 			schedule_jitter: TimeSpan(Span::new().minutes(10)),
 			minimum_ttl: None,
 			switchover_grace_period: TimeSpan(Span::new().minutes(5)),