Close the MinIO object body after download

internal/bucket/minio/minio.go

@@ -18,17 +18,23 @@ package minio
 
 import (
 	"context"
+	"crypto/sha256"
 	"crypto/tls"
+	"encoding/hex"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
+	"os"
+	"path/filepath"
 	"strings"
 
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
 	"github.com/minio/minio-go/v7/pkg/s3utils"
 	corev1 "k8s.io/api/core/v1"
+	ctrl "sigs.k8s.io/controller-runtime"
 
 	"github.com/fluxcd/pkg/auth"
 	awsauth "github.com/fluxcd/pkg/auth/aws"
@@ -330,12 +336,77 @@ func (c *MinioClient) FGetObject(ctx context.Context, bucketName, objectName, lo
 	if err = opts.SetMatchETag(stat.ETag); err != nil {
 		return "", err
 	}
-	if err = c.Client.FGetObject(ctx, bucketName, objectName, localPath, opts); err != nil {
+
+	st, err := os.Stat(localPath)
+	if err == nil {
+		if st.IsDir() {
+			return "", errors.New("fileName is a directory.")
+		}
+	} else if !os.IsNotExist(err) {
+		return "", err
+	}
+	objectDir, _ := filepath.Split(localPath)
+	if objectDir != "" {
+		if err = os.MkdirAll(objectDir, 0o700); err != nil {
+			return "", err
+		}
+	}
+
+	// GetObject, not Client.FGetObject: minio-go v7.2.1 leaves the GET body open.
+	object, err := c.Client.GetObject(ctx, bucketName, objectName, opts)
+	if err != nil {
+		return "", err
+	}
+	defer func() {
+		if cerr := object.Close(); cerr != nil {
+			ctrl.LoggerFrom(ctx).Error(cerr, "failed to close object reader")
+		}
+	}()
+
+	partPath := partFilePath(localPath, stat.ETag)
+	f, err := os.OpenFile(partPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
+	if err != nil {
+		return "", err
+	}
+	discardPart := func(copyErr error) (string, error) {
+		if cerr := f.Close(); cerr != nil {
+			ctrl.LoggerFrom(ctx).Error(cerr, "failed to close file")
+		}
+		if rerr := os.Remove(partPath); rerr != nil && !os.IsNotExist(rerr) {
+			ctrl.LoggerFrom(ctx).Error(rerr, "failed to remove partial download")
+		}
+		return "", copyErr
+	}
+
+	n, err := io.Copy(f, object)
+	if err != nil {
+		return discardPart(err)
+	}
+	if n != stat.Size {
+		return discardPart(fmt.Errorf("downloaded %d bytes from object %q, stat size is %d", n, objectName, stat.Size))
+	}
+	if err = f.Close(); err != nil {
+		if rerr := os.Remove(partPath); rerr != nil && !os.IsNotExist(rerr) {
+			ctrl.LoggerFrom(ctx).Error(rerr, "failed to remove partial download")
+		}
+		return "", err
+	}
+	if err = os.Rename(partPath, localPath); err != nil {
+		if rerr := os.Remove(partPath); rerr != nil && !os.IsNotExist(rerr) {
+			ctrl.LoggerFrom(ctx).Error(rerr, "failed to remove partial download")
+		}
 		return "", err
 	}
 	return stat.ETag, nil
 }
 
+// Hashed like minio-go so "<base>.part.minio" can be a different object key.
+func partFilePath(localPath, etag string) string {
+	sum := sha256.Sum256([]byte(filepath.Base(localPath) + etag))
+	name := hex.EncodeToString(sum[:]) + ".part.minio"
+	return filepath.Join(filepath.Dir(localPath), name)
+}
+
 // VisitObjects iterates over the items in the provided object storage
 // bucket, calling visit for every item.
 // If the underlying client or the visit callback returns an error,

internal/bucket/minio/minio_test.go

@@ -17,24 +17,30 @@ limitations under the License.
 package minio
 
 import (
+	"bytes"
 	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/xml"
 	"errors"
 	"fmt"
+	"io"
 	"log"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"os"
 	"path/filepath"
 	"regexp"
+	"strconv"
+	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/google/uuid"
+	miniogo "github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
@@ -313,6 +319,122 @@ func TestFGetObject(t *testing.T) {
 	g.Expect(err).NotTo(HaveOccurred())
 }
 
+type closeTrackTransport struct {
+	closed *atomic.Bool
+}
+
+func (t *closeTrackTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	resp, err := http.DefaultTransport.RoundTrip(req)
+	if err != nil || resp == nil || resp.Body == nil || req.Method != http.MethodGet {
+		return resp, err
+	}
+	resp.Body = &closeTrackBody{ReadCloser: resp.Body, closed: t.closed}
+	return resp, nil
+}
+
+type closeTrackBody struct {
+	io.ReadCloser
+	closed *atomic.Bool
+}
+
+func (b *closeTrackBody) Close() error {
+	b.closed.Store(true)
+	return b.ReadCloser.Close()
+}
+
+func newTrackingMinioClient(t *testing.T, handler http.Handler, closed *atomic.Bool) *MinioClient {
+	t.Helper()
+	srv := httptest.NewServer(handler)
+	t.Cleanup(srv.Close)
+
+	mc, err := miniogo.New(strings.TrimPrefix(srv.URL, "http://"), &miniogo.Options{
+		Creds:     credentials.NewStaticV4("fluxcd", "passw0rd!", ""),
+		Secure:    false,
+		Region:    "us-east-1",
+		Transport: &closeTrackTransport{closed: closed},
+	})
+	NewWithT(t).Expect(err).NotTo(HaveOccurred())
+	return &MinioClient{Client: mc}
+}
+
+func s3ObjectHandler(payload []byte) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT")
+		w.Header().Set("ETag", `"abc123"`)
+		w.Header().Set("Content-Length", strconv.Itoa(len(payload)))
+		if r.Method == http.MethodGet {
+			_, _ = w.Write(payload)
+		}
+	}
+}
+
+func TestMinioClient_FGetObject_ClosesResponseBody(t *testing.T) {
+	payload := bytes.Repeat([]byte("flux"), 30000)
+	var closed atomic.Bool
+	var ifMatch atomic.Value
+	dir := t.TempDir()
+	partDir := filepath.Join(dir, "nested")
+	var partMu sync.Mutex
+	var partNames []string
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodGet {
+			ifMatch.Store(r.Header.Get("If-Match"))
+			matches, _ := filepath.Glob(filepath.Join(partDir, "*.part.minio"))
+			partMu.Lock()
+			partNames = append(partNames, matches...)
+			partMu.Unlock()
+		}
+		s3ObjectHandler(payload)(w, r)
+	})
+	client := newTrackingMinioClient(t, handler, &closed)
+
+	path := filepath.Join(partDir, "obj")
+	etag, err := client.FGetObject(context.Background(), "test-bucket", "obj", path)
+	g := NewWithT(t)
+	g.Expect(err).NotTo(HaveOccurred())
+	g.Expect(etag).To(Equal("abc123"))
+	g.Expect(ifMatch.Load()).To(Equal(`"abc123"`))
+	got, err := os.ReadFile(path)
+	g.Expect(err).NotTo(HaveOccurred())
+	g.Expect(got).To(Equal(payload))
+	partMu.Lock()
+	seenParts := append([]string(nil), partNames...)
+	partMu.Unlock()
+	g.Expect(seenParts).NotTo(BeEmpty())
+	for _, name := range seenParts {
+		g.Expect(filepath.Base(name)).NotTo(Equal("obj.part.minio"))
+	}
+	matches, globErr := filepath.Glob(filepath.Join(partDir, "*.part.minio"))
+	g.Expect(globErr).NotTo(HaveOccurred())
+	g.Expect(matches).To(BeEmpty())
+	g.Eventually(closed.Load, 2*time.Second, 10*time.Millisecond).Should(BeTrue())
+}
+
+func TestMinioClient_FGetObject_CopyError(t *testing.T) {
+	var closed atomic.Bool
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT")
+		w.Header().Set("ETag", `"abc123"`)
+		w.Header().Set("Content-Length", "64")
+		if r.Method == http.MethodGet {
+			_, _ = w.Write([]byte("short"))
+		}
+	})
+	client := newTrackingMinioClient(t, handler, &closed)
+
+	dir := t.TempDir()
+	path := filepath.Join(dir, "obj")
+	_, err := client.FGetObject(context.Background(), "test-bucket", "obj", path)
+	g := NewWithT(t)
+	g.Expect(err).To(HaveOccurred())
+	_, statErr := os.Stat(path)
+	g.Expect(os.IsNotExist(statErr)).To(BeTrue())
+	matches, globErr := filepath.Glob(filepath.Join(dir, "*.part.minio"))
+	g.Expect(globErr).NotTo(HaveOccurred())
+	g.Expect(matches).To(BeEmpty())
+	g.Eventually(closed.Load, 2*time.Second, 10*time.Millisecond).Should(BeTrue())
+}
+
 func TestFGetObjectInvalidCredentials(t *testing.T) {
 	invalidSecret := secret.DeepCopy()
 	invalidSecret.Data["accesskey"] = []byte("invalid")