diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index 8969bf4411..3d4ee6b8c4 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -208,7 +208,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { dataManagementHandler := admin.NewDataManagementHandler(dataManagementService) backupObjectStoreFactory := repository.NewS3BackupStoreFactory() dbDumper := repository.NewPgDumper(configConfig) - backupService := service.ProvideBackupService(settingRepository, configConfig, secretEncryptor, backupObjectStoreFactory, dbDumper) + backupService := service.ProvideBackupService(settingRepository, configConfig, secretEncryptor, backupObjectStoreFactory, dbDumper, leaderLockCache, db) imageStorageFactory := repository.ProvideImageStorageFactory() imageStorageSettingService := service.ProvideImageStorageSettingService(settingRepository, secretEncryptor, backupService, imageStorageFactory, configConfig) backupHandler := admin.NewBackupHandler(backupService, userService, imageStorageSettingService) diff --git a/backend/internal/service/backup_service.go b/backend/internal/service/backup_service.go index f130b3653a..aea9aeeb8c 100644 --- a/backend/internal/service/backup_service.go +++ b/backend/internal/service/backup_service.go @@ -4,6 +4,7 @@ import ( "compress/gzip" "context" "crypto/sha256" + "database/sql" "encoding/hex" "encoding/json" "errors" @@ -31,6 +32,21 @@ const ( maxBackupRecords = 100 backupObjectCleanupTimeout = 2 * time.Minute + + // backupScheduledLeaderLockKey gates the scheduled full-database backup so + // that only one instance in a clustered deployment performs the + // dump-and-upload each cycle. Without it every instance runs the cron + // independently, producing N concurrent pg_dumps against the same database, + // N× peak memory while the archive is uploaded, and N identical objects that + // overwrite the same timestamped key. Every other periodic job in this + // package is already gated the same way; the scheduled backup was the last + // one that still fanned out across every instance. + backupScheduledLeaderLockKey = "backup:scheduled:leader" + // backupScheduledLeaderLockTTL bounds crash recovery only; the lock is + // released as soon as the backup finishes. It must exceed the job's + // worst-case runtime (the scheduled backup context is bounded at 30m) so the + // lock cannot expire mid-dump and let a peer start a second backup. + backupScheduledLeaderLockTTL = 35 * time.Minute ) var ( @@ -161,6 +177,14 @@ type BackupService struct { cronSched *cron.Cron cronEntryID cron.EntryID + // lockCache/db elect a single leader for the scheduled backup across + // instances; instanceID identifies this process as the lock owner. Injected + // via SetLeaderLock — when both are nil the backup runs ungated + // (single-instance / test behavior). + lockCache LeaderLockCache + db *sql.DB + instanceID string + wg sync.WaitGroup // 追踪活跃的备份/恢复 goroutine shuttingDown atomic.Bool // 阻止新备份启动 bgCtx context.Context // 所有后台操作的 parent context @@ -186,9 +210,21 @@ func NewBackupService( bgCtx: bgCtx, bgCancel: bgCancel, partSizeBytes: defaultBackupPartSizeBytes, + instanceID: uuid.NewString(), } } +// SetLeaderLock injects the leader-lock cache and DB used to elect a single +// instance for the scheduled backup. When both are nil the scheduled backup runs +// ungated (single-instance / test behavior). +func (s *BackupService) SetLeaderLock(lockCache LeaderLockCache, db *sql.DB) { + if s == nil { + return + } + s.lockCache = lockCache + s.db = db +} + // Start 启动定时备份调度器并清理孤立记录 func (s *BackupService) Start() { s.cronSched = cron.New() @@ -467,6 +503,16 @@ func (s *BackupService) runScheduledBackup() { ctx, cancel := context.WithTimeout(s.bgCtx, 30*time.Minute) defer cancel() + // 多实例保护: 集群部署时只让 leader 执行定时备份, 避免每个实例各自对同一个 + // 数据库跑一次全量 dump、上传时峰值内存翻倍、以及多份同名对象互相覆盖。 + // 手动触发的备份 (CreateBackup/StartBackup) 不受此限, 运维仍可随时在任一节点强制备份。 + release, ok := tryAcquireSingletonLeaderLock(ctx, s.lockCache, s.db, backupScheduledLeaderLockKey, s.instanceID, backupScheduledLeaderLockTTL) + if !ok { + logger.LegacyPrintf("service.backup", "[Backup] 定时备份跳过: 本实例非 leader") + return + } + defer release() + // 读取定时备份配置中的过期天数 schedule, _ := s.GetSchedule(ctx) expireDays := 14 // 默认14天过期 diff --git a/backend/internal/service/backup_service_test.go b/backend/internal/service/backup_service_test.go index c599d719f4..876e8243d4 100644 --- a/backend/internal/service/backup_service_test.go +++ b/backend/internal/service/backup_service_test.go @@ -629,6 +629,56 @@ func TestBackupService_CreateBackup_ConcurrentBlocked(t *testing.T) { require.ErrorIs(t, err, ErrBackupInProgress) } +// TestBackupService_RunScheduledBackup_LeaderElection verifies the scheduled +// backup is gated by a cross-instance leader lock: a non-leader instance skips +// the dump entirely so a clustered deployment does not run N identical backups +// against the same database, while the leader runs it and releases the lock +// afterward. Manual backups (CreateBackup/StartBackup) are intentionally left +// ungated and are covered by the other tests. +func TestBackupService_RunScheduledBackup_LeaderElection(t *testing.T) { + t.Run("non-leader skips", func(t *testing.T) { + repo := newMockSettingRepo() + seedS3Config(t, repo) + store := newMockObjectStore() + svc := newTestBackupService(repo, &mockDumper{dumpData: []byte("data")}, store) + + // A peer already owns the lock, so this instance is not the leader. + cache := &fakeLeaderLockCache{} + peerRelease, ok := tryAcquireSingletonLeaderLock(context.Background(), cache, nil, backupScheduledLeaderLockKey, "peer", time.Minute) + require.True(t, ok) + defer peerRelease() + + svc.SetLeaderLock(cache, nil) + svc.runScheduledBackup() + + store.mu.Lock() + require.Empty(t, store.objects, "non-leader must not upload a backup") + store.mu.Unlock() + + records, err := svc.ListBackups(context.Background()) + require.NoError(t, err) + require.Empty(t, records, "non-leader must not create a backup record") + require.Equal(t, "peer", cache.heldBy(backupScheduledLeaderLockKey), "peer keeps the lock") + }) + + t.Run("leader runs and releases", func(t *testing.T) { + repo := newMockSettingRepo() + seedS3Config(t, repo) + store := newMockObjectStore() + svc := newTestBackupService(repo, &mockDumper{dumpData: []byte("-- dump\n")}, store) + + cache := &fakeLeaderLockCache{} + svc.SetLeaderLock(cache, nil) + svc.runScheduledBackup() + + records, err := svc.ListBackups(context.Background()) + require.NoError(t, err) + require.Len(t, records, 1, "leader creates exactly one backup record") + require.Equal(t, "completed", records[0].Status) + require.Empty(t, cache.heldBy(backupScheduledLeaderLockKey), "leader releases the lock when done") + }) +} + func TestBackupService_RestoreBackup_Streaming(t *testing.T) { repo := newMockSettingRepo() seedS3Config(t, repo) diff --git a/backend/internal/service/wire.go b/backend/internal/service/wire.go index 93555730a7..1fe46f6740 100644 --- a/backend/internal/service/wire.go +++ b/backend/internal/service/wire.go @@ -625,8 +625,11 @@ func ProvideBackupService( encryptor SecretEncryptor, storeFactory BackupObjectStoreFactory, dumper DBDumper, + lockCache LeaderLockCache, + db *sql.DB, ) *BackupService { svc := NewBackupService(settingRepo, cfg, encryptor, storeFactory, dumper) + svc.SetLeaderLock(lockCache, db) svc.Start() return svc }