Invalidate bucket caches on object modification or deletion

Invalidate content, existence, and attributes cached for a particular object
when the object is modified or deleted.

Part of #9386

Author: 56quarters

pkg/storage/tsdb/bucketcache/caching_bucket.go

@@ -159,6 +159,38 @@ func NewCachingBucket(bucketID string, bucketClient objstore.Bucket, cfg *Cachin
 	return cb, nil
 }
 
+func (cb *CachingBucket) invalidate(ctx context.Context, name string) {
+	_, cfg := cb.cfg.findGetConfig(name)
+	if cfg != nil {
+		contentKey := cachingKeyContent(cb.bucketID, name)
+		attrsKey := cachingKeyAttributes(cb.bucketID, name)
+		existsKey := cachingKeyExists(cb.bucketID, name)
+
+		// Cache might be down or key might not exist. Ignore errors since invalidation is best-effort.
+		_ = cfg.cache.Delete(ctx, contentKey)
+		_ = cfg.cache.Delete(ctx, attrsKey)
+		_ = cfg.cache.Delete(ctx, existsKey)
+	}
+}
+
+func (cb *CachingBucket) Upload(ctx context.Context, name string, r io.Reader) error {
+	err := cb.Bucket.Upload(ctx, name, r)
+	if err == nil {
+		cb.invalidate(ctx, name)
+	}
+
+	return err
+}
+
+func (cb *CachingBucket) Delete(ctx context.Context, name string) error {
+	err := cb.Bucket.Delete(ctx, name)
+	if err == nil {
+		cb.invalidate(ctx, name)
+	}
+
+	return err
+}
+
 func (cb *CachingBucket) Name() string {
 	return "caching: " + cb.Bucket.Name()
 }

pkg/storage/tsdb/bucketcache/caching_bucket_test.go

@@ -909,6 +909,55 @@ func TestCachingKey_ShouldKeepAllocationsToMinimum(t *testing.T) {
 	}
 }
 
+func TestMutationInvalidatesCache(t *testing.T) {
+	inmem := objstore.NewInMemBucket()
+
+	// We reuse cache between tests (!)
+	c := cache.NewMockCache()
+	ctx := context.Background()
+
+	const cfgName = "test"
+	cfg := NewCachingBucketConfig()
+	cfg.CacheGet(cfgName, c, matchAll, 1024^2, time.Minute, time.Minute, time.Minute)
+	cfg.CacheExists(cfgName, c, matchAll, time.Minute, time.Minute)
+	cfg.CacheAttributes(cfgName, c, matchAll, time.Minute)
+
+	cb, err := NewCachingBucket("test", inmem, cfg, nil, nil)
+	require.NoError(t, err)
+
+	t.Run("invalidated on upload", func(t *testing.T) {
+		c.Flush()
+
+		// Initial upload bypassing the CachingBucket but read the object back to ensure it is in cache.
+		require.NoError(t, inmem.Upload(ctx, "/object-1", strings.NewReader("test content 1")))
+		verifyGet(ctx, t, cb, "/object-1", []byte("test content 1"), true, false, cfgName)
+		verifyExists(ctx, t, cb, "/object-1", true, true, true, cfgName)
+		verifyObjectAttrs(ctx, t, cb, "/object-1", 14, true, false, cfgName)
+
+		// Do an upload via the CachingBucket and ensure the first read after does not come from cache.
+		require.NoError(t, cb.Upload(ctx, "/object-1", strings.NewReader("test content 12")))
+		verifyGet(ctx, t, cb, "/object-1", []byte("test content 12"), true, false, cfgName)
+		verifyExists(ctx, t, cb, "/object-1", true, true, true, cfgName)
+		verifyObjectAttrs(ctx, t, cb, "/object-1", 15, true, false, cfgName)
+	})
+
+	t.Run("invalidated on delete", func(t *testing.T) {
+		c.Flush()
+
+		// Initial upload bypassing the CachingBucket but read the object back to ensure it is in cache.
+		require.NoError(t, inmem.Upload(ctx, "/object-1", strings.NewReader("test content 1")))
+		verifyGet(ctx, t, cb, "/object-1", []byte("test content 1"), true, false, cfgName)
+		verifyExists(ctx, t, cb, "/object-1", true, true, true, cfgName)
+		verifyObjectAttrs(ctx, t, cb, "/object-1", 14, true, false, cfgName)
+
+		// Delete via the CachingBucket and ensure the first read after does not come from cache but non-existence is cached.
+		require.NoError(t, cb.Delete(ctx, "/object-1"))
+		verifyGet(ctx, t, cb, "/object-1", nil, true, false, cfgName)
+		verifyExists(ctx, t, cb, "/object-1", false, true, true, cfgName)
+		verifyObjectAttrs(ctx, t, cb, "/object-1", -1, true, false, cfgName)
+	})
+}
+
 func BenchmarkCachingKey(b *testing.B) {
 	tests := map[string]struct {
 		run func(bucketID string)