danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix(downloads): stop phantom-completing stuck post_processing tasks

Browse source 
A task stuck in 'post_processing' past the cutoff was force-marked 'completed'
("assume it worked"). In a large batch, post-processing (AcoustID + quality +
import) is serialized and backs up, so tasks sit in post_processing while merely
QUEUED — then got falsely completed, showing as downloaded with no file on disk
(/Transfer empty).

Now: the cutoff is 30 min (was 5) so legit backlog isn't cut off, and when it
does fire the task is only completed if it actually produced a file
(final_file_path exists on disk) — otherwise marked failed (honest + retryable).
Applied at both stuck-detection sites (check_batch_completion + _v2).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
dev 2026-06-14 23:25:33 +02:00
parent 446465a833
commit 9af31c3706
2 changed files with 64 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
core/downloads/lifecycle.py
View file

@ -26,6 +26,7 @@ Lifted verbatim from web_server.py. Dependencies injected via
from __future__ import annotations
import logging
import os
import shutil
import time
import traceback
@ -44,6 +45,27 @@ from core.runtime_state import (
logger = logging.getLogger(__name__)
# A task that has been in 'post_processing' longer than this is treated as stuck.
# Post-processing (AcoustID + quality + import) is serialized, so a large batch
# legitimately backs up — keep this generous so genuinely-slow imports aren't
# cut off mid-flight (the old 5-min cutoff falsely "completed" queued tasks).
_POST_PROCESSING_STUCK_TIMEOUT = 1800 # 30 minutes
def _resolve_stuck_post_processing_status(task: dict) -> str:
"""Decide the terminal status for a task stuck in post_processing.
Only call it 'completed' if the import actually produced a file on disk
(``final_file_path`` is set at the end of successful post-processing). Without
a real file, force-completing is a lie — the task shows as a downloaded track
that isn't anywhere. Mark those 'failed' so they're retryable and honest.
"""
final_path = task.get('final_file_path')
if final_path and os.path.exists(final_path):
return 'completed'
return 'failed'
def _safe_batch_dirname(batch_id: str) -> str:
safe = ''.join(ch if ch.isalnum() or ch in ('-', '_') else '_' for ch in str(batch_id or 'batch'))
return safe or 'batch'
@ -435,9 +457,15 @@ def on_download_completed(batch_id: str, task_id: str, success: bool, deps: Life
retrying_count += 1
elif task_status == 'post_processing':
task_age = current_time - task.get('status_change_time', current_time)
if task_age > 300: # 5 minutes (post-processing should be fast)
logger.info(f"⏰ [Stuck Detection] Task {queue_task_id} stuck in post_processing for {task_age:.0f}s - forcing completion")
task['status'] = 'completed' # Assume it worked if file verification is taking too long
if task_age > _POST_PROCESSING_STUCK_TIMEOUT:
new_status = _resolve_stuck_post_processing_status(task)
if new_status == 'completed':
logger.info(f"⏰ [Stuck Detection] Task {queue_task_id} stuck in post_processing for {task_age:.0f}s but file exists — completing")
task['status'] = 'completed'
else:
logger.warning(f"⏰ [Stuck Detection] Task {queue_task_id} stuck in post_processing for {task_age:.0f}s with no output file — marking failed")
task['status'] = 'failed'
task['error_message'] = 'Post-processing timed out without producing a file'
finished_count += 1
else:
retrying_count += 1
@ -667,9 +695,15 @@ def check_batch_completion_v2(batch_id: str, deps: LifecycleDeps) -> Optional[bo
retrying_count += 1
elif task_status == 'post_processing':
task_age = current_time - task.get('status_change_time', current_time)
if task_age > 300: # 5 minutes (post-processing should be fast)
logger.info(f"⏰ [Stuck Detection V2] Task {task_id} stuck in post_processing for {task_age:.0f}s - forcing completion")
task['status'] = 'completed' # Assume it worked if file verification is taking too long
if task_age > _POST_PROCESSING_STUCK_TIMEOUT:
new_status = _resolve_stuck_post_processing_status(task)
if new_status == 'completed':
logger.info(f"⏰ [Stuck Detection V2] Task {task_id} stuck in post_processing for {task_age:.0f}s but file exists — completing")
task['status'] = 'completed'
else:
logger.warning(f"⏰ [Stuck Detection V2] Task {task_id} stuck in post_processing for {task_age:.0f}s with no output file — marking failed")
task['status'] = 'failed'
task['error_message'] = 'Post-processing timed out without producing a file'
finished_count += 1
else:
retrying_count += 1

26
tests/downloads/test_downloads_lifecycle.py
View file

@ -480,11 +480,33 @@ def test_stuck_searching_task_forced_to_not_found():
assert download_tasks['t1']['status'] == 'not_found'
def test_stuck_post_processing_task_forced_to_completed():
"""Task post_processing > 5min gets forced to completed."""
def test_stuck_post_processing_without_file_forced_to_failed():
"""Task stuck in post_processing past the timeout with NO output file must
be marked FAILED, not falsely completed — otherwise it shows as a phantom
download with nothing on disk (big batches back up post-processing)."""
download_tasks['t1'] = {
'status': 'post_processing', 'track_info': {'name': 'X'},
'status_change_time': 0, # ancient → past any timeout
}
download_batches['b1'] = {
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
'max_concurrent': 1, 'permanently_failed_tracks': [],
'cancelled_tracks': set(),
}
deps, _ = _build_deps()
lc.on_download_completed('b1', 't1', True, deps)
assert download_tasks['t1']['status'] == 'failed'
def test_stuck_post_processing_with_existing_file_completed(tmp_path):
"""If the import really finished (final_file_path exists on disk), a stuck
post_processing task is legitimately completed."""
real_file = tmp_path / 'track.flac'
real_file.write_bytes(b'x')
download_tasks['t1'] = {
'status': 'post_processing', 'track_info': {'name': 'X'},
'status_change_time': 0,
'final_file_path': str(real_file),
}
download_batches['b1'] = {
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,