From a9ed8dba27245945b0c578c71c07fb66910b21fd Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Wed, 12 Aug 2026 01:18:41 +0800 Subject: [PATCH] feat: return results from jobs --- nodejs/src/job.rs | 12 +- rust/lancedb/src/job.rs | 131 ++++++++++++++++-- rust/lancedb/src/remote/job.rs | 23 ++- rust/lancedb/src/table.rs | 5 +- rust/lancedb/src/table/create_index.rs | 12 +- ...irst_class_function_job_result_contract.rs | 4 +- 6 files changed, 162 insertions(+), 25 deletions(-) diff --git a/nodejs/src/job.rs b/nodejs/src/job.rs index 6aaeee174..6e300ec17 100644 --- a/nodejs/src/job.rs +++ b/nodejs/src/job.rs @@ -42,9 +42,19 @@ impl Job { } /// Wait until the operation reaches a terminal state. + /// + /// Jobs that complete without a resource result resolve successfully. + /// Resource results are not exposed on this binding yet; unsupported + /// success results reject with a generic error. #[napi(catch_unwind)] pub async fn wait(&self) -> napi::Result<()> { - self.inner.wait().await.default_error() + match self.inner.wait().await.default_error()? { + lancedb::JobResult::None => Ok(()), + // JobResult is non_exhaustive; Function and future variants fail closed. + _ => Err(napi::Error::from_reason( + "unsupported job result".to_string(), + )), + } } /// Request cancellation. Cancelling a finished operation is a no-op. diff --git a/rust/lancedb/src/job.rs b/rust/lancedb/src/job.rs index e8975ac63..17d8a12a3 100644 --- a/rust/lancedb/src/job.rs +++ b/rust/lancedb/src/job.rs @@ -133,7 +133,7 @@ pub(crate) trait JobHandle: Send + Sync { None } async fn status(&self) -> Result; - async fn wait(&self) -> Result<()>; + async fn wait(&self) -> Result; async fn cancel(&self) -> Result<()>; } @@ -166,7 +166,7 @@ impl Job { } /// A job running as a task in this process. - pub(crate) fn spawned(task: JoinHandle>) -> Self { + pub(crate) fn spawned(task: JoinHandle>) -> Self { Self::new(Box::new(SpawnedJob::new(task))) } @@ -195,11 +195,14 @@ impl Job { /// Waits until the operation reaches a terminal state. /// + /// On success, returns the job's [`JobResult`]. Operations that produce no + /// resource result yield [`JobResult::None`]. + /// /// Returns [`crate::Error::JobFailed`] if the operation failed and /// [`crate::Error::JobCancelled`] if it was cancelled. - pub async fn wait(&self) -> Result<()> { + pub async fn wait(&self) -> Result { match &self.handle { - None => Ok(()), + None => Ok(JobResult::None), Some(handle) => handle.wait().await, } } @@ -219,15 +222,15 @@ impl Job { /// the outcome; [`Error`] is not, so failures share one behind an [`Arc`]. #[derive(Clone)] enum Outcome { - Succeeded, + Succeeded(JobResult), Failed(Arc), Cancelled, } impl Outcome { - fn into_result(self) -> Result<()> { + fn into_result(self) -> Result { match self { - Self::Succeeded => Ok(()), + Self::Succeeded(result) => Ok(result), Self::Failed(source) => Err(Error::JobFailed { job_id: None, failure: JobFailure::from_source(source), @@ -246,16 +249,16 @@ struct SpawnedJob { } impl SpawnedJob { - fn new(task: JoinHandle>) -> Self { + fn new(task: JoinHandle>) -> Self { let abort = task.abort_handle(); let (tx, outcome) = watch::channel(None); tokio::spawn(async move { let outcome = match task.await { - Ok(Ok(())) => Outcome::Succeeded, + Ok(Ok(result)) => Outcome::Succeeded(result), Ok(Err(err)) => Outcome::Failed(Arc::new(err)), Err(err) if err.is_cancelled() => Outcome::Cancelled, Err(err) => Outcome::Failed(Arc::new(Error::Runtime { - message: format!("index job task failed: {err}"), + message: format!("job task failed: {err}"), })), }; let _ = tx.send(Some(outcome)); @@ -269,20 +272,20 @@ impl JobHandle for SpawnedJob { async fn status(&self) -> Result { let label = match &*self.outcome.borrow() { None => "running", - Some(Outcome::Succeeded) => "finished", + Some(Outcome::Succeeded(_)) => "finished", Some(Outcome::Failed(_)) => "failed", Some(Outcome::Cancelled) => "cancelled", }; Ok(label.to_string()) } - async fn wait(&self) -> Result<()> { + async fn wait(&self) -> Result { let mut outcome = self.outcome.clone(); let settled = outcome .wait_for(|outcome| outcome.is_some()) .await .map_err(|_| Error::Runtime { - message: "index job outcome was dropped before it completed".to_string(), + message: "job outcome was dropped before it completed".to_string(), })? .clone() .expect("wait_for returns once an outcome is set"); @@ -297,8 +300,110 @@ impl JobHandle for SpawnedJob { #[cfg(test)] mod tests { + use std::future::Future; + use std::pin::pin; + use std::task::{Context, Poll, Waker}; + + use arrow_schema::DataType; + use tokio::sync::oneshot; + use super::*; use crate::error::FunctionErrorCode; + use crate::function::{ + Function, FunctionId, FunctionOutput, FunctionParameter, FunctionSignature, + }; + + fn sample_success_function() -> Function { + let id = FunctionId::try_new("fn.exact.local-job-result").expect("valid FunctionId"); + let signature = FunctionSignature::try_new( + vec![FunctionParameter::new("x", DataType::Int32)], + FunctionOutput::new(DataType::Int32, true), + ) + .expect("valid FunctionSignature"); + Function::new(id, signature) + } + + fn assert_exact_function(actual: &Function, expected: &Function) { + assert_eq!(actual.id(), expected.id()); + assert_eq!(actual.signature(), expected.signature()); + } + + /// A completed-before-handle local job projects success as None. + #[tokio::test] + async fn local_job_result_new_done_wait_returns_none() { + let job = Job::new_done(); + let result = job.wait().await.expect("new_done must succeed"); + assert_eq!(result, JobResult::None); + } + + /// A local spawned unit / no-resource success projects as None. + #[tokio::test] + async fn local_job_result_spawned_unit_success_projects_none() { + let job = Job::spawned(tokio::spawn(async { Ok(JobResult::None) })); + let result = job + .wait() + .await + .expect("unit success must finish without error"); + assert_eq!(result, JobResult::None); + } + + /// Function success is cloneable and shared by concurrent + late waiters. + /// + /// Wait futures are pinned and polled once to Pending while success is still + /// gated, proving they observed the running state before publication. + #[tokio::test] + async fn local_job_result_spawned_function_shared_by_waiters() { + let expected = sample_success_function(); + let (release_tx, release_rx) = oneshot::channel(); + + let job = Job::spawned(tokio::spawn({ + let function = expected.clone(); + async move { + release_rx + .await + .expect("success task must be released by the test"); + Ok(JobResult::Function(function)) + } + })); + + let mut wait_a = pin!(job.wait()); + let mut wait_b = pin!(job.wait()); + let waker = Waker::noop(); + let mut cx = Context::from_waker(waker); + + assert!( + matches!(wait_a.as_mut().poll(&mut cx), Poll::Pending), + "waiter A must poll Pending before success publication" + ); + assert!( + matches!(wait_b.as_mut().poll(&mut cx), Poll::Pending), + "waiter B must poll Pending before success publication" + ); + + release_tx + .send(()) + .expect("success task must still be waiting on the gate"); + + let result_a = wait_a + .await + .expect("concurrent waiter A must observe success"); + let result_b = wait_b + .await + .expect("concurrent waiter B must observe success"); + let result_late = job + .wait() + .await + .expect("late waiter must observe the same success"); + + for result in [&result_a, &result_b, &result_late] { + match result { + JobResult::Function(function) => assert_exact_function(function, &expected), + JobResult::None => panic!("Function success must not project as JobResult::None"), + } + } + assert_eq!(result_a, result_b); + assert_eq!(result_a, result_late); + } #[tokio::test] async fn spawned_job_function_failure_returns_job_failed_with_same_code() { diff --git a/rust/lancedb/src/remote/job.rs b/rust/lancedb/src/remote/job.rs index 63cff248b..8bd387f96 100644 --- a/rust/lancedb/src/remote/job.rs +++ b/rust/lancedb/src/remote/job.rs @@ -11,7 +11,7 @@ use tokio::time::sleep; use serde::{Deserialize, Deserializer}; use crate::error::{Error, FunctionErrorCode, JobFailure, Result}; -use crate::job::JobHandle; +use crate::job::{JobHandle, JobResult}; use crate::remote::client::{HttpSend, RequestResultExt, RestfulLanceDbClient}; /// Delay before the second job-state poll; doubles up to [`MAX_POLL_INTERVAL`]. @@ -124,12 +124,14 @@ impl JobHandle for RemoteJob { Ok(self.describe().await?.job_state.client_label()) } - async fn wait(&self) -> Result<()> { + async fn wait(&self) -> Result { let mut interval = INITIAL_POLL_INTERVAL; loop { let description = self.describe().await?; match description.job_state { - JobState::Done => return Ok(()), + // Existing DONE responses have no success-result field; map to + // None until strict result decoding is added. + JobState::Done => return Ok(JobResult::None), JobState::Failed => { return Err(Error::JobFailed { job_id: Some(self.job_id.clone()), @@ -177,8 +179,23 @@ impl JobHandle for RemoteJob { mod tests { use super::*; use crate::error::FunctionErrorCode; + use crate::job::JobResult; use crate::remote::client::test_utils::client_with_handler; + /// Terminal DONE with no result field projects as [`JobResult::None`]. + #[tokio::test] + async fn local_job_result_remote_done_without_result_projects_none() { + let client = client_with_handler(|_| { + http::Response::builder() + .status(200) + .body(r#"{"job_id":"job-done","job_state":"DONE"}"#) + .unwrap() + }); + let job = RemoteJob::new(client, "job-done".into()); + let result = job.wait().await.expect("DONE with no result must succeed"); + assert_eq!(result, JobResult::None); + } + #[tokio::test] async fn wait_decodes_known_failure_error_code() { let client = client_with_handler(|_| { diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 5120c48b7..5a98b3ec3 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -57,7 +57,7 @@ use crate::error::{Error, Result}; use crate::index::IndexStatistics; use crate::index::{Index, IndexBuilder}; use crate::index::{IndexConfig, IndexStatisticsImpl, IndexType}; -use crate::job::Job; +use crate::job::{Job, JobResult}; use crate::query::{IntoQueryVector, Query, QueryExecutionOptions, TakeQuery, VectorQuery}; use crate::table::datafusion::insert::InsertExec; use crate::utils::{PatchReadParam, PatchWriteParam, resolve_arrow_field_path}; @@ -3201,7 +3201,8 @@ impl BaseTable for NativeTable { let prepared = self.prepare_index(&opts).await?; let table = self.clone(); Ok(Job::spawned(tokio::spawn(async move { - table.build_index(opts, prepared).await + table.build_index(opts, prepared).await?; + Ok(JobResult::None) }))) } diff --git a/rust/lancedb/src/table/create_index.rs b/rust/lancedb/src/table/create_index.rs index 144c6dbfb..5139d0499 100644 --- a/rust/lancedb/src/table/create_index.rs +++ b/rust/lancedb/src/table/create_index.rs @@ -423,6 +423,7 @@ mod tests { use futures::TryStreamExt; use tempfile::tempdir; + use crate::JobResult; use crate::connect; use crate::connection::ConnectBuilder; use crate::index::Index; @@ -538,7 +539,8 @@ mod tests { assert_eq!(job.id(), None); // The build runs as a task, so the index need not exist yet; it must // once the job resolves. - job.wait().await.unwrap(); + let result = job.wait().await.unwrap(); + assert_eq!(result, JobResult::None); assert_eq!(table.list_indices().await.unwrap().len(), 1); // Cancelling a finished job is a no-op. job.cancel().await.unwrap(); @@ -570,10 +572,12 @@ mod tests { }) .collect::>(); for waiter in waiters { - waiter.await.unwrap().unwrap(); + let result = waiter.await.unwrap().unwrap(); + assert_eq!(result, JobResult::None); } // A wait after the job settled still reports the same outcome. - job.wait().await.unwrap(); + let late = job.wait().await.unwrap(); + assert_eq!(late, JobResult::None); assert_eq!(table.list_indices().await.unwrap().len(), 1); } @@ -716,7 +720,7 @@ mod tests { match job.wait().await { Err(crate::Error::JobCancelled { .. }) => {} // The build may finish before the abort lands. - Ok(()) => {} + Ok(JobResult::None) => {} other => panic!("unexpected job outcome: {other:?}"), } } diff --git a/rust/lancedb/tests/first_class_function_job_result_contract.rs b/rust/lancedb/tests/first_class_function_job_result_contract.rs index 4e3d855e3..ba6d512fb 100644 --- a/rust/lancedb/tests/first_class_function_job_result_contract.rs +++ b/rust/lancedb/tests/first_class_function_job_result_contract.rs @@ -292,7 +292,7 @@ fn unknown_kind_field_version_and_malformed_function_fail_closed() -> Result<()> let function = sample_function()?; let function_json = - serde_json::to_value(&JobResult::Function(function)).expect("serialize Function"); + serde_json::to_value(JobResult::Function(function)).expect("serialize Function"); let mut missing_function = function_json.clone(); missing_function.as_object_mut().unwrap().remove("function"); @@ -346,7 +346,7 @@ fn outer_result_excludes_forbidden_fields() -> Result<()> { let function = sample_function()?; let function_json = - serde_json::to_value(&JobResult::Function(function)).expect("serialize Function"); + serde_json::to_value(JobResult::Function(function)).expect("serialize Function"); assert_json_object_keys_exact(&function_json, &["format_version", "kind", "function"]); assert_outer_forbidden_keys_absent(&function_json);