fix: UDP port allocation race causing link self-loop (one-way links)

PortManager.get_free_udp_port had an unguarded find-then-add sequence.
A link allocates both ends concurrently (asyncio.gather in
UDPLink._prepare -> two POST /ports/udp) and FastAPI runs the sync
route handler in a threadpool, so both threads could probe the same
'free' port before either recorded it — handing lport == rport to both
ends. uBridge sets SO_REUSEADDR on UDP NIO sockets, so the double bind
succeeds silently and the kernel delivers everything to the last-bound
socket: one node starves, the other echoes to itself.

Make every TCP/UDP allocate/reserve/release path atomic with an RLock,
and rebuild _link_data in UDPLink._prepare so reset() commits the fresh
port pair instead of re-sending the stale, already-released one.

Regression tests: threaded barrier allocation never returns duplicates
(red on the old code, UDP and TCP); reset() leaves exactly one mirrored
NIO pair per side with lport != rport (red on the old code).

tests/compute/test_port_manager.py

@@ -16,6 +16,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import pytest
+import threading
 import uuid
 
 from fastapi import HTTPException
@@ -116,6 +117,82 @@ def test_release_udp_port():
     pm.reserve_udp_port(20000, project)
 
 
+def test_concurrent_udp_port_allocation_no_duplicates():
+    """
+    Regression test for the link UDP self-loop bug (docs/bugs/link-udp-self-loop.md):
+    both ends of a link are allocated concurrently on the controller
+    (asyncio.gather -> two POST /ports/udp), and FastAPI runs the sync route
+    handler in a threadpool. The find-then-add allocation must be atomic,
+    otherwise both threads can probe and return the same "free" port —
+    handing lport == rport to both ends, which makes every packet loop back
+    to its sender (one-way link).
+    """
+
+    pm = PortManager()
+    pm.udp_port_range = (50000, 50100)
+    project = Project(project_id=str(uuid.uuid4()))
+
+    workers = 8
+    rounds = 10
+    barrier = threading.Barrier(workers)
+    results = []
+    results_lock = threading.Lock()
+
+    def worker():
+        allocated = []
+        for _ in range(rounds):
+            # start each round together to maximize the collision window
+            barrier.wait()
+            allocated.append(pm.get_free_udp_port(project))
+        with results_lock:
+            results.extend(allocated)
+
+    threads = [threading.Thread(target=worker) for _ in range(workers)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert len(results) == workers * rounds
+    assert len(set(results)) == len(results), "the same UDP port was handed to two callers"
+    assert pm.udp_ports == set(results)
+
+
+def test_concurrent_tcp_port_allocation_no_duplicates():
+    """
+    Same race class as the UDP self-loop bug, on the console/TCP side:
+    concurrent get_free_tcp_port calls must never return the same port.
+    """
+
+    pm = PortManager()
+    pm.console_port_range = (51000, 51100)
+    project = Project(project_id=str(uuid.uuid4()))
+
+    workers = 8
+    rounds = 10
+    barrier = threading.Barrier(workers)
+    results = []
+    results_lock = threading.Lock()
+
+    def worker():
+        allocated = []
+        for _ in range(rounds):
+            barrier.wait()
+            allocated.append(pm.get_free_tcp_port(project))
+        with results_lock:
+            results.extend(allocated)
+
+    threads = [threading.Thread(target=worker) for _ in range(workers)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert len(results) == workers * rounds
+    assert len(set(results)) == len(results), "the same TCP port was handed to two callers"
+    assert pm.tcp_ports == set(results)
+
+
 def test_find_unused_port():
 
     p = PortManager().find_unused_port(1000, 10000)

tests/controller/test_udp_link.py

@@ -187,6 +187,78 @@ async def test_delete(project):
     compute2.delete.assert_any_call("/projects/{}/vpcs/nodes/{}/adapters/3/ports/1/nio".format(project.id, node2.id), timeout=120)
 
 
+@pytest.mark.asyncio
+async def test_reset(project):
+    """
+    reset() re-creates the link on the same object: the fresh port pair must
+    replace the stale one instead of accumulating (the committed NIOs always
+    come from indices 0/1) — see docs/bugs/link-udp-self-loop.md.
+    """
+
+    compute1 = MagicMock()
+    compute2 = MagicMock()
+
+    node1 = Node(project, compute1, "node1", node_type="vpcs")
+    node1._ports = [EthernetPort("E0", 0, 0, 4)]
+    node2 = Node(project, compute2, "node2", node_type="vpcs")
+    node2._ports = [EthernetPort("E0", 0, 3, 1)]
+
+    async def subnet_callback(compute2):
+        """
+        Fake subnet callback
+        """
+        return ("192.168.1.1", "192.168.1.2")
+
+    compute1.get_ip_on_same_subnet.side_effect = subnet_callback
+
+    # per-compute port sequences: first create -> 1024/2048, reset -> 4096/8192
+    node1_ports = iter([1024, 4096])
+    node2_ports = iter([2048, 8192])
+
+    async def compute1_callback(path, data={}, **kwargs):
+        if "/ports/udp" in path:
+            response = MagicMock()
+            response.json = {"udp_port": next(node1_ports)}
+            return response
+
+    async def compute2_callback(path, data={}, **kwargs):
+        if "/ports/udp" in path:
+            response = MagicMock()
+            response.json = {"udp_port": next(node2_ports)}
+            return response
+
+    compute1.post.side_effect = compute1_callback
+    compute1.host = "example.com"
+    compute2.post.side_effect = compute2_callback
+    compute2.host = "example.org"
+
+    link = UDPLink(project)
+    await link.add_node(node1, 0, 4)
+    await link.add_node(node2, 3, 1)
+
+    await link.reset()
+
+    # exactly one (fresh) NIO spec per side — no stale entries left behind
+    assert len(link.debug_link_data) == 2
+    assert link.debug_link_data[0]["lport"] == 4096
+    assert link.debug_link_data[0]["rport"] == 8192
+    assert link.debug_link_data[1]["lport"] == 8192
+    assert link.debug_link_data[1]["rport"] == 4096
+    # the self-loop invariant: an end's lport must never equal its rport
+    assert link.debug_link_data[0]["lport"] != link.debug_link_data[0]["rport"]
+    assert link.debug_link_data[1]["lport"] != link.debug_link_data[1]["rport"]
+    # the committed NIO carries the fresh pair, not the released one
+    compute1.post.assert_any_call("/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/nio".format(project.id, node1.id), data={
+        "lport": 4096,
+        "rhost": "192.168.1.2",
+        "rport": 8192,
+        "type": "nio_udp",
+        "filters": {},
+        "markers": {},
+        "suspend": False,
+    }, timeout=120)
+
+
 @pytest.mark.asyncio
 async def test_choose_capture_side(project):
     """