From 5a12163f0fd73c61dd0522dbd8530fa285da8215 Mon Sep 17 00:00:00 2001
From: Tadeusz Dudkiewicz <tadeusz.dudkiewicz@rtbhouse.com>
Date: Thu, 25 Jan 2024 14:03:23 +0100
Subject: [PATCH] uploader: propagate errors in compress

---
 uploader/base.go | 16 +++++++++++-----
 1 file changed, 11 insertions(+), 5 deletions(-)

diff --git a/uploader/base.go b/uploader/base.go
index b0b3ff76..a66608ac 100644
--- a/uploader/base.go
+++ b/uploader/base.go
@@ -213,20 +213,26 @@ func (u *Base) uploadWorker(ctx context.Context) {
 	}
 }
 
-func compress(data io.Reader) io.Reader {
+func compress(data *io.PipeReader) io.Reader {
 	pr, pw := io.Pipe()
 	gw := gzip.NewWriter(pw)
 
 	go func() {
-		_, _ = io.Copy(gw, data)
-		gw.Close()
-		pw.Close()
+		_, err := io.Copy(gw, data)
+		if err != nil {
+			data.CloseWithError(err)
+			pw.CloseWithError(err)
+		} else {
+			data.Close()
+			gw.Close()
+			pw.Close()
+		}
 	}()
 
 	return pr
 }
 
-func (u *Base) insertRowBinary(table string, data io.Reader) error {
+func (u *Base) insertRowBinary(table string, data *io.PipeReader) error {
 	p, err := url.Parse(u.config.URL)
 	if err != nil {
 		return err