feat(backup): 支持月度归档与独立保留策略

backend/internal/handler/admin/backup_handler.go

@@ -140,7 +140,13 @@ func (h *BackupHandler) DeleteBackup(c *gin.Context) {
 		response.BadRequest(c, "backup ID is required")
 		return
 	}
-	if err := h.backupService.DeleteBackup(c.Request.Context(), backupID); err != nil {
+	var err error
+	if c.Query("delete_archived") == "true" {
+		err = h.backupService.DeleteArchivedBackup(c.Request.Context(), backupID)
+	} else {
+		err = h.backupService.DeleteBackup(c.Request.Context(), backupID)
+	}
+	if err != nil {
 		response.ErrorFrom(c, err)
 		return
 	}

backend/internal/service/backup_recovery_test.go

@@ -0,0 +1,222 @@
+//go:build unit
+
+package service
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+type backupDelayedReadRepo struct {
+	*mockSettingRepo
+	fired  atomic.Bool
+	loaded chan struct{}
+	resume chan struct{}
+}
+
+func (r *backupDelayedReadRepo) GetValue(ctx context.Context, key string) (string, error) {
+	value, err := r.mockSettingRepo.GetValue(ctx, key)
+	if key == settingKeyBackupRecords && r.fired.CompareAndSwap(false, true) {
+		close(r.loaded)
+		<-r.resume
+	}
+	return value, err
+}
+func TestBackupRecovery_RestartPreservesPeerArchive(t *testing.T) {
+	ctx := context.Background()
+	repo := newMockSettingRepo()
+	seedS3Config(t, repo)
+	store := newMockObjectStore()
+	writer := newTestBackupService(repo, &mockDumper{}, store)
+	started := time.Now().UTC()
+	record := archiveRecord("now-permanent", started.Format(time.RFC3339))
+	record.Status = "running"
+	record.S3Key = "now-permanent.sql.gz"
+	store.objects[record.S3Key] = []byte("important archive")
+	require.NoError(t, writer.saveRecord(ctx, record))
+
+	restarting := newTestBackupService(repo, &mockDumper{}, store)
+	delayedRepo := &backupDelayedReadRepo{mockSettingRepo: repo, loaded: make(chan struct{}), resume: make(chan struct{})}
+	restarting.settingRepo = delayedRepo
+	cache := &fakeLeaderLockCache{}
+	writer.SetLeaderLock(cache, nil)
+	restarting.SetLeaderLock(cache, nil)
+	done := make(chan struct{})
+	go func() { restarting.recoverStaleRecords(); close(done) }()
+	<-delayedRepo.loaded
+	record.Status = "completed"
+	// Recovery must own the writer lock across its read and decision. The peer
+	// completes only after recovery releases it; the young operation is active.
+	require.Equal(t, restarting.instanceID, cache.heldBy("backup:records:writer"))
+	written := make(chan error, 1)
+	go func() { written <- writer.saveRecordWithArchive(ctx, record, archiveSchedule(started.Day())) }()
+	close(delayedRepo.resume)
+	<-done
+	require.NoError(t, <-written)
+	require.NotNil(t, record.MonthlyArchive)
+	restarting.recoverStaleRecords()
+	saved, err := writer.GetBackupRecord(ctx, record.ID)
+	require.NoError(t, err)
+	require.NotNil(t, saved.MonthlyArchive)
+	require.Equal(t, "completed", saved.Status, "completed permanent archive was overwritten by stale restart snapshot")
+	require.Contains(t, store.objects, record.S3Key, "restart must not delete completed permanent archive")
+}
+
+func TestBackupRecovery_RestoreRejectsRemovedArchive(t *testing.T) {
+	ctx := context.Background()
+	repo := newMockSettingRepo()
+	seedS3Config(t, repo)
+	store := newMockObjectStore()
+	cleaner := newTestBackupService(repo, &mockDumper{}, store)
+	for i, id := range []string{"old", "new"} {
+		record := archiveRecord(id, []string{"2026-09-01T04:00:00Z", "2026-09-15T04:00:00Z"}[i])
+		record.MonthlyArchive = &BackupMonthlyArchive{RetainCount: 1}
+		record.S3Key = id
+		store.objects[id] = []byte("payload")
+		require.NoError(t, cleaner.saveRecord(ctx, record))
+	}
+	restoring := newTestBackupService(repo, &mockDumper{}, store)
+	delayedRepo := &backupDelayedReadRepo{mockSettingRepo: repo, loaded: make(chan struct{}), resume: make(chan struct{})}
+	restoring.settingRepo = delayedRepo
+	cache := &fakeLeaderLockCache{}
+	cleaner.SetLeaderLock(cache, nil)
+	restoring.SetLeaderLock(cache, nil)
+	result := make(chan error)
+	go func() { _, err := restoring.StartRestore(ctx, "old"); result <- err }()
+	<-delayedRepo.loaded
+	cfg := archiveSchedule(1, 15)
+	cfg.MonthlyArchive.RetainCount = 1
+	require.NoError(t, cleaner.cleanupOldBackups(ctx, cfg))
+	require.NotContains(t, store.objects, "old")
+	close(delayedRepo.resume)
+	err := <-result
+	restoring.wg.Wait()
+	saved, getErr := cleaner.GetBackupRecord(ctx, "old")
+	require.Nil(t, saved)
+	require.ErrorIs(t, err, ErrBackupNotFound, "restore must not start against a deleted archive")
+	require.ErrorIs(t, getErr, ErrBackupNotFound, "cleanup-deleted archive must not be resurrected")
+}
+
+type backupFailRecordWriteRepo struct{ *mockSettingRepo }
+
+func (r *backupFailRecordWriteRepo) Set(ctx context.Context, key, value string) error {
+	if key == settingKeyBackupRecords {
+		return context.DeadlineExceeded
+	}
+	return r.mockSettingRepo.Set(ctx, key, value)
+}
+func TestBackupRecovery_RestoreRequiresDurableProtection(t *testing.T) {
+	ctx := context.Background()
+	repo := newMockSettingRepo()
+	seedS3Config(t, repo)
+	store := newMockObjectStore()
+	svc := newTestBackupService(repo, &mockDumper{}, store)
+	record := archiveRecord("finite-archive", "2026-09-01T04:00:00Z")
+	record.MonthlyArchive = &BackupMonthlyArchive{RetainCount: 1}
+	record.S3Key = record.ID
+	require.NoError(t, svc.saveRecord(ctx, record))
+	svc.settingRepo = &backupFailRecordWriteRepo{repo}
+	_, err := svc.StartRestore(ctx, record.ID)
+	svc.wg.Wait()
+	require.Error(t, err, "failed restore-running marker must prevent restore from starting")
+}
+
+func TestBackupRecovery_ActiveAndLegacyRestoresKeepProtection(t *testing.T) {
+	ctx := context.Background()
+	svc := newTestBackupService(newMockSettingRepo(), &mockDumper{}, newMockObjectStore())
+	for _, id := range []string{"active", "legacy"} {
+		record := archiveRecord(id, time.Now().AddDate(-1, 0, 0).Format(time.RFC3339))
+		record.MonthlyArchive = &BackupMonthlyArchive{RetainCount: 0}
+		record.RestoreStatus = "running"
+		if id == "active" {
+			record.RestoreStartedAt = time.Now().Format(time.RFC3339)
+		}
+		require.NoError(t, svc.saveRecord(ctx, record))
+	}
+	svc.recoverStaleRecords()
+	for _, id := range []string{"active", "legacy"} {
+		record, err := svc.GetBackupRecord(ctx, id)
+		require.NoError(t, err)
+		require.Equal(t, "running", record.RestoreStatus)
+		require.NotEmpty(t, record.RestoreStartedAt)
+		require.ErrorIs(t, svc.DeleteArchivedBackup(ctx, id), ErrRestoreInProgress)
+	}
+}
+
+func TestBackupRecovery_FailedStateWriteNeverDeletesObjects(t *testing.T) {
+	ctx := context.Background()
+	repo := newMockSettingRepo()
+	seedS3Config(t, repo)
+	store := newMockObjectStore()
+	svc := newTestBackupService(repo, &mockDumper{}, store)
+	record := archiveRecord("interrupted", time.Now().Add(-time.Hour).Format(time.RFC3339))
+	record.Status, record.S3Key = "running", "interrupted.sql.gz"
+	store.objects[record.S3Key] = []byte("payload")
+	require.NoError(t, svc.saveRecord(ctx, record))
+	svc.settingRepo = &backupFailRecordWriteRepo{repo}
+	svc.recoverStaleRecords()
+	saved, err := svc.GetBackupRecord(ctx, record.ID)
+	require.NoError(t, err)
+	require.Equal(t, "running", saved.Status)
+	require.Contains(t, store.objects, record.S3Key)
+	require.Empty(t, store.deletedKeys)
+}
+
+func TestBackupRecovery_RestoreReservationProtectsFromPeerCleanup(t *testing.T) {
+	ctx := context.Background()
+	repo := newMockSettingRepo()
+	seedS3Config(t, repo)
+	store := newMockObjectStore()
+	svc := newTestBackupService(repo, &mockDumper{}, store)
+	peer := newTestBackupService(repo, &mockDumper{}, store)
+	cache := &fakeLeaderLockCache{}
+	svc.SetLeaderLock(cache, nil)
+	peer.SetLeaderLock(cache, nil)
+	record := archiveRecord("old", time.Now().AddDate(-1, 0, 0).Format(time.RFC3339))
+	record.S3Key = record.ID
+	store.objects[record.ID] = []byte("payload")
+	require.NoError(t, svc.saveRecord(ctx, record))
+	restoring, err := svc.beginBackupRestore(ctx, record.ID)
+	require.NoError(t, err)
+	require.Equal(t, "running", restoring.RestoreStatus)
+	_, err = peer.beginBackupRestore(ctx, record.ID)
+	require.ErrorIs(t, err, ErrRestoreInProgress)
+	require.NoError(t, peer.cleanupOldBackups(ctx, archiveSchedule(1)))
+	require.Contains(t, store.objects, record.ID)
+	require.ErrorIs(t, peer.DeleteBackup(ctx, record.ID), ErrRestoreInProgress)
+}
+
+func TestBackupRecovery_FailedRestoreCannotRecreateMissingRecord(t *testing.T) {
+	ctx := context.Background()
+	svc := newTestBackupService(newMockSettingRepo(), &mockDumper{}, newMockObjectStore())
+	record := archiveRecord("removed", time.Now().Format(time.RFC3339))
+	record.RestoreStatus = "failed"
+	require.ErrorIs(t, svc.saveRestoreRecord(ctx, record), ErrBackupNotFound)
+	_, err := svc.GetBackupRecord(ctx, record.ID)
+	require.ErrorIs(t, err, ErrBackupNotFound)
+}
+
+func TestBackupRecovery_LateRestoreResultCannotReleaseNewProtection(t *testing.T) {
+	ctx := context.Background()
+	svc := newTestBackupService(newMockSettingRepo(), &mockDumper{}, newMockObjectStore())
+	record := archiveRecord("restoring", time.Now().AddDate(-1, 0, 0).Format(time.RFC3339))
+	record.RestoreStatus = "running"
+	record.RestoreStartedAt = time.Now().Add(-time.Hour).Format(time.RFC3339)
+	require.NoError(t, svc.saveRecord(ctx, record))
+	svc.recoverStaleRecords()
+	current, err := svc.beginBackupRestore(ctx, record.ID)
+	require.NoError(t, err)
+	for _, status := range []string{"failed", "completed"} {
+		record.RestoreStatus = status
+		require.Error(t, svc.saveRestoreRecord(ctx, record))
+		saved, err := svc.GetBackupRecord(ctx, record.ID)
+		require.NoError(t, err)
+		require.Equal(t, "running", saved.RestoreStatus)
+		require.Equal(t, current.RestoreStartedAt, saved.RestoreStartedAt)
+		require.ErrorIs(t, svc.DeleteBackup(ctx, record.ID), ErrRestoreInProgress)
+	}
+}

backend/internal/service/backup_restore_state.go

@@ -0,0 +1,85 @@
+package service
+
+import (
+	"context"
+	"time"
+
+	infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
+)
+
+func (s *BackupService) beginBackupRestore(ctx context.Context, id string) (*BackupRecord, error) {
+	ctx, release, err := s.lockBackupRecordUpdates(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer release()
+	records, err := s.loadRecordsLocked(ctx)
+	if err != nil {
+		return nil, err
+	}
+	for i := range records {
+		if records[i].ID != id {
+			continue
+		}
+		if records[i].Status != "completed" {
+			return nil, infraerrors.BadRequest("BACKUP_NOT_COMPLETED", "can only restore from a completed backup")
+		}
+		if records[i].RestoreStatus == "running" {
+			return nil, ErrRestoreInProgress
+		}
+		records[i].RestoreStatus = "running"
+		records[i].RestoreStartedAt = time.Now().Format(time.RFC3339)
+		records[i].RestoreError = ""
+		if err := s.saveRecordsLocked(ctx, records); err != nil {
+			return nil, err
+		}
+		return &records[i], nil
+	}
+	return nil, ErrBackupNotFound
+}
+
+// Failed/panicked restores must not resurrect a removed record. A successful
+// full database restore may itself roll back settings and remove its own record,
+// so that completion keeps the existing explicit re-registration behavior.
+func (s *BackupService) saveRestoreRecord(ctx context.Context, record *BackupRecord) error {
+	ctx, release, err := s.lockBackupRecordUpdates(ctx)
+	if err != nil {
+		return err
+	}
+	defer release()
+	records, err := s.loadRecordsLocked(ctx)
+	if err != nil {
+		return err
+	}
+	for i := range records {
+		if records[i].ID != record.ID {
+			continue
+		}
+		if records[i].RestoreStartedAt != record.RestoreStartedAt &&
+			(record.RestoreStatus != "completed" || (records[i].RestoreStatus == "running" && records[i].RestoreStartedAt != "")) {
+			return infraerrors.Conflict("BACKUP_RESTORE_STATE_CHANGED", "a newer restore operation owns this backup")
+		}
+		if record.RestoreStatus == "completed" {
+			// Restoring the full database may roll the backup record itself back
+			// to its pre-upload state. Re-register the completed snapshot while
+			// retaining any archive protection already persisted in the database.
+			snapshot := *record
+			if records[i].MonthlyArchive != nil {
+				snapshot.MonthlyArchive = records[i].MonthlyArchive
+				snapshot.ExpiresAt = ""
+			}
+			records[i] = snapshot
+			return s.saveRecordsLocked(ctx, records)
+		}
+		// Only the restore operation's fields change, preserving current archive
+		// metadata and any unrelated progress recorded by another writer.
+		records[i].RestoreStatus = record.RestoreStatus
+		records[i].RestoreError = record.RestoreError
+		records[i].RestoredAt = record.RestoredAt
+		return s.saveRecordsLocked(ctx, records)
+	}
+	if record.RestoreStatus == "completed" {
+		return s.saveRecordsLocked(ctx, append(records, *record))
+	}
+	return ErrBackupNotFound
+}

backend/internal/service/backup_retention.go

@@ -0,0 +1,193 @@
+package service
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+	"time"
+
+	infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
+	"github.com/robfig/cron/v3"
+)
+
+const settingKeyBackupArchiveCheckpoint = "backup_monthly_archive_checkpoint"
+
+// Records and the archive checkpoint share a read/modify/write boundary across
+// instances, including manual backups and restore status updates. Use the same
+// backend consistently; a Redis error must not silently switch to a DB lock
+// while another instance still owns the Redis lock.
+func (s *BackupService) lockBackupRecordUpdates(ctx context.Context) (context.Context, func(), error) {
+	s.recordsMu.Lock()
+	writeCtx, cancel := context.WithTimeout(ctx, 30*time.Minute)
+	waitCtx, stopWaiting := context.WithTimeout(writeCtx, 10*time.Second)
+	defer stopWaiting()
+	const key = "backup:records:writer"
+	for {
+		var release func()
+		var ok bool
+		if s.lockCache != nil {
+			var err error
+			ok, err = s.lockCache.TryAcquireLeaderLock(waitCtx, key, s.instanceID, backupScheduledLeaderLockTTL)
+			if err != nil {
+				cancel()
+				s.recordsMu.Unlock()
+				return nil, nil, fmt.Errorf("lock backup records: %w", err)
+			}
+			release = func() {
+				ctx, done := context.WithTimeout(context.Background(), 2*time.Second)
+				defer done()
+				_ = s.lockCache.ReleaseLeaderLock(ctx, key, s.instanceID)
+			}
+		} else if s.db != nil {
+			// A session advisory lock holds a connection while the repository
+			// needs another. Redis avoids this requirement in production.
+			if s.db.Stats().MaxOpenConnections == 1 {
+				cancel()
+				s.recordsMu.Unlock()
+				return nil, nil, fmt.Errorf("backup record locking requires Redis or at least two database connections")
+			}
+			var err error
+			release, ok, err = tryAcquireDBAdvisoryLockWithError(waitCtx, s.db, hashAdvisoryLockID(key))
+			if err != nil {
+				cancel()
+				s.recordsMu.Unlock()
+				return nil, nil, fmt.Errorf("lock backup records: %w", err)
+			}
+		} else {
+			release, ok = func() {}, true
+		}
+		if ok {
+			return writeCtx, func() { release(); cancel(); s.recordsMu.Unlock() }, nil
+		}
+		select {
+		case <-waitCtx.Done():
+			cancel()
+			s.recordsMu.Unlock()
+			return nil, nil, fmt.Errorf("lock backup records: %w", waitCtx.Err())
+		case <-time.After(25 * time.Millisecond):
+		}
+	}
+}
+
+type BackupMonthlyArchiveConfig struct {
+	Enabled         bool  `json:"enabled"`
+	Days            []int `json:"days"`
+	IncludeMonthEnd bool  `json:"include_month_end"`
+	RetainCount     int   `json:"retain_count"` // 0 = permanent; positive = finite archive pool size.
+}
+
+type BackupMonthlyArchive struct {
+	Dates       []string `json:"dates"`        // Deduplicated target dates covered by this backup.
+	RetainCount int      `json:"retain_count"` // 0 is permanent and never implicitly downgraded.
+}
+
+func validateBackupRetention(cfg *BackupScheduleConfig) error {
+	if cfg.RetainDays < 0 || cfg.RetainCount < 0 {
+		return infraerrors.BadRequest("INVALID_BACKUP_RETENTION", "backup retention must not be negative")
+	}
+	archive := cfg.MonthlyArchive
+	if archive == nil {
+		return nil
+	}
+	if archive.RetainCount < 0 || len(archive.Days) > 31 {
+		return infraerrors.BadRequest("INVALID_MONTHLY_ARCHIVE", "invalid archive retention or dates")
+	}
+	seen := make(map[int]bool)
+	days := make([]int, 0, len(archive.Days))
+	for _, day := range archive.Days {
+		if day < 1 || day > 31 {
+			return infraerrors.BadRequest("INVALID_MONTHLY_ARCHIVE", "archive dates must be between 1 and 31")
+		}
+		if !seen[day] {
+			days = append(days, day)
+			seen[day] = true
+		}
+	}
+	sort.Ints(days)
+	archive.Days = days
+	if archive.Enabled && len(days) == 0 && !archive.IncludeMonthEnd {
+		return infraerrors.BadRequest("INVALID_MONTHLY_ARCHIVE", "select at least one archive date")
+	}
+	return nil
+}
+
+// assignMonthlyArchive runs under recordsMu and only for a completed scheduled
+// backup. Successful processing advances a durable date checkpoint even when
+// no date is due. Removing a record or reducing retention cannot reset it.
+func (s *BackupService) assignMonthlyArchive(ctx context.Context, record *BackupRecord, schedule *BackupScheduleConfig) (string, error) {
+	if record.Status != "completed" || record.TriggeredBy != "scheduled" || record.MonthlyArchive != nil || schedule == nil || schedule.MonthlyArchive == nil || !schedule.MonthlyArchive.Enabled {
+		return "", nil
+	}
+	if err := validateBackupRetention(schedule); err != nil {
+		return "", err
+	}
+	startedAt, err := time.Parse(time.RFC3339, record.StartedAt)
+	if err != nil {
+		return "", fmt.Errorf("parse archive backup start: %w", err)
+	}
+	parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
+	spec, err := parser.Parse(schedule.CronExpr)
+	if err != nil {
+		return "", fmt.Errorf("parse archive schedule: %w", err)
+	}
+	location := time.Local
+	if parsed, ok := spec.(*cron.SpecSchedule); ok {
+		location = parsed.Location
+	}
+	startedAt = startedAt.In(location)
+	today := startedAt.Format(time.DateOnly)
+	checkpoint, err := s.settingRepo.GetValue(ctx, settingKeyBackupArchiveCheckpoint)
+	if err != nil && !errors.Is(err, ErrSettingNotFound) {
+		return "", fmt.Errorf("load archive checkpoint: %w", err)
+	}
+	if checkpoint != "" {
+		if _, err := time.Parse(time.DateOnly, checkpoint); err != nil {
+			return "", infraerrors.InternalServer("BACKUP_ARCHIVE_CHECKPOINT_CORRUPT", "archive checkpoint data is corrupted")
+		}
+		if checkpoint >= today {
+			return "", nil
+		}
+	}
+	dates := monthlyArchiveDates(startedAt, schedule.MonthlyArchive)
+	var due []string
+	for _, date := range dates {
+		if date > checkpoint && date <= today {
+			due = append(due, date)
+		}
+	}
+	if len(due) > 0 {
+		record.MonthlyArchive = &BackupMonthlyArchive{Dates: due, RetainCount: schedule.MonthlyArchive.RetainCount}
+		record.ExpiresAt = ""
+	}
+	return today, nil
+}
+
+func monthlyArchiveDates(at time.Time, cfg *BackupMonthlyArchiveConfig) []string {
+	lastDay := time.Date(at.Year(), at.Month()+1, 0, 0, 0, 0, 0, at.Location()).Day()
+	unique := make(map[int]bool)
+	for _, day := range cfg.Days {
+		unique[min(day, lastDay)] = true
+	}
+	if cfg.IncludeMonthEnd {
+		unique[lastDay] = true
+	}
+	dates := make([]string, 0, len(unique))
+	for day := range unique {
+		dates = append(dates, time.Date(at.Year(), at.Month(), day, 0, 0, 0, 0, at.Location()).Format(time.DateOnly))
+	}
+	sort.Strings(dates)
+	return dates
+}
+
+func backupStartedAfter(a, b BackupRecord) bool {
+	left, leftErr := time.Parse(time.RFC3339, a.StartedAt)
+	right, rightErr := time.Parse(time.RFC3339, b.StartedAt)
+	if leftErr == nil && rightErr == nil && !left.Equal(right) {
+		return left.After(right)
+	}
+	if a.StartedAt != b.StartedAt {
+		return a.StartedAt > b.StartedAt
+	}
+	return a.ID > b.ID
+}

backend/internal/service/backup_retention_test.go

@@ -0,0 +1,296 @@
+//go:build unit
+
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/require"
+)
+
+func archiveSchedule(days ...int) *BackupScheduleConfig {
+	return &BackupScheduleConfig{CronExpr: "CRON_TZ=UTC 0 4 * * *", RetainDays: 14, RetainCount: 10,
+		MonthlyArchive: &BackupMonthlyArchiveConfig{Enabled: true, Days: days}}
+}
+
+func archiveRecord(id, started string) *BackupRecord {
+	return &BackupRecord{ID: id, Status: "completed", TriggeredBy: "scheduled", StartedAt: started, ExpiresAt: "2099-01-01T00:00:00Z"}
+}
+
+func TestBackupRetention_ValidationAndRoundTrip(t *testing.T) {
+	for _, cfg := range []BackupScheduleConfig{
+		{RetainDays: -1}, {RetainCount: -1},
+		{MonthlyArchive: &BackupMonthlyArchiveConfig{Enabled: true}},
+		{MonthlyArchive: &BackupMonthlyArchiveConfig{Days: []int{0}}},
+		{MonthlyArchive: &BackupMonthlyArchiveConfig{Days: []int{32}}},
+		{MonthlyArchive: &BackupMonthlyArchiveConfig{RetainCount: -1}},
+	} {
+		require.Error(t, validateBackupRetention(&cfg))
+	}
+	repo := newMockSettingRepo()
+	svc := newTestBackupService(repo, &mockDumper{}, newMockObjectStore())
+	cfg := archiveSchedule(15, 1, 15)
+	cfg.RetainDays, cfg.RetainCount = 0, 0
+	cfg.MonthlyArchive.IncludeMonthEnd = true
+	cfg.MonthlyArchive.RetainCount = 17
+	saved, err := svc.UpdateSchedule(context.Background(), *cfg)
+	require.NoError(t, err)
+	require.Equal(t, []int{1, 15}, saved.MonthlyArchive.Days)
+	loaded, err := svc.GetSchedule(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, saved, loaded)
+	require.Zero(t, loaded.RetainDays)
+	require.Zero(t, loaded.RetainCount)
+}
+
+func TestBackupRetention_MonthEndAndLeapYears(t *testing.T) {
+	for _, tc := range []struct{ date, end string }{
+		{"2026-02-01", "2026-02-28"}, {"2028-02-01", "2028-02-29"}, {"2026-04-01", "2026-04-30"},
+	} {
+		t.Run(tc.date, func(t *testing.T) {
+			at, err := time.Parse(time.DateOnly, tc.date)
+			require.NoError(t, err)
+			cfg := &BackupMonthlyArchiveConfig{Days: []int{30, 31}, IncludeMonthEnd: true}
+			require.Equal(t, []string{tc.end}, monthlyArchiveDates(at, cfg))
+		})
+	}
+}
+
+func TestBackupRetention_SuccessOnlyFallbackAndMultipleDates(t *testing.T) {
+	repo := newMockSettingRepo()
+	svc := newTestBackupService(repo, &mockDumper{}, newMockObjectStore())
+	cfg := archiveSchedule(1, 15)
+	ctx := context.Background()
+	failed := archiveRecord("failed", "2026-09-01T04:00:00Z")
+	failed.Status = "failed"
+	require.NoError(t, svc.saveRecordWithArchive(ctx, failed, cfg))
+	manual := archiveRecord("manual", "2026-09-01T05:00:00Z")
+	manual.TriggeredBy = "manual"
+	require.NoError(t, svc.saveRecordWithArchive(ctx, manual, cfg))
+	require.Nil(t, failed.MonthlyArchive)
+	require.Nil(t, manual.MonthlyArchive)
+	first := archiveRecord("first", "2026-09-02T04:00:00Z")
+	require.NoError(t, svc.saveRecordWithArchive(ctx, first, cfg))
+	require.Equal(t, []string{"2026-09-01"}, first.MonthlyArchive.Dates)
+	require.Empty(t, first.ExpiresAt)
+	again := archiveRecord("again", "2026-09-02T06:00:00Z")
+	require.NoError(t, svc.saveRecordWithArchive(ctx, again, cfg))
+	require.Nil(t, again.MonthlyArchive)
+	next := archiveRecord("next", "2026-09-16T04:00:00Z")
+	require.NoError(t, svc.saveRecordWithArchive(ctx, next, cfg))
+	require.Equal(t, []string{"2026-09-15"}, next.MonthlyArchive.Dates)
+	// A later month does not manufacture archives for the missed previous month.
+	newMonth := archiveRecord("oct", "2026-10-20T04:00:00Z")
+	require.NoError(t, svc.saveRecordWithArchive(ctx, newMonth, cfg))
+	require.Equal(t, []string{"2026-10-01", "2026-10-15"}, newMonth.MonthlyArchive.Dates)
+}
+
+func TestBackupRetention_ScheduleTimezoneAndStartDate(t *testing.T) {
+	repo := newMockSettingRepo()
+	svc := newTestBackupService(repo, &mockDumper{}, newMockObjectStore())
+	cfg := archiveSchedule(1, 31)
+	cfg.CronExpr = "CRON_TZ=Asia/Tokyo 0 4 * * *"
+	first := archiveRecord("tokyo", "2026-08-31T16:00:00Z")
+	require.NoError(t, svc.saveRecordWithArchive(context.Background(), first, cfg))
+	require.Equal(t, []string{"2026-09-01"}, first.MonthlyArchive.Dates)
+	last := archiveRecord("cross-midnight", "2026-09-30T14:59:00Z")
+	last.FinishedAt = "2026-09-30T15:10:00Z"
+	require.NoError(t, svc.saveRecordWithArchive(context.Background(), last, cfg))
+	require.Equal(t, []string{"2026-09-30"}, last.MonthlyArchive.Dates)
+}
+
+func TestBackupRetention_CheckpointSurvivesRecordRemovalAndRestart(t *testing.T) {
+	repo := newMockSettingRepo()
+	seedS3Config(t, repo)
+	store := newMockObjectStore()
+	svc := newTestBackupService(repo, &mockDumper{}, store)
+	cfg := archiveSchedule(1, 15)
+	ctx := context.Background()
+	record := archiveRecord("archive", "2026-09-16T04:00:00Z")
+	require.NoError(t, svc.saveRecordWithArchive(ctx, record, cfg))
+	require.ErrorIs(t, svc.DeleteBackup(ctx, record.ID), ErrBackupArchiveProtected)
+	require.NoError(t, svc.DeleteArchivedBackup(ctx, record.ID))
+	svc = newTestBackupService(repo, &mockDumper{}, store)
+	retry := archiveRecord("retry", "2026-09-17T04:00:00Z")
+	require.NoError(t, svc.saveRecordWithArchive(ctx, retry, cfg))
+	require.Nil(t, retry.MonthlyArchive)
+}
+
+func TestBackupRetention_CleanupSeparatePoolsAndPermanentProtection(t *testing.T) {
+	repo := newMockSettingRepo()
+	seedS3Config(t, repo)
+	store := newMockObjectStore()
+	svc := newTestBackupService(repo, &mockDumper{}, store)
+	ctx := context.Background()
+	now := time.Now().UTC()
+	for i, id := range []string{"running", "failed", "ordinary-new", "archive-new", "permanent", "ordinary-old", "archive-old", "restoring"} {
+		r := &BackupRecord{ID: id, Status: "completed", StartedAt: now.Add(-time.Duration(i) * time.Hour).Format(time.RFC3339), S3Key: id}
+		switch id {
+		case "running", "failed":
+			r.Status = id
+		case "archive-new", "archive-old":
+			r.MonthlyArchive = &BackupMonthlyArchive{RetainCount: 12}
+		case "permanent":
+			r.MonthlyArchive = &BackupMonthlyArchive{}
+		case "restoring":
+			r.RestoreStatus = "running"
+		}
+		store.objects[id] = []byte(id)
+		require.NoError(t, svc.saveRecord(ctx, r))
+	}
+	cfg := &BackupScheduleConfig{RetainDays: 1, RetainCount: 1, MonthlyArchive: &BackupMonthlyArchiveConfig{RetainCount: 1}}
+	require.NoError(t, svc.cleanupOldBackups(ctx, cfg))
+	for _, id := range []string{"running", "failed", "ordinary-new", "archive-new", "permanent", "restoring"} {
+		_, err := svc.GetBackupRecord(ctx, id)
+		require.NoError(t, err, id)
+	}
+	for _, id := range []string{"ordinary-old", "archive-old"} {
+		_, err := svc.GetBackupRecord(ctx, id)
+		require.ErrorIs(t, err, ErrBackupNotFound)
+		require.NotContains(t, store.objects, id)
+	}
+	finite, err := svc.GetBackupRecord(ctx, "archive-new")
+	require.NoError(t, err)
+	require.Equal(t, 1, finite.MonthlyArchive.RetainCount)
+	// Switching the rule off or to permanent does not downgrade prior permanent archives.
+	cfg.MonthlyArchive = &BackupMonthlyArchiveConfig{}
+	cfg.RetainDays = 0
+	require.NoError(t, svc.cleanupOldBackups(ctx, cfg))
+	_, err = svc.GetBackupRecord(ctx, "permanent")
+	require.NoError(t, err)
+}
+
+func TestBackupRetention_Over100ArchivesRemainDiscoverable(t *testing.T) {
+	repo := newMockSettingRepo()
+	svc := newTestBackupService(repo, &mockDumper{}, newMockObjectStore())
+	ctx := context.Background()
+	for i := range 105 {
+		record := archiveRecord(fmt.Sprint(i), "2020-01-01T00:00:00Z")
+		record.MonthlyArchive = &BackupMonthlyArchive{}
+		require.NoError(t, svc.saveRecord(ctx, record))
+	}
+	require.NoError(t, svc.cleanupOldBackups(ctx, &BackupScheduleConfig{RetainDays: 1, RetainCount: 1}))
+	records, err := svc.ListBackups(ctx)
+	require.NoError(t, err)
+	require.Len(t, records, 105)
+}
+
+type failingArchiveRepo struct {
+	*mockSettingRepo
+	fail bool
+}
+
+func (r *failingArchiveRepo) SetMultiple(ctx context.Context, values map[string]string) error {
+	if r.fail {
+		return errors.New("injected atomic save failure")
+	}
+	return r.mockSettingRepo.SetMultiple(ctx, values)
+}
+
+func TestBackupRetention_PersistenceFailuresFailClosed(t *testing.T) {
+	repo := newMockSettingRepo()
+	svc := newTestBackupService(repo, &mockDumper{}, newMockObjectStore())
+	ctx := context.Background()
+	failing := &failingArchiveRepo{mockSettingRepo: repo, fail: true}
+	svc.settingRepo = failing
+	record := archiveRecord("new", "2026-09-01T04:00:00Z")
+	cfg := archiveSchedule(1)
+	require.Error(t, svc.saveRecordWithArchive(ctx, record, cfg))
+	require.Nil(t, record.MonthlyArchive, "failed commit must not mutate the caller's retry state")
+	require.Empty(t, repo.data[settingKeyBackupArchiveCheckpoint])
+	failing.fail = false
+	require.NoError(t, svc.saveRecordWithArchive(ctx, record, cfg))
+	require.NotNil(t, record.MonthlyArchive)
+	oldRecords := repo.data[settingKeyBackupRecords]
+	repo.getValueErr = errors.New("read failed")
+	require.Error(t, svc.saveRecord(ctx, archiveRecord("other", "2026-09-02T04:00:00Z")))
+	require.Equal(t, oldRecords, repo.data[settingKeyBackupRecords])
+	require.Error(t, svc.cleanupOldBackups(ctx, cfg))
+	require.Error(t, func() error { _, err := svc.GetSchedule(ctx); return err }())
+	repo.getValueErr = nil
+	repo.data[settingKeyBackupRecords] = "corrupted"
+	require.ErrorIs(t, svc.saveRecord(ctx, record), ErrBackupRecordsCorrupt)
+	require.Equal(t, "corrupted", repo.data[settingKeyBackupRecords])
+	repo.data[settingKeyBackupRecords] = oldRecords
+	repo.data[settingKeyBackupArchiveCheckpoint] = "bad date"
+	require.Error(t, svc.saveRecordWithArchive(ctx, archiveRecord("later", "2026-09-02T04:00:00Z"), cfg))
+	require.Equal(t, oldRecords, repo.data[settingKeyBackupRecords])
+}
+
+func TestBackupRetention_ScheduledExecutionAndZeroDays(t *testing.T) {
+	repo := newMockSettingRepo()
+	seedS3Config(t, repo)
+	svc := newTestBackupService(repo, &mockDumper{dumpData: []byte("backup")}, newMockObjectStore())
+	cfg := archiveSchedule(time.Now().UTC().Day())
+	cfg.RetainDays = 0
+	raw, err := json.Marshal(cfg)
+	require.NoError(t, err)
+	require.NoError(t, repo.Set(context.Background(), settingKeyBackupSchedule, string(raw)))
+	svc.runScheduledBackup()
+	records, err := svc.ListBackups(context.Background())
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	require.NotNil(t, records[0].MonthlyArchive)
+	require.Empty(t, records[0].ExpiresAt)
+	// The following successful backup is ordinary and must still honor zero days.
+	svc.runScheduledBackup()
+	records, err = svc.ListBackups(context.Background())
+	require.NoError(t, err)
+	require.Len(t, records, 2)
+	for _, record := range records {
+		require.Empty(t, record.ExpiresAt)
+	}
+}
+
+func TestBackupRetention_CrossInstanceRecordUpdates(t *testing.T) {
+	repo := newMockSettingRepo()
+	cache := &fakeLeaderLockCache{}
+	services := []*BackupService{
+		newTestBackupService(repo, &mockDumper{}, newMockObjectStore()),
+		newTestBackupService(repo, &mockDumper{}, newMockObjectStore()),
+	}
+	for _, svc := range services {
+		svc.SetLeaderLock(cache, nil)
+	}
+	var wg sync.WaitGroup
+	errors := make(chan error, 24)
+	for i := range 24 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			errors <- services[i%2].saveRecord(context.Background(), archiveRecord(fmt.Sprint(i), "2026-09-01T00:00:00Z"))
+		}()
+	}
+	wg.Wait()
+	close(errors)
+	for err := range errors {
+		require.NoError(t, err)
+	}
+	records, err := services[0].ListBackups(context.Background())
+	require.NoError(t, err)
+	require.Len(t, records, 24)
+}
+
+func TestBackupRetention_RecordLockWithSingleConnectionPool(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	defer db.Close()
+	db.SetMaxOpenConns(1)
+	repo := newMockSettingRepo()
+	svc := newTestBackupService(repo, &mockDumper{}, newMockObjectStore())
+	svc.SetLeaderLock(&fakeLeaderLockCache{}, db)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	record := archiveRecord("single-connection", "2026-09-01T00:00:00Z")
+	require.NoError(t, svc.saveRecord(ctx, record), "Redis locking must not occupy a database connection")
+	require.NoError(t, mock.ExpectationsWereMet())
+	svc.SetLeaderLock(nil, db)
+	require.ErrorContains(t, svc.saveRecord(ctx, record), "at least two database connections")
+}

backend/internal/service/backup_service.go

@@ -30,7 +30,6 @@ const (
 	settingKeyBackupSchedule = "backup_schedule"
 	settingKeyBackupRecords  = "backup_records"
 
-	maxBackupRecords           = 100
 	backupObjectCleanupTimeout = 2 * time.Minute
 
 	// backupScheduledLeaderLockKey gates the scheduled full-database backup so
@@ -50,12 +49,13 @@ const (
 )
 
 var (
-	ErrBackupS3NotConfigured = infraerrors.BadRequest("BACKUP_S3_NOT_CONFIGURED", "backup S3 storage is not configured")
-	ErrBackupNotFound        = infraerrors.NotFound("BACKUP_NOT_FOUND", "backup record not found")
-	ErrBackupInProgress      = infraerrors.Conflict("BACKUP_IN_PROGRESS", "a backup is already in progress")
-	ErrRestoreInProgress     = infraerrors.Conflict("RESTORE_IN_PROGRESS", "a restore is already in progress")
-	ErrBackupRecordsCorrupt  = infraerrors.InternalServer("BACKUP_RECORDS_CORRUPT", "backup records data is corrupted")
-	ErrBackupS3ConfigCorrupt = infraerrors.InternalServer("BACKUP_S3_CONFIG_CORRUPT", "backup S3 config data is corrupted")
+	ErrBackupS3NotConfigured  = infraerrors.BadRequest("BACKUP_S3_NOT_CONFIGURED", "backup S3 storage is not configured")
+	ErrBackupNotFound         = infraerrors.NotFound("BACKUP_NOT_FOUND", "backup record not found")
+	ErrBackupInProgress       = infraerrors.Conflict("BACKUP_IN_PROGRESS", "a backup is already in progress")
+	ErrRestoreInProgress      = infraerrors.Conflict("RESTORE_IN_PROGRESS", "a restore is already in progress")
+	ErrBackupRecordsCorrupt   = infraerrors.InternalServer("BACKUP_RECORDS_CORRUPT", "backup records data is corrupted")
+	ErrBackupS3ConfigCorrupt  = infraerrors.InternalServer("BACKUP_S3_CONFIG_CORRUPT", "backup S3 config data is corrupted")
+	ErrBackupArchiveProtected = infraerrors.Conflict("BACKUP_ARCHIVE_PROTECTED", "explicit confirmation is required to delete an archived backup")
 
 	// ErrSecretEncryptionKeyNotConfigured is returned when an S3 SecretAccessKey
 	// would be encrypted with an auto-generated (ephemeral) key. That key is
@@ -111,30 +111,33 @@ func (c *BackupS3Config) IsConfigured() bool {
 
 // BackupScheduleConfig 定时备份配置
 type BackupScheduleConfig struct {
-	Enabled     bool   `json:"enabled"`
-	CronExpr    string `json:"cron_expr"`    // cron 表达式，如 "0 2 * * *" 每天凌晨2点
-	RetainDays  int    `json:"retain_days"`  // 备份文件过期天数，默认14，0=不自动清理
-	RetainCount int    `json:"retain_count"` // 最多保留份数，0=不限制
+	Enabled        bool                        `json:"enabled"`
+	CronExpr       string                      `json:"cron_expr"`    // cron 表达式，如 "0 2 * * *" 每天凌晨2点
+	RetainDays     int                         `json:"retain_days"`  // 备份文件过期天数，默认14，0=不自动清理
+	RetainCount    int                         `json:"retain_count"` // 最多保留份数，0=不限制
+	MonthlyArchive *BackupMonthlyArchiveConfig `json:"monthly_archive,omitempty"`
 }
 
 // BackupRecord 备份记录
 type BackupRecord struct {
-	ID            string       `json:"id"`
-	Status        string       `json:"status"`      // pending, running, completed, failed
-	BackupType    string       `json:"backup_type"` // postgres
-	FileName      string       `json:"file_name"`
-	S3Key         string       `json:"s3_key"`
-	Parts         []BackupPart `json:"parts,omitempty"`
-	SizeBytes     int64        `json:"size_bytes"`
-	TriggeredBy   string       `json:"triggered_by"` // manual, scheduled
-	ErrorMsg      string       `json:"error_message,omitempty"`
-	StartedAt     string       `json:"started_at"`
-	FinishedAt    string       `json:"finished_at,omitempty"`
-	ExpiresAt     string       `json:"expires_at,omitempty"`     // 过期时间
-	Progress      string       `json:"progress,omitempty"`       // "dumping", "uploading", ""
-	RestoreStatus string       `json:"restore_status,omitempty"` // "", "running", "completed", "failed"
-	RestoreError  string       `json:"restore_error,omitempty"`
-	RestoredAt    string       `json:"restored_at,omitempty"`
+	ID               string                `json:"id"`
+	Status           string                `json:"status"`      // pending, running, completed, failed
+	BackupType       string                `json:"backup_type"` // postgres
+	FileName         string                `json:"file_name"`
+	S3Key            string                `json:"s3_key"`
+	Parts            []BackupPart          `json:"parts,omitempty"`
+	SizeBytes        int64                 `json:"size_bytes"`
+	TriggeredBy      string                `json:"triggered_by"` // manual, scheduled
+	ErrorMsg         string                `json:"error_message,omitempty"`
+	StartedAt        string                `json:"started_at"`
+	FinishedAt       string                `json:"finished_at,omitempty"`
+	ExpiresAt        string                `json:"expires_at,omitempty"`     // 过期时间
+	Progress         string                `json:"progress,omitempty"`       // "dumping", "uploading", ""
+	RestoreStatus    string                `json:"restore_status,omitempty"` // "", "running", "completed", "failed"
+	RestoreStartedAt string                `json:"restore_started_at,omitempty"`
+	RestoreError     string                `json:"restore_error,omitempty"`
+	RestoredAt       string                `json:"restored_at,omitempty"`
+	MonthlyArchive   *BackupMonthlyArchive `json:"monthly_archive,omitempty"`
 }
 
 // BackupDownloadPart 描述一个可下载的备份分卷。
@@ -228,6 +231,9 @@ func (s *BackupService) SetLeaderLock(lockCache LeaderLockCache, db *sql.DB) {
 // Start 启动定时备份调度器并清理孤立记录
 func (s *BackupService) Start() {
 	s.cronSched = cron.New()
+	// A recently interrupted job may still belong to a live peer. Revisit it
+	// after the maximum execution window rather than deleting it at startup.
+	_, _ = s.cronSched.AddFunc("@every 5m", s.recoverStaleRecords)
 	s.cronSched.Start()
 
 	// 清理重启后孤立的 running 记录
@@ -248,53 +254,78 @@ func (s *BackupService) Start() {
 	}
 }
 
-// recoverStaleRecords 启动时将孤立的 running 记录标记为 failed，并清理已上传对象。
+// recoverStaleRecords expires interrupted operations only after their maximum
+// runtime. Re-read, status persistence and object deletion share the writer lock.
 func (s *BackupService) recoverStaleRecords() {
-	loadCtx, loadCancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer loadCancel()
-
-	records, err := s.loadRecords(loadCtx)
+	ctx, cancel := context.WithTimeout(s.bgCtx, 5*time.Minute)
+	defer cancel()
+	ctx, release, err := s.lockBackupRecordUpdates(ctx)
 	if err != nil {
 		return
 	}
+	defer release()
+	records, err := s.loadRecordsLocked(ctx)
+	if err != nil {
+		return
+	}
+	now := time.Now()
+	changed := false
+	var stale []int
 	for i := range records {
-		if records[i].Status == "running" {
-			staleRecord := records[i]
+		if records[i].Status == "running" && records[i].MonthlyArchive == nil && backupOperationExpired(records[i].StartedAt, now) {
 			records[i].Status = "failed"
 			records[i].ErrorMsg = "interrupted by server restart"
 			records[i].Progress = ""
-			records[i].FinishedAt = time.Now().Format(time.RFC3339)
-			s.saveRecoveredRecord(&records[i])
-
-			if cleanupErr := s.cleanupStaleBackupObjects(&staleRecord); cleanupErr != nil {
-				records[i].ErrorMsg = fmt.Sprintf("interrupted by server restart; cleanup failed, manual deletion may be required: %v", cleanupErr)
-				s.saveRecoveredRecord(&records[i])
-				logger.LegacyPrintf("service.backup", "[Backup] failed to clean stale backup objects for %s: %v", records[i].ID, cleanupErr)
-			}
-			logger.LegacyPrintf("service.backup", "[Backup] recovered stale running record: %s", records[i].ID)
+			records[i].FinishedAt = now.Format(time.RFC3339)
+			changed = true
+			stale = append(stale, i)
 		}
 		if records[i].RestoreStatus == "running" {
+			if records[i].RestoreStartedAt == "" {
+				// Legacy restores have no operation timestamp. Give them a full
+				// grace period; the backup's creation date says nothing about it.
+				records[i].RestoreStartedAt = now.Format(time.RFC3339)
+				changed = true
+				continue
+			}
+			if !backupOperationExpired(records[i].RestoreStartedAt, now) {
+				continue
+			}
 			records[i].RestoreStatus = "failed"
 			records[i].RestoreError = "interrupted by server restart"
-			s.saveRecoveredRecord(&records[i])
-			logger.LegacyPrintf("service.backup", "[Backup] recovered stale restoring record: %s", records[i].ID)
+			changed = true
+		}
+	}
+	if !changed {
+		return
+	}
+	if err := s.saveRecordsLocked(ctx, records); err != nil {
+		logger.LegacyPrintf("service.backup", "[Backup] 保存中断记录失败: %v", err)
+		return // Never delete objects if the state transition did not commit.
+	}
+	for _, i := range stale {
+		if cleanupErr := s.cleanupStaleBackupObjects(ctx, &records[i]); cleanupErr != nil {
+			records[i].ErrorMsg = fmt.Sprintf("interrupted by server restart; cleanup failed, manual deletion may be required: %v", cleanupErr)
+			logger.LegacyPrintf("service.backup", "[Backup] failed to clean stale backup objects for %s: %v", records[i].ID, cleanupErr)
+		}
+	}
+	if len(stale) > 0 {
+		if err := s.saveRecordsLocked(ctx, records); err != nil {
+			logger.LegacyPrintf("service.backup", "[Backup] 保存中断清理结果失败: %v", err)
 		}
 	}
 }
 
-func (s *BackupService) saveRecoveredRecord(record *BackupRecord) {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	if err := s.saveRecord(ctx, record); err != nil {
-		logger.LegacyPrintf("service.backup", "[Backup] 保存恢复后的备份记录失败 %s: %v", record.ID, err)
-	}
+func backupOperationExpired(started string, now time.Time) bool {
+	at, err := time.Parse(time.RFC3339, started)
+	return err == nil && now.Sub(at) > backupScheduledLeaderLockTTL
 }
 
-func (s *BackupService) cleanupStaleBackupObjects(record *BackupRecord) error {
+func (s *BackupService) cleanupStaleBackupObjects(ctx context.Context, record *BackupRecord) error {
 	if len(backupObjectKeys(record)) == 0 {
 		return nil
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), backupObjectCleanupTimeout)
+	ctx, cancel := context.WithTimeout(ctx, backupObjectCleanupTimeout)
 	defer cancel()
 	return s.deleteBackupObjects(ctx, record)
 }
@@ -303,15 +334,19 @@ func (s *BackupService) cleanupStaleBackupObjects(record *BackupRecord) error {
 func (s *BackupService) Stop() {
 	s.shuttingDown.Store(true)
 
+	var cronStopped context.Context
 	s.cronMu.Lock()
 	if s.cronSched != nil {
-		s.cronSched.Stop()
+		cronStopped = s.cronSched.Stop()
 	}
 	s.cronMu.Unlock()
 
 	// 等待活跃备份/恢复完成（最多 5 分钟）
 	done := make(chan struct{})
 	go func() {
+		if cronStopped != nil {
+			<-cronStopped.Done()
+		}
 		s.wg.Wait()
 		close(done)
 	}()
@@ -419,17 +454,23 @@ func (s *BackupService) TestS3Connection(ctx context.Context, cfg BackupS3Config
 
 func (s *BackupService) GetSchedule(ctx context.Context) (*BackupScheduleConfig, error) {
 	raw, err := s.settingRepo.GetValue(ctx, settingKeyBackupSchedule)
-	if err != nil || raw == "" {
-		return &BackupScheduleConfig{}, nil
+	if err != nil && !errors.Is(err, ErrSettingNotFound) {
+		return nil, fmt.Errorf("load backup schedule: %w", err)
+	}
+	if raw == "" {
+		return &BackupScheduleConfig{RetainDays: 14, RetainCount: 10}, nil
 	}
 	var cfg BackupScheduleConfig
 	if err := json.Unmarshal([]byte(raw), &cfg); err != nil {
-		return &BackupScheduleConfig{}, nil
+		return nil, infraerrors.InternalServer("BACKUP_SCHEDULE_CORRUPT", "backup schedule data is corrupted")
 	}
 	return &cfg, nil
 }
 
 func (s *BackupService) UpdateSchedule(ctx context.Context, cfg BackupScheduleConfig) (*BackupScheduleConfig, error) {
+	if err := validateBackupRetention(&cfg); err != nil {
+		return nil, err
+	}
 	if cfg.Enabled && cfg.CronExpr == "" {
 		return nil, infraerrors.BadRequest("INVALID_CRON", "cron expression is required when schedule is enabled")
 	}
@@ -514,14 +555,18 @@ func (s *BackupService) runScheduledBackup() {
 	defer release()
 
 	// 读取定时备份配置中的过期天数
-	schedule, _ := s.GetSchedule(ctx)
+	schedule, err := s.GetSchedule(ctx)
+	if err != nil {
+		logger.LegacyPrintf("service.backup", "[Backup] 加载备份配置失败: %v", err)
+		return
+	}
 	expireDays := 14 // 默认14天过期
-	if schedule != nil && schedule.RetainDays > 0 {
+	if schedule != nil {
 		expireDays = schedule.RetainDays
 	}
 
 	logger.LegacyPrintf("service.backup", "[Backup] 开始执行定时备份, 过期天数: %d", expireDays)
-	record, err := s.CreateBackup(ctx, "scheduled", expireDays)
+	record, err := s.createBackup(ctx, "scheduled", expireDays, schedule)
 	if err != nil {
 		if errors.Is(err, ErrBackupInProgress) {
 			logger.LegacyPrintf("service.backup", "[Backup] 定时备份跳过: 已有备份正在进行中")
@@ -546,6 +591,10 @@ func (s *BackupService) runScheduledBackup() {
 // CreateBackup 创建全量数据库备份并上传到 S3。
 // expireDays: 备份过期天数，0=永不过期，默认14天
 func (s *BackupService) CreateBackup(ctx context.Context, triggeredBy string, expireDays int) (*BackupRecord, error) {
+	return s.createBackup(ctx, triggeredBy, expireDays, nil)
+}
+
+func (s *BackupService) createBackup(ctx context.Context, triggeredBy string, expireDays int, schedule *BackupScheduleConfig) (*BackupRecord, error) {
 	if s.shuttingDown.Load() {
 		return nil, infraerrors.ServiceUnavailable("SERVER_SHUTTING_DOWN", "server is shutting down")
 	}
@@ -578,7 +627,7 @@ func (s *BackupService) CreateBackup(ctx context.Context, triggeredBy string, ex
 
 	now := time.Now()
 	backupID := uuid.New().String()[:8]
-	fileName := fmt.Sprintf("%s_%s.sql.gz", s.dbCfg.DBName, now.Format("20060102_150405"))
+	fileName := fmt.Sprintf("%s_%s_%s.sql.gz", s.dbCfg.DBName, now.Format("20060102_150405"), backupID)
 	s3Key := s.buildS3Key(s3Cfg, fileName)
 
 	var expiresAt string
@@ -620,8 +669,8 @@ func (s *BackupService) CreateBackup(ctx context.Context, triggeredBy string, ex
 
 	record.Status = "completed"
 	record.FinishedAt = time.Now().Format(time.RFC3339)
-	if err := s.saveRecord(ctx, record); err != nil {
-		logger.LegacyPrintf("service.backup", "[Backup] 保存备份记录失败: %v", err)
+	if err := s.saveRecordWithArchive(ctx, record, schedule); err != nil {
+		return record, fmt.Errorf("save completed backup: %w", err)
 	}
 
 	return record, nil
@@ -667,7 +716,7 @@ func (s *BackupService) StartBackup(ctx context.Context, triggeredBy string, exp
 
 	now := time.Now()
 	backupID := uuid.New().String()[:8]
-	fileName := fmt.Sprintf("%s_%s.sql.gz", s.dbCfg.DBName, now.Format("20060102_150405"))
+	fileName := fmt.Sprintf("%s_%s_%s.sql.gz", s.dbCfg.DBName, now.Format("20060102_150405"), backupID)
 	s3Key := s.buildS3Key(s3Cfg, fileName)
 
 	var expiresAt string
@@ -966,8 +1015,12 @@ func (s *BackupService) StartRestore(ctx context.Context, backupID string) (*Bac
 		return nil, fmt.Errorf("init object store: %w", err)
 	}
 
-	record.RestoreStatus = "running"
-	_ = s.saveRecord(ctx, record)
+	// Re-read and reserve under the same lock used by cleanup/deletion. A prior
+	// read above may have become stale while S3 configuration was being loaded.
+	record, err = s.beginBackupRestore(ctx, backupID)
+	if err != nil {
+		return nil, err
+	}
 
 	launched = true
 	result := *record
@@ -985,7 +1038,7 @@ func (s *BackupService) StartRestore(ctx context.Context, backupID string) (*Bac
 				logger.LegacyPrintf("service.backup", "[Backup] restore panic recovered: %v", r)
 				record.RestoreStatus = "failed"
 				record.RestoreError = fmt.Sprintf("internal panic: %v", r)
-				_ = s.saveRecord(context.Background(), record)
+				_ = s.saveRestoreRecord(context.Background(), record)
 			}
 		}()
 		s.executeRestore(record, objectStore)
@@ -1004,19 +1057,19 @@ func (s *BackupService) executeRestore(record *BackupRecord, objectStore BackupO
 		if err != nil {
 			record.RestoreStatus = "failed"
 			record.RestoreError = err.Error()
-			_ = s.saveRecord(context.Background(), record)
+			_ = s.saveRestoreRecord(context.Background(), record)
 			return
 		}
 		defer func() { _ = cleanupBackupFiles(archivePath) }()
 		if err := s.restoreArchive(ctx, archivePath); err != nil {
 			record.RestoreStatus = "failed"
 			record.RestoreError = fmt.Sprintf("pg restore: %v", err)
-			_ = s.saveRecord(context.Background(), record)
+			_ = s.saveRestoreRecord(context.Background(), record)
 			return
 		}
 		record.RestoreStatus = "completed"
 		record.RestoredAt = time.Now().Format(time.RFC3339)
-		if err := s.saveRecord(context.Background(), record); err != nil {
+		if err := s.saveRestoreRecord(context.Background(), record); err != nil {
 			logger.LegacyPrintf("service.backup", "[Backup] 保存恢复记录失败: %v", err)
 		}
 		return
@@ -1026,7 +1079,7 @@ func (s *BackupService) executeRestore(record *BackupRecord, objectStore BackupO
 	if err != nil {
 		record.RestoreStatus = "failed"
 		record.RestoreError = fmt.Sprintf("S3 download failed: %v", err)
-		_ = s.saveRecord(context.Background(), record)
+		_ = s.saveRestoreRecord(context.Background(), record)
 		return
 	}
 	defer func() { _ = body.Close() }()
@@ -1035,7 +1088,7 @@ func (s *BackupService) executeRestore(record *BackupRecord, objectStore BackupO
 	if err != nil {
 		record.RestoreStatus = "failed"
 		record.RestoreError = fmt.Sprintf("gzip reader: %v", err)
-		_ = s.saveRecord(context.Background(), record)
+		_ = s.saveRestoreRecord(context.Background(), record)
 		return
 	}
 	defer func() { _ = gzReader.Close() }()
@@ -1043,13 +1096,13 @@ func (s *BackupService) executeRestore(record *BackupRecord, objectStore BackupO
 	if err := s.dumper.Restore(ctx, gzReader); err != nil {
 		record.RestoreStatus = "failed"
 		record.RestoreError = fmt.Sprintf("pg restore: %v", err)
-		_ = s.saveRecord(context.Background(), record)
+		_ = s.saveRestoreRecord(context.Background(), record)
 		return
 	}
 
 	record.RestoreStatus = "completed"
 	record.RestoredAt = time.Now().Format(time.RFC3339)
-	if err := s.saveRecord(context.Background(), record); err != nil {
+	if err := s.saveRestoreRecord(context.Background(), record); err != nil {
 		logger.LegacyPrintf("service.backup", "[Backup] 保存恢复记录失败: %v", err)
 	}
 }
@@ -1136,7 +1189,7 @@ func (s *BackupService) ListBackups(ctx context.Context) ([]BackupRecord, error)
 	}
 	// 倒序返回（最新在前）
 	sort.Slice(records, func(i, j int) bool {
-		return records[i].StartedAt > records[j].StartedAt
+		return backupStartedAfter(records[i], records[j])
 	})
 	return records, nil
 }
@@ -1155,8 +1208,20 @@ func (s *BackupService) GetBackupRecord(ctx context.Context, backupID string) (*
 }
 
 func (s *BackupService) DeleteBackup(ctx context.Context, backupID string) error {
-	s.recordsMu.Lock()
-	defer s.recordsMu.Unlock()
+	return s.deleteBackup(ctx, backupID, false)
+}
+
+// DeleteArchivedBackup requires the caller to explicitly confirm archive removal.
+func (s *BackupService) DeleteArchivedBackup(ctx context.Context, backupID string) error {
+	return s.deleteBackup(ctx, backupID, true)
+}
+
+func (s *BackupService) deleteBackup(ctx context.Context, backupID string, deleteArchived bool) error {
+	ctx, release, err := s.lockBackupRecordUpdates(ctx)
+	if err != nil {
+		return err
+	}
+	defer release()
 
 	records, err := s.loadRecordsLocked(ctx)
 	if err != nil {
@@ -1179,6 +1244,12 @@ func (s *BackupService) DeleteBackup(ctx context.Context, backupID string) error
 		// 后台上传仍可能依赖 Parts 计划；删除对象会让随后完成的记录引用失效卷。
 		return ErrBackupInProgress
 	}
+	if found.MonthlyArchive != nil && !deleteArchived {
+		return ErrBackupArchiveProtected
+	}
+	if found.RestoreStatus == "running" {
+		return ErrRestoreInProgress
+	}
 
 	// 从对象存储删除所有单文件或分卷对象。删除不完整时保留记录，便于重试。
 	if err := s.deleteBackupObjects(ctx, found); err != nil {
@@ -1301,7 +1372,10 @@ func (s *BackupService) loadRecords(ctx context.Context) ([]BackupRecord, error)
 // loadRecordsLocked 在已持有 recordsMu 锁的情况下加载记录
 func (s *BackupService) loadRecordsLocked(ctx context.Context) ([]BackupRecord, error) {
 	raw, err := s.settingRepo.GetValue(ctx, settingKeyBackupRecords)
-	if err != nil || raw == "" {
+	if err != nil && !errors.Is(err, ErrSettingNotFound) {
+		return nil, fmt.Errorf("load backup records: %w", err)
+	}
+	if raw == "" {
 		return nil, nil //nolint:nilnil // no records is a valid state
 	}
 	var records []BackupRecord
@@ -1322,27 +1396,64 @@ func (s *BackupService) saveRecordsLocked(ctx context.Context, records []BackupR
 
 // saveRecord 保存单条记录（带互斥锁保护）
 func (s *BackupService) saveRecord(ctx context.Context, record *BackupRecord) error {
-	s.recordsMu.Lock()
-	defer s.recordsMu.Unlock()
+	return s.saveRecordWithArchive(ctx, record, nil)
+}
 
-	records, _ := s.loadRecordsLocked(ctx)
+func (s *BackupService) saveRecordWithArchive(ctx context.Context, record *BackupRecord, schedule *BackupScheduleConfig) error {
+	ctx, release, err := s.lockBackupRecordUpdates(ctx)
+	if err != nil {
+		return err
+	}
+	defer release()
+
+	records, err := s.loadRecordsLocked(ctx)
+	if err != nil {
+		return err
+	}
+	snapshot := *record
+	for _, existing := range records {
+		if existing.ID == record.ID && existing.MonthlyArchive != nil {
+			// Preserve durable protection when a progress/restore writer or retry
+			// holds an older record snapshot.
+			snapshot.MonthlyArchive = existing.MonthlyArchive
+			snapshot.ExpiresAt = ""
+			break
+		}
+	}
+	checkpoint, err := s.assignMonthlyArchive(ctx, &snapshot, schedule)
+	if err != nil {
+		return err
+	}
 
 	// 更新已有记录或追加
 	found := false
 	for i := range records {
 		if records[i].ID == record.ID {
-			records[i] = *record
+			records[i] = snapshot
 			found = true
 			break
 		}
 	}
 	if !found {
-		records = append(records, *record)
+		records = append(records, snapshot)
 	}
 
-	// 限制记录数量
-	if len(records) > maxBackupRecords {
-		records = records[len(records)-maxBackupRecords:]
+	// Never discard metadata for retained objects, including permanent archives.
+	// The checkpoint and completed record must commit together so retries cannot
+	// consume a date without an archive, or archive the same date repeatedly.
+	if checkpoint != "" {
+		data, err := json.Marshal(records)
+		if err != nil {
+			return err
+		}
+		if err := s.settingRepo.SetMultiple(ctx, map[string]string{
+			settingKeyBackupRecords:           string(data),
+			settingKeyBackupArchiveCheckpoint: checkpoint,
+		}); err != nil {
+			return err
+		}
+		*record = snapshot
+		return nil
 	}
 
 	return s.saveRecordsLocked(ctx, records)
@@ -1353,8 +1464,11 @@ func (s *BackupService) cleanupOldBackups(ctx context.Context, schedule *BackupS
 		return nil
 	}
 
-	s.recordsMu.Lock()
-	defer s.recordsMu.Unlock()
+	ctx, release, err := s.lockBackupRecordUpdates(ctx)
+	if err != nil {
+		return err
+	}
+	defer release()
 
 	records, err := s.loadRecordsLocked(ctx)
 	if err != nil {
@@ -1363,29 +1477,49 @@ func (s *BackupService) cleanupOldBackups(ctx context.Context, schedule *BackupS
 
 	// 按时间倒序
 	sort.Slice(records, func(i, j int) bool {
-		return records[i].StartedAt > records[j].StartedAt
+		return backupStartedAfter(records[i], records[j])
 	})
 
 	var toDelete []BackupRecord
 	var toKeep []BackupRecord
 
-	for i, r := range records {
+	ordinaryCount, archiveCount := 0, 0
+	metadataChanged := false
+	for _, r := range records {
 		shouldDelete := false
-
-		// 按保留份数清理
-		if schedule.RetainCount > 0 && i >= schedule.RetainCount {
-			shouldDelete = true
+		if r.Status != "completed" {
+			toKeep = append(toKeep, r)
+			continue
 		}
+		if r.MonthlyArchive != nil {
+			if r.MonthlyArchive.RetainCount > 0 {
+				archiveCount++
+				// A new finite limit applies to the finite archive pool. Permanent
+				// archives remain protected even after the configuration changes.
+				if cfg := schedule.MonthlyArchive; cfg != nil && cfg.RetainCount > 0 && cfg.RetainCount != r.MonthlyArchive.RetainCount {
+					r.MonthlyArchive.RetainCount = cfg.RetainCount
+					metadataChanged = true
+				}
+				shouldDelete = archiveCount > r.MonthlyArchive.RetainCount
+			}
+		} else {
+			ordinaryCount++
 
-		// 按保留天数清理
-		if schedule.RetainDays > 0 && r.StartedAt != "" {
-			startedAt, err := time.Parse(time.RFC3339, r.StartedAt)
-			if err == nil && time.Since(startedAt) > time.Duration(schedule.RetainDays)*24*time.Hour {
+			// 按保留份数清理
+			if schedule.RetainCount > 0 && ordinaryCount > schedule.RetainCount {
 				shouldDelete = true
 			}
+
+			// 按保留天数清理
+			if schedule.RetainDays > 0 && r.StartedAt != "" {
+				startedAt, err := time.Parse(time.RFC3339, r.StartedAt)
+				if err == nil && time.Now().After(startedAt.AddDate(0, 0, schedule.RetainDays)) {
+					shouldDelete = true
+				}
+			}
 		}
 
-		if shouldDelete && r.Status == "completed" {
+		if shouldDelete && r.RestoreStatus != "running" {
 			toDelete = append(toDelete, r)
 		} else {
 			toKeep = append(toKeep, r)
@@ -1404,7 +1538,7 @@ func (s *BackupService) cleanupOldBackups(ctx context.Context, schedule *BackupS
 		deletedCount++
 	}
 
-	if len(toDelete) > 0 {
+	if len(toDelete) > 0 || metadataChanged {
 		if err := s.saveRecordsLocked(ctx, toKeep); err != nil {
 			cleanupErrs = append(cleanupErrs, fmt.Errorf("save backup records after cleanup: %w", err))
 		}

backend/internal/service/backup_service_test.go

@@ -1146,10 +1146,11 @@ func TestRecoverStaleRecords(t *testing.T) {
 	})
 	// 模拟一条孤立的恢复中记录
 	_ = svc.saveRecord(context.Background(), &BackupRecord{
-		ID:            "stale-2",
-		Status:        "completed",
-		RestoreStatus: "running",
-		StartedAt:     time.Now().Add(-1 * time.Hour).Format(time.RFC3339),
+		ID:               "stale-2",
+		Status:           "completed",
+		RestoreStatus:    "running",
+		RestoreStartedAt: time.Now().Add(-1 * time.Hour).Format(time.RFC3339),
+		StartedAt:        time.Now().Add(-1 * time.Hour).Format(time.RFC3339),
 	})
 
 	svc.recoverStaleRecords()