feat(canopy): bump bestool-canopy 0.7.0 / kopia 0.4.0 and thread run_id through restore runs

Cargo.toml

@@ -8,8 +8,8 @@ publish = false
 [dependencies]
 anyhow = "1.0.102"
 axum = "0.8.9"
-bestool-canopy = "0.6.1"
-bestool-kopia = { version = "0.3.4", features = ["proxy"] }
+bestool-canopy = "0.7.0"
+bestool-kopia = { version = "0.4.0", features = ["proxy"] }
 cronexpr = "1.5.0"
 futures = "0.3.31"
 globset = "0.4.18"

README.md

@@ -214,6 +214,7 @@ Deleting this resource will drop the restored database and prompt the Replica to
 | Field | Type | Description |
 |-------|------|-------------|
 | `phase` | `Pending` \| `Restoring` \| `Ready` \| `Switching` \| `Active` \| `Failed` | Current phase of the restore. |
+| `runId` | `string` | Canopy run-uuid minted when the restore Job is created, reused for the run's credential requests and verification report so canopy can correlate them. Canopy-backed restores only. |
 | `postgresVersion` | `string` | Detected PostgreSQL major version from the restored data. |
 | `createdAt` | `Time` | When the restore resource was created. |
 | `restoredAt` | `Time` | When the restore job completed. |

crds.yaml

@@ -1124,6 +1124,15 @@ spec:
                 format: date-time
                 nullable: true
                 type: string
+              runId:
+                description: |-
+                  Canopy run-uuid for this restore run, minted when the restore Job is
+                  created and reused for the run's credential requests and its
+                  verification report so canopy can correlate them. Stored as a string
+                  (parsed to `Uuid` when sent) to avoid pulling a schemars uuid feature
+                  into the CRD schema.
+                nullable: true
+                type: string
             type: object
         required:
         - spec

src/bin/canopy_proxy.rs

@@ -34,6 +34,10 @@ struct Config {
 	broker_url: String,
 	group: String,
 	backup_type: String,
+	/// Canopy run-uuid for this restore run (`PGRO_RUN_ID`), forwarded with
+	/// every creds request so canopy attributes the grant to the run. Optional
+	/// so a sidecar scheduled by an older operator (no env set) still works.
+	run_id: Option<String>,
 	region: String,
 	port_file: PathBuf,
 	/// URL the operator serves the stats-callback on; the sidecar POSTs
@@ -49,6 +53,7 @@ impl Config {
 			broker_url: env_required("PGRO_BROKER_URL")?,
 			group: env_required("PGRO_GROUP")?,
 			backup_type: env_required("PGRO_TYPE")?,
+			run_id: std::env::var("PGRO_RUN_ID").ok().filter(|s| !s.is_empty()),
 			region: env_required("PGRO_REGION")?,
 			port_file: env_or("PGRO_PROXY_PORT_FILE", "/var/run/pgro/proxy-port").into(),
 			stats_callback_url: env_required("PGRO_STATS_CALLBACK_URL")?,
@@ -71,11 +76,12 @@ struct BrokerCredentialProvider {
 	url: String,
 	group: String,
 	backup_type: String,
+	run_id: Option<String>,
 	cache: Mutex<Option<BackupCredentials>>,
 }
 
 impl BrokerCredentialProvider {
-	fn new(broker_url: &str, group: &str, backup_type: &str) -> Self {
+	fn new(broker_url: &str, group: &str, backup_type: &str, run_id: Option<&str>) -> Self {
 		Self {
 			http: reqwest::Client::builder()
 				.timeout(Duration::from_secs(10))
@@ -87,6 +93,7 @@ impl BrokerCredentialProvider {
 			),
 			group: group.to_string(),
 			backup_type: backup_type.to_string(),
+			run_id: run_id.map(str::to_string),
 			cache: Mutex::new(None),
 		}
 	}
@@ -96,13 +103,16 @@ impl BrokerCredentialProvider {
 		struct Req<'a> {
 			group: &'a str,
 			r#type: &'a str,
+			#[serde(skip_serializing_if = "Option::is_none")]
+			run_id: Option<&'a str>,
 		}
 		let resp = self
 			.http
 			.post(&self.url)
 			.json(&Req {
 				group: &self.group,
 				r#type: &self.backup_type,
+				run_id: self.run_id.as_deref(),
 			})
 			.send()
 			.await?
@@ -207,6 +217,7 @@ async fn run(cfg: Config) -> Result<(), String> {
 		&cfg.broker_url,
 		&group,
 		&backup_type,
+		cfg.run_id.as_deref(),
 	));
 	let s3_cfg = S3ProxyConfig {
 		upstream: format!("https://s3.{}.amazonaws.com", cfg.region),

src/bin/operator.rs

@@ -698,10 +698,15 @@ async fn register_capabilities(ctx: Arc<Context>) {
 /// State passed to the credential-broker Router. Holds the operator's canopy
 /// client and a per-(group, type) cache so concurrent Job sidecars don't
 /// multiply upstream canopy calls.
+/// Broker creds cache key: `(group, type, run_id)`. Including the run_id keeps
+/// within-run STS refreshes on a cache hit while giving each restore run its
+/// own canopy-attributed credentials.
+type CredsCacheKey = (String, String, Option<uuid::Uuid>);
+
 #[derive(Clone)]
 struct BrokerState {
 	ctx: Arc<Context>,
-	cache: Arc<tokio::sync::Mutex<std::collections::HashMap<(String, String), CachedCreds>>>,
+	cache: Arc<tokio::sync::Mutex<std::collections::HashMap<CredsCacheKey, CachedCreds>>>,
 }
 
 #[derive(Clone)]
@@ -736,6 +741,11 @@ fn broker_router(state: BrokerState) -> Router {
 struct BrokerCredsRequest {
 	group: uuid::Uuid,
 	r#type: String,
+	/// Canopy run-uuid of the restore run this sidecar serves, forwarded to
+	/// canopy so its credential grant is attributed to the run. Optional to
+	/// tolerate an older sidecar (mid-rollout) that doesn't send one.
+	#[serde(default)]
+	run_id: Option<uuid::Uuid>,
 }
 
 /// Broker endpoint the proxy sidecar hits to refresh its STS creds. Forwards
@@ -756,7 +766,7 @@ async fn post_restore_creds(
 		);
 	};
 
-	let key = (req.group.to_string(), req.r#type.clone());
+	let key = (req.group.to_string(), req.r#type.clone(), req.run_id);
 	{
 		let cache = state.cache.lock().await;
 		if let Some(cached) = cache.get(&key)
@@ -766,7 +776,10 @@ async fn post_restore_creds(
 		}
 	}
 
-	match canopy.restore_credentials(&req.r#type, req.group).await {
+	match canopy
+		.restore_credentials(&req.r#type, req.group, req.run_id)
+		.await
+	{
 		Ok(resp) => {
 			let expires_at = resp
 				.credentials

src/canopy.rs

@@ -109,9 +109,9 @@ impl Client {
 	/// carries the intent name, the canopy semantics it opts into, and its
 	/// typed parameter schema.
 	pub async fn restore_capabilities(&self, intents: &[IntentDescriptor]) -> Result<()> {
-		let body = RestoreCapabilitiesArgs {
-			intents: intents.to_vec(),
-		};
+		let body = RestoreCapabilitiesArgs::builder()
+			.intents(intents.to_vec())
+			.build();
 		self.inner
 			.restore_capabilities(&body)
 			.await
@@ -128,16 +128,21 @@ impl Client {
 	}
 
 	/// Fetch short-lived read-only STS creds plus the repo password for a
-	/// `(group, type)`. Authorized iff a declaration covers it.
+	/// `(group, type)`. Authorized iff a declaration covers it. `run_id` is the
+	/// canopy run-uuid when the fetch belongs to a restore run (the restore
+	/// job's sidecar); `None` for non-run fetches such as the reconcile-time
+	/// repo-password poll.
 	pub async fn restore_credentials(
 		&self,
 		backup_type: &str,
 		group: Uuid,
+		run_id: Option<Uuid>,
 	) -> Result<RestoreCredentials> {
-		let body = RestoreCredentialsArgs {
-			group,
-			type_: backup_type.to_string(),
-		};
+		let body = RestoreCredentialsArgs::builder()
+			.group(group)
+			.type_(backup_type.to_string())
+			.maybe_run_id(run_id)
+			.build();
 		self.inner.restore_credentials(&body).await.map_err(|err| {
 			Error::Canopy(format!(
 				"restore_credentials({backup_type}, {group}): {err}"

src/controllers/canopy.rs

@@ -259,8 +259,10 @@ async fn reconcile_entry(ctx: &Context, ns_name: &str, entry: &WorklistEntry) ->
 	let Some(canopy) = ctx.canopy.as_ref() else {
 		return Ok(());
 	};
+	// This per-tick fetch exists only to read the repo password for the creds
+	// Secret; it is not a restore run, so it carries no run_id.
 	let creds = canopy
-		.restore_credentials(&entry.type_, entry.group_id)
+		.restore_credentials(&entry.type_, entry.group_id, None)
 		.await?;
 
 	ensure_canopy_creds_secret(ctx, ns_name, entry, &creds.repo_password.0).await?;

src/controllers/canopy/intent.rs

@@ -72,7 +72,10 @@ const DEFAULT_ANALYTICS_MINIMUM_TTL_SECS: i64 = 7200;
 const DEFAULT_ANALYTICS_SWITCHOVER_GRACE_SECS: i64 = 120;
 
 fn param(type_: ParamType, default: Option<Value>) -> ParamSpec {
-	ParamSpec { type_, default }
+	ParamSpec::builder()
+		.type_(type_)
+		.maybe_default(default)
+		.build()
 }
 
 /// The `analytics` intent's parameter schema (name → typed spec + default).
@@ -112,23 +115,28 @@ fn analytics_param_schema() -> ParamSchema {
 /// operators, dispatches only these, and applies the semantics' behaviours.
 pub fn descriptors() -> Vec<IntentDescriptor> {
 	vec![
-		IntentDescriptor {
-			intent: "verify".to_string(),
-			description: Some(
+		IntentDescriptor::builder()
+			.intent("verify".to_string())
+			.description(
 				"Restore the snapshot to prove it is restorable, then discard it.".to_string(),
-			),
-			semantics: vec![semantics::CHECK.to_string(), semantics::ONCE.to_string()],
-			params: None,
-		},
-		IntentDescriptor {
-			intent: "analytics".to_string(),
-			description: Some(
+			)
+			.semantics(vec![
+				semantics::CHECK.to_string(),
+				semantics::ONCE.to_string(),
+			])
+			.build(),
+		IntentDescriptor::builder()
+			.intent("analytics".to_string())
+			.description(
 				"Keep a long-lived read-only query replica restored from the latest snapshot."
 					.to_string(),
-			),
-			semantics: vec![semantics::CHECK.to_string(), semantics::URL.to_string()],
-			params: Some(analytics_param_schema()),
-		},
+			)
+			.semantics(vec![
+				semantics::CHECK.to_string(),
+				semantics::URL.to_string(),
+			])
+			.params(analytics_param_schema())
+			.build(),
 	]
 }