Cancelling Client.list Creates Unrecoverable State for Client

[whatamithinking]

Environment

OS: Windows 10
Python Version: 3.13.7
aioftp Version: 0.26.2

Problem

I think a problem in list() is it cannot handle cancellation and have the client recover properly because the buffers still contain data from the previous command. If we cancel in the middle of the list() call, aioftp.errors.StatusCodeError is raised when we try to run another command or call quit() in this case.

This can be reproduced with the following code:

import asyncio
import aioftp
import pathlib

async def run_server():
    root = pathlib.Path.home()
    server = aioftp.Server(users=[aioftp.User("anonymous", "", base_path=root)])
    await server.start("127.0.0.1", 2121)
    print("FTP server running on 127.0.0.1:2121")
    await asyncio.Event().wait()

asyncio.run(run_server())

if you uncomment # clear_buffers(), the client.quit() call is able to run successfully.

import asyncio
import aioftp
import time
from contextlib import suppress


async def clear_buffers(client: aioftp.Client):
    print("clearing buffers...")
    while True:
        await asyncio.sleep(0.250)
        available = len(client.stream.reader._buffer)
        if available == 0:
            break
        await client.stream.read(available)


async def work():
    try:
        client = aioftp.Client()
        await client.connect("127.0.0.1", 2121)
        await client.login(user="anonymous", password="")
        lister = client.list("./", recursive=True, raw_command="MLSD")
        print("Listing...")
        results = await lister
    finally:
        print("DONE")
        # await clear_buffers(client)
        if client:
            await client.quit()


async def main():
    while True:
        task = asyncio.create_task(work())
        await asyncio.sleep(2)
        task.cancel()
        with suppress(asyncio.CancelledError):
            await task


asyncio.run(main())

output without calling clear_buffers()...

Listing...
DONE
Traceback (most recent call last):
  File "c:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\test_client.py", line 24, in work
    results = await lister
              ^^^^^^^^^^^^
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\common.py", line 182, in _to_list
    async for item in self:
        items.append(item)
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\client.py", line 980, in __anext__
    cls.stream = await cls._new_stream(current_path)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\client.py", line 955, in _new_stream
    return await self.get_stream(command, "1xx")
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\client.py", line 1403, in get_stream
    await self.command(command, expected_codes, wait_codes, censor_after)
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\client.py", line 397, in command
    code, info = await self.parse_response()
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\client.py", line 301, in parse_response
    code, rest = await self.parse_line()
                 ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\client.py", line 281, in parse_line
    line = await self.stream.readline()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\common.py", line 673, in readline
    data = await super().readline()
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\common.py", line 79, in wrapper
    return await asyncio.wait_for(f(*args, **kwargs), timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\cmaynes\AppData\Local\Programs\Python\Python313\Lib\asyncio\tasks.py", line 507, in wait_for
    return await fut
           ^^^^^^^^^
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\common.py", line 402, in readline
    return await self.reader.readline()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\cmaynes\AppData\Local\Programs\Python\Python313\Lib\asyncio\streams.py", line 562, in readline
    line = await self.readuntil(sep)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\cmaynes\AppData\Local\Programs\Python\Python313\Lib\asyncio\streams.py", line 677, in readuntil
    await self._wait_for_data('readuntil')
  File "C:\Users\cmaynes\AppData\Local\Programs\Python\Python313\Lib\asyncio\streams.py", line 539, in _wait_for_data
    await self._waiter
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "c:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\test_client.py", line 44, in <module>
    asyncio.run(main())
    ~~~~~~~~~~~^^^^^^^^
  File "C:\Users\cmaynes\AppData\Local\Programs\Python\Python313\Lib\asyncio\runners.py", line 195, in run
    return runner.run(main)
           ~~~~~~~~~~^^^^^^
  File "C:\Users\cmaynes\AppData\Local\Programs\Python\Python313\Lib\asyncio\runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "C:\Users\cmaynes\AppData\Local\Programs\Python\Python313\Lib\asyncio\base_events.py", line 725, in run_until_complete
    return future.result()
           ~~~~~~~~~~~~~^^
  File "c:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\test_client.py", line 38, in main
    await task
  File "c:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\test_client.py", line 29, in work
    await client.quit()
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\client.py", line 1314, in quit
    await self.command("QUIT", "2xx")
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\client.py", line 401, in command
    self.check_codes(expected_codes, code, info)
    ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\cmaynes\Desktop\repos\jma-wireless-oem-test-data-sync-service\.venv\Lib\site-packages\aioftp\client.py", line 331, in check_codes
    raise errors.StatusCodeError(expected_codes, received_code, info)
aioftp.errors.StatusCodeError: Waiting for ('2xx',) but got 150 [' mlsd transfer started']

[sakosha]

nevermind. will try to take a deeper look on my win10 machine

[pohmelie]

It is «by design» of FTP protocol. You have only one command connection. And you can't «interrupt» most of the commands. If you cancel your main client task, and want to continue, just recreate connection. I think most FTP servers are not ready for such live recovery.

In general, if command interrupted on client side, then I'm pretty sure connection state is corrupted, since server can send theses 150 responses as it wants, lets say 10 time per second and it is not an error in terms of protocol. So in case «I list an universe and want to interrupt» the best solution is to close all connections without QUIT command and create new one

[whatamithinking]

It is «by design» of FTP protocol. You have only one command connection. And you can't «interrupt» most of the commands. If you cancel your main client task, and want to continue, just recreate connection. I think most FTP servers are not ready for such live recovery.

In general, if command interrupted on client side, then I'm pretty sure connection state is corrupted, since server can send theses 150 responses as it wants, lets say 10 time per second and it is not an error in terms of protocol. So in case «I list an universe and want to interrupt» the best solution is to close all connections without QUIT command and create new one

I wonder if an abstraction layer where there is a Client which builds a Connection object internally or uses a ConnectionPool would allow for automatically handling of this state management, so that in cases where commands do not run to completion the Client or the ConnectionPool can either try to reset the state of the connection or just drop it and make a new one. This might help support cancellation and be more resilient to issues. A ConnectionPool might even allow for clearer lifecycle management of all connections, handling command and data connections in one central location.