feat: return results from jobs

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.

rust/lancedb/src/job.rs

@@ -133,7 +133,7 @@ pub(crate) trait JobHandle: Send + Sync {
         None
     }
     async fn status(&self) -> Result<String>;
-    async fn wait(&self) -> Result<()>;
+    async fn wait(&self) -> Result<JobResult>;
     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<Result<()>>) -> Self {
+    pub(crate) fn spawned(task: JoinHandle<Result<JobResult>>) -> 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<JobResult> {
         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<Error>),
     Cancelled,
 }
 
 impl Outcome {
-    fn into_result(self) -> Result<()> {
+    fn into_result(self) -> Result<JobResult> {
         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<Result<()>>) -> Self {
+    fn new(task: JoinHandle<Result<JobResult>>) -> 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<String> {
         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<JobResult> {
         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() {

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<S: HttpSend> JobHandle for RemoteJob<S> {
         Ok(self.describe().await?.job_state.client_label())
     }
 
-    async fn wait(&self) -> Result<()> {
+    async fn wait(&self) -> Result<JobResult> {
         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<S: HttpSend> JobHandle for RemoteJob<S> {
 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(|_| {

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)
         })))
     }
 

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::<Vec<_>>();
         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:?}"),
         }
     }

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);