danvics/docling-studio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix: preserve batch progress on analysis completion

Browse source 
Re-read the job from DB before mark_completed so that
progress_current/progress_total written during batched conversion
are not overwritten by the stale in-memory object.

Add regression unit test and e2e assertion on final progress values.
This commit is contained in:
Pier-Jean Malandrino 2026-04-09 15:51:36 +02:00
parent 0e5e088438
commit d810fda333
3 changed files with 92 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
document-parser/services/analysis_service.py
View file

@ -327,6 +327,9 @@ class AnalysisService:
chunks_json = json.dumps([_chunk_to_dict(c) for c in chunks]) chunks_json = json.dumps([_chunk_to_dict(c) for c in chunks])
logger.info("Chunking produced %d chunks for job %s", len(chunks), job_id) logger.info("Chunking produced %d chunks for job %s", len(chunks), job_id)
# Re-read the job so we don't lose progress_current/progress_total
# written to the DB during batched conversion.
job = await analysis_repo.find_by_id(job_id) or job
job.mark_completed( job.mark_completed(
markdown=result.content_markdown, markdown=result.content_markdown,
html=result.content_html, html=result.content_html,

84
document-parser/tests/test_analysis_service.py
View file

@ -385,6 +385,90 @@ class TestBatchedConversion:
batch_size=5, batch_size=5,
) )
@pytest.mark.asyncio
async def test_progress_preserved_through_full_analysis_flow(self):
"""Progress written during batches must survive the final update_status.
Regression: _run_analysis_inner used to re-read the job from DB at the
start, then call update_status(job) at the end — overwriting
progress_current/progress_total with None because the in-memory object
was stale. The fix re-reads the job before mark_completed.
"""
from domain.models import AnalysisJob, AnalysisStatus
converter = AsyncMock()
converter.convert.side_effect = [
ConversionResult(
page_count=5,
content_markdown="# B1",
content_html="<html><body><p>B1</p></body></html>",
pages=[PageDetail(page_number=i, width=612, height=792) for i in range(1, 6)],
),
ConversionResult(
page_count=3,
content_markdown="# B2",
content_html="<html><body><p>B2</p></body></html>",
pages=[PageDetail(page_number=i, width=612, height=792) for i in range(6, 9)],
),
]
service = AnalysisService(converter=converter, conversion_timeout=60)
# Simulated DB state: find_by_id is called 4 times:
# 1) start of _run_analysis_inner (initial load)
# 2) batch 1 mid-flight deletion check
# 3) batch 2 mid-flight deletion check
# 4) re-read before mark_completed (must carry progress)
initial_job = AnalysisJob(
id="job-progress",
document_id="doc-1",
status=AnalysisStatus.PENDING,
)
batch_check_job = AnalysisJob(
id="job-progress",
document_id="doc-1",
status=AnalysisStatus.RUNNING,
)
refreshed_job = AnalysisJob(
id="job-progress",
document_id="doc-1",
status=AnalysisStatus.RUNNING,
progress_current=8,
progress_total=8,
)
saved_jobs: list[AnalysisJob] = []
async def capture_update_status(job):
saved_jobs.append(job)
with (
patch("services.analysis_service.analysis_repo") as mock_repo,
patch("services.analysis_service.document_repo") as mock_doc_repo,
patch("services.analysis_service._count_pdf_pages", return_value=8),
patch("services.analysis_service.settings") as mock_settings,
):
mock_settings.batch_page_size = 5
mock_settings.default_table_mode = "accurate"
mock_repo.find_by_id = AsyncMock(
side_effect=[initial_job, batch_check_job, batch_check_job, refreshed_job]
)
mock_repo.update_status = AsyncMock(side_effect=capture_update_status)
mock_repo.update_progress = AsyncMock()
mock_doc_repo.update_page_count = AsyncMock()
await service._run_analysis_inner("job-progress", "/fake.pdf", "fake.pdf")
# The last update_status call is the COMPLETED one
completed_job = saved_jobs[-1]
assert completed_job.status == AnalysisStatus.COMPLETED
assert completed_job.progress_current == 8, (
"progress_current must be preserved (was the bug: got None)"
)
assert completed_job.progress_total == 8, (
"progress_total must be preserved (was the bug: got None)"
)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_job_deleted_mid_batch_returns_none(self): async def test_job_deleted_mid_batch_returns_none(self):
"""If the job is deleted between batches, conversion aborts with None.""" """If the job is deleted between batches, conversion aborts with None."""

5
e2e/api/src/test/resources/analyses/batch-progress.feature
View file

@ -27,6 +27,11 @@ Feature: Batched conversion with progress reporting
And match response.contentMarkdown == '#string' And match response.contentMarkdown == '#string'
And match response.pagesJson == '#string' And match response.pagesJson == '#string'
# When batched, progress must be preserved at completion (not reset to null)
# progressTotal > 0 proves batching was active
# progressCurrent == progressTotal proves the final update_status didn't erase them
* if (response.progressTotal > 0) karate.match('response.progressCurrent', response.progressTotal)
# Cleanup # Cleanup
* call read('classpath:common/helpers/cleanup.feature') { docId: '#(uploaded.docId)' } * call read('classpath:common/helpers/cleanup.feature') { docId: '#(uploaded.docId)' }