1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
//! Handles to operations a server may run asynchronously.
use std::sync::Arc;
use async_trait::async_trait;
use tokio::sync::watch;
use tokio::task::{AbortHandle, JoinHandle};
use crate::error::{Error, JobFailure, Result};
/// Backend-specific tracking for an asynchronous operation.
#[async_trait]
pub(crate) trait JobHandle: Send + Sync {
/// Server-assigned id, when the backend has one.
fn id(&self) -> Option<&str> {
None
}
async fn status(&self) -> Result<String>;
async fn wait(&self) -> Result<()>;
async fn cancel(&self) -> Result<()>;
}
/// A handle to an operation that may still be running.
///
/// The operation may already be complete when the handle is created.
pub struct Job {
handle: Option<Box<dyn JobHandle>>,
}
impl std::fmt::Debug for Job {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Job")
.field("id", &self.id())
.field("done", &self.handle.is_none())
.finish()
}
}
impl Job {
/// A job whose operation finished before the handle was created.
pub(crate) fn new_done() -> Self {
Self { handle: None }
}
pub(crate) fn new(handle: Box<dyn JobHandle>) -> Self {
Self {
handle: Some(handle),
}
}
/// A job running as a task in this process.
pub(crate) fn spawned(task: JoinHandle<Result<()>>) -> Self {
Self::new(Box::new(SpawnedJob::new(task)))
}
/// Identifies the operation on the server that is running it.
///
/// Returned for correlating with server logs or the jobs API. Operations
/// that run in this process have no server id and return `None`. The
/// value is opaque: parsing it or storing it to resume the job later is
/// not supported.
pub fn id(&self) -> Option<&str> {
self.handle.as_ref().and_then(|handle| handle.id())
}
/// The operation's current lifecycle state: "running", "finished",
/// "failed", or "cancelled".
///
/// A point snapshot; unlike [`Job::wait`] it does not block, raise on a
/// terminal failure state, or retry. States a newer server reports that
/// this client version does not know pass through as-is.
pub async fn status(&self) -> Result<String> {
match &self.handle {
None => Ok("finished".to_string()),
Some(handle) => handle.status().await,
}
}
/// Waits until the operation reaches a terminal state.
///
/// Returns [`crate::Error::JobFailed`] if the operation failed and
/// [`crate::Error::JobCancelled`] if it was cancelled.
pub async fn wait(&self) -> Result<()> {
match &self.handle {
None => Ok(()),
Some(handle) => handle.wait().await,
}
}
/// Requests cancellation of the operation.
///
/// Cancelling an operation that already finished is a no-op.
pub async fn cancel(&self) -> Result<()> {
match &self.handle {
None => Ok(()),
Some(handle) => handle.cancel().await,
}
}
}
/// How an in-process operation ended. Cloneable so every waiter can be given
/// the outcome; [`Error`] is not, so failures share one behind an [`Arc`].
#[derive(Clone)]
enum Outcome {
Succeeded,
Failed(Arc<Error>),
Cancelled,
}
impl Outcome {
fn into_result(self) -> Result<()> {
match self {
Self::Succeeded => Ok(()),
Self::Failed(source) => Err(Error::JobFailed {
job_id: None,
failure: JobFailure::from_source(source),
}),
Self::Cancelled => Err(Error::JobCancelled { job_id: None }),
}
}
}
/// Tracks an operation running as a task in this process. A second task
/// watches the first so that aborting it still produces an outcome, and so
/// that every caller of `wait` observes the same one.
struct SpawnedJob {
outcome: watch::Receiver<Option<Outcome>>,
abort: AbortHandle,
}
impl SpawnedJob {
fn new(task: JoinHandle<Result<()>>) -> 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(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}"),
})),
};
let _ = tx.send(Some(outcome));
});
Self { outcome, abort }
}
}
#[async_trait]
impl JobHandle for SpawnedJob {
async fn status(&self) -> Result<String> {
let label = match &*self.outcome.borrow() {
None => "running",
Some(Outcome::Succeeded) => "finished",
Some(Outcome::Failed(_)) => "failed",
Some(Outcome::Cancelled) => "cancelled",
};
Ok(label.to_string())
}
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(),
})?
.clone()
.expect("wait_for returns once an outcome is set");
settled.into_result()
}
async fn cancel(&self) -> Result<()> {
self.abort.abort();
Ok(())
}
}