fix: return context cancel as cause of timeout

pkg/streamer/error.go

@@ -6,12 +6,20 @@ import (
 
 type ReadTimeoutException struct {
 	LastRead []byte
+	Cause    error
 }
 
 func (m *ReadTimeoutException) Error() string {
+	if m.Cause != nil {
+		return fmt.Sprintf("read timeout error caused by %v. last seen: %q", m.Cause, string(m.LastRead))
+	}
 	return fmt.Sprintf("read timeout error. last seen: %q", string(m.LastRead))
 }
 
+func (m *ReadTimeoutException) Unwrap() error {
+	return m.Cause
+}
+
 func (m *ReadTimeoutException) Is(target error) bool {
 	if _, ok := target.(*ReadTimeoutException); ok {
 		return true
@@ -34,8 +42,12 @@ func (m *EOFException) Is(target error) bool {
 	return false
 }
 
-func ThrowReadTimeoutException(lastRead []byte) error {
-	return &ReadTimeoutException{LastRead: lastRead}
+func ThrowReadTimeoutException(lastRead []byte, cause ...error) error {
+	var err error
+	if len(cause) > 0 {
+		err = cause[0]
+	}
+	return &ReadTimeoutException{LastRead: lastRead, Cause: err}
 }
 
 func ThrowEOFException(lastRead []byte) error {

pkg/streamer/streamer.go

@@ -16,7 +16,6 @@ import (
 	"github.com/annetutil/gnetcli/pkg/credentials"
 	"github.com/annetutil/gnetcli/pkg/expr"
 	"github.com/annetutil/gnetcli/pkg/trace"
-	"go.uber.org/multierr"
 	"go.uber.org/zap"
 )
 
@@ -268,7 +267,7 @@ func GenericReadX(ctx context.Context, inBuffer []byte, readCh chan []byte, read
 		case <-ctx.Done():
 			StopTimer(maxDurationTimeout)
 			buffer = append(buffer, flushCh(readCh)...)
-			return nil, buffer, buffer[len(inBuffer):], multierr.Combine(ctx.Err(), ThrowReadTimeoutException(GetLastBytes(buffer, readSize)))
+			return nil, buffer, buffer[len(inBuffer):], ThrowReadTimeoutException(GetLastBytes(buffer, readSize), ctx.Err())
 		default:
 		}
 		readIterTimeout := NewTimerWithDefault(readTimeout)
@@ -300,7 +299,7 @@ func GenericReadX(ctx context.Context, inBuffer []byte, readCh chan []byte, read
 			StopTimer(readIterTimeout)
 			StopTimer(maxDurationTimeout)
 			buffer = append(buffer, flushCh(readCh)...)
-			return nil, buffer, buffer[len(inBuffer):], multierr.Combine(ctx.Err(), ThrowReadTimeoutException(GetLastBytes(buffer, readSize)))
+			return nil, buffer, buffer[len(inBuffer):], ThrowReadTimeoutException(GetLastBytes(buffer, readSize), ctx.Err())
 		case readData, ok := <-readCh:
 			StopTimer(readIterTimeout)
 			if ok {

pkg/streamer/streamer_test.go

@@ -96,9 +96,12 @@ func TestGenericReadXCtxDoneFlushChannel(t *testing.T) {
 	_, _, _, err := GenericReadX(ctx, nil, ch, 4096, time.Second, pat, 0, 0)
 	require.Error(t, err)
 
+	require.ErrorIs(t, err, context.Canceled)
 	var rt *ReadTimeoutException
 	require.True(t, errors.As(err, &rt))
 	require.Equal(t, "pending in channel", string(rt.LastRead))
+	require.ErrorIs(t, rt.Cause, context.Canceled)
+	require.Equal(t, `read timeout error caused by context canceled. last seen: "pending in channel"`, err.Error())
 	require.Empty(t, ch, "channel should be drained")
 }