[MM-57991] Add log message about deleted rows from Data Retention job (#26847)

server/channels/store/opentracinglayer/opentracinglayer.go

@@ -7648,7 +7648,7 @@ func (s *OpenTracingLayerReactionStore) DeleteAllWithEmojiName(emojiName string)
 	return err
 }
 
-func (s *OpenTracingLayerReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) error {
+func (s *OpenTracingLayerReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) (int64, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ReactionStore.DeleteOrphanedRowsByIds")
 	s.Root.Store.SetContext(newCtx)
@@ -7657,13 +7657,13 @@ func (s *OpenTracingLayerReactionStore) DeleteOrphanedRowsByIds(r *model.Retenti
 	}()
 
 	defer span.Finish()
-	err := s.ReactionStore.DeleteOrphanedRowsByIds(r)
+	result, err := s.ReactionStore.DeleteOrphanedRowsByIds(r)
 	if err != nil {
 		span.LogFields(spanlog.Error(err))
 		ext.Error.Set(span, true)
 	}
 
-	return err
+	return result, err
 }
 
 func (s *OpenTracingLayerReactionStore) ExistsOnPost(postId string, emojiName string) (bool, error) {

server/channels/store/retrylayer/retrylayer.go

@@ -8699,21 +8699,21 @@ func (s *RetryLayerReactionStore) DeleteAllWithEmojiName(emojiName string) error
 
 }
 
-func (s *RetryLayerReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) error {
+func (s *RetryLayerReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) (int64, error) {
 
 	tries := 0
 	for {
-		err := s.ReactionStore.DeleteOrphanedRowsByIds(r)
+		result, err := s.ReactionStore.DeleteOrphanedRowsByIds(r)
 		if err == nil {
-			return nil
+			return result, nil
 		}
 		if !isRepeatableError(err) {
-			return err
+			return result, err
 		}
 		tries++
 		if tries >= 3 {
 			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
-			return err
+			return result, err
 		}
 		timepkg.Sleep(100 * timepkg.Millisecond)
 	}

server/channels/store/sqlstore/reaction_store.go

@@ -330,10 +330,10 @@ func (s SqlReactionStore) PermanentDeleteByUser(userId string) error {
 	return nil
 }
 
-func (s *SqlReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) error {
+func (s *SqlReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) (int64, error) {
 	txn, err := s.GetMasterX().Beginx()
 	if err != nil {
-		return err
+		return 0, err
 	}
 	defer finalizeTransactionX(txn, &err)
 
@@ -343,18 +343,24 @@ func (s *SqlReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDelet
 			sq.Eq{"PostId": r.Ids},
 		)
 
-	_, err = txn.ExecBuilder(query)
+	sqlResult, err := txn.ExecBuilder(query)
 	if err != nil {
-		return errors.Wrapf(err, "failed to delete orphaned reactions with RetentionIdsForDeletion Id=%s", r.Id)
+		return 0, errors.Wrapf(err, "failed to delete orphaned reactions with RetentionIdsForDeletion Id=%s", r.Id)
 	}
 	err = deleteFromRetentionIdsTx(txn, r.Id)
 	if err != nil {
-		return err
+		return 0, err
 	}
 	if err = txn.Commit(); err != nil {
-		return err
+		return 0, err
 	}
-	return nil
+
+	rowsAffected, err := sqlResult.RowsAffected()
+	if err != nil {
+		return 0, errors.Wrap(err, "unable to retrieve rows affected")
+	}
+
+	return rowsAffected, nil
 }
 
 func (s *SqlReactionStore) PermanentDeleteBatch(endTime int64, limit int64) (int64, error) {

server/channels/store/sqlstore/retention_policy_store.go

@@ -1152,9 +1152,11 @@ func getDeleteQueriesForMySQL(r RetentionPolicyBatchDeletionInfo, query string)
 	return fmt.Sprintf("DELETE %s FROM %s INNER JOIN (%s) AS A ON %s", r.Table, r.Table, query, joinClause)
 }
 
-func deleteFromRetentionIdsTx(txn *sqlxTxWrapper, id string) (err error) {
-	if _, err := txn.Exec("DELETE FROM RetentionIdsForDeletion WHERE Id = ?", id); err != nil {
+func deleteFromRetentionIdsTx(txn *sqlxTxWrapper, id string) error {
+	_, err := txn.Exec("DELETE FROM RetentionIdsForDeletion WHERE Id = ?", id)
+	if err != nil {
 		return errors.Wrap(err, "Failed to delete from RetentionIdsForDeletion")
 	}
+
 	return nil
 }

server/channels/store/store.go

@@ -751,7 +751,7 @@ type ReactionStore interface {
 	DeleteAllWithEmojiName(emojiName string) error
 	BulkGetForPosts(postIds []string) ([]*model.Reaction, error)
 	GetSingle(userID, postID, remoteID, emojiName string) (*model.Reaction, error)
-	DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) error
+	DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) (int64, error)
 	PermanentDeleteBatch(endTime int64, limit int64) (int64, error)
 	PermanentDeleteByUser(userID string) error
 }

server/channels/store/storetest/mocks/ReactionStore.go

@@ -93,21 +93,31 @@ func (_m *ReactionStore) DeleteAllWithEmojiName(emojiName string) error {
 }
 
 // DeleteOrphanedRowsByIds provides a mock function with given fields: r
-func (_m *ReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) error {
+func (_m *ReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) (int64, error) {
 	ret := _m.Called(r)
 
 	if len(ret) == 0 {
 		panic("no return value specified for DeleteOrphanedRowsByIds")
 	}
 
-	var r0 error
-	if rf, ok := ret.Get(0).(func(*model.RetentionIdsForDeletion) error); ok {
+	var r0 int64
+	var r1 error
+	if rf, ok := ret.Get(0).(func(*model.RetentionIdsForDeletion) (int64, error)); ok {
+		return rf(r)
+	}
+	if rf, ok := ret.Get(0).(func(*model.RetentionIdsForDeletion) int64); ok {
 		r0 = rf(r)
 	} else {
-		r0 = ret.Error(0)
+		r0 = ret.Get(0).(int64)
 	}
 
-	return r0
+	if rf, ok := ret.Get(1).(func(*model.RetentionIdsForDeletion) error); ok {
+		r1 = rf(r)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
 }
 
 // ExistsOnPost provides a mock function with given fields: postId, emojiName

server/channels/store/storetest/post_store.go

@@ -4015,8 +4015,9 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 	require.Equal(t, 1, len(rows))
 	require.Equal(t, 2, len(rows[0].Ids))
 	// Clean up retention ids table
-	err = ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
+	deleted, err = ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
 	require.NoError(t, err)
+	require.Equal(t, int64(0), deleted)
 
 	t.Run("with pagination", func(t *testing.T) {
 		for i := 0; i < 3; i++ {
@@ -4040,8 +4041,9 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 		require.Equal(t, 2, len(rows[0].Ids))
 
 		// Clean up retention ids table
-		err = ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
+		deleted, err = ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
 		require.NoError(t, err)
+		require.Equal(t, int64(0), deleted)
 
 		deleted, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2, 2, cursor)
 		require.NoError(t, err)
@@ -4053,8 +4055,9 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 		require.Equal(t, 1, len(rows[0].Ids))
 
 		// Clean up retention ids table
-		err = ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
+		deleted, err = ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
 		require.NoError(t, err)
+		require.Equal(t, int64(0), deleted)
 	})
 
 	t.Run("with data retention policies", func(t *testing.T) {
@@ -4127,8 +4130,9 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 		rows, err = ss.RetentionPolicy().GetIdsForDeletionByTableName("Posts", 1000)
 		require.NoError(t, err)
 		for _, row := range rows {
-			err = ss.Reaction().DeleteOrphanedRowsByIds(row)
+			deleted, err = ss.Reaction().DeleteOrphanedRowsByIds(row)
 			require.NoError(t, err)
+			require.Equal(t, int64(0), deleted)
 		}
 	})
 
@@ -4203,8 +4207,9 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 
 		// Clean up retention ids table
 		for _, row := range rows {
-			err = ss.Reaction().DeleteOrphanedRowsByIds(row)
+			deleted, err = ss.Reaction().DeleteOrphanedRowsByIds(row)
 			require.NoError(t, err)
+			require.Equal(t, int64(0), deleted)
 		}
 	})
 }

server/channels/store/storetest/preference_store.go

@@ -390,8 +390,9 @@ func testPreferenceDeleteOrphanedRows(t *testing.T, rctx request.CTX, ss store.S
 	require.Equal(t, 1, len(rows))
 
 	// Clean up retention ids table
-	err = ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
+	deleted, err := ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
 	require.NoError(t, err)
+	require.Equal(t, int64(0), deleted)
 
 	_, nErr = ss.Preference().DeleteOrphanedRows(limit)
 	assert.NoError(t, nErr)

server/channels/store/storetest/reaction_store.go

@@ -714,8 +714,10 @@ func testReactionStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss st
 	require.Contains(t, rows[0].Ids, olderPost.Id)
 
 	for _, row := range rows {
-		err = ss.Reaction().DeleteOrphanedRowsByIds(row)
+		var deleted int64
+		deleted, err = ss.Reaction().DeleteOrphanedRowsByIds(row)
 		require.NoError(t, err)
+		require.Equal(t, int64(2), deleted)
 	}
 
 	rows, err = ss.RetentionPolicy().GetIdsForDeletionByTableName("Posts", 1000)

server/channels/store/timerlayer/timerlayer.go

@@ -6905,10 +6905,10 @@ func (s *TimerLayerReactionStore) DeleteAllWithEmojiName(emojiName string) error
 	return err
 }
 
-func (s *TimerLayerReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) error {
+func (s *TimerLayerReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsForDeletion) (int64, error) {
 	start := time.Now()
 
-	err := s.ReactionStore.DeleteOrphanedRowsByIds(r)
+	result, err := s.ReactionStore.DeleteOrphanedRowsByIds(r)
 
 	elapsed := float64(time.Since(start)) / float64(time.Second)
 	if s.Root.Metrics != nil {
@@ -6918,7 +6918,7 @@ func (s *TimerLayerReactionStore) DeleteOrphanedRowsByIds(r *model.RetentionIdsF
 		}
 		s.Root.Metrics.ObserveStoreMethodDuration("ReactionStore.DeleteOrphanedRowsByIds", success, elapsed)
 	}
-	return err
+	return result, err
 }
 
 func (s *TimerLayerReactionStore) ExistsOnPost(postId string, emojiName string) (bool, error) {