diff --git a/src/documents/management/commands/document_consumer.py b/src/documents/management/commands/document_consumer.py index 7e3900842..cc6535487 100644 --- a/src/documents/management/commands/document_consumer.py +++ b/src/documents/management/commands/document_consumer.py @@ -314,7 +314,7 @@ def _consume_file( consumption_dir: Path, *, subdirs_as_tags: bool, -) -> None: +) -> bool: """ Queue a file for consumption. @@ -322,15 +322,20 @@ def _consume_file( filepath: Path to the file to consume. consumption_dir: Base consumption directory. subdirs_as_tags: Whether to create tags from subdirectory names. + + Returns: + True if the file was successfully handed to Celery, False otherwise. + Callers must not record the file as queued on failure, or the rescan + will never retry it. """ # Verify file still exists and is accessible try: if not filepath.is_file(): logger.debug(f"Not consuming {filepath}: not a file or doesn't exist") - return + return False except OSError as e: logger.warning(f"Not consuming {filepath}: {e}") - return + return False # Get tags from path if configured tag_ids: list[int] | None = None @@ -355,6 +360,9 @@ def _consume_file( ) except Exception: logger.exception(f"Error while queuing document {filepath}") + return False + + return True class Command(BaseCommand): @@ -492,12 +500,12 @@ class Command(BaseCommand): if not consumer_filter(Change.added, str(filepath)): continue - _consume_file( + if _consume_file( filepath=filepath, consumption_dir=directory, subdirs_as_tags=subdirs_as_tags, - ) - queued.add(filepath.resolve()) + ): + queued.add(filepath.resolve()) return queued @@ -651,14 +659,16 @@ class Command(BaseCommand): # Check for stable files for stable_path in tracker.get_stable_files(): - _consume_file( + # Only remember files that were actually queued, so the + # rescan does not re-queue them while the consume task + # has yet to remove them from disk, but does retry a + # failed publish instead of stranding it + if _consume_file( filepath=stable_path, consumption_dir=directory, subdirs_as_tags=subdirs_as_tags, - ) - # Remember it so the rescan does not re-queue it while - # the consume task has yet to remove it from disk - queued.add(stable_path) + ): + queued.add(stable_path) # Exit watch loop to reconfigure timeout break diff --git a/src/documents/tests/test_management_consumer.py b/src/documents/tests/test_management_consumer.py index 3efe4490e..0c2c8f585 100644 --- a/src/documents/tests/test_management_consumer.py +++ b/src/documents/tests/test_management_consumer.py @@ -445,12 +445,13 @@ class TestConsumeFile: target = consumption_dir / "document.pdf" shutil.copy(sample_pdf, target) - _consume_file( + result = _consume_file( filepath=target, consumption_dir=consumption_dir, subdirs_as_tags=False, ) + assert result is True mock_consume_file_delay.apply_async.assert_called_once() call_args = mock_consume_file_delay.apply_async.call_args consumable_doc = call_args.kwargs["kwargs"]["input_doc"] @@ -464,11 +465,12 @@ class TestConsumeFile: mock_consume_file_delay: MagicMock, ) -> None: """Test _consume_file handles nonexistent files gracefully.""" - _consume_file( + result = _consume_file( filepath=consumption_dir / "nonexistent.pdf", consumption_dir=consumption_dir, subdirs_as_tags=False, ) + assert result is False mock_consume_file_delay.apply_async.assert_not_called() def test_consume_directory( @@ -480,11 +482,12 @@ class TestConsumeFile: subdir = consumption_dir / "subdir" subdir.mkdir() - _consume_file( + result = _consume_file( filepath=subdir, consumption_dir=consumption_dir, subdirs_as_tags=False, ) + assert result is False mock_consume_file_delay.apply_async.assert_not_called() def test_consume_with_permission_error( @@ -499,13 +502,33 @@ class TestConsumeFile: shutil.copy(sample_pdf, target) mocker.patch.object(Path, "is_file", side_effect=PermissionError("denied")) - _consume_file( + result = _consume_file( filepath=target, consumption_dir=consumption_dir, subdirs_as_tags=False, ) + assert result is False mock_consume_file_delay.apply_async.assert_not_called() + def test_consume_with_apply_async_failure( + self, + consumption_dir: Path, + sample_pdf: Path, + mock_consume_file_delay: MagicMock, + ) -> None: + """Test _consume_file reports failure when apply_async raises.""" + target = consumption_dir / "document.pdf" + shutil.copy(sample_pdf, target) + + mock_consume_file_delay.apply_async.side_effect = Exception("broker down") + + result = _consume_file( + filepath=target, + consumption_dir=consumption_dir, + subdirs_as_tags=False, + ) + assert result is False + def test_consume_with_tags_error( self, consumption_dir: Path, @@ -522,11 +545,12 @@ class TestConsumeFile: side_effect=DatabaseError("Something happened"), ) - _consume_file( + result = _consume_file( filepath=target, consumption_dir=consumption_dir, subdirs_as_tags=True, ) + assert result is True mock_consume_file_delay.apply_async.assert_called_once() call_args = mock_consume_file_delay.apply_async.call_args overrides = call_args.kwargs["kwargs"]["overrides"] @@ -1249,6 +1273,52 @@ class TestProcessExistingFilesQueued: assert target.resolve() in queued +@pytest.mark.management +@pytest.mark.django_db +class TestCommandRetryAfterQueueFailure: + """ + Regression test for GH #13923. + + A file whose ``apply_async`` publish fails (e.g. broker briefly down) + must not be marked as queued, so the periodic rescan retries it once + the broker recovers, instead of stranding it until the consumer + process is restarted. + """ + + def test_watch_loop_retries_failed_publish_on_rescan( + self, + consumption_dir: Path, + sample_pdf: Path, + mock_consume_file_delay: MagicMock, + start_consumer: Callable[..., ConsumerThread], + ) -> None: + """A publish failure from the watch loop is retried by the rescan.""" + apply_async = mock_consume_file_delay.apply_async + + def fail_first_call(*args: object, **kwargs: object) -> None: + if apply_async.call_count == 1: + raise Exception("broker down") + + apply_async.side_effect = fail_first_call + + thread = start_consumer(stability_delay=0.1, rescan_interval=0.3) + + target = consumption_dir / "document.pdf" + shutil.copy(sample_pdf, target) + + deadline = monotonic() + 5.0 + while apply_async.call_count < 2 and monotonic() < deadline: + sleep(0.1) + + if thread.exception: + raise thread.exception + + assert apply_async.call_count >= 2, ( + "Expected the failed publish to be retried by the rescan, " + f"but apply_async was only called {apply_async.call_count} time(s)" + ) + + @pytest.mark.management @pytest.mark.django_db class TestCommandRescanRecovery: