diff --git a/.github/workflows/runtests.yml b/.github/workflows/runtests.yml
index ab139e2ba..f59e5d614 100644
--- a/.github/workflows/runtests.yml
+++ b/.github/workflows/runtests.yml
@@ -154,7 +154,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-latest, macos-latest]
+        os: ${{ fromJson(needs.setup.outputs.os-matrix) }}
 
     # Keep remote-IO coverage visible without blocking unrelated changes.
     if: github.event_name == 'push' || !contains(github.event.pull_request.labels.*.name, 'no_ci')
diff --git a/dascore/constants.py b/dascore/constants.py
index de3f38e68..2491cd21b 100644
--- a/dascore/constants.py
+++ b/dascore/constants.py
@@ -66,10 +66,14 @@ def map(self, fn: Callable, iterable: Iterable, /) -> Iterable:
 # types used to represent paths
 path_types = str | Path | UPath
 
+# Protocols served over HTTP. The one spelling of this set; HDF5 tuning and
+# the remote-cache downloader both key off it.
+http_protocols = ("http", "https")
+
 # Remote protocols that should use DASCore's smaller HDF5 readahead blocks.
 # h5py performs many small metadata reads while opening files, and some S3-like
 # backends default to large readahead chunks that overfetch remote data.
-remote_hdf5_tuned_protocols = ("s3", "s3a", "s3n")
+remote_hdf5_tuned_protocols = ("s3", "s3a", "s3n", *http_protocols)
 
 # One second in numpy timedelta speak
 ONE_SECOND = np.timedelta64(1, "s")
diff --git a/dascore/io/core.py b/dascore/io/core.py
index b9eb9615e..d67dbe4be 100644
--- a/dascore/io/core.py
+++ b/dascore/io/core.py
@@ -17,6 +17,7 @@
     Mapping,
     Sequence,
 )
+from contextlib import suppress
 from functools import cached_property, wraps
 from numbers import Integral
 from pathlib import Path
@@ -59,7 +60,11 @@
     RemoteCacheError,
     UnknownFiberFormatError,
 )
-from dascore.utils.io import IOResourceManager, get_handle_from_resource
+from dascore.utils.io import (
+    IOResourceManager,
+    get_handle_from_resource,
+    release_handle,
+)
 from dascore.utils.mapping import FrozenDict
 from dascore.utils.misc import (
     _get_install_message,
@@ -860,22 +865,35 @@ def _wrapper(*args, _pre_cast=False, **kwargs):
         bound = sig.bind(*args, **kwargs)
         new_kw = bound.arguments
         resource = new_kw.pop(arg_name)
+        new_resource = None
         try:
             new_resource = get_handle_from_resource(resource, required_type)
             new_kw[arg_name] = new_resource
             # kwargs is included in bound arguments, need to re-attach
             new_kw.update(new_kw.pop("kwargs", {}))
             out = func(**new_kw)
-        except Exception as e:  # get_format can't raise; must return false.
-            if fun_name == "get_format":
-                out = False
-            else:
-                raise e
+        except BaseException as e:
+            # A handle created here must be released even on failure,
+            # including on KeyboardInterrupt: leaking a remote handle leaves
+            # garbage collection paused for as long as the traceback is
+            # retained. Abort rather than close, so a failed remote write
+            # discards its temp file instead of uploading a partial one.
+            if new_resource is not None and new_resource is not resource:
+                with suppress(Exception):
+                    release_handle(new_resource, abort=True)
+            # get_format reports "not my format" by returning False rather
+            # than raising, so an ordinary Exception becomes False here.
+            # Everything else propagates, including a BaseException raised
+            # inside get_format: the catch is only this wide so the cleanup
+            # above runs on a KeyboardInterrupt, not to swallow one.
+            if fun_name != "get_format" or not isinstance(e, Exception):
+                raise
+            out = False
         else:
             # if a new file handle was created we need to close it now. But it
             # shouldn't close any passed in, that should happen up the stack.
-            if new_resource is not resource and hasattr(new_resource, "close"):
-                new_resource.close()
+            if new_resource is not resource:
+                release_handle(new_resource)
         return out
 
     # attach the function and required type for later use
diff --git a/dascore/utils/hdf5.py b/dascore/utils/hdf5.py
index ec86e7739..fe729b081 100644
--- a/dascore/utils/hdf5.py
+++ b/dascore/utils/hdf5.py
@@ -18,7 +18,7 @@
 
 from dascore.compat import UPath
 from dascore.config import get_config
-from dascore.constants import remote_hdf5_tuned_protocols
+from dascore.constants import http_protocols, remote_hdf5_tuned_protocols
 from dascore.utils.misc import (
     _maybe_make_parent_directory,
     _maybe_unpack,
@@ -31,6 +31,8 @@
     ensure_local_file,
     get_local_handle,
     is_no_range_http_error,
+    pause_gc,
+    resume_gc,
 )
 
 ns_to_datetime = partial(pd.to_datetime, unit="ns")
@@ -55,26 +57,56 @@ class _ManagedH5pyFile:
 
     For path-backed opens, this wrapper owns only the h5py handle. For
     ``h5py.File(..., driver="fileobj")`` paths, it also owns the Python
-    file-like object DASCore created on behalf of the caller. ``close()`` is
-    therefore the point where DASCore tears down the entire HDF5 access stack.
+    file-like object underneath, whether DASCore created it or the caller
+    supplied it. ``close()`` is therefore the point where DASCore tears down
+    the entire HDF5 access stack.
     """
 
-    def __init__(self, handle: H5pyFile, owned_fileobj=None):
+    # Class defaults, so a half-built instance is still closeable rather than
+    # falling through __getattr__ to a handle that may not be set yet.
+    _closed = False
+    _gc_paused_pid = None
+
+    def __init__(self, handle: H5pyFile, owned_fileobj=None, gc_paused=False):
         self._handle = handle
         self._owned_fileobj = owned_fileobj
-        self._closed = False
+        if gc_paused:
+            # The pid that paused, so a handle inherited through a fork
+            # cannot resume a pause the child never took.
+            self._gc_paused_pid = os.getpid()
 
     def close(self):
         """Close the h5py file and, when present, the owned file object."""
         if self._closed:
             return
+        self._closed = True
         try:
             self._handle.close()
         finally:
-            if self._owned_fileobj is not None:
-                with suppress(Exception):
-                    self._owned_fileobj.close()
-            self._closed = True
+            # Nested so nothing raised by the teardown, including a
+            # BaseException, can skip the resume and strand the pause.
+            try:
+                if self._owned_fileobj is not None:
+                    with suppress(Exception):
+                        self._owned_fileobj.close()
+            finally:
+                # dict.pop is atomic, so racing closes resume exactly once.
+                pid = self.__dict__.pop("_gc_paused_pid", None)
+                if pid == os.getpid():
+                    resume_gc()
+
+    def __del__(self):
+        """
+        Release a leaked handle's pause so it cannot stop collection forever.
+
+        Only the pause: closing here would also close a caller-supplied h5py
+        file or stream that this wrapper never had permission to close.
+        Reference counting still tears the underlying handles down.
+        """
+        pid = self.__dict__.pop("_gc_paused_pid", None)
+        if pid == os.getpid():
+            with suppress(Exception):
+                resume_gc()
 
     def __enter__(self):
         return self
@@ -101,6 +133,68 @@ def __getattr__(self, item):
         return getattr(self._handle, item)
 
 
+def _is_loop_backed(resource) -> bool:
+    """
+    Return True when reads on a resource are served by an event-loop thread.
+
+    fsspec async filesystems (http, s3, ...) bridge each read onto a shared
+    event-loop thread and mark themselves with ``async_impl``; duck-typing it
+    avoids importing fsspec here. Local, memory, and other synchronous
+    backends need no GC pause. A buffered reader hides the filesystem behind
+    ``raw``, so unwrap it: missing it would leave the deadlock window open.
+    """
+    while resource is not None:
+        try:
+            if getattr(getattr(resource, "fs", None), "async_impl", False):
+                return True
+            wrapped = getattr(resource, "raw", None)
+        except Exception:
+            # Probing can fail rather than return nothing: a UPath whose
+            # backend is not installed raises from ``fs``, and a wrapper
+            # can refuse an attribute with something other than
+            # AttributeError. Either way the open below reports it.
+            return False
+        resource = None if wrapped is resource else wrapped
+    return False
+
+
+def _open_h5_fileobj(
+    fileobj, constructor, mode, *, pause: bool, close_on_error: bool
+) -> _ManagedH5pyFile:
+    """
+    Open a file object with h5py through the fileobj driver.
+
+    Loop-backed resources pause automatic collection first; the returned
+    wrapper owns that pause and releases it on close, and a failed open
+    releases it here.
+
+    ``close_on_error`` is set only for file objects DASCore created. A
+    caller-supplied one must survive a failed open: ``get_format`` offers the
+    same object to every FiberIO in turn, and an HDF5 miss is the expected
+    outcome for most of them.
+    """
+    try:
+        if pause:
+            # Inside the try, and paired unconditionally below, because
+            # pause_gc always leaves the depth consistent with the pauses it
+            # took -- even when interrupted partway.
+            pause_gc()
+        handle = constructor(fileobj, mode=mode, driver="fileobj")
+        return _ManagedH5pyFile(handle, fileobj, gc_paused=pause)
+    except BaseException:
+        # Everything the pause covers runs in here, so an interrupt at any
+        # point still rebalances. Nested so nothing raised while closing,
+        # including a BaseException, can skip the resume.
+        try:
+            if close_on_error:
+                with suppress(Exception):
+                    fileobj.close()
+        finally:
+            if pause:
+                resume_gc()
+        raise
+
+
 def get_h5py_file(handle) -> H5pyFile:
     """
     Return the underlying ``h5py.File`` for a DASCore h5 handle.
@@ -148,8 +242,16 @@ def open_h5_resource(
     if isinstance(resource, H5pyFile):
         return _ManagedH5pyFile(resource)
     if isinstance(resource, io.IOBase):
-        handle = constructor(resource, mode=mode, driver="fileobj")
-        return _ManagedH5pyFile(handle, resource)
+        # A user-supplied fsspec file object delegates reads to the same
+        # event-loop thread as the UPath branch below and needs the same
+        # GC pause; plain local/in-memory streams do not.
+        return _open_h5_fileobj(
+            resource,
+            constructor,
+            mode,
+            pause=_is_loop_backed(resource),
+            close_on_error=False,
+        )
     if isinstance(resource, UPath):
         # Reuse an already-materialized local artifact when present so later
         # HDF5 reads do not re-enter the remote fallback path unnecessarily.
@@ -160,6 +262,8 @@ def open_h5_resource(
                 constructor=constructor,
                 open_kwargs_getter=open_kwargs_getter,
             )
+        # Note: only mode == "r" is a supported remote path here; H5Writer
+        # intercepts UPath targets with its temp-file write-back handle.
         file_mode = "rb" if mode == "r" else "r+b"
         open_kwargs = open_kwargs_getter(resource)
         handle = _FallbackFileObj(
@@ -167,12 +271,17 @@ def open_h5_resource(
             local_opener=lambda: ensure_local_file(resource).open(file_mode),
             error_predicate=is_no_range_http_error,
         )
-        try:
-            h5_handle = constructor(handle, mode=mode, driver="fileobj")
-            return _ManagedH5pyFile(h5_handle, handle)
-        except Exception:
-            handle.close()
-            raise
+        # h5py holds its global lock while blocking on fsspec's event-loop
+        # thread for remote fetches; an automatic garbage collection on that
+        # thread deallocating h5py objects then deadlocks on the same lock.
+        # Pause collection for the handle's lifetime (resumed in close()).
+        return _open_h5_fileobj(
+            handle,
+            constructor,
+            mode,
+            pause=_is_loop_backed(resource),
+            close_on_error=True,
+        )
     try:
         if mode != "r":
             _maybe_make_parent_directory(resource)
@@ -205,15 +314,19 @@ class H5Reader(_H5CasterBase):
     def _get_open_kwargs(resource: UPath) -> dict[str, object]:
         """Return backend-specific kwargs for remote HDF5 file objects."""
         protocol = getattr(resource, "protocol", None)
-        if protocol in remote_hdf5_tuned_protocols:
-            # h5py performs many small seeks while opening HDF5 metadata.
-            # s3fs defaults to 50 MB readahead blocks, which can pull most of
-            # a large remote file just to satisfy metadata probes.
-            return {
-                "block_size": get_config().remote_hdf5_block_size,
-                "cache_type": "readahead",
-            }
-        return {}
+        if protocol not in remote_hdf5_tuned_protocols:
+            return {}
+        # h5py performs many small seeks while opening HDF5 metadata, and
+        # remote backends default to large readahead blocks (s3fs uses 50 MB)
+        # which can pull most of a file just to satisfy those probes.
+        out = {"block_size": get_config().remote_hdf5_block_size}
+        if protocol not in http_protocols:
+            return out | {"cache_type": "readahead"}
+        # HTTP needs a block LRU instead: the probe alternates between the
+        # file header and footer, and fsspec's default single-window cache
+        # refetches a full block (or the whole file on range-less servers) on
+        # every jump. Eight blocks keeps both ends resident and bounds memory.
+        return out | {"cache_type": "blockcache", "cache_options": {"maxblocks": 8}}
 
     @classmethod
     def get_handle(cls, resource):
diff --git a/dascore/utils/io.py b/dascore/utils/io.py
index f2f4e9431..08725c877 100644
--- a/dascore/utils/io.py
+++ b/dascore/utils/io.py
@@ -209,6 +209,20 @@ def get_handle_from_resource(uri, required_type):
     return uri
 
 
+def release_handle(handle, abort: bool = False):
+    """
+    Release a file handle, closing it or discarding its uncommitted work.
+
+    Only a few handles can ``abort``; a remote HDF5 writer does, because
+    closing it uploads whatever was written so far. Everything else is
+    closed, and a handle with no ``close`` needs no release at all.
+    """
+    if abort and hasattr(handle, "abort"):
+        handle.abort()
+    else:
+        getattr(handle, "close", lambda: None)()
+
+
 class IOResourceManager:
     """
     A class for managing opening/closing files.
@@ -254,28 +268,54 @@ def get_resource(self, required_type: RequiredType) -> RequiredType:
                 self._cache[required_type] = out
             return self._cache[required_type]
 
-    def close_all(self):
-        """Close any open file handles."""
+    def close_all(self, abort: bool = False):
+        """
+        Close any open file handles.
+
+        With ``abort=True``, handles that support it discard uncommitted
+        work (e.g. remote writers skip uploading a partial file). One
+        handle failing must not skip cleanup of the others (remote handles
+        resume garbage collection in close), so the first error is
+        re-raised only after every handle was attempted. BaseException is
+        caught for that reason too: a Ctrl-C mid-close would otherwise
+        strand the GC pause of every handle after it.
+        """
+        first_exc = None
         with self._lock:
             for handle in self._cache.values():
-                getattr(handle, "close", lambda: None)()
+                try:
+                    release_handle(handle, abort=abort)
+                except BaseException as exc:
+                    first_exc = first_exc if first_exc is not None else exc
+        if first_exc is not None:
+            raise first_exc
 
     def clear_cache(self):
         """Close and forget any cached resources so they can be reopened fresh."""
         with self._lock:
-            self.close_all()
-            self._cache.clear()
+            try:
+                self.close_all()
+            finally:
+                self._cache.clear()
 
     def __enter__(self):
         """Entering context manager."""
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        """Simply ensure all file handles are closed."""
-        self.close_all()
+        """Close all handles; on error, abort uncommitted writes instead."""
+        if exc_type is None:
+            self.close_all()
+            return
+        try:
+            self.close_all(abort=True)
+        except Exception as cleanup_error:
+            # A cleanup failure must not replace the error which caused it.
+            exc_val.add_note(f"Aborting IO resources also failed: {cleanup_error!r}")
 
     def __del__(self):
-        self.close_all()
+        with suppress(Exception):
+            self.close_all()
 
 
 def patch_to_xarray(patch: PatchType):
diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py
index 72bc6ba5b..76dac6581 100644
--- a/dascore/utils/remote_io.py
+++ b/dascore/utils/remote_io.py
@@ -2,10 +2,13 @@
 
 from __future__ import annotations
 
+import gc
 import json
 import os
 import shutil
 import tempfile
+import threading
+import time
 import warnings
 from contextlib import contextmanager
 from contextvars import ContextVar
@@ -16,15 +19,16 @@
 
 from dascore.compat import UPath
 from dascore.config import get_config
+from dascore.constants import http_protocols
 from dascore.exceptions import RemoteCacheError
 from dascore.utils.misc import _reinit_after_fork
 from dascore.utils.paths import coerce_to_upath, is_local_path, is_pathlike
 
-_HTTP_PROTOCOLS = {"http", "https"}
 _NO_RANGE_HTTP_PATTERNS = (
     "doesn't appear to support range requests",
     "only reading this file from the beginning is supported",
 )
+_NO_SIZE_HTTP_PATTERN = "cannot seek streaming http file"
 _REMOTE_RESOURCE_CACHE: dict[str, UPath] = {}
 # One lock per cached resource, so two threads never download the same
 # file at once while unrelated downloads still run together. Entries are
@@ -34,6 +38,101 @@
     "remote_cache_scope", default="default"
 )
 
+_gc_pause_lock = threading.Lock()
+_gc_pause_depth = 0
+_gc_was_enabled = False
+_gc_collect_after = 0.0
+# Seconds between safety-valve collections; a full collect is O(live objects),
+# measured at 7-160 ms here, so it must not run on every remote open.
+_GC_COLLECT_INTERVAL = 10.0
+
+
+def _claim_safety_collect() -> bool:
+    """Return True when this caller wins the rate-limited safety collection."""
+    global _gc_collect_after
+    with _gc_pause_lock:
+        now = time.monotonic()
+        if now < _gc_collect_after:
+            return False
+        _gc_collect_after = now + _GC_COLLECT_INTERVAL
+        return True
+
+
+def pause_gc() -> None:
+    """
+    Pause automatic garbage collection for one remote read session.
+
+    h5py holds its process-global lock while blocking on fsspec's event-loop
+    thread for each remote fetch. An automatic collection on that thread which
+    has to deallocate a dead h5py object needs the same lock, so the two
+    threads deadlock. Disabling automatic collection for the handle's lifetime
+    closes that window; reference counting still frees non-cyclic garbage.
+
+    h5py documents this deadlock, and disabling collection as one of its two
+    mitigations, under "Python file-like objects":
+    https://docs.h5py.org/en/stable/high/file.html#python-file-like-objects
+    Its other mitigation, avoiding reference cycles which keep h5py objects
+    alive, is not available to us: the cycle can be anywhere in the process.
+
+    Calls nest; every ``pause_gc`` needs one ``resume_gc``. ``_ManagedH5pyFile``
+    pairs them with ``close``/``__del__``. An interrupted ``pause_gc`` still
+    leaves the depth consistent with the pauses it took, so a caller which
+    resumes on any failure stays balanced.
+
+    The pause is process-global, so cyclic garbage from every thread
+    accumulates until the last remote handle closes. A handle which is never
+    closed keeps it paused; one which is dropped inside a reference cycle is
+    recovered by the collection below, on the next remote open -- if the
+    program makes one. Otherwise the pause outlives every remote read.
+    """
+    global _gc_pause_depth, _gc_was_enabled
+    # Safety valve: finalize cyclic garbage, including a handle leaked by an
+    # earlier session, so a stranded pause cannot disable collection forever.
+    # It runs before this pause is taken, so an interrupt during the collect
+    # cannot strand one, and outside the lock, since finalizing such a handle
+    # calls resume_gc, which takes it.
+    if _claim_safety_collect():
+        gc.collect()
+    with _gc_pause_lock:
+        # The count must move even if ``disable`` is interrupted, or the depth
+        # stops matching the live handles: it would never fall back to zero,
+        # so no later pause would ever disable collection again.
+        try:
+            if _gc_pause_depth == 0:
+                _gc_was_enabled = gc.isenabled()
+                gc.disable()
+        finally:
+            _gc_pause_depth += 1
+
+
+def resume_gc() -> None:
+    """Undo one ``pause_gc``; the last one restores the previous gc state."""
+    global _gc_pause_depth
+    with _gc_pause_lock:
+        if _gc_pause_depth == 0:
+            return
+        # Enable first, for the mirror image of the reason pause counts first.
+        if _gc_pause_depth == 1 and _gc_was_enabled:
+            gc.enable()
+        _gc_pause_depth -= 1
+
+
+@_reinit_after_fork
+def _reset_gc_pause_state():
+    """
+    Drop a pause inherited from a fork; no close in the child can undo it.
+
+    Handles inherited by the child record the pid that paused for them, so
+    closing one there cannot resume a pause this child never took.
+    """
+    global _gc_pause_depth, _gc_pause_lock, _gc_collect_after
+    _gc_pause_lock = threading.Lock()
+    if _gc_pause_depth and _gc_was_enabled:
+        gc.enable()
+    _gc_pause_depth = 0
+    # The parent's rate limit does not describe this process.
+    _gc_collect_after = 0.0
+
 
 @_reinit_after_fork
 def _reinit_remote_cache_locks():
@@ -150,7 +249,7 @@ def _download_remote_file(path, local_path: Path):
     """Download a remote path into its cache location."""
     resource = coerce_to_upath(path)
     protocol = getattr(resource, "protocol", None)
-    open_kwargs = {"block_size": 0} if protocol in _HTTP_PROTOCOLS else {}
+    open_kwargs = {"block_size": 0} if protocol in http_protocols else {}
     local_path.parent.mkdir(parents=True, exist_ok=True)
     fd, temp_name = tempfile.mkstemp(
         dir=local_path.parent,
@@ -251,10 +350,14 @@ def get_local_handle(resource, opener):
 
 def is_no_range_http_error(exc: Exception) -> bool:
     """Return True when an exception indicates no-range HTTP random access."""
+    if not isinstance(exc, ValueError):
+        return False
     message = str(exc).lower()
-    return isinstance(exc, ValueError) and all(
-        pattern in message for pattern in _NO_RANGE_HTTP_PATTERNS
-    )
+    # A server reporting no size streams instead of ranging; seeking such a
+    # file needs the same local-file fallback as an outright no-range server.
+    if _NO_SIZE_HTTP_PATTERN in message:
+        return True
+    return all(pattern in message for pattern in _NO_RANGE_HTTP_PATTERNS)
 
 
 class _FallbackFileObj:
@@ -290,6 +393,14 @@ class _FallbackFileObj:
     This is not a general retry wrapper for arbitrary IO failures. It is meant
     for one known fallback condition where switching from remote access to a
     local cached file is safe and expected.
+
+    Load-bearing invariant: references flow one way here - h5py objects may
+    reference this wrapper and the fsspec handle, but nothing that crosses to
+    fsspec's event-loop thread may ever reference an h5py object. Otherwise
+    that thread could perform the final decref of an h5py object, whose
+    C-level deallocation takes h5py's global lock, and deadlock against a
+    caller blocked on the loop while holding that lock (``pause_gc`` only
+    stops cyclic collection, not refcount-driven deallocation).
     """
 
     def __init__(self, remote_opener, local_opener, error_predicate):
diff --git a/docs/tutorial/remote_patches.qmd b/docs/tutorial/remote_patches.qmd
index 62c01495d..c88930ca3 100644
--- a/docs/tutorial/remote_patches.qmd
+++ b/docs/tutorial/remote_patches.qmd
@@ -136,3 +136,13 @@ The same distinction applies to spools:
 - `spool.get_contents()` relies on summary metadata
 - `spool[0]` or iteration loads patch data
 - if a remote backend cannot answer the metadata side without a download, use `allow_remote_cache_for_metadata=True`
+
+## Garbage Collection During Remote HDF5 Reads
+
+While a remote HDF5 handle is open, DASCore pauses Python's automatic garbage collection for the whole process, then restores it when the last such handle closes.
+
+This avoids a deadlock which h5py documents under [Python file-like objects](https://docs.h5py.org/en/stable/high/file.html#python-file-like-objects): h5py serializes its low-level calls behind a process-global lock, holds that lock while a file-like object services a read, and needs the same lock to deallocate a dead h5py object. A remote read is serviced by the event loop thread [fsspec](https://filesystem-spec.readthedocs.io/) runs for each async backend, so a cyclic collection landing on that thread waits for a lock the reader will not release until the read completes.
+
+Of the mitigations h5py suggests, DASCore takes the second: temporarily disabling collection. The first, avoiding reference cycles which keep h5py objects alive, is not something a library can guarantee, since the cycle may be anywhere in the process.
+
+Reference counting is unaffected, so most objects are still freed immediately; only cyclic garbage accumulates, and only until the handle closes. Reads of local files and of synchronous backends such as `memory://` are unaffected.
diff --git a/tests/conftest.py b/tests/conftest.py
index 396020883..e0b8b901b 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import gc
 import os
 import shutil
 import threading
@@ -17,6 +18,7 @@
 
 import dascore as dc
 import dascore.examples as ex
+import dascore.utils.remote_io as remote_io
 from dascore.compat import random_state
 from dascore.config import get_config, set_config
 from dascore.constants import SpoolType
@@ -156,6 +158,32 @@ def use_test_config():
         yield
 
 
+@pytest.fixture(autouse=True)
+def gc_pause_is_not_leaked():
+    """
+    Blame the test which strands the remote-read gc pause, and repair it.
+
+    Remote HDF5 handles disable automatic collection process-wide (see
+    `dascore.utils.remote_io.pause_gc`). Repairing here keeps one leak from
+    cascading into every later test, which would bury the real failure.
+    """
+    was_enabled = gc.isenabled()
+    yield
+    depth = remote_io._gc_pause_depth
+    is_enabled = gc.isenabled()
+    if not depth and is_enabled == was_enabled:
+        return
+    remote_io._gc_pause_depth = 0
+    if was_enabled:
+        gc.enable()
+    else:
+        gc.disable()
+    pytest.fail(
+        f"test changed collection state: pause depth={depth}, "
+        f"gc enabled={is_enabled}, expected={was_enabled}"
+    )
+
+
 @pytest.fixture(scope="session", autouse=True)
 def allow_legacy_dasdae_coord_unpickle():
     """Test fixtures may rely on trusted historical DASDAE coord payloads."""
diff --git a/tests/test_io/conftest.py b/tests/test_io/conftest.py
index 57d188d0b..eea228674 100644
--- a/tests/test_io/conftest.py
+++ b/tests/test_io/conftest.py
@@ -14,9 +14,10 @@
 import threading
 import time
 from collections.abc import Callable
+from contextlib import contextmanager
 from functools import partial
 from http import HTTPStatus
-from http.server import HTTPServer, SimpleHTTPRequestHandler, ThreadingHTTPServer
+from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from urllib.error import URLError
 from urllib.request import urlopen
@@ -243,7 +244,7 @@ def _ensure(fetch_name: str, relative_path: str | Path | None = None) -> Path:
     return _ensure
 
 
-class _NoReverseDNSMixin:
+class _ThreadingHTTPServer(ThreadingHTTPServer):
     """Bind without the reverse DNS lookup ``HTTPServer`` normally performs.
 
     ``HTTPServer.server_bind`` calls ``socket.getfqdn(host)`` purely to fill in
@@ -261,38 +262,40 @@ def server_bind(self):
         self.server_port = port
 
 
-class _HTTPServer(_NoReverseDNSMixin, HTTPServer):
-    """Single-threaded test server that skips the reverse DNS lookup."""
-
-
-class _ThreadingHTTPServer(_NoReverseDNSMixin, ThreadingHTTPServer):
-    """Threading test server that skips the reverse DNS lookup."""
-
+@contextmanager
+def _serve_das_tree(handler_cls, root, label):
+    """
+    Serve one local tree over localhost HTTP for the life of a fixture.
 
-@pytest.fixture(scope="session")
-def http_das_path(http_test_data_root, ensure_http_fetch_file):
-    """Return a UPath pointing at a localhost HTTP view of DAS test data."""
-    handler = partial(
-        _SilentSimpleHTTPRequestHandler,
-        directory=str(http_test_data_root),
-    )
+    Threaded so one slow or abandoned connection cannot park the accept loop;
+    a parked single-threaded server left new connections in TCP SYN retry
+    backoff, which appeared as 15+ second stalls or hangs.
+    """
+    handler = partial(handler_cls, directory=str(root))
     server = _ThreadingHTTPServer(("127.0.0.1", 0), handler)
-    server.daemon_threads = True
     thread = threading.Thread(target=server.serve_forever, daemon=True)
     thread.start()
-    probe_path = "example_dasdae_event_1.h5"
     try:
         host, port = server.server_address
-        probe_url = f"http://{host}:{port}/das/{probe_path}"
-        _wait_for_http_server(probe_url, "http_das_path readiness probe")
+        probe_url = f"http://{host}:{port}/das/example_dasdae_event_1.h5"
+        _wait_for_http_server(probe_url, f"{label} readiness probe")
         yield UPath(f"http://{host}:{port}/das")
     finally:
-        with skip_on_timeout(10, "http_das_path teardown"):
+        with skip_on_timeout(10, f"{label} teardown"):
             server.shutdown()
             server.server_close()
             thread.join(timeout=5)
             if thread.is_alive():
-                raise TimeoutError("HTTP test server thread did not exit cleanly.")
+                raise TimeoutError(f"{label} server thread did not exit cleanly.")
+
+
+@pytest.fixture(scope="session")
+def http_das_path(http_test_data_root, ensure_http_fetch_file):
+    """Return a UPath pointing at a localhost HTTP view of DAS test data."""
+    with _serve_das_tree(
+        _SilentSimpleHTTPRequestHandler, http_test_data_root, "http_das_path"
+    ) as path:
+        yield path
 
 
 @pytest.fixture(scope="session")
@@ -321,51 +324,21 @@ def _ensure(fetch_name: str, relative_path: str | Path | None = None) -> Path:
 @pytest.fixture(scope="session")
 def http_regression_das_path(http_regression_data_root, ensure_http_regression_file):
     """Return an isolated HTTP tree containing only the regression fixtures."""
-    handler = partial(
+    with _serve_das_tree(
         _RegressionHTTPRequestHandler,
-        directory=str(http_regression_data_root),
-    )
-    server = _HTTPServer(("127.0.0.1", 0), handler)
-    thread = threading.Thread(target=server.serve_forever, daemon=True)
-    thread.start()
-    probe_path = "example_dasdae_event_1.h5"
-    try:
-        host, port = server.server_address
-        probe_url = f"http://{host}:{port}/das/{probe_path}"
-        _wait_for_http_server(probe_url, "http_regression_das_path readiness probe")
-        yield UPath(f"http://{host}:{port}/das")
-    finally:
-        with skip_on_timeout(10, "http_regression_das_path teardown"):
-            server.shutdown()
-            server.server_close()
-            thread.join(timeout=5)
-            if thread.is_alive():
-                raise TimeoutError(
-                    "HTTP regression server thread did not exit cleanly."
-                )
+        http_regression_data_root,
+        "http_regression_das_path",
+    ) as path:
+        yield path
 
 
 @pytest.fixture(scope="session")
 def http_range_das_path(http_test_data_root, ensure_http_fetch_file):
     """Return a UPath pointing at a localhost HTTP server with range support."""
-    handler = partial(_RangeHTTPRequestHandler, directory=str(http_test_data_root))
-    server = _HTTPServer(("127.0.0.1", 0), handler)
-    thread = threading.Thread(target=server.serve_forever, daemon=True)
-    thread.start()
-    try:
-        host, port = server.server_address
-        probe_url = f"http://{host}:{port}/das/example_dasdae_event_1.h5"
-        _wait_for_http_server(probe_url, "http_range_das_path readiness probe")
-        yield UPath(f"http://{host}:{port}/das")
-    finally:
-        with skip_on_timeout(10, "http_range_das_path teardown"):
-            server.shutdown()
-            server.server_close()
-            thread.join(timeout=5)
-            if thread.is_alive():
-                raise TimeoutError(
-                    "Range-capable HTTP server thread did not exit cleanly."
-                )
+    with _serve_das_tree(
+        _RangeHTTPRequestHandler, http_test_data_root, "http_range_das_path"
+    ) as path:
+        yield path
 
 
 @pytest.fixture(scope="session")
diff --git a/tests/test_io/test_io_core.py b/tests/test_io/test_io_core.py
index 375afbd91..efe194df9 100644
--- a/tests/test_io/test_io_core.py
+++ b/tests/test_io/test_io_core.py
@@ -1599,6 +1599,63 @@ def test_unsupported_type(self, dummy_text_file):
         out = dc.read(dummy_text_file, name, version)
         assert out == Path(dummy_text_file).read_text()
 
+    def test_handle_closed_when_method_raises(self, dummy_text_file, monkeypatch):
+        """A handle opened by the caster must close when the method raises."""
+
+        class _Recorder:
+            closed = False
+
+            def close(self):
+                self.closed = True
+
+        recorder = _Recorder()
+        monkeypatch.setattr(
+            "dascore.io.core.get_handle_from_resource",
+            lambda resource, required_type: recorder,
+        )
+
+        class _Exploder(FiberIO):
+            name = "_ExploderIO"
+            version = "1"
+
+            def read(self, resource: BinaryReader, **kwargs):
+                raise ValueError("mid-read failure")
+
+        with pytest.raises(ValueError, match="mid-read failure"):
+            _Exploder().read(dummy_text_file)
+        assert recorder.closed
+
+    def test_handle_aborted_when_write_raises(self, dummy_text_file, monkeypatch):
+        """A failed write must discard its handle, not commit a partial file."""
+
+        class _Recorder:
+            aborted = False
+            closed = False
+
+            def abort(self):
+                self.aborted = True
+
+            def close(self):
+                self.closed = True
+
+        recorder = _Recorder()
+        monkeypatch.setattr(
+            "dascore.io.core.get_handle_from_resource",
+            lambda resource, required_type: recorder,
+        )
+
+        class _BadWriter(FiberIO):
+            name = "_BadWriterIO"
+            version = "1"
+
+            def write(self, patch, resource: BinaryWriter, **kwargs):
+                raise ValueError("mid-write failure")
+
+        with pytest.raises(ValueError, match="mid-write failure"):
+            _BadWriter().write(None, dummy_text_file)
+        assert recorder.aborted
+        assert not recorder.closed
+
 
 class TestGetSupportedIOTable:
     """A test for creating the supported io table."""
diff --git a/tests/test_io/test_remote_http.py b/tests/test_io/test_remote_http.py
index 3c3f82e0f..319baf7b6 100644
--- a/tests/test_io/test_remote_http.py
+++ b/tests/test_io/test_remote_http.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import sys
 from urllib.request import Request, urlopen
 
 import pytest
@@ -13,20 +12,14 @@
 from dascore.exceptions import InvalidSpoolError, RemoteCacheError
 from dascore.utils.misc import suppress_warnings
 from dascore.utils.remote_io import clear_remote_file_cache, get_remote_cache_path
-from tests.test_io._common_io_test_utils import skip_on_timeout
 
-# The localhost HTTP + fsspec/aiohttp streaming path intermittently deadlocks on
-# Windows (the async read stalls while h5py probes remote HDF5 metadata, which
-# pytest-timeout then aborts). This is a known Windows flakiness in that fallback
-# path, not a DASCore logic issue. Skip the localhost-HTTP tests on Windows to
-# keep CI deterministic; Linux and macOS still exercise them fully.
+# The intermittent stall formerly attributed to Windows was a platform-agnostic
+# deadlock between h5py's global lock and garbage collection on fsspec's
+# event-loop thread, fixed by pausing collection around remote h5py handles
+# (see dascore.utils.remote_io.pause_gc), so these tests run on all platforms.
 pytestmark = [
     pytest.mark.network,
     pytest.mark.timeout(30),
-    pytest.mark.skipif(
-        sys.platform == "win32",
-        reason="Flaky localhost-HTTP fsspec/aiohttp streaming on Windows.",
-    ),
 ]
 
 
@@ -181,11 +174,12 @@ def test_http_range_hdf5_read_succeeds(
         path = http_range_das_path / "prodml_2.1.h5"
         fmt = dc.get_format(path)
         assert fmt == ("PRODML", "2.1")
-        # Note: this path is intermittently hanging in CI/local repro during
-        # the live ranged HDF5 read, so keep the skip narrowly scoped here.
-        # TODO: root-cause the ranged HTTP/fsspec/h5py stall and remove this.
-        with skip_on_timeout(15, "http_range_hdf5_read_succeeds dc.read"):
-            spool = dc.read(path)
+        # This read used to deadlock: h5py holds its global lock while
+        # delegating fetches to fsspec's event-loop thread, and a garbage
+        # collection on that thread deallocating h5py objects needed the same
+        # lock. Remote h5py handles now pause automatic collection, see
+        # dascore.utils.remote_io.pause_gc.
+        spool = dc.read(path)
         assert spool
         cached = list(get_remote_cache_path().rglob("prodml_2.1.h5"))
         assert not cached
diff --git a/tests/test_utils/test_gc_pause.py b/tests/test_utils/test_gc_pause.py
new file mode 100644
index 000000000..bfa0fdfee
--- /dev/null
+++ b/tests/test_utils/test_gc_pause.py
@@ -0,0 +1,279 @@
+"""Tests for the garbage-collection pause used by remote HDF5 reads."""
+
+from __future__ import annotations
+
+import gc
+import io
+import os
+import threading
+from contextlib import suppress
+from types import SimpleNamespace
+
+import pytest
+
+import dascore.utils.remote_io as remote_io
+from dascore.utils.hdf5 import (
+    _is_loop_backed,
+    _ManagedH5pyFile,
+    _open_h5_fileobj,
+)
+from dascore.utils.remote_io import pause_gc, resume_gc
+
+
+class _FakeFS:
+    """Stand-in for an fsspec filesystem."""
+
+    def __init__(self, async_impl):
+        self.async_impl = async_impl
+
+
+class _FakeRemoteFile(io.RawIOBase):
+    """A file object whose filesystem serves reads on a loop thread."""
+
+    def __init__(self):
+        self.fs = _FakeFS(True)
+
+    def readable(self):
+        return True
+
+
+def _raise_keyboard_interrupt(*args, **kwargs):
+    """Stand in for an interrupt landing inside an uninterruptible step."""
+    raise KeyboardInterrupt("interrupted")
+
+
+def _open_paused(fileobj, constructor=lambda fileobj, **kwargs: object()):
+    """Open a fake loop-backed file object the way the UPath branch does."""
+    return _open_h5_fileobj(fileobj, constructor, "r", pause=True, close_on_error=True)
+
+
+class TestDeadlockProperty:
+    """The pause must actually prevent the h5py/loop-thread deadlock."""
+
+    @pytest.mark.concurrency
+    def test_pause_prevents_loop_thread_deadlock(self):
+        """A collection on the loop thread cannot wedge a lock-holding reader.
+
+        Models the real cycle: a reader holds h5py's global lock and waits on
+        a loop thread, while that thread's collection finalizes an object
+        needing the same lock. Without the pause this deadlocks.
+
+        This calls pause_gc directly; that the HDF5 open paths reach it is
+        covered by test_remote_h5_handle_pauses_gc and its neighbours.
+        """
+        phil = threading.RLock()
+        request = threading.Semaphore(0)
+        answer = threading.Semaphore(0)
+        stop = threading.Event()
+
+        class _NeedsPhil:
+            """Its finalizer takes the lock, as a dead h5py object would."""
+
+            def __init__(self):
+                self.self_ref = self  # a cycle: only gc can free it
+
+            def __del__(self):
+                with phil:
+                    pass
+
+        def loop_thread():
+            """Serve reads and make cyclic garbage while doing it."""
+            while not stop.is_set():
+                if not request.acquire(timeout=0.5):
+                    continue
+                for _ in range(200):
+                    _NeedsPhil()
+                answer.release()
+
+        server = threading.Thread(target=loop_thread, daemon=True)
+        server.start()
+        pause_gc()
+        try:
+            for _ in range(20):
+                with phil:  # h5py holds its lock across the fetch
+                    request.release()
+                    assert answer.acquire(timeout=20), "deadlocked"
+        finally:
+            stop.set()
+            resume_gc()
+            server.join(timeout=5)
+        gc.collect()
+
+
+class TestPauseAccounting:
+    """The pause must never be left on, and never lifted early."""
+
+    def test_nests(self):
+        """Only the outermost resume re-enables collection."""
+        pause_gc()
+        pause_gc()
+        try:
+            assert not gc.isenabled()
+            resume_gc()
+            assert not gc.isenabled()
+        finally:
+            resume_gc()
+        assert gc.isenabled()
+
+    def test_unbalanced_resume_is_a_no_op(self):
+        """A stray resume cannot enable collection during a live session."""
+        resume_gc()
+        pause_gc()
+        try:
+            assert not gc.isenabled()
+        finally:
+            resume_gc()
+
+    def test_user_disabled_gc_is_not_re_enabled(self):
+        """A caller who disabled collection keeps it disabled."""
+        was_enabled = gc.isenabled()
+        gc.disable()
+        try:
+            pause_gc()
+            resume_gc()
+            assert not gc.isenabled()
+        finally:
+            if was_enabled:
+                gc.enable()
+
+    def test_leaked_handle_resumes(self):
+        """Dropping a handle without closing it releases the pause."""
+        _open_paused(_FakeRemoteFile())
+        gc.collect()
+        assert gc.isenabled()
+
+    def test_stranded_cyclic_handle_is_recovered(self, monkeypatch):
+        """A handle leaked inside a cycle is healed by the next remote open."""
+        holder = {}
+        handle = _open_paused(_FakeRemoteFile())
+        holder["handle"], holder["self"] = handle, holder  # unreachable cycle
+        del handle, holder
+        assert not gc.isenabled()
+        monkeypatch.setattr(remote_io, "_gc_collect_after", 0.0)  # rate limited
+        pause_gc()
+        resume_gc()
+        assert gc.isenabled()
+
+    def test_interrupted_safety_collect_takes_no_pause(self, monkeypatch):
+        """An interrupt during the safety collect cannot strand a pause."""
+        fake_gc = SimpleNamespace(
+            collect=_raise_keyboard_interrupt,
+            isenabled=gc.isenabled,
+            disable=gc.disable,
+            enable=gc.enable,
+        )
+        monkeypatch.setattr(remote_io, "gc", fake_gc)
+        monkeypatch.setattr(remote_io, "_gc_collect_after", 0.0)  # rate limited
+        with pytest.raises(KeyboardInterrupt):
+            pause_gc()
+        assert gc.isenabled()
+
+    def test_open_failure_releases_pause_and_fileobj(self):
+        """A failed open leaves neither the pause nor the file object behind."""
+        fileobj = _FakeRemoteFile()
+
+        def _raise(*args, **kwargs):
+            raise ValueError("no")
+
+        with pytest.raises(ValueError):
+            _open_paused(fileobj, _raise)
+        assert gc.isenabled()
+        assert fileobj.closed
+
+    def test_teardown_error_still_resumes(self):
+        """A BaseException from the owned file object cannot strand the pause."""
+
+        class _BadClose:
+            def close(self):
+                raise KeyboardInterrupt("interrupted mid-close")
+
+        class _Handle:
+            def close(self):
+                pass
+
+        pause_gc()
+        managed = _ManagedH5pyFile(_Handle(), _BadClose(), gc_paused=True)
+        with suppress(KeyboardInterrupt):
+            managed.close()
+        assert gc.isenabled()
+
+    def test_inherited_handle_does_not_resume_after_fork(self):
+        """A handle carried through a fork cannot resume the child's session."""
+        pause_gc()
+        try:
+
+            class _Handle:
+                def close(self):
+                    pass
+
+            managed = _ManagedH5pyFile(_Handle(), None, gc_paused=True)
+            managed._gc_paused_pid = os.getpid() + 1  # as if inherited
+            managed.close()
+            assert not gc.isenabled(), "an inherited close stole a live pause"
+        finally:
+            resume_gc()
+
+    def test_fork_reset_clears_inherited_pause(self):
+        """The fork hook drops a pause no child close can rebalance."""
+        pause_gc()
+        remote_io._reset_gc_pause_state()
+        assert gc.isenabled()
+        assert remote_io._gc_pause_depth == 0
+
+    @pytest.mark.concurrency
+    def test_concurrent_sessions_keep_the_count_exact(self, run_in_threads):
+        """Overlapping pauses must not lose or double-count each other."""
+
+        def _pause_and_resume(_index):
+            for _ in range(50):
+                pause_gc()
+                assert not gc.isenabled()
+                resume_gc()
+
+        run_in_threads(_pause_and_resume)
+        assert remote_io._gc_pause_depth == 0
+        assert gc.isenabled()
+
+
+class TestLoopBackedDetection:
+    """Missing a loop-backed object leaves the deadlock window open."""
+
+    def test_async_filesystem_detected(self):
+        """A file object over an async filesystem is loop backed."""
+        assert _is_loop_backed(_FakeRemoteFile())
+
+    def test_wrapped_async_filesystem_detected(self):
+        """Buffering wrappers hide the filesystem but not the loop thread."""
+        wrapped = io.BufferedReader(_FakeRemoteFile())
+        assert _is_loop_backed(wrapped)
+
+    def test_plain_stream_is_not_loop_backed(self):
+        """Local and in-memory streams must not pause collection."""
+        assert not _is_loop_backed(io.BytesIO(b"abc"))
+
+    def test_unavailable_backend_is_not_loop_backed(self):
+        """A path whose backend is not installed raises on ``fs``, not here."""
+
+        class _MissingBackend:
+            @property
+            def fs(self):
+                raise ImportError("please install some-fs")
+
+        assert not _is_loop_backed(_MissingBackend())
+
+    def test_refused_attribute_is_not_loop_backed(self):
+        """A wrapper may refuse an attribute with more than AttributeError."""
+
+        class _Refusing:
+            def __getattr__(self, name):
+                raise io.UnsupportedOperation(name)
+
+        assert not _is_loop_backed(_Refusing())
+
+    def test_sync_filesystem_is_not_loop_backed(self):
+        """A synchronous fsspec filesystem needs no pause."""
+
+        class _SyncFile:
+            fs = _FakeFS(False)
+
+        assert not _is_loop_backed(_SyncFile())
diff --git a/tests/test_utils/test_io_utils.py b/tests/test_utils/test_io_utils.py
index 0f2b6f995..77d0afee3 100644
--- a/tests/test_utils/test_io_utils.py
+++ b/tests/test_utils/test_io_utils.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import gc
 import threading
 from contextlib import closing
 from io import BufferedReader, BufferedWriter, BytesIO, StringIO, TextIOBase
@@ -9,6 +10,7 @@
 
 import h5py
 import pytest
+from fsspec.asyn import AsyncFileSystem
 from upath import UPath
 
 import dascore as dc
@@ -46,6 +48,28 @@
 )
 
 
+class _DummyHandle:
+    """A file-like stand-in that only records being closed."""
+
+    closed = False
+
+    def close(self):
+        self.closed = True
+
+
+class _FakeAsyncFS(AsyncFileSystem):
+    """A stand-in async fsspec filesystem which needs no event loop."""
+
+    def __init__(self):
+        pass
+
+
+class _FakeRemoteFile(BytesIO):
+    """A file object whose fsspec filesystem serves reads on a loop thread."""
+
+    fs = _FakeAsyncFS()
+
+
 class _BadType:
     """A dummy type for testing."""
 
@@ -359,13 +383,6 @@ def test_h5_reader_closes_upath_handle_on_constructor_error(
         self, tmp_path, monkeypatch
     ):
         """Ensure constructor failures close UPath-opened file handles."""
-
-        class _DummyHandle:
-            closed = False
-
-            def close(self):
-                self.closed = True
-
         handle = _DummyHandle()
         path = UPath(tmp_path / "error.h5")
         path.write_bytes(b"not an hdf5")
@@ -381,33 +398,132 @@ def close(self):
             H5Reader.get_handle(path)
         assert handle.closed
 
-    def test_h5_reader_uses_small_blocks_for_s3_upath(self, monkeypatch):
-        """S3-backed HDF5 readers should override s3fs's large default block."""
-
-        class _DummyHandle:
-            closed = False
-
-            def close(self):
-                self.closed = True
-
+    @pytest.mark.parametrize(
+        ("url", "options", "expected"),
+        [
+            (
+                "s3://example-bucket/example.h5",
+                {"anon": True},
+                {"cache_type": "readahead"},
+            ),
+            (
+                "http://example.com/example.h5",
+                {},
+                {"cache_type": "blockcache", "cache_options": {"maxblocks": 8}},
+            ),
+            (
+                "https://example.com/example.h5",
+                {},
+                {"cache_type": "blockcache", "cache_options": {"maxblocks": 8}},
+            ),
+        ],
+    )
+    def test_remote_h5_open_kwargs_are_tuned(self, monkeypatch, url, options, expected):
+        """Remote HDF5 opens must override backend defaults that overfetch.
+
+        s3fs defaults to 50 MB readahead blocks. HTTP instead needs a block
+        LRU: the h5py metadata probe alternates between the file header and
+        footer, and a single-window cache refetches on every jump.
+        """
         opened = {}
-        handle = _DummyHandle()
-        path = UPath("s3://example-bucket/example.h5", anon=True)
+        path = UPath(url, **options)
 
         def _open(_self, _mode, **kwargs):
             opened.update(kwargs)
-            return handle
+            return _DummyHandle()
 
         monkeypatch.setattr(type(path), "open", _open)
         monkeypatch.setattr(
             H5Reader,
             "constructor",
-            staticmethod(lambda *args, **kwargs: object()),
+            staticmethod(lambda *args, **kwargs: _DummyHandle()),
         )
         with config_context(remote_hdf5_block_size=1234):
-            H5Reader.get_handle(path)
+            H5Reader.get_handle(path).close()
         assert opened["block_size"] == 1234
-        assert opened["cache_type"] == "readahead"
+        for key, value in expected.items():
+            assert opened[key] == value
+
+    def test_remote_h5_handle_pauses_gc(self, monkeypatch):
+        """Automatic collection stays paused while a remote handle is open."""
+        path = UPath("http://example.com/gc-pause.h5")
+        # Stand in for the real filesystem, which needs aiohttp; some
+        # platforms DASCore supports (wasm, free-threaded) do not have it.
+        monkeypatch.setattr(type(path), "fs", property(lambda _self: _FakeAsyncFS()))
+        monkeypatch.setattr(type(path), "open", lambda *a, **k: _DummyHandle())
+        monkeypatch.setattr(
+            H5Reader,
+            "constructor",
+            staticmethod(lambda *args, **kwargs: _DummyHandle()),
+        )
+        assert gc.isenabled()
+        handle = H5Reader.get_handle(path)
+        try:
+            assert not gc.isenabled()
+        finally:
+            handle.close()
+        assert gc.isenabled()
+        # Closing twice must not unbalance the pause bookkeeping.
+        handle.close()
+        assert gc.isenabled()
+
+    def test_dropped_wrapper_leaves_caller_handle_open(self, generic_hdf5):
+        """Collecting a wrapper must not close the handle its caller owns."""
+        file = h5py.File(generic_hdf5, "r")
+        try:
+            wrapper = H5Reader.get_handle(file)
+            assert wrapper is not file
+            del wrapper
+            gc.collect()
+            assert file  # h5py files are falsey once closed
+        finally:
+            file.close()
+
+    def test_failed_open_leaves_caller_fileobj_usable(self):
+        """A caller's file object must survive a failed HDF5 open.
+
+        get_format hands the same object to every FiberIO in turn, so an
+        HDF5 miss cannot close it out from under the next one.
+        """
+        buffer = BytesIO(b"not an hdf5 file")
+        with pytest.raises(OSError):
+            H5Reader.get_handle(buffer)
+        assert not buffer.closed
+
+    def test_local_upath_does_not_pause_gc(self, generic_hdf5):
+        """A synchronous backend has no loop thread, so it must not pause gc."""
+        handle = H5Reader.get_handle(UPath(generic_hdf5))
+        try:
+            assert gc.isenabled()
+        finally:
+            handle.close()
+
+    def test_loop_backed_fileobj_pauses_gc(self, monkeypatch):
+        """User-supplied fsspec async file objects need the GC pause too."""
+        monkeypatch.setattr(
+            H5Reader,
+            "constructor",
+            staticmethod(lambda *args, **kwargs: BytesIO()),
+        )
+        assert gc.isenabled()
+        handle = H5Reader.get_handle(_FakeRemoteFile())
+        try:
+            assert not gc.isenabled()
+        finally:
+            handle.close()
+        assert gc.isenabled()
+
+    def test_loop_backed_fileobj_constructor_error_resumes_gc(self, monkeypatch):
+        """A failed h5py construction must rebalance the GC pause."""
+
+        def _explode(*args, **kwargs):
+            raise ValueError("not an hdf5 fileobj")
+
+        monkeypatch.setattr(H5Reader, "constructor", staticmethod(_explode))
+        assert gc.isenabled()
+        with pytest.raises(ValueError, match="not an hdf5 fileobj"):
+            H5Reader.get_handle(_FakeRemoteFile())
+        assert gc.isenabled()
 
     def test_h5_writer_to_remote_upath(self):
         """HDF5 writers should create remote UPath files via write-back."""
@@ -534,6 +650,71 @@ def test_get_none_resource_returns_source(self):
         with IOResourceManager(source) as man:
             assert man.get_resource(None) is source
 
+    def test_error_in_context_aborts_handles(self):
+        """An exception inside the context must abort, not commit, handles."""
+
+        class _Recorder:
+            aborted = False
+            closed = False
+
+            def abort(self):
+                self.aborted = True
+
+            def close(self):
+                self.closed = True
+
+        recorder = _Recorder()
+        man = IOResourceManager("unused")
+        man._cache["key"] = recorder
+        with pytest.raises(ValueError, match="boom"):
+            with man:
+                raise ValueError("boom")
+        assert recorder.aborted
+        assert not recorder.closed
+        # A clean exit closes normally.
+        recorder2 = _Recorder()
+        man2 = IOResourceManager("unused")
+        man2._cache["key"] = recorder2
+        with man2:
+            pass
+        assert recorder2.closed
+        assert not recorder2.aborted
+
+    def test_failed_abort_does_not_mask_original_error(self):
+        """A cleanup failure must not replace the error which caused it."""
+
+        class _BadAbort:
+            def abort(self):
+                raise OSError("abort failed")
+
+        man = IOResourceManager("unused")
+        man._cache["key"] = _BadAbort()
+        with pytest.raises(ValueError, match="boom") as exc_info:
+            with man:
+                raise ValueError("boom")
+        assert any("abort failed" in note for note in exc_info.value.__notes__)
+
+    def test_close_all_survives_failing_handle(self):
+        """One handle raising must not skip cleanup of the others."""
+
+        class _Exploder:
+            def close(self):
+                raise OSError("close failed")
+
+        class _Recorder:
+            closed = False
+
+            def close(self):
+                self.closed = True
+
+        recorder = _Recorder()
+        man = IOResourceManager("unused")
+        man._cache["bad"] = _Exploder()
+        man._cache["good"] = recorder
+        with pytest.raises(OSError, match="close failed"):
+            man.close_all()
+        assert recorder.closed
+
     def test_non_pathlike_resource_passthrough(self):
         """Non-pathlike resources should bypass path coercion entirely."""
         source = BytesIO(b"abc")
@@ -922,6 +1103,12 @@ def test_no_range_error_predicate_matches_expected_message(self):
         assert not is_no_range_http_error(ValueError("different error"))
         assert not is_no_range_http_error(RuntimeError("range requests"))
 
+    def test_no_range_error_predicate_matches_streaming_seek(self):
+        """Seeking a streaming (size-less) HTTP file needs the same fallback."""
+        exc = ValueError("Cannot seek streaming HTTP file")
+        assert is_no_range_http_error(exc)
+        assert not is_no_range_http_error(RuntimeError(str(exc)))
+
 
 class TestFallbackFileObj:
     """Tests for switching failed remote handles to local cache files."""