fix: backport non-breaking 0.16 fixes and advisories to 0.15

.github/workflows/ci.yml

@@ -2,7 +2,7 @@ name: CI
 
 on:
   push:
-    branches: [main]
+    branches: [main, "release/**"]
     # Code paths only. `benches/**` is spelled out so the results document
     # and the bench README stay out, and the committed charts are absent on
     # purpose: a chart-only or results-only change is checked by

docs/pages/backends/kafka.mdx

@@ -137,12 +137,21 @@ Provision the topic before the consumer starts.
 A consumer then needs these ACLs: `Describe` and `Read` on the topic, and `Describe` and `Read` on the consumer group.
 The group is `{queue}-consumer` unless `for_consumer_group` or `with_group_id` names another.
 A publisher additionally needs `Write` on the topic.
+shove's producer is idempotent.
+librdkafka requests its producer id 500 ms after the producer is created, before any publish, with an `InitProducerId` request.
+A broker authorizes that request as `IdempotentWrite` on the cluster, or since Kafka 2.8 (KIP-679) as `Write` on a topic, and a consumer holds neither.
+The client builds that producer at the first publish, not at connect.
+The first publish may come from `publisher()` or from a consumer's retry, defer or DLQ republish.
+A service that never publishes, through `publisher()` or a republish, therefore never builds a producer and needs no permission beyond the consumer's.
+`connect` still builds the consumer-type client that `ping` uses, so a connection setting librdkafka refuses fails at connect as before.
 A declared DLQ needs `Create`, `Describe` and `Write` on the DLQ topic, so that shove can create it and route to it.
 
 On an external topic, shove's consumer never writes into the topic when it settles an outcome: a `Retry` or a `Defer` waits in place, and a `Reject` publishes to the shove-owned DLQ.
 That is the guarantee, and it covers declaring and consuming.
 Publishing is outside it: a `Publisher` on this topology writes to the topic.
-Until the producer pins `allow.auto.create.topics=false`, a separate change in the same release, a publish to a topic that does not exist can create it with broker defaults on a broker with `auto.create.topics.enable=true`, so provision the topic before any publisher starts.
+The producer pins `allow.auto.create.topics=false`, so a publish to a topic that does not exist fails after the produce timeout and creates nothing.
+Only `with_producer_auto_create_topics(true)` lets a publish create a topic, with broker defaults, on a broker running `auto.create.topics.enable=true`.
+Provision the topic before any publisher starts.
 `Ack` and `Reject` behave as everywhere else, because a `Reject` publishes to the shove-owned DLQ and not into the topic.
 How `Retry` and `Defer` are carried out is the consumer's **retry strategy**, `RetryStrategy::Republish` or `RetryStrategy::InPlace`, set with `with_retry_strategy` on `ConsumerOptions::<Kafka>` or `KafkaConsumerGroupConfig`.
 Ownership implies it: an external topology runs with `InPlace` and refuses `Republish`, because the republish would write into a topic infra owns.
@@ -167,7 +176,14 @@ A partition the group assigns to the member during a pause is paused too.
 A record it delivers first is handed back to the broker and arrives again, in order, once a slot frees.
 The pause lasts until a slot frees: the slot that ends the wait serves the record already in hand, so with one slot the assignment stays paused through that record's handler as well.
 A shutdown during a wait completes nothing: the record stays uncommitted and is redelivered on restart.
-A waiting handler holds its slot for the whole delay, the contract a broadcast subscription has always had.
+A record the consumer had fetched but not yet handed to a handler when shutdown fired is dropped unhandled too.
+It is redelivered on restart, behind the record that was waiting.
+A revoke of the partition ends the wait at once with nothing completed.
+The partition's next owner is handed the record again from the committed offset.
+A handler result carries the assignment it was made under, and one from an earlier assignment of the partition is ignored.
+With one slot, no later record of a partition reaches the handler while a record of that partition waits in place.
+That holds while the consumer runs, on a stop, and across a rebalance.
+A waiting handler holds its slot until the delay ends, shutdown starts or its partition is revoked.
 Size `prefetch_count` for the number of records you accept to have waiting at once.
 
 `external()` is incompatible with `sequenced()` and with every topic-config method.
@@ -204,7 +220,10 @@ let config = KafkaConfig::new(brokers)
 Unset throughput settings keep librdkafka's defaults.
 `acks`, `enable.idempotence`, and `max.in.flight.requests.per.connection` are deliberately not exposed.
 
-The client holds a single producer, shared by `publisher()` and by the consumer's retry, defer, and DLQ republishes — a long linger also delays those republishes (and the offset commits gated on them for sequenced topics), so keep linger modest when consumers republish on the same client.
+The client holds a single producer, built at the first publish.
+`publisher()` and the consumer's retry, defer, and DLQ republishes share it.
+A long linger also delays those republishes, and the offset commits gated on them for sequenced topics.
+Keep linger modest when consumers republish on the same client.
 
 ## Consume
 
@@ -443,7 +462,8 @@ A trailing control record is different: with nothing delivered after it, the com
 An idle transactional topic therefore reads a lag of 1 until the next record arrives, and nothing is redelivered.
 
 On shutdown the consumer first waits for its in-flight handlers to finish.
-That wait is bounded by the handler timeout and by any in-place delays.
+That wait is bounded by the handler timeout.
+An in-place delay does not extend it: shutdown cuts the delay short, and the record stays uncommitted.
 It then drains the republish queue and commits its final position **synchronously**.
 The commit runs on a dedicated thread that also owns the consumer's close.
 The thread is spawned before it is handed the consumer, so a failed spawn never closes the consumer on the runtime thread.
@@ -464,6 +484,7 @@ What that means for replay:
 - After a rebalance, a moved partition replays from its last accepted commit.
 - After a clean shutdown, a record whose handler returned a terminal outcome and whose offset reached the final commit is not redelivered.
   A record whose handler had not settled when shutdown fired stays uncommitted and is redelivered.
+  A record the consumer had fetched but not yet handed to a handler is not handled during the drain, and is redelivered too.
   So is the last batch when the final commit hits its deadline.
   An asynchronous commit librdkafka was still retrying can also land after the final one.
   A coordinator hiccup overlapping the shutdown can therefore replay a few records.

docs/pages/concepts/broadcast.mdx

@@ -141,6 +141,8 @@ An id under the granted prefix avoids provoking it.
 A deferred record is redelivered to the deferring subscription before any record received after it.
 The subscription's single slot is held across the delay and the same bytes are decoded again.
 The order is therefore the original delivery, the redelivery, then whatever came next.
+A stop during the delay hands nothing over: the deferred record is dropped with nothing completed.
+A record the subscription had read while it waited for its slot is dropped too, not handled during the drain.
 
 Because the assignment is taken at the tail, a reconnect re-assigns at the *then*-current tail: messages published while a subscriber was disconnected are not replayed. That is deliver-new applied to the reconnect window rather than a separate caveat.
 With `with_broadcast_start` the reconnect re-resolves the configured start instead, so a `Head` subscription replays the retained log again.

docs/pages/guides/liveness.mdx

@@ -27,7 +27,7 @@ Every production backend issues a real round-trip. There are no local-state shor
 
 | Backend   | RPC                                                | Side effects             |
 |-----------|----------------------------------------------------|--------------------------|
-| Kafka     | `fetch_metadata(None, timeout)` on producer client | none                     |
+| Kafka     | `fetch_metadata(None, timeout)` on the metadata client `connect` builds | none                     |
 | Redis     | `PING` on the multiplexed connection               | none                     |
 | NATS      | subscribe + unsubscribe-after(1) + flush + publish + await echo on a unique inbox | none |
 | RabbitMQ  | transient channel open + close                     | none                     |