Perf: speed up client-side queue filtering in job list

CHANGELOG.md

@@ -1,5 +1,14 @@
 ## lifebit-ai/cloudos-cli: changelog
 
+## v2.95.1 (2026-07-16)
+
+### Fix:
+
+- `cloudos job list --filter-queue` appeared to hang forever on large workspaces. Queue filtering is client-side (the jobs API has no queue parameter), so all workspace jobs must be scanned; the scan now uses the API maximum page size (100 jobs per request instead of `--page-size`, typically 10), reuses a single HTTP connection and prints scan progress
+- The queue-filtered scan now always starts from the first API page, so results are no longer silently missing when combined with `--page`
+- `--page` is now honoured as the initially displayed page of the filtered results when combined with `--filter-queue`
+- Documentation now correctly states that `--filter-queue` returns only batch jobs running on the specified queue (non-batch jobs were never preserved)
+
 ## v2.95.0 (2026-06-30)
 
 ### Breaking:

README.md

@@ -231,9 +231,8 @@ cloudos job list --help
 │                                                workflow (by date).                                       │
 │    --filter-job-id                 TEXT        Filter jobs by specific job ID.                           │
 │    --filter-only-mine                          Filter to show only jobs belonging to the current user.   │
-│    --filter-queue                  TEXT        Filter jobs by queue name. Only applies to jobs running   │
-│                                                in batch environment. Non-batch jobs are preserved in     │
-│                                                results.                                                  │
+│    --filter-queue                  TEXT        Filter jobs by queue name. Only batch jobs running on     │
+│                                                the specified queue are returned.                         │
 │    --filter-owner                  TEXT        Filter jobs by owner username.                            │
 │    --verbose                                   Whether to print information messages or not.             │
 │    --disable-ssl-verification                  Disable SSL certificate verification. Please, remember    │
@@ -912,7 +911,7 @@ You can find specific jobs within your workspace using the filtering options. Fi
 - **`--filter-job-id`**: Filter jobs by specific job ID (exact match required)
 - **`--filter-only-mine`**: Show only jobs belonging to the current user
 - **`--filter-owner`**: Show only jobs for the specified owner (exact match required, e.g., "John Doe")
-- **`--filter-queue`**: Filter jobs by queue name (works with both regular and system queues; only applies to batch jobs)
+- **`--filter-queue`**: Filter jobs by queue name (works with both regular and system queues; only applies to batch jobs). Queue filtering is performed client-side, so all the workspace jobs are scanned; a progress message is displayed during the scan, which can take a while on workspaces with many jobs
 
 **Filtering Examples**
 

cloudos_cli/_version.py

@@ -1 +1 @@
-__version__ = '2.95.0'
+__version__ = '2.95.1'

cloudos_cli/clos.py

@@ -3,12 +3,14 @@
 """
 
 import requests
+import sys
 import time
 import json
 from dataclasses import dataclass
 from cloudos_cli.utils.cloud import find_cloud
 from cloudos_cli.utils.errors import BadRequestException, JoBNotCompletedException, NotAuthorisedException, JobAccessDeniedException
-from cloudos_cli.utils.requests import retry_requests_get, retry_requests_post, retry_requests_put
+from cloudos_cli.utils.requests import (retry_requests_get, retry_requests_post,
+                                        retry_requests_put, create_retry_session)
 import pandas as pd
 from cloudos_cli.utils.last_wf import youngest_workflow_id_by_name
 from datetime import datetime, timezone
@@ -1049,8 +1051,10 @@ def get_job_list(self, workspace_id, last_n_jobs=None, page=None, page_size=None
             Filter jobs by owner username (will be resolved to user ID).
         filter_queue : string, optional
             Filter jobs by queue name (will be resolved to queue ID).
-            Only applies to jobs running in batch environment.
-            Non-batch jobs are preserved in results as they don't use queues.
+            Only batch jobs running on the specified queue are returned;
+            jobs without a batch queue are excluded.
+            Note: the API does not support server-side queue filtering, so
+            all the workspace jobs are scanned and filtered client-side.
         last : bool, optional
             When workflows are duplicated, use the latest imported workflow (by date).
 
@@ -1199,57 +1203,85 @@ def get_job_list(self, workspace_id, last_n_jobs=None, page=None, page_size=None
 
         # --- Fetch jobs page by page ---
         all_jobs = []
-        params["limit"] = current_page_size
+        # The queue filter is applied client-side, so all the workspace jobs must
+        # be scanned. Scan from the first page using the API maximum page size and
+        # a reusable HTTP session to minimise the number of requests, and report
+        # progress as the scan can take a while on large workspaces.
+        scan_all_pages = bool(filter_queue)
+        if scan_all_pages:
+            current_page = 1
+            params["limit"] = 100
+        else:
+            params["limit"] = current_page_size
         last_pagination_metadata = None  # Track the last pagination metadata
+        scanned_job_count = 0
 
-        while True:
-            params["page"] = current_page
-
-            r = retry_requests_get(f"{self.cloudos_url}/api/v2/jobs", params=params, headers=headers, verify=verify)
-            if r.status_code >= 400:
-                raise BadRequestException(r)
+        with create_retry_session() as session:
+            while True:
+                params["page"] = current_page
 
-            content = r.json()
-            page_jobs = content.get('jobs', [])
+                r = retry_requests_get(f"{self.cloudos_url}/api/v2/jobs", params=params,
+                                       headers=headers, verify=verify, session=session)
+                if r.status_code >= 400:
+                    raise BadRequestException(r)
 
-            # Capture pagination metadata
-            last_pagination_metadata = content.get('paginationMetadata', None)
+                content = r.json()
+                page_jobs = content.get('jobs', [])
+                raw_page_job_count = len(page_jobs)
 
-            # No jobs returned, we've reached the end
-            if not page_jobs:
-                break
+                # Capture pagination metadata
+                last_pagination_metadata = content.get('paginationMetadata', None)
 
-            # Apply queue filter during pagination (if specified)
-            # jobQueue is a dict with "id" and "name" keys, extract the id for comparison
-            if filter_queue and queue_id:
-                filtered_jobs = []
-                for job in page_jobs:
-                    job_queue = job.get("batch", {}).get("jobQueue", {})
-                    # jobQueue is a dict like {"id": "...", "name": "...", ...}
-                    job_queue_id = job_queue.get("id") if isinstance(job_queue, dict) else job_queue
-                    if job_queue_id == queue_id:
-                        filtered_jobs.append(job)
-                page_jobs = filtered_jobs
-
-            all_jobs.extend(page_jobs)
-
-            # Check stopping conditions based on mode
-            if use_pagination_mode:
-                # In pagination mode (last_n_jobs), continue until we have enough jobs (after filtering)
-                if target_job_count != 'all' and len(all_jobs) >= target_job_count:
-                    break
-            else:
-                if not filter_queue and len(all_jobs) >= current_page_size:
+                # No jobs returned, we've reached the end
+                if not page_jobs:
                     break
 
-            # Check if we reached the last page (fewer jobs than requested page size)
-            # Note: For queue filter, we check the unfiltered page_jobs count from the API
-            # This ensures we stop when the API has exhausted results
-            raw_page_jobs = content.get('jobs', [])
-            if len(raw_page_jobs) < params["limit"]:
-                break  # Last page
+                # Apply queue filter during pagination (if specified)
+                # jobQueue is a dict with "id" and "name" keys, extract the id for comparison
+                if filter_queue and queue_id:
+                    filtered_jobs = []
+                    for job in page_jobs:
+                        job_queue = job.get("batch", {}).get("jobQueue", {})
+                        # jobQueue is a dict like {"id": "...", "name": "...", ...}
+                        job_queue_id = job_queue.get("id") if isinstance(job_queue, dict) else job_queue
+                        if job_queue_id == queue_id:
+                            filtered_jobs.append(job)
+                    page_jobs = filtered_jobs
+
+                all_jobs.extend(page_jobs)
+
+                if scan_all_pages:
+                    scanned_job_count += raw_page_job_count
+                    total_job_count = (last_pagination_metadata or {}).get('Pagination-Count', '?')
+                    progress_msg = (f"\tScanning workspace jobs for queue '{filter_queue}': "
+                                    f"{scanned_job_count}/{total_job_count} jobs scanned, "
+                                    f"{len(all_jobs)} matching...")
+                    if sys.stderr.isatty():
+                        # Self-updating single line on interactive terminals
+                        print(f"\r{progress_msg}", end='', flush=True, file=sys.stderr)
+                    else:
+                        # Newline-terminated lines when redirected (e.g. CI logs)
+                        print(progress_msg, flush=True, file=sys.stderr)
+
+                # Check stopping conditions based on mode
+                if use_pagination_mode:
+                    # In pagination mode (last_n_jobs), continue until we have enough jobs (after filtering)
+                    if target_job_count != 'all' and len(all_jobs) >= target_job_count:
+                        break
+                else:
+                    if not filter_queue and len(all_jobs) >= current_page_size:
+                        break
+
+                # Check if we reached the last page (fewer jobs than requested page size)
+                # Note: For queue filter, we check the unfiltered job count from the API
+                # This ensures we stop when the API has exhausted results
+                if raw_page_job_count < params["limit"]:
+                    break  # Last page
+
+                current_page += 1
 
-            current_page += 1
+        if scan_all_pages and scanned_job_count and sys.stderr.isatty():
+            print(file=sys.stderr)  # End the self-updating progress line
 
         # --- Apply limit after all filtering ---
         if use_pagination_mode and target_job_count != 'all' and isinstance(target_job_count, int) and target_job_count > 0:

cloudos_cli/jobs/cli.py

@@ -1185,7 +1185,7 @@ def job_details(ctx,
               help='Filter to show only jobs belonging to the current user.',
               is_flag=True)
 @click.option('--filter-queue',
-              help='Filter jobs by queue name. Only applies to jobs running in batch environment. Non-batch jobs are preserved in results.')
+              help='Filter jobs by queue name. Only batch jobs running on the specified queue are returned.')
 @click.option('--filter-owner',
               help='Filter jobs by owner username.')
 @click.option('--verbose',
@@ -1326,15 +1326,11 @@ def list_jobs(ctx,
             # For client-filtered results, we have all jobs already
             # Create a callback that paginates them client-side using helper function
             fetch_page = create_client_pagination_callback(my_jobs_r, page_size)
-
-            # Show first page of filtered results
-            first_page_jobs = my_jobs_r[:page_size]
-            first_page_metadata = {
-                'Pagination-Count': len(my_jobs_r),
-                'Pagination-Page': 1,
-                'Pagination-Limit': page_size
-            }
-            create_job_list_table(first_page_jobs, cloudos_url, first_page_metadata, selected_columns, fetch_page_callback=fetch_page)
+
+            # Show the requested page of filtered results
+            initial_page = fetch_page(page)
+            create_job_list_table(initial_page['jobs'], cloudos_url, initial_page['pagination_metadata'],
+                                  selected_columns, fetch_page_callback=fetch_page)
         else:
             # For normal (non-filtered) results, use API pagination with helper function
             fetch_page = create_api_pagination_callback(

cloudos_cli/utils/details.py

@@ -756,7 +756,7 @@ def create_job_list_table(jobs, cloudos_url, pagination_metadata=None, selected_
         # Show pagination controls only if there are multiple pages
         if total_pages > 1:
             if not sys.stdin.isatty():
-                console.print("\n[yellow]Note: Pagination not available in non-interactive mode. Showing page 1 of {0}.[/yellow]".format(total_pages))
+                console.print("\n[yellow]Note: Pagination not available in non-interactive mode. Showing page {0} of {1}.[/yellow]".format(current_page, total_pages))
                 console.print("[yellow]Run in an interactive terminal to navigate through all pages.[/yellow]")
                 break