diff --git a/cli/command_index_inspect.go b/cli/command_index_inspect.go index 22e2b87af..9c987fd58 100644 --- a/cli/command_index_inspect.go +++ b/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) } diff --git a/cli/command_index_list.go b/cli/command_index_list.go index 5c02bdffa..b01e5c2fe 100644 --- a/cli/command_index_list.go +++ b/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") } diff --git a/internal/epoch/epoch_manager.go b/internal/epoch/epoch_manager.go index 36bf9b3d9..36a253119 100644 --- a/internal/epoch/epoch_manager.go +++ b/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") } diff --git a/internal/epoch/epoch_manager_test.go b/internal/epoch/epoch_manager_test.go index c8d8ef010..ce8365777 100644 --- a/internal/epoch/epoch_manager_test.go +++ b/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) } diff --git a/repo/content/committed_read_manager.go b/repo/content/committed_read_manager.go index f0d8f0602..53ba458fc 100644 --- a/repo/content/committed_read_manager.go +++ b/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 } diff --git a/repo/content/content_index_reader.go b/repo/content/content_index_reader.go deleted file mode 100644 index 763304422..000000000 --- a/repo/content/content_index_reader.go +++ /dev/null @@ -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) -} diff --git a/repo/content/content_manager_indexes.go b/repo/content/content_manager_indexes.go index 3e980e941..5468619c7 100644 --- a/repo/content/content_manager_indexes.go +++ b/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") } diff --git a/repo/content/encrypted_blob_mgr.go b/repo/content/encrypted_blob_mgr.go new file mode 100644 index 000000000..2d1a10c68 --- /dev/null +++ b/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 +} diff --git a/repo/content/index_blob_manager.go b/repo/content/index_blob_manager_v0.go similarity index 71% rename from repo/content/index_blob_manager.go rename to repo/content/index_blob_manager_v0.go index 6bbb152c9..4bddc40a6 100644 --- a/repo/content/index_blob_manager.go +++ b/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) } } } diff --git a/repo/content/index_blob_manager_test.go b/repo/content/index_blob_manager_v0_test.go similarity index 91% rename from repo/content/index_blob_manager_test.go rename to repo/content/index_blob_manager_v0_test.go index a72992658..81edba479 100644 --- a/repo/content/index_blob_manager_test.go +++ b/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 diff --git a/repo/repository.go b/repo/repository.go index 21359dd82..b6dede263 100644 --- a/repo/repository.go +++ b/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.