mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-12 14:52:26 +00:00
feat: pause and resume jobs from the Python SDK (#4112)
Cancellation is the only job control the SDK exposes, and it is terminal, so a long-running server-side job cannot be parked and picked up again. This adds `pause_job` and `resume_job` to the Rust core connection and the Python bindings (sync and async), posting to the server's `/v1/jobs/pause` and `/v1/jobs/resume` endpoints. A pause parks the job until it is resumed: its workers drain and stop. The outcome strings mirror the server's answers — a job finalizing its results reports `"committing"` and cannot be parked, and a resume before the drain is confirmed reports `"still_pausing"`; both are retried rather than failed. Resuming re-queues the job and its workers pick their work back up from checkpoints. Local connections report the operations as unsupported, like the rest of the jobs API.
This commit is contained in:
@@ -1183,6 +1183,24 @@ describe("remote connection jobs surface", () => {
|
||||
res
|
||||
.writeHead(200, { "Content-Type": "application/json" })
|
||||
.end('{"job_id": "job-1"}');
|
||||
} else if (req.url === "/v1/jobs/pause") {
|
||||
if (payload["job_id"] !== "job-1") {
|
||||
res.writeHead(404).end("no such job");
|
||||
return;
|
||||
}
|
||||
res
|
||||
.writeHead(200, { "Content-Type": "application/json" })
|
||||
.end('{"job_id": "job-1", "paused": true}');
|
||||
} else if (req.url === "/v1/jobs/resume") {
|
||||
if (payload["job_id"] !== "job-1") {
|
||||
res.writeHead(404).end("no such job");
|
||||
return;
|
||||
}
|
||||
res
|
||||
.writeHead(200, { "Content-Type": "application/json" })
|
||||
.end(
|
||||
'{"job_id": "job-1", "resumed": false, "still_pausing": true}',
|
||||
);
|
||||
} else if (req.url === "/v1/jobs/query_events") {
|
||||
queryEventsPayloads.push(payload);
|
||||
res
|
||||
@@ -1204,6 +1222,9 @@ describe("remote connection jobs surface", () => {
|
||||
expect(await db.cancelJob("job-1")).toBe(true);
|
||||
expect(await db.cancelJob("missing")).toBe(false);
|
||||
|
||||
expect(await db.pauseJob("job-1")).toEqual("pausing");
|
||||
expect(await db.resumeJob("job-1")).toEqual("still_pausing");
|
||||
|
||||
// Opening a job hands back a populated handle; a missing one rejects.
|
||||
await expect(db.openJob("missing")).rejects.toThrow("not found");
|
||||
const finished = await db.openJob("job-2");
|
||||
|
||||
@@ -650,6 +650,24 @@ export abstract class Connection {
|
||||
* such job exists. Cancelling an already-terminal job is a no-op success.
|
||||
*/
|
||||
abstract cancelJob(jobId: string): Promise<boolean>;
|
||||
|
||||
/**
|
||||
* Pause a server-side job by id.
|
||||
*
|
||||
* The job's workers drain and it stays parked until resumed. Resolves to
|
||||
* "pausing", "already_paused", or "committing" -- a job finalizing its
|
||||
* results cannot be parked; retry shortly.
|
||||
*/
|
||||
abstract pauseJob(jobId: string): Promise<string>;
|
||||
|
||||
/**
|
||||
* Resume a paused server-side job by id.
|
||||
*
|
||||
* Its workers pick their work back up from checkpoints. Resolves to
|
||||
* "resumed", "still_pausing" -- the pause's worker drain is not confirmed
|
||||
* yet; retry shortly -- or "not_paused".
|
||||
*/
|
||||
abstract resumeJob(jobId: string): Promise<string>;
|
||||
}
|
||||
|
||||
/** @hideconstructor */
|
||||
@@ -1049,6 +1067,14 @@ export class LocalConnection extends Connection {
|
||||
async cancelJob(jobId: string): Promise<boolean> {
|
||||
return this.inner.cancelJob(jobId);
|
||||
}
|
||||
|
||||
async pauseJob(jobId: string): Promise<string> {
|
||||
return this.inner.pauseJob(jobId);
|
||||
}
|
||||
|
||||
async resumeJob(jobId: string): Promise<string> {
|
||||
return this.inner.resumeJob(jobId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -600,6 +600,34 @@ impl Connection {
|
||||
self.get_inner()?.cancel_job(&job_id).await.default_error()
|
||||
}
|
||||
|
||||
/// Pause a server-side job by id: its workers drain and it stays parked
|
||||
/// until resumed. Returns "pausing", "already_paused", or "committing".
|
||||
#[napi(catch_unwind)]
|
||||
pub async fn pause_job(&self, job_id: String) -> napi::Result<String> {
|
||||
let status = self.get_inner()?.pause_job(&job_id).await.default_error()?;
|
||||
Ok(match status {
|
||||
lancedb::database::PauseJobStatus::Pausing => "pausing".to_string(),
|
||||
lancedb::database::PauseJobStatus::AlreadyPaused => "already_paused".to_string(),
|
||||
lancedb::database::PauseJobStatus::Committing => "committing".to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Resume a paused server-side job by id. Returns "resumed",
|
||||
/// "still_pausing", or "not_paused".
|
||||
#[napi(catch_unwind)]
|
||||
pub async fn resume_job(&self, job_id: String) -> napi::Result<String> {
|
||||
let status = self
|
||||
.get_inner()?
|
||||
.resume_job(&job_id)
|
||||
.await
|
||||
.default_error()?;
|
||||
Ok(match status {
|
||||
lancedb::database::ResumeJobStatus::Resumed => "resumed".to_string(),
|
||||
lancedb::database::ResumeJobStatus::StillPausing => "still_pausing".to_string(),
|
||||
lancedb::database::ResumeJobStatus::NotPaused => "not_paused".to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
#[napi(catch_unwind)]
|
||||
/// Describe a namespace and return its properties.
|
||||
pub async fn describe_namespace(
|
||||
|
||||
Reference in New Issue
Block a user