Refactor: extract QuerySetStream, shared by _StreamedDocuments and _DocumentViewerStream

docs/migration-v3.md

@@ -341,10 +341,7 @@ citing that these instructions have been present in over 99.7% of CPUs since 200
 
 NumPy is a dependency of the document classifier (via scikit-learn), so any CPU that
 predates SSE4.2 support will crash with `SIGILL` (illegal instruction) when the classifier
-is loaded or trained, regardless of whether AI features are enabled. NumPy is also pulled in
-transitively by `ocrmypdf`'s default page-finalization step (via the `fpdf2` package), which
-runs during normal OCR/archive generation - so affected hardware can crash during ordinary
-document consumption as well, not only during classifier training.
+is loaded or trained, regardless of whether AI features are enabled.
 
 This differs from NumPy's optional SIMD dispatch (e.g. AVX2, AVX512), which is detected and
 selected safely at runtime - the `x86-64-v2` requirement above is a hard floor baked into the
@@ -368,64 +365,28 @@ If this prints nothing, your CPU is affected.
 
 The Celery worker (and potentially the web server) repeatedly crashes and restarts with a
 `SIGILL` error, typically visible in `dmesg`/`journalctl` as a `trap invalid opcode` inside
-`_multiarray_umath...so`. This can surface either as an intermittent crash when the
-classifier's periodic training task runs (hourly, by default), or as a `RuntimeError: NumPy
-was built with baseline optimizations... but your machine doesn't support` raised from
-`documents.parsers.ParseError` during OCR of a document being consumed, since `ocrmypdf`
-imports NumPy indirectly through `fpdf2` while finalizing pages.
+`_multiarray_umath...so`. Because the classifier is trained on a periodic schedule
+(hourly, by default), affected instances see intermittent, hard-to-reproduce document
+consumption failures whenever that scheduled task runs and takes down the worker process
+mid-task.
 
-### Current mitigation: NumPy is pinned below 2.4
+### Action Required (for affected hardware only)
 
-Because so much affected hardware is still in active use, Paperless-ngx pins NumPy to
-`<2.4` (currently resolving to the 2.3.x series) rather than requiring affected users to
-work around the problem themselves. This is a **temporary accommodation, not a permanent
-fix** - we don't control NumPy's wheel baseline, and this pin holds back a transitive
-dependency indefinitely.
-
-**Check whether your hardware will keep being supported.** Use the SSE4.2 check under
-[Affected hardware](#affected-hardware) above. If your CPU has SSE4.2, this issue never
-applies to you, now or later - the pin doesn't change anything for you. If your CPU lacks
-SSE4.2, the pin is the only reason Paperless-ngx works for you today, and that protection is
-not indefinite - see below.
-
-**This pin has an expiration date.** NumPy 2.3.x is only
-[supported upstream until 2027-06-08](https://endoflife.date/numpy) under the
-[SPEC 0](https://scientific-python.org/specs/spec-0000/) policy (roughly 24 months from its
-June 2025 release). After that date, 2.3.x stops receiving security fixes, and we will need
-to either drop the pin (reintroducing this crash on affected hardware) or find another way
-forward. If your CPU lacks SSE4.2, plan around that date rather than assuming indefinite
-support - watch the
-[paperless-ngx release notes](https://github.com/paperless-ngx/paperless-ngx/releases) as
-it approaches, since we'll announce there if/when the pin is lifted.
-
-If you build your own image or otherwise manage dependencies independently of our lockfile,
-make sure your own NumPy pin matches (`numpy<2.4`) - nothing stops your build tooling from
-picking up 2.4+ on its own.
-
-### If you're stuck on NumPy 2.4+ (custom builds, or after the pin is eventually dropped)
-
-There is no way to make NumPy 2.4+ itself work on affected CPUs on the official images - it
-requires an unofficial build with `cpu-baseline=none`, which is not something we can ship.
-
-Setting
+There is no way to make the classifier itself work on such CPUs - it requires an unofficial
+NumPy build with `cpu-baseline=none`, which is not something we can ship. The practical
+path forward is to stop the classifier from ever loading or training, which avoids
+importing NumPy at all:
 
 ```bash
 PAPERLESS_TRAIN_TASK_CRON=disable
 ```
 
-disables the periodic classifier training task (see
-[`PAPERLESS_TRAIN_TASK_CRON`](configuration.md#PAPERLESS_TRAIN_TASK_CRON)) and stops the
-classifier itself from ever loading or training. Automatic matching based on the classifier
-(suggested correspondents, document types, tags, and storage paths from trained rules) will
-no longer be available, but rule-based matching is unaffected.
-
-**This setting alone does not fully resolve the issue**, because `ocrmypdf` also imports
-NumPy indirectly (via `fpdf2`) as part of its normal page-finalization step, independent of
-the classifier. On affected hardware, OCR of documents can therefore still crash the worker
-even with the classifier disabled. Advanced users may be able to work around this by building
-a custom image that installs an alternate NumPy build compiled without the `x86-64-v2`
-baseline (e.g. from source with `-Dcpu-baseline=none`), but this is unsupported and not
-something the project can provide guidance for.
+This disables the periodic classifier training task (see
+[`PAPERLESS_TRAIN_TASK_CRON`](configuration.md#PAPERLESS_TRAIN_TASK_CRON)). Automatic
+matching based on the classifier (suggested correspondents, document types, tags, and
+storage paths from trained rules) will no longer be available, but rule-based matching is
+unaffected, and document consumption itself will no longer be at risk of crashing the
+worker.
 
 ## Database Migrations
 

pyproject.toml

@@ -149,12 +149,6 @@ typing = [
 
 [tool.uv]
 required-version = ">=0.9.0"
-# TEMPORARY: numpy>=2.4 raises the manylinux x86_64 wheel's CPU baseline to
-# x86-64-v2 (SSE4.2+), which crashes on older/embedded CPUs. Pinning below
-# 2.4 keeps that hardware working. numpy 2.3.x is supported upstream until
-# 2027-06-08 (https://endoflife.date/numpy); revisit this pin before then.
-# See docs/migration-v3.md#minimum-cpu-requirements-numpy-baseline
-constraint-dependencies = [ "numpy<2.4" ]
 environments = [
   "sys_platform == 'darwin'",
   "sys_platform == 'linux'",

src/documents/search/_backend.py

@@ -39,6 +39,7 @@ from documents.search._tokenizer import ascii_fold
 from documents.search._tokenizer import autocomplete_tokens
 from documents.search._tokenizer import register_tokenizers
 from documents.utils import IterWrapper
+from documents.utils import QuerySetStream
 from documents.utils import identity
 
 if TYPE_CHECKING:
@@ -1035,14 +1036,15 @@ _EMPTY_VIEWER_GRANT: Final[ViewerGrant] = ViewerGrant(
 )
 
 
-class _DocumentViewerStream:
+class _DocumentViewerStream(QuerySetStream["Document"]):
     """Yield document permission data while batch-loading grants.
 
     Viewer permissions are fetched in batches (see
     ``_bulk_get_viewer_permissions``), but documents are yielded individually so a
     progress bar wrapped around this stream advances per document rather than
-    jumping a whole chunk at a time. ``__len__`` lets the progress helper still
-    discover the total (it inspects ``QuerySet``/``Sized``).
+    jumping a whole chunk at a time. ``__len__`` (inherited from
+    ``QuerySetStream``) lets the progress helper still discover the total (it
+    inspects ``QuerySet``/``Sized``).
 
     The viewer and group ids travel with each document in the yielded pair
     rather than through a separate mutable attribute, so the pairing survives
@@ -1051,18 +1053,11 @@ class _DocumentViewerStream:
     generator in lock-step.
     """
 
-    def __init__(self, documents: QuerySet[Document], *, chunk_size: int) -> None:
-        self._documents = documents
-        self._chunk_size = chunk_size
-
-    def __len__(self) -> int:
-        return self._documents.count()
-
     def __iter__(self) -> Iterator[tuple[Document, ViewerGrant]]:
         # iterator(chunk_size=…) streams from a server-side cursor instead of
         # materialising the whole queryset in memory; since Django 4.1 it still
         # honours prefetch_related, running the prefetches one batch at a time.
-        documents = self._documents.iterator(chunk_size=self._chunk_size)
+        documents = self._queryset.iterator(chunk_size=self._chunk_size)
         for chunk in chunked(documents, self._chunk_size):
             grants_by_pk = _bulk_get_viewer_permissions([doc.pk for doc in chunk])
             for doc in chunk:

src/documents/tests/test_utils.py

@@ -0,0 +1,35 @@
+import pytest_mock
+
+from documents.utils import QuerySetStream
+
+
+class TestQuerySetStream:
+    def test_len_and_iter_delegate_to_streaming_queryset_methods(
+        self,
+        mocker: pytest_mock.MockerFixture,
+    ) -> None:
+        """
+        GIVEN:
+            - A mock queryset
+        WHEN:
+            - A QuerySetStream wrapping it is measured and iterated
+        THEN:
+            - len() uses count() (not a materializing len()), and iteration
+              uses .iterator(chunk_size=...) (not plain iteration, which
+              would materialize the whole queryset, plus any prefetch
+              caches, into Django's own result cache at once)
+        """
+        mock_queryset = mocker.MagicMock()
+        mock_queryset.count.return_value = 42
+        mock_queryset.iterator.return_value = iter(["row-1", "row-2"])
+        streamed = QuerySetStream(mock_queryset, chunk_size=1000)
+
+        assert len(streamed) == 42
+        assert list(streamed) == ["row-1", "row-2"]
+        # count.call_count isn't asserted exactly: list()'s own size-hint
+        # optimization calls len(streamed) again internally, on top of the
+        # explicit len() call above -- both legitimately delegate to
+        # count(), so only the delegation itself (not the call count) is
+        # the thing being verified here.
+        mock_queryset.count.assert_called_with()
+        mock_queryset.iterator.assert_called_once_with(chunk_size=1000)

src/documents/utils.py

@@ -3,16 +3,24 @@ import logging
 import shutil
 from collections.abc import Callable
 from collections.abc import Iterable
+from collections.abc import Iterator
 from os import utime
 from pathlib import Path
 from subprocess import CompletedProcess
 from subprocess import run
+from typing import TYPE_CHECKING
+from typing import Generic
 from typing import TypeVar
 
 from django.conf import settings
 from PIL import Image
 
+if TYPE_CHECKING:
+    from django.db.models import Model
+    from django.db.models import QuerySet
+
 _T = TypeVar("_T")
+_M = TypeVar("_M", bound="Model")
 
 # A function that wraps an iterable — typically used to inject a progress bar.
 IterWrapper = Callable[[Iterable[_T]], Iterable[_T]]
@@ -23,6 +31,40 @@ def identity(iterable: Iterable[_T]) -> Iterable[_T]:
     return iterable
 
 
+class QuerySetStream(Generic[_M]):
+    """Stream a QuerySet via .iterator(chunk_size=...) instead of
+    materializing it (plus any prefetch caches) all at once, while still
+    supporting len() via count() so a progress bar wrapped around this
+    (e.g. via IterWrapper) shows a real total instead of falling back to
+    indeterminate.
+
+    Plain QuerySet iteration (``for row in queryset:``) is not lazy: Django
+    fetches every matching row in one query and caches the fully-hydrated
+    result in the queryset's own ``_result_cache`` before yielding the
+    first item -- wrapping that in a progress bar or any other iterable
+    adapter doesn't change this, since none of them alter how the
+    underlying queryset produces items. ``.iterator(chunk_size=...)`` is
+    the specific Django API that bypasses ``_result_cache`` and streams
+    from a server-side cursor instead, discarding each chunk once consumed
+    (and, since Django 4.1, still honours ``prefetch_related``, running the
+    prefetches one batch at a time rather than for the whole queryset).
+
+    Subclass to layer additional per-batch work on top (see
+    ``documents.search._backend._DocumentViewerStream``) by overriding
+    ``__iter__`` -- ``__len__`` and the constructor are inherited for free.
+    """
+
+    def __init__(self, queryset: "QuerySet[_M]", *, chunk_size: int) -> None:
+        self._queryset = queryset
+        self._chunk_size = chunk_size
+
+    def __len__(self) -> int:
+        return self._queryset.count()
+
+    def __iter__(self) -> Iterator[_M]:
+        return iter(self._queryset.iterator(chunk_size=self._chunk_size))
+
+
 def _coerce_to_path(
     source: Path | str,
     dest: Path | str,

src/paperless_ai/indexing.py

@@ -14,6 +14,7 @@ from filelock import Timeout
 from documents.models import Document
 from documents.models import PaperlessTask
 from documents.utils import IterWrapper
+from documents.utils import QuerySetStream
 from documents.utils import identity
 from paperless.config import AIConfig
 from paperless_ai.db import db_connection_released
@@ -32,6 +33,11 @@ logger = logging.getLogger("paperless_ai.indexing")
 RAG_NUM_OUTPUT = 512
 RAG_CHUNK_OVERLAP = 200
 
+# update_llm_index(): row count per .iterator() batch when streaming
+# documents for a rebuild/update via QuerySetStream, matching
+# _DocumentViewerStream's chunk size in documents/search/_backend.py.
+_INDEX_STREAM_CHUNK_SIZE = 1000
+
 
 def queue_llm_index_update_if_needed(*, rebuild: bool, reason: str) -> bool:
     # NOTE: The check-then-enqueue sequence below is non-atomic (TOCTOU): two
@@ -385,7 +391,9 @@ def update_llm_index(
         if rebuild or not store.table_exists():
             logger.info("Rebuilding LLM index.")
             store.drop_table()
-            for document in iter_wrapper(documents):
+            for document in iter_wrapper(
+                QuerySetStream(documents, chunk_size=_INDEX_STREAM_CHUNK_SIZE),
+            ):
                 nodes = build_document_node(document, chunk_size=chunk_size)
                 _embed_nodes(nodes, embed_model)
                 store.add(nodes)
@@ -398,7 +406,9 @@ def update_llm_index(
             )
             existing = store.get_modified_times()
             changed = 0
-            for document in iter_wrapper(scoped_documents):
+            for document in iter_wrapper(
+                QuerySetStream(scoped_documents, chunk_size=_INDEX_STREAM_CHUNK_SIZE),
+            ):
                 doc_id = str(document.id)
                 if existing.get(doc_id) == document.modified.isoformat():
                     continue