test(fabric): bound port manager queue tests and update warning log

Author: littlebullGit

src/lightning/fabric/utilities/port_manager.py

@@ -233,8 +233,9 @@ def allocate_port(self, preferred_port: Optional[int] = None, max_attempts: int
 
                     queue_count = len(state.recently_released)
                     if queue_count > 800:  # >78% of typical 1024 capacity
+                        utilization_pct = (queue_count / _RECENTLY_RELEASED_PORTS_MAXLEN) * 100
                         log.warning(
-                            f"Port queue utilization high: {queue_count} entries. "
+                            f"Port queue utilization high: {queue_count} entries ({utilization_pct:.1f}% of capacity). "
                             f"Allocated port {port}. Active allocations: {len(state.allocated_ports)}"
                         )
 

tests/tests_fabric/utilities/test_port_manager.py

@@ -13,15 +13,20 @@
 # limitations under the License.
 """Tests for the PortManager utility and port allocation integration."""
 
+import collections
 import os
 import socket
 import threading
-from collections import Counter
 
 import pytest
 
-from lightning.fabric.plugins.environments.lightning import find_free_network_port
-from lightning.fabric.utilities.port_manager import PortManager, get_port_manager
+import lightning.fabric.utilities.port_manager as port_manager_module
+import lightning.fabric.utilities.port_state as port_state_module
+from lightning.fabric.utilities.port_manager import (
+    PortManager,
+    find_free_network_port,
+    get_port_manager,
+)
 
 # =============================================================================
 # Fixtures
@@ -143,7 +148,7 @@ def allocate_ports():
     assert len(set(ports)) == 100, f"Expected 100 unique ports, got {len(set(ports))}"
 
     # Check for any duplicates
-    counts = Counter(ports)
+    counts = collections.Counter(ports)
     duplicates = {port: count for port, count in counts.items() if count > 1}
     assert not duplicates, f"Found duplicate ports: {duplicates}"
 
@@ -714,9 +719,14 @@ def test_port_manager_recently_released_prevents_immediate_reuse():
     manager.release_port(new_port)
 
 
-def test_port_manager_recently_released_queue_cycles():
+def _set_recently_released_limit(monkeypatch, value: int) -> None:
+    monkeypatch.setattr(port_manager_module, "_RECENTLY_RELEASED_PORTS_MAXLEN", value, raising=True)
+    monkeypatch.setattr(port_state_module, "_RECENTLY_RELEASED_MAX_LEN", value, raising=True)
+
+
+def test_port_manager_recently_released_queue_cycles(monkeypatch):
     """Test that recently_released queue cycles after maxlen allocations."""
-    from lightning.fabric.utilities.port_manager import _RECENTLY_RELEASED_PORTS_MAXLEN
+    _set_recently_released_limit(monkeypatch, 64)
 
     manager = PortManager()
 
@@ -727,8 +737,10 @@ def test_port_manager_recently_released_queue_cycles():
     # Port should be in recently_released queue
     assert first_port in manager._recently_released
 
+    queue_limit = port_manager_module._RECENTLY_RELEASED_PORTS_MAXLEN
+
     # Allocate and release many ports to fill the queue beyond maxlen
-    for _ in range(_RECENTLY_RELEASED_PORTS_MAXLEN + 10):
+    for _ in range(queue_limit + 10):
         port = manager.allocate_port()
         manager.release_port(port)
 
@@ -755,14 +767,20 @@ def test_port_manager_reserve_clears_recently_released():
     manager.release_port(port)
 
 
-def test_port_manager_high_queue_utilization_warning(caplog):
+def test_port_manager_high_queue_utilization_warning(monkeypatch, caplog):
     """Test that warning is logged when queue utilization exceeds 80%."""
     import logging
 
+    _set_recently_released_limit(monkeypatch, 64)
+
+    queue_limit = port_manager_module._RECENTLY_RELEASED_PORTS_MAXLEN
+    trigger_count = int(queue_limit * 0.8) + 1  # Just over 80%
+    expected_pct = (trigger_count / queue_limit) * 100
+
     manager = PortManager()
 
-    # Fill queue to >80% (821/1024 = 80.2%)
-    for _ in range(821):
+    # Fill queue to just over 80%
+    for _ in range(trigger_count):
         port = manager.allocate_port()
         manager.release_port(port)
 
@@ -773,4 +791,4 @@ def test_port_manager_high_queue_utilization_warning(caplog):
 
     # Verify warning was logged
     assert any("Port queue utilization high" in record.message for record in caplog.records)
-    assert any("80." in record.message for record in caplog.records)  # Should show 80.x%
+    assert any(f"{expected_pct:.1f}%" in record.message for record in caplog.records)

tests/tests_fabric/utilities/test_port_manager_process_safe.py

@@ -22,6 +22,7 @@
 
 import pytest
 
+import lightning.fabric.utilities.port_manager as port_manager_module
 from lightning.fabric.utilities.file_lock import UnixFileLock, WindowsFileLock, create_file_lock
 from lightning.fabric.utilities.port_manager import PortManager, _get_lock_dir, _get_lock_file
 from lightning.fabric.utilities.port_state import PortAllocation, PortState
@@ -106,6 +107,43 @@ def test_file_lock_context_manager_timeout(tmpdir):
     lock1.release()
 
 
+def test_get_lock_dir_handles_permission_error(monkeypatch, tmp_path):
+    """_get_lock_dir should tolerate probe unlink permission errors and register cleanup."""
+
+    monkeypatch.setenv("LIGHTNING_PORT_LOCK_DIR", str(tmp_path))
+
+    registered_calls = []
+
+    def fake_register(func, *args, **kwargs):
+        registered_calls.append((func, args, kwargs))
+        return func
+
+    monkeypatch.setattr(port_manager_module.atexit, "register", fake_register)
+
+    original_unlink = Path.unlink
+    call_state = {"count": 0}
+
+    def fake_unlink(self, *args, **kwargs):
+        if self.name.startswith(".lightning_port_manager_write_test_") and call_state["count"] == 0:
+            call_state["count"] += 1
+            raise PermissionError("locked")
+        return original_unlink(self, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", fake_unlink)
+
+    lock_dir = _get_lock_dir()
+    assert Path(lock_dir) == tmp_path
+    assert registered_calls, "Cleanup should be registered when unlink fails"
+
+    cleanup_func, args, kwargs = registered_calls[0]
+    probe_path = args[0]
+    assert isinstance(probe_path, Path)
+    assert probe_path.exists()
+
+    cleanup_func(*args, **kwargs)
+    assert not probe_path.exists()
+
+
 # =============================================================================
 # Tests for PortState
 # =============================================================================