changed how index compaction works:

instead of having time-based naming, block manager will perform occasional compaction at startup time and delete unwanted blocks

The protocol is safe from concurrency standpoint and provides eventual consistency.
We only delete blocks after creating compacted blocks.
We retry loading index until (any) consistent index is fetched (possibly from cache) and all underlying blocks are also available, not necessarily the latest ones.

TODO - we need to periodically snapshot block index contents, so that if we have a bug somewhere in compaction code, we have a way of restoring working indexes.

block/block_cache.go

@@ -6,15 +6,16 @@ import (
 	"path/filepath"
 	"time"
 
-	"github.com/kopia/kopia/storage/filesystem"
-
 	"github.com/kopia/kopia/storage"
+	"github.com/kopia/kopia/storage/filesystem"
 )
 
 type blockCache interface {
 	getBlock(ctx context.Context, cacheKey string, physicalBlockID PhysicalBlockID, offset, length int64) ([]byte, error)
 	putBlock(ctx context.Context, blockID PhysicalBlockID, data []byte) error
-	listIndexBlocks(ctx context.Context, full bool, extraTime time.Duration) ([]IndexInfo, error)
+	listIndexBlocks(ctx context.Context) ([]IndexInfo, error)
+	deleteListCache(ctx context.Context)
+	deleteBlock(ctx context.Context, blockID PhysicalBlockID) error
 	close() error
 }
 

block/block_manager.go

@@ -8,7 +8,6 @@ import (
 	"encoding/hex"
 	"fmt"
 	"io"
-	"math"
 	"math/rand"
 	"os"
 	"path/filepath"
@@ -25,19 +24,22 @@ import (
 )
 
 const (
-	parallelFetches              = 5                // number of parallel reads goroutines
-	flushPackIndexTimeout        = 10 * time.Minute // time after which all pending indexes are flushes
-	newIndexBlockPrefix          = "n"
-	compactedBlockSuffix         = "-z"
-	defaultMinPreambleLength     = 32
-	defaultMaxPreambleLength     = 32
-	defaultPaddingUnit           = 4096
-	autoCompactionBlockCount     = 16
+	parallelFetches             = 5                // number of parallel reads goroutines
+	flushPackIndexTimeout       = 10 * time.Minute // time after which all pending indexes are flushes
+	newIndexBlockPrefix         = "n"
+	defaultMinPreambleLength    = 32
+	defaultMaxPreambleLength    = 32
+	defaultPaddingUnit          = 4096
+	autoCompactionMinBlockCount = 4 * parallelFetches
+	autoCompactionMaxBlockCount = 64
+
 	defaultActiveBlocksExtraTime = 10 * time.Minute
 
 	currentWriteVersion     = 1
 	minSupportedReadVersion = 0
 	maxSupportedReadVersion = currentWriteVersion
+
+	indexLoadAttempts = 10
 )
 
 // Info is an information about a single block managed by Manager.
@@ -244,12 +246,12 @@ func (bm *Manager) flushPackIndexesLocked(ctx context.Context) error {
 		data := buf.Bytes()
 		dataCopy := append([]byte(nil), data...)
 
-		indexBlockID, err := bm.writePackIndexesNew(ctx, data, false)
+		indexBlockID, err := bm.writePackIndexesNew(ctx, data)
 		if err != nil {
 			return err
 		}
 
-		if err := bm.committedBlocks.addBlock(indexBlockID, dataCopy); err != nil {
+		if err := bm.committedBlocks.addBlock(indexBlockID, dataCopy, true); err != nil {
 			return fmt.Errorf("unable to add committed block: %v", err)
 		}
 		bm.packIndexBuilder = packindex.NewBuilder()
@@ -259,14 +261,8 @@ func (bm *Manager) flushPackIndexesLocked(ctx context.Context) error {
 	return nil
 }
 
-func (bm *Manager) writePackIndexesNew(ctx context.Context, data []byte, isCompaction bool) (PhysicalBlockID, error) {
-	var suffix string
-	if isCompaction {
-		suffix = compactedBlockSuffix
-	}
-
-	inverseTimePrefix := fmt.Sprintf("%016x", math.MaxInt64-bm.timeNow().UnixNano())
-	return bm.encryptAndWriteBlockNotLocked(ctx, data, newIndexBlockPrefix+inverseTimePrefix, suffix)
+func (bm *Manager) writePackIndexesNew(ctx context.Context, data []byte) (PhysicalBlockID, error) {
+	return bm.encryptAndWriteBlockNotLocked(ctx, data, newIndexBlockPrefix)
 }
 
 func (bm *Manager) finishPackLocked(ctx context.Context) error {
@@ -370,29 +366,12 @@ func appendRandomBytes(b []byte, count int) ([]byte, error) {
 	return append(b, rnd...), nil
 }
 
-// ListIndexBlocks returns the list of all index blocks, including inactive, sorted by time.
-func (bm *Manager) ListIndexBlocks(ctx context.Context) ([]IndexInfo, error) {
-	blocks, err := bm.cache.listIndexBlocks(ctx, true, 0)
-	if err != nil {
-		return nil, fmt.Errorf("error listing index blocks: %v", err)
-	}
-
-	sortBlocksByTime(blocks)
-	return blocks, nil
-}
-
-// ActiveIndexBlocks returns the list of active index blocks, sorted by time.
-func (bm *Manager) ActiveIndexBlocks(ctx context.Context) ([]IndexInfo, error) {
-	blocks, err := bm.cache.listIndexBlocks(ctx, false, bm.activeBlocksExtraTime)
+// IndexBlocks returns the list of active index blocks, sorted by time.
+func (bm *Manager) IndexBlocks(ctx context.Context) ([]IndexInfo, error) {
+	blocks, err := bm.cache.listIndexBlocks(ctx)
 	if err != nil {
 		return nil, err
 	}
-	// for i, b := range blocks {
-	// 	log.Printf("found block #%v %v %v %v", i, b.BlockID, b.Length, b.Timestamp)
-	// }
-	if len(blocks) == 0 {
-		return nil, nil
-	}
 
 	sortBlocksByTime(blocks)
 	return blocks, nil
@@ -404,19 +383,52 @@ func sortBlocksByTime(b []IndexInfo) {
 	})
 }
 
-func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]PhysicalBlockID, error) {
-	blocks, err := bm.ActiveIndexBlocks(ctx)
-	if err != nil {
-		return nil, err
+func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]IndexInfo, error) {
+	nextSleepTime := 100 * time.Millisecond
+
+	for i := 0; i < indexLoadAttempts; i++ {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+
+		if i > 0 {
+			bm.cache.deleteListCache(ctx)
+			log.Printf("encountered NOT_FOUND when loading, sleeping %v before retrying #%v", nextSleepTime, i)
+			time.Sleep(nextSleepTime)
+			nextSleepTime *= 2
+		}
+
+		blocks, err := bm.cache.listIndexBlocks(ctx)
+		if err != nil {
+			return nil, err
+		}
+
+		err = bm.tryLoadPackIndexBlocksLocked(ctx, blocks)
+		if err == nil {
+			var blockIDs []PhysicalBlockID
+			for _, b := range blocks {
+				blockIDs = append(blockIDs, b.BlockID)
+			}
+			if err = bm.committedBlocks.use(blockIDs); err != nil {
+				return nil, err
+			}
+			return blocks, nil
+		}
+		if err != storage.ErrBlockNotFound {
+			return nil, err
+		}
 	}
 
-	indexBlockIDs := indexBlockIDs(blocks)
-	ch, err := bm.unprocessedIndexBlocks(indexBlockIDs)
+	return nil, fmt.Errorf("unable to load pack indexes despite %v retries", indexLoadAttempts)
+}
+
+func (bm *Manager) tryLoadPackIndexBlocksLocked(ctx context.Context, blocks []IndexInfo) error {
+	ch, err := bm.unprocessedIndexBlocks(blocks)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	if len(ch) == 0 {
-		return indexBlockIDs, nil
+		return nil
 	}
 
 	var wg sync.WaitGroup
@@ -435,8 +447,8 @@ func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]PhysicalBlockID
 					return
 				}
 
-				if err := bm.loadPackIndexes(indexBlockID, data); err != nil {
-					errors <- err
+				if err := bm.committedBlocks.addBlock(indexBlockID, data, false); err != nil {
+					errors <- fmt.Errorf("unable to add to committed block cache: %v", err)
 					return
 				}
 			}
@@ -448,66 +460,89 @@ func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]PhysicalBlockID
 
 	// Propagate async errors, if any.
 	for err := range errors {
-		return nil, err
+		return err
 	}
 
-	return indexBlockIDs, nil
+	return nil
 }
 
 // unprocessedIndexBlocks returns a closed channel filled with block IDs that are not in committedBlocks cache.
-func (bm *Manager) unprocessedIndexBlocks(blocks []PhysicalBlockID) (<-chan PhysicalBlockID, error) {
+func (bm *Manager) unprocessedIndexBlocks(blocks []IndexInfo) (<-chan PhysicalBlockID, error) {
 	ch := make(chan PhysicalBlockID, len(blocks))
-	for _, blockID := range blocks {
-		has, err := bm.committedBlocks.hasIndexBlockID(blockID)
+	for _, block := range blocks {
+		has, err := bm.committedBlocks.hasIndexBlockID(block.BlockID)
 		if err != nil {
 			return nil, err
 		}
 		if has {
-			log.Printf("index block %q already in cache, skipping", blockID)
+			log.Printf("index block %q already in cache, skipping", block.BlockID)
 			continue
 		}
-		ch <- blockID
+		ch <- block.BlockID
 	}
 	close(ch)
 	return ch, nil
 }
 
-func indexBlockIDs(blocks []IndexInfo) []PhysicalBlockID {
-	var indexBlockIDs []PhysicalBlockID
-	for _, b := range blocks {
-		indexBlockIDs = append(indexBlockIDs, b.BlockID)
-	}
-	return indexBlockIDs
-}
-
-func (bm *Manager) loadPackIndexes(indexBlockID PhysicalBlockID, data []byte) error {
-	if err := bm.committedBlocks.addBlock(indexBlockID, data); err != nil {
-		return fmt.Errorf("unable to add to committed block cache: %v", err)
-	}
-
-	return nil
-}
-
-func (bm *Manager) fetchCommittedIndexAndMaybeCompact(ctx context.Context, compactThresold int) error {
+// CompactIndexes performs compaction of index blocks ensuring that # of small blocks is between minSmallBlockCount and maxSmallBlockCount
+func (bm *Manager) CompactIndexes(ctx context.Context, minSmallBlockCount int, maxSmallBlockCount int) error {
 	bm.lock()
 	defer bm.unlock()
 
-	indexBlocks, err := bm.loadPackIndexesLocked(ctx)
-	if err != nil {
-		return err
+	if maxSmallBlockCount < minSmallBlockCount {
+		return fmt.Errorf("invalid block counts")
 	}
 
-	if len(indexBlocks) > compactThresold {
-		log.Info().Msgf("compacting block indexes (block count %v exceeds threshold of %v)", len(indexBlocks), compactThresold)
-		return bm.compactIndexes(ctx, indexBlocks)
+	indexBlocks, err := bm.loadPackIndexesLocked(ctx)
+	if err != nil {
+		return fmt.Errorf("error loading indexes: %v", err)
+	}
+
+	blocksToCompact := bm.getBlocksToCompact(indexBlocks, minSmallBlockCount, maxSmallBlockCount)
+
+	if err := bm.compactAndDeleteIndexBlocks(ctx, blocksToCompact); err != nil {
+		log.Warn().Msgf("error performing quick compaction: %v", err)
 	}
 
 	return nil
 }
 
-// CompactIndexes performs compaction of index blocks.
-func (bm *Manager) CompactIndexes(ctx context.Context) error {
-	return bm.fetchCommittedIndexAndMaybeCompact(ctx, 1)
+func (bm *Manager) getBlocksToCompact(indexBlocks []IndexInfo, minSmallBlockCount int, maxSmallBlockCount int) []IndexInfo {
+	var nonCompactedBlocks []IndexInfo
+	var totalSizeNonCompactedBlocks int64
+
+	var verySmallBlocks []IndexInfo
+	var totalSizeVerySmallBlocks int64
+
+	var mediumSizedBlocks []IndexInfo
+	var totalSizeMediumSizedBlocks int64
+
+	for _, b := range indexBlocks {
+		if b.Length > int64(bm.maxPackSize) {
+			continue
+		}
+
+		nonCompactedBlocks = append(nonCompactedBlocks, b)
+		if b.Length < int64(bm.maxPackSize/20) {
+			verySmallBlocks = append(verySmallBlocks, b)
+			totalSizeVerySmallBlocks += b.Length
+		} else {
+			mediumSizedBlocks = append(mediumSizedBlocks, b)
+			totalSizeMediumSizedBlocks += b.Length
+		}
+		totalSizeNonCompactedBlocks += b.Length
+	}
+
+	if len(nonCompactedBlocks) < minSmallBlockCount {
+		// current count is below min allowed - nothing to do
+		return nil
+	}
+
+	if len(verySmallBlocks) > len(nonCompactedBlocks)/2 && len(mediumSizedBlocks)+1 < minSmallBlockCount {
+		return verySmallBlocks
+	}
+
+	return nonCompactedBlocks
 }
 
 // ListBlocks returns IDs of blocks matching given prefix.
@@ -563,16 +598,16 @@ func (bm *Manager) ListBlockInfos(prefix string) ([]Info, error) {
 	return result, nil
 }
 
-func (bm *Manager) compactIndexes(ctx context.Context, indexBlocks []PhysicalBlockID) error {
+func (bm *Manager) compactAndDeleteIndexBlocks(ctx context.Context, indexBlocks []IndexInfo) error {
 	if len(indexBlocks) <= 1 {
-		log.Printf("skipping index compaction - already compacted")
 		return nil
 	}
+	log.Debug().Msgf("compacting %v blocks", len(indexBlocks))
 	t0 := time.Now()
 
 	bld := packindex.NewBuilder()
 	for _, indexBlock := range indexBlocks {
-		data, err := bm.getPhysicalBlockInternal(ctx, indexBlock)
+		data, err := bm.getPhysicalBlockInternal(ctx, indexBlock.BlockID)
 		if err != nil {
 			return err
 		}
@@ -592,9 +627,25 @@ func (bm *Manager) compactIndexes(ctx context.Context, indexBlocks []PhysicalBlo
 	if err := bld.Build(&buf); err != nil {
 		return fmt.Errorf("unable to build an index: %v", err)
 	}
-	compactedIndexBlock, err := bm.writePackIndexesNew(ctx, buf.Bytes(), true)
-	log.Info().Msgf("wrote compacted index to %v (%v bytes) in %v", compactedIndexBlock, buf.Len(), time.Since(t0))
-	return err
+
+	compactedIndexBlock, err := bm.writePackIndexesNew(ctx, buf.Bytes())
+	if err != nil {
+		return fmt.Errorf("unable to write compacted indexes: %v", err)
+	}
+
+	log.Debug().Msgf("wrote compacted index (%v bytes) in %v", compactedIndexBlock, time.Since(t0))
+
+	for _, indexBlock := range indexBlocks {
+		if indexBlock.BlockID == compactedIndexBlock {
+			continue
+		}
+
+		if err := bm.cache.deleteBlock(ctx, indexBlock.BlockID); err != nil {
+			log.Warn().Msgf("unable to delete compacted block %q: %v", indexBlock.BlockID, err)
+		}
+	}
+
+	return nil
 }
 
 // Flush completes writing any pending packs and writes pack indexes to the underlyign storage.
@@ -653,7 +704,7 @@ func validatePrefix(prefix string) error {
 func (bm *Manager) writePackDataNotLocked(ctx context.Context, data []byte) (PhysicalBlockID, error) {
 	if bm.writeFormatVersion == 0 {
 		// 0 blocks are encrypted together
-		return bm.encryptAndWriteBlockNotLocked(ctx, data, "", "")
+		return bm.encryptAndWriteBlockNotLocked(ctx, data, "")
 	}
 
 	suffix := make([]byte, 16)
@@ -672,9 +723,9 @@ func (bm *Manager) writePackDataNotLocked(ctx context.Context, data []byte) (Phy
 	return physicalBlockID, nil
 }
 
-func (bm *Manager) encryptAndWriteBlockNotLocked(ctx context.Context, data []byte, prefix string, suffix string) (PhysicalBlockID, error) {
+func (bm *Manager) encryptAndWriteBlockNotLocked(ctx context.Context, data []byte, prefix string) (PhysicalBlockID, error) {
 	hash := bm.hashData(data)
-	physicalBlockID := PhysicalBlockID(prefix + hex.EncodeToString(hash) + suffix)
+	physicalBlockID := PhysicalBlockID(prefix + hex.EncodeToString(hash))
 
 	// Encrypt the block in-place.
 	atomic.AddInt64(&bm.stats.EncryptedBytes, int64(len(data)))
@@ -898,26 +949,14 @@ type cachedList struct {
 // If 'full' is set to true, this function lists and returns all blocks,
 // if 'full' is false, the function returns only blocks from the last 2 compactions.
 // The list of blocks is not guaranteed to be sorted.
-func listIndexBlocksFromStorage(ctx context.Context, st storage.Storage, full bool, extraTime time.Duration) ([]IndexInfo, error) {
-	maxCompactions := 1
-	if full {
-		maxCompactions = math.MaxInt32
-	}
-
+func listIndexBlocksFromStorage(ctx context.Context, st storage.Storage) ([]IndexInfo, error) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	ch := st.ListBlocks(ctx, newIndexBlockPrefix)
 
 	var results []IndexInfo
-	numCompactions := 0
-
-	var timestampCutoff time.Time
 	for it := range ch {
-		if !timestampCutoff.IsZero() && it.TimeStamp.Before(timestampCutoff) {
-			break
-		}
-
 		if it.Error != nil {
 			return nil, it.Error
 		}
@@ -928,16 +967,14 @@ func listIndexBlocksFromStorage(ctx context.Context, st storage.Storage, full bo
 			Length:    it.Length,
 		}
 		results = append(results, ii)
-
-		if strings.Contains(string(ii.BlockID), compactedBlockSuffix) {
-			numCompactions++
-			if numCompactions == maxCompactions {
-				timestampCutoff = it.TimeStamp.Add(-extraTime)
-			}
-		}
 	}
 
-	return results, nil
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	default:
+		return results, nil
+	}
 }
 
 // NewManager creates new block manager with given packing options and a formatter.
@@ -997,8 +1034,8 @@ func newManagerWithOptions(ctx context.Context, st storage.Storage, f Formatting
 
 	m.startPackIndexLocked()
 
-	if err := m.fetchCommittedIndexAndMaybeCompact(ctx, autoCompactionBlockCount); err != nil {
-		return nil, fmt.Errorf("unable initialize indexes: %v", err)
+	if err := m.CompactIndexes(ctx, autoCompactionMinBlockCount, autoCompactionMaxBlockCount); err != nil {
+		return nil, fmt.Errorf("error initializing block manager: %v", err)
 	}
 
 	return m, nil

block/block_manager_test.go

@@ -29,7 +29,7 @@ const (
 var fakeTime = time.Date(2017, 1, 1, 0, 0, 0, 0, time.UTC)
 
 func init() {
-	//zerolog.SetGlobalLevel(zerolog.InfoLevel)
+	zerolog.SetGlobalLevel(zerolog.DebugLevel)
 	log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr})
 }
 
@@ -54,7 +54,7 @@ func TestBlockZeroBytes1(t *testing.T) {
 	if got, want := len(data), 2; got != want {
 		t.Errorf("unexpected number of blocks: %v, wanted %v", got, want)
 	}
-	//dumpBlockManagerData(t, data)
+	dumpBlockManagerData(t, data)
 	bm = newTestBlockManager(data, keyTime, nil)
 	verifyBlock(ctx, t, bm, blockID, []byte{})
 }
@@ -67,7 +67,6 @@ func TestBlockZeroBytes2(t *testing.T) {
 	writeBlockAndVerify(ctx, t, bm, seededRandomData(10, 10))
 	writeBlockAndVerify(ctx, t, bm, []byte{})
 	bm.Flush(ctx)
-	//dumpBlockManagerData(t, data)
 	if got, want := len(data), 2; got != want {
 		t.Errorf("unexpected number of blocks: %v, wanted %v", got, want)
 		dumpBlockManagerData(t, data)
@@ -166,7 +165,7 @@ func TestBlockManagerEmpty(t *testing.T) {
 func verifyActiveIndexBlockCount(ctx context.Context, t *testing.T, bm *Manager, expected int) {
 	t.Helper()
 
-	blks, err := bm.ActiveIndexBlocks(ctx)
+	blks, err := bm.IndexBlocks(ctx)
 	if err != nil {
 		t.Errorf("error listing active index blocks: %v", err)
 		return
@@ -318,11 +317,11 @@ func TestBlockManagerConcurrency(t *testing.T) {
 		t.Errorf("unexpected index count before compaction: %v, wanted %v", got, want)
 	}
 
-	if err := bm4.CompactIndexes(ctx); err != nil {
+	if err := bm4.CompactIndexes(ctx, 1, 1); err != nil {
 		t.Errorf("compaction error: %v", err)
 	}
-	if got, want := getIndexCount(data), 5; got != want {
-		t.Errorf("unexpected index count after partial compaction: %v, wanted %v", got, want)
+	if got, want := getIndexCount(data), 1; got != want {
+		t.Errorf("unexpected index count after compaction: %v, wanted %v", got, want)
 	}
 
 	// new block manager at this point can see all data.
@@ -332,7 +331,7 @@ func TestBlockManagerConcurrency(t *testing.T) {
 	verifyBlock(ctx, t, bm5, bm1block, seededRandomData(31, 100))
 	verifyBlock(ctx, t, bm5, bm2block, seededRandomData(32, 100))
 	verifyBlock(ctx, t, bm5, bm3block, seededRandomData(33, 100))
-	if err := bm5.CompactIndexes(ctx); err != nil {
+	if err := bm5.CompactIndexes(ctx, 1, 1); err != nil {
 		t.Errorf("compaction error: %v", err)
 	}
 }
@@ -471,7 +470,7 @@ func verifyVersionCompat(t *testing.T, writeVersion int) {
 	// make sure we can read everything
 	verifyBlockManagerDataSet(ctx, t, mgr, dataSet)
 
-	if err := mgr.CompactIndexes(ctx); err != nil {
+	if err := mgr.CompactIndexes(ctx, 1, 1); err != nil {
 		t.Fatalf("unable to compact indexes: %v", err)
 	}
 	if err := mgr.Flush(ctx); err != nil {
@@ -508,11 +507,10 @@ func newTestBlockManager(data map[string][]byte, keyTime map[string]time.Time, t
 		BlockFormat: "TESTONLY_MD5",
 		MaxPackSize: maxPackSize,
 	}, CachingOptions{}, timeFunc, 0)
-	bm.checkInvariantsOnUnlock = true
-
 	if err != nil {
 		panic("can't create block manager: " + err.Error())
 	}
+	bm.checkInvariantsOnUnlock = true
 	return bm
 }
 

block/committed_block_index.go

@@ -1,15 +1,21 @@
 package block
 
+import (
+	"github.com/kopia/kopia/internal/packindex"
+)
+
 type committedBlockIndex interface {
 	getBlock(blockID string) (Info, error)
 	listBlocks(prefix string, cb func(i Info) error) error
 
-	addBlock(indexBlockID PhysicalBlockID, indexData []byte) error
+	addBlock(indexBlockID PhysicalBlockID, indexData []byte, use bool) error
 	hasIndexBlockID(indexBlockID PhysicalBlockID) (bool, error)
+
+	use(indexBlockIDs []PhysicalBlockID) error
 }
 
 func newCommittedBlockIndex() committedBlockIndex {
 	return &inMemoryCommittedBlockIndex{
-		physicalBlocks: make(map[PhysicalBlockID]bool),
+		physicalBlocks: make(map[PhysicalBlockID]packindex.Index),
 	}
 }

block/inmemory_committed_block_index.go

@@ -2,6 +2,7 @@ package block
 
 import (
 	"bytes"
+	"fmt"
 	"sync"
 
 	"github.com/kopia/kopia/internal/packindex"
@@ -11,7 +12,7 @@ import (
 type inMemoryCommittedBlockIndex struct {
 	mu             sync.Mutex
 	indexes        packindex.Merged
-	physicalBlocks map[PhysicalBlockID]bool
+	physicalBlocks map[PhysicalBlockID]packindex.Index
 }
 
 func (b *inMemoryCommittedBlockIndex) getBlock(blockID string) (Info, error) {
@@ -31,18 +32,20 @@ func (b *inMemoryCommittedBlockIndex) getBlock(blockID string) (Info, error) {
 
 }
 
-func (b *inMemoryCommittedBlockIndex) addBlock(indexBlockID PhysicalBlockID, data []byte) error {
+func (b *inMemoryCommittedBlockIndex) addBlock(indexBlockID PhysicalBlockID, data []byte, use bool) error {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	ndx, err := packindex.Open(bytes.NewReader(data))
-	if err != nil {
-		return err
-	}
+	if b.physicalBlocks[indexBlockID] == nil {
+		ndx, err := packindex.Open(bytes.NewReader(data))
+		if err != nil {
+			return err
+		}
+		b.physicalBlocks[indexBlockID] = ndx
 
-	if !b.physicalBlocks[indexBlockID] {
-		b.physicalBlocks[indexBlockID] = true
-		b.indexes = append(b.indexes, ndx)
+		if use {
+			b.indexes = append(b.indexes, ndx)
+		}
 	}
 
 	return nil
@@ -59,5 +62,27 @@ func (b *inMemoryCommittedBlockIndex) hasIndexBlockID(indexBlockID PhysicalBlock
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	return b.physicalBlocks[indexBlockID], nil
+	return b.physicalBlocks[indexBlockID] != nil, nil
+}
+
+func (b *inMemoryCommittedBlockIndex) use(packBlockIDs []PhysicalBlockID) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	var newIndexes packindex.Merged
+	defer func() {
+		newIndexes.Close() //nolint:errcheck
+	}()
+	for _, e := range packBlockIDs {
+		ndx := b.physicalBlocks[e]
+		if ndx == nil {
+			return fmt.Errorf("unable to open pack index %q", e)
+		}
+
+		//log.Printf("opened %v with %v entries", e, ndx.EntryCount())
+		newIndexes = append(newIndexes, ndx)
+	}
+	b.indexes = newIndexes
+	newIndexes = nil
+	return nil
 }

block/local_storage_cache.go

@@ -20,8 +20,7 @@ import (
 
 const (
 	sweepCacheFrequency = 1 * time.Minute
-	fullListCacheItem   = "list-full"
-	activeListCacheItem = "list-active"
+	fullListCacheItem   = "list"
 )
 
 type localStorageCache struct {
@@ -80,30 +79,27 @@ func (c *localStorageCache) putBlock(ctx context.Context, blockID PhysicalBlockI
 	return c.st.PutBlock(ctx, string(blockID), bytes.NewReader(data))
 }
 
-func (c *localStorageCache) listIndexBlocks(ctx context.Context, full bool, extraTime time.Duration) ([]IndexInfo, error) {
-	var cachedListBlockID string
+func (c *localStorageCache) deleteBlock(ctx context.Context, blockID PhysicalBlockID) error {
+	c.deleteListCache(ctx)
+	return c.st.DeleteBlock(ctx, string(blockID))
+}
 
-	if full {
-		cachedListBlockID = fullListCacheItem
-	} else {
-		cachedListBlockID = activeListCacheItem
-	}
-
-	ci, err := c.readBlocksFromCacheBlock(ctx, cachedListBlockID)
+func (c *localStorageCache) listIndexBlocks(ctx context.Context) ([]IndexInfo, error) {
+	ci, err := c.readBlocksFromCacheBlock(ctx, fullListCacheItem)
 	if err == nil {
 		expirationTime := ci.Timestamp.Add(c.listCacheDuration)
 		if time.Now().Before(expirationTime) {
-			log.Debug().Bool("full", full).Str("blockID", cachedListBlockID).Msg("retrieved index blocks from cache")
+			log.Debug().Str("blockID", fullListCacheItem).Msg("retrieved index blocks from cache")
 			return ci.Blocks, nil
 		}
 	} else if err != storage.ErrBlockNotFound {
-		log.Warn().Str("blockID", cachedListBlockID).Err(err).Msgf("unable to open cache file")
+		log.Warn().Err(err).Msgf("unable to open cache file")
 	}
 
-	log.Debug().Bool("full", full).Msg("listing index blocks from source")
-	blocks, err := listIndexBlocksFromStorage(ctx, c.st, full, extraTime)
+	log.Debug().Msg("listing index blocks from source")
+	blocks, err := listIndexBlocksFromStorage(ctx, c.st)
 	if err == nil {
-		c.saveListToCache(ctx, cachedListBlockID, &cachedList{
+		c.saveListToCache(ctx, fullListCacheItem, &cachedList{
 			Blocks:    blocks,
 			Timestamp: time.Now(),
 		})
@@ -229,9 +225,7 @@ func (c *localStorageCache) sweepDirectory(ctx context.Context) (err error) {
 }
 
 func (c *localStorageCache) deleteListCache(ctx context.Context) {
-	for _, it := range []string{fullListCacheItem, activeListCacheItem} {
-		if err := c.cacheStorage.DeleteBlock(ctx, it); err != nil && err != storage.ErrBlockNotFound {
-			log.Warn().Err(err).Msg("unable to delete cache item")
-		}
+	if err := c.cacheStorage.DeleteBlock(ctx, fullListCacheItem); err != nil && err != storage.ErrBlockNotFound {
+		log.Warn().Err(err).Msg("unable to delete cache item")
 	}
 }

block/null_block_cache.go

@@ -3,7 +3,6 @@ package block
 import (
 	"bytes"
 	"context"
-	"time"
 
 	"github.com/kopia/kopia/storage"
 )
@@ -20,8 +19,15 @@ func (c nullBlockCache) putBlock(ctx context.Context, blockID PhysicalBlockID, d
 	return c.st.PutBlock(ctx, string(blockID), bytes.NewReader(data))
 }
 
-func (c nullBlockCache) listIndexBlocks(ctx context.Context, full bool, extraTime time.Duration) ([]IndexInfo, error) {
-	return listIndexBlocksFromStorage(ctx, c.st, full, extraTime)
+func (c nullBlockCache) listIndexBlocks(ctx context.Context) ([]IndexInfo, error) {
+	return listIndexBlocksFromStorage(ctx, c.st)
+}
+
+func (c nullBlockCache) deleteListCache(ctx context.Context) {
+}
+
+func (c nullBlockCache) deleteBlock(ctx context.Context, blockID PhysicalBlockID) error {
+	return c.st.DeleteBlock(ctx, string(blockID))
 }
 
 func (c nullBlockCache) close() error {

block/simple_committed_block_index.go

@@ -5,7 +5,6 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
-	"strings"
 	"sync"
 
 	"github.com/kopia/kopia/internal/packindex"
@@ -17,13 +16,17 @@ import (
 const simpleIndexSuffix = ".sndx"
 
 type simpleCommittedBlockIndex struct {
-	dirname      string
-	indexesMutex sync.Mutex
-	indexBlocks  map[PhysicalBlockID]bool
-	merged       packindex.Merged
+	dirname string
+
+	mu          sync.Mutex
+	indexBlocks map[PhysicalBlockID]bool
+	merged      packindex.Merged
 }
 
 func (b *simpleCommittedBlockIndex) getBlock(blockID string) (Info, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
 	info, err := b.merged.GetInfo(blockID)
 	if info != nil {
 		return *info, nil
@@ -35,75 +38,44 @@ func (b *simpleCommittedBlockIndex) getBlock(blockID string) (Info, error) {
 }
 
 func (b *simpleCommittedBlockIndex) hasIndexBlockID(indexBlockID PhysicalBlockID) (bool, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
 	return b.indexBlocks[indexBlockID], nil
 }
 
-func (b *simpleCommittedBlockIndex) addBlock(indexBlockID PhysicalBlockID, data []byte) error {
+func (b *simpleCommittedBlockIndex) addBlock(indexBlockID PhysicalBlockID, data []byte, use bool) error {
 	fullPath := filepath.Join(b.dirname, string(indexBlockID+simpleIndexSuffix))
 
 	if err := ioutil.WriteFile(fullPath, data, 0600); err != nil {
 		return err
 	}
 
-	b.indexesMutex.Lock()
-	defer b.indexesMutex.Unlock()
+	b.mu.Lock()
+	defer b.mu.Unlock()
 
-	if !b.indexBlocks[indexBlockID] {
-		ndx, err := b.openIndex(fullPath)
-		if err != nil {
-			return fmt.Errorf("unable to open pack index %q: %v", fullPath, err)
-		}
+	b.indexBlocks[indexBlockID] = true
 
-		b.indexBlocks[indexBlockID] = true
-		b.merged = append(b.merged, ndx)
+	if !use {
+		return nil
 	}
 
+	ndx, err := b.openIndex(fullPath)
+	if err != nil {
+		return fmt.Errorf("unable to open pack index %q: %v", fullPath, err)
+	}
+
+	b.merged = append(b.merged, ndx)
+
 	return nil
 }
 
 func (b *simpleCommittedBlockIndex) listBlocks(prefix string, cb func(i Info) error) error {
-	return b.merged.Iterate(prefix, cb)
-}
+	b.mu.Lock()
+	m := b.merged
+	b.mu.Unlock()
 
-func (b *simpleCommittedBlockIndex) loadIndexes() error {
-	b.indexesMutex.Lock()
-	defer b.indexesMutex.Unlock()
-
-	entries, err := ioutil.ReadDir(b.dirname)
-	if err != nil {
-		return err
-	}
-
-	newIndexes := map[PhysicalBlockID]bool{}
-	var newMerged packindex.Merged
-	defer func() {
-		newMerged.Close() //nolint:errcheck
-	}()
-	for _, e := range entries {
-		if !strings.HasSuffix(e.Name(), simpleIndexSuffix) {
-			continue
-		}
-
-		fullpath := filepath.Join(b.dirname, e.Name())
-		ndx, err := b.openIndex(fullpath)
-		if err != nil {
-			return fmt.Errorf("unable to open pack index %q: %v", fullpath, err)
-		}
-
-		log.Printf("opened %v with %v entries", fullpath, ndx.EntryCount())
-
-		// ndx.Iterate("", func(i Info) error {
-		// 	log.Info().Msgf("i: %v blk:%v off:%v len:%v", i.BlockID, i.PackBlockID, i.PackOffset, i.Length)
-		// 	return nil
-		// })
-
-		newIndexes[PhysicalBlockID(strings.TrimSuffix(e.Name(), simpleIndexSuffix))] = true
-		newMerged = append(newMerged, ndx)
-	}
-	b.indexBlocks = newIndexes
-	b.merged = newMerged
-	newMerged = nil
-	return nil
+	return m.Iterate(prefix, cb)
 }
 
 func (b *simpleCommittedBlockIndex) openIndex(fullpath string) (packindex.Index, error) {
@@ -116,12 +88,38 @@ func (b *simpleCommittedBlockIndex) openIndex(fullpath string) (packindex.Index,
 	return packindex.Open(f)
 }
 
+func (b *simpleCommittedBlockIndex) use(packBlockIDs []PhysicalBlockID) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	newIndexes := map[PhysicalBlockID]bool{}
+	var newMerged packindex.Merged
+	defer func() {
+		newMerged.Close() //nolint:errcheck
+	}()
+	for _, e := range packBlockIDs {
+		fullpath := filepath.Join(b.dirname, string(e)+simpleIndexSuffix)
+		ndx, err := b.openIndex(fullpath)
+		if err != nil {
+			return fmt.Errorf("unable to open pack index %q: %v", fullpath, err)
+		}
+
+		log.Printf("opened %v with %v entries", fullpath, ndx.EntryCount())
+		newIndexes[e] = true
+		newMerged = append(newMerged, ndx)
+	}
+	b.indexBlocks = newIndexes
+	b.merged = newMerged
+	newMerged = nil
+	return nil
+}
+
 func newSimpleCommittedBlockIndex(dirname string) (committedBlockIndex, error) {
 	_ = os.MkdirAll(dirname, 0700)
 
-	s := &simpleCommittedBlockIndex{dirname: dirname}
-	if err := s.loadIndexes(); err != nil {
-		return nil, err
+	s := &simpleCommittedBlockIndex{
+		dirname:     dirname,
+		indexBlocks: map[PhysicalBlockID]bool{},
 	}
 	return s, nil
 }

cli/command_block_index_list.go

@@ -4,25 +4,16 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/kopia/kopia/block"
 	"github.com/kopia/kopia/repo"
 )
 
 var (
 	blockIndexListCommand = blockIndexCommands.Command("list", "List block indexes").Alias("ls").Default()
-	blockIndexListAll     = blockIndexListCommand.Flag("all", "List all blocks, not just active ones").Short('a').Bool()
 	blockIndexListSummary = blockIndexListCommand.Flag("summary", "Display block summary").Bool()
 )
 
 func runListBlockIndexesAction(ctx context.Context, rep *repo.Repository) error {
-	var blks []block.IndexInfo
-	var err error
-
-	if !*blockIndexListAll {
-		blks, err = rep.Blocks.ActiveIndexBlocks(ctx)
-	} else {
-		blks, err = rep.Blocks.ListIndexBlocks(ctx)
-	}
+	blks, err := rep.Blocks.IndexBlocks(ctx)
 
 	if err != nil {
 		return err

cli/command_block_index_optimize.go

@@ -7,11 +7,13 @@ import (
 )
 
 var (
-	optimizeCommand = blockIndexCommands.Command("optimize", "Optimize block indexes.")
+	optimizeCommand        = blockIndexCommands.Command("optimize", "Optimize block indexes.")
+	optimizeMinSmallBlocks = blockIndexCommands.Flag("min-small-blocks", "Minimum number of small blocks that can be left after compaction.").Default("1").Int()
+	optimizeMaxSmallBlocks = blockIndexCommands.Flag("max-small-blocks", "Maximum number of small blocks that can be left after compaction.").Default("1").Int()
 )
 
 func runOptimizeCommand(ctx context.Context, rep *repo.Repository) error {
-	return rep.Blocks.CompactIndexes(ctx)
+	return rep.Blocks.CompactIndexes(ctx, *optimizeMinSmallBlocks, *optimizeMaxSmallBlocks)
 }
 
 func init() {

cli/command_block_index_show.go

@@ -26,7 +26,7 @@ func getIndexBlocksToShow(ctx context.Context, rep *repo.Repository) ([]block.Ph
 	}
 
 	if len(blockIDs) == 1 && blockIDs[0] == "active" {
-		b, err := rep.Blocks.ActiveIndexBlocks(ctx)
+		b, err := rep.Blocks.IndexBlocks(ctx)
 		if err != nil {
 			return nil, err
 		}

internal/packindex/packindex_test.go

@@ -3,6 +3,7 @@ package packindex_test
 import (
 	"bytes"
 	"crypto/sha1"
+	"encoding/hex"
 	"fmt"
 	"math/rand"
 	"reflect"
@@ -13,8 +14,6 @@ import (
 )
 
 func TestPackIndex(t *testing.T) {
-	b := packindex.NewBuilder()
-
 	blockNumber := 0
 
 	deterministicBlockID := func(prefix string, id int) string {
@@ -96,19 +95,41 @@ func TestPackIndex(t *testing.T) {
 	}
 
 	infoMap := map[string]packindex.Info{}
+	b1 := packindex.NewBuilder()
+	b2 := packindex.NewBuilder()
+	b3 := packindex.NewBuilder()
 
 	for _, info := range infos {
 		infoMap[info.BlockID] = info
-		b.Add(info)
+		b1.Add(info)
+		b2.Add(info)
+		b3.Add(info)
 	}
 
-	var buf bytes.Buffer
-	if err := b.Build(&buf); err != nil {
+	var buf1 bytes.Buffer
+	var buf2 bytes.Buffer
+	var buf3 bytes.Buffer
+	if err := b1.Build(&buf1); err != nil {
 		t.Errorf("unable to build: %v", err)
 	}
+	if err := b1.Build(&buf2); err != nil {
+		t.Errorf("unable to build: %v", err)
+	}
+	if err := b1.Build(&buf3); err != nil {
+		t.Errorf("unable to build: %v", err)
+	}
+	data1 := buf1.Bytes()
+	data2 := buf2.Bytes()
+	data3 := buf3.Bytes()
 
-	data := buf.Bytes()
-	ndx, err := packindex.Open(bytes.NewReader(data))
+	if !reflect.DeepEqual(data1, data2) {
+		t.Errorf("builder output not stable: %x vs %x", hex.Dump(data1), hex.Dump(data2))
+	}
+	if !reflect.DeepEqual(data2, data3) {
+		t.Errorf("builder output not stable: %x vs %x", hex.Dump(data2), hex.Dump(data3))
+	}
+
+	ndx, err := packindex.Open(bytes.NewReader(data1))
 	if err != nil {
 		t.Fatalf("can't open index: %v", err)
 	}

repo/repository_test.go

@@ -186,7 +186,7 @@ func TestPackingSimple(t *testing.T) {
 	verify(ctx, t, repo, oid2a, []byte(content2), "packed-object-2")
 	verify(ctx, t, repo, oid3a, []byte(content3), "packed-object-3")
 
-	if err := repo.Blocks.CompactIndexes(ctx); err != nil {
+	if err := repo.Blocks.CompactIndexes(ctx, 1, 1); err != nil {
 		t.Errorf("optimize error: %v", err)
 	}
 	data, _, repo = setupTestWithData(t, data, keyTime, func(n *NewRepositoryOptions) {
@@ -196,7 +196,7 @@ func TestPackingSimple(t *testing.T) {
 	verify(ctx, t, repo, oid2a, []byte(content2), "packed-object-2")
 	verify(ctx, t, repo, oid3a, []byte(content3), "packed-object-3")
 
-	if err := repo.Blocks.CompactIndexes(ctx); err != nil {
+	if err := repo.Blocks.CompactIndexes(ctx, 1, 1); err != nil {
 		t.Errorf("optimize error: %v", err)
 	}
 	_, _, repo = setupTestWithData(t, data, keyTime, func(n *NewRepositoryOptions) {

tests/end_to_end_test/end_to_end_test.go

@@ -112,11 +112,10 @@ func TestEndToEnd(t *testing.T) {
 	// expect 5 blocks, each snapshot creation adds one index block.
 	e.runAndVerifyOutputLineCount(t, 5, "blockindex", "ls")
 	e.runAndExpectSuccess(t, "blockindex", "optimize")
-	e.runAndVerifyOutputLineCount(t, 6, "blockindex", "ls")
-	e.runAndVerifyOutputLineCount(t, 6, "blockindex", "ls", "--all")
+	e.runAndVerifyOutputLineCount(t, 1, "blockindex", "ls")
 
 	e.runAndExpectSuccess(t, "snapshot", "create", ".", dir1, dir2)
-	e.runAndVerifyOutputLineCount(t, 7, "blockindex", "ls")
+	e.runAndVerifyOutputLineCount(t, 2, "blockindex", "ls")
 }
 
 func (e *testenv) runAndExpectSuccess(t *testing.T, args ...string) []string {

tests/stress_test/stress_test.go

@@ -16,7 +16,7 @@ import (
 	"github.com/kopia/kopia/storage"
 )
 
-const goroutineCount = 8
+const goroutineCount = 16
 
 func TestStressBlockManager(t *testing.T) {
 	data := map[string][]byte{}