added block.Manager.Refresh method

block/block_manager.go

@@ -76,6 +76,8 @@ type Manager struct {
 	flushPackIndexesAfter time.Time // time when those indexes should be flushed
 	activeBlocksExtraTime time.Duration
 
+	closed chan struct{}
+
 	writeFormatVersion int32 // format version to write
 
 	maxPackSize int
@@ -403,12 +405,12 @@ func sortBlocksByTime(b []IndexInfo) {
 	})
 }
 
-func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]IndexInfo, error) {
+func (bm *Manager) loadPackIndexesUnlocked(ctx context.Context) ([]IndexInfo, bool, error) {
 	nextSleepTime := 100 * time.Millisecond
 
 	for i := 0; i < indexLoadAttempts; i++ {
 		if err := ctx.Err(); err != nil {
-			return nil, err
+			return nil, false, err
 		}
 
 		if i > 0 {
@@ -420,7 +422,7 @@ func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]IndexInfo, erro
 
 		blocks, err := bm.listCache.listIndexBlocks(ctx)
 		if err != nil {
-			return nil, err
+			return nil, false, err
 		}
 
 		err = bm.tryLoadPackIndexBlocksLocked(ctx, blocks)
@@ -429,17 +431,19 @@ func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]IndexInfo, erro
 			for _, b := range blocks {
 				blockIDs = append(blockIDs, b.FileName)
 			}
-			if err = bm.committedBlocks.use(blockIDs); err != nil {
-				return nil, err
+			var updated bool
+			updated, err = bm.committedBlocks.use(blockIDs)
+			if err != nil {
+				return nil, false, err
 			}
-			return blocks, nil
+			return blocks, updated, nil
 		}
 		if err != storage.ErrBlockNotFound {
-			return nil, err
+			return nil, false, err
 		}
 	}
 
-	return nil, fmt.Errorf("unable to load pack indexes despite %v retries", indexLoadAttempts)
+	return nil, false, fmt.Errorf("unable to load pack indexes despite %v retries", indexLoadAttempts)
 }
 
 func (bm *Manager) tryLoadPackIndexBlocksLocked(ctx context.Context, blocks []IndexInfo) error {
@@ -507,18 +511,16 @@ func (bm *Manager) unprocessedIndexBlocks(blocks []IndexInfo) (<-chan string, er
 // Close closes the block manager.
 func (bm *Manager) Close() {
 	bm.blockCache.close()
+	close(bm.closed)
 }
 
 // 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()
-
 	if maxSmallBlockCount < minSmallBlockCount {
 		return fmt.Errorf("invalid block counts")
 	}
 
-	indexBlocks, err := bm.loadPackIndexesLocked(ctx)
+	indexBlocks, _, err := bm.loadPackIndexesUnlocked(ctx)
 	if err != nil {
 		return fmt.Errorf("error loading indexes: %v", err)
 	}
@@ -982,6 +984,12 @@ func (bm *Manager) assertLocked() {
 	}
 }
 
+// Refresh reloads the committed block indexes.
+func (bm *Manager) Refresh(ctx context.Context) (bool, error) {
+	_, updated, err := bm.loadPackIndexesUnlocked(ctx)
+	return updated, err
+}
+
 type cachedList struct {
 	Timestamp time.Time   `json:"timestamp"`
 	Blocks    []IndexInfo `json:"blocks"`
@@ -1075,6 +1083,7 @@ func newManagerWithOptions(ctx context.Context, st storage.Storage, f Formatting
 		st:                    st,
 		activeBlocksExtraTime: activeBlocksExtraTime,
 		writeFormatVersion:    int32(f.Version),
+		closed:                make(chan struct{}),
 	}
 
 	if os.Getenv("KOPIA_VERIFY_INVARIANTS") != "" {

block/block_manager_test.go

@@ -266,6 +266,7 @@ func TestBlockManagerConcurrency(t *testing.T) {
 	preexistingBlock := writeBlockAndVerify(ctx, t, bm, seededRandomData(10, 100))
 	bm.Flush(ctx)
 
+	dumpBlockManagerData(t, data)
 	bm1 := newTestBlockManager(data, keyTime, nil)
 	bm2 := newTestBlockManager(data, keyTime, nil)
 	bm3 := newTestBlockManager(data, keyTime, fakeTimeNowWithAutoAdvance(fakeTime.Add(1), 1*time.Second))

block/committed_block_index.go

@@ -11,11 +11,12 @@ type committedBlockIndex interface {
 	addBlock(indexBlockID string, indexData []byte, use bool) error
 	hasIndexBlockID(indexBlockID string) (bool, error)
 
-	use(indexBlockIDs []string) error
+	use(indexBlockIDs []string) (bool, error)
 }
 
 func newCommittedBlockIndex() committedBlockIndex {
 	return &inMemoryCommittedBlockIndex{
-		physicalBlocks: make(map[string]packindex.Index),
+		cachedPhysicalBlocks: make(map[string]packindex.Index),
+		usedPhysicalBlocks:   make(map[string]packindex.Index),
 	}
 }

block/inmemory_committed_block_index.go

@@ -7,12 +7,14 @@ import (
 
 	"github.com/kopia/kopia/internal/packindex"
 	"github.com/kopia/kopia/storage"
+	"github.com/rs/zerolog/log"
 )
 
 type inMemoryCommittedBlockIndex struct {
-	mu             sync.Mutex
-	indexes        packindex.Merged
-	physicalBlocks map[string]packindex.Index
+	mu                   sync.Mutex
+	indexes              packindex.Merged
+	usedPhysicalBlocks   map[string]packindex.Index
+	cachedPhysicalBlocks map[string]packindex.Index
 }
 
 func (b *inMemoryCommittedBlockIndex) getBlock(blockID string) (Info, error) {
@@ -36,18 +38,19 @@ func (b *inMemoryCommittedBlockIndex) addBlock(indexBlockID string, data []byte,
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	if b.physicalBlocks[indexBlockID] == nil {
+	if b.cachedPhysicalBlocks[indexBlockID] == nil {
 		ndx, err := packindex.Open(bytes.NewReader(data))
 		if err != nil {
 			return err
 		}
-		b.physicalBlocks[indexBlockID] = ndx
-
-		if use {
-			b.indexes = append(b.indexes, ndx)
-		}
+		b.cachedPhysicalBlocks[indexBlockID] = ndx
 	}
 
+	if use && b.usedPhysicalBlocks[indexBlockID] == nil {
+		ndx := b.cachedPhysicalBlocks[indexBlockID]
+		b.usedPhysicalBlocks[indexBlockID] = ndx
+		b.indexes = append(b.indexes, ndx)
+	}
 	return nil
 }
 
@@ -62,27 +65,50 @@ func (b *inMemoryCommittedBlockIndex) hasIndexBlockID(indexBlockID string) (bool
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	return b.physicalBlocks[indexBlockID] != nil, nil
+	return b.cachedPhysicalBlocks[indexBlockID] != nil, nil
 }
 
-func (b *inMemoryCommittedBlockIndex) use(packFiles []string) error {
+func (b *inMemoryCommittedBlockIndex) packFilesChanged(packFiles []string) bool {
+	if len(packFiles) != len(b.usedPhysicalBlocks) {
+		return true
+	}
+
+	for _, packFile := range packFiles {
+		if b.usedPhysicalBlocks[packFile] == nil {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (b *inMemoryCommittedBlockIndex) use(packFiles []string) (bool, error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	if !b.packFilesChanged(packFiles) {
+		return false, nil
+	}
+
+	log.Printf("using %v", packFiles)
+
 	var newIndexes packindex.Merged
+	newUsedBlocks := map[string]packindex.Index{}
 	defer func() {
 		newIndexes.Close() //nolint:errcheck
 	}()
 	for _, e := range packFiles {
-		ndx := b.physicalBlocks[e]
+		ndx := b.cachedPhysicalBlocks[e]
 		if ndx == nil {
-			return fmt.Errorf("unable to open pack index %q", e)
+			return false, fmt.Errorf("unable to open pack index %q", e)
 		}
 
 		//log.Printf("opened %v with %v entries", e, ndx.EntryCount())
 		newIndexes = append(newIndexes, ndx)
+		newUsedBlocks[e] = ndx
 	}
 	b.indexes = newIndexes
+	b.usedPhysicalBlocks = newUsedBlocks
 	newIndexes = nil
-	return nil
+	return true, nil
 }

block/list_cache.go

@@ -34,7 +34,6 @@ func (c *listCache) listIndexBlocks(ctx context.Context) ([]IndexInfo, error) {
 		}
 	}
 
-	log.Debug().Msg("listing index blocks from source")
 	blocks, err := listIndexBlocksFromStorage(ctx, c.st)
 	if err == nil {
 		c.saveListToCache(ctx, &cachedList{
@@ -42,6 +41,7 @@ func (c *listCache) listIndexBlocks(ctx context.Context) ([]IndexInfo, error) {
 			Timestamp: time.Now(),
 		})
 	}
+	log.Debug().Msgf("found %v index blocks from source", len(blocks))
 
 	return blocks, err
 }

block/simple_committed_block_index.go

@@ -24,6 +24,7 @@ type simpleCommittedBlockIndex struct {
 	dirname string
 
 	mu     sync.Mutex
+	inUse  map[string]packindex.Index
 	merged packindex.Merged
 }
 
@@ -111,11 +112,15 @@ func (b *simpleCommittedBlockIndex) addBlock(indexBlockID string, data []byte, u
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	if b.inUse[indexBlockID] != nil {
+		return nil
+	}
+
 	ndx, err := b.openIndex(b.indexBlockPath(indexBlockID))
 	if err != nil {
 		return fmt.Errorf("unable to open pack index %q: %v", indexBlockID, err)
 	}
-
+	b.inUse[indexBlockID] = ndx
 	b.merged = append(b.merged, ndx)
 	return nil
 }
@@ -137,11 +142,31 @@ func (b *simpleCommittedBlockIndex) openIndex(fullpath string) (packindex.Index,
 	return packindex.Open(f)
 }
 
-func (b *simpleCommittedBlockIndex) use(packFiles []string) error {
+func (b *simpleCommittedBlockIndex) packFilesChanged(packFiles []string) bool {
+	if len(packFiles) != len(b.inUse) {
+		return true
+	}
+
+	for _, packFile := range packFiles {
+		if b.inUse[packFile] == nil {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (b *simpleCommittedBlockIndex) use(packFiles []string) (bool, error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	if !b.packFilesChanged(packFiles) {
+		return false, nil
+	}
+	log.Printf("set of index files has changed (had %v, now %v)", len(b.inUse), len(packFiles))
+
 	var newMerged packindex.Merged
+	newInUse := map[string]packindex.Index{}
 	defer func() {
 		newMerged.Close() //nolint:errcheck
 	}()
@@ -149,7 +174,7 @@ func (b *simpleCommittedBlockIndex) use(packFiles []string) error {
 	remaining := map[string]os.FileInfo{}
 	entries, err := ioutil.ReadDir(b.dirname)
 	if err != nil {
-		return fmt.Errorf("can't list cache: %v", err)
+		return false, fmt.Errorf("can't list cache: %v", err)
 	}
 
 	for _, ent := range entries {
@@ -164,14 +189,17 @@ func (b *simpleCommittedBlockIndex) use(packFiles []string) error {
 		fullpath := filepath.Join(b.dirname, fname)
 		ndx, err := b.openIndex(fullpath)
 		if err != nil {
-			return fmt.Errorf("unable to open pack index %q: %v", fullpath, err)
+			return false, fmt.Errorf("unable to open pack index %q: %v", fullpath, err)
 		}
 
 		log.Printf("opened %v with %v entries", fullpath, ndx.EntryCount())
 		newMerged = append(newMerged, ndx)
+		newInUse[e] = ndx
 	}
 	b.merged = newMerged
+	b.inUse = newInUse
 	newMerged = nil
+
 	for _, rem := range remaining {
 		if time.Since(rem.ModTime()) > unusedCommittedBlockIndexCleanupTime {
 			log.Printf("removing unused %v %v", rem.Name(), rem.ModTime())
@@ -182,7 +210,7 @@ func (b *simpleCommittedBlockIndex) use(packFiles []string) error {
 			log.Printf("keeping unused %v because it's too new %v", rem.Name(), rem.ModTime())
 		}
 	}
-	return nil
+	return true, nil
 }
 
 func newSimpleCommittedBlockIndex(dirname string) (committedBlockIndex, error) {
@@ -190,6 +218,7 @@ func newSimpleCommittedBlockIndex(dirname string) (committedBlockIndex, error) {
 
 	s := &simpleCommittedBlockIndex{
 		dirname: dirname,
+		inUse:   map[string]packindex.Index{},
 	}
 	return s, nil
 }