diff --git a/deploy/rustfs-operator/crds/tenant-crd.yaml b/deploy/rustfs-operator/crds/tenant-crd.yaml index e8aab6b4..b9da9347 100644 --- a/deploy/rustfs-operator/crds/tenant-crd.yaml +++ b/deploy/rustfs-operator/crds/tenant-crd.yaml @@ -2592,6 +2592,9 @@ spec: type: array users: items: + description: |- + User-specific provisioning status. The flattened item preserves the existing status wire + format while keeping ownership metadata out of policy and bucket status schemas. properties: desiredHash: nullable: true @@ -2623,6 +2626,27 @@ spec: observedSecretResourceVersion: nullable: true type: string + ownership: + description: Durable proof that the operator claimed a RustFS user identity before mutating it. + nullable: true + properties: + accessKeyHash: + type: string + state: + enum: + - PendingCreate + - Managed + type: string + tenantUid: + type: string + userName: + type: string + required: + - accessKeyHash + - state + - tenantUid + - userName + type: object policies: items: type: string diff --git a/e2e/src/cases/mod.rs b/e2e/src/cases/mod.rs index 51a68dcb..19e55c81 100644 --- a/e2e/src/cases/mod.rs +++ b/e2e/src/cases/mod.rs @@ -145,7 +145,7 @@ mod tests { }); assert_eq!(counts.get(&Suite::Smoke).copied().unwrap_or_default(), 3); - assert_eq!(counts.get(&Suite::Operator).copied().unwrap_or_default(), 1); + assert_eq!(counts.get(&Suite::Operator).copied().unwrap_or_default(), 2); assert_eq!(counts.get(&Suite::Sts).copied().unwrap_or_default(), 2); assert_eq!(counts.get(&Suite::Console).copied().unwrap_or_default(), 1); assert_eq!( diff --git a/e2e/src/cases/operator.rs b/e2e/src/cases/operator.rs index 2f80cda9..1fc097d3 100644 --- a/e2e/src/cases/operator.rs +++ b/e2e/src/cases/operator.rs @@ -15,13 +15,22 @@ use super::{CaseSpec, Suite}; pub fn cases() -> Vec { - vec![CaseSpec::new( - Suite::Operator, - "operator_live_tenant_is_ready_and_observed", - "Assert the live Tenant is Ready, not Degraded, and has observed the current generation.", - "operator/status", - "operator", - )] + vec![ + CaseSpec::new( + Suite::Operator, + "operator_live_tenant_is_ready_and_observed", + "Assert the live Tenant is Ready, not Degraded, and has observed the current generation.", + "operator/status", + "operator", + ), + CaseSpec::new( + Suite::Operator, + "operator_live_status_subresource_enforces_cas_and_pruning", + "Verify Kubernetes rejects stale status writes and prunes fields omitted by the CRD schema.", + "operator/status", + "operator", + ), + ] } #[cfg(test)] @@ -35,6 +44,12 @@ mod tests { .map(|case| case.name) .collect::>(); - assert_eq!(names, vec!["operator_live_tenant_is_ready_and_observed"]); + assert_eq!( + names, + vec![ + "operator_live_tenant_is_ready_and_observed", + "operator_live_status_subresource_enforces_cas_and_pruning", + ] + ); } } diff --git a/e2e/tests/operator.rs b/e2e/tests/operator.rs index 6a1de3a3..266ffb97 100644 --- a/e2e/tests/operator.rs +++ b/e2e/tests/operator.rs @@ -14,10 +14,59 @@ use anyhow::{Result, ensure}; use kube::Api; -use rustfs_operator_e2e::framework::{assertions, config::E2eConfig, kube_client, live}; +use rustfs_operator_e2e::framework::{ + assertions, config::E2eConfig, kube_client, kubectl::Kubectl, live, +}; +use serde_json::{Value, json}; use operator::types::v1alpha1::tenant::Tenant; +const CHECKPOINT_TEST_CRD: &str = "ownershipcheckpointtests.e2e.rustfs.com"; +const CHECKPOINT_TEST_RESOURCE: &str = "ownershipcheckpointtests.e2e.rustfs.com"; +const CHECKPOINT_TEST_NAME: &str = "ownership-checkpoint-contract"; +const CHECKPOINT_TEST_CRD_YAML: &str = r#" +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: ownershipcheckpointtests.e2e.rustfs.com +spec: + group: e2e.rustfs.com + scope: Namespaced + names: + plural: ownershipcheckpointtests + singular: ownershipcheckpointtest + kind: OwnershipCheckpointTest + versions: + - name: v1 + served: true + storage: true + schema: + openAPIV3Schema: + type: object + properties: + spec: + type: object + status: + type: object + properties: + marker: + type: string + users: + type: array + items: + type: object + required: + - name + - state + properties: + name: + type: string + state: + type: string + subresources: + status: {} +"#; + #[tokio::test] #[ignore = "requires a live Tenant; run through `make e2e-live-run`"] async fn operator_live_tenant_is_ready_and_observed() -> Result<()> { @@ -42,3 +91,151 @@ async fn operator_live_tenant_is_ready_and_observed() -> Result<()> { Ok(()) } + +#[tokio::test] +#[ignore = "requires a dedicated live cluster; run through `make e2e-live-run`"] +async fn operator_live_status_subresource_enforces_cas_and_pruning() -> Result<()> { + let config = E2eConfig::from_env(); + live::require_live_enabled(&config)?; + live::ensure_dedicated_context(&config)?; + + let kubectl = Kubectl::new(&config); + kubectl + .command([ + "delete", + "crd", + CHECKPOINT_TEST_CRD, + "--ignore-not-found=true", + ]) + .run_checked()?; + let result = verify_status_subresource_contract(&kubectl, &config.test_namespace); + let cleanup_result = kubectl + .command([ + "delete", + "crd", + CHECKPOINT_TEST_CRD, + "--ignore-not-found=true", + ]) + .run_checked(); + + result?; + cleanup_result?; + Ok(()) +} + +fn verify_status_subresource_contract(kubectl: &Kubectl, namespace: &str) -> Result<()> { + kubectl + .apply_yaml_command(CHECKPOINT_TEST_CRD_YAML) + .run_checked()?; + kubectl + .command([ + "wait".to_string(), + "--for=condition=Established".to_string(), + format!("crd/{CHECKPOINT_TEST_CRD}"), + "--timeout=60s".to_string(), + ]) + .run_checked()?; + + let namespaced = kubectl.clone().namespaced(namespace); + namespaced + .create_yaml_command(format!( + r#" +apiVersion: e2e.rustfs.com/v1 +kind: OwnershipCheckpointTest +metadata: + name: {CHECKPOINT_TEST_NAME} +spec: {{}} +"# + )) + .run_checked()?; + let created = namespaced + .command([ + "get", + CHECKPOINT_TEST_RESOURCE, + CHECKPOINT_TEST_NAME, + "-o", + "json", + ]) + .run_checked()?; + let created: Value = serde_json::from_str(&created.stdout)?; + let initial_resource_version = created["metadata"]["resourceVersion"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("test resource did not receive a resourceVersion"))?; + + let winner_patch = json!({ + "metadata": { "resourceVersion": initial_resource_version }, + "status": { + "marker": "winner", + "users": [{ + "name": "app-user", + "state": "Pending", + "ownership": { + "state": "PendingCreate", + "tenantUid": "tenant-uid", + }, + }], + }, + }) + .to_string(); + let winner = namespaced + .command([ + "patch".to_string(), + CHECKPOINT_TEST_RESOURCE.to_string(), + CHECKPOINT_TEST_NAME.to_string(), + "--subresource=status".to_string(), + "--type=merge".to_string(), + "-p".to_string(), + winner_patch, + "-o".to_string(), + "json".to_string(), + ]) + .run_checked()?; + let winner: Value = serde_json::from_str(&winner.stdout)?; + ensure!( + winner["status"]["users"][0].get("ownership").is_none(), + "the API server preserved an ownership field omitted by the CRD schema" + ); + + let stale_patch = json!({ + "metadata": { "resourceVersion": initial_resource_version }, + "status": { "marker": "loser" }, + }) + .to_string(); + let stale = namespaced + .command([ + "patch".to_string(), + CHECKPOINT_TEST_RESOURCE.to_string(), + CHECKPOINT_TEST_NAME.to_string(), + "--subresource=status".to_string(), + "--type=merge".to_string(), + "-p".to_string(), + stale_patch, + ]) + .run()?; + ensure!( + stale.code != Some(0), + "the API server accepted a status patch with a stale resourceVersion" + ); + let stale_output = format!("{}\n{}", stale.stdout, stale.stderr).to_ascii_lowercase(); + ensure!( + stale_output.contains("conflict") || stale_output.contains("object has been modified"), + "the stale status patch failed without a Kubernetes conflict: {stale_output}" + ); + + let current = namespaced + .command([ + "get", + CHECKPOINT_TEST_RESOURCE, + CHECKPOINT_TEST_NAME, + "-o", + "json", + ]) + .run_checked()?; + let current: Value = serde_json::from_str(¤t.stdout)?; + ensure!( + current["status"]["marker"] == "winner", + "the rejected stale patch changed the persisted status" + ); + + Ok(()) +} diff --git a/src/reconcile/phases.rs b/src/reconcile/phases.rs index d6e808b7..eae77747 100644 --- a/src/reconcile/phases.rs +++ b/src/reconcile/phases.rs @@ -1290,6 +1290,19 @@ pub(super) async fn finalize_tenant_status( message, ) } + ProvisioningOutcome::Retry { + message, + retry_after, + } => { + warn!( + tenant = %tenant.name(), + namespace = %namespace, + message = %message, + retry_after_seconds = retry_after.as_secs(), + "retrying after RustFS user ownership checkpoint contention or transient failure" + ); + return Ok(Action::requeue(retry_after)); + } } } else { builder.finish_reconciling( diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs index 5897f49c..6fdf633f 100644 --- a/src/reconcile/provisioning.rs +++ b/src/reconcile/provisioning.rs @@ -21,15 +21,22 @@ use crate::types::v1alpha1::provisioning::{ use crate::types::v1alpha1::status::Reason; use crate::types::v1alpha1::status::provisioning::{ ProvisioningItemState, ProvisioningItemStatus, ProvisioningPhase, ProvisioningStatus, + ProvisioningUserOwnershipState, ProvisioningUserOwnershipStatus, ProvisioningUserStatus, }; use crate::types::v1alpha1::tenant::Tenant; use k8s_openapi::ByteString; use k8s_openapi::api::core::v1::{ConfigMap, Secret}; +use kube::Api; +use kube::api::{Patch, PatchParams}; use serde_json::Value; use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, BTreeSet}; +use std::time::Duration; use tracing::{info, warn}; +const CHECKPOINT_CONFLICT_RETRY: Duration = Duration::from_secs(2); +const CHECKPOINT_TRANSIENT_RETRY: Duration = Duration::from_secs(10); + pub(super) struct ProvisioningReconcileResult { pub status: ProvisioningStatus, pub outcome: ProvisioningOutcome, @@ -37,8 +44,29 @@ pub(super) struct ProvisioningReconcileResult { pub(super) enum ProvisioningOutcome { Ready, - Pending { message: String }, - Failed { reason: Reason, message: String }, + Pending { + message: String, + }, + Failed { + reason: Reason, + message: String, + }, + Retry { + message: String, + retry_after: Duration, + }, +} + +#[derive(Clone, Debug)] +struct CheckpointRetry { + message: String, + retry_after: Duration, +} + +#[derive(Debug)] +enum CheckpointError { + Permanent { message: String }, + Retry(CheckpointRetry), } struct ProvisioningRun<'a> { @@ -51,6 +79,7 @@ struct ProvisioningRun<'a> { failures: Vec<(Reason, String)>, } +#[derive(Clone)] struct UserCredentials { access_key: String, secret_key: String, @@ -71,6 +100,19 @@ struct UserCredentialsPreflight { duplicate_access_key_hashes: BTreeSet, } +enum UserReconcilePlan { + Complete(Box), + Prepared(Box), +} + +struct PreparedUserReconcile { + user: ProvisioningUser, + credentials: UserCredentials, + exists: bool, + ownership: ProvisioningUserOwnershipStatus, + checkpoint: Option, +} + struct PolicyDocument { raw: String, normalized: String, @@ -101,7 +143,7 @@ impl ProvisioningRun<'_> { self.previous.policies.iter().find(|item| item.name == name) } - fn previous_user(&self, name: &str) -> Option<&ProvisioningItemStatus> { + fn previous_user(&self, name: &str) -> Option<&ProvisioningUserStatus> { self.previous.users.iter().find(|item| item.name == name) } @@ -118,8 +160,12 @@ impl ProvisioningRun<'_> { self.status.policies.push(item); } - fn push_user(&mut self, item: ProvisioningItemStatus) { - self.log_item_transition("user", self.previous_user(&item.name), &item); + fn push_user(&mut self, item: ProvisioningUserStatus) { + self.log_item_transition( + "user", + self.previous_user(&item.name).map(AsRef::as_ref), + item.as_ref(), + ); if item.state == ProvisioningItemState::Failed.as_str() { self.failures .push((reason_from_str(&item.reason), item_message(&item))); @@ -180,14 +226,18 @@ impl ProvisioningRun<'_> { } } - fn item( + fn item

( &self, - previous: Option<&ProvisioningItemStatus>, + previous: Option<&P>, name: &str, state: ProvisioningItemState, reason: Reason, message: impl Into, - ) -> ProvisioningItemStatus { + ) -> ProvisioningItemStatus + where + P: AsRef + ?Sized, + { + let previous = previous.map(AsRef::as_ref); let message = message.into(); let mut item = ProvisioningItemStatus::new(name, state, reason.as_str()); item.message = Some(message.clone()); @@ -224,6 +274,12 @@ impl ProvisioningRun<'_> { item } + fn retained_user(&self, previous: &ProvisioningUserStatus) -> ProvisioningUserStatus { + let mut item = ProvisioningUserStatus::new(self.retained_item(previous.as_ref())); + item.ownership = previous.ownership.clone(); + item + } + fn mark_all_active(&mut self, state: ProvisioningItemState, reason: Reason, message: &str) { for policy in &self.tenant.spec.policies { let mut item = self.item( @@ -255,6 +311,10 @@ impl ProvisioningRun<'_> { item.last_applied_access_key_hash = previous.last_applied_access_key_hash.clone(); item.policies = previous.policies.clone(); } + let mut item = ProvisioningUserStatus::new(item); + item.ownership = self + .previous_user(&user.name) + .and_then(|previous| previous.ownership.clone()); self.push_user(item); } for bucket in &self.tenant.spec.buckets { @@ -284,7 +344,7 @@ impl ProvisioningRun<'_> { let users = desired_names(self.tenant.spec.users.iter().map(|user| &user.name)); for previous in &self.previous.users { if !users.contains(&previous.name) { - self.status.users.push(self.retained_item(previous)); + self.status.users.push(self.retained_user(previous)); } } @@ -320,6 +380,7 @@ impl ProvisioningRun<'_> { ProvisioningOutcome::Ready => ProvisioningPhase::Ready, ProvisioningOutcome::Pending { .. } => ProvisioningPhase::Pending, ProvisioningOutcome::Failed { .. } => ProvisioningPhase::Failed, + ProvisioningOutcome::Retry { .. } => ProvisioningPhase::Pending, }; self.prepare_status(phase); @@ -417,7 +478,16 @@ pub(super) async fn reconcile_provisioning( }; reconcile_policies(&mut run, &client, &mut live_policies).await; - reconcile_users(&mut run, &client, &live_policies, &user_credentials).await; + if let Some(retry) = reconcile_users(&mut run, &client, &live_policies, &user_credentials).await + { + return ProvisioningReconcileResult { + status: run.status, + outcome: ProvisioningOutcome::Retry { + message: retry.message, + retry_after: retry.retry_after, + }, + }; + } reconcile_buckets(&mut run, &client).await; run.finish() } @@ -711,7 +781,7 @@ async fn reconcile_users( client: &RustfsAdminClient, live_policies: &BTreeMap, credentials_preflight: &UserCredentialsPreflight, -) { +) -> Option { let failed_spec_policies = run .status .policies @@ -719,6 +789,7 @@ async fn reconcile_users( .filter(|item| item.state == ProvisioningItemState::Failed.as_str()) .map(|item| item.name.clone()) .collect::>(); + let mut plans = Vec::with_capacity(run.tenant.spec.users.len()); for (user, preflight) in run .tenant @@ -741,7 +812,7 @@ async fn reconcile_users( ), ); let item = annotate_user_item(item, user, previous, None); - run.push_user(item); + plans.push(UserReconcilePlan::Complete(Box::new(item))); continue; } UserCredentialsCheck::Checked { @@ -766,7 +837,7 @@ async fn reconcile_users( ), ); let item = annotate_user_item(item, user, previous, None); - run.push_user(item); + plans.push(UserReconcilePlan::Complete(Box::new(item))); continue; } if let Some(message) = policy_error { @@ -779,7 +850,7 @@ async fn reconcile_users( message, ); let item = annotate_user_item(item, user, previous, None); - run.push_user(item); + plans.push(UserReconcilePlan::Complete(Box::new(item))); continue; } let credentials = match credentials { @@ -794,22 +865,75 @@ async fn reconcile_users( message, ); let item = annotate_user_item(item, user, previous, None); - run.push_user(item); + plans.push(UserReconcilePlan::Complete(Box::new(item))); continue; } }; - let item = reconcile_user( - run, - client, - live_policies, - &failed_spec_policies, - user, - credentials, - ) - .await; + plans.push( + prepare_user_reconcile( + run, + client, + live_policies, + &failed_spec_policies, + user, + credentials, + ) + .await, + ); + } + + let checkpoints = plans + .iter() + .filter_map(|plan| match plan { + UserReconcilePlan::Prepared(prepared) => prepared.checkpoint.clone(), + UserReconcilePlan::Complete(_) => None, + }) + .collect::>(); + if !checkpoints.is_empty() + && let Err(error) = persist_user_ownership_checkpoints(run, &checkpoints).await + { + match error { + CheckpointError::Retry(retry) => return Some(retry), + CheckpointError::Permanent { message } => { + for plan in &mut plans { + let replacement = match plan { + UserReconcilePlan::Prepared(prepared) if prepared.checkpoint.is_some() => { + let previous = run.previous_user(&prepared.user.name); + let item = run.item( + previous, + &prepared.user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipCheckpointFailed, + message.clone(), + ); + Some(UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, + &prepared.user, + previous, + None, + )))) + } + UserReconcilePlan::Prepared(_) | UserReconcilePlan::Complete(_) => None, + }; + if let Some(replacement) = replacement { + *plan = replacement; + } + } + } + } + } + + for plan in plans { + let item = match plan { + UserReconcilePlan::Complete(item) => *item, + UserReconcilePlan::Prepared(prepared) => { + execute_prepared_user(run, client, *prepared).await + } + }; run.push_user(item); } + None } async fn preflight_user_credentials(run: &ProvisioningRun<'_>) -> UserCredentialsPreflight { @@ -856,14 +980,14 @@ fn duplicate_user_access_key_hashes( .collect() } -async fn reconcile_user( +async fn prepare_user_reconcile( run: &ProvisioningRun<'_>, client: &RustfsAdminClient, live_policies: &BTreeMap, failed_spec_policies: &BTreeSet, user: &ProvisioningUser, credentials: &UserCredentials, -) -> ProvisioningItemStatus { +) -> UserReconcilePlan { let previous = run.previous_user(&user.name); if user_access_key_changed(previous, credentials) { let item = run.item( @@ -873,7 +997,9 @@ async fn reconcile_user( Reason::ImmutableFieldModified, "user access key is immutable after provisioning; create a new user entry to migrate it", ); - return annotate_user_item(item, user, previous, None); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); } if let Some(policy_name) = user @@ -888,7 +1014,9 @@ async fn reconcile_user( Reason::UserPolicySetFailed, format!("referenced policy '{policy_name}' is not ready"), ); - return annotate_user_item(item, user, previous, None); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); } if let Some(policy_name) = user @@ -903,7 +1031,9 @@ async fn reconcile_user( Reason::UserPolicyNotFound, format!("referenced policy '{policy_name}' does not exist"), ); - return annotate_user_item(item, user, previous, None); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); } let exists = match client.user_exists(&credentials.access_key).await { @@ -916,12 +1046,171 @@ async fn reconcile_user( Reason::UserSecretInvalid, format!("failed to query RustFS user: {error}"), ); - return annotate_user_item(item, user, previous, None); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); + } + }; + + let mut ownership = match matching_user_ownership(previous, run.tenant, user, credentials) { + Ok(ownership) => ownership, + Err(message) => { + let item = run.item( + previous, + &user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipConflict, + message, + ); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); + } + }; + let mut checkpoint_update = None; + + if exists && ownership.is_none() { + if legacy_user_status_can_migrate(previous, user, credentials) { + let managed_ownership = match user_ownership( + run.tenant, + user, + credentials, + ProvisioningUserOwnershipState::Managed, + ) { + Ok(ownership) => ownership, + Err(message) => { + let item = run.item( + previous, + &user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipCheckpointFailed, + message, + ); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); + } + }; + let managed_checkpoint = run.item( + previous, + &user.name, + ProvisioningItemState::Ready, + Reason::ProvisioningConfigured, + "Legacy operator-managed RustFS user ownership was migrated", + ); + // Preserve the legacy observed Secret version so a concurrently rotated Secret is + // still applied after the ownership checkpoint has been persisted. + let mut managed_checkpoint = + annotate_user_item(managed_checkpoint, user, previous, None); + managed_checkpoint.ownership = Some(managed_ownership.clone()); + checkpoint_update = Some(managed_checkpoint); + ownership = Some(managed_ownership); + } else { + let item = run.item( + previous, + &user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipConflict, + "RustFS user already exists without a matching operator ownership checkpoint; choose a different access key or remove the unmanaged user", + ); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); } + } + + if !exists && ownership.is_none() { + let pending_ownership = match user_ownership( + run.tenant, + user, + credentials, + ProvisioningUserOwnershipState::PendingCreate, + ) { + Ok(ownership) => ownership, + Err(message) => { + let item = run.item( + previous, + &user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipCheckpointFailed, + message, + ); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); + } + }; + let pending_checkpoint = run.item( + previous, + &user.name, + ProvisioningItemState::Pending, + Reason::ProvisioningPending, + "Operator ownership checkpoint was persisted before creating the RustFS user", + ); + let mut pending_checkpoint = + annotate_user_item(pending_checkpoint, user, previous, Some(credentials)); + pending_checkpoint.ownership = Some(pending_ownership.clone()); + checkpoint_update = Some(pending_checkpoint); + ownership = Some(pending_ownership); + } else if !exists + && ownership.as_ref().is_some_and(|ownership| { + ownership.state == ProvisioningUserOwnershipState::PendingCreate + }) + { + // Refresh the persisted intent before retrying an external create after a process crash. + // This recovery relies on per-Tenant controller serialization; it does not provide + // exactly-once delivery across Kubernetes and independent RustFS actors. + let pending_checkpoint = run.item( + previous, + &user.name, + ProvisioningItemState::Pending, + Reason::ProvisioningPending, + "Operator is resuming a pending RustFS user creation", + ); + let mut pending_checkpoint = + annotate_user_item(pending_checkpoint, user, previous, Some(credentials)); + pending_checkpoint.ownership = ownership.clone(); + checkpoint_update = Some(pending_checkpoint); + } + + let Some(ownership) = ownership else { + let item = run.item( + previous, + &user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipCheckpointFailed, + "Operator ownership checkpoint is required before synchronizing RustFS user credentials", + ); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); }; + UserReconcilePlan::Prepared(Box::new(PreparedUserReconcile { + user: user.clone(), + credentials: credentials.clone(), + exists, + ownership, + checkpoint: checkpoint_update, + })) +} + +async fn execute_prepared_user( + run: &ProvisioningRun<'_>, + client: &RustfsAdminClient, + prepared: PreparedUserReconcile, +) -> ProvisioningUserStatus { + let PreparedUserReconcile { + user, + credentials, + exists, + mut ownership, + .. + } = prepared; + let previous = run.previous_user(&user.name); + let credentials_applied = - match sync_user_credentials(client, previous, credentials, exists).await { + match sync_user_credentials(client, previous, &credentials, exists).await { Ok(applied) => applied, Err(error) => { let item = run.item( @@ -931,10 +1220,14 @@ async fn reconcile_user( Reason::UserSecretInvalid, format!("failed to update RustFS user credentials: {error}"), ); - return annotate_user_item(item, user, previous, None); + let mut item = annotate_user_item(item, &user, previous, None); + item.ownership = Some(ownership); + return item; } }; + ownership.state = ProvisioningUserOwnershipState::Managed; + if let Err(error) = client .set_user_policy(&credentials.access_key, &user.policies) .await @@ -946,7 +1239,9 @@ async fn reconcile_user( Reason::UserPolicySetFailed, format!("failed to set RustFS user policy mapping: {error}"), ); - return annotate_user_item(item, user, previous, Some(credentials)); + let mut item = annotate_user_item(item, &user, previous, Some(&credentials)); + item.ownership = Some(ownership); + return item; } let message = if !exists { @@ -972,15 +1267,305 @@ async fn reconcile_user( } _ => Some(run.now.clone()), }; - annotate_user_item(item, user, previous, Some(credentials)) + let mut item = annotate_user_item(item, &user, previous, Some(&credentials)); + item.ownership = Some(ownership); + item +} + +#[cfg(test)] +async fn reconcile_user( + run: &ProvisioningRun<'_>, + client: &RustfsAdminClient, + live_policies: &BTreeMap, + failed_spec_policies: &BTreeSet, + user: &ProvisioningUser, + credentials: &UserCredentials, +) -> ProvisioningUserStatus { + match prepare_user_reconcile( + run, + client, + live_policies, + failed_spec_policies, + user, + credentials, + ) + .await + { + UserReconcilePlan::Complete(item) => *item, + UserReconcilePlan::Prepared(prepared) => { + if let Some(checkpoint) = prepared.checkpoint.as_ref() + && let Err(error) = + persist_user_ownership_checkpoints(run, std::slice::from_ref(checkpoint)).await + { + let previous = run.previous_user(&prepared.user.name); + let item = run.item( + previous, + &prepared.user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipCheckpointFailed, + checkpoint_error_message(error), + ); + return annotate_user_item(item, &prepared.user, previous, None); + } + execute_prepared_user(run, client, *prepared).await + } + } +} + +fn matching_user_ownership( + previous: Option<&ProvisioningUserStatus>, + tenant: &Tenant, + user: &ProvisioningUser, + credentials: &UserCredentials, +) -> Result, &'static str> { + let Some(ownership) = previous.and_then(|item| item.ownership.as_ref()) else { + return Ok(None); + }; + let Some(tenant_uid) = tenant.metadata.uid.as_deref() else { + return Err( + "Tenant UID is unavailable, so the operator cannot verify RustFS user ownership", + ); + }; + let current_access_key_hash = access_key_hash(&credentials.access_key); + if ownership.tenant_uid != tenant_uid + || ownership.user_name != user.name + || ownership.access_key_hash != current_access_key_hash + { + return Err( + "RustFS user ownership checkpoint does not match this Tenant UID, provisioning user, or access key", + ); + } + + Ok(Some(ownership.clone())) +} + +fn legacy_user_status_can_migrate( + previous: Option<&ProvisioningUserStatus>, + user: &ProvisioningUser, + credentials: &UserCredentials, +) -> bool { + let Some(previous) = previous else { + return false; + }; + let current_access_key_hash = access_key_hash(&credentials.access_key); + matches!( + previous.state.as_str(), + state if state == ProvisioningItemState::Ready.as_str() + || state == ProvisioningItemState::Retained.as_str() + ) && previous.last_applied_access_key_hash.as_deref() == Some(current_access_key_hash.as_str()) + && previous.observed_secret_name.as_deref() == Some(user.credentials_secret_name()) +} + +fn user_ownership( + tenant: &Tenant, + user: &ProvisioningUser, + credentials: &UserCredentials, + state: ProvisioningUserOwnershipState, +) -> Result { + let Some(tenant_uid) = tenant.metadata.uid.as_deref() else { + return Err( + "Tenant UID is unavailable, so the operator cannot persist a RustFS user ownership checkpoint", + ); + }; + Ok(ProvisioningUserOwnershipStatus { + state, + tenant_uid: tenant_uid.to_string(), + user_name: user.name.clone(), + access_key_hash: access_key_hash(&credentials.access_key), + }) +} + +async fn persist_user_ownership_checkpoints( + run: &ProvisioningRun<'_>, + checkpoints: &[ProvisioningUserStatus], +) -> Result<(), CheckpointError> { + if checkpoints.is_empty() { + return Ok(()); + } + let api: Api = Api::namespaced(run.ctx.client.clone(), run.namespace); + let latest = api + .get(&run.tenant.name()) + .await + .map_err(classify_checkpoint_kube_error)?; + if latest.metadata.uid != run.tenant.metadata.uid + || latest.metadata.generation != run.tenant.metadata.generation + { + return Err(CheckpointError::Retry(CheckpointRetry { + message: "Tenant identity or generation changed before persisting the RustFS user ownership checkpoint" + .to_string(), + retry_after: CHECKPOINT_CONFLICT_RETRY, + })); + } + for checkpoint in checkpoints { + let previous_user = run + .previous + .users + .iter() + .find(|item| item.name == checkpoint.name); + let latest_user = latest.status.as_ref().and_then(|status| { + status + .provisioning + .users + .iter() + .find(|item| item.name == checkpoint.name) + }); + if latest_user != previous_user { + return Err(CheckpointError::Retry(CheckpointRetry { + message: format!( + "Tenant user '{}' provisioning status changed before persisting the RustFS user ownership checkpoint", + checkpoint.name + ), + retry_after: CHECKPOINT_CONFLICT_RETRY, + })); + } + } + let Some(resource_version) = latest.metadata.resource_version.clone() else { + return Err(CheckpointError::Permanent { + message: "Tenant resourceVersion is unavailable, so the operator cannot safely persist the RustFS user ownership checkpoint" + .to_string(), + }); + }; + + let mut provisioning = latest + .status + .as_ref() + .map(|status| status.provisioning.clone()) + .unwrap_or_default(); + merge_provisioning_items(&mut provisioning.policies, &run.status.policies); + merge_provisioning_user_items(&mut provisioning.users, &run.status.users); + merge_provisioning_items(&mut provisioning.buckets, &run.status.buckets); + merge_provisioning_user_items(&mut provisioning.users, checkpoints); + provisioning.observed_generation = run.tenant.metadata.generation; + provisioning.phase = Some(ProvisioningPhase::Pending); + provisioning + .policies + .sort_by(|left, right| left.name.cmp(&right.name)); + provisioning + .users + .sort_by(|left, right| left.name.cmp(&right.name)); + provisioning + .buckets + .sort_by(|left, right| left.name.cmp(&right.name)); + + let mut status = latest.status.unwrap_or_default(); + status.provisioning = provisioning; + let status_patch = serde_json::json!({ + "metadata": { "resourceVersion": resource_version }, + "status": status, + }); + let updated = api + .patch_status( + &run.tenant.name(), + &PatchParams::default(), + &Patch::Merge(&status_patch), + ) + .await + .map_err(classify_checkpoint_kube_error)?; + if updated.metadata.resource_version.is_none() { + return Err(CheckpointError::Retry(CheckpointRetry { + message: "Kubernetes accepted the RustFS user ownership checkpoint but omitted resourceVersion; retrying from fresh state" + .to_string(), + retry_after: CHECKPOINT_TRANSIENT_RETRY, + })); + } + for checkpoint in checkpoints { + let persisted_checkpoint = updated.status.as_ref().and_then(|status| { + status + .provisioning + .users + .iter() + .find(|item| item.name == checkpoint.name) + }); + if persisted_checkpoint.is_none_or(|persisted| { + persisted.state != checkpoint.state || persisted.ownership != checkpoint.ownership + }) { + return Err(CheckpointError::Permanent { + message: format!( + "Kubernetes accepted the RustFS user ownership checkpoint request but did not persist the expected state and ownership proof for user '{}'; ensure the Tenant CRD is upgraded before the Operator", + checkpoint.name + ), + }); + } + } + Ok(()) +} + +fn classify_checkpoint_kube_error(error: kube::Error) -> CheckpointError { + match error { + kube::Error::Api(response) if response.code == 409 => { + CheckpointError::Retry(CheckpointRetry { + message: "Tenant status changed while persisting the RustFS user ownership checkpoint" + .to_string(), + retry_after: CHECKPOINT_CONFLICT_RETRY, + }) + } + kube::Error::Api(response) + if response.code == 408 || response.code == 429 || response.code >= 500 => + { + CheckpointError::Retry(CheckpointRetry { + message: format!( + "Kubernetes temporarily rejected the RustFS user ownership checkpoint ({} {})", + response.code, response.reason + ), + retry_after: CHECKPOINT_TRANSIENT_RETRY, + }) + } + kube::Error::Api(response) if (400..500).contains(&response.code) => { + CheckpointError::Permanent { + message: format!( + "Kubernetes rejected the RustFS user ownership checkpoint ({} {})", + response.code, response.reason + ), + } + } + _ => CheckpointError::Retry(CheckpointRetry { + message: "The Kubernetes result for the RustFS user ownership checkpoint is uncertain; retrying from fresh state" + .to_string(), + retry_after: CHECKPOINT_TRANSIENT_RETRY, + }), + } +} + +#[cfg(test)] +fn checkpoint_error_message(error: CheckpointError) -> String { + match error { + CheckpointError::Permanent { message } => message, + CheckpointError::Retry(retry) => retry.message, + } +} + +fn merge_provisioning_items( + destination: &mut Vec, + updates: &[ProvisioningItemStatus], +) { + for update in updates { + if let Some(item) = destination.iter_mut().find(|item| item.name == update.name) { + *item = update.clone(); + } else { + destination.push(update.clone()); + } + } +} + +fn merge_provisioning_user_items( + destination: &mut Vec, + updates: &[ProvisioningUserStatus], +) { + for update in updates { + if let Some(item) = destination.iter_mut().find(|item| item.name == update.name) { + *item = update.clone(); + } else { + destination.push(update.clone()); + } + } } fn annotate_user_item( mut item: ProvisioningItemStatus, user: &ProvisioningUser, - previous: Option<&ProvisioningItemStatus>, + previous: Option<&ProvisioningUserStatus>, applied_credentials: Option<&UserCredentials>, -) -> ProvisioningItemStatus { +) -> ProvisioningUserStatus { match applied_credentials { Some(credentials) => { item.observed_secret_resource_version = credentials.resource_version.clone(); @@ -996,11 +1581,13 @@ fn annotate_user_item( } } item.policies = user.policies.clone(); + let mut item = ProvisioningUserStatus::new(item); + item.ownership = previous.and_then(|item| item.ownership.clone()); item } fn user_access_key_changed( - previous: Option<&ProvisioningItemStatus>, + previous: Option<&ProvisioningUserStatus>, credentials: &UserCredentials, ) -> bool { let current_hash = access_key_hash(&credentials.access_key); @@ -1011,7 +1598,7 @@ fn user_access_key_changed( async fn sync_user_credentials( client: &RustfsAdminClient, - previous: Option<&ProvisioningItemStatus>, + previous: Option<&ProvisioningUserStatus>, credentials: &UserCredentials, exists: bool, ) -> Result { @@ -1026,7 +1613,7 @@ async fn sync_user_credentials( } fn user_credentials_need_apply( - previous: Option<&ProvisioningItemStatus>, + previous: Option<&ProvisioningUserStatus>, credentials: &UserCredentials, exists: bool, ) -> bool { @@ -1471,6 +2058,8 @@ fn reason_from_str(reason: &str) -> Reason { "UserPolicyNotFound" => Reason::UserPolicyNotFound, "UserPolicyInvalid" => Reason::UserPolicyInvalid, "UserPolicySetFailed" => Reason::UserPolicySetFailed, + "UserOwnershipConflict" => Reason::UserOwnershipConflict, + "UserOwnershipCheckpointFailed" => Reason::UserOwnershipCheckpointFailed, "BucketCreateFailed" => Reason::BucketCreateFailed, "BucketObjectLockConflict" => Reason::BucketObjectLockConflict, _ => Reason::ProvisioningFailed, @@ -1487,6 +2076,7 @@ mod tests { http::{Request, StatusCode}, routing::{any, get, put}, }; + use http_body_util::BodyExt; use k8s_openapi::ByteString; use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta; use kube::{Client, client::Body as KubeBody}; @@ -1745,6 +2335,997 @@ mod tests { } } + fn provisioning_test_tenant( + user: ProvisioningUser, + provisioning: ProvisioningStatus, + ) -> Tenant { + Tenant { + metadata: ObjectMeta { + name: Some("tenant-a".to_string()), + namespace: Some("storage".to_string()), + uid: Some("tenant-uid-a".to_string()), + resource_version: Some("17".to_string()), + generation: Some(3), + ..Default::default() + }, + spec: crate::types::v1alpha1::tenant::TenantSpec { + users: vec![user], + ..Default::default() + }, + status: Some(crate::types::v1alpha1::status::Status { + provisioning, + ..Default::default() + }), + } + } + + fn user_credentials(resource_version: &str) -> UserCredentials { + UserCredentials { + access_key: "appuser01".to_string(), + secret_key: "super-secret-value".to_string(), + secret_name: "app-user-secret".to_string(), + resource_version: Some(resource_version.to_string()), + } + } + + fn owned_user_status( + state: ProvisioningUserOwnershipState, + secret_resource_version: &str, + ) -> ProvisioningUserStatus { + let item = ProvisioningItemStatus::new( + "app-user", + ProvisioningItemState::Ready, + Reason::ProvisioningConfigured.as_str(), + ); + let mut item = ProvisioningUserStatus::new(item); + item.observed_secret_resource_version = Some(secret_resource_version.to_string()); + item.observed_secret_name = Some("app-user-secret".to_string()); + item.last_applied_access_key_hash = Some(access_key_hash("appuser01")); + item.ownership = Some(ProvisioningUserOwnershipStatus { + state, + tenant_uid: "tenant-uid-a".to_string(), + user_name: "app-user".to_string(), + access_key_hash: access_key_hash("appuser01"), + }); + item + } + + #[test] + fn legacy_user_migration_requires_complete_matching_ready_or_retained_status() { + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let credentials = user_credentials("5"); + let mut previous = owned_user_status(ProvisioningUserOwnershipState::Managed, "4"); + previous.ownership = None; + + assert!(legacy_user_status_can_migrate( + Some(&previous), + &user, + &credentials + )); + previous.state = ProvisioningItemState::Retained.as_str().to_string(); + assert!(legacy_user_status_can_migrate( + Some(&previous), + &user, + &credentials + )); + + previous.state = ProvisioningItemState::Failed.as_str().to_string(); + assert!(!legacy_user_status_can_migrate( + Some(&previous), + &user, + &credentials + )); + previous.state = ProvisioningItemState::Ready.as_str().to_string(); + previous.last_applied_access_key_hash = None; + assert!(!legacy_user_status_can_migrate( + Some(&previous), + &user, + &credentials + )); + previous.last_applied_access_key_hash = Some(access_key_hash("different-user")); + assert!(!legacy_user_status_can_migrate( + Some(&previous), + &user, + &credentials + )); + previous.last_applied_access_key_hash = Some(access_key_hash("appuser01")); + previous.observed_secret_name = Some("different-secret".to_string()); + assert!(!legacy_user_status_can_migrate( + Some(&previous), + &user, + &credentials + )); + } + + #[tokio::test] + async fn stale_reconciler_retries_without_returning_overwriting_status() { + let requests = Arc::new(AtomicUsize::new(0)); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut checkpoint = owned_user_status(ProvisioningUserOwnershipState::PendingCreate, "5"); + checkpoint.state = ProvisioningItemState::Pending.as_str().to_string(); + let mut latest_tenant = tenant.clone(); + latest_tenant.metadata.resource_version = Some("18".to_string()); + let mut persisted_tenant = latest_tenant.clone(); + persisted_tenant.metadata.resource_version = Some("19".to_string()); + persisted_tenant + .status + .as_mut() + .expect("Tenant should have status") + .provisioning + .users = vec![checkpoint.clone()]; + let service_requests = requests.clone(); + let kube_service = service_fn(move |request: http::Request| { + let service_requests = service_requests.clone(); + let latest_tenant = latest_tenant.clone(); + let persisted_tenant = persisted_tenant.clone(); + async move { + let attempt = service_requests.fetch_add(1, Ordering::SeqCst); + let response = match attempt { + 0 => { + assert_eq!(request.method(), http::Method::GET); + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&latest_tenant) + .expect("Tenant response should serialize"), + )) + } + 1 => { + assert_eq!(request.method(), http::Method::PATCH); + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&persisted_tenant) + .expect("Tenant response should serialize"), + )) + } + 2 => { + assert_eq!(request.method(), http::Method::GET); + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&persisted_tenant) + .expect("Tenant response should serialize"), + )) + } + _ => panic!("unexpected Kubernetes request {attempt}"), + }; + Ok::<_, Infallible>(response.expect("response should build")) + } + }); + let ctx = Context::new(Client::new(kube_service, "default")); + let make_run = || ProvisioningRun { + ctx: &ctx, + tenant: &tenant, + namespace: "storage", + previous: ProvisioningStatus::default(), + now: "2026-08-02T00:00:00Z".to_string(), + status: ProvisioningStatus::default(), + failures: Vec::new(), + }; + let winner = make_run(); + let loser = make_run(); + + persist_user_ownership_checkpoints(&winner, std::slice::from_ref(&checkpoint)) + .await + .expect("first reconciler should persist its checkpoint"); + let error = persist_user_ownership_checkpoints(&loser, std::slice::from_ref(&checkpoint)) + .await + .expect_err("stale reconciler should lose the CAS"); + + match error { + CheckpointError::Retry(CheckpointRetry { retry_after, .. }) => { + assert_eq!(retry_after, CHECKPOINT_CONFLICT_RETRY); + } + _ => panic!("stale checkpoint writer should retry"), + } + assert_eq!(requests.load(Ordering::SeqCst), 3); + } + + #[tokio::test] + async fn multiple_user_ownership_checkpoints_use_one_cas_patch() { + let requests = Arc::new(AtomicUsize::new(0)); + let captured_patch = Arc::new(Mutex::new(Value::Null)); + let user = provisioning_user("app-user-a", "app-user-secret-a", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut first = owned_user_status(ProvisioningUserOwnershipState::PendingCreate, "5"); + first.name = "app-user-a".to_string(); + first.state = ProvisioningItemState::Pending.as_str().to_string(); + first + .ownership + .as_mut() + .expect("ownership should exist") + .user_name = first.name.clone(); + let mut second = first.clone(); + second.name = "app-user-b".to_string(); + second + .ownership + .as_mut() + .expect("ownership should exist") + .user_name = second.name.clone(); + + let mut latest_tenant = tenant.clone(); + latest_tenant.metadata.resource_version = Some("18".to_string()); + let mut persisted_tenant = latest_tenant.clone(); + persisted_tenant.metadata.resource_version = Some("19".to_string()); + persisted_tenant + .status + .as_mut() + .expect("Tenant should have status") + .provisioning + .users = vec![first.clone(), second.clone()]; + + let service_requests = requests.clone(); + let service_patch = captured_patch.clone(); + let kube_service = service_fn(move |request: http::Request| { + let service_requests = service_requests.clone(); + let service_patch = service_patch.clone(); + let latest_tenant = latest_tenant.clone(); + let persisted_tenant = persisted_tenant.clone(); + async move { + let attempt = service_requests.fetch_add(1, Ordering::SeqCst); + let response_tenant = match attempt { + 0 => { + assert_eq!(request.method(), http::Method::GET); + latest_tenant + } + 1 => { + assert_eq!(request.method(), http::Method::PATCH); + let body = request + .into_body() + .collect() + .await + .expect("status patch body should be readable") + .to_bytes(); + *service_patch.lock().await = + serde_json::from_slice(&body).expect("status patch should be JSON"); + persisted_tenant + } + _ => panic!("unexpected Kubernetes request {attempt}"), + }; + Ok::<_, Infallible>( + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&response_tenant) + .expect("Tenant response should serialize"), + )) + .expect("response should build"), + ) + } + }); + let ctx = Context::new(Client::new(kube_service, "default")); + let run = ProvisioningRun { + ctx: &ctx, + tenant: &tenant, + namespace: "storage", + previous: ProvisioningStatus::default(), + now: "2026-08-02T00:00:00Z".to_string(), + status: ProvisioningStatus::default(), + failures: Vec::new(), + }; + + persist_user_ownership_checkpoints(&run, &[first, second]) + .await + .expect("all checkpoints should be persisted together"); + + assert_eq!(requests.load(Ordering::SeqCst), 2); + let patch = captured_patch.lock().await; + assert_eq!(patch["metadata"]["resourceVersion"], "18"); + assert_eq!( + patch["status"]["provisioning"]["users"] + .as_array() + .expect("users should be an array") + .len(), + 2 + ); + } + + #[test] + fn forbidden_checkpoint_write_is_permanent_but_conflict_retries() { + let api_error = |code, reason: &str| { + kube::Error::Api(kube::error::ErrorResponse { + status: "Failure".to_string(), + message: reason.to_string(), + reason: reason.to_string(), + code, + }) + }; + + assert!(matches!( + classify_checkpoint_kube_error(api_error(403, "Forbidden")), + CheckpointError::Permanent { .. } + )); + assert!(matches!( + classify_checkpoint_kube_error(api_error(409, "Conflict")), + CheckpointError::Retry(CheckpointRetry { + retry_after: CHECKPOINT_CONFLICT_RETRY, + .. + }) + )); + assert!(matches!( + classify_checkpoint_kube_error(api_error(503, "ServiceUnavailable")), + CheckpointError::Retry(CheckpointRetry { + retry_after: CHECKPOINT_TRANSIENT_RETRY, + .. + }) + )); + } + + #[tokio::test] + async fn unmanaged_existing_user_fails_before_credentials_or_policy_writes() { + let kube_requests = Arc::new(AtomicUsize::new(0)); + let kube_request_count = kube_requests.clone(); + let kube_service = service_fn(move |_request: http::Request| { + let kube_request_count = kube_request_count.clone(); + async move { + kube_request_count.fetch_add(1, Ordering::SeqCst); + Ok::<_, Infallible>( + http::Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .body(KubeBody::empty()) + .expect("response should build"), + ) + } + }); + let ctx = Context::new(Client::new(kube_service, "default")); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user.clone(), ProvisioningStatus::default()); + let run = ProvisioningRun { + ctx: &ctx, + tenant: &tenant, + namespace: "storage", + previous: ProvisioningStatus::default(), + now: "2026-08-02T00:00:00Z".to_string(), + status: ProvisioningStatus::default(), + failures: Vec::new(), + }; + + let write_requests = Arc::new(AtomicUsize::new(0)); + let add_requests = write_requests.clone(); + let policy_requests = write_requests.clone(); + let router = Router::new() + .route( + "/rustfs/admin/v3/user-info", + get(|| async { StatusCode::OK }), + ) + .route( + "/rustfs/admin/v3/add-user", + put(move || { + let add_requests = add_requests.clone(); + async move { + add_requests.fetch_add(1, Ordering::SeqCst); + StatusCode::OK + } + }), + ) + .route( + "/rustfs/admin/v3/set-policy", + put(move || { + let policy_requests = policy_requests.clone(); + async move { + policy_requests.fetch_add(1, Ordering::SeqCst); + StatusCode::OK + } + }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("test server should bind"); + let addr = listener.local_addr().expect("listener should have address"); + let server = tokio::spawn(async move { + axum::serve(listener, router) + .await + .expect("test server should serve") + }); + let client = + RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let credentials = user_credentials("1"); + + let item = reconcile_user( + &run, + &client, + &BTreeMap::from([("readwrite".to_string(), "{}".to_string())]), + &BTreeSet::new(), + &user, + &credentials, + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::UserOwnershipConflict.as_str()); + assert_eq!(write_requests.load(Ordering::SeqCst), 0); + assert_eq!(kube_requests.load(Ordering::SeqCst), 0); + let serialized = serde_json::to_string(&item).expect("status should serialize"); + assert!(!serialized.contains(&credentials.access_key)); + assert!(!serialized.contains(&credentials.secret_key)); + server.abort(); + } + + #[tokio::test] + async fn new_user_persists_pending_ownership_before_external_writes() { + let sequence = Arc::new(AtomicUsize::new(0)); + let captured_patch = Arc::new(Mutex::new(Value::Null)); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user.clone(), ProvisioningStatus::default()); + let mut latest_tenant = tenant.clone(); + latest_tenant.metadata.resource_version = Some("18".to_string()); + latest_tenant + .status + .as_mut() + .expect("Tenant should have status") + .current_state = "latest-controller-state".to_string(); + let mut persisted_tenant = latest_tenant.clone(); + persisted_tenant.metadata.resource_version = Some("19".to_string()); + let mut persisted_checkpoint = + owned_user_status(ProvisioningUserOwnershipState::PendingCreate, "5"); + persisted_checkpoint.state = ProvisioningItemState::Pending.as_str().to_string(); + persisted_tenant + .status + .as_mut() + .expect("Tenant should have status") + .provisioning + .users = vec![persisted_checkpoint]; + let kube_sequence = sequence.clone(); + let kube_patch = captured_patch.clone(); + let kube_service = service_fn(move |request: http::Request| { + let kube_sequence = kube_sequence.clone(); + let kube_patch = kube_patch.clone(); + let latest_tenant = latest_tenant.clone(); + let persisted_tenant = persisted_tenant.clone(); + async move { + let response_tenant = if request.method() == http::Method::GET { + assert!(request.uri().path().ends_with("/tenants/tenant-a")); + assert_eq!(kube_sequence.load(Ordering::SeqCst), 1); + kube_sequence.store(2, Ordering::SeqCst); + latest_tenant + } else { + assert_eq!(request.method(), http::Method::PATCH); + assert!(request.uri().path().ends_with("/tenants/tenant-a/status")); + assert_eq!(kube_sequence.load(Ordering::SeqCst), 2); + let body = request + .into_body() + .collect() + .await + .expect("status patch body should be readable") + .to_bytes(); + let patch: Value = + serde_json::from_slice(&body).expect("status patch should be JSON"); + *kube_patch.lock().await = patch; + kube_sequence.store(3, Ordering::SeqCst); + persisted_tenant + }; + Ok::<_, Infallible>( + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&response_tenant) + .expect("Tenant response should serialize"), + )) + .expect("response should build"), + ) + } + }); + let ctx = Context::new(Client::new(kube_service, "default")); + let run = ProvisioningRun { + ctx: &ctx, + tenant: &tenant, + namespace: "storage", + previous: ProvisioningStatus::default(), + now: "2026-08-02T00:00:00Z".to_string(), + status: ProvisioningStatus::default(), + failures: Vec::new(), + }; + + let get_sequence = sequence.clone(); + let add_sequence = sequence.clone(); + let policy_sequence = sequence.clone(); + let router = Router::new() + .route( + "/rustfs/admin/v3/user-info", + get(move || { + let get_sequence = get_sequence.clone(); + async move { + assert_eq!(get_sequence.load(Ordering::SeqCst), 0); + get_sequence.store(1, Ordering::SeqCst); + StatusCode::NOT_FOUND + } + }), + ) + .route( + "/rustfs/admin/v3/add-user", + put(move || { + let add_sequence = add_sequence.clone(); + async move { + assert_eq!(add_sequence.load(Ordering::SeqCst), 3); + add_sequence.store(4, Ordering::SeqCst); + StatusCode::OK + } + }), + ) + .route( + "/rustfs/admin/v3/set-policy", + put(move || { + let policy_sequence = policy_sequence.clone(); + async move { + assert_eq!(policy_sequence.load(Ordering::SeqCst), 4); + policy_sequence.store(5, Ordering::SeqCst); + StatusCode::OK + } + }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("test server should bind"); + let addr = listener.local_addr().expect("listener should have address"); + let server = tokio::spawn(async move { + axum::serve(listener, router) + .await + .expect("test server should serve") + }); + let client = + RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let credentials = user_credentials("5"); + + let item = reconcile_user( + &run, + &client, + &BTreeMap::from([("readwrite".to_string(), "{}".to_string())]), + &BTreeSet::new(), + &user, + &credentials, + ) + .await; + + assert_eq!(sequence.load(Ordering::SeqCst), 5); + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!( + item.ownership.as_ref().map(|ownership| ownership.state), + Some(ProvisioningUserOwnershipState::Managed) + ); + let patch = captured_patch.lock().await; + assert_eq!(patch["metadata"]["resourceVersion"], "18"); + assert_eq!(patch["status"]["currentState"], "latest-controller-state"); + let checkpoint = &patch["status"]["provisioning"]["users"][0]; + assert_eq!(checkpoint["name"], "app-user"); + assert_eq!(checkpoint["state"], "Pending"); + assert_eq!(checkpoint["ownership"]["state"], "PendingCreate"); + assert_eq!(checkpoint["ownership"]["tenantUid"], "tenant-uid-a"); + assert_eq!(checkpoint["ownership"]["userName"], "app-user"); + assert_eq!( + checkpoint["ownership"]["accessKeyHash"], + access_key_hash("appuser01") + ); + let serialized_patch = patch.to_string(); + assert!(!serialized_patch.contains(&credentials.access_key)); + assert!(!serialized_patch.contains(&credentials.secret_key)); + server.abort(); + } + + #[tokio::test] + async fn pruned_checkpoint_response_fails_before_external_user_writes() { + let kube_requests = Arc::new(AtomicUsize::new(0)); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user.clone(), ProvisioningStatus::default()); + let mut latest_tenant = tenant.clone(); + latest_tenant.metadata.resource_version = Some("18".to_string()); + let mut pruned_tenant = latest_tenant.clone(); + pruned_tenant.metadata.resource_version = Some("19".to_string()); + pruned_tenant + .status + .as_mut() + .expect("Tenant should have status") + .provisioning + .users = vec![ProvisioningUserStatus::new(ProvisioningItemStatus::new( + "app-user", + ProvisioningItemState::Pending, + Reason::ProvisioningPending.as_str(), + ))]; + let service_requests = kube_requests.clone(); + let kube_service = service_fn(move |request: http::Request| { + let service_requests = service_requests.clone(); + let latest_tenant = latest_tenant.clone(); + let pruned_tenant = pruned_tenant.clone(); + async move { + let attempt = service_requests.fetch_add(1, Ordering::SeqCst); + let response_tenant = match attempt { + 0 => { + assert_eq!(request.method(), http::Method::GET); + latest_tenant + } + 1 => { + assert_eq!(request.method(), http::Method::PATCH); + pruned_tenant + } + _ => panic!("unexpected Kubernetes request {attempt}"), + }; + Ok::<_, Infallible>( + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&response_tenant) + .expect("Tenant response should serialize"), + )) + .expect("response should build"), + ) + } + }); + let ctx = Context::new(Client::new(kube_service, "default")); + let run = ProvisioningRun { + ctx: &ctx, + tenant: &tenant, + namespace: "storage", + previous: ProvisioningStatus::default(), + now: "2026-08-02T00:00:00Z".to_string(), + status: ProvisioningStatus::default(), + failures: Vec::new(), + }; + + let write_requests = Arc::new(AtomicUsize::new(0)); + let add_requests = write_requests.clone(); + let policy_requests = write_requests.clone(); + let router = Router::new() + .route( + "/rustfs/admin/v3/user-info", + get(|| async { StatusCode::NOT_FOUND }), + ) + .route( + "/rustfs/admin/v3/add-user", + put(move || { + let add_requests = add_requests.clone(); + async move { + add_requests.fetch_add(1, Ordering::SeqCst); + StatusCode::OK + } + }), + ) + .route( + "/rustfs/admin/v3/set-policy", + put(move || { + let policy_requests = policy_requests.clone(); + async move { + policy_requests.fetch_add(1, Ordering::SeqCst); + StatusCode::OK + } + }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("test server should bind"); + let addr = listener.local_addr().expect("listener should have address"); + let server = tokio::spawn(async move { + axum::serve(listener, router) + .await + .expect("test server should serve") + }); + let client = + RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + + let item = reconcile_user( + &run, + &client, + &BTreeMap::from([("readwrite".to_string(), "{}".to_string())]), + &BTreeSet::new(), + &user, + &user_credentials("5"), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::UserOwnershipCheckpointFailed.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains("CRD")) + ); + assert_eq!(kube_requests.load(Ordering::SeqCst), 2); + assert_eq!(write_requests.load(Ordering::SeqCst), 0); + server.abort(); + } + + #[tokio::test] + async fn checkpoint_response_with_rewritten_state_is_rejected() { + let requests = Arc::new(AtomicUsize::new(0)); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut checkpoint = owned_user_status(ProvisioningUserOwnershipState::PendingCreate, "5"); + checkpoint.state = ProvisioningItemState::Pending.as_str().to_string(); + let mut latest_tenant = tenant.clone(); + latest_tenant.metadata.resource_version = Some("18".to_string()); + let mut rewritten_tenant = latest_tenant.clone(); + rewritten_tenant.metadata.resource_version = Some("19".to_string()); + let mut rewritten_checkpoint = checkpoint.clone(); + rewritten_checkpoint.state = ProvisioningItemState::Ready.as_str().to_string(); + rewritten_tenant + .status + .as_mut() + .expect("Tenant should have status") + .provisioning + .users = vec![rewritten_checkpoint]; + let service_requests = requests.clone(); + let kube_service = service_fn(move |request: http::Request| { + let service_requests = service_requests.clone(); + let latest_tenant = latest_tenant.clone(); + let rewritten_tenant = rewritten_tenant.clone(); + async move { + let attempt = service_requests.fetch_add(1, Ordering::SeqCst); + let response_tenant = match attempt { + 0 => { + assert_eq!(request.method(), http::Method::GET); + latest_tenant + } + 1 => { + assert_eq!(request.method(), http::Method::PATCH); + rewritten_tenant + } + _ => panic!("unexpected Kubernetes request {attempt}"), + }; + Ok::<_, Infallible>( + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&response_tenant) + .expect("Tenant response should serialize"), + )) + .expect("response should build"), + ) + } + }); + let ctx = Context::new(Client::new(kube_service, "default")); + let run = ProvisioningRun { + ctx: &ctx, + tenant: &tenant, + namespace: "storage", + previous: ProvisioningStatus::default(), + now: "2026-08-02T00:00:00Z".to_string(), + status: ProvisioningStatus::default(), + failures: Vec::new(), + }; + + let error = persist_user_ownership_checkpoints(&run, std::slice::from_ref(&checkpoint)) + .await + .expect_err("rewritten checkpoint state must be rejected"); + + assert!(matches!(error, CheckpointError::Permanent { .. })); + assert_eq!(requests.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn pending_user_checkpoint_recovers_after_create_before_status_crash() { + let kube_requests = Arc::new(AtomicUsize::new(0)); + let kube_request_count = kube_requests.clone(); + let kube_service = service_fn(move |_request: http::Request| { + let kube_request_count = kube_request_count.clone(); + async move { + kube_request_count.fetch_add(1, Ordering::SeqCst); + Ok::<_, Infallible>( + http::Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .body(KubeBody::empty()) + .expect("response should build"), + ) + } + }); + let ctx = Context::new(Client::new(kube_service, "default")); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let previous_user = owned_user_status(ProvisioningUserOwnershipState::PendingCreate, "5"); + let previous = ProvisioningStatus { + users: vec![previous_user], + ..Default::default() + }; + let tenant = provisioning_test_tenant(user.clone(), previous.clone()); + let run = ProvisioningRun { + ctx: &ctx, + tenant: &tenant, + namespace: "storage", + previous, + now: "2026-08-02T00:00:00Z".to_string(), + status: ProvisioningStatus::default(), + failures: Vec::new(), + }; + + let add_requests = Arc::new(AtomicUsize::new(0)); + let policy_requests = Arc::new(AtomicUsize::new(0)); + let add_count = add_requests.clone(); + let policy_count = policy_requests.clone(); + let router = Router::new() + .route( + "/rustfs/admin/v3/user-info", + get(|| async { StatusCode::OK }), + ) + .route( + "/rustfs/admin/v3/add-user", + put(move || { + let add_count = add_count.clone(); + async move { + add_count.fetch_add(1, Ordering::SeqCst); + StatusCode::OK + } + }), + ) + .route( + "/rustfs/admin/v3/set-policy", + put(move || { + let policy_count = policy_count.clone(); + async move { + policy_count.fetch_add(1, Ordering::SeqCst); + StatusCode::OK + } + }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("test server should bind"); + let addr = listener.local_addr().expect("listener should have address"); + let server = tokio::spawn(async move { + axum::serve(listener, router) + .await + .expect("test server should serve") + }); + let client = + RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + + let item = reconcile_user( + &run, + &client, + &BTreeMap::from([("readwrite".to_string(), "{}".to_string())]), + &BTreeSet::new(), + &user, + &user_credentials("5"), + ) + .await; + + assert_eq!(add_requests.load(Ordering::SeqCst), 0); + assert_eq!(policy_requests.load(Ordering::SeqCst), 1); + assert_eq!(kube_requests.load(Ordering::SeqCst), 0); + assert_eq!( + item.ownership.as_ref().map(|ownership| ownership.state), + Some(ProvisioningUserOwnershipState::Managed) + ); + server.abort(); + } + + #[tokio::test] + async fn retained_legacy_user_is_checkpointed_before_readd_and_secret_rotation() { + let sequence = Arc::new(AtomicUsize::new(0)); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let mut previous_user = owned_user_status(ProvisioningUserOwnershipState::Managed, "4"); + previous_user.ownership = None; + previous_user.state = ProvisioningItemState::Retained.as_str().to_string(); + let previous = ProvisioningStatus { + users: vec![previous_user], + ..Default::default() + }; + let tenant = provisioning_test_tenant(user.clone(), previous.clone()); + let mut latest_tenant = tenant.clone(); + latest_tenant.metadata.resource_version = Some("18".to_string()); + let mut persisted_tenant = latest_tenant.clone(); + persisted_tenant.metadata.resource_version = Some("19".to_string()); + let persisted_checkpoint = owned_user_status(ProvisioningUserOwnershipState::Managed, "4"); + persisted_tenant + .status + .as_mut() + .expect("Tenant should have status") + .provisioning + .users = vec![persisted_checkpoint]; + let kube_sequence = sequence.clone(); + let kube_service = service_fn(move |request: http::Request| { + let kube_sequence = kube_sequence.clone(); + let latest_tenant = latest_tenant.clone(); + let persisted_tenant = persisted_tenant.clone(); + async move { + let response_tenant = if request.method() == http::Method::GET { + assert_eq!(kube_sequence.load(Ordering::SeqCst), 1); + kube_sequence.store(2, Ordering::SeqCst); + latest_tenant + } else { + assert_eq!(request.method(), http::Method::PATCH); + assert_eq!(kube_sequence.load(Ordering::SeqCst), 2); + kube_sequence.store(3, Ordering::SeqCst); + persisted_tenant + }; + Ok::<_, Infallible>( + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&response_tenant) + .expect("Tenant response should serialize"), + )) + .expect("response should build"), + ) + } + }); + let ctx = Context::new(Client::new(kube_service, "default")); + let run = ProvisioningRun { + ctx: &ctx, + tenant: &tenant, + namespace: "storage", + previous, + now: "2026-08-02T00:00:00Z".to_string(), + status: ProvisioningStatus::default(), + failures: Vec::new(), + }; + + let add_requests = Arc::new(AtomicUsize::new(0)); + let policy_requests = Arc::new(AtomicUsize::new(0)); + let add_count = add_requests.clone(); + let policy_count = policy_requests.clone(); + let get_sequence = sequence.clone(); + let add_sequence = sequence.clone(); + let policy_sequence = sequence.clone(); + let router = Router::new() + .route( + "/rustfs/admin/v3/user-info", + get(move || { + let get_sequence = get_sequence.clone(); + async move { + assert_eq!(get_sequence.load(Ordering::SeqCst), 0); + get_sequence.store(1, Ordering::SeqCst); + StatusCode::OK + } + }), + ) + .route( + "/rustfs/admin/v3/add-user", + put(move || { + let add_count = add_count.clone(); + let add_sequence = add_sequence.clone(); + async move { + assert_eq!(add_sequence.load(Ordering::SeqCst), 3); + add_sequence.store(4, Ordering::SeqCst); + add_count.fetch_add(1, Ordering::SeqCst); + StatusCode::OK + } + }), + ) + .route( + "/rustfs/admin/v3/set-policy", + put(move || { + let policy_count = policy_count.clone(); + let policy_sequence = policy_sequence.clone(); + async move { + assert_eq!(policy_sequence.load(Ordering::SeqCst), 4); + policy_sequence.store(5, Ordering::SeqCst); + policy_count.fetch_add(1, Ordering::SeqCst); + StatusCode::OK + } + }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("test server should bind"); + let addr = listener.local_addr().expect("listener should have address"); + let server = tokio::spawn(async move { + axum::serve(listener, router) + .await + .expect("test server should serve") + }); + let client = + RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + + let item = reconcile_user( + &run, + &client, + &BTreeMap::from([("readwrite".to_string(), "{}".to_string())]), + &BTreeSet::new(), + &user, + &user_credentials("5"), + ) + .await; + + assert_eq!(add_requests.load(Ordering::SeqCst), 1); + assert_eq!(policy_requests.load(Ordering::SeqCst), 1); + assert_eq!(sequence.load(Ordering::SeqCst), 5); + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!(item.observed_secret_resource_version.as_deref(), Some("5")); + assert_eq!( + item.ownership.as_ref().map(|ownership| ownership.state), + Some(ProvisioningUserOwnershipState::Managed) + ); + server.abort(); + } + #[test] fn policy_document_hash_uses_compact_json() { let normalized = normalize_policy_document( @@ -1928,6 +3509,7 @@ mod tests { previous.observed_secret_resource_version = Some("1".to_string()); previous.observed_secret_name = Some("app-user".to_string()); previous.last_applied_access_key_hash = Some(access_key_hash("appuser01")); + let previous = ProvisioningUserStatus::new(previous); let credentials = UserCredentials { access_key: "appuser01".to_string(), secret_key: "rotated-secret".to_string(), @@ -1958,6 +3540,7 @@ mod tests { previous.observed_secret_resource_version = Some("1".to_string()); previous.observed_secret_name = Some("app-user".to_string()); previous.last_applied_access_key_hash = Some(access_key_hash("appuser01")); + let previous = ProvisioningUserStatus::new(previous); let credentials = UserCredentials { access_key: "appuser01".to_string(), secret_key: "unchanged-secret".to_string(), @@ -1982,6 +3565,7 @@ mod tests { previous.observed_secret_resource_version = Some("1".to_string()); previous.observed_secret_name = Some("app-user".to_string()); previous.last_applied_access_key_hash = Some(access_key_hash("appuser01")); + let previous = ProvisioningUserStatus::new(previous); let credentials = UserCredentials { access_key: "appuser01".to_string(), secret_key: "rotated-secret".to_string(), @@ -2005,6 +3589,7 @@ mod tests { ); previous.observed_secret_resource_version = Some("1".to_string()); previous.last_applied_access_key_hash = Some(access_key_hash("appuser01")); + let previous = ProvisioningUserStatus::new(previous); let credentials = UserCredentials { access_key: "appuser01".to_string(), secret_key: "unchanged-secret".to_string(), @@ -2027,6 +3612,7 @@ mod tests { Reason::ProvisioningConfigured.as_str(), ); previous.last_applied_access_key_hash = Some(access_key_hash("appuser01")); + let previous = ProvisioningUserStatus::new(previous); let credentials = UserCredentials { access_key: "otheruser".to_string(), secret_key: "rotated-secret".to_string(), diff --git a/src/status.rs b/src/status.rs index 701f5d86..6715d958 100644 --- a/src/status.rs +++ b/src/status.rs @@ -665,6 +665,32 @@ mod tests { use super::*; use crate::types::v1alpha1::status::Condition; + #[test] + fn user_ownership_provisioning_failures_block_top_level_status() { + for reason in [ + Reason::UserOwnershipConflict, + Reason::UserOwnershipCheckpointFailed, + ] { + let tenant = crate::tests::create_test_tenant(None, None); + let mut builder = StatusBuilder::from_tenant(&tenant); + builder.finish_provisioning_failed(reason, "ownership safety check failed".to_string()); + let status = builder.build(); + + assert_eq!(status.current_state, "Blocked"); + assert_eq!( + crate::types::v1alpha1::status::primary_condition(&status) + .map(|condition| condition.reason.as_str()), + Some(reason.as_str()) + ); + assert_eq!( + status + .condition(ConditionType::ProvisioningReady) + .map(|condition| condition.status.as_str()), + Some("False") + ); + } + } + #[test] fn status_builder_maps_credential_missing_key() { let tenant = crate::tests::create_test_tenant(None, None); diff --git a/src/types/v1alpha1.rs b/src/types/v1alpha1.rs index f5e65a00..f2c73006 100755 --- a/src/types/v1alpha1.rs +++ b/src/types/v1alpha1.rs @@ -279,6 +279,26 @@ mod tenant_provisioning_crd_tests { ["nullable"], json!(true) ); + let ownership = &status["properties"]["provisioning"]["properties"]["users"]["items"]["properties"] + ["ownership"]; + assert_eq!(ownership["type"], json!("object")); + assert_eq!(ownership["nullable"], json!(true)); + assert_eq!( + ownership["required"], + json!(["accessKeyHash", "state", "tenantUid", "userName"]) + ); + assert_eq!( + ownership["properties"]["state"]["enum"], + json!(["PendingCreate", "Managed"]) + ); + for kind in ["policies", "buckets"] { + assert!( + status["properties"]["provisioning"]["properties"][kind]["items"] + ["properties"]["ownership"] + .is_null(), + "ownership must only be exposed on provisioning user status" + ); + } assert!( spec["properties"]["policies"]["x-kubernetes-validations"].is_null(), diff --git a/src/types/v1alpha1/status.rs b/src/types/v1alpha1/status.rs index 7f785de3..c7860bf4 100755 --- a/src/types/v1alpha1/status.rs +++ b/src/types/v1alpha1/status.rs @@ -171,6 +171,8 @@ pub enum Reason { UserPolicyNotFound, UserPolicyInvalid, UserPolicySetFailed, + UserOwnershipConflict, + UserOwnershipCheckpointFailed, BucketCreateFailed, BucketObjectLockConflict, KubernetesApiError, @@ -240,6 +242,8 @@ impl Reason { Self::UserPolicyNotFound => "UserPolicyNotFound", Self::UserPolicyInvalid => "UserPolicyInvalid", Self::UserPolicySetFailed => "UserPolicySetFailed", + Self::UserOwnershipConflict => "UserOwnershipConflict", + Self::UserOwnershipCheckpointFailed => "UserOwnershipCheckpointFailed", Self::BucketCreateFailed => "BucketCreateFailed", Self::BucketObjectLockConflict => "BucketObjectLockConflict", Self::KubernetesApiError => "KubernetesApiError", @@ -512,6 +516,8 @@ pub fn is_blocked_reason(reason: &str) -> bool { | "UserPolicyNotFound" | "UserPolicyInvalid" | "UserPolicySetFailed" + | "UserOwnershipConflict" + | "UserOwnershipCheckpointFailed" | "BucketCreateFailed" | "BucketObjectLockConflict" ) @@ -600,6 +606,14 @@ pub fn next_actions_for_reason(reason: &str) -> Vec<&'static str> { "UserPolicyNotFound" => vec!["createPolicy", "fixUserPolicyList"], "UserPolicyInvalid" => vec!["fixUserPolicyList"], "UserPolicySetFailed" => vec!["inspectUserPolicyMapping", "inspectOperatorLogs"], + "UserOwnershipConflict" => vec!["inspectRustfsUser", "chooseDifferentAccessKey"], + "UserOwnershipCheckpointFailed" => { + vec![ + "upgradeTenantCrd", + "inspectOperatorRbac", + "inspectOperatorLogs", + ] + } "BucketCreateFailed" => vec!["inspectBucket", "inspectOperatorLogs"], "BucketObjectLockConflict" => vec!["createObjectLockBucket", "fixBucketSpec"], "KubernetesApiError" => vec!["retry", "inspectOperatorLogs"], @@ -677,6 +691,24 @@ mod tests { ); } + #[test] + fn user_ownership_failures_are_blocked_and_actionable() { + assert!(is_blocked_reason("UserOwnershipConflict")); + assert!(is_blocked_reason("UserOwnershipCheckpointFailed")); + assert_eq!( + next_actions_for_reason("UserOwnershipConflict"), + vec!["inspectRustfsUser", "chooseDifferentAccessKey"] + ); + assert_eq!( + next_actions_for_reason("UserOwnershipCheckpointFailed"), + vec![ + "upgradeTenantCrd", + "inspectOperatorRbac", + "inspectOperatorLogs" + ] + ); + } + #[test] fn tls_blocked_reasons_are_primary_and_actionable() { let status = Status { diff --git a/src/types/v1alpha1/status/provisioning.rs b/src/types/v1alpha1/status/provisioning.rs index af976bb4..bcc56fb2 100644 --- a/src/types/v1alpha1/status/provisioning.rs +++ b/src/types/v1alpha1/status/provisioning.rs @@ -14,6 +14,7 @@ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +use std::ops::{Deref, DerefMut}; use utoipa::ToSchema; #[derive(Deserialize, Serialize, Clone, Debug, JsonSchema, ToSchema, Default, PartialEq, Eq)] @@ -29,7 +30,7 @@ pub struct ProvisioningStatus { pub policies: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub users: Vec, + pub users: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub buckets: Vec, @@ -41,7 +42,7 @@ impl ProvisioningStatus { } } -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema, ToSchema, PartialEq, Eq)] +#[derive(Deserialize, Serialize, Clone, Copy, Debug, JsonSchema, ToSchema, PartialEq, Eq)] #[serde(rename_all = "PascalCase")] pub enum ProvisioningPhase { Pending, @@ -58,6 +59,64 @@ pub enum ProvisioningItemState { Retained, } +#[derive(Deserialize, Serialize, Clone, Copy, Debug, JsonSchema, ToSchema, PartialEq, Eq)] +#[serde(rename_all = "PascalCase")] +pub enum ProvisioningUserOwnershipState { + PendingCreate, + Managed, +} + +/// Durable proof that the operator claimed a RustFS user identity before mutating it. +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema, ToSchema, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct ProvisioningUserOwnershipStatus { + pub state: ProvisioningUserOwnershipState, + pub tenant_uid: String, + pub user_name: String, + pub access_key_hash: String, +} + +/// User-specific provisioning status. The flattened item preserves the existing status wire +/// format while keeping ownership metadata out of policy and bucket status schemas. +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema, ToSchema, Default, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct ProvisioningUserStatus { + #[serde(flatten)] + pub item: ProvisioningItemStatus, + + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ownership: Option, +} + +impl ProvisioningUserStatus { + pub fn new(item: ProvisioningItemStatus) -> Self { + Self { + item, + ownership: None, + } + } +} + +impl Deref for ProvisioningUserStatus { + type Target = ProvisioningItemStatus; + + fn deref(&self) -> &Self::Target { + &self.item + } +} + +impl DerefMut for ProvisioningUserStatus { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.item + } +} + +impl AsRef for ProvisioningUserStatus { + fn as_ref(&self) -> &ProvisioningItemStatus { + &self.item + } +} + impl ProvisioningItemState { pub const fn as_str(&self) -> &'static str { match self { @@ -126,3 +185,9 @@ impl ProvisioningItemStatus { } } } + +impl AsRef for ProvisioningItemStatus { + fn as_ref(&self) -> &ProvisioningItemStatus { + self + } +}