feat(digitize): add cancel endpoint and cooperative pipeline cancellation - #1300
Open
dharaneeshvrd wants to merge 1 commit into
Open
feat(digitize): add cancel endpoint and cooperative pipeline cancellation#1300dharaneeshvrd wants to merge 1 commit into
dharaneeshvrd wants to merge 1 commit into
Conversation
…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>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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 ---
--- Pipeline checkpoints ---
--- LLM / processing layer ---
--- Correctness fixes ---
--- Tests ---