From ead5601aec2a6ec887993d09c75fdfc60e020e81 Mon Sep 17 00:00:00 2001 From: Shurik <10572093+shurkus@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:46:02 +0300 Subject: [PATCH 1/2] fix(provisioning): skip CreateBucket for buckets that already exist Closes #286 --- CHANGELOG.md | 2 + crates/rustfs-admin/src/s3_ops.rs | 27 ++++++++++++ src/reconcile/provisioning.rs | 70 ++++++++++++++++++++++++++++++- 3 files changed, 98 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b35f216..b8fe98a 100644 --- a/CHANGELOG.md +++ b/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, diff --git a/crates/rustfs-admin/src/s3_ops.rs b/crates/rustfs-admin/src/s3_ops.rs index ff43c3e..20b0221 100644 --- a/crates/rustfs-admin/src/s3_ops.rs +++ b/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 { + 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() { diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs index 6380c48..d16f42e 100644 --- a/src/reconcile/provisioning.rs +++ b/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| { let live_policy = route_live_policy.clone(); async move { From b9801199e0a238d64d7dbfad18e5f6f1a80ef954 Mon Sep 17 00:00:00 2001 From: Shurik <10572093+shurkus@users.noreply.github.com> Date: Mon, 26 Oct 2026 11:58:51 +0200 Subject: [PATCH 2/2] fix(provisioning): keep CreateBucket ownership classification for COSI HeadBucket success proves access, not ownership, so the shared client's create_bucket no longer answers AlreadyOwnedByYou from it. Tenant provisioning, which owns its buckets, checks HeadBucket itself before CreateBucket; COSI keeps reporting a readable bucket owned by another account as a name conflict and never records ownership for it. --- Cargo.lock | 4 + crates/cosi-driver/Cargo.toml | 6 ++ crates/cosi-driver/src/bucket.rs | 122 ++++++++++++++++++++++++++++++ crates/rustfs-admin/src/s3_ops.rs | 16 ++-- src/reconcile/provisioning.rs | 15 ++++ 5 files changed, 154 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index dfb3ef6..3dcdac7 100755 --- a/Cargo.lock +++ b/Cargo.lock @@ -2582,7 +2582,10 @@ dependencies = [ name = "rustfs-cosi-driver" version = "0.1.0" dependencies = [ + "axum", "hex", + "http", + "http-body-util", "k8s-openapi", "kube", "prost", @@ -2598,6 +2601,7 @@ dependencies = [ "tokio-stream", "tonic", "tonic-build", + "tower 0.5.3", "tracing", "tracing-subscriber", ] diff --git a/crates/cosi-driver/Cargo.toml b/crates/cosi-driver/Cargo.toml index 78346ae..4deff83 100644 --- a/crates/cosi-driver/Cargo.toml +++ b/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" diff --git a/crates/cosi-driver/src/bucket.rs b/crates/cosi-driver/src/bucket.rs index 52fdf15..da6fb6b 100644 --- a/crates/cosi-driver/src/bucket.rs +++ b/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::::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, + "BucketAlreadyExists".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::::new())); + let store = config_maps.clone(); + let kube_service = tower::service_fn(move |request: http::Request| { + 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(); + } } diff --git a/crates/rustfs-admin/src/s3_ops.rs b/crates/rustfs-admin/src/s3_ops.rs index 20b0221..ff4c84c 100644 --- a/crates/rustfs-admin/src/s3_ops.rs +++ b/crates/rustfs-admin/src/s3_ops.rs @@ -748,12 +748,6 @@ 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"); @@ -813,9 +807,13 @@ 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 { + /// 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 { + 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()?; diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs index d16f42e..e22b690 100644 --- a/src/reconcile/provisioning.rs +++ b/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,