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

[josh-ag2]

Problem

Three related issues in S3FileSystem.set_session() that together cause aiohttp Unclosed client session / Unclosed connector warnings when the server closes connections mid-session:

1. Stale _s3creator is silently abandoned on refresh. When set_session(refresh=True) replaces self._s3creator with a new S3BucketRegionCache, the old cache is never closed. It holds aiobotocore-created aiohttp.ClientSession objects (1 general + up to N regional). GC'd without an explicit close() → aiohttp emits warnings.

2. Vacuous-truth refresh on empty _sessions. all(_.closed for _ in []) is always True, so a freshly-entered AIOHTTPSession with _sessions = {} (no requests sent yet) incorrectly triggers refresh=True. This causes the next issue to fire spuriously.

3. Concurrent callers (e.g. asyncio.gather) each create their own creator. When self._s3 is None, multiple concurrent coroutines all fall through to the setup path. Without serialization, Task B's Fix 1 (close stale creator) tears down Task A's creator while Task A is mid-request — setting _sessions = None on an active AIOHTTPSession, producing:

   AttributeError: 'NoneType' object has no attribute 'get'
   

   at aiobotocore/httpsession.py.

The stale-check also needs to guard against newer aiobotocore setting _sessions = None (rather than {}) after __aexit__.

Changes

s3fs/core.py

• Fix 1 — Close stale _s3creator on refresh: Before self._s3creator = s3creator, call await old_creator.__aexit__(None, None, None) so all regional aiobotocore clients and their aiohttp sessions are closed cleanly.

• Fix 2 — Guard stale-session check: Change the stale-session refresh condition to:

  if sessions is not None and sessions and all(_.closed for _ in sessions.values()):
      refresh = True

  • sessions is not None — newer aiobotocore sets _sessions = None after __aexit__; treat None as "not stale, just closed"
  • sessions (non-empty) — an empty dict means no requests have been sent yet; not stale

• Fix 3 — Serialize creator setup under asyncio.Lock: Wrap the entire creator-setup path in async with self._setup_lock. Concurrent tasks wait at the lock, then re-check self._s3 is not None and reuse the already-created session rather than creating their own. This makes Fix 1 safe: the old creator is only closed once, by the task that wins the lock, before any new sessions are open.

s3fs/tests/test_s3fs.py

• test_stale_creator_closed_on_refresh — patches __aexit__ on the old creator and asserts it was called during set_session(refresh=True)

docs/source/changelog.rst — Unreleased section added.

Motivation

Discovered while integrating zarr v3's FsspecStore.close() (zarr-developers/zarr-python#4003). zarr's asyncio.gather fires 4 concurrent FsspecStore.get() calls on the same S3FileSystem instance, triggering issue 3. Issues 1 and 2 produce residual unclosed-session warnings once the render completes.

Related: aiobotocore/aiobotocore#866

Test

All existing tests pass. New test confirms __aexit__ is called on the old creator during refresh.

[martindurant]

OK, I think I have wrapped my head around it now. It's a pity that indenting such a big block messes us the look of the diff so badly; it's only the bit with awaits that needs to go inside the lock block correct? Maybe the kwargs can be worked out beforehand with the same indentation as before.

[martindurant]

Since there was no await between creation of the lock and acquiring it, no other coroutine could have run. The only way the lock might have been acquired is in another thread - but asyncio threads are not thread-safe anyway (and all async code should be running in the one event loop on the one thread)

OK, so the model is, that one coroutine might have run to the awaits below (get_client, __aenter__) while this coroutine is waiting for the lock?

[josh-ag2]

Yes, that's exactly right. A coroutine can yield control at the await get_client(...) and await __aenter__() calls below the lock, allowing a sibling coroutine (e.g. launched via asyncio.gather) to also enter set_session() while the first is suspended. Without the lock they'd each create a separate _s3creator, and when the first resumes and replaces self._s3creator, it calls __aexit__ on the stale one — tearing down a session the sibling is still actively using.

[martindurant]

Prefer to use self._setup_lock = None in __init__.
Note that we don't remove the lock - should it be made in __init__ instead?

[josh-ag2]

Good point on declaring it in __init__ — happy to add self._setup_lock = None there for clarity. The reason the actual asyncio.Lock() creation stays lazy rather than moving entirely to __init__ is that AbstractFileSystem uses instance caching via __new__, so an S3FileSystem instance may be constructed outside any async context and potentially reused across different event loops. Creating the Lock lazily inside set_session() (which is always called from within an async context) ensures it's bound to the right loop. On Python 3.10+ this is less of a concern since locks no longer bind at construction time, but keeping it lazy is the safer cross-version choice.

[martindurant]

On Python 3.10+ this is less of a concern

We don't support 3.9 any more

[martindurant]

Can you check with #1020 - does that also offer the right guard now?