Misc index blob manager refactorings (#1138)

* content: extracted encryptedBlobMgr component

* content: renamed files

* content: refactored ParseIndexBlob

* epoch: adjusted API to return blob.Metadata

* content: removed IndexBlobReader interface

* content: cleaned up indexBlobManager API

cli/command_index_inspect.go

@@ -53,7 +53,12 @@ func (c *commandIndexInspect) inspectSingleIndexBlob(ctx context.Context, rep re
 		return errors.Wrapf(err, "unable to get metadata for %v", blobID)
 	}
 
-	entries, err := rep.IndexBlobReader().ParseIndexBlob(ctx, blobID)
+	data, err := rep.BlobReader().GetBlob(ctx, blobID, 0, -1)
+	if err != nil {
+		return errors.Wrapf(err, "unable to get data for %v", blobID)
+	}
+
+	entries, err := content.ParseIndexBlob(ctx, blobID, data, rep.Crypter())
 	if err != nil {
 		return errors.Wrapf(err, "unable to recover index from %v", blobID)
 	}

cli/command_index_list.go

@@ -34,7 +34,7 @@ func (c *commandIndexList) run(ctx context.Context, rep repo.DirectRepository) e
 	jl.begin(&c.jo)
 	defer jl.end()
 
-	blks, err := rep.IndexBlobReader().IndexBlobs(ctx, c.blockIndexListIncludeSuperseded)
+	blks, err := rep.IndexBlobs(ctx, c.blockIndexListIncludeSuperseded)
 	if err != nil {
 		return errors.Wrap(err, "error listing index blobs")
 	}

internal/epoch/epoch_manager.go

@@ -20,6 +20,9 @@ import (
 	"github.com/kopia/kopia/repo/logging"
 )
 
+// LatestEpoch represents the current epoch number in GetCompleteIndexSet.
+const LatestEpoch = -1
+
 // Parameters encapsulates all parameters that influence the behavior of epoch manager.
 type Parameters struct {
 	// how frequently each client will list blobs to determine the current epoch.
@@ -376,13 +379,17 @@ func (e *Manager) Current(ctx context.Context) (int, error) {
 }
 
 // GetCompleteIndexSet returns the set of blobs forming a complete index set up to the provided epoch number.
-func (e *Manager) GetCompleteIndexSet(ctx context.Context, maxEpoch int) ([]blob.ID, error) {
+func (e *Manager) GetCompleteIndexSet(ctx context.Context, maxEpoch int) ([]blob.Metadata, error) {
 	for {
 		cs, err := e.committedState(ctx)
 		if err != nil {
 			return nil, err
 		}
 
+		if maxEpoch == LatestEpoch {
+			maxEpoch = cs.WriteEpoch + 1
+		}
+
 		result, err := e.getCompleteIndexSetForCommittedState(ctx, cs, maxEpoch)
 		if e.timeFunc().Before(cs.ValidUntil) {
 			return result, err
@@ -401,17 +408,17 @@ func (e *Manager) GetCompleteIndexSet(ctx context.Context, maxEpoch int) ([]blob
 	}
 }
 
-func (e *Manager) getCompleteIndexSetForCommittedState(ctx context.Context, cs snapshot, maxEpoch int) ([]blob.ID, error) {
+func (e *Manager) getCompleteIndexSetForCommittedState(ctx context.Context, cs snapshot, maxEpoch int) ([]blob.Metadata, error) {
 	var (
 		startEpoch int
 
 		resultMutex sync.Mutex
-		result      []blob.ID
+		result      []blob.Metadata
 	)
 
 	for i := maxEpoch; i >= 0; i-- {
 		if blobs := cs.FullCheckpointSets[i]; blobs != nil {
-			result = append(result, blob.IDsFromMetadata(blobs)...)
+			result = append(result, blobs...)
 			startEpoch = i + 1
 
 			e.log.Debugf("using full checkpoint at epoch %v", i)
@@ -466,11 +473,11 @@ func (e *Manager) WroteIndex(ctx context.Context, bm blob.Metadata) error {
 	return nil
 }
 
-func (e *Manager) getIndexesFromEpochInternal(ctx context.Context, cs snapshot, epoch int) ([]blob.ID, error) {
+func (e *Manager) getIndexesFromEpochInternal(ctx context.Context, cs snapshot, epoch int) ([]blob.Metadata, error) {
 	// check if the epoch is old enough to possibly have compacted blobs
 	epochSettled := cs.isSettledEpochNumber(epoch)
 	if epochSettled && cs.SingleEpochCompactionSets[epoch] != nil {
-		return blob.IDsFromMetadata(cs.SingleEpochCompactionSets[epoch]), nil
+		return cs.SingleEpochCompactionSets[epoch], nil
 	}
 
 	// load uncompacted blobs for this epoch
@@ -516,7 +523,7 @@ func (e *Manager) getIndexesFromEpochInternal(ctx context.Context, cs snapshot,
 	}
 
 	// return uncompacted blobs to the caller while we're compacting them in background
-	return blob.IDsFromMetadata(uncompactedBlobs), nil
+	return uncompactedBlobs, nil
 }
 
 func (e *Manager) advanceEpoch(ctx context.Context, newEpoch int) error {
@@ -549,7 +556,7 @@ func (e *Manager) generateFullCheckpointFromCommittedState(ctx context.Context,
 		return errors.Errorf("not generating full checkpoint - the committed state is no longer valid")
 	}
 
-	if err := e.compact(ctx, completeSet, fullCheckpointBlobPrefix(epoch)); err != nil {
+	if err := e.compact(ctx, blob.IDsFromMetadata(completeSet), fullCheckpointBlobPrefix(epoch)); err != nil {
 		return errors.Wrap(err, "unable to compact blobs")
 	}
 

internal/epoch/epoch_manager_test.go

@@ -22,8 +22,6 @@ import (
 	"github.com/kopia/kopia/repo/blob/logging"
 )
 
-const verifyAllEpochs = -1
-
 type fakeIndex struct {
 	Entries []int `json:"entries"`
 }
@@ -333,10 +331,10 @@ func TestLateWriteIsIgnored(t *testing.T) {
 	require.NoError(t, err)
 
 	// blobID1 will be included in the index.
-	require.Contains(t, iset, blobID1)
+	require.Contains(t, blob.IDsFromMetadata(iset), blobID1)
 
 	// blobID2 will be excluded from the index.
-	require.NotContains(t, iset, blobID2)
+	require.NotContains(t, blob.IDsFromMetadata(iset), blobID2)
 }
 
 // nolint:thelper
@@ -354,7 +352,7 @@ func verifySequentialWrites(t *testing.T, te *epochManagerTestEnv) {
 		te.mustWriteIndexFile(ctx, t, newFakeIndexWithEntries(indexNum))
 
 		expected.Entries = append(expected.Entries, indexNum)
-		te.verifyCompleteIndexSet(ctx, t, verifyAllEpochs, expected)
+		te.verifyCompleteIndexSet(ctx, t, LatestEpoch, expected)
 
 		dt := randomTime(1*time.Minute, 8*time.Hour)
 		t.Logf("advancing time by %v", dt)
@@ -387,18 +385,11 @@ func randomTime(min, max time.Duration) time.Duration {
 func (te *epochManagerTestEnv) verifyCompleteIndexSet(ctx context.Context, t *testing.T, maxEpoch int, want *fakeIndex) {
 	t.Helper()
 
-	if maxEpoch == verifyAllEpochs {
-		n, err := te.mgr.Current(ctx)
-		require.NoError(t, err)
-
-		maxEpoch = n + 1
-	}
-
 	blobs, err := te.mgr.GetCompleteIndexSet(ctx, maxEpoch)
 	t.Logf("complete set length: %v", len(blobs))
 	require.NoError(t, err)
 
-	merged, err := te.getMergedIndexContents(ctx, blobs)
+	merged, err := te.getMergedIndexContents(ctx, blob.IDsFromMetadata(blobs))
 	require.NoError(t, err)
 	require.Equal(t, want.Entries, merged.Entries)
 }

repo/content/committed_read_manager.go

@@ -30,6 +30,14 @@ const ownWritesCacheDuration = 15 * time.Minute
 
 var cachedIndexBlobPrefixes = []blob.ID{IndexBlobPrefix, compactionLogBlobPrefix, cleanupBlobPrefix}
 
+// indexBlobManager is the API of index blob manager as used by content manager.
+type indexBlobManager interface {
+	writeIndexBlob(ctx context.Context, data []byte, sessionID SessionID) (blob.Metadata, error)
+	listActiveIndexBlobs(ctx context.Context) ([]IndexBlobInfo, error)
+	compact(ctx context.Context, opts CompactOptions) error
+	flushCache(ctx context.Context)
+}
+
 // SharedManager is responsible for read-only access to committed data.
 type SharedManager struct {
 	refCount int32 // number of Manager objects that refer to this SharedManager
@@ -42,6 +50,7 @@ type SharedManager struct {
 	metadataCache     contentCache
 	committedContents *committedContentIndex
 	crypter           *Crypter
+	enc               *encryptedBlobMgr
 	timeNow           func() time.Time
 
 	format                  FormattingOptions
@@ -236,8 +245,22 @@ func (sm *SharedManager) decryptAndVerify(encrypted, iv []byte) ([]byte, error)
 // IndexBlobs returns the list of active index blobs.
 func (sm *SharedManager) IndexBlobs(ctx context.Context, includeInactive bool) ([]IndexBlobInfo, error) {
 	if includeInactive {
-		// nolint:wrapcheck
-		return sm.indexBlobManager.listAllIndexBlobs(ctx)
+		var result []IndexBlobInfo
+
+		prefixes := []blob.ID{IndexBlobPrefix}
+
+		for _, prefix := range prefixes {
+			blobs, err := blob.ListAllBlobs(ctx, sm.st, prefix)
+			if err != nil {
+				return nil, errors.Wrapf(err, "error listing %v blogs", prefix)
+			}
+
+			for _, bm := range blobs {
+				result = append(result, IndexBlobInfo{Metadata: bm})
+			}
+		}
+
+		return result, nil
 	}
 
 	// nolint:wrapcheck
@@ -318,20 +341,26 @@ func (sm *SharedManager) setupReadManagerCaches(ctx context.Context, caching *Ca
 		return errors.Wrap(err, "unable to initialize list cache")
 	}
 
-	sm.indexBlobManager = &indexBlobManagerImpl{
+	sm.enc = &encryptedBlobMgr{
 		st:             cachedSt,
 		crypter:        sm.crypter,
-		timeNow:        sm.timeNow,
 		indexBlobCache: metadataCache,
-		maxPackSize:    sm.maxPackSize,
-		indexVersion:   sm.indexVersion,
-		log:            logging.WithPrefix("[index-blob-manager] ", sm.sharedBaseLogger),
+		log:            logging.WithPrefix("[encrypted-blob-manager] ", sm.sharedBaseLogger),
+	}
+
+	sm.indexBlobManager = &indexBlobManagerV0{
+		st:           cachedSt,
+		enc:          sm.enc,
+		timeNow:      sm.timeNow,
+		maxPackSize:  sm.maxPackSize,
+		indexVersion: sm.indexVersion,
+		log:          logging.WithPrefix("[index-blob-manager] ", sm.sharedBaseLogger),
 	}
 
 	// once everything is ready, set it up
 	sm.contentCache = dataCache
 	sm.metadataCache = metadataCache
-	sm.committedContents = newCommittedContentIndex(caching, uint32(sm.crypter.Encryptor.Overhead()), sm.indexVersion, sm.indexBlobManager.getIndexBlob, sm.sharedBaseLogger)
+	sm.committedContents = newCommittedContentIndex(caching, uint32(sm.crypter.Encryptor.Overhead()), sm.indexVersion, sm.enc.getEncryptedBlob, sm.sharedBaseLogger)
 
 	return nil
 }

repo/content/content_index_reader.go

@@ -1,13 +0,0 @@
-package content
-
-import (
-	"context"
-
-	"github.com/kopia/kopia/repo/blob"
-)
-
-// IndexBlobReader defines content read API.
-type IndexBlobReader interface {
-	ParseIndexBlob(ctx context.Context, blobID blob.ID) ([]Info, error)
-	IndexBlobs(ctx context.Context, includeInactive bool) ([]IndexBlobInfo, error)
-}

repo/content/content_manager_indexes.go

@@ -46,13 +46,13 @@ func (bm *WriteManager) CompactIndexes(ctx context.Context, opt CompactOptions)
 }
 
 // ParseIndexBlob loads entries in a given index blob and returns them.
-func (sm *SharedManager) ParseIndexBlob(ctx context.Context, blobID blob.ID) ([]Info, error) {
-	data, err := sm.indexBlobManager.getIndexBlob(ctx, blobID)
+func ParseIndexBlob(ctx context.Context, blobID blob.ID, data []byte, crypter *Crypter) ([]Info, error) {
+	data, err := crypter.DecryptBLOB(data, blobID)
 	if err != nil {
-		return nil, errors.Wrapf(err, "error getting index %q", blobID)
+		return nil, errors.Wrap(err, "unable to decrypt index blob")
 	}
 
-	index, err := openPackIndex(bytes.NewReader(data), uint32(sm.crypter.Encryptor.Overhead()))
+	index, err := openPackIndex(bytes.NewReader(data), uint32(crypter.Encryptor.Overhead()))
 	if err != nil {
 		return nil, errors.Wrapf(err, "unable to open index blob")
 	}

repo/content/encrypted_blob_mgr.go

@@ -0,0 +1,50 @@
+package content
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+
+	"github.com/kopia/kopia/internal/gather"
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/logging"
+)
+
+type encryptedBlobMgr struct {
+	st             blob.Storage
+	crypter        *Crypter
+	indexBlobCache contentCache
+	log            logging.Logger
+}
+
+func (m *encryptedBlobMgr) getEncryptedBlob(ctx context.Context, blobID blob.ID) ([]byte, error) {
+	payload, err := m.indexBlobCache.getContent(ctx, cacheKey(blobID), blobID, 0, -1)
+	if err != nil {
+		return nil, errors.Wrap(err, "getContent")
+	}
+
+	return m.crypter.DecryptBLOB(payload, blobID)
+}
+
+func (m *encryptedBlobMgr) encryptAndWriteBlob(ctx context.Context, data []byte, prefix blob.ID, sessionID SessionID) (blob.Metadata, error) {
+	blobID, data2, err := m.crypter.EncryptBLOB(data, prefix, sessionID)
+	if err != nil {
+		return blob.Metadata{}, errors.Wrap(err, "error encrypting")
+	}
+
+	err = m.st.PutBlob(ctx, blobID, gather.FromSlice(data2))
+	if err != nil {
+		m.log.Debugf("write-index-blob %v failed %v", blobID, err)
+		return blob.Metadata{}, errors.Wrapf(err, "error writing blob %v", blobID)
+	}
+
+	bm, err := m.st.GetMetadata(ctx, blobID)
+	if err != nil {
+		m.log.Debugf("write-index-blob-get-metadata %v failed %v", blobID, err)
+		return blob.Metadata{}, errors.Wrap(err, "unable to get blob metadata")
+	}
+
+	m.log.Debugf("write-index-blob %v %v %v", blobID, bm.Length, bm.Timestamp)
+
+	return bm, nil
+}

repo/content/index_blob_manager_v0.go

@@ -9,22 +9,10 @@ import (
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 
-	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/logging"
 )
 
-// indexBlobManager is the API of index blob manager as used by content manager.
-type indexBlobManager interface {
-	writeIndexBlob(ctx context.Context, data []byte, sessionID SessionID) (blob.Metadata, error)
-	listActiveIndexBlobs(ctx context.Context) ([]IndexBlobInfo, error)
-	listAllIndexBlobs(ctx context.Context) ([]IndexBlobInfo, error)
-	getIndexBlob(ctx context.Context, blobID blob.ID) ([]byte, error)
-	compact(ctx context.Context, opts CompactOptions) error
-	registerCompaction(ctx context.Context, inputs, outputs []blob.Metadata, maxEventualConsistencySettleTime time.Duration) error
-	flushCache(ctx context.Context)
-}
-
 const (
 	defaultEventualConsistencySettleTime = 1 * time.Hour
 	compactionLogBlobPrefix              = "m"
@@ -56,25 +44,16 @@ type cleanupEntry struct {
 	age time.Duration // not serialized, computed on load
 }
 
-type indexBlobManagerImpl struct {
-	st             blob.Storage
-	crypter        *Crypter
-	timeNow        func() time.Time
-	indexBlobCache contentCache
-	log            logging.Logger
-	maxPackSize    int
-	indexVersion   int
+type indexBlobManagerV0 struct {
+	st           blob.Storage
+	enc          *encryptedBlobMgr
+	timeNow      func() time.Time
+	log          logging.Logger
+	maxPackSize  int
+	indexVersion int
 }
 
-func (m *indexBlobManagerImpl) listActiveIndexBlobs(ctx context.Context) ([]IndexBlobInfo, error) {
-	return m.listIndexBlobs(ctx, false)
-}
-
-func (m *indexBlobManagerImpl) listAllIndexBlobs(ctx context.Context) ([]IndexBlobInfo, error) {
-	return m.listIndexBlobs(ctx, true)
-}
-
-func (m *indexBlobManagerImpl) listIndexBlobs(ctx context.Context, includeInactive bool) ([]IndexBlobInfo, error) {
+func (m *indexBlobManagerV0) listActiveIndexBlobs(ctx context.Context) ([]IndexBlobInfo, error) {
 	var compactionLogMetadata, storageIndexBlobs []blob.Metadata
 
 	var eg errgroup.Group
@@ -115,7 +94,7 @@ func (m *indexBlobManagerImpl) listIndexBlobs(ctx context.Context, includeInacti
 	}
 
 	// remove entries from indexMap that have been compacted and replaced by other indexes.
-	m.removeCompactedIndexes(indexMap, compactionLogs, includeInactive)
+	m.removeCompactedIndexes(indexMap, compactionLogs)
 
 	var results []IndexBlobInfo
 	for _, v := range indexMap {
@@ -129,13 +108,13 @@ func (m *indexBlobManagerImpl) listIndexBlobs(ctx context.Context, includeInacti
 	return results, nil
 }
 
-func (m *indexBlobManagerImpl) flushCache(ctx context.Context) {
+func (m *indexBlobManagerV0) flushCache(ctx context.Context) {
 	if err := m.st.FlushCaches(ctx); err != nil {
 		m.log.Debugf("error flushing caches: %v", err)
 	}
 }
 
-func (m *indexBlobManagerImpl) compact(ctx context.Context, opt CompactOptions) error {
+func (m *indexBlobManagerV0) compact(ctx context.Context, opt CompactOptions) error {
 	indexBlobs, err := m.listActiveIndexBlobs(ctx)
 	if err != nil {
 		return errors.Wrap(err, "error listing active index blobs")
@@ -154,7 +133,7 @@ func (m *indexBlobManagerImpl) compact(ctx context.Context, opt CompactOptions)
 	return nil
 }
 
-func (m *indexBlobManagerImpl) registerCompaction(ctx context.Context, inputs, outputs []blob.Metadata, maxEventualConsistencySettleTime time.Duration) error {
+func (m *indexBlobManagerV0) registerCompaction(ctx context.Context, inputs, outputs []blob.Metadata, maxEventualConsistencySettleTime time.Duration) error {
 	logEntryBytes, err := json.Marshal(&compactionLogEntry{
 		InputMetadata:  inputs,
 		OutputMetadata: outputs,
@@ -163,7 +142,7 @@ func (m *indexBlobManagerImpl) registerCompaction(ctx context.Context, inputs, o
 		return errors.Wrap(err, "unable to marshal log entry bytes")
 	}
 
-	compactionLogBlobMetadata, err := m.encryptAndWriteBlob(ctx, logEntryBytes, compactionLogBlobPrefix, "")
+	compactionLogBlobMetadata, err := m.enc.encryptAndWriteBlob(ctx, logEntryBytes, compactionLogBlobPrefix, "")
 	if err != nil {
 		return errors.Wrap(err, "unable to write compaction log")
 	}
@@ -185,51 +164,19 @@ func (m *indexBlobManagerImpl) registerCompaction(ctx context.Context, inputs, o
 	return nil
 }
 
-func (m *indexBlobManagerImpl) getIndexBlob(ctx context.Context, blobID blob.ID) ([]byte, error) {
-	return m.getEncryptedBlob(ctx, blobID)
+func (m *indexBlobManagerV0) getIndexBlob(ctx context.Context, blobID blob.ID) ([]byte, error) {
+	return m.enc.getEncryptedBlob(ctx, blobID)
 }
 
-func (m *indexBlobManagerImpl) getEncryptedBlob(ctx context.Context, blobID blob.ID) ([]byte, error) {
-	payload, err := m.indexBlobCache.getContent(ctx, cacheKey(blobID), blobID, 0, -1)
-	if err != nil {
-		return nil, errors.Wrap(err, "getContent")
-	}
-
-	return m.crypter.DecryptBLOB(payload, blobID)
+func (m *indexBlobManagerV0) writeIndexBlob(ctx context.Context, data []byte, sessionID SessionID) (blob.Metadata, error) {
+	return m.enc.encryptAndWriteBlob(ctx, data, IndexBlobPrefix, sessionID)
 }
 
-func (m *indexBlobManagerImpl) writeIndexBlob(ctx context.Context, data []byte, sessionID SessionID) (blob.Metadata, error) {
-	return m.encryptAndWriteBlob(ctx, data, IndexBlobPrefix, sessionID)
-}
-
-func (m *indexBlobManagerImpl) encryptAndWriteBlob(ctx context.Context, data []byte, prefix blob.ID, sessionID SessionID) (blob.Metadata, error) {
-	blobID, data2, err := m.crypter.EncryptBLOB(data, prefix, sessionID)
-	if err != nil {
-		return blob.Metadata{}, errors.Wrap(err, "error encrypting")
-	}
-
-	err = m.st.PutBlob(ctx, blobID, gather.FromSlice(data2))
-	if err != nil {
-		m.log.Debugf("write-index-blob %v failed %v", blobID, err)
-		return blob.Metadata{}, errors.Wrapf(err, "error writing blob %v", blobID)
-	}
-
-	bm, err := m.st.GetMetadata(ctx, blobID)
-	if err != nil {
-		m.log.Debugf("write-index-blob-get-metadata %v failed %v", blobID, err)
-		return blob.Metadata{}, errors.Wrap(err, "unable to get blob metadata")
-	}
-
-	m.log.Debugf("write-index-blob %v %v %v", blobID, bm.Length, bm.Timestamp)
-
-	return bm, nil
-}
-
-func (m *indexBlobManagerImpl) getCompactionLogEntries(ctx context.Context, blobs []blob.Metadata) (map[blob.ID]*compactionLogEntry, error) {
+func (m *indexBlobManagerV0) getCompactionLogEntries(ctx context.Context, blobs []blob.Metadata) (map[blob.ID]*compactionLogEntry, error) {
 	results := map[blob.ID]*compactionLogEntry{}
 
 	for _, cb := range blobs {
-		data, err := m.getEncryptedBlob(ctx, cb.BlobID)
+		data, err := m.enc.getEncryptedBlob(ctx, cb.BlobID)
 
 		if errors.Is(err, blob.ErrBlobNotFound) {
 			continue
@@ -253,11 +200,11 @@ func (m *indexBlobManagerImpl) getCompactionLogEntries(ctx context.Context, blob
 	return results, nil
 }
 
-func (m *indexBlobManagerImpl) getCleanupEntries(ctx context.Context, latestServerBlobTime time.Time, blobs []blob.Metadata) (map[blob.ID]*cleanupEntry, error) {
+func (m *indexBlobManagerV0) getCleanupEntries(ctx context.Context, latestServerBlobTime time.Time, blobs []blob.Metadata) (map[blob.ID]*cleanupEntry, error) {
 	results := map[blob.ID]*cleanupEntry{}
 
 	for _, cb := range blobs {
-		data, err := m.getEncryptedBlob(ctx, cb.BlobID)
+		data, err := m.enc.getEncryptedBlob(ctx, cb.BlobID)
 
 		if errors.Is(err, blob.ErrBlobNotFound) {
 			continue
@@ -281,7 +228,7 @@ func (m *indexBlobManagerImpl) getCleanupEntries(ctx context.Context, latestServ
 	return results, nil
 }
 
-func (m *indexBlobManagerImpl) deleteOldBlobs(ctx context.Context, latestBlob blob.Metadata, maxEventualConsistencySettleTime time.Duration) error {
+func (m *indexBlobManagerV0) deleteOldBlobs(ctx context.Context, latestBlob blob.Metadata, maxEventualConsistencySettleTime time.Duration) error {
 	allCompactionLogBlobs, err := blob.ListAllBlobs(ctx, m.st, compactionLogBlobPrefix)
 	if err != nil {
 		return errors.Wrap(err, "error listing compaction log blobs")
@@ -316,7 +263,7 @@ func (m *indexBlobManagerImpl) deleteOldBlobs(ctx context.Context, latestBlob bl
 	return nil
 }
 
-func (m *indexBlobManagerImpl) findIndexBlobsToDelete(latestServerBlobTime time.Time, entries map[blob.ID]*compactionLogEntry, maxEventualConsistencySettleTime time.Duration) []blob.ID {
+func (m *indexBlobManagerV0) findIndexBlobsToDelete(latestServerBlobTime time.Time, entries map[blob.ID]*compactionLogEntry, maxEventualConsistencySettleTime time.Duration) []blob.ID {
 	tmp := map[blob.ID]bool{}
 
 	for _, cl := range entries {
@@ -342,7 +289,7 @@ func (m *indexBlobManagerImpl) findIndexBlobsToDelete(latestServerBlobTime time.
 	return result
 }
 
-func (m *indexBlobManagerImpl) findCompactionLogBlobsToDelayCleanup(compactionBlobs []blob.Metadata) []blob.ID {
+func (m *indexBlobManagerV0) findCompactionLogBlobsToDelayCleanup(compactionBlobs []blob.Metadata) []blob.ID {
 	var result []blob.ID
 
 	for _, cb := range compactionBlobs {
@@ -353,7 +300,7 @@ func (m *indexBlobManagerImpl) findCompactionLogBlobsToDelayCleanup(compactionBl
 	return result
 }
 
-func (m *indexBlobManagerImpl) findBlobsToDelete(entries map[blob.ID]*cleanupEntry, maxEventualConsistencySettleTime time.Duration) (compactionLogs, cleanupBlobs []blob.ID) {
+func (m *indexBlobManagerV0) findBlobsToDelete(entries map[blob.ID]*cleanupEntry, maxEventualConsistencySettleTime time.Duration) (compactionLogs, cleanupBlobs []blob.ID) {
 	for k, e := range entries {
 		if e.age >= maxEventualConsistencySettleTime {
 			compactionLogs = append(compactionLogs, e.BlobIDs...)
@@ -364,7 +311,7 @@ func (m *indexBlobManagerImpl) findBlobsToDelete(entries map[blob.ID]*cleanupEnt
 	return
 }
 
-func (m *indexBlobManagerImpl) delayCleanupBlobs(ctx context.Context, blobIDs []blob.ID, cleanupScheduleTime time.Time) error {
+func (m *indexBlobManagerV0) delayCleanupBlobs(ctx context.Context, blobIDs []blob.ID, cleanupScheduleTime time.Time) error {
 	if len(blobIDs) == 0 {
 		return nil
 	}
@@ -377,14 +324,14 @@ func (m *indexBlobManagerImpl) delayCleanupBlobs(ctx context.Context, blobIDs []
 		return errors.Wrap(err, "unable to marshal cleanup log bytes")
 	}
 
-	if _, err := m.encryptAndWriteBlob(ctx, payload, cleanupBlobPrefix, ""); err != nil {
+	if _, err := m.enc.encryptAndWriteBlob(ctx, payload, cleanupBlobPrefix, ""); err != nil {
 		return errors.Wrap(err, "unable to cleanup log")
 	}
 
 	return nil
 }
 
-func (m *indexBlobManagerImpl) deleteBlobsFromStorageAndCache(ctx context.Context, blobIDs []blob.ID) error {
+func (m *indexBlobManagerV0) deleteBlobsFromStorageAndCache(ctx context.Context, blobIDs []blob.ID) error {
 	for _, blobID := range blobIDs {
 		if err := m.st.DeleteBlob(ctx, blobID); err != nil && !errors.Is(err, blob.ErrBlobNotFound) {
 			m.log.Debugf("delete-blob failed %v %v", blobID, err)
@@ -397,7 +344,7 @@ func (m *indexBlobManagerImpl) deleteBlobsFromStorageAndCache(ctx context.Contex
 	return nil
 }
 
-func (m *indexBlobManagerImpl) cleanup(ctx context.Context, maxEventualConsistencySettleTime time.Duration) error {
+func (m *indexBlobManagerV0) cleanup(ctx context.Context, maxEventualConsistencySettleTime time.Duration) error {
 	allCleanupBlobs, err := blob.ListAllBlobs(ctx, m.st, cleanupBlobPrefix)
 	if err != nil {
 		return errors.Wrap(err, "error listing cleanup blobs")
@@ -434,7 +381,7 @@ func (m *indexBlobManagerImpl) cleanup(ctx context.Context, maxEventualConsisten
 	return nil
 }
 
-func (m *indexBlobManagerImpl) getBlobsToCompact(indexBlobs []IndexBlobInfo, opt CompactOptions) []IndexBlobInfo {
+func (m *indexBlobManagerV0) getBlobsToCompact(indexBlobs []IndexBlobInfo, opt CompactOptions) []IndexBlobInfo {
 	var nonCompactedBlobs, verySmallBlobs []IndexBlobInfo
 
 	var totalSizeNonCompactedBlobs, totalSizeVerySmallBlobs, totalSizeMediumSizedBlobs int64
@@ -474,7 +421,7 @@ func (m *indexBlobManagerImpl) getBlobsToCompact(indexBlobs []IndexBlobInfo, opt
 	return nonCompactedBlobs
 }
 
-func (m *indexBlobManagerImpl) compactIndexBlobs(ctx context.Context, indexBlobs []IndexBlobInfo, opt CompactOptions) error {
+func (m *indexBlobManagerV0) compactIndexBlobs(ctx context.Context, indexBlobs []IndexBlobInfo, opt CompactOptions) error {
 	if len(indexBlobs) <= 1 && opt.DropDeletedBefore.IsZero() && len(opt.DropContents) == 0 {
 		return nil
 	}
@@ -525,7 +472,7 @@ func (m *indexBlobManagerImpl) compactIndexBlobs(ctx context.Context, indexBlobs
 	return nil
 }
 
-func (m *indexBlobManagerImpl) dropContentsFromBuilder(bld packIndexBuilder, opt CompactOptions) {
+func (m *indexBlobManagerV0) dropContentsFromBuilder(bld packIndexBuilder, opt CompactOptions) {
 	for _, dc := range opt.DropContents {
 		if _, ok := bld[dc]; ok {
 			m.log.Debugf("manual-drop-from-index %v", dc)
@@ -547,13 +494,13 @@ func (m *indexBlobManagerImpl) dropContentsFromBuilder(bld packIndexBuilder, opt
 	}
 }
 
-func (m *indexBlobManagerImpl) addIndexBlobsToBuilder(ctx context.Context, bld packIndexBuilder, indexBlob IndexBlobInfo) error {
+func (m *indexBlobManagerV0) addIndexBlobsToBuilder(ctx context.Context, bld packIndexBuilder, indexBlob IndexBlobInfo) error {
 	data, err := m.getIndexBlob(ctx, indexBlob.BlobID)
 	if err != nil {
 		return errors.Wrapf(err, "error getting index %q", indexBlob.BlobID)
 	}
 
-	index, err := openPackIndex(bytes.NewReader(data), uint32(m.crypter.Encryptor.Overhead()))
+	index, err := openPackIndex(bytes.NewReader(data), uint32(m.enc.crypter.Encryptor.Overhead()))
 	if err != nil {
 		return errors.Wrapf(err, "unable to open index blob %q", indexBlob)
 	}
@@ -578,7 +525,7 @@ func blobsOlderThan(m []blob.Metadata, cutoffTime time.Time) []blob.Metadata {
 	return res
 }
 
-func (m *indexBlobManagerImpl) removeCompactedIndexes(bimap map[blob.ID]*IndexBlobInfo, compactionLogs map[blob.ID]*compactionLogEntry, markAsSuperseded bool) {
+func (m *indexBlobManagerV0) removeCompactedIndexes(bimap map[blob.ID]*IndexBlobInfo, compactionLogs map[blob.ID]*compactionLogEntry) {
 	var validCompactionLogs []*compactionLogEntry
 
 	for _, cl := range compactionLogs {
@@ -606,11 +553,7 @@ func (m *indexBlobManagerImpl) removeCompactedIndexes(bimap map[blob.ID]*IndexBl
 			if md := bimap[ib.BlobID]; md != nil && md.Superseded == nil {
 				m.log.Debugf("ignore-index-blob %v compacted to %v", ib, cl.OutputMetadata)
 
-				if markAsSuperseded {
-					md.Superseded = cl.OutputMetadata
-				} else {
-					delete(bimap, ib.BlobID)
-				}
+				delete(bimap, ib.BlobID)
 			}
 		}
 	}

repo/content/index_blob_manager_v0_test.go

@@ -418,7 +418,7 @@ type fakeContentIndexEntry struct {
 	Deleted bool
 }
 
-func verifyFakeContentsWritten(ctx context.Context, t *testing.T, m indexBlobManager, numWritten int, contentPrefix string, deletedContents map[string]bool) error {
+func verifyFakeContentsWritten(ctx context.Context, t *testing.T, m *indexBlobManagerV0, numWritten int, contentPrefix string, deletedContents map[string]bool) error {
 	t.Helper()
 
 	if numWritten == 0 {
@@ -452,7 +452,7 @@ func verifyFakeContentsWritten(ctx context.Context, t *testing.T, m indexBlobMan
 	return nil
 }
 
-func fakeCompaction(ctx context.Context, t *testing.T, m indexBlobManager, dropDeleted bool) error {
+func fakeCompaction(ctx context.Context, t *testing.T, m *indexBlobManagerV0, dropDeleted bool) error {
 	t.Helper()
 
 	t.Logf("fakeCompaction(dropDeleted=%v)", dropDeleted)
@@ -513,7 +513,7 @@ func fakeContentID(prefix string, n int) string {
 	return fmt.Sprintf("%v-%06v", prefix, n)
 }
 
-func deleteFakeContents(ctx context.Context, t *testing.T, m indexBlobManager, prefix string, numWritten int, deleted map[string]bool, timeFunc func() time.Time) error {
+func deleteFakeContents(ctx context.Context, t *testing.T, m *indexBlobManagerV0, prefix string, numWritten int, deleted map[string]bool, timeFunc func() time.Time) error {
 	t.Helper()
 
 	if numWritten == 0 {
@@ -550,7 +550,7 @@ func deleteFakeContents(ctx context.Context, t *testing.T, m indexBlobManager, p
 	return err
 }
 
-func undeleteFakeContents(ctx context.Context, t *testing.T, m indexBlobManager, deleted map[string]bool, timeFunc func() time.Time) error {
+func undeleteFakeContents(ctx context.Context, t *testing.T, m *indexBlobManagerV0, deleted map[string]bool, timeFunc func() time.Time) error {
 	t.Helper()
 
 	if len(deleted) == 0 {
@@ -588,7 +588,7 @@ func undeleteFakeContents(ctx context.Context, t *testing.T, m indexBlobManager,
 	return err
 }
 
-func writeFakeContents(ctx context.Context, t *testing.T, m indexBlobManager, prefix string, count int, numWritten *int, timeFunc func() time.Time) error {
+func writeFakeContents(ctx context.Context, t *testing.T, m *indexBlobManagerV0, prefix string, count int, numWritten *int, timeFunc func() time.Time) error {
 	t.Helper()
 
 	t.Logf("writeFakeContents()")
@@ -615,7 +615,7 @@ type fakeIndexData struct {
 	Entries  map[string]fakeContentIndexEntry
 }
 
-func writeFakeIndex(ctx context.Context, t *testing.T, m indexBlobManager, ndx map[string]fakeContentIndexEntry) (blob.Metadata, error) {
+func writeFakeIndex(ctx context.Context, t *testing.T, m *indexBlobManagerV0, ndx map[string]fakeContentIndexEntry) (blob.Metadata, error) {
 	t.Helper()
 
 	j, err := json.Marshal(fakeIndexData{
@@ -640,7 +640,7 @@ func writeFakeIndex(ctx context.Context, t *testing.T, m indexBlobManager, ndx m
 
 var errGetAllFakeContentsRetry = errors.New("retry")
 
-func getAllFakeContents(ctx context.Context, t *testing.T, m indexBlobManager) (map[string]fakeContentIndexEntry, []IndexBlobInfo, error) {
+func getAllFakeContents(ctx context.Context, t *testing.T, m *indexBlobManagerV0) (map[string]fakeContentIndexEntry, []IndexBlobInfo, error) {
 	t.Helper()
 
 	allContents, allBlobs, err := getAllFakeContentsInternal(ctx, t, m)
@@ -652,7 +652,7 @@ func getAllFakeContents(ctx context.Context, t *testing.T, m indexBlobManager) (
 	return allContents, allBlobs, err
 }
 
-func getAllFakeContentsInternal(ctx context.Context, t *testing.T, m indexBlobManager) (map[string]fakeContentIndexEntry, []IndexBlobInfo, error) {
+func getAllFakeContentsInternal(ctx context.Context, t *testing.T, m *indexBlobManagerV0) (map[string]fakeContentIndexEntry, []IndexBlobInfo, error) {
 	t.Helper()
 
 	blobs, err := m.listActiveIndexBlobs(ctx)
@@ -715,7 +715,7 @@ func keysWithPrefix(data blobtesting.DataMap, prefix blob.ID) []blob.ID {
 	return res
 }
 
-func mustRegisterCompaction(t *testing.T, m indexBlobManager, inputs, outputs []blob.Metadata) {
+func mustRegisterCompaction(t *testing.T, m *indexBlobManagerV0, inputs, outputs []blob.Metadata) {
 	t.Helper()
 
 	t.Logf("compacting %v to %v", inputs, outputs)
@@ -726,7 +726,7 @@ func mustRegisterCompaction(t *testing.T, m indexBlobManager, inputs, outputs []
 	}
 }
 
-func mustWriteIndexBlob(t *testing.T, m indexBlobManager, data string) blob.Metadata {
+func mustWriteIndexBlob(t *testing.T, m *indexBlobManagerV0, data string) blob.Metadata {
 	t.Helper()
 
 	t.Logf("writing index blob %q", data)
@@ -739,7 +739,7 @@ func mustWriteIndexBlob(t *testing.T, m indexBlobManager, data string) blob.Meta
 	return blobMD
 }
 
-func assertIndexBlobList(t *testing.T, m indexBlobManager, wantMD ...blob.Metadata) {
+func assertIndexBlobList(t *testing.T, m *indexBlobManagerV0, wantMD ...blob.Metadata) {
 	t.Helper()
 
 	var want []blob.ID
@@ -762,7 +762,7 @@ func assertIndexBlobList(t *testing.T, m indexBlobManager, wantMD ...blob.Metada
 	require.ElementsMatch(t, got, want)
 }
 
-func newIndexBlobManagerForTesting(t *testing.T, st blob.Storage, localTimeNow func() time.Time) indexBlobManager {
+func newIndexBlobManagerForTesting(t *testing.T, st blob.Storage, localTimeNow func() time.Time) *indexBlobManagerV0 {
 	t.Helper()
 
 	p := &FormattingOptions{
@@ -780,22 +780,30 @@ func newIndexBlobManagerForTesting(t *testing.T, st blob.Storage, localTimeNow f
 		t.Fatalf("unable to create hash: %v", err)
 	}
 
-	m := &indexBlobManagerImpl{
-		st: ownwrites.NewWrapper(
-			st,
-			blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil),
-			[]blob.ID{IndexBlobPrefix, compactionLogBlobPrefix, cleanupBlobPrefix},
-			15*time.Minute,
-		),
-		indexBlobCache: passthroughContentCache{st},
-		crypter: &Crypter{
-			HashFunction: hf,
-			Encryptor:    enc,
+	st = ownwrites.NewWrapper(
+		st,
+		blobtesting.NewMapStorage(blobtesting.DataMap{}, nil, nil),
+		[]blob.ID{IndexBlobPrefix, compactionLogBlobPrefix, cleanupBlobPrefix},
+		15*time.Minute,
+	)
+
+	log := repologging.Printf(t.Logf)("test")
+
+	m := &indexBlobManagerV0{
+		st: st,
+		enc: &encryptedBlobMgr{
+			st:             st,
+			indexBlobCache: passthroughContentCache{st},
+			crypter: &Crypter{
+				HashFunction: hf,
+				Encryptor:    enc,
+			},
+			log: log,
 		},
 		timeNow:      localTimeNow,
 		maxPackSize:  20 << 20,
 		indexVersion: 1,
-		log:          repologging.Printf(t.Logf)("test"),
+		log:          log,
 	}
 
 	return m

repo/repository.go

@@ -53,7 +53,7 @@ type DirectRepository interface {
 	ObjectFormat() object.Format
 	BlobReader() blob.Reader
 	ContentReader() content.Reader
-	IndexBlobReader() content.IndexBlobReader
+	IndexBlobs(ctx context.Context, includeInactive bool) ([]content.IndexBlobInfo, error)
 	Crypter() *content.Crypter
 
 	NewDirectWriter(ctx context.Context, opt WriteSessionOptions) (context.Context, DirectRepositoryWriter, error)
@@ -280,9 +280,10 @@ func (r *directRepository) ContentReader() content.Reader {
 	return r.cmgr
 }
 
-// IndexBlobReader returns the index blob reader.
-func (r *directRepository) IndexBlobReader() content.IndexBlobReader {
-	return r.cmgr
+// IndexBlobs returns the index blobs in use.
+func (r *directRepository) IndexBlobs(ctx context.Context, includeInactive bool) ([]content.IndexBlobInfo, error) {
+	// nolint:wrapcheck
+	return r.cmgr.IndexBlobs(ctx, includeInactive)
 }
 
 // Refresh makes external changes visible to repository.