feat(server): publish [:grpc, :server, :rpc, :abort] telemetry

grpc_core/lib/grpc/telemetry.ex

@@ -47,12 +47,19 @@ defmodule GRPC.Telemetry do
     * `[:grpc, :server, :rpc, :exception]` - Published if any exception occurs while receiving a message.
       * `:duration` - the duration as measured through `System.monotonic_time()`
         for the execution since the start of the pipeline until the exception happened.
+    * `[:grpc, :server, :rpc, :abort]` - Published when the adapter stops an in-flight
+      RPC: expired deadline, client cancellation or a dropped connection. The exit
+      signal doesn't unwind the RPC process, so `:stop` and `:exception` are not
+      published for such a call; this event comes from the adapter's process.
+      * `:duration` - the duration as measured through `System.monotonic_time()`
+        from the arrival of the request until the abort.
 
   | event        | measurements | metadata |
   |--------------|--------------|----------|
   | `[:rpc, :start]`  | `:count`     | `:stream`, `:server`, `:endpoint`, `:function_name` |
   | `[:rpc, :stop]`  | `:duration`  | `:stream`, `:server`, `:endpoint`, `:function_name` , `:result` |
   | `[:rpc, :exception]` | `:duration`  | `:stream`, `:server`, `:endpoint`, `:function_name`, `:kind`, `:reason`, `:stacktrace` |
+  | `[:rpc, :abort]` | `:duration`  | `:stream`, `:server`, `:endpoint`, `:path`, `:pid`, `:reason` |
 
   ### Metadata
 
@@ -62,6 +69,12 @@ defmodule GRPC.Telemetry do
     * `:endpoint` - the endpoint module name.
     * `:result` - the result returned from the interceptor pipeline.
 
+  `:abort` events also include `:path` (the request path), `:pid` (the RPC
+  process being stopped) and `:reason` (the exit reason, e.g. `:timeout`).
+  They carry no `:function_name`, which is resolved in the RPC process; for
+  the same reason `:stream` holds only what the adapter built on arrival, so
+  read `:server` and `:endpoint` from the metadata rather than from it.
+
   `:exception` events also include some error metadata:
 
     * `:reason` is the error value in case of `catch` or the actual exception in case of `rescue`.

grpc_server/CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+### Enhancements
+
+  * `[:grpc, :server, :rpc, :abort]` is now published when the adapter stops an in-flight RPC — an expired deadline, a client cancellation, a dropped connection. The exit signal that stops the RPC process does not unwind it, so `:stop` and `:exception` cannot be published for such a call; the new event carries the stream, server, endpoint, request path, RPC pid and exit reason.
+
 ## v1.0.4 (2026-0-15)
 
 ### Bug Fixes

grpc_server/lib/grpc/server/adapters/cowboy/handler.ex

@@ -14,6 +14,8 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
   @default_trailers HTTP2.server_trailers()
   @trailers_flag 0b1000_0000
 
+  @abort_published :"$grpc_abort_published"
+
   # 4 MB – matches gRPC-Go's default max receive message size.
   # Override per-server with the :max_body_size option (bytes).
   @default_max_body_size 4 * 1024 * 1024
@@ -35,6 +37,7 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
          {:ok, codec} <- find_codec(sub_type, content_type, server),
          {:ok, compressor} <- find_compressor(req, server) do
       stream_pid = self()
+      started_at = System.monotonic_time()
       http_transcode = access_mode == :http_transcoding
       request_headers = :cowboy_req.headers(req)
 
@@ -81,6 +84,10 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
         req,
         %{
           pid: server_rpc_pid,
+          stream: stream,
+          endpoint: endpoint,
+          route: route,
+          started_at: started_at,
           handling_timer: timer_ref,
           pending_reader: nil,
           access_mode: access_mode,
@@ -525,8 +532,8 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
     {:stop, req, state}
   end
 
-  def terminate(reason, _req, %{pid: pid}) do
-    exit_handler(pid, reason)
+  def terminate(reason, _req, state = %{pid: _pid}) do
+    abort_rpc(state, reason)
     :ok
   end
 
@@ -644,6 +651,51 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
     end
   end
 
+  # Stops an RPC that is still running, and publishes it. The exit signal does
+  # not unwind the RPC process, so `:telemetry.span/3` around the call publishes
+  # neither `:stop` nor `:exception`; this process outlives it and publishes
+  # `:abort` before signalling.
+  #
+  # Both abort paths can run for one call — `send_error/4` signals the RPC
+  # process and cowboy then calls `terminate/3` — and the signal is
+  # asynchronous, so the process may still be alive for the second. The flag
+  # keeps the event to one per call; it lives in the cowboy request process,
+  # which handles this call only.
+  defp abort_rpc(state, reason) do
+    case Map.get(state, :pid) do
+      pid when is_pid(pid) ->
+        if Process.alive?(pid) do
+          unless Process.put(@abort_published, true) do
+            publish_abort(state, pid, reason)
+          end
+
+          exit_handler(pid, reason)
+        end
+
+      _no_rpc_process ->
+        :ok
+    end
+
+    :ok
+  end
+
+  defp publish_abort(state, pid, reason) do
+    %{stream: stream, endpoint: endpoint, route: route, started_at: started_at} = state
+
+    :telemetry.execute(
+      GRPC.Telemetry.server_rpc_prefix() ++ [:abort],
+      %{duration: System.monotonic_time() - started_at},
+      %{
+        stream: stream,
+        server: stream.server,
+        endpoint: endpoint,
+        path: route,
+        pid: pid,
+        reason: reason
+      }
+    )
+  end
+
   defp timeout_left_opt(timer, opts \\ %{}) do
     case timer do
       nil ->
@@ -709,9 +761,7 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
         do: GRPC.Status.http_code(error.status),
         else: 200
 
-    if pid = Map.get(state, :pid) do
-      exit_handler(pid, reason)
-    end
+    abort_rpc(state, reason)
 
     send_error_trailers(req, status, trailers, state)
   end

grpc_server/test/grpc/server/adapters/cowboy/handler_test.exs

@@ -2,6 +2,7 @@ defmodule GRPC.Server.Adapters.Cowboy.HandlerTest do
   use ExUnit.Case, async: false
 
   import ExUnit.CaptureLog
+  import GRPC.DataCase, only: [attach_telemetry: 1]
 
   # --------------------------------------------------------------------------
   # Minimal server used across all tests
@@ -15,6 +16,25 @@ defmodule GRPC.Server.Adapters.Cowboy.HandlerTest do
     end
   end
 
+  defmodule SlowServer do
+    use GRPC.Server, service: Helloworld.Greeter.Service
+
+    def say_hello(req, _stream) do
+      Process.sleep(5_000)
+      %Helloworld.HelloReply{message: "Hello, #{req.name}"}
+    end
+  end
+
+  defmodule TrappingServer do
+    use GRPC.Server, service: Helloworld.Greeter.Service
+
+    def say_hello(req, _stream) do
+      Process.flag(:trap_exit, true)
+      Process.sleep(3_000)
+      %Helloworld.HelloReply{message: "Hello, #{req.name}"}
+    end
+  end
+
   # --------------------------------------------------------------------------
   # Helpers
   # --------------------------------------------------------------------------
@@ -155,6 +175,111 @@ defmodule GRPC.Server.Adapters.Cowboy.HandlerTest do
     end
   end
 
+  # --------------------------------------------------------------------------
+  # Tests: telemetry for RPCs the adapter stops before they return
+  # --------------------------------------------------------------------------
+
+  describe "aborted RPCs" do
+    test "an expired deadline publishes :abort and no :stop" do
+      attach_telemetry([:grpc, :server, :rpc, :abort])
+      attach_telemetry([:grpc, :server, :rpc, :stop])
+      attach_telemetry([:grpc, :server, :rpc, :exception])
+
+      capture_log(fn ->
+        run_server_with_opts([SlowServer], [], fn port ->
+          headers = [{"grpc-timeout", "50m"} | grpc_request_headers()]
+          body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "slow"}))
+
+          conn = open_h2(port)
+          ref = :gun.post(conn, "/helloworld.Greeter/SayHello", headers, body)
+
+          assert collect_grpc_status(conn, ref) == "4"
+
+          :gun.close(conn)
+        end)
+      end)
+
+      assert_receive {:telemetry, [:grpc, :server, :rpc, :abort], measurements, metadata}
+      assert metadata.reason == :timeout
+      assert metadata.path == "/helloworld.Greeter/SayHello"
+      assert metadata.server == SlowServer
+      assert metadata.endpoint == nil
+      assert is_pid(metadata.pid)
+      assert metadata.stream.http_request_headers["grpc-timeout"] == "50m"
+      assert metadata.stream.deadline
+      assert measurements.duration > 0
+
+      # The RPC process is stopped by a signal it cannot unwind, so the span
+      # around the call publishes nothing: without `:abort` the call would be
+      # absent from telemetry entirely.
+      refute_receive {:telemetry, [:grpc, :server, :rpc, :stop], _, _}, 200
+      refute_receive {:telemetry, [:grpc, :server, :rpc, :exception], _, _}, 10
+    end
+
+    test "publishes :abort once when both abort paths run for one call" do
+      attach_telemetry([:grpc, :server, :rpc, :abort])
+
+      capture_log(fn ->
+        run_server_with_opts([TrappingServer], [], fn port ->
+          headers = [{"grpc-timeout", "50m"} | grpc_request_headers()]
+          body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "trap"}))
+
+          conn = open_h2(port)
+          ref = :gun.post(conn, "/helloworld.Greeter/SayHello", headers, body)
+
+          assert collect_grpc_status(conn, ref) == "4"
+
+          :gun.close(conn)
+        end)
+      end)
+
+      # `send_error/4` publishes and signals; cowboy then calls `terminate/3`,
+      # by which point this RPC process has not died.
+      assert_receive {:telemetry, [:grpc, :server, :rpc, :abort], _, %{reason: :timeout}}
+      refute_receive {:telemetry, [:grpc, :server, :rpc, :abort], _, _}, 500
+    end
+
+    test "a dropped connection publishes :abort" do
+      attach_telemetry([:grpc, :server, :rpc, :abort])
+
+      capture_log(fn ->
+        run_server_with_opts([SlowServer], [], fn port ->
+          body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "slow"}))
+
+          conn = open_h2(port)
+          _ref = :gun.post(conn, "/helloworld.Greeter/SayHello", grpc_request_headers(), body)
+
+          # No deadline: the RPC is still running when the client goes away.
+          Process.sleep(100)
+          :gun.close(conn)
+
+          assert_receive {:telemetry, [:grpc, :server, :rpc, :abort], _, metadata}, 1_000
+          assert metadata.path == "/helloworld.Greeter/SayHello"
+          refute metadata.reason == :timeout
+        end)
+      end)
+    end
+
+    test "a call that returns publishes :stop and no :abort" do
+      attach_telemetry([:grpc, :server, :rpc, :abort])
+      attach_telemetry([:grpc, :server, :rpc, :stop])
+
+      run_server_with_opts([HelloServer], [], fn port ->
+        body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "hi"}))
+
+        conn = open_h2(port)
+        ref = :gun.post(conn, "/helloworld.Greeter/SayHello", grpc_request_headers(), body)
+
+        assert collect_grpc_status(conn, ref) == "0"
+
+        :gun.close(conn)
+      end)
+
+      assert_receive {:telemetry, [:grpc, :server, :rpc, :stop], _measurements, _metadata}
+      refute_receive {:telemetry, [:grpc, :server, :rpc, :abort], _, _}, 200
+    end
+  end
+
   # --------------------------------------------------------------------------
   # Private helper: start a server with specific opts and run a test function
   # --------------------------------------------------------------------------

grpc_server/test/support/data_case.ex

@@ -6,4 +6,25 @@ defmodule GRPC.DataCase do
       import GRPC.Factory
     end
   end
+
+  @doc """
+  Attaches a telemetry handler for `event` that forwards emissions to the
+  test process as `{:telemetry, event, measurements, metadata}` messages,
+  for use with `assert_receive`. The handler is detached on test exit.
+  """
+  def attach_telemetry(event) do
+    handler_id = {__MODULE__, self(), System.unique_integer()}
+    test_pid = self()
+
+    :telemetry.attach(
+      handler_id,
+      event,
+      fn event, measurements, metadata, _config ->
+        send(test_pid, {:telemetry, event, measurements, metadata})
+      end,
+      nil
+    )
+
+    ExUnit.Callbacks.on_exit(fn -> :telemetry.detach(handler_id) end)
+  end
 end