Merge pull request #153 from anyproto/GO-5827-delete-s3-keys

GO-5827 remove keys from s3

Author: cheggaaa

index/delete.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"slices"
 	"strings"
+	"time"
 
 	"github.com/redis/go-redis/v9"
 )
@@ -66,7 +67,7 @@ func (ri *redisIndex) MarkSpaceAsDeleted(ctx context.Context, key Key) (ok bool,
 	}
 	defer release()
 	if !exists {
-		if err = ri.cl.Set(ctx, DelKey(key), 1, 0).Err(); err != nil {
+		if err = ri.cl.Set(ctx, DelKey(key), time.Now().Unix(), 0).Err(); err != nil {
 			return
 		}
 		return true, nil

index/loader.go

@@ -35,6 +35,7 @@ func init() {
 type persistentStore interface {
 	IndexGet(ctx context.Context, key string) (value []byte, err error)
 	IndexPut(ctx context.Context, key string, value []byte) (err error)
+	IndexDelete(ctx context.Context, key string) (err error)
 
 	Get(ctx context.Context, k cid.Cid) (blocks.Block, error)
 }
@@ -271,9 +272,12 @@ func (ri *redisIndex) persistKey(ctx context.Context, storeKey, key string, dead
 		stat.errors.Add(1)
 		return
 	}
-	// key was removed - just remove it from store queue
+	// the key was removed - remove it from the persistent store and the queue
 	if errors.Is(err, redis.Nil) {
 		stat.deleted.Add(1)
+		if err = ri.persistStore.IndexDelete(ctx, key); err != nil {
+			stat.errors.Add(1)
+		}
 		return ri.cl.ZRem(ctx, storeKey, key).Err()
 	}
 

index/loader_test.go

@@ -21,6 +21,18 @@ func TestRedisIndex_PersistKeys(t *testing.T) {
 		require.NoError(t, fx.BlocksAdd(ctx, bs))
 		fx.PersistKeys(ctx)
 	})
+	t.Run("delete", func(t *testing.T) {
+		fx := newFixtureConfig(t, &config.Config{PersistTtl: 1})
+		defer fx.Finish(t)
+		bs := testutil.NewRandBlocks(5)
+		for _, b := range bs {
+			_, release, _ := fx.AcquireKey(ctx, CidKey(b.Cid()))
+			release()
+			fx.persistStore.EXPECT().IndexDelete(ctx, CidKey(b.Cid()))
+		}
+		time.Sleep(time.Second * 3)
+		fx.PersistKeys(ctx)
+	})
 	t.Run("persist", func(t *testing.T) {
 		fx := newFixtureConfig(t, &config.Config{PersistTtl: 1})
 		defer fx.Finish(t)

store/filedevstore/filedevstore.go

@@ -106,3 +106,7 @@ func (s *fsstore) IndexGet(ctx context.Context, key string) (value []byte, err e
 func (s *fsstore) IndexPut(ctx context.Context, key string, value []byte) (err error) {
 	return os.WriteFile(filepath.Join(s.path, key), value, 0644)
 }
+
+func (s *fsstore) IndexDelete(ctx context.Context, key string) (err error) {
+	return os.Remove(filepath.Join(s.path, key))
+}

store/mock_store/mock_store.go

@@ -239,6 +239,14 @@ func (s *s3store) IndexPut(ctx context.Context, key string, data []byte) (err er
 	return
 }
 
+func (s *s3store) IndexDelete(ctx context.Context, key string) (err error) {
+	_, err = s.client.DeleteObjectWithContext(ctx, &s3.DeleteObjectInput{
+		Key:    aws.String(key),
+		Bucket: s.indexBucket,
+	})
+	return
+}
+
 func (s *s3store) Close(ctx context.Context) (err error) {
 	return nil
 }

store/s3store/s3store.go

@@ -15,5 +15,6 @@ type Store interface {
 
 	IndexGet(ctx context.Context, key string) (value []byte, err error)
 	IndexPut(ctx context.Context, key string, value []byte) (err error)
+	IndexDelete(ctx context.Context, key string) (err error)
 	app.Component
 }