From 1798ece362060d7cff996f7244d3fa7c8db800b3 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Tue, 11 Aug 2026 22:43:39 +0800 Subject: [PATCH] feat: add stable function error codes --- rust/lancedb/src/error.rs | 104 ++++++++++ rust/lancedb/src/job.rs | 61 ++++++ rust/lancedb/src/remote/db.rs | 105 ++++++++++ rust/lancedb/src/remote/job.rs | 90 ++++++++- .../first_class_function_error_contract.rs | 186 ++++++++++++++++++ 5 files changed, 545 insertions(+), 1 deletion(-) create mode 100644 rust/lancedb/tests/first_class_function_error_contract.rs diff --git a/rust/lancedb/src/error.rs b/rust/lancedb/src/error.rs index 4a6e6d8d9..deb91ef3a 100644 --- a/rust/lancedb/src/error.rs +++ b/rust/lancedb/src/error.rs @@ -6,10 +6,91 @@ use std::sync::{Arc, PoisonError}; use arrow_schema::ArrowError; use datafusion_common::DataFusionError; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; use snafu::Snafu; pub(crate) type BoxError = Box; +/// Stable Function error category (FF-006). +/// +/// The known variants serialize to fixed JSON strings. Any other wire string +/// decodes as [`Self::Unrecognized`] with the exact value preserved, and +/// re-serializes unchanged. Category judgment is structural: do not infer a +/// code from diagnostic message text, HTTP status, job phase, or retryability. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub enum FunctionErrorCode { + /// Function definition failed validation. + DefinitionValidationFailure, + /// A named Function or Function reference was not found. + NameOrFunctionNotFound, + /// A Function name conflicts with an existing name. + NameConflict, + /// The requested runtime or capability is not supported. + UnsupportedRuntimeOrCapability, + /// The Function has been revoked and cannot be used. + RevokedFunction, + /// User-defined Function execution failed. + UdfExecutionFailure, + /// A generated column was not fully materialized. + GeneratedColumnIncomplete, + /// Input was stale or conflicted with the current state. + StaleOrConflictingInput, + /// A wire string this client version does not recognize. + /// + /// The inner value is preserved exactly for forward compatibility. + Unrecognized(String), +} + +impl FunctionErrorCode { + /// The stable JSON / wire string for this code. + pub fn as_str(&self) -> &str { + match self { + Self::DefinitionValidationFailure => "definition_validation_failure", + Self::NameOrFunctionNotFound => "name_or_function_not_found", + Self::NameConflict => "name_conflict", + Self::UnsupportedRuntimeOrCapability => "unsupported_runtime_or_capability", + Self::RevokedFunction => "revoked_function", + Self::UdfExecutionFailure => "udf_execution_failure", + Self::GeneratedColumnIncomplete => "generated_column_incomplete", + Self::StaleOrConflictingInput => "stale_or_conflicting_input", + Self::Unrecognized(raw) => raw.as_str(), + } + } + + fn from_wire(value: &str) -> Self { + match value { + "definition_validation_failure" => Self::DefinitionValidationFailure, + "name_or_function_not_found" => Self::NameOrFunctionNotFound, + "name_conflict" => Self::NameConflict, + "unsupported_runtime_or_capability" => Self::UnsupportedRuntimeOrCapability, + "revoked_function" => Self::RevokedFunction, + "udf_execution_failure" => Self::UdfExecutionFailure, + "generated_column_incomplete" => Self::GeneratedColumnIncomplete, + "stale_or_conflicting_input" => Self::StaleOrConflictingInput, + other => Self::Unrecognized(other.to_string()), + } + } +} + +impl Display for FunctionErrorCode { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl Serialize for FunctionErrorCode { + fn serialize(&self, serializer: S) -> std::result::Result { + serializer.serialize_str(self.as_str()) + } +} + +impl<'de> Deserialize<'de> for FunctionErrorCode { + fn deserialize>(deserializer: D) -> std::result::Result { + Ok(Self::from_wire(String::deserialize(deserializer)?.as_str())) + } +} + /// Why a job failed, to whatever precision the backend provides. /// /// A job run in this process carries the error it failed with in [`Self::source`]. @@ -18,6 +99,12 @@ pub(crate) type BoxError = Box; /// backend does not supply it. #[derive(Debug, Clone, Default)] pub struct JobFailure { + /// Stable Function error category, when the backend supplied one. + /// + /// Present only when copied from [`Error::Function`] or decoded from a + /// remote `error_code` field. Never inferred from message, phase, + /// retryable, HTTP status, or other diagnostics. + pub error_code: Option, /// The stage the job was in, when known. pub phase: Option, /// A human-readable reason, when known. @@ -30,8 +117,16 @@ pub struct JobFailure { impl JobFailure { /// A failure whose only known detail is the error that caused it. + /// + /// When `source` is [`Error::Function`], [`Self::error_code`] is copied + /// from that error. Other error kinds leave `error_code` as [`None`]. pub(crate) fn from_source(source: Arc) -> Self { + let error_code = match source.as_ref() { + Error::Function { code, .. } => Some(code.clone()), + _ => None, + }; Self { + error_code, message: Some(source.to_string()), source: Some(source), ..Default::default() @@ -92,6 +187,15 @@ pub enum Error { }, #[snafu(display("Job{} was cancelled", job_id.as_ref().map(|id| format!(" {id}")).unwrap_or_default()))] JobCancelled { job_id: Option }, + /// A first-class Function operation failed with a stable category. + /// + /// [`Self::Function::code`] is the semantic category. [`Self::Function::message`] + /// is diagnostic only and must not be used to recover or override the code. + #[snafu(display("Function error ({code}): {message}"))] + Function { + code: FunctionErrorCode, + message: String, + }, // 3rd party / external errors #[snafu(display("object_store error: {source}"))] diff --git a/rust/lancedb/src/job.rs b/rust/lancedb/src/job.rs index 789ce8312..b9d9bd7f2 100644 --- a/rust/lancedb/src/job.rs +++ b/rust/lancedb/src/job.rs @@ -180,3 +180,64 @@ impl JobHandle for SpawnedJob { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::error::FunctionErrorCode; + + #[tokio::test] + async fn spawned_job_function_failure_returns_job_failed_with_same_code() { + let job = Job::spawned(tokio::spawn(async { + Err(Error::Function { + code: FunctionErrorCode::UdfExecutionFailure, + // Message names a different category on purpose; code is structural. + message: "looks like name_conflict to a string parser".to_string(), + }) + })); + + let err = job + .wait() + .await + .expect_err("Function failure must fail the job"); + match err { + Error::JobFailed { failure, .. } => match &failure.error_code { + Some(code) => { + assert_eq!(code, &FunctionErrorCode::UdfExecutionFailure); + assert_ne!(code, &FunctionErrorCode::NameConflict); + } + None => panic!("local Function failure must project error_code onto JobFailure"), + }, + other => panic!("expected Error::JobFailed, got {other:?}"), + } + } + + #[tokio::test] + async fn spawned_job_preserves_unrecognized_function_error_code() { + let raw = "enterprise_future_category_xyz"; + let job = Job::spawned(tokio::spawn({ + let raw = raw.to_string(); + async move { + Err(Error::Function { + code: FunctionErrorCode::Unrecognized(raw), + message: "future server category".to_string(), + }) + } + })); + + let err = job + .wait() + .await + .expect_err("Function failure must fail the job"); + match err { + Error::JobFailed { failure, .. } => match &failure.error_code { + Some(FunctionErrorCode::Unrecognized(preserved)) => { + assert_eq!(preserved, raw); + } + Some(other) => panic!("unrecognized code must not become known: {other:?}"), + None => panic!("unrecognized Function code must be preserved on JobFailure"), + }, + other => panic!("expected Error::JobFailed, got {other:?}"), + } + } +} diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 839cb3797..262b7d211 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -457,6 +457,9 @@ struct RemoteListJobsResponse { /// which report only the terminal state. #[derive(serde::Deserialize)] struct RemoteReportedFailure { + /// Stable Function error category when the server supplied one. + #[serde(default)] + error_code: Option, #[serde(default)] phase: Option, #[serde(default)] @@ -575,6 +578,7 @@ impl Database for RemoteDatabase { creation_ms: body.creation_ms, spec: body.spec, failure: body.failure.map(|reported| crate::error::JobFailure { + error_code: reported.error_code, phase: reported.phase, message: reported.message, retryable: reported.retryable, @@ -2424,4 +2428,105 @@ mod tests { assert_eq!(job.status().await.unwrap(), "finished"); assert!(polls.load(Ordering::SeqCst) >= 3); } + + #[tokio::test] + async fn test_get_job_decodes_known_failure_error_code() { + use crate::error::FunctionErrorCode; + + let conn = Connection::new_with_handler(|request| { + assert_eq!(request.url().path(), "/v1/jobs/describe"); + http::Response::builder() + .status(200) + .body( + r#"{"job_id":"job-1","job_type":"create_index","job_state":"FAILED","creation_ms":1000,"spec":{},"failure":{"error_code":"name_or_function_not_found","phase":"validate","message":"looks like definition_validation_failure","retryable":false}}"#, + ) + .unwrap() + }); + let job = conn.get_job("job-1").await.unwrap().unwrap(); + let failure = job.failure.expect("failure payload present"); + match &failure.error_code { + Some(code) => { + assert_eq!(code, &FunctionErrorCode::NameOrFunctionNotFound); + assert_ne!(code, &FunctionErrorCode::DefinitionValidationFailure); + } + None => panic!("known error_code must be decoded by get_job"), + } + assert_eq!(failure.phase.as_deref(), Some("validate")); + assert_eq!(failure.retryable, Some(false)); + } + + #[tokio::test] + async fn test_get_job_preserves_unrecognized_failure_error_code() { + use crate::error::FunctionErrorCode; + + let conn = Connection::new_with_handler(|_| { + http::Response::builder() + .status(200) + .body( + r#"{"job_id":"job-1","job_type":"create_index","job_state":"FAILED","creation_ms":1000,"spec":{},"failure":{"error_code":"enterprise_future_category_xyz","phase":"execute","message":"new category","retryable":true}}"#, + ) + .unwrap() + }); + let job = conn.get_job("job-1").await.unwrap().unwrap(); + let failure = job.failure.expect("failure payload present"); + match &failure.error_code { + Some(FunctionErrorCode::Unrecognized(raw)) => { + assert_eq!(raw, "enterprise_future_category_xyz"); + } + Some(other) => panic!("unknown error_code must stay Unrecognized, got {other:?}"), + None => panic!("unknown error_code must not be dropped by get_job"), + } + } + + #[tokio::test] + async fn test_get_job_allows_older_failure_payload_without_error_code() { + let conn = Connection::new_with_handler(|_| { + http::Response::builder() + .status(200) + .body( + r#"{"job_id":"job-1","job_type":"create_index","job_state":"FAILED","creation_ms":1000,"spec":{},"failure":{"phase":"execute","message":"stale_or_conflicting_input in logs","retryable":true}}"#, + ) + .unwrap() + }); + let job = conn.get_job("job-1").await.unwrap().unwrap(); + let failure = job.failure.expect("failure payload present"); + assert!( + failure.error_code.is_none(), + "older get_job payloads without error_code must not invent a category: {failure:?}" + ); + assert_eq!(failure.phase.as_deref(), Some("execute")); + assert_eq!(failure.retryable, Some(true)); + } + + #[tokio::test] + async fn test_conn_job_wait_decodes_failure_error_code_without_transport_override() { + use crate::error::FunctionErrorCode; + + let conn = Connection::new_with_handler(|request| { + assert_eq!(request.url().path(), "/v1/jobs/describe"); + // Transport is HTTP 200 with FAILED state; category comes only from error_code. + http::Response::builder() + .status(200) + .body( + r#"{"job_id":"job-1","job_type":"create_index","job_state":"FAILED","creation_ms":1,"failure":{"error_code":"unsupported_runtime_or_capability","phase":"dispatch","message":"revoked_function in transport logs","retryable":false}}"#, + ) + .unwrap() + }); + let err = conn + .job("job-1") + .unwrap() + .wait() + .await + .expect_err("FAILED must surface JobFailed"); + match err { + Error::JobFailed { failure, .. } => match &failure.error_code { + Some(code) => { + assert_eq!(code, &FunctionErrorCode::UnsupportedRuntimeOrCapability); + assert_ne!(code, &FunctionErrorCode::RevokedFunction); + } + None => panic!("Database job wait must decode failure.error_code"), + }, + other => panic!("expected Error::JobFailed, got {other:?}"), + } + } } diff --git a/rust/lancedb/src/remote/job.rs b/rust/lancedb/src/remote/job.rs index 2fc99da59..63cff248b 100644 --- a/rust/lancedb/src/remote/job.rs +++ b/rust/lancedb/src/remote/job.rs @@ -10,7 +10,7 @@ use tokio::time::sleep; use serde::{Deserialize, Deserializer}; -use crate::error::{Error, JobFailure, Result}; +use crate::error::{Error, FunctionErrorCode, JobFailure, Result}; use crate::job::JobHandle; use crate::remote::client::{HttpSend, RequestResultExt, RestfulLanceDbClient}; @@ -67,6 +67,9 @@ impl From<&str> for JobState { /// report only the terminal state. #[derive(Deserialize)] struct ReportedFailure { + /// Stable Function error category when the server supplied one. + #[serde(default)] + error_code: Option, #[serde(default)] phase: Option, #[serde(default)] @@ -133,6 +136,7 @@ impl JobHandle for RemoteJob { failure: description .failure .map(|reported| JobFailure { + error_code: reported.error_code, phase: reported.phase, message: reported.message, retryable: reported.retryable, @@ -168,3 +172,87 @@ impl JobHandle for RemoteJob { .map(|_| ()) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::error::FunctionErrorCode; + use crate::remote::client::test_utils::client_with_handler; + + #[tokio::test] + async fn wait_decodes_known_failure_error_code() { + let client = client_with_handler(|_| { + http::Response::builder() + .status(200) + .body( + r#"{"job_id":"job-err","job_state":"FAILED","failure":{"error_code":"stale_or_conflicting_input","phase":"commit","message":"looks like udf_execution_failure","retryable":true}}"#, + ) + .unwrap() + }); + let job = RemoteJob::new(client, "job-err".into()); + let err = job.wait().await.expect_err("FAILED must surface JobFailed"); + match err { + Error::JobFailed { failure, .. } => { + match &failure.error_code { + Some(code) => { + assert_eq!(code, &FunctionErrorCode::StaleOrConflictingInput); + assert_ne!(code, &FunctionErrorCode::UdfExecutionFailure); + } + None => panic!("known error_code must be decoded"), + } + assert_eq!(failure.phase.as_deref(), Some("commit")); + assert_eq!(failure.retryable, Some(true)); + } + other => panic!("expected Error::JobFailed, got {other:?}"), + } + } + + #[tokio::test] + async fn wait_preserves_unrecognized_failure_error_code() { + let client = client_with_handler(|_| { + http::Response::builder() + .status(200) + .body( + r#"{"job_id":"job-err","job_state":"FAILED","failure":{"error_code":"enterprise_future_category_xyz","phase":"execute","message":"new category","retryable":false}}"#, + ) + .unwrap() + }); + let job = RemoteJob::new(client, "job-err".into()); + let err = job.wait().await.expect_err("FAILED must surface JobFailed"); + match err { + Error::JobFailed { failure, .. } => match &failure.error_code { + Some(FunctionErrorCode::Unrecognized(raw)) => { + assert_eq!(raw, "enterprise_future_category_xyz"); + } + Some(other) => panic!("unknown error_code must stay Unrecognized, got {other:?}"), + None => panic!("unknown error_code must not be dropped"), + }, + other => panic!("expected Error::JobFailed, got {other:?}"), + } + } + + #[tokio::test] + async fn wait_allows_older_failure_payload_without_error_code() { + let client = client_with_handler(|_| { + http::Response::builder() + .status(200) + .body( + r#"{"job_id":"job-err","job_state":"FAILED","failure":{"phase":"execute","message":"generated_column_incomplete in logs","retryable":true}}"#, + ) + .unwrap() + }); + let job = RemoteJob::new(client, "job-err".into()); + let err = job.wait().await.expect_err("FAILED must surface JobFailed"); + match err { + Error::JobFailed { failure, .. } => { + assert!( + failure.error_code.is_none(), + "older payloads without error_code must not invent a category from message/phase/retryable: {failure:?}" + ); + assert_eq!(failure.phase.as_deref(), Some("execute")); + assert_eq!(failure.retryable, Some(true)); + } + other => panic!("expected Error::JobFailed, got {other:?}"), + } + } +} diff --git a/rust/lancedb/tests/first_class_function_error_contract.rs b/rust/lancedb/tests/first_class_function_error_contract.rs new file mode 100644 index 000000000..133ba61ca --- /dev/null +++ b/rust/lancedb/tests/first_class_function_error_contract.rs @@ -0,0 +1,186 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Public contract tests for first-class Function error codes (FF-006). +//! +//! These tests pin the stable `FunctionErrorCode` wire strings, direct +//! `Error::Function` projection, and optional `JobFailure.error_code`. +//! They intentionally fail to compile until that public API exists. +//! +//! Categories are judged only by structural enum matching / equality, never +//! by parsing diagnostic message text. + +use lancedb::error::FunctionErrorCode; +use lancedb::{Error, JobFailure}; +use serde_json::{Value, json}; + +/// Exact stable wire strings for the eight known Function error categories. +const KNOWN_WIRE_CODES: &[(&str, FunctionErrorCode)] = &[ + ( + "definition_validation_failure", + FunctionErrorCode::DefinitionValidationFailure, + ), + ( + "name_or_function_not_found", + FunctionErrorCode::NameOrFunctionNotFound, + ), + ("name_conflict", FunctionErrorCode::NameConflict), + ( + "unsupported_runtime_or_capability", + FunctionErrorCode::UnsupportedRuntimeOrCapability, + ), + ("revoked_function", FunctionErrorCode::RevokedFunction), + ( + "udf_execution_failure", + FunctionErrorCode::UdfExecutionFailure, + ), + ( + "generated_column_incomplete", + FunctionErrorCode::GeneratedColumnIncomplete, + ), + ( + "stale_or_conflicting_input", + FunctionErrorCode::StaleOrConflictingInput, + ), +]; + +fn assert_known_variant(code: &FunctionErrorCode, expected: &FunctionErrorCode) { + assert_eq!( + code, expected, + "FunctionErrorCode must match structurally; got {code:?}, expected {expected:?}" + ); + assert!( + !matches!(code, FunctionErrorCode::Unrecognized(_)), + "known wire string must not deserialize as Unrecognized: {code:?}" + ); +} + +#[test] +fn function_error_code_known_variants_use_exact_stable_json_strings() { + for (wire, expected) in KNOWN_WIRE_CODES { + let encoded = serde_json::to_value(expected).expect("serialize FunctionErrorCode"); + assert_eq!( + encoded, + Value::String((*wire).to_string()), + "stable JSON string for {expected:?}" + ); + + let decoded: FunctionErrorCode = serde_json::from_value(Value::String((*wire).to_string())) + .unwrap_or_else(|e| panic!("deserialize `{wire}`: {e}")); + assert_known_variant(&decoded, expected); + + let round_trip = serde_json::to_value(&decoded).expect("re-serialize"); + assert_eq!(round_trip, Value::String((*wire).to_string())); + } +} + +#[test] +fn unrecognized_error_code_preserves_exact_string_and_does_not_become_known() { + let raw = "enterprise_future_category_xyz"; + let decoded: FunctionErrorCode = serde_json::from_value(json!(raw)) + .unwrap_or_else(|e| panic!("unknown code must deserialize, not fail: {e}")); + + match &decoded { + FunctionErrorCode::Unrecognized(preserved) => { + assert_eq!(preserved, raw, "unknown code must be preserved verbatim"); + } + other => panic!("expected FunctionErrorCode::Unrecognized, got {other:?}"), + } + + for (_, known) in KNOWN_WIRE_CODES { + assert_ne!( + &decoded, known, + "unrecognized code must not equal known variant {known:?}" + ); + } + + let encoded = serde_json::to_value(&decoded).expect("serialize Unrecognized"); + assert_eq!(encoded, json!(raw)); + + let again: FunctionErrorCode = + serde_json::from_value(encoded).expect("Unrecognized must round-trip"); + match again { + FunctionErrorCode::Unrecognized(preserved) => assert_eq!(preserved, raw), + other => panic!("round-trip must stay Unrecognized, got {other:?}"), + } +} + +#[test] +fn error_function_carries_code_plus_diagnostic_message() { + let err = Error::Function { + code: FunctionErrorCode::NameConflict, + message: "sanitized diagnostic only".to_string(), + }; + + match err { + Error::Function { code, message } => { + assert_known_variant(&code, &FunctionErrorCode::NameConflict); + assert_eq!(message, "sanitized diagnostic only"); + } + other => panic!("expected Error::Function, got {other:?}"), + } +} + +#[test] +fn error_function_category_is_the_code_field_not_the_message() { + // Message text deliberately names a different category; structural code wins. + let err = Error::Function { + code: FunctionErrorCode::GeneratedColumnIncomplete, + message: "looks like udf_execution_failure to a string parser".to_string(), + }; + + match err { + Error::Function { code, .. } => { + assert_known_variant(&code, &FunctionErrorCode::GeneratedColumnIncomplete); + assert_ne!(code, FunctionErrorCode::UdfExecutionFailure); + } + other => panic!("expected Error::Function, got {other:?}"), + } +} + +#[test] +fn job_failure_has_optional_error_code() { + let with_code = JobFailure { + error_code: Some(FunctionErrorCode::RevokedFunction), + phase: Some("execute".to_string()), + message: Some("revoked".to_string()), + retryable: Some(false), + source: None, + }; + match &with_code.error_code { + Some(code) => assert_known_variant(code, &FunctionErrorCode::RevokedFunction), + None => panic!("error_code must be present when set"), + } + + let without_code = JobFailure { + phase: Some("execute".to_string()), + message: Some("older backend failure without a category".to_string()), + retryable: Some(true), + ..Default::default() + }; + assert!( + without_code.error_code.is_none(), + "missing error_code must stay None; diagnostics must not invent a category" + ); +} + +#[test] +fn job_failure_diagnostics_do_not_overwrite_error_code() { + let failure = JobFailure { + error_code: Some(FunctionErrorCode::StaleOrConflictingInput), + phase: Some("commit".to_string()), + message: Some("definition_validation_failure in worker logs".to_string()), + retryable: Some(true), + source: None, + }; + + match &failure.error_code { + Some(code) => { + assert_known_variant(code, &FunctionErrorCode::StaleOrConflictingInput); + assert_ne!(code, &FunctionErrorCode::DefinitionValidationFailure); + } + None => panic!("explicit error_code must remain set"), + } + assert_eq!(failure.phase.as_deref(), Some("commit")); + assert_eq!(failure.retryable, Some(true)); +}