diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index e00d899e0..da79ddd89 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -590,6 +590,33 @@ impl Connection { self.internal.lookup_function_by_id(function_id).await } + /// Conditionally remove a database-scoped Function catalog name. + /// + /// This is a direct synchronous catalog compare-and-swap (CAS), not a + /// [`crate::job::Job`], not physical [`Function`] deletion, and not + /// revocation. The caller supplies an observed immutable [`Function`] + /// handle; only [`Function::id`] is authority for the CAS precondition. + /// + /// Empty names return [`Error::InvalidInput`] before backend dispatch. + /// Nonempty names on local/default backends return [`Error::NotSupported`]. + /// Remote backends complete only when the server reports durable CAS + /// success for the `(name, current.id)` pair. + pub async fn remove_function_name( + &self, + name: impl AsRef, + current: &Function, + ) -> Result<()> { + let name = name.as_ref(); + // Public nonempty invariant: validate before any Database backend sees + // the call so local and remote Connections agree on InvalidInput. + if name.is_empty() { + return Err(Error::InvalidInput { + message: "function name removal name must be non-empty".into(), + }); + } + self.internal.remove_function_name(name, current).await + } + /// Drop a table in the database. /// /// # Arguments diff --git a/rust/lancedb/src/database.rs b/rust/lancedb/src/database.rs index 3d4534aa3..cf8dc98d7 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -350,6 +350,24 @@ pub trait Database: async fn lookup_function_by_id(&self, _function_id: &FunctionId) -> Result { job_op_not_supported("lookup_function_by_id") } + /// Conditionally remove a database-scoped Function catalog name. + /// + /// Direct synchronous catalog CAS, not a Job and not physical Function + /// deletion. Empty names return [`crate::Error::InvalidInput`] before the + /// unsupported fallback so local and remote backends agree. Nonempty names + /// on databases without enterprise catalog mutation return + /// [`crate::Error::NotSupported`]. + async fn remove_function_name(&self, name: &str, _current: &Function) -> Result<()> { + // Public nonempty invariant on the Database trait seam itself: + // Connection::database() exposes Arc, so empty-name + // rejection cannot rely solely on Connection prevalidation. + if name.is_empty() { + return Err(crate::error::Error::InvalidInput { + message: "function name removal name must be non-empty".into(), + }); + } + job_op_not_supported("remove_function_name") + } /// Open a table in the database async fn open_table(&self, request: OpenTableRequest) -> Result>; /// Rename a table in the database diff --git a/rust/lancedb/src/remote/client.rs b/rust/lancedb/src/remote/client.rs index 2abac50d2..bab965223 100644 --- a/rust/lancedb/src/remote/client.rs +++ b/rust/lancedb/src/remote/client.rs @@ -793,10 +793,10 @@ impl RestfulLanceDbClient { /// Send one attempt with a caller-owned request id. /// - /// Used by Function lookup so the helper can inspect non-success bodies for - /// an explicit `error_code` before deciding whether to retry. Does not log - /// request bodies or headers (lookup selectors must stay out of logs) and - /// does not interpret HTTP status. + /// Used by Function catalog helpers (lookup/remove) so they can inspect + /// non-success bodies for an explicit `error_code` before deciding whether + /// to retry. Does not log request bodies or headers (selectors must stay + /// out of logs) and does not interpret HTTP status. pub(crate) async fn send_attempt_with_request_id( &self, req_builder: RequestBuilder, diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 8871d537d..56b764f72 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -627,6 +627,10 @@ impl Database for RemoteDatabase { super::function::lookup_function(&self.client, selector).await } + async fn remove_function_name(&self, name: &str, current: &Function) -> Result<()> { + super::function::remove_function_name(&self.client, name, current).await + } + async fn table_names(&self, request: TableNamesRequest) -> Result> { let mut req = if !request.namespace_path.is_empty() { let namespace_id = @@ -4178,4 +4182,730 @@ mod tests { "nonretryable client/header error must not be repeated" ); } + + // ------------------------------------------------------------------------- + // Conditional Function name removal + // + // Direct synchronous catalog CAS via POST /v1/functions/remove. Not a Job + // and not physical Function deletion. Caller supplies an observed immutable + // Function handle; only current.id is authority on the wire. + // ------------------------------------------------------------------------- + + const REMOVE_CATALOG_NAME: &str = "text.normalize.remove-name"; + const REMOVE_FUNCTION_ID: &str = "fn.exact.remove-handle"; + const REMOVE_SERVER_MESSAGE_MARKER: &str = + "SERVER_REMOVE_DIAGNOSTIC_MARKER name=text.normalize.remove-name id=fn.exact.remove-handle"; + + fn sample_remove_function() -> Function { + let id = FunctionId::try_new(REMOVE_FUNCTION_ID).expect("valid FunctionId"); + let signature = FunctionSignature::try_new( + vec![ + FunctionParameter::new("text", DataType::Utf8), + FunctionParameter::new("limit", DataType::Int32), + ], + FunctionOutput::new(DataType::Utf8, true), + ) + .expect("valid FunctionSignature"); + Function::new(id, signature) + } + + fn remove_error_chain_text(err: &Error) -> String { + let mut text = format!("{err}\n{err:?}"); + let mut current: Option<&(dyn std::error::Error + 'static)> = Some(err); + while let Some(e) = current { + text.push('\n'); + text.push_str(&e.to_string()); + text.push('\n'); + text.push_str(&format!("{e:?}")); + current = e.source(); + } + text + } + + fn assert_remove_payload_free(err: &Error) { + let text = remove_error_chain_text(err); + assert!( + !text.contains(REMOVE_SERVER_MESSAGE_MARKER), + "server diagnostic marker must be absent from error/debug/source chain: {text}" + ); + assert!( + !text.contains(REMOVE_CATALOG_NAME), + "catalog name must be absent from error/debug/source chain: {text}" + ); + assert!( + !text.contains(REMOVE_FUNCTION_ID), + "FunctionId must be absent from error/debug/source chain: {text}" + ); + assert!( + !text.contains("SENSITIVE_REMOVE_BODY_MARKER"), + "non-success/malformed body marker must be absent from error/debug/source chain: {text}" + ); + } + + fn assert_remove_request( + request: &reqwest::Request, + expected_name: &str, + expected_function_id: &str, + ) { + assert_eq!(request.method(), &reqwest::Method::POST); + assert_eq!(request.url().path(), "/v1/functions/remove"); + assert!( + request.url().query().is_none(), + "remove selectors must stay out of the URL query: {}", + request.url() + ); + let body = request + .body() + .and_then(|b| b.as_bytes()) + .expect("remove request must carry a JSON body"); + let actual: Value = serde_json::from_slice(body).expect("remove body must be JSON"); + assert_eq!( + actual, + json!({ + "name": expected_name, + "expected_current_function_id": expected_function_id, + }), + "remove body must be exactly {{\"name\":...,\"expected_current_function_id\":...}}" + ); + let object = actual.as_object().expect("remove body must be an object"); + assert_eq!( + object.len(), + 2, + "remove body must not carry format_version or extra user fields: {actual}" + ); + assert!( + object.get("format_version").is_none(), + "remove body must not include format_version: {actual}" + ); + assert!( + object.get("function_id").is_none(), + "remove body must use expected_current_function_id, not function_id: {actual}" + ); + assert!( + object.get("current").is_none() && object.get("function").is_none(), + "remove must not send a Function record: {actual}" + ); + assert!( + object.get("job_id").is_none() && object.get("idempotency_key").is_none(), + "remove is not a Job and must not send user idempotency keys: {actual}" + ); + } + + /// Exact path/body and 204 success use only the observed Function id. + #[tokio::test] + async fn remove_function_name_posts_exact_body_and_succeeds_on_204() { + let current = sample_remove_function(); + let expected_id = current.id().as_str().to_string(); + let conn = Connection::new_with_handler(move |request| { + assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id); + // Illegal body on 204 must be ignored; success is status-driven only. + http::Response::builder() + .status(204) + .body(format!( + "{{\"SENSITIVE_REMOVE_BODY_MARKER\":true,\"message\":{REMOVE_SERVER_MESSAGE_MARKER:?}}}" + )) + .unwrap() + }); + + conn.remove_function_name(REMOVE_CATALOG_NAME, ¤t) + .await + .expect("HTTP 204 must complete the CAS"); + } + + /// One configured 5xx retry then 204 keeps identical request id/body. + #[tokio::test] + async fn remove_function_name_retry_preserves_request_id_and_body() { + let current = sample_remove_function(); + let expected_id = current.id().as_str().to_string(); + let seen_request_id = Arc::new(OnceLock::new()); + let seen_request_id_ref = seen_request_id.clone(); + let attempts = Arc::new(AtomicUsize::new(0)); + let attempts_ref = attempts.clone(); + + let conn = Connection::new_with_handler_and_config( + move |request| { + assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id); + let request_id = request.headers()["x-request-id"] + .to_str() + .unwrap() + .to_string(); + assert!(!request_id.is_empty(), "SDK must generate a request id"); + let seen = seen_request_id_ref.get_or_init(|| request_id.clone()); + assert_eq!( + &request_id, seen, + "request id must be identical across retries" + ); + + let n = attempts_ref.fetch_add(1, Ordering::SeqCst); + if n == 0 { + http::Response::builder() + .status(500) + .body(format!( + "{REMOVE_SERVER_MESSAGE_MARKER} SENSITIVE_REMOVE_BODY_MARKER" + )) + .unwrap() + } else { + http::Response::builder() + .status(204) + .body(String::new()) + .unwrap() + } + }, + ClientConfig { + retry_config: RetryConfig { + retries: Some(2), + backoff_factor: Some(0.0), + backoff_jitter: Some(0.0), + ..Default::default() + }, + ..Default::default() + }, + ); + + conn.remove_function_name(REMOVE_CATALOG_NAME, ¤t) + .await + .expect("remove must succeed after one retry"); + assert_eq!( + attempts.load(Ordering::SeqCst), + 2, + "one 5xx then 204 must be exactly two attempts" + ); + assert!(seen_request_id.get().is_some()); + } + + /// Exhausted always-retryable 5xx without explicit error_code surfaces Error::Retry. + /// + /// retries=2 is max request failures: exactly two identical attempts, then + /// Retry with request_failures == max_request_failures == 2, zero connect/read + /// failures, the retryable status retained, and a payload-free source chain. + #[tokio::test] + async fn remove_function_name_exhausted_retryable_5xx_returns_retry_with_request_counters() { + let current = sample_remove_function(); + let expected_id = current.id().as_str().to_string(); + let seen_request_id = Arc::new(OnceLock::new()); + let seen_request_id_ref = seen_request_id.clone(); + let attempts = Arc::new(AtomicUsize::new(0)); + let attempts_ref = attempts.clone(); + let body = format!("{REMOVE_SERVER_MESSAGE_MARKER} SENSITIVE_REMOVE_BODY_MARKER"); + + let conn = Connection::new_with_handler_and_config( + move |request| { + assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id); + let request_id = request.headers()["x-request-id"] + .to_str() + .unwrap() + .to_string(); + assert!(!request_id.is_empty(), "SDK must generate a request id"); + let seen = seen_request_id_ref.get_or_init(|| request_id.clone()); + assert_eq!( + &request_id, seen, + "request id must be identical across exhausted retries" + ); + + attempts_ref.fetch_add(1, Ordering::SeqCst); + http::Response::builder() + .status(500) + .body(body.clone()) + .unwrap() + }, + ClientConfig { + retry_config: RetryConfig { + // RetryCounter treats `retries` as max request failures, so + // retries=2 yields exactly two transport attempts before Error::Retry. + retries: Some(2), + backoff_factor: Some(0.0), + backoff_jitter: Some(0.0), + ..Default::default() + }, + ..Default::default() + }, + ); + + let err = conn + .remove_function_name(REMOVE_CATALOG_NAME, ¤t) + .await + .expect_err("exhausted retryable 5xx must fail"); + match &err { + Error::Retry { + request_failures, + max_request_failures, + connect_failures, + read_failures, + status_code, + .. + } => { + assert_eq!(*request_failures, 2); + assert_eq!(*max_request_failures, 2); + assert_eq!( + request_failures, max_request_failures, + "request budget must be fully exhausted" + ); + assert_eq!(*connect_failures, 0, "5xx must not consume connect budget"); + assert_eq!(*read_failures, 0, "5xx must not consume read budget"); + assert_eq!( + status_code.map(|s| s.as_u16()), + Some(500), + "retryable status must be retained on Error::Retry" + ); + } + other => panic!("exhausted 5xx must surface as Error::Retry, got {other:?}"), + } + assert_eq!( + attempts.load(Ordering::SeqCst), + 2, + "retries=2 must make exactly two identical attempts" + ); + assert!(seen_request_id.get().is_some()); + assert_remove_payload_free(&err); + } + + /// Explicit name_conflict on a retryable status is terminal Error::Function. + #[tokio::test] + async fn remove_function_name_explicit_name_conflict_is_terminal_on_retryable_status() { + let current = sample_remove_function(); + let expected_id = current.id().as_str().to_string(); + let attempts = Arc::new(AtomicUsize::new(0)); + let attempts_ref = attempts.clone(); + let body = json!({ + "error_code": "name_conflict", + "message": format!( + "{REMOVE_SERVER_MESSAGE_MARKER} looks_like revoked_function" + ), + "SENSITIVE_REMOVE_BODY_MARKER": true, + }) + .to_string(); + + let conn = Connection::new_with_handler_and_config( + move |request| { + assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id); + attempts_ref.fetch_add(1, Ordering::SeqCst); + http::Response::builder() + .status(503) + .body(body.clone()) + .unwrap() + }, + ClientConfig { + retry_config: RetryConfig { + retries: Some(3), + backoff_factor: Some(0.0), + backoff_jitter: Some(0.0), + ..Default::default() + }, + ..Default::default() + }, + ); + + let err = conn + .remove_function_name(REMOVE_CATALOG_NAME, ¤t) + .await + .expect_err("explicit name_conflict must fail"); + match &err { + Error::Function { code, message } => { + assert_eq!(code.as_str(), "name_conflict"); + assert!( + matches!(code, FunctionErrorCode::NameConflict), + "expected NameConflict, got {code:?}" + ); + assert_ne!(code.as_str(), "revoked_function"); + assert!( + !message.contains(REMOVE_SERVER_MESSAGE_MARKER), + "Function error message must be sanitized, got {message}" + ); + assert!( + !message.contains(REMOVE_CATALOG_NAME), + "Function error message must not echo the catalog name, got {message}" + ); + assert!( + !message.contains(REMOVE_FUNCTION_ID), + "Function error message must not echo the FunctionId, got {message}" + ); + } + other => panic!("expected Error::Function, got {other:?}"), + } + assert_remove_payload_free(&err); + assert_eq!( + attempts.load(Ordering::SeqCst), + 1, + "explicit semantic code must not consume request retries" + ); + } + + /// Unknown nonempty explicit code is preserved; status/message do not override. + #[tokio::test] + async fn remove_function_name_preserves_unknown_explicit_code_despite_status_and_message() { + let current = sample_remove_function(); + let expected_id = current.id().as_str().to_string(); + let raw = "enterprise_future_remove_category_xyz"; + let body = json!({ + "error_code": raw, + "message": format!( + "{REMOVE_SERVER_MESSAGE_MARKER} name_conflict revoked_function" + ), + "SENSITIVE_REMOVE_BODY_MARKER": true, + }) + .to_string(); + let conn = Connection::new_with_handler(move |request| { + assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id); + http::Response::builder() + .status(409) + .body(body.clone()) + .unwrap() + }); + + let err = conn + .remove_function_name(REMOVE_CATALOG_NAME, ¤t) + .await + .expect_err("unknown explicit code must surface"); + match &err { + Error::Function { code, message } => { + assert_eq!(code.as_str(), raw); + assert!( + matches!(code, FunctionErrorCode::Unrecognized(_)), + "unknown code must stay Unrecognized, got {code:?}" + ); + assert_ne!(code.as_str(), "name_conflict"); + assert_ne!(code.as_str(), "revoked_function"); + assert!( + !message.contains(REMOVE_SERVER_MESSAGE_MARKER), + "diagnostic message must be sanitized" + ); + } + other => panic!("expected Error::Function, got {other:?}"), + } + assert_remove_payload_free(&err); + } + + /// Missing/empty/null/wrong-type/malformed error_code stays payload-free Http. + #[tokio::test] + async fn remove_function_name_missing_or_invalid_error_code_is_payload_free_http() { + let cases: Vec<(&str, u16, String)> = vec![ + ( + "missing_code_404", + 404, + json!({ + "message": REMOVE_SERVER_MESSAGE_MARKER, + "SENSITIVE_REMOVE_BODY_MARKER": true, + }) + .to_string(), + ), + ( + "empty_code", + 400, + json!({ + "error_code": "", + "message": REMOVE_SERVER_MESSAGE_MARKER, + "SENSITIVE_REMOVE_BODY_MARKER": true, + }) + .to_string(), + ), + ( + "wrong_type_code", + 400, + json!({ + "error_code": 123, + "message": REMOVE_SERVER_MESSAGE_MARKER, + "SENSITIVE_REMOVE_BODY_MARKER": true, + }) + .to_string(), + ), + ( + // Non-retryable status: invalid/null error_code must stay immediate Http. + // Exhausted retryable 5xx is covered by + // remove_function_name_exhausted_retryable_5xx_returns_retry_with_request_counters. + "null_code", + 400, + json!({ + "error_code": null, + "message": REMOVE_SERVER_MESSAGE_MARKER, + "SENSITIVE_REMOVE_BODY_MARKER": true, + }) + .to_string(), + ), + ( + // Non-retryable status: this case proves invalid-code -> Http only. + // Exhausted retryable 5xx without error_code is covered separately + // by remove_function_name_exhausted_retryable_5xx_returns_retry_with_request_counters. + "non_json", + 400, + format!("not-json {REMOVE_SERVER_MESSAGE_MARKER} SENSITIVE_REMOVE_BODY_MARKER"), + ), + ]; + + let mut unexpected = Vec::new(); + for (label, status, response_body) in cases { + let current = sample_remove_function(); + let expected_id = current.id().as_str().to_string(); + let body_for_handler = response_body.clone(); + let conn = Connection::new_with_handler(move |request| { + assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id); + http::Response::builder() + .status(status) + .body(body_for_handler.clone()) + .unwrap() + }); + match conn + .remove_function_name(REMOVE_CATALOG_NAME, ¤t) + .await + { + Err(err @ Error::Http { .. }) => assert_remove_payload_free(&err), + Err(Error::Function { .. }) => unexpected.push(format!( + "{label}: must not invent Error::Function without explicit code" + )), + other => unexpected.push(format!("{label}: {other:?}")), + } + } + assert!( + unexpected.is_empty(), + "invalid/missing error_code must stay payload-free Http: {unexpected:?}" + ); + } + + /// 200/202 are payload-free protocol Http failures, never CAS success. + #[tokio::test] + async fn remove_function_name_other_2xx_are_payload_free_http_failures() { + let cases: Vec<(&str, u16, String)> = vec![ + ( + "200_with_body", + 200, + json!({ + "ok": true, + "message": REMOVE_SERVER_MESSAGE_MARKER, + "SENSITIVE_REMOVE_BODY_MARKER": true, + "job_id": "must-not-infer-job", + }) + .to_string(), + ), + ( + "202_empty", + 202, + format!("{REMOVE_SERVER_MESSAGE_MARKER} SENSITIVE_REMOVE_BODY_MARKER"), + ), + ("200_empty", 200, String::new()), + ]; + + let mut unexpected = Vec::new(); + for (label, status, response_body) in cases { + let current = sample_remove_function(); + let expected_id = current.id().as_str().to_string(); + let body_for_handler = response_body.clone(); + let conn = Connection::new_with_handler(move |request| { + assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id); + http::Response::builder() + .status(status) + .body(body_for_handler.clone()) + .unwrap() + }); + match conn + .remove_function_name(REMOVE_CATALOG_NAME, ¤t) + .await + { + Ok(()) => { + unexpected.push(format!("{label}: must not treat non-204 2xx as success")) + } + Err(err @ Error::Http { .. }) => assert_remove_payload_free(&err), + Err(Error::Function { .. }) => unexpected.push(format!( + "{label}: must not invent Error::Function from 2xx body" + )), + other => unexpected.push(format!("{label}: {other:?}")), + } + } + assert!( + unexpected.is_empty(), + "200/202 must be payload-free Http failures: {unexpected:?}" + ); + } + + /// Non-204 success must fail from status alone without reading the body. + /// + /// A protocol-invalid HTTP 200 whose body stream fails if read must return + /// payload-free Error::Http with status 200 on exactly one attempt, even when + /// read/request retry budgets are configured above one. Body must not be + /// read and no retry budget may be consumed. + #[tokio::test] + async fn remove_function_name_non_204_success_does_not_read_failing_body() { + let current = sample_remove_function(); + let expected_id = current.id().as_str().to_string(); + let attempts = Arc::new(AtomicUsize::new(0)); + let attempts_ref = attempts.clone(); + + let conn = Connection::new_with_handler_and_config( + move |request| { + assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id); + attempts_ref.fetch_add(1, Ordering::SeqCst); + let stream = futures::stream::once(async { + Err::(std::io::Error::other( + "simulated remove response body read failure SENSITIVE_REMOVE_BODY_MARKER", + )) + }); + http::Response::builder() + .status(200) + .body(reqwest::Body::wrap_stream(stream)) + .unwrap() + }, + ClientConfig { + // Budgets above one must not be consumed: status 200 is terminal + // protocol Http before any body read/retry classification. + retry_config: RetryConfig { + retries: Some(3), + read_retries: Some(3), + connect_retries: Some(3), + backoff_factor: Some(0.0), + backoff_jitter: Some(0.0), + ..Default::default() + }, + ..Default::default() + }, + ); + + let err = conn + .remove_function_name(REMOVE_CATALOG_NAME, ¤t) + .await + .expect_err("non-204 success must be protocol Http"); + assert_eq!( + attempts.load(Ordering::SeqCst), + 1, + "invalid 2xx must not read the body or consume retry budget" + ); + match &err { + Error::Http { status_code, .. } => { + assert_eq!( + status_code.map(|s| s.as_u16()), + Some(200), + "protocol Http must retain status 200 from the response alone" + ); + } + other => panic!("expected payload-free Error::Http, got {other:?}"), + } + assert_remove_payload_free(&err); + } + + /// Empty name is InvalidInput before backend through Connection. + #[tokio::test] + async fn remove_function_name_empty_name_fails_before_transport() { + let current = sample_remove_function(); + let attempts = Arc::new(AtomicUsize::new(0)); + let attempts_ref = attempts.clone(); + let conn = Connection::new_with_handler(move |_request| -> http::Response { + attempts_ref.fetch_add(1, Ordering::SeqCst); + panic!("empty name must not issue an HTTP request"); + }); + + let err = conn + .remove_function_name("", ¤t) + .await + .expect_err("empty name must fail before transport"); + assert!( + matches!(err, Error::InvalidInput { .. }), + "empty name must be InvalidInput, got {err:?}" + ); + assert_eq!( + attempts.load(Ordering::SeqCst), + 0, + "empty name must not touch transport" + ); + assert_remove_payload_free(&err); + } + + /// Database trait seam must reject empty names as InvalidInput (not NotSupported). + #[tokio::test] + async fn remove_function_name_database_trait_empty_name_is_invalid_input_without_mutation() { + let dir = tempfile::tempdir().expect("tempdir"); + let conn = ConnectBuilder::new(dir.path().to_str().unwrap()) + .execute() + .await + .expect("local connect"); + let before = conn + .table_names() + .execute() + .await + .expect("table_names before"); + assert!(before.is_empty()); + + let current = sample_remove_function(); + let err = conn + .database() + .remove_function_name("", ¤t) + .await + .expect_err("empty name must fail on Database trait default"); + assert!( + matches!(err, Error::InvalidInput { .. }), + "empty name via Database trait must be InvalidInput, got {err:?}" + ); + + let after = conn + .table_names() + .execute() + .await + .expect("table_names after"); + assert_eq!( + before, after, + "Database-trait empty-name rejection must not mutate tables" + ); + } + + /// Valid local removal is NotSupported and does not mutate tables. + #[tokio::test] + async fn remove_function_name_local_database_returns_not_supported_without_mutation() { + let dir = tempfile::tempdir().expect("tempdir"); + let conn = ConnectBuilder::new(dir.path().to_str().unwrap()) + .execute() + .await + .expect("local connect"); + let before = conn + .table_names() + .execute() + .await + .expect("table_names before"); + assert!(before.is_empty()); + + let current = sample_remove_function(); + let err = conn + .remove_function_name(REMOVE_CATALOG_NAME, ¤t) + .await + .expect_err("local remove_function_name must be unsupported"); + assert!( + matches!(err, Error::NotSupported { .. }), + "expected NotSupported for local remove, got {err:?}" + ); + + let after = conn + .table_names() + .execute() + .await + .expect("table_names after"); + assert_eq!(before, after, "unsupported remove must not mutate tables"); + } + + /// Empty name on local Connection is InvalidInput before NotSupported. + #[tokio::test] + async fn remove_function_name_local_empty_name_is_invalid_input_without_mutation() { + let dir = tempfile::tempdir().expect("tempdir"); + let conn = ConnectBuilder::new(dir.path().to_str().unwrap()) + .execute() + .await + .expect("local connect"); + let before = conn + .table_names() + .execute() + .await + .expect("table_names before"); + assert!(before.is_empty()); + + let current = sample_remove_function(); + let err = conn + .remove_function_name("", ¤t) + .await + .expect_err("empty name must fail before backend dispatch"); + assert!( + matches!(err, Error::InvalidInput { .. }), + "empty name must be InvalidInput on local Connection, got {err:?}" + ); + + let after = conn + .table_names() + .execute() + .await + .expect("table_names after"); + assert_eq!(before, after, "empty-name rejection must not mutate tables"); + } } diff --git a/rust/lancedb/src/remote/function.rs b/rust/lancedb/src/remote/function.rs index a46b5d2d8..4021b9d3e 100644 --- a/rust/lancedb/src/remote/function.rs +++ b/rust/lancedb/src/remote/function.rs @@ -1,12 +1,17 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright The LanceDB Authors -//! Remote first-class Function catalog lookup wire helper. +//! Remote first-class Function catalog wire helpers. //! //! POST `/v1/functions/lookup` resolves a database-scoped name or exact //! [`FunctionId`] to an immutable [`Function`] value. Name is lookup //! indirection only and never becomes part of the returned handle. +//! +//! POST `/v1/functions/remove` performs a direct synchronous catalog CAS that +//! unbinds a name when the caller's observed [`Function`] id still matches. +//! This is not a Job, not physical Function deletion, and not revocation. +use reqwest::{RequestBuilder, StatusCode}; use serde::Deserialize; use serde_json::Value; @@ -17,6 +22,7 @@ use super::client::{HttpSend, RestfulLanceDbClient}; use super::retry::RetryCounter; const LOOKUP_PATH: &str = "/v1/functions/lookup"; +const REMOVE_PATH: &str = "/v1/functions/remove"; /// Fixed client diagnostic for [`Error::Function`]. Never carry server text, /// selector values, or response payload bytes. @@ -29,6 +35,13 @@ const LOOKUP_HTTP_ERROR_MESSAGE: &str = "function lookup request failed"; /// Fixed client diagnostic for malformed success payloads. const LOOKUP_INVALID_SUCCESS_MESSAGE: &str = "function lookup response missing or invalid function"; +/// Fixed client diagnostic for remove [`Error::Function`]. Never carry server +/// text, catalog name, Function id, or response payload bytes. +const REMOVE_FUNCTION_ERROR_MESSAGE: &str = "function name removal failed"; + +/// Fixed client diagnostic for remove protocol / HTTP failures. +const REMOVE_HTTP_ERROR_MESSAGE: &str = "function name removal request failed"; + /// One exact lookup selector. Exactly one variant is serialized on the wire. pub enum FunctionLookupSelector { Name(String), @@ -65,6 +78,23 @@ struct LookupSuccessResponse { function: Function, } +/// Decision before reading response bytes. +enum BeforeBody { + /// Finish without reading or interpreting any body (HTTP 204 remove CAS). + Done(Result), + /// Read bytes and continue classification. + ReadBody, +} + +/// How a catalog POST treats a successfully read response body. +enum CatalogBodyAction { + /// Terminal success or failure for this attempt. + Done(Result), + /// Configured retryable status without an explicit `error_code`: consume + /// the request budget and retry with the same request id and body. + RetryRequest, +} + /// Resolve a Function via POST `/v1/functions/lookup`. /// /// Transport classification matches [`RestfulLanceDbClient::send_with_retry`]: @@ -78,16 +108,125 @@ pub async fn lookup_function( selector: FunctionLookupSelector, ) -> Result { let req_builder = client.post(LOOKUP_PATH).json(&selector.to_wire()); + catalog_post_with_retry( + client, + req_builder, + LOOKUP_HTTP_ERROR_MESSAGE, + |_status, _request_id| BeforeBody::ReadBody, + |status, bytes, request_id| { + if status.is_success() { + return CatalogBodyAction::Done(decode_lookup_success(bytes, request_id)); + } + if let Some(code) = explicit_error_code(bytes) { + return CatalogBodyAction::Done(Err(Error::Function { + code, + message: LOOKUP_FUNCTION_ERROR_MESSAGE.to_string(), + })); + } + if client.retry_config.statuses.contains(&status) { + return CatalogBodyAction::RetryRequest; + } + CatalogBodyAction::Done(Err(Error::Http { + source: LOOKUP_HTTP_ERROR_MESSAGE.into(), + request_id, + status_code: Some(status), + })) + }, + ) + .await +} - let tmp_req = req_builder.try_clone().ok_or_else(|| Error::Runtime { - message: "Attempted to retry a request that cannot be cloned".to_string(), - })?; - let (_, built) = tmp_req.build_split(); - let mut built = built.map_err(|e| Error::Runtime { - message: format!("Failed to build request: {}", e), - })?; - let request_id = client.extract_request_id(&mut built); - let mut retry_counter = RetryCounter::new(&client.retry_config, request_id.clone()); +/// Conditionally remove a database-scoped Function name via POST +/// `/v1/functions/remove`. +/// +/// Direct synchronous catalog CAS: the wire body is exactly +/// `{"name","expected_current_function_id"}` using only `current.id`. Only +/// HTTP 204 means the CAS completed; other 2xx are payload-free protocol +/// [`Error::Http`]. Empty names are [`Error::InvalidInput`] before transport. +/// +/// Retry budgets match lookup: stable internal request id and exact cloned +/// body across attempts; response-byte failures consume read budget; configured +/// retryable status without explicit `error_code` consumes request budget; +/// header/client errors are immediate. Sophon deduplicates the internal request +/// id; it is not a user-facing idempotency key. +pub async fn remove_function_name( + client: &RestfulLanceDbClient, + name: &str, + current: &Function, +) -> Result<()> { + if name.is_empty() { + return Err(Error::InvalidInput { + message: "function name removal name must be non-empty".into(), + }); + } + + // Authority is the observed immutable Function id only; never send + // signature, raw Function objects, Job fields, or user idempotency keys. + let body = serde_json::json!({ + "name": name, + "expected_current_function_id": current.id().as_str(), + }); + let req_builder = client.post(REMOVE_PATH).json(&body); + + catalog_post_with_retry( + client, + req_builder, + REMOVE_HTTP_ERROR_MESSAGE, + |status, request_id| { + // Exact HTTP 204 completes the CAS; do not read or interpret any body. + if status == StatusCode::NO_CONTENT { + BeforeBody::Done(Ok(())) + } else if status.is_success() { + // Other 2xx are payload-free protocol failures from status alone. + BeforeBody::Done(Err(Error::Http { + source: REMOVE_HTTP_ERROR_MESSAGE.into(), + request_id: request_id.to_string(), + status_code: Some(status), + })) + } else { + BeforeBody::ReadBody + } + }, + |status, bytes, request_id| { + // Explicit nonempty error_code wins over HTTP status and precludes retry. + if let Some(code) = explicit_error_code(bytes) { + return CatalogBodyAction::Done(Err(Error::Function { + code, + message: REMOVE_FUNCTION_ERROR_MESSAGE.to_string(), + })); + } + + if client.retry_config.statuses.contains(&status) { + return CatalogBodyAction::RetryRequest; + } + + CatalogBodyAction::Done(Err(Error::Http { + source: REMOVE_HTTP_ERROR_MESSAGE.into(), + request_id, + status_code: Some(status), + })) + }, + ) + .await +} + +/// Shared Function-catalog POST retry loop used by lookup and remove. +/// +/// Sensitive attempt sending logs no body/header selectors. One SDK-generated +/// request id and the exact cloned JSON body are reused across attempts. +async fn catalog_post_with_retry( + client: &RestfulLanceDbClient, + req_builder: RequestBuilder, + http_error_message: &'static str, + mut before_body: Before, + mut after_body: After, +) -> Result +where + S: HttpSend, + Before: FnMut(StatusCode, &str) -> BeforeBody, + After: FnMut(StatusCode, &[u8], String) -> CatalogBodyAction, +{ + let mut retry_counter = prepare_catalog_retry(client, &req_builder)?; loop { let attempt = req_builder.try_clone().ok_or_else(|| Error::Runtime { @@ -100,13 +239,18 @@ pub async fn lookup_function( { Ok(rsp) => rsp, Err(err) => { - classify_lookup_send_error(&mut retry_counter, err)?; + classify_catalog_send_error(&mut retry_counter, err)?; tokio::time::sleep(retry_counter.next_sleep_time()).await; continue; } }; let status = rsp.status(); + match before_body(status, &retry_counter.request_id) { + BeforeBody::Done(result) => return result, + BeforeBody::ReadBody => {} + } + let bytes = match rsp.bytes().await { Ok(bytes) => bytes, Err(err) => { @@ -118,42 +262,41 @@ pub async fn lookup_function( } }; - if status.is_success() { - return decode_lookup_success(&bytes, retry_counter.request_id); + match after_body(status, &bytes, retry_counter.request_id.clone()) { + CatalogBodyAction::Done(result) => return result, + CatalogBodyAction::RetryRequest => { + let source = Error::Http { + source: http_error_message.into(), + request_id: retry_counter.request_id.clone(), + status_code: Some(status), + }; + retry_counter.increment_request_failures(source)?; + tokio::time::sleep(retry_counter.next_sleep_time()).await; + } } - - // Explicit nonempty error_code wins over HTTP status and precludes retry. - if let Some(code) = explicit_error_code(&bytes) { - return Err(Error::Function { - code, - message: LOOKUP_FUNCTION_ERROR_MESSAGE.to_string(), - }); - } - - if client.retry_config.statuses.contains(&status) { - let source = Error::Http { - source: LOOKUP_HTTP_ERROR_MESSAGE.into(), - request_id: retry_counter.request_id.clone(), - status_code: Some(status), - }; - retry_counter.increment_request_failures(source)?; - tokio::time::sleep(retry_counter.next_sleep_time()).await; - continue; - } - - return Err(Error::Http { - source: LOOKUP_HTTP_ERROR_MESSAGE.into(), - request_id: retry_counter.request_id, - status_code: Some(status), - }); } } +fn prepare_catalog_retry<'a, S: HttpSend>( + client: &'a RestfulLanceDbClient, + req_builder: &RequestBuilder, +) -> Result> { + let tmp_req = req_builder.try_clone().ok_or_else(|| Error::Runtime { + message: "Attempted to retry a request that cannot be cloned".to_string(), + })?; + let (_, built) = tmp_req.build_split(); + let mut built = built.map_err(|e| Error::Runtime { + message: format!("Failed to build request: {}", e), + })?; + let request_id = client.extract_request_id(&mut built); + Ok(RetryCounter::new(&client.retry_config, request_id)) +} + /// Classify a send-attempt error using the same buckets as `send_with_retry`. /// /// Returns `Ok(())` when the caller should sleep and retry. Returns `Err` for /// nonretryable failures or when a retry budget is exhausted (no extra attempt). -fn classify_lookup_send_error(retry_counter: &mut RetryCounter<'_>, err: Error) -> Result<()> { +fn classify_catalog_send_error(retry_counter: &mut RetryCounter<'_>, err: Error) -> Result<()> { match err { Error::Http { source,