diff --git a/distributed/worker.py b/distributed/worker.py
index 9f940f02..33010836 100644
--- a/distributed/worker.py
+++ b/distributed/worker.py
@@ -996,7 +996,12 @@ class Worker(ServerNode):
             self.status = "closed"
 
             if nanny and "nanny" in self.service_ports:
-                with self.rpc((self.ip, self.service_ports["nanny"])) as r:
+                nanny_address = "%s%s:%d" % (
+                    self.listener.prefix,
+                    self.ip,
+                    self.service_ports["nanny"],
+                )
+                with self.rpc(nanny_address) as r:
                     yield r.terminate()
 
             if self.batched_stream and not self.batched_stream.comm.closed():

`retire_workers` fails to close workers if TLS is enabled
When calling `retire_workers` on a cluster with `TLS` security, the worker is successfully removed from the cluster, but will fail in shutdown and endup in a hung state.

Steps to reproduce:

#### 1. Create Keypair:

```python
# create_keypair.py
from datetime import datetime, timedelta

from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID


def write_keypair():
    """Create a new self-signed certificate & key pair."""
    key = rsa.generate_private_key(public_exponent=65537,
                                   key_size=2048,
                                   backend=default_backend())
    key_bytes = key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption())

    dask_internal = x509.Name(
        [x509.NameAttribute(NameOID.COMMON_NAME, 'dask-internal')]
    )
    now = datetime.utcnow()
    cert = (x509.CertificateBuilder()
                .subject_name(dask_internal)
                .issuer_name(dask_internal)
                .public_key(key.public_key())
                .serial_number(x509.random_serial_number())
                .not_valid_before(now)
                .not_valid_after(now + timedelta(days=365))
                .sign(key, hashes.SHA256(), default_backend()))

    cert_bytes = cert.public_bytes(serialization.Encoding.PEM)

    with open('dask.crt', 'wb') as f:
        f.write(cert_bytes)

    with open('dask.key', 'wb') as f:
        f.write(key_bytes)


if __name__ == '__main__':
    write_keypair()
```

#### 2. Start scheduler

```bash
$ dask-scheduler --tls-ca-file dask.crt --tls-key dask.key --tls-cert dask.crt
```

#### 3. Start worker

```bash
$ dask-worker <SCHEDULER_ADDRESS> --tls-ca-file dask.crt --tls-key dask.key --tls-cert dask.crt
```

#### 4. Create client, and call retire_workers

```python
from distributed.security import Security
from distributed import Client
sec = Security(tls_client_key='dask.key', tls_client_cert='dask.crt', tls_ca_file='dask.crt')
client = Client('<SCHEDULER_ADDRESS>', security=sec)
client.retire_workers(n=1, close_workers=True)
```

#### 5. Observer error from worker during shutdown:

```bash
$ dask-worker tls://192.168.1.5:8786  --tls-ca-file dask.crt --tls-key dask.key --tls-cert dask.crt
distributed.nanny - INFO -         Start Nanny at: 'tls://192.168.1.5:61830'
distributed.worker - INFO -       Start worker at:    tls://192.168.1.5:61831
distributed.worker - INFO -          Listening to:    tls://192.168.1.5:61831
distributed.worker - INFO -              nanny at:          192.168.1.5:61830
distributed.worker - INFO -              bokeh at:          192.168.1.5:61832
distributed.worker - INFO - Waiting to connect to:     tls://192.168.1.5:8786
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -               Threads:                          8
distributed.worker - INFO -                Memory:                   17.18 GB
distributed.worker - INFO -       Local Directory: /Users/jcrist/Code/dask-gateway/worker-n6oomr1q
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -         Registered to:     tls://192.168.1.5:8786
distributed.worker - INFO - -------------------------------------------------
distributed.core - INFO - Starting established connection
distributed.worker - INFO - Stopping worker at tls://192.168.1.5:61831
distributed.comm.tcp - WARNING - Listener on 'tls://192.168.1.5:61830': TLS handshake failed with remote 'tls://192.168.1.5:61841': [SSL: WRONG_VERSION_NUMBER] wrong version number (_ssl.c:1056)
Future exception was never retrieved
future: <Future finished exception=CommClosedError('in <closed TCP>: Stream is closed')>
Traceback (most recent call last):
  File "/Users/jcrist/Code/distributed/distributed/comm/tcp.py", line 181, in read
    n_frames = yield stream.read_bytes(8)
  File "/Users/jcrist/anaconda/envs/dask-gateway/lib/python3.7/site-packages/tornado/gen.py", line 1133, in run
    value = future.result()
tornado.iostream.StreamClosedError: Stream is closed

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/Users/jcrist/anaconda/envs/dask-gateway/lib/python3.7/site-packages/tornado/gen.py", line 1141, in run
    yielded = self.gen.throw(*exc_info)
  File "/Users/jcrist/Code/distributed/distributed/worker.py", line 863, in _close
    yield r.terminate()
  File "/Users/jcrist/anaconda/envs/dask-gateway/lib/python3.7/site-packages/tornado/gen.py", line 1133, in run
    value = future.result()
  File "/Users/jcrist/anaconda/envs/dask-gateway/lib/python3.7/site-packages/tornado/gen.py", line 1141, in run
    yielded = self.gen.throw(*exc_info)
  File "/Users/jcrist/Code/distributed/distributed/core.py", line 657, in send_recv_from_rpc
    result = yield send_recv(comm=comm, op=key, **kwargs)
  File "/Users/jcrist/anaconda/envs/dask-gateway/lib/python3.7/site-packages/tornado/gen.py", line 1133, in run
    value = future.result()
  File "/Users/jcrist/anaconda/envs/dask-gateway/lib/python3.7/site-packages/tornado/gen.py", line 1141, in run
    yielded = self.gen.throw(*exc_info)
  File "/Users/jcrist/Code/distributed/distributed/core.py", line 472, in send_recv
    response = yield comm.read(deserializers=deserializers)
  File "/Users/jcrist/anaconda/envs/dask-gateway/lib/python3.7/site-packages/tornado/gen.py", line 1133, in run
    value = future.result()
  File "/Users/jcrist/anaconda/envs/dask-gateway/lib/python3.7/site-packages/tornado/gen.py", line 1141, in run
    yielded = self.gen.throw(*exc_info)
  File "/Users/jcrist/Code/distributed/distributed/comm/tcp.py", line 201, in read
    convert_stream_closed_error(self, e)
  File "/Users/jcrist/Code/distributed/distributed/comm/tcp.py", line 130, in convert_stream_closed_error
    raise CommClosedError("in %s: %s" % (obj, exc))
distributed.comm.core.CommClosedError: in <closed TCP>: Stream is closed

```
diff --git a/distributed/tests/test_tls_functional.py b/distributed/tests/test_tls_functional.py
index 6b719412..74a9cf3c 100644
--- a/distributed/tests/test_tls_functional.py
+++ b/distributed/tests/test_tls_functional.py
@@ -10,6 +10,8 @@ from tornado import gen
 
 from distributed import Nanny, worker_client, Queue
 from distributed.client import wait
+from distributed.metrics import time
+from distributed.nanny import Nanny
 from distributed.utils_test import gen_tls_cluster, inc, double, slowinc, slowadd
 
 
@@ -157,3 +159,15 @@ def test_worker_client_executor(c, s, a, b):
     future = c.submit(mysum)
     result = yield future
     assert result == 30 * 29
+
+
+@gen_tls_cluster(client=True, Worker=Nanny)
+def test_retire_workers(c, s, a, b):
+    assert set(s.workers) == {a.worker_address, b.worker_address}
+    yield c.retire_workers(workers=[a.worker_address], close_workers=True)
+    assert set(s.workers) == {b.worker_address}
+
+    start = time()
+    while a.status != "closed":
+        yield gen.sleep(0.01)
+        assert time() < start + 5

pip install -e .[dev]
parse_log_pytest
requirements.txt
pytest pandas numpy bokeh requests pyzmq ipython jupyter_client ipykernel prometheus_client jupyter-server-proxy
apt-get update apt-get install -y gcc
3.6
requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
aiohttp==3.8.6
aiosignal==1.2.0
anyio==3.6.2
argon2-cffi==21.3.0
argon2-cffi-bindings==21.2.0
async-generator==1.10
async-timeout==4.0.2
asynctest==0.13.0
attrs==22.2.0
backcall==0.2.0
bleach==4.1.0
bokeh==2.3.3
certifi==2021.5.30
cffi==1.15.1
charset-normalizer==2.0.12
click==8.0.4
cloudpickle==2.2.1
contextvars==2.4
dask==2021.3.0
dataclasses==0.8
decorator==5.1.1
defusedxml==0.7.1
-e git+https://github.com/dask/distributed.git@e0cf7e7300c9dcf10b7440abb1e3efc6cea3a91a#egg=distributed
entrypoints==0.4
frozenlist==1.2.0
HeapDict==1.0.1
idna==3.10
idna-ssl==1.1.0
immutables==0.19
importlib-metadata==4.8.3
iniconfig==1.1.1
ipykernel==5.5.6
ipython==7.16.3
ipython-genutils==0.2.0
jedi==0.17.2
Jinja2==3.0.3
jsonschema==3.2.0
jupyter-client==7.1.2
jupyter-core==4.9.2
jupyter-server==1.13.1
jupyter-server-proxy==3.2.4
jupyterlab-pygments==0.1.2
MarkupSafe==2.0.1
mistune==0.8.4
msgpack==1.0.5
multidict==5.2.0
nbclient==0.5.9
nbconvert==6.0.7
nbformat==5.1.3
nest-asyncio==1.6.0
numpy==1.19.5
packaging==21.3
pandas==1.1.5
pandocfilters==1.5.1
parso==0.7.1
pexpect==4.9.0
pickleshare==0.7.5
Pillow==8.4.0
pluggy==1.0.0
prometheus-client==0.17.1
prompt-toolkit==3.0.36
psutil==7.0.0
ptyprocess==0.7.0
py==1.11.0
pycparser==2.21
Pygments==2.14.0
pyparsing==3.1.4
pyrsistent==0.18.0
pytest==7.0.1
python-dateutil==2.9.0.post0
pytz==2025.2
PyYAML==6.0.1
pyzmq==25.1.2
requests==2.27.1
Send2Trash==1.8.3
simpervisor==0.4
six==1.17.0
sniffio==1.2.0
sortedcontainers==2.4.0
tblib==1.7.0
terminado==0.12.1
testpath==0.6.0
tomli==1.2.3
toolz==0.12.0
tornado==6.1
traitlets==4.3.3
typing_extensions==4.1.1
urllib3==1.26.20
wcwidth==0.2.13
webencodings==0.5.1
websocket-client==1.3.1
yarl==1.7.2
zict==2.1.0
zipp==3.6.0

name: distributed
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - aiohttp==3.8.6
      - aiosignal==1.2.0
      - anyio==3.6.2
      - argon2-cffi==21.3.0
      - argon2-cffi-bindings==21.2.0
      - async-generator==1.10
      - async-timeout==4.0.2
      - asynctest==0.13.0
      - attrs==22.2.0
      - backcall==0.2.0
      - bleach==4.1.0
      - bokeh==2.3.3
      - cffi==1.15.1
      - charset-normalizer==2.0.12
      - click==8.0.4
      - cloudpickle==2.2.1
      - contextvars==2.4
      - dask==2021.3.0
      - dataclasses==0.8
      - decorator==5.1.1
      - defusedxml==0.7.1
      - entrypoints==0.4
      - frozenlist==1.2.0
      - heapdict==1.0.1
      - idna==3.10
      - idna-ssl==1.1.0
      - immutables==0.19
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - ipykernel==5.5.6
      - ipython==7.16.3
      - ipython-genutils==0.2.0
      - jedi==0.17.2
      - jinja2==3.0.3
      - jsonschema==3.2.0
      - jupyter-client==7.1.2
      - jupyter-core==4.9.2
      - jupyter-server==1.13.1
      - jupyter-server-proxy==3.2.4
      - jupyterlab-pygments==0.1.2
      - markupsafe==2.0.1
      - mistune==0.8.4
      - msgpack==1.0.5
      - multidict==5.2.0
      - nbclient==0.5.9
      - nbconvert==6.0.7
      - nbformat==5.1.3
      - nest-asyncio==1.6.0
      - numpy==1.19.5
      - packaging==21.3
      - pandas==1.1.5
      - pandocfilters==1.5.1
      - parso==0.7.1
      - pexpect==4.9.0
      - pickleshare==0.7.5
      - pillow==8.4.0
      - pluggy==1.0.0
      - prometheus-client==0.17.1
      - prompt-toolkit==3.0.36
      - psutil==7.0.0
      - ptyprocess==0.7.0
      - py==1.11.0
      - pycparser==2.21
      - pygments==2.14.0
      - pyparsing==3.1.4
      - pyrsistent==0.18.0
      - pytest==7.0.1
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - pyyaml==6.0.1
      - pyzmq==25.1.2
      - requests==2.27.1
      - send2trash==1.8.3
      - simpervisor==0.4
      - six==1.17.0
      - sniffio==1.2.0
      - sortedcontainers==2.4.0
      - tblib==1.7.0
      - terminado==0.12.1
      - testpath==0.6.0
      - tomli==1.2.3
      - toolz==0.12.0
      - tornado==6.1
      - traitlets==4.3.3
      - typing-extensions==4.1.1
      - urllib3==1.26.20
      - wcwidth==0.2.13
      - webencodings==0.5.1
      - websocket-client==1.3.1
      - yarl==1.7.2
      - zict==2.1.0
      - zipp==3.6.0
prefix: /opt/conda/envs/distributed

distributed/tests/test_tls_functional.py::test_retire_workers
distributed/tests/test_tls_functional.py::test_client_submit distributed/tests/test_tls_functional.py::test_gather distributed/tests/test_tls_functional.py::test_scatter distributed/tests/test_tls_functional.py::test_nanny distributed/tests/test_tls_functional.py::test_rebalance distributed/tests/test_tls_functional.py::test_work_stealing distributed/tests/test_tls_functional.py::test_worker_client distributed/tests/test_tls_functional.py::test_worker_client_gather distributed/tests/test_tls_functional.py::test_worker_client_executor
distributed/tests/test_tls_functional.py::test_Queue
distributed/worker.py
distributed/worker.py