diff --git a/api/api.go b/api/api.go
index 9336241594..a58292dff2 100644
--- a/api/api.go
+++ b/api/api.go
@@ -75,7 +75,13 @@ type Options struct {
 	// limit of GOMAXPROCS or 8, whichever is larger. Status code 503 is served
 	// for GET requests that would exceed the concurrency limit; Connect calls
 	// receive ResourceExhausted.
-	Concurrency int
+	Concurrency                int
+	ConnectUnaryConcurrency    int
+	ConnectStreamConcurrency   int
+	ConnectUnaryTimeout        time.Duration
+	ConnectReadMaxBytes        int
+	ConnectSendMaxBytes        int
+	ConnectMaxRequestBodyBytes int64
 	// Logger is used for logging, if nil, no logging will happen.
 	Logger *slog.Logger
 	// Registry is used to register Prometheus metrics. If nil, no metrics
@@ -133,12 +139,31 @@ func New(opts Options) (*API, error) {
 	if err != nil {
 		return nil, err
 	}
-	connect := apiconnect.NewAPI(apiconnect.Options{
-		Peer:              opts.Peer,
-		UnaryConcurrency:  concurrency,
-		StreamConcurrency: concurrency,
-		UnaryTimeout:      opts.Timeout,
+	unaryConcurrency := opts.ConnectUnaryConcurrency
+	if unaryConcurrency < 1 {
+		unaryConcurrency = concurrency
+	}
+	streamConcurrency := opts.ConnectStreamConcurrency
+	if streamConcurrency < 1 {
+		streamConcurrency = concurrency
+	}
+	unaryTimeout := opts.ConnectUnaryTimeout
+	if unaryTimeout == 0 {
+		unaryTimeout = opts.Timeout
+	}
+	connect, err := apiconnect.NewAPI(apiconnect.Options{
+		Peer:                opts.Peer,
+		Registerer:          opts.Registry,
+		UnaryConcurrency:    unaryConcurrency,
+		StreamConcurrency:   streamConcurrency,
+		UnaryTimeout:        unaryTimeout,
+		ReadMaxBytes:        opts.ConnectReadMaxBytes,
+		SendMaxBytes:        opts.ConnectSendMaxBytes,
+		MaxRequestBodyBytes: opts.ConnectMaxRequestBodyBytes,
 	})
+	if err != nil {
+		return nil, err
+	}
 
 	requestsInFlight := prometheus.NewGauge(prometheus.GaugeOpts{
 		Name:        "alertmanager_http_requests_in_flight",
diff --git a/api/api_test.go b/api/api_test.go
index 4a86faf6f6..ce6ffa08fc 100644
--- a/api/api_test.go
+++ b/api/api_test.go
@@ -85,7 +85,8 @@ func TestConcurrencyLimitHandler(t *testing.T) {
 func TestConnectProceduresRegistered(t *testing.T) {
 	for _, routePrefix := range []string{"/", "/alertmanager"} {
 		t.Run(routePrefix, func(t *testing.T) {
-			connectAPI := apiconnect.NewAPI(apiconnect.Options{})
+			connectAPI, err := apiconnect.NewAPI(apiconnect.Options{})
+			require.NoError(t, err)
 			requestDuration := prometheus.NewHistogramVec(
 				prometheus.HistogramOpts{Name: "test_registered_http_request_duration_seconds"},
 				[]string{"handler", "method", "code"},
diff --git a/api/connect/connect.go b/api/connect/connect.go
index 684694c9a0..edf7a92ad4 100644
--- a/api/connect/connect.go
+++ b/api/connect/connect.go
@@ -21,6 +21,7 @@ package apiconnect
 import (
 	"context"
 	"errors"
+	"fmt"
 	"net/http"
 	"runtime"
 	"sync/atomic"
@@ -29,6 +30,7 @@ import (
 	"connectrpc.com/connect"
 	"connectrpc.com/grpchealth"
 	"connectrpc.com/grpcreflect"
+	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/prometheus/alertmanager/api/status/v3alpha/statusv3alphaconnect"
 	"github.com/prometheus/alertmanager/cluster"
@@ -37,14 +39,21 @@ import (
 
 // Options configures the Connect API.
 type Options struct {
-	Peer              cluster.ClusterPeer
-	UnaryConcurrency  int
-	StreamConcurrency int
-	UnaryTimeout      time.Duration
+	Peer                cluster.ClusterPeer
+	Registerer          prometheus.Registerer
+	UnaryConcurrency    int
+	StreamConcurrency   int
+	UnaryTimeout        time.Duration
+	ReadMaxBytes        int
+	SendMaxBytes        int
+	MaxRequestBodyBytes int64
 }
 
 type procedureDescriptor struct {
-	path string
+	path       string
+	service    string
+	procedure  string
+	streamType connect.StreamType
 }
 
 type serviceDescriptor struct {
@@ -53,6 +62,62 @@ type serviceDescriptor struct {
 	handler    func(...connect.HandlerOption) (string, http.Handler)
 }
 
+type rpcMetrics struct {
+	unaryInFlight   *prometheus.GaugeVec
+	unaryRejected   *prometheus.CounterVec
+	unaryDuration   *prometheus.HistogramVec
+	unaryDeadlines  *prometheus.CounterVec
+	streamsActive   *prometheus.GaugeVec
+	streamsRejected *prometheus.CounterVec
+}
+
+func newRPCMetrics(reg prometheus.Registerer) (*rpcMetrics, error) {
+	labels := []string{"service", "procedure"}
+	outcomeLabels := []string{"service", "procedure", "outcome"}
+	metrics := &rpcMetrics{
+		unaryInFlight: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "alertmanager_api_connect_unary_requests_in_flight",
+			Help: "Current number of admitted Connect unary RPCs.",
+		}, labels),
+		unaryRejected: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "alertmanager_api_connect_unary_admission_rejections_total",
+			Help: "Total number of Connect unary RPCs rejected by admission control.",
+		}, labels),
+		unaryDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "alertmanager_api_connect_unary_request_duration_seconds",
+			Help:    "Duration of Connect unary RPCs.",
+			Buckets: prometheus.DefBuckets,
+		}, outcomeLabels),
+		unaryDeadlines: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "alertmanager_api_connect_unary_deadline_exceeded_total",
+			Help: "Total number of Connect unary RPCs that exceeded their deadline.",
+		}, labels),
+		streamsActive: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "alertmanager_api_connect_streams_active",
+			Help: "Current number of admitted Connect streams.",
+		}, labels),
+		streamsRejected: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "alertmanager_api_connect_stream_admission_rejections_total",
+			Help: "Total number of Connect streams rejected by admission control.",
+		}, labels),
+	}
+	if reg != nil {
+		for _, collector := range []prometheus.Collector{
+			metrics.unaryInFlight,
+			metrics.unaryRejected,
+			metrics.unaryDuration,
+			metrics.unaryDeadlines,
+			metrics.streamsActive,
+			metrics.streamsRejected,
+		} {
+			if err := reg.Register(collector); err != nil {
+				return nil, fmt.Errorf("register Connect API metrics: %w", err)
+			}
+		}
+	}
+	return metrics, nil
+}
+
 // API implements the ConnectRPC service handlers for the Connect API.
 type API struct {
 	peer            cluster.ClusterPeer
@@ -61,25 +126,28 @@ type API struct {
 	peerSnapshotSem chan struct{}
 	services        []serviceDescriptor
 	procedures      map[string]procedureDescriptor
+	readMaxBytes    int
+	sendMaxBytes    int
+	maxRequestBytes int64
 
 	configSnapshot atomic.Pointer[string]
 }
 
 // NewAPI returns a new Connect API handler. Peer may be nil when clustering
 // is disabled.
-func NewAPI(opts Options) *API {
-	unaryConcurrency := defaultConcurrency(opts.UnaryConcurrency)
-	streamConcurrency := defaultConcurrency(opts.StreamConcurrency)
+func NewAPI(opts Options) (*API, error) {
+	metrics, err := newRPCMetrics(opts.Registerer)
+	if err != nil {
+		return nil, err
+	}
 	api := &API{
 		peer:            opts.Peer,
 		uptime:          time.Now(),
 		peerSnapshotSem: make(chan struct{}, 1),
 		procedures:      make(map[string]procedureDescriptor),
-		admission: &admissionInterceptor{
-			unary:        make(chan struct{}, unaryConcurrency),
-			streams:      make(chan struct{}, streamConcurrency),
-			unaryTimeout: opts.UnaryTimeout,
-		},
+		readMaxBytes:    opts.ReadMaxBytes,
+		sendMaxBytes:    opts.SendMaxBytes,
+		maxRequestBytes: opts.MaxRequestBodyBytes,
 	}
 	api.services = api.serviceDescriptors()
 	for _, service := range api.services {
@@ -87,7 +155,14 @@ func NewAPI(opts Options) *API {
 			api.procedures[procedure.path] = procedure
 		}
 	}
-	return api
+	api.admission = &admissionInterceptor{
+		unary:        make(chan struct{}, defaultConcurrency(opts.UnaryConcurrency)),
+		streams:      make(chan struct{}, defaultConcurrency(opts.StreamConcurrency)),
+		unaryTimeout: opts.UnaryTimeout,
+		procedures:   api.procedures,
+		metrics:      metrics,
+	}
+	return api, nil
 }
 
 func defaultConcurrency(concurrency int) int {
@@ -97,15 +172,20 @@ func defaultConcurrency(concurrency int) int {
 	return concurrency
 }
 
-func procedure(service, name string) procedureDescriptor {
-	return procedureDescriptor{path: "/" + service + "/" + name}
+func procedure(service, name string, streamType connect.StreamType) procedureDescriptor {
+	return procedureDescriptor{
+		path:       "/" + service + "/" + name,
+		service:    service,
+		procedure:  name,
+		streamType: streamType,
+	}
 }
 
 func (api *API) serviceDescriptors() []serviceDescriptor {
 	status := serviceDescriptor{
 		name: statusv3alphaconnect.StatusServiceName,
 		procedures: []procedureDescriptor{
-			procedure(statusv3alphaconnect.StatusServiceName, "GetStatus"),
+			procedure(statusv3alphaconnect.StatusServiceName, "GetStatus", connect.StreamTypeUnary),
 		},
 		handler: func(opts ...connect.HandlerOption) (string, http.Handler) {
 			return statusv3alphaconnect.NewStatusServiceHandler(api, opts...)
@@ -119,8 +199,8 @@ func (api *API) serviceDescriptors() []serviceDescriptor {
 		{
 			name: grpchealth.HealthV1ServiceName,
 			procedures: []procedureDescriptor{
-				procedure(grpchealth.HealthV1ServiceName, "Check"),
-				procedure(grpchealth.HealthV1ServiceName, "Watch"),
+				procedure(grpchealth.HealthV1ServiceName, "Check", connect.StreamTypeUnary),
+				procedure(grpchealth.HealthV1ServiceName, "Watch", connect.StreamTypeServer),
 			},
 			handler: func(opts ...connect.HandlerOption) (string, http.Handler) {
 				return grpchealth.NewHandler(checker, opts...)
@@ -129,7 +209,7 @@ func (api *API) serviceDescriptors() []serviceDescriptor {
 		{
 			name: grpcreflect.ReflectV1ServiceName,
 			procedures: []procedureDescriptor{
-				procedure(grpcreflect.ReflectV1ServiceName, "ServerReflectionInfo"),
+				procedure(grpcreflect.ReflectV1ServiceName, "ServerReflectionInfo", connect.StreamTypeBidi),
 			},
 			handler: func(opts ...connect.HandlerOption) (string, http.Handler) {
 				return grpcreflect.NewHandlerV1(reflector, opts...)
@@ -138,7 +218,7 @@ func (api *API) serviceDescriptors() []serviceDescriptor {
 		{
 			name: grpcreflect.ReflectV1AlphaServiceName,
 			procedures: []procedureDescriptor{
-				procedure(grpcreflect.ReflectV1AlphaServiceName, "ServerReflectionInfo"),
+				procedure(grpcreflect.ReflectV1AlphaServiceName, "ServerReflectionInfo", connect.StreamTypeBidi),
 			},
 			handler: func(opts ...connect.HandlerOption) (string, http.Handler) {
 				return grpcreflect.NewHandlerV1Alpha(reflector, opts...)
@@ -147,39 +227,141 @@ func (api *API) serviceDescriptors() []serviceDescriptor {
 	}
 }
 
+type (
+	admittedContextKey           struct{}
+	observationContextKey        struct{}
+	requestStartContextKey       struct{}
+	responseControllerContextKey struct{}
+)
+
 // admissionInterceptor gives unary RPCs and streams independent capacity so
 // slow Connect clients cannot consume the API v2 GET request allowance.
 type admissionInterceptor struct {
 	unary        chan struct{}
 	streams      chan struct{}
 	unaryTimeout time.Duration
+	procedures   map[string]procedureDescriptor
+	metrics      *rpcMetrics
 }
 
-func (i *admissionInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
-	return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
+func (i *admissionInterceptor) descriptor(path string, streamType connect.StreamType) procedureDescriptor {
+	if procedure, ok := i.procedures[path]; ok {
+		return procedure
+	}
+	return procedureDescriptor{service: "unknown", procedure: "unknown", streamType: streamType}
+}
+
+func (i *admissionInterceptor) enter(desc procedureDescriptor) (func(), error) {
+	labels := prometheus.Labels{"service": desc.service, "procedure": desc.procedure}
+	if desc.streamType == connect.StreamTypeUnary {
 		select {
 		case i.unary <- struct{}{}:
-			defer func() { <-i.unary }()
+			if i.metrics != nil {
+				i.metrics.unaryInFlight.With(labels).Inc()
+			}
+			return func() {
+				<-i.unary
+				if i.metrics != nil {
+					i.metrics.unaryInFlight.With(labels).Dec()
+				}
+			}, nil
 		default:
+			if i.metrics != nil {
+				i.metrics.unaryRejected.With(labels).Inc()
+			}
 			return nil, connect.NewError(connect.CodeResourceExhausted, errors.New("maximum concurrent unary RPCs reached"))
 		}
-		if i.unaryTimeout > 0 {
-			var cancel context.CancelFunc
-			ctx, cancel = context.WithTimeout(ctx, i.unaryTimeout)
-			defer cancel()
+	}
+	select {
+	case i.streams <- struct{}{}:
+		if i.metrics != nil {
+			i.metrics.streamsActive.With(labels).Inc()
+		}
+		return func() {
+			<-i.streams
+			if i.metrics != nil {
+				i.metrics.streamsActive.With(labels).Dec()
+			}
+		}, nil
+	default:
+		if i.metrics != nil {
+			i.metrics.streamsRejected.With(labels).Inc()
+		}
+		return nil, connect.NewError(connect.CodeResourceExhausted, errors.New("maximum concurrent streams reached"))
+	}
+}
+
+func (i *admissionInterceptor) startTime(ctx context.Context) time.Time {
+	if started, ok := ctx.Value(requestStartContextKey{}).(time.Time); ok {
+		return started
+	}
+	return time.Now()
+}
+
+func (i *admissionInterceptor) observe(desc procedureDescriptor, started time.Time, err error) {
+	if i.metrics == nil {
+		return
+	}
+	outcome := "ok"
+	if err != nil {
+		outcome = connect.CodeOf(err).String()
+	}
+	if desc.streamType == connect.StreamTypeUnary {
+		labels := prometheus.Labels{"service": desc.service, "procedure": desc.procedure, "outcome": outcome}
+		i.metrics.unaryDuration.With(labels).Observe(time.Since(started).Seconds())
+	}
+}
+
+func normalizeContextError(ctx context.Context, err error) error {
+	if err == nil {
+		return nil
+	}
+	if connect.CodeOf(err) != connect.CodeUnknown {
+		return err
+	}
+	cause := context.Cause(ctx)
+	if errors.Is(cause, context.DeadlineExceeded) {
+		return connect.NewError(connect.CodeDeadlineExceeded, context.DeadlineExceeded)
+	}
+	switch {
+	case errors.Is(err, context.DeadlineExceeded):
+		return connect.NewError(connect.CodeDeadlineExceeded, context.DeadlineExceeded)
+	case errors.Is(err, context.Canceled), errors.Is(cause, context.Canceled):
+		return connect.NewError(connect.CodeCanceled, context.Canceled)
+	default:
+		return err
+	}
+}
+
+func (i *admissionInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
+	return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
+		desc := i.descriptor("", connect.StreamTypeUnary)
+		if req != nil {
+			desc = i.descriptor(req.Spec().Procedure, connect.StreamTypeUnary)
+		}
+		started := i.startTime(ctx)
+		if controller, ok := ctx.Value(responseControllerContextKey{}).(*http.ResponseController); ok {
+			_ = controller.SetReadDeadline(time.Time{})
+		}
+		if _, admitted := ctx.Value(admittedContextKey{}).(struct{}); !admitted {
+			release, err := i.enter(desc)
+			if err != nil {
+				return nil, err
+			}
+			defer release()
+			if i.unaryTimeout > 0 {
+				var cancel context.CancelFunc
+				ctx, cancel = context.WithTimeout(ctx, i.unaryTimeout)
+				defer cancel()
+			}
 		}
 		response, err := next(ctx, req)
-		if err == nil || connect.CodeOf(err) != connect.CodeUnknown {
-			return response, err
-		}
-		switch {
-		case errors.Is(err, context.DeadlineExceeded):
-			return nil, connect.NewError(connect.CodeDeadlineExceeded, err)
-		case errors.Is(err, context.Canceled):
-			return nil, connect.NewError(connect.CodeCanceled, err)
-		default:
-			return response, err
+		err = normalizeContextError(ctx, err)
+		i.observe(desc, started, err)
+		if observed, ok := ctx.Value(observationContextKey{}).(*atomic.Bool); ok {
+			observed.Store(true)
 		}
+		return response, err
 	}
 }
 
@@ -189,13 +371,18 @@ func (i *admissionInterceptor) WrapStreamingClient(next connect.StreamingClientF
 
 func (i *admissionInterceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
 	return func(ctx context.Context, conn connect.StreamingHandlerConn) error {
-		select {
-		case i.streams <- struct{}{}:
-			defer func() { <-i.streams }()
-		default:
-			return connect.NewError(connect.CodeResourceExhausted, errors.New("maximum concurrent streams reached"))
+		desc := i.descriptor("", connect.StreamTypeBidi)
+		if conn != nil {
+			desc = i.descriptor(conn.Spec().Procedure, conn.Spec().StreamType)
+		}
+		if _, admitted := ctx.Value(admittedContextKey{}).(struct{}); !admitted {
+			release, err := i.enter(desc)
+			if err != nil {
+				return err
+			}
+			defer release()
 		}
-		return next(ctx, conn)
+		return normalizeContextError(ctx, next(ctx, conn))
 	}
 }
 
@@ -233,9 +420,71 @@ func (api *API) Procedures() []string {
 	return procedures
 }
 
+func (api *API) controlHandler(next http.Handler, errorWriter *connect.ErrorWriter) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		desc, ok := api.procedures[r.URL.Path]
+		if !ok {
+			next.ServeHTTP(w, r)
+			return
+		}
+		release, err := api.admission.enter(desc)
+		if err != nil {
+			_ = r.Body.Close()
+			_ = errorWriter.Write(w, r, err)
+			return
+		}
+		defer release()
+
+		controller := http.NewResponseController(w)
+		started := time.Now()
+		observed := &atomic.Bool{}
+		ctx := context.WithValue(r.Context(), admittedContextKey{}, struct{}{})
+		ctx = context.WithValue(ctx, observationContextKey{}, observed)
+		ctx = context.WithValue(ctx, requestStartContextKey{}, started)
+		ctx = context.WithValue(ctx, responseControllerContextKey{}, controller)
+		if desc.streamType == connect.StreamTypeUnary && api.admission.unaryTimeout > 0 {
+			var cancel context.CancelFunc
+			ctx, cancel = context.WithTimeout(ctx, api.admission.unaryTimeout)
+			defer cancel()
+			if deadline, ok := ctx.Deadline(); ok {
+				_ = controller.SetReadDeadline(deadline)
+			}
+		}
+		defer func() {
+			if desc.streamType == connect.StreamTypeUnary && !observed.Load() {
+				err := normalizeContextError(ctx, errors.New("rpc ended before handler execution"))
+				api.admission.observe(desc, started, err)
+			}
+		}()
+		if desc.streamType == connect.StreamTypeUnary {
+			defer func() {
+				if errors.Is(context.Cause(ctx), context.DeadlineExceeded) {
+					api.admission.metrics.unaryDeadlines.With(prometheus.Labels{"service": desc.service, "procedure": desc.procedure}).Inc()
+				}
+			}()
+		}
+		defer func() {
+			if ctx.Err() == nil {
+				_ = controller.SetReadDeadline(time.Time{})
+			}
+		}()
+
+		request := r.WithContext(ctx)
+		handler := next
+		if desc.streamType == connect.StreamTypeUnary && api.maxRequestBytes > 0 {
+			handler = http.MaxBytesHandler(handler, api.maxRequestBytes)
+		}
+		handler.ServeHTTP(w, request)
+	})
+}
+
 // buildHandler registers every ConnectRPC service on a fresh mux.
 func (api *API) buildHandler(opts ...connect.HandlerOption) http.Handler {
-	opts = append([]connect.HandlerOption{connect.WithInterceptors(api.admission)}, opts...)
+	opts = append(opts,
+		connect.WithReadMaxBytes(api.readMaxBytes),
+		connect.WithSendMaxBytes(api.sendMaxBytes),
+		connect.WithInterceptors(api.admission),
+	)
 
 	mux := http.NewServeMux()
 	for _, service := range api.services {
@@ -245,5 +494,5 @@ func (api *API) buildHandler(opts ...connect.HandlerOption) http.Handler {
 		}
 		mux.Handle(path, handler)
 	}
-	return mux
+	return api.controlHandler(mux, connect.NewErrorWriter(opts...))
 }
diff --git a/api/connect/connect_suite_test.go b/api/connect/connect_suite_test.go
index b12041a4b0..96bc497d0f 100644
--- a/api/connect/connect_suite_test.go
+++ b/api/connect/connect_suite_test.go
@@ -24,3 +24,10 @@ func TestConnectAPI(t *testing.T) {
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "Connect API Suite")
 }
+
+func newTestAPI(opts Options) *API {
+	GinkgoHelper()
+	api, err := NewAPI(opts)
+	Expect(err).NotTo(HaveOccurred())
+	return api
+}
diff --git a/api/connect/health_test.go b/api/connect/health_test.go
index bd0090912b..2bba462e1c 100644
--- a/api/connect/health_test.go
+++ b/api/connect/health_test.go
@@ -33,7 +33,7 @@ import (
 // protocol with JSON, which needs only an HTTP/1.1 client.
 var _ = Describe("gRPC health", func() {
 	It("reports serving for the server and StatusService", func() {
-		api := NewAPI(Options{})
+		api := newTestAPI(Options{})
 		api.Update(&config.Config{})
 
 		srv := httptest.NewServer(api.Handler())
diff --git a/api/connect/status_test.go b/api/connect/status_test.go
index d5dbeb33f8..b1cd1edace 100644
--- a/api/connect/status_test.go
+++ b/api/connect/status_test.go
@@ -15,6 +15,7 @@ package apiconnect
 
 import (
 	"context"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"sync"
@@ -24,6 +25,7 @@ import (
 	"connectrpc.com/connect"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/version"
 
 	statusv3alpha "github.com/prometheus/alertmanager/api/status/v3alpha"
@@ -72,7 +74,7 @@ func (p *blockingPeer) unblock() { p.releaseOnce.Do(func() { close(p.release) })
 
 var _ = Describe("StatusService", func() {
 	It("returns status when clustering is disabled", func() {
-		api := NewAPI(Options{})
+		api := newTestAPI(Options{})
 		api.Update(&config.Config{})
 
 		resp, err := api.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{}))
@@ -102,7 +104,7 @@ var _ = Describe("StatusService", func() {
 			},
 		}
 
-		api := NewAPI(Options{Peer: peer})
+		api := newTestAPI(Options{Peer: peer})
 		api.Update(&config.Config{})
 
 		resp, err := api.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{}))
@@ -126,7 +128,7 @@ var _ = Describe("StatusService", func() {
 		}
 		DeferCleanup(peer.unblock)
 
-		api := NewAPI(Options{Peer: peer})
+		api := newTestAPI(Options{Peer: peer})
 		api.Update(&config.Config{})
 
 		statusDone := make(chan error, 1)
@@ -154,7 +156,7 @@ var _ = Describe("StatusService", func() {
 			entered: make(chan struct{}),
 			release: make(chan struct{}),
 		}
-		api := NewAPI(Options{Peer: peer, UnaryTimeout: 20 * time.Millisecond})
+		api := newTestAPI(Options{Peer: peer, UnaryConcurrency: 1, UnaryTimeout: 20 * time.Millisecond})
 		api.Update(&config.Config{})
 
 		srv := httptest.NewServer(api.Handler())
@@ -188,7 +190,7 @@ var _ = Describe("StatusService", func() {
 	// standard library's http.Protocols.
 	DescribeTable("serves status over HTTP",
 		func(wantMethod string, opts []connect.ClientOption) {
-			api := NewAPI(Options{})
+			api := newTestAPI(Options{})
 			api.Update(&config.Config{})
 
 			methods := make(chan string, 1)
@@ -228,11 +230,87 @@ var _ = Describe("StatusService", func() {
 		Entry("gRPC-Web", http.MethodPost, []connect.ClientOption{connect.WithGRPCWeb()}),
 		Entry("gRPC", http.MethodPost, []connect.ClientOption{connect.WithGRPC()}),
 	)
+
+	It("rejects oversized incoming messages before handler execution", func() {
+		peer := &blockingPeer{entered: make(chan struct{}), release: make(chan struct{})}
+		api := newTestAPI(Options{Peer: peer, ReadMaxBytes: 1, MaxRequestBodyBytes: 1024})
+		api.Update(&config.Config{})
+		srv := httptest.NewServer(api.Handler())
+		DeferCleanup(srv.Close)
+		DeferCleanup(peer.unblock)
+
+		request := &statusv3alpha.GetStatusRequest{}
+		request.ProtoReflect().SetUnknown([]byte{0x08, 0x01})
+		client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL)
+		_, err := client.GetStatus(context.Background(), connect.NewRequest(request))
+		Expect(connect.CodeOf(err)).To(Equal(connect.CodeResourceExhausted))
+		Consistently(peer.entered, 50*time.Millisecond).ShouldNot(BeClosed())
+	})
+
+	It("rejects oversized unary request bodies", func() {
+		api := newTestAPI(Options{ReadMaxBytes: 1024, MaxRequestBodyBytes: 1})
+		api.Update(&config.Config{})
+		srv := httptest.NewServer(api.Handler())
+		DeferCleanup(srv.Close)
+
+		request := &statusv3alpha.GetStatusRequest{}
+		request.ProtoReflect().SetUnknown([]byte{0x08, 0x01})
+		client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL)
+		_, err := client.GetStatus(context.Background(), connect.NewRequest(request))
+		Expect(connect.CodeOf(err)).To(Equal(connect.CodeResourceExhausted))
+	})
+
+	It("rejects oversized outgoing messages", func() {
+		api := newTestAPI(Options{SendMaxBytes: 1})
+		api.Update(&config.Config{})
+		srv := httptest.NewServer(api.Handler())
+		DeferCleanup(srv.Close)
+		client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL)
+
+		_, err := client.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{}))
+		Expect(connect.CodeOf(err)).To(Equal(connect.CodeResourceExhausted))
+	})
+
+	It("bounds slow unary uploads before handler execution", func() {
+		peer := &blockingPeer{entered: make(chan struct{}), release: make(chan struct{})}
+		api := newTestAPI(Options{Peer: peer, UnaryConcurrency: 1, UnaryTimeout: 500 * time.Millisecond})
+		api.Update(&config.Config{})
+		srv := httptest.NewServer(api.Handler())
+		DeferCleanup(srv.Close)
+		DeferCleanup(peer.unblock)
+
+		reader, writer := io.Pipe()
+		DeferCleanup(writer.Close)
+		req, err := http.NewRequest(http.MethodPost, srv.URL+statusv3alphaconnect.StatusServiceGetStatusProcedure, reader)
+		Expect(err).NotTo(HaveOccurred())
+		req.Header.Set("Content-Type", "application/proto")
+		req.ContentLength = 1
+		done := make(chan *http.Response, 1)
+		go func() {
+			resp, _ := srv.Client().Do(req)
+			done <- resp
+		}()
+		Eventually(func() int { return len(api.admission.unary) }).Should(Equal(1))
+
+		client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL)
+		_, err = client.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{}))
+		Expect(connect.CodeOf(err)).To(Equal(connect.CodeResourceExhausted))
+
+		var resp *http.Response
+		Eventually(done, time.Second).Should(Receive(&resp))
+		Expect(resp).NotTo(BeNil())
+		Expect(resp.Body.Close()).To(Succeed())
+		Expect(peer.calls.Load()).To(BeZero())
+
+		peer.unblock()
+		_, err = client.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{}))
+		Expect(err).NotTo(HaveOccurred())
+	})
 })
 
 var _ = Describe("Connect API", func() {
 	It("pins registered procedures", func() {
-		Expect(NewAPI(Options{}).Procedures()).To(Equal([]string{
+		Expect(newTestAPI(Options{}).Procedures()).To(Equal([]string{
 			"/status.v3alpha.StatusService/GetStatus",
 			"/grpc.health.v1.Health/Check",
 			"/grpc.health.v1.Health/Watch",
@@ -240,15 +318,59 @@ var _ = Describe("Connect API", func() {
 			"/grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo",
 		}))
 	})
+
+	It("allows unlimited message and request body sizes by default", func() {
+		api := newTestAPI(Options{})
+		Expect(api.readMaxBytes).To(BeZero())
+		Expect(api.sendMaxBytes).To(BeZero())
+		Expect(api.maxRequestBytes).To(BeZero())
+	})
+
+	It("returns metric registration errors", func() {
+		reg := prometheus.NewRegistry()
+		newTestAPI(Options{Registerer: reg})
+		_, err := NewAPI(Options{Registerer: reg})
+		Expect(err).To(HaveOccurred())
+	})
+
+	It("registers bounded unary lifecycle metrics", func() {
+		reg := prometheus.NewRegistry()
+		api := newTestAPI(Options{Registerer: reg})
+		api.Update(&config.Config{})
+		srv := httptest.NewServer(api.Handler())
+		DeferCleanup(srv.Close)
+		client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL)
+
+		_, err := client.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{}))
+		Expect(err).NotTo(HaveOccurred())
+		families, err := reg.Gather()
+		Expect(err).NotTo(HaveOccurred())
+
+		var labels map[string]string
+		for _, family := range families {
+			if family.GetName() != "alertmanager_api_connect_unary_request_duration_seconds" {
+				continue
+			}
+			labels = map[string]string{}
+			for _, pair := range family.GetMetric()[0].GetLabel() {
+				labels[pair.GetName()] = pair.GetValue()
+			}
+		}
+		Expect(labels).To(Equal(map[string]string{
+			"outcome":   "ok",
+			"procedure": "GetStatus",
+			"service":   statusv3alphaconnect.StatusServiceName,
+		}))
+	})
 })
 
 var _ = Describe("RPC admission", func() {
 	It("defaults unary and stream concurrency independently", func() {
-		api := NewAPI(Options{UnaryConcurrency: 1})
+		api := newTestAPI(Options{UnaryConcurrency: 1})
 		Expect(cap(api.admission.unary)).To(Equal(1))
 		Expect(cap(api.admission.streams)).To(BeNumerically(">=", 8))
 
-		api = NewAPI(Options{StreamConcurrency: 1})
+		api = newTestAPI(Options{StreamConcurrency: 1})
 		Expect(cap(api.admission.unary)).To(BeNumerically(">=", 8))
 		Expect(cap(api.admission.streams)).To(Equal(1))
 	})
@@ -323,6 +445,15 @@ var _ = Describe("RPC admission", func() {
 		Expect(wrapped(context.Background(), nil)).To(Succeed())
 	})
 
+	It("preserves successful and specifically coded results after context expiration", func() {
+		ctx, cancel := context.WithCancelCause(context.Background())
+		cancel(context.DeadlineExceeded)
+
+		Expect(normalizeContextError(ctx, nil)).To(Succeed())
+		specific := connect.NewError(connect.CodeInvalidArgument, context.Canceled)
+		Expect(normalizeContextError(ctx, specific)).To(BeIdenticalTo(specific))
+	})
+
 	It("sets configured unary deadlines", func() {
 		admission := &admissionInterceptor{
 			unary:        make(chan struct{}, 1),
diff --git a/app/app.go b/app/app.go
index 67e3ec80aa..01df329878 100644
--- a/app/app.go
+++ b/app/app.go
@@ -383,16 +383,22 @@ func (a *App) setup() error {
 	}
 
 	apih, err := api.New(api.Options{
-		Alerts:          alerts,
-		Silences:        silences,
-		GroupMutedFunc:  groupMarker.Muted,
-		Peer:            clusterPeer,
-		Timeout:         opts.HTTPTimeout,
-		Concurrency:     opts.GetConcurrency,
-		Logger:          logger.With("component", "api"),
-		Registry:        reg,
-		RequestDuration: m.requestDuration,
-		GroupFunc:       groupFn,
+		Alerts:                     alerts,
+		Silences:                   silences,
+		GroupMutedFunc:             groupMarker.Muted,
+		Peer:                       clusterPeer,
+		Timeout:                    opts.HTTPTimeout,
+		Concurrency:                opts.GetConcurrency,
+		ConnectUnaryConcurrency:    opts.ConnectUnaryConcurrency,
+		ConnectStreamConcurrency:   opts.ConnectStreamConcurrency,
+		ConnectUnaryTimeout:        opts.ConnectUnaryTimeout,
+		ConnectReadMaxBytes:        opts.ConnectReadMaxBytes,
+		ConnectSendMaxBytes:        opts.ConnectSendMaxBytes,
+		ConnectMaxRequestBodyBytes: opts.ConnectMaxRequestBodyBytes,
+		Logger:                     logger.With("component", "api"),
+		Registry:                   reg,
+		RequestDuration:            m.requestDuration,
+		GroupFunc:                  groupFn,
 	})
 	if err != nil {
 		return fmt.Errorf("failed to create API: %w", err)
diff --git a/app/options.go b/app/options.go
index aba831f8a3..d1fe90361e 100644
--- a/app/options.go
+++ b/app/options.go
@@ -64,11 +64,17 @@ type Options struct {
 	DispatchStartDelay          time.Duration
 
 	// Web server.
-	WebConfig      *web.FlagConfig
-	ExternalURL    string
-	RoutePrefix    string
-	GetConcurrency int
-	HTTPTimeout    time.Duration
+	WebConfig                  *web.FlagConfig
+	ExternalURL                string
+	RoutePrefix                string
+	GetConcurrency             int
+	HTTPTimeout                time.Duration
+	ConnectUnaryConcurrency    int
+	ConnectStreamConcurrency   int
+	ConnectUnaryTimeout        time.Duration
+	ConnectReadMaxBytes        int
+	ConnectSendMaxBytes        int
+	ConnectMaxRequestBodyBytes int64
 
 	// Cluster.
 	ClusterBindAddr        string
diff --git a/app/options_test.go b/app/options_test.go
index eb40a18127..aa255e8340 100644
--- a/app/options_test.go
+++ b/app/options_test.go
@@ -55,6 +55,9 @@ func TestOptions_Validate(t *testing.T) {
 
 	base := valid()
 	require.NoError(t, base.validate())
+	require.Zero(t, base.ConnectReadMaxBytes)
+	require.Zero(t, base.ConnectSendMaxBytes)
+	require.Zero(t, base.ConnectMaxRequestBodyBytes)
 
 	for _, tc := range []struct {
 		name   string
diff --git a/cmd/alertmanager/main.go b/cmd/alertmanager/main.go
index cd2afd6972..cd722ccec0 100644
--- a/cmd/alertmanager/main.go
+++ b/cmd/alertmanager/main.go
@@ -60,11 +60,17 @@ func run() int {
 		dispatchMaintenanceInterval = kingpin.Flag("dispatch.maintenance-interval", "Interval between maintenance of aggregation groups in the dispatcher.").Default("30s").Duration()
 		dispatchStartDelay          = kingpin.Flag("dispatch.start-delay", "Minimum amount of time to wait before dispatching alerts. This option should be synced with value of --rules.alert.resend-delay on Prometheus.").Default("0s").Duration()
 
-		webConfig      = webflag.AddFlags(kingpin.CommandLine, ":9093")
-		externalURL    = kingpin.Flag("web.external-url", "The URL under which Alertmanager is externally reachable (for example, if Alertmanager is served via a reverse proxy). Used for generating relative and absolute links back to Alertmanager itself. If the URL has a path portion, it will be used to prefix all HTTP endpoints served by Alertmanager. If omitted, relevant URL components will be derived automatically.").String()
-		routePrefix    = kingpin.Flag("web.route-prefix", "Prefix for the internal routes of web endpoints. Defaults to path of --web.external-url.").String()
-		getConcurrency = kingpin.Flag("web.get-concurrency", "Maximum number of GET requests processed concurrently. If negative or zero, the limit is GOMAXPROC or 8, whichever is larger.").Default("0").Int()
-		httpTimeout    = kingpin.Flag("web.timeout", "Timeout for HTTP requests. If negative or zero, no timeout is set.").Default("0").Duration()
+		webConfig                  = webflag.AddFlags(kingpin.CommandLine, ":9093")
+		externalURL                = kingpin.Flag("web.external-url", "The URL under which Alertmanager is externally reachable (for example, if Alertmanager is served via a reverse proxy). Used for generating relative and absolute links back to Alertmanager itself. If the URL has a path portion, it will be used to prefix all HTTP endpoints served by Alertmanager. If omitted, relevant URL components will be derived automatically.").String()
+		routePrefix                = kingpin.Flag("web.route-prefix", "Prefix for the internal routes of web endpoints. Defaults to path of --web.external-url.").String()
+		getConcurrency             = kingpin.Flag("web.get-concurrency", "Maximum number of GET requests processed concurrently. If negative or zero, the limit is GOMAXPROC or 8, whichever is larger.").Default("0").Int()
+		httpTimeout                = kingpin.Flag("web.timeout", "Timeout for HTTP requests. If negative or zero, no timeout is set.").Default("0").Duration()
+		connectUnaryConcurrency    = kingpin.Flag("api.connect.unary-concurrency", "Maximum number of Connect unary RPCs processed concurrently. Defaults to --web.get-concurrency.").Default("0").Int()
+		connectStreamConcurrency   = kingpin.Flag("api.connect.stream-concurrency", "Maximum number of Connect streams processed concurrently. Defaults to --web.get-concurrency.").Default("0").Int()
+		connectUnaryTimeout        = kingpin.Flag("api.connect.unary-timeout", "Timeout for Connect unary RPCs, including request reads. Defaults to --web.timeout.").Default("0").Duration()
+		connectReadMaxBytes        = kingpin.Flag("api.connect.read-max-bytes", "Maximum size of each incoming Connect protobuf message. If zero or negative, no limit is set.").Default("0").Int()
+		connectSendMaxBytes        = kingpin.Flag("api.connect.send-max-bytes", "Maximum size of each outgoing Connect protobuf message. If zero or negative, no limit is set.").Default("0").Int()
+		connectMaxRequestBodyBytes = kingpin.Flag("api.connect.max-request-body-bytes", "Maximum wire size of a Connect unary request body. If zero or negative, no limit is set.").Default("0").Int64()
 
 		memlimitEnable = kingpin.Flag("auto-gomemlimit", "Automatically set GOMEMLIMIT to match Linux container or system memory limit").
 				Default("false").Bool()
@@ -175,11 +181,17 @@ func run() int {
 		DispatchMaintenanceInterval: *dispatchMaintenanceInterval,
 		DispatchStartDelay:          *dispatchStartDelay,
 
-		WebConfig:      webConfig,
-		ExternalURL:    *externalURL,
-		RoutePrefix:    *routePrefix,
-		GetConcurrency: *getConcurrency,
-		HTTPTimeout:    *httpTimeout,
+		WebConfig:                  webConfig,
+		ExternalURL:                *externalURL,
+		RoutePrefix:                *routePrefix,
+		GetConcurrency:             *getConcurrency,
+		HTTPTimeout:                *httpTimeout,
+		ConnectUnaryConcurrency:    *connectUnaryConcurrency,
+		ConnectStreamConcurrency:   *connectStreamConcurrency,
+		ConnectUnaryTimeout:        *connectUnaryTimeout,
+		ConnectReadMaxBytes:        *connectReadMaxBytes,
+		ConnectSendMaxBytes:        *connectSendMaxBytes,
+		ConnectMaxRequestBodyBytes: *connectMaxRequestBodyBytes,
 
 		ClusterBindAddr:        *clusterBindAddr,
 		ClusterAdvertiseAddr:   *clusterAdvertiseAddr,