deletion log: handle ownership transfer changes

Author: cheggaaa

nodespace/spacedeleter/spacedeleter.go

@@ -87,8 +87,8 @@ func (s *spaceDeleter) delete(ctx context.Context) (err error) {
 	case <-ctx.Done():
 		return ctx.Err()
 	}
-	lastRecordId, err := s.deletionStorage.LastRecordId(ctx)
-	if err != nil && !errors.Is(err, nodestorage.ErrNoLastRecordId) {
+	lastRecordId, err := s.deletionStorage.DeletionLogId(ctx)
+	if err != nil && !errors.Is(err, nodestorage.ErrNoDeletionLogId) {
 		return err
 	}
 	log.Debug("getting deletion log", zap.Int("limit", logLimit), zap.String("lastRecordId", lastRecordId))
@@ -135,6 +135,12 @@ func (s *spaceDeleter) processDeletionRecord(ctx context.Context, rec *coordinat
 		if err != nil {
 			return err
 		}
+	case coordinatorproto.DeletionLogRecordStatus_OwnershipChange:
+		log.Debug("received ownership change record")
+		err := s.deletionStorage.SetDeletionLogId(ctx, rec.Id)
+		if err != nil {
+			return err
+		}
 	}
 	return nil
 }

nodespace/spacedeleter/spacedeleter_test.go

@@ -42,7 +42,7 @@ func TestSpaceDeleter_Run_Ok(t *testing.T) {
 	close(fx.waiterChan)
 	<-fx.deleter.testChan
 
-	id, err := fx.storage.IndexStorage().LastRecordId(ctx)
+	id, err := fx.storage.IndexStorage().DeletionLogId(ctx)
 	require.NoError(t, err)
 	require.Equal(t, lg[2].Id, id)
 	store, err = fx.storage.WaitSpaceStorage(ctx, payload.SpaceHeaderWithId.Id)
@@ -72,7 +72,7 @@ func TestSpaceDeleter_Run_Ok_NewPush(t *testing.T) {
 	close(fx.waiterChan)
 	<-fx.deleter.testChan
 
-	id, err := fx.storage.IndexStorage().LastRecordId(ctx)
+	id, err := fx.storage.IndexStorage().DeletionLogId(ctx)
 	require.NoError(t, err)
 	require.Equal(t, lg[3].Id, id)
 	status, err := fx.storage.IndexStorage().SpaceStatus(ctx, payload.SpaceHeaderWithId.Id)
@@ -90,7 +90,7 @@ func TestSpaceDeleter_Run_Ok_NoStorage(t *testing.T) {
 	close(fx.waiterChan)
 	<-fx.deleter.testChan
 
-	id, err := fx.storage.IndexStorage().LastRecordId(ctx)
+	id, err := fx.storage.IndexStorage().DeletionLogId(ctx)
 	require.NoError(t, err)
 	require.Equal(t, lg[2].Id, id)
 	status, err := fx.storage.IndexStorage().SpaceStatus(ctx, "space3")
@@ -130,7 +130,7 @@ func TestSpaceDeleter_Run_Ok_EmptyStorage(t *testing.T) {
 	close(fx.waiterChan)
 	<-fx.deleter.testChan
 
-	id, err := fx.storage.IndexStorage().LastRecordId(ctx)
+	id, err := fx.storage.IndexStorage().DeletionLogId(ctx)
 	require.NoError(t, err)
 	require.Equal(t, lg[2].Id, id)
 	store, err = fx.storage.WaitSpaceStorage(context.Background(), payload.SpaceHeaderWithId.Id)

nodestorage/indexstorage.go

@@ -26,8 +26,8 @@ const (
 )
 
 var (
-	ErrUnknownSpaceId = errors.New("unknown space id")
-	ErrNoLastRecordId = errors.New("no last record id")
+	ErrUnknownSpaceId  = errors.New("unknown space id")
+	ErrNoDeletionLogId = errors.New("no last record id")
 )
 
 const (
@@ -57,7 +57,8 @@ type IndexStorage interface {
 	SpaceStatus(ctx context.Context, spaceId string) (status SpaceStatus, err error)
 	MarkArchived(ctx context.Context, spaceId string, compressedSize, uncompressedSize int64) (err error)
 	MarkError(ctx context.Context, spaceId string, errString string) (err error)
-	LastRecordId(ctx context.Context) (id string, err error)
+	DeletionLogId(ctx context.Context) (id string, err error)
+	SetDeletionLogId(ctx context.Context, id string) (err error)
 	FindOldestInactiveSpace(ctx context.Context, olderThan time.Duration, skip int) (spaceId string, err error)
 
 	UpdateLastAccess(ctx context.Context, spaceId string) (err error)
@@ -205,17 +206,29 @@ func (d *indexStorage) MarkArchived(ctx context.Context, spaceId string, compres
 	return err
 }
 
-func (d *indexStorage) LastRecordId(ctx context.Context) (id string, err error) {
+func (d *indexStorage) DeletionLogId(ctx context.Context) (id string, err error) {
 	doc, err := d.settingsColl.FindId(ctx, lastDeletionIdKey)
 	if err != nil {
 		if errors.Is(err, anystore.ErrDocNotFound) {
-			return "", ErrNoLastRecordId
+			return "", ErrNoDeletionLogId
 		}
 		return "", err
 	}
 	return doc.Value().GetString(valueKey), nil
 }
 
+func (d *indexStorage) SetDeletionLogId(ctx context.Context, id string) (err error) {
+	_, err = d.settingsColl.UpsertId(ctx, lastDeletionIdKey, query.ModifyFunc(func(a *anyenc.Arena, v *anyenc.Value) (result *anyenc.Value, modified bool, err error) {
+		prevKey := v.GetString(valueKey)
+		if prevKey < id {
+			v.Set(valueKey, a.NewString(id))
+			return v, true, nil
+		}
+		return v, false, nil
+	}))
+	return
+}
+
 func (d *indexStorage) UpdateLastAccess(ctx context.Context, spaceId string) (err error) {
 	now := time.Now()
 	if val, ok := d.lastAccessCache.Load(spaceId); ok {