fix/csv-scanner-infinite-loop

[diwang01]

Hi benthos team,

This is a real issue I met in our production env, so I created this PR to handle it.

scanner/csv: don't skip read errors with continue_on_error

Problem

With continue_on_error: true, the csv scanner skips every error that isn't
io.EOF, including errors coming from the underlying reader, not only CSV parse errors:

recordStrs, err := c.c.Read()
if err != nil {
    if errors.Is(err, io.EOF) || !c.continueOnError {
        return nil, err
    }
}
// otherwise: emit a message flagged with the error and carry on

A read error from the source is not a bad row that can be stepped over. Readers such as
an HTTP response body (net/http keeps the first read error, bodyEOFSignal.rerr)
return the same error on every later Read(). encoding/csv then returns a
single empty field plus that error each time. The result:

• the scanner emits an endless stream of empty, error-flagged messages
  ({"<first header>": ""}) as fast as the pipeline can consume them;
• io.EOF is never reached, so the input never finishes the object, never acks it and
  never moves on;
• the read error itself is never logged, because it is never returned. It only
  lives on the message's error flag.

The loop only stops when the process restarts.

How I hit it

An aws_s3 input in SQS mode reading a large |-delimited CSV with
continue_on_error: true (Redpanda Connect 4.48 / benthos v4.44.1). The S3 download
connection broke partway through the object. The pod logged real rows, then switched to
nothing but empty rows for about 24 hours, at full CPU, until it was restarted. A restart
fixed it because the redelivered SQS message triggered a new GetObject on a new
connection.

Reproduced locally with an HTTP server that sends two rows and then resets the
connection, read through http_client with stream.scanner.csv:

continue_on_error	Result in 4 seconds
true	rows 1 and 2, then 192,414 empty messages
false	rows 1 and 2, then Failed to read message: read tcp ...: connection reset by peer

Fix

Skip an error only when it is a CSV parse error (*csv.ParseError, e.g. ErrFieldCount,
ErrQuote, ErrBareQuote), which is what the option is documented for. Any other
error is returned as normal, as if continue_on_error were false:

var pErr *csv.ParseError
if !c.continueOnError || !errors.As(err, &pErr) {
    return nil, err
}

io.EOF is not a *csv.ParseError, so end-of-input behaviour is unchanged and the
explicit io.EOF check is no longer needed.

The continue_on_error field description now says "fails to parse" instead of "fails to
parse due to any error", and states that errors reading from the source are not skipped.

Behaviour change

• Malformed rows: unchanged. Still emitted as messages flagged with the parse error, and
  reading continues.
• Source read errors with continue_on_error: true: now returned from NextBatch, so the
  input handles them like any other read error. They get logged, and the input
  reconnects, retries or nacks depending on the input. Before, they caused an infinite
  loop that logged nothing.
• continue_on_error: false: unchanged.

Tests

Added TestCSVScannerContinueOnError in internal/impl/pure/scanner_csv_test.go. It
reads a valid row, a wrong-field-count row and another valid row, then hits a reader that
returns connection reset by peer on every read (iotest.ErrReader). It asserts that:

• the bad row is still emitted, flagged with csv.ErrFieldCount;
• the read error is returned by NextBatch.

On the old code the test fails with
Expected error with "connection reset by peer" in chain but got nil.

go test ./internal/impl/pure/ -count=1