uploader: propagate errors in compress (#137)

Co-authored-by: Tadeusz Dudkiewicz <[email protected]>

uploader/base.go

@@ -213,20 +213,33 @@ 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()
+		_, err := io.Copy(gw, data)
+		if err != nil {
+			pw.CloseWithError(err)
+			data.CloseWithError(err)
+			return
+		}
+
+		err = gw.Close()
+		if err != nil {
+			pw.CloseWithError(err)
+			data.CloseWithError(err)
+			return
+		}
+
 		pw.Close()
+		data.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