sinks: a value the driver cannot encode fails once, as a user error

README.md

@@ -534,6 +534,49 @@ sink:
 
 `sink.format.type: parquet` is parsed and ignored.
 
+### Sink retries
+
+ClickHouse and Iceberg flushes retry when the destination is not answering.
+Omit the block to accept the defaults. Set `max_attempts: 1` to turn retrying
+off. The Kafka sink ignores this block: franz-go already retries a produce
+with its own backoff.
+
+```yaml
+sink:
+  type: clickhouse
+  retry:
+    max_attempts: 4           # total attempts, including the first
+    initial_backoff_ms: 100   # doubles each attempt
+    max_backoff_ms: 2000      # ceiling on the backoff
+    deadline_seconds: 10      # bounds the whole ladder, not one attempt
+```
+
+The values shown are the defaults.
+
+Keep `deadline_seconds` below `pipeline.flush_interval_seconds`. The retry
+runs inside the open state transaction, and a ladder that outlives the flush
+interval freezes the window clock.
+
+The ladder retries only what another attempt could change. The error code
+decides:
+
+| Error | Retried | Why |
+| --- | --- | --- |
+| Any `user.*` code, including `user.sink.encode_failed` and `user.sink.type_unsupported` | No | The config, SQL or a value is wrong. It fails identically every time. |
+| `system.sink.write_failed` | No | The destination answered and refused the write. |
+| `system.sink.unreachable` | Yes | The destination may come back. |
+| An error with no code | Yes | A driver's timeout or reset arrives unclassified. The deadline bounds the cost. |
+| Any other `system.*` code | Yes | |
+
+A failure that is not retried keeps its own code and exit code. A retried
+failure that outlasts the ladder is reported as `system.sink.unreachable`,
+exit 12, and `sink_retry_count_total` counts each attempt after the first.
+
+`user.sink.encode_failed` is the sink's client refusing a value before
+anything reaches the network, such as a timestamp string the ClickHouse
+driver's `DateTime` layout cannot parse. Cast or format the column in the
+handler SQL. The message names the column and quotes the value.
+
 ## Error policies
 
 `pipeline.on_error.policy` is `RAISE` (default), `IGNORE`, or `DLQ`. It is

docs/coverage/invariants.yml

@@ -120,8 +120,10 @@ invariants:
     class: safety
     applies_to: sink
     claim: >
-      The sink's errors classify as unreachable or rejected, so the retry
-      ladder retries the right ones.
+      The sink's errors classify as unreachable, rejected, or a user fault,
+      so the retry ladder retries only the first. A value the sink's client
+      cannot encode is a user fault: it never reaches the network and fails
+      the same way every attempt.
     verified_by: harness
     requires: []
 

docs/coverage/matrix.md

@@ -108,7 +108,7 @@ until an invariant's `requires` is filled in, and none is yet.
 | `sink.flush.honours_context` | Flush returns ctx.Err() when the context ends. Rows stay buffered. *(violated once: #219)* | ✅ i | — exempt | — exempt | ✅ i | — exempt | — exempt |
 | `sink.flush.empty_is_noop` | Flush with nothing buffered returns nil and touches nothing. | ✅ i | ✅ u | ✅ u | ✅ i | — exempt | ✅ u |
 | `sink.buffer.reports_depth` | A sink reports how many rows it is holding, and the count rises when a flush fails and falls to zero when one succeeds. | ✅ i | ✅ u | ✅ u | ✅ i | — exempt | ✅ u |
-| `sink.error.classifies` | The sink's errors classify as unreachable or rejected, so the retry ladder retries the right ones. | ❌ missing | ❌ missing | ❌ missing | ❌ missing | — exempt | ❌ missing |
+| `sink.error.classifies` | The sink's errors classify as unreachable, rejected, or a user fault, so the retry ladder retries only the first. A value the sink's client cannot encode is a user fault: it never reaches the network and fails the same way every attempt. | ❌ missing | ❌ missing | ❌ missing | ❌ missing | — exempt | ❌ missing |
 | `sink.probe.fails_start` | A Prober whose destination is unreachable fails the start once, without retrying. | ✅ i | — exempt | — exempt | ✅ i | — exempt | — exempt |
 
 ## Safety invariants: checkpoint