From b0f010ad8359246d789a444e80d3ac09ba7e369b Mon Sep 17 00:00:00 2001 From: Wyatt Alt Date: Tue, 1 Sep 2026 13:28:57 +0000 Subject: [PATCH] feat(nodejs): pause and resume jobs from the TypeScript SDK Companion to the Python surface: pauseJob and resumeJob on Connection, 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, and a resume re-queues it to pick work back up from checkpoints. The outcome strings mirror the server's answers -- "pausing", "already_paused", or "committing" (a job finalizing its results cannot be parked; retry shortly), and "resumed", "still_pausing", or "not_paused" -- so a caller can retry the transient refusals rather than treat them as failures. --- nodejs/__test__/remote.test.ts | 19 +++++++++++++++++++ nodejs/lancedb/connection.ts | 26 ++++++++++++++++++++++++++ nodejs/src/connection.rs | 28 ++++++++++++++++++++++++++++ 3 files changed, 73 insertions(+) diff --git a/nodejs/__test__/remote.test.ts b/nodejs/__test__/remote.test.ts index 708559b7b..13aed2908 100644 --- a/nodejs/__test__/remote.test.ts +++ b/nodejs/__test__/remote.test.ts @@ -987,6 +987,22 @@ 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") { res .writeHead(200, { @@ -1015,6 +1031,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"); + const history = await db.jobHistory("job-1"); expect(history.numRows).toEqual(2); diff --git a/nodejs/lancedb/connection.ts b/nodejs/lancedb/connection.ts index 263a338ab..835b1dc5b 100644 --- a/nodejs/lancedb/connection.ts +++ b/nodejs/lancedb/connection.ts @@ -583,6 +583,24 @@ export abstract class Connection { */ abstract cancelJob(jobId: string): Promise; + /** + * 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; + + /** + * 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; + /** * The lifecycle event history of a server-side job, as an Arrow table. * @@ -944,6 +962,14 @@ export class LocalConnection extends Connection { return this.inner.cancelJob(jobId); } + async pauseJob(jobId: string): Promise { + return this.inner.pauseJob(jobId); + } + + async resumeJob(jobId: string): Promise { + return this.inner.resumeJob(jobId); + } + async jobHistory(jobId?: string): Promise { const buf = await this.inner.jobHistory(jobId); if (buf.length === 0) { diff --git a/nodejs/src/connection.rs b/nodejs/src/connection.rs index 5cf676256..94e41ded7 100644 --- a/nodejs/src/connection.rs +++ b/nodejs/src/connection.rs @@ -477,6 +477,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 { + 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 { + 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(), + }) + } + /// The lifecycle event history of a server-side job (all jobs when /// `job_id` is null), as an Arrow IPC stream buffer. Empty when there is /// no history.