From ca92e9fd7639e1e7485f2590acb785030c571cdc Mon Sep 17 00:00:00 2001
From: tlhsec <talhaouy.web3@gmail.com>
Date: Wed, 5 Aug 2026 12:49:30 +0100
Subject: [PATCH] fix: pdcp data loss, pipeline conn leak, Close() race,
 FilterCustom error swallowing

---
 common/httpx/filter.go      |  7 +++--
 common/httpx/filter_test.go | 55 +++++++++++++++++++++++++++++++++++++
 common/httpx/pipeline.go    |  1 +
 internal/pdcp/writer.go     | 11 +++++---
 4 files changed, 68 insertions(+), 6 deletions(-)
 create mode 100644 common/httpx/filter_test.go

diff --git a/common/httpx/filter.go b/common/httpx/filter.go
index e553abcbb..44ac72ef0 100644
--- a/common/httpx/filter.go
+++ b/common/httpx/filter.go
@@ -53,8 +53,11 @@ type FilterCustom struct {
 func (f FilterCustom) Filter(response *Response) (bool, error) {
 	for _, callback := range f.CallBacks {
 		ok, err := callback(response)
-		if ok && err == nil {
-			return true, err
+		if err != nil {
+			return false, err
+		}
+		if ok {
+			return true, nil
 		}
 	}
 
diff --git a/common/httpx/filter_test.go b/common/httpx/filter_test.go
new file mode 100644
index 000000000..9a163183a
--- /dev/null
+++ b/common/httpx/filter_test.go
@@ -0,0 +1,55 @@
+package httpx
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestFilterCustomErrorPropagation(t *testing.T) {
+	t.Run("error from callback is returned, not swallowed", func(t *testing.T) {
+		expectedErr := errors.New("callback failure")
+		callback := func(response *Response) (bool, error) {
+			return true, expectedErr
+		}
+		filter := FilterCustom{CallBacks: []CustomCallback{callback}}
+		ok, err := filter.Filter(&Response{})
+		require.False(t, ok, "ok should be false when callback returns an error")
+		require.ErrorIs(t, err, expectedErr, "error from callback should be propagated")
+	})
+
+	t.Run("error from callback with ok=false is returned", func(t *testing.T) {
+		expectedErr := errors.New("callback failure")
+		callback := func(response *Response) (bool, error) {
+			return false, expectedErr
+		}
+		filter := FilterCustom{CallBacks: []CustomCallback{callback}}
+		ok, err := filter.Filter(&Response{})
+		require.False(t, ok)
+		require.ErrorIs(t, err, expectedErr)
+	})
+
+	t.Run("first matching callback without error returns true", func(t *testing.T) {
+		callbacks := []CustomCallback{
+			func(response *Response) (bool, error) { return false, nil },
+			func(response *Response) (bool, error) { return true, nil },
+			func(response *Response) (bool, error) { return true, nil },
+		}
+		filter := FilterCustom{CallBacks: callbacks}
+		ok, err := filter.Filter(&Response{})
+		require.True(t, ok)
+		require.NoError(t, err)
+	})
+
+	t.Run("no callbacks match returns false with nil error", func(t *testing.T) {
+		callbacks := []CustomCallback{
+			func(response *Response) (bool, error) { return false, nil },
+			func(response *Response) (bool, error) { return false, nil },
+		}
+		filter := FilterCustom{CallBacks: callbacks}
+		ok, err := filter.Filter(&Response{})
+		require.False(t, ok)
+		require.NoError(t, err)
+	})
+}
diff --git a/common/httpx/pipeline.go b/common/httpx/pipeline.go
index b6b7b7817..d804833f5 100644
--- a/common/httpx/pipeline.go
+++ b/common/httpx/pipeline.go
@@ -29,6 +29,7 @@ func (h *HTTPX) SupportPipeline(protocol, method, host string, port int) bool {
 	if err != nil {
 		return false
 	}
+	defer conn.Close()
 	// send some probes
 	nprobes := 10
 	for i := 0; i < nprobes; i++ {
diff --git a/internal/pdcp/writer.go b/internal/pdcp/writer.go
index fbad91abd..06ee03999 100644
--- a/internal/pdcp/writer.go
+++ b/internal/pdcp/writer.go
@@ -174,6 +174,9 @@ func (u *UploadWriter) autoCommit(ctx context.Context) {
 				if err := u.uploadChunk(buff); err != nil {
 					gologger.Error().Msgf("Failed to upload asset results on cloud: %v", err)
 				}
+				// write the current line to the now-empty buffer so it is not lost
+				buff.WriteString(line)
+				buff.WriteString("\n")
 			} else {
 				buff.WriteString(line)
 				buff.WriteString("\n")
@@ -261,10 +264,10 @@ func (u *UploadWriter) getRequest(bin []byte) (*retryablehttp.Request, error) {
 
 // Close closes the upload writer
 func (u *UploadWriter) Close() {
-	if !u.closed.Load() {
-		// protect to avoid channel closed twice error
-		close(u.data)
-		u.closed.Store(true)
+	// atomically ensure we only close the channel once
+	if !u.closed.CompareAndSwap(false, true) {
+		return
 	}
+	close(u.data)
 	<-u.done
 }