fix(connectors): stop rendering counter metrics with a doubled _total

.claude/skills/connector-runtime/SKILL.md

@@ -211,7 +211,7 @@ Fatal errors propagate to `main` and exit. Per-connector / per-message errors ar
 
 All families labeled by `connector_key` + `connector_type` (histogram adds `stage`):
 
-- **Counters**: `iggy_connector_messages_{produced,sent,consumed,processed,filtered,errors}_total`.
+- **Counters**: `iggy_connector_messages_{produced,sent,consumed,processed,filtered}_total` and `iggy_connector_errors_total`. These are the *rendered* names; each is registered without the `_total`, which the OpenMetrics encoder appends.
   - `messages_filtered_total` - intentional drops via transform `Ok(None)`.
   - `errors_total` - unexpected drops (decode/encode/build failure, missing field, ...) + batch-level failures.
 - **Histograms**: `iggy_connector_stage_duration_seconds{stage}` (snake_case stage labels - `prepare`, `ffi`, `decode`, `iggy_send`, `state_save`, `total`). Buckets `STAGE_BUCKETS_SECONDS`. Always populated regardless of any flag. Scraped at `/metrics` when `[http.metrics] enabled = true`.
@@ -221,7 +221,7 @@ All families labeled by `connector_key` + `connector_type` (histogram adds `stag
 
 When adding a metric:
 
-- Add family to `Metrics` struct + `init`, register with name + help text.
+- Add family to `Metrics` struct + `init`, register with name + help text. Never end a `Counter` family's registered name in `_total`: the encoder appends it and the series renders `_total_total`. Gauges get no suffix, so a gauge name may end in `_total` literally.
 - New label sets define `EncodeLabelSet` struct + label enum (hand-impl `EncodeLabelValue` for snake_case values - the derive emits PascalCase).
 - Histograms: pass `fn() -> Histogram` to `Family::new_with_constructor`.
 - Add unit tests under `mod tests` with `given_*_when_*_should_*` BDD names.

core/connectors/runtime/src/metrics.rs

@@ -208,33 +208,35 @@ impl Metrics {
             "Sinks in Running status",
             sinks_running.clone(),
         );
+        // Counter families are registered without the `_total` suffix: the
+        // OpenMetrics encoder appends it, so a literal one renders doubled.
         registry.register(
-            "iggy_connector_messages_produced_total",
+            "iggy_connector_messages_produced",
             "Messages received from source plugin poll",
             messages_produced.clone(),
         );
         registry.register(
-            "iggy_connector_messages_sent_total",
+            "iggy_connector_messages_sent",
             "Messages sent to Iggy (source)",
             messages_sent.clone(),
         );
         registry.register(
-            "iggy_connector_messages_consumed_total",
+            "iggy_connector_messages_consumed",
             "Messages consumed from Iggy (sink)",
             messages_consumed.clone(),
         );
         registry.register(
-            "iggy_connector_messages_processed_total",
+            "iggy_connector_messages_processed",
             "Messages processed and sent to sink plugin",
             messages_processed.clone(),
         );
         registry.register(
-            "iggy_connector_messages_filtered_total",
+            "iggy_connector_messages_filtered",
             "Messages intentionally dropped by transforms returning Ok(None)",
             messages_filtered.clone(),
         );
         registry.register(
-            "iggy_connector_errors_total",
+            "iggy_connector_errors",
             "Errors encountered",
             errors.clone(),
         );
@@ -520,6 +522,83 @@ fn stage_histogram() -> Histogram {
 mod tests {
     use super::*;
 
+    /// Every series the registry renders once each family holds a sample.
+    /// Counters carry exactly one `_total`, gauges and histogram buckets carry
+    /// the suffixes the OpenMetrics encoder gives them.
+    const RENDERED_SERIES_NAMES: [&str; 13] = [
+        "iggy_connectors_sources_total",
+        "iggy_connectors_sources_running",
+        "iggy_connectors_sinks_total",
+        "iggy_connectors_sinks_running",
+        "iggy_connector_messages_produced_total",
+        "iggy_connector_messages_sent_total",
+        "iggy_connector_messages_consumed_total",
+        "iggy_connector_messages_processed_total",
+        "iggy_connector_messages_filtered_total",
+        "iggy_connector_errors_total",
+        "iggy_connector_stage_duration_seconds_sum",
+        "iggy_connector_stage_duration_seconds_count",
+        "iggy_connector_stage_duration_seconds_bucket",
+    ];
+
+    /// Exercise every family so each one renders at least one sample: a
+    /// `Family` with no label set encodes its `# TYPE` header and nothing else,
+    /// which would hide the rendered series names from any assertion.
+    fn populated_metrics() -> Metrics {
+        let metrics = Metrics::init();
+        metrics.set_sources_total(1);
+        metrics.set_sinks_total(1);
+        metrics.increment_sources_running();
+        metrics.increment_sinks_running();
+        metrics.increment_messages_produced("k", 1);
+        metrics.increment_messages_sent("k", 1);
+        metrics.increment_messages_consumed("k", 1);
+        metrics.increment_messages_processed("k", 1);
+        metrics.increment_messages_filtered("k", ConnectorType::Sink, 1);
+        metrics.increment_errors("k", ConnectorType::Sink);
+        metrics.observe_stage_duration(
+            "k",
+            ConnectorType::Sink,
+            Stage::Ffi,
+            Duration::from_micros(120),
+        );
+        metrics
+    }
+
+    /// Sample lines are `name{labels} value` or `name value`; comments and the
+    /// `# EOF` trailer are skipped.
+    fn rendered_series_names(output: &str) -> Vec<String> {
+        let mut names: Vec<String> = output
+            .lines()
+            .filter(|line| !line.is_empty() && !line.starts_with('#'))
+            .map(|line| line.split(['{', ' ']).next().unwrap_or_default().to_owned())
+            .collect();
+        names.sort_unstable();
+        names.dedup();
+        names
+    }
+
+    #[test]
+    fn given_every_family_when_encoded_should_render_one_total_suffix() {
+        let output = populated_metrics().get_formatted_output();
+
+        assert!(
+            !output.contains("_total_total"),
+            "counter registered with a literal `_total` renders doubled:\n{output}"
+        );
+
+        let mut expected: Vec<String> = RENDERED_SERIES_NAMES
+            .iter()
+            .map(|name| (*name).to_owned())
+            .collect();
+        expected.sort_unstable();
+        assert_eq!(
+            rendered_series_names(&output),
+            expected,
+            "rendered series names drifted; dashboards key off these:\n{output}"
+        );
+    }
+
     #[test]
     fn test_metrics_init() {
         let metrics = Metrics::init();

core/integration/tests/connectors/runtime/benchmark.rs

@@ -435,6 +435,13 @@ async fn given_runtime_processing_batches_when_metrics_scraped_should_expose_sta
         "expected stage histogram in /metrics, got:\n{metrics_body}"
     );
 
+    // Both connectors have processed batches by now, so every counter family
+    // renders a sample and a doubled suffix would be visible on the wire.
+    assert!(
+        !metrics_body.contains("_total_total"),
+        "counter names must carry exactly one `_total`, got:\n{metrics_body}"
+    );
+
     let sink_total_count = parse_metric_value(
         &metrics_body,
         "iggy_connector_stage_duration_seconds_count",