debug: http deleteion checker

Author: cheggaaa

deletelog/deletelog.go

@@ -54,14 +54,14 @@ func (d *deleteLog) Name() (name string) {
 
 func (d *deleteLog) Run(ctx context.Context) (err error) {
 	if !d.disableTicker {
-		d.ticker = periodicsync.NewPeriodicSync(60, time.Hour, d.checkLog, log)
+		d.ticker = periodicsync.NewPeriodicSync(30, time.Hour*2, d.checkLog, log)
 		d.ticker.Run()
 	}
 	return
 }
 
 func (d *deleteLog) checkLog(ctx context.Context) (err error) {
-	mu := d.redsync.NewMutex("_lock:deletion", redsync.WithExpiry(time.Hour))
+	mu := d.redsync.NewMutex("_lock:deletion", redsync.WithExpiry(time.Hour*2))
 	if err = mu.LockContext(ctx); err != nil {
 		return
 	}

index/check.go

@@ -362,3 +362,44 @@ func (gc *groupContent) Check(cidRefs map[string]uint64, sumSize uint64) (checkR
 	}
 	return
 }
+
+func (ri *redisIndex) CheckDeletedSpaces(
+	ctx context.Context,
+	key Key,
+	resolve func(spaceIds []string) (deletedIds []string, err error),
+	doFix bool,
+) (toBeDeleted []string, err error) {
+	gExists, gRelease, err := ri.AcquireKey(ctx, GroupKey(key))
+	if err != nil {
+		return
+	}
+	if !gExists {
+		gRelease()
+		return
+	}
+	gEntry, err := ri.getGroupEntry(ctx, key)
+	if err != nil {
+		gRelease()
+		return
+	}
+
+	deletedIds, err := resolve(gEntry.GetSpaceIds())
+	if err != nil {
+		gRelease()
+		return
+	}
+
+	gRelease()
+
+	if doFix {
+		ctx = context.WithValue(ctx, ctxForceSpaceGet, true)
+		for _, toDeleteSpaceId := range deletedIds {
+			key := Key{GroupId: key.GroupId, SpaceId: toDeleteSpaceId}
+			if _, err = ri.SpaceDelete(ctx, key); err != nil {
+				return nil, err
+			}
+			_, _ = ri.MarkSpaceAsDeleted(ctx, key)
+		}
+	}
+	return deletedIds, nil
+}

index/check_test.go

@@ -89,3 +89,28 @@ func TestRedisIndex_Check(t *testing.T) {
 	})
 
 }
+
+func TestRedisIndex_CheckDeletedSpaces(t *testing.T) {
+	fx := newFixture(t)
+	defer fx.Finish(t)
+	bs := testutil.NewRandBlocks(3)
+	fileId1 := testutil.NewRandCid().String()
+
+	require.NoError(t, fx.BlocksAdd(ctx, bs))
+	key := newRandKey()
+	cids, err := fx.CidEntriesByBlocks(ctx, bs[:2])
+	require.NoError(t, err)
+	require.NoError(t, fx.FileBind(ctx, key, fileId1, cids))
+	cids.Release()
+
+	result, err := fx.CheckDeletedSpaces(ctx, key, func(spaceIds []string) (deletedIds []string, err error) {
+		return spaceIds, nil
+	}, true)
+	require.NoError(t, err)
+	require.Len(t, result, 1)
+	assert.Equal(t, key.SpaceId, result[0])
+
+	info, err := fx.GroupInfo(ctx, key.GroupId)
+	require.NoError(t, err)
+	assert.Len(t, info.SpaceIds, 0)
+}

index/index.go

@@ -33,6 +33,12 @@ var (
 	ErrSpaceIsDeleted = errors.New("space is deleted")
 )
 
+type ctxKey int
+
+const (
+	ctxForceSpaceGet ctxKey = iota
+)
+
 type Index interface {
 	FileBind(ctx context.Context, key Key, fileId string, cidEntries *CidEntries) (err error)
 	FileUnbind(ctx context.Context, kye Key, fileIds ...string) (err error)
@@ -62,6 +68,7 @@ type Index interface {
 	Migrate(ctx context.Context, key Key) error
 
 	Check(ctx context.Context, key Key, doFix bool) (checkResults []CheckResult, err error)
+	CheckDeletedSpaces(ctx context.Context, key Key, resolve func(spaceIds []string) (deletedIds []string, err error), doFix bool) (toBeDeleted []string, err error)
 
 	SpaceDelete(ctx context.Context, key Key) (ok bool, err error)
 	MarkSpaceAsDeleted(ctx context.Context, key Key) (ok bool, err error)

index/loader.go

@@ -87,7 +87,8 @@ func (ri *redisIndex) AcquireSpace(ctx context.Context, key Key) (entry groupSpa
 		return
 	}
 
-	if !sExists {
+	isForceLoad, _ := ctx.Value(ctxForceSpaceGet).(bool)
+	if !isForceLoad {
 		// check if space marked as deleted
 		var (
 			delKey     = DelKey(key)

index/mock_index/mock_index.go

@@ -8,6 +8,8 @@ import (
 
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/commonfile/fileproto"
+	"github.com/anyproto/any-sync/coordinator/coordinatorclient"
+	"github.com/anyproto/any-sync/coordinator/coordinatorproto"
 
 	"github.com/anyproto/any-sync-filenode/index"
 )
@@ -30,11 +32,13 @@ func New() Stat {
 type statService struct {
 	accountInfoProvider accountInfoProvider
 	index               index.Index
+	coordinator         coordinatorclient.CoordinatorClient
 }
 
 func (i *statService) Init(a *app.App) (err error) {
 	i.accountInfoProvider = app.MustComponent[accountInfoProvider](a)
 	i.index = app.MustComponent[index.Index](a)
+	i.coordinator = app.MustComponent[coordinatorclient.CoordinatorClient](a)
 	return
 }
 
@@ -118,6 +122,46 @@ func (i *statService) Run(ctx context.Context) (err error) {
 			return
 		}
 	})
+	http.HandleFunc("/stat/check_deletion/{identity}", func(writer http.ResponseWriter, request *http.Request) {
+		identity := request.PathValue("identity")
+		if identity == "" {
+			http.Error(writer, "identity is empty", http.StatusBadRequest)
+			return
+		}
+		isDoFix := request.URL.Query().Get("fix") != ""
+
+		st := time.Now()
+		res, err := i.index.CheckDeletedSpaces(ctx, index.Key{GroupId: identity}, func(spaceIds []string) (deletedIds []string, err error) {
+			statuses, _, err := i.coordinator.StatusCheckMany(ctx, spaceIds)
+			if err != nil {
+				return nil, err
+			}
+			for idx, status := range statuses {
+				if status.Status == coordinatorproto.SpaceStatus_SpaceStatusDeleted {
+					deletedIds = append(deletedIds, spaceIds[idx])
+				}
+			}
+			return
+		}, isDoFix)
+		if err != nil {
+			http.Error(writer, err.Error(), http.StatusInternalServerError)
+		}
+		resp := struct {
+			Results  []string `json:"results"`
+			Duration string   `json:"duration"`
+		}{
+			Results:  res,
+			Duration: time.Since(st).String(),
+		}
+
+		writer.Header().Set("Content-Type", "application/json")
+		writer.WriteHeader(http.StatusOK)
+		err = json.NewEncoder(writer).Encode(resp)
+		if err != nil {
+			http.Error(writer, err.Error(), http.StatusInternalServerError)
+			return
+		}
+	})
 	return nil
 }