From 5157ed1576e88d714c26e06552f223fe8a4ae193 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Wed, 21 Oct 2026 17:56:45 +0200 Subject: [PATCH] fix(cas): handle disconnected HTTP and gRPC downloads Signed-off-by: Miguel Martinez Trivino --- .../internal/service/bytestream.go | 9 ++- .../service/bytestream_download_test.go | 68 ++++++++++++++++- app/artifact-cas/internal/service/download.go | 5 +- .../internal/service/download_test.go | 75 +++++++++++++++++++ 4 files changed, 154 insertions(+), 3 deletions(-) diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go index 38d6ee77a..75364903f 100644 --- a/app/artifact-cas/internal/service/bytestream.go +++ b/app/artifact-cas/internal/service/bytestream.go @@ -437,7 +437,14 @@ type sendWriter struct { } func (sw sendWriter) Write(data []byte) (int, error) { - return len(data), sw.stream.Send(&bytestream.ReadResponse{Data: data}) + err := sw.stream.Send(&bytestream.ReadResponse{Data: data}) + // Unavailable from Send means the connection to the download client is + // gone. The transport can close before it cancels the stream context. + // Classify it here so backend Unavailable errors remain server failures. + if status.Code(err) == codes.Unavailable { + return 0, errors.Join(context.Canceled, err) + } + return len(data), err } // pacedWriter reports each write's cumulative bytes to a pacing hook. When the diff --git a/app/artifact-cas/internal/service/bytestream_download_test.go b/app/artifact-cas/internal/service/bytestream_download_test.go index 3d15054a7..4f7bd6678 100644 --- a/app/artifact-cas/internal/service/bytestream_download_test.go +++ b/app/artifact-cas/internal/service/bytestream_download_test.go @@ -23,11 +23,13 @@ import ( "os" "testing" + "github.com/getsentry/sentry-go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "google.golang.org/genproto/googleapis/bytestream" "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // These tests lock down the DOWNLOAD digest-verification behavior — the server @@ -39,10 +41,14 @@ import ( // data chunks the sendWriter sends. Only Send is exercised by sendWriter. type fakeReadServer struct { bytestream.ByteStream_ReadServer - sent [][]byte + sent [][]byte + sendErr error } func (f *fakeReadServer) Send(r *bytestream.ReadResponse) error { + if f.sendErr != nil { + return f.sendErr + } // copy: the underlying buffer may be reused by the caller cp := make([]byte, len(r.Data)) copy(cp, r.Data) @@ -50,6 +56,66 @@ func (f *fakeReadServer) Send(r *bytestream.ReadResponse) error { return nil } +func TestByteStreamDownloadTransportClosing(t *testing.T) { + tests := []struct { + name string + err error + fromSend bool + wantDisconnect bool + }{ + { + name: "client transport closes before request cancellation", + err: status.Error(codes.Unavailable, "transport is closing"), + fromSend: true, + wantDisconnect: true, + }, + { + name: "unavailable client transport", + err: status.Error(codes.Unavailable, "connection error"), + fromSend: true, + wantDisconnect: true, + }, + { + name: "internal send error is still reported", + err: status.Error(codes.Internal, "send failed"), + fromSend: true, + }, + { + name: "backend transport closing is still reported", + err: status.Error(codes.Unavailable, "transport is closing"), + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var reported int + client, err := sentry.NewClient(sentry.ClientOptions{ + BeforeSend: func(_ *sentry.Event, _ *sentry.EventHint) *sentry.Event { + reported++ + return nil + }, + }) + require.NoError(t, err) + ctx := sentry.SetHubOnContext(context.Background(), sentry.NewHub(client, sentry.NewScope())) + downloadErr := tc.err + if tc.fromSend { + sw := sendWriter{stream: &fakeReadServer{sendErr: tc.err}} + _, downloadErr = sw.Write([]byte(downloadContent)) + require.ErrorIs(t, downloadErr, tc.err) + } + require.NoError(t, ctx.Err(), "transport can close before the request context is canceled") + svc := NewByteStreamService(nil) + err = svc.downloadError(ctx, downloadErr, downloadDigestHex) + if tc.wantDisconnect { + require.NoError(t, err) + assert.Zero(t, reported, "client disconnects must not reach Sentry") + } else { + require.Error(t, err) + assert.Equal(t, 1, reported) + } + }) + } +} + // TestSendWriter_ForwardsChunksInOrder verifies the download adapter turns // every Write into one ReadResponse carrying exactly those bytes, in order. // Hashing is not its job: content is verified on disk before the first Send. diff --git a/app/artifact-cas/internal/service/download.go b/app/artifact-cas/internal/service/download.go index 5afa71434..9b3e67f69 100644 --- a/app/artifact-cas/internal/service/download.go +++ b/app/artifact-cas/internal/service/download.go @@ -177,7 +177,10 @@ func (s *DownloadService) writeDownloadError(ctx context.Context, w http.Respons http.Error(w, err.Error(), http.StatusInternalServerError) return } - if isClientDisconnect(err) { + // HTTP/2 writes can return an unexported transport error such as + // "client disconnected" instead of context.Canceled. The request context + // still records the cancellation, so use it rather than matching messages. + if errors.Is(ctx.Err(), context.Canceled) || isClientDisconnect(err) { s.log.Infow("msg", "download canceled", "digest", digest) return } diff --git a/app/artifact-cas/internal/service/download_test.go b/app/artifact-cas/internal/service/download_test.go index a51e0fde6..5fa8def1f 100644 --- a/app/artifact-cas/internal/service/download_test.go +++ b/app/artifact-cas/internal/service/download_test.go @@ -16,6 +16,8 @@ package service import ( + "context" + "errors" "io" "net/http" "net/http/httptest" @@ -27,6 +29,7 @@ import ( casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" "github.com/chainloop-dev/chainloop/pkg/blobmanager/mocks" + "github.com/getsentry/sentry-go" "github.com/go-kratos/kratos/v2/log" jwtMiddleware "github.com/go-kratos/kratos/v2/middleware/auth/jwt" "github.com/gorilla/mux" @@ -35,6 +38,78 @@ import ( "github.com/stretchr/testify/require" ) +func TestWriteDownloadErrorCancellation(t *testing.T) { + tests := []struct { + name string + err error + canceled bool + wantError bool + }{ + { + name: "HTTP2 connection disconnect with canceled request", + err: errors.New("client disconnected"), + canceled: true, + }, + { + name: "HTTP2 stream close with canceled request", + err: errors.New("http2: stream closed"), + canceled: true, + }, + { + name: "context cancellation error", + err: context.Canceled, + }, + { + name: "disconnect message alone does not suppress errors", + err: errors.New("client disconnected"), + wantError: true, + }, + { + name: "backend error on active request", + err: errors.New("backend unavailable"), + wantError: true, + }, + { + name: "deadline exceeded is still reported", + err: context.DeadlineExceeded, + wantError: true, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var reported int + client, err := sentry.NewClient(sentry.ClientOptions{ + BeforeSend: func(_ *sentry.Event, _ *sentry.EventHint) *sentry.Event { + reported++ + return nil + }, + }) + require.NoError(t, err) + ctx := sentry.SetHubOnContext(context.Background(), sentry.NewHub(client, sentry.NewScope())) + ctx, cancel := context.WithCancel(ctx) + defer cancel() + if tc.canceled { + cancel() + } + + svc := NewDownloadService(nil) + w := httptest.NewRecorder() + svc.writeDownloadError(ctx, w, tc.err, downloadDigestHex) + + if tc.wantError { + assert.Equal(t, 1, reported) + assert.Equal(t, http.StatusInternalServerError, w.Code) + assert.Contains(t, w.Body.String(), "server error") + } else { + assert.Zero(t, reported, "canceled downloads must not reach Sentry") + assert.False(t, w.Flushed) + assert.Empty(t, w.Header(), "do not attempt an error response after disconnect") + assert.Empty(t, w.Body.String()) + } + }) + } +} + const ( downloadBackendType = "backend-type" downloadContent = "hello world"