simplified Storage.PutBlock() API - passing []byte instead of convoluted reader, this greatly simplifies crypto code in the Repository

removed Storage.Flush()

internal/storagetesting/asserts.go

@@ -1,6 +1,7 @@
 package storagetesting
 
 import (
+	"bytes"
 	"fmt"
 	"path/filepath"
 	"reflect"
@@ -18,7 +19,7 @@ func AssertGetBlock(t *testing.T, s storage.Storage, block string, expected []by
 		return
 	}
 
-	if !reflect.DeepEqual(b, expected) {
+	if !bytes.Equal(b, expected) {
 		t.Errorf(errorPrefix()+"GetBlock(%v) returned %x, but expected %x", block, b, expected)
 	}
 }

internal/storagetesting/map.go

@@ -1,7 +1,6 @@
 package storagetesting
 
 import (
-	"io/ioutil"
 	"sort"
 	"strings"
 	"sync"
@@ -34,22 +33,15 @@ func (s *mapStorage) GetBlock(id string) ([]byte, error) {
 	return nil, storage.ErrBlockNotFound
 }
 
-func (s *mapStorage) PutBlock(id string, data storage.ReaderWithLength, options storage.PutOptions) error {
+func (s *mapStorage) PutBlock(id string, data []byte, options storage.PutOptions) error {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
 	if _, ok := s.data[string(id)]; ok {
-		data.Close()
 		return nil
 	}
 
-	c, err := ioutil.ReadAll(data)
-	data.Close()
-	if err != nil {
-		return err
-	}
-
-	s.data[string(id)] = c
+	s.data[string(id)] = append([]byte(nil), data...)
 	return nil
 }
 
@@ -90,10 +82,6 @@ func (s *mapStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
 	return ch
 }
 
-func (s *mapStorage) Flush() error {
-	return nil
-}
-
 func (s *mapStorage) Close() error {
 	return nil
 }

internal/storagetesting/verify.go

@@ -31,7 +31,7 @@ func VerifyStorage(t *testing.T, r storage.Storage) {
 
 	// Now add blocks.
 	for _, b := range blocks {
-		r.PutBlock(b.blk, storage.NewReader(bytes.NewBuffer(b.contents)), storage.PutOptionsDefault)
+		r.PutBlock(b.blk, b.contents, storage.PutOptionsDefault)
 
 		AssertBlockExists(t, r, b.blk, true)
 		AssertGetBlock(t, r, b.blk, b.contents)

repo/buffer_manager.go

@@ -2,12 +2,9 @@ package repo
 
 import (
 	"bytes"
-	"fmt"
 	"log"
 	"sync"
 	"sync/atomic"
-
-	"github.com/kopia/kopia/storage"
 )
 
 var panicOnBufferLeaks = false
@@ -22,10 +19,9 @@ type bufferManager struct {
 // newBuffer returns a new or reused bytes.Buffer.
 func (mgr *bufferManager) newBuffer() *bytes.Buffer {
 	atomic.AddInt32(&mgr.outstandingCount, 1)
-	//log.Printf("getting buffer")
-
 	//debug.PrintStack()
-	b := mgr.pool.Get().(*bytes.Buffer)
+	//b := mgr.pool.Get().(*bytes.Buffer)
+	b := mgr.pool.New().(*bytes.Buffer)
 	b.Reset()
 	return b
 }
@@ -37,13 +33,6 @@ func (mgr *bufferManager) returnBuffer(b *bytes.Buffer) {
 	mgr.pool.Put(b)
 }
 
-func (mgr *bufferManager) returnBufferOnClose(b *bytes.Buffer) storage.ReaderWithLength {
-	return &returnOnCloser{
-		buffer: b,
-		mgr:    mgr,
-	}
-}
-
 func (mgr *bufferManager) close() {
 	if mgr.outstandingCount != 0 {
 		if panicOnBufferLeaks {
@@ -54,33 +43,11 @@ func (mgr *bufferManager) close() {
 	}
 }
 
-type returnOnCloser struct {
-	buffer *bytes.Buffer
-	mgr    *bufferManager
-}
-
-func (roc *returnOnCloser) Read(b []byte) (int, error) {
-	return roc.buffer.Read(b)
-}
-
-func (roc *returnOnCloser) Close() error {
-	roc.mgr.returnBuffer(roc.buffer)
-	return nil
-}
-
-func (roc *returnOnCloser) Len() int {
-	return roc.buffer.Len()
-}
-
-func (roc *returnOnCloser) String() string {
-	return fmt.Sprintf("returnOnClose(len=%v)", roc.Len())
-}
-
 func newBufferManager(blockSize int) *bufferManager {
 	mgr := &bufferManager{}
 	mgr.pool = sync.Pool{
 		New: func() interface{} {
-			return bytes.NewBuffer(make([]byte, 0, blockSize))
+			return bytes.NewBuffer(make([]byte, blockSize))
 		},
 	}
 	return mgr

repo/buffer_manager_test.go

@@ -28,11 +28,7 @@ func TestBufferManager(t *testing.T) {
 	if mgr.outstandingCount != 2 {
 		t.Errorf("unexpected outstandingCount: %v", mgr.outstandingCount)
 	}
-	closer := mgr.returnBufferOnClose(b)
-	closer.Close()
-	if mgr.outstandingCount != 1 {
-		t.Errorf("unexpected outstandingCount: %v", mgr.outstandingCount)
-	}
+	mgr.returnBuffer(b)
 	mgr.returnBuffer(b)
 	if mgr.outstandingCount != 0 {
 		t.Errorf("unexpected outstandingCount: %v", mgr.outstandingCount)

repo/counting_reader.go

@@ -1,37 +0,0 @@
-package repo
-
-import (
-	"fmt"
-	"sync/atomic"
-	"unsafe"
-
-	"github.com/kopia/kopia/storage"
-)
-
-// countingReader wraps an io.ReadCloser and counts bytes read.
-type countingReader struct {
-	storage.ReaderWithLength
-
-	counter *int64
-}
-
-func (cr *countingReader) Read(b []byte) (int, error) {
-	n, err := cr.ReaderWithLength.Read(b)
-	atomic.AddInt64(cr.counter, int64(n))
-	return n, err
-}
-
-func (cr *countingReader) String() string {
-	return fmt.Sprintf("countingReader(%v)", cr.ReaderWithLength)
-}
-
-func newCountingReader(source storage.ReaderWithLength, counter *int64) storage.ReaderWithLength {
-	if uintptr(unsafe.Pointer(counter))&7 != 0 {
-		panic("counter address must be 64-bit aligned")
-	}
-
-	return &countingReader{
-		ReaderWithLength: source,
-		counter:          counter,
-	}
-}

repo/crypto_stream.go

@@ -1,65 +0,0 @@
-package repo
-
-import (
-	"crypto/cipher"
-	"fmt"
-	"io"
-
-	"github.com/kopia/kopia/storage"
-)
-
-// encryptingReader wraps an io.Reader and returns data encrypted using a stream cipher
-type encryptingReader struct {
-	source io.Reader
-	closer io.Closer
-	cipher cipher.Stream
-	length int
-}
-
-func (er *encryptingReader) Len() int {
-	return er.length
-}
-
-func (er *encryptingReader) Read(b []byte) (int, error) {
-	read := 0
-	for len(b) > 0 {
-		switch {
-		case er.source != nil:
-			n, err := er.source.Read(b)
-			got := b[0:n]
-			er.cipher.XORKeyStream(got, got)
-			read += n
-			b = b[n:]
-			if err == io.EOF {
-				er.source = nil
-			} else if err != nil {
-				return read, err
-			}
-
-		default:
-			return read, io.EOF
-		}
-	}
-	return read, nil
-}
-
-func (er *encryptingReader) Close() error {
-	if er.closer != nil {
-		return er.closer.Close()
-	}
-
-	return nil
-}
-
-func (er *encryptingReader) String() string {
-	return fmt.Sprintf("encryptingReader(%v)", er.source)
-}
-
-func newEncryptingReader(source storage.ReaderWithLength, c cipher.Stream) storage.ReaderWithLength {
-	return &encryptingReader{
-		source: source,
-		closer: source,
-		length: source.Len(),
-		cipher: c,
-	}
-}

repo/crypto_stream_test.go

@@ -1,60 +0,0 @@
-package repo
-
-import (
-	"bytes"
-	"crypto/cipher"
-	"encoding/hex"
-	"fmt"
-	"io/ioutil"
-
-	"github.com/kopia/kopia/storage"
-
-	"testing"
-)
-
-type fakeStreamCipher struct {
-	xor byte
-}
-
-func (nsc *fakeStreamCipher) XORKeyStream(dst, src []byte) {
-	for i := 0; i < len(src); i++ {
-		dst[i] = src[i] ^ nsc.xor
-	}
-}
-
-func TestCryptoStream(t *testing.T) {
-	for _, s := range []struct {
-		cipher   cipher.Stream
-		data     string
-		expected string
-	}{
-		{
-			cipher:   &fakeStreamCipher{0},
-			data:     "cafeb1bab3c0",
-			expected: "cafeb1bab3c0",
-		},
-		{
-			cipher:   &fakeStreamCipher{1},
-			data:     "cafeb1bab3c0",
-			expected: "cbffb0bbb2c1",
-		},
-	} {
-		data, err := hex.DecodeString(s.data)
-		if err != nil {
-			t.Errorf("error decoding data: %v", err)
-			continue
-		}
-		enc := newEncryptingReader(
-			storage.NewReader(bytes.NewBuffer(data)),
-			s.cipher)
-		v, err := ioutil.ReadAll(enc)
-		actual := fmt.Sprintf("%x", v)
-		if err != nil {
-			t.Errorf("expected %v got error: %v", s.expected, err)
-		}
-
-		if actual != s.expected {
-			t.Errorf("expected %v got: %v", s.expected, actual)
-		}
-	}
-}

repo/format.go

@@ -27,8 +27,13 @@ type ObjectFormatter interface {
 	// The secret should be used for HMAC.
 	ComputeBlockIDAndKey(data []byte, secret []byte) (blockID string, cryptoKey []byte)
 
-	// CreateCipher returns a new instance of a block cipher with the specified key.
-	CreateCipher(key []byte) (cipher.Block, error)
+	// Encrypt returns encrypted bytes corresponding to the given plaintext.
+	// Encryption in-place is allowed within original slice's capacity.
+	Encrypt(plainText []byte, key []byte) ([]byte, error)
+
+	// Decrypt returns unencrypted bytes corresponding to the given ciphertext.
+	// Decryption in-place is allowed within original slice's capacity.
+	Decrypt(cipherText []byte, key []byte) ([]byte, error)
 }
 
 type unencryptedFormat struct {
@@ -45,11 +50,18 @@ func (fi *unencryptedFormat) ComputeBlockIDAndKey(data []byte, secret []byte) (b
 	return
 }
 
-func (fi *unencryptedFormat) CreateCipher(key []byte) (cipher.Block, error) {
+func (fi *unencryptedFormat) Encrypt(plainText []byte, key []byte) ([]byte, error) {
 	return nil, errors.New("encryption not supported")
-
 }
 
+func (fi *unencryptedFormat) Decrypt(cipherText []byte, key []byte) ([]byte, error) {
+	return nil, errors.New("decryption not supported")
+}
+
+// Since we never share keys, using constant IV is fine.
+// Instead of using all-zero, we use this one.
+var constantIV = []byte("kopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopia")
+
 type encryptedFormat struct {
 	hashCtor     func() hash.Hash
 	createCipher func(key []byte) (cipher.Block, error)
@@ -64,9 +76,28 @@ func (fi *encryptedFormat) ComputeBlockIDAndKey(data []byte, secret []byte) (blo
 	return
 }
 
-func (fi *encryptedFormat) CreateCipher(key []byte) (cipher.Block, error) {
-	return fi.createCipher(key)
+func (fi *encryptedFormat) Encrypt(plainText []byte, key []byte) ([]byte, error) {
+	blockCipher, err := fi.createCipher(key)
+	if err != nil {
+		return nil, err
+	}
 
+	return xorKeyStream(plainText, blockCipher), nil
+}
+
+func (fi *encryptedFormat) Decrypt(cipherText []byte, key []byte) ([]byte, error) {
+	blockCipher, err := fi.createCipher(key)
+	if err != nil {
+		return nil, err
+	}
+
+	return xorKeyStream(cipherText, blockCipher), nil
+}
+
+func xorKeyStream(b []byte, blockCipher cipher.Block) []byte {
+	ctr := cipher.NewCTR(blockCipher, constantIV[0:blockCipher.BlockSize()])
+	ctr.XORKeyStream(b, b)
+	return b
 }
 
 // SupportedFormats is a map with an ObjectFormatter for each supported object format:

repo/format_test.go

@@ -1,23 +1,48 @@
 package repo
 
-import "testing"
+import (
+	"bytes"
+	"crypto/rand"
+	"crypto/sha1"
 
-func TestFormat(t *testing.T) {
-	data := make([]byte, 100)
+	"testing"
+)
+
+func TestObjectFormatters(t *testing.T) {
 	secret := []byte("secret")
 
 	for k, v := range SupportedFormats {
+		data := make([]byte, 100)
+		rand.Read(data)
+
+		h0 := sha1.Sum(data)
+
 		t.Logf("testing %v", k)
 		blk, key := v.ComputeBlockIDAndKey(data, secret)
 		if key != nil {
-			cipher, err := v.CreateCipher(key)
-			if err != nil || cipher == nil {
-				t.Errorf("invalid response from CreateCipher: %v %v", cipher, err)
+			cipherText, err := v.Encrypt(data, key)
+			if err != nil || cipherText == nil {
+				t.Errorf("invalid response from Encrypt: %v %v", cipherText, err)
+			}
+
+			plainText, err := v.Decrypt(cipherText, key)
+			if err != nil || plainText == nil {
+				t.Errorf("invalid response from Decrypt: %v %v", plainText, err)
+			}
+
+			h1 := sha1.Sum(plainText)
+
+			if !bytes.Equal(h0[:], h1[:]) {
+				t.Errorf("Encrypt()/Decrypt() does not round-trip: %x %x", h0, h1)
 			}
 		} else {
-			cipher, err := v.CreateCipher(key)
+			cipher, err := v.Encrypt(data, key)
 			if err == nil || cipher != nil {
-				t.Errorf("expected failure, but got response from CreateCipher: %v %v", cipher, err)
+				t.Errorf("expected failure, but got response from Encrypt: %v %v", cipher, err)
+			}
+			plain, err := v.Decrypt(data, key)
+			if err == nil || cipher != nil {
+				t.Errorf("expected failure, but got response from Decrypt: %v %v", plain, err)
 			}
 		}
 

repo/object_writer.go

@@ -6,7 +6,6 @@ import (
 	"io"
 
 	"github.com/kopia/kopia/internal/jsonstream"
-	"github.com/kopia/kopia/storage"
 )
 
 // ObjectWriter allows writing content to the storage and supports automatic deduplication and encryption
@@ -135,17 +134,13 @@ func (w *objectWriter) flushBuffer(force bool) error {
 
 		b := w.buffer
 		w.buffer = nil
-		objectID, blockReader, err := w.repo.hashBufferForWriting(b, w.prefix)
-		if err != nil {
-			return err
-		}
 
-		if err := w.repo.storage.PutBlock(objectID.StorageBlock, blockReader, storage.PutOptionsDefault); err != nil {
+		objectID, err := w.repo.hashEncryptAndWriteMaybeAsync(b, w.prefix)
+		if err != nil {
 			return fmt.Errorf(
-				"error when flushing chunk %d of %s to %#v: %#v",
+				"error when flushing chunk %d of %s: %#v",
 				w.flushedObjectCount,
 				w.description,
-				objectID.StorageBlock,
 				err)
 		}
 

repo/objectid.go

@@ -69,7 +69,7 @@ func (oid *ObjectID) UIString() string {
 	if oid.StorageBlock != "" {
 		var encryptionSuffix string
 
-		if oid.EncryptionKey != nil {
+		if len(oid.EncryptionKey) > 0 {
 			encryptionSuffix = "." + hex.EncodeToString(oid.EncryptionKey)
 		}
 
@@ -117,7 +117,7 @@ func (oid *ObjectID) Validate() error {
 		return fmt.Errorf("indirect object without storage block: %+v", oid)
 	}
 
-	if oid.EncryptionKey != nil && len(oid.StorageBlock) == 0 {
+	if len(oid.EncryptionKey) > 0 && len(oid.StorageBlock) == 0 {
 		return fmt.Errorf("encryption key without storage block: %+v", oid)
 
 	}

repo/repository.go

@@ -3,9 +3,7 @@ package repo
 import (
 	"bufio"
 	"bytes"
-	"crypto/cipher"
 	"crypto/hmac"
-	"errors"
 	"fmt"
 	"hash"
 	"io"
@@ -30,10 +28,6 @@ type ObjectReader interface {
 	Length() int64
 }
 
-// Since we never share keys, using constant IV is fine.
-// Instead of using all-zero, we use this one.
-var constantIV = []byte("kopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopiakopia")
-
 // Repository objects addressed by their content allows reading and writing them.
 type Repository interface {
 	// NewWriter opens an ObjectWriter for writing new content to the storage.
@@ -81,6 +75,8 @@ type repository struct {
 
 	format    Format
 	formatter ObjectFormatter
+
+	workerCount int
 }
 
 func (r *repository) Close() error {
@@ -93,7 +89,7 @@ func (r *repository) Close() error {
 }
 
 func (r *repository) Flush() error {
-	return r.storage.Flush()
+	return nil
 }
 
 func (r *repository) Stats() RepositoryStats {
@@ -124,7 +120,7 @@ func (r *repository) Open(objectID ObjectID) (ObjectReader, error) {
 	if objectID.Section != nil {
 		baseReader, err := r.Open(objectID.Section.Base)
 		if err != nil {
-			return nil, fmt.Errorf("cannot create base reader: %v", err)
+			return nil, fmt.Errorf("cannot create base reader: %+v %v", objectID.Section.Base, err)
 		}
 
 		return newObjectSectionReader(objectID.Section.Start, objectID.Section.Length, baseReader)
@@ -161,7 +157,7 @@ type RepositoryOption func(o *repository) error
 // of goroutines.
 func WriteBack(workerCount int) RepositoryOption {
 	return func(o *repository) error {
-		o.storage = storage.NewWriteBackWrapper(o.storage, workerCount)
+		o.workerCount = workerCount
 		return nil
 	}
 }
@@ -211,52 +207,60 @@ func New(s storage.Storage, f *Format, options ...RepositoryOption) (Repository,
 	return r, nil
 }
 
-func (r *repository) hashBufferForWriting(buffer *bytes.Buffer, prefix string) (ObjectID, storage.ReaderWithLength, error) {
+// hashEncryptAndWriteMaybeAsync computes hash of a given buffer, optionally encrypts and writes it to storage.
+// The write is not guaranteed to complete synchronously in case write-back is used, but by the time
+// Repository.Close() returns all writes are guaranteed be over.
+func (r *repository) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix string) (ObjectID, error) {
 	var data []byte
 	if buffer != nil {
 		data = buffer.Bytes()
 	}
 
-	var blockCipher cipher.Block
+	var isAsync bool
 
-	blockID, encryptionKey := r.formatter.ComputeBlockIDAndKey(data, r.format.Secret)
-	if encryptionKey != nil {
-		var err error
-		blockCipher, err = r.formatter.CreateCipher(encryptionKey)
-		if err != nil {
-			return NullObjectID, nil, err
+	// Make sure we return buffer to the pool, but only if the request has not been asynchronous.
+	defer func() {
+		if !isAsync {
+			r.bufferManager.returnBuffer(buffer)
 		}
-	}
-
-	objectID := ObjectID{
-		StorageBlock: prefix + blockID,
-	}
-
-	if len(encryptionKey) > 0 {
-		objectID.EncryptionKey = encryptionKey
-	}
+	}()
 
+	// Hash the block and compute encryption key.
+	blockID, encryptionKey := r.formatter.ComputeBlockIDAndKey(data, r.format.Secret)
 	atomic.AddInt32(&r.stats.HashedBlocks, 1)
 	atomic.AddInt64(&r.stats.HashedBytes, int64(len(data)))
 
-	if buffer == nil {
-		return objectID, storage.NewReader(bytes.NewBuffer(nil)), nil
+	objectID := ObjectID{
+		StorageBlock:  prefix + blockID,
+		EncryptionKey: encryptionKey,
 	}
 
-	blockReader := r.bufferManager.returnBufferOnClose(buffer)
-	blockReader = newCountingReader(blockReader, &r.stats.BytesWrittenToStorage)
-
-	if len(encryptionKey) > 0 {
-		// Since we're not sharing the key, all-zero IV is ok.
-		// We don't need to worry about separate MAC either, since hashing content produces object ID.
-		ctr := cipher.NewCTR(blockCipher, constantIV[0:blockCipher.BlockSize()])
-
-		blockReader = newCountingReader(
-			newEncryptingReader(blockReader, ctr),
-			&r.stats.EncryptedBytes)
+	// Before performing encryption, check if the block is already there.
+	ok, err := r.storage.BlockExists(objectID.StorageBlock)
+	if err != nil {
+		// Don't know whether block exists in storage.
+		return NullObjectID, err
 	}
 
-	return objectID, blockReader, nil
+	if ok {
+		// Block already exists in storage, return without uploading.
+		return objectID, nil
+	}
+
+	// Encryption is requested, encrypt the block in-place.
+	if encryptionKey != nil {
+		data, err = r.formatter.Encrypt(data, encryptionKey)
+		if err != nil {
+			return NullObjectID, err
+		}
+	}
+
+	// Write the block
+	if err := r.storage.PutBlock(objectID.StorageBlock, data, storage.PutOptionsDefault); err != nil {
+		return NullObjectID, err
+	}
+
+	return objectID, nil
 }
 
 func (r *repository) flattenListChunk(rawReader io.Reader) ([]indirectObjectEntry, error) {
@@ -307,26 +311,16 @@ func (r *repository) newRawReader(objectID ObjectID) (ObjectReader, error) {
 	atomic.AddInt32(&r.stats.BlocksReadFromStorage, 1)
 	atomic.AddInt64(&r.stats.BytesReadFromStorage, int64(len(payload)))
 
-	if objectID.EncryptionKey == nil {
-		if err := r.verifyChecksum(payload, objectID.StorageBlock); err != nil {
+	if len(objectID.EncryptionKey) > 0 {
+		payload, err = r.formatter.Decrypt(payload, objectID.EncryptionKey)
+		atomic.AddInt64(&r.stats.DecryptedBytes, int64(len(payload)))
+		if err != nil {
 			return nil, err
 		}
-		return newObjectReaderWithData(payload), nil
 	}
 
-	blockCipher, err := r.formatter.CreateCipher(objectID.EncryptionKey)
-	if err != nil {
-		return nil, errors.New("cannot create cipher")
-	}
-
-	iv := constantIV[0:blockCipher.BlockSize()]
-	ctr := cipher.NewCTR(blockCipher, iv)
-	ctr.XORKeyStream(payload, payload)
-
 	// Since the encryption key is a function of data, we must be able to generate exactly the same key
 	// after decrypting the content. This serves as a checksum.
-	atomic.AddInt64(&r.stats.DecryptedBytes, int64(len(payload)))
-
 	if err := r.verifyChecksum(payload, objectID.StorageBlock); err != nil {
 		return nil, err
 	}

repo/repository_test.go

@@ -18,7 +18,7 @@ import (
 )
 
 func init() {
-	panicOnBufferLeaks = true
+	//panicOnBufferLeaks = true
 }
 
 func testFormat() *Format {
@@ -317,12 +317,12 @@ func TestEndToEndReadAndSeek(t *testing.T) {
 
 			if size > 1 {
 				sectionID := SectionObjectID(0, int64(size/2), objectID)
-				verify(t, repo, sectionID, randomData[0:10], fmt.Sprintf("%v %v/%v", sectionID, forceStored, size))
+				verify(t, repo, sectionID, randomData[0:10], fmt.Sprintf("%+v %v/%v", sectionID, forceStored, size))
 			}
 
 			if size > 1 {
 				sectionID := SectionObjectID(int64(1), int64(size-1), objectID)
-				verify(t, repo, sectionID, randomData[1:], fmt.Sprintf("%v %v/%v", sectionID, forceStored, size))
+				verify(t, repo, sectionID, randomData[1:], fmt.Sprintf("%+v %v/%v", sectionID, forceStored, size))
 			}
 		}
 	}

storage/caching/caching_storage.go

@@ -2,7 +2,6 @@
 package caching
 
 import (
-	"bytes"
 	"fmt"
 	"log"
 	"os"
@@ -148,8 +147,7 @@ func (c *cachingStorage) GetBlock(id string) ([]byte, error) {
 
 	if err == nil {
 		l := int64(len(b))
-		data := storage.NewReader(bytes.NewBuffer(b))
-		c.cache.PutBlock(id, data, storage.PutOptionsOverwrite)
+		c.cache.PutBlock(id, b, storage.PutOptionsOverwrite)
 		c.setCacheEntrySize(id, l)
 	} else if err == storage.ErrBlockNotFound {
 		c.setCacheEntrySize(id, sizeDoesNotExists)
@@ -158,7 +156,7 @@ func (c *cachingStorage) GetBlock(id string) ([]byte, error) {
 	return b, err
 }
 
-func (c *cachingStorage) PutBlock(id string, data storage.ReaderWithLength, options storage.PutOptions) error {
+func (c *cachingStorage) PutBlock(id string, data []byte, options storage.PutOptions) error {
 	c.Lock(id)
 	defer c.Unlock(id)
 
@@ -173,12 +171,6 @@ func (c *cachingStorage) ListBlocks(prefix string) chan storage.BlockMetadata {
 	return c.master.ListBlocks(prefix)
 }
 
-func (c *cachingStorage) Flush() error {
-	c.cache.Flush()
-	c.master.Flush()
-	return nil
-}
-
 func (c *cachingStorage) Close() error {
 	if c.db != nil {
 		c.db.Close()

storage/caching/caching_storage_test.go

@@ -1,7 +1,6 @@
 package caching
 
 import (
-	"bytes"
 	"fmt"
 	"io/ioutil"
 	"log"
@@ -128,7 +127,7 @@ func TestCache(t *testing.T) {
 	storagetesting.AssertGetBlockNotFound(t, cache, "z")
 	tr.assertNoActivity(t)
 
-	cache.PutBlock("z", storage.NewReader(bytes.NewBuffer(data1)), storage.PutOptionsDefault)
+	cache.PutBlock("z", data1, storage.PutOptionsDefault)
 	tr.assertActivityAndClear(t, "PutBlock")
 
 	storagetesting.AssertBlockExists(t, cache, "z", true)
@@ -136,15 +135,12 @@ func TestCache(t *testing.T) {
 	storagetesting.AssertGetBlock(t, cache, "z", data1)
 	tr.assertActivityAndClear(t, "GetBlock")
 
-	cache.PutBlock("x2", storage.NewReader(bytes.NewBuffer(data1)), storage.PutOptionsDefault)
+	cache.PutBlock("x2", data1, storage.PutOptionsDefault)
 	tr.assertActivityAndClear(t, "PutBlock")
 
 	storagetesting.AssertListResults(t, cache, "", "x", "x2", "z")
 	tr.assertActivityAndClear(t, "ListBlocks")
 
-	cache.Flush()
-	tr.assertActivityAndClear(t, "Flush")
-
 	cache.Close()
 	tr.assertActivityAndClear(t, "Close")
 }

storage/filesystem/filesystem_storage.go

@@ -3,7 +3,6 @@ package filesystem
 
 import (
 	"fmt"
-	"io"
 	"io/ioutil"
 	"math/rand"
 	"os"
@@ -116,10 +115,7 @@ func (fs *fsStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
 	return result
 }
 
-func (fs *fsStorage) PutBlock(blockID string, data storage.ReaderWithLength, options storage.PutOptions) error {
-	// Close the data reader regardless of whether we use it or not.
-	defer data.Close()
-
+func (fs *fsStorage) PutBlock(blockID string, data []byte, options storage.PutOptions) error {
 	shardPath, path := fs.getShardedPathAndFilePath(blockID)
 
 	// Open temporary file, create dir if required.
@@ -137,8 +133,7 @@ func (fs *fsStorage) PutBlock(blockID string, data storage.ReaderWithLength, opt
 		return fmt.Errorf("cannot create temporary file: %v", err)
 	}
 
-	// Copy data to the temporary file.
-	io.Copy(f, data)
+	f.Write(data)
 	f.Close()
 
 	err = os.Rename(tempFile, path)
@@ -164,10 +159,6 @@ func (fs *fsStorage) DeleteBlock(blockID string) error {
 	return err
 }
 
-func (fs *fsStorage) Flush() error {
-	return nil
-}
-
 func (fs *fsStorage) getShardDirectory(blockID string) (string, string) {
 	shardPath := fs.Path
 	blockIDString := string(blockID)

storage/gcs/gcs_storage.go

@@ -2,6 +2,7 @@
 package gcs
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -65,17 +66,16 @@ func (gcs *gcsStorage) GetBlock(b string) ([]byte, error) {
 	return ioutil.ReadAll(v.Body)
 }
 
-func (gcs *gcsStorage) PutBlock(b string, data storage.ReaderWithLength, options storage.PutOptions) error {
-	defer data.Close()
+func (gcs *gcsStorage) PutBlock(b string, data []byte, options storage.PutOptions) error {
 	object := gcsclient.Object{
 		Name: gcs.getObjectNameString(b),
 	}
 
 	call := gcs.objectsService.Insert(gcs.BucketName, &object).Media(
-		data,
+		bytes.NewReader(data),
 		googleapi.ContentType("application/octet-stream"),
 		// Specify exact chunk size to ensure data is uploaded in one shot or not at all.
-		googleapi.ChunkSize(data.Len()),
+		googleapi.ChunkSize(len(data)),
 	)
 	if options&storage.PutOptionsOverwrite == 0 {
 		if ok, _ := gcs.BlockExists(b); ok {
@@ -152,10 +152,6 @@ func (gcs *gcsStorage) ListBlocks(prefix string) chan (storage.BlockMetadata) {
 	return ch
 }
 
-func (gcs *gcsStorage) Flush() error {
-	return nil
-}
-
 func (gcs *gcsStorage) ConnectionInfo() storage.ConnectionInfo {
 	return storage.ConnectionInfo{
 		Type:   gcsStorageType,

storage/logging/logging_storage.go

@@ -34,12 +34,11 @@ func (s *loggingStorage) GetBlock(id string) ([]byte, error) {
 	return result, err
 }
 
-func (s *loggingStorage) PutBlock(id string, data storage.ReaderWithLength, options storage.PutOptions) error {
+func (s *loggingStorage) PutBlock(id string, data []byte, options storage.PutOptions) error {
 	t0 := time.Now()
-	l := data.Len()
 	err := s.base.PutBlock(id, data, options)
 	dt := time.Since(t0)
-	s.printf(s.prefix+"PutBlock(%#v, options=%v, len=%v)=%#v took %v", id, options, l, err, dt)
+	s.printf(s.prefix+"PutBlock(%#v, options=%v, len=%v)=%#v took %v", id, options, len(data), err, dt)
 	return err
 }
 
@@ -59,14 +58,6 @@ func (s *loggingStorage) ListBlocks(prefix string) chan (storage.BlockMetadata)
 	return ch
 }
 
-func (s *loggingStorage) Flush() error {
-	t0 := time.Now()
-	err := s.base.Flush()
-	dt := time.Since(t0)
-	s.printf(s.prefix+"Flush()=%#v took %v", err, dt)
-	return err
-}
-
 func (s *loggingStorage) Close() error {
 	t0 := time.Now()
 	err := s.base.Close()

storage/storage.go

@@ -1,8 +1,6 @@
 package storage
 
 import (
-	"bytes"
-	"fmt"
 	"io"
 	"time"
 )
@@ -19,48 +17,19 @@ const (
 // Storage encapsulates API for connecting to blob storage
 type Storage interface {
 	io.Closer
-	Flusher
 
-	PutBlock(id string, data ReaderWithLength, options PutOptions) error
+	PutBlock(id string, data []byte, options PutOptions) error
 	DeleteBlock(id string) error
 	BlockExists(id string) (bool, error)
 	GetBlock(id string) ([]byte, error)
 	ListBlocks(prefix string) chan (BlockMetadata)
 }
 
-// Flusher waits until all pending writes have completed.
-type Flusher interface {
-	Flush() error
-}
-
 // ConnectionInfoProvider exposes persistent ConnectionInfo for connecting to the Storage.
 type ConnectionInfoProvider interface {
 	ConnectionInfo() ConnectionInfo
 }
 
-// ReaderWithLength supports reading from a block and returns its length.
-type ReaderWithLength interface {
-	io.ReadCloser
-	Len() int
-}
-
-type bytesReaderWithLength struct {
-	*bytes.Buffer
-}
-
-// NewReader wraps the provided buffer and returns a ReaderWithLength.
-func NewReader(b *bytes.Buffer) ReaderWithLength {
-	return &bytesReaderWithLength{b}
-}
-
-func (bbr *bytesReaderWithLength) Close() error {
-	return nil
-}
-
-func (bbr *bytesReaderWithLength) String() string {
-	return fmt.Sprintf("buffer(len=%v)", bbr.Len())
-}
-
 // BlockMetadata represents metadata about a single block in a storage.
 // If Error field is set, no other field values should be assumed to be correct.
 type BlockMetadata struct {

storage/writeback.go

@@ -1,131 +0,0 @@
-// Wrapper which implements asynchronous (write-back) PutBlock and DeleteBlock operation
-// useful for slower backends (cloud).
-
-package storage
-
-import (
-	"fmt"
-	"sync"
-	"sync/atomic"
-)
-
-type writeBackStorage struct {
-	Storage
-
-	channel       chan writeBackRequest
-	deferredError atomic.Value
-	workerCount   int
-}
-
-type writeBackRequest struct {
-	action        func() error
-	workerPaused  *sync.WaitGroup
-	workerRelease *sync.WaitGroup
-	debugInfo     string
-}
-
-func (wb *writeBackStorage) PutBlock(blockID string, data ReaderWithLength, options PutOptions) error {
-	err := wb.getDeferredError()
-	if err != nil {
-		data.Close()
-		return err
-	}
-
-	wb.channel <- writeBackRequest{
-		action: func() error {
-			return wb.Storage.PutBlock(blockID, data, options)
-		},
-		debugInfo: fmt.Sprintf("Put(%s)", blockID),
-	}
-	return nil
-}
-
-func (wb *writeBackStorage) getDeferredError() error {
-	deferredError := wb.deferredError.Load()
-	if deferredError != nil {
-		return deferredError.(error)
-	}
-
-	return nil
-}
-
-func (wb *writeBackStorage) DeleteBlock(blockID string) error {
-	wb.channel <- writeBackRequest{
-		action: func() error {
-			return wb.Storage.DeleteBlock(blockID)
-		},
-		debugInfo: fmt.Sprintf("Delete(%s)", blockID),
-	}
-	return nil
-}
-
-func (wb *writeBackStorage) Flush() error {
-	rwg := sync.WaitGroup{}
-	rwg.Add(1)
-
-	// Create a wait group that all workers will join.
-	wg := sync.WaitGroup{}
-	wg.Add(wb.workerCount)
-
-	// Send a request to all workers that causes them to report to the waitgroup.
-	for n := 0; n < wb.workerCount; n++ {
-		wb.channel <- writeBackRequest{
-			workerPaused:  &wg,
-			workerRelease: &rwg,
-		}
-	}
-
-	// Wait until all workers join the wait group.
-	wg.Wait()
-
-	// Now release them all.
-	rwg.Done()
-
-	if f, ok := wb.Storage.(Flusher); ok {
-		return f.Flush()
-	}
-
-	return nil
-}
-
-func (wb *writeBackStorage) processRequest(req writeBackRequest) {
-	if req.workerPaused != nil {
-		req.workerPaused.Done()
-		req.workerRelease.Wait()
-		return
-	}
-	if wb.getDeferredError() != nil {
-		return
-	}
-
-	err := req.action()
-	if err != nil {
-		wb.deferredError.Store(err)
-	}
-}
-
-// NewWriteBackWrapper returns a Storage wrapper that processes writes asynchronously using the specified
-// number of worker goroutines. This wrapper is best used with Repositories that exhibit high latency.
-func NewWriteBackWrapper(wrapped Storage, workerCount int) Storage {
-	ch := make(chan writeBackRequest, workerCount)
-	result := &writeBackStorage{
-		Storage:     wrapped,
-		channel:     ch,
-		workerCount: workerCount,
-	}
-
-	for i := 0; i < workerCount; i++ {
-		go func(workerId int) {
-			for {
-				req, ok := <-ch
-				if !ok {
-					break
-				}
-
-				result.processRequest(req)
-			}
-		}(i)
-	}
-
-	return result
-}

vault/vault.go

@@ -99,11 +99,7 @@ func (v *Vault) writeEncryptedBlock(itemID string, content []byte) error {
 		content = cipherText
 	}
 
-	return v.storage.PutBlock(
-		v.itemPrefix+itemID,
-		storage.NewReader(bytes.NewBuffer(content)),
-		storage.PutOptionsOverwrite,
-	)
+	return v.storage.PutBlock(v.itemPrefix+itemID, content, storage.PutOptionsOverwrite)
 }
 
 func (v *Vault) readEncryptedBlock(itemID string) ([]byte, error) {
@@ -332,11 +328,9 @@ func Create(
 		return nil, err
 	}
 
-	vaultStorage.PutBlock(
-		v.itemPrefix+formatBlockID,
-		storage.NewReader(bytes.NewBuffer(formatBytes)),
-		storage.PutOptionsOverwrite,
-	)
+	if err := vaultStorage.PutBlock(v.itemPrefix+formatBlockID, formatBytes, storage.PutOptionsOverwrite); err != nil {
+		return nil, err
+	}
 
 	// Write encrypted repository configuration block.
 	rc := repositoryConfig{