fix(redirector): improve error handling; switch to 302 redirects

site/cds_rdm/legacy/errors.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2024 CERN.
+# Copyright (C) 2024-2026 CERN.
 #
 # CDS-RDM is free software; you can redistribute it and/or modify it under
 # the terms of the GPL-2.0 License; see LICENSE file for more details.
@@ -15,3 +15,12 @@ def __init__(self, version, latest_record):
         """Initialise error."""
         self.version = version
         self.latest_record = latest_record
+
+
+class FileNotFound(Exception):
+    """The requested file was not found in any version of the record."""
+
+    def __init__(self, file_key, record) -> None:
+        """Constructor."""
+        self.file_key = file_key
+        self.record = record

site/cds_rdm/legacy/redirector.py

@@ -30,14 +30,14 @@
 from invenio_records_resources.services.errors import PermissionDeniedError
 from sqlalchemy.orm.exc import NoResultFound
 
-from .errors import VersionNotFound
+from .errors import FileNotFound, VersionNotFound
 from .resolver import (
     get_pid_by_legacy_recid,
     get_record_by_version,
     get_record_versions,
 )
 
-HTTP_MOVED_PERMANENTLY = 301
+HTTP_FOUND = 302
 
 
 def version_not_found_error(error):
@@ -52,13 +52,25 @@ def version_not_found_error(error):
     )
 
 
+def file_not_found_error(error: FileNotFound):
+    """Handler for the requested file not being found in any record version."""
+    return (
+        render_template(
+            "cds_rdm/file_not_found.html",
+            file_key=error.file_key,
+            record=error.record,
+        ),
+        404,
+    )
+
+
 def legacy_record_redirect(legacy_id):
     """Redirect legacy recid."""
     pid = get_pid_by_legacy_recid(legacy_id)
     url_path = invenio_url_for(
         "invenio_app_rdm_records.record_detail", pid_value=pid.pid_value
     )
-    return redirect(url_path, HTTP_MOVED_PERMANENTLY)
+    return redirect(url_path, HTTP_FOUND)
 
 
 def legacy_files_redirect(legacy_id, filename):
@@ -72,7 +84,7 @@ def legacy_files_redirect(legacy_id, filename):
         record = get_record_by_version(parent_pid.pid_value, version)
         # Directly download files from redirected link to replicate the `allfiles-` behaviour from legacy
         if filename.startswith("allfiles-"):
-            return redirect(record["links"]["archive"], HTTP_MOVED_PERMANENTLY)
+            return redirect(record["links"]["archive"], HTTP_FOUND)
 
         # If no version is provided, trickle down the versions and find the newest version that contains the file
         if version is None:
@@ -89,6 +101,10 @@ def legacy_files_redirect(legacy_id, filename):
                     break
         else:
             record_pid_value = record["id"]
+
+        if record_pid_value is None:
+            # No explicit version was requested, and none of the record versions contained the requested file.
+            raise FileNotFound(file_key=filename, record=record)
     except PermissionDeniedError:
         if not current_user.is_authenticated:
             # trigger the flask-login unauthorized handler
@@ -114,7 +130,7 @@ def legacy_files_redirect(legacy_id, filename):
             filename=filename,
             **query_params,
         )
-    return redirect(url_path, HTTP_MOVED_PERMANENTLY)
+    return redirect(url_path, HTTP_FOUND)
 
 
 def legacy_comments_redirect(legacy_id):
@@ -131,14 +147,14 @@ def legacy_comments_redirect(legacy_id):
             url_for(
                 "invenio_app_rdm_records.record_detail", pid_value=parent_pid.pid_value
             ),
-            HTTP_MOVED_PERMANENTLY,
+            HTTP_FOUND,
         )
     return redirect(
         url_for(
             "invenio_app_rdm_requests.user_dashboard_request_view",
             request_pid_value=community_relation.request_id,
         ),
-        HTTP_MOVED_PERMANENTLY,
+        HTTP_FOUND,
     )
 
 
@@ -236,6 +252,7 @@ def create_blueprint(app):
     blueprint.register_error_handler(NoResultFound, not_found_error)
     blueprint.register_error_handler(VersionNotFound, version_not_found_error)
     blueprint.register_error_handler(PIDDeletedError, record_tombstone_error)
+    blueprint.register_error_handler(FileNotFound, file_not_found_error)
 
     # Add URL rules
     return blueprint

site/cds_rdm/templates/semantic-ui/cds_rdm/file_not_found.html

@@ -0,0 +1,23 @@
+{#
+    Copyright (C) 2026 CERN.
+
+    CDS-RDM is free software; you can redistribute it and/or modify it
+    under the terms of the GPL-2.0 License; see LICENSE file for more details.
+#}
+
+
+{% extends config.THEME_404_TEMPLATE %}
+
+{% block message %}
+  <h1><i class="bolt icon"></i> {{_('File not found')}}</h1>
+  <p>
+    {{
+      _('We could not find the file "%(file_key)s" in any version of the record.', file_key=file_key)
+    }}
+  </p>
+  <p>
+    <a href="{{ record["links"]["self_html"] }}">
+      {{ _("View the latest version of the record") }}
+    </a>
+  </p>
+{% endblock message %}

site/tests/conftest.py

@@ -1430,6 +1430,18 @@ def minimal_record_with_files():
     }
 
 
+@pytest.fixture(params=["public", "restricted"])
+def minimal_record_with_public_and_restricted_files(request, minimal_record_with_files):
+    """Same as minimal_record_with_files, but the files are marked as public/restricted."""
+    return {
+        **minimal_record_with_files,
+        "access": {
+            **minimal_record_with_files["access"],
+            "files": request.param,
+        },
+    }
+
+
 @pytest.fixture(scope="function")
 def add_pid(db):
     """Fixture to add a row to the pidstore_pid table."""

site/tests/legacy/test_redirector.py

@@ -8,6 +8,7 @@
 
 from io import BytesIO
 
+from invenio_access.permissions import system_identity
 from invenio_pidstore.models import PersistentIdentifier
 from invenio_rdm_records.proxies import current_rdm_records
 from invenio_rdm_records.records.api import RDMRecord
@@ -26,13 +27,15 @@ def add_file_to_draft(draft_file_service, uploader, draft, file_id):
 
 
 def test_legacy_record_redirection(
-    uploader, client, minimal_record_with_files, add_pid
+    uploader, client, minimal_record_with_public_and_restricted_files, add_pid
 ):
     """Test legacy redirection mechanism."""
     client = uploader.login(client)
     service = current_rdm_records.records_service
     # create draft
-    draft = service.create(uploader.identity, minimal_record_with_files)
+    draft = service.create(
+        uploader.identity, minimal_record_with_public_and_restricted_files
+    )
     add_file_to_draft(service.draft_files, uploader, draft, "test.pdf")
     # publish record with file
     record = service.publish(uploader.identity, draft.id)
@@ -53,7 +56,7 @@ def test_legacy_record_redirection(
 
     # Test record redirection
     response = client.get("/legacy/record/123456")
-    assert response.status_code == 301
+    assert response.status_code == 302
     # Resolves to parent, so get response from /records/parent_pid
     # The parent always redirects with a 302
     response = client.get(response.location)
@@ -62,7 +65,7 @@ def test_legacy_record_redirection(
 
     query_params = "?test=check&foo=bar"
     response = client.get("/legacy/record/123456" + query_params)
-    assert response.status_code == 301
+    assert response.status_code == 302
     # Resolves to parent, so get response from /records/parent_pid
     # The parent always redirects with a 302
     response = client.get(response.location)
@@ -76,31 +79,33 @@ def test_legacy_record_redirection(
     # Test files redirection
     file_route = "/preview/test.pdf"
     response = client.get("/legacy/record/123456/files/test.pdf")
-    assert response.status_code == 301
+    assert response.status_code == 302
     assert response.location == rdm_record_url + file_route
 
     response = client.get("/legacy/record/123456/files/")
-    assert response.status_code == 301
+    assert response.status_code == 302
     # Resolves to parent, so get response from /records/parent_pid
     response = client.get(response.location)
     assert response.status_code == 302
     assert response.location == rdm_record_url
 
     response = client.get("/legacy/record/123456/files/test.pdf" + query_params)
-    assert response.status_code == 301
+    assert response.status_code == 302
     assert response.location == rdm_record_url + file_route + query_params
 
     # Add new version of record
     draft_v2 = service.new_version(uploader.identity, draft.id)
-    service.update_draft(uploader.identity, draft_v2.id, minimal_record_with_files)
+    service.update_draft(
+        uploader.identity, draft_v2.id, minimal_record_with_public_and_restricted_files
+    )
     add_file_to_draft(service.draft_files, uploader, draft_v2, "test_v2.pdf")
     record_v2 = service.publish(uploader.identity, draft_v2.id)
     RDMRecord.index.refresh()
     rdm_record_v2_url = "/records/" + record_v2.id
 
     # Always redirect to latest version
     response = client.get("/legacy/record/123456")
-    assert response.status_code == 301
+    assert response.status_code == 302
     # Resolves to parent, so get response from /records/parent_pid
     # The parent always redirects with a 302
     response = client.get(response.location)
@@ -110,16 +115,16 @@ def test_legacy_record_redirection(
     # Test files redirection without version
     file_route_v2 = "/preview/test_v2.pdf"
     response = client.get("/legacy/record/123456/files/test_v2.pdf" + query_params)
-    assert response.status_code == 301
+    assert response.status_code == 302
     assert response.location == rdm_record_v2_url + file_route_v2 + query_params
 
     # Test files redirection with version
     response = client.get("/legacy/record/123456/files/test.pdf?version=1")
-    assert response.status_code == 301
+    assert response.status_code == 302
     assert response.location == rdm_record_url + file_route
 
     response = client.get("/legacy/record/123456/files/test.pdf?version=2")
-    assert response.status_code == 301
+    assert response.status_code == 302
     assert response.location == rdm_record_v2_url + file_route
 
     # v3 doesn't exist, throws an error
@@ -128,9 +133,21 @@ def test_legacy_record_redirection(
 
     # files download redirection case
     response = client.get("/legacy/record/123456/files/allfiles-small" + query_params)
-    assert response.status_code == 301
+    assert response.status_code == 302
     assert response.location == record_v2.links["archive"]
 
+    # Test non-existent file on existent record
+    response = client.get("/legacy/record/123456/files/non_existent_file.pdf")
+    assert response.status_code == 404
+
+    # Soft delete the record (no tombstone for this example)
+    service.delete_record(system_identity, record.id, {})
+    RDMRecord.index.refresh()
+
+    # Test (existent) file on soft-deleted record
+    response = client.get("/legacy/record/123456/files/test.pdf")
+    assert response.status_code == 404
+
 
 # def test_legacy_collection_redirection(
 #     superuser_identity,