Read snapshot artifacts via async fsspec API - #21
Merged
Conversation
Reading artifacts with the synchronous fs.open() crashed on the gs:// backend with "RuntimeError: got Future attached to a different loop". Pravda binds gcsfs's aiohttp session to the pipeline event loop by awaiting its async API during capture; the subsequent sync fs.open() dispatched the same cached filesystem instance's coroutine to fsspec's background loop, mixing loops on one session. Wrap synchronous backends in AsyncFileSystemWrapper (as Pravda's Storage.from_url does) and read via `await fs._cat_file()`, keeping all storage I/O on the running loop. Local (sync) backends are unaffected; this only manifested on remote async storage in the cluster. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The read-side fix moved the pipeline past capture and extraction, but it
then crashed identically ("got Future attached to a different loop") in
write_outputs, which used the synchronous fs.makedirs()/fs.open() on the
gs:// backend from inside the pipeline event loop.
Give write_outputs the same treatment as read_artifact: wrap sync
backends and write via `await fs._makedirs()` / `await fs._pipe_file()`,
keeping all storage I/O on the running loop. load_inputs is unaffected —
it runs before asyncio.run(), outside the pipeline loop.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Read Pravda snapshot artifacts through the async fsspec API instead of the synchronous
fs.open().Why
A
kolkhoz-manualjob on GKE crashed in the extraction phase reading artifacts back from thegs://bucket:Captures themselves succeeded — the crash was on the first
read_artifact. Root cause is an event-loop mismatch on a shared, fsspec-cachedgcsfsinstance:asyncio.runloop, binding gcsfs's aiohttp session to that loop.read_artifactcalled the synchronousfs.open(), which fsspec dispatches to its own background loop — reusing the session bound to the pipeline loop from a different loop.It only manifests on a remote async backend: the local (
LocalFileSystem) path is synchronous, so dev/tests never hit it — which is why this reached the cluster.Change
storage_filesystem: wrap synchronous backends inAsyncFileSystemWrapper, mirroring Pravda's ownStorage.from_url, so reads use the async API on every backend.read_artifact: nowasync, reads viaawait fs._cat_file(path); the three call sites inextract_snapshotawaitit.This keeps all storage I/O on the running event loop, so Pravda's writes and Kolkhoz's reads share one session.
Testing
ruff checkclean.gcsfsboth expose an awaitable_cat_file.gs://on GKE — the definitive check is akolkhoz-manualjob run. Recommend confirming there before relying on it.🤖 Generated with Claude Code