Distributor: return HTTP 499 instead of 500 on client push cancellation

CHANGELOG.md

@@ -78,6 +78,7 @@
 * [BUGFIX] Store Gateway: Fix misleading "no index cache backend addresses" validation error being reported for chunks-cache, metadata-cache, and parquet caches when their memcached or redis backend is configured without addresses. The message is now the cache-type-agnostic "no cache backend addresses". #7675
 * [BUGFIX] Querier/Query Frontend: Fix DNS watcher dropping all query-frontend/scheduler worker connections on a transient DNS lookup failure. #7698
 * [BUGFIX] Ring: Fix DynamoDB KV CAS not retrying on transactional conditional check failures. `TransactWriteItems` reports condition failures as `TransactionCanceledException` with a `ConditionalCheckFailed` cancellation reason, which was not recognized as retryable, so any concurrent ring update conflict (e.g. many ingesters joining during a rolling update) failed immediately instead of re-reading and retrying. `TransactionConflict` cancellation reasons are also treated as retryable. #7706
+* [BUGFIX] Distributor: Return HTTP 499 (Client Closed Request) instead of 500 when a remote-write or OTLP push is canceled by the client, so client-side cancellations are no longer counted as server-side errors. #7717
 
 ## 1.21.1 2026-06-04
 

pkg/util/push/otlp.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"compress/gzip"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -25,6 +26,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/distributor"
 	"github.com/cortexproject/cortex/pkg/util"
+	util_api "github.com/cortexproject/cortex/pkg/util/api"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
 	"github.com/cortexproject/cortex/pkg/util/users"
 	"github.com/cortexproject/cortex/pkg/util/validation"
@@ -94,14 +96,17 @@ func OTLPHandler(maxRecvMsgSize int, overrides *validation.Overrides, cfg distri
 		prwReq.Metadata = metadata
 
 		if _, err := push(ctx, &prwReq); err != nil {
+			if errors.Is(err, context.Canceled) {
+				err = httpgrpc.Errorf(util_api.StatusClientClosedRequest, "%s", err.Error())
+			}
 			resp, ok := httpgrpc.HTTPResponseFromError(err)
 			if !ok {
 				http.Error(w, err.Error(), http.StatusInternalServerError)
 				return
 			}
 			if resp.GetCode()/100 == 5 {
 				level.Error(logger).Log("msg", "push error", "err", err)
-			} else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests {
+			} else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests && resp.GetCode() != util_api.StatusClientClosedRequest {
 				level.Warn(logger).Log("msg", "push refused", "err", err)
 			}
 			http.Error(w, string(resp.Body), int(resp.Code))

pkg/util/push/otlp_test.go

@@ -1048,6 +1048,41 @@ func TestOTLPHandler_MetricCollection(t *testing.T) {
 	assert.Equal(t, 1.0, val)
 }
 
+func TestOTLPHandler_ClientCancellation(t *testing.T) {
+	cfg := distributor.OTLPConfig{
+		ConvertAllAttributes: false,
+		DisableTargetInfo:    false,
+	}
+	overrides := validation.NewOverrides(querier.DefaultLimitsConfig(), nil)
+	exportRequest := generateOTLPWriteRequest()
+
+	// A client-canceled push (bare or wrapped context.Canceled) should return 499
+	// (Client Closed Request) instead of 500, matching the remote-write handlers.
+	for _, tc := range []struct {
+		name string
+		err  error
+	}{
+		{name: "bare cancellation", err: context.Canceled},
+		{name: "wrapped cancellation", err: fmt.Errorf("send failed: %w", context.Canceled)},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			pushErr := tc.err
+			pushFunc := func(ctx context.Context, req *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error) {
+				return nil, pushErr
+			}
+
+			req, err := getOTLPHttpRequest(&exportRequest, pbContentType, "")
+			require.NoError(t, err)
+
+			handler := OTLPHandler(100000, overrides, cfg, nil, pushFunc, nil)
+			recorder := httptest.NewRecorder()
+			handler.ServeHTTP(recorder, req)
+
+			assert.Equal(t, 499, recorder.Result().StatusCode)
+		})
+	}
+}
+
 func TestOTLPWriteHandler(t *testing.T) {
 	cfg := distributor.OTLPConfig{
 		ConvertAllAttributes: false,

pkg/util/push/push.go

@@ -2,6 +2,7 @@ package push
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"strconv"
@@ -18,6 +19,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/util"
+	util_api "github.com/cortexproject/cortex/pkg/util/api"
 	"github.com/cortexproject/cortex/pkg/util/extract"
 	"github.com/cortexproject/cortex/pkg/util/log"
 	"github.com/cortexproject/cortex/pkg/util/users"
@@ -73,14 +75,17 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool,
 			}
 
 			if _, err := push(ctx, &req.WriteRequest); err != nil {
+				if errors.Is(err, context.Canceled) {
+					err = httpgrpc.Errorf(util_api.StatusClientClosedRequest, "%s", err.Error())
+				}
 				resp, ok := httpgrpc.HTTPResponseFromError(err)
 				if !ok {
 					http.Error(w, err.Error(), http.StatusInternalServerError)
 					return
 				}
 				if resp.GetCode()/100 == 5 {
 					level.Error(logger).Log("msg", "push error", "err", err)
-				} else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests {
+				} else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests && resp.GetCode() != util_api.StatusClientClosedRequest {
 					level.Warn(logger).Log("msg", "push refused", "err", err)
 				}
 				http.Error(w, string(resp.Body), int(resp.Code))
@@ -123,6 +128,9 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool,
 			}
 
 			if writeResp, err := push(ctx, &v1Req.WriteRequest); err != nil {
+				if errors.Is(err, context.Canceled) {
+					err = httpgrpc.Errorf(util_api.StatusClientClosedRequest, "%s", err.Error())
+				}
 				resp, ok := httpgrpc.HTTPResponseFromError(err)
 				if !ok {
 					setPRW2RespHeader(w, 0, 0, 0)
@@ -137,7 +145,7 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool,
 				}
 				if resp.GetCode()/100 == 5 {
 					level.Error(logger).Log("msg", "push error", "err", err)
-				} else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests {
+				} else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests && resp.GetCode() != util_api.StatusClientClosedRequest {
 					level.Warn(logger).Log("msg", "push refused", "err", err)
 				}
 				http.Error(w, string(resp.Body), int(resp.Code))

pkg/util/push/push_test.go

@@ -855,6 +855,53 @@ func TestHandler_remoteWrite(t *testing.T) {
 		assert.Equal(t, "1", respHeader[rw20WrittenExemplarsHeader][0])
 		assert.Contains(t, resp.Body.String(), "HA deduplication")
 	})
+
+	// A client-canceled push (bare or wrapped context.Canceled) should return 499
+	// (Client Closed Request) instead of 500, matching the query-frontend convention.
+	cancellationTests := []struct {
+		name string
+		err  error
+		isV2 bool
+	}{
+		{name: "remote write v1 client cancellation", err: context.Canceled, isV2: false},
+		{name: "remote write v2 client cancellation", err: context.Canceled, isV2: true},
+		{name: "remote write v1 wrapped client cancellation", err: fmt.Errorf("send failed: %w", context.Canceled), isV2: false},
+		{name: "remote write v2 wrapped client cancellation", err: fmt.Errorf("send failed: %w", context.Canceled), isV2: true},
+	}
+
+	for _, test := range cancellationTests {
+		t.Run(test.name, func(t *testing.T) {
+			pushErr := test.err
+			pushFunc := func(ctx context.Context, req *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error) {
+				return nil, pushErr
+			}
+
+			ctx := context.Background()
+			ctx = user.InjectOrgID(ctx, "user-1")
+			handler := Handler(true, false, 100000, overrides, nil, pushFunc, nil)
+
+			var body []byte
+			if test.isV2 {
+				body = createPrometheusRemoteWriteV2Protobuf(t)
+			} else {
+				body = createPrometheusRemoteWriteProtobuf(t)
+			}
+			req := createRequest(t, body, test.isV2)
+			req = req.WithContext(ctx)
+			resp := httptest.NewRecorder()
+			handler.ServeHTTP(resp, req)
+
+			assert.Equal(t, 499, resp.Code)
+
+			if test.isV2 {
+				// Written-stats headers should still be set (to zero) on the error path.
+				respHeader := resp.Header()
+				assert.Equal(t, "0", respHeader[rw20WrittenSamplesHeader][0])
+				assert.Equal(t, "0", respHeader[rw20WrittenHistogramsHeader][0])
+				assert.Equal(t, "0", respHeader[rw20WrittenExemplarsHeader][0])
+			}
+		})
+	}
 }
 
 func TestHandler_ContentTypeAndEncoding(t *testing.T) {