cleverage/flysystem-process-bundle#24 SFTP connection timeout causing archiving step failure

.claude/CLAUDE.md

@@ -6,6 +6,8 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 A Symfony 7.4 demo project showcasing [`cleverage/process-bundle`](https://github.com/cleverage/process-bundle) v5 and its bridge bundles (archive, cache, doctrine, flysystem, rest, soap, ui). The "product" here is not PHP code — it is the **process configurations** under `config/packages/process/`. Each file demonstrates one feature of the process ecosystem. `src/` contains only a handful of glue classes (entities, fixtures, a few custom adapters/DTOs).
 
+> **Convention — how features are exercised here.** Features (including bug reproductions and regression checks) must be demonstrated **primarily through `clever_age_process` configurations**, each named with a `demo.` prefix and placed under `config/packages/process/` (one file per process, auto-imported). Reach for a throwaway PHP script or a custom `src/` command only as a last resort when a behaviour genuinely cannot be expressed as a process; prefer composing existing tasks and (generic) transformers instead. Example: issue-#24 SFTP reproduction lives in `config/packages/process/demo.sftp_stale_connection.yaml`.
+
 ## Environment & commands
 
 Everything runs inside Docker (`.docker/compose.yaml`), driven by the `Makefile`. Do **not** run PHP/composer/console directly on the host — wrap them in the container. The Makefile targets already do this via `docker compose ... run --rm php-fpm`.

.docker/compose.yaml

@@ -80,11 +80,35 @@ services:
       retries: 100
     ports:
       - "${MYSQL_PORT:-3306}:3306"
+
   sftp:
     <<: *base-sftp
     tty: true
-    ports:
-      - "${SFTP_PORT:-22}:22"
+
+  # Same atmoz/sftp image as `sftp`, but with a deliberately short sshd idle
+  # timeout (see sftp/short-timeout.sh). Used by demo.sftp_stale_connection to
+  # reproduce flysystem-process-bundle issue #24; all other demo.sftp_* processes
+  # keep using the normal `sftp` service.
+  sftp-short:
+    <<: *base-sftp
+    tty: true
+    volumes:
+      - ./sftp/data:/home/sftp/data
+      - ./sftp/short-timeout.sh:/etc/sftp.d/short-timeout.sh:ro
+
+  # Mock SFTP server (paramiko) that, after a short idle, injects a stray SFTP
+  # packet while keeping the SSH transport open - reproducing the client's SAP
+  # server behaviour and thus the VERBATIM issue #24 error
+  # ("Expected NET_SFTP_HANDLE or NET_SFTP_STATUS. Got packet type").
+  sftp-mock:
+    build:
+      context: sftp-mock
+    tty: true
+    volumes:
+      - ./sftp/data:/data
+    environment:
+      SFTP_MOCK_ROOT: /
+      SFTP_MOCK_IDLE: "4"
 
 volumes:
   process_bundle_demo_data:

.docker/sftp-mock/Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.12-slim
+
+RUN pip install --no-cache-dir "paramiko>=3,<4"
+
+COPY server.py /server.py
+
+EXPOSE 22
+CMD ["python", "/server.py"]

.docker/sftp-mock/server.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""
+Minimal mock SFTP server (paramiko) used to reproduce the EXACT flysystem-process-bundle
+issue #24 error "Expected NET_SFTP_HANDLE or NET_SFTP_STATUS. Got packet type".
+
+It serves files normally, but after a short idle period on an established SFTP
+session it injects one stray SFTP packet (SSH_FXP_NAME) onto the channel WHILE
+KEEPING THE SSH TRANSPORT OPEN. This mimics enterprise SFTP servers (e.g. the
+client's SAP server) that expire the SFTP session / leave a stray or late response
+instead of tearing down the TCP transport the way OpenSSH does. phpseclib then reads
+that stray packet on the next reused operation and raises the issue #24 error
+(whereas a clean OpenSSH disconnect yields "Connection closed prematurely").
+"""
+import os
+import socket
+import threading
+import time
+
+import paramiko
+from paramiko.sftp import CMD_NAME
+
+ROOT = os.environ.get("SFTP_MOCK_ROOT", "/data")
+IDLE = float(os.environ.get("SFTP_MOCK_IDLE", "4"))
+HOST_KEY = paramiko.RSAKey.generate(2048)
+
+
+class Server(paramiko.ServerInterface):
+    def check_auth_password(self, username, password):
+        if username == "sftp" and password == "password":
+            return paramiko.AUTH_SUCCESSFUL
+        return paramiko.AUTH_FAILED
+
+    def get_allowed_auths(self, username):
+        return "password"
+
+    def check_channel_request(self, kind, chanid):
+        if kind == "session":
+            return paramiko.OPEN_SUCCEEDED
+        return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
+
+
+class StubSFTPHandle(paramiko.SFTPHandle):
+    def stat(self):
+        try:
+            return paramiko.SFTPAttributes.from_stat(os.fstat(self.readfile.fileno()))
+        except OSError as e:
+            return paramiko.SFTPServer.convert_errno(e.errno)
+
+
+class StubSFTPServer(paramiko.SFTPServerInterface):
+    def _realpath(self, path):
+        return ROOT + self.canonicalize(path)
+
+    def list_folder(self, path):
+        p = self._realpath(path)
+        out = []
+        for fname in os.listdir(p):
+            attr = paramiko.SFTPAttributes.from_stat(os.stat(os.path.join(p, fname)))
+            attr.filename = fname
+            out.append(attr)
+        return out
+
+    def stat(self, path):
+        try:
+            return paramiko.SFTPAttributes.from_stat(os.stat(self._realpath(path)))
+        except OSError as e:
+            return paramiko.SFTPServer.convert_errno(e.errno)
+
+    def lstat(self, path):
+        try:
+            return paramiko.SFTPAttributes.from_stat(os.lstat(self._realpath(path)))
+        except OSError as e:
+            return paramiko.SFTPServer.convert_errno(e.errno)
+
+    def open(self, path, flags, attr):
+        p = self._realpath(path)
+        try:
+            fd = os.open(p, flags, 0o666)
+        except OSError as e:
+            return paramiko.SFTPServer.convert_errno(e.errno)
+        if flags & os.O_WRONLY:
+            fstr = "ab" if flags & os.O_APPEND else "wb"
+        elif flags & os.O_RDWR:
+            fstr = "a+b" if flags & os.O_APPEND else "r+b"
+        else:
+            fstr = "rb"
+        try:
+            f = os.fdopen(fd, fstr)
+        except OSError as e:
+            return paramiko.SFTPServer.convert_errno(e.errno)
+        fobj = StubSFTPHandle(flags)
+        fobj.filename = p
+        fobj.readfile = f
+        fobj.writefile = f
+        return fobj
+
+
+class InjectingSFTPServer(paramiko.SFTPServer):
+    """Serves normally, then injects one stray SFTP packet after IDLE seconds."""
+
+    def start_subsystem(self, name, transport, channel):
+        self._last = time.time()
+        self._seen = False
+        self._injected = False
+        self._alive = True
+        threading.Thread(target=self._watch, daemon=True).start()
+        try:
+            super().start_subsystem(name, transport, channel)
+        finally:
+            self._alive = False
+
+    def _read_packet(self):
+        self._last = time.time()
+        self._seen = True
+        return super()._read_packet()
+
+    def _watch(self):
+        while getattr(self, "_alive", False):
+            time.sleep(0.5)
+            if self._seen and not self._injected and (time.time() - self._last) > IDLE:
+                try:
+                    # Stray SSH_FXP_NAME (type 104): body just needs valid framing;
+                    # phpseclib reads the type, expects HANDLE/STATUS, and throws.
+                    msg = paramiko.Message()
+                    msg.add_int(0)  # request id
+                    msg.add_int(0)  # count
+                    self._send_packet(CMD_NAME, msg)
+                    print("[sftp-mock] injected stray NAME packet after idle", flush=True)
+                except Exception as exc:  # noqa: BLE001
+                    print(f"[sftp-mock] inject failed: {exc}", flush=True)
+                self._injected = True
+
+
+def handle(client, addr):
+    try:
+        t = paramiko.Transport(client)
+        t.add_server_key(HOST_KEY)
+        t.set_subsystem_handler("sftp", InjectingSFTPServer, StubSFTPServer)
+        t.start_server(server=Server())
+        while t.is_active():
+            time.sleep(1)
+    except Exception as exc:  # noqa: BLE001
+        print(f"[sftp-mock] session error: {exc}", flush=True)
+
+
+def main():
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind(("", 22))
+    sock.listen(10)
+    print(f"[sftp-mock] listening on :22 root={ROOT} idle={IDLE}s", flush=True)
+    while True:
+        client, addr = sock.accept()
+        threading.Thread(target=handle, args=(client, addr), daemon=True).start()
+
+
+if __name__ == "__main__":
+    main()

.docker/sftp/short-timeout.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+# Shorten the sshd idle timeout for this dedicated SFTP service so that
+# demo.sftp_stale_connection can reproduce flysystem-process-bundle issue #24
+# (stale cached SFTP connection) in a few seconds instead of the ~60 minutes
+# reported in the wild. Run by the atmoz/sftp entrypoint (/etc/sftp.d/*) before
+# sshd starts. Only this "short timeout" SFTP service mounts it; the normal
+# `sftp` service keeps OpenSSH defaults.
+set -e
+{
+    echo "ClientAliveInterval 3"
+    echo "ClientAliveCountMax 1"
+} >> /etc/ssh/sshd_config

.env

@@ -39,10 +39,14 @@ DATABASE_URL="mysql://app:app@mysql:3306/app?serverVersion=9.1.0&charset=utf8mb4
 
 ###> docker/atmoz-sftp ###
 SFTP_HOST=sftp
-SFTP_PORT=22
 SFTP_USERNAME=sftp
 SFTP_PASSWORD=password
 SFTP_ROOT=data
+# Dedicated SFTP service with a short sshd idle timeout (see .docker/sftp/short-timeout.sh).
+SFTP_SHORT_HOST=sftp-short
+# Mock SFTP server that injects a stray packet after idle (see .docker/sftp-mock/), to
+# reproduce the verbatim issue #24 error. Serves /data as its root (SFTP root is empty).
+SFTP_MOCK_HOST=sftp-mock
 ###< docker/atmoz-sftp ###
 
 ###> symfony/routing ###

.gitignore

@@ -1,6 +1,7 @@
 /composer.lock
 .idea
 .composer
+.bash_history
 
 ###> symfony/framework-bundle ###
 /.env.local

config/packages/cleverage_process.yaml

@@ -31,4 +31,8 @@ clever_age_process:
                 callback:
                     callback: substr
                     right_parameters: [ '{{ offset }}', '{{ length }}' ]
+        sleep: # pause for $value seconds (PHP sleep) then pass the value through
+            transformers:
+                callback:
+                    callback: sleep
 

config/packages/flysystem.yaml

@@ -11,7 +11,57 @@ flysystem:
       adapter: 'sftp'
       options:
         host: '%env(string:SFTP_HOST)%'
-        port: '%env(int:SFTP_PORT)%'
+        port: 22
         username: '%env(string:SFTP_USERNAME)%'
         password: '%env(string:SFTP_PASSWORD)%'
-        root: '%env(string:SFTP_ROOT)%'
+        root: '%env(string:SFTP_ROOT)%'
+
+    # `sftp-short` container (OpenSSH with a short sshd idle timeout). Two distinct
+    # storages (different roots) on the same host, following the <base>.incoming /
+    # <base>.archive convention. Because OpenSSH closes the idle transport cleanly,
+    # this yields "Connection closed prematurely". Used by demo.sftp_stale_connection.
+    remote.storage.short_timeout.incoming:
+      adapter: 'sftp'
+      options:
+        host: '%env(string:SFTP_SHORT_HOST)%'
+        port: 22
+        username: '%env(string:SFTP_USERNAME)%'
+        password: '%env(string:SFTP_PASSWORD)%'
+        root: '%env(string:SFTP_ROOT)%/incoming'
+        connectivityChecker: App\Flysystem\SftpConnectivityChecker
+
+    remote.storage.short_timeout.archive:
+      adapter: 'sftp'
+      options:
+        host: '%env(string:SFTP_SHORT_HOST)%'
+        port: 22
+        username: '%env(string:SFTP_USERNAME)%'
+        password: '%env(string:SFTP_PASSWORD)%'
+        root: '%env(string:SFTP_ROOT)%/archive'
+        connectivityChecker: App\Flysystem\SftpConnectivityChecker
+
+    # `sftp-mock` container: a paramiko server that injects a stray packet after idle
+    # while keeping the SSH transport open (like the client's SAP server). Two distinct
+    # storages (different roots) on the same host, following the <base>.incoming /
+    # <base>.archive convention. Being different storages, each gets its OWN
+    # SftpConnectionProvider (its OWN phpseclib connection). This yields the VERBATIM
+    # issue #24 error. Used by demo.sftp_fetch_archive_timeout.
+    remote.storage.mock.incoming:
+      adapter: 'sftp'
+      options:
+        host: '%env(string:SFTP_MOCK_HOST)%'
+        port: 22
+        username: '%env(string:SFTP_USERNAME)%'
+        password: '%env(string:SFTP_PASSWORD)%'
+        root: '%env(string:SFTP_ROOT)%/incoming'
+        connectivityChecker: App\Flysystem\SftpConnectivityChecker
+
+    remote.storage.mock.archive:
+      adapter: 'sftp'
+      options:
+        host: '%env(string:SFTP_MOCK_HOST)%'
+        port: 22
+        username: '%env(string:SFTP_USERNAME)%'
+        password: '%env(string:SFTP_PASSWORD)%'
+        root: '%env(string:SFTP_ROOT)%/archive'
+        connectivityChecker: App\Flysystem\SftpConnectivityChecker

config/packages/process/demo.sftp_fetch_archive_timeout.yaml

@@ -0,0 +1,60 @@
+clever_age_process:
+    configurations:
+        demo.sftp_fetch_archive_timeout:
+            description: >
+                Minimal reproduction of the client's erp_contact_import failure (flysystem-process-bundle issue #24),
+                keeping only the essential shape: get_file fans out (in parallel) to [treatment, archive_file].
+                Pass the get_file storage BASE via the `filesystem` context; get_file uses `{{ filesystem }}.incoming`
+                and archive_file uses `{{ filesystem }}.archive` (the archive storage is deduced from the base - option
+                templating is a plain substitution, so the suffix is appended). They are two DIFFERENT flysystem
+                storages (different roots) on the SAME host, so each has its OWN cached phpseclib connection;
+                archive_file opens a FRESH connection and does NOT reuse get_file's one.
+                The stale connection is get_file's own (`{{ filesystem }}.incoming`): opened by get_file.execute, left
+                idle during the treatment sleep, and re-hit by get_file.next() - FileFetchTask is iterable, so after the
+                outputs are processed it lists the source again on the now-stale connection and fails. This matches the
+                client report ("archiving task and subsequent file listing operations").
+                With `filesystem: remote.storage.mock` the sftp-mock server injects a stray packet after idle while
+                keeping the transport open (like the client's SAP server), yielding the VERBATIM issue #24 error
+                "Expected NET_SFTP_HANDLE or NET_SFTP_STATUS. Got packet type".
+            help: >
+                bin/console cleverage:process:execute demo.sftp_fetch_archive_timeout -c filesystem:remote.storage.mock
+            tasks:
+                get_file:
+                    service: '@CleverAge\FlysystemProcessBundle\Task\FileFetchTask'
+                    description: >
+                        Fetch the source file from `{{ filesystem }}.incoming` (opens + caches connection C_A). Mirrors
+                        the client's get_zip; iterable (file_pattern), fans out to treatment and archive_file (LAST).
+                    options:
+                        source_filesystem: '{{ filesystem }}.incoming'
+                        destination_filesystem: 'local.storage'
+                        file_pattern: '/sample.*\.txt$/'
+                        remove_source: false
+                        ignore_missing: false
+                    outputs: [treatment, debug, archive_file]
+                treatment:
+                    service: '@CleverAge\ProcessBundle\Task\ConstantOutputTask'
+                    description: >
+                        Long-running treatment placeholder (feeds the idle duration in seconds to the sleep
+                        transformer). Must exceed the server idle threshold.
+                    options:
+                        output: 10
+                    outputs: [treatment_sleep]
+                treatment_sleep:
+                    service: '@CleverAge\ProcessBundle\Task\TransformerTask'
+                    description: >
+                        Keeps connection C_A idle (generic `sleep` transformer). Dead-end branch.
+                    options:
+                        transformers:
+                            sleep: ~
+                debug:
+                    service: '@CleverAge\ProcessBundle\Task\Debug\DebugTask'
+                archive_file:
+                    service: '@CleverAge\FlysystemProcessBundle\Task\FileFetchTask'
+                    description: >
+                        Archive the file by reading from `{{ filesystem }}.archive` - a DIFFERENT storage (deduced from
+                        the base), so this opens a FRESH connection C_B. Mirrors the client's archive_file.
+                    options:
+                        source_filesystem: '{{ filesystem }}.incoming'
+                        destination_filesystem: '{{ filesystem }}.archive'
+                        remove_source: false
+                        ignore_missing: false