diff --git a/backend/cmd/main.go b/backend/cmd/main.go
index bec1b38..d56730e 100644
--- a/backend/cmd/main.go
+++ b/backend/cmd/main.go
@@ -272,6 +272,10 @@ func runBackgroundTasks(log *slog.Logger) {
 			backuping.GetBackupsScheduler().Run(ctx)
 		})
 
+		go runWithPanicLogging(log, "backup cleaner background service", func() {
+			backuping.GetBackupCleaner().Run(ctx)
+		})
+
 		go runWithPanicLogging(log, "restore background service", func() {
 			restoring.GetRestoresScheduler().Run(ctx)
 		})
diff --git a/backend/internal/features/backups/backups/backuping/backuper.go b/backend/internal/features/backups/backups/backuping/backuper.go
index 5a17e16..2495c4c 100644
--- a/backend/internal/features/backups/backups/backuping/backuper.go
+++ b/backend/internal/features/backups/backups/backuping/backuper.go
@@ -157,21 +157,41 @@ func (n *BackuperNode) MakeBackup(backupID uuid.UUID, isCallNotifier bool) {
 
 	start := time.Now().UTC()
 
+	ctx, cancel := context.WithCancel(context.Background())
+	n.backupCancelManager.RegisterTask(backup.ID, cancel)
+	defer n.backupCancelManager.UnregisterTask(backup.ID)
+
 	backupProgressListener := func(
 		completedMBs float64,
 	) {
 		backup.BackupSizeMb = completedMBs
 		backup.BackupDurationMs = time.Since(start).Milliseconds()
 
+		// Check size limit (0 = unlimited)
+		if backupConfig.MaxBackupSizeMB > 0 &&
+			completedMBs > float64(backupConfig.MaxBackupSizeMB) {
+			errMsg := fmt.Sprintf(
+				"backup size (%.2f MB) exceeded maximum allowed size (%d MB)",
+				completedMBs,
+				backupConfig.MaxBackupSizeMB,
+			)
+
+			backup.Status = backups_core.BackupStatusFailed
+			backup.IsSkipRetry = true
+			backup.FailMessage = &errMsg
+			if err := n.backupRepository.Save(backup); err != nil {
+				n.logger.Error("Failed to save backup with size exceeded error", "error", err)
+			}
+			cancel() // Cancel the backup context
+
+			return
+		}
+
 		if err := n.backupRepository.Save(backup); err != nil {
 			n.logger.Error("Failed to update backup progress", "error", err)
 		}
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-	n.backupCancelManager.RegisterTask(backup.ID, cancel)
-	defer n.backupCancelManager.UnregisterTask(backup.ID)
-
 	backupMetadata, err := n.createBackupUseCase.Execute(
 		ctx,
 		backup.ID,
@@ -181,6 +201,29 @@ func (n *BackuperNode) MakeBackup(backupID uuid.UUID, isCallNotifier bool) {
 		backupProgressListener,
 	)
 	if err != nil {
+		// Check if backup was already marked as failed by progress listener (e.g., size limit exceeded)
+		// If so, skip error handling to avoid overwriting the status
+		currentBackup, fetchErr := n.backupRepository.FindByID(backup.ID)
+		if fetchErr == nil && currentBackup.Status == backups_core.BackupStatusFailed {
+			n.logger.Warn(
+				"Backup already marked as failed by progress listener, skipping error handling",
+				"backupId",
+				backup.ID,
+				"failMessage",
+				*currentBackup.FailMessage,
+			)
+
+			// Still call notification for size limit failures
+			n.SendBackupNotification(
+				backupConfig,
+				currentBackup,
+				backups_config.NotificationBackupFailed,
+				currentBackup.FailMessage,
+			)
+
+			return
+		}
+
 		errMsg := err.Error()
 
 		// Log detailed error information for debugging
diff --git a/backend/internal/features/backups/backups/backuping/backuper_test.go b/backend/internal/features/backups/backups/backuping/backuper_test.go
index 1275120..45b9ed6 100644
--- a/backend/internal/features/backups/backups/backuping/backuper_test.go
+++ b/backend/internal/features/backups/backups/backuping/backuper_test.go
@@ -1,13 +1,10 @@
 package backuping
 
 import (
-	"context"
-	"errors"
 	"strings"
 	"testing"
 	"time"
 
-	common "databasus-backend/internal/features/backups/backups/common"
 	backups_core "databasus-backend/internal/features/backups/backups/core"
 	backups_config "databasus-backend/internal/features/backups/config"
 	"databasus-backend/internal/features/databases"
@@ -18,7 +15,6 @@ import (
 	workspaces_testing "databasus-backend/internal/features/workspaces/testing"
 	cache_utils "databasus-backend/internal/util/cache"
 
-	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 )
@@ -158,35 +154,120 @@ func Test_BackupExecuted_NotificationSent(t *testing.T) {
 	})
 }
 
-type CreateFailedBackupUsecase struct {
-}
+func Test_BackupSizeLimits(t *testing.T) {
+	cache_utils.ClearAllCache()
+	user := users_testing.CreateTestUser(users_enums.UserRoleAdmin)
+	router := CreateTestRouter()
+	workspace := workspaces_testing.CreateTestWorkspace("Test Workspace", user, router)
+	storage := storages.CreateTestStorage(workspace.ID)
+	notifier := notifiers.CreateTestNotifier(workspace.ID)
+	database := databases.CreateTestDatabase(workspace.ID, storage, notifier)
 
-func (uc *CreateFailedBackupUsecase) Execute(
-	ctx context.Context,
-	backupID uuid.UUID,
-	backupConfig *backups_config.BackupConfig,
-	database *databases.Database,
-	storage *storages.Storage,
-	backupProgressListener func(completedMBs float64),
-) (*common.BackupMetadata, error) {
-	backupProgressListener(10)
-	return nil, errors.New("backup failed")
-}
+	defer func() {
+		// cleanup backups first
+		backups, _ := backupRepository.FindByDatabaseID(database.ID)
+		for _, backup := range backups {
+			backupRepository.DeleteByID(backup.ID)
+		}
 
-type CreateSuccessBackupUsecase struct{}
+		databases.RemoveTestDatabase(database)
+		time.Sleep(50 * time.Millisecond) // Wait for cascading deletes
+		notifiers.RemoveTestNotifier(notifier)
+		storages.RemoveTestStorage(storage.ID)
+		workspaces_testing.RemoveTestWorkspace(workspace, router)
+	}()
 
-func (uc *CreateSuccessBackupUsecase) Execute(
-	ctx context.Context,
-	backupID uuid.UUID,
-	backupConfig *backups_config.BackupConfig,
-	database *databases.Database,
-	storage *storages.Storage,
-	backupProgressListener func(completedMBs float64),
-) (*common.BackupMetadata, error) {
-	backupProgressListener(10)
-	return &common.BackupMetadata{
-		EncryptionSalt: nil,
-		EncryptionIV:   nil,
-		Encryption:     backups_config.BackupEncryptionNone,
-	}, nil
+	t.Run("UnlimitedSize_MaxBackupSizeMBIsZero_BackupCompletes", func(t *testing.T) {
+		// Enable backups with unlimited size (0)
+		backupConfig := backups_config.EnableBackupsForTestDatabase(database.ID, storage)
+		backupConfig.MaxBackupSizeMB = 0 // unlimited
+		backupConfig, err := backups_config.GetBackupConfigService().SaveBackupConfig(backupConfig)
+		assert.NoError(t, err)
+
+		backuperNode := CreateTestBackuperNode()
+		backuperNode.createBackupUseCase = &CreateLargeBackupUsecase{}
+
+		// Create a backup record
+		backup := &backups_core.Backup{
+			DatabaseID: database.ID,
+			StorageID:  storage.ID,
+			Status:     backups_core.BackupStatusInProgress,
+			CreatedAt:  time.Now().UTC(),
+		}
+		err = backupRepository.Save(backup)
+		assert.NoError(t, err)
+
+		backuperNode.MakeBackup(backup.ID, false)
+
+		// Verify backup completed successfully even with large size
+		updatedBackup, err := backupRepository.FindByID(backup.ID)
+		assert.NoError(t, err)
+		assert.Equal(t, backups_core.BackupStatusCompleted, updatedBackup.Status)
+		assert.Equal(t, float64(10000), updatedBackup.BackupSizeMb)
+		assert.Nil(t, updatedBackup.FailMessage)
+	})
+
+	t.Run("SizeExceeded_BackupFailedWithIsSkipRetry", func(t *testing.T) {
+		// Enable backups with 5 MB limit
+		backupConfig := backups_config.EnableBackupsForTestDatabase(database.ID, storage)
+		backupConfig.MaxBackupSizeMB = 5
+		backupConfig, err := backups_config.GetBackupConfigService().SaveBackupConfig(backupConfig)
+		assert.NoError(t, err)
+
+		backuperNode := CreateTestBackuperNode()
+		backuperNode.createBackupUseCase = &CreateProgressiveBackupUsecase{}
+
+		// Create a backup record
+		backup := &backups_core.Backup{
+			DatabaseID: database.ID,
+			StorageID:  storage.ID,
+			Status:     backups_core.BackupStatusInProgress,
+			CreatedAt:  time.Now().UTC(),
+		}
+		err = backupRepository.Save(backup)
+		assert.NoError(t, err)
+
+		backuperNode.MakeBackup(backup.ID, false)
+
+		// Verify backup was marked as failed with IsSkipRetry=true
+		updatedBackup, err := backupRepository.FindByID(backup.ID)
+		assert.NoError(t, err)
+		assert.Equal(t, backups_core.BackupStatusFailed, updatedBackup.Status)
+		assert.True(t, updatedBackup.IsSkipRetry)
+		assert.NotNil(t, updatedBackup.FailMessage)
+		assert.Contains(t, *updatedBackup.FailMessage, "exceeded maximum allowed size")
+		assert.Contains(t, *updatedBackup.FailMessage, "10.00 MB")
+		assert.Contains(t, *updatedBackup.FailMessage, "5 MB")
+		assert.Greater(t, updatedBackup.BackupSizeMb, float64(5))
+	})
+
+	t.Run("SizeWithinLimit_BackupCompletes", func(t *testing.T) {
+		// Enable backups with 100 MB limit
+		backupConfig := backups_config.EnableBackupsForTestDatabase(database.ID, storage)
+		backupConfig.MaxBackupSizeMB = 100
+		backupConfig, err := backups_config.GetBackupConfigService().SaveBackupConfig(backupConfig)
+		assert.NoError(t, err)
+
+		backuperNode := CreateTestBackuperNode()
+		backuperNode.createBackupUseCase = &CreateMediumBackupUsecase{}
+
+		// Create a backup record
+		backup := &backups_core.Backup{
+			DatabaseID: database.ID,
+			StorageID:  storage.ID,
+			Status:     backups_core.BackupStatusInProgress,
+			CreatedAt:  time.Now().UTC(),
+		}
+		err = backupRepository.Save(backup)
+		assert.NoError(t, err)
+
+		backuperNode.MakeBackup(backup.ID, false)
+
+		// Verify backup completed successfully
+		updatedBackup, err := backupRepository.FindByID(backup.ID)
+		assert.NoError(t, err)
+		assert.Equal(t, backups_core.BackupStatusCompleted, updatedBackup.Status)
+		assert.Equal(t, float64(50), updatedBackup.BackupSizeMb)
+		assert.Nil(t, updatedBackup.FailMessage)
+	})
 }
diff --git a/backend/internal/features/backups/backups/backuping/cleaner.go b/backend/internal/features/backups/backups/backuping/cleaner.go
new file mode 100644
index 0000000..7f0de96
--- /dev/null
+++ b/backend/internal/features/backups/backups/backuping/cleaner.go
@@ -0,0 +1,242 @@
+package backuping
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/google/uuid"
+
+	backups_core "databasus-backend/internal/features/backups/backups/core"
+	backups_config "databasus-backend/internal/features/backups/config"
+	"databasus-backend/internal/features/storages"
+	util_encryption "databasus-backend/internal/util/encryption"
+	"databasus-backend/internal/util/period"
+)
+
+const (
+	cleanerTickerInterval = 1 * time.Minute
+)
+
+type BackupCleaner struct {
+	backupRepository      *backups_core.BackupRepository
+	storageService        *storages.StorageService
+	backupConfigService   *backups_config.BackupConfigService
+	fieldEncryptor        util_encryption.FieldEncryptor
+	logger                *slog.Logger
+	backupRemoveListeners []backups_core.BackupRemoveListener
+
+	runOnce sync.Once
+	hasRun  atomic.Bool
+}
+
+func (c *BackupCleaner) Run(ctx context.Context) {
+	wasAlreadyRun := c.hasRun.Load()
+
+	c.runOnce.Do(func() {
+		c.hasRun.Store(true)
+
+		if ctx.Err() != nil {
+			return
+		}
+
+		ticker := time.NewTicker(cleanerTickerInterval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				if err := c.cleanOldBackups(); err != nil {
+					c.logger.Error("Failed to clean old backups", "error", err)
+				}
+
+				if err := c.cleanExceededBackups(); err != nil {
+					c.logger.Error("Failed to clean exceeded backups", "error", err)
+				}
+			}
+		}
+	})
+
+	if wasAlreadyRun {
+		panic(fmt.Sprintf("%T.Run() called multiple times", c))
+	}
+}
+
+func (c *BackupCleaner) DeleteBackup(backup *backups_core.Backup) error {
+	for _, listener := range c.backupRemoveListeners {
+		if err := listener.OnBeforeBackupRemove(backup); err != nil {
+			return err
+		}
+	}
+
+	storage, err := c.storageService.GetStorageByID(backup.StorageID)
+	if err != nil {
+		return err
+	}
+
+	err = storage.DeleteFile(c.fieldEncryptor, backup.ID)
+	if err != nil {
+		// we do not return error here, because sometimes clean up performed
+		// before unavailable storage removal or change - therefore we should
+		// proceed even in case of error. It's possible that some S3 or
+		// storage is not available yet, it should not block us
+		c.logger.Error("Failed to delete backup file", "error", err)
+	}
+
+	return c.backupRepository.DeleteByID(backup.ID)
+}
+
+func (c *BackupCleaner) AddBackupRemoveListener(listener backups_core.BackupRemoveListener) {
+	c.backupRemoveListeners = append(c.backupRemoveListeners, listener)
+}
+
+func (c *BackupCleaner) cleanOldBackups() error {
+	enabledBackupConfigs, err := c.backupConfigService.GetBackupConfigsWithEnabledBackups()
+	if err != nil {
+		return err
+	}
+
+	for _, backupConfig := range enabledBackupConfigs {
+		backupStorePeriod := backupConfig.StorePeriod
+
+		if backupStorePeriod == period.PeriodForever {
+			continue
+		}
+
+		storeDuration := backupStorePeriod.ToDuration()
+		dateBeforeBackupsShouldBeDeleted := time.Now().UTC().Add(-storeDuration)
+
+		oldBackups, err := c.backupRepository.FindBackupsBeforeDate(
+			backupConfig.DatabaseID,
+			dateBeforeBackupsShouldBeDeleted,
+		)
+		if err != nil {
+			c.logger.Error(
+				"Failed to find old backups for database",
+				"databaseId",
+				backupConfig.DatabaseID,
+				"error",
+				err,
+			)
+			continue
+		}
+
+		for _, backup := range oldBackups {
+			if err := c.DeleteBackup(backup); err != nil {
+				c.logger.Error("Failed to delete old backup", "backupId", backup.ID, "error", err)
+				continue
+			}
+
+			c.logger.Info(
+				"Deleted old backup",
+				"backupId",
+				backup.ID,
+				"databaseId",
+				backupConfig.DatabaseID,
+			)
+		}
+	}
+
+	return nil
+}
+
+func (c *BackupCleaner) cleanExceededBackups() error {
+	enabledBackupConfigs, err := c.backupConfigService.GetBackupConfigsWithEnabledBackups()
+	if err != nil {
+		return err
+	}
+
+	for _, backupConfig := range enabledBackupConfigs {
+		if backupConfig.MaxBackupsTotalSizeMB <= 0 {
+			continue
+		}
+
+		if err := c.cleanExceededBackupsForDatabase(
+			backupConfig.DatabaseID,
+			backupConfig.MaxBackupsTotalSizeMB,
+		); err != nil {
+			c.logger.Error(
+				"Failed to clean exceeded backups for database",
+				"databaseId",
+				backupConfig.DatabaseID,
+				"error",
+				err,
+			)
+			continue
+		}
+	}
+
+	return nil
+}
+
+func (c *BackupCleaner) cleanExceededBackupsForDatabase(
+	databaseID uuid.UUID,
+	limitperDbMB int64,
+) error {
+	for {
+		backupsTotalSizeMB, err := c.backupRepository.GetTotalSizeByDatabase(databaseID)
+		if err != nil {
+			return err
+		}
+
+		if backupsTotalSizeMB <= float64(limitperDbMB) {
+			break
+		}
+
+		oldestBackups, err := c.backupRepository.FindOldestByDatabaseExcludingInProgress(
+			databaseID,
+			1,
+		)
+		if err != nil {
+			return err
+		}
+
+		if len(oldestBackups) == 0 {
+			c.logger.Warn(
+				"No backups to delete but still over limit",
+				"databaseId",
+				databaseID,
+				"totalSizeMB",
+				backupsTotalSizeMB,
+				"limitMB",
+				limitperDbMB,
+			)
+			break
+		}
+
+		backup := oldestBackups[0]
+		if err := c.DeleteBackup(backup); err != nil {
+			c.logger.Error(
+				"Failed to delete exceeded backup",
+				"backupId",
+				backup.ID,
+				"databaseId",
+				databaseID,
+				"error",
+				err,
+			)
+			return err
+		}
+
+		c.logger.Info(
+			"Deleted exceeded backup",
+			"backupId",
+			backup.ID,
+			"databaseId",
+			databaseID,
+			"backupSizeMB",
+			backup.BackupSizeMb,
+			"totalSizeMB",
+			backupsTotalSizeMB,
+			"limitMB",
+			limitperDbMB,
+		)
+	}
+
+	return nil
+}
diff --git a/backend/internal/features/backups/backups/backuping/cleaner_test.go b/backend/internal/features/backups/backups/backuping/cleaner_test.go
new file mode 100644
index 0000000..942f6c8
--- /dev/null
+++ b/backend/internal/features/backups/backups/backuping/cleaner_test.go
@@ -0,0 +1,491 @@
+package backuping
+
+import (
+	"testing"
+	"time"
+
+	backups_core "databasus-backend/internal/features/backups/backups/core"
+	backups_config "databasus-backend/internal/features/backups/config"
+	"databasus-backend/internal/features/databases"
+	"databasus-backend/internal/features/intervals"
+	"databasus-backend/internal/features/notifiers"
+	"databasus-backend/internal/features/storages"
+	users_enums "databasus-backend/internal/features/users/enums"
+	users_testing "databasus-backend/internal/features/users/testing"
+	workspaces_testing "databasus-backend/internal/features/workspaces/testing"
+	"databasus-backend/internal/storage"
+	"databasus-backend/internal/util/period"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_CleanOldBackups_DeletesBackupsOlderThanStorePeriod(t *testing.T) {
+	router := CreateTestRouter()
+	owner := users_testing.CreateTestUser(users_enums.UserRoleMember)
+	workspace := workspaces_testing.CreateTestWorkspace("Test Workspace", owner, router)
+	storage := storages.CreateTestStorage(workspace.ID)
+	notifier := notifiers.CreateTestNotifier(workspace.ID)
+	database := databases.CreateTestDatabase(workspace.ID, storage, notifier)
+
+	// Create backup interval
+	interval := createTestInterval()
+
+	backupConfig := &backups_config.BackupConfig{
+		DatabaseID:       database.ID,
+		IsBackupsEnabled: true,
+		StorePeriod:      period.PeriodWeek,
+		StorageID:        &storage.ID,
+		BackupIntervalID: interval.ID,
+		BackupInterval:   interval,
+	}
+	_, err := backups_config.GetBackupConfigService().SaveBackupConfig(backupConfig)
+	assert.NoError(t, err)
+
+	// Create backups with different ages
+	now := time.Now().UTC()
+	oldBackup1 := &backups_core.Backup{
+		ID:           uuid.New(),
+		DatabaseID:   database.ID,
+		StorageID:    storage.ID,
+		Status:       backups_core.BackupStatusCompleted,
+		BackupSizeMb: 10,
+		CreatedAt:    now.Add(-10 * 24 * time.Hour), // 10 days old
+	}
+	oldBackup2 := &backups_core.Backup{
+		ID:           uuid.New(),
+		DatabaseID:   database.ID,
+		StorageID:    storage.ID,
+		Status:       backups_core.BackupStatusCompleted,
+		BackupSizeMb: 10,
+		CreatedAt:    now.Add(-8 * 24 * time.Hour), // 8 days old
+	}
+	recentBackup := &backups_core.Backup{
+		ID:           uuid.New(),
+		DatabaseID:   database.ID,
+		StorageID:    storage.ID,
+		Status:       backups_core.BackupStatusCompleted,
+		BackupSizeMb: 10,
+		CreatedAt:    now.Add(-3 * 24 * time.Hour), // 3 days old
+	}
+
+	err = backupRepository.Save(oldBackup1)
+	assert.NoError(t, err)
+	err = backupRepository.Save(oldBackup2)
+	assert.NoError(t, err)
+	err = backupRepository.Save(recentBackup)
+	assert.NoError(t, err)
+
+	// Run cleanup
+	cleaner := GetBackupCleaner()
+	err = cleaner.cleanOldBackups()
+	assert.NoError(t, err)
+
+	// Verify old backups deleted, recent backup remains
+	remainingBackups, err := backupRepository.FindByDatabaseID(database.ID)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(remainingBackups))
+	assert.Equal(t, recentBackup.ID, remainingBackups[0].ID)
+}
+
+func Test_CleanOldBackups_SkipsDatabaseWithForeverStorePeriod(t *testing.T) {
+	router := CreateTestRouter()
+	owner := users_testing.CreateTestUser(users_enums.UserRoleMember)
+	workspace := workspaces_testing.CreateTestWorkspace("Test Workspace", owner, router)
+	storage := storages.CreateTestStorage(workspace.ID)
+	notifier := notifiers.CreateTestNotifier(workspace.ID)
+	database := databases.CreateTestDatabase(workspace.ID, storage, notifier)
+
+	// Create backup interval
+	interval := createTestInterval()
+
+	backupConfig := &backups_config.BackupConfig{
+		DatabaseID:       database.ID,
+		IsBackupsEnabled: true,
+		StorePeriod:      period.PeriodForever,
+		StorageID:        &storage.ID,
+		BackupIntervalID: interval.ID,
+		BackupInterval:   interval,
+	}
+	_, err := backups_config.GetBackupConfigService().SaveBackupConfig(backupConfig)
+	assert.NoError(t, err)
+
+	// Create very old backup
+	oldBackup := &backups_core.Backup{
+		ID:           uuid.New(),
+		DatabaseID:   database.ID,
+		StorageID:    storage.ID,
+		Status:       backups_core.BackupStatusCompleted,
+		BackupSizeMb: 10,
+		CreatedAt:    time.Now().UTC().Add(-365 * 24 * time.Hour), // 1 year old
+	}
+	err = backupRepository.Save(oldBackup)
+	assert.NoError(t, err)
+
+	// Run cleanup
+	cleaner := GetBackupCleaner()
+	err = cleaner.cleanOldBackups()
+	assert.NoError(t, err)
+
+	// Verify backup still exists
+	remainingBackups, err := backupRepository.FindByDatabaseID(database.ID)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(remainingBackups))
+	assert.Equal(t, oldBackup.ID, remainingBackups[0].ID)
+}
+
+func Test_CleanExceededBackups_WhenUnderLimit_NoBackupsDeleted(t *testing.T) {
+	router := CreateTestRouter()
+	owner := users_testing.CreateTestUser(users_enums.UserRoleMember)
+	workspace := workspaces_testing.CreateTestWorkspace("Test Workspace", owner, router)
+	storage := storages.CreateTestStorage(workspace.ID)
+	notifier := notifiers.CreateTestNotifier(workspace.ID)
+	database := databases.CreateTestDatabase(workspace.ID, storage, notifier)
+
+	// Create backup interval
+	interval := createTestInterval()
+
+	backupConfig := &backups_config.BackupConfig{
+		DatabaseID:            database.ID,
+		IsBackupsEnabled:      true,
+		StorePeriod:           period.PeriodForever,
+		StorageID:             &storage.ID,
+		MaxBackupsTotalSizeMB: 100, // 100 MB limit
+		BackupIntervalID:      interval.ID,
+		BackupInterval:        interval,
+	}
+	_, err := backups_config.GetBackupConfigService().SaveBackupConfig(backupConfig)
+	assert.NoError(t, err)
+
+	// Create 3 backups totaling 50MB (under limit)
+	for i := 0; i < 3; i++ {
+		backup := &backups_core.Backup{
+			ID:           uuid.New(),
+			DatabaseID:   database.ID,
+			StorageID:    storage.ID,
+			Status:       backups_core.BackupStatusCompleted,
+			BackupSizeMb: 16.67,
+			CreatedAt:    time.Now().UTC().Add(-time.Duration(i) * time.Hour),
+		}
+		err = backupRepository.Save(backup)
+		assert.NoError(t, err)
+	}
+
+	// Run cleanup
+	cleaner := GetBackupCleaner()
+	err = cleaner.cleanExceededBackups()
+	assert.NoError(t, err)
+
+	// Verify all backups remain
+	remainingBackups, err := backupRepository.FindByDatabaseID(database.ID)
+	assert.NoError(t, err)
+	assert.Equal(t, 3, len(remainingBackups))
+}
+
+func Test_CleanExceededBackups_WhenOverLimit_DeletesOldestBackups(t *testing.T) {
+	router := CreateTestRouter()
+	owner := users_testing.CreateTestUser(users_enums.UserRoleMember)
+	workspace := workspaces_testing.CreateTestWorkspace("Test Workspace", owner, router)
+	storage := storages.CreateTestStorage(workspace.ID)
+	notifier := notifiers.CreateTestNotifier(workspace.ID)
+	database := databases.CreateTestDatabase(workspace.ID, storage, notifier)
+
+	// Create backup interval
+	interval := createTestInterval()
+
+	backupConfig := &backups_config.BackupConfig{
+		DatabaseID:            database.ID,
+		IsBackupsEnabled:      true,
+		StorePeriod:           period.PeriodForever,
+		StorageID:             &storage.ID,
+		MaxBackupsTotalSizeMB: 30, // 30 MB limit
+		BackupIntervalID:      interval.ID,
+		BackupInterval:        interval,
+	}
+	_, err := backups_config.GetBackupConfigService().SaveBackupConfig(backupConfig)
+	assert.NoError(t, err)
+
+	// Create 5 backups of 10MB each (total 50MB, over 30MB limit)
+	now := time.Now().UTC()
+	var backupIDs []uuid.UUID
+	for i := 0; i < 5; i++ {
+		backup := &backups_core.Backup{
+			ID:           uuid.New(),
+			DatabaseID:   database.ID,
+			StorageID:    storage.ID,
+			Status:       backups_core.BackupStatusCompleted,
+			BackupSizeMb: 10,
+			CreatedAt:    now.Add(-time.Duration(4-i) * time.Hour), // Oldest first
+		}
+		err = backupRepository.Save(backup)
+		assert.NoError(t, err)
+		backupIDs = append(backupIDs, backup.ID)
+	}
+
+	// Run cleanup
+	cleaner := GetBackupCleaner()
+	err = cleaner.cleanExceededBackups()
+	assert.NoError(t, err)
+
+	// Verify 2 oldest backups deleted, 3 newest remain
+	remainingBackups, err := backupRepository.FindByDatabaseID(database.ID)
+	assert.NoError(t, err)
+	assert.Equal(t, 3, len(remainingBackups))
+
+	// Check that the newest 3 backups remain
+	remainingIDs := make(map[uuid.UUID]bool)
+	for _, backup := range remainingBackups {
+		remainingIDs[backup.ID] = true
+	}
+	assert.False(t, remainingIDs[backupIDs[0]]) // Oldest deleted
+	assert.False(t, remainingIDs[backupIDs[1]]) // 2nd oldest deleted
+	assert.True(t, remainingIDs[backupIDs[2]])  // 3rd remains
+	assert.True(t, remainingIDs[backupIDs[3]])  // 4th remains
+	assert.True(t, remainingIDs[backupIDs[4]])  // Newest remains
+}
+
+func Test_CleanExceededBackups_SkipsInProgressBackups(t *testing.T) {
+	router := CreateTestRouter()
+	owner := users_testing.CreateTestUser(users_enums.UserRoleMember)
+	workspace := workspaces_testing.CreateTestWorkspace("Test Workspace", owner, router)
+	storage := storages.CreateTestStorage(workspace.ID)
+	notifier := notifiers.CreateTestNotifier(workspace.ID)
+	database := databases.CreateTestDatabase(workspace.ID, storage, notifier)
+
+	// Create backup interval
+	interval := createTestInterval()
+
+	backupConfig := &backups_config.BackupConfig{
+		DatabaseID:            database.ID,
+		IsBackupsEnabled:      true,
+		StorePeriod:           period.PeriodForever,
+		StorageID:             &storage.ID,
+		MaxBackupsTotalSizeMB: 50, // 50 MB limit
+		BackupIntervalID:      interval.ID,
+		BackupInterval:        interval,
+	}
+	_, err := backups_config.GetBackupConfigService().SaveBackupConfig(backupConfig)
+	assert.NoError(t, err)
+
+	now := time.Now().UTC()
+
+	// Create 3 completed backups of 30MB each
+	completedBackups := make([]*backups_core.Backup, 3)
+	for i := 0; i < 3; i++ {
+		backup := &backups_core.Backup{
+			ID:           uuid.New(),
+			DatabaseID:   database.ID,
+			StorageID:    storage.ID,
+			Status:       backups_core.BackupStatusCompleted,
+			BackupSizeMb: 30,
+			CreatedAt:    now.Add(-time.Duration(3-i) * time.Hour),
+		}
+		err = backupRepository.Save(backup)
+		assert.NoError(t, err)
+		completedBackups[i] = backup
+	}
+
+	// Create 1 in-progress backup (should be excluded from size calculation and deletion)
+	inProgressBackup := &backups_core.Backup{
+		ID:           uuid.New(),
+		DatabaseID:   database.ID,
+		StorageID:    storage.ID,
+		Status:       backups_core.BackupStatusInProgress,
+		BackupSizeMb: 10,
+		CreatedAt:    now,
+	}
+	err = backupRepository.Save(inProgressBackup)
+	assert.NoError(t, err)
+
+	// Run cleanup
+	cleaner := GetBackupCleaner()
+	err = cleaner.cleanExceededBackups()
+	assert.NoError(t, err)
+
+	// Verify: only completed backups deleted, in-progress remains
+	remainingBackups, err := backupRepository.FindByDatabaseID(database.ID)
+	assert.NoError(t, err)
+
+	// Should have in-progress + 1 completed (total 40MB completed + 10MB in-progress)
+	assert.GreaterOrEqual(t, len(remainingBackups), 2)
+
+	// Verify in-progress backup still exists
+	var inProgressFound bool
+	for _, backup := range remainingBackups {
+		if backup.ID == inProgressBackup.ID {
+			inProgressFound = true
+			assert.Equal(t, backups_core.BackupStatusInProgress, backup.Status)
+		}
+	}
+	assert.True(t, inProgressFound, "In-progress backup should not be deleted")
+}
+
+func Test_CleanExceededBackups_WithZeroLimit_SkipsDatabase(t *testing.T) {
+	router := CreateTestRouter()
+	owner := users_testing.CreateTestUser(users_enums.UserRoleMember)
+	workspace := workspaces_testing.CreateTestWorkspace("Test Workspace", owner, router)
+	storage := storages.CreateTestStorage(workspace.ID)
+	notifier := notifiers.CreateTestNotifier(workspace.ID)
+	database := databases.CreateTestDatabase(workspace.ID, storage, notifier)
+
+	// Create backup interval
+	interval := createTestInterval()
+
+	backupConfig := &backups_config.BackupConfig{
+		DatabaseID:            database.ID,
+		IsBackupsEnabled:      true,
+		StorePeriod:           period.PeriodForever,
+		StorageID:             &storage.ID,
+		MaxBackupsTotalSizeMB: 0, // No size limit
+		BackupIntervalID:      interval.ID,
+		BackupInterval:        interval,
+	}
+	_, err := backups_config.GetBackupConfigService().SaveBackupConfig(backupConfig)
+	assert.NoError(t, err)
+
+	// Create large backups
+	for i := 0; i < 10; i++ {
+		backup := &backups_core.Backup{
+			ID:           uuid.New(),
+			DatabaseID:   database.ID,
+			StorageID:    storage.ID,
+			Status:       backups_core.BackupStatusCompleted,
+			BackupSizeMb: 100,
+			CreatedAt:    time.Now().UTC().Add(-time.Duration(i) * time.Hour),
+		}
+		err = backupRepository.Save(backup)
+		assert.NoError(t, err)
+	}
+
+	// Run cleanup
+	cleaner := GetBackupCleaner()
+	err = cleaner.cleanExceededBackups()
+	assert.NoError(t, err)
+
+	// Verify all backups remain
+	remainingBackups, err := backupRepository.FindByDatabaseID(database.ID)
+	assert.NoError(t, err)
+	assert.Equal(t, 10, len(remainingBackups))
+}
+
+func Test_GetTotalSizeByDatabase_CalculatesCorrectly(t *testing.T) {
+	router := CreateTestRouter()
+	owner := users_testing.CreateTestUser(users_enums.UserRoleMember)
+	workspace := workspaces_testing.CreateTestWorkspace("Test Workspace", owner, router)
+	storage := storages.CreateTestStorage(workspace.ID)
+	notifier := notifiers.CreateTestNotifier(workspace.ID)
+	database := databases.CreateTestDatabase(workspace.ID, storage, notifier)
+
+	// Create completed backups
+	completedBackup1 := &backups_core.Backup{
+		ID:           uuid.New(),
+		DatabaseID:   database.ID,
+		StorageID:    storage.ID,
+		Status:       backups_core.BackupStatusCompleted,
+		BackupSizeMb: 10.5,
+		CreatedAt:    time.Now().UTC(),
+	}
+	completedBackup2 := &backups_core.Backup{
+		ID:           uuid.New(),
+		DatabaseID:   database.ID,
+		StorageID:    storage.ID,
+		Status:       backups_core.BackupStatusCompleted,
+		BackupSizeMb: 20.3,
+		CreatedAt:    time.Now().UTC(),
+	}
+	// Create failed backup (should be included)
+	failedBackup := &backups_core.Backup{
+		ID:           uuid.New(),
+		DatabaseID:   database.ID,
+		StorageID:    storage.ID,
+		Status:       backups_core.BackupStatusFailed,
+		BackupSizeMb: 5.2,
+		CreatedAt:    time.Now().UTC(),
+	}
+	// Create in-progress backup (should be excluded)
+	inProgressBackup := &backups_core.Backup{
+		ID:           uuid.New(),
+		DatabaseID:   database.ID,
+		StorageID:    storage.ID,
+		Status:       backups_core.BackupStatusInProgress,
+		BackupSizeMb: 100,
+		CreatedAt:    time.Now().UTC(),
+	}
+
+	err := backupRepository.Save(completedBackup1)
+	assert.NoError(t, err)
+	err = backupRepository.Save(completedBackup2)
+	assert.NoError(t, err)
+	err = backupRepository.Save(failedBackup)
+	assert.NoError(t, err)
+	err = backupRepository.Save(inProgressBackup)
+	assert.NoError(t, err)
+
+	// Calculate total size
+	totalSize, err := backupRepository.GetTotalSizeByDatabase(database.ID)
+	assert.NoError(t, err)
+
+	// Should be 10.5 + 20.3 + 5.2 = 36.0 (excluding in-progress 100)
+	assert.InDelta(t, 36.0, totalSize, 0.1)
+}
+
+// Mock listener for testing
+type mockBackupRemoveListener struct {
+	onBeforeBackupRemove func(*backups_core.Backup) error
+}
+
+func (m *mockBackupRemoveListener) OnBeforeBackupRemove(backup *backups_core.Backup) error {
+	if m.onBeforeBackupRemove != nil {
+		return m.onBeforeBackupRemove(backup)
+	}
+
+	return nil
+}
+
+// Test_DeleteBackup_WhenStorageDeleteFails_BackupStillRemovedFromDatabase verifies resilience
+// when storage becomes unavailable. Even if storage.DeleteFile fails (e.g., storage is offline,
+// credentials changed, or storage was deleted), the backup record should still be removed from
+// the database. This prevents orphaned backup records when storage is no longer accessible.
+func Test_DeleteBackup_WhenStorageDeleteFails_BackupStillRemovedFromDatabase(t *testing.T) {
+	router := CreateTestRouter()
+	owner := users_testing.CreateTestUser(users_enums.UserRoleMember)
+	workspace := workspaces_testing.CreateTestWorkspace("Test Workspace", owner, router)
+	testStorage := storages.CreateTestStorage(workspace.ID)
+	notifier := notifiers.CreateTestNotifier(workspace.ID)
+	database := databases.CreateTestDatabase(workspace.ID, testStorage, notifier)
+
+	backup := &backups_core.Backup{
+		ID:           uuid.New(),
+		DatabaseID:   database.ID,
+		StorageID:    testStorage.ID,
+		Status:       backups_core.BackupStatusCompleted,
+		BackupSizeMb: 10,
+		CreatedAt:    time.Now().UTC(),
+	}
+	err := backupRepository.Save(backup)
+	assert.NoError(t, err)
+
+	cleaner := GetBackupCleaner()
+
+	err = cleaner.DeleteBackup(backup)
+	assert.NoError(t, err, "DeleteBackup should succeed even when storage file doesn't exist")
+
+	deletedBackup, err := backupRepository.FindByID(backup.ID)
+	assert.Error(t, err, "Backup should not exist in database")
+	assert.Nil(t, deletedBackup)
+}
+
+func createTestInterval() *intervals.Interval {
+	timeOfDay := "04:00"
+	interval := &intervals.Interval{
+		Interval:  intervals.IntervalDaily,
+		TimeOfDay: &timeOfDay,
+	}
+
+	err := storage.GetDb().Create(interval).Error
+	if err != nil {
+		panic(err)
+	}
+
+	return interval
+}
diff --git a/backend/internal/features/backups/backups/backuping/di.go b/backend/internal/features/backups/backups/backuping/di.go
index 86b3893..b9235de 100644
--- a/backend/internal/features/backups/backups/backuping/di.go
+++ b/backend/internal/features/backups/backups/backuping/di.go
@@ -24,6 +24,17 @@ var backupRepository = &backups_core.BackupRepository{}
 
 var taskCancelManager = tasks_cancellation.GetTaskCancelManager()
 
+var backupCleaner = &BackupCleaner{
+	backupRepository:      backupRepository,
+	storageService:        storages.GetStorageService(),
+	backupConfigService:   backups_config.GetBackupConfigService(),
+	fieldEncryptor:        encryption.GetFieldEncryptor(),
+	logger:                logger.GetLogger(),
+	backupRemoveListeners: []backups_core.BackupRemoveListener{},
+	runOnce:               sync.Once{},
+	hasRun:                atomic.Bool{},
+}
+
 var backupNodesRegistry = &BackupNodesRegistry{
 	client:            cache_utils.GetValkeyClient(),
 	logger:            logger.GetLogger(),
@@ -59,7 +70,6 @@ var backuperNode = &BackuperNode{
 var backupsScheduler = &BackupsScheduler{
 	backupRepository:      backupRepository,
 	backupConfigService:   backups_config.GetBackupConfigService(),
-	storageService:        storages.GetStorageService(),
 	taskCancelManager:     taskCancelManager,
 	backupNodesRegistry:   backupNodesRegistry,
 	lastBackupTime:        time.Now().UTC(),
@@ -81,3 +91,7 @@ func GetBackuperNode() *BackuperNode {
 func GetBackupNodesRegistry() *BackupNodesRegistry {
 	return backupNodesRegistry
 }
+
+func GetBackupCleaner() *BackupCleaner {
+	return backupCleaner
+}
diff --git a/backend/internal/features/backups/backups/backuping/mocks.go b/backend/internal/features/backups/backups/backuping/mocks.go
index 5bd84a4..7807f06 100644
--- a/backend/internal/features/backups/backups/backuping/mocks.go
+++ b/backend/internal/features/backups/backups/backuping/mocks.go
@@ -1,8 +1,16 @@
 package backuping
 
 import (
-	"databasus-backend/internal/features/notifiers"
+	"context"
+	"errors"
 
+	common "databasus-backend/internal/features/backups/backups/common"
+	backups_config "databasus-backend/internal/features/backups/config"
+	"databasus-backend/internal/features/databases"
+	"databasus-backend/internal/features/notifiers"
+	"databasus-backend/internal/features/storages"
+
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/mock"
 )
 
@@ -17,3 +25,113 @@ func (m *MockNotificationSender) SendNotification(
 ) {
 	m.Called(notifier, title, message)
 }
+
+type CreateFailedBackupUsecase struct{}
+
+func (uc *CreateFailedBackupUsecase) Execute(
+	ctx context.Context,
+	backupID uuid.UUID,
+	backupConfig *backups_config.BackupConfig,
+	database *databases.Database,
+	storage *storages.Storage,
+	backupProgressListener func(completedMBs float64),
+) (*common.BackupMetadata, error) {
+	backupProgressListener(10)
+	return nil, errors.New("backup failed")
+}
+
+type CreateSuccessBackupUsecase struct{}
+
+func (uc *CreateSuccessBackupUsecase) Execute(
+	ctx context.Context,
+	backupID uuid.UUID,
+	backupConfig *backups_config.BackupConfig,
+	database *databases.Database,
+	storage *storages.Storage,
+	backupProgressListener func(completedMBs float64),
+) (*common.BackupMetadata, error) {
+	backupProgressListener(10)
+	return &common.BackupMetadata{
+		EncryptionSalt: nil,
+		EncryptionIV:   nil,
+		Encryption:     backups_config.BackupEncryptionNone,
+	}, nil
+}
+
+// CreateLargeBackupUsecase simulates a large backup (10000 MB)
+type CreateLargeBackupUsecase struct{}
+
+func (uc *CreateLargeBackupUsecase) Execute(
+	ctx context.Context,
+	backupID uuid.UUID,
+	backupConfig *backups_config.BackupConfig,
+	database *databases.Database,
+	storage *storages.Storage,
+	backupProgressListener func(completedMBs float64),
+) (*common.BackupMetadata, error) {
+	backupProgressListener(10000)
+	return &common.BackupMetadata{
+		EncryptionSalt: nil,
+		EncryptionIV:   nil,
+		Encryption:     backups_config.BackupEncryptionNone,
+	}, nil
+}
+
+// CreateProgressiveBackupUsecase simulates progressive size updates that exceed limit
+type CreateProgressiveBackupUsecase struct{}
+
+func (uc *CreateProgressiveBackupUsecase) Execute(
+	ctx context.Context,
+	backupID uuid.UUID,
+	backupConfig *backups_config.BackupConfig,
+	database *databases.Database,
+	storage *storages.Storage,
+	backupProgressListener func(completedMBs float64),
+) (*common.BackupMetadata, error) {
+	// Simulate progressive backup that grows beyond limit
+	backupProgressListener(1)
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+
+	backupProgressListener(3)
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+
+	backupProgressListener(5)
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+
+	backupProgressListener(10) // This exceeds the 5 MB limit
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+
+	// Should not reach here due to cancellation
+	return &common.BackupMetadata{
+		EncryptionSalt: nil,
+		EncryptionIV:   nil,
+		Encryption:     backups_config.BackupEncryptionNone,
+	}, nil
+}
+
+// CreateMediumBackupUsecase simulates a 50 MB backup
+type CreateMediumBackupUsecase struct{}
+
+func (uc *CreateMediumBackupUsecase) Execute(
+	ctx context.Context,
+	backupID uuid.UUID,
+	backupConfig *backups_config.BackupConfig,
+	database *databases.Database,
+	storage *storages.Storage,
+	backupProgressListener func(completedMBs float64),
+) (*common.BackupMetadata, error) {
+	backupProgressListener(50)
+	return &common.BackupMetadata{
+		EncryptionSalt: nil,
+		EncryptionIV:   nil,
+		Encryption:     backups_config.BackupEncryptionNone,
+	}, nil
+}
diff --git a/backend/internal/features/backups/backups/backuping/scheduler.go b/backend/internal/features/backups/backups/backuping/scheduler.go
index deb9f78..f774fec 100644
--- a/backend/internal/features/backups/backups/backuping/scheduler.go
+++ b/backend/internal/features/backups/backups/backuping/scheduler.go
@@ -13,10 +13,7 @@ import (
 	"databasus-backend/internal/config"
 	backups_core "databasus-backend/internal/features/backups/backups/core"
 	backups_config "databasus-backend/internal/features/backups/config"
-	"databasus-backend/internal/features/storages"
 	task_cancellation "databasus-backend/internal/features/tasks/cancellation"
-	"databasus-backend/internal/util/encryption"
-	"databasus-backend/internal/util/period"
 )
 
 const (
@@ -28,7 +25,6 @@ const (
 type BackupsScheduler struct {
 	backupRepository    *backups_core.BackupRepository
 	backupConfigService *backups_config.BackupConfigService
-	storageService      *storages.StorageService
 	taskCancelManager   *task_cancellation.TaskCancelManager
 	backupNodesRegistry *BackupNodesRegistry
 
@@ -84,10 +80,6 @@ func (s *BackupsScheduler) Run(ctx context.Context) {
 			case <-ctx.Done():
 				return
 			case <-ticker.C:
-				if err := s.cleanOldBackups(); err != nil {
-					s.logger.Error("Failed to clean old backups", "error", err)
-				}
-
 				if err := s.checkDeadNodesAndFailBackups(); err != nil {
 					s.logger.Error("Failed to check dead nodes and fail backups", "error", err)
 				}
@@ -293,6 +285,10 @@ func (s *BackupsScheduler) GetRemainedBackupTryCount(lastBackup *backups_core.Ba
 		return 0
 	}
 
+	if lastBackup.IsSkipRetry {
+		return 0
+	}
+
 	backupConfig, err := s.backupConfigService.GetBackupConfigByDbId(lastBackup.DatabaseID)
 	if err != nil {
 		s.logger.Error("Failed to get backup config by database ID", "error", err)
@@ -325,74 +321,6 @@ func (s *BackupsScheduler) GetRemainedBackupTryCount(lastBackup *backups_core.Ba
 	return maxFailedTriesCount - len(lastFailedBackups)
 }
 
-func (s *BackupsScheduler) cleanOldBackups() error {
-	enabledBackupConfigs, err := s.backupConfigService.GetBackupConfigsWithEnabledBackups()
-	if err != nil {
-		return err
-	}
-
-	for _, backupConfig := range enabledBackupConfigs {
-		backupStorePeriod := backupConfig.StorePeriod
-
-		if backupStorePeriod == period.PeriodForever {
-			continue
-		}
-
-		storeDuration := backupStorePeriod.ToDuration()
-		dateBeforeBackupsShouldBeDeleted := time.Now().UTC().Add(-storeDuration)
-
-		oldBackups, err := s.backupRepository.FindBackupsBeforeDate(
-			backupConfig.DatabaseID,
-			dateBeforeBackupsShouldBeDeleted,
-		)
-		if err != nil {
-			s.logger.Error(
-				"Failed to find old backups for database",
-				"databaseId",
-				backupConfig.DatabaseID,
-				"error",
-				err,
-			)
-			continue
-		}
-
-		for _, backup := range oldBackups {
-			storage, err := s.storageService.GetStorageByID(backup.StorageID)
-			if err != nil {
-				s.logger.Error(
-					"Failed to get storage by ID",
-					"storageId",
-					backup.StorageID,
-					"error",
-					err,
-				)
-				continue
-			}
-
-			encryptor := encryption.GetFieldEncryptor()
-			err = storage.DeleteFile(encryptor, backup.ID)
-			if err != nil {
-				s.logger.Error("Failed to delete backup file", "backupId", backup.ID, "error", err)
-			}
-
-			if err := s.backupRepository.DeleteByID(backup.ID); err != nil {
-				s.logger.Error("Failed to delete old backup", "backupId", backup.ID, "error", err)
-				continue
-			}
-
-			s.logger.Info(
-				"Deleted old backup",
-				"backupId",
-				backup.ID,
-				"databaseId",
-				backupConfig.DatabaseID,
-			)
-		}
-	}
-
-	return nil
-}
-
 func (s *BackupsScheduler) runPendingBackups() error {
 	enabledBackupConfigs, err := s.backupConfigService.GetBackupConfigsWithEnabledBackups()
 	if err != nil {
diff --git a/backend/internal/features/backups/backups/backuping/scheduler_test.go b/backend/internal/features/backups/backups/backuping/scheduler_test.go
index b3e8b6f..a0d6cc7 100644
--- a/backend/internal/features/backups/backups/backuping/scheduler_test.go
+++ b/backend/internal/features/backups/backups/backuping/scheduler_test.go
@@ -1101,3 +1101,84 @@ func Test_StartBackup_WhenBackupAlreadyInProgress_SkipsNewBackup(t *testing.T) {
 
 	time.Sleep(200 * time.Millisecond)
 }
+
+func Test_RunPendingBackups_WhenLastBackupFailedWithIsSkipRetry_SkipsBackupEvenWithRetriesEnabled(
+	t *testing.T,
+) {
+	cache_utils.ClearAllCache()
+	backuperNode := CreateTestBackuperNode()
+	cancel := StartBackuperNodeForTest(t, backuperNode)
+	defer StopBackuperNodeForTest(t, cancel, backuperNode)
+
+	user := users_testing.CreateTestUser(users_enums.UserRoleAdmin)
+	router := CreateTestRouter()
+	workspace := workspaces_testing.CreateTestWorkspace("Test Workspace", user, router)
+	storage := storages.CreateTestStorage(workspace.ID)
+	notifier := notifiers.CreateTestNotifier(workspace.ID)
+	database := databases.CreateTestDatabase(workspace.ID, storage, notifier)
+
+	defer func() {
+		backups, _ := backupRepository.FindByDatabaseID(database.ID)
+		for _, backup := range backups {
+			backupRepository.DeleteByID(backup.ID)
+		}
+
+		databases.RemoveTestDatabase(database)
+		time.Sleep(50 * time.Millisecond)
+		storages.RemoveTestStorage(storage.ID)
+		notifiers.RemoveTestNotifier(notifier)
+		workspaces_testing.RemoveTestWorkspace(workspace, router)
+	}()
+
+	// Enable backups with retries enabled and high retry count
+	backupConfig, err := backups_config.GetBackupConfigService().GetBackupConfigByDbId(database.ID)
+	assert.NoError(t, err)
+
+	timeOfDay := "04:00"
+	backupConfig.BackupInterval = &intervals.Interval{
+		Interval:  intervals.IntervalDaily,
+		TimeOfDay: &timeOfDay,
+	}
+	backupConfig.IsBackupsEnabled = true
+	backupConfig.StorePeriod = period.PeriodWeek
+	backupConfig.Storage = storage
+	backupConfig.StorageID = &storage.ID
+	backupConfig.IsRetryIfFailed = true
+	backupConfig.MaxFailedTriesCount = 5
+
+	_, err = backups_config.GetBackupConfigService().SaveBackupConfig(backupConfig)
+	assert.NoError(t, err)
+
+	// Create a failed backup with IsSkipRetry set to true
+	failMessage := "backup failed due to size limit exceeded"
+	backupRepository.Save(&backups_core.Backup{
+		DatabaseID: database.ID,
+		StorageID:  storage.ID,
+
+		Status:      backups_core.BackupStatusFailed,
+		FailMessage: &failMessage,
+		IsSkipRetry: true,
+
+		CreatedAt: time.Now().UTC().Add(-1 * time.Hour),
+	})
+
+	// Verify GetRemainedBackupTryCount returns 0 even though retries are enabled
+	lastBackup, err := backupRepository.FindLastByDatabaseID(database.ID)
+	assert.NoError(t, err)
+	assert.NotNil(t, lastBackup)
+
+	remainedTries := GetBackupsScheduler().GetRemainedBackupTryCount(lastBackup)
+	assert.Equal(t, 0, remainedTries, "Should return 0 tries when IsSkipRetry is true")
+
+	// Run the scheduler
+	GetBackupsScheduler().runPendingBackups()
+
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify no new backup was created (still only 1 backup exists)
+	backups, err := backupRepository.FindByDatabaseID(database.ID)
+	assert.NoError(t, err)
+	assert.Len(t, backups, 1, "No retry should be attempted when IsSkipRetry is true")
+
+	time.Sleep(200 * time.Millisecond)
+}
diff --git a/backend/internal/features/backups/backups/backuping/testing.go b/backend/internal/features/backups/backups/backuping/testing.go
index fa06770..22aa39e 100644
--- a/backend/internal/features/backups/backups/backuping/testing.go
+++ b/backend/internal/features/backups/backups/backuping/testing.go
@@ -57,17 +57,16 @@ func CreateTestBackuperNode() *BackuperNode {
 
 func CreateTestScheduler() *BackupsScheduler {
 	return &BackupsScheduler{
-		backupRepository,
-		backups_config.GetBackupConfigService(),
-		storages.GetStorageService(),
-		taskCancelManager,
-		backupNodesRegistry,
-		time.Now().UTC(),
-		logger.GetLogger(),
-		make(map[uuid.UUID]BackupToNodeRelation),
-		CreateTestBackuperNode(),
-		sync.Once{},
-		atomic.Bool{},
+		backupRepository:      backupRepository,
+		backupConfigService:   backups_config.GetBackupConfigService(),
+		taskCancelManager:     taskCancelManager,
+		backupNodesRegistry:   backupNodesRegistry,
+		lastBackupTime:        time.Now().UTC(),
+		logger:                logger.GetLogger(),
+		backupToNodeRelations: make(map[uuid.UUID]BackupToNodeRelation),
+		backuperNode:          CreateTestBackuperNode(),
+		runOnce:               sync.Once{},
+		hasRun:                atomic.Bool{},
 	}
 }
 
diff --git a/backend/internal/features/backups/backups/core/model.go b/backend/internal/features/backups/backups/core/model.go
index afff310..f060ab2 100644
--- a/backend/internal/features/backups/backups/core/model.go
+++ b/backend/internal/features/backups/backups/core/model.go
@@ -15,6 +15,7 @@ type Backup struct {
 
 	Status      BackupStatus `json:"status"      gorm:"column:status;not null"`
 	FailMessage *string      `json:"failMessage" gorm:"column:fail_message"`
+	IsSkipRetry bool         `json:"isSkipRetry" gorm:"column:is_skip_retry;type:boolean;not null"`
 
 	BackupSizeMb float64 `json:"backupSizeMb" gorm:"column:backup_size_mb;default:0"`
 
diff --git a/backend/internal/features/backups/backups/core/repository.go b/backend/internal/features/backups/backups/core/repository.go
index 8282348..dfa81f0 100644
--- a/backend/internal/features/backups/backups/core/repository.go
+++ b/backend/internal/features/backups/backups/core/repository.go
@@ -212,3 +212,36 @@ func (r *BackupRepository) CountByDatabaseID(databaseID uuid.UUID) (int64, error
 
 	return count, nil
 }
+
+func (r *BackupRepository) GetTotalSizeByDatabase(databaseID uuid.UUID) (float64, error) {
+	var totalSize float64
+
+	if err := storage.
+		GetDb().
+		Model(&Backup{}).
+		Select("COALESCE(SUM(backup_size_mb), 0)").
+		Where("database_id = ? AND status != ?", databaseID, BackupStatusInProgress).
+		Scan(&totalSize).Error; err != nil {
+		return 0, err
+	}
+
+	return totalSize, nil
+}
+
+func (r *BackupRepository) FindOldestByDatabaseExcludingInProgress(
+	databaseID uuid.UUID,
+	limit int,
+) ([]*Backup, error) {
+	var backups []*Backup
+
+	if err := storage.
+		GetDb().
+		Where("database_id = ? AND status != ?", databaseID, BackupStatusInProgress).
+		Order("created_at ASC").
+		Limit(limit).
+		Find(&backups).Error; err != nil {
+		return nil, err
+	}
+
+	return backups, nil
+}
diff --git a/backend/internal/features/backups/backups/di.go b/backend/internal/features/backups/backups/di.go
index cddb159..d79dc6a 100644
--- a/backend/internal/features/backups/backups/di.go
+++ b/backend/internal/features/backups/backups/di.go
@@ -25,22 +25,23 @@ var backupRepository = &backups_core.BackupRepository{}
 var taskCancelManager = task_cancellation.GetTaskCancelManager()
 
 var backupService = &BackupService{
-	databaseService:        databases.GetDatabaseService(),
-	storageService:         storages.GetStorageService(),
-	backupRepository:       backupRepository,
-	notifierService:        notifiers.GetNotifierService(),
-	notificationSender:     notifiers.GetNotifierService(),
-	backupConfigService:    backups_config.GetBackupConfigService(),
-	secretKeyService:       encryption_secrets.GetSecretKeyService(),
-	fieldEncryptor:         encryption.GetFieldEncryptor(),
-	createBackupUseCase:    usecases.GetCreateBackupUsecase(),
-	logger:                 logger.GetLogger(),
-	backupRemoveListeners:  []backups_core.BackupRemoveListener{},
-	workspaceService:       workspaces_services.GetWorkspaceService(),
-	auditLogService:        audit_logs.GetAuditLogService(),
-	taskCancelManager:      taskCancelManager,
-	downloadTokenService:   backups_download.GetDownloadTokenService(),
-	backupSchedulerService: backuping.GetBackupsScheduler(),
+	databases.GetDatabaseService(),
+	storages.GetStorageService(),
+	backupRepository,
+	notifiers.GetNotifierService(),
+	notifiers.GetNotifierService(),
+	backups_config.GetBackupConfigService(),
+	encryption_secrets.GetSecretKeyService(),
+	encryption.GetFieldEncryptor(),
+	usecases.GetCreateBackupUsecase(),
+	logger.GetLogger(),
+	[]backups_core.BackupRemoveListener{},
+	workspaces_services.GetWorkspaceService(),
+	audit_logs.GetAuditLogService(),
+	taskCancelManager,
+	backups_download.GetDownloadTokenService(),
+	backuping.GetBackupsScheduler(),
+	backuping.GetBackupCleaner(),
 }
 
 var backupController = &BackupController{
diff --git a/backend/internal/features/backups/backups/service.go b/backend/internal/features/backups/backups/service.go
index 527797c..101305a 100644
--- a/backend/internal/features/backups/backups/service.go
+++ b/backend/internal/features/backups/backups/service.go
@@ -46,6 +46,7 @@ type BackupService struct {
 	taskCancelManager      *task_cancellation.TaskCancelManager
 	downloadTokenService   *backups_download.DownloadTokenService
 	backupSchedulerService *backuping.BackupsScheduler
+	backupCleaner          *backuping.BackupCleaner
 }
 
 func (s *BackupService) AddBackupRemoveListener(listener backups_core.BackupRemoveListener) {
@@ -189,7 +190,7 @@ func (s *BackupService) DeleteBackup(
 		database.WorkspaceID,
 	)
 
-	return s.deleteBackup(backup)
+	return s.backupCleaner.DeleteBackup(backup)
 }
 
 func (s *BackupService) GetBackup(backupID uuid.UUID) (*backups_core.Backup, error) {
@@ -292,29 +293,6 @@ func (s *BackupService) GetBackupFile(
 	return reader, backup, database, nil
 }
 
-func (s *BackupService) deleteBackup(backup *backups_core.Backup) error {
-	for _, listener := range s.backupRemoveListeners {
-		if err := listener.OnBeforeBackupRemove(backup); err != nil {
-			return err
-		}
-	}
-
-	storage, err := s.storageService.GetStorageByID(backup.StorageID)
-	if err != nil {
-		return err
-	}
-
-	err = storage.DeleteFile(s.fieldEncryptor, backup.ID)
-	if err != nil {
-		// we do not return error here, because sometimes clean up performed
-		// before unavailable storage removal or change - therefore we should
-		// proceed even in case of error
-		s.logger.Error("Failed to delete backup file", "error", err)
-	}
-
-	return s.backupRepository.DeleteByID(backup.ID)
-}
-
 func (s *BackupService) deleteDbBackups(databaseID uuid.UUID) error {
 	dbBackupsInProgress, err := s.backupRepository.FindByDatabaseIdAndStatus(
 		databaseID,
@@ -336,7 +314,7 @@ func (s *BackupService) deleteDbBackups(databaseID uuid.UUID) error {
 	}
 
 	for _, dbBackup := range dbBackups {
-		err := s.deleteBackup(dbBackup)
+		err := s.backupCleaner.DeleteBackup(dbBackup)
 		if err != nil {
 			return err
 		}
diff --git a/backend/internal/features/backups/config/model.go b/backend/internal/features/backups/config/model.go
index 82b425c..7e63b2b 100644
--- a/backend/internal/features/backups/config/model.go
+++ b/backend/internal/features/backups/config/model.go
@@ -31,6 +31,11 @@ type BackupConfig struct {
 	MaxFailedTriesCount int  `json:"maxFailedTriesCount" gorm:"column:max_failed_tries_count;type:int;not null"`
 
 	Encryption BackupEncryption `json:"encryption" gorm:"column:encryption;type:text;not null;default:'NONE'"`
+
+	// MaxBackupSizeMB limits individual backup size. 0 = unlimited.
+	MaxBackupSizeMB int64 `json:"maxBackupSizeMb"       gorm:"column:max_backup_size_mb;type:int;not null"`
+	// MaxBackupsTotalSizeMB limits total size of all backups. 0 = unlimited.
+	MaxBackupsTotalSizeMB int64 `json:"maxBackupsTotalSizeMb" gorm:"column:max_backups_total_size_mb;type:int;not null"`
 }
 
 func (h *BackupConfig) TableName() string {
@@ -89,20 +94,30 @@ func (b *BackupConfig) Validate() error {
 		return errors.New("encryption must be NONE or ENCRYPTED")
 	}
 
+	if b.MaxBackupSizeMB < 0 {
+		return errors.New("max backup size must be non-negative")
+	}
+
+	if b.MaxBackupsTotalSizeMB < 0 {
+		return errors.New("max backups total size must be non-negative")
+	}
+
 	return nil
 }
 
 func (b *BackupConfig) Copy(newDatabaseID uuid.UUID) *BackupConfig {
 	return &BackupConfig{
-		DatabaseID:          newDatabaseID,
-		IsBackupsEnabled:    b.IsBackupsEnabled,
-		StorePeriod:         b.StorePeriod,
-		BackupIntervalID:    uuid.Nil,
-		BackupInterval:      b.BackupInterval.Copy(),
-		StorageID:           b.StorageID,
-		SendNotificationsOn: b.SendNotificationsOn,
-		IsRetryIfFailed:     b.IsRetryIfFailed,
-		MaxFailedTriesCount: b.MaxFailedTriesCount,
-		Encryption:          b.Encryption,
+		DatabaseID:            newDatabaseID,
+		IsBackupsEnabled:      b.IsBackupsEnabled,
+		StorePeriod:           b.StorePeriod,
+		BackupIntervalID:      uuid.Nil,
+		BackupInterval:        b.BackupInterval.Copy(),
+		StorageID:             b.StorageID,
+		SendNotificationsOn:   b.SendNotificationsOn,
+		IsRetryIfFailed:       b.IsRetryIfFailed,
+		MaxFailedTriesCount:   b.MaxFailedTriesCount,
+		Encryption:            b.Encryption,
+		MaxBackupSizeMB:       b.MaxBackupSizeMB,
+		MaxBackupsTotalSizeMB: b.MaxBackupsTotalSizeMB,
 	}
 }
diff --git a/backend/internal/features/restores/di.go b/backend/internal/features/restores/di.go
index 8670583..483549e 100644
--- a/backend/internal/features/restores/di.go
+++ b/backend/internal/features/restores/di.go
@@ -6,6 +6,7 @@ import (
 
 	audit_logs "databasus-backend/internal/features/audit_logs"
 	"databasus-backend/internal/features/backups/backups"
+	"databasus-backend/internal/features/backups/backups/backuping"
 	backups_config "databasus-backend/internal/features/backups/config"
 	"databasus-backend/internal/features/databases"
 	"databasus-backend/internal/features/disk"
@@ -51,6 +52,7 @@ func SetupDependencies() {
 
 	setupOnce.Do(func() {
 		backups.GetBackupService().AddBackupRemoveListener(restoreService)
+		backuping.GetBackupCleaner().AddBackupRemoveListener(restoreService)
 
 		isSetup.Store(true)
 	})
diff --git a/backend/migrations/20260119124334_add_backup_size_limits.sql b/backend/migrations/20260119124334_add_backup_size_limits.sql
new file mode 100644
index 0000000..3d5ac22
--- /dev/null
+++ b/backend/migrations/20260119124334_add_backup_size_limits.sql
@@ -0,0 +1,23 @@
+-- +goose Up
+-- +goose StatementBegin
+
+ALTER TABLE backup_configs
+    ADD COLUMN max_backup_size_mb        BIGINT  NOT NULL DEFAULT 0,
+    ADD COLUMN max_backups_total_size_mb BIGINT  NOT NULL DEFAULT 0;
+
+ALTER TABLE backups
+    ADD COLUMN is_skip_retry             BOOLEAN NOT NULL DEFAULT FALSE;
+
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+
+ALTER TABLE backup_configs
+    DROP COLUMN IF EXISTS max_backups_total_size_mb,
+    DROP COLUMN IF EXISTS max_backup_size_mb;
+
+ALTER TABLE backups
+    DROP COLUMN IF EXISTS is_skip_retry;
+
+-- +goose StatementEnd
diff --git a/frontend/src/entity/backups/model/BackupConfig.ts b/frontend/src/entity/backups/model/BackupConfig.ts
index 77fb00e..8fcb1f4 100644
--- a/frontend/src/entity/backups/model/BackupConfig.ts
+++ b/frontend/src/entity/backups/model/BackupConfig.ts
@@ -15,4 +15,7 @@ export interface BackupConfig {
   isRetryIfFailed: boolean;
   maxFailedTriesCount: number;
   encryption: BackupEncryption;
+
+  maxBackupSizeMb: number;
+  maxBackupsTotalSizeMb: number;
 }
diff --git a/frontend/src/features/backups/ui/EditBackupConfigComponent.tsx b/frontend/src/features/backups/ui/EditBackupConfigComponent.tsx
index 63dec88..bac77f6 100644
--- a/frontend/src/features/backups/ui/EditBackupConfigComponent.tsx
+++ b/frontend/src/features/backups/ui/EditBackupConfigComponent.tsx
@@ -79,7 +79,10 @@ export const EditBackupConfigComponent = ({
 
   const [isShowWarn, setIsShowWarn] = useState(false);
 
-  const hasAdvancedValues = !!backupConfig?.isRetryIfFailed;
+  const hasAdvancedValues =
+    !!backupConfig?.isRetryIfFailed ||
+    (backupConfig?.maxBackupSizeMb ?? 0) > 0 ||
+    (backupConfig?.maxBackupsTotalSizeMb ?? 0) > 0;
   const [isShowAdvanced, setShowAdvanced] = useState(hasAdvancedValues);
 
   const timeFormat = useMemo(() => {
@@ -162,6 +165,9 @@ export const EditBackupConfigComponent = ({
         isRetryIfFailed: true,
         maxFailedTriesCount: 3,
         encryption: BackupEncryption.ENCRYPTED,
+
+        maxBackupSizeMb: 0,
+        maxBackupsTotalSizeMb: 0,
       });
     }
     loadStorages();
@@ -565,6 +571,89 @@ export const EditBackupConfigComponent = ({
               </div>
             </div>
           )}
+
+          <div className="mt-5 mb-1 flex w-full flex-col items-start sm:flex-row sm:items-center">
+            <div className="mb-1 min-w-[150px] sm:mb-0">Max backup size limit</div>
+            <div className="flex items-center">
+              <Checkbox
+                checked={backupConfig.maxBackupSizeMb > 0}
+                onChange={(e) => {
+                  updateBackupConfig({
+                    maxBackupSizeMb: e.target.checked ? backupConfig.maxBackupSizeMb || 1000 : 0,
+                  });
+                }}
+              >
+                Enable
+              </Checkbox>
+
+              <Tooltip
+                className="cursor-pointer"
+                title="Limits the size of each individual backup. Note that backups are typically 15× smaller than the database size. For example, a 100 MB backup represents approximately 1.5 GB database."
+              >
+                <InfoCircleOutlined className="ml-2" style={{ color: 'gray' }} />
+              </Tooltip>
+            </div>
+          </div>
+
+          {backupConfig.maxBackupSizeMb > 0 && (
+            <div className="mb-1 flex w-full flex-col items-start sm:flex-row sm:items-center">
+              <div className="mb-1 min-w-[150px] sm:mb-0">Max backup size (MB)</div>
+
+              <InputNumber
+                min={1}
+                value={backupConfig.maxBackupSizeMb}
+                onChange={(value) => updateBackupConfig({ maxBackupSizeMb: value || 1 })}
+                size="small"
+                className="w-full max-w-[100px] grow"
+              />
+
+              <div className="ml-2 text-xs text-gray-600 dark:text-gray-400">
+                {(backupConfig.maxBackupSizeMb / 1024).toFixed(2)} GB
+              </div>
+            </div>
+          )}
+
+          <div className="mb-1 flex w-full flex-col items-start sm:flex-row sm:items-center">
+            <div className="mb-1 min-w-[150px] sm:mb-0">Limit total backups size</div>
+            <div className="flex items-center">
+              <Checkbox
+                checked={backupConfig.maxBackupsTotalSizeMb > 0}
+                onChange={(e) => {
+                  updateBackupConfig({
+                    maxBackupsTotalSizeMb: e.target.checked
+                      ? backupConfig.maxBackupsTotalSizeMb || 1_000_000
+                      : 0,
+                  });
+                }}
+              >
+                Enable
+              </Checkbox>
+
+              <Tooltip
+                className="cursor-pointer"
+                title="Limits the total size of all backups in storage. Once this limit is exceeded, the oldest backups are automatically removed until the total size is within the limit again."
+              >
+                <InfoCircleOutlined className="ml-2" style={{ color: 'gray' }} />
+              </Tooltip>
+            </div>
+          </div>
+
+          {backupConfig.maxBackupsTotalSizeMb > 0 && (
+            <div className="mb-1 flex w-full flex-col items-start sm:flex-row sm:items-center">
+              <div className="mb-1 min-w-[150px] sm:mb-0">Max total size (MB)</div>
+              <InputNumber
+                min={1}
+                value={backupConfig.maxBackupsTotalSizeMb}
+                onChange={(value) => updateBackupConfig({ maxBackupsTotalSizeMb: value || 1 })}
+                size="small"
+                className="w-full max-w-[100px] grow"
+              />
+
+              <div className="ml-2 text-xs text-gray-600 dark:text-gray-400">
+                {(backupConfig.maxBackupsTotalSizeMb / 1024).toFixed(2)} GB
+              </div>
+            </div>
+          )}
         </>
       )}