feat(store): Add tiered cache (disk + mem)

lib/store/disk/scoped_store.go

@@ -68,7 +68,7 @@ func (s *Store) MarkComplete(key string) error { return s.impl.MarkComplete(key)
 func (s *Store) Delete(key string) error { return s.impl.Delete(key, s.scope) }
 
 // List returns the keys of all blobs (except those out of scope).
-func (s *Store) List() []string { return s.impl.list(s.scope) }
+func (s *Store) List() []string { return s.impl.List(s.scope) }
 
 // BanEviction marks a blob as unevictable by LRU eviction. It is idempotent.
 // Needed when e.g. blobs must be written back to GCS/S3 and eviction before that is unacceptable.
@@ -88,8 +88,8 @@ func (s *Store) GetMetadata(key string, md metadata.Metadata) (ok bool, err erro
 }
 
 // DeleteMetadata removes a blob's metadata. No error returned if the metadata is not present.
-func (s *Store) DeleteMetadata(key string, md metadata.Metadata) error {
-	return s.impl.DeleteMetadata(key, md, s.scope)
+func (s *Store) DeleteMetadata(key string, mdSuffix string) error {
+	return s.impl.DeleteMetadata(key, mdSuffix, s.scope)
 }
 
 // ListMetadata returns all [metadata.Metadata] of key.
@@ -109,3 +109,6 @@ func (s *Store) ScopeComplete() *Store { return &Store{s.impl, storelib.BlobScop
 // ScopeIncomplete scopes [Store]'s APIs such that they can only operate on incomplete blobs.
 // [storelib.ErrOutOfScope] is returned if the user tries to operate on a complete blob.
 func (s *Store) ScopeIncomplete() *Store { return &Store{s.impl, storelib.BlobScopeIncomplete} }
+
+// Scoped scopes [Store]'s APIs, such that [storelib.ErrOutOfScope] is returned upon attempting to operate on blobs out of scope.
+func (s *Store) Scoped(scope storelib.BlobScope) *Store { return &Store{s.impl, scope} }

lib/store/disk/store.go

@@ -62,7 +62,7 @@ func newStore(config *Config, metrics tally.Scope) (*store, error) {
 		return nil, err
 	}
 	if !ok {
-		log.Info("Initialized a new, empty Store (did not find any previously persisted state to reboot for Store)")
+		log.Info("Initialized a new, empty disk.Store (did not find any previously persisted state to reboot for disk.Store)")
 		store := &store{
 			capacity:   config.CapacityBytes,
 			size:       0,
@@ -128,6 +128,8 @@ func (s *store) Has(key string, scope storelib.BlobScope) (inStore bool, inScope
 }
 
 func (s *store) Stat(key string, scope storelib.BlobScope) (os.FileInfo, error) {
+	// TODO - make stat return `size` and not os.FileInfo to keep its interface consistent with memory.Store and tiered.Store.
+	// TODO - consider whether Stat (and maybe other APIs) should update the access time of the blob.
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
@@ -138,7 +140,6 @@ func (s *store) Stat(key string, scope storelib.BlobScope) (os.FileInfo, error)
 	if err := isOutOfScope(b, scope); err != nil {
 		return nil, err
 	}
-	// TODO - consider whether this should update the access time of the blob.
 	blobPath := s.blobPath(key, b.complete)
 	return os.Stat(blobPath)
 }
@@ -148,13 +149,8 @@ func (s *store) Create(key string, sizeBytes uint64) (*File, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	if b, ok := s.blobs[key]; ok {
-		// TODO - consider whether we need public errors for these cases.
-		if b.complete {
-			return nil, errors.New("blob is already in store")
-		} else {
-			return nil, errors.New("blob is already in store (it is incomplete)")
-		}
+	if _, ok := s.blobs[key]; ok {
+		return nil, os.ErrExist
 	}
 
 	if err := s.reserveSpace(sizeBytes); err != nil {
@@ -346,7 +342,7 @@ func (s *store) Delete(key string, scope storelib.BlobScope) error {
 	return nil
 }
 
-func (s *store) list(scope storelib.BlobScope) []string {
+func (s *store) List(scope storelib.BlobScope) []string {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
@@ -489,8 +485,7 @@ func (s *store) GetMetadata(key string, md metadata.Metadata, scope storelib.Blo
 	return true, nil
 }
 
-func (s *store) DeleteMetadata(key string, md metadata.Metadata, scope storelib.BlobScope) error {
-	// TODO - change interface to take `mdSuffix string` instead of `md metadata.Metadata`, just like memory.Store.`
+func (s *store) DeleteMetadata(key, mdSuffix string, scope storelib.BlobScope) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -501,7 +496,7 @@ func (s *store) DeleteMetadata(key string, md metadata.Metadata, scope storelib.
 	if err := isOutOfScope(b, scope); err != nil {
 		return err
 	}
-	mdFilePath := s.sidecarFilePath(key, b.complete, md.GetSuffix())
+	mdFilePath := s.sidecarFilePath(key, b.complete, mdSuffix)
 	err := os.Remove(mdFilePath)
 	if errors.Is(err, os.ErrNotExist) {
 		// no-op

lib/store/disk/store_test.go

@@ -764,7 +764,7 @@ func TestMetadata(t *testing.T) {
 			[]string{mdList[0].GetSuffix(), mdList[1].GetSuffix()},
 		)
 
-		require.NoError(store.DeleteMetadata(key, &readMd))
+		require.NoError(store.DeleteMetadata(key, readMd.GetSuffix()))
 		ok, err = store.GetMetadata(key, &readMd)
 		require.NoError(err)
 		require.False(ok)
@@ -773,14 +773,14 @@ func TestMetadata(t *testing.T) {
 		_, err = os.Stat(mdFilePath)
 		require.ErrorIs(err, os.ErrNotExist)
 		// deleting a second time should be a no-op.
-		require.NoError(store.DeleteMetadata(key, &readMd))
+		require.NoError(store.DeleteMetadata(key, readMd.GetSuffix()))
 
 		mdList, err = store.ListMetadata(key)
 		require.NoError(err)
 		require.Len(mdList, 1)
 		require.Equal(persistMd.GetSuffix(), mdList[0].GetSuffix())
 
-		require.NoError(store.DeleteMetadata(key, persistMd))
+		require.NoError(store.DeleteMetadata(key, persistMd.GetSuffix()))
 		mdList, err = store.ListMetadata(key)
 		require.NoError(err)
 		require.Empty(mdList)
@@ -806,7 +806,7 @@ func TestMetadata(t *testing.T) {
 		err = store.WriteAtMetadata(nonExistentKey, md, []byte("data"), 0)
 		require.ErrorIs(os.ErrNotExist, err)
 
-		err = store.DeleteMetadata(nonExistentKey, md)
+		err = store.DeleteMetadata(nonExistentKey, md.GetSuffix())
 		require.ErrorIs(os.ErrNotExist, err)
 	})
 
@@ -980,7 +980,7 @@ func TestScopes(t *testing.T) {
 	_, err = store.ScopeComplete().ListMetadata(key)
 	require.ErrorIs(err, storelib.ErrOutOfScope)
 	require.ErrorIs(store.ScopeComplete().WriteAtMetadata(key, md, mdData, 0), storelib.ErrOutOfScope)
-	require.ErrorIs(store.ScopeComplete().DeleteMetadata(key, &readMd), storelib.ErrOutOfScope)
+	require.ErrorIs(store.ScopeComplete().DeleteMetadata(key, readMd.GetSuffix()), storelib.ErrOutOfScope)
 	require.NotContains(store.ScopeComplete().List(), key)
 	_, ok = store.ScopeComplete().Has(key)
 	require.False(ok)
@@ -1006,7 +1006,7 @@ func TestScopes(t *testing.T) {
 	require.Len(mdList, 1)
 	require.Equal(md.GetSuffix(), mdList[0].GetSuffix())
 	require.NoError(store.WriteAtMetadata(key, md, mdData, 0))
-	require.NoError(store.DeleteMetadata(key, &readMd))
+	require.NoError(store.DeleteMetadata(key, readMd.GetSuffix()))
 	require.Contains(store.List(), key)
 	_, ok = store.Has(key)
 	require.True(ok)
@@ -1032,7 +1032,7 @@ func TestScopes(t *testing.T) {
 	require.Len(mdList, 1)
 	require.Equal(md.GetSuffix(), mdList[0].GetSuffix())
 	require.NoError(store.ScopeIncomplete().WriteAtMetadata(key, md, mdData, 0))
-	require.NoError(store.ScopeIncomplete().DeleteMetadata(key, &readMd))
+	require.NoError(store.ScopeIncomplete().DeleteMetadata(key, readMd.GetSuffix()))
 	require.Contains(store.ScopeIncomplete().List(), key)
 	_, ok = store.ScopeIncomplete().Has(key)
 	require.True(ok)
@@ -1056,7 +1056,7 @@ func TestScopes(t *testing.T) {
 	_, err = store.ScopeIncomplete().ListMetadata(key)
 	require.ErrorIs(err, storelib.ErrOutOfScope)
 	require.ErrorIs(store.ScopeIncomplete().WriteAtMetadata(key, md, mdData, 0), storelib.ErrOutOfScope)
-	require.ErrorIs(store.ScopeIncomplete().DeleteMetadata(key, &readMd), storelib.ErrOutOfScope)
+	require.ErrorIs(store.ScopeIncomplete().DeleteMetadata(key, readMd.GetSuffix()), storelib.ErrOutOfScope)
 	require.NotContains(store.ScopeIncomplete().List(), key)
 	_, ok = store.ScopeIncomplete().Has(key)
 	require.False(ok)
@@ -1082,7 +1082,7 @@ func TestScopes(t *testing.T) {
 	require.Len(mdList, 1)
 	require.Equal(md.GetSuffix(), mdList[0].GetSuffix())
 	require.NoError(store.WriteAtMetadata(key, md, mdData, 0))
-	require.NoError(store.DeleteMetadata(key, &readMd))
+	require.NoError(store.DeleteMetadata(key, readMd.GetSuffix()))
 	require.Contains(store.List(), key)
 	_, ok = store.Has(key)
 	require.True(ok)
@@ -1108,7 +1108,7 @@ func TestScopes(t *testing.T) {
 	require.Len(mdList, 1)
 	require.Equal(md.GetSuffix(), mdList[0].GetSuffix())
 	require.NoError(store.ScopeComplete().WriteAtMetadata(key, md, mdData, 0))
-	require.NoError(store.ScopeComplete().DeleteMetadata(key, &readMd))
+	require.NoError(store.ScopeComplete().DeleteMetadata(key, readMd.GetSuffix()))
 	require.Contains(store.ScopeComplete().List(), key)
 	_, ok = store.ScopeComplete().Has(key)
 	require.True(ok)

lib/store/memory/file.go

@@ -103,11 +103,11 @@ func (f *File) Seek(off int64, whence int) (int64, error) {
 }
 
 // Stat returns the blob's actual size, even if it differs from the size reported during Create.
+// A return value of -1 represents [ErrEvicted].
 func (f *File) Size() int64 {
 	buf, evicted := f.getData()
 	if evicted {
-		// TODO - consider whether this is ok or whether we need to store the user-provided blob size in [*File].
-		return 0
+		return -1
 	}
 	return int64(len(buf))
 }

lib/store/memory/scoped_store.go

@@ -11,8 +11,9 @@ import (
 // ErrNoSpace means the store could not free enough space for a new entry.
 var ErrNoSpace error = errors.New("cannot free enough memory for new entry")
 
-// ErrEvicted is returned when a user tries operating on a blob's [*File] after the blob has been evicted.
-var ErrEvicted error = errors.New("the blob has been evicted from the store")
+// ErrEvicted is returned when a user tries operating on a blob's [*File] after the blob has been evicted or deleted.
+var ErrEvicted error = errors.New("the blob has been evicted or deleted from the store") // TODO - rename to ErrGone
+// or something else that shows the entry might be deleted too, not just evicted.
 
 // Store is an in-memory, thread-safe, LRU cache for blobs and their [metadata.Metadata].
 //
@@ -67,7 +68,7 @@ func (s *Store) MarkComplete(key string) error { return s.impl.MarkComplete(key)
 func (s *Store) Delete(key string) error { return s.impl.Delete(key, s.scope) }
 
 // List returns the keys of all blobs (except those out of scope).
-func (s *Store) List() []string { return s.impl.list(s.scope) }
+func (s *Store) List() []string { return s.impl.List(s.scope) }
 
 // BanEviction marks a blob as unevictable by LRU eviction. It is idempotent.
 // Usually used by clients to ensure a blob is not evicted before being flushed to disk.
@@ -103,3 +104,6 @@ func (s *Store) ScopeComplete() *Store { return &Store{s.impl, storelib.BlobScop
 // ScopeIncomplete scopes [Store]'s APIs such that they can only operate on incomplete blobs.
 // [storelib.ErrOutOfScope] is returned if the user tries to operate on a complete blob.
 func (s *Store) ScopeIncomplete() *Store { return &Store{s.impl, storelib.BlobScopeIncomplete} }
+
+// Scoped scopes [Store]'s APIs, such that [storelib.ErrOutOfScope] is returned upon attempting to operate on blobs out of scope.
+func (s *Store) Scoped(scope storelib.BlobScope) *Store { return &Store{s.impl, scope} }

lib/store/memory/store.go

@@ -48,7 +48,7 @@ func newStore(capacityBytes uint64, metrics tally.Scope) (*store, error) {
 
 	log := log.Default().With("module", "memory_store")
 
-	log.Info("Initialized new, empty *memory.Store")
+	log.Info("Initialized new, empty memory.Store")
 	s := &store{
 		blobs:      make(map[string]*blob, 0),
 		evictQueue: list.New(),
@@ -175,7 +175,7 @@ func (s *store) Delete(key string, scope storelib.BlobScope) error {
 	return nil
 }
 
-func (s *store) list(scope storelib.BlobScope) []string {
+func (s *store) List(scope storelib.BlobScope) []string {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 

lib/store/memory/store_test.go

@@ -172,7 +172,7 @@ func TestEvictedBlobImmediatelyNotAvailable(t *testing.T) {
 	require.NoError(b.Close())
 	require.NoError(b.Commit())
 
-	require.Equal(int64(0), b.Size())
+	require.Equal(int64(-1), b.Size())
 }
 
 func TestParallelAccessToSingleFile(t *testing.T) {