Ensure zlib buffer is closed before writing objects to storage

storage/chaintree/object.go

@@ -116,25 +116,10 @@ func (s *ObjectStorage) SetEncodedObjectTxn(o plumbing.EncodedObject) (*transact
 		return nil, plumbing.ErrInvalidType
 	}
 
-	buf := bytes.NewBuffer(nil)
-
-	writer := objfile.NewWriter(buf)
-	defer writer.Close()
-
-	reader, err := o.Reader()
+	buf, err := storage.ZlibBufferForObject(o)
 	if err != nil {
 		return nil, err
 	}
-	defer reader.Close()
-
-	if err := writer.WriteHeader(o.Type(), o.Size()); err != nil {
-		return nil, err
-	}
-
-	if _, err = io.Copy(writer, reader); err != nil {
-		return nil, err
-	}
-	writer.Close()
 
 	objectBytes, err := ioutil.ReadAll(buf)
 	if err != nil {

storage/object.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/go-git/go-git/v5/plumbing"
+	"github.com/go-git/go-git/v5/plumbing/format/objfile"
 	"github.com/go-git/go-git/v5/plumbing/format/packfile"
 	"github.com/go-git/go-git/v5/plumbing/storer"
 	logging "github.com/ipfs/go-log"
@@ -37,6 +38,29 @@ func ObjectWritePath(h plumbing.Hash) string {
 	return strings.Join(ObjectReadPath(h)[2:], "/")
 }
 
+func ZlibBufferForObject(o plumbing.EncodedObject) (*bytes.Buffer, error) {
+	buf := bytes.NewBuffer(nil)
+
+	writer := objfile.NewWriter(buf)
+	defer writer.Close()
+
+	reader, err := o.Reader()
+	if err != nil {
+		return nil, err
+	}
+	defer reader.Close()
+
+	if err := writer.WriteHeader(o.Type(), o.Size()); err != nil {
+		return nil, err
+	}
+
+	if _, err = io.Copy(writer, reader); err != nil {
+		return nil, err
+	}
+
+	return buf, err
+}
+
 func (s *ChaintreeObjectStorage) Chaintree() *chaintree.ChainTree {
 	return s.ChainTree.ChainTree
 }

storage/object_test.go

@@ -0,0 +1,25 @@
+package storage
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/go-git/go-git/v5/plumbing"
+	"github.com/stretchr/testify/require"
+)
+
+func TestZlibBufferForObject(t *testing.T) {
+	o := &plumbing.MemoryObject{}
+	o.SetType(plumbing.BlobObject)
+	o.SetSize(14)
+	_, err := o.Write([]byte("Hello, World!\n"))
+	require.Nil(t, err)
+	require.Equal(t, o.Hash().String(), "8ab686eafeb1f44702738c8b0f24f2567c36da6d")
+
+	buf, err := ZlibBufferForObject(o)
+	require.Nil(t, err)
+
+	fmt.Printf("%v", buf.Bytes())
+
+	require.Equal(t, buf.Bytes(), []byte{120, 156, 74, 202, 201, 79, 82, 48, 52, 97, 240, 72, 205, 201, 201, 215, 81, 8, 207, 47, 202, 73, 81, 228, 2, 4, 0, 0, 255, 255, 78, 21, 6, 152})
+}

storage/siaskynet/net.go

@@ -1,13 +1,13 @@
 package siaskynet
 
 import (
-	"bytes"
 	"io"
 	"sync"
 
 	"github.com/NebulousLabs/go-skynet"
 	"github.com/go-git/go-git/v5/plumbing"
 	"github.com/go-git/go-git/v5/plumbing/format/objfile"
+	"github.com/quorumcontrol/dgit/storage"
 	"go.uber.org/zap"
 )
 
@@ -47,24 +47,11 @@ func InitSkynet(uploaderCount, downloaderCount int) *Skynet {
 }
 
 func (s *Skynet) uploadObject(o plumbing.EncodedObject) (string, error) {
-	buf := bytes.NewBuffer(nil)
-
-	writer := objfile.NewWriter(buf)
-	defer writer.Close()
-
-	reader, err := o.Reader()
+	buf, err := storage.ZlibBufferForObject(o)
 	if err != nil {
 		return "", err
 	}
 
-	if err = writer.WriteHeader(o.Type(), o.Size()); err != nil {
-		return "", err
-	}
-
-	if _, err = io.Copy(writer, reader); err != nil {
-		return "", err
-	}
-
 	uploadData := make(skynet.UploadData)
 	uploadData[o.Hash().String()] = buf