chore(profiling): confirm Stage 2 loop-resolution reduces query count and wall time at scale

Guardian permission subplans now actually execute (previously
'never executed' due to unowned seed documents short-circuiting the
owner_id IS NULL branch). best_seconds rose from 0.0251s to 31.4648s,
confirming the varchar-cast nested-loop-semi-join pathology is real
and now measured under realistic conditions.

profiling/harness.py

@@ -0,0 +1,123 @@
+# profiling/harness.py
+from __future__ import annotations
+
+import json
+import subprocess
+import time
+from dataclasses import dataclass
+from datetime import UTC
+from datetime import datetime
+from pathlib import Path
+from typing import TYPE_CHECKING
+from typing import Any
+from typing import Generic
+from typing import TypeVar
+
+import pytest
+from django.db import connection
+from django.test.utils import CaptureQueriesContext
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from django.db.models import QuerySet
+
+    from profiling.seed import ScaleProfile
+
+T = TypeVar("T")
+
+
+def require_postgres() -> None:
+    if connection.vendor != "postgresql":
+        pytest.skip(
+            "Profiling requires PostgreSQL to reflect production query "
+            "planning; SQLite does not reproduce the varchar-cast planner "
+            "pathology this work fixes.",
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ProfileResult(Generic[T]):
+    best_seconds: float
+    all_seconds: tuple[float, ...]
+    query_count: int
+    result: T
+
+
+def run_profile(fn: Callable[[], T], *, repeat: int = 5) -> ProfileResult[T]:
+    require_postgres()
+    all_seconds: list[float] = []
+    result: T | None = None
+    query_count = 0
+    for i in range(repeat):
+        with CaptureQueriesContext(connection) as ctx:
+            start = time.perf_counter()
+            result = fn()
+            all_seconds.append(time.perf_counter() - start)
+        if i == repeat - 1:
+            query_count = len(ctx.captured_queries)
+    assert result is not None  # repeat >= 1 guarantees at least one assignment
+    return ProfileResult(
+        best_seconds=min(all_seconds),
+        all_seconds=tuple(all_seconds),
+        query_count=query_count,
+        result=result,
+    )
+
+
+def capture_explain_analyze(queryset: QuerySet[Any]) -> str:
+    if connection.vendor != "postgresql":
+        raise RuntimeError("EXPLAIN ANALYZE capture is Postgres-only")
+    sql, params = queryset.query.sql_with_params()
+    with connection.cursor() as cur:
+        cur.execute(f"EXPLAIN ANALYZE {sql}", params)
+        return "\n".join(row[0] for row in cur.fetchall())
+
+
+def _current_git_ref() -> str:
+    result = subprocess.run(
+        ["git", "rev-parse", "--short", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    return result.stdout.strip() or "unknown"
+
+
+def append_profiling_history(
+    *,
+    stage: str,
+    scenario: str,
+    best_seconds: float,
+    query_count: int,
+    scale: ScaleProfile,
+    code_ref: str | None = None,
+) -> None:
+    """
+    Append one line to profiling/results/history.jsonl -- an append-only,
+    cross-session record of every profiling run. Unlike the per-stage
+    before/after snapshot JSON files (which the next run of the same stage
+    overwrites), this never gets clobbered, so a profiling effort picked
+    back up days later has a full timeline to resume from instead of only
+    the single most recent comparison.
+
+    ``code_ref`` should identify the *production* code under test (a short
+    SHA on the feature branch, not this `profiling` branch) -- pass it
+    explicitly when known. Left as ``None``, it falls back to this repo's
+    current ``git rev-parse --short HEAD``, which is only meaningful if
+    you're actually running from a checkout where that reflects the code
+    being measured.
+    """
+    results_dir = Path(__file__).parent / "results"
+    results_dir.mkdir(exist_ok=True)
+    entry = {
+        "timestamp": datetime.now(UTC).isoformat(),
+        "code_ref": code_ref or _current_git_ref(),
+        "stage": stage,
+        "scenario": scenario,
+        "scale": scale,
+        "best_seconds": best_seconds,
+        "query_count": query_count,
+    }
+    with (results_dir / "history.jsonl").open("a") as f:
+        f.write(json.dumps(entry) + "\n")

profiling/pyproject.toml

@@ -0,0 +1,18 @@
+[tool.pytest]
+ini_options.pythonpath = [ "../src", "..", "." ]
+ini_options.markers = [
+  """\
+  profiling: Profiling scripts comparing old vs new query patterns; require a throwaway PostgreSQL container, not run \
+  as part of the normal test suite\
+  """,
+]
+ini_options.DJANGO_SETTINGS_MODULE = "paperless.settings"
+
+[tool.pytest_env]
+# Standalone config: this directory must be independently runnable without
+# the rest of the dev stack (Redis, etc) present -- only Postgres is required,
+# via run_with_postgres.sh.
+PAPERLESS_SECRET_KEY = "profiling-only-insecure-key"
+PAPERLESS_DISABLE_DBHANDLER = "true"
+PAPERLESS_CACHE_BACKEND = "django.core.cache.backends.locmem.LocMemCache"
+PAPERLESS_CHANNELS_BACKEND = "channels.layers.InMemoryChannelLayer"

profiling/results/history.jsonl

@@ -0,0 +1,8 @@
+{"timestamp": "2026-07-27T22:47:46.677504+00:00", "code_ref": "test-run", "stage": "self-check", "scenario": "test_append_profiling_history_writes_a_valid_json_line", "scale": "medium", "best_seconds": 0.001, "query_count": 0}
+{"timestamp": "2026-07-27T22:49:03.762638+00:00", "code_ref": "test-run", "stage": "self-check", "scenario": "test_append_profiling_history_writes_a_valid_json_line", "scale": "medium", "best_seconds": 0.001, "query_count": 0}
+{"timestamp": "2026-07-27T23:01:49.777269+00:00", "code_ref": "0506dcad0", "stage": "stage1", "scenario": "get_objects_for_user_owner_aware_document_baseline", "scale": "medium", "best_seconds": 0.025123266968876123, "query_count": 1}
+{"timestamp": "2026-07-28T01:38:32.706519+00:00", "code_ref": "e80ede4cb", "stage": "stage1", "scenario": "get_objects_for_user_owner_aware_document_baseline", "scale": "medium", "best_seconds": 31.46481539506931, "query_count": 1}
+{"timestamp": "2026-07-28T03:53:46.770669+00:00", "code_ref": "d4fa85237", "stage": "stage1", "scenario": "get_objects_for_user_owner_aware_document_baseline", "scale": "medium", "best_seconds": 31.671881378046237, "query_count": 1}
+{"timestamp": "2026-07-28T04:02:41.942168+00:00", "code_ref": "d4fa85237", "stage": "stage1", "scenario": "permitted_document_ids_after_stage1", "scale": "medium", "best_seconds": 0.07956207206007093, "query_count": 1}
+{"timestamp": "2026-07-28T23:30:28.602501+00:00", "code_ref": "86846255f", "stage": "stage2", "scenario": "per_row_checker_vs_resolved_set_old", "scale": "medium", "best_seconds": 3.4628250940004364, "query_count": 692}
+{"timestamp": "2026-07-28T23:30:28.606987+00:00", "code_ref": "86846255f", "stage": "stage2", "scenario": "per_row_checker_vs_resolved_set_new", "scale": "medium", "best_seconds": 0.10436739504802972, "query_count": 1}

profiling/results/stage1_after.json

@@ -0,0 +1,9 @@
+{
+  "permitted_document_ids": {
+    "best_seconds": 0.07956207206007093,
+    "query_count": 1
+  },
+  "baseline_best_seconds": 31.671881378046237,
+  "baseline_query_count": 1,
+  "speedup_x": 398.07763370131113
+}

profiling/results/stage1_after_explain.txt

@@ -0,0 +1,58 @@
+Index Scan Backward using documents_document_created_bedd0818 on documents_document v0  (cost=91.31..99.33 rows=1 width=2900) (actual time=68.340..77.517 rows=12024.00 loops=1)
+  Filter: ((deleted_at IS NULL) AND ((owner_id = 12) OR (owner_id IS NULL) OR (ANY (id = (hashed SubPlan 1).col1))))
+  Rows Removed by Filter: 7976
+  Index Searches: 1
+  Buffers: shared hit=74978
+  SubPlan 1
+    ->  Unique  (cost=91.04..91.05 rows=2 width=4) (actual time=64.698..66.622 rows=9136.00 loops=1)
+          Buffers: shared hit=73487
+          ->  Sort  (cost=91.04..91.05 rows=2 width=4) (actual time=64.697..65.202 rows=9262.00 loops=1)
+                Sort Key: ((u0.object_pk)::integer)
+                Sort Method: quicksort  Memory: 385kB
+                Buffers: shared hit=73487
+                ->  Append  (cost=0.27..91.03 rows=2 width=4) (actual time=0.026..63.201 rows=9262.00 loops=1)
+                      Buffers: shared hit=73487
+                      ->  Nested Loop  (cost=0.27..12.79 rows=1 width=4) (actual time=0.026..1.834 rows=237.00 loops=1)
+                            Join Filter: (u0.permission_id = u1.id)
+                            Buffers: shared hit=696
+                            ->  Index Only Scan using guardian_userobjectpermi_user_id_permission_id_ob_b0b3d2fc_uniq on guardian_userobjectpermission u0  (cost=0.27..8.29 rows=1 width=520) (actual time=0.012..0.131 rows=237.00 loops=1)
+                                  Index Cond: (user_id = 12)
+                                  Heap Fetches: 237
+                                  Index Searches: 1
+                                  Buffers: shared hit=222
+                            ->  Seq Scan on auth_permission u1  (cost=0.00..4.49 rows=1 width=4) (actual time=0.006..0.006 rows=1.00 loops=237)
+                                  Filter: (((codename)::text = 'view_document'::text) AND (content_type_id = 17))
+                                  Rows Removed by Filter: 67
+                                  Buffers: shared hit=474
+                      ->  Nested Loop  (cost=4.73..78.23 rows=1 width=4) (actual time=1.119..60.528 rows=9025.00 loops=1)
+                            Buffers: shared hit=72791
+                            ->  Nested Loop  (cost=4.59..78.05 rows=1 width=524) (actual time=1.115..50.521 rows=9025.00 loops=1)
+                                  Buffers: shared hit=54741
+                                  ->  Nested Loop  (cost=4.44..73.62 rows=24 width=520) (actual time=1.109..13.226 rows=45387.00 loops=1)
+                                        Buffers: shared hit=329
+                                        ->  Seq Scan on auth_permission u4  (cost=0.00..4.49 rows=1 width=4) (actual time=0.008..0.017 rows=1.00 loops=1)
+                                              Filter: (((codename)::text = 'view_document'::text) AND (content_type_id = 17))
+                                              Rows Removed by Filter: 165
+                                              Buffers: shared hit=2
+                                        ->  Bitmap Heap Scan on guardian_groupobjectpermission u0_1  (cost=4.44..68.93 rows=20 width=524) (actual time=1.100..8.095 rows=45387.00 loops=1)
+                                              Recheck Cond: (permission_id = u4.id)
+                                              Heap Blocks: exact=290
+                                              Buffers: shared hit=327
+                                              ->  Bitmap Index Scan on guardian_groupobjectpermission_permission_id_36572738  (cost=0.00..4.43 rows=20 width=0) (actual time=1.052..1.053 rows=45387.00 loops=1)
+                                                    Index Cond: (permission_id = u4.id)
+                                                    Index Searches: 1
+                                                    Buffers: shared hit=37
+                                  ->  Index Only Scan using auth_user_groups_user_id_group_id_94350c0c_uniq on auth_user_groups u2  (cost=0.15..0.18 rows=1 width=4) (actual time=0.001..0.001 rows=0.20 loops=45387)
+                                        Index Cond: ((user_id = 12) AND (group_id = u0_1.group_id))
+                                        Heap Fetches: 9025
+                                        Index Searches: 45387
+                                        Buffers: shared hit=54412
+                            ->  Index Only Scan using auth_group_pkey on auth_group u1_1  (cost=0.14..0.17 rows=1 width=4) (actual time=0.001..0.001 rows=1.00 loops=9025)
+                                  Index Cond: (id = u0_1.group_id)
+                                  Heap Fetches: 9025
+                                  Index Searches: 9025
+                                  Buffers: shared hit=18050
+Planning:
+  Buffers: shared hit=2
+Planning Time: 1.089 ms
+Execution Time: 78.020 ms

profiling/results/stage1_baseline.json

@@ -0,0 +1,6 @@
+{
+  "get_objects_for_user_owner_aware_document": {
+    "best_seconds": 31.671881378046237,
+    "query_count": 1
+  }
+}

profiling/results/stage1_baseline_explain.txt

@@ -0,0 +1,68 @@
+Sort  (cost=3790.80..3790.82 rows=9 width=2900) (actual time=39120.166..39120.831 rows=12024.00 loops=1)
+  Sort Key: documents_document.created DESC
+  Sort Method: quicksort  Memory: 3230kB
+  Buffers: shared hit=8598270
+  ->  Seq Scan on documents_document  (cost=2550.72..3790.65 rows=9 width=2900) (actual time=0.009..39114.720 rows=12024.00 loops=1)
+        Filter: ((deleted_at IS NULL) AND ((owner_id = 2) OR (owner_id IS NULL) OR (ANY (id = (hashed SubPlan 1).col1)) OR (ANY (id = (hashed SubPlan 2).col1))))
+        Rows Removed by Filter: 7976
+        Buffers: shared hit=8598270
+        SubPlan 1
+          ->  Nested Loop Semi Join  (cost=0.00..1245.63 rows=1 width=8) (actual time=4.254..969.033 rows=237.00 loops=1)
+                Join Filter: ((((v0.object_pk)::bigint)::character varying)::text = (((u0.id)::bigint)::character varying)::text)
+                Rows Removed by Join Filter: 2539276
+                Buffers: shared hit=214835
+                ->  Nested Loop  (cost=0.00..23.30 rows=1 width=516) (actual time=0.031..2.425 rows=237.00 loops=1)
+                      Join Filter: (v0.permission_id = v2.id)
+                      Buffers: shared hit=490
+                      ->  Seq Scan on guardian_userobjectpermission v0  (cost=0.00..18.80 rows=1 width=520) (actual time=0.021..0.424 rows=237.00 loops=1)
+                            Filter: (user_id = 2)
+                            Rows Removed by Filter: 2120
+                            Buffers: shared hit=16
+                      ->  Seq Scan on auth_permission v2  (cost=0.00..4.49 rows=1 width=4) (actual time=0.006..0.006 rows=1.00 loops=237)
+                            Filter: ((content_type_id = 17) AND ((codename)::text = 'view_document'::text))
+                            Rows Removed by Filter: 67
+                            Buffers: shared hit=474
+                ->  Seq Scan on documents_document u0  (cost=0.00..1221.96 rows=12 width=4) (actual time=0.002..1.917 rows=10715.24 loops=237)
+                      Filter: (deleted_at IS NULL)
+                      Buffers: shared hit=214345
+        SubPlan 2
+          ->  Nested Loop Semi Join  (cost=4.73..1305.09 rows=1 width=8) (actual time=6.729..38127.912 rows=9025.00 loops=1)
+                Join Filter: ((((v0_1.object_pk)::bigint)::character varying)::text = (((u0_2.id)::bigint)::character varying)::text)
+                Rows Removed by Join Filter: 98955406
+                Buffers: shared hit=8382237
+                ->  Nested Loop Semi Join  (cost=4.73..82.77 rows=1 width=516) (actual time=1.109..137.661 rows=9025.00 loops=1)
+                      Buffers: shared hit=145515
+                      ->  Nested Loop  (cost=4.44..73.62 rows=24 width=520) (actual time=1.090..18.002 rows=45387.00 loops=1)
+                            Buffers: shared hit=329
+                            ->  Seq Scan on auth_permission v2_1  (cost=0.00..4.49 rows=1 width=4) (actual time=0.011..0.023 rows=1.00 loops=1)
+                                  Filter: (((codename)::text = 'view_document'::text) AND (content_type_id = 17))
+                                  Rows Removed by Filter: 165
+                                  Buffers: shared hit=2
+                            ->  Bitmap Heap Scan on guardian_groupobjectpermission v0_1  (cost=4.44..68.93 rows=20 width=524) (actual time=1.077..10.777 rows=45387.00 loops=1)
+                                  Recheck Cond: (permission_id = v2_1.id)
+                                  Heap Blocks: exact=290
+                                  Buffers: shared hit=327
+                                  ->  Bitmap Index Scan on guardian_groupobjectpermission_permission_id_36572738  (cost=0.00..4.43 rows=20 width=0) (actual time=1.026..1.026 rows=45387.00 loops=1)
+                                        Index Cond: (permission_id = v2_1.id)
+                                        Index Searches: 1
+                                        Buffers: shared hit=37
+                      ->  Nested Loop  (cost=0.30..0.37 rows=1 width=8) (actual time=0.002..0.002 rows=0.20 loops=45387)
+                            Join Filter: (u0_1.id = u1.group_id)
+                            Buffers: shared hit=145186
+                            ->  Index Only Scan using auth_group_pkey on auth_group u0_1  (cost=0.14..0.17 rows=1 width=4) (actual time=0.001..0.001 rows=1.00 loops=45387)
+                                  Index Cond: (id = v0_1.group_id)
+                                  Heap Fetches: 45387
+                                  Index Searches: 45387
+                                  Buffers: shared hit=90774
+                            ->  Index Only Scan using auth_user_groups_user_id_group_id_94350c0c_uniq on auth_user_groups u1  (cost=0.15..0.18 rows=1 width=4) (actual time=0.001..0.001 rows=0.20 loops=45387)
+                                  Index Cond: ((user_id = 2) AND (group_id = v0_1.group_id))
+                                  Heap Fetches: 9025
+                                  Index Searches: 45387
+                                  Buffers: shared hit=54412
+                ->  Seq Scan on documents_document u0_2  (cost=0.00..1221.96 rows=12 width=4) (actual time=0.002..1.956 rows=10965.59 loops=9025)
+                      Filter: (deleted_at IS NULL)
+                      Buffers: shared hit=8236722
+Planning:
+  Buffers: shared hit=3
+Planning Time: 1.119 ms
+Execution Time: 39121.312 ms

profiling/run_with_postgres.sh

@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+# profiling/run_with_postgres.sh
+#
+# Starts a PostgreSQL 18 container (matching
+# docker/compose/docker-compose.postgres.yml's image) if one isn't already
+# running, waits for it to accept connections, force-terminates any
+# leftover connections from a previous run that crashed or was killed, then
+# runs the given command against it with the right PAPERLESS_DB* env vars
+# set. The container is intentionally left running afterward -- see
+# stop_postgres.sh to tear it down explicitly once you're done profiling
+# for the session.
+set -euo pipefail
+
+CONTAINER_NAME="pngx-profiling-pg"
+HOST_PORT="55432"
+
+if ! docker inspect "$CONTAINER_NAME" >/dev/null 2>&1; then
+	docker run --rm -d \
+		--name "$CONTAINER_NAME" \
+		-e POSTGRES_DB=paperless \
+		-e POSTGRES_USER=paperless \
+		-e POSTGRES_PASSWORD=paperless \
+		-p "${HOST_PORT}:5432" \
+		docker.io/library/postgres:18 >/dev/null
+	echo "Started profiling Postgres container '$CONTAINER_NAME' on port $HOST_PORT (left running -- see profiling/stop_postgres.sh)." >&2
+fi
+
+until docker exec "$CONTAINER_NAME" pg_isready -U paperless >/dev/null 2>&1; do
+	sleep 1
+done
+
+# Reap any connections left over from a previous run that crashed, was
+# killed, or timed out mid-query -- without this, a persistent container
+# can accumulate stuck backends that block later test-database
+# create/drop operations indefinitely.
+docker exec "$CONTAINER_NAME" psql -U paperless -d paperless -v ON_ERROR_STOP=0 -c "
+  SELECT pg_terminate_backend(pid)
+  FROM pg_stat_activity
+  WHERE datname IN ('paperless', 'test_paperless')
+    AND pid <> pg_backend_pid();
+" >/dev/null 2>&1 || true
+
+PAPERLESS_DBHOST=localhost \
+	PAPERLESS_DBNAME=paperless \
+	PAPERLESS_DBUSER=paperless \
+	PAPERLESS_DBPASS=paperless \
+	PAPERLESS_DBPORT="${HOST_PORT}" \
+	"$@"

profiling/seed.py

@@ -0,0 +1,199 @@
+# profiling/seed.py
+from __future__ import annotations
+
+import random
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+from typing import Literal
+
+from django.contrib.auth.models import Group
+from django.contrib.auth.models import User
+from guardian.shortcuts import assign_perm
+
+from documents.models import Document
+from documents.tests.factories import CorrespondentFactory
+from documents.tests.factories import DocumentFactory
+from documents.tests.factories import DocumentTypeFactory
+from documents.tests.factories import StoragePathFactory
+from documents.tests.factories import TagFactory
+
+if TYPE_CHECKING:
+    from documents.models import Correspondent
+    from documents.models import DocumentType
+    from documents.models import StoragePath
+    from documents.models import Tag
+
+ScaleProfile = Literal["medium", "large"]
+
+# Fraction of seeded documents given a real owner; the remainder stays
+# owner=None, mirroring a real install's mix of owned and legacy/imported
+# unowned documents.
+_OWNED_FRACTION = 0.9
+
+
+class _ScaleCounts:
+    __slots__ = (
+        "correspondents",
+        "document_types",
+        "documents",
+        "groups",
+        "storage_paths",
+        "tags",
+        "users",
+    )
+
+    def __init__(
+        self,
+        *,
+        documents: int,
+        tags: int,
+        correspondents: int,
+        document_types: int,
+        storage_paths: int,
+        users: int,
+        groups: int,
+    ) -> None:
+        self.documents = documents
+        self.tags = tags
+        self.correspondents = correspondents
+        self.document_types = document_types
+        self.storage_paths = storage_paths
+        self.users = users
+        self.groups = groups
+
+
+_SCALE_PROFILES: dict[ScaleProfile, _ScaleCounts] = {
+    "medium": _ScaleCounts(
+        documents=20_000,
+        tags=100,
+        correspondents=300,
+        document_types=50,
+        storage_paths=20,
+        users=10,
+        groups=5,
+    ),
+    "large": _ScaleCounts(
+        documents=360_000,
+        tags=1_000,
+        correspondents=5_000,
+        document_types=300,
+        storage_paths=50,
+        users=25,
+        groups=10,
+    ),
+}
+# Ratios measured from a real install (discussion #13276): 1,414 user-perm
+# rows / 27,232 group-perm rows over 12,000 documents.
+_USER_PERM_ROWS_PER_DOC = 1_414 / 12_000
+_GROUP_PERM_ROWS_PER_DOC = 27_232 / 12_000
+
+
+def _sample_distinct_pairs(
+    rng: random.Random,
+    n_subjects: int,
+    n_objects: int,
+    count: int,
+) -> set[tuple[int, int]]:
+    """
+    Rejection-sample `count` distinct (subject_index, object_index) pairs.
+    `count` is always well under `n_subjects * n_objects` for every scale
+    profile, so this terminates quickly.
+    """
+    if count > n_subjects * n_objects:
+        msg = (
+            f"cannot sample {count} distinct pairs from only "
+            f"{n_subjects * n_objects} possible (subject, object) slots"
+        )
+        raise ValueError(msg)
+    pairs: set[tuple[int, int]] = set()
+    while len(pairs) < count:
+        pairs.add((rng.randrange(n_subjects), rng.randrange(n_objects)))
+    return pairs
+
+
+@dataclass(frozen=True, slots=True)
+class SeededData:
+    users: tuple[User, ...]
+    groups: tuple[Group, ...]
+    documents: tuple[Document, ...]
+    tags: tuple[Tag, ...]
+    correspondents: tuple[Correspondent, ...]
+    document_types: tuple[DocumentType, ...]
+    storage_paths: tuple[StoragePath, ...]
+
+
+def seed_permission_dataset(
+    scale: ScaleProfile = "medium",
+    *,
+    seed: int = 1337,
+) -> SeededData:
+    """
+    Build a dataset whose document count and guardian-permission-row ratios
+    mirror real bug reports, so profiling results are representative of
+    actual large installs rather than an arbitrary small fixture.
+    """
+    counts = _SCALE_PROFILES[scale]
+    rng = random.Random(seed)
+
+    users = tuple(
+        User.objects.create_user(username=f"profile_user_{i}")
+        for i in range(counts.users)
+    )
+    groups = tuple(
+        Group.objects.create(name=f"profile_group_{i}") for i in range(counts.groups)
+    )
+    for user in users:
+        user.groups.add(rng.choice(groups))
+
+    documents = list(DocumentFactory.create_batch(counts.documents))
+
+    # Assign realistic ownership -- without this, every document stays
+    # owner=None and the "owned or unowned" visibility clause is trivially
+    # true for every row, so no guardian permission check ever needs to run
+    # (see the amendment note above this code block).
+    owned_documents = []
+    for document in documents:
+        if rng.random() < _OWNED_FRACTION:
+            document.owner = rng.choice(users)
+            owned_documents.append(document)
+    Document.objects.bulk_update(owned_documents, ["owner"], batch_size=2_000)
+    documents = tuple(documents)
+
+    tags = tuple(TagFactory.create_batch(counts.tags))
+    correspondents = tuple(CorrespondentFactory.create_batch(counts.correspondents))
+    document_types = tuple(DocumentTypeFactory.create_batch(counts.document_types))
+    storage_paths = tuple(StoragePathFactory.create_batch(counts.storage_paths))
+
+    n_user_perms = round(counts.documents * _USER_PERM_ROWS_PER_DOC)
+    n_group_perms = round(counts.documents * _GROUP_PERM_ROWS_PER_DOC)
+
+    # Grant permissions only on OWNED documents -- a grant on an already-
+    # unowned document is a no-op for visibility (unowned documents are
+    # visible to everyone regardless), so restricting the pool here is what
+    # makes each assign_perm() call actually matter for someone, and is
+    # what exercises the guardian-permission-join code path this profiling
+    # effort exists to measure.
+    for user_idx, owned_idx in _sample_distinct_pairs(
+        rng,
+        len(users),
+        len(owned_documents),
+        n_user_perms,
+    ):
+        assign_perm("view_document", users[user_idx], owned_documents[owned_idx])
+    for group_idx, owned_idx in _sample_distinct_pairs(
+        rng,
+        len(groups),
+        len(owned_documents),
+        n_group_perms,
+    ):
+        assign_perm("view_document", groups[group_idx], owned_documents[owned_idx])
+
+    return SeededData(
+        users=users,
+        groups=groups,
+        documents=documents,
+        tags=tags,
+        correspondents=correspondents,
+        document_types=document_types,
+        storage_paths=storage_paths,
+    )

profiling/stop_postgres.sh

@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+# profiling/stop_postgres.sh
+#
+# Explicit, manual teardown of the profiling Postgres container. Not called
+# automatically by run_with_postgres.sh -- run this yourself once you're
+# done with a profiling session, so a seeded `large`-scale dataset can be
+# reused across several script runs first. Terminates any active backends
+# before stopping, so a stuck query can't stall `docker stop` waiting for
+# a graceful Postgres shutdown that never comes.
+set -euo pipefail
+
+CONTAINER_NAME="pngx-profiling-pg"
+
+if docker inspect "$CONTAINER_NAME" >/dev/null 2>&1; then
+	docker exec "$CONTAINER_NAME" psql -U paperless -d paperless -v ON_ERROR_STOP=0 -c "
+    SELECT pg_terminate_backend(pid)
+    FROM pg_stat_activity
+    WHERE pid <> pg_backend_pid();
+  " >/dev/null 2>&1 || true
+	docker stop "$CONTAINER_NAME" >/dev/null
+	echo "Stopped and removed profiling Postgres container."
+else
+	echo "No profiling Postgres container was running."
+fi

profiling/test_harness_self_check.py

@@ -0,0 +1,74 @@
+# profiling/test_harness_self_check.py
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from profiling.harness import append_profiling_history
+from profiling.harness import require_postgres
+from profiling.harness import run_profile
+from profiling.seed import seed_permission_dataset
+
+
+@pytest.mark.profiling
+@pytest.mark.django_db
+def test_seed_medium_scale_produces_expected_counts() -> None:
+    require_postgres()
+    data = seed_permission_dataset(scale="medium")
+    assert len(data.documents) == 20_000
+    assert len(data.tags) == 100
+    assert len(data.correspondents) == 300
+    assert len(data.document_types) == 50
+    assert len(data.storage_paths) == 20
+
+    from guardian.models import GroupObjectPermission
+    from guardian.models import UserObjectPermission
+
+    user_perm_count = UserObjectPermission.objects.count()
+    group_perm_count = GroupObjectPermission.objects.count()
+    # within 5% of the ratio-derived target -- exact counts depend on random
+    # sampling without replacement, see seed.py
+    assert 2_242 <= user_perm_count <= 2_478
+    assert 43_111 <= group_perm_count <= 47_649
+
+
+@pytest.mark.profiling
+@pytest.mark.django_db
+def test_run_profile_reports_query_count_and_timing() -> None:
+    require_postgres()
+
+    def trivial() -> list[int]:
+        from documents.models import Document
+
+        return list(Document.objects.values_list("id", flat=True)[:1])
+
+    result = run_profile(trivial, repeat=3)
+    assert result.best_seconds >= 0
+    assert len(result.all_seconds) == 3
+    assert result.query_count >= 1
+
+
+def test_append_profiling_history_writes_a_valid_json_line() -> None:
+    history_path = Path(__file__).parent / "results" / "history.jsonl"
+    before_lines = (
+        history_path.read_text().splitlines() if history_path.exists() else []
+    )
+
+    append_profiling_history(
+        stage="self-check",
+        scenario="test_append_profiling_history_writes_a_valid_json_line",
+        best_seconds=0.001,
+        query_count=0,
+        scale="medium",
+        code_ref="test-run",
+    )
+
+    after_lines = history_path.read_text().splitlines()
+    assert len(after_lines) == len(before_lines) + 1
+    entry = json.loads(after_lines[-1])
+    assert entry["stage"] == "self-check"
+    assert entry["code_ref"] == "test-run"
+    assert entry["scale"] == "medium"
+    assert "timestamp" in entry

profiling/test_stage1_document_calls.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from documents.models import Document
+from documents.permissions import get_objects_for_user_owner_aware
+from documents.permissions import permitted_document_ids
+from profiling.harness import append_profiling_history
+from profiling.harness import capture_explain_analyze
+from profiling.harness import require_postgres
+from profiling.harness import run_profile
+from profiling.seed import seed_permission_dataset
+
+RESULTS_DIR = Path(__file__).parent / "results"
+
+
+@pytest.mark.profiling
+@pytest.mark.django_db
+def test_profile_get_objects_for_user_owner_aware_document_baseline() -> None:
+    require_postgres()
+    data = seed_permission_dataset(scale="medium")
+    user = data.users[0]
+
+    def call() -> list[int]:
+        return list(
+            get_objects_for_user_owner_aware(
+                user,
+                "documents.view_document",
+                Document,
+            ).values_list("id", flat=True),
+        )
+
+    profile = run_profile(call, repeat=3)
+    plan = capture_explain_analyze(
+        get_objects_for_user_owner_aware(user, "documents.view_document", Document),
+    )
+
+    RESULTS_DIR.mkdir(exist_ok=True)
+    (RESULTS_DIR / "stage1_baseline.json").write_text(
+        json.dumps(
+            {
+                "get_objects_for_user_owner_aware_document": {
+                    "best_seconds": profile.best_seconds,
+                    "query_count": profile.query_count,
+                },
+            },
+            indent=2,
+        ),
+    )
+    (RESULTS_DIR / "stage1_baseline_explain.txt").write_text(plan)
+    append_profiling_history(
+        stage="stage1",
+        scenario="get_objects_for_user_owner_aware_document_baseline",
+        best_seconds=profile.best_seconds,
+        query_count=profile.query_count,
+        scale="medium",
+    )
+
+    print(f"\nBASELINE best={profile.best_seconds:.4f}s queries={profile.query_count}")  # noqa: T201
+    print(plan)  # noqa: T201
+
+
+@pytest.mark.profiling
+@pytest.mark.django_db
+def test_profile_permitted_document_ids_after_stage1() -> None:
+    require_postgres()
+    data = seed_permission_dataset(scale="medium")
+    user = data.users[0]
+
+    def call():
+        return list(
+            Document.objects.filter(id__in=permitted_document_ids(user)).values_list(
+                "id",
+                flat=True,
+            ),
+        )
+
+    profile = run_profile(call, repeat=3)
+    plan = capture_explain_analyze(
+        Document.objects.filter(id__in=permitted_document_ids(user)),
+    )
+
+    baseline = json.loads((RESULTS_DIR / "stage1_baseline.json").read_text())
+    baseline_seconds = baseline["get_objects_for_user_owner_aware_document"][
+        "best_seconds"
+    ]
+    baseline_queries = baseline["get_objects_for_user_owner_aware_document"][
+        "query_count"
+    ]
+
+    (RESULTS_DIR / "stage1_after.json").write_text(
+        json.dumps(
+            {
+                "permitted_document_ids": {
+                    "best_seconds": profile.best_seconds,
+                    "query_count": profile.query_count,
+                },
+                "baseline_best_seconds": baseline_seconds,
+                "baseline_query_count": baseline_queries,
+                "speedup_x": baseline_seconds / profile.best_seconds
+                if profile.best_seconds
+                else None,
+            },
+            indent=2,
+        ),
+    )
+    (RESULTS_DIR / "stage1_after_explain.txt").write_text(plan)
+    append_profiling_history(
+        stage="stage1",
+        scenario="permitted_document_ids_after_stage1",
+        best_seconds=profile.best_seconds,
+        query_count=profile.query_count,
+        scale="medium",
+    )
+
+    print(f"\nBEFORE best={baseline_seconds:.4f}s queries={baseline_queries}")  # noqa: T201
+    print(f"AFTER  best={profile.best_seconds:.4f}s queries={profile.query_count}")  # noqa: T201
+
+    # Hard gate: must not be a regression, and should be meaningfully faster
+    # at this scale (the whole point of this stage).
+    assert profile.best_seconds < baseline_seconds
+    assert profile.query_count <= baseline_queries

profiling/test_stage2_document_loops.py

@@ -0,0 +1,64 @@
+# profiling/test_stage2_document_loops.py
+from __future__ import annotations
+
+import pytest
+
+from documents.permissions import has_perms_owner_aware
+from documents.permissions import permitted_document_ids
+from profiling.harness import append_profiling_history
+from profiling.harness import require_postgres
+from profiling.harness import run_profile
+from profiling.seed import seed_permission_dataset
+
+
+@pytest.mark.profiling
+@pytest.mark.django_db
+def test_profile_per_row_checker_vs_resolved_set_for_bulk_operation() -> None:
+    require_postgres()
+    data = seed_permission_dataset(scale="medium")
+    user = data.users[0]
+    # simulate a bulk operation over 500 documents the user can actually see
+    # (bulk-edit/share/trash all operate on a user's own selection) -- built
+    # from confirmed-visible documents so all() can't short-circuit unfairly
+    # in either old_style or new_style, see amendment note above
+    permitted_ids = set(permitted_document_ids(user))
+    batch = [doc for doc in data.documents if doc.pk in permitted_ids][:500]
+    assert len(batch) == 500, (
+        f"expected at least 500 visible documents to build a fair batch, "
+        f"got {len(batch)} -- scale profile or seed ratios may need adjusting"
+    )
+
+    def old_style():
+        return all(has_perms_owner_aware(user, "view_document", doc) for doc in batch)
+
+    def new_style():
+        permitted = set(permitted_document_ids(user))
+        return all(doc.pk in permitted for doc in batch)
+
+    old_profile = run_profile(old_style, repeat=3)
+    new_profile = run_profile(new_style, repeat=3)
+
+    append_profiling_history(
+        stage="stage2",
+        scenario="per_row_checker_vs_resolved_set_old",
+        best_seconds=old_profile.best_seconds,
+        query_count=old_profile.query_count,
+        scale="medium",
+    )
+    append_profiling_history(
+        stage="stage2",
+        scenario="per_row_checker_vs_resolved_set_new",
+        best_seconds=new_profile.best_seconds,
+        query_count=new_profile.query_count,
+        scale="medium",
+    )
+
+    print(  # noqa: T201
+        f"\nOLD (per-row checker): best={old_profile.best_seconds:.4f}s queries={old_profile.query_count}",
+    )
+    print(  # noqa: T201
+        f"NEW (resolved set):    best={new_profile.best_seconds:.4f}s queries={new_profile.query_count}",
+    )
+
+    assert new_profile.query_count < old_profile.query_count
+    assert new_profile.best_seconds < old_profile.best_seconds

src/documents/permissions.py

@@ -163,23 +163,14 @@ def set_permissions_for_object(
                         )
 
 
-def permitted_document_ids(
-    user,
-    *,
-    perm: str = "view_document",
-    include_deleted: bool = False,
-):
+def permitted_document_ids(user):
     """
-    Return a queryset of document IDs the user has ``perm`` on (default
-    ``"view_document"``). By default limited to non-deleted documents; pass
-    ``include_deleted=True`` for callers that need to check permission on
-    soft-deleted documents (e.g. trash restore). This intentionally avoids
-    ``get_objects_for_user`` to keep the subquery small and index-friendly.
+    Return a queryset of document IDs the user may view, limited to non-deleted
+    documents. This intentionally avoids ``get_objects_for_user`` to keep the
+    subquery small and index-friendly.
     """
 
-    manager = Document.global_objects if include_deleted else Document.objects
-    base_docs = manager.all()
-    base_docs = base_docs.only("id", "owner")
+    base_docs = Document.objects.filter(deleted_at__isnull=True).only("id", "owner")
 
     if user is None or not getattr(user, "is_authenticated", False):
         # Just Anonymous user e.g. for drf-spectacular
@@ -190,7 +181,7 @@ def permitted_document_ids(
 
     document_ct = ContentType.objects.get_for_model(Document)
     perm_filter = {
-        "permission__codename": perm,
+        "permission__codename": "view_document",
         "permission__content_type": document_ct,
     }
 

src/documents/serialisers.py

@@ -39,6 +39,7 @@ from drf_spectacular.utils import extend_schema_field
 from drf_spectacular.utils import extend_schema_serializer
 from drf_writable_nested.serializers import NestedUpdateMixin
 from guardian.core import ObjectPermissionChecker
+from guardian.shortcuts import get_objects_for_user
 from guardian.shortcuts import get_users_with_perms
 from guardian.utils import get_group_obj_perms_model
 from guardian.utils import get_user_obj_perms_model
@@ -79,8 +80,8 @@ from documents.models import WorkflowTrigger
 from documents.parsers import is_mime_type_supported
 from documents.permissions import get_document_count_filter_for_user
 from documents.permissions import get_groups_with_only_permission
+from documents.permissions import get_objects_for_user_owner_aware
 from documents.permissions import has_perms_owner_aware
-from documents.permissions import permitted_document_ids
 from documents.permissions import set_permissions_for_object
 from documents.regex import validate_regex_pattern
 from documents.templating.filepath import validate_filepath_template_and_render
@@ -864,8 +865,11 @@ def validate_documentlink_targets(user, doc_ids):
     if user is None:
         return
 
-    permitted_change_ids = set(permitted_document_ids(user, perm="change_document"))
-    if not set(doc_ids) <= permitted_change_ids:
+    target_documents = Document.objects.filter(id__in=doc_ids).select_related("owner")
+    if not all(
+        has_perms_owner_aware(user, "change_document", document)
+        for document in target_documents
+    ):
         raise PermissionDenied(
             _("Insufficient permissions."),
         )
@@ -1007,8 +1011,13 @@ def _get_viewable_duplicates(
     ).exclude(pk=document.pk)
     duplicates = duplicates.filter(root_document__isnull=True)
     duplicates = duplicates.order_by("-created")
-    allowed_ids = permitted_document_ids(user, include_deleted=True)
-    return duplicates.filter(id__in=allowed_ids)
+    allowed = get_objects_for_user_owner_aware(
+        user,
+        "documents.view_document",
+        Document,
+        include_deleted=True,
+    )
+    return duplicates.filter(id__in=allowed)
 
 
 class DuplicateDocumentSummarySerializer(serializers.Serializer[dict[str, Any]]):
@@ -2650,8 +2659,13 @@ class TaskSerializerV9(serializers.ModelSerializer[PaperlessTask]):
         user = request.user
         qs = Document.global_objects.filter(pk=dup_of)
         if not user.is_staff:
-            allowed_ids = permitted_document_ids(user, include_deleted=True)
-            qs = qs.filter(pk__in=allowed_ids)
+            with_perms = get_objects_for_user(
+                user,
+                "documents.view_document",
+                qs,
+                accept_global_perms=False,
+            )
+            qs = with_perms | qs.filter(owner=user) | qs.filter(owner__isnull=True)
         return list(qs.values("id", "title", "deleted_at"))
 
 
@@ -3501,6 +3515,8 @@ class StoragePathTestSerializer(SerializerWithPerms):
             document_field = self.fields.get("document")
             if not isinstance(document_field, serializers.PrimaryKeyRelatedField):
                 return
-            document_field.queryset = Document.objects.filter(
-                id__in=permitted_document_ids(user),
+            document_field.queryset = get_objects_for_user_owner_aware(
+                user,
+                "documents.view_document",
+                Document,
             )

src/documents/tests/test_permission_filtering_security.py

@@ -1,392 +0,0 @@
-from __future__ import annotations
-
-from http import HTTPStatus
-from unittest.mock import patch
-
-import pytest
-from django.contrib.auth.models import AnonymousUser
-from django.contrib.auth.models import Group
-from django.contrib.auth.models import Permission
-from django.contrib.auth.models import User
-from django.test import override_settings
-from guardian.shortcuts import assign_perm
-from rest_framework.test import APIClient
-
-from documents.permissions import permitted_document_ids
-from documents.serialisers import _get_viewable_duplicates
-from documents.tests.factories import DocumentFactory
-
-
-def assert_visible_document_ids(actual_ids, *, expected_visible, expected_hidden):
-    actual_ids = set(actual_ids)
-    for doc_id in expected_visible:
-        assert doc_id in actual_ids, (
-            f"document {doc_id} should be visible but was hidden"
-        )
-    for doc_id in expected_hidden:
-        assert doc_id not in actual_ids, (
-            f"document {doc_id} leaked but should be hidden"
-        )
-
-
-@pytest.mark.django_db
-class TestPermittedDocumentIdsSecurity:
-    def test_owner_sees_own_document(self):
-        user = User.objects.create_user(username="alice")
-        stranger = User.objects.create_user(username="mallory")
-        owned = DocumentFactory(owner=user)
-        strangers_doc = DocumentFactory(owner=stranger)
-
-        visible = permitted_document_ids(user)
-
-        assert_visible_document_ids(
-            visible,
-            expected_visible=[owned.pk],
-            expected_hidden=[strangers_doc.pk],
-        )
-
-    def test_unowned_document_visible_to_everyone(self):
-        user = User.objects.create_user(username="alice")
-        unowned = DocumentFactory(owner=None)
-
-        assert_visible_document_ids(
-            permitted_document_ids(user),
-            expected_visible=[unowned.pk],
-            expected_hidden=[],
-        )
-
-    def test_explicit_user_permission_grants_visibility(self):
-        grantee = User.objects.create_user(username="alice")
-        stranger = User.objects.create_user(username="mallory")
-        owner = User.objects.create_user(username="owner")
-        shared = DocumentFactory(owner=owner)
-        not_shared = DocumentFactory(owner=owner)
-        assign_perm("view_document", grantee, shared)
-
-        assert_visible_document_ids(
-            permitted_document_ids(grantee),
-            expected_visible=[shared.pk],
-            expected_hidden=[not_shared.pk],
-        )
-        assert_visible_document_ids(
-            permitted_document_ids(stranger),
-            expected_visible=[],
-            expected_hidden=[shared.pk, not_shared.pk],
-        )
-
-    def test_explicit_group_permission_grants_visibility_to_members_only(self):
-        owner = User.objects.create_user(username="owner")
-        member = User.objects.create_user(username="member")
-        non_member = User.objects.create_user(username="non_member")
-        group = Group.objects.create(name="finance")
-        member.groups.add(group)
-        shared = DocumentFactory(owner=owner)
-        assign_perm("view_document", group, shared)
-
-        assert_visible_document_ids(
-            permitted_document_ids(member),
-            expected_visible=[shared.pk],
-            expected_hidden=[],
-        )
-        assert_visible_document_ids(
-            permitted_document_ids(non_member),
-            expected_visible=[],
-            expected_hidden=[shared.pk],
-        )
-
-    def test_soft_deleted_document_excluded_by_default(self):
-        owner = User.objects.create_user(username="owner")
-        doc = DocumentFactory(owner=owner)
-        doc.delete()  # soft delete
-
-        assert_visible_document_ids(
-            permitted_document_ids(owner),
-            expected_visible=[],
-            expected_hidden=[doc.pk],
-        )
-
-    def test_superuser_sees_everything_including_no_perm_documents(self):
-        superuser = User.objects.create_superuser(username="root")
-        owner = User.objects.create_user(username="owner")
-        doc = DocumentFactory(owner=owner)
-
-        assert_visible_document_ids(
-            permitted_document_ids(superuser),
-            expected_visible=[doc.pk],
-            expected_hidden=[],
-        )
-
-    def test_anonymous_user_sees_only_unowned_documents(self):
-        owner = User.objects.create_user(username="owner")
-        owned = DocumentFactory(owner=owner)
-        unowned = DocumentFactory(owner=None)
-
-        assert_visible_document_ids(
-            permitted_document_ids(AnonymousUser()),
-            expected_visible=[unowned.pk],
-            expected_hidden=[owned.pk],
-        )
-
-
-@pytest.mark.django_db
-class TestPermittedDocumentIdsIncludeDeleted:
-    def test_include_deleted_true_reveals_soft_deleted_owned_document(self):
-        owner = User.objects.create_user(username="owner")
-        doc = DocumentFactory(owner=owner)
-        doc.delete()
-
-        assert_visible_document_ids(
-            permitted_document_ids(owner, include_deleted=True),
-            expected_visible=[doc.pk],
-            expected_hidden=[],
-        )
-
-    def test_include_deleted_true_still_respects_permission_boundary(self):
-        owner = User.objects.create_user(username="owner")
-        stranger = User.objects.create_user(username="mallory")
-        doc = DocumentFactory(owner=owner)
-        doc.delete()
-
-        assert_visible_document_ids(
-            permitted_document_ids(stranger, include_deleted=True),
-            expected_visible=[],
-            expected_hidden=[doc.pk],
-        )
-
-
-@pytest.mark.django_db
-class TestAiChatAllDocumentsPermissionBoundary:
-    """
-    Regression test pinning the "ask across all documents" AI chat behavior
-    (ChatStreamingView.post, no document_id) to the same owner/permission
-    boundary enforced by permitted_document_ids(). This call site was
-    migrated from get_objects_for_user_owner_aware() to
-    permitted_document_ids() in Task 5; this test must stay green across
-    that swap.
-    """
-
-    ENDPOINT = "/api/documents/chat/"
-
-    @override_settings(AI_ENABLED=True)
-    @patch("documents.views.stream_chat_with_documents")
-    def test_chat_all_documents_excludes_unshared_document(self, mock_stream_chat):
-        mock_stream_chat.return_value = iter([b"data"])
-
-        owner = User.objects.create_user(username="owner")
-        asker = User.objects.create_user(username="asker")
-        asker.user_permissions.add(
-            *Permission.objects.filter(codename="view_document"),
-        )
-        shared = DocumentFactory(owner=owner)
-        not_shared = DocumentFactory(owner=owner)
-        assign_perm("view_document", asker, shared)
-
-        client = APIClient()
-        client.force_authenticate(user=asker)
-        response = client.post(
-            self.ENDPOINT,
-            data={"q": "question"},
-            format="json",
-        )
-
-        assert response.status_code == HTTPStatus.OK
-        mock_stream_chat.assert_called_once()
-        _, kwargs = mock_stream_chat.call_args
-        visible_ids = {doc.pk for doc in kwargs["documents"]}
-        assert shared.pk in visible_ids
-        assert not_shared.pk not in visible_ids
-
-
-@pytest.mark.django_db
-class TestDuplicateDocumentsPermissionBoundary:
-    def test_get_viewable_duplicates_includes_soft_deleted_but_respects_perms(self):
-        owner = User.objects.create_user(username="owner")
-        stranger = User.objects.create_user(username="mallory")
-        original = DocumentFactory(owner=owner, checksum="dupe-checksum")
-        dup_visible = DocumentFactory(owner=owner, checksum="dupe-checksum")
-        dup_hidden = DocumentFactory(owner=owner, checksum="dupe-checksum")
-        dup_hidden.delete()  # soft delete, should still be found (include_deleted=True)
-        assign_perm("view_document", stranger, dup_visible)
-
-        result_owner = _get_viewable_duplicates(original, owner)
-        assert {d.pk for d in result_owner} == {dup_visible.pk, dup_hidden.pk}
-
-        result_stranger = _get_viewable_duplicates(original, stranger)
-        assert {d.pk for d in result_stranger} == {dup_visible.pk}
-
-
-@pytest.mark.django_db
-class TestPermittedDocumentIdsArbitraryPermission:
-    def test_change_document_permission_is_distinct_from_view(self):
-        owner = User.objects.create_user(username="owner")
-        viewer_only = User.objects.create_user(username="viewer")
-        editor = User.objects.create_user(username="editor")
-        doc = DocumentFactory(owner=owner)
-        assign_perm("view_document", viewer_only, doc)
-        assign_perm("change_document", editor, doc)
-        assign_perm("view_document", editor, doc)
-
-        assert_visible_document_ids(
-            permitted_document_ids(editor, perm="change_document"),
-            expected_visible=[doc.pk],
-            expected_hidden=[],
-        )
-        assert_visible_document_ids(
-            permitted_document_ids(viewer_only, perm="change_document"),
-            expected_visible=[],
-            expected_hidden=[doc.pk],
-        )
-
-    def test_delete_permission_with_include_deleted_for_trash_restore(self):
-        owner = User.objects.create_user(username="owner")
-        stranger = User.objects.create_user(username="mallory")
-        doc = DocumentFactory(owner=owner)
-        doc.delete()
-
-        assert_visible_document_ids(
-            permitted_document_ids(owner, perm="delete_document", include_deleted=True),
-            expected_visible=[doc.pk],
-            expected_hidden=[],
-        )
-        assert_visible_document_ids(
-            permitted_document_ids(
-                stranger,
-                perm="delete_document",
-                include_deleted=True,
-            ),
-            expected_visible=[],
-            expected_hidden=[doc.pk],
-        )
-
-
-@pytest.mark.django_db
-class TestEmailDocumentPermissionBoundary:
-    def test_email_action_rejects_document_without_view_permission(
-        self,
-        rest_api_client,
-    ):
-        owner = User.objects.create_user(username="owner")
-        requester = User.objects.create_user(username="requester")
-        rest_api_client.force_authenticate(user=requester)
-        hidden = DocumentFactory(owner=owner)
-
-        response = rest_api_client.post(
-            "/api/documents/email/",
-            {
-                "documents": [hidden.pk],
-                "addresses": "someone@example.com",
-                "subject": "test",
-                "message": "test",
-            },
-            format="json",
-        )
-        assert response.status_code == HTTPStatus.FORBIDDEN
-
-
-@pytest.mark.django_db
-class TestBulkEditChangePermissionBoundary:
-    def test_bulk_edit_rejects_document_without_change_permission(
-        self,
-        rest_api_client,
-    ):
-        owner = User.objects.create_user(username="owner")
-        requester = User.objects.create_user(username="requester")
-        # grant the global change_document permission so the object-level
-        # check (not the global has_perm check) is what's under test
-        requester.user_permissions.add(
-            Permission.objects.get(codename="change_document"),
-        )
-        rest_api_client.force_authenticate(user=requester)
-        assign_perm(
-            "view_document",
-            requester,
-            DocumentFactory(owner=owner),
-        )  # unrelated grant
-        target = DocumentFactory(owner=owner)
-        assign_perm("view_document", requester, target)  # view only, NOT change
-
-        response = rest_api_client.post(
-            "/api/documents/bulk_edit/",
-            {
-                "documents": [target.pk],
-                "method": "modify_tags",
-                "parameters": {"add_tags": [], "remove_tags": []},
-            },
-            format="json",
-        )
-        assert response.status_code == HTTPStatus.FORBIDDEN
-
-
-@pytest.mark.django_db
-class TestBulkDownloadPermissionChecksRootDocument:
-    def test_permission_checked_on_root_not_on_version(
-        self,
-        rest_api_client,
-        paperless_dirs,
-        _media_settings,
-    ):
-        owner = User.objects.create_user(username="owner")
-        requester = User.objects.create_user(username="requester")
-        rest_api_client.force_authenticate(user=requester)
-        root = DocumentFactory(owner=owner)
-        # a version of root that the requester has NOT been individually granted
-        version = DocumentFactory(owner=owner, root_document=root, version_index=1)
-        version.source_path.write_bytes(b"%PDF-1.4 test")
-        assign_perm("view_document", requester, root)  # granted on ROOT only
-
-        response = rest_api_client.post(
-            "/api/documents/bulk_download/",
-            {"documents": [version.pk]},
-            format="json",
-        )
-        assert (
-            response.status_code == HTTPStatus.OK
-        )  # visible because root is permitted
-
-        stranger = User.objects.create_user(username="mallory")
-        rest_api_client.force_authenticate(user=stranger)
-        response = rest_api_client.post(
-            "/api/documents/bulk_download/",
-            {"documents": [version.pk]},
-            format="json",
-        )
-        assert response.status_code == HTTPStatus.FORBIDDEN
-
-
-@pytest.mark.django_db
-class TestTrashRestorePermissionBoundary:
-    def test_restore_rejects_document_without_delete_permission(
-        self,
-        rest_api_client,
-    ):
-        owner = User.objects.create_user(username="owner")
-        requester = User.objects.create_user(username="requester")
-        rest_api_client.force_authenticate(user=requester)
-        doc = DocumentFactory(owner=owner)
-        assign_perm("view_document", requester, doc)  # view only, NOT delete
-        doc.delete()
-
-        response = rest_api_client.post(
-            "/api/trash/",
-            {"documents": [doc.pk], "action": "restore"},
-            format="json",
-        )
-        assert response.status_code == HTTPStatus.FORBIDDEN
-
-    def test_restore_allows_document_with_explicit_delete_permission(
-        self,
-        rest_api_client,
-    ):
-        owner = User.objects.create_user(username="owner")
-        requester = User.objects.create_user(username="requester")
-        rest_api_client.force_authenticate(user=requester)
-        doc = DocumentFactory(owner=owner)
-        assign_perm("delete_document", requester, doc)
-        doc.delete()
-
-        response = rest_api_client.post(
-            "/api/trash/",
-            {"documents": [doc.pk], "action": "restore"},
-            format="json",
-        )
-        assert response.status_code == HTTPStatus.OK

src/documents/tests/test_share_link_bundles.py

@@ -60,7 +60,7 @@ class ShareLinkBundleAPITests(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
         self.assertIn("document_ids", response.data)
 
-    @mock.patch("documents.views.permitted_document_ids", return_value=set())
+    @mock.patch("documents.views.has_perms_owner_aware", return_value=False)
     def test_create_bundle_rejects_insufficient_permissions(self, perms_mock) -> None:
         payload = {
             "document_ids": [self.document.pk],

src/documents/tests/test_views.py

@@ -612,11 +612,11 @@ class TestAIChatStreamingView(DirectoriesMixin, TestCase):
         self.assertIn(b"AI is required for this feature", response.content)
 
     @patch("documents.views.stream_chat_with_documents")
-    @patch("documents.views.permitted_document_ids")
+    @patch("documents.views.get_objects_for_user_owner_aware")
     @override_settings(AI_ENABLED=True)
-    def test_post_no_document_id(self, mock_permitted_ids, mock_stream_chat) -> None:
+    def test_post_no_document_id(self, mock_get_objects, mock_stream_chat) -> None:
         self.grant_view_document_permission()
-        mock_permitted_ids.return_value = [self.document.pk]
+        mock_get_objects.return_value = [self.document]
         mock_stream_chat.return_value = iter([b"data"])
         response = self.client.post(
             self.ENDPOINT,
@@ -625,23 +625,23 @@ class TestAIChatStreamingView(DirectoriesMixin, TestCase):
         )
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response["Content-Type"], "text/event-stream")
-        mock_stream_chat.assert_called_once()
-        call_kwargs = mock_stream_chat.call_args.kwargs
-        self.assertEqual(call_kwargs["query_str"], "question")
-        self.assertEqual(list(call_kwargs["documents"]), [self.document])
-        self.assertIsNone(call_kwargs["output_language"])
+        mock_stream_chat.assert_called_once_with(
+            query_str="question",
+            documents=[self.document],
+            output_language=None,
+        )
 
     @patch("documents.views.stream_chat_with_documents")
-    @patch("documents.views.permitted_document_ids")
+    @patch("documents.views.get_objects_for_user_owner_aware")
     @override_settings(AI_ENABLED=True)
     def test_post_uses_user_display_language(
         self,
-        mock_permitted_ids,
+        mock_get_objects,
         mock_stream_chat,
     ) -> None:
         UiSettings.objects.create(user=self.user, settings={"language": "de-de"})
         self.grant_view_document_permission()
-        mock_permitted_ids.return_value = [self.document.pk]
+        mock_get_objects.return_value = [self.document]
         mock_stream_chat.return_value = iter([b"data"])
 
         response = self.client.post(
@@ -651,11 +651,11 @@ class TestAIChatStreamingView(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(response.status_code, 200)
-        mock_stream_chat.assert_called_once()
-        call_kwargs = mock_stream_chat.call_args.kwargs
-        self.assertEqual(call_kwargs["query_str"], "question")
-        self.assertEqual(list(call_kwargs["documents"]), [self.document])
-        self.assertEqual(call_kwargs["output_language"], "de-de")
+        mock_stream_chat.assert_called_once_with(
+            query_str="question",
+            documents=[self.document],
+            output_language="de-de",
+        )
 
     @patch("documents.views.stream_chat_with_documents")
     @override_settings(AI_ENABLED=True)

src/documents/views.py

@@ -177,7 +177,6 @@ from documents.permissions import get_objects_for_user_owner_aware
 from documents.permissions import has_global_statistics_permission
 from documents.permissions import has_perms_owner_aware
 from documents.permissions import has_system_status_permission
-from documents.permissions import permitted_document_ids
 from documents.permissions import set_permissions_for_object
 from documents.plugins.date_parsing import get_date_parser
 from documents.schema import generate_object_with_permissions_schema
@@ -1925,9 +1924,12 @@ class DocumentViewSet(
         use_archive_version = validated_data.get("use_archive_version", True)
 
         documents = Document.objects.select_related("owner").filter(pk__in=document_ids)
-        if request.user is not None:
-            permitted_ids = set(permitted_document_ids(request.user))
-            if not all(document.pk in permitted_ids for document in documents):
+        for document in documents:
+            if request.user is not None and not has_perms_owner_aware(
+                request.user,
+                "view_document",
+                document,
+            ):
                 return HttpResponseForbidden("Insufficient permissions")
 
         try:
@@ -2263,8 +2265,10 @@ class ChatStreamingView(GenericAPIView[Any]):
 
             documents = [document]
         else:
-            documents = Document.objects.filter(
-                id__in=permitted_document_ids(request.user),
+            documents = get_objects_for_user_owner_aware(
+                request.user,
+                "view_document",
+                Document,
             )
 
         output_language = _get_llm_output_language(ai_config=ai_config, request=request)
@@ -2732,8 +2736,10 @@ class DocumentSelectionMixin:
             for key, value in filters.items()
             if key not in _TANTIVY_SEARCH_PARAM_NAMES
         }
-        permitted_documents = Document.objects.filter(
-            id__in=permitted_document_ids(user),
+        permitted_documents = get_objects_for_user_owner_aware(
+            user,
+            permission_codename,
+            Document,
         )
         # orm-filtered docs
         filtered_documents = DocumentFilterSet(
@@ -2782,9 +2788,8 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
         )
 
         # check global and object permissions for all documents
-        permitted_change_ids = set(permitted_document_ids(user, perm="change_document"))
         has_perms = user.has_perm("documents.change_document") and all(
-            doc.pk in permitted_change_ids for doc in document_objs
+            has_perms_owner_aware(user, "change_document", doc) for doc in document_objs
         )
 
         # check ownership for methods that change original document
@@ -3342,8 +3347,10 @@ class SelectionDataView(GenericAPIView[Any]):
         serializer.is_valid(raise_exception=True)
 
         ids = serializer.validated_data.get("documents")
-        permitted_documents = Document.objects.filter(
-            id__in=permitted_document_ids(request.user),
+        permitted_documents = get_objects_for_user_owner_aware(
+            request.user,
+            "documents.view_document",
+            Document,
         )
         if permitted_documents.filter(pk__in=ids).count() != len(ids):
             return HttpResponseForbidden("Insufficient permissions")
@@ -3515,8 +3522,10 @@ class GlobalSearchView(PassUserMixin):
         OBJECT_LIMIT = 3
         docs = []
         if request.user.has_perm("documents.view_document"):
-            all_docs = Document.objects.filter(
-                id__in=permitted_document_ids(request.user),
+            all_docs = get_objects_for_user_owner_aware(
+                request.user,
+                "view_document",
+                Document,
             )
             if db_only:
                 docs = all_docs.filter(title__icontains=query)[:OBJECT_LIMIT]
@@ -3720,7 +3729,11 @@ class StatisticsView(GenericAPIView[Any]):
         documents = (
             Document.objects.all()
             if can_view_global_stats
-            else Document.objects.filter(id__in=permitted_document_ids(user))
+            else get_objects_for_user_owner_aware(
+                user,
+                "documents.view_document",
+                Document,
+            )
         ).filter(root_document__isnull=True)
         tags = (
             Tag.objects.all()
@@ -3837,10 +3850,9 @@ class BulkDownloadView(DocumentSelectionMixin, GenericAPIView[Any]):
         content = serializer.validated_data.get("content")
         follow_filename_format = serializer.validated_data.get("follow_formatting")
 
-        permitted_ids = set(permitted_document_ids(request.user))
         for document in documents:
             root_doc = get_root_document(document)
-            if root_doc.pk not in permitted_ids:
+            if not has_perms_owner_aware(request.user, "view_document", root_doc):
                 return HttpResponseForbidden("Insufficient permissions")
             versioned_documents.append(
                 get_latest_version_for_root(
@@ -4504,9 +4516,8 @@ class ShareLinkBundleViewSet(PassUserMixin, ModelViewSet[ShareLinkBundle]):
             )
 
         documents = list(documents_qs)
-        permitted_ids = set(permitted_document_ids(request.user))
         for document in documents:
-            if document.pk not in permitted_ids:
+            if not has_perms_owner_aware(request.user, "view_document", document):
                 raise ValidationError(
                     {
                         "document_ids": _(
@@ -5308,15 +5319,9 @@ class TrashView(ListModelMixin, PassUserMixin):
             if doc_ids is not None
             else self.filter_queryset(self.get_queryset()).all()
         )
-        permitted_ids = set(
-            permitted_document_ids(
-                request.user,
-                perm="delete_document",
-                include_deleted=True,
-            ),
-        )
-        if not all(doc.pk in permitted_ids for doc in docs):
-            return HttpResponseForbidden("Insufficient permissions")
+        for doc in docs:
+            if not has_perms_owner_aware(request.user, "delete_document", doc):
+                return HttpResponseForbidden("Insufficient permissions")
         action = serializer.validated_data.get("action")
         if action == "restore":
             for doc in Document.deleted_objects.filter(id__in=doc_ids).all():