Hi Community, I have encountered a very strange behavior with retries on the flow level. Here is my flow that connects to an SFTP server and loads data from it. If a file is not found on the SFTP server, I close the connection and raise an exception. I expect that the flow will retry to re-run from the beginning, starting from the connection to the SFTP server. But it fails with the error “Socket is closed.” It appears to me that Prefect caches the connection and does not re-run the flow from the beginning. What is the best practice for restarting connection on retry?

class SFTPClient(Base):
    def __init__(self, block_name: str, broker_name: str = None) -> None:
        self.rfs = RemoteFileSystem.load(block_name)
    
    def load_file_data(self, remote_path: str) -> pd.DataFrame:
        ...


@flow(
    name="Transfer files to S3 bucket",
    retries=20,
    retry_delay_seconds=5,
)
def flowrun(date: str = None):
    context = get_run_context()

    sftclient = SFTPClient(block_name="rjo-sftp", broker_name="RJO")
    data = sftclient.load_file_data(remote_path="/tmp/test.csv")

    if not data:
        sftclient.rfs.filesystem.client.close()
        sftclient = None
        raise Exception("No data")