feat(canopy): tear down ephemeral (verify) replicas after verification

.github/workflows/integration.yml

@@ -56,6 +56,11 @@ jobs:
               test-ps-all-missing
             needs_non_pg_snapshot: false
 
+          - name: ephemeral
+            namespaces: >-
+              test-ephemeral
+            needs_non_pg_snapshot: false
+
           # canopy_integration is a scaffold in tests/canopy_integration.rs
           # but no stub-canopy HTTP server exists yet, so the happy-path
           # test times out. Re-add this matrix entry when the stub lands:

README.md

@@ -103,6 +103,7 @@ Defines a continuously-refreshed replica of a PostgreSQL database restored from
 | `affinity` | `Affinity` | No | — | Pod scheduling affinity rules. |
 | `tolerations` | `[]Toleration` | No | `[]` | Pod tolerations. |
 | `readOnly` | `bool` | No | `true` | Set the restored database to read-only mode. |
+| `ephemeral` | `bool` | No | `false` | Tear the restore down once it reaches `Active` (postgres came up healthy) instead of keeping it running. The replica only restores again when a newer snapshot is offered (canopy path) or the schedule next fires (legacy path). Used by the `verify` intent, whose job is just to prove the snapshot restores. |
 | `postgresExtraConfig` | `string` | No | — | Extra lines appended to `postgresql.conf` (e.g. `shared_preload_libraries`). |
 | `notifications` | `[]NotificationConfig` | No | `[]` | Notification targets called on restore events. |
 | `persistentSchemas` | `[]string` | No | — | List of schema names to migrate from the previous restore to the new restore on each switchover. See [Persistent schemas](#persistent-schemas) below for the migration time budget and what happens on timeout. |
@@ -180,6 +181,7 @@ Additional fields for `target: graphQL`:
 | `nextScheduledRestore` | `Time` | When the next scheduled restore will occur. |
 | `latestAvailableSnapshot` | `string` | Snapshot ID of the latest available snapshot matching the filter. |
 | `canopyDesiredSnapshotId` | `string` | For canopy-sourced replicas: the snapshot the canopy worklist syncer wants restored. The reconciler triggers a new restore when this differs from the current one. |
+| `verifiedSnapshotId` | `string` | For `ephemeral` replicas: the last snapshot that was verified and then torn down. Gates re-restore — the reconciler only restores again when the desired snapshot differs from this. |
 | `connectionInfo` | `ConnectionInfo` | Connection details (host, port, database, username, password secret). |
 | `queuePosition` | `uint32` | Position in the global restore queue. |
 | `notifications` | `[]NotificationStatus` | Status of each configured notification target. |

crds.yaml

@@ -571,6 +571,17 @@ spec:
                 - group
                 - type
                 type: object
+              ephemeral:
+                default: false
+                description: |-
+                  Ephemeral replica: once a restore reaches `Active` (postgres came up
+                  healthy and, for canopy replicas, the verification was reported),
+                  tear the restore down instead of keeping it running. The replica CR
+                  stays; it only restores again when a new snapshot is offered (canopy
+                  path) or the schedule next fires (legacy path). Used by the `verify`
+                  intent, whose whole job is "prove the snapshot restores" — keeping
+                  the database idling afterward just wastes cluster resources.
+                type: boolean
               kopiaSecretRef:
                 description: |-
                   Reference to a Secret containing kopia repository credentials.
@@ -934,6 +945,15 @@ spec:
               serviceName:
                 nullable: true
                 type: string
+              verifiedSnapshotId:
+                description: |-
+                  Last snapshot id an ephemeral replica (`spec.ephemeral`) verified
+                  and then tore down. After teardown there is no `currentRestore` to
+                  compare against, so this marker is what stops the reconciler from
+                  immediately re-restoring the same snapshot; a restore is only
+                  re-triggered when the desired snapshot differs from this.
+                nullable: true
+                type: string
             type: object
         required:
         - spec

src/controllers/canopy/intent.rs

@@ -41,6 +41,11 @@ pub struct IntentConfig {
 	pub service_annotations: Option<BTreeMap<String, String>>,
 	pub switchover_grace_period: TimeSpan,
 	pub storage_size_override: Quantity,
+	/// Tear the restore down once it's verified healthy rather than keeping
+	/// it running. Materialised into `PostgresPhysicalReplicaSpec.ephemeral`.
+	/// True for `verify` (throwaway snapshot check), false for the
+	/// analytics intents (long-lived query replicas).
+	pub ephemeral: bool,
 	/// Floor on the postgres pod's `/dev/shm` sizing. Materialised into
 	/// `PostgresPhysicalReplicaSpec.shm_size_floor` so the shared
 	/// Deployment builder picks `max(computed_from_resources, floor)`.
@@ -79,6 +84,7 @@ pub fn config_for(intent: &str) -> Option<IntentConfig> {
 			service_annotations: None,
 			switchover_grace_period: TimeSpan(Span::new().minutes(5)),
 			storage_size_override: Quantity("20Gi".to_string()),
+			ephemeral: true,
 			shm_size_floor: Quantity("512Mi".to_string()),
 		}),
 		"analytics-dev" => Some(IntentConfig {
@@ -89,6 +95,7 @@ pub fn config_for(intent: &str) -> Option<IntentConfig> {
 			service_annotations: None,
 			switchover_grace_period: TimeSpan(Span::new().minutes(5)),
 			storage_size_override: Quantity("50Gi".to_string()),
+			ephemeral: false,
 			shm_size_floor: Quantity("2Gi".to_string()),
 		}),
 		"analytics-dbt" => Some(IntentConfig {
@@ -105,6 +112,7 @@ pub fn config_for(intent: &str) -> Option<IntentConfig> {
 			])),
 			switchover_grace_period: TimeSpan(Span::new().minutes(2)),
 			storage_size_override: Quantity("50Gi".to_string()),
+			ephemeral: false,
 			shm_size_floor: Quantity("2Gi".to_string()),
 		}),
 		_ => None,
@@ -160,6 +168,7 @@ impl IntentConfig {
 			affinity: None,
 			tolerations: Vec::new(),
 			read_only: self.read_only,
+			ephemeral: self.ephemeral,
 			postgres_extra_config: None,
 			notifications,
 			persistent_schemas: self.persistent_schemas.clone(),
@@ -213,6 +222,38 @@ mod tests {
 		assert!(cfg.read_only);
 		assert!(cfg.minimum_ttl.is_none());
 		assert!(cfg.persistent_schemas.is_none());
+		assert!(cfg.ephemeral, "verify replicas are torn down after verify");
+	}
+
+	#[test]
+	fn only_verify_is_ephemeral() {
+		assert!(config_for("verify").unwrap().ephemeral);
+		assert!(
+			!config_for("analytics-dev").unwrap().ephemeral,
+			"analytics-dev is a long-lived query replica"
+		);
+		assert!(
+			!config_for("analytics-dbt").unwrap().ephemeral,
+			"analytics-dbt is a long-lived query replica"
+		);
+	}
+
+	#[test]
+	fn to_replica_spec_carries_ephemeral() {
+		let e = entry("verify", "test");
+		assert!(
+			config_for("verify")
+				.unwrap()
+				.to_replica_spec(&e, vec![])
+				.ephemeral
+		);
+		let e = entry("analytics-dev", "test");
+		assert!(
+			!config_for("analytics-dev")
+				.unwrap()
+				.to_replica_spec(&e, vec![])
+				.ephemeral
+		);
 	}
 
 	#[test]

src/controllers/replica.rs

@@ -357,6 +357,60 @@ pub async fn reconcile(replica: Arc<PostgresPhysicalReplica>, ctx: Arc<Context>)
 		return Ok(Action::requeue(Duration::from_secs(10)));
 	}
 
+	// Ephemeral teardown. For a `spec.ephemeral` replica (e.g. the `verify`
+	// intent) the point of the restore is to prove the snapshot comes up —
+	// once it's Active (the switchover block above ran and, for canopy
+	// replicas, reported the verification), keeping the database running
+	// just wastes resources. Record the verified snapshot and delete the
+	// restore. The replica CR stays; the should_restore logic below only
+	// re-triggers when a newer snapshot is offered (canopy) or the schedule
+	// fires (legacy), gated by `verifiedSnapshotId`.
+	if replica.spec.ephemeral
+		&& let Some(active) = active_restore
+		&& active.metadata.deletion_timestamp.is_none()
+	{
+		let active_name = active.name_any();
+		let verified_snapshot = active.spec.snapshot.clone();
+		info!(
+			replica = name,
+			restore = active_name,
+			snapshot = verified_snapshot,
+			"ephemeral replica: snapshot verified, tearing down restore"
+		);
+
+		// Record the marker BEFORE deleting so that even if the delete or a
+		// crash interleaves, the reconciler won't treat the vanished
+		// restore as an accidental deletion and immediately re-restore.
+		let replicas: Api<PostgresPhysicalReplica> = Api::namespaced(client.clone(), &namespace);
+		let patch = serde_json::json!({
+			"status": {
+				"verifiedSnapshotId": verified_snapshot,
+				"currentRestore": null,
+				"previousRestore": null,
+			}
+		});
+		replicas
+			.patch_status(
+				&name,
+				&PatchParams::apply("postgres-restore-operator"),
+				&Patch::Merge(&patch),
+			)
+			.await?;
+
+		if let Err(e) = restores.delete(&active_name, &Default::default()).await {
+			warn!(
+				replica = name,
+				restore = %active_name,
+				error = %e,
+				"failed to delete ephemeral restore; will retry"
+			);
+		}
+		if let Some(promoted) = ctx.release_restore_slot(&name).await {
+			info!(promoted = %promoted, "promoted queued restore after ephemeral teardown");
+		}
+		return Ok(Action::requeue(Duration::from_secs(30)));
+	}
+
 	// Sweep stale Active restores after grace period.
 	//
 	// Any Active restore for this replica that isn't the current one is a
@@ -909,17 +963,30 @@ pub async fn reconcile(replica: Arc<PostgresPhysicalReplica>, ctx: Arc<Context>)
 	// On the canopy path, the worklist syncer updates
 	// `status.canopyDesiredSnapshotId` when canopy offers a newer snapshot.
 	// Trigger a restore whenever the desired snapshot differs from what the
-	// active restore already carries, in addition to the usual schedule /
+	// replica already carries, in addition to the usual schedule /
 	// never-restored / active-deleted triggers. minimum_ttl still gates:
 	// the intent may declare an explicit lower bound on restore frequency
 	// even in the face of a newer canopy snapshot.
+	//
+	// "What the replica already carries" is normally the active restore's
+	// snapshot, but an ephemeral replica has no active restore after it
+	// tears one down — there we fall back to `verifiedSnapshotId`, the
+	// marker recording the last snapshot we verified. Without that
+	// fallback the `(Some, None)` arm would re-fire forever.
+	let effective_current_snapshot =
+		active_restore.map(|r| r.spec.snapshot.clone()).or_else(|| {
+			replica
+				.status
+				.as_ref()
+				.and_then(|s| s.verified_snapshot_id.clone())
+		});
 	let canopy_desired_changed = is_canopy
 		&& match (
 			replica
 				.status
 				.as_ref()
 				.and_then(|s| s.canopy_desired_snapshot_id.as_ref()),
-			active_restore.map(|r| r.spec.snapshot.as_str()),
+			effective_current_snapshot.as_deref(),
 		) {
 			(Some(desired), Some(current)) => desired != current,
 			(Some(_), None) => true,

src/controllers/replica/scheduling.rs

@@ -273,6 +273,7 @@ mod tests {
 				affinity: None,
 				tolerations: Vec::new(),
 				read_only: true,
+				ephemeral: false,
 				postgres_extra_config: None,
 				notifications: Vec::new(),
 				storage_size_maximum: k8s_openapi::apimachinery::pkg::api::resource::Quantity(

src/controllers/replica/schema_migration.rs

@@ -302,6 +302,7 @@ mod tests {
 				affinity: None,
 				tolerations: vec![],
 				read_only: true,
+				ephemeral: false,
 				postgres_extra_config: None,
 				notifications: vec![],
 				storage_size_maximum: k8s_openapi::apimachinery::pkg::api::resource::Quantity(

src/controllers/replica/tests.rs

@@ -38,6 +38,7 @@ fn make_replica(
 			affinity: None,
 			tolerations: vec![],
 			read_only: true,
+			ephemeral: false,
 			postgres_extra_config: None,
 			notifications: vec![],
 			persistent_schemas,
@@ -198,6 +199,7 @@ fn snapshot_list_job_rotates_kopia_logs() {
 			affinity: None,
 			tolerations: vec![],
 			read_only: true,
+			ephemeral: false,
 			postgres_extra_config: None,
 			notifications: vec![],
 			persistent_schemas: None,

src/controllers/restore/tests.rs

@@ -37,6 +37,7 @@ fn deployment_uses_affinity_not_node_selector() {
 			affinity: None,
 			tolerations: vec![],
 			read_only: true,
+			ephemeral: false,
 			postgres_extra_config: None,
 			notifications: vec![],
 
@@ -123,6 +124,7 @@ fn test_restore_and_replica() -> (PostgresPhysicalRestore, PostgresPhysicalRepli
 			affinity: None,
 			tolerations: vec![],
 			read_only: true,
+			ephemeral: false,
 			postgres_extra_config: None,
 			notifications: vec![],
 

src/types/replica.rs

@@ -118,6 +118,16 @@ pub struct PostgresPhysicalReplicaSpec {
 	#[serde(default = "default_read_only")]
 	pub read_only: bool,
 
+	/// Ephemeral replica: once a restore reaches `Active` (postgres came up
+	/// healthy and, for canopy replicas, the verification was reported),
+	/// tear the restore down instead of keeping it running. The replica CR
+	/// stays; it only restores again when a new snapshot is offered (canopy
+	/// path) or the schedule next fires (legacy path). Used by the `verify`
+	/// intent, whose whole job is "prove the snapshot restores" — keeping
+	/// the database idling afterward just wastes cluster resources.
+	#[serde(default)]
+	pub ephemeral: bool,
+
 	/// Extra lines appended to postgresql.conf (e.g. shared_preload_libraries)
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub postgres_extra_config: Option<String>,
@@ -313,6 +323,14 @@ pub struct PostgresPhysicalReplicaStatus {
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub canopy_desired_snapshot_id: Option<String>,
 
+	/// Last snapshot id an ephemeral replica (`spec.ephemeral`) verified
+	/// and then tore down. After teardown there is no `currentRestore` to
+	/// compare against, so this marker is what stops the reconciler from
+	/// immediately re-restoring the same snapshot; a restore is only
+	/// re-triggered when the desired snapshot differs from this.
+	#[serde(default, skip_serializing_if = "Option::is_none")]
+	pub verified_snapshot_id: Option<String>,
+
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub connection_info: Option<ConnectionInfo>,