diff --git a/CHANGES/+pull-through-digest-failure.bugfix b/CHANGES/+pull-through-digest-failure.bugfix
new file mode 100644
index 0000000000..180f44acfc
--- /dev/null
+++ b/CHANGES/+pull-through-digest-failure.bugfix
@@ -0,0 +1 @@
+Fixed a foreign key error when a first-time pull-through download fails digest validation; the content app now reports the intended checksum error and closes the connection.
diff --git a/pulpcore/content/handler.py b/pulpcore/content/handler.py
index 5abe50f807..2a50d9b177 100644
--- a/pulpcore/content/handler.py
+++ b/pulpcore/content/handler.py
@@ -1442,8 +1442,11 @@ async def finalize():
                 extra_data={"disable_retry_list": (DigestValidationError,)}
             )
         except DigestValidationError:
-            remote_artifact.failed_at = timezone.now()
-            await remote_artifact.asave()
+            # First-time pull-through RAs point at an unsaved ContentArtifact, so saving would
+            # violate the foreign key.
+            if not remote_artifact.content_artifact._state.adding:
+                remote_artifact.failed_at = timezone.now()
+                await remote_artifact.asave()
             close_tcp_connection(request.transport._sock)
             REMOTE_CONTENT_FETCH_FAILURE_COOLDOWN = settings.REMOTE_CONTENT_FETCH_FAILURE_COOLDOWN
             raise RuntimeError(
diff --git a/pulpcore/tests/unit/content/test_handler.py b/pulpcore/tests/unit/content/test_handler.py
index b80e8af1dd..273816cb75 100644
--- a/pulpcore/tests/unit/content/test_handler.py
+++ b/pulpcore/tests/unit/content/test_handler.py
@@ -20,6 +20,7 @@
     PathNotResolved,
     _current_distribution,
 )
+from pulpcore.exceptions import DigestValidationError
 from pulpcore.plugin.models import (
     Artifact,
     Content,
@@ -788,3 +789,78 @@ async def test_async_pull_through_add(ca1, monkeypatch, app_status):
         await repo.adelete()
         if task:
             await task.adelete()
+
+
+@pytest.fixture
+def digest_failure_downloader(monkeypatch):
+    class FailingDownloader:
+        path = None
+
+        async def handle_data(self, data):
+            pass
+
+        async def finalize(self):
+            pass
+
+        async def run(self, extra_data=None):
+            raise DigestValidationError("actual", "expected", url="https://123/c123")
+
+    monkeypatch.setattr(Remote, "get_downloader", lambda self, **kwargs: FailingDownloader())
+    close_mock = Mock()
+    monkeypatch.setattr("pulpcore.content.handler.close_tcp_connection", close_mock)
+    return close_mock
+
+
+def _failing_stream_args():
+    request = Mock(match_info={"path": "c123"}, http_range=Mock(start=None, stop=None))
+    response = web.StreamResponse()
+    response.write_eof = AsyncMock()
+    return request, response
+
+
+def _assert_not_completed_cleanly(close_mock, request, response):
+    """The connection is force-closed and the response is never finished for the client."""
+    close_mock.assert_called_once_with(request.transport._sock)
+    response.write_eof.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.django_db
+async def test_pull_through_digest_failure_new_content_artifact(digest_failure_downloader):
+    """A failed digest on a first-time pull-through raises the checksum error, not IntegrityError."""
+    remote = await create_remote()
+    ca = ContentArtifact(relative_path="c123")
+    ra = RemoteArtifact(remote=remote, url="https://123/c123", content_artifact=ca, sha256="bad")
+    request, response = _failing_stream_args()
+
+    try:
+        with pytest.raises(RuntimeError, match="failed checksum validation"):
+            await Handler()._stream_remote_artifact(request, response, ra, save_artifact=True)
+        _assert_not_completed_cleanly(digest_failure_downloader, request, response)
+        assert ra.failed_at is None
+        assert not await RemoteArtifact.objects.filter(url="https://123/c123").aexists()
+    finally:
+        await remote.adelete()
+
+
+@pytest.mark.asyncio
+@pytest.mark.django_db
+async def test_pull_through_digest_failure_existing_content_artifact(digest_failure_downloader):
+    """A failed digest on an existing ContentArtifact still records failed_at."""
+    content = await create_content()
+    ca = await create_content_artifact(content)
+    remote = await create_remote()
+    ra = await RemoteArtifact.objects.acreate(
+        remote=remote, url="https://123/c123", content_artifact=ca, sha256="bad"
+    )
+    request, response = _failing_stream_args()
+
+    try:
+        with pytest.raises(RuntimeError, match="failed checksum validation"):
+            await Handler()._stream_remote_artifact(request, response, ra, save_artifact=True)
+        _assert_not_completed_cleanly(digest_failure_downloader, request, response)
+        await ra.arefresh_from_db()
+        assert ra.failed_at is not None
+    finally:
+        await content.adelete()
+        await remote.adelete()