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.
 - Update source release metadata to `0.0.8` so source Helm installs select the
   `0.0.8` Operator and Console images, and verify release tags and chart images in CI.
 - Use `docker.io/rustfs/operator` as the default Helm image repository for the Operator,

crates/cosi-driver/Cargo.toml

@@ -41,6 +41,12 @@ rand = "0.8"
 snafu = { version = "0.8.9", features = ["futures"] }
 rustls = { version = "0.23", default-features = false, features = ["ring"] }
 
+[dev-dependencies]
+axum = "0.7"
+http = "1.2"
+http-body-util = "0.1"
+tower = { version = "0.5", features = ["util"] }
+
 [build-dependencies]
 tonic-build = "0.12"
 protoc-bin-vendored = "3"

crates/cosi-driver/src/bucket.rs

@@ -459,4 +459,126 @@ mod tests {
         ));
         assert!(authorize_delete(false, Some(BucketProofState::Ready), "b").is_ok());
     }
+
+    /// HeadBucket success proves access, not ownership. A readable bucket owned by
+    /// another account must stay a name conflict: no proof survives and delete is refused.
+    #[tokio::test]
+    async fn readable_bucket_owned_by_another_account_is_a_name_conflict() {
+        use axum::{Router, http::StatusCode, routing::any};
+        use http_body_util::BodyExt;
+        use kube::client::Body as KubeBody;
+        use std::convert::Infallible;
+        use std::sync::{Arc, Mutex};
+
+        let s3_requests = Arc::new(Mutex::new(Vec::<String>::new()));
+        let recorded = s3_requests.clone();
+        let router = Router::new().route(
+            "/cosi-bucket",
+            any(move |method: axum::http::Method| {
+                let recorded = recorded.clone();
+                async move {
+                    recorded.lock().unwrap().push(method.to_string());
+                    match method.as_str() {
+                        "HEAD" => (StatusCode::OK, String::new()),
+                        "PUT" => (
+                            StatusCode::CONFLICT,
+                            "<Error><Code>BucketAlreadyExists</Code></Error>".to_string(),
+                        ),
+                        _ => (StatusCode::NO_CONTENT, String::new()),
+                    }
+                }
+            }),
+        );
+        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 config_maps = Arc::new(Mutex::new(HashMap::<String, ConfigMap>::new()));
+        let store = config_maps.clone();
+        let kube_service = tower::service_fn(move |request: http::Request<KubeBody>| {
+            let store = store.clone();
+            async move {
+                let method = request.method().clone();
+                let path = request.uri().path().to_string();
+                let body = request
+                    .into_body()
+                    .collect()
+                    .await
+                    .expect("request body should read")
+                    .to_bytes();
+                let name = path.rsplit('/').next().unwrap_or_default().to_string();
+                let not_found = || {
+                    http::Response::builder()
+                        .status(404)
+                        .body(KubeBody::from(
+                            br#"{"kind":"Status","apiVersion":"v1","status":"Failure","reason":"NotFound","code":404}"#.to_vec(),
+                        ))
+                        .expect("response should build")
+                };
+                let json = |cm: &ConfigMap| {
+                    http::Response::builder()
+                        .body(KubeBody::from(
+                            serde_json::to_vec(cm).expect("ConfigMap should serialize"),
+                        ))
+                        .expect("response should build")
+                };
+                let response = match method.as_str() {
+                    "POST" => {
+                        let mut cm: ConfigMap =
+                            serde_json::from_slice(&body).expect("ConfigMap should parse");
+                        cm.metadata.resource_version = Some("1".to_string());
+                        let cm_name = cm.metadata.name.clone().expect("ConfigMap needs a name");
+                        store.lock().unwrap().insert(cm_name, cm.clone());
+                        json(&cm)
+                    }
+                    "PUT" => {
+                        let cm: ConfigMap =
+                            serde_json::from_slice(&body).expect("ConfigMap should parse");
+                        store.lock().unwrap().insert(name, cm.clone());
+                        json(&cm)
+                    }
+                    "GET" => match store.lock().unwrap().get(&name) {
+                        Some(cm) => json(cm),
+                        None => not_found(),
+                    },
+                    "DELETE" => match store.lock().unwrap().remove(&name) {
+                        Some(_) => http::Response::builder()
+                            .body(KubeBody::from(
+                                br#"{"kind":"Status","apiVersion":"v1","status":"Success"}"#
+                                    .to_vec(),
+                            ))
+                            .expect("response should build"),
+                        None => not_found(),
+                    },
+                    other => panic!("unexpected Kubernetes request: {other} {path}"),
+                };
+                Ok::<_, Infallible>(response)
+            }
+        });
+        let kube = Client::new(kube_service, "cosi-system");
+        let client =
+            RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret");
+        let p = params(&[]);
+
+        let err = create_bucket(&kube, &client, &p, "cosi-bucket")
+            .await
+            .expect_err("a bucket owned by another account must not be created");
+        assert!(matches!(err, BucketError::NameConflict { .. }));
+        assert!(
+            config_maps.lock().unwrap().is_empty(),
+            "the pending ownership proof must be removed"
+        );
+
+        delete_bucket(&kube, &client, &p, "cosi-bucket")
+            .await
+            .expect_err("deleting a bucket without a Ready proof must be refused");
+        assert_eq!(*s3_requests.lock().unwrap(), vec!["PUT".to_string()]);
+        server.abort();
+    }
 }

crates/rustfs-admin/src/s3_ops.rs

@@ -807,6 +807,31 @@ impl RustfsAdminClient {
         ))
     }
 
+    /// Report whether HeadBucket succeeds with these credentials. Success proves
+    /// access, not ownership; callers that need ownership must use CreateBucket.
+    pub async fn bucket_exists(&self, bucket: &str) -> Result<bool, RustfsClientError> {
+        if bucket.trim().is_empty() {
+            return Err(RustfsClientError::RequestBuildFailed);
+        }
+
+        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

@@ -2121,6 +2121,21 @@ async fn ensure_bucket(
         return annotate_bucket_item(item, bucket);
     }
 
+    // RustFS takes the bucket's exclusive locks before it finds an existing
+    // bucket, so a repeated CreateBucket times out with 503 while objects are written.
+    if matches!(client.bucket_exists(&bucket.name).await, Ok(true)) {
+        return annotate_bucket_item(
+            run.item(
+                previous.as_ref(),
+                &bucket.name,
+                ProvisioningItemState::Ready,
+                Reason::ProvisioningConfigured,
+                "RustFS bucket already exists",
+            ),
+            bucket,
+        );
+    }
+
     let create_result = match client
         .create_bucket(
             &bucket.name,
@@ -3761,7 +3776,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 +6545,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 +7158,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 {