From 414ad0ad4c89454a533232c9d4ef345c32f94298 Mon Sep 17 00:00:00 2001 From: GatewayJ <18332154+GatewayJ@users.noreply.github.com> Date: Sun, 2 Aug 2026 04:06:05 +0800 Subject: [PATCH 1/3] fix(provisioning): protect existing RustFS users --- deploy/rustfs-operator/crds/tenant-crd.yaml | 63 ++ src/reconcile/phases.rs | 13 + src/reconcile/provisioning.rs | 1078 ++++++++++++++++++- src/types/v1alpha1.rs | 12 + src/types/v1alpha1/status.rs | 5 + src/types/v1alpha1/status/provisioning.rs | 22 +- 6 files changed, 1187 insertions(+), 6 deletions(-) diff --git a/deploy/rustfs-operator/crds/tenant-crd.yaml b/deploy/rustfs-operator/crds/tenant-crd.yaml index e8aab6b..f97e860 100644 --- a/deploy/rustfs-operator/crds/tenant-crd.yaml +++ b/deploy/rustfs-operator/crds/tenant-crd.yaml @@ -2511,6 +2511,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 @@ -2573,6 +2594,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 @@ -2623,6 +2665,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/src/reconcile/phases.rs b/src/reconcile/phases.rs index d6e808b..eae7774 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 5897f49..e225191 100644 --- a/src/reconcile/provisioning.rs +++ b/src/reconcile/provisioning.rs @@ -21,15 +21,23 @@ use crate::types::v1alpha1::provisioning::{ use crate::types::v1alpha1::status::Reason; use crate::types::v1alpha1::status::provisioning::{ ProvisioningItemState, ProvisioningItemStatus, ProvisioningPhase, ProvisioningStatus, + ProvisioningUserOwnershipState, ProvisioningUserOwnershipStatus, }; 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::sync::Mutex as StdMutex; +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 +45,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> { @@ -49,6 +78,8 @@ struct ProvisioningRun<'a> { now: String, status: ProvisioningStatus, failures: Vec<(Reason, String)>, + status_resource_version: StdMutex>, + checkpoint_retry: StdMutex>, } struct UserCredentials { @@ -218,6 +249,7 @@ impl ProvisioningRun<'_> { item.observed_secret_resource_version = previous.observed_secret_resource_version.clone(); item.observed_secret_name = previous.observed_secret_name.clone(); item.last_applied_access_key_hash = previous.last_applied_access_key_hash.clone(); + item.ownership = previous.ownership.clone(); item.policies = previous.policies.clone(); item.region = previous.region.clone(); item.object_lock = previous.object_lock; @@ -253,6 +285,7 @@ impl ProvisioningRun<'_> { previous.observed_secret_resource_version.clone(); item.observed_secret_name = previous.observed_secret_name.clone(); item.last_applied_access_key_hash = previous.last_applied_access_key_hash.clone(); + item.ownership = previous.ownership.clone(); item.policies = previous.policies.clone(); } self.push_user(item); @@ -308,6 +341,15 @@ impl ProvisioningRun<'_> { } fn finish(mut self) -> ProvisioningReconcileResult { + if let Some(retry) = self.checkpoint_retry() { + return ProvisioningReconcileResult { + status: self.status, + outcome: ProvisioningOutcome::Retry { + message: retry.message, + retry_after: retry.retry_after, + }, + }; + } let outcome = self .failures .first() @@ -320,6 +362,7 @@ impl ProvisioningRun<'_> { ProvisioningOutcome::Ready => ProvisioningPhase::Ready, ProvisioningOutcome::Pending { .. } => ProvisioningPhase::Pending, ProvisioningOutcome::Failed { .. } => ProvisioningPhase::Failed, + ProvisioningOutcome::Retry { .. } => ProvisioningPhase::Pending, }; self.prepare_status(phase); @@ -328,6 +371,25 @@ impl ProvisioningRun<'_> { outcome, } } + + fn request_checkpoint_retry(&self, pending_retry: CheckpointRetry) { + if let Ok(mut retry) = self.checkpoint_retry.lock() { + *retry = Some(pending_retry); + } + } + + fn checkpoint_retry(&self) -> Option { + self.checkpoint_retry + .lock() + .map(|retry| retry.clone()) + .unwrap_or_else(|_| { + Some(CheckpointRetry { + message: "RustFS user ownership checkpoint retry state is unavailable" + .to_string(), + retry_after: CHECKPOINT_TRANSIENT_RETRY, + }) + }) + } } pub(super) async fn reconcile_provisioning( @@ -349,6 +411,8 @@ pub(super) async fn reconcile_provisioning( now, status: ProvisioningStatus::default(), failures: Vec::new(), + status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), + checkpoint_retry: StdMutex::new(None), }; if !has_active_spec(tenant) { @@ -418,6 +482,9 @@ 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 run.checkpoint_retry().is_some() { + return run.finish(); + } reconcile_buckets(&mut run, &client).await; run.finish() } @@ -809,6 +876,9 @@ async fn reconcile_users( ) .await; run.push_user(item); + if run.checkpoint_retry().is_some() { + break; + } } } @@ -920,6 +990,156 @@ async fn reconcile_user( } }; + 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 annotate_user_item(item, user, previous, 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 annotate_user_item(item, user, previous, None); + } + }; + let mut 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. + checkpoint = annotate_user_item(checkpoint, user, previous, None); + checkpoint.ownership = Some(managed_ownership.clone()); + if let Err(error) = persist_user_ownership_checkpoint(run, checkpoint).await { + let message = handle_checkpoint_error(run, error); + let item = run.item( + previous, + &user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipCheckpointFailed, + message, + ); + return annotate_user_item(item, user, previous, None); + } + 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 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 annotate_user_item(item, user, previous, None); + } + }; + let mut checkpoint = run.item( + previous, + &user.name, + ProvisioningItemState::Pending, + Reason::ProvisioningPending, + "Operator ownership checkpoint was persisted before creating the RustFS user", + ); + checkpoint = annotate_user_item(checkpoint, user, previous, Some(credentials)); + checkpoint.ownership = Some(pending_ownership.clone()); + if let Err(error) = persist_user_ownership_checkpoint(run, checkpoint).await { + let message = handle_checkpoint_error(run, error); + let item = run.item( + previous, + &user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipCheckpointFailed, + message, + ); + return annotate_user_item(item, user, previous, None); + } + ownership = Some(pending_ownership); + } else if !exists + && ownership.as_ref().is_some_and(|ownership| { + ownership.state == ProvisioningUserOwnershipState::PendingCreate + }) + { + // Refresh the persisted intent with a resourceVersion precondition before retrying an + // external create after a process crash. This keeps concurrent reconcilers from both + // crossing the Kubernetes-to-RustFS side-effect boundary. + let mut checkpoint = run.item( + previous, + &user.name, + ProvisioningItemState::Pending, + Reason::ProvisioningPending, + "Operator is resuming a pending RustFS user creation", + ); + checkpoint = annotate_user_item(checkpoint, user, previous, Some(credentials)); + checkpoint.ownership = ownership.clone(); + if let Err(error) = persist_user_ownership_checkpoint(run, checkpoint).await { + let message = handle_checkpoint_error(run, error); + let item = run.item( + previous, + &user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipCheckpointFailed, + message, + ); + return annotate_user_item(item, user, previous, None); + } + } + + let Some(mut 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 annotate_user_item(item, user, previous, None); + }; + let credentials_applied = match sync_user_credentials(client, previous, credentials, exists).await { Ok(applied) => applied, @@ -931,10 +1151,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 +1170,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,7 +1198,198 @@ 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 +} + +fn matching_user_ownership( + previous: Option<&ProvisioningItemStatus>, + 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<&ProvisioningItemStatus>, + user: &ProvisioningUser, + credentials: &UserCredentials, +) -> bool { + let Some(previous) = previous else { + return false; + }; + let current_access_key_hash = access_key_hash(&credentials.access_key); + previous.state == ProvisioningItemState::Ready.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_checkpoint( + run: &ProvisioningRun<'_>, + checkpoint: ProvisioningItemStatus, +) -> Result<(), CheckpointError> { + let resource_version = run + .status_resource_version + .lock() + .map_err(|_| CheckpointError::Permanent { + message: "Tenant status checkpoint lock is unavailable".to_string(), + })? + .clone(); + let Some(resource_version) = resource_version 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 = run.previous.clone(); + merge_provisioning_items(&mut provisioning.policies, &run.status.policies); + merge_provisioning_items(&mut provisioning.users, &run.status.users); + merge_provisioning_items(&mut provisioning.buckets, &run.status.buckets); + merge_provisioning_items(&mut provisioning.users, std::slice::from_ref(&checkpoint)); + 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 = run.tenant.status.clone().unwrap_or_default(); + status.provisioning = provisioning; + let status_patch = serde_json::json!({ + "metadata": { "resourceVersion": resource_version }, + "status": status, + }); + let api: Api = Api::namespaced(run.ctx.client.clone(), run.namespace); + let updated = api + .patch_status( + &run.tenant.name(), + &PatchParams::default(), + &Patch::Merge(&status_patch), + ) + .await + .map_err(classify_checkpoint_kube_error)?; + let Some(next_resource_version) = updated.metadata.resource_version else { + 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, + })); + }; + *run.status_resource_version + .lock() + .map_err(|_| CheckpointError::Retry(CheckpointRetry { + message: "RustFS user ownership checkpoint was persisted, but its new resourceVersion could not be retained locally" + .to_string(), + retry_after: CHECKPOINT_TRANSIENT_RETRY, + }))? = + Some(next_resource_version); + 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, + }), + } +} + +fn handle_checkpoint_error(run: &ProvisioningRun<'_>, error: CheckpointError) -> String { + match error { + CheckpointError::Permanent { message } => message, + CheckpointError::Retry(retry) => { + let message = retry.message.clone(); + run.request_checkpoint_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 annotate_user_item( @@ -995,6 +1412,7 @@ fn annotate_user_item( previous.and_then(|item| item.last_applied_access_key_hash.clone()); } } + item.ownership = previous.and_then(|item| item.ownership.clone()); item.policies = user.policies.clone(); item } @@ -1471,6 +1889,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 +1907,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}; @@ -1606,6 +2027,8 @@ mod tests { now: "2026-07-18T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), + checkpoint_retry: StdMutex::new(None), }; let request_count = Arc::new(AtomicUsize::new(0)); @@ -1692,6 +2115,8 @@ mod tests { now: "2026-07-18T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), + checkpoint_retry: StdMutex::new(None), }; let client = RustfsAdminClient::new_with_base_url( @@ -1745,6 +2170,649 @@ 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, + ) -> ProvisioningItemStatus { + let mut item = ProvisioningItemStatus::new( + "app-user", + ProvisioningItemState::Ready, + Reason::ProvisioningConfigured.as_str(), + ); + 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_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::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 updated_tenant = tenant.clone(); + updated_tenant.metadata.resource_version = Some("18".to_string()); + let service_requests = requests.clone(); + let kube_service = service_fn(move |_request: http::Request| { + let service_requests = service_requests.clone(); + let updated_tenant = updated_tenant.clone(); + async move { + let attempt = service_requests.fetch_add(1, Ordering::SeqCst); + let response = if attempt == 0 { + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&updated_tenant) + .expect("Tenant response should serialize"), + )) + } else { + http::Response::builder() + .status(StatusCode::CONFLICT) + .header("content-type", "application/json") + .body(KubeBody::from( + br#"{"kind":"Status","apiVersion":"v1","status":"Failure","message":"resourceVersion conflict","reason":"Conflict","code":409}"#.to_vec(), + )) + }; + 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(), + status_resource_version: StdMutex::new(Some("17".to_string())), + checkpoint_retry: StdMutex::new(None), + }; + let winner = make_run(); + let loser = make_run(); + let mut checkpoint = owned_user_status(ProvisioningUserOwnershipState::PendingCreate, "5"); + checkpoint.state = ProvisioningItemState::Pending.as_str().to_string(); + + persist_user_ownership_checkpoint(&winner, checkpoint.clone()) + .await + .expect("first reconciler should persist its checkpoint"); + let error = persist_user_ownership_checkpoint(&loser, checkpoint) + .await + .expect_err("stale reconciler should lose the CAS"); + handle_checkpoint_error(&loser, error); + + match loser.finish().outcome { + ProvisioningOutcome::Retry { retry_after, .. } => { + assert_eq!(retry_after, CHECKPOINT_CONFLICT_RETRY); + } + _ => panic!("stale checkpoint writer should retry"), + } + assert_eq!(requests.load(Ordering::SeqCst), 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(), + status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), + checkpoint_retry: StdMutex::new(None), + }; + + 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 response_tenant = tenant.clone(); + response_tenant.metadata.resource_version = Some("18".to_string()); + 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 response_tenant = response_tenant.clone(); + async move { + assert!(request.uri().path().ends_with("/tenants/tenant-a/status")); + assert_eq!(kube_sequence.load(Ordering::SeqCst), 1); + 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(2, Ordering::SeqCst); + 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(), + status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), + checkpoint_retry: StdMutex::new(None), + }; + + 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), 2); + add_sequence.store(3, 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), 3); + policy_sequence.store(4, 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), 4); + assert_eq!( + run.status_resource_version + .lock() + .expect("status resourceVersion lock should be available") + .as_deref(), + Some("18") + ); + 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"], "17"); + let checkpoint = &patch["status"]["provisioning"]["users"][0]; + 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 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(), + status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), + checkpoint_retry: StdMutex::new(None), + }; + + 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 legacy_managed_user_is_checkpointed_before_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; + let previous = ProvisioningStatus { + users: vec![previous_user], + ..Default::default() + }; + let tenant = provisioning_test_tenant(user.clone(), previous.clone()); + let mut response_tenant = tenant.clone(); + response_tenant.metadata.resource_version = Some("18".to_string()); + let kube_sequence = sequence.clone(); + let kube_service = service_fn(move |_request: http::Request| { + let kube_sequence = kube_sequence.clone(); + let response_tenant = response_tenant.clone(); + async move { + assert_eq!(kube_sequence.load(Ordering::SeqCst), 1); + kube_sequence.store(2, Ordering::SeqCst); + 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(), + status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), + checkpoint_retry: StdMutex::new(None), + }; + + 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), 2); + add_sequence.store(3, 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), 3); + policy_sequence.store(4, 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), 4); + 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( diff --git a/src/types/v1alpha1.rs b/src/types/v1alpha1.rs index f5e65a0..06f01da 100755 --- a/src/types/v1alpha1.rs +++ b/src/types/v1alpha1.rs @@ -279,6 +279,18 @@ 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"]) + ); assert!( spec["properties"]["policies"]["x-kubernetes-validations"].is_null(), diff --git a/src/types/v1alpha1/status.rs b/src/types/v1alpha1/status.rs index 7f785de..f4f8256 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,7 @@ pub fn is_blocked_reason(reason: &str) -> bool { | "UserPolicyNotFound" | "UserPolicyInvalid" | "UserPolicySetFailed" + | "UserOwnershipConflict" | "BucketCreateFailed" | "BucketObjectLockConflict" ) diff --git a/src/types/v1alpha1/status/provisioning.rs b/src/types/v1alpha1/status/provisioning.rs index af976bb..34d574a 100644 --- a/src/types/v1alpha1/status/provisioning.rs +++ b/src/types/v1alpha1/status/provisioning.rs @@ -41,7 +41,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 +58,23 @@ 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, +} + impl ProvisioningItemState { pub const fn as_str(&self) -> &'static str { match self { @@ -102,6 +119,9 @@ pub struct ProvisioningItemStatus { #[serde(default, skip_serializing_if = "Option::is_none")] pub last_applied_access_key_hash: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ownership: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub policies: Vec, From 3deb26eddd85e0c1e9110bbeced5cbfd75dd6cf7 Mon Sep 17 00:00:00 2001 From: GatewayJ <18332154+GatewayJ@users.noreply.github.com> Date: Sun, 2 Aug 2026 12:17:26 +0800 Subject: [PATCH 2/3] fix(provisioning): harden ownership checkpoints --- deploy/rustfs-operator/crds/tenant-crd.yaml | 45 +- src/reconcile/provisioning.rs | 580 +++++++++++++++----- src/status.rs | 26 + src/types/v1alpha1.rs | 8 + src/types/v1alpha1/status.rs | 27 + src/types/v1alpha1/status/provisioning.rs | 53 +- 6 files changed, 566 insertions(+), 173 deletions(-) diff --git a/deploy/rustfs-operator/crds/tenant-crd.yaml b/deploy/rustfs-operator/crds/tenant-crd.yaml index f97e860..b9da934 100644 --- a/deploy/rustfs-operator/crds/tenant-crd.yaml +++ b/deploy/rustfs-operator/crds/tenant-crd.yaml @@ -2511,27 +2511,6 @@ 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 @@ -2594,27 +2573,6 @@ 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 @@ -2634,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 diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs index e225191..f57e924 100644 --- a/src/reconcile/provisioning.rs +++ b/src/reconcile/provisioning.rs @@ -21,7 +21,7 @@ use crate::types::v1alpha1::provisioning::{ use crate::types::v1alpha1::status::Reason; use crate::types::v1alpha1::status::provisioning::{ ProvisioningItemState, ProvisioningItemStatus, ProvisioningPhase, ProvisioningStatus, - ProvisioningUserOwnershipState, ProvisioningUserOwnershipStatus, + ProvisioningUserOwnershipState, ProvisioningUserOwnershipStatus, ProvisioningUserStatus, }; use crate::types::v1alpha1::tenant::Tenant; use k8s_openapi::ByteString; @@ -78,7 +78,6 @@ struct ProvisioningRun<'a> { now: String, status: ProvisioningStatus, failures: Vec<(Reason, String)>, - status_resource_version: StdMutex>, checkpoint_retry: StdMutex>, } @@ -132,7 +131,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) } @@ -149,8 +148,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))); @@ -211,14 +214,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()); @@ -249,13 +256,18 @@ impl ProvisioningRun<'_> { item.observed_secret_resource_version = previous.observed_secret_resource_version.clone(); item.observed_secret_name = previous.observed_secret_name.clone(); item.last_applied_access_key_hash = previous.last_applied_access_key_hash.clone(); - item.ownership = previous.ownership.clone(); item.policies = previous.policies.clone(); item.region = previous.region.clone(); item.object_lock = previous.object_lock; 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( @@ -285,9 +297,12 @@ impl ProvisioningRun<'_> { previous.observed_secret_resource_version.clone(); item.observed_secret_name = previous.observed_secret_name.clone(); item.last_applied_access_key_hash = previous.last_applied_access_key_hash.clone(); - item.ownership = previous.ownership.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 { @@ -317,7 +332,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)); } } @@ -411,7 +426,6 @@ pub(super) async fn reconcile_provisioning( now, status: ProvisioningStatus::default(), failures: Vec::new(), - status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), checkpoint_retry: StdMutex::new(None), }; @@ -933,7 +947,7 @@ async fn reconcile_user( failed_spec_policies: &BTreeSet, user: &ProvisioningUser, credentials: &UserCredentials, -) -> ProvisioningItemStatus { +) -> ProvisioningUserStatus { let previous = run.previous_user(&user.name); if user_access_key_changed(previous, credentials) { let item = run.item( @@ -1024,7 +1038,7 @@ async fn reconcile_user( return annotate_user_item(item, user, previous, None); } }; - let mut checkpoint = run.item( + let checkpoint = run.item( previous, &user.name, ProvisioningItemState::Ready, @@ -1033,7 +1047,7 @@ async fn reconcile_user( ); // Preserve the legacy observed Secret version so a concurrently rotated Secret is // still applied after the ownership checkpoint has been persisted. - checkpoint = annotate_user_item(checkpoint, user, previous, None); + let mut checkpoint = annotate_user_item(checkpoint, user, previous, None); checkpoint.ownership = Some(managed_ownership.clone()); if let Err(error) = persist_user_ownership_checkpoint(run, checkpoint).await { let message = handle_checkpoint_error(run, error); @@ -1078,14 +1092,14 @@ async fn reconcile_user( return annotate_user_item(item, user, previous, None); } }; - let mut checkpoint = run.item( + let checkpoint = run.item( previous, &user.name, ProvisioningItemState::Pending, Reason::ProvisioningPending, "Operator ownership checkpoint was persisted before creating the RustFS user", ); - checkpoint = annotate_user_item(checkpoint, user, previous, Some(credentials)); + let mut checkpoint = annotate_user_item(checkpoint, user, previous, Some(credentials)); checkpoint.ownership = Some(pending_ownership.clone()); if let Err(error) = persist_user_ownership_checkpoint(run, checkpoint).await { let message = handle_checkpoint_error(run, error); @@ -1104,17 +1118,17 @@ async fn reconcile_user( ownership.state == ProvisioningUserOwnershipState::PendingCreate }) { - // Refresh the persisted intent with a resourceVersion precondition before retrying an - // external create after a process crash. This keeps concurrent reconcilers from both - // crossing the Kubernetes-to-RustFS side-effect boundary. - let mut checkpoint = run.item( + // 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 checkpoint = run.item( previous, &user.name, ProvisioningItemState::Pending, Reason::ProvisioningPending, "Operator is resuming a pending RustFS user creation", ); - checkpoint = annotate_user_item(checkpoint, user, previous, Some(credentials)); + let mut checkpoint = annotate_user_item(checkpoint, user, previous, Some(credentials)); checkpoint.ownership = ownership.clone(); if let Err(error) = persist_user_ownership_checkpoint(run, checkpoint).await { let message = handle_checkpoint_error(run, error); @@ -1204,7 +1218,7 @@ async fn reconcile_user( } fn matching_user_ownership( - previous: Option<&ProvisioningItemStatus>, + previous: Option<&ProvisioningUserStatus>, tenant: &Tenant, user: &ProvisioningUser, credentials: &UserCredentials, @@ -1231,7 +1245,7 @@ fn matching_user_ownership( } fn legacy_user_status_can_migrate( - previous: Option<&ProvisioningItemStatus>, + previous: Option<&ProvisioningUserStatus>, user: &ProvisioningUser, credentials: &UserCredentials, ) -> bool { @@ -1239,9 +1253,11 @@ fn legacy_user_status_can_migrate( return false; }; let current_access_key_hash = access_key_hash(&credentials.access_key); - previous.state == ProvisioningItemState::Ready.as_str() - && previous.last_applied_access_key_hash.as_deref() - == Some(current_access_key_hash.as_str()) + 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()) } @@ -1266,27 +1282,57 @@ fn user_ownership( async fn persist_user_ownership_checkpoint( run: &ProvisioningRun<'_>, - checkpoint: ProvisioningItemStatus, + checkpoint: ProvisioningUserStatus, ) -> Result<(), CheckpointError> { - let resource_version = run - .status_resource_version - .lock() - .map_err(|_| CheckpointError::Permanent { - message: "Tenant status checkpoint lock is unavailable".to_string(), - })? - .clone(); - let Some(resource_version) = resource_version else { + 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, + })); + } + 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: "Tenant user provisioning status changed before persisting the RustFS user ownership checkpoint" + .to_string(), + 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 = run.previous.clone(); + 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_items(&mut provisioning.users, &run.status.users); + merge_provisioning_user_items(&mut provisioning.users, &run.status.users); merge_provisioning_items(&mut provisioning.buckets, &run.status.buckets); - merge_provisioning_items(&mut provisioning.users, std::slice::from_ref(&checkpoint)); + merge_provisioning_user_items(&mut provisioning.users, std::slice::from_ref(&checkpoint)); provisioning.observed_generation = run.tenant.metadata.generation; provisioning.phase = Some(ProvisioningPhase::Pending); provisioning @@ -1299,13 +1345,12 @@ async fn persist_user_ownership_checkpoint( .buckets .sort_by(|left, right| left.name.cmp(&right.name)); - let mut status = run.tenant.status.clone().unwrap_or_default(); + let mut status = latest.status.unwrap_or_default(); status.provisioning = provisioning; let status_patch = serde_json::json!({ "metadata": { "resourceVersion": resource_version }, "status": status, }); - let api: Api = Api::namespaced(run.ctx.client.clone(), run.namespace); let updated = api .patch_status( &run.tenant.name(), @@ -1314,21 +1359,28 @@ async fn persist_user_ownership_checkpoint( ) .await .map_err(classify_checkpoint_kube_error)?; - let Some(next_resource_version) = updated.metadata.resource_version else { + 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, })); - }; - *run.status_resource_version - .lock() - .map_err(|_| CheckpointError::Retry(CheckpointRetry { - message: "RustFS user ownership checkpoint was persisted, but its new resourceVersion could not be retained locally" + } + 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: "Kubernetes accepted the RustFS user ownership checkpoint request but did not persist the expected state and ownership proof; ensure the Tenant CRD is upgraded before the Operator" .to_string(), - retry_after: CHECKPOINT_TRANSIENT_RETRY, - }))? = - Some(next_resource_version); + }); + } Ok(()) } @@ -1392,12 +1444,25 @@ fn merge_provisioning_items( } } +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(); @@ -1412,13 +1477,14 @@ fn annotate_user_item( previous.and_then(|item| item.last_applied_access_key_hash.clone()); } } - item.ownership = previous.and_then(|item| item.ownership.clone()); 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); @@ -1429,7 +1495,7 @@ fn user_access_key_changed( async fn sync_user_credentials( client: &RustfsAdminClient, - previous: Option<&ProvisioningItemStatus>, + previous: Option<&ProvisioningUserStatus>, credentials: &UserCredentials, exists: bool, ) -> Result { @@ -1444,7 +1510,7 @@ async fn sync_user_credentials( } fn user_credentials_need_apply( - previous: Option<&ProvisioningItemStatus>, + previous: Option<&ProvisioningUserStatus>, credentials: &UserCredentials, exists: bool, ) -> bool { @@ -2027,7 +2093,6 @@ mod tests { now: "2026-07-18T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), checkpoint_retry: StdMutex::new(None), }; @@ -2115,7 +2180,6 @@ mod tests { now: "2026-07-18T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), checkpoint_retry: StdMutex::new(None), }; @@ -2206,12 +2270,13 @@ mod tests { fn owned_user_status( state: ProvisioningUserOwnershipState, secret_resource_version: &str, - ) -> ProvisioningItemStatus { - let mut item = ProvisioningItemStatus::new( + ) -> 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")); @@ -2225,12 +2290,18 @@ mod tests { } #[test] - fn legacy_user_migration_requires_complete_matching_ready_status() { + 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, @@ -2270,28 +2341,54 @@ mod tests { 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 updated_tenant = tenant.clone(); - updated_tenant.metadata.resource_version = Some("18".to_string()); + 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 kube_service = service_fn(move |request: http::Request| { let service_requests = service_requests.clone(); - let updated_tenant = updated_tenant.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 = if attempt == 0 { - http::Response::builder() - .header("content-type", "application/json") - .body(KubeBody::from( - serde_json::to_vec(&updated_tenant) - .expect("Tenant response should serialize"), - )) - } else { - http::Response::builder() - .status(StatusCode::CONFLICT) - .header("content-type", "application/json") - .body(KubeBody::from( - br#"{"kind":"Status","apiVersion":"v1","status":"Failure","message":"resourceVersion conflict","reason":"Conflict","code":409}"#.to_vec(), - )) + 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")) } @@ -2305,13 +2402,10 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - status_resource_version: StdMutex::new(Some("17".to_string())), checkpoint_retry: StdMutex::new(None), }; let winner = make_run(); let loser = make_run(); - let mut checkpoint = owned_user_status(ProvisioningUserOwnershipState::PendingCreate, "5"); - checkpoint.state = ProvisioningItemState::Pending.as_str().to_string(); persist_user_ownership_checkpoint(&winner, checkpoint.clone()) .await @@ -2327,7 +2421,7 @@ mod tests { } _ => panic!("stale checkpoint writer should retry"), } - assert_eq!(requests.load(Ordering::SeqCst), 2); + assert_eq!(requests.load(Ordering::SeqCst), 3); } #[test] @@ -2388,7 +2482,6 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), checkpoint_retry: StdMutex::new(None), }; @@ -2459,27 +2552,53 @@ mod tests { 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 response_tenant = tenant.clone(); - response_tenant.metadata.resource_version = Some("18".to_string()); + 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 response_tenant = response_tenant.clone(); + let latest_tenant = latest_tenant.clone(); + let persisted_tenant = persisted_tenant.clone(); async move { - assert!(request.uri().path().ends_with("/tenants/tenant-a/status")); - assert_eq!(kube_sequence.load(Ordering::SeqCst), 1); - 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(2, Ordering::SeqCst); + 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") @@ -2500,7 +2619,6 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), checkpoint_retry: StdMutex::new(None), }; @@ -2524,8 +2642,8 @@ mod tests { put(move || { let add_sequence = add_sequence.clone(); async move { - assert_eq!(add_sequence.load(Ordering::SeqCst), 2); - add_sequence.store(3, Ordering::SeqCst); + assert_eq!(add_sequence.load(Ordering::SeqCst), 3); + add_sequence.store(4, Ordering::SeqCst); StatusCode::OK } }), @@ -2535,8 +2653,8 @@ mod tests { put(move || { let policy_sequence = policy_sequence.clone(); async move { - assert_eq!(policy_sequence.load(Ordering::SeqCst), 3); - policy_sequence.store(4, Ordering::SeqCst); + assert_eq!(policy_sequence.load(Ordering::SeqCst), 4); + policy_sequence.store(5, Ordering::SeqCst); StatusCode::OK } }), @@ -2564,22 +2682,18 @@ mod tests { ) .await; - assert_eq!(sequence.load(Ordering::SeqCst), 4); - assert_eq!( - run.status_resource_version - .lock() - .expect("status resourceVersion lock should be available") - .as_deref(), - Some("18") - ); + 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"], "17"); + 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"); @@ -2593,6 +2707,196 @@ mod tests { 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(), + checkpoint_retry: StdMutex::new(None), + }; + + 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(), + checkpoint_retry: StdMutex::new(None), + }; + + let error = persist_user_ownership_checkpoint(&run, 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)); @@ -2625,7 +2929,6 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), checkpoint_retry: StdMutex::new(None), }; @@ -2691,25 +2994,44 @@ mod tests { } #[tokio::test] - async fn legacy_managed_user_is_checkpointed_before_secret_rotation() { + 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 response_tenant = tenant.clone(); - response_tenant.metadata.resource_version = Some("18".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()); + 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_service = service_fn(move |request: http::Request| { let kube_sequence = kube_sequence.clone(); - let response_tenant = response_tenant.clone(); + let latest_tenant = latest_tenant.clone(); + let persisted_tenant = persisted_tenant.clone(); async move { - assert_eq!(kube_sequence.load(Ordering::SeqCst), 1); - kube_sequence.store(2, Ordering::SeqCst); + 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") @@ -2730,7 +3052,6 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - status_resource_version: StdMutex::new(tenant.metadata.resource_version.clone()), checkpoint_retry: StdMutex::new(None), }; @@ -2759,8 +3080,8 @@ mod tests { let add_count = add_count.clone(); let add_sequence = add_sequence.clone(); async move { - assert_eq!(add_sequence.load(Ordering::SeqCst), 2); - add_sequence.store(3, Ordering::SeqCst); + assert_eq!(add_sequence.load(Ordering::SeqCst), 3); + add_sequence.store(4, Ordering::SeqCst); add_count.fetch_add(1, Ordering::SeqCst); StatusCode::OK } @@ -2772,8 +3093,8 @@ mod tests { let policy_count = policy_count.clone(); let policy_sequence = policy_sequence.clone(); async move { - assert_eq!(policy_sequence.load(Ordering::SeqCst), 3); - policy_sequence.store(4, Ordering::SeqCst); + assert_eq!(policy_sequence.load(Ordering::SeqCst), 4); + policy_sequence.store(5, Ordering::SeqCst); policy_count.fetch_add(1, Ordering::SeqCst); StatusCode::OK } @@ -2803,7 +3124,7 @@ mod tests { assert_eq!(add_requests.load(Ordering::SeqCst), 1); assert_eq!(policy_requests.load(Ordering::SeqCst), 1); - assert_eq!(sequence.load(Ordering::SeqCst), 4); + 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!( @@ -2996,6 +3317,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(), @@ -3026,6 +3348,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(), @@ -3050,6 +3373,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(), @@ -3073,6 +3397,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(), @@ -3095,6 +3420,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 701f5d8..6715d95 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 06f01da..f2c7300 100755 --- a/src/types/v1alpha1.rs +++ b/src/types/v1alpha1.rs @@ -291,6 +291,14 @@ mod tenant_provisioning_crd_tests { 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 f4f8256..c7860bf 100755 --- a/src/types/v1alpha1/status.rs +++ b/src/types/v1alpha1/status.rs @@ -517,6 +517,7 @@ pub fn is_blocked_reason(reason: &str) -> bool { | "UserPolicyInvalid" | "UserPolicySetFailed" | "UserOwnershipConflict" + | "UserOwnershipCheckpointFailed" | "BucketCreateFailed" | "BucketObjectLockConflict" ) @@ -605,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"], @@ -682,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 34d574a..bcc56fb 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, @@ -75,6 +76,47 @@ pub struct ProvisioningUserOwnershipStatus { 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 { @@ -119,9 +161,6 @@ pub struct ProvisioningItemStatus { #[serde(default, skip_serializing_if = "Option::is_none")] pub last_applied_access_key_hash: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub ownership: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] pub policies: Vec, @@ -146,3 +185,9 @@ impl ProvisioningItemStatus { } } } + +impl AsRef for ProvisioningItemStatus { + fn as_ref(&self) -> &ProvisioningItemStatus { + self + } +} From c77c9a23f22b7ecec511b251f9749511e47ffd56 Mon Sep 17 00:00:00 2001 From: GatewayJ <18332154+GatewayJ@users.noreply.github.com> Date: Sun, 2 Aug 2026 18:01:39 +0800 Subject: [PATCH 3/3] fix(provisioning): batch ownership checkpoints --- e2e/src/cases/mod.rs | 2 +- e2e/src/cases/operator.rs | 31 +- e2e/tests/operator.rs | 199 ++++++++++++- src/reconcile/provisioning.rs | 516 +++++++++++++++++++++++----------- 4 files changed, 576 insertions(+), 172 deletions(-) diff --git a/e2e/src/cases/mod.rs b/e2e/src/cases/mod.rs index 51a68dc..19e55c8 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 2f80cda..1fc097d 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 6a1de3a..266ffb9 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/provisioning.rs b/src/reconcile/provisioning.rs index f57e924..6fdf633 100644 --- a/src/reconcile/provisioning.rs +++ b/src/reconcile/provisioning.rs @@ -31,7 +31,6 @@ use kube::api::{Patch, PatchParams}; use serde_json::Value; use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, BTreeSet}; -use std::sync::Mutex as StdMutex; use std::time::Duration; use tracing::{info, warn}; @@ -78,9 +77,9 @@ struct ProvisioningRun<'a> { now: String, status: ProvisioningStatus, failures: Vec<(Reason, String)>, - checkpoint_retry: StdMutex>, } +#[derive(Clone)] struct UserCredentials { access_key: String, secret_key: String, @@ -101,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, @@ -356,15 +368,6 @@ impl ProvisioningRun<'_> { } fn finish(mut self) -> ProvisioningReconcileResult { - if let Some(retry) = self.checkpoint_retry() { - return ProvisioningReconcileResult { - status: self.status, - outcome: ProvisioningOutcome::Retry { - message: retry.message, - retry_after: retry.retry_after, - }, - }; - } let outcome = self .failures .first() @@ -386,25 +389,6 @@ impl ProvisioningRun<'_> { outcome, } } - - fn request_checkpoint_retry(&self, pending_retry: CheckpointRetry) { - if let Ok(mut retry) = self.checkpoint_retry.lock() { - *retry = Some(pending_retry); - } - } - - fn checkpoint_retry(&self) -> Option { - self.checkpoint_retry - .lock() - .map(|retry| retry.clone()) - .unwrap_or_else(|_| { - Some(CheckpointRetry { - message: "RustFS user ownership checkpoint retry state is unavailable" - .to_string(), - retry_after: CHECKPOINT_TRANSIENT_RETRY, - }) - }) - } } pub(super) async fn reconcile_provisioning( @@ -426,7 +410,6 @@ pub(super) async fn reconcile_provisioning( now, status: ProvisioningStatus::default(), failures: Vec::new(), - checkpoint_retry: StdMutex::new(None), }; if !has_active_spec(tenant) { @@ -495,9 +478,15 @@ 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 run.checkpoint_retry().is_some() { - return run.finish(); + 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() @@ -792,7 +781,7 @@ async fn reconcile_users( client: &RustfsAdminClient, live_policies: &BTreeMap, credentials_preflight: &UserCredentialsPreflight, -) { +) -> Option { let failed_spec_policies = run .status .policies @@ -800,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 @@ -822,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 { @@ -847,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 { @@ -860,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 { @@ -875,25 +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; - run.push_user(item); - if run.checkpoint_retry().is_some() { - break; + 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 { @@ -940,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, -) -> ProvisioningUserStatus { +) -> UserReconcilePlan { let previous = run.previous_user(&user.name); if user_access_key_changed(previous, credentials) { let item = run.item( @@ -957,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 @@ -972,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 @@ -987,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 { @@ -1000,7 +1046,9 @@ 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, + ))); } }; @@ -1014,9 +1062,12 @@ async fn reconcile_user( Reason::UserOwnershipConflict, message, ); - return annotate_user_item(item, user, previous, None); + 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) { @@ -1035,10 +1086,12 @@ async fn reconcile_user( Reason::UserOwnershipCheckpointFailed, message, ); - return annotate_user_item(item, user, previous, None); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); } }; - let checkpoint = run.item( + let managed_checkpoint = run.item( previous, &user.name, ProvisioningItemState::Ready, @@ -1047,19 +1100,10 @@ async fn reconcile_user( ); // Preserve the legacy observed Secret version so a concurrently rotated Secret is // still applied after the ownership checkpoint has been persisted. - let mut checkpoint = annotate_user_item(checkpoint, user, previous, None); - checkpoint.ownership = Some(managed_ownership.clone()); - if let Err(error) = persist_user_ownership_checkpoint(run, checkpoint).await { - let message = handle_checkpoint_error(run, error); - let item = run.item( - previous, - &user.name, - ProvisioningItemState::Failed, - Reason::UserOwnershipCheckpointFailed, - message, - ); - return annotate_user_item(item, user, previous, None); - } + 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( @@ -1069,7 +1113,9 @@ async fn reconcile_user( Reason::UserOwnershipConflict, "RustFS user already exists without a matching operator ownership checkpoint; choose a different access key or remove the unmanaged user", ); - return annotate_user_item(item, user, previous, None); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); } } @@ -1089,29 +1135,22 @@ async fn reconcile_user( Reason::UserOwnershipCheckpointFailed, message, ); - return annotate_user_item(item, user, previous, None); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, user, previous, None, + ))); } }; - let checkpoint = run.item( + let pending_checkpoint = run.item( previous, &user.name, ProvisioningItemState::Pending, Reason::ProvisioningPending, "Operator ownership checkpoint was persisted before creating the RustFS user", ); - let mut checkpoint = annotate_user_item(checkpoint, user, previous, Some(credentials)); - checkpoint.ownership = Some(pending_ownership.clone()); - if let Err(error) = persist_user_ownership_checkpoint(run, checkpoint).await { - let message = handle_checkpoint_error(run, error); - let item = run.item( - previous, - &user.name, - ProvisioningItemState::Failed, - Reason::UserOwnershipCheckpointFailed, - message, - ); - return annotate_user_item(item, user, previous, None); - } + 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| { @@ -1121,29 +1160,20 @@ async fn reconcile_user( // 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 checkpoint = run.item( + let pending_checkpoint = run.item( previous, &user.name, ProvisioningItemState::Pending, Reason::ProvisioningPending, "Operator is resuming a pending RustFS user creation", ); - let mut checkpoint = annotate_user_item(checkpoint, user, previous, Some(credentials)); - checkpoint.ownership = ownership.clone(); - if let Err(error) = persist_user_ownership_checkpoint(run, checkpoint).await { - let message = handle_checkpoint_error(run, error); - let item = run.item( - previous, - &user.name, - ProvisioningItemState::Failed, - Reason::UserOwnershipCheckpointFailed, - message, - ); - return annotate_user_item(item, user, previous, None); - } + 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(mut ownership) = ownership else { + let Some(ownership) = ownership else { let item = run.item( previous, &user.name, @@ -1151,11 +1181,36 @@ async fn reconcile_user( Reason::UserOwnershipCheckpointFailed, "Operator ownership checkpoint is required before synchronizing RustFS user credentials", ); - return annotate_user_item(item, user, previous, None); + 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( @@ -1165,7 +1220,7 @@ async fn reconcile_user( Reason::UserSecretInvalid, format!("failed to update RustFS user credentials: {error}"), ); - let mut item = annotate_user_item(item, user, previous, None); + let mut item = annotate_user_item(item, &user, previous, None); item.ownership = Some(ownership); return item; } @@ -1184,7 +1239,7 @@ async fn reconcile_user( Reason::UserPolicySetFailed, format!("failed to set RustFS user policy mapping: {error}"), ); - let mut item = 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; } @@ -1212,11 +1267,51 @@ async fn reconcile_user( } _ => Some(run.now.clone()), }; - let mut item = 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, @@ -1280,10 +1375,13 @@ fn user_ownership( }) } -async fn persist_user_ownership_checkpoint( +async fn persist_user_ownership_checkpoints( run: &ProvisioningRun<'_>, - checkpoint: ProvisioningUserStatus, + 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()) @@ -1298,24 +1396,28 @@ async fn persist_user_ownership_checkpoint( retry_after: CHECKPOINT_CONFLICT_RETRY, })); } - 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 + for checkpoint in checkpoints { + let previous_user = run + .previous .users .iter() - .find(|item| item.name == checkpoint.name) - }); - if latest_user != previous_user { - return Err(CheckpointError::Retry(CheckpointRetry { - message: "Tenant user provisioning status changed before persisting the RustFS user ownership checkpoint" - .to_string(), - retry_after: CHECKPOINT_CONFLICT_RETRY, - })); + .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 { @@ -1332,7 +1434,7 @@ async fn persist_user_ownership_checkpoint( 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, std::slice::from_ref(&checkpoint)); + merge_provisioning_user_items(&mut provisioning.users, checkpoints); provisioning.observed_generation = run.tenant.metadata.generation; provisioning.phase = Some(ProvisioningPhase::Pending); provisioning @@ -1366,20 +1468,24 @@ async fn persist_user_ownership_checkpoint( retry_after: CHECKPOINT_TRANSIENT_RETRY, })); } - 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: "Kubernetes accepted the RustFS user ownership checkpoint request but did not persist the expected state and ownership proof; ensure the Tenant CRD is upgraded before the Operator" - .to_string(), + 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(()) } @@ -1420,14 +1526,11 @@ fn classify_checkpoint_kube_error(error: kube::Error) -> CheckpointError { } } -fn handle_checkpoint_error(run: &ProvisioningRun<'_>, error: CheckpointError) -> String { +#[cfg(test)] +fn checkpoint_error_message(error: CheckpointError) -> String { match error { CheckpointError::Permanent { message } => message, - CheckpointError::Retry(retry) => { - let message = retry.message.clone(); - run.request_checkpoint_retry(retry); - message - } + CheckpointError::Retry(retry) => retry.message, } } @@ -2093,7 +2196,6 @@ mod tests { now: "2026-07-18T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - checkpoint_retry: StdMutex::new(None), }; let request_count = Arc::new(AtomicUsize::new(0)); @@ -2180,7 +2282,6 @@ mod tests { now: "2026-07-18T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - checkpoint_retry: StdMutex::new(None), }; let client = RustfsAdminClient::new_with_base_url( @@ -2402,21 +2503,19 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - checkpoint_retry: StdMutex::new(None), }; let winner = make_run(); let loser = make_run(); - persist_user_ownership_checkpoint(&winner, checkpoint.clone()) + persist_user_ownership_checkpoints(&winner, std::slice::from_ref(&checkpoint)) .await .expect("first reconciler should persist its checkpoint"); - let error = persist_user_ownership_checkpoint(&loser, checkpoint) + let error = persist_user_ownership_checkpoints(&loser, std::slice::from_ref(&checkpoint)) .await .expect_err("stale reconciler should lose the CAS"); - handle_checkpoint_error(&loser, error); - match loser.finish().outcome { - ProvisioningOutcome::Retry { retry_after, .. } => { + match error { + CheckpointError::Retry(CheckpointRetry { retry_after, .. }) => { assert_eq!(retry_after, CHECKPOINT_CONFLICT_RETRY); } _ => panic!("stale checkpoint writer should retry"), @@ -2424,6 +2523,105 @@ mod tests { 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| { @@ -2482,7 +2680,6 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - checkpoint_retry: StdMutex::new(None), }; let write_requests = Arc::new(AtomicUsize::new(0)); @@ -2619,7 +2816,6 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - checkpoint_retry: StdMutex::new(None), }; let get_sequence = sequence.clone(); @@ -2764,7 +2960,6 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - checkpoint_retry: StdMutex::new(None), }; let write_requests = Arc::new(AtomicUsize::new(0)); @@ -2886,10 +3081,9 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - checkpoint_retry: StdMutex::new(None), }; - let error = persist_user_ownership_checkpoint(&run, checkpoint) + let error = persist_user_ownership_checkpoints(&run, std::slice::from_ref(&checkpoint)) .await .expect_err("rewritten checkpoint state must be rejected"); @@ -2929,7 +3123,6 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - checkpoint_retry: StdMutex::new(None), }; let add_requests = Arc::new(AtomicUsize::new(0)); @@ -3052,7 +3245,6 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), - checkpoint_retry: StdMutex::new(None), }; let add_requests = Arc::new(AtomicUsize::new(0));