diff --git a/geonode/geoserver/signals.py b/geonode/geoserver/signals.py
index 0f3a84768bc..96c578db392 100644
--- a/geonode/geoserver/signals.py
+++ b/geonode/geoserver/signals.py
@@ -34,8 +34,8 @@
 from geonode.layers.models import Dataset
 from geonode.services.enumerations import CASCADED
 
-from . import BACKEND_PACKAGE
-from .tasks import geoserver_cascading_delete, geoserver_post_save_datasets
+from geonode.geoserver import BACKEND_PACKAGE
+from geonode.geoserver.tasks import geoserver_cascading_delete, geoserver_post_save_datasets
 
 logger = logging.getLogger("geonode.geoserver.signals")
 
@@ -104,14 +104,22 @@ def geoserver_pre_save_maplayer(instance, sender, **kwargs):
     # Set dataset
     if instance.dataset is None:
         dataset_queryset = Dataset.objects.filter(Q(alternate=instance.name) | Q(name=instance.name))
-        if instance.local and instance.store:
+        if instance.store:
             dataset_queryset = dataset_queryset.filter(store=instance.store)
         elif instance.ows_url:
             dataset_queryset = dataset_queryset.filter(remote_service__base_url=instance.ows_url)
         try:
             instance.dataset = dataset_queryset.get()
-        except (Dataset.DoesNotExist, Dataset.MultipleObjectsReturned):
+        except Dataset.DoesNotExist:
             pass
+        except Dataset.MultipleObjectsReturned:
+            logger.error(
+                "Ambiguous dataset lookup for map layer '%s' (store=%s, ows_url=%s)",
+                instance.name,
+                instance.store,
+                instance.ows_url,
+            )
+            raise
 
 
 @deprecated(version="3.2.1", reason="Use direct calls to the ReourceManager.")
diff --git a/geonode/layers/tests.py b/geonode/layers/tests.py
index 450192f4e36..f0d3535f365 100644
--- a/geonode/layers/tests.py
+++ b/geonode/layers/tests.py
@@ -296,7 +296,7 @@ def test_dataset_styles(self):
         except UnicodeEncodeError:
             self.fail("str of the Style model throws a UnicodeEncodeError with special characters.")
 
-    def test_dataset_links(self):
+    def test_vector_links(self):
         lyr = Dataset.objects.filter(subtype="vector").first()
         self.assertEqual(lyr.subtype, "vector")
 
@@ -324,6 +324,7 @@ def test_dataset_links(self):
             links = Link.objects.filter(resource=lyr.resourcebase_ptr, link_type="image")
             self.assertIsNotNone(links)
 
+    def test_raster_links(self):
         lyr = Dataset.objects.filter(subtype="raster").first()
         self.assertEqual(lyr.subtype, "raster")
         if check_ogc_backend(geoserver.BACKEND_PACKAGE):
@@ -350,6 +351,42 @@ def test_dataset_links(self):
             links = Link.objects.filter(resource=lyr.resourcebase_ptr, link_type="image")
             self.assertIsNotNone(links)
 
+    def test_set_resource_default_links_passes_auth_config_to_service_handler(self):
+        remote_auth_config = object()
+        instance = MagicMock()
+        instance.resourcebase_ptr = MagicMock()
+        instance.srid = "EPSG:4326"
+        instance.bbox_polygon = True
+        instance.bbox_string = "0,0,1,1"
+        instance.ows_url = "http://example.com/ows"
+        instance.alternate = "remote:layer"
+        instance.can_have_wfs_links = False
+        instance.subtype = "vector"
+        instance.can_have_style = False
+        instance.prepare_wms_links.return_value = []
+        instance.get_thumbnail_url.return_value = None
+        instance.get_real_instance.return_value = MagicMock(ptype="NOT_WMS")
+        instance.remote_service = MagicMock(
+            service_url="http://example.com/ows?service=WMS",
+            type="WMS",
+            id=42,
+            auth_config=remote_auth_config,
+        )
+
+        with (
+            patch("geonode.utils.check_ogc_backend", return_value=True),
+            patch("geonode.resource.utils.is_remote_resource", return_value=False),
+            patch("geonode.base.models.Link.objects") as mock_link_objects,
+            patch("geonode.services.serviceprocessors.get_service_handler") as mock_get_service_handler,
+        ):
+            mock_link_objects.filter.return_value.count.return_value = 0
+            mock_get_service_handler.return_value = MagicMock(_create_dataset_legend_link=MagicMock())
+
+            set_resource_default_links(instance, instance)
+
+            _, kwargs = mock_get_service_handler.call_args
+            self.assertIs(kwargs.get("auth_config"), remote_auth_config)
+
     def test_get_valid_user(self):
         # Verify it accepts an admin user
         adminuser = get_user_model().objects.get(is_superuser=True)
diff --git a/geonode/services/forms.py b/geonode/services/forms.py
index 03255f53715..b105a8a9f9c 100644
--- a/geonode/services/forms.py
+++ b/geonode/services/forms.py
@@ -22,15 +22,15 @@
 from django import forms
 from django.core.exceptions import ValidationError
 from django.utils.translation import gettext_lazy as _
-import taggit
+from taggit.forms import TagField
 
 from geonode.security.auth_handlers import BasicAuthHandler
 from geonode.security.auth_registry import auth_handler_registry
 from geonode.security.models import AuthConfig
 
-from . import enumerations
-from .models import Service, get_service_type_choices
-from .serviceprocessors import get_service_handler
+from geonode.services import enumerations
+from geonode.services.models import Service, get_service_type_choices
+from geonode.services.serviceprocessors import get_service_handler
 from geonode.utils import is_safe_url
 
 logger = logging.getLogger(__name__)
@@ -72,10 +72,6 @@ def clean_url(self):
 
         if not is_safe_url(proposed_url):
             raise ValidationError(_("Invalid URL provided"))
-
-        existing = Service.objects.filter(base_url=proposed_url).exists()
-        if existing:
-            raise ValidationError(_("Service %(url)s is already registered"), params={"url": proposed_url})
         return proposed_url
 
     def clean(self):
@@ -129,7 +125,7 @@ class ServiceForm(forms.ModelForm):
     )
     description = forms.CharField(label=_("Description"), widget=forms.Textarea(attrs={"cols": 60}))
     abstract = forms.CharField(label=_("Abstract"), widget=forms.Textarea(attrs={"cols": 60}))
-    keywords = taggit.forms.TagField(required=False)
+    keywords = TagField(required=False)
 
     class Meta:
         model = Service
diff --git a/geonode/services/migrations/0061_alter_service_base_url.py b/geonode/services/migrations/0061_alter_service_base_url.py
new file mode 100644
index 00000000000..b1ff6aed39c
--- /dev/null
+++ b/geonode/services/migrations/0061_alter_service_base_url.py
@@ -0,0 +1,19 @@
+# Generated by Django 5.2.13 on 2026-07-02 10:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("services", "0060_alter_service_type"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="service",
+            name="base_url",
+            field=models.URLField(db_index=True),
+        ),
+    ]
+
diff --git a/geonode/services/models.py b/geonode/services/models.py
index b3b47e1ecb0..df1a72dbd49 100644
--- a/geonode/services/models.py
+++ b/geonode/services/models.py
@@ -31,7 +31,7 @@
 from geonode.layers.enumerations import GXP_PTYPES
 from geonode.people.enumerations import ROLE_VALUES
 from geonode.services.serviceprocessors import get_available_service_types
-from . import enumerations
+from geonode.services import enumerations
 
 
 def get_service_type_choices():
@@ -56,8 +56,9 @@ class Service(ResourceBase):
             (enumerations.OPENGEOPORTAL, _("OpenGeoPortal")),
         ),
     )
-    # with service, version and request etc stripped off
-    base_url = models.URLField(unique=True, db_index=True)
+    # URL with service, version and request etc stripped off
+    # Intentionally non-unique: the same endpoint can be registered by different owners/use-cases.
+    base_url = models.URLField(db_index=True)
     version = models.CharField(max_length=100, null=True, blank=True)
     # Should force to slug?
     name = models.CharField(max_length=255, unique=True, db_index=True)
diff --git a/geonode/services/serviceprocessors/__init__.py b/geonode/services/serviceprocessors/__init__.py
index eba68ae7302..7b8bf0e4a76 100644
--- a/geonode/services/serviceprocessors/__init__.py
+++ b/geonode/services/serviceprocessors/__init__.py
@@ -18,17 +18,21 @@
 #########################################################################
 import logging
 
-from django.conf import settings
 from geonode.services import enumerations
-from django.core.cache import caches
-from geonode.services.serviceprocessors.registry import service_type_registry
+from geonode.services.serviceprocessors.cache import service_handler_cache
+from geonode.services.serviceprocessors.registry import service_type_registry
 
-service_cache = caches["services"]
 logger = logging.getLogger(__name__)
 
 
-def get_available_service_types():
-    return service_type_registry.get_available_service_types()
+def get_service_cache_key(base_url, service_type=enumerations.AUTO, service_id=None, auth=None, auth_config=None):
+    return service_handler_cache.get_key(
+        base_url, service_type=service_type, service_id=service_id, auth=auth, auth_config=auth_config
+    )
+
+
+def get_available_service_types():
+    return service_type_registry.get_available_service_types()
 
 
 def get_service_handler(base_url, service_type=enumerations.AUTO, service_id=None, *args, **kwargs):
@@ -36,13 +40,25 @@ def get_service_handler(base_url, service_type=enumerations.AUTO, service_id=Non
     If the service type is not explicitly passed in it will be guessed from
     """
 
-    if entry := service_cache.get(base_url):
-        return entry
+    # Without a service_id the key can't tell apart two unpersisted registration
+    # attempts for the same type/url/auth, so skip caching until one exists.
+    cache_key = None
+    if service_id is not None:
+        cache_key = get_service_cache_key(
+            base_url,
+            service_type=service_type,
+            service_id=service_id,
+            auth=kwargs.get("auth"),
+            auth_config=kwargs.get("auth_config"),
+        )
+        if entry := service_handler_cache.get(cache_key):
+            return entry
 
-    handler = service_type_registry.get_handler_class(service_type)
+    handler = service_type_registry.get_handler_class(service_type)
     try:
         service_handler = handler(base_url, service_id, *args, **kwargs)
-        service_cache.set(service_handler.url, service_handler, settings.SERVICE_CACHE_EXPIRATION_TIME)
+        if cache_key is not None:
+            service_handler_cache.set(cache_key, service_handler)
     except Exception as e:
         logger.exception(e)
         logger.exception(msg=f"Could not parse service {base_url}")
diff --git a/geonode/services/serviceprocessors/arcgis.py b/geonode/services/serviceprocessors/arcgis.py
index b711153c912..86e17841a9d 100644
--- a/geonode/services/serviceprocessors/arcgis.py
+++ b/geonode/services/serviceprocessors/arcgis.py
@@ -20,9 +20,11 @@
 import os
 import logging
 import traceback
-
+from collections import namedtuple
 from uuid import uuid4
 
+from arcrest import MapService as ArcMapService, ImageService as ArcImageService
+
 from django.conf import settings
 from django.db import transaction
 from django.utils.translation import gettext_lazy as _
@@ -31,16 +33,9 @@
 from geonode import GeoNodeException
 from geonode.base.bbox_utils import BBOXHelper
 from geonode.harvesting.models import Harvester
+from geonode.services import enumerations, models, utils
+from geonode.services.serviceprocessors import base
 
-from arcrest import MapService as ArcMapService, ImageService as ArcImageService
-
-from .. import enumerations
-from ..enumerations import INDEXED
-from .. import models
-from .. import utils
-from . import base
-
-from collections import namedtuple
 
 logger = logging.getLogger(__name__)
 
@@ -86,7 +81,7 @@ def __init__(self, url, geonode_service_id=None, *args, **kwargs):
         #     extent['ymax']
         # ])
 
-        self.indexing_method = INDEXED
+        self.indexing_method = enumerations.INDEXED
         self.name = slugify(self.url)[:255]
         self.title = str(_title).encode("utf-8", "ignore").decode("utf-8")
 
@@ -96,7 +91,7 @@ def parsed_service(self):
 
     def probe(self):
         try:
-            return True if len(self.parsed_service._json_struct) > 0 else False
+            return len(self.parsed_service._json_struct) > 0
         except Exception:
             return False
 
@@ -110,39 +105,44 @@ def create_geonode_service(self, owner, parent=None):
         :type owner: geonode.people.models.Profile
 
         """
-        with transaction.atomic():
-            instance = models.Service.objects.create(
-                uuid=str(uuid4()),
-                base_url=self.url,
-                type=self.service_type,
-                method=self.indexing_method,
-                owner=owner,
-                metadata_only=True,
-                version=str(self.parsed_service._json_struct.get("currentVersion", 0.0))
-                .encode("utf-8", "ignore")
-                .decode("utf-8"),
-                name=self.name,
-                title=self.title,
-                abstract=str(self.parsed_service._json_struct.get("serviceDescription"))
-                .encode("utf-8", "ignore")
-                .decode("utf-8")
-                or _("Not provided"),
-            )
-            service_harvester = Harvester.objects.create(
-                name=self.name,
-                default_owner=owner,
-                scheduling_enabled=False,
-                remote_url=instance.service_url,
-                delete_orphan_resources_automatically=True,
-                harvester_type=enumerations.HARVESTER_TYPES[self.service_type],
-                harvester_type_specific_configuration=self.get_harvester_configuration_options(),
-            )
-            if service_harvester.update_availability():
-                service_harvester.initiate_update_harvestable_resources()
-            else:
-                logger.exception(GeoNodeException("Could not reach remote endpoint."))
-            instance.harvester = service_harvester
 
+        def _create(unique_name):
+            with transaction.atomic():
+                new_instance = models.Service.objects.create(
+                    uuid=str(uuid4()),
+                    base_url=self.url,
+                    type=self.service_type,
+                    method=self.indexing_method,
+                    owner=owner,
+                    metadata_only=True,
+                    version=str(self.parsed_service._json_struct.get("currentVersion", 0.0))
+                    .encode("utf-8", "ignore")
+                    .decode("utf-8"),
+                    name=unique_name,
+                    title=self.title,
+                    abstract=str(self.parsed_service._json_struct.get("serviceDescription"))
+                    .encode("utf-8", "ignore")
+                    .decode("utf-8")
+                    or _("Not provided"),
+                )
+                new_harvester = Harvester.objects.create(
+                    name=unique_name,
+                    default_owner=owner,
+                    scheduling_enabled=False,
+                    remote_url=new_instance.service_url,
+                    delete_orphan_resources_automatically=True,
+                    harvester_type=enumerations.HARVESTER_TYPES[self.service_type],
+                    harvester_type_specific_configuration=self.get_harvester_configuration_options(),
+                )
+                if new_harvester.update_availability():
+                    new_harvester.initiate_update_harvestable_resources()
+                else:
+                    logger.exception(GeoNodeException("Could not reach remote endpoint."))
+                new_instance.harvester = new_harvester
+                return new_instance
+
+        instance = base.create_with_unique_name(self.name, _create)
+        self.name = instance.name
         self.geonode_service_id = instance.id
         return instance
 
@@ -235,7 +235,7 @@ def __init__(self, url, geonode_service_id=None, *args, **kwargs):
         #     extent['ymax']
         # ])
 
-        self.indexing_method = INDEXED
+        self.indexing_method = enumerations.INDEXED
         self.name = slugify(self.url)[:255]
         self.title = _title
 
diff --git a/geonode/services/serviceprocessors/base.py b/geonode/services/serviceprocessors/base.py
index d78cc858398..187ca890eed 100644
--- a/geonode/services/serviceprocessors/base.py
+++ b/geonode/services/serviceprocessors/base.py
@@ -22,14 +22,14 @@
 import logging
 
 from django.conf import settings
+from django.db import IntegrityError
 
 from geonode.utils import check_ogc_backend
 from geonode import GeoNodeException, geoserver
 from geonode.harvesting.tasks import harvest_resources
-from geonode.harvesting.models import AsynchronousHarvestingSession
+from geonode.harvesting.models import AsynchronousHarvestingSession, Harvester
 
-from .. import models
-from .. import enumerations
+from geonode.services import models, enumerations
 
 if check_ogc_backend(geoserver.BACKEND_PACKAGE):
     from geonode.geoserver.helpers import gs_catalog as catalog
@@ -51,6 +51,42 @@ def get_geoserver_cascading_workspace(create=True):
     return workspace
 
 
+def build_unique_resource_name(name, max_length=255):
+    """Return a Service/Harvester name that is unique in both models."""
+    max_length = max(1, int(max_length))
+    candidate = (name or "service")[:max_length]
+    base_name = candidate
+    idx = 1
+    while models.Service.objects.filter(name=candidate).exists() or Harvester.objects.filter(name=candidate).exists():
+        suffix = f"-{idx}"
+        if len(suffix) >= max_length:
+            # When max_length is tiny, keep the most specific part of the suffix.
+            candidate = suffix[-max_length:]
+        else:
+            prefix_len = max_length - len(suffix)
+            candidate = f"{base_name[:prefix_len]}{suffix}"
+        idx += 1
+    return candidate
+
+
+def create_with_unique_name(name, create_fn, max_length=255, max_attempts=3):
+    """Call create_fn(unique_name), retrying with a fresh candidate on IntegrityError.
+
+    build_unique_resource_name's check isn't atomic with the caller's create(),
+    so two concurrent registrations can still race for the same name.
+
+    :arg create_fn: creates the object(s); must raise IntegrityError if the
+        name turned out to already be taken.
+    """
+    for attempt in range(max_attempts):
+        candidate = build_unique_resource_name(name, max_length=max_length)
+        try:
+            return create_fn(candidate)
+        except IntegrityError:
+            if attempt == max_attempts - 1:
+                raise
+
+
 class ServiceHandlerBase(object):  # LGTM: @property will not work in old-style classes
     """Base class for remote service handlers
 
diff --git a/geonode/services/serviceprocessors/cache.py b/geonode/services/serviceprocessors/cache.py
new file mode 100644
index 00000000000..582676ac2c6
--- /dev/null
+++ b/geonode/services/serviceprocessors/cache.py
@@ -0,0 +1,88 @@
+#########################################################################
+#
+# Copyright (C) 2026 OSGeo
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+#########################################################################
+import hashlib
+import hmac
+
+from django.conf import settings
+from django.core.cache import caches
+
+from geonode.services import enumerations
+
+
+class ServiceHandlerCache:
+    """Caches remote service handler instances, keyed by URL, service type and auth."""
+
+    def __init__(self):
+        self.cache = caches["services"]
+
+    @staticmethod
+    def _digest(items):
+        # HMAC-keyed, not a bare hash: cache keys can leak via backend
+        # introspection (e.g. Redis SCAN), and a bare digest of low-entropy
+        # credentials would allow offline guessing.
+        message = repr(sorted(items)).encode("utf-8")
+        return hmac.new(settings.SECRET_KEY.encode("utf-8"), message, hashlib.sha256).hexdigest()[:16]
+
+    @classmethod
+    def _build_auth_fingerprint(cls, auth=None, auth_config=None):
+        """Build a non-sensitive auth discriminator for service-handler cache keys."""
+        if auth_config is not None:
+            # Content-based, not AuthConfig.pk: service_id already scopes the key,
+            # and this keeps the fingerprint stable across its unsaved -> saved
+            # transition during handler.create_geonode_service.
+            payload_digest = cls._digest((getattr(auth_config, "payload", None) or {}).items())
+            auth_type = getattr(auth_config, "type", None)
+            return f"authcfg:{auth_type or '-'}:{payload_digest}"
+
+        if auth is not None:
+            # HashableAuthBase (geonode.security.auth_handlers) wraps the real
+            # requests.auth.AuthBase in `.auth`; unwrap to fingerprint it.
+            wrapped_auth = getattr(auth, "auth", auth)
+            if isinstance(wrapped_auth, tuple) and len(wrapped_auth) == 2:
+                digest = cls._digest({"username": wrapped_auth[0], "password": wrapped_auth[1]}.items())
+                return f"auth:basic:{digest}"
+            if hasattr(wrapped_auth, "__dict__") and wrapped_auth.__dict__:
+                digest = cls._digest(wrapped_auth.__dict__.items())
+                return f"auth:{wrapped_auth.__class__.__name__}:{digest}"
+            if hasattr(wrapped_auth, "username"):
+                return f"auth:{wrapped_auth.__class__.__name__}:{getattr(wrapped_auth, 'username', '-') or '-'}"
+            return f"auth:{auth.__class__.__name__}"
+
+        return "-"
+
+    @staticmethod
+    def _build_url_fingerprint(base_url):
+        return hashlib.sha256((base_url or "").encode("utf-8")).hexdigest()
+
+    def get_key(self, base_url, service_type=enumerations.AUTO, service_id=None, auth=None, auth_config=None):
+        auth_fingerprint = self._build_auth_fingerprint(auth=auth, auth_config=auth_config)
+        url_fingerprint = self._build_url_fingerprint(base_url)
+        return f"{service_type}|{service_id or '-'}|{auth_fingerprint}|{url_fingerprint}"
+
+    def get(self, key):
+        return self.cache.get(key)
+
+    def set(self, key, value):
+        self.cache.set(key, value, settings.SERVICE_CACHE_EXPIRATION_TIME)
+
+    def delete(self, key):
+        self.cache.delete(key)
+
+
+service_handler_cache = ServiceHandlerCache()
diff --git a/geonode/services/serviceprocessors/wms.py b/geonode/services/serviceprocessors/wms.py
index 4391edd3219..74579e3291d 100644
--- a/geonode/services/serviceprocessors/wms.py
+++ b/geonode/services/serviceprocessors/wms.py
@@ -41,13 +41,8 @@
 from geonode.harvesting.models import Harvester
 from geonode.harvesting.harvesters.wms import OgcWmsHarvester, WebMapService
 from geonode.security.auth_registry import auth_handler_registry
-
-from .. import enumerations
-from ..enumerations import CASCADED
-from ..enumerations import INDEXED
-from .. import models
-from .. import utils
-from . import base, get_service_handler
+from geonode.services import models, utils, enumerations
+from geonode.services.serviceprocessors import base, get_service_handler
 
 logger = logging.getLogger(__name__)
 
@@ -62,7 +57,7 @@ def __init__(self, url, geonode_service_id=None, *args, **kwargs):
         self.args = args
         self.kwargs = kwargs
         self._parsed_service = None
-        self.indexing_method = INDEXED if self._offers_geonode_projection() else CASCADED
+        self.indexing_method = enumerations.INDEXED if self._offers_geonode_projection() else enumerations.CASCADED
         self.name = slugify(self.url)[:255]
 
     @staticmethod
@@ -75,7 +70,7 @@ def get_cleaned_url_params(url):
         get_args = parsed_url.query
         # Converting URL arguments to dict
         parsed_get_args = dict(parse_qsl(get_args))
-        # Strip out redoundant args
+        # Strip out redundant args
         _version = parsed_get_args.pop("version", "1.3.0") if "version" in parsed_get_args else "1.3.0"
         _service = parsed_get_args.pop("service") if "service" in parsed_get_args else None
         _request = parsed_get_args.pop("request") if "request" in parsed_get_args else None
@@ -109,7 +104,7 @@ def parsed_service(self):
 
     def probe(self):
         try:
-            return True if len(self.parsed_service.contents) > 0 else False
+            return len(self.parsed_service.contents) > 0
         except Exception:
             return False
 
@@ -140,42 +135,50 @@ def create_geonode_service(self, owner, parent=None):
             if auth_config is not None and auth_config.pk is None:
                 auth_config.save()
 
-            with transaction.atomic():
-                service = models.Service.objects.create(
-                    uuid=str(uuid4()),
-                    base_url=f"{cleaned_url.scheme}://{cleaned_url.netloc}{cleaned_url.path}".encode(
-                        "utf-8", "ignore"
-                    ).decode("utf-8"),
-                    extra_queryparams=cleaned_url.query,
-                    type=self.service_type,
-                    method=self.indexing_method,
-                    owner=owner,
-                    metadata_only=True,
-                    version=str(self.parsed_service.identification.version).encode("utf-8", "ignore").decode("utf-8"),
-                    name=self.name,
-                    title=str(self.parsed_service.identification.title).encode("utf-8", "ignore").decode("utf-8")
-                    or self.name,
-                    abstract=str(self.parsed_service.identification.abstract).encode("utf-8", "ignore").decode("utf-8")
-                    or _("Not provided"),
-                    operations=OgcWmsHarvester.get_wms_operations(self.parsed_service.url, version=version),
-                    auth_config=auth_config,
-                )
-                service_harvester = Harvester.objects.create(
-                    name=self.name,
-                    default_owner=owner,
-                    scheduling_enabled=False,
-                    remote_url=service.service_url,
-                    delete_orphan_resources_automatically=True,
-                    harvester_type=enumerations.HARVESTER_TYPES[self.service_type],
-                    harvester_type_specific_configuration=self.get_harvester_configuration_options(),
-                )
-                service.harvester = service_harvester
-                service.save()
-                if service_harvester.update_availability():
-                    service_harvester.initiate_update_harvestable_resources()
-                else:
-                    logger.exception(GeoNodeException("Could not reach remote endpoint."))
-
+            def _create(unique_name):
+                with transaction.atomic():
+                    new_service = models.Service.objects.create(
+                        uuid=str(uuid4()),
+                        base_url=f"{cleaned_url.scheme}://{cleaned_url.netloc}{cleaned_url.path}".encode(
+                            "utf-8", "ignore"
+                        ).decode("utf-8"),
+                        extra_queryparams=cleaned_url.query,
+                        type=self.service_type,
+                        method=self.indexing_method,
+                        owner=owner,
+                        metadata_only=True,
+                        version=str(self.parsed_service.identification.version)
+                        .encode("utf-8", "ignore")
+                        .decode("utf-8"),
+                        name=unique_name,
+                        title=str(self.parsed_service.identification.title).encode("utf-8", "ignore").decode("utf-8")
+                        or unique_name,
+                        abstract=str(self.parsed_service.identification.abstract)
+                        .encode("utf-8", "ignore")
+                        .decode("utf-8")
+                        or _("Not provided"),
+                        operations=OgcWmsHarvester.get_wms_operations(self.parsed_service.url, version=version),
+                        auth_config=auth_config,
+                    )
+                    new_harvester = Harvester.objects.create(
+                        name=unique_name,
+                        default_owner=owner,
+                        scheduling_enabled=False,
+                        remote_url=new_service.service_url,
+                        delete_orphan_resources_automatically=True,
+                        harvester_type=enumerations.HARVESTER_TYPES[self.service_type],
+                        harvester_type_specific_configuration=self.get_harvester_configuration_options(),
+                    )
+                    new_service.harvester = new_harvester
+                    new_service.save()
+                    if new_harvester.update_availability():
+                        new_harvester.initiate_update_harvestable_resources()
+                    else:
+                        logger.exception(GeoNodeException("Could not reach remote endpoint."))
+                    return new_service
+
+            service = base.create_with_unique_name(self.name, _create)
+            self.name = service.name
             self.geonode_service_id = service.id
         except Exception as e:
             logger.exception(e)
@@ -289,20 +292,24 @@ def __init__(self, url, geonode_service_id=None, *args, **kwargs):
         self.args = args
         self.kwargs = kwargs
 
-        self.indexing_method = INDEXED
+        self.indexing_method = enumerations.INDEXED
         self.name = slugify(self.url)[:255]
 
     @property
     def parsed_service(self):
-        cleaned_url, service, version, request = WmsServiceHandler.get_cleaned_url_params(self.ows_endpoint())
-        auth = None
-        if service.needs_authentication:
-            auth = auth_handler_registry.build(service.auth_config).get_request_auth()
+        # 2nd return value is the raw `service=` query param, not a Service model;
+        # auth comes from self.kwargs, as no Service row may exist yet.
+        cleaned_url, _, version, _request = WmsServiceHandler.get_cleaned_url_params(self.ows_endpoint())
+        auth = self.kwargs.get("auth")
+        auth_config = self.kwargs.get("auth_config")
+        if auth is None and auth_config is not None:
+            auth = auth_handler_registry.build(auth_config).get_request_auth()
         _parsed_service = get_service_handler(
             cleaned_url.geturl(),
-            service.type,
-            service.id,
+            enumerations.WMS,
+            self.geonode_service_id,
             auth=auth,
+            auth_config=auth_config,
         )
         return _parsed_service
 
diff --git a/geonode/services/tests.py b/geonode/services/tests.py
index 1e16f49288c..86ee644f5ba 100644
--- a/geonode/services/tests.py
+++ b/geonode/services/tests.py
@@ -28,6 +28,7 @@
 from arcrest import MapService as ArcMapService
 from unittest import TestCase as StandardTestCase, skip
 from owslib.wms import WebMapService as OwsWebMapService
+from django.db import IntegrityError
 from django.test import Client, override_settings
 from django.urls import reverse
 from django.contrib.auth import get_user_model
@@ -47,16 +48,120 @@
 from geonode.harvesting.harvesters.wms import WebMapService
 from geonode.services.utils import parse_services_types, test_resource_table_status
 
-from . import enumerations, forms
-from .models import Service
-from .serviceprocessors import base, wms, arcgis, get_service_handler, get_available_service_types
-from .serviceprocessors.arcgis import ArcImageServiceHandler, ArcMapServiceHandler, MapLayer
-from .serviceprocessors.registry import ServiceTypeRegistry, service_type_registry
+from geonode.services import enumerations, forms, views
+from geonode.services.models import Service
+from geonode.services.serviceprocessors import (
+    base,
+    wms,
+    arcgis,
+    get_service_cache_key,
+    get_service_handler,
+    get_available_service_types,
+)
+from geonode.services.serviceprocessors.arcgis import ArcImageServiceHandler, ArcMapServiceHandler, MapLayer
+from geonode.services.serviceprocessors.registry import ServiceTypeRegistry, service_type_registry
 
 logger = logging.getLogger(__name__)
 
 
+class PickableMagicMock(mock.MagicMock):
+    # Plain MagicMocks aren't picklable, which LocMemCache.set() requires.
+    def __reduce__(self):
+        return (mock.MagicMock, ())
+
+
 class ModuleFunctionsTestCase(StandardTestCase):
+    def test_get_service_cache_key_is_service_specific(self):
+        phony_url = "http://fake"
+        key_1 = get_service_cache_key(phony_url, service_type=enumerations.WMS, service_id=1)
+        key_2 = get_service_cache_key(phony_url, service_type=enumerations.WMS, service_id=2)
+        self.assertNotEqual(key_1, key_2)
+
+    def test_get_service_cache_key_is_auth_specific(self):
+        phony_url = "http://fake"
+        auth_1 = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_1.payload = {"username": "alice", "password": "pw1"}
+        auth_2 = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_2.payload = {"username": "bob", "password": "pw1"}
+
+        key_1 = get_service_cache_key(phony_url, service_type=enumerations.WMS, service_id=1, auth_config=auth_1)
+        key_2 = get_service_cache_key(phony_url, service_type=enumerations.WMS, service_id=1, auth_config=auth_2)
+        self.assertNotEqual(key_1, key_2)
+
+    def test_get_service_cache_key_changes_when_saved_auth_config_payload_changes(self):
+        # A password change on a saved AuthConfig must bust the cache key.
+        phony_url = "http://fake"
+        auth_config = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_config.payload = {"username": "alice", "password": "pw1"}
+        auth_config.save()
+
+        key_before = get_service_cache_key(
+            phony_url, service_type=enumerations.WMS, service_id=1, auth_config=auth_config
+        )
+
+        auth_config.payload = {"username": "alice", "password": "pw2"}
+        auth_config.save()
+
+        key_after = get_service_cache_key(
+            phony_url, service_type=enumerations.WMS, service_id=1, auth_config=auth_config
+        )
+        self.assertNotEqual(key_before, key_after)
+
+    def test_get_service_cache_key_is_auth_specific_for_hashable_auth_base(self):
+        # get_request_auth() wraps the real auth object in a HashableAuthBase;
+        # the fingerprint must unwrap it instead of treating every instance alike.
+        phony_url = "http://fake"
+        auth_config_1 = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_config_1.payload = {"username": "alice", "password": "pw1"}
+        auth_config_2 = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_config_2.payload = {"username": "alice", "password": "pw2"}
+
+        auth_1 = auth_handler_registry.build(auth_config_1).get_request_auth()
+        auth_2 = auth_handler_registry.build(auth_config_2).get_request_auth()
+
+        key_1 = get_service_cache_key(phony_url, service_type=enumerations.WMS, service_id=1, auth=auth_1)
+        key_2 = get_service_cache_key(phony_url, service_type=enumerations.WMS, service_id=1, auth=auth_2)
+        self.assertNotEqual(key_1, key_2)
+        self.assertNotIn("HashableAuthBase", key_1)
+
+    def test_get_service_cache_key_is_auth_specific_for_tuple_auth(self):
+        # A password change must bust the key even with the same username.
+        phony_url = "http://fake"
+        key_1 = get_service_cache_key(phony_url, service_type=enumerations.WMS, service_id=1, auth=("bob", "pw1"))
+        key_2 = get_service_cache_key(phony_url, service_type=enumerations.WMS, service_id=1, auth=("bob", "pw2"))
+        self.assertNotEqual(key_1, key_2)
+
+    def test_get_service_cache_key_does_not_embed_plaintext_credentials(self):
+        # Cache keys can leak via backend introspection, so no raw credentials.
+        phony_url = "http://fake"
+        auth_config = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_config.payload = {"username": "alice", "password": "super-secret-password"}
+
+        key = get_service_cache_key(phony_url, service_type=enumerations.WMS, service_id=1, auth_config=auth_config)
+        self.assertNotIn("super-secret-password", key)
+        self.assertNotIn("alice", key)
+
+    def test_get_service_cache_key_is_stable_across_auth_config_save(self):
+        # Saving an AuthConfig must not change its fingerprint (avoids an
+        # orphaned cache entry from before create_geonode_service saves it).
+        phony_url = "http://fake"
+        auth_config = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_config.payload = {"username": "alice", "password": "pw1"}
+
+        key_before_save = get_service_cache_key(
+            phony_url, service_type=enumerations.WMS, service_id=1, auth_config=auth_config
+        )
+        auth_config.save()
+        key_after_save = get_service_cache_key(
+            phony_url, service_type=enumerations.WMS, service_id=1, auth_config=auth_config
+        )
+        self.assertEqual(key_before_save, key_after_save)
+
+    def test_get_service_cache_key_has_bounded_length(self):
+        very_long_url = "http://example.com/" + ("a" * 2000)
+        key = get_service_cache_key(very_long_url, service_type=enumerations.WMS, service_id=1)
+        self.assertLess(len(key), 200)
+
     @mock.patch("geonode.services.serviceprocessors.base.catalog", autospec=True)
     @mock.patch("geonode.services.serviceprocessors.base.settings", autospec=True)
     def test_get_cascading_workspace_returns_existing(self, mock_settings, mock_catalog):
@@ -97,14 +202,81 @@ def test_get_cascading_workspace_creates_new_workspace(self, mock_settings, mock
             mock_settings.CASCADE_WORKSPACE, f"http://www.geonode.org/{mock_settings.CASCADE_WORKSPACE}"
         )
 
-    @mock.patch("geonode.services.serviceprocessors.service_type_registry.get_handler_class", autospec=True)
-    def test_get_service_handler_wms(self, mock_wms_handler):
-        class PickableMagicMock(mock.MagicMock):
-            def __reduce__(self):
-                return (mock.MagicMock, ())
+    @mock.patch("geonode.services.serviceprocessors.base.build_unique_resource_name")
+    def test_create_with_unique_name_retries_on_integrity_error(self, mock_build_unique_resource_name):
+        # Must retry with a fresh candidate instead of surfacing IntegrityError.
+        mock_build_unique_resource_name.side_effect = ["candidate-1", "candidate-2"]
+        calls = []
+
+        def flaky_create(name):
+            calls.append(name)
+            if len(calls) < 2:
+                raise IntegrityError("simulated race")
+            return name
+
+        result = base.create_with_unique_name("svc", flaky_create)
+        self.assertEqual(result, "candidate-2")
+        self.assertEqual(calls, ["candidate-1", "candidate-2"])
+
+    @mock.patch("geonode.services.serviceprocessors.base.build_unique_resource_name")
+    def test_create_with_unique_name_gives_up_after_max_attempts(self, mock_build_unique_resource_name):
+        mock_build_unique_resource_name.side_effect = ["c1", "c2", "c3"]
+
+        def always_fails(name):
+            raise IntegrityError("simulated race")
+
+        with self.assertRaises(IntegrityError):
+            base.create_with_unique_name("svc", always_fails, max_attempts=3)
+
+    @mock.patch("geonode.services.serviceprocessors.registry.service_type_registry.get_handler_class", autospec=True)
+    def test_get_service_handler_does_not_cache_without_service_id(self, mock_get_handler_class):
+        # Without a service_id, two unrelated calls must not share a cached handler.
+        handler_class = mock.MagicMock(side_effect=lambda *a, **k: mock.MagicMock())
+        mock_get_handler_class.return_value = handler_class
+        phony_url = "http://fake"
+
+        handler_1 = get_service_handler(phony_url, service_type=enumerations.WMS)
+        handler_2 = get_service_handler(phony_url, service_type=enumerations.WMS)
+
+        self.assertEqual(handler_class.call_count, 2)
+        self.assertIsNot(handler_1, handler_2)
+
+    @mock.patch("geonode.services.serviceprocessors.registry.service_type_registry.get_handler_class", autospec=True)
+    def test_get_service_handler_caches_when_service_id_present(self, mock_get_handler_class):
+        # Cached values are pickled (LocMemCache), so the constructed handler must be too.
+        handler_class = mock.MagicMock(side_effect=lambda *a, **k: PickableMagicMock())
+        mock_get_handler_class.return_value = handler_class
+        # Unique URL: LocMemCache isn't reset between test runs.
+        phony_url = f"http://fake/{uuid4()}"
+
+        get_service_handler(phony_url, service_type=enumerations.WMS, service_id=1)
+        get_service_handler(phony_url, service_type=enumerations.WMS, service_id=1)
+
+        # Second call must be a cache hit, not a second construction.
+        self.assertEqual(handler_class.call_count, 1)
+
+    @mock.patch("geonode.services.serviceprocessors.registry.service_type_registry.get_handler_class", autospec=True)
+    def test_get_service_handler_separates_cache_by_auth(self, mock_get_handler_class):
+        # Same url/type/service_id but different credentials must not collide,
+        # since relaxed base_url uniqueness allows distinct Services to share a URL.
+        handler_class = mock.MagicMock(side_effect=lambda *a, **k: PickableMagicMock())
+        mock_get_handler_class.return_value = handler_class
+        phony_url = f"http://fake/{uuid4()}"
+
+        auth_config_1 = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_config_1.payload = {"username": "alice", "password": "pw1"}
+        auth_config_2 = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_config_2.payload = {"username": "bob", "password": "pw2"}
+
+        get_service_handler(phony_url, service_type=enumerations.WMS, service_id=1, auth_config=auth_config_1)
+        get_service_handler(phony_url, service_type=enumerations.WMS, service_id=1, auth_config=auth_config_2)
+
+        self.assertEqual(handler_class.call_count, 2)
 
+    @mock.patch("geonode.services.serviceprocessors.registry.service_type_registry.get_handler_class", autospec=True)
+    def test_get_service_handler_wms(self, mock_get_handler_class):
         _handler = PickableMagicMock()
-        mock_wms_handler.return_value = _handler
+        mock_get_handler_class.return_value = _handler
         phony_url = "http://fake"
         get_service_handler(phony_url, service_type=enumerations.WMS)
         _handler.assert_called_with(phony_url, None)
@@ -449,7 +621,7 @@ def test_get_arcgis_alternative_structure(self, mock_map_service):
             test_user.save()
         try:
             result = handler.create_geonode_service(test_user)
-            geonode_service, created = Service.objects.get_or_create(base_url=result.base_url, owner=test_user)
+            geonode_service = Service.objects.get(id=result.id)
             for _d in Dataset.objects.filter(remote_service=geonode_service):
                 resource_manager_registry.get_for_instance(_d).delete(_d.uuid, instance=_d)
 
@@ -516,6 +688,33 @@ def setUp(self):
             self.local_user.set_password("somepassword")
             self.local_user.save()
 
+    @mock.patch("geonode.services.serviceprocessors.wms.get_service_handler")
+    @mock.patch("geonode.services.serviceprocessors.wms.WmsServiceHandler.get_cleaned_url_params")
+    @mock.patch.object(wms.GeoNodeServiceHandler, "ows_endpoint")
+    def test_geonode_service_handler_parsed_service_passes_auth_config(
+        self, mock_ows_endpoint, mock_get_cleaned_url_params, mock_get_service_handler
+    ):
+        # `service` from get_cleaned_url_params is a query-string value, not a
+        # Service model; auth must come from self.kwargs instead.
+        mock_ows_endpoint.return_value = self.phony_url
+
+        auth_config = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_config.payload = {"username": "test_user", "password": "test_password"}
+        auth_config.save()
+
+        cleaned_url = mock.MagicMock()
+        cleaned_url.geturl.return_value = self.phony_url
+        mock_get_cleaned_url_params.return_value = (cleaned_url, None, self.phony_version, None)
+
+        handler = wms.GeoNodeServiceHandler(self.phony_url, geonode_service_id=42, auth_config=auth_config)
+        handler.parsed_service
+
+        args, kwargs = mock_get_service_handler.call_args
+        self.assertEqual(args[0], self.phony_url)
+        self.assertEqual(args[1], enumerations.WMS)
+        self.assertEqual(args[2], 42)
+        self.assertEqual(kwargs.get("auth_config"), auth_config)
+
     @mock.patch("geonode.harvesting.harvesters.wms.WebMapService")
     @mock.patch("geonode.services.serviceprocessors.wms.WmsServiceHandler.parsed_service", autospec=True)
     def test_has_correct_url(self, mock_wms_parsed_service, mock_wms):
@@ -705,7 +904,7 @@ def test_get_resources(self, mock_wms_get_resource, mock_wms_get_resources, mock
             test_user.save()
         result = handler.create_geonode_service(test_user)
         try:
-            geonode_service, created = Service.objects.get_or_create(base_url=result.base_url, owner=test_user)
+            geonode_service = Service.objects.get(id=result.id)
             for _d in Dataset.objects.filter(remote_service=geonode_service):
                 resource_manager_registry.get_for_instance(_d).delete(_d.uuid, instance=_d)
 
@@ -851,13 +1050,12 @@ def test_add_duplicate_remote_service_url(self):
             self.client.post(reverse("register_service"), data=form_data)
             self.assertEqual(Service.objects.count(), 1)
 
-            # Try adding the same URL again
+            # Adding the same URL again should now create a second Service
             form = forms.CreateServiceForm(form_data)
             self.assertEqual(Service.objects.count(), 1)
             response = self.client.post(reverse("register_service"), data=form_data)
-            # The service is None since there is already a created service from the first call
-            self.assertEqual(response.status_code, 404)
-            self.assertEqual(Service.objects.count(), 1)
+            self.assertEqual(response.status_code, 302)
+            self.assertEqual(Service.objects.count(), 2)
 
 
 class WmsServiceHarvestingTestCase(StaticLiveServerTestCase):
@@ -964,6 +1162,22 @@ def setUp(self):
         )
         self.sut.clear_dirty_state()
 
+    @mock.patch("geonode.services.views.get_service_handler")
+    def test_get_service_handler_view_passes_auth_config_for_cache_key(self, mock_get_service_handler):
+        # Must pass auth_config= too, not just auth=, so the cache key varies by credentials.
+        auth_config = AuthConfig(type=BasicAuthHandler.handled_type)
+        auth_config.payload = {"username": "test_user", "password": "test_password"}
+        auth_config.save()
+        self.sut.auth_config = auth_config
+        self.sut.save()
+
+        mock_get_service_handler.return_value = mock.MagicMock(geonode_service_id=self.sut.id)
+
+        views._get_service_handler(None, self.sut)
+
+        _, kwargs = mock_get_service_handler.call_args
+        self.assertEqual(kwargs.get("auth_config"), auth_config)
+
     def test_user_admin_can_access_to_page(self):
         self.client.login(username="admin", password="admin")
         response = self.client.get(reverse("services"))
diff --git a/geonode/services/views.py b/geonode/services/views.py
index 93ac7d0fe2a..5b8d8f15eed 100644
--- a/geonode/services/views.py
+++ b/geonode/services/views.py
@@ -35,17 +35,15 @@
 from geonode.harvesting.models import Harvester
 from geonode.security.views import _perms_info_json
 from geonode.security.utils import get_visible_resources, check_add_remote_resource_perm
-from django.core.cache import caches
 from django.core.exceptions import PermissionDenied
 
-from .models import Service
-from . import forms, enumerations
-from .serviceprocessors import get_service_handler
+from geonode.services.models import Service
+from geonode.services import forms, enumerations
+from geonode.services.serviceprocessors import get_service_cache_key, get_service_handler
+from geonode.services.serviceprocessors.cache import service_handler_cache
 from geonode.security.registry import permissions_registry
 from geonode.views import err403
 
-service_cache = caches["services"]
-
 logger = logging.getLogger(__name__)
 
 
@@ -86,7 +84,15 @@ def register_service(request):
             if service_handler.indexing_method == enumerations.CASCADED:
                 service_handler.create_cascaded_store(service)
             service_handler.geonode_service_id = service.id
-            service_cache.set(service_handler.url, service_handler, settings.SERVICE_CACHE_EXPIRATION_TIME)
+            service_handler_cache.set(
+                get_service_cache_key(
+                    service.service_url,
+                    service_type=service.type,
+                    service_id=service.id,
+                    auth_config=service.auth_config,
+                ),
+                service_handler,
+            )
             # commented out due to jsonserializer error, will be replaced with cache
             # request.session[service_handler.url] = service_handler
             # logger.debug("Added handler to the session")
@@ -114,7 +120,9 @@ def _get_service_handler(request, service):
 
         auth = auth_handler_registry.build(service.auth_config).get_request_auth()
 
-    service_handler = get_service_handler(service.service_url, service.type, service.id, auth=auth)
+    service_handler = get_service_handler(
+        service.service_url, service.type, service.id, auth=auth, auth_config=service.auth_config
+    )
     if not service_handler.geonode_service_id:
         service_handler.geonode_service_id = service.id
     # commented out due to jsonserializer error, will be replaced with cache
@@ -337,7 +345,14 @@ def remove_service(request, service_id):
     elif request.method == "POST":
         service.dataset_set.all().delete()
         # by deleting the harvester we delete also the service
-        service_cache.delete(service.base_url)
+        service_handler_cache.delete(
+            get_service_cache_key(
+                service.service_url,
+                service_type=service.type,
+                service_id=service.id,
+                auth_config=service.auth_config,
+            )
+        )
         service.harvester.delete()
         messages.add_message(request, messages.INFO, _(f"Service {service.title} has been deleted"))
         return HttpResponseRedirect(reverse("services"))
diff --git a/geonode/utils.py b/geonode/utils.py
index 572bdfd3e2c..e502112e733 100755
--- a/geonode/utils.py
+++ b/geonode/utils.py
@@ -1418,7 +1418,10 @@ def set_resource_default_links(instance, layer, prune=False, **kwargs):
                 from geonode.services.serviceprocessors import get_service_handler
 
                 handler = get_service_handler(
-                    instance.remote_service.service_url, service_type=instance.remote_service.type
+                    instance.remote_service.service_url,
+                    service_type=instance.remote_service.type,
+                    service_id=instance.remote_service.id,
+                    auth_config=instance.remote_service.auth_config,
                 )
                 if handler and hasattr(handler, "_create_dataset_legend_link"):
                     handler._create_dataset_legend_link(instance)