feat(telemetry): export per-status serve_shape request counter (#4500)

## What

Adds a new unsampled, per-request metric from the shape-serving path:

```
electric.plug.serve_shape.requests.count   tags: [status, known_error, live]
```

It hangs off the existing `[:electric, :plug, :serve_shape]` telemetry
event (no new emission point) and threads a `known_error` flag into that
event's metadata, read straight off the `electric-internal-known-error`
response header.

## Why

We sample shape-request **spans** aggressively (head sampling + tail
overrides) to stay within our tracing event budget. That makes the span
dataset great for drill-down but unreliable as a health signal:

- It under-represents true volume (sampled), and
- It **drops some request classes entirely** — admission-control
rejections are marked `known_error` and are intentionally excluded from
span export, so a request plot built from spans can look perfectly
healthy while the system is actually shedding load under overload.

Admission rejection counts already exist as a metric
(`electric.admission_control.reject.count`), but there was no general,
status-dimensioned request counter. The existing `serve_shape` metrics
also (a) drop live/long-poll requests and (b) aren't tagged by response
status, so they can't express request mix or error rate.

This counter fills that gap:

- **Unsampled** — every request is counted, so there's no detection
floor. Errors and rejections are visible the moment they rise, not once
they cross a sampling threshold.
- **Counts live requests too** — unlike the other `serve_shape.*`
metrics, which `keep: live != true`. Cheap to do as an aggregated metric
(the reason we couldn't do it for spans doesn't apply).
- **Admission rejections appear inline** as `status=503,
known_error=true`. `check_admission` halts the conn, but a halt isn't a
raise, so the halted conn still flows through `emit_shape_telemetry/1`
and gets counted.
- **`known_error` matches the wire signal** — it's derived from the
`electric-internal-known-error` response header, the same byte
downstream consumers (e.g. the edge worker's tracing) key on, so the
classification is consistent end to end.

Intended use: this becomes the authoritative "requests by status / error
rate" dashboard panel and alert source, leaving the sampled span dataset
for exemplar drill-down.

## Changes

- `electric-telemetry`: define
`counter("electric.plug.serve_shape.requests.count", tags: [:status,
:known_error, :live])` against the existing event (explicit
`event_name`/`measurement` to avoid colliding with the existing
`serve_shape.count`).
- `sync-service`: add `known_error` to the `serve_shape` event metadata,
derived from the `electric-internal-known-error` response header. The
header check lives next to the code that sets it so the expected values
stay single-sourced.

## Cardinality

Bounded: `status` (~6 codes) × `known_error` (2) × `live` (2), per
stack.

## Notes

- One pre-existing gap is unchanged: the mid-stream re-raise path
(`Plug.Conn.chunk/2` raising after the response is committed) doesn't
emit the `serve_shape` event, so those requests aren't counted here —
same limitation that already affects every `serve_shape.*` metric.
- No rollout/rollback concerns: additive metric only, no behavior change
on the request path.

---------

Co-authored-by: Oleksii Sholik <oleksii@sholik.dev>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: 3 people

.changeset/serve-shape-request-metrics.md

@@ -0,0 +1,6 @@
+---
+'@core/electric-telemetry': patch
+'@core/sync-service': patch
+---
+
+Export a per-request `electric.plug.serve_shape.requests.count` metric tagged by `status`, `known_error` and `live`.

packages/electric-telemetry/lib/electric/telemetry/stack_telemetry.ex

@@ -93,6 +93,11 @@ defmodule ElectricTelemetry.StackTelemetry do
         unit: :byte,
         keep: fn metadata -> metadata[:live] != true end
       ),
+      counter("electric.plug.serve_shape.requests.count",
+        event_name: [:electric, :plug, :serve_shape],
+        measurement: :count,
+        tags: [:status, :known_error, :live]
+      ),
       distribution("electric.shape.response_size.bytes",
         unit: :byte,
         tags: [:root_table, :is_live, :stack_id]

packages/sync-service/lib/electric/plug/serve_shape_plug.ex

@@ -451,7 +451,8 @@ defmodule Electric.Plug.ServeShapePlug do
         shape_handle: get_handle(assigns) || conn.query_params["handle"],
         client_ip: conn.remote_ip,
         status: conn.status,
-        stack_id: stack_id
+        stack_id: stack_id,
+        known_error: Api.Response.conn_has_known_error?(conn)
       }
     )
 

packages/sync-service/lib/electric/shapes/api/response.ex

@@ -390,6 +390,13 @@ defmodule Electric.Shapes.Api.Response do
     Plug.Conn.put_resp_header(conn, @electric_known_error_header, "#{known_error}")
   end
 
+  # keeping this function close to `put_known_error_header/2` above so that we know exactly
+  # which value to expect for a set known_error header: i.e. "true" or "false" (and absent when
+  # known_error is nil), as opposed to e.g. "1" etc.
+  def conn_has_known_error?(conn) do
+    Plug.Conn.get_resp_header(conn, @electric_known_error_header) == ["true"]
+  end
+
   defp put_retry_after_header(conn, %__MODULE__{retry_after: nil}) do
     conn
   end

packages/sync-service/test/electric/plug/serve_shape_plug_test.exs

@@ -86,7 +86,8 @@ defmodule Electric.Plug.ServeShapePlugTest do
       sse_timeout: sse_timeout(ctx),
       max_age: max_age(ctx),
       stale_age: stale_age(ctx),
-      max_concurrent_requests: %{initial: 300, existing: 10_000}
+      max_concurrent_requests:
+        Access.get(ctx, :max_concurrent_requests, %{initial: 300, existing: 10_000})
     )
   end
 
@@ -1044,6 +1045,101 @@ defmodule Electric.Plug.ServeShapePlugTest do
         :telemetry.detach(handler_id)
       end
     end
+
+    test "[:electric, :plug, :serve_shape] tags a successful response with known_error: false",
+         ctx do
+      expect_shape_cache(
+        get_or_create_shape_handle: fn @test_shape, _stack_id, _opts ->
+          {@test_shape_handle, @test_offset}
+        end,
+        await_snapshot_start: fn @test_shape_handle, _ -> :started end
+      )
+
+      patch_shape_cache(has_shape?: fn @test_shape_handle, _opts -> true end)
+
+      next_offset = LogOffset.increment(@first_offset)
+
+      expect_storage(
+        get_chunk_end_log_offset: fn @before_all_offset, _ ->
+          @first_offset
+        end,
+        get_log_stream: fn @before_all_offset, @first_offset, @test_opts ->
+          [Jason.encode!(%{key: "log", value: "foo", headers: %{}, offset: next_offset})]
+        end
+      )
+
+      test_pid = self()
+      ref = make_ref()
+      handler_id = "test-serve-shape-known-error-false-#{inspect(ref)}"
+
+      :telemetry.attach(
+        handler_id,
+        [:electric, :plug, :serve_shape],
+        fn event, measurements, metadata, _config ->
+          send(test_pid, {:telemetry_serve_shape, event, measurements, metadata})
+        end,
+        nil
+      )
+
+      stack_id = ctx.stack_id
+
+      try do
+        conn =
+          ctx
+          |> conn(:get, %{"table" => "public.users"}, "?offset=-1")
+          |> call_serve_shape_plug(ctx)
+
+        assert conn.status == 200
+
+        assert_receive {:telemetry_serve_shape, [:electric, :plug, :serve_shape], _measurements,
+                        %{stack_id: ^stack_id} = metadata}
+
+        assert metadata.status == 200
+        assert metadata.known_error == false
+      after
+        :telemetry.detach(handler_id)
+      end
+    end
+
+    test "[:electric, :plug, :serve_shape] tags an admission-rejected 503 with known_error: true",
+         ctx do
+      # Force the load-shedding path: with a zero concurrency limit, check_admission
+      # rejects immediately with a 503 carrying the `electric-internal-known-error`
+      # header.
+      ctx = Map.put(ctx, :max_concurrent_requests, %{initial: 0, existing: 0})
+
+      test_pid = self()
+      ref = make_ref()
+      handler_id = "test-serve-shape-known-error-true-#{inspect(ref)}"
+
+      :telemetry.attach(
+        handler_id,
+        [:electric, :plug, :serve_shape],
+        fn event, measurements, metadata, _config ->
+          send(test_pid, {:telemetry_serve_shape, event, measurements, metadata})
+        end,
+        nil
+      )
+
+      stack_id = ctx.stack_id
+
+      try do
+        conn =
+          ctx
+          |> conn(:get, %{"table" => "public.users"}, "?offset=-1")
+          |> call_serve_shape_plug(ctx)
+
+        assert conn.status == 503
+
+        assert_receive {:telemetry_serve_shape, [:electric, :plug, :serve_shape], _measurements,
+                        %{stack_id: ^stack_id} = metadata}
+
+        assert metadata.status == 503
+        assert metadata.known_error == true
+      after
+        :telemetry.detach(handler_id)
+      end
+    end
   end
 
   describe "serving shapes with sse mode" do