diff --git a/authentik/stages/user_write/stage.py b/authentik/stages/user_write/stage.py
index 9bcbd9cb1..61f3eb275 100644
--- a/authentik/stages/user_write/stage.py
+++ b/authentik/stages/user_write/stage.py
@@ -2,6 +2,7 @@
 from django.contrib import messages
 from django.contrib.auth import update_session_auth_hash
 from django.contrib.auth.backends import ModelBackend
+from django.db import transaction
 from django.db.utils import IntegrityError
 from django.http import HttpRequest, HttpResponse
 from django.utils.translation import gettext as _
@@ -86,10 +87,11 @@ class UserWriteStageView(StageView):
             ]
             user.attributes[USER_ATTRIBUTE_SOURCES].append(connection.source.name)
         try:
-            user.save()
+            with transaction.atomic():
+                user.save()
         except IntegrityError as exc:
             LOGGER.warning("Failed to save user", exc=exc)
-            self.executor.stage_invalid()
+            return self.executor.stage_invalid()
         user_write.send(
             sender=self, request=request, user=user, data=data, created=user_created
         )
diff --git a/authentik/stages/user_write/tests.py b/authentik/stages/user_write/tests.py
index 55c217b56..b78bb2fe1 100644
--- a/authentik/stages/user_write/tests.py
+++ b/authentik/stages/user_write/tests.py
@@ -7,7 +7,13 @@ from django.test import Client, TestCase
 from django.urls import reverse
 from django.utils.encoding import force_str
 
-from authentik.core.models import User
+from authentik.core.models import (
+    USER_ATTRIBUTE_SOURCES,
+    Source,
+    User,
+    UserSourceConnection,
+)
+from authentik.core.sources.stage import PLAN_CONTEXT_SOURCES_CONNECTION
 from authentik.flows.challenge import ChallengeTypes
 from authentik.flows.markers import StageMarker
 from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding
@@ -32,6 +38,7 @@ class TestUserWriteStage(TestCase):
         )
         self.stage = UserWriteStage.objects.create(name="write")
         FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)
+        self.source = Source.objects.create(name="fake_source")
 
     def test_user_create(self):
         """Test creation of user"""
@@ -49,6 +56,9 @@ class TestUserWriteStage(TestCase):
             "email": "test@beryju.org",
             "password": password,
         }
+        plan.context[PLAN_CONTEXT_SOURCES_CONNECTION] = UserSourceConnection(
+            source=self.source
+        )
         session = self.client.session
         session[SESSION_KEY_PLAN] = plan
         session.save()
@@ -71,6 +81,9 @@ class TestUserWriteStage(TestCase):
         )
         self.assertTrue(user_qs.exists())
         self.assertTrue(user_qs.first().check_password(password))
+        self.assertEqual(
+            user_qs.first().attributes, {USER_ATTRIBUTE_SOURCES: [self.source.name]}
+        )
 
     def test_user_update(self):
         """Test update of existing user"""
@@ -147,7 +160,7 @@ class TestUserWriteStage(TestCase):
         "authentik.flows.views.to_stage_response",
         TO_STAGE_RESPONSE_MOCK,
     )
-    def test_with_blank_username(self):
+    def test_blank_username(self):
         """Test with blank username results in error"""
         plan = FlowPlan(
             flow_pk=self.flow.pk.hex, stages=[self.stage], markers=[StageMarker()]
@@ -175,3 +188,36 @@ class TestUserWriteStage(TestCase):
                 "type": ChallengeTypes.NATIVE.value,
             },
         )
+
+    @patch(
+        "authentik.flows.views.to_stage_response",
+        TO_STAGE_RESPONSE_MOCK,
+    )
+    def test_duplicate_data(self):
+        """Test with duplicate data, should trigger error"""
+        plan = FlowPlan(
+            flow_pk=self.flow.pk.hex, stages=[self.stage], markers=[StageMarker()]
+        )
+        session = self.client.session
+        plan.context[PLAN_CONTEXT_PROMPT] = {
+            "username": "akadmin",
+            "attribute_some-custom-attribute": "test",
+            "some_ignored_attribute": "bar",
+        }
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+
+        response = self.client.get(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertJSONEqual(
+            force_str(response.content),
+            {
+                "component": "ak-stage-access-denied",
+                "error_message": None,
+                "title": "",
+                "type": ChallengeTypes.NATIVE.value,
+            },
+        )