fix(kafka): verify external dead-letter topics and check topics at consumer startup

docs/pages/backends/kafka.mdx

@@ -131,8 +131,23 @@ It therefore needs no group permission and cannot trigger the broker's topic aut
 shove never expands the topic's partitions and never reconciles its config.
 A consumer group registered with more `max_consumers` than the topic has partitions runs with the extra members idle.
 `declare` logs a warning when that happens.
-shove still creates and owns its DLQ topic, and still manages its consumer group.
-Provision the topic before the consumer starts.
+The DLQ topic of an external topology is infra's as well, under its default `{queue}-dlq` name or the one `dlq_named` gives it.
+`declare` probes it exactly like the main topic and returns `ShoveError::Topology` when it is missing; shove never creates it and never expands its partitions.
+On a topology that is not external, `declare` creates its DLQ with eight partitions and expands an existing one to that.
+shove still manages its consumer group.
+Provision the topic and its DLQ before the consumer starts.
+
+The registry path is the only one that runs `declare` for you.
+A consumer that starts without it, through `KafkaConsumer::run`, a `ConsumerSupervisor`, `run_batch`, `run_fifo`, a broadcast subscription or `run_dlq`, runs the same metadata probe once at startup, before its reconnect loop and before it consumes a record.
+It probes the main topic when the topology is external, and the DLQ topic wherever the path publishes to it, owned or external: the standard, batch and FIFO consumers.
+A broadcast subscription publishes no dead letters, so it probes the external main topic only.
+`run_dlq` reads from the DLQ and publishes nowhere. On an external topology it probes the main topic and the DLQ topic, both infra's; on a shove-owned one it probes nothing and waits for the DLQ topic, as a consumer waits for an owned main topic.
+A missing topic is `ShoveError::Topology` naming it, returned before any record is consumed.
+A broker that cannot be reached, or does not answer in time, says nothing about the topic and is not a `Topology` error: the probe retries under the consumer's usual backoff and `max_reconnect_attempts`, and a stop ends it.
+A shove-owned main topic is never probed, because another process may still declare it.
+With `with_producer_auto_create_topics(true)` a missing shove-owned DLQ is not probed either, because the first dead-letter publish creates it.
+An external topology with a DLQ is refused outright on a client with `with_producer_auto_create_topics(true)`, at `register` and at every consumer start that publishes dead letters (a publish-only `declare` is not refused, since it never writes a dead letter): a dead-letter publish after that topic disappeared would recreate the infra-owned topic with broker defaults. Consume it through a client without producer auto-creation.
+The probe is a metadata request, so it builds no producer and needs no permission beyond `Describe` on the topic.
 
 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.
@@ -145,14 +160,15 @@ The first publish may come from `publisher()` or from a consumer's retry, defer
 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 topology infra creates the DLQ, so shove needs only `Describe` and `Write` on 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.
+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 DLQ.
 That is the guarantee, and it covers declaring and consuming.
 Publishing is outside it: a `Publisher` on this topology writes to the topic.
 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.
+`Ack` and `Reject` behave as everywhere else, because a `Reject` publishes to the 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.
 A shove-owned topology runs with `Republish` by default and may opt into `InPlace`, so a consumer that must not duplicate a retried record for the other groups on a fan-out topic can wait in place too.
@@ -161,7 +177,7 @@ A FIFO consumer refuses `InPlace`, which it does not implement.
 | Outcome | `RetryStrategy::Republish`, the shove-owned default | `RetryStrategy::InPlace`, implied by an external topology |
 |---|---|---|
 | `Ack` | commits the offset | the same |
-| `Reject` | publishes to the DLQ when one is declared, then commits | the same, the DLQ is shove's own topic |
+| `Reject` | publishes to the DLQ when one is declared, then commits | the same, to the DLQ topic infra provisioned |
 | `Retry` | republishes into the topic after the tier's delay, with the retry count in a header | waits the tier's delay inside the handler's task, holding its prefetch slot, then hands the same record back with the retry count kept in memory; an exhausted budget goes to the DLQ as usual, with the in-memory count written to the dead letter's `Shove-Retry-Count` header |
 | `Defer` | republishes into the topic after the first tier's delay | waits the first tier's delay in place, then hands the same record back |
 

docs/pages/concepts/topics.mdx

@@ -160,7 +160,8 @@ See [Sequenced Topics](/guides/sequenced) for the full story.
 Before publishing or consuming, call `broker.topology().declare::<MyTopic>().await?` once at startup. This call is idempotent - the backend creates queues, exchanges, streams, and any other infrastructure the topology requires, and does nothing if they already exist. It is safe to call on every restart.
 
 `TopologyBuilder::external()` binds to infrastructure that infra owns instead of creating it: one flag on every backend, with each declarer deciding what verification means.
-On Kafka and NATS, `declare` verifies that the topic or stream exists and fails fast when it does not, while shove still creates its own DLQ.
+On Kafka and NATS, `declare` verifies that the topic or stream exists and fails fast when it does not.
+On Kafka the DLQ topic is infra's too, so `declare` verifies it the same way and never creates or expands it; on NATS shove still creates its own DLQ stream.
 RabbitMQ, SQS and Redis refuse an external topology at `declare` on this version, until their declarers gain a verification step, because creating the resource is the write the flag promises never to make; the in-process broker treats the flag as a no-op.
 Their consumers refuse it too, at every entry point and before any request, because a consumer can run without `declare` against a resource infra created, and a `Retry` on those three backends republishes into the consumed resource until each gains its in-place redelivery.
 `nats_external_stream()` remains for one release as a deprecated alias.

src/backends/kafka/backend.rs

@@ -221,7 +221,7 @@ impl ConsumerImpl for KafkaConsumer {
         T: SequencedTopic,
         H: MessageHandler<T>,
     {
-        KafkaConsumer::spawn_fifo_shards::<T, H>(self, handler, ctx, options)
+        KafkaConsumer::spawn_fifo_shards_checked::<T, H>(self, handler, ctx, options).await
     }
 }
 

src/backends/kafka/client.rs

@@ -794,17 +794,18 @@ impl KafkaClient {
             .map_err(|e| ShoveError::Connection(format!("kafka ping failed: {e}")))
     }
 
-    /// Confirm an infra-owned topic exists and return its partition count,
-    /// creating nothing; see `probe_external_topic_blocking`.
-    pub(super) async fn verify_external_topic(&self, name: &str) -> Result<i32> {
+    /// What the broker says of one topic, creating nothing; see
+    /// `probe_topic_blocking`. `Err` is a metadata fetch that failed, so
+    /// nothing was learned about the topic.
+    pub(super) async fn probe_topic(&self, name: &str) -> Result<TopicProbe> {
         let base = (*self.base_config).clone();
         let topic_name = name.to_string();
         #[cfg(feature = "kafka-msk-iam")]
         let msk_ctx = self.msk_context();
         #[cfg(feature = "kafka-msk-iam")]
         let shutdown = self.shutdown_token();
-        let probed = tokio::task::spawn_blocking(move || {
-            probe_external_topic_blocking(
+        tokio::task::spawn_blocking(move || {
+            probe_topic_blocking(
                 base,
                 &topic_name,
                 #[cfg(feature = "kafka-msk-iam")]
@@ -814,18 +815,27 @@ impl KafkaClient {
             )
         })
         .await
-        .map_err(|e| ShoveError::Topology(format!("metadata task failed: {e}")))??;
-        probed.ok_or_else(|| {
-            metrics::record_backend_error(
-                metrics::BackendLabel::Kafka,
-                metrics::BackendErrorKind::Topology,
-            );
-            ShoveError::Topology(format!(
-                "external(): topic `{name}` must be provisioned before the consumer starts, \
-                 but the broker has no topic by that name; shove never creates or alters an \
-                 external topic"
-            ))
-        })
+        .map_err(|e| ShoveError::Topology(format!("metadata task failed: {e}")))?
+    }
+
+    /// Whether the producer would create a topic it publishes to.
+    pub(super) fn producer_creates_topics(&self) -> bool {
+        self.producer_config
+            .config_map()
+            .get("allow.auto.create.topics")
+            .is_some_and(|v| *v == "true")
+    }
+
+    /// Confirm an infra-owned topic exists and return its partition count,
+    /// creating nothing. `what` names the topic's role in the refusal.
+    pub(super) async fn verify_external_topic(&self, name: &str, what: &str) -> Result<i32> {
+        match self.probe_topic(name).await? {
+            TopicProbe::Present(partitions) => Ok(partitions),
+            TopicProbe::Missing => Err(external_topic_missing(what, name)),
+            TopicProbe::Failed(code) => Err(ShoveError::Topology(format!(
+                "metadata for topic {name} returned {code:?}"
+            ))),
+        }
     }
 
     pub(super) async fn create_admin_default(&self) -> Result<AdminClient<DefaultClientContext>> {
@@ -1159,9 +1169,35 @@ fn fetch_topic_partition_count_blocking(
     Ok(topic.partitions().len() as i32)
 }
 
-/// Whether an infra-owned topic exists, and with how many partitions,
-/// without creating it. `Ok(None)` is the broker's answer that no such topic
-/// exists.
+/// What the broker's metadata says of one topic.
+pub(super) enum TopicProbe {
+    /// The broker has no topic by that name.
+    Missing,
+    /// The topic exists, with this many partitions.
+    Present(i32),
+    /// The broker answered, but with an error for the topic that is not
+    /// "unknown topic": nothing says whether it exists.
+    Failed(RDKafkaRespErr),
+}
+
+/// The `Topology` refusal for an external topic the broker does not have.
+/// `what` is the topic's role: "topic" or "dead-letter topic".
+pub(super) fn external_topic_missing(what: &str, name: &str) -> ShoveError {
+    metrics::record_backend_error(
+        metrics::BackendLabel::Kafka,
+        metrics::BackendErrorKind::Topology,
+    );
+    ShoveError::Topology(format!(
+        "external(): {what} `{name}` must be provisioned before the consumer starts, \
+         but the broker has no topic by that name; shove never creates or alters an \
+         external topic"
+    ))
+}
+
+/// What the broker says of a topic, without creating it. `Err` is a metadata
+/// fetch that failed (`Connection`: the broker is unreachable or timed out),
+/// which says nothing about the topic; `TopicProbe::Missing` is the broker's
+/// answer that no such topic exists.
 ///
 /// The probe is a consumer-type client with no `group.id`, and both halves
 /// matter. librdkafka defaults `allow.auto.create.topics` to true for a
@@ -1172,12 +1208,12 @@ fn fetch_topic_partition_count_blocking(
 /// consumer-type client defaults it to false, and it is set explicitly here
 /// anyway. Without a `group.id` there is no coordinator lookup, so the probe
 /// needs no group permission under a group-scoped ACL.
-fn probe_external_topic_blocking(
+fn probe_topic_blocking(
     base: ClientConfig,
     topic_name: &str,
     #[cfg(feature = "kafka-msk-iam")] msk_ctx: Option<MskIamContext>,
     #[cfg(feature = "kafka-msk-iam")] shutdown: CancellationToken,
-) -> Result<Option<i32>> {
+) -> Result<TopicProbe> {
     let mut cfg = base;
     cfg.set("allow.auto.create.topics", "false");
     let md = fetch_topic_metadata_blocking(
@@ -1189,16 +1225,16 @@ fn probe_external_topic_blocking(
         shutdown,
     )?;
     let Some(topic) = md.topics().iter().find(|t| t.name() == topic_name) else {
-        return Ok(None);
+        return Ok(TopicProbe::Missing);
     };
-    match topic.error() {
-        Some(RDKafkaRespErr::RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART) => Ok(None),
-        Some(other) => Err(ShoveError::Topology(format!(
-            "metadata for topic {topic_name} returned {other:?}"
-        ))),
-        None if topic.partitions().is_empty() => Ok(None),
-        None => Ok(Some(topic.partitions().len() as i32)),
-    }
+    Ok(match topic.error() {
+        Some(RDKafkaRespErr::RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART) => TopicProbe::Missing,
+        Some(other) => TopicProbe::Failed(other),
+        None if topic.partitions().is_empty() => TopicProbe::Missing,
+        None => TopicProbe::Present(i32::try_from(topic.partitions().len()).map_err(|_| {
+            ShoveError::Topology(format!("topic {topic_name} reports too many partitions"))
+        })?),
+    })
 }
 
 /// A consumer-type librdkafka client with no `group.id`, used for metadata