diff --git a/geonode/base/api/tests.py b/geonode/base/api/tests.py
index aabf8c055d6..92ea47db3bf 100644
--- a/geonode/base/api/tests.py
+++ b/geonode/base/api/tests.py
@@ -516,7 +516,7 @@ def test_base_resources(self):
         response = self.client.get(url, format="json")
         self.assertEqual(response.status_code, 200)
         self.assertEqual(len(response.data), 5)
-        self.assertEqual(response.data["total"], 28)
+        self.assertEqual(response.data["total"], 29)
 
         url = f"{reverse('base-resources-list')}?filter{{metadata_only}}=false"
         # Anonymous
@@ -984,7 +984,7 @@ def test_sort_resources(self):
         response = self.client.get(f"{url}?sort[]=title", format="json")
         self.assertEqual(response.status_code, 200)
         self.assertEqual(len(response.data), 5)
-        self.assertEqual(response.data["total"], 28)
+        self.assertEqual(response.data["total"], 29)
         # Pagination
         self.assertEqual(len(response.data["resources"]), 10)
 
@@ -997,7 +997,7 @@ def test_sort_resources(self):
         response = self.client.get(f"{url}?sort[]=-title", format="json")
         self.assertEqual(response.status_code, 200)
         self.assertEqual(len(response.data), 5)
-        self.assertEqual(response.data["total"], 28)
+        self.assertEqual(response.data["total"], 29)
         # Pagination
         self.assertEqual(len(response.data["resources"]), 10)
 
diff --git a/geonode/geoserver/signals.py b/geonode/geoserver/signals.py
index 0f3a84768bc..02250abe67d 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,23 @@ 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:
+            # Expected since #14381: the same endpoint can be registered by several services, so a
+            # name can match one dataset per service. Nothing on the map layer tells them apart.
+            logger.warning(
+                "Ambiguous dataset lookup for map layer '%s' (store=%s, ows_url=%s), leaving it unresolved",
+                instance.name,
+                instance.store,
+                instance.ows_url,
+            )
 
 
 @deprecated(version="3.2.1", reason="Use direct calls to the ReourceManager.")
diff --git a/geonode/harvesting/harvesters/arcgis.py b/geonode/harvesting/harvesters/arcgis.py
index a3bba50849a..11361ca4090 100644
--- a/geonode/harvesting/harvesters/arcgis.py
+++ b/geonode/harvesting/harvesters/arcgis.py
@@ -34,7 +34,6 @@
 import arcrest
 import requests
 from django.contrib.gis import geos
-from django.template.defaultfilters import slugify
 
 from geonode.layers.enumerations import GXP_PTYPES
 from geonode.layers.models import Dataset
@@ -204,7 +203,6 @@ def _get_resource_descriptor(
         _, service_name, service_type = parse_remote_url(harvestable_resource.unique_identifier)
         epsg_code, spatial_extent = _parse_spatial_extent(layer_representation["extent"])
         ows_url = harvestable_resource.harvester.remote_url
-        store = slugify(ows_url)
         name = layer_representation.get("id", layer_representation.get("name", "Undefined"))
         title = layer_representation.get("name", layer_representation.get("title", "Undefined"))
         workspace = "remoteWorkspace"
@@ -228,7 +226,8 @@ def _get_resource_descriptor(
             ),
             reference_systems=[epsg_code],
             additional_parameters={
-                "store": store,
+                # store is a GeoServer grouping: remote datasets have none
+                "store": None,
                 "workspace": workspace,
                 "alternate": alternate,
                 "ows_url": ows_url,
@@ -306,7 +305,6 @@ def _get_resource_descriptor(
         _, service_name, service_type = parse_remote_url(harvestable_resource.unique_identifier)
         epsg_code, spatial_extent = _parse_spatial_extent(layer_representation["extent"])
         ows_url = harvestable_resource.harvester.remote_url
-        store = slugify(ows_url)
         name = layer_representation.get("id", layer_representation.get("name", "Undefined"))
         title = layer_representation.get("name", layer_representation.get("title", "Undefined"))
         workspace = "remoteWorkspace"
@@ -330,7 +328,8 @@ def _get_resource_descriptor(
             ),
             reference_systems=[epsg_code],
             additional_parameters={
-                "store": store,
+                # store is a GeoServer grouping: remote datasets have none
+                "store": None,
                 "workspace": workspace,
                 "alternate": alternate,
                 "ows_url": ows_url,
diff --git a/geonode/harvesting/harvesters/geonodeharvester.py b/geonode/harvesting/harvesters/geonodeharvester.py
index 92a8160b711..27d7a6e32f4 100644
--- a/geonode/harvesting/harvesters/geonodeharvester.py
+++ b/geonode/harvesting/harvesters/geonodeharvester.py
@@ -392,6 +392,8 @@ def _get_resource_descriptor(
                 {
                     "alternate": resource["alternate"],
                     "workspace": resource["workspace"],
+                    # store is a GeoServer grouping: remote datasets have none
+                    "store": None,
                 }
             )
         return descriptor
@@ -782,6 +784,8 @@ def _get_dataset_additional_parameters(
             "resource_type": "dataset",
             "alternate": api_record.get("alternate", descriptor.identification.name),
             "workspace": api_record.get("workspace"),
+            # store is a GeoServer grouping: remote datasets have none
+            "store": None,
         }
         if descriptor.identification.native_format.lower() == RemoteDatasetType.VECTOR.value:
             result["subtype"] = GeoNodeDatasetType.VECTOR.value
diff --git a/geonode/harvesting/harvesters/wms.py b/geonode/harvesting/harvesters/wms.py
index 4a3c5af7a77..a33f656f582 100644
--- a/geonode/harvesting/harvesters/wms.py
+++ b/geonode/harvesting/harvesters/wms.py
@@ -32,7 +32,6 @@
 
 from django.conf import settings
 from django.contrib.gis import geos
-from django.template.defaultfilters import slugify
 from geonode.layers.models import Dataset
 from geonode.base.models import Link, ResourceBase
 from geonode.layers.enumerations import GXP_PTYPES
@@ -309,7 +308,6 @@ def get_resource(
             # WMS does not provide the date of the resource.
             # Use current time for the date stamp and resource time.
             time = datetime.now()
-            service_name = slugify(self.remote_url)[:255]
             contact = resourcedescriptor.RecordDescriptionContact(**data["contact"])
             result = base.HarvestedResourceInfo(
                 resource_descriptor=resourcedescriptor.RecordDescription(
@@ -335,7 +333,7 @@ def get_resource(
                     reference_systems=[relevant_layer["crs"]],
                     additional_parameters={
                         "alternate": relevant_layer["name"],
-                        "store": service_name,
+                        "store": None,
                         "workspace": "remoteWorkspace",
                         "ows_url": relevant_layer["wms_url"],
                         "ptype": GXP_PTYPES["WMS"],
diff --git a/geonode/harvesting/tests/test_harvester_worker_wms.py b/geonode/harvesting/tests/test_harvester_worker_wms.py
index 705ed4b2de5..32b1b7863cb 100644
--- a/geonode/harvesting/tests/test_harvester_worker_wms.py
+++ b/geonode/harvesting/tests/test_harvester_worker_wms.py
@@ -16,6 +16,8 @@
 # along with this program. If not, see <http://www.gnu.org/licenses/>.
 #
 #########################################################################
+from unittest import mock
+
 from geonode.harvesting.harvesters import wms
 from geonode.tests.base import GeoNodeBaseSimpleTestSupport
 
@@ -29,3 +31,25 @@ def test_get_nsmap(self):
         for original, expected in fixtures:
             result = wms._get_nsmap(original)
             self.assertEqual(result, expected)
+
+    def test_harvested_resource_has_no_store(self):
+        # store is a GeoServer grouping, so harvested datasets must not claim one (#14381)
+        data = {
+            "contact": {"role": "", "name": "someone"},
+            "layers": [
+                {
+                    "name": "bahra",
+                    "title": "Bahra",
+                    "abstract": "",
+                    "keywords": [],
+                    "spatial_extent": None,
+                    "crs": "EPSG:4326",
+                    "wms_url": "https://wms.example.org/geoserver/wms?layers=bahra",
+                }
+            ],
+        }
+        worker = wms.OgcWmsHarvester("https://wms.example.org/geoserver/wms", 1)
+        harvestable_resource = mock.MagicMock(unique_identifier="bahra", geonode_resource=None)
+        with mock.patch.object(worker, "_get_data", return_value=data):
+            harvested = worker.get_resource(harvestable_resource)
+        self.assertIsNone(harvested.resource_descriptor.additional_parameters["store"])
diff --git a/geonode/layers/api/tests.py b/geonode/layers/api/tests.py
index 3c71f320c06..f7afb8f10e4 100644
--- a/geonode/layers/api/tests.py
+++ b/geonode/layers/api/tests.py
@@ -141,8 +141,8 @@ def test_filter_dirty_state(self):
         self.assertEqual(response.status_code, 200)
         dataset_list = response.data["datasets"]
 
-        # dirty resource is now included in count
-        resource_count_clean = Dataset.objects.count()
+        # dirty resource is now included in count; the listing excludes metadata-only datasets
+        resource_count_clean = Dataset.objects.filter(metadata_only=False).count()
         self.assertEqual(len(dataset_list), resource_count_clean)
         # ensure that the updated dirty dataset is included in the response
         self.assertTrue(dirty_dataset.pk in [int(dataset["pk"]) for dataset in dataset_list])
@@ -157,10 +157,11 @@ def test_filter_dirty_state_include_dirty(self):
         dirty_dataset.dirty_state = True
         dirty_dataset.save()
 
+        # the listing excludes metadata-only datasets
         # clean resources
-        resource_count_clean = Dataset.objects.filter(dirty_state=False).count()
+        resource_count_clean = Dataset.objects.filter(dirty_state=False, metadata_only=False).count()
         # dirty resources
-        resource_count_dirty = Dataset.objects.filter(dirty_state=True).count()
+        resource_count_dirty = Dataset.objects.filter(dirty_state=True, metadata_only=False).count()
 
         resource_count_all = resource_count_clean + resource_count_dirty
 
@@ -187,7 +188,7 @@ def test_dataset_listing_advertised(self):
 
         prev_count = payload.json().get("total")
         # the user can see only the advertised resources
-        self.assertEqual(Dataset.objects.filter(advertised=True).count(), prev_count)
+        self.assertEqual(Dataset.objects.filter(advertised=True, metadata_only=False).count(), prev_count)
 
         payload = self.client.get(f"{url}?advertised=True")
         # so if advertised is True, we dont see the advertised=False resource
diff --git a/geonode/layers/migrations/0047_alter_dataset_store.py b/geonode/layers/migrations/0047_alter_dataset_store.py
new file mode 100644
index 00000000000..7e396b096ab
--- /dev/null
+++ b/geonode/layers/migrations/0047_alter_dataset_store.py
@@ -0,0 +1,17 @@
+# Generated by Django 5.2.13 on 2026-08-14 10:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("layers", "0046_remove_modeltranslation"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="dataset",
+            name="store",
+            field=models.CharField(blank=True, max_length=255, null=True, verbose_name="Store"),
+        ),
+    ]
diff --git a/geonode/layers/models.py b/geonode/layers/models.py
index ada0787e9dc..0f605243819 100644
--- a/geonode/layers/models.py
+++ b/geonode/layers/models.py
@@ -118,7 +118,8 @@ class Dataset(ResourceBase):
     # internal fields
     objects = DatasetManager()
     workspace = models.CharField(_("Workspace"), max_length=255)
-    store = models.CharField(_("Store"), max_length=255)
+    # Store is nullable for remote layers
+    store = models.CharField(_("Store"), max_length=255, null=True, blank=True)
     name = models.CharField(_("Name"), max_length=255)
     typename = models.CharField(_("Typename"), max_length=255, null=True, blank=True)
     ows_url = models.URLField(
diff --git a/geonode/layers/tests.py b/geonode/layers/tests.py
index 450192f4e36..8480815ba4b 100644
--- a/geonode/layers/tests.py
+++ b/geonode/layers/tests.py
@@ -284,6 +284,22 @@ def test_dataset_attribute_config(self):
         attribute_config = lyr.attribute_config()
         self.assertTrue("ftInfoTemplate" not in attribute_config)
 
+    def test_remote_datasets_are_exempt_from_store_workspace_name_uniqueness(self):
+        # store is NULL for remote datasets, and PostgreSQL does not consider two NULLs equal,
+        # so the (store, workspace, name) uniqueness no longer applies to them: the same remote
+        # endpoint can be registered by more than one service (#14381).
+        common = dict(
+            owner=get_user_model().objects.get(username="admin"),
+            workspace="remoteWorkspace",
+            name="bahra",
+            subtype="remote",
+            store=None,
+        )
+        first = Dataset.objects.create(uuid=str(uuid4()), title="bahra", alternate="bahra", **common)
+        second = Dataset.objects.create(uuid=str(uuid4()), title="bahra", alternate="bahra", **common)
+        self.assertNotEqual(first.pk, second.pk)
+        self.assertEqual(Dataset.objects.filter(workspace="remoteWorkspace", name="bahra").count(), 2)
+
     def test_dataset_styles(self):
         lyr = Dataset.objects.all().first()
         # There should be a total of 3 styles
@@ -296,7 +312,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 +340,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 +367,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/security/tests.py b/geonode/security/tests.py
index 952824c2ff7..d5957a46304 100644
--- a/geonode/security/tests.py
+++ b/geonode/security/tests.py
@@ -1438,7 +1438,8 @@ def test_get_visible_resources_should_return_resource_with_metadata_only_true(se
             actual = get_visible_resources(
                 queryset=layers, metadata_only=True, user=get_user_model().objects.get(username=self.user)
             )
-            self.assertEqual(1, actual.count())
+            # plus "dataset metadata true" from the fixtures
+            self.assertEqual(2, actual.count())
         finally:
             if dataset:
                 dataset.delete()
@@ -1470,7 +1471,8 @@ def test_get_visible_resources_advanced_workflow(self):
         self.assertIsNotNone(standard_user)
         admin_user.is_superuser = True
         admin_user.save()
-        layers = Dataset.objects.all()
+        # get_visible_resources only returns metadata_only=False resources
+        layers = Dataset.objects.filter(metadata_only=False)
 
         actual = get_visible_resources(
             queryset=Dataset.objects.all(),
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..3107ae38afb 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
 
@@ -195,7 +195,8 @@ def _get_indexed_dataset_fields(self, dataset_meta):
         typename = slugify(f"{dataset_meta.id}-{''.join(c for c in dataset_meta.title if ord(c) < 128)}")
         return {
             "name": dataset_meta.title,
-            "store": self.name,
+            # store is a GeoServer grouping: remote datasets have none
+            "store": None,
             "subtype": "remote",
             "workspace": "remoteWorkspace",
             "typename": typename,
@@ -235,7 +236,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..d761ba803b7
--- /dev/null
+++ b/geonode/services/serviceprocessors/cache.py
@@ -0,0 +1,89 @@
+#########################################################################
+#
+# 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"):
+                digest = cls._digest({"username": getattr(wrapped_auth, "username", None) or "-"}.items())
+                return f"auth:{wrapped_auth.__class__.__name__}:{digest}"
+            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..43584d6abad 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)
@@ -214,7 +217,7 @@ def _get_indexed_dataset_fields(self, dataset_meta):
             raise RuntimeError(f"Resource BBOX is not valid: {bbox}")
         return {
             "name": dataset_meta.name,
-            "store": self.name,
+            "store": None,
             "subtype": "remote",
             "workspace": "remoteWorkspace",
             "typename": dataset_meta.name,
@@ -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..cd145e7004f 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,139 @@
 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_does_not_embed_username_for_username_only_auth(self):
+        # Auth objects that expose a bare `username` with no usable __dict__ (e.g.
+        # __slots__-based classes) must still be hashed, not embedded raw in the key.
+        class SlottedAuth:
+            __slots__ = ("username",)
+
+            def __init__(self, username):
+                self.username = username
+
+        phony_url = "http://fake"
+        key_alice = get_service_cache_key(
+            phony_url, service_type=enumerations.WMS, service_id=1, auth=SlottedAuth("alice")
+        )
+        key_bob = get_service_cache_key(phony_url, service_type=enumerations.WMS, service_id=1, auth=SlottedAuth("bob"))
+
+        self.assertNotIn("alice", key_alice)
+        self.assertNotIn("bob", key_bob)
+        self.assertNotEqual(key_alice, key_bob)
+
+    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 +221,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 +640,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 +707,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 +923,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 +1069,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 +1181,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 a7f143c0b29..127ac528580 100755
--- a/geonode/utils.py
+++ b/geonode/utils.py
@@ -1492,7 +1492,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)