fix: close stale S3BucketRegionCache on session refresh; guard _sessions=None; serialize set_session() under asyncio.Lock

docs/source/changelog.rst

@@ -1,6 +1,20 @@
 Changelog
 =========
 
+Unreleased
+----------
+
+- close stale ``S3BucketRegionCache`` on session refresh, eliminating
+  ``aiohttp`` "Unclosed client session" warnings when the server closes
+  connections mid-session (#NNNN)
+- guard stale-session check against ``_sessions = None`` (newer aiobotocore
+  sets ``_sessions`` to ``None`` after close rather than leaving it as an
+  empty dict) and against empty ``_sessions`` (no requests yet, not stale)
+  (#NNNN)
+- serialize ``set_session()`` creator setup under an ``asyncio.Lock`` to
+  prevent concurrent callers (e.g. ``asyncio.gather``) from each creating a
+  separate creator and then closing each other's in-flight sessions (#NNNN)
+
 2026.4.0
 --------
 

s3fs/core.py

@@ -425,6 +425,7 @@ def __init__(
         self.use_ssl = use_ssl
         self.cache_regions = cache_regions
         self._s3 = None
+        self._setup_lock = asyncio.Lock()
         self.session = session
         self.fixed_upload_size = fixed_upload_size
         self.local_expiry_check = local_expiry_check
@@ -599,12 +600,24 @@ async def set_session(self, refresh=False, kwargs={}):
         if self._s3 is not None and not refresh:
             hsess = getattr(getattr(self._s3, "_endpoint", None), "http_session", None)
             if hsess is not None:
-                if all(_.closed for _ in hsess._sessions.values()):
+                sessions = hsess._sessions  # None after __aexit__ in newer aiobotocore
+                # Treat sessions=None (aiobotocore 3.x post-close) and all
+                # server-closed sessions as stale.  Empty dict = not yet used.
+                if sessions is None or (
+                    sessions and all(_.closed for _ in sessions.values())
+                ):
                     refresh = True
             if not refresh:
                 return self._s3
-        logger.debug("Setting up s3fs instance")
 
+        # Serialize creator setup.  Callers such as zarr may launch concurrent
+        # coroutines via asyncio.gather that all see self._s3 is None and each
+        # try to build their own creator.  Without this lock every concurrent
+        # task would create a separate creator and then Fix 1 (close stale
+        # creator) would tear down a creator that a sibling task is still using,
+        # causing ``_sessions = None`` mid-request.
+        # Compute kwargs before the lock — pure reads, no awaits, safe to
+        # redo if another coroutine wins the race and we return early inside.
         client_kwargs = self.client_kwargs.copy()
         init_kwargs = dict(
             aws_access_key_id=self.key,
@@ -639,41 +652,65 @@ async def set_session(self, refresh=False, kwargs={}):
             config_kwargs["signature_version"] = UNSIGNED
 
         conf = AioConfig(**config_kwargs)
-        if self.session is None or refresh:
-            self.session = aiobotocore.session.AioSession(**self.kwargs)
 
-        for parameters in (config_kwargs, self.kwargs, init_kwargs, client_kwargs):
-            for option in ("region_name", "endpoint_url"):
-                if parameters.get(option):
-                    self.cache_regions = False
-                    break
-        else:
-            cache_regions = self.cache_regions
+        async with self._setup_lock:
+            # Re-check under the lock: a concurrent task may have set up the
+            # session while we were waiting.
+            if self._s3 is not None and not refresh:
+                return self._s3
 
-        logger.debug(
-            "RC: caching enabled? %r (explicit option is %r)",
-            cache_regions,
-            self.cache_regions,
-        )
-        self.cache_regions = cache_regions
-        if self.cache_regions:
-            s3creator = S3BucketRegionCache(
-                self.session, config=conf, **init_kwargs, **client_kwargs
-            )
-            self._s3 = await s3creator.get_client()
-        else:
-            s3creator = self.session.create_client(
-                "s3", config=conf, **init_kwargs, **client_kwargs
+            logger.debug("Setting up s3fs instance")
+
+            if self.session is None or refresh:
+                self.session = aiobotocore.session.AioSession(**self.kwargs)
+
+            for parameters in (config_kwargs, self.kwargs, init_kwargs, client_kwargs):
+                for option in ("region_name", "endpoint_url"):
+                    if parameters.get(option):
+                        self.cache_regions = False
+                        break
+            else:
+                cache_regions = self.cache_regions
+
+            logger.debug(
+                "RC: caching enabled? %r (explicit option is %r)",
+                cache_regions,
+                self.cache_regions,
             )
-            self._s3 = await s3creator.__aenter__()
-
-        self._s3creator = s3creator
-        # the following actually closes the aiohttp connection; use of privates
-        # might break in the future, would cause exception at gc time
-        if not self.asynchronous:
-            weakref.finalize(self, self.close_session, self.loop, self._s3creator)
-        self._kwargs_helper = ParamKwargsHelper(self._s3)
-        return self._s3
+            self.cache_regions = cache_regions
+            if self.cache_regions:
+                s3creator = S3BucketRegionCache(
+                    self.session, config=conf, **init_kwargs, **client_kwargs
+                )
+                self._s3 = await s3creator.get_client()
+            else:
+                s3creator = self.session.create_client(
+                    "s3", config=conf, **init_kwargs, **client_kwargs
+                )
+                self._s3 = await s3creator.__aenter__()
+
+            # Close the stale _s3creator before replacing it.  The old cache holds
+            # aiobotocore ClientSessions that are garbage-collected without an
+            # explicit close(), causing aiohttp "Unclosed client session" warnings.
+            # See: https://github.com/aio-libs/aiobotocore/issues/866
+            old_creator = getattr(self, "_s3creator", None)
+            if old_creator is not None:
+                try:
+                    await old_creator.__aexit__(None, None, None)
+                except Exception:
+                    pass
+            self._s3creator = s3creator
+            # the following actually closes the aiohttp connection; use of privates
+            # might break in the future, would cause exception at gc time
+            if not self.asynchronous:
+                old_finalizer = getattr(self, "_finalizer", None)
+                if old_finalizer is not None:
+                    old_finalizer.detach()
+                self._finalizer = weakref.finalize(
+                    self, self.close_session, self.loop, self._s3creator
+                )
+            self._kwargs_helper = ParamKwargsHelper(self._s3)
+            return self._s3
 
     _connect = set_session
 

s3fs/tests/test_s3fs.py

@@ -3189,6 +3189,38 @@ async def run_program(run):
     asyncio.run(run_program(False))
 
 
+def test_stale_creator_closed_on_refresh(s3):
+    """Old _s3creator must have __aexit__ called when set_session refreshes.
+
+    Without the fix, the stale aiobotocore ClientSession held by the old
+    _s3creator is garbage-collected without explicit close(), producing
+    aiohttp 'Unclosed client session' warnings at process exit.
+    """
+
+    async def run():
+        await s3.set_session()
+        assert s3._s3creator is not None
+
+        # Replace the real creator with a spy that records whether __aexit__
+        # is called.  The real client is already in s3._s3 so this is safe.
+        class _SpyCreator:
+            exited = False
+
+            async def __aexit__(self, *args):
+                _SpyCreator.exited = True
+
+        s3._s3creator = _SpyCreator()
+        await s3.set_session(refresh=True)
+
+        assert _SpyCreator.exited, (
+            "Old _s3creator.__aexit__ was not called during session refresh; "
+            "aiohttp sessions will leak"
+        )
+
+    asyncio.run(run())
+
+
+
 def test_rm_recursive_prfix(s3):
     prefix = "logs/"  # must end with "/"