Issue Opening Files via SFTP with fsspec in Zarr v3

[dgegen]

I'm currently using Zarr and fsspec to access a Zarr file stored on a server via SFTP. However, I've encountered an issue while transitioning from Zarr v2 to v3.

Description

import zarr
import fsspec

# Specify the SFTP filesystem
storage_options = {
    "key_filename": "/path/to/key",
    "username": "username",
    "host": "host",
}
sftp_fs = fsspec.filesystem('sftp', **storage_options)
zarr_path = "/path/to/file/on/server.zarr"

fs_map = fsspec.FSMap(zarr_path, sftp_fs)

print(f"File exists: {sftp_fs.exists(zarr_path)}")  # True

In Zarr v2.18.3 I am able to open this file using

# Zarr v2.18.3
zarr_file = zarr.open(fs_map)

Now, attempting to achieve the same thing using Zarr v3.0.9 (including an additional import because of #3195), I use:

# Zarr v3.0.9
import fsspec.implementations.asyn_wrapper  # missing import in zarr.storage.FsspecStore
from zarr.storage import FsspecStore

store = FsspecStore.from_mapper(fs_map, read_only=True)
zarr_file = zarr.open(store)

However, the call zarr_file = zarr.open(store), never finishes, i.e. Python freezes.

Is this a bug? Or will SFPT no longer be supported in Zarr v3? Or am I using FsspecStore incorrectly?

Thanks a lot in advance!

[d-v-b]

However, the call zarr_file = zarr.open(store), never finishes, i.e. Python freezes.

that's a surprising failure mode. As our 3.0.9 release has been a bit bumpy, can you try running the same thing in 3.0.8? You will need to skip the fsmap and just give the filesystem to the fsspecstore I think.

[dgegen]

For me, Python also freezes using the following in v3.0.8

# Zarr v3.0.8
from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
from zarr.storage import FsspecStore

store = FsspecStore(AsyncFileSystemWrapper(sftp_fs, asynchronous=True), path=remote_file_path)
zarr.open(store)

[d-v-b]

ok,that's "good" in the sense that it's not related to the 3.0.9 release, but it's bad insofar as it's a bug. Can you confirm that this isn't an fsspec issue?

[dgegen]

I'm not sure I understand what you mean by that. However, using the same environment, merely downgrading Numcodecs and Zarr using

pip install zarr==2.18.3
pip install numcodecs==0.13.1

and of course using the snippet for Zarr v2 resolves the issue. So in principal the file can be opened with fsspec v2025.5.1.

[d-v-b]

I'm not sure I understand what you mean by that.

Fsspecstore uses fsspec for all of its IO. So if fsspec's sftp implementation has a problem, then zarr's usage of fsspec's sftp implementation will reveal that problem, even if it comes from fsspec. It's worth noting that there is no sftp-specific logic in zarr's use of fsspec -- we are just using the fsspec filesystem object, no matter what storage backend it uses. So that's why I am wondering if the basic problem here comes from fsspec.

[dgegen]

However, considering that I can successfully open the file with the same version of fsspec using zarr v2.18.3, doesn't this suggest that the issue is related to how zarr v3.0.8/v3.0.9 is utilizing fsspec?

[d-v-b]

This would be much easier with a reproducible example that one of the devs could run. Could you create one with the dependencies in the header (like in our issue template)? You could try using the address of a public sftp server

[dgegen]

I encountered difficulties creating directories on the public SFTP servers, which prevented me from writing and reading Zarr files for testing. However, I've adapted a Python package to create a local SFTP server to test this, which can be used to create a reproducible example.

That being said, the issue stems from the use of asyncio.gather in zarr.core.array.get_array_metadata, which attempts to read multiple metadata files (e.g., .zarray, .zattrs, zarr.json) concurrently. This works well for truly asynchronous filesystems, but breaks when using systems like SFTPFileSystem, which does not seem to be concurrency-safe in async contexts (potentially relying on blocking I/O internally or managing connection states using global locks) leading to deadlocks or indefinite hangs when asyncio.gather is used to perform multiple reads simultaneously. I have opened #3207 to propose a solution for this.

Reproducible example for completeness

Make sure to first install LocalSFTPServer:
pip install git+https://github.com/dgegen/LocalSFTPServer.git

import os
from pathlib import Path

import numpy as np
import fsspec
from localsftpserver import SFTPServer

import zarr
import zarr.storage


def create_zarr_file(server_dir):
    """Create a Zarr file in the specified server directory."""
    zarr_path = server_dir / "example.zarr"
    if not zarr_path.exists():
        z = zarr.open(zarr_path, mode='w', shape=(4, 4), chunks=(1, 1), dtype='i1')
        z[:, :] = np.random.randint(low=-128, high=128, size=(4, 4), dtype=np.int8)
    return zarr_path


def setup_paths():
    tmp_dir = Path(os.getcwd()) / "tmp"
    tmp_dir.mkdir(parents=True, exist_ok=True)
    server_dir = tmp_dir / "sftp_server"
    server_dir.mkdir(exist_ok=True)
    key_path = tmp_dir / "key.pem"
    return tmp_dir, server_dir, key_path


tmp_dir, server_dir, key_path = setup_paths()


with SFTPServer(
    root_path=str(server_dir),
) as server:
    print(f"SFTP server started at {server.host}:{server.port} with key {server.key_filename}")
    
    zarr_path = create_zarr_file(server_dir)

    # Connect to the SFTP server and open the Zarr file
    storage_options = {
        "host": server.host,
        "port": server.port,
        "username": "",
        "password": "",
        "key_filename": str(server.key_filename),
        "timeout": 10,
    }

    sftp_fs = fsspec.filesystem("sftp", **storage_options)
    fs_map = fsspec.FSMap(str(zarr_path), sftp_fs)

    remote_file_path = str(zarr_path.relative_to(server_dir))

    if os.path.exists(zarr_path):
        print(f"Zarr file exists.")
    else:
        raise FileNotFoundError(f"Remote file {remote_file_path} does not exist.")
    
    try:
        from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
        store = zarr.storage.FsspecStore(
            AsyncFileSystemWrapper(sftp_fs, asynchronous=False), path=str(zarr_path)
        )
    except ImportError:
        store = fs_map


    print(f"Opening Zarr store at {zarr_path} on SFTP server...")
    opened_zarr = zarr.open(store)
    print(f"Zarr store successfully opened!")

[d-v-b]

@dgegen thanks for the reproducer! I modified your script so that it declares its dependencies and can then be run by uv / hatch, and I can replicate the freezing you observed:

# /// script
# requires-python = ">=3.11"
# dependencies = [
#   "zarr@git+https://github.com/zarr-developers/zarr-python.git@main",
#  "localsftpserver @ git+https://github.com/dgegen/LocalSFTPServer.git"
# ]
# ///
#
import os
from pathlib import Path

import numpy as np
import fsspec
from localsftpserver import SFTPServer

import zarr
import zarr.storage


def create_zarr_file(server_dir):
    """Create a Zarr file in the specified server directory."""
    zarr_path = server_dir / "example.zarr"
    if not zarr_path.exists():
        z = zarr.open(zarr_path, mode='w', shape=(4, 4), chunks=(1, 1), dtype='i1')
        z[:, :] = np.random.randint(low=-128, high=128, size=(4, 4), dtype=np.int8)
    return zarr_path


def setup_paths():
    tmp_dir = Path(os.getcwd()) / "tmp"
    tmp_dir.mkdir(parents=True, exist_ok=True)
    server_dir = tmp_dir / "sftp_server"
    server_dir.mkdir(exist_ok=True)
    key_path = tmp_dir / "key.pem"
    return tmp_dir, server_dir, key_path


tmp_dir, server_dir, key_path = setup_paths()


with SFTPServer(
    root_path=str(server_dir),
) as server:
    print(f"SFTP server started at {server.host}:{server.port} with key {server.key_filename}")
    
    zarr_path = create_zarr_file(server_dir)

    # Connect to the SFTP server and open the Zarr file
    storage_options = {
        "host": server.host,
        "port": server.port,
        "username": "",
        "password": "",
        "key_filename": str(server.key_filename),
        "timeout": 10,
    }

    sftp_fs = fsspec.filesystem("sftp", **storage_options)
    fs_map = fsspec.FSMap(str(zarr_path), sftp_fs)

    remote_file_path = str(zarr_path.relative_to(server_dir))

    if os.path.exists(zarr_path):
        print(f"Zarr file exists.")
    else:
        raise FileNotFoundError(f"Remote file {remote_file_path} does not exist.")
    
    try:
        from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
        store = zarr.storage.FsspecStore(
            AsyncFileSystemWrapper(sftp_fs, asynchronous=False), path=str(zarr_path)
        )
    except ImportError:
        store = fs_map


    print(f"Opening Zarr store at {zarr_path} on SFTP server...")
    opened_zarr = zarr.open(store)
    print(f"Zarr store successfully opened!")

[martindurant]

I can confirm that the script freezes. However, since it is also running the SFTP server in the same loop, it is more complicated than it needs to be.

Some notes:

• I thought we had added code to allow FSMap objects to be passed in the place of store objects (for legacy compatibility)
• the async wrapper should be asynchronous, because zarr will call it in an asynchronous context
• an asynchronous FS should be created only in a coroutine; this is why you are expected in most cases to pass a URL to zarr, not an instantiated store
• LocalFileSystem works fine, and that should have exactly the same wrapping
• using gather(*coroutines) where those coroutines call blocking functions, will result in sequential calls, although the order is not guaranteed.
• fsspec already implements cat/cat_ranges for fetching many file/pieces concurrently, which will result in sequential calls for blocking IO. It may well not be tested - so I created the following in fsspec/implementations/tests/test_sftp.py , and it passes

@pytest.mark.asyncio
async def test_async(ssh):
    from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
    fs0 = fsspec.get_filesystem_class("sftp")(**ssh)
    fs = AsyncFileSystemWrapper(fs0, asynchronous=True)
    await fs._pipe({"a": b"b", "b": b"c"})
    out = await fs._cat(["a", "b"])
    assert out == {"a": b"b", "b": b"c"}

[dgegen]

@martindurant It's great to have you join the conversation and share your insights on the inner workings of fsspec! :)

I created the following in fsspec/implementations/tests/test_sftp.py , and it passes

Interesting. The test you provided does not pass on either of my MacBooks. If I append it to fsspec/implementations/tests/test_sftp.py, I get an OSError: Failure:

Click here to see full traceback

================================================================== FAILURES ===================================================================
_________________________________________________________________ test_async __________________________________________________________________

ssh = {'host': 'localhost', 'password': 'pass', 'port': 9200, 'username': 'root'}

    @pytest.mark.asyncio
    async def test_async(ssh):
        from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
        fs0 = fsspec.get_filesystem_class("sftp")(**ssh)
        fs = AsyncFileSystemWrapper(fs0, asynchronous=True)
>       await fs._pipe({"a": b"b", "b": b"c"})


fsspec/implementations/tests/test_sftp.py:240:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
# [...]
fsspec/implementations/sftp.py:161: in _open
    f = self.ftp.open(path, mode, bufsize=block_size if block_size else -1)
# [...]
E           OSError: Failure

Furthermore, I can say that merely ensuring everything is called from within a coroutine doesn't resolve the issue I'm facing in the actual used case of an external server, where I end up in a deadlock.

Click here to see example where everything is in coroutines

import asyncio
import fsspec
import zarr
from zarr.storage import FsspecStore
from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper


async def get_zarr(remote_file_path, ssh={}):
    sftp_fs = fsspec.filesystem('sftp', **ssh)
    store = FsspecStore(AsyncFileSystemWrapper(sftp_fs), path=remote_file_path)
    result = zarr.open(store)
    return result

async def main():
    remote_path = # ...
    storage_options = # {...}
    result = await get_zarr(remote_path, storage_options)
    print(result)

asyncio.run(main())

[martindurant]

If you are getting an error and not a deadlock, that's progress - perhaps the docker container didn't come up properly or on time?

I'll try to find time to play with second, zarr-specific example; although it would be helpful to activate logger "fsspec.sftp" so you can see what calls are actually being made.

[martindurant]

The issue is, apparently paramiko itself. It is not designed to be thread-safe, and in this context it is not clear which thread work happens in (I think because of zarr's use of to_thread()). I don't think the deadlock is an async issue - it happens on an sftp blocking call.

https://github.com/fsspec/sshfs might do better in this respect.