Repository: kopia/kopia
Pull request #373 (2020-03-20 05:38:33)

Support for gather writes, where blob.Storage.PutBlob gets a list of slices and writes them sequentially
This amortizes memory allocations when assembling content packs.

This PR only optimizes code paths critical to snapshot creations.


Patch:
diff --git a/Makefile b/Makefile
index 32370327..3b5051eb 100644
--- a/Makefile
+++ b/Makefile
@@ -187,6 +187,7 @@ ifneq ($(uname),Windows)
 	             -e github.com/kopia/kopia/internal/buf \
 	             -e github.com/kopia/kopia/internal/throttle \
 	             -e github.com/kopia/kopia/internal/iocopy \
+	             -e github.com/kopia/kopia/internal/gather \
 	             -e github.com/kopia/kopia/internal/blobtesting \
 	             -e github.com/kopia/kopia/internal/repotesting \
 	             -e github.com/kopia/kopia/internal/testlogging \
diff --git a/cli/command_policy_set.go b/cli/command_policy_set.go
index fcf570fc..b63571f3 100644
--- a/cli/command_policy_set.go
+++ b/cli/command_policy_set.go
@@ -82,8 +82,12 @@ func setPolicy(ctx context.Context, rep *repo.Repository) error {
 
 	for _, target := range targets {
 		p, err := policy.GetDefinedPolicy(ctx, rep, target)
-		if err == policy.ErrPolicyNotFound {
+
+		switch {
+		case err == policy.ErrPolicyNotFound:
 			p = &policy.Policy{}
+		case err != nil:
+			return errors.Wrap(err, "could not get defined policy")
 		}
 
 		printStderr("Setting policy for %v\n", target)
diff --git a/cli/command_repository_repair.go b/cli/command_repository_repair.go
index e9b9d633..1ee69324 100644
--- a/cli/command_repository_repair.go
+++ b/cli/command_repository_repair.go
@@ -5,6 +5,7 @@
 
 	"github.com/pkg/errors"
 
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/repo"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/content"
@@ -66,7 +67,7 @@ func recoverFormatBlob(ctx context.Context, st blob.Storage, prefixes []string)
 			log(ctx).Infof("looking for replica of format blob in %v...", bi.BlobID)
 			if b, err := repo.RecoverFormatBlob(ctx, st, bi.BlobID, bi.Length); err == nil {
 				if !*repairDryDrun {
-					if puterr := st.PutBlob(ctx, repo.FormatBlobID, b); puterr != nil {
+					if puterr := st.PutBlob(ctx, repo.FormatBlobID, gather.FromSlice(b)); puterr != nil {
 						return puterr
 					}
 				}
diff --git a/internal/gather/gather_bytes.go b/internal/gather/gather_bytes.go
new file mode 100644
index 00000000..5912ce47
--- /dev/null
+++ b/internal/gather/gather_bytes.go
@@ -0,0 +1,133 @@
+// Package gather implements data structures storing binary data organized
+// in a series of byte slices of fixed size that only gathered together by the user.
+package gather
+
+import (
+	"bytes"
+	"io"
+)
+
+// Bytes represents a sequence of bytes split into slices.
+type Bytes struct {
+	Slices [][]byte
+
+	// for common case where there's one slice, store the slice itself here
+	// to avoid allocation
+	sliceBuf [1][]byte
+}
+
+// AppendSectionTo appends the section of the buffer to the provided slice and returns it.
+func (b *Bytes) AppendSectionTo(output []byte, offset, size int) []byte {
+	// find the index of starting slice
+	sliceNdx := -1
+
+	for i, p := range b.Slices {
+		if offset < len(p) {
+			sliceNdx = i
+			break
+		}
+
+		offset -= len(p)
+	}
+
+	// not found
+	if sliceNdx == -1 {
+		return nil
+	}
+
+	// first slice, possibly with offset zero
+	var firstChunkSize int
+	if offset+size <= len(b.Slices[sliceNdx]) {
+		firstChunkSize = size
+	} else {
+		// slice shorter
+		firstChunkSize = len(b.Slices[sliceNdx]) - offset
+	}
+
+	output = append(output, b.Slices[sliceNdx][offset:offset+firstChunkSize]...)
+	size -= firstChunkSize
+	sliceNdx++
+
+	// at this point we're staying at offset 0
+	for size > 0 && sliceNdx < len(b.Slices) {
+		s := b.Slices[sliceNdx]
+
+		// l is how many bytes we consume out of the current slice
+		l := size
+		if l > len(s) {
+			l = len(s)
+		}
+
+		output = append(output, s[0:l]...)
+		size -= l
+		sliceNdx++
+	}
+
+	return output
+}
+
+// Length returns the combined length of all slices.
+func (b Bytes) Length() int {
+	l := 0
+
+	for _, data := range b.Slices {
+		l += len(data)
+	}
+
+	return l
+}
+
+// Reader returns a reader for the data.
+func (b Bytes) Reader() io.Reader {
+	switch len(b.Slices) {
+	case 0:
+		return bytes.NewReader(nil)
+
+	case 1:
+		return bytes.NewReader(b.Slices[0])
+
+	default:
+		readers := make([]io.Reader, 0, len(b.Slices))
+
+		for _, v := range b.Slices {
+			readers = append(readers, bytes.NewReader(v))
+		}
+
+		return io.MultiReader(readers...)
+	}
+}
+
+// GetBytes appends all bytes to the provided slice and returns it.
+func (b Bytes) GetBytes(output []byte) []byte {
+	for _, v := range b.Slices {
+		output = append(output, v...)
+	}
+
+	return output
+}
+
+// WriteTo writes contents to the specified writer and returns number of bytes written.
+func (b Bytes) WriteTo(w io.Writer) (int64, error) {
+	var totalN int64
+
+	for _, v := range b.Slices {
+		n, err := w.Write(v)
+
+		totalN += int64(n)
+
+		if err != nil {
+			return totalN, err
+		}
+	}
+
+	return totalN, nil
+}
+
+// FromSlice creates Bytes from the specified slice.
+func FromSlice(b []byte) Bytes {
+	var r Bytes
+	r.sliceBuf[0] = b
+	r.Slices = r.sliceBuf[:]
+
+	return r
+}
diff --git a/internal/gather/gather_write_buffer.go b/internal/gather/gather_write_buffer.go
new file mode 100644
index 00000000..32693c8a
--- /dev/null
+++ b/internal/gather/gather_write_buffer.go
@@ -0,0 +1,63 @@
+package gather
+
+// WriteBuffer is a write buffer for content of unknown size that manages
+// data in a series of byte slices of uniform size.
+type WriteBuffer struct {
+	Bytes
+}
+
+// Close releases all memory allocated by this buffer.
+func (b *WriteBuffer) Close() {
+	for _, s := range b.Slices {
+		releaseChunk(s)
+	}
+
+	b.Slices = nil
+}
+
+// Reset resets buffer back to empty.
+func (b *WriteBuffer) Reset() {
+	for _, s := range b.Slices {
+		releaseChunk(s)
+	}
+
+	b.Slices = nil
+}
+
+// Write implements io.Writer for appending to the buffer.
+func (b *WriteBuffer) Write(data []byte) (n int, err error) {
+	b.Append(data)
+	return len(data), nil
+}
+
+// Append appends the specified slice of bytes to the buffer.
+func (b *WriteBuffer) Append(data []byte) {
+	if len(b.Slices) == 0 {
+		b.sliceBuf[0] = allocChunk()
+		b.Slices = b.sliceBuf[0:1]
+	}
+
+	for len(data) > 0 {
+		ndx := len(b.Slices) - 1
+		remaining := cap(b.Slices[ndx]) - len(b.Slices[ndx])
+
+		if remaining == 0 {
+			b.Slices = append(b.Slices, allocChunk())
+			ndx = len(b.Slices) - 1
+			remaining = cap(b.Slices[ndx]) - len(b.Slices[ndx])
+		}
+
+		chunkSize := remaining
+		if chunkSize > len(data) {
+			chunkSize = len(data)
+		}
+
+		b.Slices[ndx] = append(b.Slices[ndx], data[0:chunkSize]...)
+		data = data[chunkSize:]
+	}
+}
+
+// NewWriteBuffer creates new write buffer.
+func NewWriteBuffer() *WriteBuffer {
+	return &WriteBuffer{}
+}
diff --git a/internal/gather/gather_write_buffer_chunk.go b/internal/gather/gather_write_buffer_chunk.go
new file mode 100644
index 00000000..9a4ed28a
--- /dev/null
+++ b/internal/gather/gather_write_buffer_chunk.go
@@ -0,0 +1,42 @@
+package gather
+
+import (
+	"sync"
+)
+
+const chunkSize = 1 << 20 // 1MB chunks
+
+var (
+	freeListMutex         sync.Mutex
+	freeList              [][]byte
+	freeListHighWaterMark int
+)
+
+func allocChunk() []byte {
+	freeListMutex.Lock()
+	defer freeListMutex.Unlock()
+
+	l := len(freeList)
+	if l == 0 {
+		return make([]byte, 0, chunkSize)
+	}
+
+	ch := freeList[l-1]
+	freeList = freeList[0 : l-1]
+
+	return ch
+}
+
+func releaseChunk(s []byte) {
+	if cap(s) != chunkSize {
+		return
+	}
+
+	freeListMutex.Lock()
+	defer freeListMutex.Unlock()
+
+	freeList = append(freeList, s[:0])
+	if len(freeList) > freeListHighWaterMark {
+		freeListHighWaterMark = len(freeList)
+	}
+}
diff --git a/repo/blob/azure/azure_storage.go b/repo/blob/azure/azure_storage.go
index e5514eef..7db6ef47 100644
--- a/repo/blob/azure/azure_storage.go
+++ b/repo/blob/azure/azure_storage.go
@@ -2,7 +2,6 @@
 package azure
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"io"
@@ -105,11 +104,11 @@ func translateError(err error) error {
 	return err
 }
 
-func (az *azStorage) PutBlob(ctx context.Context, b blob.ID, data []byte) error {
+func (az *azStorage) PutBlob(ctx context.Context, b blob.ID, data blob.Bytes) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	throttled, err := az.uploadThrottler.AddReader(ioutil.NopCloser(bytes.NewReader(data)))
+	throttled, err := az.uploadThrottler.AddReader(ioutil.NopCloser(data.Reader()))
 	if err != nil {
 		return err
 	}
diff --git a/repo/blob/filesystem/filesystem_storage.go b/repo/blob/filesystem/filesystem_storage.go
index af14599b..86d8d984 100644
--- a/repo/blob/filesystem/filesystem_storage.go
+++ b/repo/blob/filesystem/filesystem_storage.go
@@ -97,7 +97,7 @@ func (fs *fsImpl) GetBlobFromPath(ctx context.Context, dirPath, path string, off
 	return b, nil
 }
 
-func (fs *fsImpl) PutBlobInPath(ctx context.Context, dirPath, path string, data []byte) error {
+func (fs *fsImpl) PutBlobInPath(ctx context.Context, dirPath, path string, data blob.Bytes) error {
 	return retry.WithExponentialBackoffNoValue(ctx, "PutBlobInPath:"+path, func() error {
 		randSuffix := make([]byte, 8)
 		if _, err := rand.Read(randSuffix); err != nil {
@@ -106,9 +106,11 @@ func (fs *fsImpl) PutBlobInPath(ctx context.Context, dirPath, path string, data
 
 		progressCallback := blob.ProgressCallback(ctx)
 
+		combinedLength := data.Length()
+
 		if progressCallback != nil {
-			progressCallback(path, 0, int64(len(data)))
-			defer progressCallback(path, int64(len(data)), int64(len(data)))
+			progressCallback(path, 0, int64(combinedLength))
+			defer progressCallback(path, int64(combinedLength), int64(combinedLength))
 		}
 
 		tempFile := fmt.Sprintf("%s.tmp.%x", path, randSuffix)
@@ -118,7 +120,7 @@ func (fs *fsImpl) PutBlobInPath(ctx context.Context, dirPath, path string, data
 			return errors.Wrap(err, "cannot create temporary file")
 		}
 
-		if _, err = f.Write(data); err != nil {
+		if _, err = data.WriteTo(f); err != nil {
 			return errors.Wrap(err, "can't write temporary file")
 		}
 
diff --git a/repo/blob/gcs/gcs_storage.go b/repo/blob/gcs/gcs_storage.go
index 2b2a46f5..e68611a5 100644
--- a/repo/blob/gcs/gcs_storage.go
+++ b/repo/blob/gcs/gcs_storage.go
@@ -2,7 +2,6 @@
 package gcs
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -100,7 +99,7 @@ func translateError(err error) error {
 		return errors.Wrap(err, "unexpected GCS error")
 	}
 }
-func (gcs *gcsStorage) PutBlob(ctx context.Context, b blob.ID, data []byte) error {
+func (gcs *gcsStorage) PutBlob(ctx context.Context, b blob.ID, data blob.Bytes) error {
 	ctx, cancel := context.WithCancel(ctx)
 
 	obj := gcs.bucket.Object(gcs.getObjectNameString(b))
@@ -108,20 +107,21 @@ func (gcs *gcsStorage) PutBlob(ctx context.Context, b blob.ID, data []byte) erro
 	writer.ChunkSize = writerChunkSize
 	writer.ContentType = "application/x-kopia"
 
+	combinedLength := data.Length()
 	progressCallback := blob.ProgressCallback(ctx)
 
 	if progressCallback != nil {
-		progressCallback(string(b), 0, int64(len(data)))
-		defer progressCallback(string(b), int64(len(data)), int64(len(data)))
+		progressCallback(string(b), 0, int64(combinedLength))
+		defer progressCallback(string(b), int64(combinedLength), int64(combinedLength))
 
 		writer.ProgressFunc = func(completed int64) {
-			if completed != int64(len(data)) {
-				progressCallback(string(b), completed, int64(len(data)))
+			if completed != int64(combinedLength) {
+				progressCallback(string(b), completed, int64(combinedLength))
 			}
 		}
 	}
 
-	_, err := iocopy.Copy(writer, bytes.NewReader(data))
+	_, err := iocopy.Copy(writer, data.Reader())
 	if err != nil {
 		// cancel context before closing the writer causes it to abandon the upload.
 		cancel()
diff --git a/repo/blob/logging/logging_storage.go b/repo/blob/logging/logging_storage.go
index de343e9c..cda9c570 100644
--- a/repo/blob/logging/logging_storage.go
+++ b/repo/blob/logging/logging_storage.go
@@ -30,11 +30,11 @@ func (s *loggingStorage) GetBlob(ctx context.Context, id blob.ID, offset, length
 	return result, err
 }
 
-func (s *loggingStorage) PutBlob(ctx context.Context, id blob.ID, data []byte) error {
+func (s *loggingStorage) PutBlob(ctx context.Context, id blob.ID, data blob.Bytes) error {
 	t0 := time.Now()
 	err := s.base.PutBlob(ctx, id, data)
 	dt := time.Since(t0)
-	s.printf(s.prefix+"PutBlob(%q,len=%v)=%#v took %v", id, len(data), err, dt)
+	s.printf(s.prefix+"PutBlob(%q,len=%v)=%#v took %v", id, data.Length(), err, dt)
 
 	return err
 }
diff --git a/repo/blob/s3/s3_storage.go b/repo/blob/s3/s3_storage.go
index 2a1a21f2..864a49f2 100644
--- a/repo/blob/s3/s3_storage.go
+++ b/repo/blob/s3/s3_storage.go
@@ -113,22 +113,24 @@ func translateError(err error) error {
 	return err
 }
 
-func (s *s3Storage) PutBlob(ctx context.Context, b blob.ID, data []byte) error {
+func (s *s3Storage) PutBlob(ctx context.Context, b blob.ID, data blob.Bytes) error {
 	return translateError(retry.WithExponentialBackoffNoValue(ctx, fmt.Sprintf("PutBlob(%v)", b), func() error {
-		throttled, err := s.uploadThrottler.AddReader(ioutil.NopCloser(bytes.NewReader(data)))
+		throttled, err := s.uploadThrottler.AddReader(ioutil.NopCloser(data.Reader()))
 		if err != nil {
 			return err
 		}
 
+		combinedLength := data.Length()
+
 		progressCallback := blob.ProgressCallback(ctx)
 		if progressCallback != nil {
-			progressCallback(string(b), 0, int64(len(data)))
-			defer progressCallback(string(b), int64(len(data)), int64(len(data)))
+			progressCallback(string(b), 0, int64(combinedLength))
+			defer progressCallback(string(b), int64(combinedLength), int64(combinedLength))
 		}
 
-		n, err := s.cli.PutObject(s.BucketName, s.getObjectNameString(b), throttled, int64(len(data)), minio.PutObjectOptions{
+		n, err := s.cli.PutObject(s.BucketName, s.getObjectNameString(b), throttled, int64(combinedLength), minio.PutObjectOptions{
 			ContentType: "application/x-kopia",
-			Progress:    newProgressReader(progressCallback, string(b), int64(len(data))),
+			Progress:    newProgressReader(progressCallback, string(b), int64(combinedLength)),
 		})
 
 		if err == io.EOF && n == 0 {
diff --git a/repo/blob/sftp/sftp_storage.go b/repo/blob/sftp/sftp_storage.go
index 0325d771..d134b382 100644
--- a/repo/blob/sftp/sftp_storage.go
+++ b/repo/blob/sftp/sftp_storage.go
@@ -80,17 +80,18 @@ func (s *sftpImpl) GetBlobFromPath(ctx context.Context, dirPath, fullPath string
 	return data[0:length], nil
 }
 
-func (s *sftpImpl) PutBlobInPath(ctx context.Context, dirPath, fullPath string, data []byte) error {
+func (s *sftpImpl) PutBlobInPath(ctx context.Context, dirPath, fullPath string, data blob.Bytes) error {
 	randSuffix := make([]byte, 8)
 	if _, err := rand.Read(randSuffix); err != nil {
 		return errors.Wrap(err, "can't get random bytes")
 	}
 
 	progressCallback := blob.ProgressCallback(ctx)
+	combinedLength := data.Length()
 
 	if progressCallback != nil {
-		progressCallback(fullPath, 0, int64(len(data)))
-		defer progressCallback(fullPath, int64(len(data)), int64(len(data)))
+		progressCallback(fullPath, 0, int64(combinedLength))
+		defer progressCallback(fullPath, int64(combinedLength), int64(combinedLength))
 	}
 
 	tempFile := fmt.Sprintf("%s.tmp.%x", fullPath, randSuffix)
@@ -100,7 +101,7 @@ func (s *sftpImpl) PutBlobInPath(ctx context.Context, dirPath, fullPath string,
 		return errors.Wrap(err, "cannot create temporary file")
 	}
 
-	if _, err = f.Write(data); err != nil {
+	if _, err = data.WriteTo(f); err != nil {
 		return errors.Wrap(err, "can't write temporary file")
 	}
 
diff --git a/repo/blob/sharded/sharded.go b/repo/blob/sharded/sharded.go
index 46a89ee9..9b503af3 100644
--- a/repo/blob/sharded/sharded.go
+++ b/repo/blob/sharded/sharded.go
@@ -15,7 +15,7 @@
 // Impl must be implemented by underlying provided.
 type Impl interface {
 	GetBlobFromPath(ctx context.Context, dirPath, filePath string, offset, length int64) ([]byte, error)
-	PutBlobInPath(ctx context.Context, dirPath, filePath string, data []byte) error
+	PutBlobInPath(ctx context.Context, dirPath, filePath string, dataSlices blob.Bytes) error
 	DeleteBlobInPath(ctx context.Context, dirPath, filePath string) error
 	ReadDir(ctx context.Context, path string) ([]os.FileInfo, error)
 }
@@ -93,7 +93,7 @@ func (s Storage) ListBlobs(ctx context.Context, prefix blob.ID, callback func(bl
 }
 
 // PutBlob implements blob.Storage
-func (s Storage) PutBlob(ctx context.Context, blobID blob.ID, data []byte) error {
+func (s Storage) PutBlob(ctx context.Context, blobID blob.ID, data blob.Bytes) error {
 	dirPath, filePath := s.GetShardedPathAndFilePath(blobID)
 
 	return s.Impl.PutBlobInPath(ctx, dirPath, filePath, data)
diff --git a/repo/blob/storage.go b/repo/blob/storage.go
index 77834529..8fb5365e 100644
--- a/repo/blob/storage.go
+++ b/repo/blob/storage.go
@@ -2,6 +2,7 @@
 
 import (
 	"context"
+	"io"
 	"log"
 	"sync"
 	"time"
@@ -9,6 +10,15 @@
 	"github.com/pkg/errors"
 )
 
+// Bytes encapsulates a sequence of bytes, possibly stored in a non-contiguous buffers,
+// which can be written sequentially or treated as a io.Reader.
+type Bytes interface {
+	io.WriterTo
+
+	Length() int
+	Reader() io.Reader
+}
+
 // Storage encapsulates API for connecting to blob storage.
 //
 // The underlying storage system must provide:
@@ -22,8 +32,8 @@
 // The required semantics are provided by existing commercial cloud storage products (Google Cloud, AWS, Azure).
 type Storage interface {
 	// PutBlob uploads the blob with given data to the repository or replaces existing blob with the provided
-	// id with given contents.
-	PutBlob(ctx context.Context, blobID ID, data []byte) error
+	// id with contents gathered from the specified list of slices.
+	PutBlob(ctx context.Context, blobID ID, data Bytes) error
 
 	// DeleteBlob removes the blob from storage. Future Get() operations will fail with ErrNotFound.
 	DeleteBlob(ctx context.Context, blobID ID) error
diff --git a/repo/blob/webdav/webdav_storage.go b/repo/blob/webdav/webdav_storage.go
index 1cb9c5c5..5d0746e5 100644
--- a/repo/blob/webdav/webdav_storage.go
+++ b/repo/blob/webdav/webdav_storage.go
@@ -2,6 +2,7 @@
 package webdav
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"math/rand"
@@ -106,11 +107,17 @@ func (d *davStorageImpl) ReadDir(ctx context.Context, dir string) ([]os.FileInfo
 	return nil, err
 }
 
-func (d *davStorageImpl) PutBlobInPath(ctx context.Context, dirPath, filePath string, data []byte) error {
+func (d *davStorageImpl) PutBlobInPath(ctx context.Context, dirPath, filePath string, data blob.Bytes) error {
 	tmpPath := fmt.Sprintf("%v-%v", filePath, rand.Int63())
 
+	var buf bytes.Buffer
+
+	data.WriteTo(&buf) // nolint:errcheck
+
+	b := buf.Bytes()
+
 	if err := d.translateError(retry.WithExponentialBackoffNoValue(ctx, "Write", func() error {
-		return d.cli.Write(tmpPath, data, defaultFilePerm)
+		return d.cli.Write(tmpPath, b, defaultFilePerm)
 	}, isRetriable)); err != nil {
 		if err != blob.ErrBlobNotFound {
 			return err
@@ -121,7 +128,7 @@ func (d *davStorageImpl) PutBlobInPath(ctx context.Context, dirPath, filePath st
 		}, isRetriable)
 
 		if err := d.translateError(retry.WithExponentialBackoffNoValue(ctx, "Write", func() error {
-			return d.cli.Write(tmpPath, data, defaultFilePerm)
+			return d.cli.Write(tmpPath, b, defaultFilePerm)
 		}, isRetriable)); err != nil {
 			return err
 		}
diff --git a/repo/content/content_cache.go b/repo/content/content_cache.go
index 63c33329..ad89e7bc 100644
--- a/repo/content/content_cache.go
+++ b/repo/content/content_cache.go
@@ -12,6 +12,7 @@
 	"go.opencensus.io/stats"
 
 	"github.com/kopia/kopia/internal/ctxutil"
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/internal/hmac"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/blob/filesystem"
@@ -87,7 +88,7 @@ func (c *contentCache) getContent(ctx context.Context, cacheKey cacheKey, blobID
 		if puterr := c.cacheStorage.PutBlob(
 			blob.WithUploadProgressCallback(ctx, nil),
 			blob.ID(cacheKey),
-			hmac.Append(b, c.hmacSecret),
+			gather.FromSlice(hmac.Append(b, c.hmacSecret)),
 		); puterr != nil {
 			stats.Record(ctx, metricContentCacheStoreErrors.M(1))
 			log(ctx).Warningf("unable to write cache item %v: %v", cacheKey, puterr)
diff --git a/repo/content/content_index_recovery.go b/repo/content/content_index_recovery.go
index a752d440..795a30c5 100644
--- a/repo/content/content_index_recovery.go
+++ b/repo/content/content_index_recovery.go
@@ -8,6 +8,7 @@
 
 	"github.com/pkg/errors"
 
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/repo/blob"
 )
 
@@ -166,9 +167,9 @@ func (bm *lockFreeManager) buildLocalIndex(pending packIndexBuilder) ([]byte, er
 }
 
 // writePackFileIndexRecoveryData appends data designed to help with recovery of pack index in case it gets damaged or lost.
-func (bm *lockFreeManager) writePackFileIndexRecoveryData(buf *bytes.Buffer, pending packIndexBuilder) error {
+func (bm *lockFreeManager) writePackFileIndexRecoveryData(buf *gather.WriteBuffer, pending packIndexBuilder) error {
 	// build, encrypt and append local index
-	localIndexOffset := buf.Len()
+	localIndexOffset := buf.Length()
 
 	localIndex, err := bm.buildLocalIndex(pending)
 	if err != nil {
@@ -188,14 +189,14 @@ func (bm *lockFreeManager) writePackFileIndexRecoveryData(buf *bytes.Buffer, pen
 		localIndexLength: uint32(len(encryptedLocalIndex)),
 	}
 
-	writeToBuffer(buf, encryptedLocalIndex)
+	buf.Append(encryptedLocalIndex)
 
 	postambleBytes, err := postamble.toBytes()
 	if err != nil {
 		return err
 	}
 
-	writeToBuffer(buf, postambleBytes)
+	buf.Append(postambleBytes)
 
 	return nil
 }
diff --git a/repo/content/content_manager.go b/repo/content/content_manager.go
index cf303632..776b223b 100644
--- a/repo/content/content_manager.go
+++ b/repo/content/content_manager.go
@@ -16,6 +16,7 @@
 	"go.opencensus.io/stats"
 
 	"github.com/kopia/kopia/internal/buf"
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/logging"
 )
@@ -83,7 +84,6 @@ type Manager struct {
 	disableIndexFlushCount int
 	flushPackIndexesAfter  time.Time // time when those indexes should be flushed
 	closed                 chan struct{}
-	bufferPool             sync.Pool
 
 	lockFreeManager
 }
@@ -91,9 +91,9 @@ type Manager struct {
 type pendingPackInfo struct {
 	prefix           blob.ID
 	packBlobID       blob.ID
-	currentPackItems map[ID]Info   // contents that are in the pack content currently being built (all inline)
-	currentPackData  *bytes.Buffer // total length of all items in the current pack content
-	finalized        bool          // indicates whether currentPackData has local index appended to it
+	currentPackItems map[ID]Info         // contents that are in the pack content currently being built (all inline)
+	currentPackData  *gather.WriteBuffer // total length of all items in the current pack content
+	finalized        bool                // indicates whether currentPackData has local index appended to it
 }
 
 // DeleteContent marks the given contentID as deleted.
@@ -195,7 +195,7 @@ func (bm *Manager) addToPackUnlocked(ctx context.Context, contentID ID, data []b
 		Deleted:          isDeleted,
 		ID:               contentID,
 		PackBlobID:       pp.packBlobID,
-		PackOffset:       uint32(pp.currentPackData.Len()),
+		PackOffset:       uint32(pp.currentPackData.Length()),
 		TimestampSeconds: bm.timeNow().Unix(),
 		FormatVersion:    byte(bm.writeFormatVersion),
 	}
@@ -204,11 +204,11 @@ func (bm *Manager) addToPackUnlocked(ctx context.Context, contentID ID, data []b
 		return errors.Wrapf(err, "unable to encrypt %q", contentID)
 	}
 
-	info.Length = uint32(pp.currentPackData.Len()) - info.PackOffset
+	info.Length = uint32(pp.currentPackData.Length()) - info.PackOffset
 
 	pp.currentPackItems[contentID] = info
 
-	shouldWrite := pp.currentPackData.Len() >= bm.maxPackSize
+	shouldWrite := pp.currentPackData.Length() >= bm.maxPackSize
 	if shouldWrite {
 		// we're about to write to storage without holding a lock
 		// remove from pendingPacks so other goroutine tries to mess with this pending pack.
@@ -360,6 +360,8 @@ func (bm *Manager) writePackAndAddToIndex(ctx context.Context, pp *pendingPackIn
 			bm.packIndexBuilder.Add(*info)
 		}
 
+		pp.currentPackData.Close()
+
 		return nil
 	}
 
@@ -375,12 +377,12 @@ func (bm *Manager) prepareAndWritePackInternal(ctx context.Context, pp *pendingP
 		return nil, errors.Wrap(err, "error preparing data content")
 	}
 
-	if pp.currentPackData.Len() > 0 {
-		if err := bm.writePackFileNotLocked(ctx, pp.packBlobID, pp.currentPackData.Bytes()); err != nil {
+	if pp.currentPackData.Length() > 0 {
+		if err := bm.writePackFileNotLocked(ctx, pp.packBlobID, pp.currentPackData.Bytes); err != nil {
 			return nil, errors.Wrap(err, "can't save pack data content")
 		}
 
-		formatLog(ctx).Debugf("wrote pack file: %v (%v bytes)", pp.packBlobID, pp.currentPackData.Len())
+		formatLog(ctx).Debugf("wrote pack file: %v (%v bytes)", pp.packBlobID, pp.currentPackData.Length())
 	}
 
 	return packFileIndex, nil
@@ -474,15 +476,14 @@ func packPrefixForContentID(contentID ID) blob.ID {
 
 func (bm *Manager) getOrCreatePendingPackInfoLocked(prefix blob.ID) (*pendingPackInfo, error) {
 	if bm.pendingPacks[prefix] == nil {
-		b := bm.bufferPool.Get().(*bytes.Buffer)
-		b.Reset()
+		b := gather.NewWriteBuffer()
 
 		contentID := make([]byte, 16)
 		if _, err := cryptorand.Read(contentID); err != nil {
 			return nil, errors.Wrap(err, "unable to read crypto bytes")
 		}
 
-		writeToBuffer(b, bm.repositoryFormatBytes)
+		b.Append(bm.repositoryFormatBytes)
 
 		if err := writeRandomBytesToBuffer(b, rand.Intn(bm.maxPreambleLength-bm.minPreambleLength+1)+bm.minPreambleLength); err != nil {
 			return nil, errors.Wrap(err, "unable to prepare content preamble")
@@ -708,11 +709,6 @@ func newManagerWithOptions(ctx context.Context, st blob.Storage, f *FormattingOp
 		pendingPacks:          map[blob.ID]*pendingPackInfo{},
 		packIndexBuilder:      make(packIndexBuilder),
 		closed:                make(chan struct{}),
-		bufferPool: sync.Pool{
-			New: func() interface{} {
-				return &bytes.Buffer{}
-			},
-		},
 	}
 
 	if err := m.CompactIndexes(ctx, autoCompactionOptions); err != nil {
diff --git a/repo/content/content_manager_lock_free.go b/repo/content/content_manager_lock_free.go
index dc4d9745..5124a57c 100644
--- a/repo/content/content_manager_lock_free.go
+++ b/repo/content/content_manager_lock_free.go
@@ -14,6 +14,7 @@
 	"github.com/pkg/errors"
 
 	"github.com/kopia/kopia/internal/buf"
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/encryption"
 	"github.com/kopia/kopia/repo/hashing"
@@ -50,7 +51,7 @@ type lockFreeManager struct {
 	encryptionBufferPool *buf.Pool
 }
 
-func (bm *lockFreeManager) maybeEncryptContentDataForPacking(output *bytes.Buffer, data []byte, contentID ID) error {
+func (bm *lockFreeManager) maybeEncryptContentDataForPacking(output *gather.WriteBuffer, data []byte, contentID ID) error {
 	var hashOutput [maxHashSize]byte
 
 	iv, err := getPackedContentIV(hashOutput[:], contentID)
@@ -66,24 +67,19 @@ func (bm *lockFreeManager) maybeEncryptContentDataForPacking(output *bytes.Buffe
 		return errors.Wrap(err, "unable to encrypt")
 	}
 
-	writeToBuffer(output, cipherText)
+	output.Append(cipherText)
 
 	return nil
 }
 
-func writeToBuffer(b *bytes.Buffer, data []byte) {
-	// buffer writes never fail, safe to ignore errors
-	_, _ = b.Write(data)
-}
-
-func writeRandomBytesToBuffer(b *bytes.Buffer, count int) error {
+func writeRandomBytesToBuffer(b *gather.WriteBuffer, count int) error {
 	var rnd [defaultPaddingUnit]byte
 
 	if _, err := io.ReadFull(cryptorand.Reader, rnd[0:count]); err != nil {
 		return err
 	}
 
-	writeToBuffer(b, rnd[0:count])
+	b.Append(rnd[0:count])
 
 	return nil
 }
@@ -231,7 +227,7 @@ func (bm *lockFreeManager) getContentDataUnlocked(ctx context.Context, pp *pendi
 	var payload []byte
 
 	if pp != nil && pp.packBlobID == bi.PackBlobID {
-		payload = pp.currentPackData.Bytes()[bi.PackOffset : bi.PackOffset+bi.Length]
+		payload = pp.currentPackData.AppendSectionTo(nil, int(bi.PackOffset), int(bi.Length))
 	} else {
 		var err error
 
@@ -277,7 +273,7 @@ func (bm *lockFreeManager) decryptAndVerify(encrypted, iv []byte) ([]byte, error
 }
 
 func (bm *lockFreeManager) preparePackDataContent(ctx context.Context, pp *pendingPackInfo) (packIndexBuilder, error) {
-	formatLog(ctx).Debugf("preparing content data with %v items (contents %v)", len(pp.currentPackItems), pp.currentPackData.Len())
+	formatLog(ctx).Debugf("preparing content data with %v items (contents %v)", len(pp.currentPackItems), pp.currentPackData.Length())
 
 	packFileIndex := packIndexBuilder{}
 	haveContent := false
@@ -307,7 +303,7 @@ func (bm *lockFreeManager) preparePackDataContent(ctx context.Context, pp *pendi
 	pp.finalized = true
 
 	if bm.paddingUnit > 0 {
-		if missing := bm.paddingUnit - (pp.currentPackData.Len() % bm.paddingUnit); missing > 0 {
+		if missing := bm.paddingUnit - (pp.currentPackData.Length() % bm.paddingUnit); missing > 0 {
 			if err := writeRandomBytesToBuffer(pp.currentPackData, missing); err != nil {
 				return nil, errors.Wrap(err, "unable to prepare content postamble")
 			}
@@ -370,8 +366,8 @@ func getIndexBlobIV(s blob.ID) ([]byte, error) {
 	return hex.DecodeString(string(s[len(s)-(aes.BlockSize*2):]))
 }
 
-func (bm *lockFreeManager) writePackFileNotLocked(ctx context.Context, packFile blob.ID, data []byte) error {
-	bm.Stats.wroteContent(len(data))
+func (bm *lockFreeManager) writePackFileNotLocked(ctx context.Context, packFile blob.ID, data gather.Bytes) error {
+	bm.Stats.wroteContent(data.Length())
 	bm.listCache.deleteListCache()
 
 	return bm.st.PutBlob(ctx, packFile, data)
@@ -398,7 +394,7 @@ func (bm *lockFreeManager) encryptAndWriteBlobNotLocked(ctx context.Context, dat
 	bm.Stats.wroteContent(len(data2))
 	bm.listCache.deleteListCache()
 
-	if err := bm.st.PutBlob(ctx, blobID, data2); err != nil {
+	if err := bm.st.PutBlob(ctx, blobID, gather.FromSlice(data2)); err != nil {
 		return "", err
 	}
 
diff --git a/repo/format_block.go b/repo/format_block.go
index cbd2efd8..6486d094 100644
--- a/repo/format_block.go
+++ b/repo/format_block.go
@@ -1,7 +1,6 @@
 package repo
 
 import (
-	"bytes"
 	"context"
 	"crypto/aes"
 	"crypto/cipher"
@@ -13,6 +12,7 @@
 
 	"github.com/pkg/errors"
 
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/repo/blob"
 )
 
@@ -155,15 +155,15 @@ func verifyFormatBlobChecksum(b []byte) ([]byte, bool) {
 }
 
 func writeFormatBlob(ctx context.Context, st blob.Storage, f *formatBlob) error {
-	var buf bytes.Buffer
-	e := json.NewEncoder(&buf)
+	buf := gather.NewWriteBuffer()
+	e := json.NewEncoder(buf)
 	e.SetIndent("", "  ")
 
 	if err := e.Encode(f); err != nil {
 		return errors.Wrap(err, "unable to marshal format blob")
 	}
 
-	if err := st.PutBlob(ctx, FormatBlobID, buf.Bytes()); err != nil {
+	if err := st.PutBlob(ctx, FormatBlobID, buf.Bytes); err != nil {
 		return errors.Wrap(err, "unable to write format blob")
 	}
 
diff --git a/repo/open.go b/repo/open.go
index b3f1c43f..4b657689 100644
--- a/repo/open.go
+++ b/repo/open.go
@@ -1,12 +1,14 @@
 package repo
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"io/ioutil"
 	"path/filepath"
 	"time"
 
+	"github.com/natefinch/atomic"
 	"github.com/pkg/errors"
 
 	"github.com/kopia/kopia/repo/blob"
@@ -196,7 +198,7 @@ func readAndCacheFormatBlobBytes(ctx context.Context, st blob.Storage, cacheDire
 	}
 
 	if cacheDirectory != "" {
-		if err := ioutil.WriteFile(cachedFile, b, 0600); err != nil {
+		if err := atomic.WriteFile(cachedFile, bytes.NewReader(b)); err != nil {
 			log(ctx).Warningf("warning: unable to write cache: %v", err)
 		}
 	}
diff --git a/snapshot/policy/policy_manager.go b/snapshot/policy/policy_manager.go
index f2061bcd..9d18e313 100644
--- a/snapshot/policy/policy_manager.go
+++ b/snapshot/policy/policy_manager.go
@@ -98,29 +98,32 @@ func GetDefinedPolicy(ctx context.Context, rep *repo.Repository, si snapshot.Sou
 		return nil, ErrPolicyNotFound
 	}
 
-	if len(md) == 1 {
-		p := &Policy{}
+	// arbitrality pick first pick ID to return in case there's more than one
+	// this is possible when two repository clients independently create manifests at approximately the same time
+	// so it should not really matter which one we pick.
+	// see https://github.com/kopia/kopia/issues/391
+	manifestID := md[0].ID
 
-		err := rep.Manifests.Get(ctx, md[0].ID, p)
-		if err == manifest.ErrNotFound {
-			return nil, ErrPolicyNotFound
-		}
+	p := &Policy{}
 
-		if err != nil {
-			return nil, err
-		}
+	err = rep.Manifests.Get(ctx, manifestID, p)
 
-		em, err := rep.Manifests.GetMetadata(ctx, md[0].ID)
-		if err != nil {
+	if err != nil {
+		if err == manifest.ErrNotFound {
 			return nil, ErrPolicyNotFound
 		}
 
-		p.Labels = em.Labels
+		return nil, err
+	}
 
-		return p, nil
+	em, err := rep.Manifests.GetMetadata(ctx, manifestID)
+	if err != nil {
+		return nil, ErrPolicyNotFound
 	}
 
-	return nil, errors.New("ambiguous policy")
+	p.Labels = em.Labels
+
+	return p, nil
 }
 
 // SetPolicy sets the policy on a given source.


Test patch:
diff --git a/internal/blobtesting/concurrent.go b/internal/blobtesting/concurrent.go
index ea7c42b7..343a0edc 100644
--- a/internal/blobtesting/concurrent.go
+++ b/internal/blobtesting/concurrent.go
@@ -10,6 +10,7 @@
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/internal/testlogging"
 	"github.com/kopia/kopia/repo/blob"
 )
@@ -87,7 +88,7 @@ func VerifyConcurrentAccess(t testingT, st blob.Storage, options ConcurrentAcces
 			for i := 0; i < options.Iterations; i++ {
 				blobID := randomBlobID()
 				data := fmt.Sprintf("%v-%v", blobID, rand.Int63())
-				err := st.PutBlob(ctx, blobID, []byte(data))
+				err := st.PutBlob(ctx, blobID, gather.FromSlice([]byte(data)))
 				switch err {
 				case nil:
 					// clean success
diff --git a/internal/blobtesting/faulty.go b/internal/blobtesting/faulty.go
index c1417001..a87358e6 100644
--- a/internal/blobtesting/faulty.go
+++ b/internal/blobtesting/faulty.go
@@ -38,8 +38,8 @@ func (s *FaultyStorage) GetBlob(ctx context.Context, id blob.ID, offset, length
 }
 
 // PutBlob implements blob.Storage
-func (s *FaultyStorage) PutBlob(ctx context.Context, id blob.ID, data []byte) error {
-	if err := s.getNextFault(ctx, "PutBlob", id, len(data)); err != nil {
+func (s *FaultyStorage) PutBlob(ctx context.Context, id blob.ID, data blob.Bytes) error {
+	if err := s.getNextFault(ctx, "PutBlob", id); err != nil {
 		return err
 	}
 
diff --git a/internal/blobtesting/map.go b/internal/blobtesting/map.go
index ba13be5a..bdd7acde 100644
--- a/internal/blobtesting/map.go
+++ b/internal/blobtesting/map.go
@@ -1,6 +1,7 @@
 package blobtesting
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"sort"
@@ -48,7 +49,7 @@ func (s *mapStorage) GetBlob(ctx context.Context, id blob.ID, offset, length int
 	return nil, blob.ErrBlobNotFound
 }
 
-func (s *mapStorage) PutBlob(ctx context.Context, id blob.ID, data []byte) error {
+func (s *mapStorage) PutBlob(ctx context.Context, id blob.ID, data blob.Bytes) error {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
@@ -58,7 +59,11 @@ func (s *mapStorage) PutBlob(ctx context.Context, id blob.ID, data []byte) error
 
 	s.keyTime[id] = s.timeNow()
 
-	s.data[id] = append([]byte{}, data...)
+	var b bytes.Buffer
+
+	data.WriteTo(&b) //nolint:errcheck
+
+	s.data[id] = b.Bytes()
 
 	return nil
 }
diff --git a/internal/blobtesting/verify.go b/internal/blobtesting/verify.go
index 9efceb03..7aabdb24 100644
--- a/internal/blobtesting/verify.go
+++ b/internal/blobtesting/verify.go
@@ -5,6 +5,7 @@
 	"context"
 	"reflect"
 
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/repo/blob"
 )
 
@@ -30,7 +31,7 @@ func VerifyStorage(ctx context.Context, t testingT, r blob.Storage) {
 
 	// Now add blocks.
 	for _, b := range blocks {
-		if err := r.PutBlob(ctx, b.blk, b.contents); err != nil {
+		if err := r.PutBlob(ctx, b.blk, gather.FromSlice(b.contents)); err != nil {
 			t.Errorf("can't put blob: %v", err)
 		}
 
@@ -42,7 +43,7 @@ func VerifyStorage(ctx context.Context, t testingT, r blob.Storage) {
 
 	// Overwrite blocks.
 	for _, b := range blocks {
-		if err := r.PutBlob(ctx, b.blk, b.contents); err != nil {
+		if err := r.PutBlob(ctx, b.blk, gather.FromSlice(b.contents)); err != nil {
 			t.Errorf("can't put blob: %v", err)
 		}
 
diff --git a/internal/gather/gather_bytes_test.go b/internal/gather/gather_bytes_test.go
new file mode 100644
index 00000000..9cf3a8a6
--- /dev/null
+++ b/internal/gather/gather_bytes_test.go
@@ -0,0 +1,109 @@
+package gather
+
+import (
+	"bytes"
+	"io/ioutil"
+	"testing"
+)
+
+var sample1 = []byte("hello! how are you? nice to meet you.")
+
+func TestGatherBytes(t *testing.T) {
+	// split the 'whole' into equivalent Bytes slicings in some interesting ways
+	cases := []struct {
+		whole  []byte
+		sliced Bytes
+	}{
+		{
+			whole:  nil,
+			sliced: Bytes{},
+		},
+		{
+			whole: nil,
+			sliced: Bytes{Slices: [][]byte{
+				nil,
+			}},
+		},
+		{
+			whole: nil,
+			sliced: Bytes{Slices: [][]byte{
+				nil,
+				{},
+				nil,
+			}},
+		},
+		{
+			whole:  sample1,
+			sliced: FromSlice(sample1),
+		},
+		{
+			whole: sample1,
+			sliced: Bytes{Slices: [][]byte{
+				nil,
+				sample1,
+				nil,
+			}},
+		},
+		{
+			whole: sample1,
+			sliced: Bytes{Slices: [][]byte{
+				sample1[0:20],
+				sample1[20:],
+			}},
+		},
+		{
+			whole: sample1,
+			sliced: Bytes{Slices: [][]byte{
+				sample1[0:20],
+				nil, // zero-length
+				{},  // zero-length
+				sample1[20:],
+			}},
+		},
+		{
+			whole: sample1,
+			sliced: Bytes{Slices: [][]byte{
+				sample1[0:10],
+				sample1[10:25],
+				sample1[25:30],
+				sample1[30:31],
+				sample1[31:],
+			}},
+		},
+	}
+
+	for _, tc := range cases {
+		b := tc.sliced
+
+		// length
+		if got, want := b.Length(), len(tc.whole); got != want {
+			t.Errorf("unexpected length: %v, want %v", got, want)
+		}
+
+		// reader
+		all, err := ioutil.ReadAll(b.Reader())
+		if err != nil {
+			t.Errorf("unable to read: %v", err)
+		}
+
+		if !bytes.Equal(all, tc.whole) {
+			t.Errorf("unexpected data read %v, want %v", string(all), string(tc.whole))
+		}
+
+		// GetBytes
+		all = b.GetBytes(nil)
+		if !bytes.Equal(all, tc.whole) {
+			t.Errorf("unexpected data from GetBytes() %v, want %v", string(all), string(tc.whole))
+		}
+
+		// AppendSectionTo - test exhaustively all combinationf os start, length
+		for i := 0; i <= len(tc.whole); i++ {
+			for j := i; j <= len(tc.whole); j++ {
+				result := b.AppendSectionTo(nil, i, j-i)
+				if !bytes.Equal(result, tc.whole[i:j]) {
+					t.Fatalf("invalid section")
+				}
+			}
+		}
+	}
+}
diff --git a/internal/gather/gather_write_buffer_chunk_test.go b/internal/gather/gather_write_buffer_chunk_test.go
new file mode 100644
index 00000000..f3c4cec7
--- /dev/null
+++ b/internal/gather/gather_write_buffer_chunk_test.go
@@ -0,0 +1,58 @@
+package gather
+
+import (
+	"bytes"
+	"testing"
+)
+
+func TestWriteBufferChunk(t *testing.T) {
+	// reset for testing
+	freeList = nil
+	freeListHighWaterMark = 0
+
+	chunk1 := allocChunk()
+	_ = append(chunk1, []byte("chunk1")...)
+
+	if got, want := len(chunk1), 0; got != want {
+		t.Errorf("invalid chunk len: %v, want %v", got, want)
+	}
+
+	if got, want := cap(chunk1), chunkSize; got != want {
+		t.Errorf("invalid chunk cap: %v, want %v", got, want)
+	}
+
+	if got, want := freeListHighWaterMark, 0; got != want {
+		t.Errorf("unexpected high water mark %v, want %v", got, want)
+	}
+
+	chunk2 := allocChunk()
+	_ = append(chunk2, []byte("chunk2")...)
+
+	if got, want := freeListHighWaterMark, 0; got != want {
+		t.Errorf("unexpected high water mark %v, want %v", got, want)
+	}
+
+	releaseChunk(chunk2)
+
+	if got, want := freeListHighWaterMark, 1; got != want {
+		t.Errorf("unexpected high water mark %v, want %v", got, want)
+	}
+
+	releaseChunk(chunk1)
+
+	if got, want := freeListHighWaterMark, 2; got != want {
+		t.Errorf("unexpected high water mark %v, want %v", got, want)
+	}
+
+	// allocate chunk3 - make sure we got the same slice as chunk1 (LIFO)
+	chunk3 := allocChunk()
+	if got, want := chunk3[0:6], []byte("chunk1"); !bytes.Equal(got, want) {
+		t.Errorf("got wrong chunk data %q, want %q", string(got), string(want))
+	}
+
+	// allocate chunk4 - make sure we got the same slice as chunk1 (LIFO)
+	chunk4 := allocChunk()
+	if got, want := chunk4[0:6], []byte("chunk2"); !bytes.Equal(got, want) {
+		t.Errorf("got wrong chunk data %q, want %q", string(got), string(want))
+	}
+}
diff --git a/internal/gather/gather_write_buffer_test.go b/internal/gather/gather_write_buffer_test.go
new file mode 100644
index 00000000..afdfe8c9
--- /dev/null
+++ b/internal/gather/gather_write_buffer_test.go
@@ -0,0 +1,67 @@
+package gather
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+)
+
+func TestGatherWriteBuffer(t *testing.T) {
+	// reset for testing
+	freeList = nil
+	freeListHighWaterMark = 0
+
+	w := NewWriteBuffer()
+	defer w.Close()
+
+	w.Append([]byte("hello "))
+	fmt.Fprintf(w, "world!")
+
+	if got, want := w.GetBytes(nil), []byte("hello world!"); !bytes.Equal(got, want) {
+		t.Errorf("invaldi bytes %v, want %v", string(got), string(want))
+	}
+
+	if got, want := len(w.Slices), 1; got != want {
+		t.Errorf("invalid number of slices %v, want %v", got, want)
+	}
+
+	w.Append(bytes.Repeat([]byte("x"), chunkSize))
+
+	if got, want := w.Length(), chunkSize+12; got != want {
+		t.Errorf("invalid length: %v, want %v", got, want)
+	}
+
+	// one more slice was allocated
+	if got, want := len(w.Slices), 2; got != want {
+		t.Errorf("invalid number of slices %v, want %v", got, want)
+	}
+
+	// write to fill the remainder of 2nd slice
+	w.Append(bytes.Repeat([]byte("x"), chunkSize-12))
+
+	// still 2 slices
+	if got, want := len(w.Slices), 2; got != want {
+		t.Errorf("invalid number of slices %v, want %v", got, want)
+	}
+
+	// one more byte allocates new slice
+	w.Append([]byte("x"))
+
+	// still 3 slices
+	if got, want := len(w.Slices), 3; got != want {
+		t.Errorf("invalid number of slices %v, want %v", got, want)
+	}
+
+	w.Reset()
+}
+
+func TestGatherDefaultWriteBuffer(t *testing.T) {
+	var w WriteBuffer
+
+	// one more byte allocates new slice
+	w.Append([]byte("x"))
+
+	if got, want := len(w.Slices), 1; got != want {
+		t.Errorf("invalid number of slices %v, want %v", got, want)
+	}
+}
diff --git a/internal/repotesting/repotesting.go b/internal/repotesting/repotesting.go
index 9e185e8c..f989098b 100644
--- a/internal/repotesting/repotesting.go
+++ b/internal/repotesting/repotesting.go
@@ -123,6 +123,16 @@ func (e *Environment) MustReopen(t *testing.T) {
 	}
 }
 
+// MustOpenAnother opens another repository backend by the same storage.
+func (e *Environment) MustOpenAnother(t *testing.T) *repo.Repository {
+	rep2, err := repo.Open(testlogging.Context(t), e.configFile(), masterPassword, &repo.Options{})
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	return rep2
+}
+
 // VerifyBlobCount verifies that the underlying storage contains the specified number of blobs.
 func (e *Environment) VerifyBlobCount(t *testing.T, want int) {
 	var got int
diff --git a/repo/blob/filesystem/filesystem_storage_test.go b/repo/blob/filesystem/filesystem_storage_test.go
index 0e67884f..773f524c 100644
--- a/repo/blob/filesystem/filesystem_storage_test.go
+++ b/repo/blob/filesystem/filesystem_storage_test.go
@@ -12,6 +12,7 @@
 	"github.com/kopia/kopia/repo/blob"
 
 	"github.com/kopia/kopia/internal/blobtesting"
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/internal/testlogging"
 )
 
@@ -74,11 +75,11 @@ func TestFileStorageTouch(t *testing.T) {
 	}
 
 	fs := r.(*fsStorage)
-	assertNoError(t, fs.PutBlob(ctx, t1, []byte{1}))
+	assertNoError(t, fs.PutBlob(ctx, t1, gather.FromSlice([]byte{1})))
 	time.Sleep(1 * time.Second) // sleep a bit to accommodate Apple filesystems with low timestamp resolution
-	assertNoError(t, fs.PutBlob(ctx, t2, []byte{1}))
+	assertNoError(t, fs.PutBlob(ctx, t2, gather.FromSlice([]byte{1})))
 	time.Sleep(1 * time.Second)
-	assertNoError(t, fs.PutBlob(ctx, t3, []byte{1}))
+	assertNoError(t, fs.PutBlob(ctx, t3, gather.FromSlice([]byte{1})))
 
 	verifyBlobTimestampOrder(t, fs, t1, t2, t3)
 
diff --git a/repo/blob/storage_test.go b/repo/blob/storage_test.go
index fa8c401f..bf001d4c 100644
--- a/repo/blob/storage_test.go
+++ b/repo/blob/storage_test.go
@@ -5,6 +5,7 @@
 	"time"
 
 	"github.com/kopia/kopia/internal/blobtesting"
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/internal/testlogging"
 	"github.com/kopia/kopia/repo/blob"
 )
@@ -13,9 +14,9 @@ func TestListAllBlobsConsistent(t *testing.T) {
 	ctx := testlogging.Context(t)
 	data := blobtesting.DataMap{}
 	st := blobtesting.NewMapStorage(data, nil, time.Now)
-	st.PutBlob(ctx, "foo1", []byte{1, 2, 3}) //nolint:errcheck
-	st.PutBlob(ctx, "foo2", []byte{1, 2, 3}) //nolint:errcheck
-	st.PutBlob(ctx, "foo3", []byte{1, 2, 3}) //nolint:errcheck
+	st.PutBlob(ctx, "foo1", gather.FromSlice([]byte{1, 2, 3})) //nolint:errcheck
+	st.PutBlob(ctx, "foo2", gather.FromSlice([]byte{1, 2, 3})) //nolint:errcheck
+	st.PutBlob(ctx, "foo3", gather.FromSlice([]byte{1, 2, 3})) //nolint:errcheck
 
 	// set up faulty storage that will add a blob while a scan is in progress.
 	f := &blobtesting.FaultyStorage{
@@ -23,7 +24,7 @@ func TestListAllBlobsConsistent(t *testing.T) {
 		Faults: map[string][]*blobtesting.Fault{
 			"ListBlobsItem": {
 				{ErrCallback: func() error {
-					st.PutBlob(ctx, "foo0", []byte{1, 2, 3}) //nolint:errcheck
+					st.PutBlob(ctx, "foo0", gather.FromSlice([]byte{1, 2, 3})) //nolint:errcheck
 					return nil
 				}},
 			},
diff --git a/repo/content/content_cache_test.go b/repo/content/content_cache_test.go
index d0543e9e..01ab68c8 100644
--- a/repo/content/content_cache_test.go
+++ b/repo/content/content_cache_test.go
@@ -14,6 +14,7 @@
 	"github.com/pkg/errors"
 
 	"github.com/kopia/kopia/internal/blobtesting"
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/internal/testlogging"
 	"github.com/kopia/kopia/repo/blob"
 )
@@ -22,8 +23,8 @@ func newUnderlyingStorageForContentCacheTesting(t *testing.T) blob.Storage {
 	ctx := testlogging.Context(t)
 	data := blobtesting.DataMap{}
 	st := blobtesting.NewMapStorage(data, nil, nil)
-	assertNoError(t, st.PutBlob(ctx, "content-1", []byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}))
-	assertNoError(t, st.PutBlob(ctx, "content-4k", bytes.Repeat([]byte{1, 2, 3, 4}, 1000))) // 4000 bytes
+	assertNoError(t, st.PutBlob(ctx, "content-1", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10})))
+	assertNoError(t, st.PutBlob(ctx, "content-4k", gather.FromSlice(bytes.Repeat([]byte{1, 2, 3, 4}, 1000)))) // 4000 bytes
 
 	return st
 }
@@ -166,7 +167,7 @@ func verifyContentCache(t *testing.T, cache *contentCache) {
 		// corrupt the data and write back
 		d[0] ^= 1
 
-		if puterr := cache.cacheStorage.PutBlob(ctx, cacheKey, d); puterr != nil {
+		if puterr := cache.cacheStorage.PutBlob(ctx, cacheKey, gather.FromSlice(d)); puterr != nil {
 			t.Fatalf("unable to write corrupted content: %v", puterr)
 		}
 
diff --git a/repo/format_block_test.go b/repo/format_block_test.go
index 472199ef..9865cbdc 100644
--- a/repo/format_block_test.go
+++ b/repo/format_block_test.go
@@ -6,6 +6,7 @@
 	"testing"
 
 	"github.com/kopia/kopia/internal/blobtesting"
+	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/internal/testlogging"
 	"github.com/kopia/kopia/repo/blob"
 )
@@ -26,19 +27,19 @@ func TestFormatBlobRecovery(t *testing.T) {
 		t.Errorf("unexpected checksummed length: %v, want %v", got, want)
 	}
 
-	assertNoError(t, st.PutBlob(ctx, "some-blob-by-itself", checksummed))
-	assertNoError(t, st.PutBlob(ctx, "some-blob-suffix", append(append([]byte(nil), 1, 2, 3), checksummed...)))
-	assertNoError(t, st.PutBlob(ctx, "some-blob-prefix", append(append([]byte(nil), checksummed...), 1, 2, 3)))
+	assertNoError(t, st.PutBlob(ctx, "some-blob-by-itself", gather.FromSlice(checksummed)))
+	assertNoError(t, st.PutBlob(ctx, "some-blob-suffix", gather.FromSlice(append(append([]byte(nil), 1, 2, 3), checksummed...))))
+	assertNoError(t, st.PutBlob(ctx, "some-blob-prefix", gather.FromSlice(append(append([]byte(nil), checksummed...), 1, 2, 3))))
 
 	// mess up checksum
 	checksummed[len(checksummed)-3] ^= 1
-	assertNoError(t, st.PutBlob(ctx, "bad-checksum", checksummed))
-	assertNoError(t, st.PutBlob(ctx, "zero-len", []byte{}))
-	assertNoError(t, st.PutBlob(ctx, "one-len", []byte{1}))
-	assertNoError(t, st.PutBlob(ctx, "two-len", []byte{1, 2}))
-	assertNoError(t, st.PutBlob(ctx, "three-len", []byte{1, 2, 3}))
-	assertNoError(t, st.PutBlob(ctx, "four-len", []byte{1, 2, 3, 4}))
-	assertNoError(t, st.PutBlob(ctx, "five-len", []byte{1, 2, 3, 4, 5}))
+	assertNoError(t, st.PutBlob(ctx, "bad-checksum", gather.FromSlice(checksummed)))
+	assertNoError(t, st.PutBlob(ctx, "zero-len", gather.FromSlice([]byte{})))
+	assertNoError(t, st.PutBlob(ctx, "one-len", gather.FromSlice([]byte{1})))
+	assertNoError(t, st.PutBlob(ctx, "two-len", gather.FromSlice([]byte{1, 2})))
+	assertNoError(t, st.PutBlob(ctx, "three-len", gather.FromSlice([]byte{1, 2, 3})))
+	assertNoError(t, st.PutBlob(ctx, "four-len", gather.FromSlice([]byte{1, 2, 3, 4})))
+	assertNoError(t, st.PutBlob(ctx, "five-len", gather.FromSlice([]byte{1, 2, 3, 4, 5})))
 
 	cases := []struct {
 		blobID blob.ID
diff --git a/snapshot/policy/policy_manager_test.go b/snapshot/policy/policy_manager_test.go
new file mode 100644
index 00000000..c5e617c0
--- /dev/null
+++ b/snapshot/policy/policy_manager_test.go
@@ -0,0 +1,52 @@
+package policy
+
+import (
+	"context"
+	"testing"
+
+	"github.com/kopia/kopia/internal/repotesting"
+)
+
+func TestPolicyManager(t *testing.T) {
+	ctx := context.Background()
+
+	var env repotesting.Environment
+
+	defer env.Setup(t).Close(ctx, t)
+
+	r1 := env.Repository
+	r2 := env.MustOpenAnother(t)
+	sourceInfo := GlobalPolicySourceInfo
+
+	must(t, SetPolicy(ctx, r1, sourceInfo, &Policy{
+		RetentionPolicy: RetentionPolicy{
+			KeepDaily: intPtr(44),
+		},
+	}))
+
+	must(t, SetPolicy(ctx, r2, sourceInfo, &Policy{
+		RetentionPolicy: RetentionPolicy{
+			KeepDaily: intPtr(33),
+		},
+	}))
+
+	must(t, r1.Flush(ctx))
+	must(t, r2.Flush(ctx))
+
+	r3 := env.MustOpenAnother(t)
+
+	pi, err := GetDefinedPolicy(ctx, r3, sourceInfo)
+	must(t, err)
+
+	if got := *pi.RetentionPolicy.KeepDaily; got != 33 && got != 44 {
+		t.Errorf("unexpected policy returned")
+	}
+}
+
+func must(t *testing.T, err error) {
+	t.Helper()
+
+	if err != nil {
+		t.Fatal(err)
+	}
+}