sftp: advance to the next file without a reconnect

[Leward]

The problem

The sftp input returned service.ErrNotConnected when the current file reached EOF, to make the AsyncReader call ReadBatch again and open the next file (although no connection was actually lost).

Benthos adds a fixed wait before each reconnect (ticket VM-74, PR redpanda-data/benthos#496). With that change, the input would pay the wait once per file.

The solution

Move the file rotation into tryReadBatch. On EOF the reader closes the current scanner and opens the next file in the same call. It returns service.ErrEndOfInput only when the path provider has no more files. The watcher provider still blocks in Next until a new file appears or the context is cancelled.

The rotation loop checks the context at the top of each iteration so that a shutdown stops the input before it opens the next file. This was handled by the AsyncReader before, but now it's done directly in the reader.

Real connection loss still returns service.ErrNotConnected.

This mirrors the csv input fix in benthos (redpanda-data/benthos#497).

[claude]

readOneFile is invoked from a non-test goroutine here, but it uses require.* (integration_test.go#L355-L365). This violates the project test pattern (.claude/agents/tester.md, "Polling" / "Async Operations"): "Do not use require inside ... require calls FailNow() which panics when called from a non-test goroutine."

Two concrete failure modes:

1. If ReadBatch returns an error, require.NoError calls t.FailNow() → runtime.Goexit(), so nothing is ever sent on results. The test then blocks on the select at L493-L498 for 5s and fails with the misleading message "watcher did not pick up the new file" instead of the real error.
2. If the subtest ends first (e.g. via the t.Fatalf at L488 or t.Fatal at L497), defer cancel() fires, the blocked ReadBatch returns context.Canceled, and the still-running goroutine calls t.Errorf on a finished test → panic: Log in goroutine after test has completed.

Suggested fix: have the goroutine send both the content and the error over a channel (or a small struct) and do all the require assertions on the main test goroutine after receiving.

[claude]

Flaky-test risk: the watcher is configured with minimum_age: 0s and poll_interval: 100ms (integration_test.go#L336-L340), and writeSFTPFile creates the file and only then writes its contents (L314-L320). minimum_age exists precisely to avoid this — see the field description in input.go: "The minimum period of time since a file was last updated before attempting to consume it. Increasing this period decreases the likelihood that a file will be consumed whilst it is still being written to."

If a poll lands in the window between client.Create returning and fmt.Fprint completing, watcherPathProvider.findNewPaths globs 3.txt at size 0, appends it, and marks it ! in the cache. The to_the_end scanner then yields EOF with no message, so no ack ever runs, the cache entry stays !, and because followUpPoll is already true the path is never re-offered (input.go#L598-L610). The test then hard-fails after 5s with "watcher did not pick up the new file".

Suggested fix: write to a temporary name and Rename it into place so the file appears atomically at full size, or set a non-zero minimum_age for this subtest.