Skip to content

feat(digitize): add cancel endpoint and cooperative pipeline cancellation - #1300

Open
dharaneeshvrd wants to merge 1 commit into
IBM:mainfrom
dharaneeshvrd:cancel-digitize-job
Open

feat(digitize): add cancel endpoint and cooperative pipeline cancellation#1300
dharaneeshvrd wants to merge 1 commit into
IBM:mainfrom
dharaneeshvrd:cancel-digitize-job

Conversation

@dharaneeshvrd

@dharaneeshvrd dharaneeshvrd commented Aug 21, 2026

Copy link
Copy Markdown
Member

Introduces a new POST /{job_id}/cancel endpoint and the supporting pipeline machinery required to cleanly stop an in-flight digitize or ingest job at the next safe checkpoint.

--- API & status model ---

  • New JobStatus values: CANCEL_PENDING (set immediately by the endpoint) and CANCELLED (set by the background task after draining).
  • New DocStatus value: CANCELLED for non-terminal documents at job cancellation time.
  • DB check constraints and init_schema.sql updated to accept the new status strings.
  • POST /{job_id}/cancel (202 Accepted):
    • Returns 404 if the job does not exist.
    • Returns 409 if the job is already in a terminal/pending-cancel state.
    • Persists CANCEL_PENDING + optional clean_files flag in job stats.
  • _run_digitize / _run_ingest catch JobCancelledError and mark all non-terminal documents and the job itself as CANCELLED. Ingest additionally removes already-indexed VDB chunks when clean_files=true.

--- Pipeline checkpoints ---

  • New digitize/exceptions.py: JobCancelledError — a clean signal that propagates up the call stack without being treated as an error.
  • db_manager.is_job_cancelled(): lightweight single-column SELECT used at every pipeline checkpoint without loading the full job row.
  • ingest.py: pre-pipeline CHECK 1 (before process_documents) and post-pipeline CHECK 5 (before writing final status).
  • digitize.py: pre-flight check before the ProcessPoolExecutor starts.
  • orchestrator.process_documents: all four as_completed loops replaced with cancellation-aware while-pending polling loops:
    • Conversion stage: pending futures cancelled; running ones drained; cancel_event (multiprocessing.Manager proxy) propagated to worker processes so long multi-chunk conversions abort between page chunks.
    • Processing stage: pending futures cancelled; running ones drained; _process_stop_event (threading.Event) propagated to process_converted_document / process_table so in-flight LLM table calls are aborted after the current call completes.
    • Chunking stage: same pattern.
    • Indexing stage: pending futures cancelled; already-running index futures are let to complete to avoid leaving stale VDB entries.
    • is_cancelled flag is latched True across stages; if all pending work has been cancelled, JobCancelledError is raised immediately to skip remaining stages.

--- LLM / processing layer ---

  • summarize_and_classify_tables accepts stop_event; checked after each individual table LLM call; queued futures cancelled on signal.
  • process_table / process_converted_document: stop_event threaded through so table summarization can be cut short mid-document.
  • convert_doc / convert_document accept cancel_event; checked between 100-page chunks so a multi-chunk conversion can be aborted early without any DB access from the worker process.

--- Correctness fixes ---

  • DatabaseStatusManager.update_job_progress: protected-status guard prevents any IN_PROGRESS update from overwriting cancel_pending, cancelled, completed, or failed.
  • Extra job-stat keys (e.g. clean_files) are preserved during stats merge so the flag is readable by the background cleanup code.
  • completed_at is now stamped for CANCELLED jobs as well as COMPLETED/FAILED ones.
  • recover_zombie_jobs: CANCEL_PENDING added to orphan statuses so interrupted cancel operations are recovered on restart.

--- Tests ---

  • services/digitize/exceptions.py: new module.
  • services/digitize/tests/test_cancel_job.py: 1 344-line test suite covering the cancel endpoint (404/409/202/500), _run_digitize and _run_ingest cancellation paths (including VDB cleanup), and all four orchestrator pipeline stages with cancellation scenarios.
  • services/digitize/tests/test_types_models.py: extended for the two new JobStatus and one new DocStatus values.

…tion

Introduces a new POST /{job_id}/cancel endpoint and the supporting
pipeline machinery required to cleanly stop an in-flight digitize or
ingest job at the next safe checkpoint.

--- API & status model ---
* New JobStatus values: CANCEL_PENDING (set immediately by the endpoint)
  and CANCELLED (set by the background task after draining).
* New DocStatus value: CANCELLED for non-terminal documents at job
  cancellation time.
* DB check constraints and init_schema.sql updated to accept the new
  status strings.
* POST /{job_id}/cancel (202 Accepted):
  - Returns 404 if the job does not exist.
  - Returns 409 if the job is already in a terminal/pending-cancel state.
  - Persists CANCEL_PENDING + optional clean_files flag in job stats.
* _run_digitize / _run_ingest catch JobCancelledError and mark
  all non-terminal documents and the job itself as CANCELLED.
  Ingest additionally removes already-indexed VDB chunks when
  clean_files=true.

--- Pipeline checkpoints ---
* New digitize/exceptions.py: JobCancelledError — a clean signal
  that propagates up the call stack without being treated as an error.
* db_manager.is_job_cancelled(): lightweight single-column SELECT
  used at every pipeline checkpoint without loading the full job row.
* ingest.py: pre-pipeline CHECK 1 (before process_documents) and
  post-pipeline CHECK 5 (before writing final status).
* digitize.py: pre-flight check before the ProcessPoolExecutor starts.
* orchestrator.process_documents: all four as_completed loops replaced
  with cancellation-aware while-pending polling loops:
  - Conversion stage: pending futures cancelled; running ones drained;
    cancel_event (multiprocessing.Manager proxy) propagated to worker
    processes so long multi-chunk conversions abort between page chunks.
  - Processing stage: pending futures cancelled; running ones drained;
    _process_stop_event (threading.Event) propagated to
    process_converted_document / process_table so in-flight LLM table
    calls are aborted after the current call completes.
  - Chunking stage: same pattern.
  - Indexing stage: pending futures cancelled; already-running index
    futures are let to complete to avoid leaving stale VDB entries.
  - is_cancelled flag is latched True across stages; if all pending
    work has been cancelled, JobCancelledError is raised immediately
    to skip remaining stages.

--- LLM / processing layer ---
* summarize_and_classify_tables accepts stop_event; checked after each
  individual table LLM call; queued futures cancelled on signal.
* process_table / process_converted_document: stop_event threaded
  through so table summarization can be cut short mid-document.
* convert_doc / convert_document accept cancel_event; checked between
  100-page chunks so a multi-chunk conversion can be aborted early
  without any DB access from the worker process.

--- Correctness fixes ---
* DatabaseStatusManager.update_job_progress: protected-status guard
  prevents any IN_PROGRESS update from overwriting cancel_pending,
  cancelled, completed, or failed.
* Extra job-stat keys (e.g. clean_files) are preserved during stats
  merge so the flag is readable by the background cleanup code.
* completed_at is now stamped for CANCELLED jobs as well as
  COMPLETED/FAILED ones.
* recover_zombie_jobs: CANCEL_PENDING added to orphan statuses so
  interrupted cancel operations are recovered on restart.

--- Tests ---
* services/digitize/exceptions.py: new module.
* services/digitize/tests/test_cancel_job.py: 1 344-line test suite
  covering the cancel endpoint (404/409/202/500), _run_digitize and
  _run_ingest cancellation paths (including VDB cleanup), and all four
  orchestrator pipeline stages with cancellation scenarios.
* services/digitize/tests/test_types_models.py: extended for the two
  new JobStatus and one new DocStatus values.

Signed-off-by: Dharaneesharan Ravichandran <dharaneeshwaran.ravichandran@ibm.com>
@dharaneeshvrd dharaneeshvrd changed the title feat(digitize): add cancel endpoint and cooperative pipeline cancella… feat(digitize): add cancel endpoint and cooperative pipeline cancellation Aug 21, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant