feat: add max_batch_age to drop stale batches instead of retrying

[zmstone]

Add a new producer option max_batch_age (default infinity, disabled).

Motivation

When a connection to Kafka drops (request timeout / broker restart), buffered and in-flight batches are (re)sent on recovery no matter how old they are. For long outages this delivers very stale data, and there is no way to bound the delivery latency of buffered messages.

What this does

When max_batch_age is set to a number of milliseconds, a batch is dropped instead of sent if ALL of its messages are older than max_batch_age (measured from when they were appended to the buffer — the queue-item timestamp, an absolute wall-clock value that survives a disk-replayq reload).

Whole-batch granularity is deliberate: a Kafka produce request is a single encoded batch, so partial drops would require re-encoding. A batch is dropped only when every message in it has expired.

The check runs in two places:

• maybe_send_to_kafka/1 — peeks the front of the queue and drops expired batches before the next send. Queue items are appended in timestamp order, so once the front is not expired nothing behind it is either → an O(#expired) front drain. It runs regardless of connection or inflight-window state, so stale data is shed (and memory reclaimed) even while disconnected. These items are still in the pending-ack backlog → dropped via wolff_pendack:drop_backlog/2. Structurally this is handle_overflow/3’s twin, but triggered by age at send time rather than by byte-overflow.
• resend_sent_reqs/2 — drops in-flight batches (in sent_reqs) that expired while a connection was down, instead of retrying them on reconnect. These are the oldest / most-stale batches and are not in replayq, so the drain above cannot see them. They are inflight acks → released via wolff_pendack:take_inflight/2.

Each dropped message has its ack callback evaluated with the reason message_expired (a new sentinel alongside buffer_overflow_discarded / partition_lost), so async callers are notified rather than waiting forever, and bumps the existing dropped counter plus a new dropped_expired telemetry counter.

infinity (the default) preserves current behavior exactly, with no now_ts/0 call on the hot path.

Notes

• New public reason atom message_expired reaches the ack callback; downstream callers that pattern-match ack results should add a clause for it.
• Relies on queue-item timestamps being non-decreasing (FIFO append order). A backwards system-clock step (NTP) could momentarily mis-time a drop decision, but not corrupt state.

Tests

• is_batch_expired_test_/0 — pure unit tests of the predicate (infinity, all-expired, one-fresh, boundary).
• drop_expired_batch_on_reconnect_test_/0 — in-flight batch ages out during a connection drop and is dropped (not retried) on reconnect.
• drop_expired_queued_batch_test_/0 — a queued (never-dispatched) batch ages out and is dropped from the front on the next send.

Draft: compiles clean (incl. warnings_as_errors); the eunit/integration suite has not been run in this environment yet and will be validated in CI / once a broker is free.