fix(mediorum): avoid blocked replication pipe goroutines

pkg/mediorum/server/replicate.go

@@ -177,23 +177,33 @@ func (ss *MediorumServer) replicateFileToHost(ctx context.Context, peer string,
 
 	r, w := io.Pipe()
 	m := multipart.NewWriter(w)
-	errChan := make(chan error)
+	errChan := make(chan error, 1)
 
 	go func() {
-		defer w.Close()
-		defer m.Close()
 		part, err := m.CreateFormFile(filesFormFieldName, fileName)
 		if err != nil {
+			_ = w.CloseWithError(err)
 			errChan <- err
 			return
 		}
 		if _, err = io.Copy(part, file); err != nil {
+			_ = w.CloseWithError(err)
 			errChan <- err
 			return
 		}
-		close(errChan)
+		if err := m.Close(); err != nil {
+			_ = w.CloseWithError(err)
+			errChan <- err
+			return
+		}
+		errChan <- w.Close()
 	}()
 
+	closeBodyWithError := func(err error) error {
+		_ = r.CloseWithError(err)
+		return err
+	}
+
 	req, err := signature.SignedPost(
 		ctx,
 		peer+"/internal/blobs",
@@ -203,7 +213,7 @@ func (ss *MediorumServer) replicateFileToHost(ctx context.Context, peer string,
 		ss.Config.Self.Host,
 	)
 	if err != nil {
-		return err
+		return closeBodyWithError(err)
 	}
 	if len(placementHosts) > 0 {
 		req.Header.Set(placementHostsHeader, encodePlacementHosts(placementHosts))
@@ -212,12 +222,12 @@ func (ss *MediorumServer) replicateFileToHost(ctx context.Context, peer string,
 	// send it
 	resp, err := ss.peerHTTPClient.Do(req)
 	if err != nil {
-		return err
+		return closeBodyWithError(err)
 	}
 
 	defer resp.Body.Close()
 	if resp.StatusCode != 200 {
-		return errors.New(resp.Status)
+		return closeBodyWithError(errors.New(resp.Status))
 	}
 
 	return <-errChan

pkg/mediorum/server/replicate_test.go

@@ -0,0 +1,66 @@
+package server
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+	"runtime"
+	"runtime/pprof"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type roundTripFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}
+
+func countReplicateFileToHostGoroutines() int {
+	var stacks bytes.Buffer
+	_ = pprof.Lookup("goroutine").WriteTo(&stacks, 2)
+	return strings.Count(stacks.String(), "replicateFileToHost.func1")
+}
+
+func TestReplicateFileToHostClosesPipeOnEarlyHTTPError(t *testing.T) {
+	ss := testNetwork[0]
+	originalClient := ss.peerHTTPClient
+	ss.peerHTTPClient = &http.Client{
+		Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+			return &http.Response{
+				StatusCode: http.StatusInternalServerError,
+				Status:     "500 Internal Server Error",
+				Header:     make(http.Header),
+				Body:       io.NopCloser(strings.NewReader("")),
+				Request:    req,
+			}, nil
+		}),
+	}
+	t.Cleanup(func() {
+		ss.peerHTTPClient = originalClient
+	})
+
+	before := countReplicateFileToHostGoroutines()
+	err := ss.replicateFileToHost(
+		context.Background(),
+		"http://unread-peer.test",
+		"leak-regression-cid",
+		strings.NewReader(strings.Repeat("x", 1024*1024)),
+		nil,
+	)
+	assert.Error(t, err)
+
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		if countReplicateFileToHostGoroutines() <= before {
+			return
+		}
+		runtime.Gosched()
+		time.Sleep(10 * time.Millisecond)
+	}
+	assert.LessOrEqual(t, countReplicateFileToHostGoroutines(), before)
+}