fix: PUC-1725 DuplicateSubPort error when adding subnet to router if previous attempt failed

python/neutron-understack/neutron_understack/routers.py

@@ -6,6 +6,7 @@
 from neutron.objects.network import NetworkSegment
 from neutron.objects.ports import Port
 from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client import OVNClient
+from neutron.services.trunk import exceptions as trunk_exc
 from neutron_lib import constants as p_const
 from neutron_lib import context as n_context
 from neutron_lib.api.definitions import segment as segment_def
@@ -97,11 +98,18 @@ def add_subport_to_trunk(shared_port: PortDict, segment: NetworkSegmentDict) ->
     }
     trunk_id = utils.fetch_network_node_trunk_id()
 
-    utils.fetch_trunk_plugin().add_subports(
-        context=n_context.get_admin_context(),
-        trunk_id=trunk_id,
-        subports=subports,
-    )
+    try:
+        utils.fetch_trunk_plugin().add_subports(
+            context=n_context.get_admin_context(),
+            trunk_id=trunk_id,
+            subports=subports,
+        )
+    except trunk_exc.DuplicateSubPort:
+        LOG.debug(
+            "subport with segmentation_id %(seg_id)s already exists on trunk "
+            "%(trunk_id)s, skipping",
+            {"seg_id": segment["segmentation_id"], "trunk_id": trunk_id},
+        )
 
 
 def fetch_or_create_router_segment(context: PortContext) -> NetworkSegmentDict:

python/neutron-understack/neutron_understack/tests/test_routers.py

@@ -1,4 +1,5 @@
 import pytest
+from neutron.services.trunk import exceptions as trunk_exc
 from neutron_lib import constants as p_const
 
 from neutron_understack.routers import add_subport_to_trunk
@@ -66,6 +67,25 @@ def test_when_successful(self, mocker):
             },
         )
 
+    def test_duplicate_subport_is_ignored(self, mocker):
+        """A stranded sub-port from a previous failed attempt must not block retries."""
+        mocker.patch(
+            "neutron_understack.utils.fetch_network_node_trunk_id",
+            return_value="trunk-uuid",
+        )
+        mocker.patch(
+            "neutron_lib.context.get_admin_context", return_value="admin_context"
+        )
+        mock_trunk_plugin = mocker.Mock()
+        mock_trunk_plugin.add_subports.side_effect = trunk_exc.DuplicateSubPort()
+        mocker.patch(
+            "neutron_understack.utils.fetch_trunk_plugin",
+            return_value=mock_trunk_plugin,
+        )
+
+        # should not raise
+        add_subport_to_trunk({"id": "port-123"}, {"segmentation_id": 42})
+
 
 class TestHandleSubportRemoval:
     def test_when_successful(self, mocker, port_id, trunk_id):