simplified MetadataManager API for listing, added caching of metadata items (unbounded for now)

disabled go <1.9 since we're now using t.Helper(), which is awesome

.travis.yml

@@ -1,6 +1,5 @@
 language: go
 go:
-- 1.8
 - 1.9
 os:
 - linux
@@ -17,4 +16,4 @@ deploy:
     repo: kopia/kopia
     go: 1.6
 git:
-  depth: 9999999
+  depth: 9999999

cli/command_metadata_ls.go

@@ -18,7 +18,7 @@ func init() {
 func listMetadataItems(context *kingpin.ParseContext) error {
 	rep := mustOpenRepository(nil)
 
-	entries, err := rep.ListMetadata(*metadataListPrefix, -1)
+	entries, err := rep.ListMetadata(*metadataListPrefix)
 	if err != nil {
 		return err
 	}

cli/command_object_cleanup.go

@@ -155,7 +155,7 @@ func runCleanupCommand(context *kingpin.ParseContext) error {
 	mgr := snapshot.NewManager(rep)
 
 	log.Printf("Listing active snapshots...")
-	snapshotNames, err := mgr.ListSnapshotManifests(nil, -1)
+	snapshotNames, err := mgr.ListSnapshotManifests(nil)
 	if err != nil {
 		return err
 	}

cli/command_repository_migrate.go

@@ -48,7 +48,7 @@ func runMigrateCommand(context *kingpin.ParseContext) error {
 
 		log.Printf("migrating source %v", s)
 
-		manifests, err := sourceSM.ListSnapshotManifests(s, -1)
+		manifests, err := sourceSM.ListSnapshotManifests(s)
 		if err != nil {
 			return fmt.Errorf("unable to list snapshot manifests for %v: %v", s, err)
 		}

cli/command_snapshot_create.go

@@ -83,7 +83,7 @@ func runBackupCommand(c *kingpin.ParseContext) error {
 			return fmt.Errorf("description too long")
 		}
 
-		previous, err := mgr.ListSnapshots(sourceInfo, 1)
+		previous, err := mgr.ListSnapshots(sourceInfo)
 		if err != nil {
 			return fmt.Errorf("error listing previous backups: %v", err)
 		}

cli/command_snapshot_expire.go

@@ -110,7 +110,7 @@ func getSnapshotNamesToExpire(mgr *snapshot.Manager) ([]string, error) {
 
 	if *snapshotExpireAll {
 		fmt.Fprintf(os.Stderr, "Scanning all active snapshots...\n")
-		return mgr.ListSnapshotManifests(nil, -1)
+		return mgr.ListSnapshotManifests(nil)
 	}
 
 	var result []string
@@ -123,7 +123,7 @@ func getSnapshotNamesToExpire(mgr *snapshot.Manager) ([]string, error) {
 
 		log.Printf("Looking for snapshots of %v", src)
 
-		matches, err := mgr.ListSnapshotManifests(&src, -1)
+		matches, err := mgr.ListSnapshotManifests(&src)
 		if err != nil {
 			return nil, fmt.Errorf("error listing snapshots for %v: %v", src, err)
 		}

cli/command_snapshot_list.go

@@ -24,7 +24,7 @@ var (
 
 func findBackups(mgr *snapshot.Manager, sourceInfo snapshot.SourceInfo) (manifestIDs []string, relPath string, err error) {
 	for len(sourceInfo.Path) > 0 {
-		list, err := mgr.ListSnapshotManifests(&sourceInfo, -1)
+		list, err := mgr.ListSnapshotManifests(&sourceInfo)
 		if err != nil {
 			return nil, "", err
 		}
@@ -72,7 +72,7 @@ func runBackupsCommand(context *kingpin.ParseContext) error {
 			relPath = "/" + relPath
 		}
 	} else {
-		previous, err = mgr.ListSnapshotManifests(nil, -1)
+		previous, err = mgr.ListSnapshotManifests(nil)
 	}
 
 	if err != nil {

fs/repofs/source_snapshots.go

@@ -35,7 +35,7 @@ func safeName(path string) string {
 }
 
 func (s *sourceSnapshots) Readdir() (fs.Entries, error) {
-	manifests, err := s.snapshotManager.ListSnapshots(s.src, -1)
+	manifests, err := s.snapshotManager.ListSnapshots(s.src)
 	if err != nil {
 		return nil, err
 	}

repo/initialize.go

@@ -39,12 +39,17 @@ func Initialize(st blob.Storage, opt *NewRepositoryOptions, creds auth.Credentia
 		opt = &NewRepositoryOptions{}
 	}
 
+	cache, err := newMetadataCache(st)
+	if err != nil {
+		return err
+	}
+
 	mm := MetadataManager{
 		storage: st,
+		cache:   cache,
 		format:  metadataFormatFromOptions(opt),
 	}
 
-	var err error
 	mm.masterKey, err = creds.GetMasterKey(mm.format.SecurityOptions)
 	if err != nil {
 		return err

repo/metadata_cache.go

@@ -0,0 +1,157 @@
+package repo
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+
+	"github.com/kopia/kopia/blob"
+)
+
+const (
+	// MetadataBlockPrefix is a prefix used for metadata blocks in repository storage.
+	MetadataBlockPrefix = "VLT"
+)
+
+type metadataCache struct {
+	st blob.Storage
+
+	mu            sync.Mutex
+	sortedNames   []string
+	nameToCacheID map[string]string
+	cachedData    map[string][]byte
+}
+
+func (mc *metadataCache) ListBlocks(prefix string) ([]string, error) {
+	var result []string
+
+	mc.mu.Lock()
+	p := sort.SearchStrings(mc.sortedNames, prefix)
+	for p < len(mc.sortedNames) && strings.HasPrefix(mc.sortedNames[p], prefix) {
+		if !isReservedName(mc.sortedNames[p]) {
+			result = append(result, mc.sortedNames[p])
+		}
+		p++
+	}
+	mc.mu.Unlock()
+	return result, nil
+}
+
+func (mc *metadataCache) GetBlock(name string) ([]byte, error) {
+	mc.mu.Lock()
+	cid := mc.nameToCacheID[name]
+	if cid == "" {
+		mc.mu.Unlock()
+		return nil, blob.ErrBlockNotFound
+	}
+
+	// see if the data is cached
+	if data, ok := mc.cachedData[cid]; ok {
+		mc.mu.Unlock()
+		return cloneBytes(data), nil
+	}
+
+	mc.mu.Unlock()
+
+	// not cached, fetch from the storage
+	b, err := mc.st.GetBlock(MetadataBlockPrefix+name, 0, -1)
+	if err != nil {
+		return nil, err
+	}
+
+	// now race to add to cache, does not matter who wins.
+	mc.mu.Lock()
+	mc.cachedData[cid] = b
+	mc.mu.Unlock()
+	return cloneBytes(b), nil
+}
+
+func (mc *metadataCache) PutBlock(name string, data []byte) error {
+	if err := mc.st.PutBlock(MetadataBlockPrefix+name, data); err != nil {
+		return err
+	}
+
+	mc.mu.Lock()
+	cid := fmt.Sprintf("%v-new-%x", name, randomBytes(8))
+	mc.nameToCacheID[name] = cid
+	p := sort.SearchStrings(mc.sortedNames, name)
+	if p >= len(mc.sortedNames) || mc.sortedNames[p] != name {
+		// Name not present, p is the index where p should be inserted.
+		mc.sortedNames = append(append(
+			append([]string(nil), mc.sortedNames[0:p]...),
+			name),
+			mc.sortedNames[p:]...)
+	}
+
+	mc.mu.Unlock()
+
+	return nil
+}
+
+func (mc *metadataCache) DeleteBlock(name string) error {
+	mc.mu.Lock()
+	if cid := mc.nameToCacheID[name]; cid != "" {
+		delete(mc.nameToCacheID, name)
+		delete(mc.cachedData, cid)
+
+		// Delete from sortedNames
+		p := sort.SearchStrings(mc.sortedNames, name)
+		if p < len(mc.sortedNames) && mc.sortedNames[p] == name {
+			// Found at index 'p' build a new slice from [0..p), [p+1,...)
+			newSlice := mc.sortedNames[0:p]
+			for _, n := range mc.sortedNames[p+1:] {
+				newSlice = append(newSlice, n)
+			}
+			mc.sortedNames = newSlice
+		}
+	}
+	mc.mu.Unlock()
+
+	return mc.st.DeleteBlock(MetadataBlockPrefix + name)
+}
+
+// Refresh refreshes the list of blocks in the cache, but does not load or expire previously cached.
+func (mc *metadataCache) Refresh() error {
+	var sortedNames []string
+	nameToCacheID := map[string]string{}
+
+	ch, cancel := mc.st.ListBlocks(MetadataBlockPrefix)
+	defer cancel()
+	for it := range ch {
+		if it.Error != nil {
+			return it.Error
+		}
+
+		n := strings.TrimPrefix(it.BlockID, MetadataBlockPrefix)
+		sortedNames = append(sortedNames, n)
+		nameToCacheID[n] = fmt.Sprintf("%v-%v-%v", it.BlockID, it.Length, it.TimeStamp.UnixNano())
+	}
+
+	mc.setLoaded(sortedNames, nameToCacheID)
+	return nil
+}
+
+func (mc *metadataCache) setLoaded(sortedNames []string, nameToCacheID map[string]string) {
+	mc.mu.Lock()
+	mc.sortedNames = sortedNames
+	mc.nameToCacheID = nameToCacheID
+	mc.mu.Unlock()
+}
+
+func cloneBytes(d []byte) []byte {
+	return append([]byte(nil), d...)
+}
+
+func newMetadataCache(st blob.Storage) (*metadataCache, error) {
+	c := &metadataCache{
+		st:            st,
+		nameToCacheID: make(map[string]string),
+		cachedData:    make(map[string][]byte),
+	}
+	if err := c.Refresh(); err != nil {
+		return nil, err
+	}
+
+	return c, nil
+}

repo/metadata_manager.go

@@ -9,7 +9,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"strings"
 	"sync"
 
 	"github.com/kopia/kopia/auth"
@@ -25,9 +24,6 @@ const (
 )
 
 const (
-	// MetadataBlockPrefix is a prefix used for metadata blocks in repository storage.
-	MetadataBlockPrefix = "VLT"
-
 	parallelFetches = 5
 )
 
@@ -59,6 +55,7 @@ func init() {
 // in a repository.
 type MetadataManager struct {
 	storage    blob.Storage
+	cache      *metadataCache
 	format     config.MetadataFormat
 	repoConfig config.EncryptedRepositoryConfig
 
@@ -94,11 +91,11 @@ func (mm *MetadataManager) writeEncryptedBlock(itemID string, content []byte) er
 		content = nonce[0 : nonceLength+len(b)]
 	}
 
-	return mm.storage.PutBlock(MetadataBlockPrefix+itemID, content)
+	return mm.cache.PutBlock(itemID, content)
 }
 
 func (mm *MetadataManager) readEncryptedBlock(itemID string) ([]byte, error) {
-	content, err := mm.storage.GetBlock(MetadataBlockPrefix+itemID, 0, -1)
+	content, err := mm.cache.GetBlock(itemID)
 	if err != nil {
 		if err == blob.ErrBlockNotFound {
 			return nil, ErrMetadataNotFound
@@ -203,34 +200,13 @@ func (mm *MetadataManager) putJSON(id string, content interface{}) error {
 }
 
 // ListMetadata returns the list of metadata items matching the specified prefix.
-// The 'limit' parameter specifies the maximum number of items to retrieve (-1 == unlimited).
-func (mm *MetadataManager) ListMetadata(prefix string, limit int) ([]string, error) {
-	var result []string
-
-	ch, cancel := mm.storage.ListBlocks(MetadataBlockPrefix + prefix)
-	defer cancel()
-	for b := range ch {
-		if limit == 0 {
-			break
-		}
-		if b.Error != nil {
-			return result, b.Error
-		}
-
-		itemID := strings.TrimPrefix(b.BlockID, MetadataBlockPrefix)
-		if !isReservedName(itemID) {
-			result = append(result, itemID)
-		}
-		if limit > 0 {
-			limit--
-		}
-	}
-	return result, nil
+func (mm *MetadataManager) ListMetadata(prefix string) ([]string, error) {
+	return mm.cache.ListBlocks(prefix)
 }
 
-// ListMetadataContents retrieves metadata contents for all items starting with a given prefix and up to a specified limit.
-func (mm *MetadataManager) ListMetadataContents(prefix string, limit int) (map[string][]byte, error) {
-	itemIDs, err := mm.ListMetadata(prefix, limit)
+// ListMetadataContents retrieves metadata contents for all items starting with a given prefix.
+func (mm *MetadataManager) ListMetadataContents(prefix string) (map[string][]byte, error) {
+	itemIDs, err := mm.ListMetadata(prefix)
 	if err != nil {
 		return nil, err
 	}
@@ -260,7 +236,7 @@ func (mm *MetadataManager) RemoveMetadata(itemID string) error {
 		return err
 	}
 
-	return mm.storage.DeleteBlock(MetadataBlockPrefix + itemID)
+	return mm.cache.DeleteBlock(itemID)
 }
 
 // RemoveMany efficiently removes multiple metadata items in parallel.
@@ -295,8 +271,14 @@ func (mm *MetadataManager) RemoveMany(itemIDs []string) error {
 
 // newMetadataManager opens a MetadataManager for given storage and credentials.
 func newMetadataManager(st blob.Storage, creds auth.Credentials) (*MetadataManager, error) {
+	cache, err := newMetadataCache(st)
+	if err != nil {
+		return nil, err
+	}
+
 	mm := MetadataManager{
 		storage: st,
+		cache:   cache,
 	}
 
 	var wg sync.WaitGroup
@@ -304,13 +286,13 @@ func newMetadataManager(st blob.Storage, creds auth.Credentials) (*MetadataManag
 	var blocks [4][]byte
 
 	f := func(index int, name string) {
-		blocks[index], _ = st.GetBlock(name, 0, -1)
+		blocks[index], _ = mm.cache.GetBlock(name)
 		wg.Done()
 	}
 
 	wg.Add(2)
-	go f(0, MetadataBlockPrefix+formatBlockID)
-	go f(1, MetadataBlockPrefix+repositoryConfigBlockID)
+	go f(0, formatBlockID)
+	go f(1, repositoryConfigBlockID)
 	wg.Wait()
 
 	if blocks[0] == nil {
@@ -318,7 +300,7 @@ func newMetadataManager(st blob.Storage, creds auth.Credentials) (*MetadataManag
 	}
 
 	var offset = 0
-	err := json.Unmarshal(blocks[offset], &mm.format)
+	err = json.Unmarshal(blocks[offset], &mm.format)
 	if err != nil {
 		return nil, err
 	}

repo/metadata_manager_test.go

@@ -41,30 +41,19 @@ func TestMetadataManager(t *testing.T) {
 		return
 	}
 
-	v1, err := newMetadataManager(st, creds)
+	v, err := newMetadataManager(st, creds)
 	if err != nil {
 		t.Errorf("can't open first metadata manager: %v", err)
 		return
 	}
 
-	v2, err := newMetadataManager(st, creds)
-	if err != nil {
-		t.Errorf("can't open second metadata manager: %v", err)
-		return
-	}
-
-	cfg, err := v1.connectionConfiguration()
+	cfg, err := v.connectionConfiguration()
 	if err != nil {
 		t.Errorf("error getting token1 %v", err)
 	}
 
-	cfg2, err := v2.connectionConfiguration()
-	if err != nil {
-		t.Errorf("error getting token2 %v", err)
-	}
-
-	if !reflect.DeepEqual(cfg, cfg2) {
-		t.Errorf("configurations are different: %+v vs %+v", cfg, cfg2)
+	if got, want := cfg.ConnectionInfo.Type, "filesystem"; got != want {
+		t.Errorf("invalid connection type: %q, want %q", got, want)
 	}
 
 	_, err = newMetadataManager(st, otherCreds)
@@ -73,54 +62,72 @@ func TestMetadataManager(t *testing.T) {
 		return
 	}
 
-	if err := v1.PutMetadata("foo", []byte("test1")); err != nil {
+	if err := v.PutMetadata("foo", []byte("test1")); err != nil {
 		t.Errorf("error putting: %v", err)
 	}
-	if err := v2.PutMetadata("bar", []byte("test2")); err != nil {
+	if err := v.PutMetadata("bar", []byte("test2")); err != nil {
 		t.Errorf("error putting: %v", err)
 	}
-	if err := v1.PutMetadata("baz", []byte("test3")); err != nil {
+	if err := v.PutMetadata("baz", []byte("test3")); err != nil {
 		t.Errorf("error putting: %v", err)
 	}
 
-	// Verify contents of metadata items for both managers.
-	for _, v := range []*MetadataManager{v1, v2} {
-		assertMetadataItem(t, v, "foo", "test1")
-		assertMetadataItem(t, v, "bar", "test2")
-		assertMetadataItem(t, v, "baz", "test3")
-		assertMetadataItemNotFound(t, v, "x")
+	assertMetadataItem(t, v, "foo", "test1")
+	assertMetadataItem(t, v, "bar", "test2")
+	assertMetadataItem(t, v, "baz", "test3")
+	assertMetadataItemNotFound(t, v, "x")
 
-		assertMetadataItems(t, v, "x", nil)
-		assertMetadataItems(t, v, "f", []string{"foo"})
-		assertMetadataItems(t, v, "ba", []string{"bar", "baz"})
-		assertMetadataItems(t, v, "be", nil)
-		assertMetadataItems(t, v, "baz", []string{"baz"})
-		assertMetadataItems(t, v, "bazx", nil)
+	assertMetadataItems(t, v, "x", nil)
+	assertMetadataItems(t, v, "f", []string{"foo"})
+	assertMetadataItems(t, v, "ba", []string{"bar", "baz"})
+	assertMetadataItems(t, v, "be", nil)
+	assertMetadataItems(t, v, "baz", []string{"baz"})
+	assertMetadataItems(t, v, "bazx", nil)
 
-		assertReservedName(t, v, formatBlockID)
-		assertReservedName(t, v, repositoryConfigBlockID)
+	assertReservedName(t, v, formatBlockID)
+	assertReservedName(t, v, repositoryConfigBlockID)
+
+	v.RemoveMetadata("bar")
+
+	assertMetadataItem(t, v, "foo", "test1")
+	assertMetadataItemNotFound(t, v, "bar")
+	assertMetadataItem(t, v, "baz", "test3")
+
+	assertMetadataItems(t, v, "x", nil)
+	assertMetadataItems(t, v, "f", []string{"foo"})
+	assertMetadataItems(t, v, "ba", []string{"baz"})
+	assertMetadataItems(t, v, "be", nil)
+	assertMetadataItems(t, v, "baz", []string{"baz"})
+	assertMetadataItems(t, v, "bazx", nil)
+
+	v.RemoveMetadata("baz")
+	assertMetadataItemNotFound(t, v, "baz")
+	v.RemoveMetadata("baz")
+	assertMetadataItemNotFound(t, v, "baz")
+
+	assertMetadataItem(t, v, "foo", "test1")
+	if err := v.PutMetadata("baz", []byte("test4")); err != nil {
+		t.Errorf("error putting: %v", err)
+	}
+	assertMetadataItem(t, v, "baz", "test4")
+
+	v2, err := newMetadataManager(st, creds)
+	if err != nil {
+		t.Errorf("can't open first metadata manager: %v", err)
+		return
 	}
 
-	v1.RemoveMetadata("bar")
-
-	for _, v := range []*MetadataManager{v1, v2} {
-		assertMetadataItem(t, v, "foo", "test1")
-		assertMetadataItemNotFound(t, v, "bar")
-		assertMetadataItem(t, v, "baz", "test3")
-
-		assertMetadataItems(t, v, "x", nil)
-		assertMetadataItems(t, v, "f", []string{"foo"})
-		assertMetadataItems(t, v, "ba", []string{"baz"})
-		assertMetadataItems(t, v, "be", nil)
-		assertMetadataItems(t, v, "baz", []string{"baz"})
-		assertMetadataItems(t, v, "bazx", nil)
-	}
+	assertMetadataItem(t, v2, "foo", "test1")
+	assertMetadataItemNotFound(t, v2, "bar")
+	assertMetadataItem(t, v2, "baz", "test4")
 }
 
 func assertMetadataItem(t *testing.T, v *MetadataManager, itemID string, expectedData string) {
+	t.Helper()
 	b, err := v.GetMetadata(itemID)
 	if err != nil {
 		t.Errorf("error getting item %v: %v", itemID, err)
+		return
 	}
 
 	bs := string(b)
@@ -155,7 +162,8 @@ func assertReservedNameError(t *testing.T, method string, itemID string, err err
 }
 
 func assertMetadataItems(t *testing.T, v *MetadataManager, prefix string, expected []string) {
-	res, err := v.ListMetadata(prefix, -1)
+	t.Helper()
+	res, err := v.ListMetadata(prefix)
 	if err != nil {
 		t.Errorf("error listing items beginning with %v: %v", prefix, err)
 	}

repo/pack_manager.go

@@ -208,7 +208,7 @@ func (p *packManager) ensurePackIndexesLoaded() (map[string]*packIndex, error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	m, err := p.metadataManager.ListMetadataContents(packIDPrefix, -1)
+	m, err := p.metadataManager.ListMetadataContents(packIDPrefix)
 	if err != nil {
 		return nil, err
 	}

snapshot/manager.go

@@ -33,7 +33,7 @@ type Manager struct {
 
 // ListSources lists all snapshot sources.
 func (m *Manager) ListSources() ([]*SourceInfo, error) {
-	names, err := m.repository.ListMetadata(snapshotPrefix, -1)
+	names, err := m.repository.ListMetadata(snapshotPrefix)
 	if err != nil {
 		return nil, err
 	}
@@ -63,8 +63,8 @@ func (m *Manager) ListSources() ([]*SourceInfo, error) {
 }
 
 // ListSnapshots lists all snapshots for a given source.
-func (m *Manager) ListSnapshots(si *SourceInfo, limit int) ([]*Manifest, error) {
-	names, err := m.repository.ListMetadata(m.snapshotIDPrefix(si), limit)
+func (m *Manager) ListSnapshots(si *SourceInfo) ([]*Manifest, error) {
+	names, err := m.repository.ListMetadata(m.snapshotIDPrefix(si))
 	if err != nil {
 		return nil, err
 	}
@@ -141,14 +141,13 @@ func (m *Manager) LoadSnapshots(names []string) ([]*Manifest, error) {
 }
 
 // ListSnapshotManifests returns the list of snapshot manifests for a given source or all sources if nil.
-// Limit specifies how mamy manifests to retrieve (-1 == unlimited).
-func (m *Manager) ListSnapshotManifests(src *SourceInfo, limit int) ([]string, error) {
+func (m *Manager) ListSnapshotManifests(src *SourceInfo) ([]string, error) {
 	prefix := snapshotPrefix
 	if src != nil {
 		prefix = m.snapshotIDPrefix(src)
 	}
 
-	return m.repository.ListMetadata(prefix, limit)
+	return m.repository.ListMetadata(prefix)
 }
 
 // GetEffectivePolicy calculates effective snapshot policy for a given source by combining the source-specifc policy (if any)
@@ -271,7 +270,7 @@ func (m *Manager) getPolicyItem(itemID string) (*Policy, error) {
 
 // ListPolicies returns a list of all snapshot policies.
 func (m *Manager) ListPolicies() ([]*Policy, error) {
-	names, err := m.repository.ListMetadata(policyPrefix, -1)
+	names, err := m.repository.ListMetadata(policyPrefix)
 	if err != nil {
 		return nil, err
 	}