Use coordinated docs port with conflict checks and deployment-independent cache keys

scripts/ci.py

@@ -13,7 +13,8 @@ import time
 ROOT = Path(__file__).resolve().parents[1]
 APP = "endorphin-tabularium-docs"
 INPUTS = ["Dockerfile", "Dockerfile.ci", ".dockerignore", ".npmrc", "package.json",
-          "pnpm-lock.yaml", "pnpm-workspace.yaml", "apps/docs", "content", "tools/figures", "scripts"]
+          "pnpm-lock.yaml", "pnpm-workspace.yaml", "apps/docs", "content", "tools/figures",
+          "scripts/check-container.sh", "scripts/ci.py"]
 
 
 def run(*args: str, capture: bool = False) -> str:
@@ -82,6 +83,7 @@ def main() -> None:
                     "--mount", f"type=volume,src={APP}-{scope}-pnpm,dst=/pnpm/store",
                     "--mount", f"type=volume,src={APP}-{scope}-uv,dst=/root/.cache/uv",
                     "--mount", f"type=volume,src={APP}-{scope}-next,dst=/workspace/apps/docs/.next/cache",
+                    "--mount", f"type=volume,src={APP}-{scope}-swc,dst=/root/.cache/next-swc",
                     checks)
                 run("docker", "cp", f"{name}:/workspace/apps/docs/.next/standalone/.", str(output))
                 app = output / "apps/docs"

scripts/deploy.py

@@ -5,6 +5,7 @@ import os
 from pathlib import Path
 import re
 import shutil
+import socket
 import subprocess
 import urllib.request
 from ci import APP, ROOT, fingerprint, run
@@ -29,6 +30,25 @@ def compose(release: Path, *args: str) -> None:
         "--env-file", str(release / "release.env"), "-f", str(release / "compose.yml"), *args)
 
 
+def check_port(port: int) -> None:
+    # An existing container belonging to this Compose service may retain its port.
+    own = run("docker", "ps", "-q", "--filter", f"label=com.docker.compose.project={APP}",
+              "--filter", "label=com.docker.compose.service=docs", capture=True)
+    if own:
+        for container in own.splitlines():
+            details = json.loads(run("docker", "inspect", container, capture=True))[0]
+            ports = details["NetworkSettings"]["Ports"].get("3000/tcp") or []
+            if any(item["HostPort"] == str(port) for item in ports):
+                print(f"Port {port} belongs to the current Tabularium service", flush=True)
+                return
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
+        try:
+            probe.bind(("0.0.0.0", port))
+        except OSError as error:
+            raise RuntimeError(f"Port {port} is already occupied; no service was changed") from error
+    print(f"Port {port} is available", flush=True)
+
+
 def main() -> None:
     if os.geteuid() != 0:
         raise RuntimeError("Deployment requires root on ces-control")
@@ -48,13 +68,14 @@ def main() -> None:
         run("docker", "image", "inspect", actual, capture=True)
         CONFIG.parent.mkdir(parents=True, exist_ok=True)
         if not CONFIG.exists():
-            CONFIG.write_text("TABULARIUM_BIND=0.0.0.0\nTABULARIUM_PORT=18132\n")
+            CONFIG.write_text("TABULARIUM_BIND=0.0.0.0\nTABULARIUM_PORT=18135\n")
             CONFIG.chmod(0o600)
         settings = dict(line.split("=", 1) for line in CONFIG.read_text().splitlines()
                         if line and not line.startswith("#"))
-        port = int(settings.get("TABULARIUM_PORT", "18132"))
+        port = int(settings.get("TABULARIUM_PORT", "18135"))
         if not 1024 <= port <= 65535:
             raise ValueError("Invalid port")
+        check_port(port)
         current = DEPLOY_ROOT / "current"
         previous = current.resolve() if current.is_symlink() else None
         release = DEPLOY_ROOT / "releases" / sha

scripts/test_ci.py

@@ -24,6 +24,9 @@ class FingerprintTests(unittest.TestCase):
                 cache = root / "apps/docs/.next/cache/state"
                 cache.parent.mkdir(parents=True)
                 cache.write_text("compiler cache\n")
+                deploy = root / "scripts/deploy.py"
+                deploy.parent.mkdir(parents=True)
+                deploy.write_text("# deployment-only configuration\n")
                 self.assertEqual(original, ci.fingerprint())
                 doc.write_text("# Updated\n")
                 self.assertNotEqual(original, ci.fingerprint())

scripts/test_deploy.py

@@ -0,0 +1,32 @@
+"""Ensure port coordination never treats another service's port as ours."""
+import json
+import unittest
+from unittest.mock import MagicMock, patch
+import deploy
+
+
+class PortGuardTests(unittest.TestCase):
+    def test_foreign_listener_blocks_deployment(self):
+        probe = MagicMock()
+        probe.__enter__.return_value.bind.side_effect = OSError("Address already in use")
+        with patch.object(deploy, "run", return_value=""), patch.object(deploy.socket, "socket", return_value=probe):
+            with self.assertRaisesRegex(RuntimeError, "already occupied"):
+                deploy.check_port(18135)
+
+    def test_current_service_can_keep_its_port(self):
+        details = [{"NetworkSettings": {"Ports": {"3000/tcp": [{"HostPort": "18135"}]}}}]
+        with patch.object(deploy, "run", side_effect=["our-container", json.dumps(details)]), patch.object(deploy.socket, "socket") as probe:
+            deploy.check_port(18135)
+            probe.assert_not_called()
+
+    def test_current_service_on_another_port_does_not_bypass_conflict(self):
+        details = [{"NetworkSettings": {"Ports": {"3000/tcp": [{"HostPort": "18134"}]}}}]
+        probe = MagicMock()
+        probe.__enter__.return_value.bind.side_effect = OSError("Address already in use")
+        with patch.object(deploy, "run", side_effect=["our-container", json.dumps(details)]), patch.object(deploy.socket, "socket", return_value=probe):
+            with self.assertRaisesRegex(RuntimeError, "already occupied"):
+                deploy.check_port(18135)
+
+
+if __name__ == "__main__":
+    unittest.main()