From 20079e6894f54ce55bd33e13d25ca9a5d2a1da2b Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Wed, 18 Mar 2026 22:19:22 +0100 Subject: [PATCH] fix(cas): handle client disconnect errors gracefully in download service Client disconnects (connection reset by peer, broken pipe) during HTTP downloads were being logged as errors and reported to Sentry. These are normal operational events that should be logged at info level instead. Refs: PFM-5015 Signed-off-by: Miguel Martinez Trivino --- .../internal/service/bytestream.go | 4 +- app/artifact-cas/internal/service/download.go | 11 ++- app/artifact-cas/internal/service/service.go | 32 ++++++- .../internal/service/service_test.go | 95 ++++++++++++++++++- 4 files changed, 134 insertions(+), 8 deletions(-) diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go index 748fe898a..ddf737612 100644 --- a/app/artifact-cas/internal/service/bytestream.go +++ b/app/artifact-cas/internal/service/bytestream.go @@ -114,7 +114,7 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro // Now it's time to check if the data provider has sent an error if err != nil { - if errors.Is(err, context.Canceled) || status.Code(err) == codes.Canceled { + if isClientDisconnect(err) { s.log.Infow("msg", "upload canceled", "digest", req.resource.Digest, "name", req.resource.FileName) return nil } @@ -164,7 +164,7 @@ func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream. // streamwriter will stream chunks of data to the client sw := &streamWriter{stream, s.log, req.ResourceName, sha256.New()} if err := backend.Download(ctx, sw, req.ResourceName); err != nil { - if errors.Is(err, context.Canceled) { + if isClientDisconnect(err) { s.log.Infow("msg", "download canceled", "digest", req.ResourceName) return nil } diff --git a/app/artifact-cas/internal/service/download.go b/app/artifact-cas/internal/service/download.go index fc6f49a47..1d1f67c83 100644 --- a/app/artifact-cas/internal/service/download.go +++ b/app/artifact-cas/internal/service/download.go @@ -1,5 +1,5 @@ // -// Copyright 2023 The Chainloop Authors. +// Copyright 2023-2026 The Chainloop Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -17,9 +17,7 @@ package service import ( "bytes" - "context" "crypto/sha256" - "errors" "fmt" "io" "net/http" @@ -112,7 +110,7 @@ func (s *DownloadService) ServeHTTP(w http.ResponseWriter, r *http.Request) { // and don't require client-side verification mw := io.MultiWriter(buf, gotChecksum) if err := b.Download(ctx, mw, wantChecksum.Hex); err != nil { - if errors.Is(err, context.Canceled) { + if isClientDisconnect(err) { s.log.Infow("msg", "download canceled", "digest", wantChecksum) return } @@ -130,6 +128,11 @@ func (s *DownloadService) ServeHTTP(w http.ResponseWriter, r *http.Request) { } if _, err := io.Copy(w, buf); err != nil { + if isClientDisconnect(err) { + s.log.Infow("msg", "download canceled during response write", "digest", wantChecksum) + return + } + http.Error(w, sl.LogAndMaskErr(err, s.log).Error(), http.StatusInternalServerError) return } diff --git a/app/artifact-cas/internal/service/service.go b/app/artifact-cas/internal/service/service.go index f8676a836..ed25510aa 100644 --- a/app/artifact-cas/internal/service/service.go +++ b/app/artifact-cas/internal/service/service.go @@ -1,5 +1,5 @@ // -// Copyright 2023 The Chainloop Authors. +// Copyright 2023-2026 The Chainloop Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -17,7 +17,9 @@ package service import ( "context" + "errors" "fmt" + "syscall" casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" @@ -26,6 +28,8 @@ import ( "github.com/go-kratos/kratos/v2/log" "github.com/go-kratos/kratos/v2/middleware/auth/jwt" "github.com/google/wire" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // ProviderSet is service providers. @@ -75,6 +79,32 @@ func newCommonService(backends backend.Providers, opts ...NewOpt) *commonService return s } +// isClientDisconnect returns true if the error indicates the client has disconnected. +// This includes context cancellation, gRPC canceled status, and network-level +// errors such as "connection reset by peer" and "broken pipe". +func isClientDisconnect(err error) bool { + if err == nil { + return false + } + + // Context cancellation (e.g. client canceled the request) + if errors.Is(err, context.Canceled) { + return true + } + + // gRPC canceled status (client disconnect in gRPC streaming) + if status.Code(err) == codes.Canceled { + return true + } + + // Network-level disconnects: connection reset by peer, broken pipe + if errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.EPIPE) { + return true + } + + return false +} + // Extract the JWT claims from the context, note that the JWT verification has happened in the middleware func infoFromAuth(ctx context.Context) (*casJWT.Claims, error) { rawClaims, ok := jwt.FromContext(ctx) diff --git a/app/artifact-cas/internal/service/service_test.go b/app/artifact-cas/internal/service/service_test.go index a0a756220..149479355 100644 --- a/app/artifact-cas/internal/service/service_test.go +++ b/app/artifact-cas/internal/service/service_test.go @@ -1,5 +1,5 @@ // -// Copyright 2023 The Chainloop Authors. +// Copyright 2023-2026 The Chainloop Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -18,6 +18,10 @@ package service import ( "context" "errors" + "fmt" + "net" + "os" + "syscall" "testing" casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" @@ -28,6 +32,8 @@ import ( "github.com/golang-jwt/jwt/v4" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) func TestInfoFromAuth(t *testing.T) { @@ -155,3 +161,90 @@ func TestLoadBackend(t *testing.T) { }) } } + +func TestIsClientDisconnect(t *testing.T) { + testCases := []struct { + name string + err error + want bool + }{ + { + name: "nil error", + err: nil, + want: false, + }, + { + name: "context canceled", + err: context.Canceled, + want: true, + }, + { + name: "wrapped context canceled", + err: fmt.Errorf("download failed: %w", context.Canceled), + want: true, + }, + { + name: "grpc canceled status", + err: status.Error(codes.Canceled, "canceled"), + want: true, + }, + { + name: "connection reset by peer (syscall)", + err: &net.OpError{ + Op: "write", + Net: "tcp", + Err: &os.SyscallError{ + Syscall: "write", + Err: syscall.ECONNRESET, + }, + }, + want: true, + }, + { + name: "broken pipe (syscall)", + err: &net.OpError{ + Op: "write", + Net: "tcp", + Err: &os.SyscallError{ + Syscall: "write", + Err: syscall.EPIPE, + }, + }, + want: true, + }, + { + name: "wrapped connection reset", + err: fmt.Errorf("copying data: %w", &net.OpError{ + Op: "write", + Net: "tcp", + Err: &os.SyscallError{ + Syscall: "write", + Err: syscall.ECONNRESET, + }, + }), + want: true, + }, + { + name: "generic error", + err: errors.New("something went wrong"), + want: false, + }, + { + name: "grpc internal error", + err: status.Error(codes.Internal, "internal"), + want: false, + }, + { + name: "grpc unavailable", + err: status.Error(codes.Unavailable, "unavailable"), + want: false, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + got := isClientDisconnect(tc.err) + assert.Equal(t, tc.want, got) + }) + } +}