fix(provisioning): skip CreateBucket for buckets that already exist

CHANGELOG.md

@@ -47,6 +47,8 @@ All notable changes to RustFS Operator are documented in this file. The format i
 
 ### Fixed
 
+- Check an existing Tenant bucket with HeadBucket before CreateBucket, so a bucket under
+  continuous writes no longer stays Pending on RustFS lock-timeout 503 responses.
 - Use `docker.io/rustfs/operator` as the default Helm image repository for the Operator,
   Console, and COSI driver, while preserving repository overrides through Helm values.
 - Reject multi-pool Tenants containing a single-node single-disk pool before applying pool workloads, matching RustFS startup constraints.

crates/rustfs-admin/src/s3_ops.rs

@@ -748,6 +748,12 @@ impl RustfsAdminClient {
             return Err(RustfsClientError::RequestBuildFailed);
         }
 
+        // RustFS takes the bucket's exclusive locks before it finds an existing
+        // bucket, so a repeated PUT times out with 503 while objects are written.
+        if self.bucket_exists(bucket).await? {
+            return Ok(CreateBucketResult::AlreadyOwnedByYou);
+        }
+
         let path = format!("/{bucket}");
         let body = create_bucket_body(region);
         let content_type = (!body.is_empty()).then_some("application/xml");
@@ -807,6 +813,27 @@ impl RustfsAdminClient {
         ))
     }
 
+    /// Report whether HeadBucket succeeds with these credentials. Any other
+    /// response is left for CreateBucket to classify.
+    async fn bucket_exists(&self, bucket: &str) -> Result<bool, RustfsClientError> {
+        let path = format!("/{bucket}");
+        let signed = self.sign_request("HEAD", &path, "", "", None, ADMIN_SIGNING_SERVICE)?;
+        let host = self.host()?;
+
+        let response = self
+            .http_client
+            .head(format!("{}{}", self.base_url.trim_end_matches('/'), path))
+            .header("x-amz-date", &signed.amz_date)
+            .header("x-amz-content-sha256", &signed.payload_hash)
+            .header("authorization", &signed.authorization)
+            .header("host", host)
+            .send()
+            .await
+            .map_err(|_| RustfsClientError::RequestFailed)?;
+
+        Ok(response.status().is_success())
+    }
+
     /// Delete a bucket. Missing buckets are treated as success (idempotent).
     pub async fn delete_bucket(&self, bucket: &str) -> Result<(), RustfsClientError> {
         if bucket.trim().is_empty() {

src/reconcile/provisioning.rs

@@ -3761,7 +3761,7 @@ mod tests {
         extract::State,
         http::{Request, StatusCode},
         response::{IntoResponse, Response},
-        routing::{any, get, put},
+        routing::{any, get, head, put},
     };
     use http_body_util::BodyExt;
     use k8s_openapi::ByteString;
@@ -6530,6 +6530,73 @@ mod tests {
         server.abort();
     }
 
+    #[tokio::test]
+    async fn existing_bucket_skips_create_bucket() {
+        let router = Router::new().route(
+            "/app-data",
+            head(|| async { StatusCode::OK }).put(|| async { StatusCode::SERVICE_UNAVAILABLE }),
+        );
+        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 ctx = empty_kube_context();
+        let user = provisioning_user("app-user", "app-user-secret", "readwrite");
+        let tenant = provisioning_test_tenant(user, ProvisioningStatus::default());
+        let mut run = empty_run(&ctx, &tenant);
+        let bucket = ProvisioningBucket {
+            name: "app-data".to_string(),
+            ..Default::default()
+        };
+
+        let item = ensure_bucket(&mut run, &client, &bucket).await;
+        assert_eq!(item.state, ProvisioningItemState::Ready.as_str());
+        assert_eq!(
+            item.message.as_deref(),
+            Some("RustFS bucket already exists")
+        );
+        server.abort();
+    }
+
+    #[tokio::test]
+    async fn missing_bucket_is_created() {
+        let router = Router::new().route(
+            "/app-data",
+            head(|| async { StatusCode::NOT_FOUND }).put(|| async { 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 ctx = empty_kube_context();
+        let user = provisioning_user("app-user", "app-user-secret", "readwrite");
+        let tenant = provisioning_test_tenant(user, ProvisioningStatus::default());
+        let mut run = empty_run(&ctx, &tenant);
+        let bucket = ProvisioningBucket {
+            name: "app-data".to_string(),
+            ..Default::default()
+        };
+
+        let item = ensure_bucket(&mut run, &client, &bucket).await;
+        assert_eq!(item.state, ProvisioningItemState::Ready.as_str());
+        assert_eq!(item.message.as_deref(), Some("RustFS bucket was created"));
+        server.abort();
+    }
+
     #[tokio::test]
     async fn create_bucket_400_fails_without_retry() {
         let router = Router::new().route("/app-data", put(|| async { StatusCode::BAD_REQUEST }));
@@ -7076,6 +7143,7 @@ mod tests {
         let router = Router::new().route(
             "/app-data",
             put(|| async { StatusCode::OK })
+                .head(|| async { StatusCode::NOT_FOUND })
                 .get(move |req: Request<Body>| {
                     let live_policy = route_live_policy.clone();
                     async move {