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) + }) + } +}