fix: reject repeated job pagination tokens

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.39.0-beta.2"
+current_version = "0.39.0-beta.3"
 parse = """(?x)
     (?P<major>0|[1-9]\\d*)\\.
     (?P<minor>0|[1-9]\\d*)\\.

Cargo.lock

@@ -5474,7 +5474,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb"
-version = "0.39.0-beta.2"
+version = "0.39.0-beta.3"
 dependencies = [
  "ahash",
  "anyhow",
@@ -5565,7 +5565,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb-nodejs"
-version = "0.39.0-beta.2"
+version = "0.39.0-beta.3"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5590,7 +5590,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb-python"
-version = "0.39.0-beta.2"
+version = "0.39.0-beta.3"
 dependencies = [
  "arrow",
  "async-trait",

docs/src/java/java.md

@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
     <groupId>com.lancedb</groupId>
     <artifactId>lancedb-core</artifactId>
-    <version>0.39.0-beta.2</version>
+    <version>0.39.0-beta.3</version>
 </dependency>
 ```
 

java/lancedb-core/pom.xml

@@ -8,7 +8,7 @@
     <parent>
       <groupId>com.lancedb</groupId>
       <artifactId>lancedb-parent</artifactId>
-      <version>0.39.0-beta.2</version>
+      <version>0.39.0-beta.3</version>
       <relativePath>../pom.xml</relativePath>
     </parent>
 

java/pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.lancedb</groupId>
     <artifactId>lancedb-parent</artifactId>
-    <version>0.39.0-beta.2</version>
+    <version>0.39.0-beta.3</version>
     <packaging>pom</packaging>
     <name>${project.artifactId}</name>
     <description>LanceDB Java SDK Parent POM</description>

nodejs/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.39.0-beta.2"
+version = "0.39.0-beta.3"
 publish = false
 license.workspace = true
 description.workspace = true

nodejs/npm/darwin-arm64/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.39.0-beta.2",
+	"version": "0.39.0-beta.3",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",

nodejs/npm/linux-arm64-gnu/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.39.0-beta.2",
+	"version": "0.39.0-beta.3",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",

nodejs/npm/linux-arm64-musl/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.39.0-beta.2",
+	"version": "0.39.0-beta.3",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",

nodejs/npm/linux-x64-gnu/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.39.0-beta.2",
+	"version": "0.39.0-beta.3",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",

nodejs/npm/linux-x64-musl/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.39.0-beta.2",
+	"version": "0.39.0-beta.3",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",

nodejs/npm/win32-arm64-msvc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.39.0-beta.2",
+  "version": "0.39.0-beta.3",
   "os": [
     "win32"
   ],

nodejs/npm/win32-x64-msvc/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.39.0-beta.2",
+	"version": "0.39.0-beta.3",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",

nodejs/package.json

@@ -11,7 +11,7 @@
     "ann"
   ],
   "private": false,
-  "version": "0.39.0-beta.2",
+  "version": "0.39.0-beta.3",
   "main": "dist/index.js",
   "exports": {
     ".": "./dist/index.js",

python/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.39.0-beta.2"
+version = "0.39.0-beta.3"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"

rust/lancedb/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.39.0-beta.2"
+version = "0.39.0-beta.3"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true

rust/lancedb/src/remote/db.rs

@@ -688,6 +688,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
     async fn list_jobs(&self) -> Result<Vec<JobInfo>> {
         let mut out = Vec::new();
         let mut page_token: Option<String> = None;
+        let mut seen_page_tokens = HashSet::new();
         for page in 0..MAX_LIST_JOBS_PAGES {
             let mut body = serde_json::json!({});
             if let Some(token) = &page_token {
@@ -696,7 +697,8 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
             let req = self.client.post("/v1/jobs/list").json(&body);
             let (request_id, rsp) = self.client.send(req).await?;
             let rsp = self.client.check_response(&request_id, rsp).await?;
-            let body: RemoteListJobsResponse = rsp.json().await.err_to_http(request_id)?;
+            let status = rsp.status();
+            let body: RemoteListJobsResponse = rsp.json().await.err_to_http(request_id.clone())?;
             out.extend(body.jobs.into_iter().map(|row| JobInfo {
                 job_id: row.job_id,
                 table: row.table,
@@ -704,10 +706,17 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
                 state: job_state_to_client(&row.state),
                 created_at_millis: row.created_at_millis,
             }));
-            page_token = body.page_token;
-            if page_token.is_none() {
+            let Some(next_page_token) = body.page_token.filter(|token| !token.is_empty()) else {
                 break;
+            };
+            if !seen_page_tokens.insert(next_page_token.clone()) {
+                return Err(Error::Http {
+                    source: "Job listing response repeated a page_token".into(),
+                    request_id,
+                    status_code: Some(status),
+                });
             }
+            page_token = Some(next_page_token);
             if page + 1 == MAX_LIST_JOBS_PAGES {
                 log::warn!(
                     "list_jobs truncated after {} pages ({} jobs)",
@@ -2622,6 +2631,37 @@ mod tests {
         assert_eq!(jobs[2].state, "failed");
     }
 
+    #[tokio::test]
+    async fn test_list_jobs_rejects_a_page_token_cycle() {
+        let requests = Arc::new(AtomicUsize::new(0));
+        let seen = requests.clone();
+        let conn = Connection::new_with_handler(move |request| {
+            let body: serde_json::Value =
+                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
+            match seen.fetch_add(1, Ordering::SeqCst) {
+                0 => assert!(body.get("page_token").is_none()),
+                _ => assert_eq!(body["page_token"], "loop"),
+            }
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"jobs": [], "page_token": "loop"}"#)
+                .unwrap()
+        });
+
+        let error = conn.list_jobs().await.unwrap_err();
+        assert!(
+            matches!(
+                &error,
+                Error::Http {
+                    status_code: Some(http::StatusCode::OK),
+                    ..
+                }
+            ),
+            "got {error:?}"
+        );
+        assert_eq!(requests.load(Ordering::SeqCst), 2);
+    }
+
     #[tokio::test]
     async fn test_open_job() {
         let conn = Connection::new_with_handler(|request| {