pipe_file fails to invalidate directory cache

[TomAugspurger]

The sequence of calls ls -> pipe_file -> ls causes the newly written file to not be found in the second ls. Here's a failing test (using Zarr's test setup, which I think was taken from s3fs).

async def test_invalidate_cache(s3: s3fs.S3FileSystem) -> None:
    await s3._call_s3("put_object", Bucket=test_bucket_name, Key="a/b.txt", Body=b"abc")
    before = await s3._ls(f"{test_bucket_name}/a/")
    assert sorted(before) == ["test/a/b.txt"]
    await s3._pipe_file(f"{test_bucket_name}/a/c.txt", data=b"abc")
    after = await s3._ls(f"{test_bucket_name}/a/")
    assert sorted(after) == ["test/a/b.txt", "test/a/c.txt"]

That fails with

    async def test_invalidate_cache(s3: s3fs.S3FileSystem) -> None:
        await s3._call_s3("put_object", Bucket=test_bucket_name, Key="a/b.txt", Body=b"abc")
        before = await s3._ls(f"{test_bucket_name}/a/")
        assert sorted(before) == ["test/a/b.txt"]
        await s3._pipe_file(f"{test_bucket_name}/a/c.txt", data=b"abc")
        after = await s3._ls(f"{test_bucket_name}/a/")
>       assert sorted(after) == ["test/a/b.txt", "test/a/c.txt"]
E       AssertionError: assert ['test/a/b.txt'] == ['test/a/b.tx...test/a/c.txt']
E         
E         Right contains one more item: 'test/a/c.txt'
E         
E         Full diff:
E           [
E               'test/a/b.txt',
E         -     'test/a/c.txt',
E           ]

tests/test_store/test_remote.py:275: AssertionError

I think the issue is inpipe_file, which has an early return at

s3fs/s3fs/core.py

Lines 1141 to 1143 in dd75a1a

	return await self._call_s3(
	"put_object", Bucket=bucket, Key=key, Body=data, **kwargs
	)

. The self.invalidate_cache at the end of the function isn't run.

xref pydata/xarray#9515