Add negative cache for folders

internal/cache/metadata/stat_cache.go

@@ -248,16 +248,11 @@ func (sc *statCacheBucketView) LookUpFolder(
  // Look up in the LRU cache.
  hit, entry := sc.sharedCacheLookup(folderName, now)
 
- if !hit {
- return false, nil
- }
- // Adds scenario to check folder as well even if object with same name is already present
- // TODO: should be removed once integration for lookup is completed
- if entry.f == nil {
- return false, nil
+ if hit {
+ return hit, entry.f
  }
 
- return true, entry.f
+ return false, nil
 }
 
 func (sc *statCacheBucketView) sharedCacheLookup(key string, now time.Time) (bool, *entry) {

internal/cache/metadata/stat_cache_test.go

@@ -419,7 +419,7 @@ func (t *MultiBucketStatCacheTest) Test_ExpiresLeastRecentlyUsed() {
 }
 
 func (t *StatCacheTest) Test_InsertFolderCreateEntryWhenNoEntryIsPresent() {
- const name = "key1"
+ const name = "key1/"
  newEntry := &gcs.Folder{
  Name: name,
  }
@@ -428,11 +428,11 @@ func (t *StatCacheTest) Test_InsertFolderCreateEntryWhenNoEntryIsPresent() {
 
  hit, entry := t.statCache.LookUpFolder(name, someTime)
  assert.True(t.T(), hit)
- assert.Equal(t.T(), "key1", entry.Name)
+ assert.Equal(t.T(), name, entry.Name)
 }
 
 func (t *StatCacheTest) Test_InsertFolderOverrideEntryOldEntryIsAlreadyPresent() {
- const name = "key1"
+ const name = "key1/"
  existingEntry := &gcs.Folder{
  Name: name,
  }
@@ -445,11 +445,11 @@ func (t *StatCacheTest) Test_InsertFolderOverrideEntryOldEntryIsAlreadyPresent()
 
  hit, entry := t.statCache.LookUpFolder(name, someTime)
  assert.True(t.T(), hit)
- assert.Equal(t.T(), "key1", entry.Name)
+ assert.Equal(t.T(), name, entry.Name)
 }
 
 func (t *StatCacheTest) Test_LookupReturnFalseIfExpirationIsPassed() {
- const name = "key1"
+ const name = "key1/"
  entry := &gcs.Folder{
  Name: name,
  }
@@ -462,7 +462,7 @@ func (t *StatCacheTest) Test_LookupReturnFalseIfExpirationIsPassed() {
 }
 
 func (t *StatCacheTest) Test_LookupReturnFalseWhenIsNotPresent() {
- const name = "key1"
+ const name = "key1/"
 
  hit, result := t.statCache.LookUpFolder(name, expiration.Add(time.Second))
 
@@ -471,7 +471,7 @@ func (t *StatCacheTest) Test_LookupReturnFalseWhenIsNotPresent() {
 }
 
 func (t *StatCacheTest) Test_InsertFolderShouldNotOverrideEntryIfMetagenerationIsOld() {
- const name = "key1"
+ const name = "key1/"
  existingEntry := &gcs.Folder{
  Name: name,
  }
@@ -484,11 +484,11 @@ func (t *StatCacheTest) Test_InsertFolderShouldNotOverrideEntryIfMetagenerationI
 
  hit, entry := t.statCache.LookUpFolder(name, someTime)
  assert.True(t.T(), hit)
- assert.Equal(t.T(), "key1", entry.Name)
+ assert.Equal(t.T(), name, entry.Name)
 }
 
 func (t *StatCacheTest) Test_AddNegativeEntryForFolderShouldAddNegativeEntryForFolder() {
- const name = "key1"
+ const name = "key1/"
  existingEntry := &gcs.Folder{
  Name: name,
  }
@@ -497,11 +497,11 @@ func (t *StatCacheTest) Test_AddNegativeEntryForFolderShouldAddNegativeEntryForF
  t.statCache.AddNegativeEntryForFolder(name, expiration)
 
  hit, entry := t.statCache.LookUpFolder(name, someTime)
- assert.False(t.T(), hit)
+ assert.True(t.T(), hit)
  assert.Nil(t.T(), entry)
 }
 
-func (t *StatCacheTest) Test_ShouldReturnHitFalseWhenOnlyObjectAlreadyHasEntry() {
+func (t *StatCacheTest) Test_ShouldReturnHitTrueWhenOnlyObjectAlreadyHasEntry() {
  const name = "key1"
  existingEntry := &gcs.MinObject{
  Name: name,
@@ -511,7 +511,8 @@ func (t *StatCacheTest) Test_ShouldReturnHitFalseWhenOnlyObjectAlreadyHasEntry()
 
  hit, entry := t.statCache.LookUpFolder(name, someTime)
 
- assert.False(t.T(), hit)
+ // If "key1" object exist then corresponding folder entry will be nil, but hit will be true as key have entry in the cache for object.
+ assert.True(t.T(), hit)
  assert.Nil(t.T(), entry)
 }
 
@@ -521,7 +522,7 @@ func (t *StatCacheTest) Test_ShouldEvictEntryOnFullCapacityIncludingFolderSize()
  objectEntry1 := &gcs.MinObject{Name: "1"}
  objectEntry2 := &gcs.MinObject{Name: "2"}
  folderEntry := &gcs.Folder{
- Name: "3",
+ Name: "3/",
  }
  t.statCache.Insert(objectEntry1, expiration) // adds size of 1428
  t.statCache.Insert(objectEntry2, expiration) // adds size of 1428
@@ -538,14 +539,14 @@ func (t *StatCacheTest) Test_ShouldEvictEntryOnFullCapacityIncludingFolderSize()
 
  hit1, entry1 = t.statCache.LookUp("1", someTime)
  hit2, entry2 = t.statCache.LookUp("2", someTime)
- hit3, entry3 := t.statCache.LookUpFolder("3", someTime)
+ hit3, entry3 := t.statCache.LookUpFolder("3/", someTime)
 
  assert.False(t.T(), hit1)
  assert.Nil(t.T(), entry1)
  assert.True(t.T(), hit2)
  assert.Equal(t.T(), "2", entry2.Name)
  assert.True(t.T(), hit3)
- assert.Equal(t.T(), "3", entry3.Name)
+ assert.Equal(t.T(), "3/", entry3.Name)
 }
 
 func (t *StatCacheTest) Test_ShouldEvictAllEntriesWithPrefixFolder() {

internal/fs/hns_bucket_test.go

@@ -188,8 +188,11 @@ func (t *HNSBucketTests) TestRenameFolderWithSourceDirectoryHaveLocalFiles() {
 func (t *HNSBucketTests) TestRenameFolderWithSameParent() {
  oldDirPath := path.Join(mntDir, "foo")
  _, err = os.Stat(oldDirPath)
- assert.NoError(t.T(), err)
+ require.NoError(t.T(), err)
  newDirPath := path.Join(mntDir, "foo_rename")
+ _, err = os.Stat(newDirPath)
+ require.Error(t.T(), err)
+ assert.True(t.T(), strings.Contains(err.Error(), "no such file or directory"))
 
  err = os.Rename(oldDirPath, newDirPath)
 

internal/gcsx/prefix_bucket.go

@@ -245,6 +245,12 @@ func (b *prefixBucket) CreateFolder(ctx context.Context, folderName string) (*gc
 func (b *prefixBucket) RenameFolder(ctx context.Context, folderName string, destinationFolderId string) (*gcs.Folder, error) {
  mFolderName := b.wrappedName(folderName)
  mDestinationFolderId := b.wrappedName(destinationFolderId)
- o, err := b.wrapped.RenameFolder(ctx, mFolderName, mDestinationFolderId)
- return o, err
+ f, err := b.wrapped.RenameFolder(ctx, mFolderName, mDestinationFolderId)
+
+ // Modify the returned folder.
+ if f != nil {
+ f.Name = b.localName(f.Name)
+ }
+
+ return f, err
 }

internal/gcsx/prefix_bucket_test.go

@@ -456,8 +456,9 @@ func TestRenameFolder(t *testing.T) {
  _, err = wrapped.CreateFolder(ctx, name)
  assert.Nil(t, err)
 
- _, err = bucket.RenameFolder(ctx, old_suffix, new_suffix)
+ f, err := bucket.RenameFolder(ctx, old_suffix, new_suffix)
  assert.Nil(t, err)
+ assert.Equal(t, new_suffix, f.Name)
 
  // New folder should get created
  _, err = bucket.GetFolder(ctx, new_suffix)

internal/storage/caching/fast_stat_bucket.go

@@ -344,9 +344,7 @@ func (b *fastStatBucket) DeleteFolder(ctx context.Context, folderName string) er
  if err != nil {
  return err
  }
- // Add negative entry in the cache.
- b.addNegativeEntryForFolder(folderName)
-
+ b.invalidate(folderName)
  return err
 }
 
@@ -369,10 +367,7 @@ func (b *fastStatBucket) StatObjectFromGcs(ctx context.Context,
  return
 }
 
-func (b *fastStatBucket) GetFolder(
- ctx context.Context,
- prefix string) (*gcs.Folder, error) {
-
+func (b *fastStatBucket) GetFolder(ctx context.Context, prefix string) (*gcs.Folder, error) {
  if hit, entry := b.lookUpFolder(prefix); hit {
  // Negative entries result in NotFoundError.
  if entry == nil {
@@ -386,16 +381,23 @@ func (b *fastStatBucket) GetFolder(
  return entry, nil
  }
 
- // Fetch the Folder
- folder, error := b.wrapped.GetFolder(ctx, prefix)
+ // Fetch the Folder from GCS
+ return b.getFolderFromGCS(ctx, prefix)
+}
+
+func (b *fastStatBucket) getFolderFromGCS(ctx context.Context, prefix string) (*gcs.Folder, error) {
+ f, err := b.wrapped.GetFolder(ctx, prefix)
 
- if error != nil {
- return nil, error
+ if err == nil {
+ b.insertFolder(f)
+ return f, nil
  }
 
- // Record the new folder.
- b.insertFolder(folder)
- return folder, nil
+ // Special case: NotFoundError -> negative entry.
+ if _, ok := err.(*gcs.NotFoundError); ok {
+ b.addNegativeEntryForFolder(prefix)
+ }
+ return nil, err
 }
 
 func (b *fastStatBucket) CreateFolder(ctx context.Context, folderName string) (f *gcs.Folder, err error) {
@@ -413,11 +415,16 @@ func (b *fastStatBucket) CreateFolder(ctx context.Context, folderName string) (f
  return
 }
 
-func (b *fastStatBucket) RenameFolder(ctx context.Context, folderName string, destinationFolderId string) (o *gcs.Folder, err error) {
- o, err = b.wrapped.RenameFolder(ctx, folderName, destinationFolderId)
+func (b *fastStatBucket) RenameFolder(ctx context.Context, folderName string, destinationFolderId string) (*gcs.Folder, error) {
+ f, err := b.wrapped.RenameFolder(ctx, folderName, destinationFolderId)
+ if err != nil {
+ return nil, err
+ }
 
  // Invalidate cache for old directory.
  b.eraseEntriesWithGivenPrefix(folderName)
+ // Insert destination folder.
+ b.insertFolder(f)
 
- return o, err
+ return f, err
 }

internal/storage/caching/fast_stat_bucket_test.go

@@ -844,7 +844,9 @@ func (t *StatObjectTest) TestRenameFolder() {
  var folder = &gcs.Folder{
  Name: newName,
  }
+
  ExpectCall(t.cache, "EraseEntriesWithGivenPrefix")(name).WillOnce(Return())
+ ExpectCall(t.cache, "InsertFolder")(folder, Any()).WillOnce(Return())
  ExpectCall(t.wrapped, "RenameFolder")(Any(), name, newName).WillOnce(Return(folder, nil))
 
  result, err := t.bucket.RenameFolder(context.Background(), name, newName)
@@ -861,10 +863,9 @@ func init() { RegisterTestSuite(&DeleteFolderTest{}) }
 
 func (t *DeleteFolderTest) Test_DeleteFolder_Success() {
  const name = "some-name"
- ExpectCall(t.cache, "AddNegativeEntryForFolder")(name, Any()).
- WillOnce(Return())
  ExpectCall(t.wrapped, "DeleteFolder")(Any(), name).
  WillOnce(Return(nil))
+ ExpectCall(t.cache, "Erase")(name).WillOnce(Return())
 
  err := t.bucket.DeleteFolder(context.TODO(), name)