Multiple Postgres output connector improvements. - #6868

Chain the `postgres::Error` as the source of the `BackoffError` instead
of interpolating it into a fresh `anyhow!` message.

`Display for postgres::Error` reports only the kind, so interpolating it
yielded "postgres error: permanent: SqlState: Some(SqlState(E23502)): db
error" and nothing more: the server's message and DETAIL live in the
error's source, which the interpolation discarded. Chaining keeps them,
and `BackoffError::inner` already formats the whole chain.

tokio-postgres appended the source to `Display` up to 0.7.13 and stopped
in 0.7.14, so this silently emptied out every postgres connector error
when we upgraded.

Reported errors now carry the diagnosis:

    while executing insert statement for 1721 record(s) ...

    Caused by:
        0: postgres error: permanent: SqlState: Some(SqlState(E23502))
        1: db error
        2: ERROR: null value in column "freshness_timestamp" of relation
           "..." violates not-null constraint
           DETAIL: Failing row contains (...).

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>

Report the record count, the payload size and a 4 KiB prefix when a
statement fails, instead of the entire payload.

The payload holds one buffered batch, up to `max_buffer_size_bytes`
(1 MiB by default), and the endpoint retains up to 100 messages per tag,
so quoting it in full grew `/output_endpoints/{name}/stats` past the
50 MB limit the pipeline manager applies while proxying the response.
A customer hit exactly that and could not read the 872 errors their
connector reported; the offending `fatal_error` alone was 1,049,481
bytes, and it also reached every plain `/stats` poll.

Bound the two other places that echo unbounded text into an error: the
generated query, which names every column of the target table, and the
JSON body of a `set_extra_columns` command.

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>

Count rows and bytes per statement that `execute` accepted, and report
them to the endpoint only once the transaction commits cleanly.

`transmitted_records` counted rows as `encode_cursor` buffered them, so
it measured what the connector attempted rather than what reached the
table. A failing statement aborts the whole transaction, and Postgres
answers the following COMMIT with ROLLBACK instead of an error, so the
connector took the commit as success and reported the batch as
transmitted.

Counting per successful statement is not enough on its own, because the
rows of a statement that succeeded before a later one failed are rolled
back too. Track whether the transaction has been poisoned and roll it
back explicitly rather than issuing a commit whose success means
nothing.

Report the totals whether or not every worker succeeded. Each worker
commits its own transaction, so a batch that failed in one worker still
wrote the rows the others committed; reporting only on the success path
also carried those rows into the next batch's total.

The error the rollback reports names the target table and says that the
rows PostgreSQL had already accepted went with it, since someone who
reads only the rejected statement expects the rest of the rows to have
landed. Reporting a single failed writer thread no longer prefixes the
count, which is noise in the default single-thread configuration.

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>

Retain each worker's partition of the batch until it commits, and encode
it into a fresh transaction if the connection drops partway.

A dropped connection lost the whole batch. `retry_connecting` clears the
open transaction and nothing reopened one, so the statement being retried
and every statement after it failed with "transaction that hasn't been
created yet", and the commit failed too. Even with a transaction the
retry could not have recovered the batch: the payloads that carried the
earlier rows had been drained from the buffers as they flushed, so the
connector no longer held them.

`exec_statement` no longer retries a statement in place, which could
never work, and stops issuing statements once the batch is doomed instead
of reporting one failure per remaining flush.

Two limits are worth knowing. A commit that fails on a lost connection
leaves it unknowable whether the server applied it, and this writes the
batch again: harmless in materialized mode, a duplicate set of rows in
CDC mode, which is documented on `write_batch`. And the
batch_records_written gauge counts the abandoned attempt's rows as well
as the rewrite's until the next batch boundary resets it.

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>

Report a connection failure as fatal only when the connector gives up on
it, not when it is about to retry.

`fatal` does two things: it stamps `fatal_error` on the endpoint status,
which is write-once and never cleared, and it prefixes the logged message
with "FATAL". So a blip the connector recovered from within a second
marked the endpoint as fatally failed for the life of the pipeline, and
said so in the log.

`retry_connecting_with_backoff` reported every failure as fatal, one per
iteration, including the ones it went on to retry successfully.
`exec_statement` did the same, which became wrong when a retryable
statement failure started leading to the batch being written again rather
than dropped.

The reconnect tests now assert that an endpoint that recovered carries no
fatal error.

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>

Classify a failure to open a connection with a new
`BackoffError::connecting`, which denies a short list of permanent
failures instead of allowing a short list of transient ones.

The shared classification was written for statements failing on an
established connection, where allowing a handful of connection-lost codes
is right. Applied to connecting, it called everything else permanent, so
the connector gave up on exactly the conditions the retry loop exists
for. `57P03`, which is what postgres answers while it is starting up
after a restart, was treated as permanent, as were `57P02` after a crash,
`53300` when connection slots run out, and the refusal a database under
maintenance returns. Measured: the connector abandoned the batch and
wrote 0 of 3000 rows.

Connecting keeps retrying unless the configuration is one the connector
cannot outlast: a wrong password, a missing database, or insufficient
privilege. Changing any of those restarts the connector anyway.

Fold the two failure flags this leaves behind into one `TransactionState`
of Open, Lost, or Rejected. They were never independent: any statement
failure set `txn_poisoned`, and a retryable one then also set
`needs_replay`, so a lost connection left both true and `batch_end_inner`
had to test `needs_replay` first to reach the recoverable case. That
precedence was load-bearing and only a comment said so. The states are
now mutually exclusive by construction, one assignment replaces two, and
it happens where the error is built and `should_retry` is already known.

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>

Watch the pipeline state, not the controller's liveness, when deciding
whether to keep waiting for postgres, and break the backoff into slices
so the wait ends promptly.

A worker that cannot reach postgres sits in `retry_connecting_with_backoff`.
The loop meant to abandon that wait on shutdown by checking whether its
`Weak<ControllerInner>` still upgraded, but that check could never fire.
The endpoint's drop joins this thread, and the output thread is blocked
on the endpoint, so the controller stays alive exactly because the worker
has not finished: the loop waited for a drop that was waiting for the
loop. `ControllerInner::stop` sets `Terminated` before it touches
anything else, so that is the signal that actually arrives.

The minute-long backoff was also one uninterruptible sleep, which would
have delayed the stop even with a signal that worked. It now wakes every
100ms to check.

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>

Release any transaction still held before opening the next one, and drop
the one a failed rewrite leaves behind.

`batch_start_inner` transmutes a borrow of `self.client` to 'static, so
nothing stops two transactions from borrowing the client at once except
the promise that each is finished before the next begins. Every path did
honor that until `restart_batch` arrived: it opens a transaction and then
encodes into it, and if the encode fails it returns an error that ends
the batch, leaving the transaction on the worker. The next batch would
then borrow the client while that one was still alive.

Release it in two places. `restart_batch` drops it when the encode fails,
which is the point where the batch actually ends, so the session does not
sit idle in a transaction until the next batch starts.
`batch_start_inner` drops whatever it finds before borrowing the client
again, which makes overlapping borrows impossible however a future caller
behaves, and the SAFETY comment now says so.

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>