From 09f6ea8291275ea2b60a25016a8a8445118f3390 Mon Sep 17 00:00:00 2001 From: yuxin00j Date: Mon, 2 Nov 2026 09:31:50 +0000 Subject: [PATCH 1/5] fix(zonal): release MRD pool when ZonalFile initialization fails --- gcsfs/tests/test_zonal_file.py | 38 ++++++++++++++++++++ gcsfs/zonal_file.py | 63 ++++++++++++++++++++-------------- 2 files changed, 76 insertions(+), 25 deletions(-) diff --git a/gcsfs/tests/test_zonal_file.py b/gcsfs/tests/test_zonal_file.py index 965e80b40..ead1e1c68 100644 --- a/gcsfs/tests/test_zonal_file.py +++ b/gcsfs/tests/test_zonal_file.py @@ -1074,6 +1074,44 @@ async def coro_func(): assert result == ["done"] +def test_zonal_file_failed_init_closes_mrd_pool(mock_gcsfs): + """If GCSFile.__init__ fails after the MRD pool is checked out, the pool is closed.""" + mock_gcsfs.loop = fsspec.asyn.get_loop() + pool = mock_gcsfs._mrd_pool_cache.get.return_value + pool.close = mock.AsyncMock() + zf = ZonalFile.__new__(ZonalFile) + + with pytest.raises(KeyError): + zf.__init__( + gcsfs=mock_gcsfs, + path="gs://test-bucket/test-key", + mode="rb", + cache_type="no-such-cache", + ) + + mock_gcsfs._mrd_pool_cache.get.assert_awaited_once() + pool.close.assert_awaited_once() + assert zf.mrd_pool is None + assert zf.closed is True + + +@pytest.mark.parametrize("mode", ["r+", "rb+", "w+", "wb+", "a+", "ab+", "x+", "xb+"]) +def test_zonal_file_plus_modes_rejected_before_pool_or_super_init(mock_gcsfs, mode): + """Unsupported '+' modes fail before checking out an MRD pool or arming __del__.""" + mock_gcsfs.loop = fsspec.asyn.get_loop() + zf = ZonalFile.__new__(ZonalFile) + + with pytest.raises(NotImplementedError): + zf.__init__( + gcsfs=mock_gcsfs, + path="gs://test-bucket/test-key", + mode=mode, + ) + + mock_gcsfs._mrd_pool_cache.get.assert_not_called() + assert getattr(zf, "closed", True) is True + + @mock.patch("gcsfs.zonal_file.asyn.sync") def test_close_impl_mrd_pool_failure_does_not_skip_aaow(mock_sync, mock_gcsfs): """Verify that if mrd_pool.close fails, aaow teardown is still executed.""" diff --git a/gcsfs/zonal_file.py b/gcsfs/zonal_file.py index e4cfc128e..d8b2109ab 100644 --- a/gcsfs/zonal_file.py +++ b/gcsfs/zonal_file.py @@ -76,6 +76,7 @@ def __init__( # fails without a lookup. _resolve_open_generation_match(mode, if_generation_match) self.aaow = None + self.mrd_pool = None self.finalize_on_close = finalize_on_close self.finalized = False self.mode = mode @@ -83,7 +84,7 @@ def __init__( self.gcsfs = gcsfs self.pool_size = pool_size object_size = None - if "r" in self.mode: + if self.mode == "rb": resolved_cache_type, resolved_cache_source = ( _get_prefetcher_and_cache_config(cache_type) ) @@ -106,36 +107,48 @@ def __init__( "AsyncMultiRangeDownloader (MRD) exists but has no 'persisted_size'. " "This may result in incorrect behavior for unfinalized objects." ) - elif "w" in self.mode or "a" in self.mode or "x" in self.mode: + elif self.mode in ("wb", "ab", "xb"): pass else: raise NotImplementedError( "Only read, write and append operations are currently supported for Zonal buckets." ) - super().__init__( - gcsfs, - path, - mode, - block_size, - autocommit, - cache_type, - cache_options, - acl, - consistency, - metadata, - content_type, - timeout, - fixed_key_metadata, - generation, - kms_key_name, - if_generation_match=if_generation_match, - # Zonal buckets support append; this prevents GCSFile from forcing 'w' mode - _supports_append="a" in mode, - # pass persisted_size here so that Cache is initialized with correct object size - size=object_size, - **kwargs, - ) + try: + super().__init__( + gcsfs, + path, + mode, + block_size, + autocommit, + cache_type, + cache_options, + acl, + consistency, + metadata, + content_type, + timeout, + fixed_key_metadata, + generation, + kms_key_name, + if_generation_match=if_generation_match, + # Zonal buckets support append; this prevents GCSFile from forcing 'w' mode + _supports_append="a" in mode, + # pass persisted_size here so that Cache is initialized with correct object size + size=object_size, + **kwargs, + ) + except BaseException: + self.closed = True + if self.mrd_pool is not None: + pool, self.mrd_pool = self.mrd_pool, None + sync_teardown( + self.gcsfs.loop, + pool.close, + timeout=timeout or DEFAULT_TEARDOWN_TIMEOUT_SECONDS, + description=f"closing mrd_pool for {path} after failed open", + ) + raise async def _init_mrd(self, bucket_name, object_name, generation=None): """ From f68a41fca48be056297e361f7b552071b80c126c Mon Sep 17 00:00:00 2001 From: yuxin00j Date: Mon, 2 Nov 2026 10:23:16 +0000 Subject: [PATCH 2/5] fix(zonal): log and suppress MRD pool teardown errors on failed init --- gcsfs/tests/test_zonal_file.py | 27 +++++++++++++++++++++++++++ gcsfs/zonal_file.py | 18 ++++++++++++------ 2 files changed, 39 insertions(+), 6 deletions(-) diff --git a/gcsfs/tests/test_zonal_file.py b/gcsfs/tests/test_zonal_file.py index ead1e1c68..35074a9a8 100644 --- a/gcsfs/tests/test_zonal_file.py +++ b/gcsfs/tests/test_zonal_file.py @@ -1095,6 +1095,33 @@ def test_zonal_file_failed_init_closes_mrd_pool(mock_gcsfs): assert zf.closed is True +def test_zonal_file_failed_init_preserves_exception_when_pool_close_fails( + mock_gcsfs, caplog +): + """If MRD pool teardown fails during failed __init__, log a warning and propagate the init error.""" + mock_gcsfs.loop = fsspec.asyn.get_loop() + pool = mock_gcsfs._mrd_pool_cache.get.return_value + pool.close = mock.AsyncMock(side_effect=RuntimeError("pool close boom")) + zf = ZonalFile.__new__(ZonalFile) + + with caplog.at_level(logging.WARNING, logger="gcsfs.zonal_file"): + with pytest.raises(KeyError): + zf.__init__( + gcsfs=mock_gcsfs, + path="gs://test-bucket/test-key", + mode="rb", + cache_type="no-such-cache", + ) + + pool.close.assert_awaited_once() + assert zf.mrd_pool is None + assert zf.closed is True + assert any( + r.levelno == logging.WARNING and "pool close boom" in r.message + for r in caplog.records + ) + + @pytest.mark.parametrize("mode", ["r+", "rb+", "w+", "wb+", "a+", "ab+", "x+", "xb+"]) def test_zonal_file_plus_modes_rejected_before_pool_or_super_init(mock_gcsfs, mode): """Unsupported '+' modes fail before checking out an MRD pool or arming __del__.""" diff --git a/gcsfs/zonal_file.py b/gcsfs/zonal_file.py index d8b2109ab..72d778f2b 100644 --- a/gcsfs/zonal_file.py +++ b/gcsfs/zonal_file.py @@ -142,12 +142,18 @@ def __init__( self.closed = True if self.mrd_pool is not None: pool, self.mrd_pool = self.mrd_pool, None - sync_teardown( - self.gcsfs.loop, - pool.close, - timeout=timeout or DEFAULT_TEARDOWN_TIMEOUT_SECONDS, - description=f"closing mrd_pool for {path} after failed open", - ) + try: + sync_teardown( + self.gcsfs.loop, + pool.close, + timeout=timeout or DEFAULT_TEARDOWN_TIMEOUT_SECONDS, + description=f"closing mrd_pool for {path} after failed open", + ) + except Exception as e: + logger.warning( + "Failed to close MRD pool during failed initialization cleanup: %s", + e, + ) raise async def _init_mrd(self, bucket_name, object_name, generation=None): From e09f0db83d96701ad2dea1e392aeb0f5838ae7b4 Mon Sep 17 00:00:00 2001 From: yuxin00j Date: Tue, 3 Nov 2026 03:44:11 +0000 Subject: [PATCH 3/5] fix(zonal): accept size= and allow pickling ZonalFile read handles fs.open(path, "rb", size=n) and unpickling an open read handle raised "TypeError: GCSFile.__init__() got multiple values for keyword argument 'size'": fsspec's reopen() replays size= plus the file's kwargs, which include _supports_append. Drop both from kwargs in ZonalFile.__init__ and fall back to the caller's size when the MRD reports no persisted size. --- gcsfs/tests/test_zonal.py | 19 +++++++++++++++++++ gcsfs/zonal_file.py | 4 +++- 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/gcsfs/tests/test_zonal.py b/gcsfs/tests/test_zonal.py index a2a90dd3c..02730a0fb 100644 --- a/gcsfs/tests/test_zonal.py +++ b/gcsfs/tests/test_zonal.py @@ -7,6 +7,7 @@ import io import multiprocessing import os +import pickle import random import threading import uuid @@ -38,6 +39,7 @@ from gcsfs.tests.settings import TEST_BUCKET, TEST_ZONAL_BUCKET from gcsfs.tests.utils import is_real_gcs, tempdir, tmpfile from gcsfs.zb_hns_utils import MRDPool +from gcsfs.zonal_file import ZonalFile file = "test/accounts.1.json" file_path = f"{TEST_ZONAL_BUCKET}/{file}" @@ -243,6 +245,23 @@ def test_readline_blocksize_zb(extended_gcsfs, gcs_bucket_mocks): assert result == expected +def test_open_with_size_zb(extended_gcsfs, gcs_bucket_mocks): + """A caller-supplied size= is accepted when opening a zonal read handle.""" + with gcs_bucket_mocks(json_data, bucket_type_val=layouts.ZONAL_HIERARCHICAL): + with extended_gcsfs.open(file_path, "rb", size=file_size) as f: + assert isinstance(f, ZonalFile) + assert f.read() == json_data + + +def test_pickle_read_handle_zb(extended_gcsfs, gcs_bucket_mocks): + """A zonal read handle can be pickled and reopened, as DataLoader and Dask do.""" + with gcs_bucket_mocks(json_data, bucket_type_val=layouts.ZONAL_HIERARCHICAL): + with extended_gcsfs.open(file_path, "rb") as f: + assert isinstance(f, ZonalFile) + with pickle.loads(pickle.dumps(f)) as f2: + assert f2.read() == json_data + + def test_read_unfinalized_file_using_mrd(extended_gcsfs, file_path): "Tests that mrd can read from an unfinalized file successfully" if not extended_gcsfs.on_google: diff --git a/gcsfs/zonal_file.py b/gcsfs/zonal_file.py index 72d778f2b..f98726b3c 100644 --- a/gcsfs/zonal_file.py +++ b/gcsfs/zonal_file.py @@ -83,6 +83,8 @@ def __init__( self.flush_interval_bytes = flush_interval_bytes self.gcsfs = gcsfs self.pool_size = pool_size + caller_size = kwargs.pop("size", None) + kwargs.pop("_supports_append", None) # replayed by fsspec.spec.reopen object_size = None if self.mode == "rb": resolved_cache_type, resolved_cache_source = ( @@ -135,7 +137,7 @@ def __init__( # Zonal buckets support append; this prevents GCSFile from forcing 'w' mode _supports_append="a" in mode, # pass persisted_size here so that Cache is initialized with correct object size - size=object_size, + size=object_size if object_size is not None else caller_size, **kwargs, ) except BaseException: From fe05b157de03b231bd4dec1536dd30af648fdec6 Mon Sep 17 00:00:00 2001 From: yuxin00j Date: Tue, 3 Nov 2026 07:08:47 +0000 Subject: [PATCH 4/5] fix(zonal): close the prefetcher on a failed ZonalFile init and reject unknown cache types early - When GCSFile.__init__ fails after the adaptive cache is built, stop the cache's prefetcher in the cleanup path. GCSFile leaves it to _close_impl(), which a file marked closed never reaches. - Reject an unknown cache_type before checking out an MRD pool, so a typo no longer opens a gRPC stream that stays in MRDPoolCache. Tests: the failed-init tests now fail the open after the cache is built (fixed_key_metadata=123) and check that the prefetcher is closed; a new test covers an unknown cache_type; the mode test covers the strings fs.open sends ("r+b", "w+b", "a+b", "x+b") and "rwb", and checks that __del__ was never armed. --- gcsfs/tests/test_zonal_file.py | 52 ++++++++++++++++++++++++++++------ gcsfs/zonal_file.py | 17 +++++++++++ 2 files changed, 61 insertions(+), 8 deletions(-) diff --git a/gcsfs/tests/test_zonal_file.py b/gcsfs/tests/test_zonal_file.py index 35074a9a8..c75c8b4b9 100644 --- a/gcsfs/tests/test_zonal_file.py +++ b/gcsfs/tests/test_zonal_file.py @@ -1075,24 +1075,26 @@ async def coro_func(): def test_zonal_file_failed_init_closes_mrd_pool(mock_gcsfs): - """If GCSFile.__init__ fails after the MRD pool is checked out, the pool is closed.""" + """If GCSFile.__init__ fails after the MRD pool is checked out, the pool and prefetcher are closed.""" mock_gcsfs.loop = fsspec.asyn.get_loop() pool = mock_gcsfs._mrd_pool_cache.get.return_value pool.close = mock.AsyncMock() zf = ZonalFile.__new__(ZonalFile) - with pytest.raises(KeyError): + # An invalid fixed_key_metadata fails in GCSFile.__init__ after the cache is built. + with pytest.raises(TypeError): zf.__init__( gcsfs=mock_gcsfs, path="gs://test-bucket/test-key", mode="rb", - cache_type="no-such-cache", + fixed_key_metadata=123, ) mock_gcsfs._mrd_pool_cache.get.assert_awaited_once() pool.close.assert_awaited_once() assert zf.mrd_pool is None assert zf.closed is True + assert zf.cache._prefetcher is None def test_zonal_file_failed_init_preserves_exception_when_pool_close_fails( @@ -1105,12 +1107,12 @@ def test_zonal_file_failed_init_preserves_exception_when_pool_close_fails( zf = ZonalFile.__new__(ZonalFile) with caplog.at_level(logging.WARNING, logger="gcsfs.zonal_file"): - with pytest.raises(KeyError): + with pytest.raises(TypeError): zf.__init__( gcsfs=mock_gcsfs, path="gs://test-bucket/test-key", mode="rb", - cache_type="no-such-cache", + fixed_key_metadata=123, ) pool.close.assert_awaited_once() @@ -1122,9 +1124,43 @@ def test_zonal_file_failed_init_preserves_exception_when_pool_close_fails( ) -@pytest.mark.parametrize("mode", ["r+", "rb+", "w+", "wb+", "a+", "ab+", "x+", "xb+"]) +def test_zonal_file_unknown_cache_type_fails_before_pool_checkout(mock_gcsfs): + """An unknown cache_type fails before checking out an MRD pool or arming __del__.""" + mock_gcsfs.loop = fsspec.asyn.get_loop() + zf = ZonalFile.__new__(ZonalFile) + + with pytest.raises(KeyError, match="no-such-cache"): + zf.__init__( + gcsfs=mock_gcsfs, + path="gs://test-bucket/test-key", + mode="rb", + cache_type="no-such-cache", + ) + + mock_gcsfs._mrd_pool_cache.get.assert_not_called() + assert "_closed" not in zf.__dict__ + + +@pytest.mark.parametrize( + "mode", + [ + "r+", + "rb+", + "r+b", + "w+", + "wb+", + "w+b", + "a+", + "ab+", + "a+b", + "x+", + "xb+", + "x+b", + "rwb", + ], +) def test_zonal_file_plus_modes_rejected_before_pool_or_super_init(mock_gcsfs, mode): - """Unsupported '+' modes fail before checking out an MRD pool or arming __del__.""" + """Unsupported modes ('+' modes or 'rwb') fail before checking out an MRD pool or arming __del__.""" mock_gcsfs.loop = fsspec.asyn.get_loop() zf = ZonalFile.__new__(ZonalFile) @@ -1136,7 +1172,7 @@ def test_zonal_file_plus_modes_rejected_before_pool_or_super_init(mock_gcsfs, mo ) mock_gcsfs._mrd_pool_cache.get.assert_not_called() - assert getattr(zf, "closed", True) is True + assert "_closed" not in zf.__dict__ @mock.patch("gcsfs.zonal_file.asyn.sync") diff --git a/gcsfs/zonal_file.py b/gcsfs/zonal_file.py index f98726b3c..fca0feeac 100644 --- a/gcsfs/zonal_file.py +++ b/gcsfs/zonal_file.py @@ -1,6 +1,7 @@ import logging from fsspec import asyn +from fsspec.caching import caches from google.cloud.storage.asyncio.async_appendable_object_writer import ( _DEFAULT_FLUSH_INTERVAL_BYTES, ) @@ -90,6 +91,9 @@ def __init__( resolved_cache_type, resolved_cache_source = ( _get_prefetcher_and_cache_config(cache_type) ) + # Reject an unknown cache type before checking out a pool and stream. + if resolved_cache_type not in caches: + raise KeyError(resolved_cache_type) self.mrd_pool = asyn.sync( self.gcsfs.loop, self.gcsfs._mrd_pool_cache.get, @@ -142,6 +146,19 @@ def __init__( ) except BaseException: self.closed = True + # GCSFile.__init__ leaves the prefetcher to _close_impl(), which a + # file marked closed never reaches, so stop it here. + cache = getattr(self, "cache", None) + prefetcher = getattr(cache, "_prefetcher", None) + if prefetcher is not None: + cache._prefetcher = None + try: + prefetcher.close() + except Exception as e: + logger.warning( + "Failed to close prefetcher during failed initialization cleanup: %s", + e, + ) if self.mrd_pool is not None: pool, self.mrd_pool = self.mrd_pool, None try: From 8b1cc3e1b3be1679b7da374b4835e547c138f993 Mon Sep 17 00:00:00 2001 From: yuxin00j Date: Tue, 3 Nov 2026 09:39:48 +0000 Subject: [PATCH 5/5] fix(zonal): cover the post-checkout window and tighten failed-init tests - Read the checked-out pool's details and persisted size inside the try block, so an interrupt (Ctrl-C) right after checkout still releases the pool. - Call BackgroundPrefetcher.close() directly in the failed-init cleanup, as GCSFile._close_impl() does. It already logs its own teardown errors, so the extra except branch could never run. Tests: assert that BackgroundPrefetcher.close() is called, not only that the cache dropped its reference; add a write-mode failed-init test (a failed "wb" open is marked closed, so __del__ never flushes and creates the object); add an interrupt-after-checkout test; and open with a wrong size= to pin that persisted_size takes precedence. --- gcsfs/tests/test_zonal.py | 5 +-- gcsfs/tests/test_zonal_file.py | 60 +++++++++++++++++++++++++++++----- gcsfs/zonal_file.py | 33 +++++++++---------- 3 files changed, 71 insertions(+), 27 deletions(-) diff --git a/gcsfs/tests/test_zonal.py b/gcsfs/tests/test_zonal.py index 02730a0fb..0e5113870 100644 --- a/gcsfs/tests/test_zonal.py +++ b/gcsfs/tests/test_zonal.py @@ -246,9 +246,10 @@ def test_readline_blocksize_zb(extended_gcsfs, gcs_bucket_mocks): def test_open_with_size_zb(extended_gcsfs, gcs_bucket_mocks): - """A caller-supplied size= is accepted when opening a zonal read handle.""" + """A caller-supplied size= is accepted, and the MRD's persisted_size takes precedence.""" with gcs_bucket_mocks(json_data, bucket_type_val=layouts.ZONAL_HIERARCHICAL): - with extended_gcsfs.open(file_path, "rb", size=file_size) as f: + # A wrong size must not truncate the read. + with extended_gcsfs.open(file_path, "rb", size=1) as f: assert isinstance(f, ZonalFile) assert f.read() == json_data diff --git a/gcsfs/tests/test_zonal_file.py b/gcsfs/tests/test_zonal_file.py index c75c8b4b9..56c081b51 100644 --- a/gcsfs/tests/test_zonal_file.py +++ b/gcsfs/tests/test_zonal_file.py @@ -16,6 +16,7 @@ import fsspec.asyn import pytest from fsspec.asyn import FSTimeoutError +from fsspec.prefetcher import BackgroundPrefetcher from google.cloud.storage.asyncio.async_appendable_object_writer import ( _DEFAULT_FLUSH_INTERVAL_BYTES, ) @@ -1080,18 +1081,23 @@ def test_zonal_file_failed_init_closes_mrd_pool(mock_gcsfs): pool = mock_gcsfs._mrd_pool_cache.get.return_value pool.close = mock.AsyncMock() zf = ZonalFile.__new__(ZonalFile) + real_close = BackgroundPrefetcher.close - # An invalid fixed_key_metadata fails in GCSFile.__init__ after the cache is built. - with pytest.raises(TypeError): - zf.__init__( - gcsfs=mock_gcsfs, - path="gs://test-bucket/test-key", - mode="rb", - fixed_key_metadata=123, - ) + with mock.patch.object( + BackgroundPrefetcher, "close", autospec=True, side_effect=real_close + ) as prefetcher_close: + # An invalid fixed_key_metadata fails in GCSFile.__init__ after the cache is built. + with pytest.raises(TypeError): + zf.__init__( + gcsfs=mock_gcsfs, + path="gs://test-bucket/test-key", + mode="rb", + fixed_key_metadata=123, + ) mock_gcsfs._mrd_pool_cache.get.assert_awaited_once() pool.close.assert_awaited_once() + prefetcher_close.assert_called_once() assert zf.mrd_pool is None assert zf.closed is True assert zf.cache._prefetcher is None @@ -1124,6 +1130,44 @@ def test_zonal_file_failed_init_preserves_exception_when_pool_close_fails( ) +def test_zonal_file_interrupt_after_pool_checkout_closes_mrd_pool(mock_gcsfs): + """A KeyboardInterrupt right after the MRD pool is checked out still releases the pool.""" + mock_gcsfs.loop = fsspec.asyn.get_loop() + pool = mock_gcsfs._mrd_pool_cache.get.return_value + pool.close = mock.AsyncMock() + # Ctrl-C arriving while the pool's size is read, before GCSFile.__init__ runs. + type(pool).persisted_size = mock.PropertyMock(side_effect=KeyboardInterrupt) + zf = ZonalFile.__new__(ZonalFile) + + with pytest.raises(KeyboardInterrupt): + zf.__init__( + gcsfs=mock_gcsfs, + path="gs://test-bucket/test-key", + mode="rb", + ) + + pool.close.assert_awaited_once() + assert zf.mrd_pool is None + + +def test_zonal_file_failed_write_init_is_marked_closed(mock_gcsfs): + """A failed write-mode open is marked closed so __del__ never flushes (creates) the object.""" + mock_gcsfs.loop = fsspec.asyn.get_loop() + zf = ZonalFile.__new__(ZonalFile) + + with pytest.raises(TypeError): + zf.__init__( + gcsfs=mock_gcsfs, + path="gs://test-bucket/test-key", + mode="wb", + fixed_key_metadata=123, + ) + + mock_gcsfs._mrd_pool_cache.get.assert_not_called() + # AbstractBufferedFile.__init__ already set _closed = False, so this is not the default. + assert zf.closed is True + + def test_zonal_file_unknown_cache_type_fails_before_pool_checkout(mock_gcsfs): """An unknown cache_type fails before checking out an MRD pool or arming __del__.""" mock_gcsfs.loop = fsspec.asyn.get_loop() diff --git a/gcsfs/zonal_file.py b/gcsfs/zonal_file.py index fca0feeac..af26bf1cc 100644 --- a/gcsfs/zonal_file.py +++ b/gcsfs/zonal_file.py @@ -104,15 +104,6 @@ def __init__( cache_type=resolved_cache_type, cache_source=resolved_cache_source, ) - if getattr(self.mrd_pool, "details", None) is not None: - self._details = self.mrd_pool.details - object_size = self.mrd_pool.persisted_size - - if object_size is None: - logger.warning( - "AsyncMultiRangeDownloader (MRD) exists but has no 'persisted_size'. " - "This may result in incorrect behavior for unfinalized objects." - ) elif self.mode in ("wb", "ab", "xb"): pass else: @@ -120,7 +111,19 @@ def __init__( "Only read, write and append operations are currently supported for Zonal buckets." ) + # Everything after the pool checkout runs under the cleanup below. try: + if self.mrd_pool is not None: + if getattr(self.mrd_pool, "details", None) is not None: + self._details = self.mrd_pool.details + object_size = self.mrd_pool.persisted_size + + if object_size is None: + logger.warning( + "AsyncMultiRangeDownloader (MRD) exists but has no 'persisted_size'. " + "This may result in incorrect behavior for unfinalized objects." + ) + super().__init__( gcsfs, path, @@ -147,18 +150,14 @@ def __init__( except BaseException: self.closed = True # GCSFile.__init__ leaves the prefetcher to _close_impl(), which a - # file marked closed never reaches, so stop it here. + # file marked closed never reaches, so stop it here. Like + # _close_impl(), rely on BackgroundPrefetcher.close() to log its + # own teardown errors. cache = getattr(self, "cache", None) prefetcher = getattr(cache, "_prefetcher", None) if prefetcher is not None: cache._prefetcher = None - try: - prefetcher.close() - except Exception as e: - logger.warning( - "Failed to close prefetcher during failed initialization cleanup: %s", - e, - ) + prefetcher.close() if self.mrd_pool is not None: pool, self.mrd_pool = self.mrd_pool, None try: