Refactor list_all_keys into reusable list_keys

Extract the single-page VSS listing logic into a list_keys method
that accepts page_token and page_size parameters. list_internal
now drives the pagination loop itself, calling list_keys per page.

This prepares for PaginatedKVStore support which will reuse
list_keys for single-page queries.

This also fixes a potential issue where if the VSS server returned None
for the page token we could enter into an infinite loop.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: benthecarman

src/io/vss_store.rs

@@ -391,35 +391,34 @@ impl VssStoreInner {
 		}
 	}
 
-	async fn list_all_keys(
+	async fn list_keys(
 		&self, client: &VssClient<CustomRetryPolicy>, primary_namespace: &str,
-		secondary_namespace: &str,
-	) -> io::Result<Vec<String>> {
-		let mut page_token = None;
-		let mut keys = vec![];
+		secondary_namespace: &str, page_token: Option<String>, page_size: Option<i32>,
+	) -> io::Result<(Vec<String>, Option<String>)> {
 		let key_prefix = self.build_obfuscated_prefix(primary_namespace, secondary_namespace);
-		while page_token != Some("".to_string()) {
-			let request = ListKeyVersionsRequest {
-				store_id: self.store_id.clone(),
-				key_prefix: Some(key_prefix.clone()),
-				page_token,
-				page_size: None,
-			};
+		let request = ListKeyVersionsRequest {
+			store_id: self.store_id.clone(),
+			key_prefix: Some(key_prefix),
+			page_token,
+			page_size,
+		};
 
-			let response = client.list_key_versions(&request).await.map_err(|e| {
-				let msg = format!(
-					"Failed to list keys in {}/{}: {}",
-					primary_namespace, secondary_namespace, e
-				);
-				Error::new(ErrorKind::Other, msg)
-			})?;
+		let response = client.list_key_versions(&request).await.map_err(|e| {
+			let msg = format!(
+				"Failed to list keys in {}/{}: {}",
+				primary_namespace, secondary_namespace, e
+			);
+			Error::new(ErrorKind::Other, msg)
+		})?;
 
-			for kv in response.key_versions {
-				keys.push(self.extract_key(&kv.key)?);
-			}
-			page_token = response.next_page_token;
+		let mut keys = Vec::with_capacity(response.key_versions.len());
+		for kv in response.key_versions {
+			keys.push(self.extract_key(&kv.key)?);
 		}
-		Ok(keys)
+
+		// VSS may return an empty string instead of None to signal the last page.
+		let next_page_token = response.next_page_token.filter(|t| !t.is_empty());
+		Ok((keys, next_page_token))
 	}
 
 	async fn read_internal(
@@ -543,17 +542,18 @@ impl VssStoreInner {
 	) -> io::Result<Vec<String>> {
 		check_namespace_key_validity(&primary_namespace, &secondary_namespace, None, "list")?;
 
-		let keys = self
-			.list_all_keys(client, &primary_namespace, &secondary_namespace)
-			.await
-			.map_err(|e| {
-				let msg = format!(
-					"Failed to retrieve keys in namespace: {}/{} : {}",
-					primary_namespace, secondary_namespace, e
-				);
-				Error::new(ErrorKind::Other, msg)
-			})?;
-
+		let mut page_token: Option<String> = None;
+		let mut keys = vec![];
+		loop {
+			let (page_keys, next_page_token) = self
+				.list_keys(client, &primary_namespace, &secondary_namespace, page_token, None)
+				.await?;
+			keys.extend(page_keys);
+			match next_page_token {
+				Some(t) => page_token = Some(t),
+				None => break,
+			}
+		}
 		Ok(keys)
 	}