repository: implement pack compression

internal/repository/index.go

@@ -75,12 +75,12 @@ const maxuint32 = 1<<32 - 1
 
 func (idx *Index) store(packIndex int, blob restic.Blob) {
 	// assert that offset and length fit into uint32!
-	if blob.Offset > maxuint32 || blob.Length > maxuint32 {
+	if blob.Offset > maxuint32 || blob.Length > maxuint32 || blob.UncompressedLength > maxuint32 {
 		panic("offset or length does not fit in uint32. You have packs > 4GB!")
 	}
 
 	m := &idx.byType[blob.Type]
-	m.add(blob.ID, packIndex, uint32(blob.Offset), uint32(blob.Length))
+	m.add(blob.ID, packIndex, uint32(blob.Offset), uint32(blob.Length), uint32(blob.UncompressedLength))
 }
 
 // Final returns true iff the index is already written to the repository, it is
@@ -169,8 +169,9 @@ func (idx *Index) toPackedBlob(e *indexEntry, t restic.BlobType) restic.PackedBl
 			BlobHandle: restic.BlobHandle{
 				ID:   e.id,
 				Type: t},
-			Length: uint(e.length),
-			Offset: uint(e.offset),
+			Length:             uint(e.length),
+			Offset:             uint(e.offset),
+			UncompressedLength: uint(e.uncompressedLength),
 		},
 		PackID: idx.packs[e.packIndex],
 	}
@@ -225,6 +226,9 @@ func (idx *Index) LookupSize(bh restic.BlobHandle) (plaintextLength uint, found
 	if e == nil {
 		return 0, false
 	}
+	if e.uncompressedLength != 0 {
+		return uint(e.uncompressedLength), true
+	}
 	return uint(restic.PlaintextLength(int(e.length))), true
 }
 
@@ -357,10 +361,11 @@ type packJSON struct {
 }
 
 type blobJSON struct {
-	ID     restic.ID       `json:"id"`
-	Type   restic.BlobType `json:"type"`
-	Offset uint            `json:"offset"`
-	Length uint            `json:"length"`
+	ID                 restic.ID       `json:"id"`
+	Type               restic.BlobType `json:"type"`
+	Offset             uint            `json:"offset"`
+	Length             uint            `json:"length"`
+	UncompressedLength uint            `json:"uncompressed_length,omitempty"`
 }
 
 // generatePackList returns a list of packs.
@@ -391,10 +396,11 @@ func (idx *Index) generatePackList() ([]*packJSON, error) {
 
 			// add blob
 			p.Blobs = append(p.Blobs, blobJSON{
-				ID:     e.id,
-				Type:   restic.BlobType(typ),
-				Offset: uint(e.offset),
-				Length: uint(e.length),
+				ID:                 e.id,
+				Type:               restic.BlobType(typ),
+				Offset:             uint(e.offset),
+				Length:             uint(e.length),
+				UncompressedLength: uint(e.uncompressedLength),
 			})
 
 			return true
@@ -553,7 +559,7 @@ func (idx *Index) merge(idx2 *Index) error {
 		m2.foreach(func(e2 *indexEntry) bool {
 			if !hasIdenticalEntry(e2) {
 				// packIndex needs to be changed as idx2.pack was appended to idx.pack, see above
-				m.add(e2.id, e2.packIndex+packlen, e2.offset, e2.length)
+				m.add(e2.id, e2.packIndex+packlen, e2.offset, e2.length, e2.uncompressedLength)
 			}
 			return true
 		})
@@ -601,8 +607,9 @@ func DecodeIndex(buf []byte, id restic.ID) (idx *Index, oldFormat bool, err erro
 				BlobHandle: restic.BlobHandle{
 					Type: blob.Type,
 					ID:   blob.ID},
-				Offset: blob.Offset,
-				Length: blob.Length,
+				Offset:             blob.Offset,
+				Length:             blob.Length,
+				UncompressedLength: blob.UncompressedLength,
 			})
 
 			switch blob.Type {
@@ -648,6 +655,7 @@ func decodeOldIndex(buf []byte) (idx *Index, err error) {
 					ID:   blob.ID},
 				Offset: blob.Offset,
 				Length: blob.Length,
+				// no compressed length in the old index format
 			})
 
 			switch blob.Type {

internal/repository/indexmap.go

@@ -32,7 +32,7 @@ const (
 
 // add inserts an indexEntry for the given arguments into the map,
 // using id as the key.
-func (m *indexMap) add(id restic.ID, packIdx int, offset, length uint32) {
+func (m *indexMap) add(id restic.ID, packIdx int, offset, length uint32, uncompressedLength uint32) {
 	switch {
 	case m.numentries == 0: // Lazy initialization.
 		m.init()
@@ -47,6 +47,7 @@ func (m *indexMap) add(id restic.ID, packIdx int, offset, length uint32) {
 	e.packIndex = packIdx
 	e.offset = offset
 	e.length = length
+	e.uncompressedLength = uncompressedLength
 
 	m.buckets[h] = e
 	m.numentries++
@@ -152,9 +153,10 @@ func (m *indexMap) newEntry() *indexEntry {
 }
 
 type indexEntry struct {
-	id        restic.ID
-	next      *indexEntry
-	packIndex int // Position in containing Index's packs field.
-	offset    uint32
-	length    uint32
+	id                 restic.ID
+	next               *indexEntry
+	packIndex          int // Position in containing Index's packs field.
+	offset             uint32
+	length             uint32
+	uncompressedLength uint32
 }

internal/repository/indexmap_test.go

@@ -22,7 +22,7 @@ func TestIndexMapBasic(t *testing.T) {
 		r.Read(id[:])
 		rtest.Assert(t, m.get(id) == nil, "%v retrieved but not added", id)
 
-		m.add(id, 0, 0, 0)
+		m.add(id, 0, 0, 0, 0)
 		rtest.Assert(t, m.get(id) != nil, "%v added but not retrieved", id)
 		rtest.Equals(t, uint(i), m.len())
 	}
@@ -41,7 +41,7 @@ func TestIndexMapForeach(t *testing.T) {
 	for i := 0; i < N; i++ {
 		var id restic.ID
 		id[0] = byte(i)
-		m.add(id, i, uint32(i), uint32(i))
+		m.add(id, i, uint32(i), uint32(i), uint32(i/2))
 	}
 
 	seen := make(map[int]struct{})
@@ -51,6 +51,7 @@ func TestIndexMapForeach(t *testing.T) {
 		rtest.Equals(t, i, e.packIndex)
 		rtest.Equals(t, i, int(e.length))
 		rtest.Equals(t, i, int(e.offset))
+		rtest.Equals(t, i/2, int(e.uncompressedLength))
 
 		seen[i] = struct{}{}
 		return true
@@ -85,13 +86,13 @@ func TestIndexMapForeachWithID(t *testing.T) {
 
 	// Test insertion and retrieval of duplicates.
 	for i := 0; i < ndups; i++ {
-		m.add(id, i, 0, 0)
+		m.add(id, i, 0, 0, 0)
 	}
 
 	for i := 0; i < 100; i++ {
 		var otherid restic.ID
 		r.Read(otherid[:])
-		m.add(otherid, -1, 0, 0)
+		m.add(otherid, -1, 0, 0, 0)
 	}
 
 	n = 0
@@ -109,7 +110,7 @@ func TestIndexMapForeachWithID(t *testing.T) {
 
 func BenchmarkIndexMapHash(b *testing.B) {
 	var m indexMap
-	m.add(restic.ID{}, 0, 0, 0) // Trigger lazy initialization.
+	m.add(restic.ID{}, 0, 0, 0, 0) // Trigger lazy initialization.
 
 	ids := make([]restic.ID, 128) // 4 KiB.
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))

internal/repository/packer_manager_test.go

@@ -70,7 +70,7 @@ func fillPacks(t testing.TB, rnd *rand.Rand, be Saver, pm *packerManager, buf []
 		// Only change a few bytes so we know we're not benchmarking the RNG.
 		rnd.Read(buf[:min(l, 4)])
 
-		n, err := packer.Add(restic.DataBlob, id, buf)
+		n, err := packer.Add(restic.DataBlob, id, buf, 0)
 		if err != nil {
 			t.Fatal(err)
 		}

internal/repository/repository.go

@@ -235,12 +235,23 @@ func (r *Repository) LoadBlob(ctx context.Context, t restic.BlobType, id restic.
 			continue
 		}
 
+		if blob.IsCompressed() {
+			plaintext, err = r.dec.DecodeAll(plaintext, make([]byte, 0, blob.DataLength()))
+			if err != nil {
+				lastError = errors.Errorf("decompressing blob %v failed: %v", id, err)
+				continue
+			}
+		}
+
 		// check hash
 		if !restic.Hash(plaintext).Equal(id) {
 			lastError = errors.Errorf("blob %v returned invalid hash", id)
 			continue
 		}
 
+		if len(plaintext) > cap(buf) {
+			return plaintext, nil
+		}
 		// move decrypted data to the start of the buffer
 		copy(buf, plaintext)
 		return buf[:len(plaintext)], nil
@@ -275,6 +286,12 @@ func (r *Repository) LookupBlobSize(id restic.ID, tpe restic.BlobType) (uint, bo
 func (r *Repository) saveAndEncrypt(ctx context.Context, t restic.BlobType, data []byte, id restic.ID) error {
 	debug.Log("save id %v (%v, %d bytes)", id, t, len(data))
 
+	uncompressedLength := 0
+	if r.cfg.Version > 1 {
+		uncompressedLength = len(data)
+		data = r.enc.EncodeAll(data, nil)
+	}
+
 	nonce := crypto.NewRandomNonce()
 
 	ciphertext := make([]byte, 0, restic.CiphertextLength(len(data)))
@@ -301,7 +318,7 @@ func (r *Repository) saveAndEncrypt(ctx context.Context, t restic.BlobType, data
 	}
 
 	// save ciphertext
-	_, err = packer.Add(t, id, ciphertext)
+	_, err = packer.Add(t, id, ciphertext, uncompressedLength)
 	if err != nil {
 		return err
 	}
@@ -536,6 +553,17 @@ func (r *Repository) LoadIndex(ctx context.Context) error {
 		return err
 	}
 
+	if r.cfg.Version < 2 {
+		// sanity check
+		ctx, cancel := context.WithCancel(ctx)
+		defer cancel()
+		for blob := range r.idx.Each(ctx) {
+			if blob.IsCompressed() {
+				return errors.Fatal("index uses feature not supported by repository version 1")
+			}
+		}
+	}
+
 	// remove index files from the cache which have been removed in the repo
 	return r.PrepareCache(validIndex)
 }
@@ -834,9 +862,15 @@ func StreamPack(ctx context.Context, beLoad BackendLoadFn, key *crypto.Key, pack
 
 	debug.Log("streaming pack %v (%d to %d bytes), blobs: %v", packID, dataStart, dataEnd, len(blobs))
 
+	dec, err := zstd.NewReader(nil)
+	if err != nil {
+		panic(dec)
+	}
+	defer dec.Close()
+
 	ctx, cancel := context.WithCancel(ctx)
 	// stream blobs in pack
-	err := beLoad(ctx, h, int(dataEnd-dataStart), int64(dataStart), func(rd io.Reader) error {
+	err = beLoad(ctx, h, int(dataEnd-dataStart), int64(dataStart), func(rd io.Reader) error {
 		// prevent callbacks after cancelation
 		if ctx.Err() != nil {
 			return ctx.Err()
@@ -849,6 +883,7 @@ func StreamPack(ctx context.Context, beLoad BackendLoadFn, key *crypto.Key, pack
 		bufRd := bufio.NewReaderSize(rd, bufferSize)
 		currentBlobEnd := dataStart
 		var buf []byte
+		var decode []byte
 		for _, entry := range blobs {
 			skipBytes := int(entry.Offset - currentBlobEnd)
 			if skipBytes < 0 {
@@ -888,6 +923,16 @@ func StreamPack(ctx context.Context, beLoad BackendLoadFn, key *crypto.Key, pack
 			// decryption errors are likely permanent, give the caller a chance to skip them
 			nonce, ciphertext := buf[:key.NonceSize()], buf[key.NonceSize():]
 			plaintext, err := key.Open(ciphertext[:0], nonce, ciphertext, nil)
+			if err == nil && entry.IsCompressed() {
+				if cap(decode) < int(entry.DataLength()) {
+					decode = make([]byte, 0, entry.DataLength())
+				}
+				decode, err = dec.DecodeAll(plaintext, decode[:0])
+				plaintext = decode
+				if err != nil {
+					err = errors.Errorf("decompressing blob %v failed: %v", h, err)
+				}
+			}
 			if err == nil {
 				id := restic.Hash(plaintext)
 				if !id.Equal(entry.ID) {