Implementation of Global Weighted Queue to process Digitize/Ingestion requests by manju956 · Pull Request #1280 · IBM/project-ai-services

manju956 · 2026-08-18T15:07:47Z

Implements the design proposal https://github.com/IBM/project-ai-services/blob/main/docs/proposals/docling_conversion_queue.md
Covers various scenarios like baseline, mixed normal and large files, Queue overflow, round robin fairness, HOL blocking etc.

…eline execution Replaces the in-process ConcurrencyManager semaphores with a durable, Postgres-backed conversion queue driven by a single async dispatcher. Pipelines are moved off the event loop by making background task helpers plain def functions dispatched to Starlette's thread pool. Conversion queue (feat) ----------------------- db/models.py - Add ConversionTask ORM model (conversion_tasks table) with operation, cached_file, output_format, page_count, is_large, status, result_path, error, timestamps. - Indexes: idx_ct_status_op_queued (dispatcher pick), idx_ct_job_id (pipeline poll). db/manager.py - Add ConversionTask CRUD: create_conversion_task, get_conversion_task, get_conversion_task_by_job_id, get_conversion_tasks_by_job_id, get_conversion_tasks, get_queued_count, get_queued_counts, update_task_status, peek_head, claim_head (SELECT FOR UPDATE SKIP LOCKED + RETURNING), promote_pending. settings.py - Replace digitization_concurrency_limit / ingestion_concurrency_limit with ingestion_queue_quota (10), digitization_queue_quota (5), conversion_poll_interval (2.0s). workers/conversion_semaphore.py [new] - WeightedSemaphore: capacity=doc_worker_size (4); weight 1 for normal files, 2 for large files (>heavy_doc_page_threshold pages). workers/conversion_dispatcher.py [new] - Round-robin dispatch loop: claims one ingestion task + one digitization task per tick subject to semaphore capacity. - Head-of-line blocking: reserves budget for a blocked large file so the other operation type cannot consume the units it is waiting for. - Single shared ProcessPoolExecutor (max_workers=4); semaphore and pool size agree so no internal pool queuing occurs. - promote_pending() after each tick to backfill quota headroom. api/v1/jobs.py - Remove ConcurrencyManager semaphore acquire/release, has_active_jobs() block. - Add per-op queue quota gate (get_queued_counts + quota check). - Insert conversion_tasks rows at admission time: min(N, free_slots) as queued, remainder as pending. pipeline/digitize.py - Remove ProcessPoolExecutor + convert_document_format call. - Poll db_manager.get_conversion_task_by_job_id() until dispatcher marks task completed or failed. processing/orchestrator.py - Remove ProcessPoolExecutor, _run_batch, light/heavy split. - process_documents() builds task_id→path map from existing conversion_tasks rows; reactive while-loop polls task status and feeds completions immediately into process→chunk→index. pipeline/ingest.py - ingest() calls process_documents() synchronously. utils/recovery.py - Add recover_conversion_tasks(): running→failed (cleans chunk dirs), queued/pending→failed if cached file missing, keeps the rest. app.py - Call recover_conversion_tasks() on startup. - Start asyncio.create_task(dispatch_loop()) after recovery. - Cancel dispatcher gracefully on shutdown. tests/test_digitize_app_endpoints.py - Update fixture: patch db_manager.get_queued_counts, db_manager.create_conversion_task, get_document_page_count instead of ConcurrencyManager. - Replace test_rejects_when_ingestion_job_already_active with test_rejects_when_ingestion_queue_full (quota-full path). Pipeline background task fix (fix) ----------------------------------- - Restore _run_digitize and _run_ingest helpers in api/v1/jobs.py; each calls the pipeline and invokes cleanup_staging_directory in its finally block. - create_job calls background_tasks.add_task(_run_digitize|_run_ingest) after inserting conversion_tasks rows so the dispatcher converts and the pipeline reacts. - Test fixture stubs _run_digitize and _run_ingest to avoid executing real pipeline logic under TestClient. Thread-pool pipeline execution (refactor) ------------------------------------------ - _run_digitize and _run_ingest converted from async def to plain def so Starlette's BackgroundTasks dispatches them to a thread-pool worker. - pipeline/digitize.py: async def -> def, await asyncio.sleep -> time.sleep, drop asyncio import. - pipeline/ingest.py: async def -> def, drop await on process_documents. - processing/orchestrator.py: async def -> def, await asyncio.sleep -> time.sleep, drop asyncio import. The digitize pipeline blocks via time.sleep while polling for conversion completion, and the ingest pipeline spawns ThreadPoolExecutors internally — neither belongs on the event loop. Resolves: docs/proposals/docling_conversion_queue.md Signed-off-by: Dharaneesharan Ravichandran <dharaneeshwaran.ravichandran@ibm.com>

… changes - Consolidate db_manager mock into a single object covering get_queued_counts, find_completed_document_by_hash, and delete_job. - Replace stale jobs_router.get_document_page_count patch (symbol no longer imported in jobs.py) with dg_util.get_document_page_count. - Stub enqueue_conversion_tasks and generate_file_checksum in fixture. - Fix test_delete_completed_job_succeeds to patch delete_job on the module- level mock rather than the real singleton. - Add get_queued_counts to the test_mixed_batch local db_manager mock. Signed-off-by: Dharaneesharan Ravichandran <dharaneeshwaran.ravichandran@ibm.com>

Replace removed concurrency_manager / ingestion_concurrency_limit / digitization_concurrency_limit references with the queue-based system introduced in the dispatcher commit: - test_deduplication.py: jobs_test_client fixture now uses ingestion_queue_quota / digitization_queue_quota / heavy_doc_page_threshold / conversion_poll_interval, mocks db_manager with get_queued_counts, stubs enqueue_conversion_tasks + generate_file_checksum + pipeline helpers. All 8 per-test mock_hash_db instances gain get_queued_counts. - test_connector_endpoints.py: connector_test_client fixture same treatment — drop concurrency_manager / has_active_jobs stubs, add queue-quota fields and the unified db_manager mock. Signed-off-by: Dharaneesharan Ravichandran <dharaneeshwaran.ravichandran@ibm.com>

Signed-off-by: manju956 <manjunath.ac956@gmail.com>

## Summary Three production-quality fixes to the Postgres-backed conversion queue. ### 1. Atomic admission quota gate (jobs.py + db/manager.py) Replace the non-atomic get_queued_counts() read with check_quota_atomic() which takes a session-level advisory lock (pg_advisory_xact_lock) before counting queued tasks. Prevents two concurrent POST /v1/jobs calls from both reading 'queued=9 < 10' and both being admitted, pushing the count to 11 and silently exceeding the quota. ### 2. Staged-file cleanup on conversion completion (conversion_dispatcher.py) Add _safe_remove() helper and call it in _run_conversion() finally block. Deletes the staged input PDF after the conversion worker finishes (success or failure). Prevents unbounded accumulation of staged files. Also removes the post-conversion get_document_page_count() call — page_count is now read from the conversion_tasks row (stored at enqueue time) so the file need not exist after conversion. ### 3. Semaphore weight threshold boundary fix (conversion_semaphore.py) Change '> heavy_doc_page_threshold' to '>= heavy_doc_page_threshold' so a file with exactly page_count == threshold is classified as heavy (weight=2) consistently with the DB admission check. ## Tests - TestCheckQuotaAtomic: advisory-lock path, quota-full, fail-open on DB error - TestSafeRemove: deletes existing, ignores missing, ignores PermissionError - TestRunConversionUpdated: uses task.page_count not file read; file deleted - Updated all mocks from get_queued_counts to check_quota_atomic signature Signed-off-by: manju956 <manjunath.ac956@gmail.com>

Signed-off-by: manju956 <manjunath.ac956@gmail.com>

sats-23 · 2026-08-19T08:43:24Z

+        else:
+            quota = settings.digitize.digitization_queue_quota
+
+        quota_ok, queued_for_op = db_manager.check_quota_atomic(op_key, quota)


Will this be the entry point for a connector job as well? Are there plans to reject connector jobs as well with 429?

We have to design how the connector needs to be integrated into this queuing logic. I was thinking of introducing a new op type for connector jobs so that it would round robin between these op types. Let's discuss more on this separately.

dharaneeshvrd

@manju956 Need to rebase with main to review effectively.

dharaneeshvrd · 2026-08-21T06:10:22Z

-    """Run the digitization pipeline and release the semaphore slot."""
-    status_mgr = get_status_manager(job_id)
-    job_staging_path = settings.digitize.staging_dir / job_id
+    """


We need to use the quotes inline with the comment as per ruff rules added for MCP work
Please be aware of this @sats-23 @manalilatkar @Niharika0306

dharaneeshvrd · 2026-08-21T06:19:54Z

-        logger.info(f"🚀 Digitization started for job: {job_id}")
-        await asyncio.to_thread(digitize, job_staging_path, job_id, doc_id_dict, output_format, file_checksum_dict)
-        logger.info(f"Digitization for job {job_id} completed successfully")
+        logger.info(f"🚀 Digitization pipeline started for job: {job_id}")


@manju956 Some of these log messages and earlier structures were heavily modified by bob even though when its not required. Can we please revert those and keep the earlier changes and allow only the changes which are absolutely needed?
I know some of these are from my commit, apologies for doing this.

dharaneeshvrd · 2026-08-21T06:24:42Z

+        else:
+            quota = settings.digitize.digitization_queue_quota
+
+        quota_ok, queued_for_op = db_manager.check_quota_atomic(op_key, quota)


We have to design how the connector needs to be integrated into this queuing logic. I was thinking of introducing a new op type for connector jobs so that it would round robin between these op types. Let's discuss more on this separately.

dharaneeshvrd · 2026-08-21T06:34:48Z

    return len(affected_ids)
+
+
+def recover_conversion_tasks() -> int:


I think now we need to redesign the recovery a bit different since we would have track of what are the documents yet to be converted. Maybe we need to give them a try if necessary staged files are present and continue the pipeline.
Maybe for future, I ll create a JIRA story for this.

dharaneeshvrd · 2026-08-21T06:36:39Z

        logger.error(f"Error during zombie job recovery: {exc}", exc_info=True)

-
-def _shutdown():


No need to remove this method
Just extend this method to call cancel on the dispatcher

dharaneeshvrd · 2026-08-21T06:38:00Z

+    queued_at       TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    started_at      TIMESTAMPTZ,
+    completed_at    TIMESTAMPTZ,
+    updated_at      TIMESTAMPTZ     NOT NULL DEFAULT NOW(),


dont think this field is needed since we would have only one operation, it won't be updated regularly to be that useful

manju956 requested review from dharaneeshvrd and sats-23 August 18, 2026 15:07

manju956 self-assigned this Aug 18, 2026

manju956 added enhancement New feature or request squad/usecases labels Aug 18, 2026

manju956 force-pushed the globalQueue branch 2 times, most recently from b5dde5b to 030708c Compare August 19, 2026 04:49

dharaneeshvrd and others added 6 commits August 19, 2026 10:25

Add tests for conversion queue

e4c2cb5

Signed-off-by: manju956 <manjunath.ac956@gmail.com>

[chore] bump up versions

51f2ff7

Signed-off-by: manju956 <manjunath.ac956@gmail.com>

manju956 force-pushed the globalQueue branch from 030708c to 51f2ff7 Compare August 19, 2026 05:34

sats-23 reviewed Aug 19, 2026

View reviewed changes

dharaneeshvrd requested a review from manalilatkar August 21, 2026 05:45

dharaneeshvrd reviewed Aug 21, 2026

View reviewed changes

Navigation Menu

Uh oh!

Implementation of Global Weighted Queue to process Digitize/Ingestion requests - #1280

Implementation of Global Weighted Queue to process Digitize/Ingestion requests#1280
manju956 wants to merge 6 commits into
IBM:mainfrom
manju956:globalQueue

manju956 commented Aug 18, 2026

Uh oh!

sats-23 Aug 19, 2026

Uh oh!

dharaneeshvrd Aug 21, 2026

Uh oh!

dharaneeshvrd left a comment

Uh oh!

dharaneeshvrd Aug 21, 2026

Uh oh!

dharaneeshvrd Aug 21, 2026

Uh oh!

dharaneeshvrd Aug 21, 2026

Uh oh!

dharaneeshvrd Aug 21, 2026

Uh oh!

dharaneeshvrd Aug 21, 2026

Uh oh!

dharaneeshvrd Aug 21, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		return len(affected_ids)


		def recover_conversion_tasks() -> int:

		logger.error(f"Error during zombie job recovery: {exc}", exc_info=True)


		def _shutdown():

Uh oh!

Conversation

manju956 commented Aug 18, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dharaneeshvrd left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants